feat(review): bind queue cursors to the storage epoch only

A decision or a new item (the monitor records one per generation) no
longer makes an open cursor stale: the (created_at, id) keyset stays
ordered when items leave or are appended. A restore still answers
storage-changed and an unverifiable cursor review-conflict.
This commit is contained in:
Vito Sansevero
2026-10-22 18:12:26 -07:00
parent 7f1b5b64bd
commit f00b0232a0
3 changed files with 56 additions and 29 deletions
+9 -6
View File
@@ -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
+10 -12
View File
@@ -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(
{
+37 -11
View File
@@ -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