fix: shutdown problems with db.stop when exit signal is received

  * remove all awaits on queue.drained calls in stop methods
  * better error handling around pool/worker stops in db.stop

src/database/standalone-sqlite.ts

@@ -2778,26 +2778,34 @@ export class StandaloneSqliteDatabase
   async stop() {
     const log = this.log.child({ method: 'stop' });
     const promises: Promise<void>[] = [];
-    WORKER_POOL_NAMES.forEach((pool) => {
-      WORKER_ROLE_NAMES.forEach((role) => {
-        this.workers[pool][role].forEach(() => {
-          promises.push(
-            new Promise((resolve, reject) => {
-              this.workQueues[pool][role].push({
-                resolve,
-                reject,
-                message: {
-                  method: 'terminate',
-                },
-              });
-              this.drainQueue();
-            }),
-          );
-        });
-      });
-    });
 
+    for (const pool of WORKER_POOL_NAMES) {
+      for (const role of WORKER_ROLE_NAMES) {
+        if (this.workers[pool][role] !== undefined) {
+          for (const worker of this.workers[pool][role]) {
+            if (worker && typeof worker.push === 'function') {
+              log.debug('Creating stop job for worker', { pool, role });
+              promises.push(
+                new Promise((resolve) => {
+                  worker.push({
+                    resolve,
+                    // by always resolving, we prevent shutdown procedure from
+                    // perpetually waiting for a response from the worker
+                    reject: resolve,
+                    message: {
+                      method: 'terminate',
+                    },
+                  });
+                }),
+              );
+              this.drainQueue();
+            }
+          }
+        }
+      }
+    }
     await Promise.all(promises);
+
     log.debug('Stopped successfully.');
   }
 
@@ -3460,6 +3468,7 @@ if (!isMainThread) {
           parentPort?.postMessage(walCheckpoint);
           break;
         case 'terminate':
+          console.log('Worker terminating', method, args);
           parentPort?.postMessage(null);
           process.exit(0);
       }

src/system.ts

@@ -754,11 +754,10 @@ export const shutdown = async (exitCode = 0) => {
       await ans104DataIndexer.stop();
       await ans104Unbundler.stop();
       await webhookEmitter.stop();
-      await db.stop();
       await headerFsCacheCleanupWorker?.stop();
       await contiguousDataFsCacheCleanupWorker?.stop();
       await dataVerificationWorker?.stop();
-
+      await db.stop();
       process.exit(exitCode);
     });
   }

src/workers/ans104-data-indexer.ts

@@ -157,7 +157,7 @@ export class Ans104DataIndexer {
   async stop(): Promise<void> {
     const log = this.log.child({ method: 'stop' });
     this.queue.kill();
-    await this.queue.drained();
+    this.queue.drained();
     log.debug('Stopped successfully.');
   }
 }

src/workers/ans104-unbundler.ts

@@ -189,7 +189,7 @@ export class Ans104Unbundler {
   async stop(): Promise<void> {
     const log = this.log.child({ method: 'stop' });
     this.queue.kill();
-    await this.queue.drained();
+    this.queue.drained();
     await this.ans104Parser.stop();
     log.debug('Stopped successfully.');
   }

src/workers/data-content-attribute-importer.ts

@@ -87,7 +87,7 @@ export class DataContentAttributeImporter {
   async stop(): Promise<void> {
     const log = this.log.child({ method: 'stop' });
     this.queue.kill();
-    await this.queue.drained();
+    this.queue.drained();
     log.debug('Stopped successfully.');
   }
 

src/workers/data-importer.ts

@@ -143,7 +143,7 @@ export class DataImporter {
   async stop(): Promise<void> {
     const log = this.log.child({ method: 'stop' });
     this.queue.kill();
-    await this.queue.drained();
+    this.queue.drained();
     log.debug('Stopped successfully.');
   }
 

src/workers/data-item-indexer.ts

@@ -107,7 +107,7 @@ export class DataItemIndexer {
   async stop(): Promise<void> {
     const log = this.log.child({ method: 'stop' });
     this.queue.kill();
-    await this.queue.drained();
+    this.queue.drained();
     log.debug('Stopped successfully.');
   }
 }

src/workers/data-verification.ts

@@ -163,7 +163,7 @@ export class DataVerificationWorker {
     const log = this.log.child({ method: 'stop' });
     clearInterval(this.intervalId);
     this.queue.kill();
-    await this.queue.drained();
+    this.queue.drained();
     await this.dataRootComputer.stop();
     log.debug('Stopped successfully.');
   }

src/workers/transaction-fetcher.ts

@@ -118,7 +118,7 @@ export class TransactionFetcher {
   async stop(): Promise<void> {
     const log = this.log.child({ method: 'stop' });
     this.queue.kill();
-    await this.queue.drained();
+    this.queue.drained();
     log.debug('Stopped successfully.');
   }
 

src/workers/transaction-importer.ts

@@ -75,7 +75,7 @@ export class TransactionImporter {
   async stop(): Promise<void> {
     const log = this.log.child({ method: 'stop' });
     this.queue.kill();
-    await this.queue.drained();
+    this.queue.drained();
     log.debug('Stopped successfully.');
   }
 

src/workers/transaction-offset-importer.ts

@@ -93,7 +93,7 @@ export class TransactionOffsetImporter {
   async stop(): Promise<void> {
     const log = this.log.child({ method: 'stop' });
     this.queue.kill();
-    await this.queue.drained();
+    this.queue.drained();
     log.debug('Stopped successfully.');
   }
 

src/workers/webhook-emitter.test.ts

@@ -153,7 +153,7 @@ describe('WebhookEmitter', () => {
 
       await wait(0);
 
-      await webhookEmitter.emissionQueue.drained();
+      webhookEmitter.emissionQueue.drained();
 
       assert.equal(
         (webhookEmitter.emitWebhookToTargetServer as any).mock.callCount(),

src/workers/webhook-emitter.ts

@@ -155,7 +155,7 @@ export class WebhookEmitter {
     this.listenerReferences.clear();
 
     this.emissionQueue.kill();
-    await this.emissionQueue.drained();
+    this.emissionQueue.drained();
     log.debug('Stopped successfully.');
   }