Cron error recovery (#544)

Author: benjie

src/cron.ts

@@ -17,17 +17,27 @@ import {
   WorkerEvents,
 } from "./interfaces";
 import {
+  calculateDelay,
   coerceError,
   CompiledOptions,
   CompiledSharedOptions,
   Releasers,
+  RetryOptions,
+  sleep,
 } from "./lib";
 
 interface CronRequirements {
   pgPool: Pool;
   events: WorkerEvents;
 }
 
+const CRON_RETRY: RetryOptions = {
+  maxAttempts: Infinity,
+  minDelay: 200,
+  maxDelay: 60_000,
+  multiplier: 1.5,
+};
+
 /**
  * This function looks through all the cron items we have (e.g. from our
  * crontab file) and compares them to the items we already know about. If the
@@ -117,13 +127,15 @@ async function scheduleCronJobs(
   jobsAndIdentifiers: JobAndCronIdentifier[],
   ts: string,
   useNodeTime: boolean,
+  workerSchema: string,
+  preparedStatements: boolean,
 ) {
   // TODO: refactor this to use `add_jobs`
 
   // Note that `identifier` is guaranteed to be unique for every record
   // in `specs`.
-  await pgPool.query(
-    `
+  await pgPool.query({
+    text: `
       with specs as (
         select
           index,
@@ -166,26 +178,45 @@ async function scheduleCronJobs(
       inner join locks on (locks.identifier = specs.identifier)
       order by specs.index asc
     `,
-    [
+    values: [
       JSON.stringify(jobsAndIdentifiers),
       ts,
       useNodeTime ? new Date().toISOString() : null,
     ],
-  );
+    name: !preparedStatements
+      ? undefined
+      : `cron${useNodeTime ? "N" : ""}/${workerSchema}`,
+  });
 }
 
 /**
  * Marks any previously unknown crontab identifiers as now being known. Then
  * performs backfilling on any crontab tasks that need it.
  */
-async function registerAndBackfillItems(
-  ctx: CompiledSharedOptions,
-  { pgPool, events, cron }: { pgPool: Pool; events: WorkerEvents; cron: Cron },
-  escapedWorkerSchema: string,
-  parsedCronItems: ParsedCronItem[],
-  startTime: Date,
-  useNodeTime: boolean,
-) {
+async function registerAndBackfillItems(details: {
+  ctx: CompiledSharedOptions;
+  pgPool: Pool;
+  events: WorkerEvents;
+  cron: Cron;
+  workerSchema: string;
+  preparedStatements: boolean;
+  escapedWorkerSchema: string;
+  parsedCronItems: ParsedCronItem[];
+  startTime: Date;
+  useNodeTime: boolean;
+}) {
+  const {
+    ctx,
+    pgPool,
+    events,
+    cron,
+    workerSchema,
+    preparedStatements,
+    escapedWorkerSchema,
+    parsedCronItems,
+    startTime,
+    useNodeTime,
+  } = details;
   // First, scan the DB to get our starting point.
   const { rows } = await pgPool.query<KnownCrontab>(
     `SELECT * FROM ${escapedWorkerSchema}._private_known_crontabs as known_crontabs`,
@@ -273,6 +304,8 @@ async function registerAndBackfillItems(
           itemsToBackfill,
           ts,
           useNodeTime,
+          workerSchema,
+          preparedStatements,
         );
       }
 
@@ -305,9 +338,10 @@ export const runCron = (
   const {
     logger,
     escapedWorkerSchema,
+    workerSchema,
     events,
     resolvedPreset: {
-      worker: { useNodeTime },
+      worker: { useNodeTime, preparedStatements = true },
     },
   } = compiledSharedOptions;
 
@@ -323,7 +357,7 @@ export const runCron = (
     if (!stopCalled) {
       stopCalled = true;
       if (e) {
-        promise.reject(e);
+        promise.reject(coerceError(e));
       } else {
         promise.resolve();
       }
@@ -334,6 +368,17 @@ export const runCron = (
     }
   }
 
+  let attempts = 0;
+  function restartCronAfterDelay(error: Error) {
+    ++attempts;
+    const delay = calculateDelay(attempts - 1, CRON_RETRY);
+    logger.error(
+      `Cron hit an error; restarting in ${delay}ms (attempt ${attempts}): ${error}`,
+      { error, attempts },
+    );
+    sleep(delay).then(cronMain).catch(restartCronAfterDelay);
+  }
+
   async function cronMain() {
     if (!cron._active) {
       return stop();
@@ -345,14 +390,18 @@ export const runCron = (
 
     // We must backfill BEFORE scheduling any new jobs otherwise backfill won't
     // work due to known_crontabs.last_execution having been updated.
-    await registerAndBackfillItems(
+    await registerAndBackfillItems({
       ctx,
-      { pgPool, events, cron },
+      pgPool,
+      events,
+      cron,
+      workerSchema,
+      preparedStatements,
       escapedWorkerSchema,
       parsedCronItems,
-      new Date(+start),
+      startTime: new Date(+start),
       useNodeTime,
-    );
+    });
 
     events.emit("cron:started", { ctx, cron, start });
 
@@ -383,6 +432,8 @@ export const runCron = (
         if (!cron._active) {
           return stop();
         }
+        // Healthy!
+        attempts = 0;
 
         // THIS MUST COME BEFORE nextTimestamp IS MUTATED
         const digest = digestTimestamp(nextTimestamp);
@@ -466,6 +517,8 @@ export const runCron = (
             jobsAndIdentifiers,
             ts,
             useNodeTime,
+            workerSchema,
+            preparedStatements,
           );
           events.emit("cron:scheduled", {
             ctx,
@@ -486,9 +539,10 @@ export const runCron = (
         // timestamps on error).
         scheduleNextLoop();
       } catch (e) {
-        // If something goes wrong; abort. The calling code should re-schedule
-        // which will re-trigger the backfilling code.
-        return stop(coerceError(e));
+        // If something goes wrong; abort the current loop and restart cron
+        // after an exponential back-off. This is essential because we need to
+        // re-trigger the backfilling code.
+        return restartCronAfterDelay(coerceError(e));
       }
     }
 
@@ -510,7 +564,7 @@ export const runCron = (
     promise,
   };
 
-  cronMain().catch(stop);
+  cronMain().catch(restartCronAfterDelay);
 
   return cron;
 };

src/interfaces.ts

@@ -568,9 +568,9 @@ export interface WorkerPool {
 
 export interface Runner {
   /** Attempts to cleanly shut down the runner */
-  stop: () => Promise<void>;
+  stop: (reason?: string) => Promise<void>;
   /** Use .stop() instead, unless you know what you're doing */
-  kill: () => Promise<void>;
+  kill: (reason?: string) => Promise<void>;
   addJob: AddJobFunction;
   promise: Promise<void>;
   events: WorkerEvents;

src/runner.ts

@@ -164,8 +164,10 @@ function buildRunner(input: {
   });
 
   let running = true;
-  const stop = async () => {
-    compiledOptions.logger.debug("Runner stopping");
+  const stop = async (reason: string | null, itsFine = reason === null) => {
+    compiledOptions.logger[itsFine ? "debug" : "warn"](
+      `Runner stopping${reason ? ` (reason: ${reason})` : ""}`,
+    );
     if (running) {
       running = false;
       events.emit("stop", { ctx });
@@ -190,25 +192,19 @@ function buildRunner(input: {
       throw new Error("Runner is already stopped");
     }
   };
-  const kill = async () => {
-    if (running) {
-      stop().catch(() => {});
-    }
-    if (workerPool._active) {
-      await workerPool.forcefulShutdown(`Terminated through .kill() command`);
-    }
-  };
 
-  const wp = workerPool.promise.finally(() => {
-    if (running) {
-      stop();
-    }
-  });
-  const cp = cron.promise.finally(() => {
-    if (running) {
-      stop();
-    }
-  });
+  const wp = workerPool.promise
+    .then(
+      () => (running ? stop("worker pool exited cleanly", true) : void 0),
+      (e) => (running ? stop(`worker pool exited with error: ${e}`) : void 0),
+    )
+    .catch(noop);
+  const cp = cron.promise
+    .then(
+      () => (running ? stop("cron exited cleanly", true) : void 0),
+      (e) => (running ? stop(`cron exited with error: ${e}`) : void 0),
+    )
+    .catch(noop);
 
   const promise = Promise.all([cp, wp]).then(
     () => {
@@ -217,7 +213,7 @@ function buildRunner(input: {
     async (error) => {
       if (running) {
         logger.error(`Stopping worker due to an error: ${error}`, { error });
-        await stop();
+        await stop(String(error));
       } else {
         logger.error(
           `Error occurred, but worker is already stopping: ${error}`,
@@ -229,10 +225,22 @@ function buildRunner(input: {
   );
 
   return {
-    stop,
-    kill,
+    // It's fine - the user told us to exit
+    stop: (reason) => stop(reason ?? null, true),
+    async kill(reason?: string) {
+      if (running) {
+        stop(`runner.kill() called${reason ? `: ${reason}` : ""}`).catch(noop);
+      }
+      if (workerPool._active) {
+        await workerPool.forcefulShutdown(`Terminated through .kill() command`);
+      }
+    },
     addJob,
     promise,
     events,
   };
 }
+
+function noop() {
+  /* NOOP */
+}