test(review): wait for a closed stream to be dropped at a later write (#314)

The hidden-tab and teardown specs assumed the one scan addition after a
page closed its change stream is the write that finds the stream closed.
The server notices a closed stream only when a write fails, and nothing
orders the browser's close before that write: on WebKit under load the
close arrived after the change was written, the write succeeded, and the
stream stayed counted until the 25 s keepalive (CI job 120735035017).

Both specs now send no-op change notices (new fixture route
POST /__test__/review/notify) until the server has dropped the stream. A
new spec makes the CI ordering certain by holding the page's close until
the change has reached it.

tests/browser/workspace-review.spec.mjs

@@ -28,6 +28,17 @@ async function pending(request) {
 }
 const links = async request => (await (await request.get('/__test__/review/links')).json()).links;
 const streams = async request => (await (await request.get('/__test__/review/streams')).json());
+/**
+ * Wait until the server has dropped every stream a page closed. It notices a closed stream only when a write to it
+ * fails, and the close can reach it after any one write (WebKit under load): each poll sends one more change notice,
+ * which changes nothing in the queue, until the count is 0.
+ */
+async function closedStreamsDropped(request) {
+    await expect.poll(async () => {
+        await expect(await request.post('/__test__/review/notify')).toBeOK();
+        return (await streams(request)).streams;
+    }).toBe(0);
+}
 const apply = page => panel(page).getByRole('button', { name: 'Apply association', exact: true });
 const warning = page => panel(page).locator('.warning-note');
 /** The page with its first item's sampler paths loaded. */
@@ -444,9 +455,8 @@ test('a hidden tab closes its stream and a visible one reads only what changed m
     await open(page);
     await expect.poll(async () => (await streams(request)).streams).toBe(1);
     await setVisible(page, false);
-    // The server drops a closed stream at its next write: this scan addition is one.
     await expect(await request.post('/__test__/review/add', { data: { kind: 'scan' } })).toBeOK();
-    await expect.poll(async () => (await streams(request)).streams).toBe(0);
+    await closedStreamsDropped(request);
     await expect(cards(page)).toHaveCount(1);
     expect(reads.filter(search => search === '?limit=40')).toHaveLength(1);
     // Visible again: the reopened stream's ready names a newer version, so the page reads once.
@@ -459,6 +469,30 @@ test('a hidden tab closes its stream and a visible one reads only what changed m
     expect(reads.filter(search => search === '?limit=40')).toHaveLength(2);
 });
 
+test('a stream whose close reaches the server after a change is dropped at a later write', async ({ page, request }) => {
+    // The CI ordering made certain: the hidden page's close arrives only after the scan addition's change was
+    // written to its stream, so that write succeeds and a later one finds the stream closed.
+    await seed(request, 1);
+    await open(page);
+    await expect.poll(async () => (await streams(request)).streams).toBe(1);
+    await page.evaluate(() => {
+        const close = EventSource.prototype.close;
+        window.lateChanges = 0;
+        window.heldCloses = [];
+        EventSource.prototype.close = function () {
+            this.addEventListener('change', () => { window.lateChanges += 1; });
+            window.heldCloses.push(() => close.call(this));
+        };
+    });
+    await setVisible(page, false);
+    await expect(await request.post('/__test__/review/add', { data: { kind: 'scan' } })).toBeOK();
+    await expect.poll(() => page.evaluate(() => window.lateChanges)).toBe(1);
+    await page.evaluate(() => window.heldCloses.splice(0).forEach(close => close()));
+    expect((await streams(request)).streams).toBe(1);
+    await closedStreamsDropped(request);
+    await expect(cards(page)).toHaveCount(1);
+});
+
 test('a re-read refreshes a card whose media or target changed', async ({ page, request }) => {
     // M6 and M10: a root came or went between reads; the card follows, and a known thumbnail is its media.
     await seed(request, 2);
@@ -549,8 +583,6 @@ test('teardown releases captures listeners and requests', async ({ page, request
     await expect(main(page).getByRole('heading', { level: 1 })).toHaveText('Tags & categories');
     await expect.poll(async () => (await streams(request)).captures).toBe(0);
     await route.abort().catch(() => {});
-    // The server drops a closed stream at its next write: a monitor addition is one.
-    await expect(await request.post('/__test__/review/add', { data: { kind: 'scan' } })).toBeOK();
-    await expect.poll(async () => (await streams(request)).streams).toBe(0);
+    await closedStreamsDropped(request);
     await expect.poll(() => inFlight.size).toBe(0);
 });

tests/browser/workspace_review_fixture.py

@@ -34,6 +34,9 @@ through the scan's recorder (as a running scan's batch does; the monitor
 never adds items), ``file`` only writes one (a metadata scan records it). ``GET
 /__test__/review/links`` answers each review file's prompt and pair links, and
 ``GET /__test__/review/streams`` the open change streams and captures.
+``POST /__test__/review/notify`` sends one queue change notice without
+changing the queue: each open stream writes a ``change``, and a stream whose
+page closed is dropped at such a write.
 """
 
 from __future__ import annotations
@@ -356,6 +359,15 @@ def register_review_commands(
         )
         return web.json_response({"success": True, **added})
 
+    @routes.post("/__test__/review/notify")
+    async def notify(request: web.Request) -> web.Response:
+        if request.query or request.can_read_body:
+            raise web.HTTPBadRequest()
+        from database.review_changes import notify_review_change
+
+        notify_review_change(api.db.model.db_path)
+        return web.json_response({"success": True})
+
     @routes.get("/__test__/review/links")
     async def links(request: web.Request) -> web.Response:
         if request.query: