fix(orchestration): close recovery scheduler gaps

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<void> {
     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())
+      )
     )
   }
 

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)
     )

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<void>((_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)