openvibers/openvibe.games Commits
Recover instance leases after database errors, and restart a dead tick loop (#281)
Changes
@@@ -1,5 +1,5 @@import { encodeServerMessage, type ClientHello, type ClientMessage } from '@openvibe/protocol'--import type { InstanceLease, InstanceRegistry } from '@openvibe/persistence'++import type { InstanceLease, InstanceRegistry, LeaseHeartbeat } from '@openvibe/persistence'import { FixedTimestep, type Logger } from '@openvibe/shared'import type { GameConnection } from './systems/context.js'import type { GameServer } from './gameServer.js'@@@ -12,9 +12,64 @@ export interface InstanceRoom {dispose(): void | Promise<void>}++/**++ * The row lease's lifetime. The heartbeat runs every 5 s (main.ts): three missed renewals and another++ * process may claim the slot.++ */++export const LEASE_TTL_MS = 15_000++++/**++ * How long a room whose lease was lost to a database error keeps its world and players while it tries++ * to regain the lease before it is stopped. 60 s rides out a connection timeout or a PgBouncer++ * failover (production saw a heartbeat UPDATE wait 8.7 s), and is short enough that a frozen room does++ * not hold a slot, its players and its memory for minutes. A database outage longer than this is++ * better served by the process restarting (the tick watchdog in main.ts).++ */++export const LEASE_RECOVERY_GRACE_MS = 60_000++++/**++ * First backoff between regain attempts. Retrying on every 5 s heartbeat would hammer a database that++ * is already timing out; 2 s puts the first retry on the next heartbeat.++ */++const LEASE_REGAIN_INITIAL_BACKOFF_MS = 2_000++++/** Regain retries double to this cap: a short outage is caught on the first or second retry. */++const LEASE_REGAIN_MAX_BACKOFF_MS = 30_000++++/**++ * After a room is stopped for good its slot is claimed again through the normal start path, even with++ * nobody waiting. The first retry comes one heartbeat later; the backoff then doubles to a minute so a++ * slot another process legitimately owns is only polled cheaply.++ */++const ROOM_RESTART_INITIAL_BACKOFF_MS = 5_000++const ROOM_RESTART_MAX_BACKOFF_MS = 60_000++++/** The same experience as a deploy restart: clients reload and reconnect when /healthz answers again. */++const RECONNECT_MESSAGE = '🔄 The world is restarting: you will reconnect in a moment…'++++/** A room whose lease could not be confirmed: it is neither simulated nor admitted while it retries. */++interface RecoveringRoom {++ room: InstanceRoom++ /** When the first heartbeat could not confirm the lease. */++ since: number++ attempts: number++ nextAttemptAt: number++}++++/** A stopped slot the hub is claiming again through the normal start path. */++interface RestartAttempt {++ attempts: number++ nextAttemptAt: number++ inFlight: boolean++}++/** One-process room router. The socket's upgrade identity remains attached across transfers. */export class InstanceHub {private readonly rooms = new Map<number, InstanceRoom>()++ /** Rooms whose lease is in doubt. They keep their world and players, but not the tick or admissions. */++ private readonly recovering = new Map<number, RecoveringRoom>()++ /** Slots stopped for good that are being claimed again with backoff. */++ private readonly restarts = new Map<number, RestartAttempt>()/** One accumulator per room: 100 Hz surf and 30 Hz everything else can share a hub. */private readonly clocks = new Map<number, FixedTimestep>()private readonly connections = new Map<GameConnection, { slot: number; hello: ClientHello }>()@@@ -84,6 +139,12 @@ export class InstanceHub {releaseAdmission()}if (!room || this.closed.has(conn)) {++ // A slot being regained cannot take players: close like a restart so the client retries++ // instead of reading "server full" and giving up on this place.++ if (!this.closed.has(conn) && this.recovering.size > 0) {++ conn.close(1012, 'instance_recovering')++ return++ }conn.send(encodeServerMessage({ t: 'reject', reason: 'server_full' }))conn.close(4003, 'server_full')return@@@ -136,6 +197,8 @@ export class InstanceHub {** Without an argument this keeps the old test/manual contract: exactly one* step per room per call, regardless of its rate.++ *++ * Rooms whose lease is in doubt are not in `rooms`: a place that cannot save must not simulate.*/step(elapsedSeconds?: number): void {if (elapsedSeconds === undefined) {@@@ -158,41 +221,217 @@ export class InstanceHub {if (this.heartbeating || this.stopping) returnthis.heartbeating = truetry {-- for (const room of [...this.rooms.values()]) {-- let renewed: InstanceLease | null = null-- try {-- renewed = await this.registry.renew(room.lease, now, 15_000, 'running')-- } catch (error) {-- this.log.error('instance heartbeat failed', { id: room.lease.id, error: String(error) })-- }-- if (renewed) {-- room.lease = renewed-- continue-- }-- this.log.error('instance lease lost', { id: room.lease.id })-- this.rooms.delete(room.lease.slot)-- this.clocks.delete(room.lease.slot)-- for (const [conn, current] of this.connections) {-- if (current.slot === room.lease.slot) conn.close(1012, 'instance_lease_lost')-- }-- await room.dispose()-- }++ // Every room heartbeats at once: one slow query (bounded by the registry's own statement++ // timeout) must not delay the next room's renewal past its deadline.++ await Promise.all([++ ...[...this.rooms.values()].map((room) => this.renewRoom(room, now)),++ ...[...this.recovering.values()].map((rec) => this.regainRoom(rec, now)),++ ])++ // Starting a room can take seconds (world restore); it must never keep the lease renewals++ // above waiting, so restarts run beside the heartbeat and schedule themselves.++ for (const [slot, attempt] of this.restarts) this.runRestart(slot, attempt, now)} finally {this.heartbeating = false}}++ private async renewRoom(room: InstanceRoom, now: number): Promise<void> {++ if (this.stopping) return++ let result: LeaseHeartbeat | null++ try {++ result = await this.registry.heartbeat(room.lease, now, LEASE_TTL_MS, 'running')++ } catch (error) {++ // The lease may still be ours, but nothing may touch the world until that is confirmed.++ this.log.error('instance heartbeat failed', { id: room.lease.id, error: String(error) })++ this.suspend(room, now)++ return++ }++ if (result) {++ room.lease = result.lease++ if (result.regained) this.log.info('instance lease regained', { id: room.lease.id })++ return++ }++ // The row no longer carries this holder and version: another process owns it. Nothing can be++ // regained, so stop at once instead of simulating a place this process cannot save.++ this.log.error('instance lease lost to another holder', { id: room.lease.id })++ await this.stopRoom(room, 'instance_lease_lost')++ }++++ /** Stop the tick and admissions, keep the players, and retry the lease with backoff. */++ private suspend(room: InstanceRoom, now: number): void {++ const slot = room.lease.slot++ if (this.recovering.has(slot)) return++ this.rooms.delete(slot)++ this.clocks.delete(slot)++ this.recovering.set(slot, {++ room,++ since: now,++ attempts: 0,++ nextAttemptAt: now + LEASE_REGAIN_INITIAL_BACKOFF_MS,++ })++ this.log.error('instance lease lost', { id: room.lease.id, graceMs: LEASE_RECOVERY_GRACE_MS })++ }++++ /** One regain attempt inside the grace; a success puts the room back into the tick. */++ private async regainRoom(rec: RecoveringRoom, now: number): Promise<void> {++ if (this.stopping) return++ const { room } = rec++ const slot = room.lease.slot++ if (now - rec.since >= LEASE_RECOVERY_GRACE_MS) {++ this.recovering.delete(slot)++ this.log.error('instance lease not regained in time: stopping room', {++ id: room.lease.id,++ graceMs: LEASE_RECOVERY_GRACE_MS,++ })++ await this.stopRoom(room, 'instance_lease_lost')++ this.scheduleRestart(slot, now)++ return++ }++ if (now < rec.nextAttemptAt) return++ let result: LeaseHeartbeat | null++ try {++ result = await this.registry.heartbeat(room.lease, now, LEASE_TTL_MS, 'running')++ } catch (error) {++ this.log.error('instance lease regain failed', { id: room.lease.id, error: String(error) })++ this.backoffRegain(rec, now)++ return++ }++ if (!result) {++ // Another holder claimed the row while the database was unreachable: this room cannot return.++ this.recovering.delete(slot)++ this.log.error('instance lease lost to another holder', { id: room.lease.id })++ await this.stopRoom(room, 'instance_lease_lost')++ this.scheduleRestart(slot, now)++ return++ }++ room.lease = result.lease++ this.recovering.delete(slot)++ this.rooms.set(slot, room)++ this.log.info('instance lease regained', {++ id: room.lease.id,++ afterMs: now - rec.since,++ reclaimed: result.regained,++ })++ }++++ private backoffRegain(rec: RecoveringRoom, now: number): void {++ rec.attempts++++ rec.nextAttemptAt =++ now +++ Math.min(++ LEASE_REGAIN_INITIAL_BACKOFF_MS * 2 ** (rec.attempts - 1),++ LEASE_REGAIN_MAX_BACKOFF_MS,++ )++ }++++ private scheduleRestart(slot: number, now: number): void {++ if (this.stopping) return++ this.restarts.set(slot, {++ attempts: 0,++ nextAttemptAt: now + ROOM_RESTART_INITIAL_BACKOFF_MS,++ inFlight: false,++ })++ }++++ private runRestart(slot: number, attempt: RestartAttempt, now: number): void {++ if (this.stopping || attempt.inFlight || now < attempt.nextAttemptAt) return++ if (this.rooms.has(slot) || this.recovering.has(slot)) {++ this.restarts.delete(slot)++ return++ }++ attempt.inFlight = true++ void this.start(slot)++ .then((room) => {++ if (room) {++ this.restarts.delete(slot)++ this.log.info('instance restarted', { id: room.lease.id })++ return++ }++ this.backoffRestart(attempt)++ })++ .catch((error: unknown) => {++ this.log.error('instance restart failed', { slot, error: String(error) })++ this.backoffRestart(attempt)++ })++ .finally(() => {++ attempt.inFlight = false++ })++ }++++ /** Back off from the moment the attempt ended: world restore can take seconds. */++ private backoffRestart(attempt: RestartAttempt): void {++ attempt.attempts++++ attempt.nextAttemptAt =++ Date.now() +++ Math.min(++ ROOM_RESTART_INITIAL_BACKOFF_MS * 2 ** (attempt.attempts - 1),++ ROOM_RESTART_MAX_BACKOFF_MS,++ )++ }++++ /** Announce the reconnect, close the room's players with 1012 and release its memory. */++ private async stopRoom(room: InstanceRoom, reason: string): Promise<void> {++ const slot = room.lease.slot++ this.rooms.delete(slot)++ this.clocks.delete(slot)++ room.game.announce(RECONNECT_MESSAGE)++ for (const [conn, current] of [...this.connections]) {++ if (current.slot !== slot) continue++ this.connections.delete(conn)++ conn.close(1012, reason)++ }++ try {++ await room.dispose()++ } catch (error) {++ // The room is already gone from every map; a failing dispose must not abort the heartbeat.++ this.log.warn('instance room dispose failed', { id: room.lease.id, error: String(error) })++ }++ }++async shutdown(): Promise<boolean> {this.stopping = true-- for (const room of this.rooms.values()) {-- const draining = await this.registry.renew(room.lease, Date.now(), 15_000, 'draining')-- if (draining) room.lease = draining++ const rooms = [...this.rooms.values(), ...[...this.recovering.values()].map((rec) => rec.room)]++ this.recovering.clear()++ this.restarts.clear()++ const draining = new Set<InstanceRoom>()++ for (const room of rooms) {++ try {++ const renewed = await this.registry.renew(room.lease, Date.now(), LEASE_TTL_MS, 'draining')++ if (renewed) {++ room.lease = renewed++ draining.add(room)++ }++ } catch (error) {++ // A database too slow to answer must not make the shutdown throw before the sockets close.++ this.log.warn('instance lease renewal failed during shutdown', {++ id: room.lease.id,++ error: String(error),++ })++ }}-- for (const room of this.rooms.values()) room.game.shutdown()-- const drained = await Promise.all([...this.rooms.values()].map((room) => room.game.drain()))-- for (const room of this.rooms.values()) {-- await this.registry.stop(room.lease)-- await room.dispose()++ // A room whose lease cannot be renewed is fenced: its writes would be refused, so do not wait++ // through the write-behind retries. It is disposed below like the rest.++ for (const room of draining) room.game.shutdown()++ const drained = await Promise.all([...draining].map((room) => room.game.drain()))++ for (const room of rooms) {++ if (draining.has(room)) {++ try {++ await this.registry.stop(room.lease)++ } catch (error) {++ this.log.warn('instance lease stop failed during shutdown', {++ id: room.lease.id,++ error: String(error),++ })++ }++ }++ try {++ await room.dispose()++ } catch (error) {++ this.log.warn('instance room dispose failed during shutdown', {++ id: room.lease.id,++ error: String(error),++ })++ }}this.rooms.clear()this.clocks.clear()@@@ -205,7 +444,8 @@ export class InstanceHub {return room}for (let slot = 0; slot < this.maxInstances; slot++) {-- if (!this.rooms.has(slot)) {++ // A recovering slot still owns its row: claiming it now would fence the room we are trying to save.++ if (!this.rooms.has(slot) && !this.recovering.has(slot)) {const room = await this.start(slot)if (room) return room}@@@ -222,12 +462,14 @@ export class InstanceHub {private async start(slot: number): Promise<InstanceRoom | null> {let result: InstanceRoom | null = nullconst prior = this.creating-- this.creating = prior.then(async () => {++ const run = prior.then(async () => {const existing = this.rooms.get(slot)if (existing) {result = existingreturn}++ if (this.recovering.has(slot)) return++ if (this.stopping) returnconst lease = await this.registry.claim(slot,this.capacity,@@@ -238,17 +480,43 @@ export class InstanceHub {if (!lease) returntry {const room = await this.createRoom(slot, lease)-- const running = await this.registry.renew(lease, Date.now(), 15_000, 'running')++ if (this.stopping) {++ // Shutdown already snapshotted the rooms: do not leave one running (and leased) behind.++ try {++ await this.registry.stop(lease)++ } catch (stopError) {++ this.log.warn('instance lease stop failed for a room started during shutdown', {++ id: lease.id,++ error: String(stopError),++ })++ }++ await room.dispose()++ return++ }++ const running = await this.registry.renew(lease, Date.now(), LEASE_TTL_MS, 'running')if (!running) throw new Error('lease expired during room start')room.lease = runningthis.rooms.set(slot, room)result = room} catch (error) {-- await this.registry.stop(lease)++ try {++ await this.registry.stop(lease)++ } catch (stopError) {++ this.log.warn('instance lease stop failed after a room start failure', {++ id: lease.id,++ error: String(stopError),++ })++ }throw error}})-- await this.creating++ // One failed start must not poison every later one: the restart backoff and the next join both++ // call start again, and the serialization chain has to keep running after this caller's error.++ this.creating = run.then(++ () => undefined,++ () => undefined,++ )++ await runreturn result}nnew file mode 100644@@@ -0,0 +1,351 @@++import type { InstanceLease, InstanceRegistry, LeaseHeartbeat } from '@openvibe/persistence'++import { PROTOCOL_VERSION, defaultAppearance, type ClientHello } from '@openvibe/protocol'++import type { LogFields, Logger } from '@openvibe/shared'++import { describe, expect, it } from 'vitest'++import type { GameServer } from './gameServer.js'++import { InstanceHub, LEASE_RECOVERY_GRACE_MS, type InstanceRoom } from './instanceHub.js'++import type { GameConnection } from './systems/context.js'++++function makeLease(overrides: Partial<InstanceLease> = {}): InstanceLease {++ return {++ id: 'lease-test:0',++ placeId: 'lease-test',++ slot: 0,++ capacity: 1,++ state: 'running',++ holder: 'holder-a',++ version: 1,++ heartbeatAt: 0,++ leaseUntil: 15_000,++ ...overrides,++ }++}++++/** The registry surface InstanceHub uses; the database behaviour is scripted per test. */++class FakeRegistry {++ readonly heartbeatCalls: number[] = []++ readonly claimCalls: number[] = []++ readonly stopCalls: string[] = []++ error: Error | null = null++ /** 'renew' | 'regain' | 'lost' */++ outcome: 'renew' | 'regain' | 'lost' = 'renew'++ claimError: Error | null = null++ claimResult: InstanceLease | null = null++++ async heartbeat(current: InstanceLease, now: number): Promise<LeaseHeartbeat | null> {++ this.heartbeatCalls.push(now)++ if (this.error) throw this.error++ if (this.outcome === 'lost') return null++ return {++ lease: { ...current, heartbeatAt: now, leaseUntil: now + 15_000 },++ regained: this.outcome === 'regain',++ }++ }++++ async claim(slot: number): Promise<InstanceLease | null> {++ this.claimCalls.push(slot)++ if (this.claimError) throw this.claimError++ return this.claimResult++ }++++ async renew(current: InstanceLease): Promise<InstanceLease | null> {++ return current++ }++++ async stop(current: InstanceLease): Promise<boolean> {++ this.stopCalls.push(current.id)++ return true++ }++}++++interface RoomProbe {++ room: InstanceRoom++ steps: number++ joins: number++ announced: string[]++ disposed: number++}++++function makeRoom(lease: InstanceLease): RoomProbe {++ const probe: RoomProbe = {++ room: null as unknown as InstanceRoom,++ steps: 0,++ joins: 0,++ announced: [],++ disposed: 0,++ }++ const game = {++ tickRate: 30,++ step: () => probe.steps++,++ join: async () => {++ probe.joins++++ return true++ },++ onMessage: () => {},++ onDisconnect: () => {},++ checkpoint: () => {},++ drain: async () => true,++ shutdown: () => {},++ announce: (text: string) => probe.announced.push(text),++ contentDigest: 'digest',++ onlineCount: () => 0,++ }++ probe.room = {++ lease,++ game: game as unknown as GameServer,++ world: {} as InstanceRoom['world'],++ dispose: () => {++ probe.disposed++++ },++ }++ return probe++}++++function makeLogger(): {++ log: Logger++ entries: { level: string; msg: string; fields?: LogFields }[]++} {++ const entries: { level: string; msg: string; fields?: LogFields }[] = []++ const log: Logger = {++ debug: (msg, fields) => entries.push({ level: 'debug', msg, ...(fields ? { fields } : {}) }),++ info: (msg, fields) => entries.push({ level: 'info', msg, ...(fields ? { fields } : {}) }),++ warn: (msg, fields) => entries.push({ level: 'warn', msg, ...(fields ? { fields } : {}) }),++ error: (msg, fields) => entries.push({ level: 'error', msg, ...(fields ? { fields } : {}) }),++ child: () => log,++ }++ return { log, entries }++}++++function makeConnection(++ hub: InstanceHub,++ closes: { code: number; reason: string }[],++): GameConnection {++ const conn: GameConnection = {++ identity: { guestKeyHash: 'a'.repeat(64) },++ send: () => {},++ close: (code, reason) => {++ closes.push({ code, reason })++ hub.onDisconnect(conn)++ },++ }++ return conn++}++++function hello(): ClientHello {++ return {++ t: 'hello',++ v: PROTOCOL_VERSION,++ contentDigest: 'digest',++ slot: 0,++ name: 'guest',++ appearance: defaultAppearance(),++ }++}++++function makeHub(registry: FakeRegistry, room: InstanceRoom, log: Logger): InstanceHub {++ return new InstanceHub(++ registry as unknown as InstanceRegistry,++ room,++ 1,++ 1,++ async () => {++ throw new Error('unexpected room creation')++ },++ log,++ )++}++++async function until(check: () => boolean): Promise<void> {++ for (let i = 0; i < 200; i++) {++ if (check()) return++ await new Promise((resolve) => setTimeout(resolve, 5))++ }++ throw new Error('condition not reached')++}++++describe('instance lease recovery', () => {++ it('stops the tick and admissions while the lease is unconfirmed, then regains it with backoff', async () => {++ const lease = makeLease()++ const registry = new FakeRegistry()++ const probe = makeRoom(lease)++ const { log, entries } = makeLogger()++ const hub = makeHub(registry, probe.room, log)++ const closes: { code: number; reason: string }[] = []++ const conn = makeConnection(hub, closes)++++ // One database error: the heartbeat can no longer confirm the lease.++ registry.error = new Error('Query read timeout')++ await hub.heartbeat(1_000)++ expect(hub.activeRooms).toHaveLength(0)++ expect(entries.some((e) => e.msg === 'instance heartbeat failed')).toBe(true)++ expect(entries.some((e) => e.msg === 'instance lease lost')).toBe(true)++++ // No simulation while the lease is in doubt.++ hub.step()++ expect(probe.steps).toBe(0)++++ // Not accepting players: the client is told to reconnect (1012), not that the server is full.++ hub.onMessage(conn, hello())++ await hub.waitForConnection(conn)++ expect(closes).toEqual([{ code: 1012, reason: 'instance_recovering' }])++ expect(probe.joins).toBe(0)++++ // The first regain attempt is scheduled 2 s after the loss; a heartbeat before that does nothing.++ await hub.heartbeat(2_000)++ expect(registry.heartbeatCalls).toEqual([1_000])++ // The database answers again; the row is still this holder's, so the room comes back.++ registry.error = null++ registry.outcome = 'regain'++ await hub.heartbeat(3_100)++ expect(entries.some((e) => e.msg === 'instance lease regained')).toBe(true)++ expect(hub.activeRooms).toHaveLength(1)++++ // The tick resumes and the regained room takes players.++ hub.step()++ expect(probe.steps).toBe(1)++ const rejoining = makeConnection(hub, [])++ hub.onMessage(rejoining, hello())++ await hub.waitForConnection(rejoining)++ expect(probe.joins).toBe(1)++ })++++ it('stops a room that cannot regain its lease inside the grace and starts it again on the next join', async () => {++ const lease = makeLease()++ const registry = new FakeRegistry()++ const probe = makeRoom(lease)++ const { log, entries } = makeLogger()++ let created = 0++ const fresh = makeRoom(makeLease({ holder: 'holder-b', version: 2, leaseUntil: 30_000 }))++ const hub = new InstanceHub(++ registry as unknown as InstanceRegistry,++ probe.room,++ 1,++ 1,++ async () => {++ created++++ return fresh.room++ },++ log,++ )++ const closes: { code: number; reason: string }[] = []++ const connected = makeConnection(hub, closes)++ hub.onMessage(connected, hello())++ await hub.waitForConnection(connected)++ expect(probe.joins).toBe(1)++++ registry.error = new Error('Connection terminated due to connection timeout')++ await hub.heartbeat(10_000)++ expect(hub.activeRooms).toHaveLength(0)++++ // The grace expires: the old room announces a reconnect, closes its players with 1012 and is disposed.++ await hub.heartbeat(10_000 + LEASE_RECOVERY_GRACE_MS)++ expect(++ entries.some((e) => e.msg === 'instance lease not regained in time: stopping room'),++ ).toBe(true)++ expect(closes).toEqual([{ code: 1012, reason: 'instance_lease_lost' }])++ expect(probe.announced[0]).toMatch(/reconnect/i)++ expect(probe.disposed).toBe(1)++ expect(probe.steps).toBe(0)++++ // The next join claims the freed slot through the normal start path and serves a fresh room.++ registry.error = null++ registry.claimResult = fresh.room.lease++ const next = makeConnection(hub, [])++ hub.onMessage(next, hello())++ await hub.waitForConnection(next)++ expect(registry.claimCalls).toEqual([0])++ expect(created).toBe(1)++ expect(fresh.joins).toBe(1)++ expect(hub.activeRooms).toHaveLength(1)++ })++++ it('claims a stopped slot again through the normal start path even with nobody waiting', async () => {++ const lease = makeLease()++ const registry = new FakeRegistry()++ const probe = makeRoom(lease)++ const { log, entries } = makeLogger()++ const fresh = makeRoom(makeLease({ holder: 'holder-b', version: 2 }))++ const hub = new InstanceHub(++ registry as unknown as InstanceRegistry,++ probe.room,++ 1,++ 1,++ async () => fresh.room,++ log,++ )++++ registry.error = new Error('Query read timeout')++ await hub.heartbeat(0)++ await hub.heartbeat(LEASE_RECOVERY_GRACE_MS)++ expect(probe.disposed).toBe(1)++ expect(hub.activeRooms).toHaveLength(0)++++ // The restart backoff (5 s) is part of the normal claim path; once it passes the hub starts the++ // room again with no join and no player waiting.++ registry.error = null++ registry.claimResult = fresh.room.lease++ await hub.heartbeat(LEASE_RECOVERY_GRACE_MS)++ expect(registry.claimCalls).toEqual([])++ await hub.heartbeat(LEASE_RECOVERY_GRACE_MS + 5_001)++ await until(() => hub.activeRooms.length === 1)++ expect(registry.claimCalls).toEqual([0])++ expect(entries.some((e) => e.msg === 'instance restarted')).toBe(true)++ })++++ it('keeps the start chain usable after a failed claim', async () => {++ const lease = makeLease()++ const registry = new FakeRegistry()++ registry.outcome = 'lost'++ const probe = makeRoom(lease)++ const { log } = makeLogger()++ const fresh = makeRoom(makeLease({ holder: 'holder-b', version: 2 }))++ const hub = new InstanceHub(++ registry as unknown as InstanceRegistry,++ probe.room,++ 1,++ 1,++ async () => fresh.room,++ log,++ )++ await hub.heartbeat(1_000)++ expect(hub.activeRooms).toHaveLength(0)++++ // The slot's first re-claim fails (the database is still coming back)...++ registry.claimError = new Error('Query read timeout')++ const failed: { code: number; reason: string }[] = []++ hub.onMessage(makeConnection(hub, failed), hello())++ await until(() => failed.length > 0)++ expect(failed[0]?.code).toBe(1011)++++ // ...and the next join still starts a room: the failed attempt did not poison the chain.++ registry.claimError = null++ registry.claimResult = fresh.room.lease++ const accepted: { code: number; reason: string }[] = []++ hub.onMessage(makeConnection(hub, accepted), hello())++ await until(() => fresh.joins === 1)++ expect(registry.claimCalls).toHaveLength(2)++ expect(hub.activeRooms).toHaveLength(1)++ expect(accepted).toEqual([])++ })++++ it('stops at once when another holder owns the lease (nothing can be regained)', async () => {++ const lease = makeLease()++ const registry = new FakeRegistry()++ registry.outcome = 'lost'++ const probe = makeRoom(lease)++ const { log, entries } = makeLogger()++ const hub = makeHub(registry, probe.room, log)++ const closes: { code: number; reason: string }[] = []++ const conn = makeConnection(hub, closes)++ hub.onMessage(conn, hello())++ await hub.waitForConnection(conn)++++ await hub.heartbeat(1_000)++ expect(entries.some((e) => e.msg === 'instance lease lost to another holder')).toBe(true)++ expect(probe.disposed).toBe(1)++ expect(closes).toEqual([{ code: 1012, reason: 'instance_lease_lost' }])++ expect(hub.activeRooms).toHaveLength(0)++ // No revive: the heartbeat is not retried as a regain for a row another holder owns.++ expect(entries.some((e) => e.msg === 'instance lease regained')).toBe(false)++ })++})@@@ -53,6 +53,7 @@ import { resolveClientAddress } from './net/clientAddress.js'import { ServerMetrics } from './observability/metrics.js'import { createReadiness } from './observability/readiness.js'import { buildRelease, releaseHandler } from './observability/release.js'++import { createTickWatchdog } from './observability/tickWatchdog.js'import { sharedAssetsHandler } from './net/sharedAssets.js'import { loadDefinitionPacks, scriptGrant } from './mods/contentPack.js'import { ModError, ModRegistry, type ModView } from './mods/registry.js'@@@ -544,6 +545,10 @@ async function main(): Promise<void> {): undefinedconst createSecondaryRoom = async (slot: number, lease: InstanceLease) => {++ // Slot 0 reuses the boot store. Point its lease guard at the lease this room is actually starting++ // with: after a lease loss the hub claims a new version, and the old closure would fence every++ // write of the restarted room.++ if (slot === 0) secondaryPrimaryLease = leaseconst roomStore =slot === 0? secondaryStore@@@ -645,6 +650,9 @@ async function main(): Promise<void> {const surfRegistry = new InstanceRegistry(db, info.placeId)let surfPrimaryLease: InstanceLeaseconst createSurfRoom = async (slot: number, lease: InstanceLease) => {++ // Slot 0's guard reads this variable; a restart after a lease loss claims a new version, so the++ // guard must follow the lease the room is actually built with.++ if (slot === 0) surfPrimaryLease = leaseconst roomStore = openPgStore(db, {placeId: info.placeId,instanceSlot: slot,@@@ -786,7 +794,15 @@ async function main(): Promise<void> {[secondaryPlace, secondaryContent],...surfContentByPlace,])-- const heartbeatTimer = setInterval(() => void router.heartbeat(), 5_000)++ const heartbeatTimer = setInterval(() => {++ // Never let a rejected heartbeat (or a dispose that throws) become an unhandled rejection that++ // takes the whole process with it: one place failing must not stop every other place's lease.++ router.heartbeat().catch((err: unknown) =>++ log.error('instance heartbeat tick failed', {++ error: String((err as Error)?.message ?? err),++ }),++ )++ }, 5_000)heartbeatTimer.unref()const syncScriptView = async (view: ModView): Promise<void> => {@@@ -1427,6 +1443,12 @@ async function main(): Promise<void> {}loop()++ // A tick loop that dies (or is never scheduled again) leaves the process answering /healthz while++ // /api/ready's required `tick` check fails forever: production stayed 8 hours like that. Exit 1++ // after the readiness threshold plus the grace, so systemd restarts a working process.++ const tickWatchdog = createTickWatchdog({ metrics, log: log.child({ system: 'tick' }) })++ tickWatchdog.start()++const metricsTimer = setInterval(() => metrics.logSummary(log), config.metricsLogSeconds * 1000)// ── Stop (roadmap WS-P lifecycle; manifests/services/games.json → lifecycle.shutdown) ──@@@ -1451,6 +1473,7 @@ async function main(): Promise<void> {hard.unref()const httpDone = drainer.close(DRAIN_MS)running = false++ tickWatchdog.stop()clearInterval(metricsTimer)if (aiMatchTimer) clearInterval(aiMatchTimer)clearInterval(heartbeatTimer)nnew file mode 100644@@@ -0,0 +1,94 @@++import type { LogFields, Logger } from '@openvibe/shared'++import { describe, expect, it } from 'vitest'++import { createTickWatchdog, TICK_WATCHDOG_GRACE_MS } from './tickWatchdog.js'++++interface Entry {++ level: 'debug' | 'info' | 'warn' | 'error'++ msg: string++ fields?: LogFields++}++++function recordingLogger(): { log: Logger; entries: Entry[] } {++ const entries: Entry[] = []++ const log: Logger = {++ debug: (msg, fields) => entries.push({ level: 'debug', msg, ...(fields ? { fields } : {}) }),++ info: (msg, fields) => entries.push({ level: 'info', msg, ...(fields ? { fields } : {}) }),++ warn: (msg, fields) => entries.push({ level: 'warn', msg, ...(fields ? { fields } : {}) }),++ error: (msg, fields) => entries.push({ level: 'error', msg, ...(fields ? { fields } : {}) }),++ child: () => log,++ }++ return { log, entries }++}++++function setup(overrides: Partial<Parameters<typeof createTickWatchdog>[0]> = {}) {++ const { log, entries } = recordingLogger()++ let clock = 1_000_000++ const exits: number[] = []++ const metrics = { tick: 0, lastTickAt: 0 }++ const watchdog = createTickWatchdog({++ metrics,++ log,++ startedAt: clock,++ stallMs: 5_000,++ graceMs: 20_000,++ now: () => clock,++ exit: (code) => exits.push(code),++ ...overrides,++ })++ return {++ watchdog,++ entries,++ exits,++ metrics,++ advance: (ms: number) => {++ clock += ms++ },++ at: () => clock,++ }++}++++describe('simulation tick watchdog', () => {++ it('does not exit before the stall threshold plus grace', () => {++ const { watchdog, exits, advance } = setup()++ advance(24_999)++ expect(watchdog.check()).toBe(false)++ expect(exits).toEqual([])++ })++++ it('exits 1 with the reason once the simulation has not ticked for the grace', () => {++ const { watchdog, exits, entries, advance } = setup()++ advance(25_001)++ expect(watchdog.check()).toBe(true)++ expect(exits).toEqual([1])++ const error = entries.find((e) => e.level === 'error')++ expect(error?.msg).toMatch(/simulation tick stalled/)++ expect(error?.fields?.ageMs).toBe(25_001)++ expect(error?.fields?.tick).toBe(0)++ // It trips once, not on every later interval.++ expect(watchdog.check()).toBe(true)++ expect(exits).toEqual([1])++ })++++ it('follows a tick timestamp that advances', () => {++ const { watchdog, exits, metrics, advance } = setup()++ advance(20_000)++ metrics.lastTickAt = 1_020_000++ metrics.tick = 700++ advance(20_000)++ expect(watchdog.check()).toBe(false)++ expect(exits).toEqual([])++ // Still 20 s after the last tick: inside the threshold.++ advance(5_002)++ expect(watchdog.check()).toBe(true)++ expect(exits).toEqual([1])++ })++++ it('stops checking after stop() (shutdown must not exit the process)', () => {++ const { watchdog, exits, advance } = setup()++ watchdog.start()++ watchdog.stop()++ advance(TICK_WATCHDOG_GRACE_MS * 2)++ expect(watchdog.check()).toBe(false)++ expect(exits).toEqual([])++ })++})nnew file mode 100644@@@ -0,0 +1,92 @@++import type { Logger } from '@openvibe/shared'++import { TICK_STALL_MS } from './readiness.js'++++/**++ * How long the simulation may stay stalled before the process exits so systemd restarts it.++ * /api/ready already fails its required `tick` check after TICK_STALL_MS (5 s); that can be a single++ * long GC pause or a map rebuild. Thirty more seconds is long enough for such a pass to finish, and++ * for a room whose lease is in doubt (LEASE_RECOVERY_GRACE_MS, 60 s) to be handled by the places that++ * still tick, but short enough that a dead tick loop never stays up for hours with /api/ready failing.++ */++export const TICK_WATCHDOG_GRACE_MS = 30_000++++/** How often the watchdog looks at the tick timestamp. */++export const TICK_WATCHDOG_INTERVAL_MS = 5_000++++export interface TickWatchdogDeps {++ /** The tick counter and when the last completed simulation tick happened (epoch ms; 0 = none). */++ metrics: { readonly tick: number; readonly lastTickAt: number }++ log: Logger++ /** The clock at construction: before the first tick, the stall is measured from here. */++ startedAt?: number++ stallMs?: number++ graceMs?: number++ intervalMs?: number++ now?: () => number++ /** `process.exit` in production; injected so a test never kills its own runner. */++ exit?: (code: number) => void++}++++export interface TickWatchdog {++ start(): void++ stop(): void++ /** Check once; true once the watchdog has asked the process to exit. */++ check(): boolean++}++++/**++ * Watch the completed-tick timestamp and exit non-zero when it stops advancing while the process is++ * alive. A tick loop that died (or never got scheduled again) leaves the process answering HTTP and++ * /healthz but serving nothing; systemd only restarts it if the process exits. The watchdog waits for++ * the readiness threshold plus its grace, logs why, and exits 1.++ */++export function createTickWatchdog(deps: TickWatchdogDeps): TickWatchdog {++ const now = deps.now ?? Date.now++ const startedAt = deps.startedAt ?? now()++ const lastAllowedGapMs =++ (deps.stallMs ?? TICK_STALL_MS) + (deps.graceMs ?? TICK_WATCHDOG_GRACE_MS)++ const intervalMs = deps.intervalMs ?? TICK_WATCHDOG_INTERVAL_MS++ const exit = deps.exit ?? ((code: number): never => process.exit(code))++ let timer: ReturnType<typeof setInterval> | null = null++ let tripped = false++ let stopped = false++++ function stop(): void {++ stopped = true++ if (timer) {++ clearInterval(timer)++ timer = null++ }++ }++++ function check(): boolean {++ if (tripped) return true++ if (stopped) return false++ const at = now()++ const last = deps.metrics.lastTickAt++ // Before the first tick the clock starts at construction, so a boot that never starts looping is++ // caught too (readiness reports "simulation has not ticked yet" from the first probe on).++ const since = last > 0 ? last : startedAt++ const ageMs = at - since++ if (ageMs <= lastAllowedGapMs) return false++ tripped = true++ stop()++ deps.log.error('simulation tick stalled: exiting 1 so the service restarts', {++ ageMs,++ thresholdMs: lastAllowedGapMs,++ tick: deps.metrics.tick,++ })++ exit(1)++ return true++ }++++ return {++ start(): void {++ if (timer || tripped || stopped) return++ timer = setInterval(check, intervalMs)++ timer.unref()++ },++ stop,++ check,++ }++}@@@ -2,7 +2,13 @@ export * from './dto.js'export * from './repositories.js'export { accountKey } from './accountKey.js'export { openPgStore, type PgPersistenceStore, type PgStoreOptions } from './pg/pgStore.js'--export { InstanceRegistry, type InstanceLease, type InstanceState } from './pg/instanceRegistry.js'++export {++ HEARTBEAT_STATEMENT_TIMEOUT_MS,++ InstanceRegistry,++ type InstanceLease,++ type InstanceState,++ type LeaseHeartbeat,++} from './pg/instanceRegistry.js'export { MIGRATIONS_DIR } from './migrations.js'export { moveCasualSaves, exportCasual, eraseCasual } from './pg/casualData.js'export {@@@ -1,7 +1,11 @@import { beforeAll, describe, expect, it } from 'vitest'import { openTestDb } from '../testing.js'import { sql, type Db } from 'openvibe-sdk/db'--import { InstanceRegistry } from './instanceRegistry.js'++import {++ HEARTBEAT_STATEMENT_TIMEOUT_MS,++ InstanceRegistry,++ type InstanceLease,++} from './instanceRegistry.js'import { openPgStore } from './pgStore.js'let db: Db@@@ -49,4 +53,83 @@ describe('PostgreSQL instance leases', () => {await expect(store.meta.set('clock', 2)).rejects.toThrow('lease lost')expect(await store.meta.get('clock')).toBe(1)})++++ it('regains a lease whose TTL lapsed while the database missed heartbeats', async () => {++ const registry = new InstanceRegistry(db, 'arena')++ const claim = await registry.claim(2, 2, 'holder-a', 100, 50)++ if (!claim) throw new Error('claim failed')++ const running = await registry.renew(claim, 110, 50, 'running')++ if (!running) throw new Error('renew failed')++++ // The database was unreachable past lease_until (160) and nobody claimed the slot meanwhile:++ // a plain renew is fenced by lease_until, but the same holder and version may take the row back.++ expect(await registry.renew(running, 300, 50)).toBeNull()++ const regained = await registry.heartbeat(running, 300, 50, 'running')++ expect(regained?.regained).toBe(true)++ expect(regained?.lease).toMatchObject({++ id: 'arena:2',++ holder: 'holder-a',++ version: running.version,++ leaseUntil: 350,++ })++ if (!regained) throw new Error('regain failed')++ // The lease guard kept working without a version bump: queued writes can land after the regain.++ await db.tx(async () => registry.assertHeld(regained.lease, 301))++ })++++ it('does not regain a lease another holder claimed while the database was unreachable', async () => {++ const registry = new InstanceRegistry(db, 'arena')++ const claim = await registry.claim(3, 2, 'holder-b', 500, 50)++ if (!claim) throw new Error('claim failed')++ const running = await registry.renew(claim, 510, 50, 'running')++ if (!running) throw new Error('renew failed')++ const stolen = await registry.claim(3, 2, 'holder-c', 600, 50)++ expect(stolen).not.toBeNull()++++ expect(await registry.heartbeat(running, 601, 50, 'running')).toBeNull()++ })++++ it('sets a short statement timeout on the heartbeat transaction', async () => {++ const executed: string[] = []++ const current: InstanceLease = {++ id: 'arena:4',++ placeId: 'arena',++ slot: 4,++ capacity: 2,++ state: 'running',++ holder: 'holder-d',++ version: 7,++ heartbeatAt: 10,++ leaseUntil: 10_000,++ }++ const row = {++ id: current.id,++ place_id: current.placeId,++ slot: current.slot,++ capacity: current.capacity,++ state: 'running',++ holder: current.holder,++ lease_version: current.version,++ heartbeat_at: 20,++ lease_until: 20_000,++ }++ const tx = {++ exec: async (q: { parts: string[] }) => {++ executed.push(q.parts.join('?'))++ return 0++ },++ // The renew matches (`lease_until > now`); the regain fallback is not reached.++ maybe: async (q: { parts: string[] }) =>++ q.parts.join('?').includes('lease_until >') ? row : null,++ }++ const stub = {++ store: 'postgresql',++ tx: async <T>(fn: (t: typeof tx) => Promise<T>) => fn(tx),++ } as unknown as Db++ const registry = new InstanceRegistry(stub, 'arena')++++ const result = await registry.heartbeat(current, 20, 15_000, 'running')++ expect(result?.regained).toBe(false)++ expect(executed).toEqual([`SET LOCAL statement_timeout = ${HEARTBEAT_STATEMENT_TIMEOUT_MS}`])++ })})@@@ -14,6 +14,22 @@ export interface InstanceLease {leaseUntil: number}++/** The outcome of one heartbeat: the renewed lease, and whether its TTL had already lapsed. */++export interface LeaseHeartbeat {++ lease: InstanceLease++ /** True when the TTL had lapsed and no other holder had claimed the row (a regain, not a renew). */++ regained: boolean++}++++/**++ * The heartbeat's own statement timeout in milliseconds. The lease TTL is 15 s and the heartbeat runs++ * every 5 s (main.ts). Without a cap one slow query waits on the connection's 15 s query timeout,++ * spans the whole TTL and loses the lease over a briefly slow database (production saw an 8.7 s++ * heartbeat UPDATE). Two seconds leaves room for a regain attempt before the next heartbeat and is++ * longer than a healthy round trip through PgBouncer.++ */++export const HEARTBEAT_STATEMENT_TIMEOUT_MS = 2_000++interface InstanceRow {id: stringplace_id: string@@@ -94,6 +110,48 @@ export class InstanceRegistry {return row ? lease(row) : null}++ /**++ * One heartbeat in a single transaction, with its own short statement timeout:++ *++ * 1. renew while the lease is still valid (`lease_until > now`), or++ * 2. take the row back when this holder and version still own it and only the TTL lapsed because++ * the database was unreachable. The UPDATE is fenced by holder and version, so a competing++ * claim either commits first (this returns null) or sees the renewed lease (that claim fails).++ *++ * On the same holder and version the version is not bumped: the store's lease guard captured this++ * lease object, and writes queued before the outage must be valid again after the regain.++ *++ * Returns null when another holder owns the row (or it was stopped): that lease cannot be regained.++ * Throws on a database error; the caller treats that as "lease unconfirmed" and retries with backoff.++ */++ async heartbeat(++ current: InstanceLease,++ now: number,++ ttlMs: number,++ state: InstanceState = current.state,++ ): Promise<LeaseHeartbeat | null> {++ if (ttlMs < 1) throw new Error('invalid instance heartbeat')++ return this.db.tx(async (tx) => {++ // SET LOCAL is transaction-scoped, so the pool connection resumes its normal timeout afterwards.++ // The value comes from a constant, never from input.++ await tx.exec(sql.raw(`SET LOCAL statement_timeout = ${HEARTBEAT_STATEMENT_TIMEOUT_MS}`))++ const renewed = await tx.maybe<InstanceRow>(sql`++ UPDATE game_instances SET heartbeat_at = ${now}, lease_until = ${now + ttlMs}, state = ${state}++ WHERE id = ${current.id} AND holder = ${current.holder}++ AND lease_version = ${current.version} AND lease_until > ${now}++ RETURNING *++ `)++ if (renewed) return { lease: lease(renewed), regained: false }++ const regained = await tx.maybe<InstanceRow>(sql`++ UPDATE game_instances SET heartbeat_at = ${now}, lease_until = ${now + ttlMs}, state = ${state}++ WHERE id = ${current.id} AND holder = ${current.holder}++ AND lease_version = ${current.version} AND state <> 'stopped'++ RETURNING *++ `)++ return regained ? { lease: lease(regained), regained: true } : null++ })++ }++async stop(current: InstanceLease): Promise<boolean> {const row = await this.db.maybe<{ id: string }>(sql`UPDATE game_instances SET state = 'stopped', lease_until = 0nnew file mode 100644@@@ -0,0 +1,3 @@++// file: 'apps/server/src/game/instanceLease.test.ts'++// covers: 'apps/server/src/game'++// Generated by node test/run.js --index; do not edit.nnew file mode 100644@@@ -0,0 +1,3 @@++// file: 'apps/server/src/observability/tickWatchdog.test.ts'++// covers: 'apps/server/src/observability'++// Generated by node test/run.js --index; do not edit.
@@ -1,5 +1,5 @@
@@ -1,5 +1,5 @@
import { encodeServerMessage, type ClientHello, type ClientMessage } from '@openvibe/protocol'
import { encodeServerMessage, type ClientHello, type ClientMessage } from '@openvibe/protocol'
-import type { InstanceLease, InstanceRegistry } from '@openvibe/persistence'
+import type { InstanceLease, InstanceRegistry, LeaseHeartbeat } from '@openvibe/persistence'
import { FixedTimestep, type Logger } from '@openvibe/shared'
import { FixedTimestep, type Logger } from '@openvibe/shared'
import type { GameConnection } from './systems/context.js'
import type { GameConnection } from './systems/context.js'
import type { GameServer } from './gameServer.js'
import type { GameServer } from './gameServer.js'
@@ -12,9 +12,64 @@ export interface InstanceRoom {
@@ -12,9 +12,64 @@ export interface InstanceRoom {
dispose(): void | Promise<void>
dispose(): void | Promise<void>
}
}
+/**
+ * The row lease's lifetime. The heartbeat runs every 5 s (main.ts): three missed renewals and another
+ * process may claim the slot.
+ */
+export const LEASE_TTL_MS = 15_000
+
+/**
+ * How long a room whose lease was lost to a database error keeps its world and players while it tries
+ * to regain the lease before it is stopped. 60 s rides out a connection timeout or a PgBouncer
+ * failover (production saw a heartbeat UPDATE wait 8.7 s), and is short enough that a frozen room does
+ * not hold a slot, its players and its memory for minutes. A database outage longer than this is
+ * better served by the process restarting (the tick watchdog in main.ts).
+ */
+export const LEASE_RECOVERY_GRACE_MS = 60_000
+
+/**
+ * First backoff between regain attempts. Retrying on every 5 s heartbeat would hammer a database that
+ * is already timing out; 2 s puts the first retry on the next heartbeat.
+ */
+const LEASE_REGAIN_INITIAL_BACKOFF_MS = 2_000
+
+/** Regain retries double to this cap: a short outage is caught on the first or second retry. */
+const LEASE_REGAIN_MAX_BACKOFF_MS = 30_000
+
+/**
+ * After a room is stopped for good its slot is claimed again through the normal start path, even with
+ * nobody waiting. The first retry comes one heartbeat later; the backoff then doubles to a minute so a
+ * slot another process legitimately owns is only polled cheaply.
+ */
+const ROOM_RESTART_INITIAL_BACKOFF_MS = 5_000
+const ROOM_RESTART_MAX_BACKOFF_MS = 60_000
+
+/** The same experience as a deploy restart: clients reload and reconnect when /healthz answers again. */
+const RECONNECT_MESSAGE = '🔄 The world is restarting: you will reconnect in a moment…'
+
+/** A room whose lease could not be confirmed: it is neither simulated nor admitted while it retries. */
+interface RecoveringRoom {
+ room: InstanceRoom
+ /** When the first heartbeat could not confirm the lease. */
+ since: number
+ attempts: number
+ nextAttemptAt: number
+}
+
+/** A stopped slot the hub is claiming again through the normal start path. */
+interface RestartAttempt {
+ attempts: number
+ nextAttemptAt: number
+ inFlight: boolean
+}
+
/** One-process room router. The socket's upgrade identity remains attached across transfers. */
/** One-process room router. The socket's upgrade identity remains attached across transfers. */
export class InstanceHub {
export class InstanceHub {
private readonly rooms = new Map<number, InstanceRoom>()
private readonly rooms = new Map<number, InstanceRoom>()
+ /** Rooms whose lease is in doubt. They keep their world and players, but not the tick or admissions. */
+ private readonly recovering = new Map<number, RecoveringRoom>()
+ /** Slots stopped for good that are being claimed again with backoff. */
+ private readonly restarts = new Map<number, RestartAttempt>()
/** One accumulator per room: 100 Hz surf and 30 Hz everything else can share a hub. */
/** One accumulator per room: 100 Hz surf and 30 Hz everything else can share a hub. */
private readonly clocks = new Map<number, FixedTimestep>()
private readonly clocks = new Map<number, FixedTimestep>()
private readonly connections = new Map<GameConnection, { slot: number; hello: ClientHello }>()
private readonly connections = new Map<GameConnection, { slot: number; hello: ClientHello }>()
@@ -84,6 +139,12 @@ export class InstanceHub {
@@ -84,6 +139,12 @@ export class InstanceHub {
releaseAdmission()
releaseAdmission()
}
}
if (!room || this.closed.has(conn)) {
if (!room || this.closed.has(conn)) {
+ // A slot being regained cannot take players: close like a restart so the client retries
+ // instead of reading "server full" and giving up on this place.
+ if (!this.closed.has(conn) && this.recovering.size > 0) {
+ conn.close(1012, 'instance_recovering')
+ return
+ }
conn.send(encodeServerMessage({ t: 'reject', reason: 'server_full' }))
conn.send(encodeServerMessage({ t: 'reject', reason: 'server_full' }))
conn.close(4003, 'server_full')
conn.close(4003, 'server_full')
return
return
@@ -136,6 +197,8 @@ export class InstanceHub {
@@ -136,6 +197,8 @@ export class InstanceHub {
*
*
* Without an argument this keeps the old test/manual contract: exactly one
* Without an argument this keeps the old test/manual contract: exactly one
* step per room per call, regardless of its rate.
* step per room per call, regardless of its rate.
+ *
+ * Rooms whose lease is in doubt are not in `rooms`: a place that cannot save must not simulate.
*/
*/
step(elapsedSeconds?: number): void {
step(elapsedSeconds?: number): void {
if (elapsedSeconds === undefined) {
if (elapsedSeconds === undefined) {
@@ -158,41 +221,217 @@ export class InstanceHub {
@@ -158,41 +221,217 @@ export class InstanceHub {
if (this.heartbeating || this.stopping) return
if (this.heartbeating || this.stopping) return
this.heartbeating = true
this.heartbeating = true
try {
try {
- for (const room of [...this.rooms.values()]) {
- let renewed: InstanceLease | null = null
- try {
- renewed = await this.registry.renew(room.lease, now, 15_000, 'running')
- } catch (error) {
- this.log.error('instance heartbeat failed', { id: room.lease.id, error: String(error) })
- }
- if (renewed) {
- room.lease = renewed
- continue
- }
- this.log.error('instance lease lost', { id: room.lease.id })
- this.rooms.delete(room.lease.slot)
- this.clocks.delete(room.lease.slot)
- for (const [conn, current] of this.connections) {
- if (current.slot === room.lease.slot) conn.close(1012, 'instance_lease_lost')
- }
- await room.dispose()
- }
+ // Every room heartbeats at once: one slow query (bounded by the registry's own statement
+ // timeout) must not delay the next room's renewal past its deadline.
+ await Promise.all([
+ ...[...this.rooms.values()].map((room) => this.renewRoom(room, now)),
+ ...[...this.recovering.values()].map((rec) => this.regainRoom(rec, now)),
+ ])
+ // Starting a room can take seconds (world restore); it must never keep the lease renewals
+ // above waiting, so restarts run beside the heartbeat and schedule themselves.
+ for (const [slot, attempt] of this.restarts) this.runRestart(slot, attempt, now)
} finally {
} finally {
this.heartbeating = false
this.heartbeating = false
}
}
}
}
+ private async renewRoom(room: InstanceRoom, now: number): Promise<void> {
+ if (this.stopping) return
+ let result: LeaseHeartbeat | null
+ try {
+ result = await this.registry.heartbeat(room.lease, now, LEASE_TTL_MS, 'running')
+ } catch (error) {
+ // The lease may still be ours, but nothing may touch the world until that is confirmed.
+ this.log.error('instance heartbeat failed', { id: room.lease.id, error: String(error) })
+ this.suspend(room, now)
+ return
+ }
+ if (result) {
+ room.lease = result.lease
+ if (result.regained) this.log.info('instance lease regained', { id: room.lease.id })
+ return
+ }
+ // The row no longer carries this holder and version: another process owns it. Nothing can be
+ // regained, so stop at once instead of simulating a place this process cannot save.
+ this.log.error('instance lease lost to another holder', { id: room.lease.id })
+ await this.stopRoom(room, 'instance_lease_lost')
+ }
+
+ /** Stop the tick and admissions, keep the players, and retry the lease with backoff. */
+ private suspend(room: InstanceRoom, now: number): void {
+ const slot = room.lease.slot
+ if (this.recovering.has(slot)) return
+ this.rooms.delete(slot)
+ this.clocks.delete(slot)
+ this.recovering.set(slot, {
+ room,
+ since: now,
+ attempts: 0,
+ nextAttemptAt: now + LEASE_REGAIN_INITIAL_BACKOFF_MS,
+ })
+ this.log.error('instance lease lost', { id: room.lease.id, graceMs: LEASE_RECOVERY_GRACE_MS })
+ }
+
+ /** One regain attempt inside the grace; a success puts the room back into the tick. */
+ private async regainRoom(rec: RecoveringRoom, now: number): Promise<void> {
+ if (this.stopping) return
+ const { room } = rec
+ const slot = room.lease.slot
+ if (now - rec.since >= LEASE_RECOVERY_GRACE_MS) {
+ this.recovering.delete(slot)
+ this.log.error('instance lease not regained in time: stopping room', {
+ id: room.lease.id,
+ graceMs: LEASE_RECOVERY_GRACE_MS,
+ })
+ await this.stopRoom(room, 'instance_lease_lost')
+ this.scheduleRestart(slot, now)
+ return
+ }
+ if (now < rec.nextAttemptAt) return
+ let result: LeaseHeartbeat | null
+ try {
+ result = await this.registry.heartbeat(room.lease, now, LEASE_TTL_MS, 'running')
+ } catch (error) {
+ this.log.error('instance lease regain failed', { id: room.lease.id, error: String(error) })
+ this.backoffRegain(rec, now)
+ return
+ }
+ if (!result) {
+ // Another holder claimed the row while the database was unreachable: this room cannot return.
+ this.recovering.delete(slot)
+ this.log.error('instance lease lost to another holder', { id: room.lease.id })
+ await this.stopRoom(room, 'instance_lease_lost')
+ this.scheduleRestart(slot, now)
+ return
+ }
+ room.lease = result.lease
+ this.recovering.delete(slot)
+ this.rooms.set(slot, room)
+ this.log.info('instance lease regained', {
+ id: room.lease.id,
+ afterMs: now - rec.since,
+ reclaimed: result.regained,
+ })
+ }
+
+ private backoffRegain(rec: RecoveringRoom, now: number): void {
+ rec.attempts++
+ rec.nextAttemptAt =
+ now +
+ Math.min(
+ LEASE_REGAIN_INITIAL_BACKOFF_MS * 2 ** (rec.attempts - 1),
+ LEASE_REGAIN_MAX_BACKOFF_MS,
+ )
+ }
+
+ private scheduleRestart(slot: number, now: number): void {
+ if (this.stopping) return
+ this.restarts.set(slot, {
+ attempts: 0,
+ nextAttemptAt: now + ROOM_RESTART_INITIAL_BACKOFF_MS,
+ inFlight: false,
+ })
+ }
+
+ private runRestart(slot: number, attempt: RestartAttempt, now: number): void {
+ if (this.stopping || attempt.inFlight || now < attempt.nextAttemptAt) return
+ if (this.rooms.has(slot) || this.recovering.has(slot)) {
+ this.restarts.delete(slot)
+ return
+ }
+ attempt.inFlight = true
+ void this.start(slot)
+ .then((room) => {
+ if (room) {
+ this.restarts.delete(slot)
+ this.log.info('instance restarted', { id: room.lease.id })
+ return
+ }
+ this.backoffRestart(attempt)
+ })
+ .catch((error: unknown) => {
+ this.log.error('instance restart failed', { slot, error: String(error) })
+ this.backoffRestart(attempt)
+ })
+ .finally(() => {
+ attempt.inFlight = false
+ })
+ }
+
+ /** Back off from the moment the attempt ended: world restore can take seconds. */
+ private backoffRestart(attempt: RestartAttempt): void {
+ attempt.attempts++
+ attempt.nextAttemptAt =
+ Date.now() +
+ Math.min(
+ ROOM_RESTART_INITIAL_BACKOFF_MS * 2 ** (attempt.attempts - 1),
+ ROOM_RESTART_MAX_BACKOFF_MS,
+ )
+ }
+
+ /** Announce the reconnect, close the room's players with 1012 and release its memory. */
+ private async stopRoom(room: InstanceRoom, reason: string): Promise<void> {
+ const slot = room.lease.slot
+ this.rooms.delete(slot)
+ this.clocks.delete(slot)
+ room.game.announce(RECONNECT_MESSAGE)
+ for (const [conn, current] of [...this.connections]) {
+ if (current.slot !== slot) continue
+ this.connections.delete(conn)
+ conn.close(1012, reason)
+ }
+ try {
+ await room.dispose()
+ } catch (error) {
+ // The room is already gone from every map; a failing dispose must not abort the heartbeat.
+ this.log.warn('instance room dispose failed', { id: room.lease.id, error: String(error) })
+ }
+ }
+
async shutdown(): Promise<boolean> {
async shutdown(): Promise<boolean> {
this.stopping = true
this.stopping = true
- for (const room of this.rooms.values()) {
- const draining = await this.registry.renew(room.lease, Date.now(), 15_000, 'draining')
- if (draining) room.lease = draining
+ const rooms = [...this.rooms.values(), ...[...this.recovering.values()].map((rec) => rec.room)]
+ this.recovering.clear()
+ this.restarts.clear()
+ const draining = new Set<InstanceRoom>()
+ for (const room of rooms) {
+ try {
+ const renewed = await this.registry.renew(room.lease, Date.now(), LEASE_TTL_MS, 'draining')
+ if (renewed) {
+ room.lease = renewed
+ draining.add(room)
+ }
+ } catch (error) {
+ // A database too slow to answer must not make the shutdown throw before the sockets close.
+ this.log.warn('instance lease renewal failed during shutdown', {
+ id: room.lease.id,
+ error: String(error),
+ })
+ }
}
}
- for (const room of this.rooms.values()) room.game.shutdown()
- const drained = await Promise.all([...this.rooms.values()].map((room) => room.game.drain()))
- for (const room of this.rooms.values()) {
- await this.registry.stop(room.lease)
- await room.dispose()
+ // A room whose lease cannot be renewed is fenced: its writes would be refused, so do not wait
+ // through the write-behind retries. It is disposed below like the rest.
+ for (const room of draining) room.game.shutdown()
+ const drained = await Promise.all([...draining].map((room) => room.game.drain()))
+ for (const room of rooms) {
+ if (draining.has(room)) {
+ try {
+ await this.registry.stop(room.lease)
+ } catch (error) {
+ this.log.warn('instance lease stop failed during shutdown', {
+ id: room.lease.id,
+ error: String(error),
+ })
+ }
+ }
+ try {
+ await room.dispose()
+ } catch (error) {
+ this.log.warn('instance room dispose failed during shutdown', {
+ id: room.lease.id,
+ error: String(error),
+ })
+ }
}
}
this.rooms.clear()
this.rooms.clear()
this.clocks.clear()
this.clocks.clear()
@@ -205,7 +444,8 @@ export class InstanceHub {
@@ -205,7 +444,8 @@ export class InstanceHub {
return room
return room
}
}
for (let slot = 0; slot < this.maxInstances; slot++) {
for (let slot = 0; slot < this.maxInstances; slot++) {
- if (!this.rooms.has(slot)) {
+ // A recovering slot still owns its row: claiming it now would fence the room we are trying to save.
+ if (!this.rooms.has(slot) && !this.recovering.has(slot)) {
const room = await this.start(slot)
const room = await this.start(slot)
if (room) return room
if (room) return room
}
}
@@ -222,12 +462,14 @@ export class InstanceHub {
@@ -222,12 +462,14 @@ export class InstanceHub {
private async start(slot: number): Promise<InstanceRoom | null> {
private async start(slot: number): Promise<InstanceRoom | null> {
let result: InstanceRoom | null = null
let result: InstanceRoom | null = null
const prior = this.creating
const prior = this.creating
- this.creating = prior.then(async () => {
+ const run = prior.then(async () => {
const existing = this.rooms.get(slot)
const existing = this.rooms.get(slot)
if (existing) {
if (existing) {
result = existing
result = existing
return
return
}
}
+ if (this.recovering.has(slot)) return
+ if (this.stopping) return
const lease = await this.registry.claim(
const lease = await this.registry.claim(
slot,
slot,
this.capacity,
this.capacity,
@@ -238,17 +480,43 @@ export class InstanceHub {
@@ -238,17 +480,43 @@ export class InstanceHub {
if (!lease) return
if (!lease) return
try {
try {
const room = await this.createRoom(slot, lease)
const room = await this.createRoom(slot, lease)
- const running = await this.registry.renew(lease, Date.now(), 15_000, 'running')
+ if (this.stopping) {
+ // Shutdown already snapshotted the rooms: do not leave one running (and leased) behind.
+ try {
+ await this.registry.stop(lease)
+ } catch (stopError) {
+ this.log.warn('instance lease stop failed for a room started during shutdown', {
+ id: lease.id,
+ error: String(stopError),
+ })
+ }
+ await room.dispose()
+ return
+ }
+ const running = await this.registry.renew(lease, Date.now(), LEASE_TTL_MS, 'running')
if (!running) throw new Error('lease expired during room start')
if (!running) throw new Error('lease expired during room start')
room.lease = running
room.lease = running
this.rooms.set(slot, room)
this.rooms.set(slot, room)
result = room
result = room
} catch (error) {
} catch (error) {
- await this.registry.stop(lease)
+ try {
+ await this.registry.stop(lease)
+ } catch (stopError) {
+ this.log.warn('instance lease stop failed after a room start failure', {
+ id: lease.id,
+ error: String(stopError),
+ })
+ }
throw error
throw error
}
}
})
})
- await this.creating
+ // One failed start must not poison every later one: the restart backoff and the next join both
+ // call start again, and the serialization chain has to keep running after this caller's error.
+ this.creating = run.then(
+ () => undefined,
+ () => undefined,
+ )
+ await run
return result
return result
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,351 @@
@@ -0,0 +1,351 @@
+import type { InstanceLease, InstanceRegistry, LeaseHeartbeat } from '@openvibe/persistence'
+import { PROTOCOL_VERSION, defaultAppearance, type ClientHello } from '@openvibe/protocol'
+import type { LogFields, Logger } from '@openvibe/shared'
+import { describe, expect, it } from 'vitest'
+import type { GameServer } from './gameServer.js'
+import { InstanceHub, LEASE_RECOVERY_GRACE_MS, type InstanceRoom } from './instanceHub.js'
+import type { GameConnection } from './systems/context.js'
+
+function makeLease(overrides: Partial<InstanceLease> = {}): InstanceLease {
+ return {
+ id: 'lease-test:0',
+ placeId: 'lease-test',
+ slot: 0,
+ capacity: 1,
+ state: 'running',
+ holder: 'holder-a',
+ version: 1,
+ heartbeatAt: 0,
+ leaseUntil: 15_000,
+ ...overrides,
+ }
+}
+
+/** The registry surface InstanceHub uses; the database behaviour is scripted per test. */
+class FakeRegistry {
+ readonly heartbeatCalls: number[] = []
+ readonly claimCalls: number[] = []
+ readonly stopCalls: string[] = []
+ error: Error | null = null
+ /** 'renew' | 'regain' | 'lost' */
+ outcome: 'renew' | 'regain' | 'lost' = 'renew'
+ claimError: Error | null = null
+ claimResult: InstanceLease | null = null
+
+ async heartbeat(current: InstanceLease, now: number): Promise<LeaseHeartbeat | null> {
+ this.heartbeatCalls.push(now)
+ if (this.error) throw this.error
+ if (this.outcome === 'lost') return null
+ return {
+ lease: { ...current, heartbeatAt: now, leaseUntil: now + 15_000 },
+ regained: this.outcome === 'regain',
+ }
+ }
+
+ async claim(slot: number): Promise<InstanceLease | null> {
+ this.claimCalls.push(slot)
+ if (this.claimError) throw this.claimError
+ return this.claimResult
+ }
+
+ async renew(current: InstanceLease): Promise<InstanceLease | null> {
+ return current
+ }
+
+ async stop(current: InstanceLease): Promise<boolean> {
+ this.stopCalls.push(current.id)
+ return true
+ }
+}
+
+interface RoomProbe {
+ room: InstanceRoom
+ steps: number
+ joins: number
+ announced: string[]
+ disposed: number
+}
+
+function makeRoom(lease: InstanceLease): RoomProbe {
+ const probe: RoomProbe = {
+ room: null as unknown as InstanceRoom,
+ steps: 0,
+ joins: 0,
+ announced: [],
+ disposed: 0,
+ }
+ const game = {
+ tickRate: 30,
+ step: () => probe.steps++,
+ join: async () => {
+ probe.joins++
+ return true
+ },
+ onMessage: () => {},
+ onDisconnect: () => {},
+ checkpoint: () => {},
+ drain: async () => true,
+ shutdown: () => {},
+ announce: (text: string) => probe.announced.push(text),
+ contentDigest: 'digest',
+ onlineCount: () => 0,
+ }
+ probe.room = {
+ lease,
+ game: game as unknown as GameServer,
+ world: {} as InstanceRoom['world'],
+ dispose: () => {
+ probe.disposed++
+ },
+ }
+ return probe
+}
+
+function makeLogger(): {
+ log: Logger
+ entries: { level: string; msg: string; fields?: LogFields }[]
+} {
+ const entries: { level: string; msg: string; fields?: LogFields }[] = []
+ const log: Logger = {
+ debug: (msg, fields) => entries.push({ level: 'debug', msg, ...(fields ? { fields } : {}) }),
+ info: (msg, fields) => entries.push({ level: 'info', msg, ...(fields ? { fields } : {}) }),
+ warn: (msg, fields) => entries.push({ level: 'warn', msg, ...(fields ? { fields } : {}) }),
+ error: (msg, fields) => entries.push({ level: 'error', msg, ...(fields ? { fields } : {}) }),
+ child: () => log,
+ }
+ return { log, entries }
+}
+
+function makeConnection(
+ hub: InstanceHub,
+ closes: { code: number; reason: string }[],
+): GameConnection {
+ const conn: GameConnection = {
+ identity: { guestKeyHash: 'a'.repeat(64) },
+ send: () => {},
+ close: (code, reason) => {
+ closes.push({ code, reason })
+ hub.onDisconnect(conn)
+ },
+ }
+ return conn
+}
+
+function hello(): ClientHello {
+ return {
+ t: 'hello',
+ v: PROTOCOL_VERSION,
+ contentDigest: 'digest',
+ slot: 0,
+ name: 'guest',
+ appearance: defaultAppearance(),
+ }
+}
+
+function makeHub(registry: FakeRegistry, room: InstanceRoom, log: Logger): InstanceHub {
+ return new InstanceHub(
+ registry as unknown as InstanceRegistry,
+ room,
+ 1,
+ 1,
+ async () => {
+ throw new Error('unexpected room creation')
+ },
+ log,
+ )
+}
+
+async function until(check: () => boolean): Promise<void> {
+ for (let i = 0; i < 200; i++) {
+ if (check()) return
+ await new Promise((resolve) => setTimeout(resolve, 5))
+ }
+ throw new Error('condition not reached')
+}
+
+describe('instance lease recovery', () => {
+ it('stops the tick and admissions while the lease is unconfirmed, then regains it with backoff', async () => {
+ const lease = makeLease()
+ const registry = new FakeRegistry()
+ const probe = makeRoom(lease)
+ const { log, entries } = makeLogger()
+ const hub = makeHub(registry, probe.room, log)
+ const closes: { code: number; reason: string }[] = []
+ const conn = makeConnection(hub, closes)
+
+ // One database error: the heartbeat can no longer confirm the lease.
+ registry.error = new Error('Query read timeout')
+ await hub.heartbeat(1_000)
+ expect(hub.activeRooms).toHaveLength(0)
+ expect(entries.some((e) => e.msg === 'instance heartbeat failed')).toBe(true)
+ expect(entries.some((e) => e.msg === 'instance lease lost')).toBe(true)
+
+ // No simulation while the lease is in doubt.
+ hub.step()
+ expect(probe.steps).toBe(0)
+
+ // Not accepting players: the client is told to reconnect (1012), not that the server is full.
+ hub.onMessage(conn, hello())
+ await hub.waitForConnection(conn)
+ expect(closes).toEqual([{ code: 1012, reason: 'instance_recovering' }])
+ expect(probe.joins).toBe(0)
+
+ // The first regain attempt is scheduled 2 s after the loss; a heartbeat before that does nothing.
+ await hub.heartbeat(2_000)
+ expect(registry.heartbeatCalls).toEqual([1_000])
+ // The database answers again; the row is still this holder's, so the room comes back.
+ registry.error = null
+ registry.outcome = 'regain'
+ await hub.heartbeat(3_100)
+ expect(entries.some((e) => e.msg === 'instance lease regained')).toBe(true)
+ expect(hub.activeRooms).toHaveLength(1)
+
+ // The tick resumes and the regained room takes players.
+ hub.step()
+ expect(probe.steps).toBe(1)
+ const rejoining = makeConnection(hub, [])
+ hub.onMessage(rejoining, hello())
+ await hub.waitForConnection(rejoining)
+ expect(probe.joins).toBe(1)
+ })
+
+ it('stops a room that cannot regain its lease inside the grace and starts it again on the next join', async () => {
+ const lease = makeLease()
+ const registry = new FakeRegistry()
+ const probe = makeRoom(lease)
+ const { log, entries } = makeLogger()
+ let created = 0
+ const fresh = makeRoom(makeLease({ holder: 'holder-b', version: 2, leaseUntil: 30_000 }))
+ const hub = new InstanceHub(
+ registry as unknown as InstanceRegistry,
+ probe.room,
+ 1,
+ 1,
+ async () => {
+ created++
+ return fresh.room
+ },
+ log,
+ )
+ const closes: { code: number; reason: string }[] = []
+ const connected = makeConnection(hub, closes)
+ hub.onMessage(connected, hello())
+ await hub.waitForConnection(connected)
+ expect(probe.joins).toBe(1)
+
+ registry.error = new Error('Connection terminated due to connection timeout')
+ await hub.heartbeat(10_000)
+ expect(hub.activeRooms).toHaveLength(0)
+
+ // The grace expires: the old room announces a reconnect, closes its players with 1012 and is disposed.
+ await hub.heartbeat(10_000 + LEASE_RECOVERY_GRACE_MS)
+ expect(
+ entries.some((e) => e.msg === 'instance lease not regained in time: stopping room'),
+ ).toBe(true)
+ expect(closes).toEqual([{ code: 1012, reason: 'instance_lease_lost' }])
+ expect(probe.announced[0]).toMatch(/reconnect/i)
+ expect(probe.disposed).toBe(1)
+ expect(probe.steps).toBe(0)
+
+ // The next join claims the freed slot through the normal start path and serves a fresh room.
+ registry.error = null
+ registry.claimResult = fresh.room.lease
+ const next = makeConnection(hub, [])
+ hub.onMessage(next, hello())
+ await hub.waitForConnection(next)
+ expect(registry.claimCalls).toEqual([0])
+ expect(created).toBe(1)
+ expect(fresh.joins).toBe(1)
+ expect(hub.activeRooms).toHaveLength(1)
+ })
+
+ it('claims a stopped slot again through the normal start path even with nobody waiting', async () => {
+ const lease = makeLease()
+ const registry = new FakeRegistry()
+ const probe = makeRoom(lease)
+ const { log, entries } = makeLogger()
+ const fresh = makeRoom(makeLease({ holder: 'holder-b', version: 2 }))
+ const hub = new InstanceHub(
+ registry as unknown as InstanceRegistry,
+ probe.room,
+ 1,
+ 1,
+ async () => fresh.room,
+ log,
+ )
+
+ registry.error = new Error('Query read timeout')
+ await hub.heartbeat(0)
+ await hub.heartbeat(LEASE_RECOVERY_GRACE_MS)
+ expect(probe.disposed).toBe(1)
+ expect(hub.activeRooms).toHaveLength(0)
+
+ // The restart backoff (5 s) is part of the normal claim path; once it passes the hub starts the
+ // room again with no join and no player waiting.
+ registry.error = null
+ registry.claimResult = fresh.room.lease
+ await hub.heartbeat(LEASE_RECOVERY_GRACE_MS)
+ expect(registry.claimCalls).toEqual([])
+ await hub.heartbeat(LEASE_RECOVERY_GRACE_MS + 5_001)
+ await until(() => hub.activeRooms.length === 1)
+ expect(registry.claimCalls).toEqual([0])
+ expect(entries.some((e) => e.msg === 'instance restarted')).toBe(true)
+ })
+
+ it('keeps the start chain usable after a failed claim', async () => {
+ const lease = makeLease()
+ const registry = new FakeRegistry()
+ registry.outcome = 'lost'
+ const probe = makeRoom(lease)
+ const { log } = makeLogger()
+ const fresh = makeRoom(makeLease({ holder: 'holder-b', version: 2 }))
+ const hub = new InstanceHub(
+ registry as unknown as InstanceRegistry,
+ probe.room,
+ 1,
+ 1,
+ async () => fresh.room,
+ log,
+ )
+ await hub.heartbeat(1_000)
+ expect(hub.activeRooms).toHaveLength(0)
+
+ // The slot's first re-claim fails (the database is still coming back)...
+ registry.claimError = new Error('Query read timeout')
+ const failed: { code: number; reason: string }[] = []
+ hub.onMessage(makeConnection(hub, failed), hello())
+ await until(() => failed.length > 0)
+ expect(failed[0]?.code).toBe(1011)
+
+ // ...and the next join still starts a room: the failed attempt did not poison the chain.
+ registry.claimError = null
+ registry.claimResult = fresh.room.lease
+ const accepted: { code: number; reason: string }[] = []
+ hub.onMessage(makeConnection(hub, accepted), hello())
+ await until(() => fresh.joins === 1)
+ expect(registry.claimCalls).toHaveLength(2)
+ expect(hub.activeRooms).toHaveLength(1)
+ expect(accepted).toEqual([])
+ })
+
+ it('stops at once when another holder owns the lease (nothing can be regained)', async () => {
+ const lease = makeLease()
+ const registry = new FakeRegistry()
+ registry.outcome = 'lost'
+ const probe = makeRoom(lease)
+ const { log, entries } = makeLogger()
+ const hub = makeHub(registry, probe.room, log)
+ const closes: { code: number; reason: string }[] = []
+ const conn = makeConnection(hub, closes)
+ hub.onMessage(conn, hello())
+ await hub.waitForConnection(conn)
+
+ await hub.heartbeat(1_000)
+ expect(entries.some((e) => e.msg === 'instance lease lost to another holder')).toBe(true)
+ expect(probe.disposed).toBe(1)
+ expect(closes).toEqual([{ code: 1012, reason: 'instance_lease_lost' }])
+ expect(hub.activeRooms).toHaveLength(0)
+ // No revive: the heartbeat is not retried as a regain for a row another holder owns.
+ expect(entries.some((e) => e.msg === 'instance lease regained')).toBe(false)
+ })
+})
@@ -53,6 +53,7 @@ import { resolveClientAddress } from './net/clientAddress.js'
@@ -53,6 +53,7 @@ import { resolveClientAddress } from './net/clientAddress.js'
import { ServerMetrics } from './observability/metrics.js'
import { ServerMetrics } from './observability/metrics.js'
import { createReadiness } from './observability/readiness.js'
import { createReadiness } from './observability/readiness.js'
import { buildRelease, releaseHandler } from './observability/release.js'
import { buildRelease, releaseHandler } from './observability/release.js'
+import { createTickWatchdog } from './observability/tickWatchdog.js'
import { sharedAssetsHandler } from './net/sharedAssets.js'
import { sharedAssetsHandler } from './net/sharedAssets.js'
import { loadDefinitionPacks, scriptGrant } from './mods/contentPack.js'
import { loadDefinitionPacks, scriptGrant } from './mods/contentPack.js'
import { ModError, ModRegistry, type ModView } from './mods/registry.js'
import { ModError, ModRegistry, type ModView } from './mods/registry.js'
@@ -544,6 +545,10 @@ async function main(): Promise<void> {
@@ -544,6 +545,10 @@ async function main(): Promise<void> {
)
)
: undefined
: undefined
const createSecondaryRoom = async (slot: number, lease: InstanceLease) => {
const createSecondaryRoom = async (slot: number, lease: InstanceLease) => {
+ // Slot 0 reuses the boot store. Point its lease guard at the lease this room is actually starting
+ // with: after a lease loss the hub claims a new version, and the old closure would fence every
+ // write of the restarted room.
+ if (slot === 0) secondaryPrimaryLease = lease
const roomStore =
const roomStore =
slot === 0
slot === 0
? secondaryStore
? secondaryStore
@@ -645,6 +650,9 @@ async function main(): Promise<void> {
@@ -645,6 +650,9 @@ async function main(): Promise<void> {
const surfRegistry = new InstanceRegistry(db, info.placeId)
const surfRegistry = new InstanceRegistry(db, info.placeId)
let surfPrimaryLease: InstanceLease
let surfPrimaryLease: InstanceLease
const createSurfRoom = async (slot: number, lease: InstanceLease) => {
const createSurfRoom = async (slot: number, lease: InstanceLease) => {
+ // Slot 0's guard reads this variable; a restart after a lease loss claims a new version, so the
+ // guard must follow the lease the room is actually built with.
+ if (slot === 0) surfPrimaryLease = lease
const roomStore = openPgStore(db, {
const roomStore = openPgStore(db, {
placeId: info.placeId,
placeId: info.placeId,
instanceSlot: slot,
instanceSlot: slot,
@@ -786,7 +794,15 @@ async function main(): Promise<void> {
@@ -786,7 +794,15 @@ async function main(): Promise<void> {
[secondaryPlace, secondaryContent],
[secondaryPlace, secondaryContent],
...surfContentByPlace,
...surfContentByPlace,
])
])
- const heartbeatTimer = setInterval(() => void router.heartbeat(), 5_000)
+ const heartbeatTimer = setInterval(() => {
+ // Never let a rejected heartbeat (or a dispose that throws) become an unhandled rejection that
+ // takes the whole process with it: one place failing must not stop every other place's lease.
+ router.heartbeat().catch((err: unknown) =>
+ log.error('instance heartbeat tick failed', {
+ error: String((err as Error)?.message ?? err),
+ }),
+ )
+ }, 5_000)
heartbeatTimer.unref()
heartbeatTimer.unref()
const syncScriptView = async (view: ModView): Promise<void> => {
const syncScriptView = async (view: ModView): Promise<void> => {
@@ -1427,6 +1443,12 @@ async function main(): Promise<void> {
@@ -1427,6 +1443,12 @@ async function main(): Promise<void> {
}
}
loop()
loop()
+ // A tick loop that dies (or is never scheduled again) leaves the process answering /healthz while
+ // /api/ready's required `tick` check fails forever: production stayed 8 hours like that. Exit 1
+ // after the readiness threshold plus the grace, so systemd restarts a working process.
+ const tickWatchdog = createTickWatchdog({ metrics, log: log.child({ system: 'tick' }) })
+ tickWatchdog.start()
+
const metricsTimer = setInterval(() => metrics.logSummary(log), config.metricsLogSeconds * 1000)
const metricsTimer = setInterval(() => metrics.logSummary(log), config.metricsLogSeconds * 1000)
// ── Stop (roadmap WS-P lifecycle; manifests/services/games.json → lifecycle.shutdown) ──
// ── Stop (roadmap WS-P lifecycle; manifests/services/games.json → lifecycle.shutdown) ──
@@ -1451,6 +1473,7 @@ async function main(): Promise<void> {
@@ -1451,6 +1473,7 @@ async function main(): Promise<void> {
hard.unref()
hard.unref()
const httpDone = drainer.close(DRAIN_MS)
const httpDone = drainer.close(DRAIN_MS)
running = false
running = false
+ tickWatchdog.stop()
clearInterval(metricsTimer)
clearInterval(metricsTimer)
if (aiMatchTimer) clearInterval(aiMatchTimer)
if (aiMatchTimer) clearInterval(aiMatchTimer)
clearInterval(heartbeatTimer)
clearInterval(heartbeatTimer)
new file mode 100644
new file mode 100644
@@ -0,0 +1,94 @@
@@ -0,0 +1,94 @@
+import type { LogFields, Logger } from '@openvibe/shared'
+import { describe, expect, it } from 'vitest'
+import { createTickWatchdog, TICK_WATCHDOG_GRACE_MS } from './tickWatchdog.js'
+
+interface Entry {
+ level: 'debug' | 'info' | 'warn' | 'error'
+ msg: string
+ fields?: LogFields
+}
+
+function recordingLogger(): { log: Logger; entries: Entry[] } {
+ const entries: Entry[] = []
+ const log: Logger = {
+ debug: (msg, fields) => entries.push({ level: 'debug', msg, ...(fields ? { fields } : {}) }),
+ info: (msg, fields) => entries.push({ level: 'info', msg, ...(fields ? { fields } : {}) }),
+ warn: (msg, fields) => entries.push({ level: 'warn', msg, ...(fields ? { fields } : {}) }),
+ error: (msg, fields) => entries.push({ level: 'error', msg, ...(fields ? { fields } : {}) }),
+ child: () => log,
+ }
+ return { log, entries }
+}
+
+function setup(overrides: Partial<Parameters<typeof createTickWatchdog>[0]> = {}) {
+ const { log, entries } = recordingLogger()
+ let clock = 1_000_000
+ const exits: number[] = []
+ const metrics = { tick: 0, lastTickAt: 0 }
+ const watchdog = createTickWatchdog({
+ metrics,
+ log,
+ startedAt: clock,
+ stallMs: 5_000,
+ graceMs: 20_000,
+ now: () => clock,
+ exit: (code) => exits.push(code),
+ ...overrides,
+ })
+ return {
+ watchdog,
+ entries,
+ exits,
+ metrics,
+ advance: (ms: number) => {
+ clock += ms
+ },
+ at: () => clock,
+ }
+}
+
+describe('simulation tick watchdog', () => {
+ it('does not exit before the stall threshold plus grace', () => {
+ const { watchdog, exits, advance } = setup()
+ advance(24_999)
+ expect(watchdog.check()).toBe(false)
+ expect(exits).toEqual([])
+ })
+
+ it('exits 1 with the reason once the simulation has not ticked for the grace', () => {
+ const { watchdog, exits, entries, advance } = setup()
+ advance(25_001)
+ expect(watchdog.check()).toBe(true)
+ expect(exits).toEqual([1])
+ const error = entries.find((e) => e.level === 'error')
+ expect(error?.msg).toMatch(/simulation tick stalled/)
+ expect(error?.fields?.ageMs).toBe(25_001)
+ expect(error?.fields?.tick).toBe(0)
+ // It trips once, not on every later interval.
+ expect(watchdog.check()).toBe(true)
+ expect(exits).toEqual([1])
+ })
+
+ it('follows a tick timestamp that advances', () => {
+ const { watchdog, exits, metrics, advance } = setup()
+ advance(20_000)
+ metrics.lastTickAt = 1_020_000
+ metrics.tick = 700
+ advance(20_000)
+ expect(watchdog.check()).toBe(false)
+ expect(exits).toEqual([])
+ // Still 20 s after the last tick: inside the threshold.
+ advance(5_002)
+ expect(watchdog.check()).toBe(true)
+ expect(exits).toEqual([1])
+ })
+
+ it('stops checking after stop() (shutdown must not exit the process)', () => {
+ const { watchdog, exits, advance } = setup()
+ watchdog.start()
+ watchdog.stop()
+ advance(TICK_WATCHDOG_GRACE_MS * 2)
+ expect(watchdog.check()).toBe(false)
+ expect(exits).toEqual([])
+ })
+})
new file mode 100644
new file mode 100644
@@ -0,0 +1,92 @@
@@ -0,0 +1,92 @@
+import type { Logger } from '@openvibe/shared'
+import { TICK_STALL_MS } from './readiness.js'
+
+/**
+ * How long the simulation may stay stalled before the process exits so systemd restarts it.
+ * /api/ready already fails its required `tick` check after TICK_STALL_MS (5 s); that can be a single
+ * long GC pause or a map rebuild. Thirty more seconds is long enough for such a pass to finish, and
+ * for a room whose lease is in doubt (LEASE_RECOVERY_GRACE_MS, 60 s) to be handled by the places that
+ * still tick, but short enough that a dead tick loop never stays up for hours with /api/ready failing.
+ */
+export const TICK_WATCHDOG_GRACE_MS = 30_000
+
+/** How often the watchdog looks at the tick timestamp. */
+export const TICK_WATCHDOG_INTERVAL_MS = 5_000
+
+export interface TickWatchdogDeps {
+ /** The tick counter and when the last completed simulation tick happened (epoch ms; 0 = none). */
+ metrics: { readonly tick: number; readonly lastTickAt: number }
+ log: Logger
+ /** The clock at construction: before the first tick, the stall is measured from here. */
+ startedAt?: number
+ stallMs?: number
+ graceMs?: number
+ intervalMs?: number
+ now?: () => number
+ /** `process.exit` in production; injected so a test never kills its own runner. */
+ exit?: (code: number) => void
+}
+
+export interface TickWatchdog {
+ start(): void
+ stop(): void
+ /** Check once; true once the watchdog has asked the process to exit. */
+ check(): boolean
+}
+
+/**
+ * Watch the completed-tick timestamp and exit non-zero when it stops advancing while the process is
+ * alive. A tick loop that died (or never got scheduled again) leaves the process answering HTTP and
+ * /healthz but serving nothing; systemd only restarts it if the process exits. The watchdog waits for
+ * the readiness threshold plus its grace, logs why, and exits 1.
+ */
+export function createTickWatchdog(deps: TickWatchdogDeps): TickWatchdog {
+ const now = deps.now ?? Date.now
+ const startedAt = deps.startedAt ?? now()
+ const lastAllowedGapMs =
+ (deps.stallMs ?? TICK_STALL_MS) + (deps.graceMs ?? TICK_WATCHDOG_GRACE_MS)
+ const intervalMs = deps.intervalMs ?? TICK_WATCHDOG_INTERVAL_MS
+ const exit = deps.exit ?? ((code: number): never => process.exit(code))
+ let timer: ReturnType<typeof setInterval> | null = null
+ let tripped = false
+ let stopped = false
+
+ function stop(): void {
+ stopped = true
+ if (timer) {
+ clearInterval(timer)
+ timer = null
+ }
+ }
+
+ function check(): boolean {
+ if (tripped) return true
+ if (stopped) return false
+ const at = now()
+ const last = deps.metrics.lastTickAt
+ // Before the first tick the clock starts at construction, so a boot that never starts looping is
+ // caught too (readiness reports "simulation has not ticked yet" from the first probe on).
+ const since = last > 0 ? last : startedAt
+ const ageMs = at - since
+ if (ageMs <= lastAllowedGapMs) return false
+ tripped = true
+ stop()
+ deps.log.error('simulation tick stalled: exiting 1 so the service restarts', {
+ ageMs,
+ thresholdMs: lastAllowedGapMs,
+ tick: deps.metrics.tick,
+ })
+ exit(1)
+ return true
+ }
+
+ return {
+ start(): void {
+ if (timer || tripped || stopped) return
+ timer = setInterval(check, intervalMs)
+ timer.unref()
+ },
+ stop,
+ check,
+ }
+}
@@ -2,7 +2,13 @@ export * from './dto.js'
@@ -2,7 +2,13 @@ export * from './dto.js'
export * from './repositories.js'
export * from './repositories.js'
export { accountKey } from './accountKey.js'
export { accountKey } from './accountKey.js'
export { openPgStore, type PgPersistenceStore, type PgStoreOptions } from './pg/pgStore.js'
export { openPgStore, type PgPersistenceStore, type PgStoreOptions } from './pg/pgStore.js'
-export { InstanceRegistry, type InstanceLease, type InstanceState } from './pg/instanceRegistry.js'
+export {
+ HEARTBEAT_STATEMENT_TIMEOUT_MS,
+ InstanceRegistry,
+ type InstanceLease,
+ type InstanceState,
+ type LeaseHeartbeat,
+} from './pg/instanceRegistry.js'
export { MIGRATIONS_DIR } from './migrations.js'
export { MIGRATIONS_DIR } from './migrations.js'
export { moveCasualSaves, exportCasual, eraseCasual } from './pg/casualData.js'
export { moveCasualSaves, exportCasual, eraseCasual } from './pg/casualData.js'
export {
export {
@@ -1,7 +1,11 @@
@@ -1,7 +1,11 @@
import { beforeAll, describe, expect, it } from 'vitest'
import { beforeAll, describe, expect, it } from 'vitest'
import { openTestDb } from '../testing.js'
import { openTestDb } from '../testing.js'
import { sql, type Db } from 'openvibe-sdk/db'
import { sql, type Db } from 'openvibe-sdk/db'
-import { InstanceRegistry } from './instanceRegistry.js'
+import {
+ HEARTBEAT_STATEMENT_TIMEOUT_MS,
+ InstanceRegistry,
+ type InstanceLease,
+} from './instanceRegistry.js'
import { openPgStore } from './pgStore.js'
import { openPgStore } from './pgStore.js'
let db: Db
let db: Db
@@ -49,4 +53,83 @@ describe('PostgreSQL instance leases', () => {
@@ -49,4 +53,83 @@ describe('PostgreSQL instance leases', () => {
await expect(store.meta.set('clock', 2)).rejects.toThrow('lease lost')
await expect(store.meta.set('clock', 2)).rejects.toThrow('lease lost')
expect(await store.meta.get('clock')).toBe(1)
expect(await store.meta.get('clock')).toBe(1)
})
})
+
+ it('regains a lease whose TTL lapsed while the database missed heartbeats', async () => {
+ const registry = new InstanceRegistry(db, 'arena')
+ const claim = await registry.claim(2, 2, 'holder-a', 100, 50)
+ if (!claim) throw new Error('claim failed')
+ const running = await registry.renew(claim, 110, 50, 'running')
+ if (!running) throw new Error('renew failed')
+
+ // The database was unreachable past lease_until (160) and nobody claimed the slot meanwhile:
+ // a plain renew is fenced by lease_until, but the same holder and version may take the row back.
+ expect(await registry.renew(running, 300, 50)).toBeNull()
+ const regained = await registry.heartbeat(running, 300, 50, 'running')
+ expect(regained?.regained).toBe(true)
+ expect(regained?.lease).toMatchObject({
+ id: 'arena:2',
+ holder: 'holder-a',
+ version: running.version,
+ leaseUntil: 350,
+ })
+ if (!regained) throw new Error('regain failed')
+ // The lease guard kept working without a version bump: queued writes can land after the regain.
+ await db.tx(async () => registry.assertHeld(regained.lease, 301))
+ })
+
+ it('does not regain a lease another holder claimed while the database was unreachable', async () => {
+ const registry = new InstanceRegistry(db, 'arena')
+ const claim = await registry.claim(3, 2, 'holder-b', 500, 50)
+ if (!claim) throw new Error('claim failed')
+ const running = await registry.renew(claim, 510, 50, 'running')
+ if (!running) throw new Error('renew failed')
+ const stolen = await registry.claim(3, 2, 'holder-c', 600, 50)
+ expect(stolen).not.toBeNull()
+
+ expect(await registry.heartbeat(running, 601, 50, 'running')).toBeNull()
+ })
+
+ it('sets a short statement timeout on the heartbeat transaction', async () => {
+ const executed: string[] = []
+ const current: InstanceLease = {
+ id: 'arena:4',
+ placeId: 'arena',
+ slot: 4,
+ capacity: 2,
+ state: 'running',
+ holder: 'holder-d',
+ version: 7,
+ heartbeatAt: 10,
+ leaseUntil: 10_000,
+ }
+ const row = {
+ id: current.id,
+ place_id: current.placeId,
+ slot: current.slot,
+ capacity: current.capacity,
+ state: 'running',
+ holder: current.holder,
+ lease_version: current.version,
+ heartbeat_at: 20,
+ lease_until: 20_000,
+ }
+ const tx = {
+ exec: async (q: { parts: string[] }) => {
+ executed.push(q.parts.join('?'))
+ return 0
+ },
+ // The renew matches (`lease_until > now`); the regain fallback is not reached.
+ maybe: async (q: { parts: string[] }) =>
+ q.parts.join('?').includes('lease_until >') ? row : null,
+ }
+ const stub = {
+ store: 'postgresql',
+ tx: async <T>(fn: (t: typeof tx) => Promise<T>) => fn(tx),
+ } as unknown as Db
+ const registry = new InstanceRegistry(stub, 'arena')
+
+ const result = await registry.heartbeat(current, 20, 15_000, 'running')
+ expect(result?.regained).toBe(false)
+ expect(executed).toEqual([`SET LOCAL statement_timeout = ${HEARTBEAT_STATEMENT_TIMEOUT_MS}`])
+ })
})
})
@@ -14,6 +14,22 @@ export interface InstanceLease {
@@ -14,6 +14,22 @@ export interface InstanceLease {
leaseUntil: number
leaseUntil: number
}
}
+/** The outcome of one heartbeat: the renewed lease, and whether its TTL had already lapsed. */
+export interface LeaseHeartbeat {
+ lease: InstanceLease
+ /** True when the TTL had lapsed and no other holder had claimed the row (a regain, not a renew). */
+ regained: boolean
+}
+
+/**
+ * The heartbeat's own statement timeout in milliseconds. The lease TTL is 15 s and the heartbeat runs
+ * every 5 s (main.ts). Without a cap one slow query waits on the connection's 15 s query timeout,
+ * spans the whole TTL and loses the lease over a briefly slow database (production saw an 8.7 s
+ * heartbeat UPDATE). Two seconds leaves room for a regain attempt before the next heartbeat and is
+ * longer than a healthy round trip through PgBouncer.
+ */
+export const HEARTBEAT_STATEMENT_TIMEOUT_MS = 2_000
+
interface InstanceRow {
interface InstanceRow {
id: string
id: string
place_id: string
place_id: string
@@ -94,6 +110,48 @@ export class InstanceRegistry {
@@ -94,6 +110,48 @@ export class InstanceRegistry {
return row ? lease(row) : null
return row ? lease(row) : null
}
}
+ /**
+ * One heartbeat in a single transaction, with its own short statement timeout:
+ *
+ * 1. renew while the lease is still valid (`lease_until > now`), or
+ * 2. take the row back when this holder and version still own it and only the TTL lapsed because
+ * the database was unreachable. The UPDATE is fenced by holder and version, so a competing
+ * claim either commits first (this returns null) or sees the renewed lease (that claim fails).
+ *
+ * On the same holder and version the version is not bumped: the store's lease guard captured this
+ * lease object, and writes queued before the outage must be valid again after the regain.
+ *
+ * Returns null when another holder owns the row (or it was stopped): that lease cannot be regained.
+ * Throws on a database error; the caller treats that as "lease unconfirmed" and retries with backoff.
+ */
+ async heartbeat(
+ current: InstanceLease,
+ now: number,
+ ttlMs: number,
+ state: InstanceState = current.state,
+ ): Promise<LeaseHeartbeat | null> {
+ if (ttlMs < 1) throw new Error('invalid instance heartbeat')
+ return this.db.tx(async (tx) => {
+ // SET LOCAL is transaction-scoped, so the pool connection resumes its normal timeout afterwards.
+ // The value comes from a constant, never from input.
+ await tx.exec(sql.raw(`SET LOCAL statement_timeout = ${HEARTBEAT_STATEMENT_TIMEOUT_MS}`))
+ const renewed = await tx.maybe<InstanceRow>(sql`
+ UPDATE game_instances SET heartbeat_at = ${now}, lease_until = ${now + ttlMs}, state = ${state}
+ WHERE id = ${current.id} AND holder = ${current.holder}
+ AND lease_version = ${current.version} AND lease_until > ${now}
+ RETURNING *
+ `)
+ if (renewed) return { lease: lease(renewed), regained: false }
+ const regained = await tx.maybe<InstanceRow>(sql`
+ UPDATE game_instances SET heartbeat_at = ${now}, lease_until = ${now + ttlMs}, state = ${state}
+ WHERE id = ${current.id} AND holder = ${current.holder}
+ AND lease_version = ${current.version} AND state <> 'stopped'
+ RETURNING *
+ `)
+ return regained ? { lease: lease(regained), regained: true } : null
+ })
+ }
+
async stop(current: InstanceLease): Promise<boolean> {
async stop(current: InstanceLease): Promise<boolean> {
const row = await this.db.maybe<{ id: string }>(sql`
const row = await this.db.maybe<{ id: string }>(sql`
UPDATE game_instances SET state = 'stopped', lease_until = 0
UPDATE game_instances SET state = 'stopped', lease_until = 0
new file mode 100644
new file mode 100644
@@ -0,0 +1,3 @@
@@ -0,0 +1,3 @@
+// file: 'apps/server/src/game/instanceLease.test.ts'
+// covers: 'apps/server/src/game'
+// Generated by node test/run.js --index; do not edit.
new file mode 100644
new file mode 100644
@@ -0,0 +1,3 @@
@@ -0,0 +1,3 @@
+// file: 'apps/server/src/observability/tickWatchdog.test.ts'
+// covers: 'apps/server/src/observability'
+// Generated by node test/run.js --index; do not edit.