diff --git a/config.template.jsonc b/config.template.jsonc index cbc0f45f7..2b691b568 100644 --- a/config.template.jsonc +++ b/config.template.jsonc @@ -351,8 +351,7 @@ // the sweep. // "notificationRetentionDays": 14, // - // How long a live notification waits for the recipient's open tabs to - // acknowledge it, in ms, before its delivery is reported as unreceived. + // Ms a live notification waits for open tabs to acknowledge it. // "notificationAckTimeoutMs": 5000, // ── Alarms / alerting ─────────────────────────────────────────────── diff --git a/src/backend/clients/event/types.ts b/src/backend/clients/event/types.ts index 267ffb975..9a7963348 100644 --- a/src/backend/clients/event/types.ts +++ b/src/backend/clients/event/types.ts @@ -551,6 +551,7 @@ export type EventMap = { createdAt: number; /** Events handler runs behind whatever created it. */ handlerDepth?: number; + sendId?: string; }; /** * Every non-silent write or update, only on the node that wrote it. The @@ -566,6 +567,8 @@ export type EventMap = { value: Record; /** An earlier notification rewritten in place. */ updated?: true; + /** Names this one send; its `notif.delivery.settled` carries the same. */ + sendId: string; }; /** * One notification, addressed at whatever sockets each region holds for the @@ -577,17 +580,21 @@ export type EventMap = { userId: number; wire: 'notif.message' | 'notif.unreads' | 'notif.ack'; response: unknown; + sendId?: string; }; - /** - * How one live `notif.message` went, on the node that sent it. `received` - * once a socket acknowledged it within the window, `unreceived` when none - * did; `ackCapable` false means no socket there acknowledges at all. - */ + /** How one live `notif.message` went, on the node that sent it. */ 'notif.delivery.settled': { + sendId?: string; uid: string; userId: number; + type?: string; outcome: NotifDeliveryOutcome; + /** A socket in this region acknowledges `notif.message`. */ ackCapable: boolean; + /** Socket counts covered every node here, not just this one. */ + complete: boolean; + /** A peer region holds a socket for the user. */ + peerRegions: boolean; responders: NotifDeliveryResponder[]; }; @@ -1009,10 +1016,7 @@ export type MatchingEvents

