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 }; }