From b526c08ffe7f8ed47106bc33902523524ee5ec3d Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Thu, 19 Mar 2026 20:56:04 -0700 Subject: [PATCH] fix: ws double messages (#2700) --- .../src/modules/broadcast/BroadcastService.js | 19 ++++-- .../BroadcastService.redisPubSub.test.js | 65 +++++++++++++++---- src/backend/src/services/WSPushService.js | 62 ++++++++---------- 3 files changed, 93 insertions(+), 53 deletions(-) diff --git a/src/backend/src/modules/broadcast/BroadcastService.js b/src/backend/src/modules/broadcast/BroadcastService.js index 496ad6c82..11f03f3e0 100644 --- a/src/backend/src/modules/broadcast/BroadcastService.js +++ b/src/backend/src/modules/broadcast/BroadcastService.js @@ -103,7 +103,7 @@ export class BroadcastService extends BaseService { const safeMeta = this.#normalizeMeta(meta); const outboundEvent = { key, data, meta: safeMeta }; - // Mirror local outer.gui/pub events to Redis so same-cluster replicas + // Mirror local outer.pub events to Redis so same-cluster replicas // receive them even when this instance is the originator. this.#publishWebhookEventsToRedis([outboundEvent]).catch(error => { console.warn('local redis pubsub publish failed', { error, key }); @@ -171,6 +171,12 @@ export class BroadcastService extends BaseService { return meta; } + #resolveLocalPeerId () { + const localPeerId = this.config?.webhook?.peerId ?? this.config?.webhook?.key; + if ( typeof localPeerId !== 'string' || localPeerId.trim() === '' ) return null; + return localPeerId.trim(); + } + #resolvePeerId (peerConfig) { if ( !peerConfig || typeof peerConfig !== 'object' ) return null; const peerId = peerConfig.peerId ?? peerConfig.key; @@ -213,9 +219,7 @@ export class BroadcastService extends BaseService { #isRedisWebhookEventKey (key) { if ( typeof key !== 'string' ) return false; - return key === 'outer.gui' || - key.startsWith('outer.gui.') || - key === 'outer.pub' || + return key === 'outer.pub' || key.startsWith('outer.pub.'); } @@ -377,6 +381,11 @@ export class BroadcastService extends BaseService { res.status(403).send({ error: { message: 'Missing X-Broadcast-Peer-Id' } }); return; } + const localPeerId = this.#resolveLocalPeerId(); + if ( localPeerId && peerId === localPeerId ) { + res.status(200).send({ ok: true, ignored: 'self-peer' }); + return; + } const peer = this.#peersByKey[peerId]; if ( !peer || !peer.webhook_secret ) { @@ -437,7 +446,7 @@ export class BroadcastService extends BaseService { this.#incomingLastNonceByPeer.set(peerId, { timestamp, nonce }); await this.#publishWebhookEventsToRedis(incomingEvents); - this.#emitIncomingEventsSequentially(incomingEvents); + await this.#emitIncomingEventsSequentially(incomingEvents); res.status(200).send({ ok: true }); } diff --git a/src/backend/src/modules/broadcast/BroadcastService.redisPubSub.test.js b/src/backend/src/modules/broadcast/BroadcastService.redisPubSub.test.js index 47655ba49..5fb11d33b 100644 --- a/src/backend/src/modules/broadcast/BroadcastService.redisPubSub.test.js +++ b/src/backend/src/modules/broadcast/BroadcastService.redisPubSub.test.js @@ -49,7 +49,7 @@ describe('BroadcastService redis pubsub', () => { eventService.emit.mockClear(); }); - it('re-emits only outer.gui/pub events from redis pubsub payloads', async () => { + it('re-emits only outer.pub events from redis pubsub payloads', async () => { await redisClient.publish('broadcast.webhook.events', JSON.stringify({ sourceId: 'other-instance', events: [ @@ -61,15 +61,9 @@ describe('BroadcastService redis pubsub', () => { await wait(); - expect(eventService.emit).toHaveBeenCalledTimes(2); + expect(eventService.emit).toHaveBeenCalledTimes(1); expect(eventService.emit).toHaveBeenNthCalledWith( 1, - 'outer.gui.notif.message', - { id: 'gui-1' }, - expect.objectContaining({ from_outside: true }), - ); - expect(eventService.emit).toHaveBeenNthCalledWith( - 2, 'outer.pub.notice', { id: 'pub-1' }, expect.objectContaining({ from_outside: true }), @@ -89,10 +83,10 @@ describe('BroadcastService redis pubsub', () => { expect(eventService.emit).not.toHaveBeenCalled(); }); - it('publishes local outer.gui/pub events to redis pubsub for replicas', async () => { + it('publishes local outer.pub events to redis pubsub for replicas', async () => { const publishSpy = vi.spyOn(redisClient, 'publish'); try { - await service.outBroadcastEventHandler('outer.gui.notif.message', { id: 'gui-local' }, {}); + await service.outBroadcastEventHandler('outer.pub.notice', { id: 'pub-local' }, {}); await wait(); const publishCall = publishSpy.mock.calls.find(([channel]) => channel === 'broadcast.webhook.events'); @@ -104,8 +98,8 @@ describe('BroadcastService redis pubsub', () => { expect(parsedPayload.sourceId).toBeDefined(); expect(parsedPayload.events).toEqual([ { - key: 'outer.gui.notif.message', - data: { id: 'gui-local' }, + key: 'outer.pub.notice', + data: { id: 'pub-local' }, meta: {}, }, ]); @@ -113,4 +107,51 @@ describe('BroadcastService redis pubsub', () => { publishSpy.mockRestore(); } }); + + it('does not publish local outer.gui events to redis pubsub', async () => { + const publishSpy = vi.spyOn(redisClient, 'publish'); + try { + await service.outBroadcastEventHandler('outer.gui.notif.message', { id: 'gui-local' }, {}); + await wait(); + + const publishCall = publishSpy.mock.calls.find(([channel]) => channel === 'broadcast.webhook.events'); + expect(publishCall).toBeUndefined(); + } finally { + publishSpy.mockRestore(); + } + }); + + it('does not rebroadcast events marked from_outside', async () => { + const publishSpy = vi.spyOn(redisClient, 'publish'); + try { + await service.outBroadcastEventHandler('outer.gui.notif.message', { id: 'outside' }, { + from_outside: true, + }); + await wait(); + + expect(publishSpy).not.toHaveBeenCalled(); + } finally { + publishSpy.mockRestore(); + } + }); + + it('ignores redis pubsub payloads with this instance sourceId', async () => { + const publishSpy = vi.spyOn(redisClient, 'publish'); + try { + await service.outBroadcastEventHandler('outer.pub.notice', { id: 'self-source' }, {}); + await wait(); + + const publishCall = publishSpy.mock.calls.find(([channel]) => channel === 'broadcast.webhook.events'); + expect(publishCall).toBeDefined(); + const [_channel, payload] = publishCall; + + eventService.emit.mockClear(); + await redisClient.publish('broadcast.webhook.events', payload); + await wait(); + + expect(eventService.emit).not.toHaveBeenCalled(); + } finally { + publishSpy.mockRestore(); + } + }); }); diff --git a/src/backend/src/services/WSPushService.js b/src/backend/src/services/WSPushService.js index 86f4c2703..e00f9c5ce 100644 --- a/src/backend/src/services/WSPushService.js +++ b/src/backend/src/services/WSPushService.js @@ -59,12 +59,10 @@ class WSPushService extends BaseService { from_new_service: true, }; - { - const svc_operationTrace = context.get('services').get('operationTrace'); - const frame = context.get(svc_operationTrace.ckey('frame')); - const gui_metadata = frame.get_attr('gui_metadata') || {}; - Object.assign(metadata, gui_metadata); - } + const svc_operationTrace = context.get('services').get('operationTrace'); + const frame = context.get(svc_operationTrace.ckey('frame')); + const gui_metadata = frame.get_attr('gui_metadata') || {}; + Object.assign(metadata, gui_metadata); const response = await node.getSafeEntry({ thumbnail: true }); @@ -109,12 +107,10 @@ class WSPushService extends BaseService { from_new_service: true, }; - { - const svc_operationTrace = context.get('services').get('operationTrace'); - const frame = context.get(svc_operationTrace.ckey('frame')); - const gui_metadata = frame?.get_attr?.('gui_metadata') || {}; - Object.assign(metadata, gui_metadata); - } + const svc_operationTrace = context.get('services').get('operationTrace'); + const frame = context.get(svc_operationTrace.ckey('frame')); + const gui_metadata = frame?.get_attr?.('gui_metadata') || {}; + Object.assign(metadata, gui_metadata); const response = await node.getSafeEntry({ debug: 'hi', thumbnail: true }); @@ -157,12 +153,10 @@ class WSPushService extends BaseService { from_new_service: true, }; - { - const svc_operationTrace = context.get('services').get('operationTrace'); - const frame = context.get(svc_operationTrace.ckey('frame')); - const gui_metadata = frame.get_attr('gui_metadata') || {}; - Object.assign(metadata, gui_metadata); - } + const svc_operationTrace = context.get('services').get('operationTrace'); + const frame = context.get(svc_operationTrace.ckey('frame')); + const gui_metadata = frame.get_attr('gui_metadata') || {}; + Object.assign(metadata, gui_metadata); const response = await moved.getSafeEntry(); @@ -205,12 +199,10 @@ class WSPushService extends BaseService { const response = { ...fsentry }; - { - const svc_operationTrace = context.get('services').get('operationTrace'); - const frame = context.get(svc_operationTrace.ckey('frame')); - const gui_metadata = frame.get_attr('gui_metadata') || {}; - Object.assign(metadata, gui_metadata); - } + const svc_operationTrace = context.get('services').get('operationTrace'); + const frame = context.get(svc_operationTrace.ckey('frame')); + const gui_metadata = frame.get_attr('gui_metadata') || {}; + Object.assign(metadata, gui_metadata); const user_id_list = await (async () => { const user_id_set = new Set(); @@ -230,7 +222,7 @@ class WSPushService extends BaseService { } /** - * Emits an upload or download progress event to the relevant socket. + * Emits an upload or download progress event to the relevant user room. * * @param {string} key - The event key that triggered this method. * @param {Object} data - Contains upload_tracker, context, and meta information. @@ -238,7 +230,7 @@ class WSPushService extends BaseService { * @param {Object} data.context - Context of the operation. * @param {Object} data.meta - Additional metadata for the event. * - * It emits a progress event to the socket if it exists, otherwise, it does nothing. + * It emits a progress event to the room if it exists, otherwise, it does nothing. */ async _on_upload_progress (key, data) { this.log.info('got upload progress event'); @@ -249,17 +241,15 @@ class WSPushService extends BaseService { from_new_service: true, }; - { - const svc_operationTrace = context.get('services').get('operationTrace'); - const frame = context.get(svc_operationTrace.ckey('frame')); - const gui_metadata = frame.get_attr('gui_metadata') || {}; - Object.assign(metadata, gui_metadata); - } + const svc_operationTrace = context.get('services').get('operationTrace'); + const frame = context.get(svc_operationTrace.ckey('frame')); + const gui_metadata = frame.get_attr('gui_metadata') || {}; + Object.assign(metadata, gui_metadata); - const { socket_id } = metadata; + const roomId = metadata.user_id ?? metadata.userId; - if ( ! socket_id ) { - console.warn('missing socket id', { metadata }); + if ( ! roomId ) { + console.warn('missing room id for upload progress', { metadata }); return; } @@ -270,7 +260,7 @@ class WSPushService extends BaseService { upload_tracker.sub(delta => { this.log.info('emitting progress event'); - svc_socketio.send({ socket: socket_id }, ws_event_name, { + svc_socketio.send({ room: roomId }, ws_event_name, { ...metadata, total: upload_tracker.total_, loaded: upload_tracker.progress_,