Fix hx-sse reader-cancel race that produces an unhandled rejection (#3935)

Fix reader-cancel race in hx-sse.js

parseSSE's `finally` released the reader's lock, but `connection.reader`
was only nulled one microtask later (via the for-await-of loop's
async-generator close protocol). Any concurrent `connection.reader.cancel()`
call (pauseOnBackground's visibilitychange handler, or cleanup()) landing in
that one-microtask window canceled an already-released reader, which per the
Streams spec rejects instead of throwing synchronously -- surfacing as an
unhandled promise rejection.

Fix: pass `connection` into parseSSE and null `connection.reader` in the
same synchronous finally block as releaseLock(), closing the window instead
of guarding each call site.

Includes a regression test that deterministically lands in the race window
by chaining the concurrent cancel trigger off the same promise the internal
`await reader.read()` is waiting on.

Author: marciomazza

src/ext/hx-sse.js

@@ -32,7 +32,9 @@
     // SSE PARSER
     // ========================================
 
-    async function* parseSSE(reader, lastEventId = '') {
+    async function* parseSSE(connection) {
+        let reader = connection.reader;
+        let lastEventId = connection.lastEventId;
         let decoder = new TextDecoder();
         let buffer = '';
         let hasData = false;
@@ -100,6 +102,9 @@
             }
         } finally {
             reader.releaseLock();
+            // On the same tick as releaseLock(), so cleanup() or visibilityHandler 
+            // calling connection.reader?.cancel() can't hit a released reader.
+            connection.reader = null;
         }
     }
 
@@ -241,7 +246,7 @@
                 try {
                     connection.reader = currentResponse.body.getReader();
 
-                    for await (let msg of parseSSE(connection.reader, connection.lastEventId)) {
+                    for await (let msg of parseSSE(connection)) {
                         if (!element.isConnected || reconnectRequested) break;
 
                         if (msg.hasId) {
@@ -285,7 +290,6 @@
                     }
                 }
 
-                connection.reader = null;
                 if (!element.isConnected) break;
 
                 connection.attempt++;

test/tests/ext/hx-sse.js

@@ -591,6 +591,52 @@ describe('hx-sse SSE extension', function() {
         stream.close();
     });
 
+    it('cancelling an already-released reader does not produce an unhandled rejection', async function() {
+        // Hand-rolled reader so the test can fire visibilitychange (→ cancel()) off the same
+        // promise the extension's `await reader.read()` is waiting on, landing it deterministically
+        // in the same tick the stream closes on.
+        let resolveRead;
+        let released = false;
+        const pendingRead = new Promise(resolve => { resolveRead = resolve; });
+        const reader = {
+            read: () => pendingRead,
+            releaseLock() { released = true; },
+            cancel() {
+                if (released) return Promise.reject(new TypeError('Cannot cancel a stream using a released reader'));
+                released = true;
+                return Promise.resolve();
+            }
+        };
+
+        fetchMock.mockResponse('GET', '/reader-cancel-race', () => {
+            const response = new MockResponse({getReader: () => reader}, {
+                headers: {'Content-Type': 'text/event-stream'}
+            });
+            response.body = {getReader: () => reader};
+            return response;
+        });
+
+        createProcessedHTML('<button hx-get="/reader-cancel-race" hx-config="sse.pauseOnBackground:true" hx-swap="innerHTML">Connect</button>');
+
+        let rejections = [];
+        let onRejection = e => rejections.push(e.reason);
+        window.addEventListener('unhandledrejection', onRejection);
+
+        find('button').click();
+        await waitForEvent('htmx:after:sse:connection');
+
+        pendingRead.then(() => {
+            Object.defineProperty(document, 'hidden', {value: true, configurable: true});
+            document.dispatchEvent(new Event('visibilitychange'));
+        });
+        resolveRead({done: true, value: undefined});
+
+        await htmx.timeout(20);
+        window.removeEventListener('unhandledrejection', onRejection);
+
+        assert.deepEqual(rejections, [], 'Attempting to cancel an already-released reader must not produce an unhandled rejection');
+    });
+
     it('custom events trigger on element and bubble', async function() {
         const stream = mockStreamResponse('/custom-events');
         createProcessedHTML('<button hx-get="/custom-events" hx-swap="innerHTML">Connect</button>');