fix: scan for stale pending messages on worker startup and increase check frequency

src/workers/index.ts

@@ -62,6 +62,7 @@ export abstract class BaseWorker {
   private lastProcessedAt: string | null = null;
   private lastErrorAt: string | null = null;
   private recoveryTimer: ReturnType<typeof setInterval> | null = null;
+  private recovering = false;
   private lastPendingCount: number | null = null;
   private readonly active = new Set<Promise<void>>();
   private readonly semaphore: AsyncSemaphore;
@@ -307,14 +308,28 @@ export abstract class BaseWorker {
     }
   }
 
+  /**
+   * `recoveryIntervalMs` is how long an entry must sit idle before it is
+   * claimed, not how often to look. Scanning only once per interval, timed from
+   * this worker's start, meant a worker restarted more often than that (a crash
+   * loop, an eager autoscaler) never scanned at all, and whatever its
+   * predecessors held stayed pending for good. So scan on start and then
+   * several times per interval; the idle threshold alone decides what is taken.
+   */
   private startRecoveryLoop(): void {
-    this.recoveryTimer = setInterval(() => {
-      if (this.state === "running") {
-        this.recover().catch((err: unknown) => {
+    const scan = () => {
+      if (this.state !== "running" || this.recovering) return;
+      this.recovering = true;
+      this.recover()
+        .catch((err: unknown) => {
           this.logger.error({ err }, "recovery loop error");
+        })
+        .finally(() => {
+          this.recovering = false;
         });
-      }
-    }, this.recoveryIntervalMs);
+    };
+    scan();
+    this.recoveryTimer = setInterval(scan, Math.max(Math.floor(this.recoveryIntervalMs / 4), 10));
   }
 
   private stopRecoveryLoop(): void {

tests/chaos/crash.test.ts

@@ -149,7 +149,7 @@ describe("Crash Testing (Chaos Monkey)", () => {
     // So wait 180s.
     const startTime = Date.now();
     while (deliveredCount + failedCount < NUM_MESSAGES) {
-      if (Date.now() - startTime > 180_000) {
+      if (Date.now() - startTime > 150_000) {
         throw new Error(
           `Timeout waiting for deliveries. Delivered: ${deliveredCount}, Failed: ${failedCount}, Expected: ${NUM_MESSAGES}`,
         );

tests/integration/worker-runtime.test.ts

@@ -515,6 +515,30 @@ describe("BaseWorker", () => {
     expect(await pending()).toBe(0);
   });
 
+  it("recovers a crashed consumer's message even when no worker outlives one recovery interval", async () => {
+    // A worker that crashed mid-message leaves its entry pending under a
+    // consumer name nobody will read again. Its replacements restart more often
+    // than the recovery interval (a crash loop, an aggressive autoscaler), so a
+    // scan timed from each worker's start never fires.
+    await infra.redis.xgroup("CREATE", STREAM, GROUP, "0", "MKSTREAM");
+    await publish(1);
+    await infra.redis.xreadgroup("GROUP", GROUP, "crashed", "COUNT", 1, "STREAMS", STREAM, ">");
+
+    const processed: number[] = [];
+    const deadline = Date.now() + 5_000;
+    while (processed.length === 0 && Date.now() < deadline) {
+      const w = worker(async (msg) => void processed.push((msg.event.payload as any).n), {
+        recoveryIntervalMs: 400,
+      });
+      await w.start();
+      await settle(250);
+      await w.stop();
+    }
+
+    expect(processed).toEqual([1]);
+    expect(await pending()).toBe(0);
+  });
+
   it("recover() reports the pending count through health()", async () => {
     const w = worker(async () => {});
     expect(w.health().pendingCount).toBeNull();