test: remove low-value relay test inventories and mocked worker suite (#38572)

Co-authored-by: OrcaWin <293788423+OrcaWin@users.noreply.github.com>
This commit is contained in:
OrcaWin
2026-11-05 06:42:22 -08:00
committed by GitHub
co-authored by OrcaWin
parent dd4db9351c
commit afe06c366a
9 changed files with 3 additions and 806 deletions
@@ -1,130 +0,0 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import type { RelayAssignmentStore } from '../assignment-store.js'
import type { RelayConfig } from '../../config.js'
import type { RegionalRehomeSafetySnapshot } from '../../observability/relay-observability.js'
import { startRegionalRehomeWorker } from './regional-rehome-worker.js'
const candidate = {
v: 1,
attemptId: '11111111-1111-4111-8111-111111111111',
userId: 'private-user',
relayHostId: 'abcdefghijklmnop',
sourceCellId: 'source',
sourceCellUrl: 'https://source.example.test',
sourceCellIncarnation: '22222222-2222-4222-8222-222222222222',
sourceAssignmentEpoch: 7,
sourceGeneration: 3,
targetCellId: 'target'
}
const safety = {
observedAt: 100,
sqlFailures: 0,
reconnects: 0,
controlActivityRecoveryFailures: 0,
databasePoolWaiting: 0,
databasePoolWaitersMax: 0,
databasePoolWaitMsMax: 0
} satisfies RegionalRehomeSafetySnapshot
const config = {
role: 'director',
regionCorrectionCohortPercent: 100,
rehomeAudience: 'https://relay.example.test/v1/admin/host-drain',
rehomeDirectorServiceAccount: 'director@example.test'
} as RelayConfig
function setup(fetch: typeof globalThis.fetch) {
const selectIdleRegionalRehomeCandidates = vi
.fn()
.mockResolvedValueOnce([])
.mockResolvedValue([candidate])
const claimRegionalRehome = vi.fn()
const recordRegionalRehomeDispatchFailure = vi.fn()
const worker = startRegionalRehomeWorker(
config,
{
selectIdleRegionalRehomeCandidates,
claimRegionalRehome,
recordRegionalRehomeDispatchFailure
} as unknown as RelayAssignmentStore,
{
safetySnapshot: () => safety,
intervalMs: 60_000,
identityToken: async () => 'private-token',
fetch
}
)!
return {
worker,
selectIdleRegionalRehomeCandidates,
claimRegionalRehome,
recordRegionalRehomeDispatchFailure
}
}
describe('idle regional worker dispatch', () => {
afterEach(() => vi.restoreAllMocks())
it('sends an idle request without claiming an assignment first', async () => {
const fetch = vi.fn<typeof globalThis.fetch>(async () =>
Response.json({ v: 1, outcome: 'committed' })
)
const c = setup(fetch)
try {
await vi.waitFor(() => expect(c.selectIdleRegionalRehomeCandidates).toHaveBeenCalledOnce())
await c.worker.run()
expect(c.claimRegionalRehome).not.toHaveBeenCalled()
expect(fetch).toHaveBeenCalledOnce()
const [url, init] = fetch.mock.calls[0]!
expect(String(url)).toBe('https://source.example.test/v1/admin/host-idle-rehome')
const { sourceCellUrl: _, ...request } = candidate
expect(JSON.parse(String(init?.body))).toEqual({
...request,
cohortPercent: 100,
directorSafety: safety
})
} finally {
c.worker.stop()
}
})
it('progresses past busy hosts without charging a dispatch failure', async () => {
const fetch = vi
.fn<typeof globalThis.fetch>()
.mockResolvedValueOnce(Response.json({ v: 1, outcome: 'busy' }))
.mockResolvedValueOnce(Response.json({ v: 1, outcome: 'committed' }))
const c = setup(fetch)
try {
await vi.waitFor(() => expect(c.selectIdleRegionalRehomeCandidates).toHaveBeenCalledOnce())
c.selectIdleRegionalRehomeCandidates.mockResolvedValue([
candidate,
{
...candidate,
relayHostId: 'ponmlkjihgfedcba',
attemptId: '33333333-3333-4333-8333-333333333333'
}
])
await c.worker.run()
expect(fetch).toHaveBeenCalledTimes(2)
expect(c.recordRegionalRehomeDispatchFailure).not.toHaveBeenCalled()
} finally {
c.worker.stop()
}
})
it('does not charge a lost response as a claimed migration failure', async () => {
const warning = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
const fetch = vi.fn<typeof globalThis.fetch>(async () => {
throw new Error('response lost')
})
const c = setup(fetch)
try {
await vi.waitFor(() => expect(c.selectIdleRegionalRehomeCandidates).toHaveBeenCalledOnce())
await c.worker.run()
expect(fetch).toHaveBeenCalledOnce()
expect(warning).toHaveBeenCalledWith(
expect.stringContaining('"event":"orca_relay_idle_rehome_request_failed"')
)
expect(warning).toHaveBeenCalledWith(expect.stringContaining('"outcomes":{"failed":1}'))
expect(c.recordRegionalRehomeDispatchFailure).not.toHaveBeenCalled()
} finally {
c.worker.stop()
}
})
})
@@ -194,43 +194,6 @@ describe('ordered region decisions and migration outcomes', () => {
])
})
it('supersedes prior windows and keeps an inconclusive tombstone immutable', async () => {
const context = await setup()
const first = await window(context)
const second = await window(context)
expect(second.generation).toBe(first.generation + 1)
const report = {
v: 1 as const,
action: 'report' as const,
assignmentEpoch: first.assignmentEpoch,
policyVersion: 1 as const,
outcome: 'conclusive' as const,
measurements: { 'us-central1': 200, 'asia-east2': 40 }
}
expect(
await context.store.exchangeRegionCorrection(
identity,
{ ...report, generation: first.generation },
first.assignmentEpoch
)
).toMatchObject({ reportStatus: 'stale' })
expect(
await context.store.exchangeRegionCorrection(
identity,
{ ...report, generation: second.generation, outcome: 'inconclusive', reason: 'jitter' },
second.assignmentEpoch
)
).toMatchObject({ reportStatus: 'accepted' })
expect(
await context.store.exchangeRegionCorrection(
identity,
{ ...report, generation: second.generation },
second.assignmentEpoch
)
).toMatchObject({ reportStatus: 'duplicate' })
expect(await context.store.previewRegionCorrection()).toEqual({ ineligible: 1 })
})
it('previews the uncapped fleet without writes, claims, or locked reads', async () => {
const context = await setup()
const query = context.database.query.bind(context.database)
@@ -3,8 +3,7 @@ import { RelayAssignmentStore } from '../assignment-store.js'
import { openRelayDatabase, type RelayDatabase } from '../../database/database.js'
import {
REGIONAL_REHOME_RECONNECTS_PER_CELL_LIMIT,
REGIONAL_REHOME_SQL_FAILURES_LIMIT,
REGIONAL_REHOME_SQL_FAILURES_PER_CELL_LIMIT
REGIONAL_REHOME_SQL_FAILURES_LIMIT
} from './regional-rehome-safety.js'
const databaseUrl = process.env.ORCA_RELAY_TEST_POSTGRES_URL
@@ -72,30 +71,6 @@ describePostgres('PostgreSQL regional rehoming', () => {
}
}
it('defaults to a closed correction cohort even with enabled durable control', async () => {
const context = await fixture()
const closed = new RelayAssignmentStore(primary, context.now, {
requireLiveCells: true,
heartbeatTtlMs: 45_000
})
expect(await cutover(closed, context.now())).toBeNull()
const preview = await closed.previewRegionalRehomeEligibility({
observedAt: context.now(),
sqlFailures: 0,
reconnects: 0,
controlActivityRecoveryFailures: 0,
databasePoolWaiting: 0,
databasePoolWaitersMax: 0,
databasePoolWaitMsMax: 0
})
expect(preview.cohortPercent).toBe(0)
expect(preview.counts['outside-cohort']).toBe(1)
expect(await attemptAndMigrationCounts(context.identity)).toEqual({
attempts: 0,
migrations: 0
})
})
it('counts existing generic migrations against the optimization cap and preview', async () => {
const context = await fixture()
const safety = {
@@ -128,61 +103,6 @@ describePostgres('PostgreSQL regional rehoming', () => {
})
})
it('preview excludes request capacity exhaustion before a claim', async () => {
const context = await fixture()
await primary.query(
`UPDATE relay_cells SET capacity_requests = reserved_requests + 1 WHERE cell_id = ?`,
[context.target.id]
)
const preview = await context.store.previewRegionalRehomeEligibility({
observedAt: context.now(),
sqlFailures: 0,
reconnects: 0,
controlActivityRecoveryFailures: 0,
databasePoolWaiting: 0,
databasePoolWaitersMax: 0,
databasePoolWaitMsMax: 0
})
expect(preview.counts['no-target-headroom']).toBe(1)
expect(await cutover(context.store, context.now())).toBeNull()
})
it('claims through ambient per-cell sql retry noise', async () => {
const context = await fixture()
await primary.query(
`UPDATE relay_cell_rehome_safety
SET sql_failures = ${REGIONAL_REHOME_SQL_FAILURES_PER_CELL_LIMIT}
WHERE cell_id IN (?, ?)`,
[context.source.id, context.target.id]
)
expect(await cutover(context.store, context.now())).not.toBeNull()
})
it('moves a us-central1 host onto a cell in its preferred asia-east2 region', async () => {
const context = await fixture()
const attempt = await cutover(context.store, context.now())
expect(attempt).toMatchObject({
preferredRegion: 'asia-east2',
sourceCellId: context.source.id,
targetCellId: context.target.id
})
expect(
await primary.query(
`SELECT preferred_region, source_cell_id, target_cell_id
FROM relay_region_rehome_attempts WHERE user_id = ?`,
[context.identity.userId]
)
).toEqual([
{
preferred_region: 'asia-east2',
source_cell_id: context.source.id,
target_cell_id: context.target.id
}
])
})
it('moves an asia-east2 host back onto a cell in its preferred us-central1 region', async () => {
const context = await fixture({
sourceRegion: 'asia-east2',
@@ -216,39 +136,6 @@ describePostgres('PostgreSQL regional rehoming', () => {
).toEqual([{ cell_id: context.target.id }])
})
it('leaves a host whose preference already matches its own region', async () => {
const context = await fixture({ preferredRegion: 'us-central1' })
await expect(cutover(context.store, context.now())).resolves.toBeNull()
await expect(context.store.inspectRegionalRehomeControl()).resolves.toMatchObject({
generation: 1,
enabled: true
})
expect(await attemptAndMigrationCounts(context.identity)).toEqual({
attempts: 0,
migrations: 0
})
})
it('leaves a host whose preference is older than the configured max age', async () => {
const context = await fixture()
await primary.query(
`UPDATE relay_region_decisions SET observed_at = ?
WHERE user_id = ? AND relay_host_id = ?`,
[context.now() - 24 * 60 * 60_000 - 1, context.identity.userId, context.identity.relayHostId]
)
await expect(cutover(context.store, context.now())).resolves.toBeNull()
await expect(context.store.inspectRegionalRehomeControl()).resolves.toMatchObject({
generation: 1,
enabled: true
})
expect(await attemptAndMigrationCounts(context.identity)).toEqual({
attempts: 0,
migrations: 0
})
})
it('leaves a host inside its per-host rehome cooldown, in either direction', async () => {
const context = await fixture({ hostCooldownMs: 3 * 24 * 60 * 60_000 })
// A move this host already made, whichever way it went.
@@ -295,42 +182,6 @@ describePostgres('PostgreSQL regional rehoming', () => {
})
})
it('leaves a host whose preferred region holds no drainable cell', async () => {
// A cell that cannot be drained cannot be a target: the host would land
// where no later rehome could move it out again.
const context = await fixture({ targetProtocol: 0 })
await expect(cutover(context.store, context.now())).resolves.toBeNull()
await expect(context.store.inspectRegionalRehomeControl()).resolves.toMatchObject({
generation: 1,
enabled: true
})
expect(await attemptAndMigrationCounts(context.identity)).toEqual({
attempts: 0,
migrations: 0
})
})
it('skips an unclean cell without latching the control off', async () => {
const context = await fixture()
await primary.query(
`UPDATE relay_cell_rehome_safety
SET sql_failures = ${REGIONAL_REHOME_SQL_FAILURES_PER_CELL_LIMIT + 1}
WHERE cell_id = ?`,
[context.target.id]
)
expect(await cutover(context.store, context.now())).toBeNull()
expect(await context.store.inspectRegionalRehomeControl()).toMatchObject({
generation: 1,
enabled: true
})
expect(await attemptAndMigrationCounts(context.identity)).toEqual({
attempts: 0,
migrations: 0
})
})
it('lets only one director claim a host', async () => {
const context = await fixture()
const claims = await Promise.all([
@@ -151,15 +151,6 @@ describe('rehome source-region filter', () => {
await context.database.close()
})
it('selects both directions when no director region is configured', async () => {
const context = await setup(undefined)
const candidates = await context.store.selectIdleRegionalRehomeCandidates(context.safety())
expect(new Set(moves(candidates).map(({ from, to }) => `${from}>${to}`))).toEqual(
new Set(['us-central1>asia-east2', 'asia-east2>us-central1'])
)
await context.database.close()
})
it('reports the filtered hosts in the preview, agreeing with selection', async () => {
const context = await setup('us-central1')
const preview = await context.store.previewRegionalRehomeEligibility(context.safety())
@@ -5,14 +5,12 @@ import {
RelayAssignmentStore as BaseRelayAssignmentStore,
type RegionalRehomeAttempt,
REGIONAL_REHOME_QUARANTINE_FAILURES,
REGIONAL_REHOME_QUARANTINE_MS,
REGIONAL_REHOME_REDRAIN_SEND_LIMIT
REGIONAL_REHOME_QUARANTINE_MS
} from '../assignment-store.js'
import {
openInMemoryRelayDatabase,
type RelayDatabase,
type RelayLockOptions,
type SqlRow
type RelayLockOptions
} from '../../database/database.js'
import {
REGIONAL_REHOME_SQL_FAILURES_LIMIT,
@@ -44,28 +42,6 @@ const targetIncarnation = '22222222-2222-4222-8222-222222222222'
const bystanderIncarnation = '33333333-3333-4333-8333-333333333333'
describe('regional rehome assignment state', () => {
it('advances past a full candidate page whose destination lacks capacity', async () => {
const context = await setup()
for (let i = 0; i < 10; i++) {
await activatePreferredSource(context, {
userId: `blocked-${i}`,
relayHostId: 'abcdefghijklmnop'
})
}
context.advance(1)
const reverse = { userId: 'healthy-reverse', relayHostId: 'abcdefghijklmnop' }
await activateReversePreferredSource(context, reverse)
await context.database.query(
'UPDATE relay_cells SET capacity_requests = reserved_requests WHERE cell_id = ?',
[target.id]
)
expect(await context.store.tryIdleRehome()).toMatchObject({
userId: reverse.userId,
sourceCellId: target.id,
targetCellId: source.id
})
await context.database.close()
})
it('defaults optional correction off even with enabled durable control', async () => {
const context = await setup()
@@ -145,32 +121,6 @@ describe('regional rehome assignment state', () => {
await context.database.close()
})
it('does not open a transaction while the worker is disabled', async () => {
const delegate = await openInMemoryRelayDatabase()
const database = new TransactionCountingDatabase(delegate)
const store = new IdleRehomeTestStore(database, () => 1_000_000)
await store.inspectRegionalRehomeControl()
database.transactionCalls = 0
await expect(store.tryIdleRehome()).resolves.toBeNull()
expect(database.transactionCalls).toBe(0)
await database.close()
})
it('initializes a missing control row without opening a transaction', async () => {
const delegate = await openInMemoryRelayDatabase()
const database = new TransactionCountingDatabase(delegate)
const store = new IdleRehomeTestStore(database, () => 1_000_000)
await expect(store.tryIdleRehome()).resolves.toBeNull()
expect(database.transactionCalls).toBe(0)
await expect(store.inspectRegionalRehomeControl()).resolves.toMatchObject({
generation: 0,
enabled: false
})
await database.close()
})
it('uses a generation-bound durable kill switch', async () => {
const context = await setup()
expect(await context.store.inspectRegionalRehomeControl()).toMatchObject({
@@ -653,22 +603,6 @@ describe('regional rehome assignment state', () => {
await context.database.close()
})
it('does not throttle or log an idle tick with no candidates', async () => {
const context = await setup()
const warnings = collectEventWarnings('orca_relay_regional_rehome_candidates_skipped')
try {
expect(await context.store.tryIdleRehome()).toBeNull()
} finally {
warnings.restore()
}
expect(warnings.entries).toEqual([])
expect(
await context.database.query(`SELECT next_dispatch_at FROM relay_region_rehome_worker_state`)
).toEqual([{ next_dispatch_at: 0 }])
await context.database.close()
})
it('pauses the claim without latching when a sql spike lands between selection and commit', async () => {
const context = await setup()
await activatePreferredSource(context, {
@@ -912,23 +846,6 @@ describe('regional rehome assignment state', () => {
})
// Only inventory contention is ordinary; other failures must propagate and count against dispatch.
it('propagates a claim failure that is not inventory contention', async () => {
const probe = new CellInventoryLockProbe()
const context = await setup({ wrap: (database) => probe.wrap(database) })
await activatePreferredSource(context, { userId: 'user-1', relayHostId: 'abcdefghijklmnop' })
probe.reset()
probe.failWith = new Error('relay_capacity_exhausted')
const busy = collectEventWarnings('orca_relay_sweep_cell_inventory_busy')
try {
await expect(context.store.tryIdleRehome()).rejects.toThrow('relay_capacity_exhausted')
} finally {
busy.restore()
}
expect(busy.entries).toEqual([])
await context.database.close()
})
// One inventory failure abandons the transaction’s whole candidate page, so count all candidates skipped.
@@ -1484,36 +1401,6 @@ describe('regional rehome assignment state', () => {
})
})
class TransactionCountingDatabase implements RelayDatabase {
transactionCalls = 0
constructor(private readonly delegate: RelayDatabase) {}
query(sql: string, params?: unknown[]): Promise<SqlRow[]> {
return this.delegate.query(sql, params)
}
queryLocked(
sql: string,
params?: unknown[],
options?: { failIfUnavailable?: boolean }
): Promise<SqlRow[]> {
return this.delegate.queryLocked(sql, params, options)
}
transaction<T>(
operation: (transaction: RelayDatabase) => Promise<T>,
options?: { reportRetries?: boolean }
): Promise<T> {
this.transactionCalls += 1
return this.delegate.transaction(operation, options)
}
close(): Promise<void> {
return this.delegate.close()
}
}
type Context = Awaited<ReturnType<typeof setup>>
function collectDisableWarnings() {
@@ -1910,21 +1797,6 @@ class CellInventoryLockProbe {
}
}
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)
}
}
class IdleRehomeTestStore extends BaseRelayAssignmentStore {
private readonly fixtureDatabase: RelayDatabase
private readonly fixtureNow: () => number
@@ -110,15 +110,6 @@ function seated(epoch: number, joinedAt = 0) {
}
describe('reserve assignment on a director', () => {
it('keeps today’s path when off, or when no cell reports reserve mode', async () => {
expect((await setup({ mode: 'off' }).assignment.plan(HOST, { reconnect: true, region: US })).kind).toBe(
'database'
)
const databaseOnly = setup({ feeds: [feed('c1', { admitMode: 'db' }), feed('c2', { admitMode: 'db' })] })
expect(await databaseOnly.assignment.plan(HOST, { reconnect: false, region: US })).toEqual({
kind: 'database'
})
})
it('reads what each cell acts on: a tripped dead-man is db, a flip back still leasing is reserve', async () => {
const tripped = setup({
@@ -232,15 +223,6 @@ describe('reserve assignment on a director', () => {
expect(reserved.every((call) => !call.sticky && call.epoch === 8)).toBe(true)
})
it('leaves a host the map last saw on a database-mode cell to today’s path, with the floor', async () => {
const { assignment } = setup({
feeds: [feed('c1', { admitMode: 'db', full: [seated(4)] }), feed('c2')]
})
const plan = await assignment.plan(HOST, { reconnect: true, region: US })
expect(plan).toMatchObject({ kind: 'database', epochFloor: 4 })
expect(plan.kind === 'database' && typeof plan.placeFresh).toBe('function')
})
it('leaves a cell without a feed (c17/c18) on today’s path without starving the reserve pool', async () => {
const cells = cellList(['c1', 'c2', 'c17'])
let rowOnOld = false
@@ -423,13 +405,6 @@ describe('reserve assignment on a director', () => {
).resolves.toMatchObject({ assignmentEpoch: 4 })
})
it('keeps today’s path while the map is incomplete, and in dry run', async () => {
const gated = setup({ cells: cellList(['c1', 'c2']), feeds: [feed('c1', { full: [seated(4)] })], startedAt: 100_500 })
expect(await gated.assignment.plan(HOST, request)).toEqual({ kind: 'database' })
const dryRun = setup({ mode: 'dry-run', feeds: [feed('c1', { full: [seated(4)] }), feed('c2')] })
expect(await dryRun.assignment.plan(HOST, request)).toEqual({ kind: 'database' })
expect([...gated.reserved, ...dryRun.reserved]).toEqual([])
})
})
describe('after the database lane found no cell anywhere', () => {
@@ -517,27 +492,6 @@ describe('reserve assignment on a director', () => {
)
})
it('counts a spill once, not as a database fallback or a pace per region', async () => {
for (const [asiaReserve, label] of [
[fullCounts, 'full'],
[paced, 'paced']
] as const) {
const lines: Record<string, unknown>[] = []
const harness = setupSpill(asiaReserve, {}, (line) => lines.push(JSON.parse(line)))
await (await spillOf(harness))?.({ skipOwnRegion: false, rowEpoch: 0 })
harness.now.value += 60_000
await harness.assignment.plan(HOST, { reconnect: true, region: ASIA })
expect(lines.find((line) => line.event === 'orca_relay_reserve_placement_summary'), label).toMatchObject({
spill: 1,
placed: 1,
database: 0,
retries: { paced: 0 }
})
expect(lines.filter((line) => line.event === 'orca_relay_reserve_spill'), label).toEqual([
{ event: 'orca_relay_reserve_spill', fromRegion: ASIA, toRegion: US, count: 1 }
])
}
})
})
describe('rehome on reconnect', () => {
@@ -667,16 +621,6 @@ describe('reserve assignment on a director', () => {
expect(gated.reserved).toEqual([])
})
it('passes a region-correction request to the database lane when rehome has no room', async () => {
const { assignment, warm } = setupRehome({ asia: asiaFull(0.98) })
await warm()
expect(await assignment.plan(HOST, { ...ask, regionCorrection: true })).toEqual({
kind: 'database',
regionReserve: expect.any(Function),
reserveSpill: expect.any(Function)
})
})
it('falls through to the sticky answer when the rehome booking is paced', async () => {
const { assignment, reserved, warm } = setupRehome({
reserve: (cellId) => (cellId === 'asia-r' ? { outcome: 'intake' } : { outcome: 'ok' })
@@ -791,40 +735,6 @@ describe('reserve assignment on a director', () => {
expect(reserved).toMatchObject([{ dryRun: true }])
})
it('in dry run counts outcomes per picked cell, and a `full` refusal while the map showed room', async () => {
const summaries: Array<{ dryRun: Record<string, unknown> }> = []
const dryRunOn = async (units: number, outcome: string) => {
const { assignment, now } = setup({
mode: 'dry-run',
cells: cellList(['c1']),
feeds: [feed('c1', { counts: { seats: units, controls: 0, bookings: 0, units, ceiling: 500 } })],
reserve: () => ({ outcome }),
log: (line) => summaries.push(JSON.parse(line))
})
for (let index = 0; index < 2; index += 1) {
const plan = await assignment.plan(HOST, { reconnect: false, region: US })
if (plan.kind === 'database') await plan.placeFresh?.()
}
// Lets the fire-and-forget dry-run reserves settle before the summary flushes.
await Promise.resolve()
now.value += 60_000
await assignment.plan(HOST, { reconnect: false, region: US })
}
// 1 free seat of 500 is under the 1% bound; 250 free is far over it.
await dryRunOn(499, 'full')
await dryRunOn(250, 'full')
await dryRunOn(250, 'ok')
expect(summaries.map((summary) => summary.dryRun)).toMatchObject([
{
outcomes: { full: 2 },
count: 2,
cells: { c1: { outcomes: { full: 2 }, fullWithRoom: 0, minFullFreeShare: 0.002 } }
},
{ outcomes: { full: 2 }, cells: { c1: { outcomes: { full: 2 }, fullWithRoom: 2, minFullFreeShare: 0.5 } } },
{ outcomes: { ok: 2 }, cells: { c1: { outcomes: { ok: 2 }, fullWithRoom: 0, minFullFreeShare: null } } }
])
})
describe('when a cell answers a demotion', () => {
const outcomes = (lines: string[]) =>
lines
@@ -854,147 +764,11 @@ describe('reserve assignment on a director', () => {
expect(directory.seatsOnCell('c1')).toHaveLength(1)
})
it('records a cell it could not reach as unreachable', async () => {
const lines: string[] = []
const logged = Promise.withResolvers<void>()
const { assignment } = setup({
feeds: [feed('c1', { full: [seated(10, 1_000)] }), feed('c2')],
demote: () => null,
log: (line) => {
lines.push(line)
if (JSON.parse(line).event === 'orca_relay_reserve_demote_outcome') {
logged.resolve()
}
}
})
assignment.demoteSeat({ ...HOST, cellId: 'c1', epoch: 10, joinedAt: 1_000 }, 'reconcile')
await logged.promise
expect(outcomes(lines)).toMatchObject([{ outcome: 'unreachable' }])
})
})
describe('for a host re-asking off a draining cell', () => {
const events = (lines: string[], event: string) =>
lines.map((line) => JSON.parse(line)).filter((entry) => entry.event === event)
const flushedSummary = async (assignment: ReserveAssignment, now: { value: number }, lines: string[]) => {
now.value += 60_001
await assignment.plan({ ...HOST, relayHostId: 'zzzzzzzzzzzzzzzz' }, { reconnect: false, region: US })
return events(lines, 'orca_relay_reserve_placement_summary')[0]
}
// graceMs 0: the host re-asks while its seat still shows, but the cell's feed already says draining.
it('counts the placement by source cell before any leave lands', async () => {
const lines: string[] = []
const { assignment, now } = setup({
feeds: [feed('c1', { draining: true, full: [seated(7)] }), feed('c2')],
log: (line) => lines.push(line)
})
expect(await assignment.plan(HOST, { reconnect: true, region: US })).toMatchObject({
kind: 'answer',
lane: 'placement',
assignment: { cellId: 'c2', assignmentEpoch: 8 }
})
expect(events(lines, 'orca_relay_reserve_drain_replaced')).toEqual([
{ event: 'orca_relay_reserve_drain_replaced', fromCellId: 'c1', toCellId: 'c2', region: US }
])
expect((await flushedSummary(assignment, now, lines)).drainReplaced).toEqual({ c1: 1 })
})
it('counts a drain-only seat, and a newest leave on a draining cell', async () => {
const lines: string[] = []
const drainOnly = setup({
feeds: [feed('c1', { full: [{ ...seated(7), state: 'drain-only' }] }), feed('c2')],
log: (line) => lines.push(line)
})
await drainOnly.assignment.plan(HOST, { reconnect: true, region: US })
const left = setup({ feeds: [feed('c1', { draining: true }), feed('c2')], log: (line) => lines.push(line) })
left.directory.apply(
'c1',
feed('c1', {
draining: true,
full: undefined,
seq: 3,
changes: [
{ seq: 2, kind: 'join', ...HOST, epoch: 7, generation: 1, at: 1 },
{ seq: 3, kind: 'leave', ...HOST, epoch: 7, generation: 1, closeCode: 4503, at: 2 }
]
}),
left.now.value,
left.now.value
)
await left.assignment.plan(HOST, { reconnect: true, region: US })
expect(events(lines, 'orca_relay_reserve_drain_replaced').map((entry) => entry.fromCellId)).toEqual([
'c1',
'c1'
])
})
it('counts a host once: after its re-booking, a second ask is not a drain re-ask', async () => {
const lines: string[] = []
const { assignment, directory, now } = setup({
feeds: [feed('c1', { draining: true, full: [{ ...seated(7), state: 'drain-only' }] }), feed('c2')],
log: (line) => lines.push(line)
})
await assignment.plan(HOST, { reconnect: true, region: US })
// c2 reports the booking before the host's join: the map still holds only the drain-only seat.
directory.apply(
'c2',
feed('c2', {
full: undefined,
seq: 2,
changes: [{ seq: 2, kind: 'reserve', ...HOST, epoch: 8, generation: 0, reservedBy: 'director', at: now.value }]
}),
now.value,
now.value
)
expect(directory.bookingOf(HOST.userId, HOST.relayHostId)).toMatchObject({ cellId: 'c2', epoch: 8 })
await assignment.plan(HOST, { reconnect: false, region: US })
expect(events(lines, 'orca_relay_reserve_drain_replaced')).toHaveLength(1)
})
it('counts a host once when it asks again before the new cell reports the booking', async () => {
const lines: string[] = []
const { assignment } = setup({
feeds: [feed('c1', { draining: true, full: [seated(7)] }), feed('c2')],
log: (line) => lines.push(line)
})
await assignment.plan(HOST, { reconnect: true, region: US })
await assignment.plan(HOST, { reconnect: false, region: US })
expect(events(lines, 'orca_relay_reserve_drain_replaced')).toHaveLength(1)
})
// Placement memories are clock-free: a host that never reached c2 is not counted again later.
it('counts a host once per drain, also when it asks again after 30 s', async () => {
const lines: string[] = []
const { assignment, directory, now } = setup({
feeds: [feed('c1', { draining: true, full: [{ ...seated(7), state: 'drain-only' }] }), feed('c2')],
log: (line) => lines.push(line)
})
await assignment.plan(HOST, { reconnect: true, region: US })
now.value += 31_000
for (const cellId of ['c1', 'c2']) {
directory.apply(cellId, feed(cellId, { draining: cellId === 'c1', full: undefined, seq: 1, changes: [] }), now.value, now.value)
}
await assignment.plan(HOST, { reconnect: true, region: US })
expect(events(lines, 'orca_relay_reserve_drain_replaced')).toHaveLength(1)
})
it('counts afresh once the drained cell restarts', async () => {
const lines: string[] = []
const { assignment, directory, now } = setup({
feeds: [feed('c1', { draining: true, full: [seated(7)] }), feed('c2')],
log: (line) => lines.push(line)
})
await assignment.plan(HOST, { reconnect: true, region: US })
directory.apply(
'c1',
feed('c1', { incarnation: 'c1-incarnation-2', draining: true, full: [seated(20)] }),
now.value,
now.value
)
await assignment.plan(HOST, { reconnect: true, region: US })
expect(events(lines, 'orca_relay_reserve_drain_replaced')).toHaveLength(2)
})
it('marks its pace as the drain’s, on the reserve lane, in the region fallback and in the spill', async () => {
const { assignment } = setup({
@@ -1026,16 +800,5 @@ describe('reserve assignment on a director', () => {
expect(events(lines, 'orca_relay_reserve_drain_replaced')).toEqual([])
})
// The database lane logs orca_relay_sticky_replaced_off_isolated_cell for these; counting both would double the canary's sum.
it('does not count a placement the database lane makes through its region fallback', async () => {
const lines: string[] = []
const { assignment } = setup({
feeds: [feed('c1', { draining: true, full: [seated(7)] }), feed('c2')],
log: (line) => lines.push(line)
})
const plan = await assignment.plan(HOST, { reconnect: true, region: US, regionCorrection: true })
await expect(plan.kind === 'database' ? plan.regionReserve?.() : null).resolves.toMatchObject({ cellId: 'c2' })
expect(events(lines, 'orca_relay_reserve_drain_replaced')).toEqual([])
})
})
})
@@ -1,9 +1,5 @@
import { RELAY_CLOSE_CODE } from '@orca-cloud/relay-contract/close-codes'
import { describe, expect, it } from 'vitest'
import { RelayAssignmentStore } from '../assignment-store.js'
import { openInMemoryRelayDatabase } from '../../database/database.js'
import {
RESERVE_REHOME_HOST_COOLDOWN_MS,
RESERVE_REHOME_SWITCH_STALE_MS,
ReserveRehome,
ReserveRehomeSwitch,
@@ -39,47 +35,6 @@ function warmed(percent = 100) {
}
describe('rehome on reconnect rules', () => {
it('leaves a host that another install dropped in the last 10 minutes', () => {
const peerDropped = { region: 'us-central1' as const, closeCode: RELAY_CLOSE_CODE.PEER_DROPPED }
expect(admitted(warmed(), { leaves: [{ ...peerDropped, at: 60_000 - 10 * 60_000 }] })).toBe(
'peer-dropped'
)
expect(
admitted(warmed(), { leaves: [{ ...peerDropped, at: 60_000 - 10 * 60_000 - 1 }] })
).toBeNull()
})
it('leaves a host whose remembered leaves span regions, or that it moved within a day', () => {
const leaves = [
{ region: 'us-central1' as const, at: 1 },
{ region: 'asia-east2' as const, at: 2 }
]
expect(admitted(warmed(), { leaves })).toBe('moved-recently')
expect(admitted(warmed(), { leaves: [{ region: undefined, at: 1 }] })).toBe(
'unknown-leave-region'
)
const rehome = warmed()
rehome.recordMoved(HOST, 0)
expect(admitted(rehome)).toBe('moved-recently')
expect(admitted(rehome, { now: RESERVE_REHOME_HOST_COOLDOWN_MS })).toBeNull()
})
it('moves no one into a region under 10% free, however full the source is', () => {
expect(admitted(warmed(), { target: { free: 20, ceiling: 1_000, bookable: true } })).toBe(
'target-room'
)
expect(admitted(warmed(), { target: { free: 99, ceiling: 1_000, bookable: true } })).toBe(
'target-room'
)
expect(admitted(warmed(), { target: { free: 100, ceiling: 1_000, bookable: true } })).toBeNull()
expect(admitted(warmed(), { target: { free: 500, ceiling: 1_000, bookable: false } })).toBe(
'target-room'
)
expect(admitted(warmed(), { target: { free: 0, ceiling: 0, bookable: false } })).toBe(
'target-room'
)
})
it('takes a source cell at 0.05% of its controls a minute, burst one, and refunds an unmade move', () => {
const rehome = warmed()
expect(admitted(rehome)).toBeNull()
@@ -156,29 +111,3 @@ describe('the rehome switch under a stalled read', () => {
expect(control.isOpen(now + 1)).toBe(false)
})
})
describe('the rehome switch read', () => {
it('reads the operator row, and nothing before it exists', async () => {
const database = await openInMemoryRelayDatabase()
try {
let now = 100_000_000
const store = new RelayAssignmentStore(database, () => now)
expect(await store.regionRehomeSwitch()).toBeNull()
await store.inspectRegionalRehomeControl()
expect(await store.regionRehomeSwitch()).toMatchObject({ enabled: false })
now += 86_400_000
await store.applyRegionalRehomeControl({
expectedGeneration: 0,
enabled: true,
notBefore: now + 1_000,
ratePerMinute: 10,
preferenceMaxAgeMs: 86_400_000,
hostCooldownMs: 604_800_000,
drainGraceMs: 60_000
})
expect(await store.regionRehomeSwitch()).toEqual({ enabled: true, notBefore: now + 1_000 })
} finally {
await database.close()
}
})
})
@@ -72,18 +72,6 @@ describe('relay token keys while auth is down', () => {
expect(await verify(await relayToken(await signingKey('unknown')))).toBeNull()
})
it('rejects once the last served set is older than the bound', async () => {
const key = await signingKey('auth-1')
const authServer = auth(key)
const jwks = createStaleIfErrorJWKSet(new URL(config.jwksUrl), { fetch: authServer.fetch })
const verify = createRelayTokenVerifier(config as RelayConfig, jwks)
expect(await verify(await relayToken(key))).not.toBeNull()
vi.setSystemTime(Date.now() + RELAY_JWKS_MAX_STALE_MS + 1)
authServer.state.up = false
expect(await verify(await relayToken(key))).toBeNull()
})
it('bounds the stale set at 30 minutes since the last served set', async () => {
expect(RELAY_JWKS_MAX_STALE_MS).toBe(30 * MINUTE)
const key = await signingKey('auth-1')
@@ -204,36 +204,6 @@ describe('relay cell admission selector', () => {
expect(await store.resolve(identity)).toBeNull()
})
it('commits atomically and resolves an ambiguous response idempotently', async () => {
const store = await setup(() => 100)
const input = {
attemptId: 'cutover_004',
expectedGeneration: 0,
expectedMembershipSha256: BASE_MEMBERSHIP_SHA256,
membership: INITIAL_MEMBERSHIP
}
await expect(store.applyCellAdmissionSelector(input)).resolves.toMatchObject({
changed: true,
selector: { generation: 1, membership: INITIAL_MEMBERSHIP }
})
await expect(store.applyCellAdmissionSelector(input)).resolves.toMatchObject({
changed: false,
selector: { generation: 1, membership: INITIAL_MEMBERSHIP }
})
await expect(store.inspectCellAdmissionSelector(input.attemptId)).resolves.toMatchObject({
intent: { state: 'committed' }
})
expect(await store.cellDeploymentStatus('c1')).toMatchObject({
enabled: false,
admissionState: 'existing-only'
})
expect(await store.cellDeploymentStatus('c2')).toMatchObject({
enabled: true,
admissionState: 'migration-only'
})
})
it('rejects a generation-zero membership change between intent and commit', async () => {
const delegate = await openInMemoryRelayDatabase()
const hooked = new AfterTransactionDatabase(delegate)