Close neo4j driver sessions

We had this error in our neo4j pod recently:

```
2019-12-02 08:29:42.680+0000 ERROR Unable to schedule bolt session 'bolt-1018230' for execution since there are no available threads to serve it at the moment. You can retry at a later time or consider increasing max thread pool size for bolt connector(s).
2019-12-02 08:29:42.680+0000 ERROR Unable to schedule bolt session 'bolt-1018224' for execution since there are no available threads to serve it at the moment. You can retry at a later time or consider increasing max thread pool size for bolt connector(s).
2019-12-02 08:29:42.681+0000 ERROR Unable to schedule bolt session 'bolt-1018352' for execution since there are no available threads to serve it at the moment. You can retry at a later time or consider increasing max thread pool size for bolt connector(s).
2019-12-02 08:29:42.682+0000 ERROR Unable to schedule bolt session 'bolt-1018243' for execution since there are no available threads to serve it at the moment. You can retry at a later time or consider increasing max thread pool size for bolt connector(s).
```

Apparently the default is 400 threads. So we must have a leak somewhere.

backend/src/jwt/decode.js

@@ -11,15 +11,21 @@ export default async (driver, authorizationHeader) => {
   } catch (err) {
     return null
   }
-  const session = driver.session()
   const query = `
     MATCH (user:User {id: $id, deleted: false, disabled: false })
     SET user.lastActiveAt = toString(datetime())
     RETURN user {.id, .slug, .name, .avatar, .email, .role, .disabled, .actorId}
     LIMIT 1
   `
-  const result = await session.run(query, { id })
+  const session = driver.session()
+  let result
+
+  try {
+    result = await session.run(query, { id })
+  } finally {
     session.close()
+  }
+
   const [currentUser] = await result.records.map(record => {
     return record.get('user')
   })

backend/src/middleware/hashtags/hashtagsMiddleware.js

