[www] token drop performance improvement

apps/notify/src/notifications-hub.ts

@@ -99,6 +99,13 @@ export interface HubState {
 	pending: Array<{ playerId: number; count: number; latest: string }>
 }
 
+/** One entry of a {@link NotificationsHub.notifyPlayers} call: a `notifyPlayer`'s arguments. */
+export interface PlayerNotification {
+	playerId: number
+	notificationType: string | number
+	data?: Record<string, unknown>
+}
+
 /**
  * How many queued notifications one offline player may accumulate. Past this the OLDEST
  * are dropped, newest kept.
@@ -464,6 +471,48 @@ export class NotificationsHub extends DurableObject<Env> {
 		return { delivered, queued: false }
 	}
 
+	/**
+	 * {@link notifyPlayer} for MANY notifications in ONE RPC, each with its own player and
+	 * payload, delivered (or queued, when that player is offline) in the order given. The
+	 * server-wide token drop pays a few hundred players at once, and each needs two frames of
+	 * their own — a balance that differs per player, then a box whose id does — so sent
+	 * through `notifyPlayer` it was two round trips per player, in series, from a request
+	 * already short of its subrequest budget. One call carries the lot.
+	 *
+	 * `notifyPlayersEphemeral` is the other bulk send: one identical payload to many, never
+	 * queued. This one keeps `notifyPlayer`'s durability, player by player. The "nobody to
+	 * deliver to" warning is logged ONCE for the whole call, naming the players, rather than
+	 * once per frame: a drop reaches everyone with an unexpired presence row, and a good
+	 * share of them are between sessions at any moment.
+	 */
+	async notifyPlayers(
+		notifications: readonly PlayerNotification[]
+	): Promise<{ delivered: number; queued: number }> {
+		let delivered = 0
+		let queued = 0
+		const unreachable = new Set<number>()
+		for (const { playerId, notificationType, data } of notifications) {
+			const payload = this.buildNotificationPayload(notificationType, data)
+			const sent = this.deliverToPlayer(playerId, payload)
+			if (sent === 0) {
+				this.queuePending(playerId, payload)
+				queued++
+				unreachable.add(playerId)
+			} else {
+				delivered += sent
+			}
+		}
+		if (queued > 0) {
+			console.warn('hub: notifications queued, nobody to deliver to', {
+				playerIds: [...unreachable],
+				queued,
+				notifications: notifications.length,
+				liveSockets: this.ctx.getWebSockets().length,
+			})
+		}
+		return { delivered, queued }
+	}
+
 	/**
 	 * 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

apps/notify/src/test/integration/api.test.ts

@@ -311,6 +311,70 @@ describe('notification delivery', () => {
 		ws.close()
 	})
 
+	// The bulk send is notifyPlayer many times over in one RPC: each entry its own player and
+	// payload, delivered live or queued per player, in the order given. What the token drop
+	// sends — two frames a player, a few hundred players.
+	test('delivers a batch to each player, live or queued, in order', async () => {
+		const live = 9005
+		const offline = 9006
+		const { ws, waitFor, records } = await connect('conn-bulk')
+		send(ws, {
+			type: 1,
+			invocationId: 's',
+			target: 'SubscribeToPlayers',
+			arguments: [{ playerIds: [live] }],
+		})
+		await waitFor((r) => r.type === 3 && r.invocationId === 's')
+
+		const stub = env.RECFLARE_NOTIFICATIONS_HUB.getByName('global')
+		const result = await runInDurableObject(stub, (instance) =>
+			instance.notifyPlayers([
+				{ playerId: live, notificationType: 61, data: { Balance: 10 } },
+				{ playerId: offline, notificationType: 61, data: { Balance: 20 } },
+				{ playerId: live, notificationType: 31, data: { Id: 1 } },
+				{ playerId: offline, notificationType: 31, data: { Id: 2 } },
+			])
+		)
+		expect(result).toEqual({ delivered: 2, queued: 2 })
+
+		await waitFor(
+			(r) =>
+				r.type === 1 &&
+				r.target === 'Notification' &&
+				(JSON.parse((r.arguments as string[])[0]) as { Id: string }).Id === '31'
+		)
+		const notes = records
+			.filter((r) => r.type === 1 && r.target === 'Notification')
+			.map((r) => JSON.parse((r.arguments as string[])[0]) as unknown)
+		expect(notes).toEqual([
+			{ Id: '61', Msg: { Balance: 10 } },
+			{ Id: '31', Msg: { Id: 1 } },
+		])
+		ws.close()
+
+		// The offline player's two are queued, and flush in order when they subscribe.
+		const late = await connect('conn-bulk-late')
+		send(late.ws, {
+			type: 1,
+			target: 'SubscribeToPlayers',
+			arguments: [{ playerIds: [offline] }],
+		})
+		await late.waitFor(
+			(r) =>
+				r.type === 1 &&
+				r.target === 'Notification' &&
+				(JSON.parse((r.arguments as string[])[0]) as { Id: string }).Id === '31'
+		)
+		const flushed = late.records
+			.filter((r) => r.type === 1 && r.target === 'Notification')
+			.map((r) => JSON.parse((r.arguments as string[])[0]) as unknown)
+		expect(flushed).toEqual([
+			{ Id: '61', Msg: { Balance: 20 } },
+			{ Id: '31', Msg: { Id: 2 } },
+		])
+		late.ws.close()
+	})
+
 	test('broadcast reaches connected clients', async () => {
 		const { ws, waitFor } = await connect('conn-broadcast')
 		const res = await post('/internal/broadcast', {