From 2f2a0bca5e3c620d94577af8eb9d989cc53ab84e Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Thu, 22 Oct 2026 22:46:38 -0400 Subject: [PATCH] fix(connectors): a connector backlog cannot fill orchestration's receipt ledger (#32130) * fix(connectors): a connector backlog cannot fill orchestration's receipt ledger Connector deliveries now refuse with delivery_backlog_full once 1,000 are unsettled, and a new seam settles a previous process's pending connector receipts as completed-unknown so the ledger can prune them. * fix(connectors): settle only a previous process's receipts, not a re-enable's The delivery receipt owner now lives for the install (one per process) and is shared by every enabled-period seam, so toggling connectors off and on no longer completes this process's still-running deliveries as unknown. --- ...nal-conversation-delivery-receipts.test.ts | 111 ++++++++++++++++++ ...external-conversation-delivery-receipts.ts | 23 +++- ...external-conversation-seam.test-fixture.ts | 3 +- .../external-conversation-seam.ts | 11 +- .../external-conversation-types.ts | 1 + .../external-conversations-install.test.ts | 41 +++++++ .../external-conversations-install.ts | 6 +- .../mutation-receipt-store.ts | 32 +++++ 8 files changed, 218 insertions(+), 10 deletions(-) create mode 100644 src/main/external-conversations/external-conversation-delivery-receipts.test.ts diff --git a/src/main/external-conversations/external-conversation-delivery-receipts.test.ts b/src/main/external-conversations/external-conversation-delivery-receipts.test.ts new file mode 100644 index 00000000000..8b3153c6ac5 --- /dev/null +++ b/src/main/external-conversations/external-conversation-delivery-receipts.test.ts @@ -0,0 +1,111 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' +import { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db' +import { MUTATION_RECEIPT_MAX_ROWS } from '../runtime/orchestration/mutation-receipt-capacity' +import { + CONNECTOR_PENDING_RECEIPT_LIMIT, + DeliveryReceipts +} from './external-conversation-delivery-receipts' +import type { ConversationKey, Delivery } from './external-conversation-types' + +const KEY: ConversationKey = { source: 'dev', externalId: 'thread' } +const ORDINARY = { + callerFingerprint: 'ordinary-user', + requestId: 'send', + method: 'orchestration.send', + payloadHash: 'hash' +} +const never = (): Promise => new Promise(() => {}) + +const directories: string[] = [] +function fileDb(): { path: string; db: OrchestrationDb } { + const directory = mkdtempSync(join(tmpdir(), 'connector-receipts-')) + directories.push(directory) + const path = join(directory, 'orchestration.db') + return { path, db: new OrchestrationDb(path) } +} + +afterEach(() => { + for (const directory of directories.splice(0)) { + rmSync(directory, { recursive: true, force: true }) + } +}) + +describe('DeliveryReceipts ledger capacity', () => { + it('cannot fill orchestration’s ledger with a backlog that outlives the process', async () => { + const { path, db } = fileDb() + const receipts = new DeliveryReceipts(() => db) + const answers: Promise[] = [] + for (let i = 0; i < MUTATION_RECEIPT_MAX_ROWS; i++) { + answers.push(receipts.run(KEY, String(i), 'connector.send', ['hello'], never)) + } + await expect(answers[CONNECTOR_PENDING_RECEIPT_LIMIT]).resolves.toEqual({ + outcome: 'refused', + reason: 'delivery_backlog_full' + }) + expect(db.countPendingMutationReceipts(['connector:dev'])).toBe(CONNECTOR_PENDING_RECEIPT_LIMIT) + db.close() + + // Process loss, then a restart that never starts connectors. + const reopened = new OrchestrationDb(path) + expect(reopened.beginMutationReceipt(ORDINARY).disposition).toBe('started') + reopened.close() + }) + + it('answers a recorded delivery even when the backlog is full', async () => { + const { db } = fileDb() + const receipts = new DeliveryReceipts(() => db) + for (let i = 0; i < CONNECTOR_PENDING_RECEIPT_LIMIT; i++) { + void receipts.run(KEY, String(i), 'connector.send', ['hello'], never) + } + const joined = receipts.run(KEY, '0', 'connector.send', ['hello'], never) + const raced = await Promise.race([joined, Promise.resolve('waiting')]) + expect(raced).toBe('waiting') + db.close() + }) + + it('settles a previous process’s pending deliveries as unknown, without replaying them', async () => { + const { path, db } = fileDb() + const before = new DeliveryReceipts(() => db) + for (let i = 0; i < CONNECTOR_PENDING_RECEIPT_LIMIT; i++) { + void before.run(KEY, String(i), 'connector.send', ['hello'], never) + } + db.close() + + const reopened = new OrchestrationDb(path) + const after = new DeliveryReceipts(() => reopened) + let executed = 0 + const execute = async (): Promise => { + executed++ + return { outcome: 'accepted' } + } + await expect(after.run(KEY, '0', 'connector.send', ['hello'], execute)).resolves.toEqual({ + outcome: 'unknown', + reason: 'replayed_after_restart' + }) + expect( + reopened.getMutationReceipt('connector:dev', JSON.stringify(['thread', '0']))?.state + ).toBe('completed') + // The settled rows no longer hold the connector's budget. + await expect(after.run(KEY, 'new', 'connector.send', ['hello'], execute)).resolves.toEqual({ + outcome: 'accepted' + }) + expect(executed).toBe(1) + expect(reopened.countPendingMutationReceipts(['connector:dev'])).toBe(0) + reopened.close() + }) + + it('leaves other callers’ pending receipts alone', async () => { + const { db } = fileDb() + db.beginMutationReceipt(ORDINARY) + await new DeliveryReceipts(() => db).run(KEY, 'd1', 'connector.send', ['hello'], async () => ({ + outcome: 'accepted' + })) + expect(db.getMutationReceipt(ORDINARY.callerFingerprint, ORDINARY.requestId)?.state).toBe( + 'pending' + ) + db.close() + }) +}) diff --git a/src/main/external-conversations/external-conversation-delivery-receipts.ts b/src/main/external-conversations/external-conversation-delivery-receipts.ts index 00cf8ef1d94..9a727093bf2 100644 --- a/src/main/external-conversations/external-conversation-delivery-receipts.ts +++ b/src/main/external-conversations/external-conversation-delivery-receipts.ts @@ -1,4 +1,5 @@ import { z } from 'zod' +import { CONNECTOR_SOURCES, type ConnectorSource } from '../../shared/external-conversation-link' import type { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db' import { OrchestrationError } from '../runtime/orchestration/orchestration-error' import { hashCanonical } from '../runtime/rpc/orchestration-mutation-receipt' @@ -16,14 +17,22 @@ const RecordedDelivery = z.discriminatedUnion('outcome', [ const LOST: Delivery = { outcome: 'unknown', reason: 'replayed_after_restart' } +/** Unsettled connector deliveries the shared ledger holds at once, so orchestration keeps the rest. */ +export const CONNECTOR_PENDING_RECEIPT_LIMIT = 1_000 + +const callerFingerprint = (source: ConnectorSource): string => `connector:${source}` +const CONNECTOR_FINGERPRINTS = CONNECTOR_SOURCES.map(callerFingerprint) + /** * A connector's delivery ids on orchestration's durable mutation receipts, under one caller * identity per source: the same id and payload share one answer, across restarts too, and another * payload is refused. A duplicate of one still running waits for it; a running one a restart * interrupted reads `unknown`. Refusals and throws wrote nothing, so they leave no receipt. + * A new delivery past `CONNECTOR_PENDING_RECEIPT_LIMIT` unsettled ones is refused unrecorded. */ export class DeliveryReceipts { private readonly inFlight = new Map>() + private leftoversSettled = false constructor(private readonly db: () => DeliveryReceiptDb) {} @@ -35,13 +44,25 @@ export class DeliveryReceipts { execute: () => Promise ): Promise { const identity = { - callerFingerprint: `connector:${key.source}`, + callerFingerprint: callerFingerprint(key.source), requestId: JSON.stringify([key.externalId, deliveryId]), method, payloadHash: hashCanonical(payload) } const flightKey = JSON.stringify([identity.callerFingerprint, identity.requestId]) const db = this.db() + if (!this.leftoversSettled) { + // Pending rows from a previous process already answer only unknown; record that, never delete + // (they may have written), so the ledger can prune them and they leave the limit. + db.settlePendingMutationReceipts(CONNECTOR_FINGERPRINTS, JSON.stringify(LOST)) + this.leftoversSettled = true + } + if ( + !db.getMutationReceipt(identity.callerFingerprint, identity.requestId) && + db.countPendingMutationReceipts(CONNECTOR_FINGERPRINTS) >= CONNECTOR_PENDING_RECEIPT_LIMIT + ) { + return Promise.resolve(refused('delivery_backlog_full')) + } let begun: ReturnType try { begun = db.beginMutationReceipt(identity) diff --git a/src/main/external-conversations/external-conversation-seam.test-fixture.ts b/src/main/external-conversations/external-conversation-seam.test-fixture.ts index 15814c782eb..e36b35270ef 100644 --- a/src/main/external-conversations/external-conversation-seam.test-fixture.ts +++ b/src/main/external-conversations/external-conversation-seam.test-fixture.ts @@ -1,5 +1,6 @@ import type { GlobalSettings } from '../../shared/global-settings-types' import { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db' +import { DeliveryReceipts } from './external-conversation-delivery-receipts' import { ExternalConversationSeam } from './external-conversation-seam' import type { SessionRows } from './external-conversation-session-capture' import { fakeLinkStore } from './external-conversation.test-fixture' @@ -19,7 +20,7 @@ export function harness( new ExternalConversationSeam({ store: records.store, readSettings: () => settings, - receiptDb: () => db, + receipts: new DeliveryReceipts(() => db), sessionRows, onWorkspaceRemoved: (listener) => { removed.add(listener) diff --git a/src/main/external-conversations/external-conversation-seam.ts b/src/main/external-conversations/external-conversation-seam.ts index 34f87618ee2..208704638c2 100644 --- a/src/main/external-conversations/external-conversation-seam.ts +++ b/src/main/external-conversations/external-conversation-seam.ts @@ -3,7 +3,7 @@ import type { ExternalConversationLink } from '../../shared/external-conversatio import type { OrcaRuntimeService } from '../runtime/orca-runtime' import { admitConnectorAgent } from './connector-agents' import { ConversationChain } from './external-conversation-chain' -import { DeliveryReceipts, type DeliveryReceiptDb } from './external-conversation-delivery-receipts' +import type { DeliveryReceipts } from './external-conversation-delivery-receipts' import { findLink, writeLink, @@ -44,7 +44,8 @@ export function runtimeConversationHost(runtime: OrcaRuntimeService): ExternalCo export type ExternalConversationSeamDeps = { store: ExternalConversationLinkStore readSettings: () => AgentStatusHooksSettings - receiptDb: () => DeliveryReceiptDb + /** Outlives the seam: a re-enable must not take this process's running deliveries for leftovers. */ + receipts: DeliveryReceipts /** Worktree removals; returns the release. */ onWorkspaceRemoved: (listener: (worktreeId: string) => void) => () => void /** The hook rows linked agents' sessions are read from. */ @@ -65,7 +66,6 @@ type StartInput = Omit & { agent: string } export class ExternalConversationSeam { private readonly abort = new AbortController() private readonly chain = new ConversationChain(this.abort.signal) - private readonly receipts: DeliveryReceipts /** Conversations this period started, so a second start for one shares its creation. */ private readonly startedHere = new Set() /** Conversations with work sent here, by the workspace their link was on. */ @@ -80,7 +80,6 @@ export class ExternalConversationSeam { private readonly releaseSessions: () => void constructor(private readonly deps: ExternalConversationSeamDeps) { - this.receipts = new DeliveryReceipts(deps.receiptDb) this.sessions = new LinkedSessionCapture(deps.store, deps.sessionRows) // A removed workspace may keep its PTY registered a while; its held and queued sends end now. // The event names no host, so a same-path twin's removal is told apart by the link surviving. @@ -116,7 +115,7 @@ export class ExternalConversationSeam { deliveryId: string, input: StartInput ): Promise { - const initial = await this.receipts.run( + const initial = await this.deps.receipts.run( key, deliveryId, 'connector.start', @@ -168,7 +167,7 @@ export class ExternalConversationSeam { text: string ): Promise { const id = conversationId(key) - return this.receipts.run(key, deliveryId, 'connector.send', [text], () => { + return this.deps.receipts.run(key, deliveryId, 'connector.send', [text], () => { const queued = findLink(this.deps.store, key) if (queued) { this.linked.set(id, { key, worktreeId: queued.workspace.worktreeId }) diff --git a/src/main/external-conversations/external-conversation-types.ts b/src/main/external-conversations/external-conversation-types.ts index 12ce95f0fad..d5a8773a970 100644 --- a/src/main/external-conversations/external-conversation-types.ts +++ b/src/main/external-conversations/external-conversation-types.ts @@ -8,6 +8,7 @@ export const CONNECTOR_REFUSALS = [ 'agent_not_reachable', 'unsupported_agent', 'delivery_id_reused', + 'delivery_backlog_full', // The three a closed agent's reopen ends on; each asks the user to open it in Orca. 'agent_asleep_in_window', 'no_session_to_reopen', diff --git a/src/main/external-conversations/external-conversations-install.test.ts b/src/main/external-conversations/external-conversations-install.test.ts index 06d301be84b..d511ff9f72b 100644 --- a/src/main/external-conversations/external-conversations-install.test.ts +++ b/src/main/external-conversations/external-conversations-install.test.ts @@ -5,7 +5,10 @@ import type { RuntimeWorktreeLifecycleEvent } from '../runtime/orca-runtime-core import { OrchestrationDb } from '../runtime/orchestration/db/orchestration-db' import { installExternalConversations } from './external-conversations-install' import { + fakeLauncher, fakeLinkStore, + fakeTerminal, + launchResult, PANE, SESSION, sessionRow, @@ -98,6 +101,44 @@ describe('installExternalConversations', () => { expect(host.lifecycle.size).toBe(1) }) + it('leaves a delivery still running across a disable and re-enable to its owner', async () => { + const { store, update } = fakeStore({ experimentalConnectors: true }) + const host = runtime() + const db = host.getOrchestrationDb() + const connectors = installExternalConversations(store, host) + let release = (): void => {} + const held = new Promise((resolve) => { + release = resolve + }) + const { launcher, launch } = fakeLauncher(async () => { + await held + return launchResult('handed-to-terminal') + }) + const conversation = { launcher, terminal: fakeTerminal().terminal } + const key = { source: 'dev', externalId: 'thread' } as const + const input = { + agent: 'claude', + target: { kind: 'existing' as const, worktree: `id:${WORKTREE}` }, + prompt: 'hi' + } + const live = connectors.current()?.start(conversation, key, 'd1', input) + await vi.waitFor(() => expect(launch).toHaveBeenCalledTimes(1)) + + update({ experimentalConnectors: false }) + update({ experimentalConnectors: true }) + const retry = connectors.current()?.start(conversation, key, 'd1', input) + const requestId = JSON.stringify(['thread', 'd1']) + expect(db.getMutationReceipt('connector:dev', requestId)?.state).toBe('pending') + + release() + const [first, second] = await Promise.all([live, retry]) + expect(second?.initial).toEqual(first?.initial) + expect(db.getMutationReceipt('connector:dev', requestId)?.receipt).toBe( + JSON.stringify(first?.initial) + ) + expect(launch).toHaveBeenCalledTimes(1) + }) + it('keeps the seam through unrelated settings changes', () => { const { store, update } = fakeStore({ experimentalConnectors: true }) const connectors = installExternalConversations(store, runtime()) diff --git a/src/main/external-conversations/external-conversations-install.ts b/src/main/external-conversations/external-conversations-install.ts index 14e0af2e6f3..e1094979706 100644 --- a/src/main/external-conversations/external-conversations-install.ts +++ b/src/main/external-conversations/external-conversations-install.ts @@ -1,6 +1,6 @@ import type { GlobalSettings } from '../../shared/global-settings-types' import type { RuntimeWorktreeLifecycleEvent } from '../runtime/orca-runtime-core' -import type { DeliveryReceiptDb } from './external-conversation-delivery-receipts' +import { DeliveryReceipts, type DeliveryReceiptDb } from './external-conversation-delivery-receipts' import type { ExternalConversationLinkStore } from './external-conversation-links' import { ExternalConversationSeam, @@ -34,6 +34,8 @@ export function installExternalConversations( if (!store) { return { current: () => null } } + // One per process, so leftover settlement only reaches a previous process's receipts. + const receipts = new DeliveryReceipts(() => runtime.getOrchestrationDb()) let seam: ExternalConversationSeam | null = null let reconciled = false const reconcile = (): void => { @@ -43,7 +45,7 @@ export function installExternalConversations( seam = new ExternalConversationSeam({ store, readSettings: () => store.getSettings(), - receiptDb: () => runtime.getOrchestrationDb(), + receipts, onWorkspaceRemoved: (listener) => runtime.onWorktreeLifecycle((event) => { if (event.kind === 'removed') { diff --git a/src/main/runtime/orchestration/db/mutation-receipts/mutation-receipt-store.ts b/src/main/runtime/orchestration/db/mutation-receipts/mutation-receipt-store.ts index 05e29f28643..eb9f1653046 100644 --- a/src/main/runtime/orchestration/db/mutation-receipts/mutation-receipt-store.ts +++ b/src/main/runtime/orchestration/db/mutation-receipts/mutation-receipt-store.ts @@ -157,6 +157,34 @@ export function discardPendingMutationReceipt( .run(callerFingerprint, requestId) } +export function countPendingMutationReceipts( + this: OrchestrationDb, + callerFingerprints: readonly string[] +): number { + const count = this.db + .prepare( + `SELECT COUNT(*) AS count FROM mutation_receipts + WHERE caller_fingerprint IN (SELECT value FROM json_each(?)) AND state = 'pending'` + ) + .get(JSON.stringify(callerFingerprints))?.count + return Number(count ?? 0) +} + +/** Completes these callers' pending rows with `receipt`, for operations no running work owns. */ +export function settlePendingMutationReceipts( + this: OrchestrationDb, + callerFingerprints: readonly string[], + receipt: string +): void { + this.db + .prepare( + `UPDATE mutation_receipts + SET state = 'completed', receipt = ?, updated_at = datetime('now') + WHERE caller_fingerprint IN (SELECT value FROM json_each(?)) AND state = 'pending'` + ) + .run(receipt, JSON.stringify(callerFingerprints)) +} + export function getMutationReceipt( this: OrchestrationDb, callerFingerprint: string, @@ -176,6 +204,8 @@ export type MutationReceiptStoreMethods = { completeMutationReceipt: typeof completeMutationReceipt checkpointPendingMutationReceipt: typeof checkpointPendingMutationReceipt discardPendingMutationReceipt: typeof discardPendingMutationReceipt + countPendingMutationReceipts: typeof countPendingMutationReceipts + settlePendingMutationReceipts: typeof settlePendingMutationReceipts getMutationReceipt: typeof getMutationReceipt } @@ -186,6 +216,8 @@ export function attachMutationReceiptStore(ctor: { prototype: object }): void { completeMutationReceipt, checkpointPendingMutationReceipt, discardPendingMutationReceipt, + countPendingMutationReceipts, + settlePendingMutationReceipts, getMutationReceipt }) }