diff --git a/apps/notify/src/notifications-hub.ts b/apps/notify/src/notifications-hub.ts index 17169c7..ae70fe7 100644 --- a/apps/notify/src/notifications-hub.ts +++ b/apps/notify/src/notifications-hub.ts @@ -99,6 +99,22 @@ export interface HubState { pending: Array<{ playerId: number; count: number; latest: string }> } +/** + * How many queued notifications one offline player may accumulate. Past this the OLDEST + * are dropped, newest kept. + * + * The queue used to be unbounded, which is fine while it only holds what one absent player + * missed — and is not fine at all when delivery breaks (a reset leaves connection rows with + * no sockets, so live players' notifications queue too; see pruneDeadConnections). Then it + * grows with every notification the server sends and `flushPending` reads the lot into + * memory and writes it to a socket in a single event, which is its own way to take the + * object down. + * + * Newest-wins because these are notifications: a player returning to 500 of them is served + * no better by the 501st-oldest, and the recent ones are the ones still worth acting on. + */ +export const MAX_PENDING_PER_PLAYER = 500 + /** The Coach system account — the `FromPlayerId` on a coach message (see coachMessageAll). */ const COACH_PLAYER_ID = 1 @@ -108,7 +124,13 @@ const COACH_MESSAGE_TYPE = 100 export class NotificationsHub extends DurableObject { constructor(ctx: DurableObjectState, env: Env) { super(ctx, env) - void ctx.blockConcurrencyWhile(async () => { + // Not floated: a rejection here is how a storage fault at wake-up surfaces + // ("Internal error in Durable Object storage caused object to be reset"), and an + // unhandled one tells us nothing about which object died or why. Catching doesn't + // prevent the reset — a throw inside blockConcurrencyWhile resets the object either + // way — it just leaves a trace when it happens. + ctx + .blockConcurrencyWhile(async () => { this.ctx.storage.sql.exec(` CREATE TABLE IF NOT EXISTS subscriptions ( connectionId TEXT NOT NULL, @@ -128,9 +150,64 @@ export class NotificationsHub extends DurableObject { ); CREATE INDEX IF NOT EXISTS idx_owner_player ON connection_owner(playerId); `) + this.pruneDeadConnections() + }) + .catch((err: unknown) => { + console.error('hub: storage init failed', { + error: err instanceof Error ? err.message : String(err), + }) + }) + } + + /** + * Drop connection rows with no socket behind them. + * + * A row is written when a socket is accepted and removed when it closes, so the two + * only diverge when sockets die without `webSocketClose` running — which is exactly + * what a Durable Object RESET does: the object is rebuilt, every socket is gone, and + * `connection_owner` / `subscriptions` still name all of them. + * + * Left alone those rows are worse than useless. `deliverToPlayer` finds connection ids + * for a player, sends to no sockets, and reports 0 delivered — so `notifyPlayer` QUEUES + * for players who are online and reconnected long ago, and `pending` grows on every + * notification the server sends. One reset then degrades delivery indefinitely. + * + * Safe at construction because hibernation does NOT lose sockets: `getWebSockets()` + * returns them on wake, so anything missing from it is genuinely gone. And a row whose + * socket has gone can never deliver anything anyway — dropping it costs nothing. + */ + private pruneDeadConnections(): void { + const live = new Set() + for (const ws of this.ctx.getWebSockets()) { + const state = ws.deserializeAttachment() as SocketState | null + if (state) live.add(state.connectionId) + } + + const known = this.ctx.storage.sql + .exec<{ connectionId: string }>( + `SELECT connectionId FROM connection_owner + UNION + SELECT connectionId FROM subscriptions` + ) + .toArray() + .map((r) => r.connectionId) + + const dead = known.filter((connectionId) => !live.has(connectionId)) + if (dead.length === 0) return + + for (const connectionId of dead) this.forgetConnection(connectionId) + console.warn('hub: pruned connections with no live socket', { + pruned: dead.length, + live: live.size, }) } + /** Forget one connection: its ownership row and everything it subscribed to. */ + private forgetConnection(connectionId: string): void { + this.ctx.storage.sql.exec('DELETE FROM subscriptions WHERE connectionId = ?', connectionId) + this.ctx.storage.sql.exec('DELETE FROM connection_owner WHERE connectionId = ?', connectionId) + } + /** WebSocket upgrade entrypoint — the worker forwards `/hub/v1` here. */ override async fetch(request: Request): Promise { if ((request.headers.get('Upgrade') ?? '').toLowerCase() !== 'websocket') { @@ -198,14 +275,7 @@ export class NotificationsHub extends DurableObject { if (state) { // Mirrors OnDisconnected: drop this connection's subscriptions, which // also removes it from every player's connection set. - this.ctx.storage.sql.exec( - 'DELETE FROM subscriptions WHERE connectionId = ?', - state.connectionId - ) - this.ctx.storage.sql.exec( - 'DELETE FROM connection_owner WHERE connectionId = ?', - state.connectionId - ) + this.forgetConnection(state.connectionId) } try { ws.close() @@ -384,16 +454,43 @@ export class NotificationsHub extends DurableObject { } if (delivered === 0) { - this.ctx.storage.sql.exec( - 'INSERT INTO pending (playerId, payload) VALUES (?, ?)', - playerId, - payload - ) + this.queuePending(playerId, payload) return { delivered: 0, queued: true } } return { delivered, queued: false } } + /** + * Queue a notification for a player who wasn't reachable, keeping the queue to + * {@link MAX_PENDING_PER_PLAYER}. Trimming is oldest-first and happens on the write, so + * the bound holds no matter how the queue got long. + */ + private queuePending(playerId: number, payload: string): void { + this.ctx.storage.sql.exec( + 'INSERT INTO pending (playerId, payload) VALUES (?, ?)', + playerId, + payload + ) + const trimmed = this.ctx.storage.sql + .exec( + `DELETE FROM pending + WHERE playerId = ?1 AND id NOT IN ( + SELECT id FROM pending WHERE playerId = ?1 ORDER BY id DESC LIMIT ?2 + ) + RETURNING id`, + playerId, + MAX_PENDING_PER_PLAYER + ) + .toArray().length + if (trimmed > 0) { + console.warn('hub: pending queue full, dropped oldest notifications', { + playerId, + dropped: trimmed, + cap: MAX_PENDING_PER_PLAYER, + }) + } + } + /** * Send a notification to a player's live sockets and, unlike {@link notifyPlayer}, * NEVER queue it when they're offline — the ephemeral "SendWebsocket" send. For @@ -580,7 +677,16 @@ export class NotificationsHub extends DurableObject { let delivered = 0 for (const connectionId of connectionIds) { - for (const ws of this.ctx.getWebSockets(connectionId)) { + const sockets = this.ctx.getWebSockets(connectionId) + // A connection id with no socket behind it is a row left over from a close we + // never saw — a reset, most likely (see pruneDeadConnections). Repaired the + // moment we trip over it, or every later send for this player queues instead of + // delivering, forever. + if (sockets.length === 0) { + this.forgetConnection(connectionId) + continue + } + for (const ws of sockets) { ws.send(this.invocation('Notification', [payload])) delivered++ } diff --git a/apps/notify/src/notify.app.ts b/apps/notify/src/notify.app.ts index c5ab65d..ba3f872 100644 --- a/apps/notify/src/notify.app.ts +++ b/apps/notify/src/notify.app.ts @@ -20,6 +20,62 @@ import type { App } from './context' /** The hub state is global → one DO instance. */ const HUB_INSTANCE = 'global' +/** + * How many times to re-issue a hub call Cloudflare aborted mid-flight. + * + * The platform occasionally resets a Durable Object under us — "Internal error in Durable + * Object storage caused object to be reset", carrying `retryable: true` and + * `durableObjectReset: true`. It is not a fault in the call: the object is rebuilt from its + * last durable state and the same call succeeds. Without a retry it surfaces as a 500 and + * the notification is simply lost, which is why these arrive periodically rather than + * predictably. + * + * Two attempts after the first is plenty — a reset that persists past that is an outage, + * not a blip, and the caller should hear about it. + */ +const HUB_RETRIES = 2 + +/** + * Whether an error is one Cloudflare says to retry. `retryable` is set on the error the + * runtime throws; `durableObjectReset` accompanies the reset flavour of it. Anything else — + * a bug in a hub method, a bad argument — is thrown straight back, since retrying it would + * only produce the same failure more slowly. + */ +function isRetryableHubError(err: unknown): boolean { + if (typeof err !== 'object' || err === null) return false + const fields = err as { retryable?: unknown; durableObjectReset?: unknown } + return fields.retryable === true || fields.durableObjectReset === true +} + +/** + * Call the hub, retrying a reset. A FRESH stub per attempt: the one that threw is bound to + * the object that just died. + * + * Safe to retry because a reset rolls the object back — the aborted call left nothing + * behind — and because every frame the hub sends is a complete, absolute statement (a + * notification, not a delta), so a duplicate is at worst a repeat and never a drift. + */ +async function hubCall( + c: Context, + call: (hub: ReturnType) => Promise +): Promise { + let lastError: unknown + for (let attempt = 0; attempt <= HUB_RETRIES; attempt++) { + try { + return await call(c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE)) + } catch (err) { + if (!isRetryableHubError(err)) throw err + lastError = err + logger.warn('hub call reset, retrying', { + attempt: attempt + 1, + of: HUB_RETRIES + 1, + error: err instanceof Error ? err.message : String(err), + }) + } + } + throw lastError +} + /** * A valid notification `Id` — a client-defined string tag (e.g. "AccountUpdate") * or a numeric code. An empty string is treated as missing. @@ -125,7 +181,9 @@ const app = new Hono() // DO's proof of identity, so a client sending its own must not be believed. request.headers.set(OWNER_HEADER, String(playerId)) - return c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).fetch(request) + // The upgrade is retried too: it is a bodyless GET, so re-issuing it is free, and a + // reset here would otherwise fail the client's connect outright. + return hubCall(c, (hub) => hub.fetch(new Request(request))) }) // ---- Internal service-to-service send/broadcast -------------------------- @@ -144,11 +202,8 @@ const app = new Hono() if (!body || typeof body.playerId !== 'number' || !isNotificationType(body.notificationType)) { return c.json({ error: 'playerId and notificationType are required' }, 400) } - const result = await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).notifyPlayer( - body.playerId, - body.notificationType, - body.data - ) + const { playerId, notificationType, data } = body + const result = await hubCall(c, (hub) => hub.notifyPlayer(playerId, notificationType, data)) return c.json({ success: true, ...result }) }) @@ -159,10 +214,8 @@ const app = new Hono() if (!body || !isNotificationType(body.notificationType)) { return c.json({ error: 'notificationType is required' }, 400) } - const result = await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).broadcast( - body.notificationType, - body.data - ) + const { notificationType, data } = body + const result = await hubCall(c, (hub) => hub.broadcast(notificationType, data)) return c.json({ success: true, ...result }) }) @@ -171,8 +224,7 @@ const app = new Hono() const body = await c.req.json<{ messageContent?: string }>().catch(() => null) const content = typeof body?.messageContent === 'string' ? body.messageContent.trim() : '' if (content === '') return c.json({ error: 'messageContent is required' }, 400) - const result = - await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).coachMessageAll(content) + const result = await hubCall(c, (hub) => hub.coachMessageAll(content)) return c.json({ success: true, ...result }) }) @@ -188,10 +240,8 @@ const app = new Hono() return c.json({ error: 'playerId is required' }, 400) } if (content === '') return c.json({ error: 'messageContent is required' }, 400) - const result = await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).coachMessage( - body.playerId, - content - ) + const { playerId } = body + const result = await hubCall(c, (hub) => hub.coachMessage(playerId, content)) return c.json({ success: true, ...result }) }) @@ -199,7 +249,7 @@ const app = new Hono() // didn't arrive: which connections are live, which players each one receives for, // and what's queued for a player who wasn't reachable. .get('/internal/hub-state', async (c) => { - return c.json(await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).inspect()) + return c.json(await hubCall(c, (hub) => hub.inspect())) }) // Discard queued notifications, for `?playerId=` or — with the explicit `?all=true`, @@ -215,8 +265,7 @@ const app = new Hono() return c.json({ error: 'pass playerId, or all=true to clear every queue' }, 400) } - const result = - await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).clearPending(playerId) + const result = await hubCall(c, (hub) => hub.clearPending(playerId)) logger.info('cleared pending notifications', { playerId: playerId ?? null, ...result }) return c.json({ success: true, ...result }) }) diff --git a/apps/notify/src/test/integration/api.test.ts b/apps/notify/src/test/integration/api.test.ts index 7688db4..ddb63f7 100644 --- a/apps/notify/src/test/integration/api.test.ts +++ b/apps/notify/src/test/integration/api.test.ts @@ -1,10 +1,12 @@ -import { adminSecretsStore, env } from 'cloudflare:test' +import { adminSecretsStore, env, runInDurableObject } from 'cloudflare:test' import { exports } from 'cloudflare:workers' import { beforeAll, describe, expect, test } from 'vitest' import '../../notify.app' import type { Env } from '../../context' +import { MAX_PENDING_PER_PLAYER } from '../../notifications-hub' + import type { HubState } from '../../notifications-hub' declare module 'cloudflare:test' { @@ -653,6 +655,70 @@ describe('clearing pending notifications', () => { }) }) +// What a Durable Object RESET leaves behind, and how the hub recovers from it. A reset +// kills every socket without running webSocketClose, so the connection rows outlive the +// sockets they name — and a stale row makes deliverToPlayer report 0 delivered, which +// queues notifications for players who are online. +describe('recovering from lost sockets', () => { + const hubState = async (): Promise => + (await ( + await exports.default.fetch(`${ORIGIN}/internal/hub-state`, { + headers: await bearer('1', ['gameClient', 'moderator']), + }) + ).json()) as HubState + + test('forgets a connection whose socket is gone, instead of queueing to it forever', async () => { + const playerId = 9301 + const { ws } = await connect('conn-reset', { + headers: await bearer(String(playerId), ['gameClient']), + }) + // Kill the socket the way a reset does — no close frame, so the DO never runs + // webSocketClose and the row survives. + await ws.close() + + // First send after the socket died: nothing to deliver to, so it queues... + const first = await post('/internal/notify', { playerId, notificationType: 40, data: {} }) + expect(await first.json()).toMatchObject({ queued: true, delivered: 0 }) + + // ...and the dead row is dropped on the way, so the player no longer looks + // connected to anything. + const state = await hubState() + expect(state.connections.find((c) => c.connectionId === 'conn-reset')).toBeUndefined() + }) +}) + +describe('pending queue bound', () => { + test('keeps the newest notifications and drops the oldest past the cap', async () => { + // Unbounded, this is what a broken delivery path fills up — and what flushPending + // then reads into memory in one go. Seeded straight into the object: the point is + // the bound, not the 500 HTTP round trips it would take to reach it. + const playerId = 9302 + const stub = env.RECFLARE_NOTIFICATIONS_HUB.getByName('global') + await runInDurableObject(stub, (_instance, state) => { + for (let i = 0; i < MAX_PENDING_PER_PLAYER + 100; i++) { + state.storage.sql.exec( + 'INSERT INTO pending (playerId, payload) VALUES (?, ?)', + playerId, + JSON.stringify({ Id: '90', Msg: { i } }) + ) + } + }) + + // The next queued notification is what enforces the bound. + await post('/internal/notify', { playerId, notificationType: 90, data: { last: true } }) + + const state = (await ( + await exports.default.fetch(`${ORIGIN}/internal/hub-state`, { + headers: await bearer('1', ['gameClient', 'moderator']), + }) + ).json()) as HubState + const queued = state.pending.find((p) => p.playerId === playerId) + expect(queued?.count).toBe(MAX_PENDING_PER_PLAYER) + // Newest-wins: the one just sent survived, the oldest hundred did not. + expect(JSON.parse(queued!.latest)).toEqual({ Id: '90', Msg: { last: true } }) + }) +}) + // The website's admin controls (maintenance countdown, coach broadcast) are a browser // calling `/internal/*` directly rather than through a `www` proxy, so these need CORS. describe('CORS', () => {