fix: Make background job shutdown cleanly

packages/base-test-utils/src/index.ts

@@ -54,6 +54,7 @@ export class BaseTestUtils {
     throw new Error(baseErrMsg + ': ' + customMsg)
   }
 
+  // TODO: De-dupe this with `delayOrAbort` in abort-signal-utils.ts
   static async delay(ms: number, signal?: AbortSignal): Promise<void> {
     return new Promise<void>((resolve, reject) => {
       const timeout = setTimeout(() => resolve(), ms)

packages/common/src/utils/abort-signal-utils.ts

@@ -66,3 +66,18 @@ export async function abortable<T>(
     original.removeEventListener('abort', onAbort)
   })
 }
+
+export async function delayOrAbort(ms: number, signal?: AbortSignal): Promise<void> {
+  return new Promise<void>((resolve, reject) => {
+    const timeout = setTimeout(() => resolve(), ms)
+    if (signal) {
+      const handleAbort = () => {
+        clearTimeout(timeout)
+        signal.removeEventListener('abort', handleAbort)
+        reject(signal.reason)
+      }
+      if (signal.aborted) handleAbort()
+      signal.addEventListener('abort', handleAbort)
+    }
+  })
+}

packages/core/src/ceramic.ts

@@ -24,6 +24,7 @@ import {
   CeramicSigner,
   StreamStateLoader,
   StreamReaderWriter,
+  delayOrAbort,
 } from '@ceramicnetwork/common'
 import {
   DEFAULT_TRACE_SAMPLE_RATIO,
@@ -683,12 +684,8 @@ export class Ceramic implements StreamReaderWriter, StreamStateLoader {
     let attemptNum = 0
     let backoffMs = 100
     const maxBackoffMs = 1000 * 60 // Caps off at checking once per minute
-    const delay = async function (ms) {
-      return new Promise<void>((resolve) => setTimeout(() => resolve(), ms))
-    }
 
-    // eslint-disable-next-line no-constant-condition
-    while (true) {
+    while (!this._shutdownSignal.isShuttingDown()) {
       try {
         await this.dispatcher.getFromIpfs(model.cid)
         if (attemptNum > 0) {
@@ -704,7 +701,7 @@ export class Ceramic implements StreamReaderWriter, StreamStateLoader {
           this._logger.err(`Error loading Model ${model} used to publish Node Metrics: ${err}`)
         }
 
-        await delay(backoffMs)
+        await this._shutdownSignal.abortable((signal) => delayOrAbort(backoffMs, signal))
         attemptNum++
         if (backoffMs <= maxBackoffMs) {
           backoffMs *= 2

packages/core/src/shutdown-signal.ts

@@ -7,6 +7,7 @@ import { Observer, Subject } from 'rxjs'
  */
 export class ShutdownSignal {
   private subject: Subject<void> = new Subject()
+  private shuttingDown = false
 
   /**
    * Subscribers to the signal.
@@ -20,6 +21,11 @@ export class ShutdownSignal {
    */
   abort(): void {
     this.subject.complete()
+    this.shuttingDown = true
+  }
+
+  isShuttingDown(): boolean {
+    return this.shuttingDown
   }
 
   /**