From 9b9a07ccef705397e738f113615d7694740cc109 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Thu, 22 Oct 2026 20:09:51 -0700 Subject: [PATCH] feat(review): keep the loaded depth on live re-reads and read once per visit A queue change re-read only the first 40 items, so the pages Load more had shown vanished and the page jumped. Each re-read now reads page after page until it holds as many items as were shown (a restore met after the first page reads again from the start), so the cards and the scroll position stay. The stream's ready made every visit read the queue twice. List reads now answer the queue's version (library_epoch, review_revision) and ready names the version read after subscribing; the page re-reads on ready only when the queue it shows is older. The SAVED VIEWS count follows the page's total, so a change costs no count read. A stream counts as open only once ready is sent. --- TESTING.md | 37 ++++-- database/review.py | 13 +++ py/api/review.py | 2 + py/api/review_events.py | 36 +++++- tests/browser/workspace-review.spec.mjs | 50 +++++++++ tests/js/workspace-review-state.test.js | 142 ++++++++++++++++++++++++ tests/test_api_review_events.py | 83 +++++++++++++- web/js/workspace/index.mjs | 21 +++- web/js/workspace/review-state.mjs | 92 ++++++++++++--- web/js/workspace/review.mjs | 38 +++++-- 10 files changed, 461 insertions(+), 53 deletions(-) diff --git a/TESTING.md b/TESTING.md index 2fe3757..eac950a 100644 --- a/TESTING.md +++ b/TESTING.md @@ -1635,20 +1635,33 @@ pending count (`GET /prompt_manager/review/items?limit=1`, its `total`). `GET /prompt_manager/review/events` (`py/api/review_events.py`): `event: ready` when it opens, then `event: change` after each committed queue change (an item the scan or the monitor recorded, a spelling refresh, a decision from any tab; `database/review_changes.py` -notifies after the commit, from whichever thread wrote it) and after a restore. Events -carry no data. The page re-reads its first page on every `ready` and `change`, keeping -the open item, its choice and an unanswered decision; re-reads asked for while one runs -collapse into one more, and one waits while a decision is in flight. The row's count is -re-read with it. Nothing polls: a `: keepalive` comment every 25 s only lets the server -drop a closed tab's subscription. Shutdown (`on_shutdown`) ends every stream and answers -a later one 503; the browser's own EventSource reconnection then stops. A duplicate -recording and a refused decision change nothing and send nothing. Residual: the stream -is one HTTP/1.1 connection per open Needs review page (browsers allow six per origin), so -it lives only while the page is open; the re-read replaces Load more's extra pages with -the first 40. +notifies after the commit, from whichever thread wrote it) and after a restore. `change` +carries no data. `ready` carries the queue's version `{library_epoch, review_revision}`, +read after the stream subscribed, and every list read answers the same pair +(`review_revision` is the review index revision, which every queue change moves; a +restore moves the epoch). **One read per visit (F3):** the page re-reads on `ready` only +when the queue it shows is older than that version (an unknown version, `{}` when it +cannot be read, re-reads); a `ready` heard while the first read runs is compared once that +read answers. The test server counts a stream as open only once it has sent `ready`, so a +browser test that changes the queue after that count is a `change`, never part of the +version. The page re-reads on every `change`, keeping the open item, its choice and an +unanswered decision; re-reads asked for while one runs collapse into one more, and one +waits while a decision is in flight. **Every re-read keeps the loaded depth (F3):** it +reads page after page (40 each, the route's limit) until it holds as many items as Load +more had shown, so the cards stay and the scroll position does not move; a restore met +after the first page reads once more from the first page. The row's count follows the +total the page shows (no count read per change). Nothing polls: a `: keepalive` comment +every 25 s only lets the server drop a closed tab's subscription. Shutdown (`on_shutdown`) +ends every stream and answers a later one 503; the browser's own EventSource reconnection +then stops. A duplicate recording and a refused decision change nothing and send nothing. +Residual: the stream is one HTTP/1.1 connection per open Needs review page (browsers allow +six per origin), so it lives only while the page is open; a re-read of a deep queue costs +one request per 40 items shown. Pinned by `tests/test_api_review_events.py`, `tests/js/workspace-review-state.test.js` -("refresh …") and `tests/browser/workspace-review.spec.mjs` ("an open page shows …"). +("refresh …", "a refresh keeps the loaded depth …", "ready naming …") and +`tests/browser/workspace-review.spec.mjs` ("an open page shows …", "a visit reads its +queue once …", "a live re-read keeps the loaded depth and the scroll position"). Leaving the page mid-decision releases that decision's capture (the server may never see the request). - **D3** and **D20** apply to the queue grid (its cards stay in the grid's flow; it is diff --git a/database/review.py b/database/review.py index b1269aa..5f15bb7 100644 --- a/database/review.py +++ b/database/review.py @@ -317,6 +317,19 @@ class ReviewOperations: raise sqlite3.DatabaseError("Missing review queue metadata") return row[0] + def review_queue_version(self) -> dict: + """``{library_epoch, review_revision}`` of the queue, from one snapshot. + + The pair a list read answers: equal pairs mean the same queue (a + restore moves the epoch, every queue change moves the revision). + """ + with self._annotation_read() as conn: + epoch = conn.execute(_EPOCH).fetchone() + revision = conn.execute(_REVIEW_INDEX).fetchone() + if epoch is None or revision is None: + raise sqlite3.DatabaseError("Missing review queue metadata") + return {"library_epoch": epoch[0], "review_revision": revision[0]} + def review_items_for(self, locator: str) -> list: """Every item of the file at ``locator``, oldest first, evidence parsed.""" _validate_locator(locator) diff --git a/py/api/review.py b/py/api/review.py index 0e3fc62..bdbcbae 100644 --- a/py/api/review.py +++ b/py/api/review.py @@ -328,6 +328,8 @@ class ReviewRoutesMixin: { "success": True, "library_epoch": page["epoch"], + # F3: the queue version the change stream's ``ready`` names. + "review_revision": page["revision"], "total": page["total"], "items": views, "next_cursor": next_cursor, diff --git a/py/api/review_events.py b/py/api/review_events.py index 61d2e0f..91a9d2a 100644 --- a/py/api/review_events.py +++ b/py/api/review_events.py @@ -5,8 +5,12 @@ open Needs review page: ``event: ready`` when it opens, then ``event: change`` after each committed queue change (an item the scan or the monitor recorded, a decision from any tab, a spelling refresh) and after a restore. Notices that arrive while one is being written collapse into one. -The page re-reads its queue on each; nothing polls. The events carry no data -(``{}``), so nothing about the queue is pushed. +The page re-reads its queue on each ``change``; nothing polls. ``change`` +carries no data (``{}``). ``ready`` carries the queue's version +``{library_epoch, review_revision}``, read after the stream subscribed, the +same pair a list read answers: a page whose first read already answered that +version skips the re-read (F3, one read per visit), and any later change is a +``change`` event. Nothing else about the queue is pushed. A ``: keepalive`` comment every ``HEARTBEAT_SECONDS`` lets the server notice a closed tab (its write fails) and drop the subscription; clients ignore @@ -15,6 +19,7 @@ every stream, and a stream asked for afterwards is a 503. """ import asyncio +import json from aiohttp import web @@ -27,11 +32,15 @@ except ImportError: EVENTS_URL = "/prompt_manager/review/events" HEARTBEAT_SECONDS = 25 -_READY = b"event: ready\ndata: {}\n\n" _CHANGE = b"event: change\ndata: {}\n\n" _KEEPALIVE = b": keepalive\n\n" +def _ready(version: dict) -> bytes: + data = json.dumps(version, separators=(",", ":"), sort_keys=True) + return f"event: ready\ndata: {data}\n\n".encode("utf-8") + + class _Stream: """One open stream: a change flag set from any thread, and a close flag.""" @@ -39,6 +48,7 @@ class _Stream: self._loop = loop self.changed = asyncio.Event() self.closed = asyncio.Event() + self.ready = False def wake(self) -> None: """Thread-safe: mark a change (the loop may already be gone).""" @@ -81,8 +91,10 @@ class ReviewEventsMixin: return streams def review_stream_count(self) -> int: - """How many change streams are open (tests and diagnostics).""" - return len(self._review_streams()) + """How many change streams have sent ``ready`` and are open (tests and + diagnostics): a queue change after this count moved is a ``change`` + event on them, never part of the version ``ready`` named.""" + return sum(1 for stream in self._review_streams() if stream.ready) async def close_review_events(self, app) -> None: """End every open stream; later requests get 503 (shutdown hook).""" @@ -90,6 +102,16 @@ class ReviewEventsMixin: for stream in list(self._review_streams()): stream.closed.set() + async def _queue_version(self) -> dict: + """The queue version for ``ready``, read after subscribing (so a change + after this read is an event); ``{}`` when it cannot be read, and the + page then re-reads.""" + try: + return await self._run_in_executor(self.db.review_queue_version) + except Exception: + self.logger.warning("Needs review version read failed", exc_info=True) + return {} + async def review_events(self, request: web.Request) -> web.StreamResponse: if getattr(self, "_review_events_closed", False): return web.json_response( @@ -113,8 +135,10 @@ class ReviewEventsMixin: }, ) try: + version = await self._queue_version() await response.prepare(request) - await response.write(_READY) + await response.write(_ready(version)) + stream.ready = True while True: outcome = await stream.next(HEARTBEAT_SECONDS) if outcome == "closed": diff --git a/tests/browser/workspace-review.spec.mjs b/tests/browser/workspace-review.spec.mjs index ddce6ad..02eabaf 100644 --- a/tests/browser/workspace-review.spec.mjs +++ b/tests/browser/workspace-review.spec.mjs @@ -358,6 +358,56 @@ test('a decision in another tab leaves the open page and opens the next image', await expect(context(page)).toBeEnabled(); }); +/** The queue reads (`?limit=40`, cursors) and count reads (`?limit=1`) the page sends from now on. */ +function reviewReads(page) { + const reads = []; + page.on('request', item => { + const url = new URL(item.url()); + if (url.pathname === '/prompt_manager/review/items') reads.push(url.searchParams.get('cursor') ? 'more' : url.search); + }); + return reads; +} + +test('a visit reads its queue once, and a change reads it once more', async ({ page, request }) => { + // F3 ruling: the stream's ready must not read a queue the first read already holds. + await seed(request, 1); + const reads = reviewReads(page); + await open(page); + await expect.poll(async () => (await streams(request)).streams).toBe(1); + await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK(); + await expect(cards(page)).toHaveCount(2); + await expect(reviewRow(page)).toHaveText('Needs review2'); + // The page's first read and the change's re-read; the shell's count read at start, and the row follows the page. + expect(reads.filter(search => search === '?limit=40')).toHaveLength(2); + expect(reads.filter(search => search === '?limit=1')).toHaveLength(1); +}); + +test('a live re-read keeps the loaded depth and the scroll position', async ({ page, request }) => { + // F3 ruling: a queue change re-reads as many items as Load more had shown, and the page does not jump. + await seed(request, 45); + await open(page); + await expect.poll(async () => (await streams(request)).streams).toBe(1); + await main(page).getByRole('button', { name: 'Load more images', exact: true }).click(); + await expect(cards(page)).toHaveCount(45); + await cards(page).last().scrollIntoViewIfNeeded(); + const scrollTop = () => cards(page).first().evaluate(node => { + for (let el = node.parentElement; el; el = el.parentElement) { + const style = getComputedStyle(el); + if (/(auto|scroll)/.test(style.overflowY) && el.scrollHeight > el.clientHeight) return el.scrollTop; + } + return document.scrollingElement.scrollTop; + }); + const before = await scrollTop(); + expect(before).toBeGreaterThan(0); + const reads = reviewReads(page); + await expect(await request.post('/__test__/review/add', { data: { kind: 'monitor' } })).toBeOK(); + await expect(main(page).locator('.notice h2')).toHaveText('46 images need a decision'); + await expect(cards(page)).toHaveCount(46); + expect(await scrollTop()).toBe(before); + expect(reads).toEqual(['?limit=40', 'more']); + await expect(main(page).locator('.infinite-continuation')).toHaveCount(0); +}); + test('teardown releases captures listeners and requests', async ({ page, request }) => { await seed(request, 1); await open(page); diff --git a/tests/js/workspace-review-state.test.js b/tests/js/workspace-review-state.test.js index 7ee0236..52678b7 100644 --- a/tests/js/workspace-review-state.test.js +++ b/tests/js/workspace-review-state.test.js @@ -464,3 +464,145 @@ test('a change heard during the first read re-reads once after it, and a failed assert.equal(broken.state.status, 'error'); assert.equal(failing.of('GET', `${ITEMS}?`).length, 1, 'Retry is the way back, not a re-read'); }); + +// F3 rulings: a live re-read keeps the loaded depth, and the stream's `ready` reads nothing when the first read is current. + +/** A keyset pager over `source()`: pages of 40 with an index cursor, each answering `version()`. */ +function pager(source, version = () => 0, { onPage } = {}) { + return call => { + const query = new URL(call.url, 'http://x').searchParams; + const cursor = query.get('cursor'); + const all = source(); + const from = cursor ? Number(cursor.slice(1)) : 0; + const answer = onPage?.(cursor); + if (answer) return answer; + const items = all.slice(from, from + 40); + const next = from + 40 < all.length ? `c${from + 40}` : null; + return response(200, page(items, { total: all.length, next_cursor: next, review_revision: version() })); + }; +} +const many = (count, prefix = 'i') => Array.from({ length: count }, (_, index) => item(`${prefix}${String(index).padStart(3, '0')}`)); +function syncOf(c, version) { + assert.equal(typeof c.sync, 'function', 'sync is not implemented'); + return c.sync(version); +} +const version = revision => ({ library_epoch: EPOCH, review_revision: revision }); + +test('a refresh keeps the loaded depth: it re-reads as many items as were loaded', async () => { + let queue = many(100); + const { c, h } = await loaded({ [`GET ${ITEMS}`]: pager(() => queue) }); + await c.loadMore(); + assert.equal(c.state.items.length, 80); + const before = h.of('GET', `${ITEMS}?`).length; + queue = [...queue, item('late')]; + await refreshOf(c); + await settle(); + assert.equal(c.state.items.length, 80, 'the Load more page is kept'); + assert.deepEqual(c.state.items.map(entry => entry.id), queue.slice(0, 80).map(entry => entry.id)); + assert.equal(c.state.total, 101); + assert.equal(c.state.nextCursor, 'c80', 'Load more continues after the kept depth'); + assert.deepEqual(h.of('GET', `${ITEMS}?`).slice(before).map(call => call.url), + [`${ITEMS}?limit=40`, `${ITEMS}?limit=40&cursor=c40`]); + assert.equal(c.state.selectedId, 'i000'); +}); + +test('a refresh after decisions elsewhere fills the loaded depth from what is left', async () => { + let queue = many(85); + const { c } = await loaded({ [`GET ${ITEMS}`]: pager(() => queue) }); + await c.loadMore(); + await c.loadMore(); + assert.equal(c.state.items.length, 85); + queue = queue.filter((_, index) => index % 10 !== 5); + await refreshOf(c); + await settle(); + assert.equal(c.state.items.length, queue.length); + assert.equal(c.state.nextCursor, null); +}); + +test('a deep refresh whose next page meets a restore reads again from the first page', async () => { + let restored = false; + let queue = many(60); + const { c, h } = await loaded({ [`GET ${ITEMS}`]: pager(() => queue, () => 0, { + onPage: cursor => { + if (cursor && restored) { restored = false; return fail(409, 'storage-changed', 'The library was restored.'); } + return null; + }, + }) }); + await c.loadMore(); + const before = h.of('GET', `${ITEMS}?`).length; + restored = true; + queue = many(60, 'r'); + await refreshOf(c); + await settle(); + assert.deepEqual(h.of('GET', `${ITEMS}?`).slice(before).map(call => call.url), + [`${ITEMS}?limit=40`, `${ITEMS}?limit=40&cursor=c40`, `${ITEMS}?limit=40`, `${ITEMS}?limit=40&cursor=c40`]); + assert.equal(c.state.items.length, 60); + assert.equal(c.state.items[0].id, 'r000'); + assert.notEqual(c.state.warning, 'The review queue could not be read again. Reload the page to see the latest images.', + 'a restore met mid-read is not a failed re-read'); + assert.equal(c.state.status, 'ready'); +}); + +test('ready naming the version the first read answered reads nothing more', async () => { + const { c, h } = await loaded({ [`GET ${ITEMS}`]: pager(() => QUEUE, () => 4) }); + await syncOf(c, version(4)); + await settle(); + assert.equal(h.of('GET', `${ITEMS}?`).length, 1, 'one list read per visit'); + assert.equal(h.of('GET', `${ITEMS}/one`).length, 1); +}); + +test('ready naming a newer version, another epoch or no version re-reads once', async () => { + const { c, h } = await loaded({ [`GET ${ITEMS}`]: pager(() => QUEUE, () => 4) }); + await syncOf(c, version(5)); + await settle(); + assert.equal(h.of('GET', `${ITEMS}?`).length, 2); + await syncOf(c, { library_epoch: '22222222-2222-4222-8222-222222222222', review_revision: 4 }); + await settle(); + assert.equal(h.of('GET', `${ITEMS}?`).length, 3); + await syncOf(c, {}); + await settle(); + assert.equal(h.of('GET', `${ITEMS}?`).length, 4, 'an unknown version is read, as before versions existed'); +}); + +test('ready heard during the first read re-reads only when that read is older', async () => { + const { createReviewController } = await load(); + for (const [readRevision, readyRevision, reads] of [[4, 4, 1], [5, 4, 1], [4, 5, 2]]) { + let release; + let lists = 0; + const answer = pager(() => QUEUE, () => (lists === 1 ? readRevision : readyRevision)); + const h = harness(routes({ [`GET ${ITEMS}`]: async call => { + lists += 1; + if (lists === 1) await new Promise(resolve => { release = resolve; }); + return answer(call); + } })); + const c = createReviewController({ api: h.api }); + const first = c.load(); + await settle(); + syncOf(c, version(readyRevision)); + release(); + await first; + await settle(); + assert.equal(lists, reads, `first read r${readRevision}, ready r${readyRevision}`); + c.dispose(); + } +}); + +test('a change heard during the first read still re-reads after a current ready', async () => { + const { createReviewController } = await load(); + let release; + let lists = 0; + const h = harness(routes({ [`GET ${ITEMS}`]: async call => { + lists += 1; + if (lists === 1) await new Promise(resolve => { release = resolve; }); + return pager(() => QUEUE, () => 4)(call); + } })); + const c = createReviewController({ api: h.api }); + const first = c.load(); + await settle(); + syncOf(c, version(4)); + refreshOf(c); + release(); + await first; + await settle(); + assert.equal(lists, 2); +}); diff --git a/tests/test_api_review_events.py b/tests/test_api_review_events.py index 08e6d3a..dba9818 100644 --- a/tests/test_api_review_events.py +++ b/tests/test_api_review_events.py @@ -9,6 +9,7 @@ decision and a duplicate recording change nothing and send nothing. """ import asyncio +import json import sqlite3 import threading @@ -51,15 +52,29 @@ class ReviewEventsTestCase(ReviewRoutesTestCase): async def _event(self, resp): """The next event's name (keepalive comments are skipped).""" - name = None + return (await self._event_with_data(resp))[0] + + async def _event_with_data(self, resp): + """The next event's name and its parsed ``data`` line.""" + name, data = None, None while True: line = await asyncio.wait_for(resp.content.readline(), READ_SECONDS) self.assertTrue(line, "the stream ended") text = line.decode("utf-8").rstrip("\n") if text.startswith("event: "): name = text[len("event: ") :] + elif text.startswith("data: "): + data = json.loads(text[len("data: ") :]) elif text == "" and name is not None: - return name + return name, data + + async def _ready_version(self): + resp = await self.client.get(EVENTS_URL) + self.assertEqual(resp.status, 200) + self.addAsyncCleanup(self._close, resp) + name, data = await self._event_with_data(resp) + self.assertEqual(name, "ready") + return data async def _record_in_thread(self, path): """Record as the monitor does: from a worker thread, not the loop.""" @@ -125,6 +140,70 @@ class TestReviewEvents(ReviewEventsTestCase): self.assertEqual(job["scope"]["review_items"], 1) self.assertEqual(await self._event(stream), "change") + async def test_ready_carries_the_version_a_list_read_answers(self): + # F3 (no double read per visit): the page skips the re-read on + # ``ready`` when its first read already answered this version. + self._require_events() + self._record(self._ambiguous()) + page = await self._list() + self.assertIsInstance(page.get("review_revision"), int, page) + version = await self._ready_version() + self.assertEqual( + version, + { + "library_epoch": page["library_epoch"], + "review_revision": page["review_revision"], + }, + ) + + async def test_a_queue_change_moves_the_version_ready_carries(self): + self._require_events() + before = await self._list() + self._record(self._ambiguous()) + after = await self._list() + self.assertIn("review_revision", after) + self.assertGreater(after["review_revision"], before["review_revision"]) + version = await self._ready_version() + self.assertEqual(version["review_revision"], after["review_revision"]) + self.assertNotEqual(version["review_revision"], before["review_revision"]) + + async def test_ready_names_no_version_when_it_cannot_be_read(self): + # The page then re-reads, as it did before versions existed. + self._require_events() + + def broken(): + raise sqlite3.DatabaseError("locked") + + self.api.db.review_queue_version = broken + self.assertEqual(await self._ready_version(), {}) + + async def test_a_stream_counts_as_open_once_it_has_sent_ready(self): + # The browser fixture waits on this count before it changes the + # queue, so a change after it is a ``change`` event and never part of + # the version ``ready`` named. + self._require_events() + reading, release = asyncio.Event(), asyncio.Event() + original = self.api._queue_version + + async def held(): + reading.set() + await release.wait() + return await original() + + self.api._queue_version = held + opening = asyncio.ensure_future(self._ready_version()) + await asyncio.wait_for(reading.wait(), READ_SECONDS) + self.assertEqual(self.api.review_stream_count(), 0) + release.set() + await asyncio.wait_for(opening, READ_SECONDS) + self.assertEqual(self.api.review_stream_count(), 1) + + async def test_change_events_carry_no_data(self): + self._require_events() + stream = await self._stream() + self._record(self._ambiguous()) + self.assertEqual(await self._event_with_data(stream), ("change", {})) + async def test_closing_the_stream_unsubscribes(self): self._require_events() path = self.api.db.model.db_path diff --git a/web/js/workspace/index.mjs b/web/js/workspace/index.mjs index 5b00388..3a84a9c 100644 --- a/web/js/workspace/index.mjs +++ b/web/js/workspace/index.mjs @@ -113,13 +113,21 @@ function mountWorkspace() { countRequests.set(id, request); // Each badge lands as its own read settles; a count that failed to refresh is removed, never left stale. loadNavCounts(api, request.signal, (_, count) => { - if (request.signal.aborted) return; - const link = sidebar.querySelector(`nav [data-route="${id}"]`); - link?.querySelector('.nav-count')?.remove(); - if (link && Number.isSafeInteger(count)) link.append(element('span', count.toLocaleString(), 'nav-count')); + if (!request.signal.aborted) showCount(id, count); }, [id]); } } + function showCount(id, count) { + const link = sidebar.querySelector(`nav [data-route="${id}"]`); + link?.querySelector('.nav-count')?.remove(); + if (link && Number.isSafeInteger(count)) link.append(element('span', count.toLocaleString(), 'nav-count')); + } + /** A count a page has just read itself: it replaces the badge, and an older read in flight no longer lands. */ + function setCount(id, count) { + countRequests.get(id)?.abort(); + countRequests.delete(id); + showCount(id, count); + } scope.cleanup(() => { for (const request of countRequests.values()) request.abort(); }); // A confirmed Library or vocabulary write cannot change the Images count, so only Library and Tags are re-read. scope.cleanup(onLibraryWrite(() => refreshCounts(LIBRARY_WRITE_COUNTS))); @@ -251,9 +259,10 @@ function mountWorkspace() { return; } if (route.id === 'review') { - // Needs review (ruling R5): a recorded decision, and a change the open page hears of, re-read the row's count. + // Needs review (ruling R5): a recorded decision re-reads the row's count, and the open page's total is the + // count (F3: a change the page hears of costs no count read). mountReview(main, { api, scope: view, onDecision: () => refreshCounts(REVIEW_WRITE_COUNTS), - onQueueChange: () => refreshCounts(REVIEW_WRITE_COUNTS) }); + onTotal: total => setCount('review', total) }); return; } if (route.id === 'inspect') { diff --git a/web/js/workspace/review-state.mjs b/web/js/workspace/review-state.mjs index 7f9fbc4..adbd22b 100644 --- a/web/js/workspace/review-state.mjs +++ b/web/js/workspace/review-state.mjs @@ -11,10 +11,14 @@ * a restored storage or a changed queue reads the queue again. A request that got no answer keeps its capture and * its exact body, and the next Apply of the same choice (`retry()`) resends that body once. * - Leave for later writes nothing. Tagging the image (the shared Tag dialog) never changes the item. - * - `refresh()` re-reads the first page quietly when the server says the queue changed (F2: the review change stream: + * - `refresh()` re-reads the queue quietly when the server says the queue changed (F2: the review change stream: * a scan or the monitor added items, another tab decided one). The open item, its choice and an unanswered decision * stay; when the open item has left the queue the next one opens with a note. Refreshes asked for while one runs - * collapse into one more read. + * collapse into one more read. Every re-read keeps the loaded depth (F3): it reads page after page until it holds as + * many items as were shown, so Load more pages stay and the page does not jump. + * - `sync(version)` is the stream's `ready`: it re-reads only when the queue read last is older than the version the + * stream names (`{library_epoch, review_revision}`, the pair every list read answers), so a visit reads its queue + * once (F3). An unknown version re-reads. */ import { noun } from './plural.mjs'; @@ -102,14 +106,18 @@ const initial = () => ({ /** * @param {{api: {get: Function, post: Function, del: Function}}} options * @returns the controller: `state`, `subscribe(listener)`, `load()`, `refresh()`, `loadMore()`, `select(id)`, `choose(value)`, - * `apply()`, `retry()`, `leave()`, `dispose()`. + * `sync(version)`, `apply()`, `retry()`, `leave()`, `dispose()`. */ export function createReviewController({ api }) { let state = Object.freeze(initial()); const listeners = new Set(); const lifetime = new AbortController(); let listRead = null, detailRead = null, pending = null, disposed = false; - let refreshing = null, refreshAgain = false, refreshAfterLoad = false; + let refreshing = null, refreshAgain = false; + // What the first read owes once it answers: nothing, a re-read ('always'), or a re-read unless it is this version. + let afterLoad = null; + // The version of the queue read last: {epoch, revision}. + let shown = null; function set(update) { if (disposed) return; @@ -127,9 +135,46 @@ export function createReviewController({ api }) { pending = null; } + const isVersion = version => typeof version?.library_epoch === 'string' && Number.isSafeInteger(version?.review_revision); + const current = version => isVersion(version) && shown !== null && shown.epoch === version.library_epoch && + shown.revision >= version.review_revision; + + /** One list page; `cursor` continues after the last item already read. */ + function readPage(read, cursor) { + const query = new URLSearchParams({ limit: String(PAGE_LIMIT) }); + if (cursor) query.set('cursor', cursor); + return api.get(`${REVIEW_URL}?${query}`, { signal: signalOf(read), cache: 'no-store' }); + } + /** - * Read the first page; `stale` re-reads the selected item's detail too (its revision or epoch moved). `quiet` (a - * refresh) keeps the page as it is while reading and after a failed read. + * The queue down to `depth` items: the first page, then the next ones while fewer than `depth` are held. A restore + * met after the first page (its cursor names the old storage) reads once more from the first page. + */ + async function readQueue(read, depth) { + for (let attempt = 0; ; attempt += 1) { + const first = await readPage(read, null); + const items = [...(first.items ?? [])]; + let { total } = first; + let next = first.next_cursor ?? null; + try { + while (next && items.length < depth) { + const page = await readPage(read, next); + const known = new Set(items.map(entry => entry.id)); + items.push(...(page.items ?? []).filter(entry => !known.has(entry.id))); + total = page.total; + next = page.next_cursor ?? null; + } + } catch (error) { + if (attempt === 0 && RELOAD.has(error?.code)) continue; + throw error; + } + return { first, items, total, next }; + } + } + + /** + * Read the queue as deep as it is shown; `stale` re-reads the selected item's detail too (its revision or epoch + * moved). `quiet` (a refresh) keeps the page as it is while reading and after a failed read. */ async function load({ keepWarning = false, stale = false, quiet = false } = {}) { listRead?.abort(); @@ -137,11 +182,14 @@ export function createReviewController({ api }) { if (!quiet) set({ status: state.items.length ? state.status : 'loading', ...(keepWarning ? {} : { warning: '' }) }); let readOk = false; try { - const page = await api.get(`${REVIEW_URL}?limit=${PAGE_LIMIT}`, { signal: signalOf(read), cache: 'no-store' }); + // One page, or as many as Load more had shown. + const queue = await readQueue(read, state.items.length > PAGE_LIMIT ? state.items.length : 0); if (read !== listRead || disposed) return; - const items = Object.freeze((page.items ?? []).map(entry => Object.freeze({ ...entry }))); + const items = Object.freeze(queue.items.map(entry => Object.freeze({ ...entry }))); const left = state.items.find(entry => entry.id === state.selectedId); - set({ status: 'ready', epoch: page.library_epoch, total: page.total, items, nextCursor: page.next_cursor ?? null, + shown = Number.isSafeInteger(queue.first.review_revision) + ? { epoch: queue.first.library_epoch, revision: queue.first.review_revision } : null; + set({ status: 'ready', epoch: queue.first.library_epoch, total: queue.total, items, nextCursor: queue.next, more: 'idle' }); readOk = true; const keep = items.find(entry => entry.id === state.selectedId); @@ -157,20 +205,34 @@ export function createReviewController({ api }) { if (quiet && state.status === 'ready') set({ warning: COPY.refreshFailed }); else set({ status: 'error' }); } finally { - // A change heard while the first read ran: that read may predate it, so read once more. - if (readOk && refreshAfterLoad && !disposed) { - refreshAfterLoad = false; - refresh(); + // A change heard while the first read ran: that read may predate it, so read once more (unless the + // stream's ready named a version this read already holds). + const owed = afterLoad; + if (readOk && owed && !disposed) { + afterLoad = null; + if (owed === 'always' || !current(owed)) refresh(); } } } + /** The stream's `ready`: re-read only when the queue shown is older than `version`. */ + function sync(version) { + if (disposed) return Promise.resolve(); + if (state.status === 'loading') { + // A second version, or a change, while the first read runs: no longer one version to compare. + afterLoad = afterLoad === null && isVersion(version) ? version : 'always'; + return Promise.resolve(); + } + if (state.status === 'ready' && current(version)) return Promise.resolve(); + return refresh(); + } + /** Re-read the queue after a change elsewhere; calls made while one runs collapse into one more read. */ function refresh() { if (disposed) return Promise.resolve(); // Before the queue has been read there is nothing to keep: the first read (or Retry) reads the latest. if (state.status !== 'ready') { - if (state.status === 'loading') refreshAfterLoad = true; + if (state.status === 'loading') afterLoad = 'always'; return Promise.resolve(); } if (refreshing) { @@ -311,7 +373,7 @@ export function createReviewController({ api }) { listeners.add(listener); return () => listeners.delete(listener); }, - load: () => load(), refresh, loadMore, select: id => select(id), choose, apply, retry, leave, + load: () => load(), refresh, sync, loadMore, select: id => select(id), choose, apply, retry, leave, dispose() { if (disposed) return; dropPending(); diff --git a/web/js/workspace/review.mjs b/web/js/workspace/review.mjs index 70bba59..c15dca1 100644 --- a/web/js/workspace/review.mjs +++ b/web/js/workspace/review.mjs @@ -13,7 +13,9 @@ * are text: `p.warning-note` in the panel, or the page's status line (D119). * - The open page re-reads its queue on each `change` of the review change stream (`/prompt_manager/review/events`, * F2): items a scan or the monitor added, and decisions made in other tabs. Each `ready` (the first one, and one - * after the browser reconnects a dropped stream) re-reads too, since a change may have come before it. Nothing polls. + * after the browser reconnects a dropped stream) names the queue's version, and the page re-reads only when the + * queue it shows is older (F3: one read per visit). A re-read keeps the depth Load more had reached. Nothing polls. + * - The SAVED VIEWS row's count follows the total the page shows (`onTotal`), so a change costs no count read. * - D119: the queue cards carry no selection checkbox and no star (this page has no selection bar, and the queue * carries no favorite state); more than 40 items add the Images page's `.infinite-continuation` Load more after the * layout; the toasts "Association accepted." and "Left unresolved." stand in for the concept's prototype toast. @@ -58,12 +60,22 @@ export function imageOf(item) { url: item.url, is_video: item.is_video === true, media_type: item.is_video ? 'video' : 'image' }; } +/** The queue version a `ready` event names, or null (the page then re-reads). */ +export function readyVersion(event) { + try { + const version = JSON.parse(event?.data ?? 'null'); + return version && typeof version === 'object' ? version : null; + } catch { + return null; + } +} + /** - * Mount the page under the shell's heading. `onDecision()` runs after a recorded decision and `onQueueChange()` after - * a change elsewhere (the shell re-reads the Needs review count); `controller` and `connect` (the change stream's - * EventSource) are injectable for tests. + * Mount the page under the shell's heading. `onDecision()` runs after a recorded decision and `onTotal(total)` each + * time the pending total the page shows changes (the shell's Needs review count); `controller` and `connect` (the + * change stream's EventSource) are injectable for tests. */ -export function mountReview(main, { api, scope, onDecision = () => {}, onQueueChange, +export function mountReview(main, { api, scope, onDecision = () => {}, onTotal = () => {}, connect = openEvents, controller = createReviewController({ api }) }) { scope.cleanup(() => controller.dispose()); const heading = pageHeading(main); @@ -248,17 +260,19 @@ export function mountReview(main, { api, scope, onDecision = () => {}, onQueueCh controller.select(item.id); }); + let shownTotal = null; + const reportTotal = state => { + if (state.status !== 'ready' || state.phase === 'done' || state.total === shownTotal) return; + shownTotal = state.total; + onTotal(state.total); + }; scope.cleanup(controller.subscribe(render)); + scope.cleanup(controller.subscribe(reportTotal)); render(controller.state); controller.load(); const events = connect(REVIEW_EVENTS_URL); scope.cleanup(() => events.close()); - const changed = () => { - if (scope.signal.aborted) return; - controller.refresh(); - onQueueChange?.(); - }; - events.addEventListener('ready', changed); - events.addEventListener('change', changed); + events.addEventListener('ready', event => { if (!scope.signal.aborted) controller.sync(readyVersion(event)); }); + events.addEventListener('change', () => { if (!scope.signal.aborted) controller.refresh(); }); return { controller }; }