fix(notifications): send live notif.message at once and report what the count covers

sendWithAck now emits to the whole room straight away as a broadcast with
ack, and lists the room's sockets alongside it only to fill in the
counts, so nobody waits on the listing. Whether the adapter can count
the cluster's servers is checked once; where it cannot, sends go out
per socket on this node with a one-time warning. The result carries
`complete`: false when only this node's sockets were counted, either
for that reason or because the listing failed or took over 1 s, which
is warned about at most once a minute.

notif.delivery.settled gains `sendId`, `type`, `complete` and
`peerRegions`. An incomplete count with no socket found settles as
unreceived rather than no-socket, and the peer-region read runs
alongside the send instead of after it. notif.written carries the same
`sendId`, minted per send (an update gets a new one) and passed through
notif.created to the socket adapter on the fold-in path.

The comment on revoking sockets no longer says fetchSockets is
unavailable with our Redis client; it works with ioredis cluster.

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 ───────────────────────────────────────────────

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<string, unknown>;
         /** 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 ListenKey> = P extends `${infer Prefix}.*`
     ? Extract<EventKey, `${Prefix}.${string}`>
     : 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';
 

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;
 

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);
         }
     });
 

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();
     });

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<string, unknown>,
-        scope: { type: string; audience: string; appUid: string | null },
+        scope: {
+            type: string;
+            audience: string;
+            appUid: string | null;
+            sendId: string;
+        },
     ): Promise<void> {
         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,
             },
             {},
         );

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<number>;
+    fetchSockets: (opts: unknown) => Promise<unknown[]>;
+    doPublish: (message: unknown) => Promise<string>;
+}
+
+/** 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<string, unknown>): Promise<ClientSocket> =>
         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<ClusterAdapter>((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<string> => {
-        const uid = await server.services.notification.notify(
-            [userId],
+    ): Promise<string> =>
+        (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<Settled> => {
         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,
+        });
     });
 });

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<boolean>,
-): Promise<NotifDeliveryOutcome> => {
-    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<void> {
-        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;

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<number>;
+        fetchSockets: (opts: unknown) => Promise<unknown[]>;
+        doPublish: (message: unknown) => Promise<string>;
+    };
+    const openSockets: ClientSocket[] = [];
+
+    const connect = (acks?: string[]): Promise<ClientSocket> =>
+        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<typeof adapter>((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<unknown>((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<unknown>((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,
+        });
+    });
+});

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<string, unknown>;
@@ -285,6 +298,8 @@ export class SocketService extends PuterService {
     #reauthTimer: ReturnType<typeof setInterval> | null = null;
     /** The sweep running now, so a slow one is joined instead of stacked. */
     #reauthSweep: Promise<void> | null = null;
+    #countsServers: Promise<boolean> | 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<Pick<AckedSend, 'responses' | 'timedOut'>>(
+            (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<boolean> {
+        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<typeof setTimeout> | undefined;
+        try {
+            const sockets = await Promise.race([
+                io.in(targets).timeout(SOCKET_LIST_TIMEOUT_MS).fetchSockets(),
+                new Promise<never>((_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<string>();
+        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<AckedSend> {
+        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<string>();
-            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<Pick<AckedSend, 'responses' | 'timedOut'>> {
-        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<Pick<AckedSend, 'responses' | 'timedOut'>> {
         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<void> {
         const io = this.#io;

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;
 
     /**

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);