Fix task identifier cache recovery after database errors

.changeset/tidy-workers-retry.md

@@ -0,0 +1,7 @@
+---
+"graphile-worker": patch
+---
+
+Allow task identifier lookups to recover after a database error instead of
+caching the rejected promise indefinitely. Pending lookups for an older task
+list can no longer overwrite a newer cached result.

__tests__/taskIdentifiers.test.ts

@@ -0,0 +1,81 @@
+import { jest } from "@jest/globals";
+
+import defer from "../src/deferred.ts";
+import type { EnhancedWithPgClient } from "../src/interfaces.ts";
+import type { CompiledSharedOptions } from "../src/lib.ts";
+import { getTaskDetails } from "../src/taskIdentifiers.ts";
+
+function setup() {
+  const options = {
+    escapedWorkerSchema: '"graphile_worker"',
+  } as CompiledSharedOptions;
+  // Control completion at the database boundary, after connection retries.
+  const withRetries = jest.fn<EnhancedWithPgClient["withRetries"]>();
+  const withPgClient = { withRetries } as unknown as EnhancedWithPgClient;
+  return {
+    withRetries,
+    lookup: (name = "task") =>
+      getTaskDetails(options, withPgClient, { [name]: async () => {} }),
+  };
+}
+
+const result = (id: number, identifier = "task") => ({
+  rows: [{ id, identifier }],
+});
+
+test.each(["53300", "ECONNRESET"])(
+  "retries a failed task lookup (%s) and caches the successful result",
+  async (code) => {
+    const { withRetries, lookup } = setup();
+    const error = Object.assign(new Error("database unavailable"), { code });
+    withRetries.mockRejectedValueOnce(error).mockResolvedValueOnce(result(1));
+
+    await expect(lookup()).rejects.toBe(error);
+    const details = await lookup();
+    expect(details.taskIds).toEqual([1]);
+    expect(lookup()).toBe(details);
+    expect(withRetries).toHaveBeenCalledTimes(2);
+  },
+);
+
+test("concurrent callers share a pending lookup and its retry", async () => {
+  const { withRetries, lookup } = setup();
+  const pending = defer();
+  withRetries.mockReturnValueOnce(pending).mockResolvedValueOnce(result(1));
+  const first = lookup();
+  expect(lookup()).toBe(first);
+  const rejected = expect(first).rejects.toThrow("unavailable");
+  pending.reject(new Error("unavailable"));
+  await rejected;
+
+  const retry = lookup();
+  expect(lookup()).toBe(retry);
+  await expect(retry).resolves.toMatchObject({ taskIds: [1] });
+  expect(withRetries).toHaveBeenCalledTimes(2);
+});
+
+test.each(["success", "failure"])(
+  "a stale lookup's %s does not replace a newer task list",
+  async (outcome) => {
+    const { withRetries, lookup } = setup();
+    const pending = defer<ReturnType<typeof result>>();
+    withRetries
+      .mockReturnValueOnce(pending)
+      .mockResolvedValue(result(2, "new"));
+    const first = lookup("old");
+    const current = await lookup("new");
+
+    if (outcome === "failure") {
+      const rejected = expect(first).rejects.toThrow("unavailable");
+      pending.reject(new Error("unavailable"));
+      await rejected;
+    } else {
+      pending.resolve(result(1, "old"));
+      await expect(first).resolves.toMatchObject({ taskIds: [1] });
+    }
+
+    expect(lookup("new")).toBe(current);
+    expect(current.taskIds).toEqual([2]);
+    expect(withRetries).toHaveBeenCalledTimes(2);
+  },
+);

src/taskIdentifiers.ts

@@ -40,7 +40,7 @@ export function getTaskDetails(
     const { escapedWorkerSchema } = compiledSharedOptions;
     assert.ok(supportedTaskNames.length, "No runnable tasks!");
     cache.lastStr = str;
-    cache.lastDigest = (async () => {
+    const pending = (async () => {
       const { rows } = await withPgClient.withRetries(async (client) => {
         await client.query({
           text: `insert into ${escapedWorkerSchema}._private_tasks as tasks (identifier) select i from unnest($1::text[]) as u(i) where not exists (select 1 from ${escapedWorkerSchema}._private_tasks as existing where existing.identifier = u.i) on conflict do nothing`,
@@ -61,14 +61,24 @@ export function getTaskDetails(
         parseInt(s, 10),
       );
 
-      // Overwrite promises with concrete values
-      cache.lastDigest = {
-        supportedTaskIdentifierByTaskId,
-        taskIds,
-      };
-      cache.lastStr = str;
-      return cache.lastDigest;
-    })();
+      return { supportedTaskIdentifierByTaskId, taskIds };
+    })().then(
+      (result) => {
+        // A changed task list may have started a newer lookup meanwhile.
+        if (cache.lastDigest === pending) {
+          cache.lastDigest = result;
+        }
+        return result;
+      },
+      (error) => {
+        // Let the next poll retry rather than caching a rejection indefinitely.
+        if (cache.lastDigest === pending) {
+          cacheByOptions.delete(compiledSharedOptions);
+        }
+        throw error;
+      },
+    );
+    cache.lastDigest = pending;
   }
   return cache.lastDigest;
 }