From d6900aaec4d01f35cd05e18149a7aef65ac77d50 Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Thu, 13 Aug 2026 16:13:55 -0400 Subject: [PATCH] fix(orchestration): close recovery scheduler gaps --- src/main/runtime/orca-runtime.ts | 45 ++++++- src/main/runtime/orchestration/db.ts | 22 ++-- .../federation-terminal-recovery.test.ts | 123 +++++++++++++++++- 3 files changed, 175 insertions(+), 15 deletions(-) diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 4d26d303018..3c711d9e480 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -166,7 +166,8 @@ import { clearFederationAckCheckpoints } from './orchestration/federation-ack-ch import { syncFederatedDispatch } from './orchestration/federation-sync' import { classifyFederationTerminalRecoveryFailure, - FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS + FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS, + FEDERATION_TERMINAL_RECOVERY_MAX_DELAY_MS } from './orchestration/federation-terminal-recovery-policy' import { formatMessagePointer } from './orchestration/formatter' import { selectExactWorkerProviderSession } from './orchestration/worker-provider-session' @@ -4747,13 +4748,14 @@ export class OrcaRuntimeService { continue } const tick = () => { - if (!this.getOrchestrationDb().isFederatedDispatchRelayEligible(dispatch.dispatch_id)) { + if (!this.getOrchestrationDb().isFederatedDispatchActive(dispatch.dispatch_id)) { const activeTimer = this.orchestrationFederationTimers.get(dispatch.dispatch_id) if (activeTimer) { clearInterval(activeTimer) } this.orchestrationFederationTimers.delete(dispatch.dispatch_id) this.orchestrationFederationWarnings.delete(dispatch.dispatch_id) + this.ensureTerminalHistoryRecovery(true) return } void this.syncOrchestrationFederatedDispatch(dispatch.dispatch_id).catch(() => undefined) @@ -4766,7 +4768,11 @@ export class OrcaRuntimeService { this.ensureTerminalHistoryRecovery() } - private ensureTerminalHistoryRecovery(): void { + private ensureTerminalHistoryRecovery(restartTimer = false): void { + if (restartTimer && this.orchestrationTerminalHistoryRecoveryTimer) { + clearTimeout(this.orchestrationTerminalHistoryRecoveryTimer) + this.orchestrationTerminalHistoryRecoveryTimer = null + } if ( this.orchestrationTerminalHistoryRecoveryTimer || this.orchestrationTerminalHistoryRecoveryInFlight @@ -4788,6 +4794,7 @@ export class OrcaRuntimeService { private async recoverNextTerminalHistoryAcknowledgment(generation: number): Promise { const db = this.getOrchestrationDb() const nowMs = Date.now() + db.capFederatedTerminalRecoveryDeadlines(nowMs + FEDERATION_TERMINAL_RECOVERY_MAX_DELAY_MS) let historical = db.findNextTerminalFederatedDispatchPendingAcknowledgment( this.orchestrationTerminalRecoveryRowId, nowMs @@ -4799,11 +4806,15 @@ export class OrcaRuntimeService { if (!historical) { const nextAt = db.getNextTerminalFederatedDispatchRecoveryAt() if (nextAt !== undefined) { - this.scheduleTerminalHistoryRecovery(generation, Math.max(0, nextAt - nowMs)) + this.scheduleTerminalHistoryRecovery( + generation, + Math.min(FEDERATION_TERMINAL_RECOVERY_MAX_DELAY_MS, Math.max(0, nextAt - nowMs)) + ) } return } this.orchestrationTerminalRecoveryRowId = historical.rowId + const before = db.getFederatedDispatch(historical.dispatchId) const result = await this.syncOrchestrationFederatedDispatch(historical.dispatchId).then( () => ({ ok: true as const }), (error: unknown) => ({ ok: false as const, error }) @@ -4811,7 +4822,26 @@ export class OrcaRuntimeService { if (generation !== this.orchestrationFederationRelayGeneration) { return } - if (result.ok) { + const recovered = db.getFederatedDispatch(historical.dispatchId) + const madeProgress = Boolean( + before && + recovered && + (recovered.to_home_imported_sequence > before.to_home_imported_sequence || + recovered.to_home_acknowledged_sequence > before.to_home_acknowledged_sequence) + ) + if ( + result.ok && + recovered && + recovered.to_home_acknowledged_sequence < recovered.to_home_imported_sequence && + !madeProgress + ) { + db.recordFederatedTerminalRecoveryFailure({ + dispatchId: historical.dispatchId, + errorCode: 'operation_unknown', + terminal: false, + nowMs: Date.now() + }) + } else if (result.ok) { db.recordFederatedTerminalRecoverySuccess(historical.dispatchId, Date.now()) } else { db.recordFederatedTerminalRecoveryFailure({ @@ -4826,7 +4856,10 @@ export class OrcaRuntimeService { } this.scheduleTerminalHistoryRecovery( generation, - Math.max(FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS, nextAt - Date.now()) + Math.min( + FEDERATION_TERMINAL_RECOVERY_MAX_DELAY_MS, + Math.max(FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS, nextAt - Date.now()) + ) ) } diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index ed2ce9f9360..2eb1e17e279 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -4630,6 +4630,18 @@ export class OrchestrationDb { return row.nextAt ?? undefined } + capFederatedTerminalRecoveryDeadlines(latestNextAtMs: number): void { + this.db + .prepare( + `UPDATE federated_dispatches + SET terminal_ack_recovery_next_at_ms = ? + WHERE terminal_ack_recovery_state != 'terminal' + AND to_home_acknowledged_sequence < to_home_imported_sequence + AND terminal_ack_recovery_next_at_ms > ?` + ) + .run(latestNextAtMs, latestNextAtMs) + } + recordFederatedTerminalRecoveryFailure(params: { dispatchId: string errorCode: string | null @@ -4675,7 +4687,7 @@ export class OrchestrationDb { .run(nowMs + getFederationTerminalRecoveryDelayMs(1), dispatchId) } - isFederatedDispatchRelayEligible(dispatchId: string): boolean { + isFederatedDispatchActive(dispatchId: string): boolean { return Boolean( this.db .prepare( @@ -4683,13 +4695,7 @@ export class OrchestrationDb { FROM federated_dispatches fd INNER JOIN worker_dispatches wd ON wd.dispatch_id = fd.dispatch_id WHERE fd.dispatch_id = ? - AND ( - wd.state IN ('starting', 'ready', 'stopping', 'start_unknown', 'stop_unknown') - OR ( - fd.to_home_acknowledged_sequence < fd.to_home_imported_sequence - AND fd.terminal_ack_recovery_state != 'terminal' - ) - )` + AND wd.state IN ('starting', 'ready', 'stopping', 'start_unknown', 'stop_unknown')` ) .get(dispatchId) ) diff --git a/src/main/runtime/orchestration/federation-terminal-recovery.test.ts b/src/main/runtime/orchestration/federation-terminal-recovery.test.ts index 3cff91d53cd..bfa9204b45f 100644 --- a/src/main/runtime/orchestration/federation-terminal-recovery.test.ts +++ b/src/main/runtime/orchestration/federation-terminal-recovery.test.ts @@ -206,6 +206,34 @@ describe('terminal federation acknowledgment recovery', () => { expect(attempts).toEqual([1_000]) }) + it('bounds a persisted retry deadline after the wall clock moves backward', async () => { + vi.useFakeTimers({ now: 100_000 }) + const { db, dbPath } = createDb(true) + const dispatchId = createHistoricalDispatch(db, 'clock-rollback') + db.recordFederatedTerminalRecoveryFailure({ + dispatchId, + errorCode: null, + terminal: false, + nowMs: Date.now() + }) + db.close() + databases.splice(databases.indexOf(db), 1) + vi.setSystemTime(0) + + const restartedDb = new OrchestrationDb(dbPath!) + databases.push(restartedDb) + const attempts: number[] = [] + createRuntime(restartedDb, async () => { + attempts.push(Date.now()) + throw new Error('transport unavailable') + }) + + await vi.advanceTimersByTimeAsync(29_999) + expect(attempts).toEqual([]) + await vi.advanceTimersByTimeAsync(1) + expect(attempts).toEqual([30_000]) + }) + it('keeps unknown failures retryable with capped backoff', async () => { vi.useFakeTimers({ now: 0 }) const { db } = createDb() @@ -231,6 +259,89 @@ describe('terminal federation acknowledgment recovery', () => { }) }) + it('backs off a fulfilled sync that makes no durable acknowledgment progress', async () => { + vi.useFakeTimers({ now: 0 }) + const { db } = createDb() + const dispatchId = createHistoricalDispatch(db, 'no-progress') + const attempts: number[] = [] + const { runtime } = createRuntime(db, async () => { + attempts.push(Date.now()) + }) + + runtime.ensureOrchestrationFederationRelay() + await vi.advanceTimersByTimeAsync(7_000) + + expect(attempts).toEqual([0, 1_000, 3_000, 7_000]) + expect(terminalRecoveryState(db, dispatchId)).toMatchObject({ + terminal_ack_recovery_state: 'retryable', + terminal_ack_recovery_attempts: 4, + terminal_ack_recovery_next_at_ms: 15_000, + terminal_ack_recovery_error_code: 'operation_unknown' + }) + }) + + it('continues at base cadence while a fulfilled sync makes durable progress', async () => { + vi.useFakeTimers({ now: 0 }) + const { db } = createDb() + const dispatchId = createHistoricalDispatch(db, 'progress') + db.setFederatedHomeImportSequence(dispatchId, 4) + const attempts: number[] = [] + const { runtime } = createRuntime(db, async () => { + attempts.push(Date.now()) + db.recordFederatedHomeAcknowledgment({ + dispatchId, + remoteRuntimeEpoch: 'worker_epoch', + sequence: attempts.length + }) + }) + + runtime.ensureOrchestrationFederationRelay() + await vi.advanceTimersByTimeAsync(3_000) + + expect(attempts).toEqual([0, 1_000, 2_000, 3_000]) + expect(terminalRecoveryState(db, dispatchId)).toMatchObject({ + to_home_acknowledged_sequence: 4, + terminal_ack_recovery_state: 'pending', + terminal_ack_recovery_attempts: 0 + }) + }) + + it('hands an active dispatch to durable recovery after it becomes terminal', async () => { + vi.useFakeTimers({ now: 0 }) + const { db } = createDb() + const dispatchId = createHistoricalDispatch(db, 'active-handoff') + db.recordWorkerStage({ dispatchId, stage: 'running', state: 'ready' }) + const attempts: number[] = [] + let rejectActive!: () => void + let active = true + createRuntime(db, async () => { + attempts.push(Date.now()) + if (active) { + active = false + return new Promise((_resolve, reject) => { + rejectActive = () => { + db.recordWorkerStage({ dispatchId, stage: 'completed', state: 'succeeded' }) + reject(new Error('transport unavailable')) + } + }) + } + throw new Error('transport unavailable') + }) + + await vi.advanceTimersByTimeAsync(0) + expect(attempts).toEqual([0]) + rejectActive() + await vi.advanceTimersByTimeAsync(4_000) + + expect(attempts).toEqual([0, 1_000, 2_000, 4_000]) + expect(terminalRecoveryState(db, dispatchId)).toMatchObject({ + terminal_ack_recovery_state: 'retryable', + terminal_ack_recovery_attempts: 3, + terminal_ack_recovery_next_at_ms: 8_000 + }) + expect(vi.getTimerCount()).toBe(1) + }) + it('terminalizes a permanent oldest row without blocking a later recoverable row', async () => { vi.useFakeTimers({ now: 0 }) const { db } = createDb() @@ -291,8 +402,13 @@ describe('terminal federation acknowledgment recovery', () => { .mockResolvedValue(undefined) runtime.setOrchestrationDb({ listActiveFederatedDispatches: () => [], + capFederatedTerminalRecoveryDeadlines: () => {}, findNextTerminalFederatedDispatchPendingAcknowledgment: (afterRowId: number) => candidates.find((candidate) => candidate.rowId > afterRowId), + getFederatedDispatch: () => ({ + to_home_acknowledged_sequence: 1, + to_home_imported_sequence: 1 + }), recordFederatedTerminalRecoverySuccess: (dispatchId: string) => { candidates.splice( candidates.findIndex((candidate) => candidate.dispatchId === dispatchId), @@ -326,8 +442,13 @@ describe('terminal federation acknowledgment recovery', () => { const sync = vi.spyOn(runtime, 'syncOrchestrationFederatedDispatch').mockReturnValue(blocked) runtime.setOrchestrationDb({ listActiveFederatedDispatches: () => [], + capFederatedTerminalRecoveryDeadlines: () => {}, findNextTerminalFederatedDispatchPendingAcknowledgment: (afterRowId: number) => - candidates.find((candidate) => candidate.rowId > afterRowId) + candidates.find((candidate) => candidate.rowId > afterRowId), + getFederatedDispatch: () => ({ + to_home_acknowledged_sequence: 0, + to_home_imported_sequence: 1 + }) } as never) expect(sync).toHaveBeenCalledTimes(1) expect(vi.getTimerCount()).toBe(0)