From 8df31d46c09849ecb1fbb5e6de8b0ae5d820a9d5 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 01:38:33 -0500 Subject: [PATCH] fix(server): preserve Discord backlog and queued scopes Retain the real company-deletion regression pending its prerequisite dependency-order repair. Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 13 +++++ .../chat-channels.integration.test.ts | 56 +++++++++++++++++++ .../chat-channels/discord/command-work.ts | 6 +- ...discord-command-registration-store.test.ts | 46 ++++++++++++++- 4 files changed, 118 insertions(+), 3 deletions(-) diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 179b1c91c14..01d6199f8ec 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1119,3 +1119,16 @@ batching cases pass; server types pass. Propagate the parent and apply the same bounded continuation to Discord, with its own regression. The Discord review also requests preserving queued scopes after malformed hints and broader producer coverage. Its initial head has Greptile 5/5 but these findings remain open. + + +Discord review repairs are in progress. Both new worker regressions failed before +repair: the 300-command backlog did not drain promptly, and an unknown hint lost +later scopes. Bounded backlog continuations and full reconstruction after malformed +scope parsing repair them; all 62 combined command/health/batching cases pass. +Real credential rotation also passes. The company-deletion producer test exposed +an existing dependency-order failure: agents are deleted before chat endpoints +whose assigned-agent foreign key is RESTRICT. Configuring the fixture lifecycle +removed the initial missing-worker failure but reproduced this actual SQL failure. +Keep the failing producer test; repair the deletion dependency in its own stack +PR before publishing the Discord follow-up. Current local changes are not final +acceptance. The original full inventory and remaining CI gates are still open. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index c2014ea8a80..207231db903 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,5 @@ +import { configureAgentLifecycle } from "../services/agent-lifecycle.js"; +import { deleteCompany } from "../services/company-deletion.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; import { readFreshWorkspaceReadiness } from "../services/fresh-workspace-continuation.js"; import { requestTaskRetry } from "../services/task-retry.js"; @@ -6781,6 +6783,60 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } }); + it.each(["credential rotation", "company deletion"] as const)("updates scheduled registration after real %s", async (mutation) => { + const f = await registrationFixture(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + let deleted = false; + let lifecycle: ReturnType | undefined; + try { + await f.service.registerCommandWork(coordinator, () => true).ready; + await f.configure(); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now())); + const before = coordinator.nextWakeAt()!; + const owners = await db.select().from(chatDiscordCommandOwners) + .where(eq(chatDiscordCommandOwners.endpointId, f.endpoint.id)); + if (mutation === "credential rotation") { + await f.service.configure(f.endpoint.id, { + action: "reconnect", credentials: { ...f.credentials, botToken: "discord-rotated-secret" }, + }, "owner-user"); + const [saved] = await db.select().from(chatActions).where(and( + eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "discord_command_registration"), + )); + const expected = Date.parse(String(saved.result!.retryAt)); + expect(expected).toBeGreaterThan(before); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(expected)); + expect(f.calls).toEqual(["GET", "POST", "GET"]); + } else { + lifecycle = configureAgentLifecycle(db, { + requiredPluginIds: async () => [], runHost: async () => "complete", runPlugin: async () => "complete", + }); + await deleteCompany(db, f.fixture.companyId); + deleted = true; + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull()); + expect(await db.select().from(chatDiscordCommandOwners) + .where(eq(chatDiscordCommandOwners.endpointId, f.endpoint.id))).toEqual(owners); + await f.service.shutdown(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + 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 { + await lifecycle?.stop(); + await coordinator.stop(); vi.useRealTimers(); + if (deleted) await f.service.shutdown(); + else await retireRegistrationFixture(f.service, f.endpoint.id); + } + }); + it("keeps mention setup usable on registration failure and upgrades automatically when due", async () => { const f = await registrationFixture("unavailable"); try { diff --git a/server/src/services/chat-channels/discord/command-work.ts b/server/src/services/chat-channels/discord/command-work.ts index dabffa73bc6..28169037824 100644 --- a/server/src/services/chat-channels/discord/command-work.ts +++ b/server/src/services/chat-channels/discord/command-work.ts @@ -54,8 +54,9 @@ export function registerDiscordCommandWork( return; } const hint = requested[index]; - const scope = decodeScope(hint); + let scope: CommandRegistrationScope | undefined; try { + scope = decodeScope(hint); const rows = await service.load(scope, signal); for (const entry of deadlines.values()) if (matches(entry, scope)) deadlines.delete(entry.id); for (const entry of rows) deadlines.set(entry.id, entry); @@ -63,7 +64,7 @@ export function registerDiscordCommandWork( const entry = deadlines.get(id); if (!entry || entry.dueAt > Date.now()) failures.delete(id); } - } catch (error) { dirty.add(hint); unavailable.push(scope); errors.push(error); } + } catch (error) { dirty.add(scope ? hint : "*"); unavailable.push(scope); errors.push(error); } } const due = [...deadlines.values()] .filter(entry => !unavailable.some(scope => matches(entry, scope)) && @@ -84,6 +85,7 @@ export function registerDiscordCommandWork( errors.push(error); } } + if (due.length > batch.entries.length) earliest = Date.now() + 1_000; if (errors.length) throw new AggregateError(errors, "Discord command registration could not settle"); }, hasPending: async () => dirty.size > 0, diff --git a/server/src/services/chat-discord-command-registration-store.test.ts b/server/src/services/chat-discord-command-registration-store.test.ts index 8960f7a0bf1..ae0a807aedc 100644 --- a/server/src/services/chat-discord-command-registration-store.test.ts +++ b/server/src/services/chat-discord-command-registration-store.test.ts @@ -1,5 +1,5 @@ import { reconcileDatabaseWork } from "@paperclipai/db"; -import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { discordCommandSchedule } from "./chat-channels/discord/command-schedule.js"; import { registerDiscordCommandWork } from "./chat-channels/discord/command-work.js"; import { notifyDiscordCommandRegistration } from "./chat-channels/discord/command-work-notifications.js"; @@ -841,4 +841,48 @@ suite("Discord command ownership store (real PostgreSQL, no network)", () => { expect(errors).toEqual([]); }); + it("drains a large due backlog without a thirty-second delay between bounded batches", 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) => ({ ...entries.find(entry => entry.id === id)!, dueAt: Date.now() + 300_000 })); + const errors: unknown[] = []; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await registerDiscordCommandWork(coordinator, { load: async () => entries, check }, () => true).ready; + expect(check).toHaveBeenCalledTimes(25); + await vi.advanceTimersByTimeAsync(15_000); + expect(check).toHaveBeenCalledTimes(300); + expect(new Set(check.mock.calls.map(([id]) => id)).size).toBe(300); + await sqlQuiet(120_000); + expect(check).toHaveBeenCalledTimes(300); + expect(errors).toEqual([]); + }); + + 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 }); + const entry = { id: "endpoint", companyId: "company", connectionId: "connection", dueAt: 0 }; + let owed = false; + const check = vi.fn(async () => { owed = false; return null; }); + const errors: unknown[] = []; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await registerDiscordCommandWork(coordinator, { load: async () => owed ? [entry] : [], check }, () => true).ready; + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatCommandRegistration, "unknown:legacy"); + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: entry.id }); + owed = true; + }); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1_000); + expect(errors).toHaveLength(1); + }, { timeout: 5_000 }); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(check).toHaveBeenCalledTimes(1)); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await sqlQuiet(120_000); + expect(errors).toHaveLength(1); + }); + });