From 962c4088246705bf87d257cd441dc953eaa046c0 Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Mon, 7 Sep 2026 04:40:42 -0400 Subject: [PATCH] fix(test): give federation tests a real read-after-write sync barrier `syncOrchestrationFederation()` coalesces onto an already-in-flight relay-tick sync, which may have pulled from the peer before the caller's mutation existed. Tests used it as a barrier, so `keeps a timed-out remote question resumable` could reply against a home DB that had never imported the worker's question: the reply failed with `Message not found`, no `to_worker` relay was enqueued, and the resume ask surfaced it 5s later as a spurious timeout. Add `syncFederationBarrier()`, which chains each active dispatch past the current round via `syncOrchestrationFederatedDispatchAfterCurrent`, and use it at every barrier-purpose sync site. The two tests whose subject is the sync machinery itself keep the raw call. Also assert the reply response, so a failed reply fails at the reply instead of masquerading as a timeout. Production is unaffected: `syncOrchestrationFederation` has no production callers, real read-after-write paths already use the after-current sync, and relay ticks retry every second. --- .../federation-control-mail.test.ts | 5 +++-- .../federation-sync-barrier.test-support.ts | 17 +++++++++++++++ .../federation/federation.test.ts | 21 +++++++++++-------- 3 files changed, 32 insertions(+), 11 deletions(-) create mode 100644 src/main/runtime/rpc/methods/orchestration/federation/federation-sync-barrier.test-support.ts diff --git a/src/main/runtime/rpc/methods/orchestration/federation/federation-control-mail.test.ts b/src/main/runtime/rpc/methods/orchestration/federation/federation-control-mail.test.ts index b351582d14b..755f85fd512 100644 --- a/src/main/runtime/rpc/methods/orchestration/federation/federation-control-mail.test.ts +++ b/src/main/runtime/rpc/methods/orchestration/federation/federation-control-mail.test.ts @@ -8,6 +8,7 @@ import type { RpcRequest } from '../../../core' import { RpcDispatcher } from '../../../dispatcher' import { fingerprintAuthenticatedPairingCredential } from '../../../orchestration-mutation-executor' import { ORCHESTRATION_METHODS } from '../../orchestration' +import { syncFederationBarrier } from './federation-sync-barrier.test-support' describe('orchestration federation control mail', () => { const homeToken = 'run-home-device-token' @@ -160,7 +161,7 @@ describe('orchestration federation control mail', () => { }) expect(homeDb.listPendingFederationRelay(dispatchId, 'to_worker')).toHaveLength(1) - await homeRuntime.syncOrchestrationFederation() + await syncFederationBarrier(homeRuntime, homeDb) const checked = await workerDispatcher.dispatch(checkRequest('check-imported')) expect(checked).toMatchObject({ @@ -252,7 +253,7 @@ describe('orchestration federation control mail', () => { settleRemoteOutcome: 'succeeded' }) - await homeRuntime.syncOrchestrationFederation() + await syncFederationBarrier(homeRuntime, homeDb) expect(homeDb.getWorkerDispatch(dispatchId)?.state).toBe('succeeded') expect(workerDb.getUnreadMessages(`dispatch:${dispatchId}`)).toHaveLength(0) diff --git a/src/main/runtime/rpc/methods/orchestration/federation/federation-sync-barrier.test-support.ts b/src/main/runtime/rpc/methods/orchestration/federation/federation-sync-barrier.test-support.ts new file mode 100644 index 00000000000..675c55068ce --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/federation/federation-sync-barrier.test-support.ts @@ -0,0 +1,17 @@ +import type { OrcaRuntimeService } from '../../../../orca-runtime' +import type { OrchestrationDb } from '../../../../orchestration/db' + +// A run-wide sync coalesces onto whatever relay tick is already in flight, and that tick may have +// read the peer before the test's latest mutation existed. Chain past the current round instead so +// awaiting the barrier really means "everything enqueued before this call has been exchanged". +export async function syncFederationBarrier( + runtime: OrcaRuntimeService, + db: OrchestrationDb +): Promise { + const dispatches = db.listActiveFederatedDispatches() + await Promise.allSettled( + dispatches.map((dispatch) => + runtime.syncOrchestrationFederatedDispatchAfterCurrent(dispatch.dispatch_id) + ) + ) +} diff --git a/src/main/runtime/rpc/methods/orchestration/federation/federation.test.ts b/src/main/runtime/rpc/methods/orchestration/federation/federation.test.ts index 8146c43ed29..e627e112530 100644 --- a/src/main/runtime/rpc/methods/orchestration/federation/federation.test.ts +++ b/src/main/runtime/rpc/methods/orchestration/federation/federation.test.ts @@ -11,6 +11,7 @@ import { RpcDispatcher } from '../../../dispatcher' import { ORCHESTRATION_METHODS } from '../../orchestration' import { createFederationWorkerStartRequest as startRequest } from './federation-request.test-support' import { configureFederationWorkerRuntime } from './federation-runtime.test-support' +import { syncFederationBarrier } from './federation-sync-barrier.test-support' describe('orchestration federation', () => { const databases: OrchestrationDb[] = [] @@ -310,7 +311,7 @@ describe('orchestration federation', () => { expect(sent).toMatchObject({ ok: true, result: { lifecycle: { action: 'completed' } } }) expect(homeDb.getTask(task.id)?.status).toBe('completed') - await homeRuntime.syncOrchestrationFederation() + await syncFederationBarrier(homeRuntime, homeDb) expect(homeDb.getTask(task.id)?.status).toBe('completed') expect(homeDb.getWorkerDispatch(dispatch.id)?.state).toBe('succeeded') @@ -362,7 +363,7 @@ describe('orchestration federation', () => { ).toHaveLength(1) ) - await homeRuntime.syncOrchestrationFederation() + await syncFederationBarrier(homeRuntime, homeDb) const question = homeDb .getRunMailboxHistory(task.run_id, 10) .find((message) => message.type === 'question') @@ -383,7 +384,7 @@ describe('orchestration federation', () => { } }) expect(reply).toMatchObject({ ok: true, result: { question: { status: 'answered' } } }) - await homeRuntime.syncOrchestrationFederation() + await syncFederationBarrier(homeRuntime, homeDb) await expect(ask).resolves.toMatchObject({ ok: true, @@ -426,8 +427,8 @@ describe('orchestration federation', () => { }) const questionId = (timedOut as { result: { messageId: string } }).result.messageId - await homeRuntime.syncOrchestrationFederation() - await homeDispatcher.dispatch({ + await syncFederationBarrier(homeRuntime, homeDb) + const lateReply = await homeDispatcher.dispatch({ id: 'rpc_home_late_reply', authToken: 'coordinator-token', orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, @@ -435,6 +436,8 @@ describe('orchestration federation', () => { method: 'orchestration.reply', params: { id: questionId, body: 'yes', from: 'term_coord' } }) + // A rejected reply enqueues no relay, which would only surface as the resume timing out. + expect(lateReply).toMatchObject({ ok: true, result: { question: { status: 'answered' } } }) restartWorkerRuntime() const resumed = workerDispatcher.dispatch({ id: 'rpc_remote_ask_resume', @@ -445,7 +448,7 @@ describe('orchestration federation', () => { method: 'orchestration.ask', params: { from: 'term_windows_worker', resume: questionId, timeoutMs: 5_000 } }) - await homeRuntime.syncOrchestrationFederation() + await syncFederationBarrier(homeRuntime, homeDb) await expect(resumed).resolves.toMatchObject({ ok: true, @@ -476,8 +479,8 @@ describe('orchestration federation', () => { loseNextAckResponse = true const remoteCall = vi.spyOn(homeRuntime, 'callOrchestrationWorkerServer') - await expect(homeRuntime.syncOrchestrationFederation()).resolves.toBeUndefined() - await homeRuntime.syncOrchestrationFederation() + await expect(syncFederationBarrier(homeRuntime, homeDb)).resolves.toBeUndefined() + await syncFederationBarrier(homeRuntime, homeDb) expect( homeDb @@ -612,7 +615,7 @@ describe('orchestration federation', () => { it('treats a worker runtime ID change as an epoch, not a new server', async () => { const task = createHomeTask() await homeDispatcher.dispatch(startRequest(task.id)) - await homeRuntime.syncOrchestrationFederation() + await syncFederationBarrier(homeRuntime, homeDb) vi.spyOn(homeRuntime, 'ensureOrchestrationFederationRelay').mockImplementation(() => {}) const dispatch = homeDb.getDispatchContext(task.id)! const oldEpoch = homeDb.getFederatedDispatch(dispatch.id)?.remote_runtime_epoch