fix(server): preserve backlog deadlines after partial failure

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-25 02:11:08 -05:00
1 parent fe1c886a3d
commit d3187f76a4
7 files changed
+126 -6

No files matched your search

+34
View File
@@ -131,3 +131,37 @@ producer integrations and six expanded deletion cases pass after propagation;
rollback restores conversation/publication history. The finding remains recorded
in the prerequisite document rather than treating the earlier endpoint-only
fixture as sufficient evidence.
## Failure-path capacity and CI isolation repair
Required CI at 58a1ac7595 found one command-inventory select after the removal test
had observed `nextWakeAt() === null`. That scheduler observation can occur during
an admitted reload; it does not prove notification settlement. The assertion then
threw before restoring DB spies and contaminated a later Slack grant-transaction
test with recursive spies. The quiet-window tests now wait for both no tracked
active work and no deadline, and restore spies in `finally`. The downstream Slack
test is included in verification. The CI failure, rather than a local reproducer,
is the evidence for the original race; no timeout or SQL-quiet assertion was relaxed.
Review also identified a genuine failed-item throughput bug: the shared
coordinator ignored `nextRunAt` after a thrown sweep and always waited its fallback
retry. A controlled 300-command regression with one persistent credential failure
reproduced only 25 attempted commands within 15 seconds. The coordinator now keeps
its fallback scheduled, then allows an explicitly future worker deadline to
preempt it while dispatch is admitted. An overdue/null/unreadable deadline cannot
cause a tight error retry loop; drain retains the ordinary fallback. A failed
deadline reader is reported together with the original failure. Workers without
a deadline keep their existing behavior. Domain retry/backoff remains owned by
the worker; no provider policy enters the coordinator.
After repair, all 300 commands are attempted within 15 seconds, the failed entry
is attempted once in that window, and its later retry does not recheck the 299
healthy entries. Five coordinator cases verify future deadlines and fallback
behavior for overdue, later, unreadable, and drain states. All 87 command,
coordinator, and health cases pass; eight real automatic-registration/Slack cases
pass (1,412 other integration cases deselected); 57 delivery/warm-standby/startup
cases pass. Server typecheck/build pass. The health parent is propagated through
company cleanup, retaining both notification paths in the three overlapping
producer sites. The six prerequisite cleanup cases pass. Remote acceptance of
the new Discord head remains required.
@@ -1180,3 +1180,15 @@ targeted health/producer/consumer cases and server typecheck/build pass. Publish
the repair to #17554, propagate through company-deletion #17561 into Discord
#17557, and require fresh remote gates. No PR in this stack is merged. The next
new owner remains chat runtime discovery; the full remaining ledger is unchanged.
Published health security repair is 9ef6e4f4c8; fresh security and Greptile checks
pass, with Actions still running at inspection. Company cleanup inherits it at
e8febb30eb; six cleanup tests pass. Discord inherits it at fe1c886a3d, preserving
both command and scoped health intents. Its previous-head CI exposed a flaky
quiet-window/spying fixture and review reproduced failure-path backlog throttling.
The repaired coordinator honors actual future deadlines without zero-delay error
loops or changing domain backoff. All 152 selected command/coordinator/health,
producer/Slack, delivery/standby/startup cases and server types/build pass. Publish
for fresh CI/review. The full inventory, #17490 CI failure, and stack/master
reconciliation remain open; no merge authority is implied.
@@ -174,3 +174,14 @@ Next gates: publish the stacked PR, complete all remote checks
and reviews, repair the health parent's fairness and producer-coverage findings,
and propagate its changes. Full original inventory and earlier unresolved CI
remain required. No merge is authorized.
Review at 58a1ac7595 identified a failed-item capacity regression in the shared
coordinator and a quiet-window test that leaked spies on assertion failure. The
300-entry failure-backlog test reproduced 25 attempts; the repair honors future
worker deadlines while preserving fallback/backoff and drain. The integration
fixture now waits for active-work settlement as well as an empty scheduler and
always restores spies. Verification: 87 worker/coordinator/health, eight producer
and affected Slack, 57 existing lifecycle/startup tests; server types/build pass.
The health security parent is inherited through company cleanup. See the owning
Discord document for failure evidence and exact scope. Fresh remote gates remain.
@@ -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();
@@ -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;
@@ -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 });
@@ -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;