Generalize StorageBackedQueue

StorageBackedQueue can wrap around a given queue to provide persistence
in a shared storage queue.
Allow parellel execution in subscribe, unsubscribe and done.
Improve flush by avoiding update visibility timeout in storage.

Author: qtomlinson

.eslintrc.json

@@ -1,7 +1,7 @@
 {
   "env": {
     "browser": false,
-    "es6": true,
+    "es2021": true,
     "node": true,
     "mocha": true
   },

ghcrawler/providers/queuing/storageBackedInMemoryQueueManager.js

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: MIT
 
 const AttenuatedQueue = require('./attenuatedQueue')
-const StorageBackedInMemoryQueue = require('./storageBackedInMemoryQueue')
+const StorageBackedQueue = require('./storageBackedQueue')
 const InMemoryCrawlQueue = require('./inmemorycrawlqueue')
 
 class StorageBackedInMemoryQueueManager {
@@ -12,7 +12,7 @@ class StorageBackedInMemoryQueueManager {
   createQueueChain(name, options) {
     const storageQueue = this._storageQueueManager.createQueue(name, options)
     const inMemoryQueue = new InMemoryCrawlQueue(name, options)
-    const queue = StorageBackedInMemoryQueue.create(inMemoryQueue, storageQueue, options)
+    const queue = StorageBackedQueue.create(inMemoryQueue, storageQueue, options)
     return new AttenuatedQueue(queue, options)
   }
 }

…rs/queuing/storageBackedInMemoryQueue.js …/providers/queuing/storageBackedQueue.jsghcrawler/providers/queuing/storageBackedInMemoryQueue.js renamed to ghcrawler/providers/queuing/storageBackedQueue.js ghcrawler/providers/queuing/storageBackedInMemoryQueue.js renamed to ghcrawler/providers/queuing/storageBackedQueue.js

@@ -6,10 +6,10 @@ const NestedQueue = require('./nestedQueue')
 const VISIBILITY_TIMEOUT_TO_REMAIN_ON_LOCAL_QUEUE = 8 * 60 * 60 // 8 hours
 const VISIBILITY_TIMEOUT_FOR_PROCESSING = 1 * 60 * 60 // 1 hours, similar to storage queue pop visibility timeout
 
-class StorageBackedInMemoryQueue extends NestedQueue {
+class StorageBackedQueue extends NestedQueue {
 
-  constructor(memoryQueue, storageQueue, options) {
-    super(memoryQueue)
+  constructor(queue, storageQueue, options) {
+    super(queue)
     this.options = options
     this.logger = options.logger
     this._sharedStorageQueue = storageQueue
@@ -45,63 +45,75 @@ class StorageBackedInMemoryQueue extends NestedQueue {
       // Message not found for the popReceipt and messageId stored in the request.  This means
       // that the message popReceipt (and possibly messageId) in the request is stale. This can
       // happen when the message visibility timeout expired and thus was visible and later
-      // updated/processed by others.Because the request is picked up by others, just log and
+      // updated/processed by others. Because the request is picked up by others, just log and
       // continue to the next.
       this._log('Failed to update stale message', request)
       return false
     }
   }
 
   async done(request) {
+    await Promise.all([
+      super.done(request),
+      this._doneInStorage(request)])
+  }
+
+  async _doneInStorage(request) {
     try {
-      await super.done(request)
       await this._sharedStorageQueue.done(request)
     } catch (error) {
       if (!this._sharedStorageQueue.isMessageNotFound(error)) throw error
       // Message not found for the popReceipt and messageId stored in the request.  This means
       // that the message popReceipt (and possibly messageId) in the request is stale. This can
       // happen when the message visibility timeout expired and thus was visible and later
-      // updated by others.This is ok because the deletion of the request can be left to the
+      // updated by others. This is ok because the deletion of the request can be left to the
       // caller with the updated popReceipt. Log and continue.
       this._log('Failed to remove stale message', request)
     }
   }
 
   async subscribe() {
-    await super.subscribe()
-    await this._sharedStorageQueue.subscribe()
+    await Promise.all([
+      super.subscribe(),
+      this._sharedStorageQueue.subscribe()])
   }
 
   async unsubscribe() {
-    await super.unsubscribe()
-    await this._sharedStorageQueue.unsubscribe()
+    const results = await Promise.allSettled([
+      super.unsubscribe(),
+      this._sharedStorageQueue.unsubscribe()])
+    this._throwIfError(results, 'Failed to unsubscribe')
   }
 
   async flush() {
     const deleteRequests = []
     const info = await this.getInfo()
     for (let count = info.count; count > 0; count--) {
-      deleteRequests.push(this.pop().then(request => request && this.done(request)))
+      const deleteOne = super.pop().then(request => this.done(request))
+      deleteRequests.push(deleteOne)
     }
-    return Promise.allSettled(deleteRequests)
-      .then(results => {
-        const found = results.find(result => result.status === 'rejected')
-        if (found) throw new Error(found.reason)
-      })
+    const results = await Promise.allSettled(deleteRequests)
+    this._throwIfError(results, 'Failed to flush')
+  }
+
+  _throwIfError(results, message) {
+    const errors = results.filter(result => result.status === 'rejected')
+      .map(rejected => new Error(rejected.reason))
+    if (errors.length) throw new AggregateError(errors, message)
   }
 
   _log(actionMessage, request) {
     this.logger.verbose(`${actionMessage} ${request.type} ${request.url}`)
   }
 
-  static create(memoryQueue, storageQueue, options = {}) {
+  static create(queue, storageQueue, options = {}) {
     const defaultOptions = {
       visibilityTimeout_remainLocal: VISIBILITY_TIMEOUT_TO_REMAIN_ON_LOCAL_QUEUE,
       visibilityTimeout: VISIBILITY_TIMEOUT_FOR_PROCESSING
     }
     const optionsWithDefaults = { ...defaultOptions, ...options }
-    return new StorageBackedInMemoryQueue(memoryQueue, storageQueue, optionsWithDefaults)
+    return new StorageBackedQueue(queue, storageQueue, optionsWithDefaults)
   }
 }
 
-module.exports = StorageBackedInMemoryQueue
+module.exports = StorageBackedQueue