web: address gemini feedback

Signed-off-by: Peter Gadfort <gadfort@zeroasic.com>

Author: gadfort

src/web/src/websocket-manager.js

@@ -65,6 +65,7 @@ export class WebSocketManager {
         this._cache = null;
         this._isConnected = false;
         this._shutdown = false; // set by server "shutdown" message
+        this._livenessTimer = undefined; // set by _startLivenessMonitor
         this._startLivenessMonitor();
         this.connect();
     }
@@ -144,6 +145,7 @@ export class WebSocketManager {
             this._queue.clear();
             if (this._shutdown) {
                 console.log('WebSocket closed (server stopped)');
+                this._stopLivenessMonitor(); // no socket will reopen
                 return; // don't reconnect after intentional shutdown
             }
             console.log('WebSocket closed, reconnecting...');
@@ -169,7 +171,14 @@ export class WebSocketManager {
         // It still counts as liveness (above) but frees no in-flight slot.
         if (id === 0 && type === 0) {
             const payload = data.slice(8);
-            const msg = JSON.parse(new TextDecoder().decode(payload));
+            let msg;
+            try {
+                msg = JSON.parse(new TextDecoder().decode(payload));
+            } catch (e) {
+                // A malformed push must not break the message loop.
+                console.error('Failed to parse server push JSON:', e);
+                return;
+            }
             // The server announces the in-flight window sized to its machine.
             // Consume it here rather than forwarding to onPush.
             if (msg && msg.type === 'config') {
@@ -268,7 +277,11 @@ export class WebSocketManager {
         // If still queued it never hit the wire — drop it for free (no server
         // work, no buffered bytes). This is the cancellation that actually
         // matters; cancelling an already-sent request can't recall the bytes.
-        if (this._queue.delete(id)) {
+        // Settle the promise so callers don't leak a forever-pending request.
+        const queued = this._queue.get(id);
+        if (queued) {
+            this._queue.delete(id);
+            queued.reject(new Error('Request cancelled'));
             this.onStatusChange();
             return;
         }
@@ -287,11 +300,22 @@ export class WebSocketManager {
         if (typeof setInterval === 'undefined') {
             return;
         }
-        const timer = setInterval(() => this._checkLiveness(),
-                                  LIVENESS_INTERVAL_MS);
+        this._livenessTimer = setInterval(() => this._checkLiveness(),
+                                          LIVENESS_INTERVAL_MS);
         // Don't keep a Node process alive for this timer (no-op in browsers).
-        if (timer && typeof timer.unref === 'function') {
-            timer.unref();
+        if (this._livenessTimer
+            && typeof this._livenessTimer.unref === 'function') {
+            this._livenessTimer.unref();
+        }
+    }
+
+    // Stop the liveness timer. Called on intentional shutdown so the interval
+    // doesn't keep firing _checkLiveness against a socket we'll never reopen.
+    _stopLivenessMonitor() {
+        if (this._livenessTimer !== undefined
+            && typeof clearInterval !== 'undefined') {
+            clearInterval(this._livenessTimer);
+            this._livenessTimer = undefined;
         }
     }
 

src/web/test/js/test-websocket-manager.js

@@ -89,6 +89,16 @@ describe('WebSocketManager', () => {
             mgr.handleMessage(buildFrame(999, 0, payload));
             assert.equal(mgr.pending.size, 0);
         });
+
+        it('survives a malformed server-push payload', () => {
+            const mgr = new WebSocketManager('ws://fake');
+            let pushed = false;
+            mgr.onPush = () => { pushed = true; };
+            // id=0 push frame with invalid JSON must not throw or call onPush.
+            const bad = new TextEncoder().encode('{ not json');
+            assert.doesNotThrow(() => mgr.handleMessage(buildFrame(0, 0, bad)));
+            assert.equal(pushed, false, 'no push delivered for malformed JSON');
+        });
     });
 
     describe('request', () => {
@@ -286,6 +296,32 @@ describe('WebSocketManager flow control', () => {
         assert.equal(mgr.socket.sent.length, sentBefore,
             'a queued cancel sends nothing');
     });
+
+    it('cancelling a queued request settles its promise', async () => {
+        const mgr = new WebSocketManager('ws://fake');
+        await mgr.readyPromise;
+
+        // Saturate the wire so the next request stays queued.
+        const ids = [];
+        for (let i = 0; i < 200; i++) {
+            const p = mgr.request({ type: 'tile', n: i });
+            p.catch(() => {});
+            ids.push(p.requestId);
+        }
+        const queuedId = ids[ids.length - 1];
+        assert.ok(mgr._queue.has(queuedId), 'last request is queued');
+
+        // Re-issue a tracked promise for the queued id so we can observe it.
+        const entry = mgr._queue.get(queuedId);
+        const observed = new Promise((resolve, reject) => {
+            entry.resolve = resolve;
+            entry.reject = reject;
+        });
+
+        mgr.cancel(queuedId);
+        // The promise must reject — not hang forever — so callers clean up.
+        await assert.rejects(observed, { message: 'Request cancelled' });
+    });
 });
 
 describe('WebSocketManager liveness', () => {
@@ -353,6 +389,21 @@ describe('WebSocketManager liveness', () => {
             performance.now = realNow;
         }
     });
+
+    it('stops the liveness timer on intentional shutdown', async () => {
+        const mgr = new WebSocketManager('ws://fake');
+        await mgr.readyPromise;
+        assert.notEqual(mgr._livenessTimer, undefined,
+            'timer runs while connected');
+
+        // Server-initiated shutdown: the socket closes and must not reconnect,
+        // so the recurring liveness check must be cleared too.
+        mgr._shutdown = true;
+        mgr.socket.readyState = 3; // CLOSED
+        mgr.socket.onclose?.();
+        assert.equal(mgr._livenessTimer, undefined,
+            'liveness timer cleared after shutdown');
+    });
 });
 
 describe('WebSocketManager.fromCache', () => {