fix(job-queue): retry promptly when an idle peek finds a ready job

packages/job-queue/src/job/JobQueueWorker.ts

@@ -43,6 +43,14 @@ import { storageToClass } from "./JobStorageConverters";
  */
 const LOOP_ERROR_LOG_INTERVAL_MS = 5_000;
 
+/**
+ * First backoff step used when the idle peek says the head of the queue is
+ * already visible but the claim we just attempted came back empty. Doubles per
+ * consecutive occurrence, capped at the poll interval — see
+ * {@link JobQueueWorker.getIdleDelay}.
+ */
+const IDLE_READY_RETRY_BASE_MS = 5;
+
 /**
  * Events emitted by JobQueueWorker
  */
@@ -183,6 +191,14 @@ export class JobQueueWorker<
    */
   private wakePending = false;
 
+  /**
+   * Consecutive idle iterations that found a ready (already-visible) job at the
+   * head of the queue without being able to claim it. Drives the backoff in
+   * {@link getIdleDelay}; reset whenever a claim succeeds or the queue is
+   * genuinely idle.
+   */
+  private readyRetryStreak = 0;
+
   /**
    * Promise for the running `processJobs` loop. Captured in {@link start} so
    * {@link stop} can await actual loop exit instead of returning while the
@@ -591,6 +607,8 @@ export class JobQueueWorker<
           continue;
         }
 
+        this.readyRetryStreak = 0;
+
         const { dispatched, limiterFull } = await this.processClaimsInternal(claims);
 
         if (!this.running) {
@@ -644,18 +662,37 @@ export class JobQueueWorker<
    * Determine how long to sleep when idle.
    *
    * Peeks at the earliest PENDING job: if it has a future `visible_at`,
-   * returns the time until it becomes ready (clamped to `pollIntervalMs`);
-   * otherwise returns `pollIntervalMs`.
+   * returns the time until it becomes ready (clamped to `pollIntervalMs`); an
+   * empty queue returns `pollIntervalMs`.
+   *
+   * A head job that is *already* visible means the claim attempt that just
+   * came back empty raced its `visible_at` deadline — the claim ran before the
+   * deadline and this peek resolved after it, which is routine when a
+   * short-deferred submit lands on a storage whose round trips are slower than
+   * the deferral. Sleeping the full poll interval there strands ready work for
+   * the whole interval (60s in the fast-wake tests), so retry promptly instead.
+   * The retry backs off exponentially per consecutive occurrence, up to the
+   * poll interval, so the pathological version of this — a head job that stays
+   * visible-but-unclaimable, e.g. under clock skew between the worker and the
+   * storage — degrades to plain polling rather than spinning on the backend.
    */
   private async getIdleDelay(): Promise<number> {
     try {
       const pending = await this.jobStore.peek(JobStatus.PENDING, 1);
-      if (pending.length > 0 && pending[0].visible_at) {
-        const delay = new Date(pending[0].visible_at).getTime() - Date.now();
+      if (pending.length > 0) {
+        const visibleAt = pending[0].visible_at;
+        const delay = visibleAt ? new Date(visibleAt).getTime() - Date.now() : 0;
         if (delay > 0) {
+          this.readyRetryStreak = 0;
           return Math.min(delay, this.pollIntervalMs);
         }
+        const step = IDLE_READY_RETRY_BASE_MS * 2 ** this.readyRetryStreak;
+        if (step < this.pollIntervalMs) {
+          this.readyRetryStreak++;
+        }
+        return Math.min(step, this.pollIntervalMs);
       }
+      this.readyRetryStreak = 0;
     } catch {
       // If peek fails, fall back to default
     }

packages/test/src/test/job-queue/JobQueueWorker.test.ts

@@ -4,7 +4,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import type { IJobExecuteContext } from "@workglow/job-queue";
+import type { IJobExecuteContext, JobStorageFormat } from "@workglow/job-queue";
 import {
   InMemoryQueueStorage,
   InMemoryRateLimiterStorage,
@@ -307,6 +307,80 @@ describe("JobQueueWorker — PR #511 follow-up regressions", () => {
   });
 });
 
+/**
+ * Storage that mirrors a backend whose round trips are slower than a short
+ * deferral, and whose change feed is unavailable (Supabase realtime when it
+ * isn't wired up) so the submit-time notify is the only wake path. The claim
+ * attempt in an idle iteration then runs *before* a deferred job's `visible_at`
+ * and the idle peek resolves *after* it.
+ */
+class SlowPeekStorage extends InMemoryQueueStorage<TI, TO> {
+  public constructor(
+    queueName: string,
+    private readonly peekDelayMs: number
+  ) {
+    super(queueName);
+  }
+
+  public override async peek(
+    status?: JobStatus,
+    num?: number
+  ): Promise<JobStorageFormat<TI, TO>[]> {
+    const rows = await super.peek(status, num);
+    await sleep(this.peekDelayMs);
+    return rows;
+  }
+
+  public override subscribeToChanges(): () => void {
+    throw new Error("change feed unavailable");
+  }
+}
+
+describe("JobQueueWorker idle delay", () => {
+  setLogger(getTestingLogger());
+
+  it("picks up a deferred job whose visible_at elapsed during the idle peek", async () => {
+    // The submit-time notify wakes the worker before the job is visible, so the
+    // claim comes back empty; the idle peek then resolves after `visible_at`
+    // has passed. Reading that as "nothing to do" would sleep the whole poll
+    // interval (60s here) on a job that is claimable right now.
+    const queueName = `idle-delay-${uuid4()}`;
+    const storage = new SlowPeekStorage(queueName, 300);
+    await storage.migrate();
+    const { messageQueue, jobStore } = wrapQueueStorage(storage);
+    const server = new JobQueueServer<TI, TO, TJob>(TJob, {
+      messageQueue,
+      jobStore,
+      queueName,
+      pollIntervalMs: 60_000,
+      stopTimeoutMs: 0,
+    });
+    const client = new JobQueueClient<TI, TO>({ messageQueue, jobStore, queueName });
+    client.attach(server);
+    await server.start();
+
+    // Let the worker settle into its idle sleep, so the submit below is what
+    // wakes it rather than being latched as a pending wake.
+    await sleep(600);
+
+    const handle = await client.send(
+      { taskType: "default", data: "deferred" },
+      { delaySeconds: 0.2 }
+    );
+
+    const start = Date.now();
+    const result = await Promise.race([
+      handle.waitFor(),
+      sleep(5_000).then(() => "TIMEOUT" as const),
+    ]);
+    expect(result).not.toBe("TIMEOUT");
+    expect(Date.now() - start).toBeLessThan(5_000);
+
+    await server.stop();
+    await storage.deleteAll();
+  }, 20_000);
+});
+
 describe("JobQueueWorker limit overrides", () => {
   it("caps the processing-time sample window at the configured maxProcessingTimeSamples", async () => {
     TJob.executeCalls = 0;