= P extends `${infer Prefix}.*` ? Extract : P & EventKey; -/** - * Where a live `notif.message` ended up: acknowledged, sent but not - * acknowledged, no socket anywhere, or sockets only in a peer region. - */ +/** Acked, not acked, no socket anywhere, or sockets only in a peer region. */ export type NotifDeliveryOutcome = 'received' | 'unreceived' | 'no-socket' | 'remote-only'; diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index 97bb77f47..3d1855d8f 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -594,6 +594,7 @@ export interface NotifDispatchInput { createdAt: number; /** Handler runs behind whatever created it; see `Actor.handlerDepth`. */ handlerDepth?: number; + sendId?: string; } /** One committed KV mutation, as the bus reports it. */ @@ -3746,7 +3747,7 @@ export class EventsService extends PuterService { // The adapter is not a subscription: the desktop is delivered to // whether or not anything subscribed, which is what keeps the wire // contract identical with the fold-in on. - this.services.notification.deliverOverSocket(context); + this.services.notification.deliverOverSocket(context, input.sendId); if (!(await this.#userHasAny(userId))) return; diff --git a/src/backend/services/events/notifications.integration.test.ts b/src/backend/services/events/notifications.integration.test.ts index 8775737ce..d6a9c02f7 100644 --- a/src/backend/services/events/notifications.integration.test.ts +++ b/src/backend/services/events/notifications.integration.test.ts @@ -109,7 +109,13 @@ const withServer = (config: IConfig) => { ).mockImplementation(async (specifier, key, data) => { const spec = Array.isArray(specifier) ? specifier[0] : specifier; state.sent.push({ room: spec.room ?? spec.socket, key, data }); - return { responses: [], timedOut: false, sockets: 0, acking: 0 }; + return { + responses: [], + timedOut: false, + sockets: 0, + acking: 0, + complete: true, + }; }); state.server.services.events.onDelivered = (envelope) => { state.delivered.push(envelope); @@ -347,7 +353,13 @@ describe('notifications with the fold-in on', () => { const d = data as { uid: string; userId: number; outcome: string }; if (d.userId === user.id) settled.push(d); }; + const sendIds: string[] = []; + const onWritten = (_k: string, data: unknown) => { + const d = data as { userId: number; sendId: string }; + if (d.userId === user.id) sendIds.push(d.sendId); + }; state.server.clients.event.on('notif.delivery.settled', handler); + state.server.clients.event.on('notif.written', onWritten); try { const uid = await state.server.services.notification.notify( [user.id], @@ -355,8 +367,14 @@ describe('notifications with the fold-in on', () => { { type: 'share.received' }, ); await waitFor(() => settled.length > 0); + // The send id rides through dispatch to the socket adapter. + expect(sendIds).toEqual([expect.any(String)]); expect(settled).toEqual([ - expect.objectContaining({ uid, outcome: 'no-socket' }), + expect.objectContaining({ + uid, + outcome: 'no-socket', + sendId: sendIds[0], + }), ]); await waitFor(() => forwardedFor(user.id).length > 0); @@ -366,6 +384,7 @@ describe('notifications with the fold-in on', () => { expect(settled).toHaveLength(1); } finally { state.server.clients.event.off?.('notif.delivery.settled', handler); + state.server.clients.event.off?.('notif.written', onWritten); } }); diff --git a/src/backend/services/notification/NotificationService.test.ts b/src/backend/services/notification/NotificationService.test.ts index ab7065f2a..2dac482a1 100644 --- a/src/backend/services/notification/NotificationService.test.ts +++ b/src/backend/services/notification/NotificationService.test.ts @@ -365,6 +365,7 @@ describe("NotificationService — 'notif.written'", () => { audience: 'app-user', appUid, value: { title: 'ended', type: 'app.events.ended' }, + sendId: expect.any(String), })); expect(written.seen).toEqual(expect.arrayContaining(expected)); written.stop(); @@ -410,6 +411,7 @@ describe("NotificationService — 'notif.written'", () => { appUid: null, value: { title: 'two shares', type: 'share.received' }, updated: true, + sendId: expect.any(String), }); written.stop(); }); diff --git a/src/backend/services/notification/NotificationService.ts b/src/backend/services/notification/NotificationService.ts index 702590899..353b92bad 100644 --- a/src/backend/services/notification/NotificationService.ts +++ b/src/backend/services/notification/NotificationService.ts @@ -171,6 +171,7 @@ export class NotificationService extends PuterService { } if (!opts.silent) { + const sendId = uuidv4(); // Listeners get their own copy; the socket sends `payload`. this.clients.event.emit( 'notif.written', @@ -181,6 +182,7 @@ export class NotificationService extends PuterService { audience: registered.audience, appUid, value: structuredClone(payload), + sendId, }, {}, ); @@ -188,6 +190,7 @@ export class NotificationService extends PuterService { type: registered.type, audience: registered.audience, appUid, + sendId, }); } return uid; @@ -230,7 +233,8 @@ export class NotificationService extends PuterService { // the wire rather than through dispatch — the row was already // published, and republishing it under its own id is a duplicate // to anything deduplicating on `event.id`. - this.#socket().message(userId, uid, payload); + const sendId = uuidv4(); + this.#socket().message(userId, uid, payload, sendId); this.#markDelivered(uid, userId); const row = await this.stores.notification.getByUid(uid, { userId, @@ -248,6 +252,7 @@ export class NotificationService extends PuterService { (typeof app?.uid === 'string' ? app.uid : null), value: structuredClone(payload), updated: true, + sendId, }, {}, ); @@ -278,12 +283,13 @@ export class NotificationService extends PuterService { * layer decides who is owed a delivery; this turns that decision into the * message the GUI has always listened for. */ - deliverOverSocket(context: NotifEventContext): void { + deliverOverSocket(context: NotifEventContext, sendId?: string): void { if (!inAccountMailbox(context.type)) return; this.#socket().message( context.userId, context.uid, context.notification, + sendId, ); } @@ -347,12 +353,17 @@ export class NotificationService extends PuterService { userId: number, uid: string, payload: Record, - scope: { type: string; audience: string; appUid: string | null }, + scope: { + type: string; + audience: string; + appUid: string | null; + sendId: string; + }, ): Promise { const desktop = inAccountMailbox(scope.type); if (!notificationsFoldInEnabled(this.config)) { if (!desktop) return; - this.#socket().message(userId, uid, payload); + this.#socket().message(userId, uid, payload, scope.sendId); this.#markDelivered(uid, userId); return; } @@ -362,7 +373,7 @@ export class NotificationService extends PuterService { const user = await this.stores.user.getById(userId); if (!user?.uuid) { if (!desktop) return; - this.#socket().message(userId, uid, payload); + this.#socket().message(userId, uid, payload, scope.sendId); this.#markDelivered(uid, userId); return; } @@ -382,6 +393,7 @@ export class NotificationService extends PuterService { value: payload, createdAt: Date.now(), ...(handlerDepth > 0 ? { handlerDepth } : {}), + sendId: scope.sendId, }, {}, ); diff --git a/src/backend/services/notification/notificationDelivery.test.ts b/src/backend/services/notification/notificationDelivery.test.ts index 7ddaf0827..fa0948fda 100644 --- a/src/backend/services/notification/notificationDelivery.test.ts +++ b/src/backend/services/notification/notificationDelivery.test.ts @@ -22,6 +22,7 @@ import { afterAll, afterEach, beforeAll, + beforeEach, describe, expect, it, @@ -37,82 +38,143 @@ import { } from '../../testUtil.js'; type Settled = EventMap['notif.delivery.settled']; +type Written = EventMap['notif.written']; const ACK_TIMEOUT_MS = 300; -describe('live notification delivery receipts', () => { - let server: PuterServer; - let origin: string; - let user: TestUserCredentials; - let userId: number; - const openSockets: ClientSocket[] = []; - const settled: Settled[] = []; +interface ClusterAdapter { + serverCount: () => Promise; + fetchSockets: (opts: unknown) => Promise; + doPublish: (message: unknown) => Promise; +} + +/** A booted server, one user, and what the bus reported for them. */ +const withServer = (opts: { cluster: boolean }) => { + const state = { + server: null as unknown as PuterServer, + origin: '', + user: null as unknown as TestUserCredentials, + userId: 0, + adapter: null as unknown as ClusterAdapter, + openSockets: [] as ClientSocket[], + settled: [] as Settled[], + written: [] as Written[], + }; const onSettled = (_key: string, data: unknown) => { - settled.push(data as Settled); + state.settled.push(data as Settled); + }; + const onWritten = (_key: string, data: unknown) => { + state.written.push(data as Written); }; - - beforeAll(async () => { - const port = await allocateEphemeralPort(); - server = await setupTestServer( - { port, notificationAckTimeoutMs: ACK_TIMEOUT_MS } as never, - { listen: true }, - ); - origin = `http://puter.localhost:${port}`; - user = await createTestUser(server, { - username: 'receipt-user', - password: 'receipt-user-password', - }); - userId = (await server.stores.user.getByUsername(user.username))!.id; - server.clients.event.on('notif.delivery.settled', onSettled); - }, 60_000); - - afterEach(async () => { - vi.restoreAllMocks(); - for (const socket of openSockets.splice(0)) socket.disconnect(); - await vi.waitFor(() => - expect(server.services.socket.has({ room: userId })).toBe(false), - ); - settled.length = 0; - }); - - afterAll(async () => { - server?.clients.event.off('notif.delivery.settled', onSettled); - await server?.shutdown(); - }, 60_000); const connect = (auth: Record): Promise => new Promise((resolve, reject) => { - const socket = ioClient(origin, { - auth: { auth_token: user.token, ...auth }, + const socket = ioClient(state.origin, { + auth: { auth_token: state.user.token, ...auth }, transports: ['websocket'], reconnection: false, }); - openSockets.push(socket); + state.openSockets.push(socket); socket.on('connect', () => resolve(socket)); socket.on('connect_error', reject); }); + const emptyRoom = () => + vi.waitFor(() => + expect( + state.server.services.socket.has({ room: state.userId }), + ).toBe(false), + ); + + beforeAll(async () => { + const port = await allocateEphemeralPort(); + state.server = await setupTestServer( + { port, notificationAckTimeoutMs: ACK_TIMEOUT_MS } as never, + { listen: true }, + ); + state.origin = `http://puter.localhost:${port}`; + state.user = await createTestUser(state.server, { + username: `receipt-${opts.cluster ? 'c' : 'l'}`, + password: 'receipt-user-password', + }); + state.userId = (await state.server.stores.user.getByUsername( + state.user.username, + ))!.id; + const seen = new Promise((resolve) => { + state.server.clients.event.on( + 'web.socket.connected', + (_k, data) => { + resolve( + (data as { socket: { nsp: { adapter: never } } }).socket + .nsp.adapter, + ); + }, + ); + }); + const first = await connect({}); + state.adapter = await seen; + first.disconnect(); + state.openSockets.length = 0; + await emptyRoom(); + state.server.clients.event.on('notif.delivery.settled', onSettled); + state.server.clients.event.on('notif.written', onWritten); + }, 60_000); + + beforeEach(() => { + if (!opts.cluster) return; + // One node counted, nothing published: the mock is shared across servers. + vi.spyOn(state.adapter, 'serverCount').mockResolvedValue(1); + vi.spyOn(state.adapter, 'doPublish').mockResolvedValue(''); + }); + + afterEach(async () => { + vi.restoreAllMocks(); + for (const socket of state.openSockets.splice(0)) socket.disconnect(); + await emptyRoom(); + state.settled.length = 0; + state.written.length = 0; + }); + + afterAll(async () => { + state.server?.clients.event.off('notif.delivery.settled', onSettled); + state.server?.clients.event.off('notif.written', onWritten); + await state.server?.shutdown(); + }, 60_000); + const notify = async ( type: 'share.received' | 'app.message' = 'share.received', - ): Promise => { - const uid = await server.services.notification.notify( - [userId], + ): Promise => + (await state.server.services.notification.notify( + [state.userId], { title: 'hello' }, type === 'app.message' ? { type, appUid: 'app-receipts' } : { type }, - ); - return uid!; - }; + ))!; const settledFor = async (uid: string): Promise => { await vi.waitFor( - () => expect(settled.some((s) => s.uid === uid)).toBe(true), + () => expect(state.settled.some((s) => s.uid === uid)).toBe(true), { timeout: 3_000 }, ); - return settled.find((s) => s.uid === uid)!; + return state.settled.find((s) => s.uid === uid)!; }; + const peerRegionHolds = () => { + const eventForward = state.server.services.eventForward; + vi.spyOn(eventForward, 'active', 'get').mockReturnValue(true); + vi.spyOn(eventForward, 'regionsFor').mockResolvedValue(['peer-a']); + return eventForward; + }; + + return { state, connect, notify, settledFor, peerRegionHolds }; +}; + +describe('live notification delivery receipts', () => { + const { state, connect, notify, settledFor, peerRegionHolds } = withServer({ + cluster: true, + }); + it('settles as received once a tab acknowledges, with its visibility', async () => { const tab = await connect({ acks: ['notif.message'] }); const got: unknown[] = []; @@ -123,13 +185,19 @@ describe('live notification delivery receipts', () => { const uid = await notify(); + const [written] = state.written; expect(await settledFor(uid)).toEqual({ + sendId: written.sendId, uid, - userId, + userId: state.userId, + type: 'share.received', outcome: 'received', ackCapable: true, + complete: true, + peerRegions: false, responders: [{ visibilityState: 'hidden' }], }); + expect(written.sendId).toEqual(expect.any(String)); expect(got).toEqual([ { uid, @@ -154,6 +222,8 @@ describe('live notification delivery receipts', () => { expect(result).toMatchObject({ outcome: 'unreceived', ackCapable: true, + complete: true, + peerRegions: false, responders: [], }); await expect(got).resolves.toMatchObject({ uid }); @@ -177,30 +247,59 @@ describe('live notification delivery receipts', () => { it('settles as no-socket when the user has no connection anywhere', async () => { const uid = await notify(); - expect(await settledFor(uid)).toEqual({ - uid, - userId, + expect(await settledFor(uid)).toMatchObject({ outcome: 'no-socket', ackCapable: false, - responders: [], + complete: true, + peerRegions: false, }); }); it('settles as remote-only when only a peer region holds a socket', async () => { - const eventForward = server.services.eventForward; - vi.spyOn(eventForward, 'active', 'get').mockReturnValue(true); - vi.spyOn(eventForward, 'regionsFor').mockResolvedValue(['peer-a']); + const eventForward = peerRegionHolds(); const uid = await notify(); expect(await settledFor(uid)).toMatchObject({ outcome: 'remote-only', ackCapable: false, + peerRegions: true, }); - expect(eventForward.regionsFor).toHaveBeenCalledWith(userId, null); + expect(eventForward.regionsFor).toHaveBeenCalledWith( + state.userId, + null, + ); }); - it('settles an app message on the desktop wire the same way', async () => { + it('says a peer region holds the user beside an unacknowledged tab here', async () => { + await connect({ acks: ['notif.message'] }); + peerRegionHolds(); + + const uid = await notify(); + + expect(await settledFor(uid)).toMatchObject({ + outcome: 'unreceived', + ackCapable: true, + peerRegions: true, + }); + }); + + it('never reads an incomplete count as nobody being here', async () => { + vi.spyOn(state.adapter, 'fetchSockets').mockRejectedValue( + new Error('timeout reached'), + ); + vi.spyOn(console, 'warn').mockImplementation(() => {}); + + const uid = await notify(); + + expect(await settledFor(uid)).toMatchObject({ + outcome: 'unreceived', + ackCapable: false, + complete: false, + }); + }); + + it('settles an app message on the desktop wire with its type', async () => { const tab = await connect({ acks: ['notif.message'] }); tab.on('notif.message', (_m: unknown, ack: (r: unknown) => void) => ack({ visibilityState: 'visible' }), @@ -209,37 +308,70 @@ describe('live notification delivery receipts', () => { const uid = await notify('app.message'); expect(await settledFor(uid)).toMatchObject({ + type: 'app.message', outcome: 'received', responders: [{ visibilityState: 'visible' }], }); }); - it('settles a rewritten notification as it is delivered again', async () => { + it('settles a rewritten notification under a send id of its own', async () => { const tab = await connect({ acks: ['notif.message'] }); tab.on('notif.message', (_m: unknown, ack: (r: unknown) => void) => ack({ visibilityState: 'visible' }), ); const uid = await notify(); - await settledFor(uid); - settled.length = 0; + const first = await settledFor(uid); + state.settled.length = 0; - await server.services.notification.notifyUpdate( + await state.server.services.notification.notifyUpdate( uid, - userId, + state.userId, { title: 'hello again' }, { type: 'share.received' }, ); - expect(await settledFor(uid)).toMatchObject({ outcome: 'received' }); + const again = await settledFor(uid); + expect(again).toMatchObject({ outcome: 'received' }); + expect(again.sendId).toEqual(expect.any(String)); + expect(again.sendId).not.toBe(first.sendId); + await vi.waitFor(() => expect(state.written).toHaveLength(2)); + expect(state.written[1].sendId).toBe(again.sendId); }); it('sends nothing and settles nothing for a silent notification', async () => { - await server.services.notification.notify( - [userId], + await state.server.services.notification.notify( + [state.userId], { title: 'quiet' }, { type: 'share.received', silent: true }, ); await new Promise((resolve) => setTimeout(resolve, 100)); - expect(settled).toEqual([]); + expect(state.settled).toEqual([]); + }); +}); + +describe('live notification receipts where the adapter cannot count servers', () => { + const { connect, notify, settledFor } = withServer({ cluster: false }); + + it('still collects this node’s acknowledgements', async () => { + const tab = await connect({ acks: ['notif.message'] }); + tab.on('notif.message', (_m: unknown, ack: (r: unknown) => void) => + ack({ visibilityState: 'visible' }), + ); + + const uid = await notify(); + + expect(await settledFor(uid)).toMatchObject({ + outcome: 'received', + complete: false, + }); + }); + + it('reports no socket here as unreceived, not as nobody anywhere', async () => { + const uid = await notify(); + expect(await settledFor(uid)).toMatchObject({ + outcome: 'unreceived', + ackCapable: false, + complete: false, + }); }); }); diff --git a/src/backend/services/notification/notificationSocket.ts b/src/backend/services/notification/notificationSocket.ts index 70ce47cb4..a546068ed 100644 --- a/src/backend/services/notification/notificationSocket.ts +++ b/src/backend/services/notification/notificationSocket.ts @@ -45,17 +45,13 @@ import type { IConfig } from '../../types.js'; * advertised (`gui_params.eventsNotifications`) and falls back to this wire * when that lapses, so the adapter is what keeps older clients and the fallback * working. It goes when the flag is on everywhere. - * - * Either way, the emitting region sends `notif.message` itself, asking sockets - * that acknowledge it for a receipt, and reports the result as the node-local - * `notif.delivery.settled`. */ export type NotifWire = 'notif.message' | 'notif.unreads' | 'notif.ack'; export const NOTIF_DELIVERY_EVENT = 'outer.notif.delivery'; -/** How long a delivered `notif.message` waits for receipts by default. */ +/** Default receipt window for a delivered `notif.message`. */ export const DEFAULT_NOTIFICATION_ACK_TIMEOUT_MS = 5_000; interface MessageResponse { @@ -63,6 +59,11 @@ interface MessageResponse { notification: unknown; } +const typeOf = (notification: unknown): string | undefined => { + const type = (notification as { type?: unknown } | null)?.type; + return typeof type === 'string' ? type : undefined; +}; + /** One unread as the replay payload carries it. */ export interface UnreadNotification { uid: string; @@ -87,14 +88,14 @@ export const notificationAckTimeoutMs = (config: IConfig): number => { : DEFAULT_NOTIFICATION_ACK_TIMEOUT_MS; }; -const deliveryOutcome = async ( +/** An incomplete count never reads as nobody being here. */ +const deliveryOutcome = ( sent: AckedSend, - heldInPeerRegion: () => Promise, -): Promise => { - if (sent.sockets === 0) { - return (await heldInPeerRegion()) ? 'remote-only' : 'no-socket'; - } - return sent.responses.length > 0 ? 'received' : 'unreceived'; + peerRegions: boolean, +): NotifDeliveryOutcome => { + if (sent.responses.length > 0) return 'received'; + if (sent.sockets > 0 || !sent.complete) return 'unreceived'; + return peerRegions ? 'remote-only' : 'no-socket'; }; const responderOf = (response: unknown): NotifDeliveryResponder => { @@ -122,16 +123,16 @@ export class NotificationSocketAdapter { this.#deps = deps; } - /** - * Start delivering what the events layer dispatches. Both the region that - * emitted and the peers that received it land here exactly once; only the - * emitting region waits on receipts for a `notif.message`. - */ + /** Only the emitting region waits on receipts for a `notif.message`. */ attach(): void { this.#deps.event.on(NOTIF_DELIVERY_EVENT, (_key, data, meta) => { - const { userId, wire, response } = data; + const { userId, wire, response, sendId } = data; if (wire === 'notif.message' && !meta?.from_outside) { - this.#deliverMessage(userId, response as MessageResponse); + this.#deliverMessage( + userId, + response as MessageResponse, + sendId, + ); return; } void this.#deps.socket @@ -143,10 +144,17 @@ export class NotificationSocketAdapter { } /** One notification, in the shape `UIDesktop` has always received. */ - message(userId: number, uid: string, notification: unknown): void { + message( + userId: number, + uid: string, + notification: unknown, + sendId?: string, + ): void { const response = { uid, notification }; - if (this.#dispatched(userId, 'notif.message', response)) return; - this.#deliverMessage(userId, response); + if (this.#dispatched(userId, 'notif.message', response, sendId)) { + return; + } + this.#deliverMessage(userId, response, sendId); this.#deps.event.emit( 'outer.gui.notif.message', { user_id_list: [userId], response }, @@ -181,32 +189,44 @@ export class NotificationSocketAdapter { return this.#deps.socket.has({ room: userId }); } - #deliverMessage(userId: number, response: MessageResponse): void { - void this.#sendWithReceipt(userId, response).catch((err: unknown) => { - console.warn('[notification] socket send failed', err); - }); + #deliverMessage( + userId: number, + response: MessageResponse, + sendId?: string, + ): void { + void this.#sendWithReceipt(userId, response, sendId).catch( + (err: unknown) => { + console.warn('[notification] socket send failed', err); + }, + ); } async #sendWithReceipt( userId: number, response: MessageResponse, + sendId?: string, ): Promise { - const sent = await this.#deps.socket.sendWithAck( - { room: userId }, - 'notif.message', - response, - { timeoutMs: this.#deps.ackTimeoutMs() }, - ); - const outcome = await deliveryOutcome(sent, () => + const [sent, peerRegions] = await Promise.all([ + this.#deps.socket.sendWithAck( + { room: userId }, + 'notif.message', + response, + { timeoutMs: this.#deps.ackTimeoutMs() }, + ), this.#deps.heldInPeerRegion(userId), - ); + ]); + const type = typeOf(response.notification); this.#deps.event.emit( 'notif.delivery.settled', { + ...(sendId ? { sendId } : {}), uid: response.uid, userId, - outcome, + ...(type ? { type } : {}), + outcome: deliveryOutcome(sent, peerRegions), ackCapable: sent.acking > 0, + complete: sent.complete, + peerRegions, responders: sent.responses.map(responderOf), }, {}, @@ -214,11 +234,16 @@ export class NotificationSocketAdapter { } /** Take the events-layer route, or report that the legacy one applies. */ - #dispatched(userId: number, wire: NotifWire, response: unknown): boolean { + #dispatched( + userId: number, + wire: NotifWire, + response: unknown, + sendId?: string, + ): boolean { if (!this.#deps.foldIn()) return false; this.#deps.event.emit( NOTIF_DELIVERY_EVENT, - { userId, wire, response }, + { userId, wire, response, ...(sendId ? { sendId } : {}) }, {}, ); return true; diff --git a/src/backend/services/socket/SocketService.test.ts b/src/backend/services/socket/SocketService.test.ts index ae9cabc86..51a0c8991 100644 --- a/src/backend/services/socket/SocketService.test.ts +++ b/src/backend/services/socket/SocketService.test.ts @@ -521,7 +521,7 @@ describe('SocketService (live socket.io)', () => { tabB.disconnect(); }); - describe('sendWithAck', () => { + describe('sendWithAck where the adapter cannot count servers', () => { const userIdOf = async () => (await server.stores.user.getByUsername(user.username))!.id; @@ -558,6 +558,7 @@ describe('SocketService (live socket.io)', () => { timedOut: false, sockets: 2, acking: 1, + complete: false, }); // A socket that never declared it gets the event with no callback. expect(await plainGot).toEqual([{ n: 1 }]); @@ -595,6 +596,7 @@ describe('SocketService (live socket.io)', () => { timedOut: true, sockets: 2, acking: 2, + complete: false, }); expect(await silentGot).toEqual({ n: 2 }); await closeAll(userId, [answers, silent]); @@ -624,6 +626,7 @@ describe('SocketService (live socket.io)', () => { timedOut: false, sockets: 1, acking: 0, + complete: false, }); expect(await got).toEqual({ n: 3 }); await closeAll(userId, [other]); @@ -642,6 +645,7 @@ describe('SocketService (live socket.io)', () => { timedOut: false, sockets: 0, acking: 0, + complete: false, }); }); }); @@ -1316,3 +1320,199 @@ describe('SocketService — before attachHttpServer', () => { expect(await service.getLastChangeTimestamp(1)).toBe(0); }); }); + +describe('SocketService.sendWithAck through the cluster adapter', () => { + let server: PuterServer; + let socketService: SocketService; + let origin: string; + let user: TestUserCredentials; + let userId: number; + let adapter: { + serverCount: () => Promise; + fetchSockets: (opts: unknown) => Promise; + doPublish: (message: unknown) => Promise; + }; + const openSockets: ClientSocket[] = []; + + const connect = (acks?: string[]): Promise => + new Promise((resolve, reject) => { + const socket = ioClient(origin, { + auth: { auth_token: user.token, ...(acks ? { acks } : {}) }, + transports: ['websocket'], + reconnection: false, + }); + openSockets.push(socket); + socket.on('connect', () => resolve(socket)); + socket.on('connect_error', reject); + }); + + beforeAll(async () => { + const port = await allocateEphemeralPort(); + server = await setupTestServer({ port } as never, { listen: true }); + socketService = server.services.socket as unknown as SocketService; + origin = `http://puter.localhost:${port}`; + user = await createTestUser(server, { + username: 'sock-cluster', + password: 'sock-cluster-password', + }); + userId = (await server.stores.user.getByUsername(user.username))!.id; + + // A one-node count stands in for a real cluster's, which the mock lacks. + const seen = new Promise((resolve) => { + server.clients.event.on('web.socket.connected', (_k, data) => { + const socket = (data as { socket: { nsp: { adapter: never } } }) + .socket; + resolve(socket.nsp.adapter); + }); + }); + const first = await connect(); + adapter = await seen; + vi.spyOn(adapter, 'serverCount').mockResolvedValue(1); + // Other test servers share the mock's channels and would answer too. + vi.spyOn(adapter, 'doPublish').mockResolvedValue(''); + first.disconnect(); + await vi.waitFor(() => + expect(socketService.has({ room: userId })).toBe(false), + ); + }, 60_000); + + afterEach(async () => { + vi.mocked(adapter.fetchSockets).mockRestore?.(); + for (const socket of openSockets.splice(0)) socket.disconnect(); + await vi.waitFor(() => + expect(socketService.has({ room: userId })).toBe(false), + ); + }); + + afterAll(async () => { + await server?.shutdown(); + }, 60_000); + + it('counts every node and returns as soon as each acking socket answered', async () => { + const tab = await connect(['demo.ack']); + tab.on('demo.ack', (_d: unknown, ack: (r: unknown) => void) => + ack('seen'), + ); + + const started = Date.now(); + const sent = await socketService.sendWithAck( + { room: userId }, + 'demo.ack', + { n: 1 }, + { timeoutMs: 5_000 }, + ); + + expect(Date.now() - started).toBeLessThan(2_000); + expect(sent).toEqual({ + responses: ['seen'], + timedOut: false, + sockets: 1, + acking: 1, + complete: true, + }); + }); + + it('delivers before the socket listing finishes', async () => { + const tab = await connect(['demo.ack']); + let release: (sockets: unknown[]) => void = () => {}; + vi.spyOn(adapter, 'fetchSockets').mockImplementation( + () => + new Promise((resolve) => { + release = resolve; + }), + ); + const got = new Promise((resolve) => + tab.once('demo.ack', (d: unknown, ack: (r: unknown) => void) => { + ack('seen'); + resolve(d); + }), + ); + + const sent = socketService.sendWithAck( + { room: userId }, + 'demo.ack', + { n: 2 }, + { timeoutMs: 5_000 }, + ); + + await expect(got).resolves.toEqual({ n: 2 }); + release([]); + await expect(sent).resolves.toMatchObject({ responses: ['seen'] }); + }); + + it('says the count is this node only when the listing fails, once a minute', async () => { + const tab = await connect(['demo.ack']); + tab.on('demo.ack', (_d: unknown, ack: (r: unknown) => void) => + ack('seen'), + ); + vi.spyOn(adapter, 'fetchSockets').mockRejectedValue( + new Error('timeout reached'), + ); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + + for (let i = 0; i < 2; i++) { + expect( + await socketService.sendWithAck( + { room: userId }, + 'demo.ack', + {}, + { timeoutMs: 2_000 }, + ), + ).toEqual({ + responses: ['seen'], + timedOut: false, + sockets: 1, + acking: 1, + complete: false, + }); + } + expect( + warn.mock.calls.filter((c) => String(c[0]).includes('listing')), + ).toHaveLength(1); + warn.mockRestore(); + }); + + it('keeps the acks it got when a socket that never acks is in the room', async () => { + const tab = await connect(['demo.ack']); + tab.on('demo.ack', (_d: unknown, ack: (r: unknown) => void) => + ack('seen'), + ); + const old = await connect(); + const oldGot = new Promise((resolve) => + old.once('demo.ack', resolve), + ); + + const sent = await socketService.sendWithAck( + { room: userId }, + 'demo.ack', + { n: 3 }, + { timeoutMs: 300 }, + ); + + expect(sent).toEqual({ + responses: ['seen'], + timedOut: true, + sockets: 2, + acking: 1, + complete: true, + }); + await expect(oldGot).resolves.toEqual({ n: 3 }); + }); + + it('reports an empty room as complete', async () => { + expect( + await socketService.sendWithAck( + { room: 'nobody-here' }, + 'demo.ack', + {}, + { timeoutMs: 1_000 }, + ), + ).toEqual({ + responses: [], + timedOut: false, + sockets: 0, + acking: 0, + complete: true, + }); + }); +}); diff --git a/src/backend/services/socket/SocketService.ts b/src/backend/services/socket/SocketService.ts index 4cf398edb..88db49682 100644 --- a/src/backend/services/socket/SocketService.ts +++ b/src/backend/services/socket/SocketService.ts @@ -218,10 +218,18 @@ export interface AckedSend { sockets: number; /** Of those, the ones that acknowledge this event. */ acking: number; + /** The counts cover every node; false when only this node's were seen. */ + complete: boolean; } /** A listing that waits on another node longer than this counts this node only. */ const SOCKET_LIST_TIMEOUT_MS = 1_000; +const LIST_WARN_INTERVAL_MS = 60_000; + +interface ListedSocket { + id: string; + data: unknown; +} const MAX_HANDSHAKE_ACKS = 16; const MAX_ACK_EVENT_LENGTH = 64; @@ -243,6 +251,11 @@ const acksEvent = (data: unknown, key: string): boolean => { return Array.isArray(acks) && acks.includes(key); }; +const countAcking = (sockets: ListedSocket[], key: string) => ({ + sockets: sockets.length, + acking: sockets.filter((socket) => acksEvent(socket.data, key)).length, +}); + interface UploadProgressPayload { upload_tracker: UploadProgressTrackerLike; meta?: Record; @@ -285,6 +298,8 @@ export class SocketService extends PuterService { #reauthTimer: ReturnType | null = null; /** The sweep running now, so a slow one is joined instead of stacked. */ #reauthSweep: Promise | null = null; + #countsServers: Promise | null = null; + #lastListWarn = -Infinity; // -- Lifecycle --------------------------------------------------- @@ -377,12 +392,7 @@ export class SocketService extends PuterService { } } - /** - * Send an event and wait up to `timeoutMs` for the sockets that declared - * they acknowledge it (`handshake.auth.acks`); the rest get a plain emit. - * Covers every node in the cluster, or this node alone when the adapter - * can't list the others. - */ + /** Sends at once; acks are awaited from sockets listing it in `auth.acks`. */ async sendWithAck( specifiers: SocketSpecifier | SocketSpecifier[], key: string, @@ -394,6 +404,7 @@ export class SocketService extends PuterService { timedOut: false, sockets: 0, acking: 0, + complete: false, }; const io = this.#io; if (!io) return none; @@ -408,70 +419,109 @@ export class SocketService extends PuterService { ); if (targets.length === 0) return none; - const { sockets, clustered } = await this.#listSockets(targets); + if (!(await this.#adapterCountsServers())) { + return this.#sendWithAckHere(targets, key, data, timeoutMs); + } + const heard = new Promise>( + (resolve) => { + io.to(targets) + .timeout(timeoutMs) + .emit( + key, + data, + (err: Error | null, responses: unknown[]) => { + resolve({ + responses: responses ?? [], + timedOut: !!err, + }); + }, + ); + }, + ); + const [listed, acks] = await Promise.all([ + this.#listSockets(targets), + heard, + ]); + return { + ...acks, + ...countAcking(listed.sockets, key), + complete: listed.complete, + }; + } + + /** Whether the adapter can reach the cluster's other nodes for acks. */ + #adapterCountsServers(): Promise { + this.#countsServers ??= Promise.resolve() + .then(() => this.#io!.of('/').adapter.serverCount()) + .then( + () => true, + () => { + console.warn( + '[socket] adapter cannot count servers; acks cover this node only', + ); + return false; + }, + ); + return this.#countsServers; + } + + async #listSockets( + targets: string[], + ): Promise<{ sockets: ListedSocket[]; complete: boolean }> { + const io = this.#io!; + let timer: ReturnType | undefined; + try { + const sockets = await Promise.race([ + io.in(targets).timeout(SOCKET_LIST_TIMEOUT_MS).fetchSockets(), + new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new Error('socket listing timed out')), + SOCKET_LIST_TIMEOUT_MS, + ); + }), + ]); + return { sockets, complete: true }; + } catch (err) { + const now = Date.now(); + if (now - this.#lastListWarn >= LIST_WARN_INTERVAL_MS) { + this.#lastListWarn = now; + console.warn('[socket] cluster socket listing failed', err); + } + return { sockets: this.#localSockets(targets), complete: false }; + } finally { + clearTimeout(timer); + } + } + + #localSockets(targets: string[]): ListedSocket[] { + const io = this.#io!; + const ids = new Set(); + for (const target of targets) { + io.sockets.adapter.rooms.get(target)?.forEach((id) => { + ids.add(id); + }); + } + return [...ids].flatMap((id) => { + const socket = io.sockets.sockets.get(id); + return socket ? [{ id, data: socket.data as unknown }] : []; + }); + } + + async #sendWithAckHere( + targets: string[], + key: string, + data: unknown, + timeoutMs: number, + ): Promise { + const io = this.#io!; + const sockets = this.#localSockets(targets); const acking = sockets .filter((socket) => acksEvent(socket.data, key)) .map((socket) => socket.id); io.to(targets).except(acking).emit(key, data); - if (acking.length === 0) return { ...none, sockets: sockets.length }; - - const heard = clustered - ? await this.#broadcastWithAck(acking, key, data, timeoutMs) - : await this.#emitWithAckHere(acking, key, data, timeoutMs); - return { ...heard, sockets: sockets.length, acking: acking.length }; - } - - /** `fetchSockets` needs the adapter's server count, which a mock lacks. */ - async #listSockets(targets: string[]): Promise<{ - sockets: { id: string; data: unknown }[]; - clustered: boolean; - }> { - const io = this.#io!; - try { - const sockets = await io - .in(targets) - .timeout(SOCKET_LIST_TIMEOUT_MS) - .fetchSockets(); - return { sockets, clustered: true }; - } catch { - const ids = new Set(); - for (const target of targets) { - io.sockets.adapter.rooms.get(target)?.forEach((id) => { - ids.add(id); - }); - } - const sockets = [...ids].flatMap((id) => { - const socket = io.sockets.sockets.get(id); - return socket ? [{ id, data: socket.data as unknown }] : []; - }); - return { sockets, clustered: false }; - } - } - - #broadcastWithAck( - ids: string[], - key: string, - data: unknown, - timeoutMs: number, - ): Promise> { - return new Promise((resolve) => { - this.#io!.to(ids) - .timeout(timeoutMs) - .emit(key, data, (err: Error | null, responses: unknown[]) => { - resolve({ responses: responses ?? [], timedOut: !!err }); - }); - }); - } - - async #emitWithAckHere( - ids: string[], - key: string, - data: unknown, - timeoutMs: number, - ): Promise> { const heard = await Promise.all( - ids.map(async (id) => { - const socket = this.#io!.sockets.sockets.get(id); + acking.map(async (id) => { + const socket = io.sockets.sockets.get(id); if (!socket) return { ok: false as const }; try { const response: unknown = await socket @@ -486,6 +536,9 @@ export class SocketService extends PuterService { return { responses: heard.flatMap((h) => (h.ok ? [h.response] : [])), timedOut: heard.some((h) => !h.ok), + sockets: sockets.length, + acking: acking.length, + complete: false, }; } @@ -890,13 +943,12 @@ export class SocketService extends PuterService { * * Every connection for the account goes, not just the revoked session's — * which is what the account room is for, since an app socket is not in the - * user room. Narrowing would mean matching each socket to its session via - * `fetchSockets`, which the adapter implements on top of `serverCount()` — - * and that path is unavailable with our Redis client. Dropping the room is - * the safe direction — a client does not reconnect by itself after a - * server-side disconnect, so one that wants to stay connected opens a new - * connection, and that handshake re-authenticates: only the revoked session - * is refused. + * user room. Narrowing would mean matching each socket to its session via a + * cluster-wide `fetchSockets` round trip. Dropping the room is the safe + * direction — a client does not reconnect by itself after a server-side + * disconnect, so one that wants to stay connected opens a new connection, + * and that handshake re-authenticates: only the revoked session is + * refused. */ async #evictUserSockets(userId: number): Promise { const io = this.#io; diff --git a/src/backend/types.ts b/src/backend/types.ts index 645ab1dee..4a74de7d3 100644 --- a/src/backend/types.ts +++ b/src/backend/types.ts @@ -971,10 +971,7 @@ interface IConfigOptional { */ notificationRetentionDays?: number; - /** - * How long a live notification waits for the recipient's sockets to - * acknowledge it before its delivery settles as unreceived. Default 5000. - */ + /** Ms a live notification waits for sockets to acknowledge it; 5000. */ notificationAckTimeoutMs?: number; /** diff --git a/src/gui/src/helpers/notificationFeed.js b/src/gui/src/helpers/notificationFeed.js index aaaa986f2..0709887e2 100644 --- a/src/gui/src/helpers/notificationFeed.js +++ b/src/gui/src/helpers/notificationFeed.js @@ -40,13 +40,7 @@ export const NOTIFICATION_SOCKET_ACKS = Object.freeze(['notif.message']); const acknowledged = new WeakSet(); -/** - * Answer a live `notif.message`'s acknowledgement callback with whether the - * page is in view. Every listener gets the same callback, so this answers once. - * - * @param {unknown} ack - The callback socket.io passes, when the server asked for one - * @param {{ visibilityState?: string }} [doc] - */ +/** Acks a live `notif.message` once, saying whether the page is in view. */ export const acknowledgeDelivery = (ack, doc = globalThis.document) => { if ( typeof ack !== 'function' || acknowledged.has(ack) ) return; acknowledged.add(ack);