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' }) ) 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()