mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 13:20:54 +01:00
fix(server): preserve backlog deadlines after partial failure
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
fe1c886a3d
commit
d3187f76a4
7 files changed
+126
-6
No files matched your search
@@ -131,3 +131,37 @@ producer integrations and six expanded deletion cases pass after propagation;
|
|||||||
rollback restores conversation/publication history. The finding remains recorded
|
rollback restores conversation/publication history. The finding remains recorded
|
||||||
in the prerequisite document rather than treating the earlier endpoint-only
|
in the prerequisite document rather than treating the earlier endpoint-only
|
||||||
fixture as sufficient evidence.
|
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
|
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
|
#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.
|
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 reviews, repair the health parent's fairness and producer-coverage findings,
|
||||||
and propagate its changes. Full original inventory and earlier unresolved CI
|
and propagate its changes. Full original inventory and earlier unresolved CI
|
||||||
remain required. No merge is authorized.
|
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);
|
expect(deadline).toBeGreaterThan(Date.now() + 200_000);
|
||||||
await f.service.configure(f.endpoint.id, { action: "remove" }, "owner-user");
|
await f.service.configure(f.endpoint.id, { action: "remove" }, "owner-user");
|
||||||
await vi.advanceTimersByTimeAsync(31_000);
|
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");
|
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
|
||||||
await vi.advanceTimersByTimeAsync(120_000);
|
try {
|
||||||
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
|
await vi.advanceTimersByTimeAsync(120_000);
|
||||||
reads.mockRestore(); writes.mockRestore();
|
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
|
||||||
|
} finally { reads.mockRestore(); writes.mockRestore(); }
|
||||||
expect(f.calls).toEqual(["GET", "POST"]);
|
expect(f.calls).toEqual(["GET", "POST"]);
|
||||||
expect(onError).not.toHaveBeenCalled();
|
expect(onError).not.toHaveBeenCalled();
|
||||||
} finally {
|
} finally {
|
||||||
@@ -6817,7 +6822,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
|||||||
await deleteCompany(db, f.fixture.companyId);
|
await deleteCompany(db, f.fixture.companyId);
|
||||||
deleted = true;
|
deleted = true;
|
||||||
await vi.advanceTimersByTimeAsync(31_000);
|
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)
|
expect(await db.select().from(chatDiscordCommandOwners)
|
||||||
.where(eq(chatDiscordCommandOwners.endpointId, f.endpoint.id))).toEqual(owners);
|
.where(eq(chatDiscordCommandOwners.endpointId, f.endpoint.id))).toEqual(owners);
|
||||||
await f.service.shutdown();
|
await f.service.shutdown();
|
||||||
|
|||||||
@@ -59,6 +59,32 @@ describe("delivery work coordinator", () => {
|
|||||||
expect(s.onError).not.toHaveBeenCalled();
|
expect(s.onError).not.toHaveBeenCalled();
|
||||||
} finally { clock.mockRestore(); }
|
} 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 () => {
|
it("schedules maintenance without holding idle intent, and lets new work preempt it", async () => {
|
||||||
const s = setup(), t = task();
|
const s = setup(), t = task();
|
||||||
let next: number | null = Date.now() + 60_000;
|
let next: number | null = Date.now() + 60_000;
|
||||||
|
|||||||
@@ -860,6 +860,27 @@ suite("Discord command ownership store (real PostgreSQL, no network)", () => {
|
|||||||
expect(errors).toEqual([]);
|
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 () => {
|
it("recovers queued scopes after an unknown hint instead of dropping the wake", async () => {
|
||||||
await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration);
|
await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration);
|
||||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||||
|
|||||||
@@ -78,9 +78,16 @@ export function createDeliveryWorkCoordinator(input: {
|
|||||||
}
|
}
|
||||||
}).catch(error => {
|
}).catch(error => {
|
||||||
schedule(task.retryMs);
|
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
|
// Logging must never turn a recoverable sweep failure into an
|
||||||
// unhandled rejection or prevent other queues from running.
|
// 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(() => {
|
}).finally(() => {
|
||||||
running = null;
|
running = null;
|
||||||
controller = null;
|
controller = null;
|
||||||
|
|||||||
Reference in new issue
Block a user