feat: ensure all cursors get closed

discue · Mar 3, 2024 · 55b0fd7 · 55b0fd7
1 parent 2d45742
commit 55b0fd7
Show file tree

Hide file tree

Showing 5 changed files with 52 additions and 19 deletions.
diff --git a/lib/one-to-few-ref-storage.js b/lib/one-to-few-ref-storage.js
@@ -2,6 +2,7 @@ const { createTracer } = require('@discue/open-telemetry-tracing')
 const { EQUALS_ANY_OF, EQUALS } = require('./aggregations.js')
 const Base = require('./simple-resource-storage.js')
 const { name } = require('../package.json')
+const { toArrayAndClose } = require('./safe-cursor.js')
 
 /**
  * @private
@@ -163,7 +164,8 @@ module.exports = class extends Base {
                 EQUALS_ANY_OF(this._resourceName, references)
             ]
 
-            return collection.aggregate(stages).toArray()
+            const cursor = collection.aggregate(stages)
+            return toArrayAndClose(cursor)
         })
     }
 

diff --git a/lib/one-to-few-resource-storage.js b/lib/one-to-few-resource-storage.js
@@ -4,6 +4,7 @@ const Base = require('./simple-resource-storage.js')
 const { Timestamp } = require('mongodb')
 const { createTracer } = require('@discue/open-telemetry-tracing')
 const { name } = require('../package.json')
+const { getFirstAndClose, toArrayAndClose } = require('./safe-cursor.js')
 
 /**
  * @private
@@ -114,12 +115,8 @@ module.exports = class extends Base {
                 aggregationStages.push(PROJECT(projection))
             }
 
-            const cursor = await collection.aggregate(aggregationStages)
-            try {
-                return await cursor.next()
-            } finally {
-                await cursor.close()
-            }
+            const cursor = collection.aggregate(aggregationStages)
+            return getFirstAndClose(cursor)
         })
     }
 
@@ -148,7 +145,8 @@ module.exports = class extends Base {
                 aggregationStages.push(PROJECT(projection))
             }
 
-            return collection.aggregate(aggregationStages).toArray()
+            const cursor = collection.aggregate(aggregationStages)
+            return toArrayAndClose(cursor)
         })
     }
 

diff --git a/lib/one-to-many-resource-storage.js b/lib/one-to-many-resource-storage.js
@@ -5,6 +5,7 @@ const usageEventTrigger = require('./usage-event-trigger.js')
 const { createTracer } = require('@discue/open-telemetry-tracing')
 const { name } = require('../package.json')
 const SpanStatusCode = require('@discue/open-telemetry-tracing/status-codes')
+const { toArrayAndClose } = require('./safe-cursor.js')
 
 /**
  * @private
@@ -168,8 +169,8 @@ module.exports = class extends Base {
     async _getResourceWithLookup(resourceIds, options = {}) {
         const collection = await this._getParentCollection()
         const aggregationStages = this._getNestedLookupStages(resourceIds, options)
-        const result = await collection.aggregate(aggregationStages).toArray()
-
+        const cursor = collection.aggregate(aggregationStages)
+        const result = await toArrayAndClose(cursor)
         return result?.at(0) ?? null
     }
 
@@ -274,7 +275,8 @@ module.exports = class extends Base {
 
             const collection = await this._getParentCollection()
             const aggregationStages = this._getNestedLookupStages(resourceIds, { withMetadata, projection })
-            const result = await collection.aggregate(aggregationStages).toArray()
+            const cursor = collection.aggregate(aggregationStages)
+            const result = await toArrayAndClose(cursor)
 
             if (result.length === 0) {
                 return result

diff --git a/lib/safe-cursor.js b/lib/safe-cursor.js
@@ -0,0 +1,29 @@
+
+
+/**
+ * 
+ * @param {import('mongodb').AbstractCursor} cursor 
+ * @returns {Promise}
+ */
+module.exports.toArrayAndClose = async function (cursor) {
+    try {
+        const array = await cursor.toArray()
+        return array
+    } finally {
+        await cursor.close()
+    }
+}
+
+/**
+ * 
+ * @param {import('mongodb').AbstractCursor} cursor 
+ * @returns {Promise}
+ */
+module.exports.getFirstAndClose = async function (cursor) {
+    try {
+        const first = await cursor.next()
+        return first
+    } finally {
+        await cursor.close()
+    }
+}
diff --git a/lib/simple-resource-storage.js b/lib/simple-resource-storage.js
@@ -4,6 +4,7 @@ const eventTrigger = require('./usage-event-trigger.js')
 const { createTracer } = require('@discue/open-telemetry-tracing')
 const { name } = require('../package.json')
 const { getSingleLookupPipeline, joinAndQueryChildResourcesPipeline } = require('./lookup-pipeline.js')
+const { toArrayAndClose, getFirstAndClose } = require('./safe-cursor.js')
 
 /**
  * @private
@@ -259,11 +260,7 @@ module.exports = class {
             }
 
             const cursor = collection.aggregate(aggregationStages)
-            try {
-                return await cursor.next()
-            } finally {
-                await cursor.close()
-            }
+            return getFirstAndClose(cursor)
         })
     }
 
@@ -289,7 +286,8 @@ module.exports = class {
                 aggregationStages.push(PROJECT(projection))
             }
 
-            return collection.aggregate(aggregationStages).toArray()
+            const cursor = collection.aggregate(aggregationStages)
+            return toArrayAndClose(cursor)
         })
     }
 
@@ -325,7 +323,9 @@ module.exports = class {
             lookupPipeline.push(...childResourcesPipeline)
 
             const collection = await this._getCollection()
-            const result = await collection.aggregate(lookupPipeline).toArray()
+
+            const cursor = collection.aggregate(lookupPipeline)
+            const result = await toArrayAndClose(cursor)
 
             if (result.length === 0) {
                 return result
@@ -352,7 +352,9 @@ module.exports = class {
     async find(aggregations = []) {
         return withActiveSpan(`${name}#find-simple-resources`, { resourceName: this._collectionName, databaseName: this._databaseName }, async () => {
             const collection = await this._getCollection()
-            return collection.aggregate(aggregations).toArray()
+
+            const cursor = collection.aggregate(aggregations)
+            return toArrayAndClose(cursor)
         })
     }