From 071a0187977aa91966612d6718f5baee8bc67671 Mon Sep 17 00:00:00 2001
From: Eric Dahl <eric@skeyelab.com>
Date: Tue, 22 Sep 2026 13:32:02 -0400
Subject: [PATCH] fix: retry unacked events over an open WebSocket

A connected node only received events when they first arrived or when it
authenticated. If a delivery failed (the node didn't ack, e.g. its local
webhook returned 404), the event sat in `leased` until the node reconnected.
Before the keepalive (#81) the ~45s reconnect cycle masked this; with stable
connections an unacked event could wait indefinitely.

Re-send pending and lease-expired events on every 30s keepalive tick, using
the existing pushPendingEventsForNode (which re-leases, so nothing is sent
twice inside a lease window). Nothing is re-sent to a node being terminated.
---
 app/lib/ws/node-socket.ts        | 16 +++++++++----
 tests/lib/ws/node-socket.test.ts | 39 ++++++++++++++++++++++++++++++++
 2 files changed, 51 insertions(+), 4 deletions(-)

diff --git a/app/lib/ws/node-socket.ts b/app/lib/ws/node-socket.ts
index 7191619..03cf337 100644
--- a/app/lib/ws/node-socket.ts
+++ b/app/lib/ws/node-socket.ts
@@ -33,6 +33,16 @@ export function createNodeSocketHandler(deps: NodeSocketDeps): (ws: WebSocket) =
       });
     }
 
+    /** Push pending and lease-expired events. Runs at auth and on every keepalive tick, so a
+     * delivery the node didn't ack is retried without waiting for a reconnect. */
+    function pushPending(): void {
+      if (!nodeId) return;
+      const id = nodeId;
+      deps.pushPendingEvents(ws, id).catch((err: Error) => {
+        console.error('[ws] Failed to push pending events for node %s: %s', id, err.message);
+      });
+    }
+
     function stopPinging(): void {
       if (pingTimer) clearInterval(pingTimer);
       pingTimer = null;
@@ -84,12 +94,10 @@ export function createNodeSocketHandler(deps: NodeSocketDeps): (ws: WebSocket) =
           }
           alive = false;
           ws.ping();
+          pushPending();
         }, deps.pingIntervalMs);
 
-        const id = nodeId;
-        deps.pushPendingEvents(ws, id).catch((err: Error) => {
-          console.error('[ws] Failed to push pending events for node %s: %s', id, err.message);
-        });
+        pushPending();
         return;
       }
 
diff --git a/tests/lib/ws/node-socket.test.ts b/tests/lib/ws/node-socket.test.ts
index 0efb0e3..52aace7 100644
--- a/tests/lib/ws/node-socket.test.ts
+++ b/tests/lib/ws/node-socket.test.ts
@@ -131,6 +131,45 @@ describe('createNodeSocketHandler', () => {
     expect(deps.markSeen).toHaveBeenCalledTimes(2);
   });
 
+  test('re-sends pending and lease-expired events on each keepalive tick', async () => {
+    const { deps, ws } = setup();
+    await authenticate(ws);
+    expect(deps.pushPendingEvents).toHaveBeenCalledTimes(1);
+
+    await vi.advanceTimersByTimeAsync(PING_MS);
+    expect(deps.pushPendingEvents).toHaveBeenCalledTimes(2);
+    expect(deps.pushPendingEvents).toHaveBeenLastCalledWith(ws, 'node-1');
+
+    ws.emit('pong');
+    await vi.advanceTimersByTimeAsync(PING_MS);
+    expect(deps.pushPendingEvents).toHaveBeenCalledTimes(3);
+  });
+
+  test('does not re-send to a node that is being terminated', async () => {
+    const { deps, ws } = setup();
+    await authenticate(ws);
+
+    await vi.advanceTimersByTimeAsync(PING_MS);
+    const callsBefore = (deps.pushPendingEvents as ReturnType<typeof vi.fn>).mock.calls.length;
+    await vi.advanceTimersByTimeAsync(PING_MS); // no pong: this tick terminates
+    expect(ws.terminate).toHaveBeenCalledTimes(1);
+    expect(deps.pushPendingEvents).toHaveBeenCalledTimes(callsBefore);
+  });
+
+  test('a failed re-send is logged, not thrown', async () => {
+    const pushPendingEvents = vi.fn()
+      .mockResolvedValueOnce(undefined)
+      .mockRejectedValueOnce(new Error('db down'));
+    const { ws } = setup({ pushPendingEvents });
+    const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {});
+    await authenticate(ws);
+
+    await vi.advanceTimersByTimeAsync(PING_MS);
+
+    expect(errorSpy).toHaveBeenCalled();
+    errorSpy.mockRestore();
+  });
+
   test('stops pinging and removes the connection on close', async () => {
     const { deps, ws } = setup();
     await authenticate(ws);