diff --git a/app/game-api/src/realtime/authenticatedConnection.ts b/app/game-api/src/realtime/authenticatedConnection.ts index da0bb12f..7d7960e9 100644 --- a/app/game-api/src/realtime/authenticatedConnection.ts +++ b/app/game-api/src/realtime/authenticatedConnection.ts @@ -28,6 +28,10 @@ export class AuthenticatedRealtimeConnection { return this.ended; } + whenIdle(): Promise { + return this.queue; + } + onClose(cleanup: () => void): void { if (this.ended) cleanup(); else this.cleanup.add(cleanup); diff --git a/app/game-api/src/realtime/connectionAdmission.ts b/app/game-api/src/realtime/connectionAdmission.ts new file mode 100644 index 00000000..9c235693 --- /dev/null +++ b/app/game-api/src/realtime/connectionAdmission.ts @@ -0,0 +1,93 @@ +const DEFAULT_LIMITS = { connections: 512, perUser: 8, opening: 32 } as const; +const OPENING_DEADLINE_MS = 5000; + +export interface RealtimeConnectionReservation { + canContinue(): boolean; + reserveUser(userId: string): boolean; + startStreaming(): boolean; + cancelOpening(): void; + release(): void; +} + +/** One game API process owns both pending setup and live streams; no waiting queue. */ +export class RealtimeConnectionAdmission { + private readonly limits: { connections: number; perUser: number; opening: number }; + private total = 0; + private opening = 0; + private closed = false; + private readonly byUser = new Map(); + private readonly pending = new Set<() => void>(); + + constructor(limits: Partial<{ connections: number; perUser: number; opening: number }> = {}) { + this.limits = { ...DEFAULT_LIMITS, ...limits }; + if (Object.values(this.limits).some((value) => !Number.isSafeInteger(value) || value < 1)) + throw new Error('Invalid realtime connection limits.'); + } + + begin(onTimeout: () => void): RealtimeConnectionReservation | null { + if (this.closed || this.total >= this.limits.connections || this.opening >= this.limits.opening) return null; + this.total += 1; + this.opening += 1; + let userId: string | null = null; + let streaming = false; + let cancelled = false; + let released = false; + const deadline = Date.now() + OPENING_DEADLINE_MS; + const expire = () => { + if (released || streaming || cancelled) return; + cancelled = true; + clearTimeout(timer); + onTimeout(); + }; + const timer = setTimeout(expire, OPENING_DEADLINE_MS); + timer.unref(); + this.pending.add(expire); + const canContinue = () => { + if (!released && !streaming && !cancelled && Date.now() >= deadline) expire(); + return !released && !streaming && !cancelled; + }; + return { + canContinue, + reserveUser: (id) => { + if (!canContinue() || userId !== null) return false; + const count = this.byUser.get(id) ?? 0; + if (count >= this.limits.perUser) return false; + this.byUser.set(id, count + 1); + userId = id; + return true; + }, + startStreaming: () => { + if (!canContinue() || userId === null) return false; + streaming = true; + this.opening -= 1; + clearTimeout(timer); + this.pending.delete(expire); + return true; + }, + cancelOpening: () => { + if (released || streaming) return; + cancelled = true; + clearTimeout(timer); + // Disconnect/timeout does not prove the awaited Redis/SQL operation has settled. + }, + release: () => { + if (released) return; + released = true; + this.total -= 1; + if (!streaming) this.opening -= 1; + clearTimeout(timer); + this.pending.delete(expire); + if (userId !== null) { + const count = (this.byUser.get(userId) ?? 1) - 1; + if (count === 0) this.byUser.delete(userId); + else this.byUser.set(userId, count); + } + }, + }; + } + + close(): void { + this.closed = true; + for (const expire of this.pending) expire(); + } +} diff --git a/app/game-api/src/server.ts b/app/game-api/src/server.ts index ca6e783d..81854b36 100644 --- a/app/game-api/src/server.ts +++ b/app/game-api/src/server.ts @@ -36,6 +36,7 @@ import { RedisBattleSimTransport } from './battleSim/redisTransport.js'; import { RedisRealtimeEventHub } from './realtime/eventHub.js'; import { isGameAccessBlocked } from '@sammo-ts/common/auth/sanctions'; import { AuthenticatedRealtimeConnection } from './realtime/authenticatedConnection.js'; +import { RealtimeConnectionAdmission } from './realtime/connectionAdmission.js'; import { formatSseFrame } from './realtime/sse.js'; import { shouldForwardRealtimeEvent, @@ -115,7 +116,9 @@ export const createGameApiServer = async () => { }); app.setErrorHandler(safeHttpErrorHandler); const realtimeConnections = new Set(); + const realtimeAdmission = new RealtimeConnectionAdmission(); app.addHook('preClose', async () => { + realtimeAdmission.close(); for (const connection of realtimeConnections) connection.close(); }); const postgres = createGamePostgresConnector(resolvePostgresConfigFromEnv({ schema: config.profile })); @@ -345,115 +348,165 @@ export const createGameApiServer = async () => { const query = request.query as { token?: string; scope?: string }; const subscriptionScope = query.scope === 'tournament' ? 'tournament' : 'dashboard'; const tokenFromHeader = extractBearerToken(request.headers.authorization); - let auth: GameSessionTokenPayload | null = null; - try { - auth = query.token === undefined ? await resolveRealtimeAuth(tokenFromHeader) : null; - } catch { - await reply.status(503).send({ ok: false, error: 'authentication_unavailable' }); - return; - } - - if (!auth || auth.profile !== config.profileName) { + if (query.token !== undefined || !tokenFromHeader) { await reply.status(401).send({ ok: false, error: 'unauthorized' }); return; } - - const loadViewerIdentity = async (): Promise => { - const general = await postgres.prisma.general.findFirst({ - where: { userId: auth.user.id, npcState: 0 }, - select: { id: true, cityId: true, nationId: true, officerLevel: true, meta: true, penalty: true }, - }); - if (!general) { - return { generalId: null, cityId: null, nationId: null, canReadDiplomacy: false }; + let expired = false; + let disconnected = false; + const opening = realtimeAdmission.begin(() => { + expired = true; + if (!reply.sent && !reply.raw.destroyed) { + reply.header('Retry-After', '15'); + void reply.status(503).send({ ok: false, error: 'realtime_unavailable' }); } - const nation = - general.nationId > 0 - ? await postgres.prisma.nation.findUnique({ - where: { id: general.nationId }, - select: { meta: true }, - }) - : null; - return { - generalId: general.id, - cityId: general.cityId, - nationId: general.nationId, - canReadDiplomacy: Boolean(nation && resolveNationPermission(general, nation.meta, false) >= 3), - }; - }; - let viewerIdentity = await loadViewerIdentity(); - - reply.hijack(); - const requestOrigin = request.headers.origin; - if (typeof requestOrigin === 'string' && requestOrigin.length > 0) { - // Hijacked SSE responses bypass Fastify's normal CORS response hook. - reply.raw.setHeader('Access-Control-Allow-Origin', requestOrigin); - reply.raw.setHeader('Access-Control-Allow-Credentials', 'true'); - reply.raw.setHeader('Vary', 'Origin'); + }); + if (!opening) { + await reply.header('Retry-After', '15').status(503).send({ ok: false, error: 'realtime_unavailable' }); + return; } - reply.raw.setHeader('Content-Type', 'text/event-stream'); - reply.raw.setHeader('Cache-Control', 'no-cache'); - reply.raw.setHeader('Connection', 'keep-alive'); - reply.raw.setHeader('X-Accel-Buffering', 'no'); - request.raw.setTimeout(0); - reply.raw.setTimeout?.(0); - reply.raw.flushHeaders?.(); + const cancelOpening = () => { + disconnected = true; + opening.cancelOpening(); + }; + request.raw.once('aborted', cancelOpening); + reply.raw.once('close', cancelOpening); + let transferred = false; + try { + let auth: GameSessionTokenPayload | null = null; + try { + auth = await resolveRealtimeAuth(tokenFromHeader); + } catch { + if (opening.canContinue() && !reply.raw.destroyed) + await reply.status(503).send({ ok: false, error: 'authentication_unavailable' }); + return; + } + if (!opening.canContinue() || reply.raw.destroyed) return; - const connection = new AuthenticatedRealtimeConnection(auth, reply.raw, () => - resolveRealtimeAuth(tokenFromHeader) - ); - realtimeConnections.add(connection); - connection.onClose(() => { - realtimeConnections.delete(connection); - }); - connection.enqueue(async () => formatSseFrame({ event: 'ready', data: '{}' })); - const unsubscribe = realtimeHub.subscribe((event) => { - if (!shouldForwardRealtimeEvent(event, subscriptionScope)) return; - connection.enqueue(async (currentAuth) => { - const identities = [viewerIdentity]; - if (shouldReloadRealtimeViewerIdentity(event, viewerIdentity)) { - const nextIdentity = await loadViewerIdentity(); - identities.push(nextIdentity); - viewerIdentity = nextIdentity; - } - let refreshGrant: string | undefined; - const publicEvent = toPublicRealtimeEvent(event, identities, () => { - refreshGrant ??= createRealtimeAccessGrant(currentAuth, config.profileName, config.gameTokenSecret); - return refreshGrant; + if (!auth || auth.profile !== config.profileName) { + await reply.status(401).send({ ok: false, error: 'unauthorized' }); + return; + } + if (!opening.reserveUser(auth.user.id)) { + if (opening.canContinue()) + await reply + .header('Retry-After', '15') + .status(429) + .send({ ok: false, error: 'realtime_connection_limit' }); + return; + } + + const loadViewerIdentity = async (): Promise => { + const general = await postgres.prisma.general.findFirst({ + where: { userId: auth.user.id, npcState: 0 }, + select: { id: true, cityId: true, nationId: true, officerLevel: true, meta: true, penalty: true }, }); - if (!publicEvent || connection.closed) return null; - if (refreshGrant) { - try { - await registerRealtimeAccessGrant(redis.client, refreshGrant, config.profileName); - } catch { - // An unregistered grant falls back to the scored refresh path. + if (!general) { + return { generalId: null, cityId: null, nationId: null, canReadDiplomacy: false }; + } + const nation = + general.nationId > 0 + ? await postgres.prisma.nation.findUnique({ + where: { id: general.nationId }, + select: { meta: true }, + }) + : null; + return { + generalId: general.id, + cityId: general.cityId, + nationId: general.nationId, + canReadDiplomacy: Boolean(nation && resolveNationPermission(general, nation.meta, false) >= 3), + }; + }; + let viewerIdentity = await loadViewerIdentity(); + if (!opening.canContinue() || reply.raw.destroyed || !opening.startStreaming()) return; + + reply.hijack(); + const requestOrigin = request.headers.origin; + if (typeof requestOrigin === 'string' && requestOrigin.length > 0) { + // Hijacked SSE responses bypass Fastify's normal CORS response hook. + reply.raw.setHeader('Access-Control-Allow-Origin', requestOrigin); + reply.raw.setHeader('Access-Control-Allow-Credentials', 'true'); + reply.raw.setHeader('Vary', 'Origin'); + } + reply.raw.setHeader('Content-Type', 'text/event-stream'); + reply.raw.setHeader('Cache-Control', 'no-cache'); + reply.raw.setHeader('Connection', 'keep-alive'); + reply.raw.setHeader('X-Accel-Buffering', 'no'); + request.raw.setTimeout(0); + reply.raw.setTimeout?.(0); + reply.raw.flushHeaders?.(); + + const connection = new AuthenticatedRealtimeConnection(auth, reply.raw, () => + resolveRealtimeAuth(tokenFromHeader) + ); + realtimeConnections.add(connection); + connection.onClose(() => { + realtimeConnections.delete(connection); + // A disconnected socket can still own an in-flight Redis/SQL operation. + void connection.whenIdle().then(opening.release, opening.release); + }); + transferred = true; + connection.enqueue(async () => formatSseFrame({ event: 'ready', data: '{}' })); + const unsubscribe = realtimeHub.subscribe((event) => { + if (!shouldForwardRealtimeEvent(event, subscriptionScope)) return; + connection.enqueue(async (currentAuth) => { + const identities = [viewerIdentity]; + if (shouldReloadRealtimeViewerIdentity(event, viewerIdentity)) { + const nextIdentity = await loadViewerIdentity(); + identities.push(nextIdentity); + viewerIdentity = nextIdentity; } - } - return formatSseFrame({ event: publicEvent.type, data: JSON.stringify(publicEvent) }); - }); - }); - connection.onClose(unsubscribe); - let heartbeatPending = false; - const heartbeat = setInterval(() => { - if (heartbeatPending || connection.closed) return; - heartbeatPending = true; - connection.enqueue(async () => { - try { - return formatSseFrame({ - event: 'ping', - data: JSON.stringify({ turnEngineRunning: await turnEngineStatus.get() }), + let refreshGrant: string | undefined; + const publicEvent = toPublicRealtimeEvent(event, identities, () => { + refreshGrant ??= createRealtimeAccessGrant( + currentAuth, + config.profileName, + config.gameTokenSecret + ); + return refreshGrant; }); - } finally { - heartbeatPending = false; - } + if (!publicEvent || connection.closed) return null; + if (refreshGrant) { + try { + await registerRealtimeAccessGrant(redis.client, refreshGrant, config.profileName); + } catch { + // An unregistered grant falls back to the scored refresh path. + } + } + return formatSseFrame({ event: publicEvent.type, data: JSON.stringify(publicEvent) }); + }); }); - }, 15_000); - connection.onClose(() => { - clearInterval(heartbeat); - }); - request.raw.once('aborted', connection.close); - connection.onClose(() => { - request.raw.off('aborted', connection.close); - }); + connection.onClose(unsubscribe); + let heartbeatPending = false; + const heartbeat = setInterval(() => { + if (heartbeatPending || connection.closed) return; + heartbeatPending = true; + connection.enqueue(async () => { + try { + return formatSseFrame({ + event: 'ping', + data: JSON.stringify({ turnEngineRunning: await turnEngineStatus.get() }), + }); + } finally { + heartbeatPending = false; + } + }); + }, 15_000); + connection.onClose(() => { + clearInterval(heartbeat); + }); + request.raw.once('aborted', connection.close); + connection.onClose(() => { + request.raw.off('aborted', connection.close); + }); + } catch (error) { + if (!expired && !disconnected) throw error; + } finally { + request.raw.off('aborted', cancelOpening); + reply.raw.off('close', cancelOpening); + if (!transferred) opening.release(); + } }); app.get('/healthz', async (_request, reply) => { diff --git a/app/game-api/test/realtimeConnectionAdmission.test.ts b/app/game-api/test/realtimeConnectionAdmission.test.ts new file mode 100644 index 00000000..aa654c95 --- /dev/null +++ b/app/game-api/test/realtimeConnectionAdmission.test.ts @@ -0,0 +1,151 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { PassThrough } from 'node:stream'; +import type { ServerResponse } from 'node:http'; +import type { GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken'; +import { RealtimeConnectionAdmission } from '../src/realtime/connectionAdmission.js'; +import { AuthenticatedRealtimeConnection } from '../src/realtime/authenticatedConnection.js'; + +afterEach(() => vi.useRealTimers()); + +describe('realtime connection admission', () => { + it('reserves the process total before authentication and counts live streams too', () => { + const budget = new RealtimeConnectionAdmission({ connections: 2, perUser: 2, opening: 2 }); + const first = budget.begin(vi.fn())!; + const second = budget.begin(vi.fn())!; + expect(first.reserveUser('actor')).toBe(true); + expect(first.startStreaming()).toBe(true); + expect(budget.begin(vi.fn())).toBeNull(); + second.release(); + const replacement = budget.begin(vi.fn())!; + expect(replacement).not.toBeNull(); + replacement.release(); + first.release(); + }); + + it('shares the authenticated actor limit while allowing another actor', () => { + const budget = new RealtimeConnectionAdmission({ connections: 5, perUser: 1, opening: 5 }); + const first = budget.begin(vi.fn())!; + expect(first.reserveUser('one')).toBe(true); + const rejected = budget.begin(vi.fn())!; + expect(rejected.reserveUser('one')).toBe(false); + expect(rejected.startStreaming()).toBe(false); + rejected.release(); + const other = budget.begin(vi.fn())!; + expect(other.reserveUser('two')).toBe(true); + first.release(); + first.release(); + const replacement = budget.begin(vi.fn())!; + expect(replacement.reserveUser('one')).toBe(true); + other.release(); + replacement.release(); + }); + + it('keeps timed-out and disconnected backend work admitted until it actually settles', async () => { + vi.useFakeTimers(); + const budget = new RealtimeConnectionAdmission({ connections: 5, perUser: 5, opening: 2 }); + const timeout = vi.fn(); + const stalled = budget.begin(timeout)!; + const disconnected = budget.begin(vi.fn())!; + expect(stalled.reserveUser('stalled')).toBe(true); + disconnected.cancelOpening(); + await vi.advanceTimersByTimeAsync(5000); + expect(timeout).toHaveBeenCalledTimes(1); + expect(stalled.canContinue()).toBe(false); + expect(stalled.startStreaming()).toBe(false); + expect(disconnected.canContinue()).toBe(false); + expect(budget.begin(vi.fn())).toBeNull(); + stalled.release(); + const recovered = budget.begin(vi.fn())!; + expect(recovered.reserveUser('stalled')).toBe(true); + expect(recovered.startStreaming()).toBe(true); + await vi.advanceTimersByTimeAsync(5000); + expect(timeout).toHaveBeenCalledTimes(1); + disconnected.release(); + recovered.release(); + }); + + it('stops pending setup on shutdown and permanently refuses new admissions', () => { + const budget = new RealtimeConnectionAdmission(); + const timeout = vi.fn(); + const pending = budget.begin(timeout)!; + budget.close(); + budget.close(); + expect(timeout).toHaveBeenCalledTimes(1); + expect(pending.canContinue()).toBe(false); + expect(budget.begin(vi.fn())).toBeNull(); + pending.release(); + }); + + it('keeps a closed stream counted while its in-flight authentication is still pending', async () => { + const budget = new RealtimeConnectionAdmission({ connections: 1, perUser: 1, opening: 1 }); + const slot = budget.begin(vi.fn())!; + expect(slot.reserveUser('actor')).toBe(true); + expect(slot.startStreaming()).toBe(true); + let finish!: () => void; + let started!: () => void; + const inFlight = new Promise((resolve) => { + finish = resolve; + }); + const running = new Promise((resolve) => { + started = resolve; + }); + const auth: GameSessionTokenPayload = { + version: 1, + sessionId: 'session', + profile: 'che:default', + issuedAt: new Date().toISOString(), + expiresAt: new Date(Date.now() + 60_000).toISOString(), + user: { + id: 'actor', + username: 'actor', + displayName: 'actor', + roles: ['user'], + createdAt: new Date().toISOString(), + }, + sanctions: {}, + }; + const response = new PassThrough(); + const connection = new AuthenticatedRealtimeConnection( + auth, + response as unknown as ServerResponse, + async () => { + started(); + await inFlight; + return auth; + } + ); + connection.onClose(() => { + void connection.whenIdle().then(slot.release, slot.release); + }); + connection.enqueue(async () => 'event: ready\ndata: {}\n\n'); + await running; + connection.close(); + expect(budget.begin(vi.fn())).toBeNull(); + finish(); + await connection.whenIdle(); + const replacement = budget.begin(vi.fn())!; + expect(replacement.reserveUser('actor')).toBe(true); + replacement.release(); + }); + + it('enforces the production 512 total and 32 pending openings without accumulating a queue', () => { + const budget = new RealtimeConnectionAdmission(); + const pending = Array.from({ length: 32 }, () => budget.begin(vi.fn())!); + expect(pending.every(Boolean)).toBe(true); + expect(budget.begin(vi.fn())).toBeNull(); + for (const slot of pending) slot.release(); + const live = Array.from({ length: 512 }, (_, index) => { + const slot = budget.begin(vi.fn())!; + expect(slot.reserveUser(`actor-${index}`)).toBe(true); + expect(slot.startStreaming()).toBe(true); + return slot; + }); + expect(budget.begin(vi.fn())).toBeNull(); + live[0]!.release(); + const replacement = budget.begin(vi.fn())!; + expect(replacement.reserveUser('replacement')).toBe(true); + expect(replacement.startStreaming()).toBe(true); + for (const slot of live) slot.release(); + replacement.release(); + }); +}); diff --git a/app/game-api/test/securityTransport.integration.test.ts b/app/game-api/test/securityTransport.integration.test.ts index d2936669..cb8ea034 100644 --- a/app/game-api/test/securityTransport.integration.test.ts +++ b/app/game-api/test/securityTransport.integration.test.ts @@ -3589,6 +3589,168 @@ integration('game API security over HTTP transport', () => { return { abort, reader }; }; + it('SSE limits simultaneous openings for one authenticated user', async () => { + const actor = `security-http-sse-concurrent-${process.pid}`; + const token = await createAccessToken('sse-concurrent', {}, actor); + const aborts = Array.from({ length: 10 }, () => new AbortController()); + try { + const responses = await Promise.all( + aborts.map((abort) => + fetch(`${baseUrl}${server!.config.eventsPath}`, { + headers: { authorization: `Bearer ${token}` }, + signal: abort.signal, + }) + ) + ); + expect(responses.filter((response) => response.status === 200)).toHaveLength(8); + expect(responses.filter((response) => response.status === 429)).toHaveLength(2); + for (const response of responses.filter((item) => item.status === 429)) { + expect(response.headers.get('retry-after')).toBe('15'); + expect(response.headers.get('content-type')).not.toBe('text/event-stream'); + expect(await response.json()).toEqual({ ok: false, error: 'realtime_connection_limit' }); + } + } finally { + for (const abort of aborts) abort.abort(); + await accessTokenStore.revoke(token); + } + }); + + it('SSE limits the same user across tokens and scopes and recovers after disconnect', async () => { + const actor = `security-http-sse-shared-${process.pid}`; + const token = await createAccessToken('sse-shared-a', {}, actor); + const otherToken = await createAccessToken('sse-shared-b', {}, actor); + const otherActorToken = await createAccessToken('sse-independent', {}, `${actor}-other`); + const streams: Awaited>[] = []; + const replacementAbort = new AbortController(); + try { + for (let index = 0; index < 8; index += 1) streams.push(await openRealtime(token)); + const rejected = await fetch(`${baseUrl}${server!.config.eventsPath}?scope=tournament`, { + headers: { authorization: `Bearer ${otherToken}` }, + }); + expect(rejected.status).toBe(429); + expect(await rejected.json()).toEqual({ ok: false, error: 'realtime_connection_limit' }); + streams.push(await openRealtime(otherActorToken)); + await streams[0]!.reader.cancel(); + streams[0]!.abort.abort(); + let replacement: Response | undefined; + await expect + .poll(async () => { + const response = await fetch(`${baseUrl}${server!.config.eventsPath}?scope=tournament`, { + headers: { authorization: `Bearer ${otherToken}` }, + signal: replacementAbort.signal, + }); + if (response.status === 200) replacement = response; + else await response.arrayBuffer(); + return response.status; + }) + .toBe(200); + const ready = await replacement!.body!.getReader().read(); + expect(new TextDecoder().decode(ready.value)).toContain('event: ready'); + } finally { + replacementAbort.abort(); + for (const stream of streams) stream.abort.abort(); + await Promise.all([token, otherToken, otherActorToken].map((value) => accessTokenStore.revoke(value))); + } + }); + + it('SSE bounds pending openings after disconnect and backend deadline until real SQL work settles', async () => { + if (!/^[A-Za-z0-9_]+$/.test(profileId)) throw new Error('Invalid isolated schema identifier'); + const actor = `security-http-sse-pending-${process.pid}`; + const tokens = await Promise.all( + Array.from({ length: 33 }, (_, index) => createAccessToken(`sse-pending-${index}`, {}, `${actor}-${index}`)) + ); + const aborts = Array.from({ length: 33 }, () => new AbortController()); + let unlock!: () => void; + const release = new Promise((resolve) => { + unlock = resolve; + }); + let ready!: (tx: GamePrisma.TransactionClient) => void; + let lockError!: (error: unknown) => void; + const locked = new Promise((resolve, reject) => { + ready = resolve; + lockError = reject; + }); + // Only the guarded, dedicated fixture schema is locked. No runtime or Ref table is touched. + const locking = db.$transaction( + async (tx) => { + await tx.$executeRawUnsafe(`LOCK TABLE "${profileId}"."general" IN ACCESS EXCLUSIVE MODE`); + ready(tx); + await release; + }, + { timeout: 20_000 } + ); + void locking.catch(lockError); + try { + const tx = await locked; + const first = fetch(`${baseUrl}${server!.config.eventsPath}`, { + headers: { authorization: `Bearer ${tokens[0]!}` }, + signal: aborts[0]!.signal, + }).then( + () => false, + () => true + ); + await expect + .poll( + async () => { + const rows = await tx.$queryRaw>` + SELECT COUNT(*)::int AS count FROM pg_locks + WHERE NOT granted AND relation = ${`${profileId}.general`}::regclass + AND database = (SELECT oid FROM pg_database WHERE datname = current_database()) + `; + return rows[0]!.count; + }, + { timeout: 3000 } + ) + .toBeGreaterThan(0); + const startedAt = Date.now(); + const waiting = tokens.slice(1, 32).map((token, index) => + fetch(`${baseUrl}${server!.config.eventsPath}`, { + headers: { authorization: `Bearer ${token}` }, + signal: aborts[index + 1]!.signal, + }) + ); + aborts[0]!.abort(); + expect(await first).toBe(true); + const responses = await Promise.all(waiting); + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(4500); + expect(Date.now() - startedAt).toBeLessThan(8500); + for (const response of responses) { + expect(response.status).toBe(503); + expect(await response.json()).toEqual({ ok: false, error: 'realtime_unavailable' }); + } + const refusedAt = Date.now(); + const refused = await fetch(`${baseUrl}${server!.config.eventsPath}`, { + headers: { authorization: `Bearer ${tokens[32]!}` }, + signal: aborts[32]!.signal, + }); + expect(refused.status).toBe(503); + expect(refused.headers.get('retry-after')).toBe('15'); + expect(Date.now() - refusedAt).toBeLessThan(2000); + await refused.arrayBuffer(); + unlock(); + await locking; + let recovered: Response | undefined; + await expect + .poll(async () => { + const response = await fetch(`${baseUrl}${server!.config.eventsPath}`, { + headers: { authorization: `Bearer ${tokens[32]!}` }, + signal: aborts[32]!.signal, + }); + if (response.status === 200) recovered = response; + else await response.arrayBuffer(); + return response.status; + }) + .toBe(200); + const frame = await recovered!.body!.getReader().read(); + expect(new TextDecoder().decode(frame.value)).toContain('event: ready'); + } finally { + unlock(); + for (const abort of aborts) abort.abort(); + await locking; + await Promise.all(tokens.map((token) => accessTokenStore.revoke(token))); + } + }, 15_000); + it('SSE accepts header auth and rejects query or cross-profile credentials', async () => { const token = await createAccessToken('sse-query', {}); const queryOnly = await fetch(`${baseUrl}${server!.config.eventsPath}?token=${encodeURIComponent(token)}`); diff --git a/docs/release-operations.md b/docs/release-operations.md index 08d40013..e000ab0e 100644 --- a/docs/release-operations.md +++ b/docs/release-operations.md @@ -557,8 +557,35 @@ user/session/profile과 게임 접근 제재를 확인한다. flush watermark는 대기 중인 연결을 닫고 Redis 장애는 fail closed로 처리한다. 연결별 대기 작업은 32개, frame과 socket buffer는 64 KiB로 제한하고 slow consumer를 닫는다. 재연결 뒤 메인은 기존 갱신 큐로 snapshot을 복구한다. 게임 계산·RNG·DB schema와 -read-model invalidation/grant의 사용자 범위는 변경하지 않는다. 연결 수 전체의 -상한이나 온라인 인증 rate limit까지 구현한 것으로 해석하지 않는다. +read-model invalidation/grant의 사용자 범위는 변경하지 않는다. 전체 연결 admission은 +아래 별도 계약을 따르며, 일반 게임 HTTP 인증의 전역 rate limit 증거로 확대하지 않는다. + +### SSE 연결과 준비 작업 상한 (2026-10-09) + +game API의 `RealtimeConnectionAdmission`은 프로세스별로 다음 용량을 소유한다. + +| 범위 | 상한 | +| --------------------------------------------------------------- | ----- | +| 같은 인증 사용자: token·session·scope를 바꾸어도 공유 | 8개 | +| 인증 중/준비 중, 열린 연결, 닫힌 연결의 미완료 작업을 합친 전체 | 512개 | +| 인증·viewer 조회가 아직 끝나지 않은 연결 준비 | 32개 | + +인증 actor는 검증한 token의 user ID로 결정하며 client query/user ID를 신뢰하지 않는다. +전체/준비 용량은 첫 async 인증 조회 전에 예약하고 사용자 용량은 viewer DB 조회 전에 +예약한다. 대기 큐를 만들지 않고 사용자 초과는 429 `realtime_connection_limit`, +전체/준비 초과는 503 `realtime_unavailable`과 `Retry-After: 15`를 반환한다. +이미 열린 정상 연결은 새 요청의 초과 때문에 닫지 않는다. + +연결 준비는 5초 뒤 503으로 실패한다. 호출자 timeout/client disconnect가 실제 Redis/SQL +완료를 뜻하지 않으므로 해당 준비 용량은 작업이 settle할 때까지 유지한다. 늦게 완료한 +작업이 이미 끝난 HTTP 요청에서 SSE를 새로 열지 못한다. 열린 socket의 종료도 구독과 +heartbeat를 즉시 정리하지만 미완료 frame 작업이 끝날 때까지 사용자/전체 용량을 +보유한다. app shutdown은 새 admission과 늦은 준비를 거절하고 기존 stream을 닫는다. + +이는 한 game API process/profile의 자원 상한이며 DB pool connection 수를 사용자 수로 +환산한 값이나 다중 replica/모든 profile에 걸친 전역 제한이 아니다. 일반 request RPS와 +backend의 모든 장애 복구를 보장하지 않는다. 기존 header auth, 만료/회수·개인정보 +필터와 refresh grant는 유지한다. 비활성화 env key와 credential은 추가하지 않는다. Game/Gateway HTTP logger는 raw URL/query/header/body와 error message/cause/stack 대신 static route, request ID, status와 허용된 error code만 남긴다. Fastify의