Use safeEmit() helper (#557)

Author: benjie

RELEASE_NOTES.md

@@ -46,6 +46,8 @@ Read more:
 - `backfillPeriod` is now marked as optional in TypeScript (defaults to 0).
 - Support for loading tasks from nested folders in crontab.
   - (`* * * * * nested/folder/task ?jobKey=my_key&jobKeyMode=preserve_run_at`)
+- Most of our event emitters now trap errors and output a log if such error were
+  to occur - useful for debugging.
 
 ## v0.16.6
 

src/cron.ts

@@ -23,6 +23,7 @@ import {
   CompiledSharedOptions,
   Releasers,
   RetryOptions,
+  safeEmit,
   sleep,
 } from "./lib";
 
@@ -208,7 +209,6 @@ async function registerAndBackfillItems(details: {
   const {
     ctx,
     pgPool,
-    events,
     cron,
     workerSchema,
     preparedStatements,
@@ -292,7 +292,7 @@ async function registerAndBackfillItems(details: {
         // the way the last_execution column of the known_crontabs table works.
         // At this time it's not expected that backfilling will be sufficiently
         // expensive to justify optimising this further.
-        events.emit("cron:backfill", {
+        safeEmit(ctx, "cron:backfill", {
           ctx,
           cron,
           itemsToBackfill,
@@ -386,7 +386,7 @@ export const runCron = (
 
     const start = new Date();
     const ctx = compiledSharedOptions;
-    events.emit("cron:starting", { ctx, cron, start });
+    safeEmit(ctx, "cron:starting", { ctx, cron, start });
 
     // We must backfill BEFORE scheduling any new jobs otherwise backfill won't
     // work due to known_crontabs.last_execution having been updated.
@@ -403,7 +403,7 @@ export const runCron = (
       useNodeTime,
     });
 
-    events.emit("cron:started", { ctx, cron, start });
+    safeEmit(ctx, "cron:started", { ctx, cron, start });
 
     if (!cron._active) {
       return stop();
@@ -465,7 +465,7 @@ export const runCron = (
               currentTimestamp,
             },
           );
-          events.emit("cron:prematureTimer", {
+          safeEmit(ctx, "cron:prematureTimer", {
             ctx,
             cron,
             currentTimestamp,
@@ -482,7 +482,7 @@ export const runCron = (
               ((currentTimestamp - expectedTimestamp) % ONE_MINUTE) / 1000,
             )}s behind)`,
           );
-          events.emit("cron:overdueTimer", {
+          safeEmit(ctx, "cron:overdueTimer", {
             ctx,
             cron,
             currentTimestamp,
@@ -505,7 +505,7 @@ export const runCron = (
 
         // Finally actually run the jobs.
         if (jobsAndIdentifiers.length) {
-          events.emit("cron:schedule", {
+          safeEmit(ctx, "cron:schedule", {
             ctx,
             cron,
             timestamp: expectedTimestamp,
@@ -520,7 +520,7 @@ export const runCron = (
             workerSchema,
             preparedStatements,
           );
-          events.emit("cron:scheduled", {
+          safeEmit(ctx, "cron:scheduled", {
             ctx,
             cron,
             timestamp: expectedTimestamp,

src/lib.ts

@@ -21,6 +21,7 @@ import {
   RunOnceOptions,
   SharedOptions,
   WithPgClient,
+  WorkerEventMap,
   WorkerOptions,
   WorkerPluginBaseContext,
   WorkerPluginContext,
@@ -632,3 +633,22 @@ export function calculateDelay(
     (0.5 + jitter)
   );
 }
+
+export function safeEmit<TEvent extends keyof WorkerEventMap>(
+  ctx: WorkerPluginContext,
+  eventName: TEvent,
+  payload: WorkerEventMap[TEvent],
+) {
+  const { events, logger } = ctx;
+  try {
+    events.emit(eventName, payload);
+  } catch (e) {
+    const error = coerceError(e);
+    logger.error(
+      `Emitting ${eventName} raised an error - this should never happen: ${
+        error.stack ?? error
+      }`,
+      { error, eventName, payload },
+    );
+  }
+}

src/localQueue.ts

@@ -15,6 +15,7 @@ import {
   coerceError,
   RETRYABLE_ERROR_CODES,
   RetryOptions,
+  safeEmit,
   sleep,
 } from "./lib";
 import { batchGetJobs } from "./sql/getJobs";
@@ -231,7 +232,7 @@ export class LocalQueue {
   private refetchDelayAbortThreshold: number = Infinity;
 
   constructor(
-    private readonly compiledSharedOptions: CompiledSharedOptions<WorkerPoolOptions>,
+    private readonly ctx: CompiledSharedOptions<WorkerPoolOptions>,
     private readonly tasks: TaskList,
     private readonly withPgClient: EnhancedWithPgClient,
     public readonly workerPool: WorkerPool,
@@ -244,13 +245,10 @@ export class LocalQueue {
     private readonly continuous: boolean,
     private readonly onMajorError: (e: unknown) => void,
   ) {
-    this.ttl =
-      compiledSharedOptions.resolvedPreset.worker.localQueue?.ttl ?? 5 * MINUTE;
-    this.pollInterval =
-      compiledSharedOptions.resolvedPreset.worker.pollInterval ?? 2 * SECOND;
+    this.ttl = ctx.resolvedPreset.worker.localQueue?.ttl ?? 5 * MINUTE;
+    this.pollInterval = ctx.resolvedPreset.worker.pollInterval ?? 2 * SECOND;
     const localQueueRefetchDelayDuration =
-      compiledSharedOptions.resolvedPreset.worker.localQueue?.refetchDelay
-        ?.durationMs;
+      ctx.resolvedPreset.worker.localQueue?.refetchDelay?.durationMs;
     if (
       localQueueRefetchDelayDuration != null &&
       localQueueRefetchDelayDuration > this.pollInterval
@@ -259,8 +257,8 @@ export class LocalQueue {
         `Invalid configuration; 'preset.worker.localQueue.refetchDelay.durationMs' (${localQueueRefetchDelayDuration}) must not be larger than 'preset.worker.pollInterval' (${this.pollInterval})`,
       );
     }
-    compiledSharedOptions.events.emit("localQueue:init", {
-      ctx: compiledSharedOptions,
+    safeEmit(ctx, "localQueue:init", {
+      ctx: ctx,
       localQueue: this,
     });
     // Immediately enter polling mode.
@@ -276,8 +274,8 @@ export class LocalQueue {
     const oldMode = this.mode;
     // Override the 'readonly'
     (this.mode as LocalQueueMode) = newMode;
-    this.compiledSharedOptions.events.emit("localQueue:setMode", {
-      ctx: this.compiledSharedOptions,
+    safeEmit(this.ctx, "localQueue:setMode", {
+      ctx: this.ctx,
       localQueue: this,
       oldMode,
       newMode,
@@ -321,7 +319,7 @@ export class LocalQueue {
     } else {
       // If we're not shutting down, view this as a temporary error (but give
       // Benjie a wrist slap anyway).
-      this.compiledSharedOptions.logger.error(
+      this.ctx.logger.error(
         `GraphileWorkerInternalError<cd483429-3372-42f0-bcf6-c78e045c760d>: Backgrounding should never yield errors when the queue is not RELEASED`,
         { error: e },
       );
@@ -444,8 +442,8 @@ export class LocalQueue {
     }
     const jobsToReturn = this.jobQueue.splice(0, l);
 
-    this.compiledSharedOptions.events.emit("localQueue:returnJobs", {
-      ctx: this.compiledSharedOptions,
+    safeEmit(this.ctx, "localQueue:returnJobs", {
+      ctx: this.ctx,
       localQueue: this,
       jobs: jobsToReturn,
     });
@@ -459,7 +457,7 @@ export class LocalQueue {
         initialError = lastError;
       }
 
-      this.compiledSharedOptions.logger.error(
+      this.ctx.logger.error(
         `Failed to return jobs from local queue to database queue (attempt ${attempts}/${maxAttempts})`,
         {
           error: e,
@@ -502,7 +500,7 @@ export class LocalQueue {
             ++attempts;
             return sleep(delay).then(() =>
               returnJobs(
-                this.compiledSharedOptions,
+                this.ctx,
                 this.withPgClient, // We'll handle the retries via onError
                 this.workerPool.id,
                 jobsToReturn,
@@ -517,7 +515,7 @@ export class LocalQueue {
     // `onError` above, since `onError` returns the next promise each time.
     this.background(
       returnJobs(
-        this.compiledSharedOptions,
+        this.ctx,
         this.withPgClient, // We'll handle the retries via onError
         this.workerPool.id,
         jobsToReturn,
@@ -554,7 +552,7 @@ export class LocalQueue {
     this.background(
       this._fetch().catch((e) => {
         // This should not happen
-        this.compiledSharedOptions.logger.error(`Error occurred during fetch`, {
+        this.ctx.logger.error(`Error occurred during fetch`, {
           error: e,
         });
       }),
@@ -578,7 +576,7 @@ export class LocalQueue {
     /** How many jobs did we fetch? (Initialize to zero in case of error.) */
     let jobCount = 0;
     const refetchDelayOptions =
-      this.compiledSharedOptions.resolvedPreset.worker.localQueue?.refetchDelay;
+      this.ctx.resolvedPreset.worker.localQueue?.refetchDelay;
     try {
       assert.equal(this.mode, POLLING, "Can only fetch when in polling mode");
       assert.equal(
@@ -605,16 +603,16 @@ export class LocalQueue {
 
       // The ONLY await in this function.
       const jobs = await batchGetJobs(
-        this.compiledSharedOptions,
+        this.ctx,
         this.withPgClient,
         this.tasks,
         this.workerPool.id,
         null, // `flagsToSkip` is not set, see `LocalQueue.getJob`
         this.getJobBatchSize,
       );
 
-      this.compiledSharedOptions.events.emit("localQueue:getJobs:complete", {
-        ctx: this.compiledSharedOptions,
+      safeEmit(this.ctx, "localQueue:getJobs:complete", {
+        ctx: this.ctx,
         localQueue: this,
         jobs,
       });
@@ -634,7 +632,7 @@ export class LocalQueue {
       this.receivedJobs(jobs);
     } catch (e) {
       // Error happened; rely on poll interval.
-      this.compiledSharedOptions.logger.error(
+      this.ctx.logger.error(
         `Error occurred fetching jobs; will try again on next poll interval. Error: ${e}`,
         { error: e },
       );
@@ -674,8 +672,8 @@ export class LocalQueue {
         this.refetchDelayCompleteOrAbort,
         refetchDelayMs,
       );
-      this.compiledSharedOptions.events.emit("localQueue:refetchDelay:start", {
-        ctx: this.compiledSharedOptions,
+      safeEmit(this.ctx, "localQueue:refetchDelay:start", {
+        ctx: this.ctx,
         localQueue: this,
         jobCount,
         threshold: refetchDelayOptions?.threshold ?? 0,
@@ -720,20 +718,17 @@ export class LocalQueue {
       // Force refetch because we've been notified of so many jobs!
       this.refetchDelayFetchOnComplete = true;
 
-      this.compiledSharedOptions.events.emit("localQueue:refetchDelay:abort", {
-        ctx: this.compiledSharedOptions,
+      safeEmit(this.ctx, "localQueue:refetchDelay:abort", {
+        ctx: this.ctx,
         localQueue: this,
         count: this.refetchDelayCounter,
         abortThreshold: this.refetchDelayAbortThreshold,
       });
     } else {
-      this.compiledSharedOptions.events.emit(
-        "localQueue:refetchDelay:expired",
-        {
-          ctx: this.compiledSharedOptions,
-          localQueue: this,
-        },
-      );
+      safeEmit(this.ctx, "localQueue:refetchDelay:expired", {
+        ctx: this.ctx,
+        localQueue: this,
+      });
     }
 
     if (this.mode === POLLING && this.refetchDelayFetchOnComplete) {
@@ -789,7 +784,7 @@ export class LocalQueue {
     if (flagsToSkip !== null) {
       // PERF: we could actually batch for similar flags, I guess.
       const jobsPromise = batchGetJobs(
-        this.compiledSharedOptions,
+        this.ctx,
         this.withPgClient,
         this.tasks,
         this.workerPool.id,