@@ -3,7 +3,6 @@ import extractHashtags from '../hashtags/extractHashtags'
 const updateHashtagsOfPost = async (postId, hashtags, context) => {
   if (!hashtags.length) return
 
-  const session = context.driver.session()
   // We need two Cypher statements, because the 'MATCH' in the 'cypherDeletePreviousRelations' statement
   //  functions as an 'if'. In case there is no previous relation, the rest of the commands are omitted
   //  and no new Hashtags and relations will be created.
@@ -19,6 +18,8 @@ const updateHashtagsOfPost = async (postId, hashtags, context) => {
     MERGE (p)-[:TAGGED]->(t)
     RETURN p, t
     `
+  const session = context.driver.session()
+  try {
     await session.run(cypherDeletePreviousRelations, {
       postId,
     })
@@ -26,7 +27,9 @@ const updateHashtagsOfPost = async (postId, hashtags, context) => {
       postId,
       hashtags,
     })
+  } finally {
     session.close()
+  }
 }
 
 const handleContentDataOfPost = async (resolve, root, args, context, resolveInfo) => {

backend/src/middleware/notifications/notificationsMiddleware.js

@@ -1,15 +1,19 @@
 import extractMentionedUsers from './mentions/extractMentionedUsers'
 
 const postAuthorOfComment = async (comment, { context }) => {
-  const session = context.driver.session()
   const cypherFindUser = `
     MATCH (user: User)-[:WROTE]->(:Post)<-[:COMMENTS]-(:Comment { id: $commentId })
     RETURN user { .id }
     `
-  const result = await session.run(cypherFindUser, {
+  const session = context.driver.session()
+  let result
+  try {
+    result = await session.run(cypherFindUser, {
       commentId: comment.id,
     })
+  } finally {
     session.close()
+  }
   const [postAuthor] = await result.records.map(record => {
     return record.get('user')
   })
@@ -31,7 +35,6 @@ const notifyUsers = async (label, id, idsOfUsers, reason, context) => {
     throw new Error('Notification does not fit the reason!')
   }
 
-  const session = context.driver.session()
   let cypher
   switch (reason) {
     case 'mentioned_in_post': {
@@ -85,12 +88,16 @@ const notifyUsers = async (label, id, idsOfUsers, reason, context) => {
       break
     }
   }
+  const session = context.driver.session()
+  try {
     await session.run(cypher, {
       id,
       idsOfUsers,
       reason,
     })
+  } finally {
     session.close()
+  }
 }
 
 const handleContentDataOfPost = async (resolve, root, args, context, resolveInfo) => {
@@ -123,15 +130,19 @@ const handleCreateComment = async (resolve, root, args, context, resolveInfo) =>
   const comment = await handleContentDataOfComment(resolve, root, args, context, resolveInfo)
 
   if (comment) {
-    const session = context.driver.session()
     const cypherFindUser = `
     MATCH (user: User)-[:WROTE]->(:Post)<-[:COMMENTS]-(:Comment { id: $commentId })
     RETURN user { .id }
     `
-    const result = await session.run(cypherFindUser, {
+    const session = context.driver.session()
+    let result
+    try {
+      result = await session.run(cypherFindUser, {
         commentId: comment.id,
       })
+    } finally {
       session.close()
+    }
     const [postAuthor] = await result.records.map(record => {
       return record.get('user')
     })

backend/src/middleware/permissionsMiddleware.js

@@ -45,8 +45,8 @@ const isAuthor = rule({
   cache: 'no_cache',
 })(async (_parent, args, { user, driver }) => {
   if (!user) return false
-  const session = driver.session()
   const { id: resourceId } = args
+  const session = driver.session()
   try {
     const result = await session.run(
       `

backend/src/middleware/sluggifyMiddleware.js

@@ -3,11 +3,14 @@ import uniqueSlug from './slugify/uniqueSlug'
 const isUniqueFor = (context, type) => {
   return async slug => {
     const session = context.driver.session()
+    try {
       const response = await session.run(`MATCH(p:${type} {slug: $slug }) return p.slug`, {
         slug,
       })
-    session.close()
       return response.records.length === 0
+    } finally {
+      session.close()
+    }
   }
 }
 

backend/src/middleware/validation/validationMiddleware.js

@@ -13,6 +13,7 @@ const validateCommentCreation = async (resolve, root, args, context, info) => {
     throw new UserInputError(`Comment must be at least ${COMMENT_MIN_LENGTH} character long!`)
   }
   const session = context.driver.session()
+  try {
     const postQueryRes = await session.run(
       `
     MATCH (post:Post {id: $postId})
@@ -21,7 +22,6 @@ const validateCommentCreation = async (resolve, root, args, context, info) => {
         postId,
       },
     )
-  session.close()
     const [post] = postQueryRes.records.map(record => {
       return record.get('post')
     })
@@ -31,6 +31,9 @@ const validateCommentCreation = async (resolve, root, args, context, info) => {
     } else {
       return resolve(root, args, context, info)
     }
+  } finally {
+    session.close()
+  }
 }
 
 const validateUpdateComment = async (resolve, root, args, context, info) => {
@@ -62,6 +65,7 @@ const validateReport = async (resolve, root, args, context, info) => {
   const { user, driver } = context
   if (resourceId === user.id) throw new Error('You cannot report yourself!')
   const session = driver.session()
+  try {
     const reportQueryRes = await session.run(
       `
       MATCH (:User {id:$submitterId})-[:REPORTED]->(resource {id:$resourceId}) 
@@ -72,7 +76,6 @@ const validateReport = async (resolve, root, args, context, info) => {
         submitterId: user.id,
       },
     )
-  session.close()
     const [existingReportedResource] = reportQueryRes.records.map(record => {
       return {
         label: record.get('label'),
@@ -81,6 +84,9 @@ const validateReport = async (resolve, root, args, context, info) => {
 
     if (existingReportedResource) throw new Error(`${existingReportedResource.label}`)
     return resolve(root, args, context, info)
+  } finally {
+    session.close()
+  }
 }
 
 export default {

backend/src/schema/resolvers/comments.js

@@ -13,6 +13,7 @@ export default {
       params.id = params.id || uuid()
 
       const session = context.driver.session()
+      try {
         const createCommentCypher = `
         MATCH (post:Post {id: $postId})
         MATCH (author:User {id: $userId})
@@ -28,14 +29,17 @@ export default {
           postId,
           params,
         })
-      session.close()
 
         const [comment] = transactionRes.records.map(record => record.get('comment').properties)
 
         return comment
+      } finally {
+        session.close()
+      }
     },
     UpdateComment: async (_parent, params, context, _resolveInfo) => {
       const session = context.driver.session()
+      try {
         const updateCommentCypher = `
         MATCH (comment:Comment {id: $params.id})
         SET comment += $params
@@ -43,12 +47,15 @@ export default {
         RETURN comment
       `
         const transactionRes = await session.run(updateCommentCypher, { params })
-      session.close()
         const [comment] = transactionRes.records.map(record => record.get('comment').properties)
         return comment
+      } finally {
+        session.close()
+      }
     },
     DeleteComment: async (_parent, args, context, _resolveInfo) => {
       const session = context.driver.session()
+      try {
         const transactionRes = await session.run(
           `
         MATCH (comment:Comment {id: $commentId})
@@ -59,9 +66,11 @@ export default {
       `,
           { commentId: args.id },
         )
-      session.close()
         const [comment] = transactionRes.records.map(record => record.get('comment').properties)
         return comment
+      } finally {
+        session.close()
+      }
     },
   },
   Comment: {

backend/src/schema/resolvers/donations.js

@@ -2,8 +2,8 @@ export default {
   Mutation: {
     UpdateDonations: async (_parent, params, context, _resolveInfo) => {
       const { driver } = context
-      const session = driver.session()
       let donations
+      const session = driver.session()
       const writeTxResultPromise = session.writeTransaction(async txc => {
         const updateDonationsTransactionResponse = await txc.run(
           ` 

backend/src/schema/resolvers/helpers/createPasswordReset.js

@@ -4,7 +4,6 @@ export default async function createPasswordReset(options) {
   const { driver, nonce, email, issuedAt = new Date() } = options
   const normalizedEmail = normalizeEmail(email)
   const session = driver.session()
-  let response = {}
   try {
     const cypher = `
       MATCH (u:User)-[:PRIMARY_EMAIL]->(e:EmailAddress {email:$email})
@@ -23,9 +22,8 @@ export default async function createPasswordReset(options) {
       const { name } = record.get('u').properties
       return { email, nonce, name }
     })
-    response = records[0] || {}
+    return records[0] || {}
   } finally {
     session.close()
   }
-  return response
 }

backend/src/schema/resolvers/moderation.js

@@ -12,13 +12,16 @@ export default {
       RETURN resource {.id}
       `
       const session = driver.session()
+      try {
         const res = await session.run(cypher, { id, userId })
-      session.close()
         const [resource] = res.records.map(record => {
           return record.get('resource')
         })
         if (!resource) return null
         return resource.id
+      } finally {
+        session.close()
+      }
     },
     enable: async (object, params, { user, driver }) => {
       const { id } = params
@@ -29,13 +32,16 @@ export default {
       RETURN resource {.id}
       `
       const session = driver.session()
+      try {
         const res = await session.run(cypher, { id })
-      session.close()
         const [resource] = res.records.map(record => {
           return record.get('resource')
         })
         if (!resource) return null
         return resource.id
+      } finally {
+        session.close()
+      }
     },
   },
 }

backend/src/schema/resolvers/notifications.js

@@ -18,7 +18,7 @@ export default {
     notifications: async (_parent, args, context, _resolveInfo) => {
       const { user: currentUser } = context
       const session = context.driver.session()
-      let notifications, whereClause, orderByClause
+      let whereClause, orderByClause
 
       switch (args.read) {
         case true:
@@ -42,7 +42,6 @@ export default {
       }
       const offset = args.offset && typeof args.offset === 'number' ? `SKIP ${args.offset}` : ''
       const limit = args.first && typeof args.first === 'number' ? `LIMIT ${args.first}` : ''
-      try {
       const cypher = `
         MATCH (resource {deleted: false, disabled: false})-[notification:NOTIFIED]->(user:User {id:$id})
         ${whereClause}
@@ -50,19 +49,18 @@ export default {
         ${orderByClause}
         ${offset} ${limit}
         `
+      try {
         const result = await session.run(cypher, { id: currentUser.id })
-        notifications = await result.records.map(transformReturnType)
+        return result.records.map(transformReturnType)
       } finally {
         session.close()
       }
-      return notifications
     },
   },
   Mutation: {
     markAsRead: async (parent, args, context, resolveInfo) => {
       const { user: currentUser } = context
       const session = context.driver.session()
-      let notification
       try {
         const cypher = `
         MATCH (resource {id: $resourceId})-[notification:NOTIFIED {read: FALSE}]->(user:User {id:$id})
@@ -71,11 +69,10 @@ export default {
         `
         const result = await session.run(cypher, { resourceId: args.id, id: currentUser.id })
         const notifications = await result.records.map(transformReturnType)
-        notification = notifications[0]
+        return notifications[0]
       } finally {
         session.close()
       }
-      return notification
     },
   },
   NOTIFIED: {

backend/src/schema/resolvers/passwordReset.js

@@ -9,7 +9,6 @@ export default {
       return createPasswordReset({ driver, nonce, email })
     },
     resetPassword: async (_parent, { email, nonce, newPassword }, { driver }) => {
-      const session = driver.session()
       const stillValid = new Date()
       stillValid.setDate(stillValid.getDate() - 1)
       const encryptedNewPassword = await bcrypt.hashSync(newPassword, 10)
@@ -21,6 +20,8 @@ export default {
       SET u.encryptedPassword = $encryptedNewPassword
       RETURN pr
       `
+      const session = driver.session()
+      try {
         const transactionRes = await session.run(cypher, {
           stillValid,
           email,
@@ -29,8 +30,10 @@ export default {
         })
         const [reset] = transactionRes.records.map(record => record.get('pr'))
         const response = !!(reset && reset.properties.usedAt)
-      session.close()
         return response
+      } finally {
+        session.close()
+      }
     },
   },
 }

backend/src/schema/resolvers/passwordReset.spec.js

@@ -15,10 +15,13 @@ let variables
 
 const getAllPasswordResets = async () => {
   const session = driver.session()
+  try {
     const transactionRes = await session.run('MATCH (r:PasswordReset) RETURN r')
     const resets = transactionRes.records.map(record => record.get('r'))
-  session.close()
     return resets
+  } finally {
+    session.close()
+  }
 }
 
 beforeEach(() => {

backend/src/schema/resolvers/posts.js

@@ -54,37 +54,41 @@ export default {
       return neo4jgraphql(object, params, context, resolveInfo)
     },
     PostsEmotionsCountByEmotion: async (object, params, context, resolveInfo) => {
-      const session = context.driver.session()
       const { postId, data } = params
+      const session = context.driver.session()
+      try {
         const transactionRes = await session.run(
           `MATCH (post:Post {id: $postId})<-[emoted:EMOTED {emotion: $data.emotion}]-()
         RETURN COUNT(DISTINCT emoted) as emotionsCount
         `,
           { postId, data },
         )
-      session.close()
 
         const [emotionsCount] = transactionRes.records.map(record => {
           return record.get('emotionsCount').low
         })
-
         return emotionsCount
+      } finally {
+        session.close()
+      }
     },
     PostsEmotionsByCurrentUser: async (object, params, context, resolveInfo) => {
-      const session = context.driver.session()
       const { postId } = params
+      const session = context.driver.session()
+      try {
         const transactionRes = await session.run(
           `MATCH (user:User {id: $userId})-[emoted:EMOTED]->(post:Post {id: $postId})
         RETURN collect(emoted.emotion) as emotion`,
           { userId: context.user.id, postId },
         )
 
-      session.close()
-
         const [emotions] = transactionRes.records.map(record => {
           return record.get('emotion')
         })
         return emotions
+      } finally {
+        session.close()
+      }
     },
   },
   Mutation: {
@@ -93,8 +97,6 @@ export default {
       delete params.categoryIds
       params = await fileUpload(params, { file: 'imageUpload', url: 'image' })
       params.id = params.id || uuid()
-      let post
-
       const createPostCypher = `CREATE (post:Post {params})
         SET post.createdAt = toString(datetime())
         SET post.updatedAt = toString(datetime())
@@ -113,7 +115,7 @@ export default {
       try {
         const transactionRes = await session.run(createPostCypher, createPostVariables)
         const posts = transactionRes.records.map(record => record.get('post').properties)
-        post = posts[0]
+        return posts[0]
       } catch (e) {
         if (e.code === 'Neo.ClientError.Schema.ConstraintValidationFailed')
           throw new UserInputError('Post with this slug already exists!')
@@ -121,20 +123,19 @@ export default {
       } finally {
         session.close()
       }
-
-      return post
     },
     UpdatePost: async (_parent, params, context, _resolveInfo) => {
       const { categoryIds } = params
       delete params.categoryIds
       params = await fileUpload(params, { file: 'imageUpload', url: 'image' })
-      const session = context.driver.session()
       let updatePostCypher = `MATCH (post:Post {id: $params.id})
       SET post += $params
       SET post.updatedAt = toString(datetime())
       WITH post
       `
 
+      const session = context.driver.session()
+      try {
         if (categoryIds && categoryIds.length) {
           const cypherDeletePreviousRelations = `
           MATCH (post:Post { id: $params.id })-[previousRelations:CATEGORIZED]->(category:Category)
@@ -159,14 +160,15 @@ export default {
         const [post] = transactionRes.records.map(record => {
           return record.get('post').properties
         })
-
-      session.close()
-
         return post
+      } finally {
+        session.close()
+      }
     },
 
     DeletePost: async (object, args, context, resolveInfo) => {
       const session = context.driver.session()
+      try {
         // we cannot set slug to 'UNAVAILABE' because of unique constraints
         const transactionRes = await session.run(
           `
@@ -182,21 +184,24 @@ export default {
       `,
           { postId: args.id },
         )
-      session.close()
         const [post] = transactionRes.records.map(record => record.get('post').properties)
         return post
+      } finally {
+        session.close()
+      }
     },
     AddPostEmotions: async (object, params, context, resolveInfo) => {
-      const session = context.driver.session()
       const { to, data } = params
       const { user } = context
+      const session = context.driver.session()
+      try {
         const transactionRes = await session.run(
           `MATCH (userFrom:User {id: $user.id}), (postTo:Post {id: $to.id})
         MERGE (userFrom)-[emotedRelation:EMOTED {emotion: $data.emotion}]->(postTo)
         RETURN userFrom, postTo, emotedRelation`,
           { user, to, data },
         )
-      session.close()
+
         const [emoted] = transactionRes.records.map(record => {
           return {
             from: { ...record.get('userFrom').properties },
@@ -205,18 +210,21 @@ export default {
           }
         })
         return emoted
+      } finally {
+        session.close()
+      }
     },
     RemovePostEmotions: async (object, params, context, resolveInfo) => {
-      const session = context.driver.session()
       const { to, data } = params
       const { id: from } = context.user
+      const session = context.driver.session()
+      try {
         const transactionRes = await session.run(
           `MATCH (userFrom:User {id: $from})-[emotedRelation:EMOTED {emotion: $data.emotion}]->(postTo:Post {id: $to.id})
         DELETE emotedRelation
         RETURN userFrom, postTo`,
           { from, to, data },
         )
-      session.close()
         const [emoted] = transactionRes.records.map(record => {
           return {
             from: { ...record.get('userFrom').properties },
@@ -225,6 +233,9 @@ export default {
           }
         })
         return emoted
+      } finally {
+        session.close()
+      }
     },
     pinPost: async (_parent, params, context, _resolveInfo) => {
       let pinnedPostWithNestedAttributes
@@ -242,6 +253,7 @@ export default {
         )
         return deletePreviousRelationsResponse.records.map(record => record.get('post').properties)
       })
+      try {
         await writeTxResultPromise
 
         writeTxResultPromise = session.writeTransaction(async transaction => {
@@ -260,7 +272,6 @@ export default {
             pinnedAt: record.get('pinnedAt'),
           }))
         })
-      try {
         const [transactionResult] = await writeTxResultPromise
         const { pinnedPost, pinnedAt } = transactionResult
         pinnedPostWithNestedAttributes = {

backend/src/schema/resolvers/reports.js

@@ -5,6 +5,7 @@ export default {
       const { resourceId, reasonCategory, reasonDescription } = params
       const { driver, user } = context
       const session = driver.session()
+      try {
         const writeTxResultPromise = session.writeTransaction(async txc => {
           const reportRelationshipTransactionResponse = await txc.run(
             `
@@ -29,7 +30,6 @@ export default {
             type: record.get('type'),
           }))
         })
-      try {
         const txResult = await writeTxResultPromise
         if (!txResult[0]) return null
         const { report, submitter, resource, type } = txResult[0]
@@ -61,7 +61,6 @@ export default {
   Query: {
     reports: async (_parent, params, context, _resolveInfo) => {
       const { driver } = context
-      const session = driver.session()
       let response
       let orderByClause
       switch (params.orderBy) {
@@ -74,6 +73,7 @@ export default {
         default:
           orderByClause = ''
       }
+      const session = driver.session()
       try {
         const cypher = `
           MATCH (submitter:User)-[report:REPORTED]->(resource)

backend/src/schema/resolvers/shout.js

@@ -4,6 +4,7 @@ export default {
       const { id, type } = params
 
       const session = context.driver.session()
+      try {
         const transactionRes = await session.run(
           `MATCH (node {id: $id})<-[:WROTE]-(userWritten:User), (user:User {id: $userId})
           WHERE $type IN labels(node) AND NOT userWritten.id = $userId
@@ -20,15 +21,16 @@ export default {
           return record.get('isShouted')
         })
 
-      session.close()
-
         return isShouted
+      } finally {
+        session.close()
+      }
     },
 
     unshout: async (_object, params, context, _resolveInfo) => {
       const { id, type } = params
       const session = context.driver.session()
-
+      try {
         const transactionRes = await session.run(
           `MATCH (user:User {id: $userId})-[relation:SHOUTED]->(node {id: $id})
           WHERE $type IN labels(node)
@@ -43,9 +45,10 @@ export default {
         const [isShouted] = transactionRes.records.map(record => {
           return record.get('isShouted')
         })
-      session.close()
-
         return isShouted
+      } finally {
+        session.close()
+      }
     },
   },
 }

backend/src/schema/resolvers/statistics.js

@@ -33,10 +33,10 @@ export default {
          * Note: invites count is calculated this way because invitation codes are not in use yet
          */
         response.countInvites = response.countEmails - response.countUsers
+        return response
       } finally {
         session.close()
       }
-      return response
     },
   },
 }

backend/src/schema/resolvers/user_management.js

@@ -24,6 +24,7 @@ export default {
       // }
       email = normalizeEmail(email)
       const session = driver.session()
+      try {
         const result = await session.run(
           `
         MATCH (user:User {deleted: false})-[:PRIMARY_EMAIL]->(e:EmailAddress {email: $userEmail})
@@ -31,7 +32,6 @@ export default {
       `,
           { userEmail: email },
         )
-      session.close()
         const [currentUser] = await result.records.map(record => {
           return record.get('user')
         })
@@ -48,6 +48,9 @@ export default {
         } else {
           throw new AuthenticationError('Incorrect email address or password.')
         }
+      } finally {
+        session.close()
+      }
     },
     changePassword: async (_, { oldPassword, newPassword }, { driver, user }) => {
       const currentUser = await instance.find('User', user.id)

backend/src/seed/factories/index.js

@@ -27,8 +27,8 @@ const factories = {
 
 export const cleanDatabase = async (options = {}) => {
   const { driver = getDriver() } = options
-  const session = driver.session()
   const cypher = 'MATCH (n) DETACH DELETE n'
+  const session = driver.session()
   try {
     return await session.run(cypher)
   } finally {