From 75dbc58628067d8b1340fc73ec3b2337e104e1d9 Mon Sep 17 00:00:00 2001
From: Juan Castro <jfcastro9208@gmail.com>
Date: Tue, 18 Aug 2026 15:31:59 -0400
Subject: [PATCH] fix: keep an undelivered broadcast event instead of dropping
 it
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The outbound queue was cleared before the send, so a peer that timed out
took its events with it. Most were survivable — a lost cache invalidation
heals when the entry's TTL lapses. A revoke's flat-perm invalidation is
not: grant-path entries carry no expiry, so a peer went on serving a
withdrawn grant until something else wrote that key.

Failed sends now go back on the queue, which the existing flush timer
retries. Anything queued since wins over the retry, and the queue is
bounded at 10,000 with the oldest dropped first, so a peer that stays
down cannot grow it without limit. Each retry is signed at send time, so
it is not rejected against the replay window.
---
 .../broadcast/BroadcastService.test.ts        | 27 +++++++++++++++++++
 .../services/broadcast/BroadcastService.ts    | 26 ++++++++++++++++++
 2 files changed, 53 insertions(+)

diff --git a/src/backend/services/broadcast/BroadcastService.test.ts b/src/backend/services/broadcast/BroadcastService.test.ts
index 54f69a1d6..2af9f2acf 100644
--- a/src/backend/services/broadcast/BroadcastService.test.ts
+++ b/src/backend/services/broadcast/BroadcastService.test.ts
@@ -603,6 +603,33 @@ describe('BroadcastService outbound flush', () => {
         );
         expect(occurrences).toBe(1);
     });
+
+    // A dropped invalidation can outlive every TTL meant to heal it.
+    it('sends a failed event again instead of dropping it', async () => {
+        const cacheKey = 'requeue-test';
+        const carries = (call: unknown) => {
+            const body = (call as [{ data?: string }])[0]?.data;
+            return typeof body === 'string' && body.includes(cacheKey);
+        };
+
+        axiosRequestMock.mockRejectedValue(new Error('peer unreachable'));
+        server.clients.event.emit(
+            'outer.cacheUpdate' as never,
+            { cacheKey: [cacheKey] } as never,
+            {},
+        );
+        await waitForFlush(() => axiosRequestMock.mock.calls.some(carries));
+
+        axiosRequestMock.mockReset();
+        axiosRequestMock.mockResolvedValue({
+            status: 200,
+            statusText: 'OK',
+            data: 'ok',
+        });
+
+        // Nothing new is emitted — only the retry can satisfy this.
+        await waitForFlush(() => axiosRequestMock.mock.calls.some(carries));
+    });
 });
 
 // -- Header validation ------------------------------------------------
diff --git a/src/backend/services/broadcast/BroadcastService.ts b/src/backend/services/broadcast/BroadcastService.ts
index b19e7edc7..30b060dcd 100644
--- a/src/backend/services/broadcast/BroadcastService.ts
+++ b/src/backend/services/broadcast/BroadcastService.ts
@@ -98,6 +98,8 @@ export class BroadcastService extends PuterService {
 
     #webhookReplayWindowSeconds = 300;
     #outboundFlushMs = 2000;
+    /** Bound on re-queued events, so a peer that stays down can't grow it. */
+    #outboundMaxQueued = 10_000;
     #webhookProtocol: 'http' | 'https' = 'https';
     #webhookHostHeader: string | null = null;
     /** Self-signed certs are common between Puter nodes — accept them. */
@@ -350,17 +352,21 @@ export class BroadcastService extends PuterService {
             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);
         } finally {
             this.#outboundIsFlushing = false;
             // Anything that arrived during flush gets the next tick.
@@ -370,6 +376,26 @@ export class BroadcastService extends PuterService {
         }
     }
 
+    /** 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);
+        }
+        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);
+        }
+    }
+
     async #sendWebhookToPeer(
         peer: IBroadcastPeerConfig,
         events: BroadcastEvent[],