mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
fix(relay): supersede a reserve seat from the feed once a later, newer join shows (#28716)
* test(relay): check invariant 4 (one demotion winner) in the step-5 simulation Every demotion now names its winner (the newer booking, or the row). The model flags a director that names a different winner for a divergence another already named, and a host whose two seats are each demoted for the other. A fault where directors break level ties by their own cell order shows the check fires. Violations are capped per invariant so one flood cannot hide another class. Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * test(relay): bound unseen duplicates by the silence close and assert all invariants on contested ties A reserve seat the booking director's map had not seen (joined, then half-open within a poll) is never superseded and has no lease to re-check, so it lasts until the cell's control silence close (75 s + one 15 s ping). Invariant 3 now uses that bound for such seats, and the level-divergence config asserts every invariant. Stale maps now judge a divergence another director already demoted, so the config's "contested" counter (divergences judged by two or more directors) is the case invariant 4 guards. The silence-close constant now matches production (90 s, not 100 s). Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * test(relay): raise the floor on map-judged demotes and count only level contests A reconcile whose map has not seen another director's demotion still demotes in production, which raises its epoch floor; the model now does too. The contested counter counts only same-epoch pairs, the only ones invariant 4 can split. The keep-this-seat rule is one helper. Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * refactor(relay): name the simulation's fault-kept seat check for the fault Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * fix(relay): supersede a reserve seat from the feed once a later, newer join shows A half-open seat the booking director's map had not yet seen was never demoted and held its cell until the 75 s control silence close. Directors now demote it once both joins are in the map; the simulation holds invariant 3 at the supersede bound for it. Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * fix(relay): outrank only past cell clock skew, reuse demoteSeat, keep invariant 3's loose clock running Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * refactor(relay): share the outrank rule with the simulation, pin the skew boundary Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * fix(relay): count only confirmed level demotions and every second judgment, and bound duplicates by placement records alone Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * fix(relay): a drained seat keeps its drain grace under a demote, and the feed reports each seat once Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f * refactor(relay): name the outranked-report bound and the reconcile backstop Claude-Session: f2fe8cb5-4467-4a25-8aab-98598b70791f
This commit is contained in:
@@ -330,6 +330,19 @@ describe('a reserve-mode cell admits from memory with its database wedged', () =
|
||||
expect(cell.book(identity, 9)).toEqual([{ outcome: 'ok' }])
|
||||
})
|
||||
|
||||
it('leaves a drained seat to its drain grace instead of the shorter demotion close', async () => {
|
||||
const cell = await startCell({ admitMode: 'reserve' })
|
||||
const identity = await cell.host()
|
||||
cell.book(identity, 9)
|
||||
await cell.connect(identity, 9)
|
||||
const page = cell.relay.sessions.seatFeed(null)
|
||||
const joinedAt = 'full' in page ? page.full[0]!.joinedAt : -1
|
||||
cell.relay.sessions.drain(120_000)
|
||||
const seat = { v: 1 as const, userId: 'user-1', relayHostId: identity.hostId, epoch: 9, joinedAt }
|
||||
expect(cell.relay.sessions.demote(seat)).toBe('draining')
|
||||
expect(cell.relay.sessions.demote(seat)).toBe('draining')
|
||||
})
|
||||
|
||||
it('gives every booking unit back when the cell starts to drain', async () => {
|
||||
const cell = await startCell({ admitMode: 'reserve' })
|
||||
const identity = await cell.host()
|
||||
|
||||
@@ -719,7 +719,7 @@ export class HostSessionRegistry {
|
||||
|
||||
// A director names this seat the loser of a duplicate: stop taking phones for it now, and
|
||||
// close it once its splices have had time to finish. Repeats are one demotion.
|
||||
demote(request: DemoteRequest): 'demoted' | 'already-demoted' | 'not-seated' | 'off' {
|
||||
demote(request: DemoteRequest): 'demoted' | 'already-demoted' | 'draining' | 'not-seated' | 'off' {
|
||||
// Only a reserve-mode cell takes demotions, so flipping admitMode back reverts the route.
|
||||
if (!this.reserveMode()) return 'off'
|
||||
const session = this.get(request)
|
||||
@@ -735,6 +735,8 @@ export class HostSessionRegistry {
|
||||
) {
|
||||
return 'not-seated'
|
||||
}
|
||||
// A drain already told this host to go: its own grace stands, never cut to the demotion's.
|
||||
if (this.drainSentHosts.has(session.relayHostId) || session.regionalDrainAttemptId) return 'draining'
|
||||
const socket = session.socket
|
||||
if (this.demotedSeats.get(key) === socket) return 'already-demoted'
|
||||
this.demotedSeats.set(key, socket)
|
||||
|
||||
@@ -255,6 +255,7 @@ export function createRelayServer(
|
||||
epoch: join.epoch
|
||||
})
|
||||
}
|
||||
directory.onOutranked = (seat) => reservePlacement?.supersedeOutranked(seat)
|
||||
reserveLedger.start()
|
||||
const reconcile = (): void => {
|
||||
const now = (options.now ?? Date.now)()
|
||||
|
||||
@@ -350,6 +350,17 @@ describe('reserve assignment on a director', () => {
|
||||
expect(demoted).toEqual([{ cellId: 'c1', epoch: 5, joinedAt: 1_000 }])
|
||||
})
|
||||
|
||||
// The booking map had not seen the c1 seat (half-open, joined within a poll lag): the feed
|
||||
// demotes it once both joins are in the map, instead of the 75 s silence close.
|
||||
it('tells an old reserve seat the booking missed to go once a newer join shows in the feed', () => {
|
||||
const { assignment, directory, demoted, now } = setup()
|
||||
directory.onOutranked = (seat) => assignment.supersedeOutranked(seat)
|
||||
directory.apply('c2', feed('c2', { full: [seated(6, 3_000)] }), now.value, now.value)
|
||||
expect(demoted).toEqual([])
|
||||
directory.apply('c1', feed('c1', { full: [seated(5, 1_000)] }), now.value, now.value)
|
||||
expect(demoted).toEqual([{ cellId: 'c1', epoch: 5, joinedAt: 1_000 }])
|
||||
})
|
||||
|
||||
it('answers a reconnect only at the newest epoch, newest join first', async () => {
|
||||
const { assignment, reserved } = setup({
|
||||
feeds: [feed('c1', { full: [seated(6, 1_000)] }), feed('c2', { full: [seated(5, 3_000)] })]
|
||||
|
||||
@@ -13,7 +13,7 @@ import {
|
||||
type PlacementCell,
|
||||
type ReservePlacer
|
||||
} from './reserve-placement.js'
|
||||
import type { SeatFeedCell, ShadowSeat, ShadowSeatDirectory } from './shadow-seat-directory.js'
|
||||
import type { OutrankedSeat, SeatFeedCell, ShadowSeat, ShadowSeatDirectory } from './shadow-seat-directory.js'
|
||||
|
||||
// Step 5 on a director: answer a desktop from the in-memory map for cells that report
|
||||
// admitMode=reserve, and fall back to today's database path for everything else. The cell
|
||||
@@ -377,10 +377,10 @@ export class ReserveAssignment {
|
||||
return undefined
|
||||
}
|
||||
|
||||
// Fire-and-forget; the cell checks the seat's epoch and join itself.
|
||||
demoteSeat(seat: { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number }): void {
|
||||
// Fire-and-forget; the cell checks the seat's epoch and join itself. False: no such cell listed.
|
||||
demoteSeat(seat: { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number }): boolean {
|
||||
const cell = this.input.cells().find((entry) => entry.cellId === seat.cellId)
|
||||
if (!cell) return
|
||||
if (!cell) return false
|
||||
void this.input.client.demote(cell, {
|
||||
v: 1,
|
||||
userId: seat.userId,
|
||||
@@ -388,6 +388,7 @@ export class ReserveAssignment {
|
||||
epoch: seat.epoch,
|
||||
joinedAt: seat.joinedAt
|
||||
})
|
||||
return true
|
||||
}
|
||||
|
||||
// Supersede at placement: the old seats this director's map holds on reserve cells are told
|
||||
@@ -396,27 +397,34 @@ export class ReserveAssignment {
|
||||
for (const seat of seats) {
|
||||
if (seat.cellId === cellId || seat.epoch >= epoch) continue
|
||||
if (this.input.directory.admitModeOf(seat.cellId) !== 'reserve') continue
|
||||
const cell = this.input.cells().find((entry) => entry.cellId === seat.cellId)
|
||||
if (!cell) continue
|
||||
this.summary.supersedes += 1
|
||||
void this.input.client.demote(cell, {
|
||||
v: 1,
|
||||
userId: identity.userId,
|
||||
relayHostId: identity.relayHostId,
|
||||
epoch: seat.epoch,
|
||||
joinedAt: seat.joinedAt
|
||||
})
|
||||
this.input.log?.(
|
||||
JSON.stringify({
|
||||
event: 'orca_relay_reserve_supersede',
|
||||
relayHostIdDigest: relayHostLogDigest(identity.relayHostId),
|
||||
fromCellId: seat.cellId,
|
||||
toCellId: cellId
|
||||
})
|
||||
)
|
||||
this.demoteSuperseded({ ...identity, ...seat }, cellId, 'placement')
|
||||
}
|
||||
}
|
||||
|
||||
// The feed's half of supersede: a seat the booking map did not hold, found once both joins
|
||||
// are in this map. Without it that seat holds its cell until the 75 s silence close.
|
||||
supersedeOutranked(seat: OutrankedSeat): void {
|
||||
this.demoteSuperseded(seat, seat.byCellId, 'feed')
|
||||
}
|
||||
|
||||
private demoteSuperseded(
|
||||
seat: { userId: string; relayHostId: string; cellId: string; epoch: number; joinedAt: number },
|
||||
toCellId: string,
|
||||
via: 'placement' | 'feed'
|
||||
): void {
|
||||
if (!this.demoteSeat(seat)) return
|
||||
this.summary.supersedes += 1
|
||||
this.input.log?.(
|
||||
JSON.stringify({
|
||||
event: 'orca_relay_reserve_supersede',
|
||||
relayHostIdDigest: relayHostLogDigest(seat.relayHostId),
|
||||
fromCellId: seat.cellId,
|
||||
toCellId,
|
||||
via
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
private flushSummary(): void {
|
||||
const now = this.now()
|
||||
if (now - this.summaryAt < SUMMARY_INTERVAL_MS) return
|
||||
|
||||
@@ -294,6 +294,8 @@ describe('step 5 deterministic simulation', () => {
|
||||
[1, nearFull({ bookPastCap: true })],
|
||||
[2, drain({ intakeBurst: 200, directorShare: 0.05 })],
|
||||
[3, drainedWithoutSupersede()],
|
||||
// Half-open seats the booking map had not seen hold their cell until the silence close.
|
||||
[3, levelSeats(11, { skipFeedSupersede: true })],
|
||||
[4, levelSeats(11, { localTieBreak: true })],
|
||||
[5, mixed({ unguardedLedger: true })],
|
||||
// A director restart mid-queue loses its booked joins' rows until its boot reconcile.
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
ReservePlacer,
|
||||
type PlacementCell
|
||||
} from './reserve-placement.js'
|
||||
import { seatOutranks } from './shadow-seat-directory.js'
|
||||
|
||||
// A seeded discrete-event model of step 5 that drives the real CellReserveBook and
|
||||
// ReservePlacer. Everything else (sockets, the database, desktops) is modelled. The checks
|
||||
@@ -111,8 +112,10 @@ export type SimulationFaults = {
|
||||
intakeBurst?: number
|
||||
// Each director assumes this share of every cell's intake, instead of a fifth.
|
||||
directorShare?: number
|
||||
// Placement does not supersede the old seat: a duplicate then outlives its bound.
|
||||
// Nothing supersedes the old seat, at placement or from the feed: a duplicate outlives its bound.
|
||||
skipSupersede?: boolean
|
||||
// Only placement supersedes: a half-open seat the booking map had not seen lasts to its silence close.
|
||||
skipFeedSupersede?: boolean
|
||||
// A restarted director forgets its ledger queue and never reconciles.
|
||||
noBootReconcile?: boolean
|
||||
// The reconcile leaves a seat whose row is ahead of it alone.
|
||||
@@ -666,10 +669,15 @@ export async function runReservePlacementSimulation(
|
||||
placement: placementCell(cell, snapshot, requestedAt)
|
||||
}
|
||||
director.views.set(cell.cellId, current)
|
||||
supersedeOutranked(director, seats!.keys())
|
||||
} else if (current && current.incarnation === snapshot.incarnation) {
|
||||
for (const change of changes) applyChange(director, cell, current, change)
|
||||
current.cursor = snapshot.cursor
|
||||
current.placement = placementCell(cell, snapshot, requestedAt)
|
||||
supersedeOutranked(
|
||||
director,
|
||||
changes.filter((change) => change.kind === 'join').map((change) => change.host)
|
||||
)
|
||||
} else {
|
||||
// A restarted director's (or a stale) delta: the next poll asks for a full snapshot.
|
||||
return
|
||||
@@ -758,6 +766,26 @@ export async function runReservePlacementSimulation(
|
||||
}
|
||||
}
|
||||
|
||||
// As shadow-seat-directory.ts onOutranked: once both joins are in this map, a reserve seat behind
|
||||
// a later join at a newer epoch is told to go. A lower epoch that joined later is the row's.
|
||||
function supersedeOutranked(director: SimDirector, hosts: Iterable<string>): void {
|
||||
if (faults.skipSupersede || faults.skipFeedSupersede) return
|
||||
for (const host of new Set(hosts)) {
|
||||
const seats = [...director.views].flatMap(([cellId, view]) => {
|
||||
const seat = view.seats.get(host)
|
||||
return seat ? [{ cellId, view, seat }] : []
|
||||
})
|
||||
if (seats.length < 2) continue
|
||||
for (const { cellId, view, seat } of seats) {
|
||||
if (seat.demoted || !view.placement.reserve) continue
|
||||
const by = seats.find((other) => seatOutranks(other.seat, seat))
|
||||
if (!by) continue
|
||||
report.supersedes += 1
|
||||
demoteFor(director, host, cellId, seat, { cellId: by.cellId, epoch: by.seat.epoch })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The seats the booking director's map held when it placed each booking: only those it supersedes.
|
||||
const heldAtBooking = new Map<string, Set<string>>()
|
||||
|
||||
@@ -1258,9 +1286,9 @@ export async function runReservePlacementSimulation(
|
||||
report.maxOccupancyShare[cell.cellId] = Math.max(report.maxOccupancyShare[cell.cellId] ?? 0, share)
|
||||
}
|
||||
}
|
||||
// A booking supersedes the old seats its director's map held within a round trip. Any other
|
||||
// duplicate (a half-open seat today's path placed past, or one the booking map had not yet
|
||||
// seen) has no lease to re-check and lasts until that seat's silence close.
|
||||
// A booking supersedes the old seats its director's map held within a round trip, and the feed
|
||||
// any seat a later join outranks. Any other duplicate (a half-open seat today's path placed past
|
||||
// at a lower epoch than it) has no lease to re-check and lasts until that seat's silence close.
|
||||
const supersedeBound =
|
||||
Math.max(...directors.map((director) => config.pollLagMs(director.index))) +
|
||||
2 * Math.max(...cells.map((cell) => config.rttMs(cell.region))) +
|
||||
@@ -1293,13 +1321,20 @@ export async function runReservePlacementSimulation(
|
||||
for (const { cell, seat } of seats) {
|
||||
if (cell.mode !== 'reserve' || seat.epoch === newest) continue
|
||||
// A leased seat goes by its renewal, as today.
|
||||
const bound = !seat.leased && supersededBy(cell.cellId, seat.joinedAt) ? supersedeBound : duplicateBound
|
||||
const tight =
|
||||
!seat.leased &&
|
||||
(supersededBy(cell.cellId, seat.joinedAt) || seats.some((other) => seatOutranks(other.seat, seat)))
|
||||
// The loose clock runs from the first outranking; the tight one from when supersede applies.
|
||||
const key = `${host}\u0000${cell.cellId}\u0000${seat.joinedAt}`
|
||||
live.add(key)
|
||||
const since = outrankedSince.get(key) ?? clock.now
|
||||
outrankedSince.set(key, since)
|
||||
if (clock.now - since > bound) {
|
||||
violate(3, `${host}@${seat.epoch} on ${cell.cellId} outranked for ${clock.now - since} ms`)
|
||||
const clocks: Array<[string, number]> = [[key, duplicateBound]]
|
||||
if (tight) clocks.push([`${key}\u0000tight`, supersedeBound])
|
||||
for (const [clockKey, bound] of clocks) {
|
||||
live.add(clockKey)
|
||||
const since = outrankedSince.get(clockKey) ?? clock.now
|
||||
outrankedSince.set(clockKey, since)
|
||||
if (clock.now - since > bound) {
|
||||
violate(3, `${host}@${seat.epoch} on ${cell.cellId} outranked for ${clock.now - since} ms`)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ import { describe, expect, it, vi } from 'vitest'
|
||||
import { RelayAssignmentStore } from './assignment-store.js'
|
||||
import { openInMemoryRelayDatabase } from './database.js'
|
||||
import {
|
||||
SHADOW_SEAT_OUTRANK_SKEW_MS,
|
||||
SHADOW_SEAT_RECENTLY_LEFT_MAX_HOSTS,
|
||||
ShadowSeatDirectory,
|
||||
startShadowSeatPoller,
|
||||
@@ -577,3 +578,76 @@ describe('ShadowSeatDirectory.placementModeOf', () => {
|
||||
expect(directory.placementModeOf('cell-a')).toBe('reserve')
|
||||
})
|
||||
})
|
||||
|
||||
describe('ShadowSeatDirectory.onOutranked', () => {
|
||||
const reserve = { generation: 3, flags: { admitMode: 'reserve' } }
|
||||
const db = { generation: 3, flags: { admitMode: 'db' } }
|
||||
const full = (cellId: string, epoch: number, joinedAt: number, flags = reserve, state = 'active') =>
|
||||
feed({
|
||||
cellId,
|
||||
seq: 1,
|
||||
flagsApplied: flags,
|
||||
full: [{ userId: 'user-a', relayHostId: 'host-1', epoch, generation: 1, state, joinedAt }]
|
||||
})
|
||||
const directory = () => {
|
||||
const map = new ShadowSeatDirectory()
|
||||
map.setCells(['cell-a', 'cell-b'])
|
||||
const outranked: unknown[] = []
|
||||
map.onOutranked = (seat) => outranked.push(seat)
|
||||
return { map, outranked }
|
||||
}
|
||||
const halfOpen = {
|
||||
cellId: 'cell-a',
|
||||
userId: 'user-a',
|
||||
relayHostId: 'host-1',
|
||||
epoch: 3,
|
||||
joinedAt: 10,
|
||||
byCellId: 'cell-b'
|
||||
}
|
||||
|
||||
it('reports a reserve seat a later, newer join elsewhere outranks, in either feed order', () => {
|
||||
const first = directory()
|
||||
first.map.apply('cell-a', full('cell-a', 3, 10), 100)
|
||||
first.map.apply('cell-b', full('cell-b', 4, 2_000), 101)
|
||||
expect(first.outranked).toEqual([halfOpen])
|
||||
const reversed = directory()
|
||||
reversed.map.apply('cell-b', full('cell-b', 4, 2_000), 100)
|
||||
reversed.map.apply('cell-a', full('cell-a', 3, 10), 101)
|
||||
expect(reversed.outranked).toEqual([halfOpen])
|
||||
const justPastSkew = directory()
|
||||
justPastSkew.map.apply('cell-a', full('cell-a', 3, 10), 100)
|
||||
justPastSkew.map.apply('cell-b', full('cell-b', 4, 11 + SHADOW_SEAT_OUTRANK_SKEW_MS), 101)
|
||||
expect(justPastSkew.outranked).toEqual([halfOpen])
|
||||
})
|
||||
|
||||
it('reports a seat once, even when a full resync re-judges its host', () => {
|
||||
const { map, outranked } = directory()
|
||||
map.apply('cell-a', full('cell-a', 3, 10), 100)
|
||||
map.apply('cell-b', full('cell-b', 4, 2_000), 101)
|
||||
map.apply('cell-b', { ...full('cell-b', 4, 2_000), incarnation: '22222222-2222-4222-8222-222222222222' }, 102)
|
||||
map.apply('cell-a', { ...full('cell-a', 3, 10), incarnation: '22222222-2222-4222-8222-222222222222' }, 103)
|
||||
expect(outranked).toEqual([halfOpen])
|
||||
})
|
||||
|
||||
it('leaves a database-cell seat, a drain-only seat, a later lower epoch, and a join within the skew', () => {
|
||||
const onDatabaseCell = directory()
|
||||
onDatabaseCell.map.apply('cell-a', full('cell-a', 3, 10, db), 100)
|
||||
onDatabaseCell.map.apply('cell-b', full('cell-b', 4, 2_000), 101)
|
||||
const drainOnly = directory()
|
||||
drainOnly.map.apply('cell-a', full('cell-a', 3, 10, reserve, 'drain-only'), 100)
|
||||
drainOnly.map.apply('cell-b', full('cell-b', 4, 2_000), 101)
|
||||
// Answered the old seat from a stale map after the newer join: the row decides that one.
|
||||
const staleSticky = directory()
|
||||
staleSticky.map.apply('cell-a', full('cell-a', 3, 3_000), 100)
|
||||
staleSticky.map.apply('cell-b', full('cell-b', 4, 2_000), 101)
|
||||
const withinSkew = directory()
|
||||
withinSkew.map.apply('cell-a', full('cell-a', 3, 10), 100)
|
||||
withinSkew.map.apply('cell-b', full('cell-b', 4, 10 + SHADOW_SEAT_OUTRANK_SKEW_MS), 101)
|
||||
expect([onDatabaseCell, drainOnly, staleSticky, withinSkew].map(({ outranked }) => outranked)).toEqual([
|
||||
[],
|
||||
[],
|
||||
[],
|
||||
[]
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -16,6 +16,16 @@ export const SHADOW_SEAT_POLL_TIMEOUT_MS = 2_000
|
||||
export const SHADOW_SEAT_CELL_LIST_REFRESH_MS = 30_000
|
||||
export const SHADOW_SEAT_RECENTLY_LEFT_TTL_MS = 24 * 60 * 60 * 1_000
|
||||
export const SHADOW_SEAT_RECENTLY_LEFT_MAX_HOSTS = 100_000
|
||||
// Cells stamp joinedAt by their own clocks: a later join must clear this to outrank across cells.
|
||||
export const SHADOW_SEAT_OUTRANK_SKEW_MS = 1_000
|
||||
const SHADOW_SEAT_OUTRANKED_REPORTS_MAX = 100_000
|
||||
|
||||
export function seatOutranks(
|
||||
newer: { epoch: number; joinedAt: number },
|
||||
seat: { epoch: number; joinedAt: number }
|
||||
): boolean {
|
||||
return newer.epoch > seat.epoch && newer.joinedAt > seat.joinedAt + SHADOW_SEAT_OUTRANK_SKEW_MS
|
||||
}
|
||||
const RECENTLY_LEFT_PER_HOST = 4
|
||||
// Google identity tokens live an hour; one per poll would be 26 metadata reads a second.
|
||||
const IDENTITY_TOKEN_REUSE_MS = 10 * 60_000
|
||||
@@ -172,6 +182,16 @@ export type BookedJoin = {
|
||||
reservedBy: string
|
||||
}
|
||||
|
||||
// A reserve seat behind a later, newer-epoch join of its host elsewhere that placement supersede missed.
|
||||
export type OutrankedSeat = {
|
||||
cellId: string
|
||||
userId: string
|
||||
relayHostId: string
|
||||
epoch: number
|
||||
joinedAt: number
|
||||
byCellId: string
|
||||
}
|
||||
|
||||
type CellCursor = SeatFeedCellState & { seats: Map<string, ShadowSeat> }
|
||||
|
||||
export type HeartbeatSnapshot = { readAt: number; expiresAt: ReadonlyMap<string, number> }
|
||||
@@ -180,6 +200,11 @@ function hostKey(userId: string, relayHostId: string): string {
|
||||
return `${userId}\u0000${relayHostId}`
|
||||
}
|
||||
|
||||
function parseHostKey(key: string): { userId: string; relayHostId: string } {
|
||||
const separator = key.indexOf('\u0000')
|
||||
return { userId: key.slice(0, separator), relayHostId: key.slice(separator + 1) }
|
||||
}
|
||||
|
||||
export class ShadowSeatDirectory {
|
||||
private readonly cells = new Map<string, CellCursor>()
|
||||
private readonly hosts = new Map<string, Map<string, ShadowSeat>>()
|
||||
@@ -193,6 +218,10 @@ export class ShadowSeatDirectory {
|
||||
// Per host, the newest booking any cell reported (step 5 epoch mint). Age-ordered.
|
||||
private readonly bookings = new Map<string, { cellId: string; epoch: number; at: number }>()
|
||||
onBookedJoin?: (join: BookedJoin) => void
|
||||
onOutranked?: (seat: OutrankedSeat) => void
|
||||
private readonly joinedSinceApply = new Set<string>()
|
||||
// Seats already reported, insertion-ordered: a full resync re-judges a host without a second demote.
|
||||
private readonly reportedOutranked = new Set<string>()
|
||||
|
||||
// Cells no longer listed are forgotten with their seats; new ones start pending.
|
||||
// `required` defaults to every listed cell.
|
||||
@@ -225,6 +254,11 @@ export class ShadowSeatDirectory {
|
||||
|
||||
// `polledAt` is when the request left; the reply holds every change before it.
|
||||
apply(cellId: string, response: SeatFeedResponse, now: number, polledAt = now): void {
|
||||
this.applyResponse(cellId, response, now, polledAt)
|
||||
this.reportOutranked()
|
||||
}
|
||||
|
||||
private applyResponse(cellId: string, response: SeatFeedResponse, now: number, polledAt: number): void {
|
||||
const cursor = this.cells.get(cellId)
|
||||
if (!cursor) return
|
||||
if (response.cellId !== cellId) {
|
||||
@@ -307,10 +341,7 @@ export class ShadowSeatDirectory {
|
||||
seatsOnCell(cellId: string): Array<{ userId: string; relayHostId: string; seat: ShadowSeat }> {
|
||||
const cursor = this.cells.get(cellId)
|
||||
if (!cursor) return []
|
||||
return [...cursor.seats].map(([key, seat]) => {
|
||||
const separator = key.indexOf('\u0000')
|
||||
return { userId: key.slice(0, separator), relayHostId: key.slice(separator + 1), seat }
|
||||
})
|
||||
return [...cursor.seats].map(([key, seat]) => ({ ...parseHostKey(key), seat }))
|
||||
}
|
||||
|
||||
// An old cell (404) has no feed; its last known seats stay, marked by status.
|
||||
@@ -558,6 +589,34 @@ export class ShadowSeatDirectory {
|
||||
this.hosts.set(key, seats)
|
||||
}
|
||||
seats.set(cursor.cellId, seat)
|
||||
if (this.onOutranked && seats.size > 1) this.joinedSinceApply.add(key)
|
||||
}
|
||||
|
||||
// After the whole response, so a cell's first snapshot is judged by the mode it carries. Only a
|
||||
// join triggers it, once per seat: a cell flipped to reserve later, or a lost demote, falls to
|
||||
// the silence close or the ledger reconcile, as does a lower epoch that joined later.
|
||||
private reportOutranked(): void {
|
||||
const report = this.onOutranked
|
||||
const keys = [...this.joinedSinceApply]
|
||||
this.joinedSinceApply.clear()
|
||||
if (!report) return
|
||||
for (const key of keys) {
|
||||
const seats = [...(this.hosts.get(key)?.values() ?? [])]
|
||||
for (const seat of seats) {
|
||||
if (seat.state !== 'active' || this.admitModeOf(seat.cellId) !== 'reserve') continue
|
||||
const by = seats.find((other) => seatOutranks(other, seat))
|
||||
if (!by) continue
|
||||
const { cellId, epoch, joinedAt } = seat
|
||||
const reported = `${key}\u0000${cellId}\u0000${epoch}\u0000${joinedAt}`
|
||||
if (this.reportedOutranked.has(reported)) continue
|
||||
this.reportedOutranked.add(reported)
|
||||
if (this.reportedOutranked.size > SHADOW_SEAT_OUTRANKED_REPORTS_MAX) {
|
||||
const oldest = this.reportedOutranked.values().next().value
|
||||
if (oldest !== undefined) this.reportedOutranked.delete(oldest)
|
||||
}
|
||||
report({ ...parseHostKey(key), cellId, epoch, joinedAt, byCellId: by.cellId })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private unseat(cursor: CellCursor, key: string): void {
|
||||
|
||||
Reference in New Issue
Block a user