From 72346053ff8bcea87b6aff9b091cbe02832b873f Mon Sep 17 00:00:00 2001 From: ProgrammerIn-wonderland <3838shah@gmail.com> Date: Fri, 9 Oct 2026 17:49:37 -0400 Subject: [PATCH] PUT-2012 fix broadcast service drops --- .../broadcast/BroadcastService.test.ts | 158 ++++++++++++++++++ .../services/broadcast/BroadcastService.ts | 137 +++++++++------ 2 files changed, 245 insertions(+), 50 deletions(-) diff --git a/src/backend/services/broadcast/BroadcastService.test.ts b/src/backend/services/broadcast/BroadcastService.test.ts index 2af9f2acf..d11395ffc 100644 --- a/src/backend/services/broadcast/BroadcastService.test.ts +++ b/src/backend/services/broadcast/BroadcastService.test.ts @@ -632,6 +632,164 @@ describe('BroadcastService outbound flush', () => { }); }); +// -- Outbound flush, more than one peer ------------------------------- + +describe('BroadcastService outbound flush with a failing peer', () => { + const HEALTHY_HOST = 'broadcast-healthy.invalid'; + const FAILING_HOST = 'broadcast-failing.invalid'; + const MAX_QUEUED = 10_000; + + let server: PuterServer; + let failingPeerDown = false; + /** Event-key lists of every request each peer accepted, in order. */ + let delivered: Record = {}; + let failedAttempts = 0; + /** While set, the failing peer's request hangs until it resolves. */ + let failingPeerGate: Promise | null = null; + let failingPeerStalled = false; + + const eventIds = (rawBody: string): string[] => + ( + JSON.parse(rawBody) as { + events: { data: { cacheKey: string[] } }[]; + } + ).events.map((e) => e.data.cacheKey[0]); + + beforeAll(async () => { + server = await setupTestServer({ + broadcast: { + webhook: { peerId: SELF_PEER_ID, secret: SELF_SECRET }, + peers: [ + { + peerId: 'peer-healthy', + webhook: true, + webhook_url: `http://${HEALTHY_HOST}/broadcast/webhook`, + webhook_secret: 'healthy-secret', + }, + { + peerId: 'peer-failing', + webhook: true, + webhook_url: `http://${FAILING_HOST}/broadcast/webhook`, + webhook_secret: 'failing-secret', + }, + ], + outbound_flush_ms: 25, + }, + } as never); + }); + + afterAll(async () => { + await server?.shutdown(); + }); + + beforeEach(() => { + failingPeerDown = false; + delivered = { [HEALTHY_HOST]: [], [FAILING_HOST]: [] }; + failedAttempts = 0; + failingPeerGate = null; + failingPeerStalled = false; + axiosRequestMock.mockReset(); + axiosRequestMock.mockImplementation( + async (request: { url: string; data: string }) => { + const host = new URL(request.url).hostname; + if (host === FAILING_HOST && failingPeerDown) { + if (failingPeerGate) { + failingPeerStalled = true; + await failingPeerGate; + } + failedAttempts += 1; + throw new Error('peer unreachable'); + } + delivered[host]?.push(eventIds(request.data)); + return { status: 200, statusText: 'OK', data: 'ok' }; + }, + ); + }); + + afterEach(() => { + axiosRequestMock.mockReset(); + }); + + const waitFor = async (predicate: () => boolean, timeoutMs = 5000) => { + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + if (predicate()) return; + await new Promise((r) => setTimeout(r, 10)); + } + throw new Error('Timed out waiting for outbound flush'); + }; + + const emit = (id: string) => + server.clients.event.emit( + 'outer.cacheUpdate' as never, + { cacheKey: [id] } as never, + {}, + ); + + const deliveriesOf = (host: string, id: string) => + delivered[host]!.filter((ids) => ids.includes(id)).length; + + it('does not resend to a healthy peer what only a failing peer missed', async () => { + failingPeerDown = true; + emit('isolation-x'); + + await waitFor( + () => + deliveriesOf(HEALTHY_HOST, 'isolation-x') === 1 && + failedAttempts >= 3, + ); + expect(deliveriesOf(HEALTHY_HOST, 'isolation-x')).toBe(1); + + failingPeerDown = false; + await waitFor(() => deliveriesOf(FAILING_HOST, 'isolation-x') > 0); + expect(deliveriesOf(HEALTHY_HOST, 'isolation-x')).toBe(1); + }); + + it('drops the oldest events on overflow, not the newest', async () => { + failingPeerDown = true; + let release!: () => void; + failingPeerGate = new Promise((r) => (release = r)); + for (let i = 0; i < 3; i++) emit(`overflow-old-${i}`); + await waitFor(() => failingPeerStalled); + + // Arrives while the old events' failing send is still in flight. + for (let i = 0; i < MAX_QUEUED; i++) emit(`overflow-new-${i}`); + failingPeerGate = null; + release(); + await waitFor(() => failedAttempts >= 3); + + failingPeerDown = false; + await waitFor( + () => + deliveriesOf(FAILING_HOST, `overflow-new-${MAX_QUEUED - 1}`) > + 0, + ); + const received = delivered[FAILING_HOST]!.flat(); + for (let i = 0; i < 3; i++) { + expect(received).not.toContain(`overflow-old-${i}`); + } + const fresh = received.filter((id) => id.startsWith('overflow-new-')); + expect(fresh).toEqual( + Array.from({ length: MAX_QUEUED }, (_, i) => `overflow-new-${i}`), + ); + }); + + it('delivers a backlog in arrival order, a repeat after what came between', async () => { + failingPeerDown = true; + emit('order-x'); + emit('order-y'); + await waitFor(() => failedAttempts >= 1); + emit('order-x'); + + failingPeerDown = false; + await waitFor(() => deliveriesOf(FAILING_HOST, 'order-x') > 0); + const received = delivered[FAILING_HOST]!.flat().filter((id) => + id.startsWith('order-'), + ); + expect(received).toEqual(['order-y', 'order-x']); + }); +}); + // -- Header validation ------------------------------------------------ describe('BroadcastService.verifyAndEmit — header validation', () => { diff --git a/src/backend/services/broadcast/BroadcastService.ts b/src/backend/services/broadcast/BroadcastService.ts index 3d12b6385..f53595f2d 100644 --- a/src/backend/services/broadcast/BroadcastService.ts +++ b/src/backend/services/broadcast/BroadcastService.ts @@ -31,6 +31,11 @@ interface BroadcastEvent { meta: Record; } +interface QueuedEvent { + event: BroadcastEvent; + seq: number; +} + interface IncomingPayload { events?: unknown; key?: string; @@ -77,9 +82,10 @@ const DEFAULT_REGION_ID = 'local'; * Cross-node event replication via signed HTTP webhooks. * * **Outbound** — subscribes to local `outer.*` events on the event bus. Each - * event is added to a small in-memory map (deduped by serialized shape), then - * flushed every `outbound_flush_ms` as a single POST per configured peer. Each - * POST carries: + * event is added to every peer's in-memory queue (deduped by serialized shape), + * then flushed every `outbound_flush_ms` as a single POST per configured peer. + * A peer's events leave its queue only once that peer accepts them. Each POST + * carries: * * - `X-Broadcast-Peer-Id` — this server's own peerId * - `X-Broadcast-Timestamp` — unix seconds, peer rejects ±5min @@ -106,15 +112,20 @@ export class BroadcastService extends PuterService { /** Identifier used to tell what server a redis fan-out is coming from. */ #redisSourceId: string = `${this.config.serverId}:${randomUUID()}`; - /** Coalesced outbound events, keyed by serialized shape. */ - #outboundEventsByDedupKey = new Map(); + /** + * Undelivered events per peer, keyed by serialized shape, oldest first — a + * peer that fails keeps its own backlog without replaying it to others. + */ + #outboundQueues = new Map>(); + /** Orders queue entries, so a delivery only clears what it actually sent. */ + #outboundSeq = 0; #outboundFlushTimer: ReturnType | null = null; #outboundIsFlushing = false; #dedupFallbackCounter = 0; #webhookReplayWindowSeconds = 300; #outboundFlushMs = 2000; - /** Bound on re-queued events, so a peer that stays down can't grow it. */ + /** Bound on one peer's queue, so a peer that stays down can't grow it. */ #outboundMaxQueued = 10_000; #webhookProtocol: 'http' | 'https' = 'https'; #webhookHostHeader: string | null = null; @@ -406,10 +417,44 @@ export class BroadcastService extends PuterService { const event: BroadcastEvent = { key, data, meta: safeMeta }; const dedupKey = this.#createDedupKey(event); - this.#outboundEventsByDedupKey.set(dedupKey, event); + const seq = ++this.#outboundSeq; + for (const peer of this.#webhookPeers) { + let queue = this.#outboundQueues.get(peer); + if (!queue) { + queue = new Map(); + this.#outboundQueues.set(peer, queue); + } + // Re-insert so a repeat moves to the tail: queue order stays + // arrival order, and the front is always the oldest. + queue.delete(dedupKey); + queue.set(dedupKey, { event, seq }); + this.#trimOutboundQueue(peer, queue); + } this.#scheduleOutboundFlush(); } + #trimOutboundQueue( + peer: IBroadcastPeerConfig, + queue: Map, + ): void { + let overflow = queue.size - this.#outboundMaxQueued; + if (overflow <= 0) return; + console.warn( + `[broadcast] outbound queue for peer ${peer.peerId ?? 'unknown'} over ${this.#outboundMaxQueued}; dropping ${overflow} oldest`, + ); + for (const key of queue.keys()) { + if (overflow-- <= 0) break; + queue.delete(key); + } + } + + #hasQueuedOutbound(): boolean { + for (const queue of this.#outboundQueues.values()) { + if (queue.size > 0) return true; + } + return false; + } + #createDedupKey(event: BroadcastEvent): string { try { return JSON.stringify(event); @@ -430,58 +475,50 @@ export class BroadcastService extends PuterService { } async #flushOutboundEvents(): Promise { - if ( - this.#outboundIsFlushing || - this.#outboundEventsByDedupKey.size === 0 - ) - return; + if (this.#outboundIsFlushing || !this.#hasQueuedOutbound()) return; this.#outboundIsFlushing = true; try { - const events = [...this.#outboundEventsByDedupKey.values()]; - this.#outboundEventsByDedupKey.clear(); - - let undelivered = false; - for (const peer of this.#webhookPeers) { - try { - await this.#sendWebhookToPeer(peer, events); - } catch (err) { - const peerId = peer.peerId ?? 'unknown'; - undelivered = true; - console.warn( - `[broadcast] webhook send to peer ${peerId} failed`, - err, - ); - } - } - // A lost flat-perm invalidation has no TTL to heal it. - if (undelivered) this.#requeueOutbound(events); + await Promise.all( + [...this.#outboundQueues].map(([peer, queue]) => + this.#flushPeer(peer, queue), + ), + ); } finally { this.#outboundIsFlushing = false; - // Anything that arrived during flush gets the next tick. - if (this.#outboundEventsByDedupKey.size > 0) { - this.#scheduleOutboundFlush(); - } + // Undelivered events, and anything that arrived during the + // flush, get the next tick. + if (this.#hasQueuedOutbound()) this.#scheduleOutboundFlush(); } } - /** Put undelivered events back for the next flush, oldest dropped first. */ - #requeueOutbound(events: BroadcastEvent[]): void { - for (const event of events) { - const dedupKey = this.#createDedupKey(event); - // Whatever arrived since is fresher — never overwrite it. - if (this.#outboundEventsByDedupKey.has(dedupKey)) continue; - this.#outboundEventsByDedupKey.set(dedupKey, event); + /** + * Send one peer its queue. Events stay queued until this peer accepts them + * — a lost flat-perm invalidation has no TTL to heal it. + */ + async #flushPeer( + peer: IBroadcastPeerConfig, + queue: Map, + ): Promise { + if (queue.size === 0) return; + const sent = [...queue]; + try { + await this.#sendWebhookToPeer( + peer, + sent.map(([, queued]) => queued.event), + ); + } catch (err) { + console.warn( + `[broadcast] webhook send to peer ${peer.peerId ?? 'unknown'} failed`, + err, + ); + return; } - let overflow = - this.#outboundEventsByDedupKey.size - this.#outboundMaxQueued; - if (overflow <= 0) return; - console.warn( - `[broadcast] outbound queue over ${this.#outboundMaxQueued}; dropping ${overflow} oldest`, - ); - for (const key of this.#outboundEventsByDedupKey.keys()) { - if (overflow-- <= 0) break; - this.#outboundEventsByDedupKey.delete(key); + for (const [dedupKey, queued] of sent) { + // A repeat that arrived mid-flush is newer than what was sent. + if (queue.get(dedupKey)?.seq === queued.seq) { + queue.delete(dedupKey); + } } }