[notify] fix intermittant drop in notifications

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<Env> {
 	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<Env> {
 				);
 				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<string>()
+		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<Response> {
 		if ((request.headers.get('Upgrade') ?? '').toLowerCase() !== 'websocket') {
@@ -198,14 +275,7 @@ export class NotificationsHub extends DurableObject<Env> {
 		if (state) {
 			// Mirrors OnDisconnected: drop this connection's subscriptions, which
 			// also removes it from every player's connection set.
-			this.ctx.storage.sql.exec(
+			this.forgetConnection(state.connectionId)
-				'DELETE FROM subscriptions WHERE connectionId = ?',
-				state.connectionId
-			)
-			this.ctx.storage.sql.exec(
-				'DELETE FROM connection_owner WHERE connectionId = ?',
-				state.connectionId
-			)
 		}
 		try {
 			ws.close()
@@ -384,16 +454,43 @@ export class NotificationsHub extends DurableObject<Env> {
 		}
 
 		if (delivered === 0) {
-			this.ctx.storage.sql.exec(
+			this.queuePending(playerId, payload)
-				'INSERT INTO pending (playerId, payload) VALUES (?, ?)',
-				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<Env> {
 
 		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++
 			}

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<T>(
+	c: Context<App>,
+	call: (hub: ReturnType<App['Bindings']['RECFLARE_NOTIFICATIONS_HUB']['getByName']>) => Promise<T>
+): Promise<T> {
+	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<App>()
 		// 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<App>()
 		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(
+		const { playerId, notificationType, data } = body
-			body.playerId,
+		const result = await hubCall(c, (hub) => hub.notifyPlayer(playerId, notificationType, data))
-			body.notificationType,
-			body.data
-		)
 		return c.json({ success: true, ...result })
 	})
 
@@ -159,10 +214,8 @@ const app = new Hono<App>()
 		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(
+		const { notificationType, data } = body
-			body.notificationType,
+		const result = await hubCall(c, (hub) => hub.broadcast(notificationType, data))
-			body.data
-		)
 		return c.json({ success: true, ...result })
 	})
 
@@ -171,8 +224,7 @@ const app = new Hono<App>()
 		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 =
+		const result = await hubCall(c, (hub) => hub.coachMessageAll(content))
-			await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).coachMessageAll(content)
 		return c.json({ success: true, ...result })
 	})
 
@@ -188,10 +240,8 @@ const app = new Hono<App>()
 			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(
+		const { playerId } = body
-			body.playerId,
+		const result = await hubCall(c, (hub) => hub.coachMessage(playerId, content))
-			content
-		)
 		return c.json({ success: true, ...result })
 	})
 
@@ -199,7 +249,7 @@ const app = new Hono<App>()
 	// 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<App>()
 			return c.json({ error: 'pass playerId, or all=true to clear every queue' }, 400)
 		}
 
-		const result =
+		const result = await hubCall(c, (hub) => hub.clearPending(playerId))
-			await c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE).clearPending(playerId)
 		logger.info('cleared pending notifications', { playerId: playerId ?? null, ...result })
 		return c.json({ success: true, ...result })
 	})

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<HubState> =>
+		(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', () => {