diff --git a/src/main/runtime/orchestration/orchestration-schema-version-skew.ts b/src/main/runtime/orchestration/orchestration-schema-version-skew.ts index 77a2293ea89..a2767e1e5a7 100644 --- a/src/main/runtime/orchestration/orchestration-schema-version-skew.ts +++ b/src/main/runtime/orchestration/orchestration-schema-version-skew.ts @@ -22,9 +22,7 @@ const POST_V6_COLUMNS = [ ['legacy_adoptions', 'source_run_id'], ['legacy_compatibility_principals', 'id'], ['legacy_operation_receipts', 'principal_id'], - ['legacy_mail_receipts', 'principal_id'], - ['worker_terminal_resources', 'recovery_attempt_count'], - ['worker_terminal_resources', 'last_recovery_at'] + ['legacy_mail_receipts', 'principal_id'] ] as const const VERSIONED_POST_V6_COLUMNS = [ @@ -36,6 +34,9 @@ const VERSIONED_POST_V6_COLUMNS = [ { version: 31, table: 'dispatch_contexts', column: 'host_scope' }, { version: 31, table: 'worker_terminal_resources', column: 'endpoint_id' }, { version: 31, table: 'worker_terminal_resources', column: 'endpoint_incarnation' }, + // Why: unversioned, these made every shipped v30 database read as v6 and replay the whole chain. + { version: 32, table: 'worker_terminal_resources', column: 'recovery_attempt_count' }, + { version: 32, table: 'worker_terminal_resources', column: 'last_recovery_at' }, { version: 33, table: 'messages', column: 'pointer_enter_pending' }, { version: 34, table: 'deliveries', column: 'mailbox_handle' }, { version: 36, table: 'dispatch_contexts', column: 'consumer_generation' }, diff --git a/src/main/runtime/orchestration/orchestration-version-skew-migration.test.ts b/src/main/runtime/orchestration/orchestration-version-skew-migration.test.ts index 923d45d701a..7a58e81920d 100644 --- a/src/main/runtime/orchestration/orchestration-version-skew-migration.test.ts +++ b/src/main/runtime/orchestration/orchestration-version-skew-migration.test.ts @@ -231,6 +231,24 @@ describe('OrchestrationDb version-skew migration', () => { ).toBeDefined() }) + // The two v32 recovery columns were listed as unversioned, so every shipped database below v32 + // read as v6 and replayed the whole chain, re-running the v23 resource backfill over live rows. + it('starts a genuine pre-v32 database at its own version, not the v6 floor', () => { + tempDir = mkdtempSync(join(tmpdir(), 'orca-db-version-skew-v31-')) + const dbPath = join(tempDir, 'orchestration.db') + db = new OrchestrationDb(dbPath) + db.close() + db = undefined + + const raw = new Database(dbPath) + raw.exec( + 'ALTER TABLE worker_terminal_resources DROP COLUMN recovery_attempt_count; ALTER TABLE worker_terminal_resources DROP COLUMN last_recovery_at;' + ) + raw.pragma('user_version = 31') + expect(resolveOrchestrationMigrationStartVersion(raw, 31, SCHEMA_VERSION)).toBe(31) + raw.close() + }) + it('creates fresh delivery mailboxes with a non-null schema invariant', () => { db = new OrchestrationDb(':memory:') diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-methods.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-methods.ts index f972954d5a4..a12428253c3 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/check-methods.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-methods.ts @@ -84,6 +84,13 @@ export const ORCHESTRATION_CHECK_METHODS: RpcMethod[] = [ }) } const consumingCheck = params.peek !== true && params.all !== true && params.unread !== false + // Why: an empty consuming check is the worker contract's "checkpoint, not a failure", so a + // caller whose Attempt moved on has to be told rather than handed an empty direct mailbox. + // This outranks the pane guard: a paneless loser cannot run-use anyway, it has to stop. + const settledDispatch = consumingCheck ? db.getLatestDispatchForTerminal(handle) : undefined + if (settledDispatch && isSupersededDispatch(settledDispatch)) { + throw dispatchFenced() + } // Why: a consuming check on a handle with no live pane and no Dispatch can never see // Run mail, so an empty inbox would read as "nothing yet" instead of a stale caller. if (!paneKey && consumingCheck) { @@ -93,12 +100,6 @@ export const ORCHESTRATION_CHECK_METHODS: RpcMethod[] = [ orchestrationSkillRecoveryData() ) } - // Why: an empty consuming check is the worker contract's "checkpoint, not a failure", so a - // caller whose Attempt moved on has to be told rather than handed an empty direct mailbox. - const settledDispatch = consumingCheck ? db.getLatestDispatchForTerminal(handle) : undefined - if (settledDispatch && isSupersededDispatch(settledDispatch)) { - throw dispatchFenced() - } return checkDirectMailbox({ params, runtime, db, handle, typeFilter, signal }) } }) diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-superseded-terminal.test.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-superseded-terminal.test.ts index ecefcbb0a0e..65cc3192b1d 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/check-superseded-terminal.test.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-superseded-terminal.test.ts @@ -92,15 +92,16 @@ describe('orchestration.check from a terminal whose Attempt was superseded', () await expect(check('term_old', PANE_OLD)).rejects.toMatchObject({ code: 'consumer_fenced' }) }) - // A terminal that lost its pane binding needs the rebind recovery, not the stop instruction. - it('reports a paneless caller as stable_pane_required before the settled-Attempt fence', async () => { - ;({ db, ctx } = h.setup()) - const task = db.createTask({ spec: 'work that failed on a pane that is gone' }) - const dispatch = createRootDispatch(db, task.id, 'term_old', PANE_OLD) - db.failDispatch(dispatch.id, 'worker terminal closed') + // A superseded worker whose pane is gone cannot run-use either; the stop signal outranks the + // rebind advice, and a caller with no settled Attempt still gets the rebind advice. + it('fences a paneless caller whose Attempt was superseded, and only that caller', async () => { + retriedOntoAnotherTerminal() await expect( h.call('orchestration.check', { terminal: 'term_old' }, ctx) + ).rejects.toMatchObject({ code: 'consumer_fenced' }) + await expect( + h.call('orchestration.check', { terminal: 'term_never_dispatched' }, ctx) ).rejects.toMatchObject({ code: 'stable_pane_required' }) })