fix(server): preserve backlog deadlines after partial failure

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/chat-channels.integration.test.ts

@@ -6770,11 +6770,16 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
         expect(deadline).toBeGreaterThan(Date.now() + 200_000);
         await f.service.configure(f.endpoint.id, { action: "remove" }, "owner-user");
         await vi.advanceTimersByTimeAsync(31_000);
-        await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull());
+        await vi.waitFor(async () => {
+          await vi.advanceTimersByTimeAsync(1000);
+          expect(idleWorkSnapshot().active).toBe(0);
+          expect(coordinator.nextWakeAt()).toBeNull();
+        });
         const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
-        await vi.advanceTimersByTimeAsync(120_000);
-        expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
-        reads.mockRestore(); writes.mockRestore();
+        try {
+          await vi.advanceTimersByTimeAsync(120_000);
+          expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
+        } finally { reads.mockRestore(); writes.mockRestore(); }
         expect(f.calls).toEqual(["GET", "POST"]);
         expect(onError).not.toHaveBeenCalled();
       } finally {
@@ -6817,7 +6822,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
           await deleteCompany(db, f.fixture.companyId);
           deleted = true;
           await vi.advanceTimersByTimeAsync(31_000);
-          await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull());
+          await vi.waitFor(async () => {
+            await vi.advanceTimersByTimeAsync(1000);
+            expect(idleWorkSnapshot().active).toBe(0);
+            expect(coordinator.nextWakeAt()).toBeNull();
+          });
           expect(await db.select().from(chatDiscordCommandOwners)
             .where(eq(chatDiscordCommandOwners.endpointId, f.endpoint.id))).toEqual(owners);
           await f.service.shutdown();

server/src/__tests__/delivery-work-coordinator.test.ts

@@ -59,6 +59,32 @@ describe("delivery work coordinator", () => {
       expect(s.onError).not.toHaveBeenCalled();
     } finally { clock.mockRestore(); }
   });
+  it("honors a future backlog deadline after failure while preserving the fallback retry", async () => {
+    const s = setup(), t = task();
+    let next: number | null = Date.now() + 1000;
+    t.run.mockRejectedValueOnce(new Error("One item failed"));
+    await s.coordinator.register(DELIVERY_QUEUES.feedback, { ...t, nextRunAt: () => next }).ready;
+    expect(s.coordinator.nextWakeAt()).toBe(next);
+    expect(s.onError).toHaveBeenCalledOnce();
+    next = null;
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(t.run).toHaveBeenCalledTimes(2);
+    expect(s.coordinator.nextWakeAt()).toBeNull();
+  });
+  it.each(["overdue", "later", "unreadable", "draining"])("keeps failure backoff for a %s deadline", async mode => {
+    const s = setup(), t = task(), now = Date.now();
+    t.run.mockImplementation(async () => {
+      if (mode === "draining") { s.canRun.mockReturnValue(false); s.canReconcile.mockReturnValue(true); }
+      throw new Error("Unavailable");
+    });
+    await s.coordinator.register(DELIVERY_QUEUES.feedback, { ...t, nextRunAt: () => {
+      if (mode === "unreadable") throw new Error("Deadline unavailable");
+      return mode === "overdue" ? now : now + (mode === "later" ? 60_000 : 1000);
+    } }).ready;
+    expect(s.coordinator.nextWakeAt()).toBe(now + 5000);
+    await vi.advanceTimersByTimeAsync(4999);
+    expect(t.run).toHaveBeenCalledOnce();
+  });
   it("schedules maintenance without holding idle intent, and lets new work preempt it", async () => {
     const s = setup(), t = task();
     let next: number | null = Date.now() + 60_000;

server/src/services/chat-discord-command-registration-store.test.ts

@@ -860,6 +860,27 @@ suite("Discord command ownership store (real PostgreSQL, no network)", () => {
     expect(errors).toEqual([]);
   });
 
+  it("drains healthy commands promptly while one entry remains in failure backoff", async () => {
+    await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration);
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const entries = Array.from({ length: 300 }, (_, index) => ({
+      id: `command-${index}`, companyId: `company-${index % 10}`, connectionId: `connection-${index}`, dueAt: 0,
+    }));
+    const check = vi.fn(async (id: string) => {
+      if (id === "command-0") throw new Error("Unavailable credential");
+      return { ...entries.find(entry => entry.id === id)!, dueAt: Date.now() + 300_000 };
+    });
+    const onError = vi.fn();
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+    await registerDiscordCommandWork(coordinator, { load: async () => entries, check }, () => true).ready;
+    await vi.advanceTimersByTimeAsync(15_000);
+    expect(new Set(check.mock.calls.map(([id]) => id)).size).toBe(300);
+    expect(check.mock.calls.filter(([id]) => id === "command-0")).toHaveLength(1);
+    expect(onError).toHaveBeenCalledOnce();
+    await vi.advanceTimersByTimeAsync(45_000);
+    expect(check.mock.calls.filter(([id]) => id === "command-0")).toHaveLength(2);
+    expect(check.mock.calls.filter(([id]) => id !== "command-0")).toHaveLength(299);
+  });
   it("recovers queued scopes after an unknown hint instead of dropping the wake", async () => {
     await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration);
     vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });

server/src/services/delivery-work-coordinator.ts

@@ -78,9 +78,16 @@ export function createDeliveryWorkCoordinator(input: {
           }
         }).catch(error => {
           schedule(task.retryMs);
+          let failure = error;
+          try {
+            const next = input.canRun() ? task.nextRunAt?.() : null;
+            if (next != null && next > Date.now()) schedule(0, next);
+          } catch (deadlineError) {
+            failure = new AggregateError([error, deadlineError], "Delivery failure deadline could not be read");
+          }
           // Logging must never turn a recoverable sweep failure into an
           // unhandled rejection or prevent other queues from running.
-          try { if (!workerStopped) input.onError(error, queue); } catch { /* Recovery remains scheduled. */ }
+          try { if (!workerStopped) input.onError(failure, queue); } catch { /* Recovery remains scheduled. */ }
         }).finally(() => {
           running = null;
           controller = null;