mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 21:49:18 +01:00
fix(server): recover Teams attempts from saved claim deadlines
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
3967c67e9d
commit
6f2f3951ca
12 files changed
+446
-53
No files matched your search
@@ -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).
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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", () => {
|
||||
|
||||
@@ -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<void>(resolve => { release = resolve; }), ready = new Promise<void>(resolve => { entered = resolve; });
|
||||
context.controls.beforeConsentReceipt = async () => { entered(); await hold; };
|
||||
let send: ReturnType<typeof context.service.publishBoardMessage> | 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[] = [];
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<boolean>) {
|
||||
async function load(id?: string, signal?: AbortSignal): Promise<Entry[]> {
|
||||
try {
|
||||
const entries: Entry[] = []; let cursor: string | undefined;
|
||||
for (;;) {
|
||||
signal?.throwIfAborted();
|
||||
const rows = await db.select({ id: chatTeamsFileTransfers.id, companyId: chatTeamsFileTransfers.companyId,
|
||||
dueAt: sql<string>`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<Entry | null> {
|
||||
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<typeof createDeliveryWorkCoordinator>, 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 };
|
||||
}
|
||||
@@ -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<typeof createDeliveryWorkCoordinator>, 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<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
|
||||
return teamsAttemptRecoveryWork(db, (companyId, id) => teamsTransferProtocol().recoverAbandonedAttempt(companyId, id))
|
||||
.register(coordinator, () => !isShuttingDown() && canRun());
|
||||
}
|
||||
return { registerTeamsAttemptRecoveryWork, registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication };
|
||||
}
|
||||
@@ -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<typeof createDeliveryWorkCoordinator> | 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<ReturnType<typeof transfer>>, 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<boolean>) {
|
||||
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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
const transferIds: string[] = [], errors: unknown[] = [];
|
||||
|
||||
@@ -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<typeof chatTeamsFileTransfers.$inferInsert>,
|
||||
): Promise<Row> {
|
||||
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,
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in new issue
Block a user