diff --git a/TESTING.md b/TESTING.md index 354e648..00a35b4 100644 --- a/TESTING.md +++ b/TESTING.md @@ -1814,12 +1814,15 @@ Needs review decisions and queue reads (#249, `py/api/review.py`, `tests/test_api_review.py`): `GET /prompt_manager/review/items?limit=1..40&cursor=` answers `{success, library_epoch, total, items, next_cursor}` with pending items oldest first; `total` is the exact pending count. The cursor is an HMAC keyset -on `(created_at, id)` bound to the storage epoch and `review_index_revision`: a -malformed one is 400, a well-formed one this process cannot verify (tampered, -or opened before a ComfyUI restart: the secret is per process) is 409 -`review-conflict` with the current `total`, one from before a restore is 409 -`storage-changed`, one from before any queue change (a decision, a new item, a -refreshed spelling) is 409 `review-conflict`. +on `(created_at, id)` bound to the storage epoch only (ruling M3, so "load more" +keeps working while the monitor adds an item per generation): a malformed one +is 400, a well-formed one this process cannot verify (tampered, or opened +before a ComfyUI restart: the secret is per process) is 409 `review-conflict` +with the current `total`, and one from before a restore is 409 +`storage-changed`. A decision or a new item leaves an open cursor valid: a +decided item drops out of later pages and new items (stamped `utc_now_iso()`, +microseconds) append after it. Residual risk: an item recorded after a wall +clock step backwards sorts before an open cursor and shows only on reload. Targets are formatted from `spelled_path` against the current roots (the roots' real paths are the only file access); a file under no current root is `unavailable` with a null target. Every read and decision lists the roots and diff --git a/py/api/review.py b/py/api/review.py index 2d3e425..414bac0 100644 --- a/py/api/review.py +++ b/py/api/review.py @@ -1,8 +1,9 @@ """Needs review routes (#249 E3): the pending queue, item detail, decisions. ``GET /prompt_manager/review/items`` lists pending items oldest first behind -an HMAC keyset cursor bound to the storage epoch and the review index -revision (any decision, new item or restore makes an old cursor stale); +an HMAC keyset cursor bound to the storage epoch only (a restore makes an +old cursor stale; a decision or a new item does not, because the +``(created_at, id)`` keyset stays ordered when items leave or are appended); ``total`` is the exact pending count. ``GET .../items/{id}`` adds the candidates and the dependency fingerprint. ``POST .../items/{id}/decision`` records one explicit decision: Apply links the chosen sampler's pair only @@ -153,9 +154,9 @@ class _StaleCursor(ValueError): """A well-formed cursor this process cannot verify (e.g. after a restart).""" -def _encode_cursor(epoch: str, revision: int, key: tuple) -> str: +def _encode_cursor(epoch: str, key: tuple) -> str: payload = json.dumps( - {"v": 1, "e": epoch, "r": revision, "k": list(key)}, + {"v": 2, "e": epoch, "k": list(key)}, separators=(",", ":"), ensure_ascii=True, ).encode() @@ -164,7 +165,7 @@ def _encode_cursor(epoch: str, revision: int, key: tuple) -> str: def _decode_cursor(token: str) -> tuple: - """``(epoch, revision, (created_at, id))`` of a cursor. + """``(epoch, (created_at, id))`` of a cursor. ``ValueError`` when malformed; :class:`_StaleCursor` when well-formed but signed by another secret (the per-process secret changes on every start, @@ -185,15 +186,14 @@ def _decode_cursor(token: str) -> tuple: data = json.loads(payload) key = data["k"] if ( - data["v"] != 1 + data["v"] != 2 or not isinstance(data["e"], str) - or type(data["r"]) is not int or not isinstance(key, list) or len(key) != 2 or not all(isinstance(part, str) for part in key) ): raise ValueError() - return data["e"], data["r"], tuple(key) + return data["e"], tuple(key) except _StaleCursor: raise except (ValueError, KeyError, TypeError, UnicodeError, binascii.Error): @@ -299,21 +299,19 @@ class ReviewRoutesMixin: return bad_request(str(exc)) try: page, views = await self._run_in_executor( - self._review_page, limit, bound[2] if bound else None + self._review_page, limit, bound[1] if bound else None ) except Exception: self.logger.exception("Review queue read failed") return _failed() if bound is not None and bound[0] != page["epoch"]: return _coded("storage-changed", library_epoch=page["epoch"]) - if bound is not None and bound[1] != page["revision"]: - return _coded("review-conflict", total=page["total"]) items = page["items"] next_cursor = None if page["more"] and items: last = items[-1] next_cursor = _encode_cursor( - page["epoch"], page["revision"], (last["created_at"], last["id"]) + page["epoch"], (last["created_at"], last["id"]) ) return web.json_response( { diff --git a/tests/test_api_review.py b/tests/test_api_review.py index 1232041..7a74f62 100644 --- a/tests/test_api_review.py +++ b/tests/test_api_review.py @@ -1,12 +1,11 @@ """Needs review decisions and queue reads (#249 E3). The queue lists pending items oldest first behind an HMAC cursor bound to the -storage epoch and the review index revision; ``total`` is the exact pending -count. A decision is explicit: Apply links the chosen sampler's pair only -under the ingester's compatible rules and only for the current, re-hashed -bytes of the reviewed file, in one transaction with its decision row; Leave -unresolved records the decision and links nothing. Responses never carry a -server path. +storage epoch; ``total`` is the exact pending count. A decision is explicit: +Apply links the chosen sampler's pair only under the ingester's compatible +rules and only for the current, re-hashed bytes of the reviewed file, in one +transaction with its decision row; Leave unresolved records the decision and +links nothing. Responses never carry a server path. """ import json @@ -222,7 +221,7 @@ class TestQueueReads(ReviewRoutesTestCase): status, data = await self._json("GET", ITEMS_URL + "?cursor=" + "x" * 5000) self.assertEqual(status, 400) - async def test_cursor_invalidated_by_decision_and_restore(self): + async def test_cursor_survives_queue_changes_and_is_invalidated_by_restore(self): self._require_routes() for index in range(3): self._record(self._ambiguous(f"f{index}.png", color=(index, 0, 0))) @@ -244,16 +243,43 @@ class TestQueueReads(ReviewRoutesTestCase): status, data = await self._json("GET", f"{ITEMS_URL}?cursor={cursor}") self.assertEqual((status, data["code"]), (409, "review-conflict"), data) self.assertEqual(data["library_epoch"], first["library_epoch"]) - # A decision moves the review index revision: the old cursor is stale. + # The cursor is bound to the storage epoch only (M3): a decision and a + # new item (the monitor records one per generation) leave "load more" + # working, because the (created_at, id) keyset is stable under both. item, detail, observed = await self._open() + self.assertEqual(item["id"], first["items"][0]["id"]) status, data = await self._decide( item["id"], {**self._body(detail, observed, "unresolved")} ) self.assertEqual(status, 200, data) + self._record(self._ambiguous("late.png", color=(9, 9, 9))) + rows = self._items() status, data = await self._json("GET", f"{ITEMS_URL}?limit=1&cursor={cursor}") - self.assertEqual(status, 409, data) - self.assertEqual(data["code"], "review-conflict") - self.assertEqual(data["total"], 2) + self.assertEqual(status, 200, data) + self.assertEqual(data["total"], 3) + self.assertEqual([i["filename"] for i in data["items"]], ["f1.png"]) + seen = ["f1.png"] + while data["next_cursor"]: + data = await self._list(f"?limit=1&cursor={data['next_cursor']}") + seen.extend(i["filename"] for i in data["items"]) + self.assertEqual(seen, ["f1.png", "f2.png", "late.png"]) + self.assertEqual(len(rows), 4) + # A decision on an item after the cursor drops it from later pages. + second = await self._list("?limit=1") + self.assertEqual([i["filename"] for i in second["items"]], ["f1.png"]) + conn = self.api.db.model.get_connection() + with conn: + conn.execute( + "UPDATE media_review_items SET state='selected', revision=1 " + "WHERE id=(SELECT i.id FROM media_review_items i " + "WHERE i.spelled_path LIKE ?)", + ("%f2.png",), + ) + data = await self._list(f"?limit=1&cursor={second['next_cursor']}") + self.assertEqual( + ([i["filename"] for i in data["items"]], data["total"]), + (["late.png"], 2), + ) fresh = (await self._list("?limit=1"))["next_cursor"] conn = self.api.db.model.get_connection() with conn: # a restore replaces the storage epoch