mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
fix(orchestration): bound terminal federation recovery
This commit is contained in:
Jinwoo-H
@@ -8401,16 +8401,17 @@
|
||||
"providers": ["local", "daemon", "ssh", "wsl", "remote-runtime"],
|
||||
"coveredPlatforms": ["macos"],
|
||||
"coveredProviders": ["local", "daemon", "ssh"],
|
||||
"coverageNotes": "A deterministic service-state-machine oracle now models a current-contract worker and coordinator whose renderer graph identities disappear across an app/runtime update. It exercises the production verifier with restored PTY and hydrated hook commitments, proves authenticated completion replay across a fresh runtime, explicit takeover, ordinary mail routing, remote-attachment process fencing, retained Task/Dispatch/terminal identity, and unchanged fixture marker bytes, and rejects foreign pane evidence. Other deterministic units cover authority-aware legacy formatting, exact legacy worker identity planning, local worker presentation, retained-output reads after adoption, reveal-failure warnings, stable-pane Run/Dispatch routing, creator pane/process/Run-generation fencing, indexed retained-Run lookup scaling, the SSH in-process CLI fallback, exact legacy Task/Dispatch/report-receipt verification, JSON rejection envelopes, and capability-negotiated Run-home settlement. The federation oracle drops the first terminal ACK, restarts Run-home twice, verifies replay through the existing polling path and the persisted acknowledgment watermark, races identical reports, accepts preexisting same-outcome ACK batches, rejects conflicting batches, and covers protocol v1/v2 completion across independently updated peers. A migration unit upgrades a v26 database with a zeroed watermark without replaying older migrations. Three isolated macOS Electron journeys launch fake Codex workers through the real RPC path. Distinct installed A/B artifacts plus live SSH, WSL, folder, Linux, and Windows cutover journeys remain explicit gaps.",
|
||||
"coverageNotes": "A deterministic service-state-machine oracle now models a current-contract worker and coordinator whose renderer graph identities disappear across an app/runtime update. It exercises the production verifier with restored PTY and hydrated hook commitments, proves authenticated completion replay across a fresh runtime, explicit takeover, ordinary mail routing, remote-attachment process fencing, retained Task/Dispatch/terminal identity, and unchanged fixture marker bytes, and rejects foreign pane evidence. Other deterministic units cover authority-aware legacy formatting, exact legacy worker identity planning, local worker presentation, retained-output reads after adoption, reveal-failure warnings, stable-pane Run/Dispatch routing, creator pane/process/Run-generation fencing, indexed retained-Run lookup scaling, the SSH in-process CLI fallback, exact legacy Task/Dispatch/report-receipt verification, JSON rejection envelopes, and capability-negotiated Run-home settlement. The federation oracle drops the first terminal ACK, restarts Run-home twice, verifies replay through the persisted acknowledgment watermark, races identical reports, accepts preexisting same-outcome ACK batches, rejects conflicting batches, and covers protocol v1/v2 completion across independently updated peers. A fake-clock SQLite recovery oracle classifies proven permanent failures, preserves unknown outcomes, verifies capped durable backoff across restart, gives later rows a turn, skips duplicate ACKs, and fences disposal. Migration units upgrade v26 acknowledgments and v27 recovery state without replaying older migrations. Three isolated macOS Electron journeys launch fake Codex workers through the real RPC path. Distinct installed A/B artifacts plus live SSH, WSL, folder, Linux, and Windows cutover journeys remain explicit gaps.",
|
||||
"motivatingLinks": [
|
||||
"https://github.com/stablyai/orca/pull/11107#discussion_r3663321387",
|
||||
"https://github.com/stablyai/orca/issues/13858",
|
||||
"https://github.com/stablyai/orca/pull/14105",
|
||||
"https://linear.app/stably/issue/STA-4020",
|
||||
"https://linear.app/stably/issue/STA-4021",
|
||||
"https://linear.app/stably/issue/STA-4022"
|
||||
"https://linear.app/stably/issue/STA-4022",
|
||||
"https://linear.app/stably/issue/STA-4174"
|
||||
],
|
||||
"invariant": "Starting a worker in the coordinator's current workspace must materialize one inactive terminal tab before worker-start returns, preserve coordinator focus, and remain exactly once after workspace re-entry. After an app update or restart, an exact live legacy worker must fence automatic provider resume, adopt its original PTY into its original background pane, retain readable output, and clear the resume record without spawning, writing, signalling, interrupting, replacing, or focusing the worker. A current-contract worker whose renderer graph identity is temporarily absent must retain its Dispatch capability and settle exactly once from exact hook-attested handle, pane, and process evidence; otherwise only an exact attested coordinator may take over. A worker_done caller may report success only after the owning runtime returns an explicit lifecycle verdict or authoritative reads prove that the exact Task, Dispatch, and worker report receipt settled the expected outcome. Federated terminal settlement must remain replay-eligible until the worker durably acknowledges it, and identical same-outcome retries must converge idempotently. Independently updated clients and worker servers must preserve the negotiated protocol: current peers use Run-home lifecycle settlement, while protocol v1/v2 peers retain their legacy completion path without receiving newer-only fields. A federated worker may accept only the authority defined by its negotiated protocol. An exact existing target workspace must receive a discoverable tab without stealing coordinator focus; if renderer reveal fails, worker-start must expose that the live worker remains background-only. Run and Dispatch checks must resolve through the caller's stable pane identity when a terminal handle is reminted, while a live handle outranks mismatched pane metadata. A nested worker's creator edge requires the current creator pane, process incarnation, and owning Run generation; reminting and rebinding that pane to another Run must remove the stale edge. Explicit legacy terminal inspection remains handle-scoped, and remote or headless worker presentation remains background-only.",
|
||||
"invariant": "Starting a worker in the coordinator's current workspace must materialize one inactive terminal tab before worker-start returns, preserve coordinator focus, and remain exactly once after workspace re-entry. After an app update or restart, an exact live legacy worker must fence automatic provider resume, adopt its original PTY into its original background pane, retain readable output, and clear the resume record without spawning, writing, signalling, interrupting, replacing, or focusing the worker. A current-contract worker whose renderer graph identity is temporarily absent must retain its Dispatch capability and settle exactly once from exact hook-attested handle, pane, and process evidence; otherwise only an exact attested coordinator may take over. A worker_done caller may report success only after the owning runtime returns an explicit lifecycle verdict or authoritative reads prove that the exact Task, Dispatch, and worker report receipt settled the expected outcome. Federated terminal settlement must remain replay-eligible until the worker durably acknowledges it, and identical same-outcome retries must converge idempotently. Historical recovery must durably terminalize only proven permanent failures, keep transient and unknown outcomes retryable with capped persisted backoff, survive restart, and never let one row starve later eligible work. Independently updated clients and worker servers must preserve the negotiated protocol: current peers use Run-home lifecycle settlement, while protocol v1/v2 peers retain their legacy completion path without receiving newer-only fields. A federated worker may accept only the authority defined by its negotiated protocol. An exact existing target workspace must receive a discoverable tab without stealing coordinator focus; if renderer reveal fails, worker-start must expose that the live worker remains background-only. Run and Dispatch checks must resolve through the caller's stable pane identity when a terminal handle is reminted, while a live handle outranks mismatched pane metadata. A nested worker's creator edge requires the current creator pane, process incarnation, and owning Run generation; reminting and rebinding that pane to another Run must remove the stale edge. Explicit legacy terminal inspection remains handle-scoped, and remote or headless worker presentation remains background-only.",
|
||||
"oracle": "Drive Run create, Task create, and worker-start through production Electron runtimes with a deterministic Codex fixture. Require append-only ledgers with one still-live PID and no interruption, a visible inactive worker tab while the coordinator stays active, Run delivery through stable pane identity, and stable PTY/incarnation, tab, leaf, worktree, Task, and Dispatch across workspace re-entry. In a restart journey, retain the original daemon PTY and PID, remove renderer ownership, retain sleeping-session evidence, mark the Dispatch legacy, relaunch, and require exact inactive tab adoption, readable ACK output, cleared resume state, one spawn, and no resume argv or Conversation interrupted text after another workspace round trip. The service oracle removes renderer lookup identity from current-contract callers while retaining real restored-PTY and hook commitments, replays authenticated completion and takeover across fresh runtimes, and requires one Task, Dispatch, terminal authority, message, mutation, ordinary-mail delivery, remote process fencing, and unchanged fixture marker bytes while foreign pane evidence remains rejected. Unit tests separately remint a creator pane and process from Run A into Run B, require the nested Run A worker to fall back to its current coordinator, require indexed query plans, and bound 300 Task reads with 50,000 retained Runs. They also assert authority-specific legacy affordances, exact identity and owner matching, retained-output fallback, pane-stable routing, federated non-activation, and SSH fallback parity.",
|
||||
"commands": [
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/orchestration-runtime-update-settlement.test.ts --reporter=dot",
|
||||
@@ -8418,6 +8419,7 @@
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/cli/handlers/orchestration-lifecycle-rejection.test.ts src/cli/handlers/orchestration-lifecycle-json-rejection.test.ts src/cli/handlers/orchestration-migration.test.ts",
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/formatter.test.ts src/main/runtime/rpc/methods/orchestration-federation.test.ts",
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts --reporter=dot",
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/federation-terminal-recovery.test.ts --reporter=dot",
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts --reporter=dot",
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/orchestration-legacy-worker-terminal-recovery.test.ts",
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/orchestration-creator-authority-performance.test.ts",
|
||||
@@ -8441,6 +8443,7 @@
|
||||
"src/main/runtime/rpc/methods/orchestration.test.ts",
|
||||
"src/main/runtime/rpc/methods/orchestration-federation.test.ts",
|
||||
"src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts",
|
||||
"src/main/runtime/orchestration/federation-terminal-recovery.test.ts",
|
||||
"src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts",
|
||||
"src/main/ssh/ssh-remote-orca-cli.test.ts",
|
||||
"tests/e2e/orchestration-worker-terminal-visibility.spec.ts",
|
||||
@@ -8551,12 +8554,21 @@
|
||||
"a rejected report replays its exact verdict without mutating durable rejection mail twice"
|
||||
]
|
||||
},
|
||||
{
|
||||
"file": "src/main/runtime/orchestration/federation-terminal-recovery.test.ts",
|
||||
"assertions": [
|
||||
"proven permanent failures become durable terminal rows and are not retried after restart",
|
||||
"transient and unknown failures use capped persisted backoff and converge when the worker returns",
|
||||
"a permanently failing oldest row does not block a later recoverable row",
|
||||
"duplicate acknowledgments and relay disposal schedule no further work"
|
||||
]
|
||||
},
|
||||
{
|
||||
"file": "src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts",
|
||||
"assertions": [
|
||||
"a v26 database gains a zeroed durable home-to-worker acknowledgment watermark",
|
||||
"a complete v26 database starts migration at v26 rather than replaying v7-v26",
|
||||
"the migration advances schema version without losing the imported relay cursor"
|
||||
"a v27 database gains zeroed terminal acknowledgment recovery state",
|
||||
"version-aware migration preserves imported and acknowledged cursors"
|
||||
]
|
||||
},
|
||||
{
|
||||
|
||||
+17
-2
@@ -77,7 +77,10 @@ import {
|
||||
type OrchestrationEnvironmentTransport
|
||||
} from './runtime/orchestration/environment-transport'
|
||||
import { callRuntimeEnvironment } from './ipc/runtime-environment-transport-routing'
|
||||
import { resolveEnvironment } from '../shared/runtime-environment-store'
|
||||
import {
|
||||
resolveEnvironment,
|
||||
RuntimeEnvironmentNotFoundError
|
||||
} from '../shared/runtime-environment-store'
|
||||
import { getPreferredPairingOffer } from '../shared/runtime-environments'
|
||||
import { OrcaRuntimeRpcServer } from './runtime/runtime-rpc'
|
||||
import {
|
||||
@@ -88,6 +91,7 @@ import { resolveAdvertisedPairingEndpoint } from './runtime/pairing-endpoint'
|
||||
import { ServeReadinessPublisher } from './server/serve-readiness'
|
||||
import { reserveServeStdoutForReadiness } from './server/serve-stdout-boundary'
|
||||
import { DesktopRelayService } from './runtime/relay/desktop-relay-service'
|
||||
import { OrchestrationError } from './runtime/orchestration/orchestration-error'
|
||||
import type { RelayBrokerStatus } from './runtime/relay/relay-session-broker'
|
||||
import { awaitRuntimeFileWatcherUnsubscribes } from './runtime/orca-runtime-files'
|
||||
import { clearRuntimeMetadataIfOwned } from './runtime/runtime-metadata'
|
||||
@@ -2488,7 +2492,18 @@ void app.whenReady().then(async () => {
|
||||
})
|
||||
const orchestrationEnvironmentTransport: OrchestrationEnvironmentTransport = {
|
||||
resolve: (selector) => {
|
||||
const environment = resolveEnvironment(app.getPath('userData'), selector)
|
||||
let environment: ReturnType<typeof resolveEnvironment>
|
||||
try {
|
||||
environment = resolveEnvironment(app.getPath('userData'), selector)
|
||||
} catch (error) {
|
||||
if (error instanceof RuntimeEnvironmentNotFoundError) {
|
||||
throw new OrchestrationError(
|
||||
'environment_not_found',
|
||||
`Saved worker environment ${selector} no longer exists.`
|
||||
)
|
||||
}
|
||||
throw error
|
||||
}
|
||||
const pairing = getPreferredPairingOffer(environment)
|
||||
return {
|
||||
environmentId: environment.id,
|
||||
|
||||
@@ -164,6 +164,10 @@ import type {
|
||||
} from './orchestration/environment-transport'
|
||||
import { clearFederationAckCheckpoints } from './orchestration/federation-ack-checkpoints'
|
||||
import { syncFederatedDispatch } from './orchestration/federation-sync'
|
||||
import {
|
||||
classifyFederationTerminalRecoveryFailure,
|
||||
FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS
|
||||
} from './orchestration/federation-terminal-recovery-policy'
|
||||
import { formatMessagePointer } from './orchestration/formatter'
|
||||
import { selectExactWorkerProviderSession } from './orchestration/worker-provider-session'
|
||||
import type {
|
||||
@@ -4783,25 +4787,57 @@ export class OrcaRuntimeService {
|
||||
|
||||
private async recoverNextTerminalHistoryAcknowledgment(generation: number): Promise<void> {
|
||||
const db = this.getOrchestrationDb()
|
||||
const nowMs = Date.now()
|
||||
let historical = db.findNextTerminalFederatedDispatchPendingAcknowledgment(
|
||||
this.orchestrationTerminalRecoveryRowId
|
||||
this.orchestrationTerminalRecoveryRowId,
|
||||
nowMs
|
||||
)
|
||||
if (!historical && this.orchestrationTerminalRecoveryRowId > 0) {
|
||||
this.orchestrationTerminalRecoveryRowId = 0
|
||||
historical = db.findNextTerminalFederatedDispatchPendingAcknowledgment(0)
|
||||
historical = db.findNextTerminalFederatedDispatchPendingAcknowledgment(0, nowMs)
|
||||
}
|
||||
if (!historical) {
|
||||
const nextAt = db.getNextTerminalFederatedDispatchRecoveryAt()
|
||||
if (nextAt !== undefined) {
|
||||
this.scheduleTerminalHistoryRecovery(generation, Math.max(0, nextAt - nowMs))
|
||||
}
|
||||
return
|
||||
}
|
||||
this.orchestrationTerminalRecoveryRowId = historical.rowId
|
||||
await this.syncOrchestrationFederatedDispatch(historical.dispatchId).catch(() => undefined)
|
||||
const result = await this.syncOrchestrationFederatedDispatch(historical.dispatchId).then(
|
||||
() => ({ ok: true as const }),
|
||||
(error: unknown) => ({ ok: false as const, error })
|
||||
)
|
||||
if (generation !== this.orchestrationFederationRelayGeneration) {
|
||||
return
|
||||
}
|
||||
if (result.ok) {
|
||||
db.recordFederatedTerminalRecoverySuccess(historical.dispatchId, Date.now())
|
||||
} else {
|
||||
db.recordFederatedTerminalRecoveryFailure({
|
||||
dispatchId: historical.dispatchId,
|
||||
...classifyFederationTerminalRecoveryFailure(result.error),
|
||||
nowMs: Date.now()
|
||||
})
|
||||
}
|
||||
const nextAt = db.getNextTerminalFederatedDispatchRecoveryAt()
|
||||
if (nextAt === undefined) {
|
||||
return
|
||||
}
|
||||
this.scheduleTerminalHistoryRecovery(
|
||||
generation,
|
||||
Math.max(FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS, nextAt - Date.now())
|
||||
)
|
||||
}
|
||||
|
||||
private scheduleTerminalHistoryRecovery(generation: number, delayMs: number): void {
|
||||
if (generation !== this.orchestrationFederationRelayGeneration) {
|
||||
return
|
||||
}
|
||||
this.orchestrationTerminalHistoryRecoveryTimer = setTimeout(() => {
|
||||
this.orchestrationTerminalHistoryRecoveryTimer = null
|
||||
this.ensureTerminalHistoryRecovery()
|
||||
}, 1_000)
|
||||
}, delayMs)
|
||||
this.orchestrationTerminalHistoryRecoveryTimer.unref?.()
|
||||
}
|
||||
|
||||
|
||||
@@ -65,6 +65,7 @@ import {
|
||||
ensureMutationReceiptCapacity,
|
||||
migrateMutationReceiptCapacity
|
||||
} from './mutation-receipt-capacity'
|
||||
import { getFederationTerminalRecoveryDelayMs } from './federation-terminal-recovery-policy'
|
||||
|
||||
function parseWorkerTerminalPriorOwnerIds(value: string): string[] | null {
|
||||
try {
|
||||
@@ -287,8 +288,8 @@ type RunListCursor = {
|
||||
id: string
|
||||
}
|
||||
|
||||
// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments.
|
||||
const SCHEMA_VERSION = 27
|
||||
// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable terminal acknowledgment recovery.
|
||||
const SCHEMA_VERSION = 28
|
||||
|
||||
function hardenOrchestrationDatabaseFiles(dbPath: (string & {}) | ':memory:'): void {
|
||||
if (dbPath === ':memory:' || process.platform === 'win32') {
|
||||
@@ -471,6 +472,11 @@ export class OrchestrationDb {
|
||||
remote_terminal_handle TEXT,
|
||||
to_home_imported_sequence INTEGER NOT NULL DEFAULT 0,
|
||||
to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0,
|
||||
terminal_ack_recovery_state TEXT NOT NULL DEFAULT 'pending'
|
||||
CHECK(terminal_ack_recovery_state IN ('pending', 'retryable', 'terminal')),
|
||||
terminal_ack_recovery_attempts INTEGER NOT NULL DEFAULT 0,
|
||||
terminal_ack_recovery_next_at_ms INTEGER NOT NULL DEFAULT 0,
|
||||
terminal_ack_recovery_error_code TEXT,
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||||
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||||
);
|
||||
@@ -875,6 +881,11 @@ export class OrchestrationDb {
|
||||
remote_terminal_handle TEXT,
|
||||
to_home_imported_sequence INTEGER NOT NULL DEFAULT 0,
|
||||
to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0,
|
||||
terminal_ack_recovery_state TEXT NOT NULL DEFAULT 'pending'
|
||||
CHECK(terminal_ack_recovery_state IN ('pending', 'retryable', 'terminal')),
|
||||
terminal_ack_recovery_attempts INTEGER NOT NULL DEFAULT 0,
|
||||
terminal_ack_recovery_next_at_ms INTEGER NOT NULL DEFAULT 0,
|
||||
terminal_ack_recovery_error_code TEXT,
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||||
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||||
);
|
||||
@@ -1002,6 +1013,28 @@ export class OrchestrationDb {
|
||||
'ALTER TABLE federated_dispatches ADD COLUMN to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0'
|
||||
)
|
||||
}
|
||||
if (current < 28) {
|
||||
if (!this.hasColumn('federated_dispatches', 'terminal_ack_recovery_state')) {
|
||||
this.db.exec(
|
||||
"ALTER TABLE federated_dispatches ADD COLUMN terminal_ack_recovery_state TEXT NOT NULL DEFAULT 'pending' CHECK(terminal_ack_recovery_state IN ('pending', 'retryable', 'terminal'))"
|
||||
)
|
||||
}
|
||||
if (!this.hasColumn('federated_dispatches', 'terminal_ack_recovery_attempts')) {
|
||||
this.db.exec(
|
||||
'ALTER TABLE federated_dispatches ADD COLUMN terminal_ack_recovery_attempts INTEGER NOT NULL DEFAULT 0'
|
||||
)
|
||||
}
|
||||
if (!this.hasColumn('federated_dispatches', 'terminal_ack_recovery_next_at_ms')) {
|
||||
this.db.exec(
|
||||
'ALTER TABLE federated_dispatches ADD COLUMN terminal_ack_recovery_next_at_ms INTEGER NOT NULL DEFAULT 0'
|
||||
)
|
||||
}
|
||||
if (!this.hasColumn('federated_dispatches', 'terminal_ack_recovery_error_code')) {
|
||||
this.db.exec(
|
||||
'ALTER TABLE federated_dispatches ADD COLUMN terminal_ack_recovery_error_code TEXT'
|
||||
)
|
||||
}
|
||||
}
|
||||
this.db.exec(`
|
||||
CREATE INDEX IF NOT EXISTS idx_dispatch_assignee_pane_leaf
|
||||
ON dispatch_contexts(${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL})
|
||||
@@ -4564,7 +4597,8 @@ export class OrchestrationDb {
|
||||
}
|
||||
|
||||
findNextTerminalFederatedDispatchPendingAcknowledgment(
|
||||
afterRowId: number
|
||||
afterRowId: number,
|
||||
nowMs = Date.now()
|
||||
): { dispatchId: string; rowId: number } | undefined {
|
||||
return this.db
|
||||
.prepare(
|
||||
@@ -4573,11 +4607,72 @@ export class OrchestrationDb {
|
||||
INNER JOIN worker_dispatches wd ON wd.dispatch_id = fd.dispatch_id
|
||||
WHERE wd.state NOT IN ('starting', 'ready', 'stopping', 'start_unknown', 'stop_unknown')
|
||||
AND fd.to_home_acknowledged_sequence < fd.to_home_imported_sequence
|
||||
AND fd.terminal_ack_recovery_state != 'terminal'
|
||||
AND fd.terminal_ack_recovery_next_at_ms <= ?
|
||||
AND fd.rowid > ?
|
||||
ORDER BY fd.rowid
|
||||
LIMIT 1`
|
||||
)
|
||||
.get(afterRowId) as { dispatchId: string; rowId: number } | undefined
|
||||
.get(nowMs, afterRowId) as { dispatchId: string; rowId: number } | undefined
|
||||
}
|
||||
|
||||
getNextTerminalFederatedDispatchRecoveryAt(): number | undefined {
|
||||
const row = this.db
|
||||
.prepare(
|
||||
`SELECT MIN(fd.terminal_ack_recovery_next_at_ms) AS nextAt
|
||||
FROM federated_dispatches fd
|
||||
INNER JOIN worker_dispatches wd ON wd.dispatch_id = fd.dispatch_id
|
||||
WHERE wd.state NOT IN ('starting', 'ready', 'stopping', 'start_unknown', 'stop_unknown')
|
||||
AND fd.to_home_acknowledged_sequence < fd.to_home_imported_sequence
|
||||
AND fd.terminal_ack_recovery_state != 'terminal'`
|
||||
)
|
||||
.get() as { nextAt: number | null }
|
||||
return row.nextAt ?? undefined
|
||||
}
|
||||
|
||||
recordFederatedTerminalRecoveryFailure(params: {
|
||||
dispatchId: string
|
||||
errorCode: string | null
|
||||
terminal: boolean
|
||||
nowMs: number
|
||||
}): void {
|
||||
const current = this.getFederatedDispatch(params.dispatchId)
|
||||
if (!current || current.to_home_acknowledged_sequence >= current.to_home_imported_sequence) {
|
||||
return
|
||||
}
|
||||
const attempts = current.terminal_ack_recovery_attempts + 1
|
||||
const nextAt = params.terminal
|
||||
? 0
|
||||
: params.nowMs + getFederationTerminalRecoveryDelayMs(attempts)
|
||||
this.db
|
||||
.prepare(
|
||||
`UPDATE federated_dispatches
|
||||
SET terminal_ack_recovery_state = ?, terminal_ack_recovery_attempts = ?,
|
||||
terminal_ack_recovery_next_at_ms = ?, terminal_ack_recovery_error_code = ?,
|
||||
updated_at = datetime('now')
|
||||
WHERE dispatch_id = ?
|
||||
AND to_home_acknowledged_sequence < to_home_imported_sequence`
|
||||
)
|
||||
.run(
|
||||
params.terminal ? 'terminal' : 'retryable',
|
||||
attempts,
|
||||
nextAt,
|
||||
params.errorCode,
|
||||
params.dispatchId
|
||||
)
|
||||
}
|
||||
|
||||
recordFederatedTerminalRecoverySuccess(dispatchId: string, nowMs: number): void {
|
||||
this.db
|
||||
.prepare(
|
||||
`UPDATE federated_dispatches
|
||||
SET terminal_ack_recovery_state = 'pending', terminal_ack_recovery_attempts = 0,
|
||||
terminal_ack_recovery_next_at_ms = CASE
|
||||
WHEN to_home_acknowledged_sequence < to_home_imported_sequence THEN ? ELSE 0 END,
|
||||
terminal_ack_recovery_error_code = NULL, updated_at = datetime('now')
|
||||
WHERE dispatch_id = ?`
|
||||
)
|
||||
.run(nowMs + getFederationTerminalRecoveryDelayMs(1), dispatchId)
|
||||
}
|
||||
|
||||
isFederatedDispatchRelayEligible(dispatchId: string): boolean {
|
||||
@@ -4592,6 +4687,7 @@ export class OrchestrationDb {
|
||||
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'
|
||||
)
|
||||
)`
|
||||
)
|
||||
@@ -5277,6 +5373,10 @@ export class OrchestrationDb {
|
||||
THEN MAX(to_home_acknowledged_sequence, ?)
|
||||
ELSE ?
|
||||
END,
|
||||
terminal_ack_recovery_state = 'pending',
|
||||
terminal_ack_recovery_attempts = 0,
|
||||
terminal_ack_recovery_next_at_ms = 0,
|
||||
terminal_ack_recovery_error_code = NULL,
|
||||
updated_at = datetime('now')
|
||||
WHERE dispatch_id = ?`
|
||||
)
|
||||
|
||||
@@ -27,7 +27,7 @@ describe('federation acknowledgment migration', () => {
|
||||
const oldDb = new Database(dbPath)
|
||||
oldDb.exec('ALTER TABLE federated_dispatches DROP COLUMN to_home_acknowledged_sequence')
|
||||
oldDb.pragma('user_version = 26')
|
||||
expect(resolveOrchestrationMigrationStartVersion(oldDb, 26, 27)).toBe(26)
|
||||
expect(resolveOrchestrationMigrationStartVersion(oldDb, 26, 28)).toBe(26)
|
||||
oldDb
|
||||
.prepare(
|
||||
`INSERT INTO federated_dispatches (
|
||||
@@ -41,10 +41,54 @@ describe('federation acknowledgment migration', () => {
|
||||
db = new OrchestrationDb(dbPath)
|
||||
const sqlite = (db as unknown as { db: Database.Database }).db
|
||||
|
||||
expect(sqlite.pragma('user_version', { simple: true })).toBe(27)
|
||||
expect(sqlite.pragma('user_version', { simple: true })).toBe(28)
|
||||
expect(db.getFederatedDispatch('ctx_migrated')).toMatchObject({
|
||||
to_home_imported_sequence: 2,
|
||||
to_home_acknowledged_sequence: 0
|
||||
to_home_acknowledged_sequence: 0,
|
||||
terminal_ack_recovery_state: 'pending',
|
||||
terminal_ack_recovery_attempts: 0,
|
||||
terminal_ack_recovery_next_at_ms: 0,
|
||||
terminal_ack_recovery_error_code: null
|
||||
})
|
||||
})
|
||||
|
||||
it('adds durable terminal recovery policy state to v27 dispatches', () => {
|
||||
tempDir = mkdtempSync(join(tmpdir(), 'orca-federation-recovery-migration-'))
|
||||
const dbPath = join(tempDir, 'orchestration.db')
|
||||
db = new OrchestrationDb(dbPath)
|
||||
db.close()
|
||||
db = undefined
|
||||
|
||||
const oldDb = new Database(dbPath)
|
||||
oldDb.exec(`
|
||||
ALTER TABLE federated_dispatches DROP COLUMN terminal_ack_recovery_error_code;
|
||||
ALTER TABLE federated_dispatches DROP COLUMN terminal_ack_recovery_next_at_ms;
|
||||
ALTER TABLE federated_dispatches DROP COLUMN terminal_ack_recovery_attempts;
|
||||
ALTER TABLE federated_dispatches DROP COLUMN terminal_ack_recovery_state;
|
||||
`)
|
||||
oldDb.pragma('user_version = 27')
|
||||
expect(resolveOrchestrationMigrationStartVersion(oldDb, 27, 28)).toBe(27)
|
||||
oldDb
|
||||
.prepare(
|
||||
`INSERT INTO federated_dispatches (
|
||||
dispatch_id, environment_id, environment_name, peer_fingerprint,
|
||||
protocol_version, to_home_imported_sequence, to_home_acknowledged_sequence
|
||||
) VALUES ('ctx_retry', 'env', 'worker', 'peer', 3, 2, 1)`
|
||||
)
|
||||
.run()
|
||||
oldDb.close()
|
||||
|
||||
db = new OrchestrationDb(dbPath)
|
||||
const sqlite = (db as unknown as { db: Database.Database }).db
|
||||
|
||||
expect(sqlite.pragma('user_version', { simple: true })).toBe(28)
|
||||
expect(db.getFederatedDispatch('ctx_retry')).toMatchObject({
|
||||
to_home_imported_sequence: 2,
|
||||
to_home_acknowledged_sequence: 1,
|
||||
terminal_ack_recovery_state: 'pending',
|
||||
terminal_ack_recovery_attempts: 0,
|
||||
terminal_ack_recovery_next_at_ms: 0,
|
||||
terminal_ack_recovery_error_code: null
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
import { OrchestrationError } from './orchestration-error'
|
||||
|
||||
export const FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS = 1_000
|
||||
export const FEDERATION_TERMINAL_RECOVERY_MAX_DELAY_MS = 30_000
|
||||
|
||||
const TERMINAL_ERROR_CODES = new Set([
|
||||
'dispatch_not_found',
|
||||
'environment_not_found',
|
||||
'peer_changed'
|
||||
])
|
||||
|
||||
export type FederationTerminalRecoveryFailure = {
|
||||
errorCode: string | null
|
||||
terminal: boolean
|
||||
}
|
||||
|
||||
export function classifyFederationTerminalRecoveryFailure(
|
||||
error: unknown
|
||||
): FederationTerminalRecoveryFailure {
|
||||
const errorCode = error instanceof OrchestrationError ? error.code : null
|
||||
return { errorCode, terminal: errorCode !== null && TERMINAL_ERROR_CODES.has(errorCode) }
|
||||
}
|
||||
|
||||
export function getFederationTerminalRecoveryDelayMs(attempts: number): number {
|
||||
const exponent = Math.max(0, Math.min(attempts - 1, 30))
|
||||
return Math.min(
|
||||
FEDERATION_TERMINAL_RECOVERY_BASE_DELAY_MS * 2 ** exponent,
|
||||
FEDERATION_TERMINAL_RECOVERY_MAX_DELAY_MS
|
||||
)
|
||||
}
|
||||
@@ -1,35 +1,308 @@
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { OrcaRuntimeService } from '../orca-runtime'
|
||||
import { OrchestrationDb } from './db'
|
||||
import { OrchestrationError } from './orchestration-error'
|
||||
import { classifyFederationTerminalRecoveryFailure } from './federation-terminal-recovery-policy'
|
||||
|
||||
type HistoricalDispatch = { dispatchId: string; rowId: number }
|
||||
|
||||
describe('terminal federation acknowledgment recovery', () => {
|
||||
const databases: OrchestrationDb[] = []
|
||||
const runtimes: OrcaRuntimeService[] = []
|
||||
const tempDirs: string[] = []
|
||||
|
||||
afterEach(() => {
|
||||
for (const runtime of runtimes.splice(0)) {
|
||||
runtime.stopOrchestrationFederationRelay()
|
||||
}
|
||||
for (const db of databases.splice(0)) {
|
||||
db.close()
|
||||
}
|
||||
for (const dir of tempDirs.splice(0)) {
|
||||
rmSync(dir, { recursive: true, force: true })
|
||||
}
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it('bounds migration replay to one oldest-first recovery', async () => {
|
||||
vi.useFakeTimers()
|
||||
function createRuntime(
|
||||
db: OrchestrationDb,
|
||||
sync?: (dispatchId: string) => Promise<void>
|
||||
): { runtime: OrcaRuntimeService; sync: ReturnType<typeof vi.spyOn> } {
|
||||
const runtime = new OrcaRuntimeService(null, undefined, {
|
||||
orchestrationEnvironmentTransport: { resolve: vi.fn(), call: vi.fn() }
|
||||
})
|
||||
runtimes.push(runtime)
|
||||
const syncSpy = vi
|
||||
.spyOn(runtime, 'syncOrchestrationFederatedDispatch')
|
||||
.mockImplementation(sync ?? (async () => {}))
|
||||
runtime.setOrchestrationDb(db)
|
||||
return { runtime, sync: syncSpy }
|
||||
}
|
||||
|
||||
function createDb(onDisk = false): { db: OrchestrationDb; dbPath?: string } {
|
||||
if (!onDisk) {
|
||||
const db = new OrchestrationDb(':memory:')
|
||||
databases.push(db)
|
||||
return { db }
|
||||
}
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-federation-recovery-'))
|
||||
const dbPath = join(dir, 'orchestration.db')
|
||||
tempDirs.push(dir)
|
||||
const db = new OrchestrationDb(dbPath)
|
||||
databases.push(db)
|
||||
return { db, dbPath }
|
||||
}
|
||||
|
||||
function createHistoricalDispatch(db: OrchestrationDb, suffix: string): string {
|
||||
const run = db.createRun({
|
||||
objective: `recover ${suffix}`,
|
||||
coordinatorHandle: `term_coord_${suffix}`,
|
||||
coordinatorPaneKey: `tab_coord_${suffix}:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa`
|
||||
})
|
||||
const task = db.createTask({ spec: `settle ${suffix}`, runId: run.id })
|
||||
const started = db.createStartingWorkerDispatch({
|
||||
taskId: task.id,
|
||||
startOptions: {},
|
||||
runtimeEpoch: 'home_epoch',
|
||||
federation: {
|
||||
environmentId: `environment_${suffix}`,
|
||||
environmentName: suffix,
|
||||
peerFingerprint: `peer_${suffix}`,
|
||||
protocolVersion: 3
|
||||
}
|
||||
})
|
||||
db.recordWorkerStage({
|
||||
dispatchId: started.dispatch.id,
|
||||
stage: 'completed',
|
||||
state: 'succeeded'
|
||||
})
|
||||
db.setFederatedHomeImportSequence(started.dispatch.id, 1)
|
||||
return started.dispatch.id
|
||||
}
|
||||
|
||||
function terminalRecoveryState(db: OrchestrationDb, dispatchId: string) {
|
||||
return db.getFederatedDispatch(dispatchId) as unknown as {
|
||||
to_home_imported_sequence: number
|
||||
to_home_acknowledged_sequence: number
|
||||
terminal_ack_recovery_state?: string
|
||||
terminal_ack_recovery_attempts?: number
|
||||
terminal_ack_recovery_next_at_ms?: number
|
||||
terminal_ack_recovery_error_code?: string | null
|
||||
}
|
||||
}
|
||||
|
||||
it.each(['dispatch_not_found', 'environment_not_found', 'peer_changed'])(
|
||||
'classifies proven permanent error %s as terminal',
|
||||
(code) => {
|
||||
expect(classifyFederationTerminalRecoveryFailure(new OrchestrationError(code, code))).toEqual(
|
||||
{ errorCode: code, terminal: true }
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
it.each(['capability_unsupported', 'operation_unknown', 'runtime_timeout'])(
|
||||
'keeps non-terminal error %s retryable',
|
||||
(code) => {
|
||||
expect(classifyFederationTerminalRecoveryFailure(new OrchestrationError(code, code))).toEqual(
|
||||
{ errorCode: code, terminal: false }
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
it('durably terminalizes a permanent failure and does not retry it after restart', async () => {
|
||||
vi.useFakeTimers({ now: 0 })
|
||||
const { db, dbPath } = createDb(true)
|
||||
const dispatchId = createHistoricalDispatch(db, 'retired')
|
||||
const attempts: number[] = []
|
||||
const { runtime } = createRuntime(db, async () => {
|
||||
attempts.push(Date.now())
|
||||
throw new OrchestrationError('environment_not_found', 'The saved worker was decommissioned.')
|
||||
})
|
||||
|
||||
runtime.ensureOrchestrationFederationRelay()
|
||||
await vi.advanceTimersByTimeAsync(60_000)
|
||||
|
||||
expect(attempts).toEqual([0])
|
||||
expect(db.findNextTerminalFederatedDispatchPendingAcknowledgment(0)).toBeUndefined()
|
||||
expect(terminalRecoveryState(db, dispatchId)).toMatchObject({
|
||||
to_home_imported_sequence: 1,
|
||||
to_home_acknowledged_sequence: 0,
|
||||
terminal_ack_recovery_state: 'terminal',
|
||||
terminal_ack_recovery_attempts: 1,
|
||||
terminal_ack_recovery_error_code: 'environment_not_found'
|
||||
})
|
||||
|
||||
runtime.stopOrchestrationFederationRelay()
|
||||
db.close()
|
||||
databases.splice(databases.indexOf(db), 1)
|
||||
const restartedDb = new OrchestrationDb(dbPath!)
|
||||
databases.push(restartedDb)
|
||||
const { sync: restartedSync } = createRuntime(restartedDb)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(60_000)
|
||||
|
||||
expect(restartedSync).not.toHaveBeenCalled()
|
||||
expect(terminalRecoveryState(restartedDb, dispatchId).terminal_ack_recovery_state).toBe(
|
||||
'terminal'
|
||||
)
|
||||
})
|
||||
|
||||
it('backs off transient failures durably and converges when the worker returns', async () => {
|
||||
vi.useFakeTimers({ now: 0 })
|
||||
const { db } = createDb()
|
||||
const dispatchId = createHistoricalDispatch(db, 'healing')
|
||||
const attempts: number[] = []
|
||||
const { runtime } = createRuntime(db, async () => {
|
||||
attempts.push(Date.now())
|
||||
if (attempts.length < 3) {
|
||||
throw new Error('worker unavailable')
|
||||
}
|
||||
db.recordFederatedHomeAcknowledgment({
|
||||
dispatchId,
|
||||
remoteRuntimeEpoch: 'worker_epoch',
|
||||
sequence: 1
|
||||
})
|
||||
})
|
||||
|
||||
runtime.ensureOrchestrationFederationRelay()
|
||||
await vi.advanceTimersByTimeAsync(3_000)
|
||||
|
||||
expect(attempts).toEqual([0, 1_000, 3_000])
|
||||
expect(db.getFederatedDispatch(dispatchId)).toMatchObject({
|
||||
to_home_acknowledged_sequence: 1
|
||||
})
|
||||
expect(db.findNextTerminalFederatedDispatchPendingAcknowledgment(0)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('persists a transient retry deadline across app restart', async () => {
|
||||
vi.useFakeTimers({ now: 0 })
|
||||
const { db, dbPath } = createDb(true)
|
||||
createHistoricalDispatch(db, 'restart')
|
||||
const { runtime, sync: firstSync } = createRuntime(db, async () => {
|
||||
throw new Error('transport unavailable')
|
||||
})
|
||||
|
||||
runtime.ensureOrchestrationFederationRelay()
|
||||
await vi.advanceTimersByTimeAsync(500)
|
||||
expect(firstSync).toHaveBeenCalledTimes(1)
|
||||
runtime.stopOrchestrationFederationRelay()
|
||||
db.close()
|
||||
databases.splice(databases.indexOf(db), 1)
|
||||
|
||||
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(499)
|
||||
expect(attempts).toEqual([])
|
||||
await vi.advanceTimersByTimeAsync(1)
|
||||
expect(attempts).toEqual([1_000])
|
||||
})
|
||||
|
||||
it('keeps unknown failures retryable with capped backoff', async () => {
|
||||
vi.useFakeTimers({ now: 0 })
|
||||
const { db } = createDb()
|
||||
const dispatchId = createHistoricalDispatch(db, 'ambiguous')
|
||||
const attempts: number[] = []
|
||||
const { runtime } = createRuntime(db, async () => {
|
||||
attempts.push(Date.now())
|
||||
throw new OrchestrationError('operation_unknown', 'The ACK outcome is unknown.')
|
||||
})
|
||||
|
||||
runtime.ensureOrchestrationFederationRelay()
|
||||
await vi.advanceTimersByTimeAsync(61_000)
|
||||
|
||||
expect(attempts).toEqual([0, 1_000, 3_000, 7_000, 15_000, 31_000, 61_000])
|
||||
expect(db.getNextTerminalFederatedDispatchRecoveryAt()).toBe(91_000)
|
||||
expect(db.findNextTerminalFederatedDispatchPendingAcknowledgment(0, 91_000)).toMatchObject({
|
||||
dispatchId
|
||||
})
|
||||
expect(terminalRecoveryState(db, dispatchId)).toMatchObject({
|
||||
terminal_ack_recovery_state: 'retryable',
|
||||
terminal_ack_recovery_attempts: 7,
|
||||
terminal_ack_recovery_next_at_ms: 91_000
|
||||
})
|
||||
})
|
||||
|
||||
it('terminalizes a permanent oldest row without blocking a later recoverable row', async () => {
|
||||
vi.useFakeTimers({ now: 0 })
|
||||
const { db } = createDb()
|
||||
const permanentId = createHistoricalDispatch(db, 'oldest')
|
||||
const recoverableId = createHistoricalDispatch(db, 'later')
|
||||
const calls: { dispatchId: string; at: number }[] = []
|
||||
const { runtime } = createRuntime(db, async (dispatchId) => {
|
||||
calls.push({ dispatchId, at: Date.now() })
|
||||
if (dispatchId === permanentId) {
|
||||
throw new OrchestrationError('dispatch_not_found', 'The old worker discarded the dispatch.')
|
||||
}
|
||||
db.recordFederatedHomeAcknowledgment({
|
||||
dispatchId,
|
||||
remoteRuntimeEpoch: 'worker_epoch',
|
||||
sequence: 1
|
||||
})
|
||||
})
|
||||
|
||||
runtime.ensureOrchestrationFederationRelay()
|
||||
await vi.advanceTimersByTimeAsync(1_000)
|
||||
|
||||
expect(calls).toEqual([
|
||||
{ dispatchId: permanentId, at: 0 },
|
||||
{ dispatchId: recoverableId, at: 1_000 }
|
||||
])
|
||||
expect(db.findNextTerminalFederatedDispatchPendingAcknowledgment(0)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('skips an already acknowledged duplicate without scheduling work', async () => {
|
||||
vi.useFakeTimers({ now: 0 })
|
||||
const { db } = createDb()
|
||||
const dispatchId = createHistoricalDispatch(db, 'duplicate')
|
||||
db.recordFederatedHomeAcknowledgment({
|
||||
dispatchId,
|
||||
remoteRuntimeEpoch: 'worker_epoch',
|
||||
sequence: 1
|
||||
})
|
||||
const { runtime, sync } = createRuntime(db)
|
||||
|
||||
runtime.ensureOrchestrationFederationRelay()
|
||||
await vi.advanceTimersByTimeAsync(60_000)
|
||||
|
||||
expect(sync).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('keeps a thousand-row backlog to one timer and one in-flight sync', async () => {
|
||||
vi.useFakeTimers({ now: 0 })
|
||||
const candidates = Array.from({ length: 1_000 }, (_, index) => ({
|
||||
dispatchId: `dispatch_${index + 1}`,
|
||||
rowId: index + 1
|
||||
}))
|
||||
const runtime = new OrcaRuntimeService(null, undefined, {
|
||||
orchestrationEnvironmentTransport: {
|
||||
resolve: vi.fn(),
|
||||
call: vi.fn()
|
||||
}
|
||||
orchestrationEnvironmentTransport: { resolve: vi.fn(), call: vi.fn() }
|
||||
})
|
||||
runtimes.push(runtime)
|
||||
const sync = vi
|
||||
.spyOn(runtime, 'syncOrchestrationFederatedDispatch')
|
||||
.mockResolvedValue(undefined)
|
||||
runtime.setOrchestrationDb({
|
||||
listActiveFederatedDispatches: () => [],
|
||||
findNextTerminalFederatedDispatchPendingAcknowledgment: (afterRowId: number) =>
|
||||
candidates.find((candidate) => candidate.rowId > afterRowId)
|
||||
candidates.find((candidate) => candidate.rowId > afterRowId),
|
||||
recordFederatedTerminalRecoverySuccess: (dispatchId: string) => {
|
||||
candidates.splice(
|
||||
candidates.findIndex((candidate) => candidate.dispatchId === dispatchId),
|
||||
1
|
||||
)
|
||||
},
|
||||
getNextTerminalFederatedDispatchRecoveryAt: () =>
|
||||
candidates.length > 0 ? Date.now() : undefined
|
||||
} as never)
|
||||
|
||||
await vi.waitFor(() => expect(sync).toHaveBeenCalledTimes(1))
|
||||
expect(sync.mock.calls).toEqual([['dispatch_1']])
|
||||
|
||||
await vi.advanceTimersByTimeAsync(3_000)
|
||||
|
||||
expect(sync.mock.calls).toEqual([
|
||||
@@ -38,65 +311,32 @@ describe('terminal federation acknowledgment recovery', () => {
|
||||
['dispatch_3'],
|
||||
['dispatch_4']
|
||||
])
|
||||
runtime.stopOrchestrationFederationRelay()
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
})
|
||||
|
||||
it('gives every unavailable terminal dispatch a turn before retrying', async () => {
|
||||
vi.useFakeTimers()
|
||||
const candidates = [1, 2, 3].map((rowId) => ({
|
||||
dispatchId: `dispatch_${rowId}`,
|
||||
rowId
|
||||
}))
|
||||
it('does not restart recovery after relay shutdown', async () => {
|
||||
vi.useFakeTimers({ now: 0 })
|
||||
let release!: () => void
|
||||
const blocked = new Promise<void>((resolve) => (release = resolve))
|
||||
const candidates: HistoricalDispatch[] = [{ dispatchId: 'dispatch_1', rowId: 1 }]
|
||||
const runtime = new OrcaRuntimeService(null, undefined, {
|
||||
orchestrationEnvironmentTransport: {
|
||||
resolve: vi.fn(),
|
||||
call: vi.fn()
|
||||
}
|
||||
orchestrationEnvironmentTransport: { resolve: vi.fn(), call: vi.fn() }
|
||||
})
|
||||
const sync = vi
|
||||
.spyOn(runtime, 'syncOrchestrationFederatedDispatch')
|
||||
.mockRejectedValue(new Error('worker unavailable'))
|
||||
runtimes.push(runtime)
|
||||
const sync = vi.spyOn(runtime, 'syncOrchestrationFederatedDispatch').mockReturnValue(blocked)
|
||||
runtime.setOrchestrationDb({
|
||||
listActiveFederatedDispatches: () => [],
|
||||
findNextTerminalFederatedDispatchPendingAcknowledgment: (afterRowId: number) =>
|
||||
candidates.find((candidate) => candidate.rowId > afterRowId)
|
||||
} as never)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(3_000)
|
||||
|
||||
expect(sync.mock.calls).toEqual([
|
||||
['dispatch_1'],
|
||||
['dispatch_2'],
|
||||
['dispatch_3'],
|
||||
['dispatch_1']
|
||||
])
|
||||
runtime.stopOrchestrationFederationRelay()
|
||||
})
|
||||
|
||||
it('does not restart recovery after relay shutdown', async () => {
|
||||
vi.useFakeTimers()
|
||||
let release!: () => void
|
||||
const blocked = new Promise<void>((resolve) => (release = resolve))
|
||||
const runtime = new OrcaRuntimeService(null, undefined, {
|
||||
orchestrationEnvironmentTransport: {
|
||||
resolve: vi.fn(),
|
||||
call: vi.fn()
|
||||
}
|
||||
})
|
||||
const sync = vi.spyOn(runtime, 'syncOrchestrationFederatedDispatch').mockReturnValue(blocked)
|
||||
runtime.setOrchestrationDb({
|
||||
listActiveFederatedDispatches: () => [],
|
||||
findNextTerminalFederatedDispatchPendingAcknowledgment: () => ({
|
||||
dispatchId: 'dispatch_1',
|
||||
rowId: 1
|
||||
})
|
||||
} as never)
|
||||
expect(sync).toHaveBeenCalledTimes(1)
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
runtime.stopOrchestrationFederationRelay()
|
||||
release()
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
await vi.advanceTimersByTimeAsync(60_000)
|
||||
|
||||
expect(sync).toHaveBeenCalledTimes(1)
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -103,7 +103,7 @@ describe('mutation receipt capacity schema', () => {
|
||||
|
||||
db = new OrchestrationDb(dbPath)
|
||||
const sqlite = sqliteFor(db)
|
||||
expect(sqlite.pragma('user_version', { simple: true })).toBe(27)
|
||||
expect(sqlite.pragma('user_version', { simple: true })).toBe(28)
|
||||
expect(sqlite.prepare('SELECT receipt_count FROM mutation_receipt_ledger').get()).toEqual({
|
||||
receipt_count: 20
|
||||
})
|
||||
|
||||
@@ -240,7 +240,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => {
|
||||
|
||||
db = new OrchestrationDb(dbPath)
|
||||
const sqlite = sqliteFor(db)
|
||||
expect(sqlite.pragma('user_version', { simple: true })).toBe(27)
|
||||
expect(sqlite.pragma('user_version', { simple: true })).toBe(28)
|
||||
expect(db.getDispatchContextById(dispatch.id)).toMatchObject({ assignee_handle: 'term_worker' })
|
||||
expect(db.getTask(task.id)).toMatchObject({
|
||||
created_by_pane_key: null,
|
||||
@@ -277,7 +277,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => {
|
||||
|
||||
db.close()
|
||||
db = new OrchestrationDb(dbPath)
|
||||
expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(27)
|
||||
expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(28)
|
||||
expect(db.getDispatchContextById(dispatch.id)).toBeDefined()
|
||||
})
|
||||
|
||||
@@ -309,7 +309,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => {
|
||||
|
||||
db = new OrchestrationDb(dbPath)
|
||||
const sqlite = sqliteFor(db)
|
||||
expect(sqlite.pragma('user_version', { simple: true })).toBe(27)
|
||||
expect(sqlite.pragma('user_version', { simple: true })).toBe(28)
|
||||
expect(db.getTask(task.id)).toMatchObject({
|
||||
created_by_pane_key: 'tab_creator:leaf_creator',
|
||||
created_by_process_incarnation: 'pty_creator:incarnation-a',
|
||||
@@ -328,7 +328,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => {
|
||||
|
||||
db.close()
|
||||
db = new OrchestrationDb(dbPath)
|
||||
expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(27)
|
||||
expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(28)
|
||||
expect(db.getTask(task.id)?.created_by_process_incarnation).toBe('pty_creator:incarnation-a')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -26,7 +26,11 @@ const POST_V6_COLUMNS = [
|
||||
] as const
|
||||
|
||||
const VERSIONED_POST_V6_COLUMNS = [
|
||||
{ version: 27, table: 'federated_dispatches', column: 'to_home_acknowledged_sequence' }
|
||||
{ version: 27, table: 'federated_dispatches', column: 'to_home_acknowledged_sequence' },
|
||||
{ version: 28, table: 'federated_dispatches', column: 'terminal_ack_recovery_state' },
|
||||
{ version: 28, table: 'federated_dispatches', column: 'terminal_ack_recovery_attempts' },
|
||||
{ version: 28, table: 'federated_dispatches', column: 'terminal_ack_recovery_next_at_ms' },
|
||||
{ version: 28, table: 'federated_dispatches', column: 'terminal_ack_recovery_error_code' }
|
||||
] as const
|
||||
|
||||
const POST_V6_INDEXES = [
|
||||
|
||||
@@ -183,6 +183,10 @@ export type FederatedDispatchRow = {
|
||||
remote_terminal_handle: string | null
|
||||
to_home_imported_sequence: number
|
||||
to_home_acknowledged_sequence: number
|
||||
terminal_ack_recovery_state: 'pending' | 'retryable' | 'terminal'
|
||||
terminal_ack_recovery_attempts: number
|
||||
terminal_ack_recovery_next_at_ms: number
|
||||
terminal_ack_recovery_error_code: string | null
|
||||
created_at: string
|
||||
updated_at: string
|
||||
}
|
||||
|
||||
@@ -77,6 +77,7 @@ const STRUCTURED_RUNTIME_PASSTHROUGH_CODES: ReadonlySet<string> = new Set([
|
||||
'source_changed',
|
||||
'transcript_required',
|
||||
'server_required',
|
||||
'environment_not_found',
|
||||
'worktree_not_found_on_server',
|
||||
'resource_server_mismatch',
|
||||
'peer_changed',
|
||||
|
||||
@@ -32,6 +32,13 @@ export class RuntimeEnvironmentStoreError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
export class RuntimeEnvironmentNotFoundError extends RuntimeEnvironmentStoreError {
|
||||
constructor(selector: string) {
|
||||
super('invalid_argument', `Unknown environment: ${selector}`)
|
||||
this.name = 'RuntimeEnvironmentNotFoundError'
|
||||
}
|
||||
}
|
||||
|
||||
export function getEnvironmentStorePath(userDataPath: string): string {
|
||||
return join(userDataPath, ENVIRONMENTS_FILE)
|
||||
}
|
||||
@@ -223,7 +230,7 @@ function resolveEnvironmentFromStore(
|
||||
`Environment name "${selector}" is ambiguous; use the environment id.`
|
||||
)
|
||||
}
|
||||
throw new RuntimeEnvironmentStoreError('invalid_argument', `Unknown environment: ${selector}`)
|
||||
throw new RuntimeEnvironmentNotFoundError(selector)
|
||||
}
|
||||
|
||||
function readEnvironmentStore(userDataPath: string): RuntimeEnvironmentStore {
|
||||
|
||||
Reference in New Issue
Block a user