diff --git a/cloud/apps/relay/src/assignment-store.ts b/cloud/apps/relay/src/assignment-store.ts index 9ead45df22e..3e81b37689f 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,7 @@ 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) => { const worker = ( await transaction.queryLocked( `SELECT * FROM relay_region_rehome_worker_state WHERE worker_id = 'global'` @@ -5952,49 +5969,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-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' }) )