mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
fix(relay): cells without a seat feed stay out of shadow coverage and switch state (#27058)
Rebased onto main as one commit after its parent squash-merged.
This commit is contained in:
@@ -7896,7 +7896,7 @@ export class RelayAssignmentStore {
|
||||
const rows = await this.database.query(
|
||||
`SELECT cell.cell_id, cell.cell_url, cell.capacity_requests, region.region,
|
||||
admission.admission_state, admission.roll_isolated_at, runtime.ready,
|
||||
runtime.last_heartbeat_at
|
||||
runtime.last_heartbeat_at, runtime.cell_incarnation
|
||||
FROM relay_cells cell
|
||||
LEFT JOIN relay_cell_regions region ON region.cell_id = cell.cell_id
|
||||
LEFT JOIN relay_cell_admission admission ON admission.cell_id = cell.cell_id
|
||||
@@ -7924,7 +7924,8 @@ export class RelayAssignmentStore {
|
||||
general:
|
||||
optionalText(row, 'admission_state') === 'general' &&
|
||||
optionalInteger(row, 'roll_isolated_at') === undefined &&
|
||||
integer(row, 'capacity_requests') > 0
|
||||
integer(row, 'capacity_requests') > 0,
|
||||
...incarnationField(optionalText(row, 'cell_incarnation'))
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -9504,6 +9505,10 @@ function optionalInteger(row: SqlRow, field: string): number | undefined {
|
||||
return integer(row, field)
|
||||
}
|
||||
|
||||
function incarnationField(incarnation: string | undefined): { incarnation?: string } {
|
||||
return incarnation === undefined ? {} : { incarnation }
|
||||
}
|
||||
|
||||
function optionalText(row: SqlRow, field: string): string | undefined {
|
||||
if (row[field] === null || row[field] === undefined) return undefined
|
||||
return text(row, field)
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import { afterAll, beforeAll, describe, expect, it } from 'vitest'
|
||||
import { RelayAssignmentStore } from './assignment-store.js'
|
||||
import type { RelayCellConfig } from './config.js'
|
||||
import { openRelayDatabase, type RelayDatabase } from './database.js'
|
||||
|
||||
// The seat-feed cell list (with the process incarnation the reserve switch keys on) on real PostgreSQL.
|
||||
|
||||
const databaseUrl = process.env.ORCA_RELAY_TEST_POSTGRES_URL
|
||||
const describePostgres = databaseUrl ? describe : describe.skip
|
||||
|
||||
const run = randomUUID().slice(0, 8)
|
||||
const cell: RelayCellConfig = {
|
||||
id: `feed-pg-${run}`,
|
||||
url: `https://feed-${run}.example.com`,
|
||||
capacityRequests: 10,
|
||||
connectionHardCap: 600,
|
||||
connectionUnobservedBound: 50
|
||||
}
|
||||
const INCARNATION = '22222222-2222-4222-8222-222222222222'
|
||||
|
||||
describePostgres('seat-feed cells on PostgreSQL', () => {
|
||||
let database: RelayDatabase
|
||||
|
||||
beforeAll(async () => {
|
||||
database = await openRelayDatabase({ databaseUrl, dataDir: '/tmp' })
|
||||
})
|
||||
|
||||
afterAll(async () => {
|
||||
for (const table of ['relay_cell_admission', 'relay_cell_regions', 'relay_cell_runtime', 'relay_cells']) {
|
||||
await database.query(`DELETE FROM ${table} WHERE cell_id = ?`, [cell.id])
|
||||
}
|
||||
await database.close()
|
||||
})
|
||||
|
||||
it('lists a heartbeating cell with its incarnation', async () => {
|
||||
const store = new RelayAssignmentStore(database, () => 1_000, { requireLiveCells: true, heartbeatTtlMs: 45_000 })
|
||||
await store.reconcileCells([cell], false)
|
||||
expect((await store.seatFeedCells()).find((entry) => entry.cellId === cell.id)).not.toHaveProperty(
|
||||
'incarnation'
|
||||
)
|
||||
await store.recordCellHeartbeat({
|
||||
cellId: cell.id,
|
||||
cellUrl: cell.url,
|
||||
cellIncarnation: INCARNATION,
|
||||
startedAt: 50,
|
||||
ready: true,
|
||||
observedRequests: 0,
|
||||
totalConnections: 0,
|
||||
inFlightConnections: 0,
|
||||
reservedConnectionUnits: 0,
|
||||
enforcedConnectionUnits: 0,
|
||||
connectionHardCap: 600,
|
||||
connectionUnobservedBound: 50
|
||||
})
|
||||
expect((await store.seatFeedCells()).find((entry) => entry.cellId === cell.id)).toMatchObject({
|
||||
cellUrl: cell.url,
|
||||
requiredForComplete: true,
|
||||
incarnation: INCARNATION
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -60,7 +60,7 @@ describe('classifyShadowSeat', () => {
|
||||
[directoryWith({ 'cell-a': [seat(3)] }), at('cell-a', 3), 'agree', true],
|
||||
[directoryWith({ 'cell-a': [] }), null, 'agree-absent', true],
|
||||
[directoryWith({ 'cell-a': [seat(3)] }), null, 'map-only', false],
|
||||
[directoryWith({ 'cell-a': [] }, ['cell-b']), at('cell-b', 3), 'cell-unpolled', true],
|
||||
[directoryWith({ 'cell-a': [] }, ['cell-b']), at('cell-b', 3), 'cell-no-feed', true],
|
||||
[directoryWith({ 'cell-a': [] }), at('cell-z', 3), 'cell-unpolled', true],
|
||||
[directoryWith({ 'cell-a': [] }), at('cell-a', 3), 'db-only-unseen', true],
|
||||
[
|
||||
@@ -153,6 +153,18 @@ describe('classifyShadowSeat', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('keeps a cell without a feed out of the coverage count, on either side of the compare', () => {
|
||||
// cell-b reported a seat, then rolled back to an image without the feed.
|
||||
const directory = directoryWith({ 'cell-b': [seat(2)], 'cell-a': [] })
|
||||
directory.markNoFeed('cell-b', 20)
|
||||
for (const answer of [at('cell-a', 3), at('cell-b', 2)]) {
|
||||
expect(classifyShadowSeat(directory, IDENTITY, answer, 30)).toEqual({
|
||||
class: 'cell-no-feed',
|
||||
explained: true
|
||||
})
|
||||
}
|
||||
})
|
||||
|
||||
it('reads a null answer on a cell whose heartbeat ran out after the list read as pending', () => {
|
||||
const directory = directoryWith({ 'cell-a': [seat(3)] })
|
||||
const listRead = (readAt: number) =>
|
||||
|
||||
@@ -12,6 +12,8 @@ export type ShadowCompareClass =
|
||||
// Neither the database nor the map has the host.
|
||||
| 'agree-absent'
|
||||
| 'map-incomplete'
|
||||
// The cell's image has no seat feed (c17/c18 by design): outside the map, so not a coverage gap.
|
||||
| 'cell-no-feed'
|
||||
| 'cell-unpolled'
|
||||
| 'db-only-left'
|
||||
| 'db-only-unseen'
|
||||
@@ -58,7 +60,11 @@ export function classifyShadowSeat(
|
||||
if (liveness.includes('unknown')) return { class: 'cell-unlive-pending', explained: true }
|
||||
return { class: 'map-only', explained: false }
|
||||
}
|
||||
if (directory.cellState(answer.cellId)?.status !== 'live') {
|
||||
const answerStatus = directory.cellState(answer.cellId)?.status
|
||||
if (answerStatus === 'no-feed') {
|
||||
return { class: 'cell-no-feed', explained: true }
|
||||
}
|
||||
if (answerStatus !== 'live') {
|
||||
return { class: 'cell-unpolled', explained: true }
|
||||
}
|
||||
if (seats.length > 1) {
|
||||
@@ -77,6 +83,9 @@ export function classifyShadowSeat(
|
||||
const mapBehind =
|
||||
answer.assignmentEpoch > seat.epoch && now - movedAt <= SHADOW_COMPARE_LAG_BOUND_MS
|
||||
if (seat.cellId !== answer.cellId) {
|
||||
if (directory.cellState(seat.cellId)?.status === 'no-feed') {
|
||||
return { class: 'cell-no-feed', explained: true }
|
||||
}
|
||||
// A stale seat cell is a coverage gap, kept visible rather than explained away.
|
||||
const seatPolledAt = directory.cellState(seat.cellId)?.lastLiveAt ?? 0
|
||||
if (now - seatPolledAt > SHADOW_COMPARE_LAG_BOUND_MS) {
|
||||
|
||||
@@ -362,6 +362,45 @@ describe('startShadowSeatPoller', () => {
|
||||
expect(cell.directory.isComplete()).toBe(true)
|
||||
})
|
||||
|
||||
it('forgets the switches a cell reported once its image no longer has the feed', async () => {
|
||||
let rolledBack = false
|
||||
const cell = poller({
|
||||
'cell-a': () => json(feed({ seq: 1, full: [] })),
|
||||
'cell-b': () =>
|
||||
rolledBack
|
||||
? new Response('Not Found', { status: 404 })
|
||||
: json(
|
||||
feed({
|
||||
cellId: 'cell-b',
|
||||
seq: 1,
|
||||
full: [],
|
||||
flagsApplied: { generation: 3, flags: { readinessLocal: true } }
|
||||
})
|
||||
)
|
||||
})
|
||||
await cell.tick()
|
||||
expect(cell.directory.cellState('cell-b')?.flagsApplied?.generation).toBe(3)
|
||||
rolledBack = true
|
||||
await cell.tick()
|
||||
cell.stop()
|
||||
expect(cell.directory.cellState('cell-b')).toMatchObject({ status: 'no-feed' })
|
||||
expect(cell.directory.cellState('cell-b')?.flagsApplied).toBeUndefined()
|
||||
})
|
||||
|
||||
it('keeps a reserve switch through a 404 until a feed says db or the process is replaced', () => {
|
||||
const directory = new ShadowSeatDirectory()
|
||||
directory.setCells(['cell-b'])
|
||||
const reserve = { generation: 4, flags: { admitMode: 'reserve' } }
|
||||
directory.apply('cell-b', feed({ cellId: 'cell-b', seq: 1, full: [], flagsApplied: reserve }), 10)
|
||||
directory.markNoFeed('cell-b', 20, INCARNATION)
|
||||
expect(directory.cellState('cell-b')?.flagsApplied).toEqual(reserve)
|
||||
directory.markNoFeed('cell-b', 30)
|
||||
expect(directory.cellState('cell-b')?.flagsApplied).toEqual(reserve)
|
||||
// relay_cell_runtime names another process: the one that reported reserve is gone.
|
||||
directory.markNoFeed('cell-b', 40, '33333333-3333-4333-8333-333333333333')
|
||||
expect(directory.cellState('cell-b')?.flagsApplied).toBeUndefined()
|
||||
})
|
||||
|
||||
it('completes without a cell that is dead, empty or isolated', async () => {
|
||||
const cell = poller({ 'cell-a': () => json(feed({ seq: 1, full: [] })) }, 'all', async () => [
|
||||
CELLS[0]!,
|
||||
|
||||
@@ -95,6 +95,8 @@ export type SeatFeedCell = {
|
||||
heartbeatExpiresAt: number | null
|
||||
// Live, with capacity, and not roll-isolated: completeness waits only on these.
|
||||
requiredForComplete: boolean
|
||||
// relay_cell_runtime's: a new one means the process that reported its switches is gone.
|
||||
incarnation?: string
|
||||
// Admission state general and not roll-isolated: step 5 places only on these.
|
||||
general?: boolean
|
||||
}
|
||||
@@ -137,6 +139,8 @@ export type SeatFeedCellState = {
|
||||
reportedSeats?: number
|
||||
reportedControls?: number
|
||||
flagsApplied?: SeatFeedResponse['flagsApplied']
|
||||
// The incarnation whose feed reported `flagsApplied`.
|
||||
flagsIncarnation?: string
|
||||
admitModeEffective?: 'db' | 'reserve'
|
||||
// Step 5: when the answered poll was sent, and the cell's own reserve numbers.
|
||||
polledAt?: number
|
||||
@@ -255,6 +259,7 @@ export class ShadowSeatDirectory {
|
||||
cursor.reportedSeats = response.counts?.seats
|
||||
cursor.reportedControls = response.counts?.controls
|
||||
cursor.flagsApplied = response.flagsApplied
|
||||
cursor.flagsIncarnation = response.incarnation
|
||||
cursor.admitModeEffective = response.admitModeEffective
|
||||
cursor.polledAt = polledAt
|
||||
cursor.bookings = response.counts?.bookings
|
||||
@@ -281,19 +286,29 @@ export class ShadowSeatDirectory {
|
||||
|
||||
|
||||
// An old cell (404) has no feed; its last known seats stay, marked by status.
|
||||
markNoFeed(cellId: string, now: number): void {
|
||||
markNoFeed(cellId: string, now: number, runtimeIncarnation?: string): void {
|
||||
const cursor = this.cells.get(cellId)
|
||||
if (!cursor) return
|
||||
cursor.status = 'no-feed'
|
||||
cursor.lastAnsweredAt = now
|
||||
cursor.incarnation = undefined
|
||||
cursor.seq = undefined
|
||||
// An image without the feed takes no booking. Its reserve switch is #27058's to keep.
|
||||
// An image without the feed takes no booking.
|
||||
cursor.polledAt = undefined
|
||||
cursor.bookings = undefined
|
||||
cursor.units = undefined
|
||||
cursor.ceiling = undefined
|
||||
cursor.intake = undefined
|
||||
// A reserve switch stands until a feed says db, or the cell's process is replaced: one 404
|
||||
// must not read a reserve cell's seated hosts as idle. Anything else is forgotten.
|
||||
const reserve = cursor.flagsApplied?.flags.admitMode === 'reserve'
|
||||
const replaced =
|
||||
runtimeIncarnation !== undefined && runtimeIncarnation !== cursor.flagsIncarnation
|
||||
if (!reserve || replaced) {
|
||||
cursor.flagsApplied = undefined
|
||||
cursor.flagsIncarnation = undefined
|
||||
cursor.admitModeEffective = undefined
|
||||
}
|
||||
}
|
||||
|
||||
fail(cellId: string, reason: string): void {
|
||||
@@ -670,7 +685,7 @@ export function startShadowSeatPoller(
|
||||
if (stopped) return false
|
||||
if (response.status === 404) {
|
||||
await response.body?.cancel().catch(() => undefined)
|
||||
directory.markNoFeed(cell.cellId, now())
|
||||
directory.markNoFeed(cell.cellId, now(), cell.incarnation)
|
||||
return false
|
||||
}
|
||||
if (!response.ok) {
|
||||
|
||||
Reference in New Issue
Block a user