fix(orchestration): scope the push stamp release to the mailbox that staged it

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.
This commit is contained in:
Jinwoo-H
2026-08-18 02:35:54 -04:00
parent 5e53ea6e71
commit c8a6aa3db2
6 changed files with 62 additions and 21 deletions
@@ -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)
@@ -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]
)
}
@@ -142,8 +142,8 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
if (flight?.enterTimer != null) {
clearTimeout(flight.enterTimer)
}
if (flight?.stagedMessageIds.length) {
this.deps.getDb()?.markAsUndelivered(flight.stagedMessageIds)
if (flight?.stagedMessageIds.length && flight.stagedMailboxHandle) {
this.deps.getDb()?.markAsUndelivered(flight.stagedMessageIds, flight.stagedMailboxHandle)
}
for (const mailboxHandle of releasedMailboxes) {
this.redrive(mailboxHandle, true)
@@ -237,6 +237,7 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
return
}
flight.stagedMessageIds = unread.map((message) => message.id)
flight.stagedMailboxHandle = mailboxHandle
db.markAsDelivered(flight.stagedMessageIds)
this.state.setWatermark(mailboxHandle, newestSequence, ptyId, this.leafKey(leaf))
if (
@@ -3,6 +3,7 @@ import type { OrchestrationMailboxLeaf } from './mailbox-owner'
export type OrchestrationMailboxDeliveryFlight = {
enterTimer: ReturnType<typeof setTimeout> | 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
}
@@ -76,7 +76,10 @@ export function submitOrchestrationMailboxPointer<TWaiter extends OrchestrationM
let released = false
if (finalizeReservation) {
if (clearAndRedrive) {
deps.getDb()?.markAsUndelivered(input.messages.map((message) => message.id))
deps.getDb()?.markAsUndelivered(
input.messages.map((message) => message.id),
input.mailboxHandle
)
}
released =
submitted || clearAndRedrive || releaseWithoutRedrive
@@ -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)