PUT-2012 fix broadcast service drops

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<string, string[][]> = {};
+    let failedAttempts = 0;
+    /** While set, the failing peer's request hangs until it resolves. */
+    let failingPeerGate: Promise<void> | 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', () => {

src/backend/services/broadcast/BroadcastService.ts

@@ -31,6 +31,11 @@ interface BroadcastEvent {
     meta: Record<string, unknown>;
 }
 
+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<string, BroadcastEvent>();
+    /**
+     * 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<IBroadcastPeerConfig, Map<string, QueuedEvent>>();
+    /** Orders queue entries, so a delivery only clears what it actually sent. */
+    #outboundSeq = 0;
     #outboundFlushTimer: ReturnType<typeof setTimeout> | 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<string, QueuedEvent>,
+    ): 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<void> {
-        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<string, QueuedEvent>,
+    ): Promise<void> {
+        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);
+            }
         }
     }