diff --git a/doc/idle-teams-attempt-recovery.md b/doc/idle-teams-attempt-recovery.md new file mode 100644 index 00000000000..a1bd96991c2 --- /dev/null +++ b/doc/idle-teams-attempt-recovery.md @@ -0,0 +1,74 @@ +# Teams abandoned attempts and idle work + +Teams transport-attempt recovery now runs from committed transfer changes and +saved claim deadlines. The shared coordinator and scheduler own its lifecycle. +Together with consent expiry, this replaces the periodic Teams maintenance +callback. It adds no table, schema or independent timer. + +## Durable obligation + +A transfer in `consent_sending`, `uploading`, `file_info_sending`, or `conflict` +with `attemptExpiresAt` has an outstanding claim. The existing protocol writes +an attempt ID and a 90-second deadline before provider I/O. A normal receipt +clears both. A conflicting callback retains the original in-flight claim until +its deadline; the conflict is not proof that the original operation stopped. + +The protocol's central versioned update registers an awaited notification before +the write, within the existing transaction. This includes normal claims, +settlement, callbacks and caller-owned operator resolution. The DB work-signal +boundary owns outer commit, rollback and uncertain transaction settlement. +Inserts create unused consent, so they do not create an attempt obligation. + +Startup reconstructs eligible attempts in 100-row UUID pages. A committed hint +reloads its transfer ID. PostgreSQL timestamps round upward to a millisecond. +An inspection does not move the saved deadline. A changed claim is reloaded +before recovery; settled or deleted rows remove the cached obligation. + +## Recovery preserves uncertainty + +At the deadline, the existing Teams owner locks publication then transfer, +checks integrity, and rechecks the current phase and deadline. It clears the +expired attempt and changes a sending phase to its corresponding unknown outcome. +Conflict stays conflict with its original reason and private evidence. Publication +projection and transfer mutation remain atomic. Lock waits remain bounded. + +Recovery does not call a provider, resend a card or file, manufacture a receipt, +or assert that an unknown effect never occurred. A delayed foreground receipt +still has to match its current attempt ID and phase. Existing operator resolution +and on-use authorization rules remain unchanged. + +Unknown outcomes with no live attempt are passive review evidence. Unused consent +is owned by the separate consent-expiry worker. Neither keeps attempt recovery +querying. Failed loads, locks or projections retain bounded retry work; fair +batches allow healthy rows and other companies to progress. Worker errors redact +SQL parameters and private transfer envelopes. + +## Lifecycle and ownership + +The Teams module selects phases and deadlines. The existing protocol performs +recovery. Shared startup only registers the worker and admission gates. The +shared scoped worker coalesces writes, preserves dirty scopes through drain, +reconstructs after restart, and joins in-flight work during shutdown. An unresolved +write may still need transaction-status reconciliation while dispatch is paused; +this is not an empty maintenance scan. + +The explicit combined `expireAndRecover` and attempts-only diagnostic methods +remain available. Production `processPublicationMaintenance` no longer invokes +Teams recovery. Other chat maintenance reasons are separate pending migrations. + +Supported writes use this process's owning DB client. Another active server or +independent SQL writer requires a separate notification contract. Cloud external +alarm coverage and conservative sleep eligibility are unchanged. A registered +local deadline cannot wake a stopped host. This migration proves worker quiet, +not measured provider suspension or fleet savings. + +## Verification + +Real PostgreSQL tests exercise all sending stages, conflict during a live send, +normal settlement, interrupted receipt persistence, saved deadlines across restart, +quiet history, drain/resume, rollback, precision, atomic projection failure, +102-entry paging/fairness and shutdown. The real Board producer integration +registers the worker before sending, observes the saved claim while the provider +receipt is held, and verifies that the committed receipt disarms recovery. Existing +Teams authority, consent and publication cases remain covered. See the +[audit and failure ledger](plans/2026-10-25-teams-attempt-recovery-sleep-audit.md). 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 e3c124bd27f..176a3a3f361 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1377,3 +1377,16 @@ Slice 26 is published as [#17590](https://github.com/paperclipai/paperclip/pull/ at 5ddb4543ed, based on #17586. It is attached; remote gates remain required. Slice 27 begins the separate Teams abandoned transport-attempt audit. The full original inventory, older findings and final master reconciliation remain open. + + +## Slice 27: Teams abandoned transport attempts + +Attempt recovery now follows committed claims and saved 90-second deadlines. +It preserves unknown outcomes and performs no provider I/O. This removes the last +Teams periodic maintenance callback; unused consent remains the separate parent +owner. The [audit](2026-10-25-teams-attempt-recovery-sleep-audit.md) records 235 +unique selected passing cases and preserved initial test failures. Workspace +typecheck and build pass; publication and remote acceptance remain open. +#17590 has Greptile 5/5; CI is running. #17586 has new review findings for writer +support and report-query indexing. Fix at that parent and propagate before +acceptance. All original remaining work and previous unresolved findings remain. diff --git a/doc/plans/2026-10-25-teams-attempt-recovery-sleep-audit.md b/doc/plans/2026-10-25-teams-attempt-recovery-sleep-audit.md index c8bb6026354..72592be941f 100644 --- a/doc/plans/2026-10-25-teams-attempt-recovery-sleep-audit.md +++ b/doc/plans/2026-10-25-teams-attempt-recovery-sleep-audit.md @@ -2,7 +2,7 @@ Baseline: consent expiry PR #17590 at 5ddb4543ed. Branch: codex/sleep-27-teams-attempt-recovery. -Status: owner audit; implementation and verification pending. +Status: implemented; targeted local acceptance and workspace types/build pass; publication and remote gates pending. ## Reason and source of truth @@ -41,3 +41,53 @@ during drain, nested rollback/outer commit, changes before deadline, and shutdow Exercise production wiring and ensure no periodic Teams scan remains. Keep provider calls outside transactions and prove recovery makes no new provider call. No change to external alarms, on-use authority or Cloud sleep eligibility. + +## Implementation and local evidence + +The new Teams-owned worker registers existing attempt deadlines. The central +versioned update signals both consent and attempt owners before writes. A narrow +recoverAbandonedAttempt method delegates to the original locked recovery logic. +The periodic scheduleTeamsFileMaintenance callback and its orchestration call are +removed. No other chat duty, schema or external wake policy is changed. + +- Five real-PostgreSQL/wiring suites pass 220 cases: 77 transfer protocol cases + and 143 publication, consent, consent-runtime and app cases. +- Fifteen selected production integration cases pass, including actual Board + claim/receipt activation and the parent expiry case plus existing Teams current + authority cases. The other 1,411 cases are skipped, not passed. +- These are 235 unique selected passing cases. The production claim test measures + 120 seconds with no select/transaction calls after the receipt settles. The + 102-entry recovery case drains healthy work from two companies while corrupt + evidence retains recovery. No recovery test causes another provider request. +- Server typecheck passes; workspace typecheck passes. Workspace build passes. Typecheck excludes test files. +- Full local test:run remains unclaimed due to the earlier embedded-PostgreSQL + host limitation. Full remote gates are required. No live Teams/suspension claim. + +Failure ledger: the first five-suite run passed 217 and failed two new assertions. +The drain fixture simulated an interrupted receipt transaction, then incorrectly +counted its pg_xact_status reconciliation as forbidden idle SQL. The rollback +fixture incorrectly rejected any future wake, including the legitimate shared +transaction-resolution deadline. It now reads committed eligibility and verifies +that no provider call occurred. Its asynchronous rejection is joined in finally. +The next run passed 219 and failed the drain fixture's wait for all idle holds to +clear while dispatch was deliberately paused. The final drain fixture uses a real +committed conflicting callback, retains its live claim through a quiet paused +window, and verifies retirement after resume. The interrupted receipt and failed +projection cases remain independently covered. Production behavior was not +relaxed to satisfy these assertions. + +Logs: /tmp/sleep27-teams-tests.log (initial failures), +/tmp/sleep27-teams-fixed-tests.log (drain hold assertion), +/tmp/sleep27-teams-final-tests.log (220 pass), +/tmp/sleep27-production-tests.log (15 pass), +/tmp/sleep27-server-types.log, /tmp/sleep27-workspace-types.log and +/tmp/sleep27-workspace-build.log (pass). + +## Stack gates + +Consent parent #17590 at 5ddb4543ed has Greptile 5/5 with no inline findings; +CI is still running. Voice-report #17586 now has two new Paperclip findings: +explicit single-active-process writer support and indexed report reconstruction. +They require owning-parent review/repair and propagation before stack acceptance. +Older #17490 E2E and #17580 deletion-policy findings and final master reconciliation +remain open. No stack PR is merged or the full objective complete. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 4c798afba0a..2aabbe2f370 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -81,10 +81,11 @@ describe("createChatReconciliationCoordinator", () => { expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,"); expect(source).toContain("chatChannels.registerTeamsConsentExpiryWork(deliveryWork,"); const teams = readFileSync(new URL("../services/chat-channels/microsoft-teams/file-delivery.ts", import.meta.url), "utf8"); - const remainingMaintenance = teams.slice(teams.indexOf("function scheduleTeamsFileMaintenance"), teams.indexOf("function registerTeamsConsentExpiryWork")); - expect(remainingMaintenance).toContain("protocol.recoverAbandonedAttempts("); - expect(remainingMaintenance).not.toContain("protocol.expireAndRecover("); - expect(remainingMaintenance).not.toContain("chatTeamsFileTransfers.expiresAt"); + expect(source).toContain("chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork,"); + expect(teams).not.toContain("scheduleTeamsFileMaintenance"); + expect(teams).not.toContain("selectDistinct"); + const chat = readFileSync(new URL("../services/chat-channels.ts", import.meta.url), "utf8"); + expect(chat).not.toContain("scheduleTeamsFileMaintenance"); }); it("separates live voice recovery from the remaining periodic maintenance", () => { diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index c3658146162..d434d69781e 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1203,6 +1203,47 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { companyId, assignedAgentId, replacementAgentId }; } + it("tracks the actual Board Teams claim and retires recovery when its receipt commits", async () => { + const context = await teamsFileAuthorityFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + context.controls.beforeConsentReceipt = async () => { entered(); await hold; }; + let send: ReturnType | undefined; + try { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await context.service.registerTeamsAttemptRecoveryWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBeNull(); + const attachment = await context.createFile(); + send = context.service.publishBoardMessage(context.endpoint.id, context.conversation.id, + "File with an active claim", `teams-attempt-${randomUUID()}`, "owner-user", [attachment.id]); + await ready; + const [claim] = await db.select().from(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.endpointId, context.endpoint.id)); + expect(claim.phase).toBe("consent_sending"); expect(context.consentCards).toHaveLength(1); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); expect(coordinator.nextWakeAt()).toBe(claim.attemptExpiresAt!.getTime()); + }, { timeout: 5000 }); + release(); await send; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + const [settled] = await db.select().from(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.id, claim.id)); + expect(settled.phase).toBe("awaiting_consent"); expect(settled.attemptId).toBeNull(); + 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(context.consentCards).toHaveLength(1); expect(context.fileCards).toEqual([]); + expect(context.uploadRequest).not.toHaveBeenCalled(); expect(errors).toEqual([]); + } finally { + release(); await send; + await coordinator.stop(); vi.useRealTimers(); + await retirePublicationFixture(context.service, context.endpoint.id); + // This test intentionally leaves unused consent; do not feed the next global expiry fixture. + await db.delete(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.endpointId, context.endpoint.id)); + } + }); + it("expires Teams consent from the committed Board transfer without periodic maintenance", async () => { const context = await teamsFileAuthorityFixture(); const errors: unknown[] = []; diff --git a/server/src/app.ts b/server/src/app.ts index 2ac96e978c7..037bc795025 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1302,6 +1302,7 @@ export async function createApp( chatChannels.voice.inbound.registerRetentionWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); emailChannels.start(); const reconcileChatPublicationMaintenance = async () => { diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index fad48a02cf1..335dff4f6ab 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1255,7 +1255,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { stageTerminalReceiptReactionRemovals, runtimeContextForRecord, scheduleMessageProcessing, processReceiptReaction, }); const { stageTeamsBoardFileIntent, deriveTeamsFileTransferAuthority } = createTeamsFileAuthority({ runtimeGeneration, credentialFingerprint, runtime, runtimeContexts, localRuntimeEpoch, runtimeVersions, lockCurrentPrincipalAuthorization, lifecycleRuntimeFence, assertFailedRetryReceipt, inboundWakeScope, failedChatRetrySource, isExplicitOperatorPublication, retryCommentIds }); - const { registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance } = createTeamsFileDelivery(db, { + const { registerTeamsAttemptRecoveryWork, registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication } = createTeamsFileDelivery(db, { endpointRecord, runtimeContextForRecord, runtimeFor, runtimeContexts, localRuntimeEpoch, runtime, deriveTeamsFileTransferAuthority, publicationFiles, scheduleMessageProcessing, processPendingPublications, withCredentialMutationLease, teamsFileRetryAfter, isShuttingDown: () => shuttingDown, @@ -3346,7 +3346,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }); async function processPublicationMaintenance(limit = 25) { - scheduleTeamsFileMaintenance(); scheduleGitHubMaintenance(); await enqueueInboundWakeupPublications(); await reconcileTerminalConfirmationActions(limit); @@ -3501,6 +3500,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor }); const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { + registerTeamsAttemptRecoveryWork, registerTeamsConsentExpiryWork, registerCommandWork, registerReceiverWork, diff --git a/server/src/services/chat-channels/microsoft-teams/attempt-recovery-work.ts b/server/src/services/chat-channels/microsoft-teams/attempt-recovery-work.ts new file mode 100644 index 00000000000..d32aff5594b --- /dev/null +++ b/server/src/services/chat-channels/microsoft-teams/attempt-recovery-work.ts @@ -0,0 +1,54 @@ +import { and, eq, gt, inArray, isNotNull, sql } from "drizzle-orm"; +import { chatTeamsFileTransfers, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js"; + +export const abandonedTeamsAttemptPhases = ["consent_sending", "uploading", "file_info_sending", "conflict"]; +type Entry = { id: string; companyId: string; dueAt: number }; + +export function notifyTeamsAttemptRecovery(tx: object, id: string) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatTeamsAttemptRecovery, id); +} + +/** Retires expired transport claims without retrying an unknown provider effect. */ +export function teamsAttemptRecoveryWork(db: Db, recover: (companyId: string, id: string) => Promise) { + async function load(id?: string, signal?: AbortSignal): Promise { + try { + const entries: Entry[] = []; let cursor: string | undefined; + for (;;) { + signal?.throwIfAborted(); + const rows = await db.select({ id: chatTeamsFileTransfers.id, companyId: chatTeamsFileTransfers.companyId, + dueAt: sql`ceil(extract(epoch from ${chatTeamsFileTransfers.attemptExpiresAt}) * 1000)`, + }).from(chatTeamsFileTransfers).where(and(inArray(chatTeamsFileTransfers.phase, abandonedTeamsAttemptPhases), + isNotNull(chatTeamsFileTransfers.attemptExpiresAt), + id ? eq(chatTeamsFileTransfers.id, id) : undefined, + cursor ? gt(chatTeamsFileTransfers.id, cursor) : undefined)).orderBy(chatTeamsFileTransfers.id).limit(100); + entries.push(...rows.map(row => ({ ...row, dueAt: Number(row.dueAt) }))); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + } catch { + // Never expose private transfer envelopes or SQL parameter errors. + throw new Error("Teams attempt recovery selection is temporarily unavailable"); + } + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + const current = (await load(entry.id, signal))[0]; + if (!current) return null; + if (current.dueAt > Date.now()) return current; + signal?.throwIfAborted(); + try { await recover(current.companyId, current.id); } + catch { throw new Error("Teams attempt recovery could not settle"); } + return (await load(entry.id, signal))[0] ?? null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatTeamsAttemptRecovery, + decodeScope: (hint: string) => hint === "*" ? undefined : hint, + matches: (entry: Entry, id?: string) => !id || entry.id === id, + load, check, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/chat-channels/microsoft-teams/file-delivery.ts b/server/src/services/chat-channels/microsoft-teams/file-delivery.ts index 396662a4e49..c87c045eb10 100644 --- a/server/src/services/chat-channels/microsoft-teams/file-delivery.ts +++ b/server/src/services/chat-channels/microsoft-teams/file-delivery.ts @@ -1,3 +1,4 @@ +import { teamsAttemptRecoveryWork } from "./attempt-recovery-work.js"; import { teamsConsentExpiryWork } from "./consent-expiry-work.js"; import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; import { createOutboundAttachments } from "../outbound-attachments.js"; @@ -14,8 +15,6 @@ import { chatCredentialMutationLease, type CredentialMutationLeaseGuard } from " import { and, eq, - inArray, - lte, } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -37,7 +36,6 @@ import { HttpError, conflict, } from "../../../errors.js"; -import { logger } from "../../../middleware/logger.js"; import { type ChatSdkCallbackEvent, @@ -376,51 +374,13 @@ export function createTeamsFileDelivery(db: Db, options: TeamsFileDeliveryOption } } - let teamsMaintenancePending = false; - let teamsMaintenanceAfter = 0; - function scheduleTeamsFileMaintenance() { - if ( - teamsMaintenancePending || - Date.now() < teamsMaintenanceAfter || - isShuttingDown() - ) - return; - teamsMaintenancePending = true; - teamsMaintenanceAfter = Date.now() + 30_000; - scheduleMessageProcessing(async () => { - try { - const scopes = await db - .selectDistinct({ companyId: chatTeamsFileTransfers.companyId }) - .from(chatTeamsFileTransfers) - .where( - and( - inArray(chatTeamsFileTransfers.phase, [ - "consent_sending", "uploading", "file_info_sending", "conflict", - ]), - lte(chatTeamsFileTransfers.attemptExpiresAt, new Date()), - ), - ); - const protocol = teamsTransferProtocol(); - // Separate from the ordinary message queue. Each company sweep bounds - // its row count and lock waits and performs no provider I/O. - for (const scope of scopes) { - if (isShuttingDown()) break; - await protocol.recoverAbandonedAttempts(scope.companyId, 100); - } - } catch { - // Selection errors are closed here too; private transfer envelopes and - // SQL parameters must not escape through the generic background logger. - logger.warn( - "Teams file maintenance will retry after a temporary failure", - ); - } finally { - teamsMaintenancePending = false; - } - }); - } function registerTeamsConsentExpiryWork(coordinator: ReturnType, canRun: () => boolean) { return teamsConsentExpiryWork(db, (companyId, id) => teamsTransferProtocol().expireUnusedConsent(companyId, id)) .register(coordinator, () => !isShuttingDown() && canRun()); } - return { registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance }; + function registerTeamsAttemptRecoveryWork(coordinator: ReturnType, canRun: () => boolean) { + return teamsAttemptRecoveryWork(db, (companyId, id) => teamsTransferProtocol().recoverAbandonedAttempt(companyId, id)) + .register(coordinator, () => !isShuttingDown() && canRun()); + } + return { registerTeamsAttemptRecoveryWork, registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication }; } diff --git a/server/src/services/chat-teams-file-transfers.test.ts b/server/src/services/chat-teams-file-transfers.test.ts index 504f921f276..d9fa8b4af22 100644 --- a/server/src/services/chat-teams-file-transfers.test.ts +++ b/server/src/services/chat-teams-file-transfers.test.ts @@ -1,3 +1,4 @@ +import { teamsAttemptRecoveryWork } from "./chat-channels/microsoft-teams/attempt-recovery-work.js"; import { teamsConsentExpiryWork } from "./chat-channels/microsoft-teams/consent-expiry-work.js"; import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { idleWorkSnapshot } from "./task-admission.js"; @@ -332,6 +333,200 @@ suite( }; } + describe("scheduled abandoned attempt recovery", () => { + let coordinator: ReturnType | undefined; + const transferIds: string[] = [], errors: unknown[] = []; + let enabled = true; + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + if (transferIds.length) await db.delete(chatTeamsFileTransfers).where(inArray(chatTeamsFileTransfers.id, transferIds.splice(0))); + errors.length = 0; enabled = true; + }); + async function transfer() { + const f = await fixture(600_000); transferIds.push(f.created.id); + f.opts.now = () => new Date(); + return f; + } + async function abandon(f: Awaited>, stage = "consent") { + if (stage !== "consent") { + await f.service().process(f.a.companyId, f.created.id); + await f.service().recordConsent(await f.event()); + if (stage === "file_info") await f.service().process(f.a.companyId, f.created.id); + } + const outcome = stage === "consent" ? "awaiting_consent" : stage === "upload" ? "file_info_pending" : "delivered"; + f.opts.project = async (_tx, row) => { if (row.phase === outcome) throw new Error("receipt projection interrupted"); }; + await expect(f.service().process(f.a.companyId, f.created.id)).rejects.toThrow("receipt projection interrupted"); + delete f.opts.project; + return (await f.read()).attemptExpiresAt!.getTime(); + } + function worker(recover: (companyId: string, id: string) => Promise) { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + const work = teamsAttemptRecoveryWork(db, recover); + return { work, ready: work.register(coordinator, () => enabled).ready }; + } + async function settled() { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 }); + } + async function quiet(ms = 10_000) { + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"), execute = vi.spyOn(db, "execute"); + try { await vi.advanceTimersByTimeAsync(ms); expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); } + finally { select.mockRestore(); transaction.mockRestore(); execute.mockRestore(); } + } + it("leaves unused consent and retained terminal history quiet", async () => { + const f = await transfer(), cancelled = await transfer(); + await cancelled.service().cancel(cancelled.a.companyId, cancelled.created.id); + await worker(f.service().recoverAbandonedAttempt).ready; await settled(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect((await f.read()).phase).toBe("consent_pending"); expect(errors).toEqual([]); + }); + it.each(["consent", "upload", "file_info"])("restores the actual %s claim deadline and quarantines without repeating provider I/O", async stage => { + const f = await transfer(); const due = await abandon(f, stage); + const calls = [f.opts.postConsent, f.opts.uploadRequest!, f.opts.postFileInfo].map(fn => vi.mocked(fn).mock.calls.length); + await worker(f.service().recoverAbandonedAttempt).ready; await settled(); + expect(coordinator!.nextWakeAt()).toBe(due); await quiet(); + await coordinator!.stop(); await worker(f.service().recoverAbandonedAttempt).ready; + expect(coordinator!.nextWakeAt()).toBe(due); await quiet(); + await vi.advanceTimersByTimeAsync(due - Date.now() + 1000); await settled(); + expect(await f.read()).toMatchObject({ phase: `${stage}_unknown`, attemptId: null, attemptExpiresAt: null, reason: "worker_outcome_unknown" }); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect([f.opts.postConsent, f.opts.uploadRequest!, f.opts.postFileInfo].map(fn => vi.mocked(fn).mock.calls.length)).toEqual(calls); + expect(errors).toEqual([]); + }); + it("wakes from the real first claim and removes the deadline after normal settlement", async () => { + const f = await transfer(); let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + f.opts.postConsent = vi.fn(async () => { entered(); await hold; return { id: "card-1" }; }); + await worker(f.service().recoverAbandonedAttempt).ready; await settled(); + expect(coordinator!.nextWakeAt()).toBeNull(); + const send = f.service().process(f.a.companyId, f.created.id); + try { + await ready; await settled(); const row = await f.read(); + expect(coordinator!.nextWakeAt()).toBe(row.attemptExpiresAt!.getTime()); await quiet(); + } finally { release(); } + expect((await send).phase).toBe("awaiting_consent"); await settled(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect(f.opts.postConsent).toHaveBeenCalledTimes(1); + }); + it("retains a first committed attempt across drain and recovers on resume", async () => { + const f = await transfer(); enabled = false; + await worker(f.service().recoverAbandonedAttempt).ready; + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + f.opts.postConsent = vi.fn(async () => { entered(); await hold; return { id: "late-card" }; }); + const send = f.service().process(f.a.companyId, f.created.id); + try { + await ready; await f.service().recordConsent(await f.event()); + await f.service().recordConsent(await f.event("decline")); + } finally { release(); await send; } + await quiet(120_000); + expect((await f.read()).phase).toBe("conflict"); expect((await f.read()).attemptId).not.toBeNull(); + enabled = true; await vi.advanceTimersByTimeAsync(30_000); await settled(); + expect((await f.read()).phase).toBe("conflict"); expect((await f.read()).attemptId).toBeNull(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); expect(errors).toEqual([]); + }); + it("retains atomic recovery after projection failure without leaking its error", async () => { + const f = await transfer(); const due = await abandon(f); let fail = true; + f.opts.project = async tx => { + await tx.update(chatPublications).set({ state: "failed" }).where(eq(chatPublications.id, f.a.publicationId)); + if (fail) throw new Error("PRIVATE-TEAMS-UPLOAD-CANARY"); + }; + await worker(f.service().recoverAbandonedAttempt).ready; + await vi.advanceTimersByTimeAsync(due - Date.now() + 1000); await settled(); + expect((await f.read()).phase).toBe("consent_sending"); + expect((await db.select().from(chatPublications).where(eq(chatPublications.id, f.a.publicationId)))[0].state).toBe("pending"); + expect(errors).toHaveLength(1); expect(String((errors[0] as AggregateError).errors[0])).not.toContain("CANARY"); + fail = false; await vi.advanceTimersByTimeAsync(30_000); await settled(); + expect((await f.read()).phase).toBe("consent_unknown"); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); expect(f.opts.postConsent).toHaveBeenCalledTimes(1); + }); + it("preserves conflict evidence and clears only its expired ownership", async () => { + const f = await transfer(); let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + f.opts.postConsent = vi.fn(async () => { entered(); await hold; return { id: "late-card" }; }); + await worker(f.service().recoverAbandonedAttempt).ready; + const send = f.service().process(f.a.companyId, f.created.id); + try { + await ready; await f.service().recordConsent(await f.event()); + await f.service().recordConsent(await f.event("decline")); + } finally { release(); } + await send; const before = await f.read(); expect(before.phase).toBe("conflict"); + await settled(); expect(coordinator!.nextWakeAt()).toBe(before.attemptExpiresAt!.getTime()); + await vi.advanceTimersByTimeAsync(before.attemptExpiresAt!.getTime() - Date.now() + 1000); await settled(); + const after = await f.read(); expect(after).toMatchObject({ phase: "conflict", attemptId: null, attemptExpiresAt: null, reason: before.reason, privateState: before.privateState }); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect(f.opts.postConsent).toHaveBeenCalledTimes(1); expect(errors).toEqual([]); + }); + it("does not observe a rolled-back claim or contact the provider before commit", async () => { + const f = await transfer(); let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + f.opts.project = async (_tx, row) => { if (row.phase === "consent_sending") { entered(); await hold; throw new Error("claim rolled back"); } }; + await worker(f.service().recoverAbandonedAttempt).ready; await settled(); + const send = f.service().process(f.a.companyId, f.created.id); + const failure = expect(send).rejects.toThrow("claim rolled back"); + try { + await ready; await vi.advanceTimersByTimeAsync(10_000); + expect(await teamsAttemptRecoveryWork(db, f.service().recoverAbandonedAttempt).load()).toEqual([]); + expect((await f.read()).phase).toBe("consent_pending"); expect(f.opts.postConsent).not.toHaveBeenCalled(); + } finally { release(); await failure; } + await settled(); + expect((await f.read()).phase).toBe("consent_pending"); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("rounds submillisecond attempts up and rechecks a changed claim", async () => { + const f = await transfer(); await abandon(f); + const whole = Date.now() + 60_000; + await db.update(chatTeamsFileTransfers).set({ attemptExpiresAt: sql`${new Date(whole).toISOString()}::timestamptz + interval '0.0005 second'` }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + vi.useFakeTimers({ toFake: ["Date"] }); + const work = teamsAttemptRecoveryWork(db, f.service().recoverAbandonedAttempt), [entry] = await work.load(); + expect(entry.dueAt).toBe(whole + 1); vi.setSystemTime(whole); await work.check(entry); + expect((await f.read()).phase).toBe("consent_sending"); + const changed = whole + 90_000; + await db.update(chatTeamsFileTransfers).set({ attemptId: randomUUID(), attemptExpiresAt: new Date(changed) }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + vi.setSystemTime(whole + 1); expect(await work.check(entry)).toMatchObject({ dueAt: changed }); + expect((await f.read()).phase).toBe("consent_sending"); + vi.setSystemTime(changed); expect(await work.check(entry)).toBeNull(); expect((await f.read()).phase).toBe("consent_unknown"); + }); + it("pages and drains healthy attempts across companies while a corrupt claim retains recovery", async () => { + const f = await transfer(), other = await transfer(); await abandon(f); await abandon(other); + for (let index = 0; index < 100; index++) { + const publicationId = randomUUID(); + await db.insert(chatPublications).values({ id: publicationId, companyId: f.a.companyId, + endpointId: f.a.endpointId, conversationId: f.a.conversationId, issueId: f.a.issueId, + commentId: f.a.commentId, idempotencyKey: `attempt-page:${publicationId}`, + payload: { text: "Same authorized file", attachmentIds: [f.a.attachmentId] } }); + const a = { ...f.a, publicationId }; + const service = teamsFileTransferService(db, { ...f.opts, authorize: async () => a, + project: async (_tx, row) => { if (row.phase === "awaiting_consent") throw new Error("receipt interrupted"); } }); + const created = await service.issue(a, new Date(Date.now() + 600_000)); transferIds.push(created.id); + await expect(service.process(a.companyId, created.id)).rejects.toThrow("receipt interrupted"); + } + await db.update(chatTeamsFileTransfers).set({ authorityDigest: "0".repeat(64) }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + const { work, ready } = worker(f.service().recoverAbandonedAttempt); await ready; + const entries = await work.load(); expect(entries).toHaveLength(102); + const calls = vi.mocked(f.opts.postConsent).mock.calls.length; + await vi.advanceTimersByTimeAsync(Math.max(...entries.map(row => row.dueAt)) - Date.now() + 1000); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); expect((await other.read()).phase).toBe("consent_unknown"); + expect((await work.load()).map(row => row.id)).toEqual([f.created.id]); + }, { timeout: 10_000 }); + expect((await f.read()).phase).toBe("consent_sending"); expect(errors.length).toBeGreaterThan(0); + expect(f.opts.postConsent).toHaveBeenCalledTimes(calls); + }, 30_000); + it("joins recovery projection before shutdown completes", async () => { + const f = await transfer(); const due = await abandon(f); let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + f.opts.project = async () => { entered(); await hold; }; + await worker(f.service().recoverAbandonedAttempt).ready; + try { + await vi.advanceTimersByTimeAsync(due - Date.now() + 1000); await ready; + let stopped = false; const stop = coordinator!.stop().then(() => { stopped = true; }); + await Promise.resolve(); expect(stopped).toBe(false); release(); await stop; + expect((await f.read()).phase).toBe("consent_unknown"); + } finally { release(); } + }); + }); + describe("scheduled unused consent expiry", () => { let coordinator: ReturnType | undefined; const transferIds: string[] = [], errors: unknown[] = []; diff --git a/server/src/services/chat-teams-file-transfers.ts b/server/src/services/chat-teams-file-transfers.ts index 0f82870da06..ef4a17f8c5d 100644 --- a/server/src/services/chat-teams-file-transfers.ts +++ b/server/src/services/chat-teams-file-transfers.ts @@ -1,3 +1,4 @@ +import { notifyTeamsAttemptRecovery } from "./chat-channels/microsoft-teams/attempt-recovery-work.js"; import { notifyTeamsConsentExpiry, unusedTeamsConsentPhases } from "./chat-channels/microsoft-teams/consent-expiry-work.js"; import { createHash, randomUUID } from "node:crypto"; import { and, eq, gt, inArray, lte, or, sql } from "drizzle-orm"; @@ -390,6 +391,7 @@ export function teamsFileTransferService( patch: Partial, ): Promise { await notifyTeamsConsentExpiry(tx, row.id); + await notifyTeamsAttemptRecovery(tx, row.id); const [next] = await tx .update(chatTeamsFileTransfers) .set({ ...patch, version: row.version + 1, updatedAt: now() }) @@ -1159,6 +1161,7 @@ export function teamsFileTransferService( get, expireAndRecover: (companyId: string, limit = 25) => expireAndRecover(companyId, limit), recoverAbandonedAttempts: (companyId: string, limit = 25) => expireAndRecover(companyId, limit, "attempts"), + recoverAbandonedAttempt: (companyId: string, id: string) => recoverTransfer(companyId, id, "attempts"), expireUnusedConsent: (companyId: string, id: string) => recoverTransfer(companyId, id, "consent"), cancel, canCancelConflict, diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 699cb7ecfb3..000bfc6f860 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -47,6 +47,7 @@ export const DELIVERY_QUEUES = { voiceInboundCleanup: "voice-inbound-cleanup", voiceReplies: "voice-reply-delivery", chatTeamsConsentExpiry: "chat-teams-consent-expiry", + chatTeamsAttemptRecovery: "chat-teams-attempt-recovery", chatCommandRegistration: "chat-command-registration", chatTelegramSubscriptions: "chat-telegram-subscriptions", chatSessionSyncs: "chat-session-syncs",