From a6e6de93c4d6d7d549068780bca8be789ef0f0a7 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Thu, 10 Sep 2026 16:04:25 -0400 Subject: [PATCH 1/2] fix(relay): keep failed rehome polls out of the durable failure budget (#19915) * fix(relay): keep failed rehome polls out of the durable failure budget The regional rehome worker polls claimRegionalRehome about once a second. Any error thrown before an attempt was claimed - in practice a director pool timeout on the pre-claim control read, 52-74 a day against a pool of 3 - was charged to relay_region_rehome_worker_state.consecutive_failures, which durably disables the control at three. That counter only ever resets on a drain receipt, so while the control is disabled it never resets: production sits at 1068 and still climbing. Enabling the control leaves the stale counter in place, so the next pool timeout latches it straight back off. That is what ended the 2026-08-28 enable after ten minutes. - A poll that never claimed an attempt drained nothing, so it no longer feeds the dispatch-failure budget and logs .._poll_failed instead of .._dispatch_failed. recordRegionalRehomeWorkerFailure had no other caller and is removed. - Enabling the control clears consecutive_failures and paused_until, so a budget spent under a previous enable cannot kill a fresh one. The dispatch interval in next_dispatch_at is deliberately left alone. - The budget's auto-disable now emits orca_relay_regional_rehome_failure_budget_disabled, matching the existing .._safety_disabled precedent. It wrote no event before, which is why this went unnoticed for two weeks. No change to region selection, the candidate query, or host eligibility. * fix(relay): serialize rehome failure accounting with control updates --- cloud/apps/relay/src/assignment-store.ts | 78 ++++++++++------- .../src/regional-rehome-postgres.test.ts | 87 ++++++++++++++++++- .../relay/src/regional-rehome-store.test.ts | 86 ++++++++++++++++++ .../relay/src/regional-rehome-worker.test.ts | 38 ++++++-- .../apps/relay/src/regional-rehome-worker.ts | 16 ++-- 5 files changed, 262 insertions(+), 43 deletions(-) diff --git a/cloud/apps/relay/src/assignment-store.ts b/cloud/apps/relay/src/assignment-store.ts index 9ead45df22e..824cb1e0b2f 100644 --- a/cloud/apps/relay/src/assignment-store.ts +++ b/cloud/apps/relay/src/assignment-store.ts @@ -362,6 +362,9 @@ export const REGIONAL_REHOME_QUARANTINE_FAILURES = 3 export const REGIONAL_REHOME_QUARANTINE_MS = 15 * 60_000 const REGIONAL_REHOME_QUARANTINE_EXCLUSION_LIMIT = 50 const REGIONAL_REHOME_QUARANTINE_MEMORY_LIMIT = 1_000 +// Consecutive drain-dispatch failures that latch the durable control off. Any +// drain receipt and any enable reset it, so it reads "dispatch is broken now". +const REGIONAL_REHOME_FAILURE_BUDGET = 3 const REGIONAL_REHOME_OBSERVATION_MS = 24 * 60 * 60_000 const ASSIGNMENT_LOCK_RETRY_MAX_DELAY_MS = 50 type AssignmentInventoryScope = 'none' | 'general' | 'all' @@ -4992,6 +4995,20 @@ export class RelayAssignmentStore { now ] ) + if (input.enabled) { + // A budget spent under a previous enable is not evidence about this one. + // Without this an old counter latches the fresh enable straight back off + // on its first transient failure. + await transaction.query( + `INSERT INTO relay_region_rehome_worker_state + (worker_id, next_dispatch_at, paused_until, consecutive_failures, updated_at) + VALUES ('global', 0, 0, 0, ?) + ON CONFLICT (worker_id) DO UPDATE + SET paused_until = 0, consecutive_failures = 0, + updated_at = excluded.updated_at`, + [now] + ) + } const updated = ( await transaction.query( `SELECT * FROM relay_region_rehome_control WHERE control_id = 'global'` @@ -5940,7 +5957,11 @@ export class RelayAssignmentStore { async recordRegionalRehomeDispatchFailure(attemptId: string): Promise { const now = this.now() - await this.database.transaction(async (transaction) => { + const disableLog = await this.database.transaction(async (transaction) => { + // Match claim and enable ordering before a spent budget updates the control. + await transaction.queryLocked( + `SELECT * FROM relay_region_rehome_control WHERE control_id = 'global'` + ) const worker = ( await transaction.queryLocked( `SELECT * FROM relay_region_rehome_worker_state WHERE worker_id = 'global'` @@ -5952,49 +5973,44 @@ export class RelayAssignmentStore { [attemptId] ) )[0] - if (!worker || !attempt) return - await this.incrementRegionalRehomeWorkerFailure(transaction, worker, now) - }) - } - - async recordRegionalRehomeWorkerFailure(): Promise { - const now = this.now() - await this.database.transaction(async (transaction) => { - await transaction.query( - `INSERT INTO relay_region_rehome_worker_state - (worker_id, next_dispatch_at, paused_until, consecutive_failures, updated_at) - VALUES ('global', 0, 0, 0, ?) - ON CONFLICT (worker_id) DO NOTHING`, - [now] - ) - const worker = ( - await transaction.queryLocked( - `SELECT * FROM relay_region_rehome_worker_state WHERE worker_id = 'global'` - ) - )[0]! - await this.incrementRegionalRehomeWorkerFailure(transaction, worker, now) + if (!worker || !attempt) return null + return await this.incrementRegionalRehomeWorkerFailure(transaction, worker, now) }) + // Logged after the commit so a rollback cannot fabricate the record. + if (disableLog) console.warn(JSON.stringify(disableLog)) } + // Returns the durable disable this failure caused, for the caller to log once + // its transaction commits; null when the budget survives or was already spent. private async incrementRegionalRehomeWorkerFailure( transaction: RelayDatabase, worker: SqlRow, now: number - ): Promise { + ): Promise | null> { const failures = integer(worker, 'consecutive_failures') + 1 + const spent = failures >= REGIONAL_REHOME_FAILURE_BUDGET await transaction.query( `UPDATE relay_region_rehome_worker_state SET consecutive_failures = ?, paused_until = ?, updated_at = ? WHERE worker_id = 'global'`, - [failures, failures >= 3 ? now + 5 * 60_000 : 0, now] + [failures, spent ? now + 5 * 60_000 : 0, now] ) - if (failures >= 3) { - await transaction.query( - `UPDATE relay_region_rehome_control - SET generation = generation + 1, enabled = 0, updated_at = ? - WHERE control_id = 'global' AND enabled = 1`, - [now] - ) + if (!spent) return null + const disabled = await transaction.query( + `UPDATE relay_region_rehome_control + SET generation = generation + 1, enabled = 0, updated_at = ? + WHERE control_id = 'global' AND enabled = 1 + RETURNING generation`, + [now] + ) + // The disable is otherwise invisible: inspection only shows enabled=false and + // nothing records that the failure budget, not an operator, turned it off. + if (disabled.length === 0) return null + return { + event: 'orca_relay_regional_rehome_failure_budget_disabled', + controlGeneration: integer(disabled[0]!, 'generation'), + consecutiveFailures: failures, + now } } diff --git a/cloud/apps/relay/src/regional-rehome-postgres.test.ts b/cloud/apps/relay/src/regional-rehome-postgres.test.ts index 44f3b3434af..fdefda54401 100644 --- a/cloud/apps/relay/src/regional-rehome-postgres.test.ts +++ b/cloud/apps/relay/src/regional-rehome-postgres.test.ts @@ -1,4 +1,4 @@ -import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest' +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' import { RelayAssignmentStore } from './assignment-store.js' import { openRelayDatabase, type RelayDatabase } from './database.js' import { @@ -267,6 +267,91 @@ describePostgres('PostgreSQL regional rehoming', () => { )).toEqual([{ count: '1' }]) }) + it('serializes an enable with a budget-exhausting failure without retries', async () => { + const context = await fixture() + const attempt = await context.store.claimRegionalRehome() + await context.store.recordRegionalRehomeDispatchFailure(attempt!.attemptId) + await context.store.recordRegionalRehomeDispatchFailure(attempt!.attemptId) + const locked = Promise.withResolvers() + const release = Promise.withResolvers() + const primaryTransaction = primary.transaction.bind(primary) + const secondaryTransaction = secondary.transaction.bind(secondary) + let enableTransactions = 0 + let failureTransactions = 0 + let enablePid = 0 + let failurePid = 0 + const enableSpy = vi.spyOn(primary, 'transaction').mockImplementation((operation, options) => + primaryTransaction(async (transaction) => { + enableTransactions++ + enablePid = Number((await transaction.query('SELECT pg_backend_pid() AS pid'))[0]!.pid) + return await operation({ + dialect: 'postgres', + query: transaction.query.bind(transaction), + queryLocked: async (sql, params, lockOptions) => { + const rows = await transaction.queryLocked(sql, params, lockOptions) + if (sql.includes('FROM relay_region_rehome_control')) { + locked.resolve() + await release.promise + } + return rows + }, + transaction: transaction.transaction.bind(transaction), + close: transaction.close.bind(transaction) + }) + }, options) + ) + const failureSpy = vi.spyOn(secondary, 'transaction').mockImplementation((operation, options) => + secondaryTransaction(async (transaction) => { + failureTransactions++ + failurePid = Number((await transaction.query('SELECT pg_backend_pid() AS pid'))[0]!.pid) + return await operation(transaction) + }, options) + ) + const enable = context.store.applyRegionalRehomeControl({ + expectedGeneration: 1, + enabled: true, + notBefore: context.now(), + ratePerMinute: 10, + preferenceMaxAgeMs: 24 * 60 * 60_000, + hostCooldownMs: 7 * 24 * 60 * 60_000, + drainGraceMs: 60_000 + }) + let failure: Promise | undefined + let outcomes: PromiseSettledResult[] = [] + try { + await Promise.race([ + locked.promise, + enable.then(() => { + throw new Error('enable completed before the control lock') + }) + ]) + failure = context.competingStore.recordRegionalRehomeDispatchFailure(attempt!.attemptId) + // Observe the actual PostgreSQL wait before letting enable acquire the worker row. + await vi.waitFor(async () => { + expect(failurePid).not.toBe(0) + const rows = await primary.query('SELECT pg_blocking_pids(?) AS blockers', [failurePid]) + expect(rows[0]!.blockers).toContain(enablePid) + }, { interval: 10, timeout: 800 }) + } finally { + release.resolve() + outcomes = await Promise.allSettled([enable, ...(failure ? [failure] : [])]) + enableSpy.mockRestore() + failureSpy.mockRestore() + } + expect(outcomes.map((outcome) => outcome.status)).toEqual(['fulfilled', 'fulfilled']) + expect({ enableTransactions, failureTransactions }).toEqual({ + enableTransactions: 1, + failureTransactions: 1 + }) + expect(await context.store.inspectRegionalRehomeControl()).toMatchObject({ + generation: 2, + enabled: true + }) + expect(await primary.query( + `SELECT consecutive_failures, paused_until FROM relay_region_rehome_worker_state` + )).toEqual([{ consecutive_failures: '1', paused_until: '0' }]) + }) + it('increments the disable generation once across competing directors', async () => { const context = await fixture() const disabled = await Promise.all([ diff --git a/cloud/apps/relay/src/regional-rehome-store.test.ts b/cloud/apps/relay/src/regional-rehome-store.test.ts index 2c1c8132266..662876ef66e 100644 --- a/cloud/apps/relay/src/regional-rehome-store.test.ts +++ b/cloud/apps/relay/src/regional-rehome-store.test.ts @@ -1709,6 +1709,77 @@ describe('regional rehome assignment state', () => { expect(await context.store.claimRegionalRehome()).toBeNull() await context.database.close() }) + + it('clears a stale failure budget when the control is enabled again', async () => { + const context = await setup() + await activatePreferredSource(context, { + userId: 'user-1', + relayHostId: 'abcdefghijklmnop' + }) + const attempt = await context.store.claimRegionalRehome() + for (let index = 0; index < 3; index++) { + await context.store.recordRegionalRehomeDispatchFailure(attempt!.attemptId) + } + expect(await workerState(context)).toMatchObject({ consecutiveFailures: 3 }) + const latched = await context.store.inspectRegionalRehomeControl() + expect(latched).toMatchObject({ generation: 2, enabled: false }) + + await context.store.applyRegionalRehomeControl({ + expectedGeneration: latched.generation, + enabled: true, + notBefore: context.now(), + ratePerMinute: 10, + preferenceMaxAgeMs: 24 * 60 * 60_000, + hostCooldownMs: 7 * 24 * 60 * 60_000, + drainGraceMs: 60 * 60_000 + }) + + // A budget spent under the previous enable is not evidence about this one. + expect(await workerState(context)).toMatchObject({ + consecutiveFailures: 0, + pausedUntil: 0 + }) + // One transient failure must not latch the fresh enable straight back off. + await context.store.recordRegionalRehomeDispatchFailure(attempt!.attemptId) + expect(await context.store.inspectRegionalRehomeControl()).toMatchObject({ + generation: 3, + enabled: true + }) + await context.database.close() + }) + + it('reports the durable disable when the failure budget latches the control off', async () => { + const context = await setup() + await activatePreferredSource(context, { + userId: 'user-1', + relayHostId: 'abcdefghijklmnop' + }) + const attempt = await context.store.claimRegionalRehome() + const warnings = collectEventWarnings( + 'orca_relay_regional_rehome_failure_budget_disabled' + ) + try { + for (let index = 0; index < 5; index++) { + await context.store.recordRegionalRehomeDispatchFailure(attempt!.attemptId) + } + } finally { + warnings.restore() + } + + // Only the transition is reported; later failures find the control already off. + expect(warnings.entries).toEqual([ + expect.objectContaining({ + event: 'orca_relay_regional_rehome_failure_budget_disabled', + controlGeneration: 2, + consecutiveFailures: 3 + }) + ]) + expect(await context.store.inspectRegionalRehomeControl()).toMatchObject({ + generation: 2, + enabled: false + }) + await context.database.close() + }) }) class TransactionCountingDatabase implements RelayDatabase { @@ -2066,3 +2137,18 @@ class CellInventoryLockProbe { return decorate(database) } } + +async function workerState( + context: Context +): Promise<{ consecutiveFailures: number; pausedUntil: number }> { + const row = ( + await context.database.query( + `SELECT consecutive_failures, paused_until + FROM relay_region_rehome_worker_state WHERE worker_id = 'global'` + ) + )[0]! + return { + consecutiveFailures: Number(row.consecutive_failures), + pausedUntil: Number(row.paused_until) + } +} diff --git a/cloud/apps/relay/src/regional-rehome-worker.test.ts b/cloud/apps/relay/src/regional-rehome-worker.test.ts index af905bb9ab2..33e7f01f737 100644 --- a/cloud/apps/relay/src/regional-rehome-worker.test.ts +++ b/cloud/apps/relay/src/regional-rehome-worker.test.ts @@ -30,11 +30,9 @@ describe('regional rehome worker', () => { } const claimRegionalRehome = vi.fn().mockResolvedValueOnce(null).mockResolvedValue(attempt) const recordRegionalRehomeDrainReceipt = vi.fn().mockResolvedValue(true) - const recordRegionalRehomeWorkerFailure = vi.fn().mockResolvedValue(undefined) const assignments = { claimRegionalRehome, - recordRegionalRehomeDrainReceipt, - recordRegionalRehomeWorkerFailure + recordRegionalRehomeDrainReceipt } as unknown as RelayAssignmentStore const requests: Array<{ url: string; init?: RequestInit }> = [] const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) @@ -99,8 +97,7 @@ describe('regional rehome worker', () => { const claimRegionalRehome = vi.fn().mockResolvedValueOnce(null).mockResolvedValue(attempt) const assignments = { claimRegionalRehome, - recordRegionalRehomeDispatchFailure: vi.fn().mockResolvedValue(undefined), - recordRegionalRehomeWorkerFailure: vi.fn().mockResolvedValue(undefined) + recordRegionalRehomeDispatchFailure: vi.fn().mockResolvedValue(undefined) } as unknown as RelayAssignmentStore const worker = startRegionalRehomeWorker(config(), assignments, { now: () => now, @@ -118,7 +115,36 @@ describe('regional rehome worker', () => { expect(assignments.recordRegionalRehomeDispatchFailure).toHaveBeenCalledWith( '11111111-1111-4111-8111-111111111111' ) - expect(assignments.recordRegionalRehomeWorkerFailure).not.toHaveBeenCalled() + }) + + it('keeps a failed poll out of the durable dispatch-failure budget', async () => { + let now = 0 + const claimRegionalRehome = vi + .fn() + .mockResolvedValueOnce(null) + .mockRejectedValue(new Error('Connection terminated due to connection timeout')) + const recordRegionalRehomeDispatchFailure = vi.fn().mockResolvedValue(undefined) + const assignments = { + claimRegionalRehome, + recordRegionalRehomeDispatchFailure + } as unknown as RelayAssignmentStore + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const worker = startRegionalRehomeWorker(config(), assignments, { + now: () => now, + safetySnapshot: () => safety(now), + intervalMs: 60_000 + })! + await settleWorker() + now = 1_000 + await expect(worker.run()).resolves.toBeUndefined() + worker.stop() + + // The poll never claimed an attempt, so nothing was drained and nothing may + // be charged to the budget that latches the durable control off. + expect(recordRegionalRehomeDispatchFailure).not.toHaveBeenCalled() + expect(warn.mock.calls.map((call) => JSON.parse(String(call[0])).event)).toEqual([ + 'orca_relay_regional_rehome_poll_failed' + ]) }) it('passes unsafe process telemetry to the durable claim gate', async () => { diff --git a/cloud/apps/relay/src/regional-rehome-worker.ts b/cloud/apps/relay/src/regional-rehome-worker.ts index 47a2748cff4..4d8fa694afd 100644 --- a/cloud/apps/relay/src/regional-rehome-worker.ts +++ b/cloud/apps/relay/src/regional-rehome-worker.ts @@ -97,13 +97,19 @@ export function startRegionalRehomeWorker( }) ) } catch (error) { - await (attemptId - ? assignments.recordRegionalRehomeDispatchFailure(attemptId) - : assignments.recordRegionalRehomeWorkerFailure() - ).catch(() => undefined) + // Only a claimed attempt was drained. A poll that failed before the claim + // - a pool timeout on the once-a-second control read - dispatched nothing, + // so it must not spend the budget that latches the durable control off. + if (attemptId) { + await assignments + .recordRegionalRehomeDispatchFailure(attemptId) + .catch(() => undefined) + } console.warn( JSON.stringify({ - event: 'orca_relay_regional_rehome_dispatch_failed', + event: attemptId + ? 'orca_relay_regional_rehome_dispatch_failed' + : 'orca_relay_regional_rehome_poll_failed', reason: error instanceof Error ? error.message : 'unknown' }) ) From 6e9de5fa58d2bacbb4d35bc754fbd4cdb8744c26 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 10 Sep 2026 13:19:20 -0700 Subject: [PATCH 2/2] fix(orchestration): revalidate an attempted Enter instead of resending it (#19911) When a PTY retires mid-delivery, every staged message was marked undelivered, which made all of them redeliverable. That is right for a pointer whose Enter never fired, but an Enter that was already written may have landed: redelivering it types the same mail into the pane a second time. The Enter timer is cleared at the top of retirement, so a RESERVED or WRITE_ATTEMPTED pointer provably never submitted and is released. An ENTER_ATTEMPTED pointer is ambiguous and now stays at its phase for the resume path to revalidate, matching the policy mailbox-pointer-submit.ts already documents for an unverifiable settlement. Co-authored-by: Merge Sim --- .../orchestration/mailbox-pointer-delivery.ts | 18 ++++- .../mailbox-pointer-stage.test.ts | 69 +++++++++++++++++++ .../orchestration/mailbox-pointer-stage.ts | 1 + .../orchestration/mailbox-pointer-state.ts | 2 + .../terminal-send-stale-leaf-liveness.test.ts | 7 +- 5 files changed, 93 insertions(+), 4 deletions(-) diff --git a/src/main/runtime/orchestration/mailbox-pointer-delivery.ts b/src/main/runtime/orchestration/mailbox-pointer-delivery.ts index 11fbc08229d..b16e5c474b3 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-delivery.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-delivery.ts @@ -9,6 +9,10 @@ import { OrchestrationMailboxPointerState, type OrchestrationMailboxDeliveryFlight } from './mailbox-pointer-state' +import { + MAILBOX_POINTER_RESERVED, + MAILBOX_POINTER_WRITE_ATTEMPTED +} from './db/messages/mailbox-pointer-enter-state' import { resumePendingOrchestrationMailboxPointer } from './mailbox-pointer-resume' import { stageOrchestrationMailboxPointer } from './mailbox-pointer-stage' @@ -163,7 +167,19 @@ export class OrchestrationMailboxPointerDelivery { db.close() }) }) + +describe('retiring a pty mid-delivery', () => { + // Why: an Enter that was already written may have landed. Releasing it would send the same + // mail a second time, so only phases that provably never submitted become redeliverable. + it('leaves an attempted Enter at its phase instead of making it redeliverable', async () => { + vi.useFakeTimers() + const db = new OrchestrationDb(':memory:') + const settlements: ((settlement: WriteSettlement) => void)[] = [] + const writePty = vi.fn( + () => + new Promise((resolve) => { + settlements.push(resolve) + }) as unknown as WriteSettlement + ) + try { + const message = db.insertMessage({ from: 'a', to: 'run:run-1', subject: 'mail' }) + const delivery = new OrchestrationMailboxPointerDelivery(pointerDeps(db, writePty) as never) + delivery.deliver(LEAF, { mailboxHandle: 'run:run-1' }) + + // Settle the pointer write, so the pane reaches WRITE_ATTEMPTED and arms the Enter. + settlements[0]?.(WRITE_ACCEPTED) + await vi.advanceTimersByTimeAsync(0) + expect(db.getMessageById(message.id)?.pointer_enter_pending).toBe(2) + + // Fire the Enter but never settle it: this is the ambiguous state. + await vi.advanceTimersByTimeAsync(600) + expect(db.getMessageById(message.id)?.pointer_enter_pending).toBe(3) + + delivery.retirePty('pty-1') + expect(db.getMessageById(message.id)).toMatchObject({ + pointer_enter_pending: 3, + read: 0 + }) + } finally { + db.close() + vi.useRealTimers() + } + }) + + it('releases a pointer whose Enter never fired', async () => { + vi.useFakeTimers() + const db = new OrchestrationDb(':memory:') + const settlements: ((settlement: WriteSettlement) => void)[] = [] + const writePty = vi.fn( + () => + new Promise((resolve) => { + settlements.push(resolve) + }) as unknown as WriteSettlement + ) + try { + const message = db.insertMessage({ from: 'a', to: 'run:run-1', subject: 'mail' }) + const delivery = new OrchestrationMailboxPointerDelivery(pointerDeps(db, writePty) as never) + delivery.deliver(LEAF, { mailboxHandle: 'run:run-1' }) + settlements[0]?.(WRITE_ACCEPTED) + await vi.advanceTimersByTimeAsync(0) + expect(db.getMessageById(message.id)?.pointer_enter_pending).toBe(2) + + delivery.retirePty('pty-1') + expect(db.getMessageById(message.id)).toMatchObject({ + pointer_enter_pending: 0, + read: 0, + delivered_at: null + }) + } finally { + db.close() + vi.useRealTimers() + } + }) +}) diff --git a/src/main/runtime/orchestration/mailbox-pointer-stage.ts b/src/main/runtime/orchestration/mailbox-pointer-stage.ts index aed3b8e06b4..3c0bea2baad 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-stage.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-stage.ts @@ -58,6 +58,7 @@ export function stageOrchestrationMailboxPointer message.id) try { if ( diff --git a/src/main/runtime/orchestration/mailbox-pointer-state.ts b/src/main/runtime/orchestration/mailbox-pointer-state.ts index 149d25057b5..d52374a4888 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-state.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-state.ts @@ -6,6 +6,8 @@ export type OrchestrationMailboxDeliveryFlight = { submitEnter: (() => void) | null deferredUntilIdle: boolean idleObservedWhileDeferred: boolean + /** The incarnation that staged this flight, so retirement can name the rows it owns. */ + processIncarnation?: string } export type ParkedOrchestrationMailboxDelivery = { diff --git a/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts b/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts index 7d173055b3a..b7c5ed95068 100644 --- a/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts +++ b/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts @@ -386,6 +386,7 @@ function makeOrchestrationDbStub(toHandle: () => string) { runMailbox, markAsDelivered, markAsUndelivered, + releaseMailboxPointerEnter, stageMailboxPointerEnter, insert(subject: string, type: StoredMessageRow['type'] = 'status'): void { rows.push({ @@ -759,7 +760,7 @@ describe('push-on-idle orchestration delivery absence gate', () => { await vi.advanceTimersByTimeAsync(500) expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0) expect(stub.stageMailboxPointerEnter).toHaveBeenCalledOnce() - expect(stub.markAsUndelivered).toHaveBeenCalledOnce() + expect(stub.releaseMailboxPointerEnter).toHaveBeenCalledOnce() expect(stub.rows[0].delivered_at).toBeNull() // The replacement's own delivery starts a fresh flight and completes — @@ -804,7 +805,7 @@ describe('push-on-idle orchestration delivery absence gate', () => { await vi.advanceTimersByTimeAsync(500) expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0) expect(stub.stageMailboxPointerEnter).toHaveBeenCalledOnce() - expect(stub.markAsUndelivered).toHaveBeenCalledOnce() + expect(stub.releaseMailboxPointerEnter).toHaveBeenCalledOnce() // No stray settle flushed the parked trigger into the dead pty. expect(write).toHaveBeenCalledTimes(1) expect(stub.rows.every((row) => row.delivered_at === null)).toBe(true) @@ -835,7 +836,7 @@ describe('push-on-idle orchestration delivery absence gate', () => { await vi.advanceTimersByTimeAsync(500) expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0) expect(stub.stageMailboxPointerEnter).toHaveBeenCalledOnce() - expect(stub.markAsUndelivered).toHaveBeenCalledOnce() + expect(stub.releaseMailboxPointerEnter).toHaveBeenCalledOnce() expect(stub.rows[0].delivered_at).toBeNull() } finally { vi.useRealTimers()