From c8a6aa3db2522756d57eaf24fd0b47de6c626f20 Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Tue, 18 Aug 2026 02:35:54 -0400 Subject: [PATCH] fix(orchestration): scope the push stamp release to the mailbox that staged it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Independent review found that clearing `delivered_at` on a mailbox ownership change made a duplicate pointer reachable. `markAsUndelivered` matched on message id alone, so an abandoned Dispatch pointer flight could release a stamp that the mail's *new* owner had since set: 1. a worker is pointed at `dispatch:D` mail, stamping it; 2. D goes inactive, the mail migrates to `run:R` and loses its stamp; 3. the idle coordinator is pointed at it and stamps it again; 4. the old Dispatch flight finally cleans up and clears that new stamp, letting the Run repoint announce the same message twice. The stamp is owner-scoped on the way in, so it is owner-scoped on the way out: a release now names the mailbox it staged. The flight records that mailbox so PTY retirement releases the same one. Also asserts the exact pointer text for a Dispatch mailbox — a bare `orca orchestration check` with no `--run` flag, which is what resolves the caller's own Dispatch — so a regression that pinned it to a Run would fail rather than pass on a substring match. message-batch-atomicity now builds each case's mutation explicitly rather than indexing a method-name union, which no longer types uniformly. --- ...rchestration-dispatch-mailbox-push.test.ts | 26 +++++++++++++++++ .../db/messages/message-inbox.ts | 16 ++++++++--- .../orchestration/mailbox-pointer-delivery.ts | 5 ++-- .../orchestration/mailbox-pointer-state.ts | 3 +- .../orchestration/mailbox-pointer-submit.ts | 5 +++- .../message-batch-atomicity.test.ts | 28 ++++++++++--------- 6 files changed, 62 insertions(+), 21 deletions(-) diff --git a/src/main/runtime/orchestration-dispatch-mailbox-push.test.ts b/src/main/runtime/orchestration-dispatch-mailbox-push.test.ts index 72c44170e68..f34eee0830c 100644 --- a/src/main/runtime/orchestration-dispatch-mailbox-push.test.ts +++ b/src/main/runtime/orchestration-dispatch-mailbox-push.test.ts @@ -71,6 +71,12 @@ describe('dispatch mailbox push-on-idle', () => { await vi.advanceTimersByTimeAsync(600) expect(pointerCount(harness.write)).toBe(1) + // A Dispatch pointer carries no --run flag: the bare command is what resolves + // the caller's own Dispatch mailbox, so pinning it to a Run would misdirect the worker. + expect(harness.write).toHaveBeenCalledWith( + PTY_ID, + '\nYou have 1 orchestration message. Run `orca orchestration check`.\n' + ) // The pointer must be backed by mail the worker's own check then returns. await expect( checkBoundMailbox(harness.runtime, { @@ -133,6 +139,26 @@ describe('dispatch mailbox push-on-idle', () => { db.close() }) + it('releases only the push stamp of the mailbox that staged it', () => { + const db = createDatabase('orca-dispatch-push-release-') + const { run, dispatch } = dispatchToWorker(db) + const message = sendToDispatch(db, run.id, dispatch.id, 'push authorization') + db.markAsDelivered([message.id]) + + // The mail moves to the Run mailbox and the coordinator is pointed at it. + db.routeUnreadDispatchMailboxToRunMailbox(dispatch.id, run.id) + db.markAsDelivered([message.id]) + // Only now does the worker's abandoned Dispatch pointer flight clean up. + db.markAsUndelivered([message.id], `dispatch:${dispatch.id}`) + + // Releasing the stale Dispatch stamp must not re-arm a push the coordinator already had. + expect(db.getMessageById(message.id)).toMatchObject({ + to_handle: `run:${run.id}`, + delivered_at: expect.any(String) + }) + db.close() + }) + it('clears the push stamp when unread mail changes mailbox owner', () => { const db = createDatabase('orca-dispatch-push-reroute-') const { run, dispatch } = dispatchToWorker(db) diff --git a/src/main/runtime/orchestration/db/messages/message-inbox.ts b/src/main/runtime/orchestration/db/messages/message-inbox.ts index af94b2e7c16..a2a08444983 100644 --- a/src/main/runtime/orchestration/db/messages/message-inbox.ts +++ b/src/main/runtime/orchestration/db/messages/message-inbox.ts @@ -9,7 +9,8 @@ const MESSAGE_MUTATION_SAVEPOINT = 'message_id_mutation' function runBatchedMessageMutation( db: OrchestrationDb, ids: string[], - sqlForPlaceholders: (placeholders: string) => string + sqlForPlaceholders: (placeholders: string) => string, + leadingParams: string[] = [] ): void { if (ids.length === 0) { return @@ -19,7 +20,7 @@ function runBatchedMessageMutation( for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) { const batch = ids.slice(offset, offset + MESSAGE_ID_UPDATE_BATCH_SIZE) const placeholders = batch.map(() => '?').join(',') - db.db.prepare(sqlForPlaceholders(placeholders)).run(...batch) + db.db.prepare(sqlForPlaceholders(placeholders)).run(...leadingParams, ...batch) } db.db.exec(`RELEASE ${MESSAGE_MUTATION_SAVEPOINT}`) } catch (error) { @@ -168,13 +169,20 @@ export function markAsDelivered(this: OrchestrationDb, ids: string[]): void { ) } -export function markAsUndelivered(this: OrchestrationDb, ids: string[]): void { +// Why: the push stamp is owner-scoped, so an abandoned flight may only release the +// mailbox it staged — the mail may since have moved on and been pointed at by its new owner. +export function markAsUndelivered( + this: OrchestrationDb, + ids: string[], + mailboxHandle: string +): void { runBatchedMessageMutation( this, ids, (placeholders) => `UPDATE messages SET delivered_at = NULL - WHERE read = 0 AND id IN (${placeholders})` + WHERE read = 0 AND to_handle = ? AND id IN (${placeholders})`, + [mailboxHandle] ) } diff --git a/src/main/runtime/orchestration/mailbox-pointer-delivery.ts b/src/main/runtime/orchestration/mailbox-pointer-delivery.ts index e9f2d7e92f8..be5f551354e 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-delivery.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-delivery.ts @@ -142,8 +142,8 @@ export class OrchestrationMailboxPointerDelivery message.id) + flight.stagedMailboxHandle = mailboxHandle db.markAsDelivered(flight.stagedMessageIds) this.state.setWatermark(mailboxHandle, newestSequence, ptyId, this.leafKey(leaf)) if ( diff --git a/src/main/runtime/orchestration/mailbox-pointer-state.ts b/src/main/runtime/orchestration/mailbox-pointer-state.ts index b0f4330b3f8..b40d86af65a 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-state.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-state.ts @@ -3,6 +3,7 @@ import type { OrchestrationMailboxLeaf } from './mailbox-owner' export type OrchestrationMailboxDeliveryFlight = { enterTimer: ReturnType | null stagedMessageIds: string[] + stagedMailboxHandle: string | null } export type ParkedOrchestrationMailboxDelivery = { @@ -28,7 +29,7 @@ export class OrchestrationMailboxPointerState { } beginFlight(ptyId: string): OrchestrationMailboxDeliveryFlight { - const flight = { enterTimer: null, stagedMessageIds: [] } + const flight = { enterTimer: null, stagedMessageIds: [], stagedMailboxHandle: null } this.flightsByPtyId.set(ptyId, flight) return flight } diff --git a/src/main/runtime/orchestration/mailbox-pointer-submit.ts b/src/main/runtime/orchestration/mailbox-pointer-submit.ts index 5d16ea6ba05..a67bb0e58cd 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-submit.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-submit.ts @@ -76,7 +76,10 @@ export function submitOrchestrationMailboxPointer message.id)) + deps.getDb()?.markAsUndelivered( + input.messages.map((message) => message.id), + input.mailboxHandle + ) } released = submitted || clearAndRedrive || releaseWithoutRedrive diff --git a/src/main/runtime/orchestration/message-batch-atomicity.test.ts b/src/main/runtime/orchestration/message-batch-atomicity.test.ts index 3c35465a3b4..3b61e73e210 100644 --- a/src/main/runtime/orchestration/message-batch-atomicity.test.ts +++ b/src/main/runtime/orchestration/message-batch-atomicity.test.ts @@ -2,12 +2,6 @@ import { afterEach, describe, expect, it } from 'vitest' import type Database from '../../sqlite/sync-database' import { OrchestrationDb } from './db' -type MessageMutation = - | 'markAsRead' - | 'markAsDelivered' - | 'markAsUndelivered' - | 'markAsReadAndDelivered' - const messageIds = Array.from( { length: 501 }, (_, index) => `m${index.toString().padStart(3, '0')}` @@ -40,26 +34,34 @@ describe('message batch atomicity', () => { afterEach(() => db?.close()) it.each<{ - method: MessageMutation + method: string setup?: string changedCountSql: string + mutate: (db: OrchestrationDb) => void }>([ - { method: 'markAsRead', changedCountSql: 'SELECT COUNT(*) count FROM messages WHERE read = 1' }, + { + method: 'markAsRead', + changedCountSql: 'SELECT COUNT(*) count FROM messages WHERE read = 1', + mutate: (db) => db.markAsRead(messageIds) + }, { method: 'markAsDelivered', - changedCountSql: 'SELECT COUNT(*) count FROM messages WHERE delivered_at IS NOT NULL' + changedCountSql: 'SELECT COUNT(*) count FROM messages WHERE delivered_at IS NOT NULL', + mutate: (db) => db.markAsDelivered(messageIds) }, { method: 'markAsUndelivered', setup: "UPDATE messages SET delivered_at = datetime('now')", - changedCountSql: 'SELECT COUNT(*) count FROM messages WHERE delivered_at IS NULL' + changedCountSql: 'SELECT COUNT(*) count FROM messages WHERE delivered_at IS NULL', + mutate: (db) => db.markAsUndelivered(messageIds, 'recipient') }, { method: 'markAsReadAndDelivered', changedCountSql: - 'SELECT COUNT(*) count FROM messages WHERE read = 1 OR delivered_at IS NOT NULL' + 'SELECT COUNT(*) count FROM messages WHERE read = 1 OR delivered_at IS NOT NULL', + mutate: (db) => db.markAsReadAndDelivered(messageIds) } - ])('rolls back $method when a later batch fails', ({ method, setup, changedCountSql }) => { + ])('rolls back $method when a later batch fails', ({ setup, changedCountSql, mutate }) => { db = new OrchestrationDb(':memory:') const sqlite = (db as unknown as { db: Database.Database }).db seedMessages(sqlite) @@ -68,7 +70,7 @@ describe('message batch atomicity', () => { } rejectLastMessageUpdate(sqlite) - expect(() => db?.[method](messageIds)).toThrow('blocked') + expect(() => mutate(db!)).toThrow('blocked') const changed = sqlite.prepare(changedCountSql).get() as { count: number } expect(changed.count).toBe(0)