From 1604a9fcbaaa231ed0c8499ba5d3cc0f6af7ea05 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 03:05:04 -0500 Subject: [PATCH] fix(server): coalesce receiver recovery without redundant stop wakes Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 17 +++++++++++ ...-25-chat-receiver-discovery-sleep-audit.md | 17 +++++++++++ .../chat-channels.integration.test.ts | 25 ++++++++++++++++- .../src/__tests__/chat-receiver-work.test.ts | 28 +++++++++++++++++++ server/src/services/chat-channels.ts | 5 ++-- .../chat-channels/runtime/ownership.ts | 6 ++-- .../chat-channels/runtime/receiver-work.ts | 11 ++++---- 7 files changed, 98 insertions(+), 11 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 aebb2957e58..9a195b95555 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1211,3 +1211,20 @@ Parent Discord #17557 at d3187f76a4 and company cleanup #17561 at e8febb30eb now have no failed or pending checks (Paperclip Review skipped). No merge was made. The older #17490 E2E cancellation failure remains unresolved. The entire original inventory stays in scope, with voice the next separate reason after receivers. + + +## Receiver review repair + +The real company-removal producer regression reproduced two inventory reads for +one observed removal. Discovery-initiated stops now suppress their own completion +wake; independent token loss still wakes discovery after listener retirement. +A full reconstruction also covers its already-pending scoped hints. Hints that +arrive during the load remain dirty, and narrower writes cannot bypass a failed +full reconstruction's retry deadline. + +Both redundant-load regressions failed before repair. All 45 selected receiver, +startup and real Discord producer tests pass after repair, including the added +full-refresh failure/backoff case. Server typecheck and build pass. Prior broad +consumer/workspace verification remains recorded above; these 45 overlap that +coverage and are not an additional unique total. Publication requires fresh +remote CI and review. The full migration inventory remains outstanding. diff --git a/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md b/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md index a2b16518398..34bd6d4dc65 100644 --- a/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md +++ b/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md @@ -231,3 +231,20 @@ Master was fetched again at 0d73505098. Company prerequisite #17561 at e8febb30e and Discord parent #17557 at d3187f76a4 now have no failed or pending checks; Paperclip Review is skipped. Fresh review-content reconciliation and final stack rebasing/verification remain distinct from these check results. + + +## Receiver review repair + +The real company-removal producer regression reproduced two inventory reads for +one observed removal. Discovery-initiated stops now suppress their own completion +wake; independent token loss still wakes discovery after listener retirement. +A full reconstruction also covers its already-pending scoped hints. Hints that +arrive during the load remain dirty, and narrower writes cannot bypass a failed +full reconstruction's retry deadline. + +Both redundant-load regressions failed before repair. All 45 selected receiver, +startup and real Discord producer tests pass after repair, including the added +full-refresh failure/backoff case. Server typecheck and build pass. Prior broad +consumer/workspace verification remains recorded above; these 45 overlap that +coverage and are not an additional unique total. Publication requires fresh +remote CI and review. The full migration inventory remains outstanding. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 65c7d4a44fd..ce85806daf8 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,4 +1,4 @@ -import { notifyChatEndpointConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; +import { notifyChatEndpointConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; import { configureAgentLifecycle } from "../services/agent-lifecycle.js"; import { deleteCompany } from "../services/company-deletion.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; @@ -6799,6 +6799,29 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } }); + it("does not rediscover a receiver stopped by a company removal refresh", async () => { + const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 600_000, discordGatewayLeaseRenewalIntervalMs: 300_000 }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + let reads: ReturnType | undefined; + try { + await f.configure(); await f.service.registerReceiverWork(coordinator, () => true).ready; + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); }); + const owned = f.runtime.get(f.endpoint.id)!; + reads = vi.spyOn(db, "select"); + await db.transaction(async tx => { + await notifyChatCompanyConfiguration(tx, f.fixture.companyId); + await tx.delete(chatEndpoints).where(eq(chatEndpoints.id, f.endpoint.id)); + }); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); }); + expect(owned.shutdown).toHaveBeenCalledOnce(); + expect(reads).toHaveBeenCalledOnce(); expect(onError).not.toHaveBeenCalled(); + } finally { + reads?.mockRestore(); await coordinator.stop(); vi.useRealTimers(); await f.service.shutdown(); + await db.update(chatEndpoints).set({ status: "archived" }).where(eq(chatEndpoints.id, f.endpoint.id)); + } + }); + it("takes over a persisted foreign receiver lease at its deadline through the worker", async () => { const f = await registrationFixture(); await f.configure(); await f.service.shutdown(); diff --git a/server/src/__tests__/chat-receiver-work.test.ts b/server/src/__tests__/chat-receiver-work.test.ts index b895a4f9a50..8cbfe0e3577 100644 --- a/server/src/__tests__/chat-receiver-work.test.ts +++ b/server/src/__tests__/chat-receiver-work.test.ts @@ -153,6 +153,34 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu service.load = original; reconcile.mockResolvedValue("owned"); await settled(); errors.length = 0; await quiet(); }); + it("coalesces a full refresh with its already pending scoped hints", async () => { + const service = chatReceiverSchedule(db, { reconcile, stop }); + await start(service); await settled(); + const load = vi.spyOn(service, "load"); + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "*"); + await endpoint({}, tx as unknown as typeof db); + await endpoint({}, tx as unknown as typeof db); + }); + await settled(); expect(reconcile).toHaveBeenCalledTimes(2); + expect(load).toHaveBeenCalledOnce(); expect(load.mock.calls[0][0]).toBeUndefined(); + load.mockRestore(); await quiet(); + }); + it("keeps full-refresh backoff when a narrower producer arrives", async () => { + const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load; + await start(service); await settled(); + const load = vi.spyOn(service, "load").mockRejectedValue(new Error("inventory unavailable")); + await db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "*")); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors).toHaveLength(1); }); + const retryAt = coordinator!.nextWakeAt(); + const row = await endpoint(); + await quiet(10_000); + expect(load).toHaveBeenCalledOnce(); expect(coordinator!.nextWakeAt()).toBe(retryAt); + load.mockImplementation(original); + await settled(); expect(reconcile).toHaveBeenCalledWith(row.id); + expect(load).toHaveBeenCalledTimes(2); expect(load.mock.calls[1][0]).toBeUndefined(); + errors.length = 0; load.mockRestore(); await quiet(); + }); it("recovers malformed scopes without losing a subsequent valid producer", async () => { await start(); await settled(); await db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "unknown:old")); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 9fbcb10ab08..0d1ae9f5ac4 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1125,7 +1125,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { runtimeVersions.delete(endpoint.id); } if (discordOwnership) { - await stopDiscordGatewayOwnership(discordOwnership, false); + await stopDiscordGatewayOwnership(discordOwnership, { removeRuntime: false }); } throw error; } @@ -1145,7 +1145,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const receiverSchedule = chatReceiverSchedule(db, { stop: async id => { const ownership = discordGatewayOwnerships.get(id); - if (ownership) await stopDiscordGatewayOwnership(ownership); + // Discovery already observed removal; only independent ownership loss needs another wake. + if (ownership) await stopDiscordGatewayOwnership(ownership, { notifyStopped: false }); }, reconcile: async id => { const record = await endpointRecord(id); diff --git a/server/src/services/chat-channels/runtime/ownership.ts b/server/src/services/chat-channels/runtime/ownership.ts index 4f2860eac30..4a673d41505 100644 --- a/server/src/services/chat-channels/runtime/ownership.ts +++ b/server/src/services/chat-channels/runtime/ownership.ts @@ -74,7 +74,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) { async function stopDiscordGatewayOwnership( ownership: DiscordGatewayOwnership, - removeRuntime = true, + stopOptions: { removeRuntime?: boolean; notifyStopped?: boolean } = {}, ): Promise { if (ownership.stopPromise) return ownership.stopPromise; ownership.stopping = true; @@ -84,7 +84,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) { } ownership.stopPromise = (async () => { if ( - removeRuntime && + stopOptions.removeRuntime !== false && ownership.context.endpointRuntime && runtime.get(ownership.endpointId) === ownership.context.endpointRuntime ) { @@ -105,7 +105,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) { if (discordGatewayOwnerships.get(ownership.endpointId) === ownership) { discordGatewayOwnerships.delete(ownership.endpointId); } - if (!options.isShuttingDown()) options.onStopped?.(ownership.endpointId); + if (!options.isShuttingDown() && stopOptions.notifyStopped !== false) options.onStopped?.(ownership.endpointId); })(); discordGatewayStopTasks.add(ownership.stopPromise); void ownership.stopPromise.finally(() => { diff --git a/server/src/services/chat-channels/runtime/receiver-work.ts b/server/src/services/chat-channels/runtime/receiver-work.ts index 49b4cb509d4..5cb664d87ab 100644 --- a/server/src/services/chat-channels/runtime/receiver-work.ts +++ b/server/src/services/chat-channels/runtime/receiver-work.ts @@ -28,7 +28,7 @@ export function registerChatReceiverWork( function next() { let nextAt: number | null = null; const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; - for (const hint of dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0)); + for (const hint of dirty.has("*") ? ["*"] : dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0)); const unavailable = dirtyScopes(); for (const entry of entries.values()) if (entry.dueAt !== null && !unavailable.some(scope => matches(entry, scope))) take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0)); @@ -42,10 +42,11 @@ export function registerChatReceiverWork( if (!canRun() || Date.now() < earliest) return; earliest = Date.now() + 1000; const errors: unknown[] = []; - const requested = [...dirty].filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); - for (const hint of requested) dirty.delete(hint); + const requested = (dirty.has("*") ? ["*"] : [...dirty]).filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); + const covered = requested.includes("*") ? [...dirty] : requested; + for (const hint of covered) dirty.delete(hint); for (let index = 0; index < requested.length; index++) { - if (signal.aborted || !canRun()) { for (const hint of requested.slice(index)) dirty.add(hint); return; } + if (signal.aborted || !canRun()) { for (const hint of requested.includes("*") ? covered : requested.slice(index)) dirty.add(hint); return; } const hint = requested[index]; let scope: ReceiverScope | undefined; let decoded = false; @@ -57,7 +58,7 @@ export function registerChatReceiverWork( entries.delete(entry.id); failures.delete(entry.id); } for (const entry of rows) entries.set(entry.id, entry); - scopeFailures.delete(hint); + if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint); } catch (error) { const retryHint = decoded ? hint : "*"; dirty.add(retryHint); failed(scopeFailures, retryHint); errors.push(error);