From 16d7708ee1f5273a8ec87e7f11d8cf55bcfcea1b Mon Sep 17 00:00:00 2001 From: TheMeinerLP Date: Sun, 23 Aug 2026 15:46:02 +0200 Subject: [PATCH] feat(api): let an administrator say what the queue should do first The claim now reads `ORDER BY priority, id`, and two endpoints write that column: a drag expressed relative to a neighbour, and two quick actions that reorder a guild's whole queue by a named rule. The unit is the session, never the job. The rows are one per speaker, so a request that took job ids could reorder four of a meeting's five speakers -- a queue that is half moved, that no page renders and that nobody would notice. A session's number goes onto every one of its outstanding jobs at once. The console never sends an integer. A drag says "before that one" and the server works out the numbers, inside the same locked transaction that writes them, so two administrators dragging at once serialise into one coherent order rather than a blend of two. A reorder only ever raises a priority. Priority is one column shared by every guild and this control is reachable by any guild administrator, so a write that could lower one would be a way for the first administrator to find it to put their whole guild ahead of everybody else's ordinary work. Going first is expressed as everything that was ahead going second. Migration 0013 said `ORDER BY priority, id` would be one forward scan of `ix_job_claim_order`. Measured against PostgreSQL 17: it is not, because `status` leads that index and a claim matches two values of it. The ordering is correct and costs what ordering by `id` already cost; the index narrows the scan but does not order it. `claim_statement` records the plan, the partial index that would fix it, and why the cheaper-looking fix must not be used. --- docs/operations.md | 174 ++++++++ src/sturnus/application/priorities.py | 269 +++++++++++++ src/sturnus/console/adapters.py | 110 +++++ src/sturnus/console/ports.py | 50 +++ src/sturnus/console/routes_queue.py | 309 +++++++++++++- src/sturnus/infrastructure/db/models.py | 39 +- src/sturnus/infrastructure/db/priority.py | 249 ++++++++++++ src/sturnus/infrastructure/db/queue.py | 122 ++++-- src/sturnus/infrastructure/db/requeue.py | 36 +- src/sturnus/observability/events.py | 12 + src/sturnus/observability/fields.py | 11 + tests/application/test_priorities.py | 285 +++++++++++++ tests/console/conftest.py | 26 +- tests/console/test_queue_overview.py | 124 ++++++ tests/console/test_queue_placement.py | 187 +++++++++ tests/console/test_queue_priority_routes.py | 387 ++++++++++++++++++ tests/console/test_queue_routes.py | 32 ++ tests/console/test_queue_stream_routes.py | 27 ++ tests/infrastructure/test_priority.py | 425 ++++++++++++++++++++ tests/infrastructure/test_queue.py | 139 ++++++- 20 files changed, 2967 insertions(+), 46 deletions(-) create mode 100644 src/sturnus/application/priorities.py create mode 100644 src/sturnus/infrastructure/db/priority.py create mode 100644 tests/application/test_priorities.py create mode 100644 tests/console/test_queue_placement.py create mode 100644 tests/console/test_queue_priority_routes.py create mode 100644 tests/infrastructure/test_priority.py diff --git a/docs/operations.md b/docs/operations.md index 0dd539a..3c5a41f 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -1672,6 +1672,10 @@ There is no re-queue control on this page. Each row links to of whether a re-queue is safe stays in one place (`plan_requeue`) rather than being made twice. +Each row also carries a **priority** — where that session sits in the +queue, and what an administrator can drag it to. §6.2.11 is what that +means and what it does not promise. + ### 6.2.8 The report **Admin View → Reporting**, per guild: how often this guild meets, how @@ -1910,6 +1914,176 @@ unvalidated `yes` would silently mean the opposite of what was typed It takes effect **immediately**: the API reads it per request, and nothing caches it. +### 6.2.11 Priority: saying what the queue should do first + +`transcription_job.priority` is one integer per job. **Lower runs first.** +Zero is ordinary and is what every job carries unless somebody has said +otherwise, so a deployment where nobody has ever touched this claims work +in exactly the order it always did: oldest meeting first. + +The claim reads `ORDER BY priority, id`. `id` still breaks every tie, so +first-in-first-out survives inside a priority — two sessions held back +equally still run in the order they ended. + +#### It is a hint, not a promise about the clock + +Priority says what a worker takes **next when it next asks**. It says +nothing about when that will be, and it is not a queue-jump: + +- A job already `running` is not interrupted. Raising a session's priority + while forty tracks are being transcribed changes what happens after + those forty finish, not what happens to them. +- `max_parallel_tracks` still applies (§4.2). A session at the front of + the queue still takes at most that many workers at once, and the ones it + leaves free go to whatever is behind it. +- A worker that is not running claims nothing at any priority. If nothing + is moving, priority will not start it; §5 is where to look. + +So "I moved it to the top and it did not start" is usually one of those +three and almost never this feature failing. + +#### A reorder holds sessions back; it never pushes one forward + +This is the part worth understanding before using it, because it is +visible in the numbers. + +`priority` is one column shared by every guild in the deployment, and any +guild administrator can reorder their own guild's queue from the console. +If a reorder could write a *smaller* number, the first administrator to +find the control could put their whole guild in front of everybody else's +ordinary work, permanently, and nobody else would be told. So the console +never writes one. Going first is expressed as everything that was ahead +going second. + +Concretely: a guild whose queue is entirely ordinary (`0`) drags its +retrospective to the top. The retrospective stays at `0`; the sessions it +overtook move to `1`. The guild's own order is now what the administrator +asked for — and the sessions at `1` now sort behind every untouched job in +the deployment, not merely behind that guild's own. That cost is real and +it is the price of one shared column. It is bounded: only the sessions +actually overtaken move, and a session leaves the queue when its work +finishes. + +Two consequences follow that look surprising and are not bugs: + +- **Numbers drift upwards** as a queue is reordered repeatedly. Nothing + resets them, and nothing needs to: sessions leave the queue when they + are done, and new ones arrive at `0`. +- **A meeting recorded after a reorder starts ahead of the held-back + ones**, because it is enqueued at the ordinary `0`. That is correct — + new work is ordinary, and the held-back sessions were held back + deliberately — but it means an order is a statement about the queue at + the moment it was made, not a standing instruction. + +An operator who genuinely wants a session ahead of *another guild's* work +can write a negative priority with SQL, exactly as §4.1 describes for +`guild_config`. The API cannot, on purpose. + +#### The two quick actions + +Beside the drag-and-drop list there are two buttons, and each reorders the +guild's **whole** outstanding queue: + +- **Most people first** — by the number of rows in `session_participant`. + A meeting of eight is eight speakers waiting on one document; a + one-person recording is one. +- **Shortest recording first** — by `transcription_job.audio_seconds`, + summed across the session's tracks. + +The second has a limitation that is worth knowing rather than +rediscovering: **`audio_seconds` is written when a job completes.** A +session that has never been transcribed has none, and null is not zero — +an unmeasured recording is *not* treated as a nought-second one and is +never promoted to the front on the strength of nothing being known about +it. Unmeasured sessions sort after every measured one and keep the order +they already had. In practice that means "shortest recording first" is +useful on a queue of **re-queued** sessions, which keep the measurements +their first pass produced, and does very little on a queue of fresh ones. + +Applying either action twice writes nothing the second time. + +#### Two administrators at once + +Each request is decided inside the same database transaction that writes +it, holding a lock on the guild's outstanding jobs, so two reorders of one +guild serialise. The second is applied to what the first left, never to +the list its browser was showing. Nothing is lost and the result is never +a blend of two orders — but the second administrator may well end up with +an order they did not picture, because the queue moved under them. Their +page is told immediately: priority is part of the queue snapshot the +`/stream` endpoint watches, so a reorder anywhere reaches every open queue +page within one poll. + +A reorder that names a session which has left the queue meanwhile — it +finished while the page was open — is refused, and the refusal carries the +queue as it now is so the page can redraw. + +#### Reading and setting it over the API + +`GET /api/guilds/{guild_id}/queue` (and its `/stream` twin) reports each +session's `priority`. It is **null**, not zero, for a session with nothing +outstanding — one that is still recording, or one that is listed only +because a job of it died. Zero is a real place in the queue; null is a row +with nothing to reorder. + +Both writes take an administrator of the guild and answer 404 to everybody +else, indistinguishably from a session that does not exist. + +``` +POST /api/sessions/{session_id}/queue/priority +{"place": "before", "session": "512"} +``` + +`place` is `first`, `last`, `before` or `after`; `session` names the +neighbour and is required by the last two and refused by the first two. +**No number is ever sent.** A drag produces "this one goes here", and +turning that into integers is the server's job — a client computing them +would be computing them from a copy of the queue that is already out of +date. + +``` +POST /api/guilds/{guild_id}/queue/priority +{"rule": "many-participants-first"} +``` + +`rule` is `many-participants-first` or `short-recordings-first`. An +unrecognised name is a 400 that lists the ones there are, rather than a +different rule being run quietly. + +Both answer with the guild's whole queue in claim order: + +```json +{ + "accepted": true, + "refusal": null, + "changed": ["512"], + "order": [ + {"session_id": "77", "priority": 0}, + {"session_id": "512", "priority": 1} + ] +} +``` + +`changed` is empty when the order asked for was already the order in +force. A stale reorder answers **409** with `accepted: false`, a `refusal` +and the same `order` field, which is what the page redraws from. + +#### One thing that is *not* true, despite what migration 0013 says + +That migration added `ix_job_claim_order (status, priority, id)` and said +the claim's `ORDER BY priority, id` would be one forward scan of it. It is +not. `status` leads that index and the claim matches two values of it — a +`pending` job and a `running` one whose lease expired — so PostgreSQL +sorts. The index still narrows the scan to outstanding work, which is what +keeps a claim off the table's history, but the ordering is not free. + +Nothing is wrong operationally: the sort is over the queue's depth and the +claim cost the same before priority existed. It is recorded here so that +nobody sizes anything on the migration's paragraph. +`sturnus.infrastructure.db.queue.claim_statement` carries the measurement, +the partial index that would fix it, and why the obvious cheaper fix +(ordering by `status` first) must not be used. + ### 6.3 Listening to a recording by hand Every automated check this system has can describe a track — its level, diff --git a/src/sturnus/application/priorities.py b/src/sturnus/application/priorities.py new file mode 100644 index 0000000..e4ca9ba --- /dev/null +++ b/src/sturnus/application/priorities.py @@ -0,0 +1,269 @@ +"""What order a guild's queued sessions should run in, as arithmetic. + +`transcription_job.priority` is one integer per row, **lower first**, and +`JobQueue.claim` reads `ORDER BY priority, id`. This module holds the two +halves of turning a human's intent into those integers: what order was +asked for, and which numbers express it. Neither half touches a database, +so both are tested against plain values and there is exactly one +definition of each rule -- the same arrangement +`sturnus.application.requeue.plan_requeue` and +`sturnus.application.retention.expired_jobs` are in, and for the same +reason. + +**A session, never a job, is the unit.** The rows are one per speaker, but +nobody in a console drags a speaker: they drag a meeting. A request that +took job ids would let a caller reorder four of a meeting's five speakers +and leave the fifth behind whatever it was behind -- a queue that is +half-reordered in a way no page renders and nobody could see. So a +session's jobs carry one priority between them, this module reasons in +sessions, and the write applies a session's number to every one of its +jobs at once. + +**A reorder only ever holds sessions back; it never moves one forward.** +`priorities_for` may raise a number and may leave one alone. It may not +lower one, and that is a safety property rather than an implementation +detail. The queue is shared by every guild in the deployment, `0` is what +untouched work carries, and this arithmetic is reachable by any guild +administrator through the console -- so a function that could write a +smaller number would be a control by which the first administrator to +find it puts their whole guild in front of everybody else's ordinary +work, permanently and without anyone being told. Going first is therefore +expressed as everything that was ahead going second, which says the same +thing about the guild's own queue and says nothing at all about anybody +else's. + +The cost of that choice, stated plainly because it is real: holding a +session back holds it back *globally*, not merely within its guild. A +session at `1` sorts behind every untouched job in the deployment, not +only behind its own guild's. See `docs/operations.md` section 6.2.11. + +**Idempotence.** `priorities_for` returns only the sessions whose number +must change, so re-sending an order that already holds writes nothing. +Two administrators who agree, a quick action applied twice, and a page +re-sending what it is already showing all cost one read and no writes -- +which matters more than it looks, because the alternative is a queue that +drifts further back every time somebody looks at it. + +**Null is not zero.** `short_recordings_first` reads +`transcription_job.audio_seconds`, which is null for a recording nothing +has ever measured -- every session that has not been transcribed yet, and +every job that predates the column. Read as nought, an unmeasured session +would be the shortest recording in the queue and would be promoted to the +front on the strength of nothing being known about it. Unmeasured +sessions therefore rank *after* every measured one and keep the order they +already had, which is the same distinction `sturnus.console.statistics` +refuses to lose. +""" + +from __future__ import annotations + +from collections.abc import Callable, Iterable, Sequence +from dataclasses import dataclass + +#: How a session may be placed relative to the queue it is in. Exactly the +#: four things a drag-and-drop list can produce, and deliberately not an +#: index: an index is an absolute claim about a list the browser was +#: showing a moment ago, and two administrators dragging at once would +#: each be numbering a different list. An anchor still means something +#: after somebody else's move landed. +PLACEMENTS: tuple[str, ...] = ("first", "last", "before", "after") + +#: The placements that name another session to sit beside. +_ANCHORED: tuple[str, ...] = ("before", "after") + + +@dataclass(frozen=True) +class QueuedSession: + """One session with transcription work outstanding, as the queue sees it. + + `priority` is the number its outstanding jobs carry; `participants` is + how many people `session_participant` recorded; `audio_seconds` is how + much audio anything has measured for it, and is `None` when nothing + has -- see the module docstring on why that is not zero. + """ + + id: int + priority: int + participants: int + audio_seconds: float | None + + +@dataclass(frozen=True) +class Placement: + """Where a dragged session was dropped. + + `anchor` is required by `before` and `after` and meaningless to the + other two; `is_valid` is what a boundary checks before building one, + so a malformed request is refused where it can still be explained. + """ + + where: str + anchor: int | None = None + + @property + def is_valid(self) -> bool: + if self.where not in PLACEMENTS: + return False + return (self.anchor is not None) == (self.where in _ANCHORED) + + +class UnknownPriorityRule(Exception): + """A quick action nobody has. Names what was asked for and what there is. + + The same shape `sturnus.domain.transcription_models.UnknownTranscriptionModel` + has, and for the same reason: the registry is closed and its names are + literals of this repository, so a caller who mistyped one can be shown + the whole list without anything being disclosed. + """ + + +#: The sort key a quick action ranks by. A rule is *only* a key function: +#: adding the third one the owner will ask for next month is one function +#: and one line in `KNOWN_RULES`, and it cannot get the tie-breaking, the +#: integers or the write wrong because it does none of them. +Rule = Callable[[QueuedSession], tuple[object, ...]] + + +def many_participants_first(session: QueuedSession) -> tuple[object, ...]: + """Biggest meeting first. + + A meeting of eight people is eight jobs and eight speakers waiting on + one document; a one-person recording is one. Negated rather than + reverse-sorted so that every rule in this module sorts ascending and + ties keep the order they arrived in. + """ + return (-session.participants,) + + +def short_recordings_first(session: QueuedSession) -> tuple[object, ...]: + """Least audio first, and unmeasured audio last. + + The leading flag is the whole of the null rule: `False` sorts before + `True`, so everything with a measurement is ranked before everything + without one, and the unmeasured sessions then keep the order they had + among themselves rather than being ordered by a zero nobody measured. + """ + return (session.audio_seconds is None, session.audio_seconds or 0.0) + + +#: Every quick action, by the name a request may use. Fixed literals of +#: this repository, which is what makes an unknown one safe to echo back. +KNOWN_RULES: dict[str, Rule] = { + "many-participants-first": many_participants_first, + "short-recordings-first": short_recordings_first, +} + + +def resolve_rule(name: str) -> Rule: + """The rule that name means, or `UnknownPriorityRule`. + + Refused rather than fallen back on. A quick action whose name was not + understood and which silently ran a different one would reorder a + guild's queue by a rule nobody asked for, and the administrator would + have no way to tell that from the rule working. + """ + rule = KNOWN_RULES.get(name) + if rule is None: + known = ", ".join(sorted(KNOWN_RULES)) + raise UnknownPriorityRule(f"no such queue rule: {name!r}; known rules are {known}") + return rule + + +def claim_order(sessions: Iterable[QueuedSession]) -> tuple[QueuedSession, ...]: + """The sessions in the order `JobQueue.claim` would reach them. + + `(priority, id)` ascending, which is the claim's `ORDER BY priority, + id` said in Python. Every other function here is defined in terms of + this one, so "where a session is now" has a single definition and the + console is shown the same order the worker will actually work in. + """ + return tuple(sorted(sessions, key=lambda session: (session.priority, session.id))) + + +def order_with( + sessions: Sequence[QueuedSession], session_id: int, placement: Placement +) -> tuple[int, ...] | None: + """The order this queue has after one session is dropped somewhere. + + `None` when the dragged session or the session it was dropped beside + is no longer in this queue -- documented while the page was open, or + never in it. That is a refusal for the caller to report along with the + queue as it now stands, never a silent no-op: a drag that appeared to + work and changed nothing is the failure this endpoint exists to make + impossible. + """ + order = [session.id for session in claim_order(sessions)] + if session_id not in order: + return None + if placement.anchor is not None and placement.anchor not in order: + return None + order.remove(session_id) + if placement.where == "first": + order.insert(0, session_id) + elif placement.where == "last": + order.append(session_id) + elif placement.where == "before": + order.insert(order.index(_anchor(placement)), session_id) + else: + order.insert(order.index(_anchor(placement)) + 1, session_id) + return tuple(order) + + +def _anchor(placement: Placement) -> int: + assert placement.anchor is not None, "an anchored placement without an anchor" + return placement.anchor + + +def order_by_rule(sessions: Sequence[QueuedSession], rule: Rule) -> tuple[int, ...]: + """The order a quick action asks for. + + Sorted over `claim_order` rather than over the argument, and with a + stable sort: two sessions a rule has no opinion between keep the order + the queue already had them in. A quick action reorders what it has + something to say about and leaves the rest exactly where it was, which + is what makes applying one twice a no-op. + """ + return tuple(session.id for session in sorted(claim_order(sessions), key=rule)) + + +def priorities_for(sessions: Iterable[QueuedSession], order: Sequence[int]) -> dict[int, int]: + """The numbers to write, for the sessions whose numbers must change. + + One forward pass over the wanted order, carrying the sort key + `(priority, id)` of the session before. A session already sorting + after its predecessor keeps the number it has; one that does not is + raised to the smallest number that puts it there -- the predecessor's + own number when this session's id is the larger (the claim breaks that + tie by id), and one more than it otherwise. + + Two properties follow directly and both are load-bearing. + + **Nothing is ever lowered.** The pass only ever raises, so a guild can + express any order at all over its own queue without a single job of + another guild's being overtaken. See the module docstring. + + **Nothing is written for an order that already holds.** Every session + keeps its number, the dictionary comes back empty, and the write is a + transaction that touches no rows. + + Sessions the order does not name are not returned and are not + touched. The callers all pass the guild's whole outstanding queue, so + in practice the order names everything; a caller that passed less + would be reordering a subset against neighbours it had not looked at, + which is why no caller does. + """ + by_id = {session.id: session for session in sessions} + changed: dict[int, int] = {} + previous: tuple[int, int] | None = None + for session_id in order: + session = by_id[session_id] + priority = session.priority + if previous is not None and (priority, session.id) <= previous: + # The least number that sorts after the session before this + # one: the same number when the tie falls our way on id, + # otherwise the next one up. + priority = previous[0] if session.id > previous[1] else previous[0] + 1 + if priority != session.priority: + changed[session.id] = priority + previous = (priority, session.id) + return changed diff --git a/src/sturnus/console/adapters.py b/src/sturnus/console/adapters.py index ca6d4bd..23dad90 100644 --- a/src/sturnus/console/adapters.py +++ b/src/sturnus/console/adapters.py @@ -37,6 +37,8 @@ MirroredRole, ) from sturnus.application.linking import new_state +from sturnus.application.priorities import Placement, order_by_rule, order_with, resolve_rule +from sturnus.application.priorities import QueuedSession as PrioritisedSession from sturnus.application.publishing import DOCUMENTED_STATUS from sturnus.console.auth import PROVIDER from sturnus.console.ports import ( @@ -52,6 +54,8 @@ OwnConsent, PersonRevocation, QueuedSession, + QueueOrder, + QueuePosition, QueueSnapshot, QueueSpeaker, RequeueOutcome, @@ -86,6 +90,7 @@ ) from sturnus.infrastructure.db.models import Session as SessionRow from sturnus.infrastructure.db.models import SessionDocument as SessionDocumentRow +from sturnus.infrastructure.db.priority import Decision, load_queued_sessions, reorder from sturnus.infrastructure.db.queue import DEFAULT_LEASE_SECONDS, TERMINAL_STATUSES from sturnus.infrastructure.db.repositories import ( AccountLinkRepository, @@ -705,6 +710,30 @@ async def requeue( model=model, ) + async def place( + self, session_id: int, *, requested_by: int, placement: Placement + ) -> QueueOrder | None: + """Moves one session to where an administrator dropped it. + + The same authorisation as `requeue`, expressed by the same call + and for the same reason: reordering a queue is an operation on the + system, not a use of one's own recording, so it takes an + administrator of the session's guild -- and `None` covers "no such + session" and "not yours" alike. + + The write is `sturnus.infrastructure.db.priority.reorder` and the + decision handed to it is `order_with`, unwrapped. Nothing here + decides where a session goes; this method knows who may ask. + """ + guild_id = await self._administered_guild(session_id, requested_by) + if guild_id is None: + return None + + def decide(sessions: Sequence[PrioritisedSession]) -> tuple[int, ...] | None: + return order_with(sessions, session_id, placement) + + return await apply_order(self._session_factory, guild_id, decide) + async def _administered_guild(self, session_id: int, discord_user_id: int) -> int | None: """The session's guild, if this person administers it. `None` otherwise. @@ -1636,6 +1665,38 @@ async def for_guild(self, guild_id: int, *, requested_by: int) -> GuildQueue | N truncated=truncated, ) + async def reprioritise( + self, guild_id: int, *, requested_by: int, rule: str + ) -> QueueOrder | None: + """Reorders this guild's whole queue by one named rule. + + The same authorisation as `for_guild` and expressed the same way, + because this is the same subject: a quick action is a statement + about a guild's queue, so it takes an administrator of that guild + and answers `None` for "no such guild" and "not yours" alike. + + `rule` is a name from `sturnus.application.priorities.KNOWN_RULES` + and is resolved at the HTTP boundary, where a caller who named one + nobody has can still be told so -- the same division `model` has + with `transcription_models.resolve`. Below that line an unknown + rule does not exist. + + **The rule reorders the whole queue and not a selection of it.** + "The biggest meetings first" is a statement about an ordering, and + an ordering that applied to some rows and not others would leave + the rest wherever a previous rule had put them -- an order nobody + chose and nobody could reconstruct. + """ + if not await self._admins.is_admin(guild_id, requested_by): + return None + + ranking = resolve_rule(rule) + + def decide(sessions: Sequence[PrioritisedSession]) -> tuple[int, ...] | None: + return order_by_rule(sessions, ranking) + + return await apply_order(self._session_factory, guild_id, decide) + def _queued(session: ActiveSession) -> QueuedSession: return QueuedSession( @@ -1650,6 +1711,55 @@ def _queued(session: ActiveSession) -> QueuedSession: running=session.counts.get("running", 0), done=session.counts.get("done", 0), dead=session.counts.get("dead", 0), + # Passed through including its absence. See `ActiveSession.priority` + # on why a session with nothing queued has no place rather than the + # ordinary one. + priority=session.priority, + ) + + +#: Why a reorder was refused, in one sentence. One string for both of the +#: ways it can happen -- the session finished, or the session it was +#: dropped beside did -- because they are the same news to the page that +#: asked: the list you were looking at has moved on. It is a fixed +#: literal, so a console may key off it and nothing a caller sent is ever +#: echoed back. +STALE_DRAG = "that session is no longer in this guild's queue" + + +async def apply_order( + session_factory: async_sessionmaker[AsyncSession], guild_id: int, decide: Decision +) -> QueueOrder: + """Runs one decision against one guild's queue and shapes the answer. + + Shared by the drag and by the quick actions, which differ only in the + decision they hand over -- and which therefore must not differ in what + they authorise, what they write, or what they say afterwards. Called + only after the administrator check has passed, so `None` from + `reorder` is always a refusal and never a permission problem. + + A refusal re-reads the queue rather than answering with nothing. The + drag that was refused was aimed at a list the browser was showing and + the refusal means precisely that the list is out of date, so sending + the current one back is what lets the page redraw instead of asking + again from the same stale picture. The re-read is a second moment and + may differ again by the time it arrives, which is why it is not + presented as the result of anything -- nothing was written. + """ + result = await reorder(session_factory, guild_id, decide) + if result is None: + current = await load_queued_sessions(session_factory, guild_id) + return QueueOrder( + accepted=False, refusal=STALE_DRAG, sessions=_positions(current), changed=() + ) + return QueueOrder( + accepted=True, refusal=None, sessions=_positions(result.sessions), changed=result.changed + ) + + +def _positions(sessions: Sequence[PrioritisedSession]) -> tuple[QueuePosition, ...]: + return tuple( + QueuePosition(session_id=session.id, priority=session.priority) for session in sessions ) diff --git a/src/sturnus/console/ports.py b/src/sturnus/console/ports.py index f74035e..bbfceaf 100644 --- a/src/sturnus/console/ports.py +++ b/src/sturnus/console/ports.py @@ -26,6 +26,7 @@ MirroredMember, MirroredRole, ) +from sturnus.application.priorities import Placement from sturnus.console.filters import SessionFilter from sturnus.console.reporting import RecordedSession from sturnus.console.statistics import ( @@ -728,6 +729,10 @@ async def requeue( self, session_id: int, *, requested_by: int, model: str ) -> RequeueOutcome | None: ... + async def place( + self, session_id: int, *, requested_by: int, placement: Placement + ) -> QueueOrder | None: ... + @dataclass(frozen=True) class ConsentHolder: @@ -1007,6 +1012,47 @@ class QueuedSession: running: int done: int dead: int + #: Where this session sits in its guild's queue, lower first, or + #: `None` when it has no outstanding jobs to sit anywhere. Zero is the + #: ordinary priority and a real place; `None` is a meeting that is + #: still recording, or one that is only still listed because a job of + #: it died. The distinction is what tells a page which rows can be + #: dragged. + priority: int | None + + +@dataclass(frozen=True) +class QueuePosition: + """One session's place in its guild's queue, after a reorder.""" + + session_id: int + priority: int + + +@dataclass(frozen=True) +class QueueOrder: + """A guild's queue order, and what a reorder did to it. + + `sessions` is the whole outstanding queue in claim order, always -- + on a refusal as much as on a success. A drag is aimed at a list the + browser was showing, and the two ways it fails (the session finished, + the session it was dropped beside finished) are both "your list is out + of date": sending the current one back with the refusal is what lets a + page redraw instead of asking again and failing again. + + `changed` names the sessions whose priority was written, and is empty + when the order asked for was the order that already held. It is not + derivable from `sessions`, and it is the difference between "done" and + "there was nothing to do" -- which an administrator who dragged + something two pixels deserves to be told apart. + """ + + accepted: bool + #: Why it was refused, in one sentence, or `None`. A fixed string, so + #: a console can key off it without any input being echoed back. + refusal: str | None + sessions: tuple[QueuePosition, ...] + changed: tuple[int, ...] @dataclass(frozen=True) @@ -1064,6 +1110,10 @@ class QueueOverview(Protocol): async def for_guild(self, guild_id: int, *, requested_by: int) -> GuildQueue | None: ... + async def reprioritise( + self, guild_id: int, *, requested_by: int, rule: str + ) -> QueueOrder | None: ... + @dataclass(frozen=True) class GuildRecording: diff --git a/src/sturnus/console/routes_queue.py b/src/sturnus/console/routes_queue.py index 8f5862b..59c7033 100644 --- a/src/sturnus/console/routes_queue.py +++ b/src/sturnus/console/routes_queue.py @@ -1,12 +1,40 @@ -"""A guild's transcription queue, and re-running one session's part of it. +"""A guild's transcription queue, what runs first, and re-running one session. - `GET /api/guilds/{guild_id}/queue` - `GET /api/guilds/{guild_id}/queue/stream` +- `POST /api/guilds/{guild_id}/queue/priority` - `GET /api/sessions/{session_id}/queue` - `GET /api/sessions/{session_id}/queue/stream` +- `POST /api/sessions/{session_id}/queue/priority` - `POST /api/sessions/{session_id}/queue/requeue` - `GET /api/models` +**Why a reorder is expressed relative to another session and never as a +number.** A drag-and-drop list produces "this one goes here", and here is +a neighbour: `{"place": "before", "session": "512"}`. It does not produce +an integer, and an API that asked for one would be asking a browser to +invent the queue's arithmetic -- with a stale copy of the queue, and with +no way to agree with the other browser doing the same thing a second +later. So the console names a session it can see and the server works out +the numbers (`sturnus.application.priorities`). + +That choice is what makes two administrators dragging at once produce +something sensible. Each request is decided inside the same lock that +writes it, against the queue as it stands at that instant, so the second +is applied to what the first left rather than to the list its browser was +showing. The result is always one of the two orders those two drags could +serialise into -- never a blend of both, and never a lost write. The +second administrator may well see an order they did not picture, because +the list moved under them; the stream tells them so immediately, and an +anchor still means something after somebody else's move landed in a way +that "put it at index 3" would not. + +**A reorder names a session, and applies to that session's jobs.** The +rows are one per speaker, so a request that took job ids could reorder +four of a meeting's five speakers -- a queue that is half moved, that no +page renders and that nobody would ever notice. Nothing in this module +can express that. + **Why the guild-wide view is here rather than in a module of its own.** It is the same subject asked at a different scale: the per-session endpoints answer "where has this one got to", and the guild one answers @@ -100,10 +128,17 @@ from aiohttp import web +from sturnus.application.priorities import ( + PLACEMENTS, + Placement, + UnknownPriorityRule, + resolve_rule, +) from sturnus.console.ports import ( GuildQueue, QueueControl, QueuedSession, + QueueOrder, QueueOverview, QueueSnapshot, RequeueOutcome, @@ -127,8 +162,10 @@ _STATUS_PATH = "/api/sessions/{session_id}/queue" _STATUS_STREAM_PATH = "/api/sessions/{session_id}/queue/stream" _REQUEUE_PATH = "/api/sessions/{session_id}/queue/requeue" +_PLACE_PATH = "/api/sessions/{session_id}/queue/priority" _GUILD_PATH = "/api/guilds/{guild_id}/queue" _GUILD_STREAM_PATH = "/api/guilds/{guild_id}/queue/stream" +_GUILD_PRIORITY_PATH = "/api/guilds/{guild_id}/queue/priority" #: Deliberately not under `/api/guilds/{guild_id}/`. The registry is a #: property of this deployment's build, not of a guild -- putting a guild #: in the path would promise a per-guild answer that does not exist and @@ -141,6 +178,11 @@ #: which names both what was asked for and what there is. _MALFORMED_BODY = "malformed request body" _MODEL_MUST_BE_A_STRING = "model must be a string naming a transcription model" +_PLACE_MUST_BE_KNOWN = "place must be one of " + ", ".join(PLACEMENTS) +_ANCHOR_MUST_BE_A_SESSION_ID = "session must be a string naming the session to sit beside" +_ANCHOR_ONLY_WITH_BEFORE_OR_AFTER = "session may only be given with place before or after" +_ANCHOR_IS_THE_SESSION_ITSELF = "a session cannot be placed relative to itself" +_RULE_MUST_BE_A_STRING = "rule must be a string naming a queue rule" @dataclass(frozen=True) @@ -576,6 +618,142 @@ async def requeue_session(request: web.Request) -> web.Response: return web.json_response(_outcome_json(outcome)) +async def place_session(request: web.Request) -> web.Response: + """Moves one session to where an administrator dropped it in the queue. + + **404 for everybody who is not an administrator of the session's + guild, and the same 404 for a session that does not exist.** Expressed + by calling `QueueControl.place`, which is the same object and the same + check the re-queue endpoint above uses -- a second copy of that rule + here would be a second rule, and the copy is the one that gets left + behind when the original changes. A 403 would confirm that the session + exists and roughly when it ran, to somebody just established as having + no business knowing. + + **409 for a drag the queue has moved out from under.** The session, or + the session it was dropped beside, has finished since the page was + drawn. That is not a malformed request -- it was perfectly good a few + seconds ago -- and it is not a permission problem, so it is neither + 400 nor 404: it is the state having changed, which is exactly what + 409 says and exactly what the re-queue refusal above uses it for. The + body carries the queue as it now is, so the page can redraw rather + than replay the drag it has just been told is stale. + """ + from sturnus.console.app import current_user + + viewer = current_user(request).discord_user_id + session_id = _session_id(request) + if session_id is None: + return _no_such_session() + + placement = await _requested_placement(request, session_id) + order = await request.app[QUEUE_CONTROL].place( + session_id, requested_by=viewer, placement=placement + ) + if order is None: + return _no_such_session() + return _order_response(order, session_id=session_id, requested_by=viewer) + + +async def prioritise_guild_queue(request: web.Request) -> web.Response: + """Reorders a whole guild's queue by one named rule. + + The quick actions beside the queue: run the meetings with the most + people in them first, or the shortest recordings first. Each is a sort + key in `sturnus.application.priorities` and nothing more, so the third + one somebody asks for is one function there and one name in a + registry -- not another endpoint, and not another way to write a + priority. + + **The name is validated here, at the boundary**, and an unknown one is + a 400 naming both what was asked for and what there is. The same trade + the `model` parameter makes on the re-queue endpoint, and for the same + reason: below this line an unknown rule does not exist, so nothing + deeper has to decide what to do about one -- and silently running a + different rule than the one asked for would reorder a guild's queue in + a way the administrator could not tell from the feature working. + + 404 for a guild this person does not administer and for one that does + not exist alike, as everywhere else in this module. + """ + from sturnus.console.app import current_user + + viewer = current_user(request).discord_user_id + try: + guild_id = int(request.match_info["guild_id"]) + except ValueError: + return _no_such_guild() + + rule = await _requested_rule(request) + try: + resolve_rule(rule) + except UnknownPriorityRule as exc: + # The message names the value a caller sent, which is unbounded + # text, so it goes into the response and never into a log line -- + # the same trade the unknown-model refusal makes above. + return _bad_request(str(exc)) + + order = await request.app[QUEUE_OVERVIEW].reprioritise(guild_id, requested_by=viewer, rule=rule) + if order is None: + return _no_such_guild() + return _order_response(order, guild_id=guild_id, requested_by=viewer, rule=rule) + + +def _order_response( + order: QueueOrder, + *, + requested_by: int, + session_id: int | None = None, + guild_id: int | None = None, + rule: str | None = None, +) -> web.Response: + """One answer for both writes, refusal included. + + The audit line is here rather than in each handler because it is the + same event: somebody changed the order work will be done in. It is + logged at WARNING when something was actually written, for the reason + the re-queue audit line is -- a queue that reordered itself under a + team needs a name attached to why -- and nothing is logged for a + reorder that changed nothing, which is what a page re-sending its + current order produces. + """ + if not order.accepted: + # INFO: a stale drag is this endpoint working. Nothing was written + # and there is nothing for an operator to do. + log_event( + log, + logging.INFO, + Event.CONSOLE_QUEUE_REORDER_REFUSED, + "Refused a queue reorder asked for from the console", + session_id=session_id, + guild_id=guild_id, + requested_by=requested_by, + ) + return web.json_response(_order_json(order), status=409) + + if order.changed: + log_event( + log, + logging.WARNING, + Event.CONSOLE_QUEUE_REORDERED, + "Reordered a guild's transcription queue from the console", + session_id=session_id, + guild_id=guild_id, + requested_by=requested_by, + # One of a fixed set of literals from this repository's own + # source, which is the standard `observability.fields` holds + # `model` to. `None` for a drag, which names no rule. + rule=rule, + sessions=len(order.changed), + ) + return web.json_response( + _order_json(order), + # It names which meetings a guild has outstanding, and it is stale + # the moment a worker claims a job. + headers={"Cache-Control": "private, no-store"}, + ) + + async def known_models(request: web.Request) -> web.Response: """Every transcription model a re-queue may name, and which is the default. @@ -657,6 +835,86 @@ async def _requested_model(request: web.Request) -> str | None: return model +async def _requested_placement(request: web.Request, session_id: int) -> Placement: + """Where a drag says a session goes, checked before anything is read. + + Strict about types rather than forgiving, exactly as + `_requested_model` is: `{"place": 1}` is not coerced and + `{"session": 512}` is refused rather than read as `"512"`. Session ids + travel as strings everywhere in this API -- see `_queued_session_json` + -- so a number here is a client that has started parsing ids as + numbers somewhere, which is a bug worth failing loudly rather than + accommodating until it reaches an id that does not survive the round + trip. + + The anchor's presence is checked against the placement rather than + ignored when it does not apply. `{"place": "first", "session": "512"}` + is somebody who believes they said where; obeying half of that and + discarding the rest is how a caller learns to distrust an API. + """ + body = await _body(request) + place = body.get("place") + if not isinstance(place, str) or place not in PLACEMENTS: + raise _bad_request_exception(_PLACE_MUST_BE_KNOWN) + + anchor = _anchor_id(body) + placement = Placement(place, anchor) + if not placement.is_valid: + raise _bad_request_exception( + _ANCHOR_MUST_BE_A_SESSION_ID if anchor is None else _ANCHOR_ONLY_WITH_BEFORE_OR_AFTER + ) + if anchor == session_id: + # "Before itself" names no position at all. Refused rather than + # treated as a no-op, because a client that sent it is computing + # the anchor wrongly and a silent success hides that for ever. + raise _bad_request_exception(_ANCHOR_IS_THE_SESSION_ITSELF) + return placement + + +def _anchor_id(body: dict[str, object]) -> int | None: + if "session" not in body: + return None + anchor = body["session"] + if not isinstance(anchor, str): + raise _bad_request_exception(_ANCHOR_MUST_BE_A_SESSION_ID) + try: + return int(anchor) + except ValueError: + raise _bad_request_exception(_ANCHOR_MUST_BE_A_SESSION_ID) from None + + +async def _requested_rule(request: web.Request) -> str: + """The quick action a request named. Required, and only ever a string. + + No default. A body without a rule is not "the usual one" -- there is + no usual one, and guessing would reorder a guild's queue by a rule + nobody chose. + """ + body = await _body(request) + rule = body.get("rule") + if not isinstance(rule, str): + raise _bad_request_exception(_RULE_MUST_BE_A_STRING) + return rule + + +async def _body(request: web.Request) -> dict[str, object]: + """The request's JSON object, or a 400. + + Unlike the re-queue's body, this one is required: a re-queue with no + body is the console's button and means "no choice", while a reorder + with no body has not said what to do. + """ + if not request.body_exists: + raise _bad_request_exception(_MALFORMED_BODY) + try: + body = await request.json() + except ValueError: + raise _bad_request_exception(_MALFORMED_BODY) from None + if not isinstance(body, dict): + raise _bad_request_exception(_MALFORMED_BODY) + return body + + def _bad_request(reason: str) -> web.Response: return web.json_response({"error": reason}, status=400) @@ -769,6 +1027,53 @@ def _queued_session_json(session: QueuedSession) -> dict[str, object]: "done": session.done, "dead": session.dead, }, + # Lower first, `0` ordinary. **Present-and-null rather than absent** + # for a session with nothing outstanding -- still recording, or + # listed only because a job of it died -- so a client never has to + # tell "no place in the queue" from "this API predates the field". + # It would guess, and it would guess wrong on one of them. The same + # shape `/api/me` gives `display_name`. + # + # Null is not zero here and the difference is what a page acts on: + # zero is the ordinary priority and a real place in the queue, null + # is a row with nothing to reorder, which is a row that must not + # offer a drag handle. + "priority": session.priority, + } + + +def _order_json(order: QueueOrder) -> dict[str, object]: + """A guild's queue order, in the shape both writes answer with. + + The whole queue every time, refusal included, because this is what a + page redraws from. It is deliberately not the *difference*: a client + that applied a diff to a list it had would be applying it to the list + that may have caused the refusal. + + `changed` is sent beside it and is the one thing the order itself does + not say -- whether this request did anything. An administrator who + dragged a session two pixels and put it back gets `[]` and can be told + "nothing to do" instead of "done". + """ + return { + "accepted": order.accepted, + # Present-and-null on success rather than absent; see `_outcome_json`. + "refusal": order.refusal, + "changed": [str(session_id) for session_id in order.changed], + "order": [ + { + # A string, like every other id in this API. Session ids do + # not need it and follow anyway: two id shapes in one + # payload is how the one that matters gets parsed with the + # wrong one. + "session_id": str(position.session_id), + # Never null here, unlike the queue listing's field of the + # same name: everything in this list has outstanding work + # by construction, which is what having a place means. + "priority": position.priority, + } + for position in order.sessions + ], } @@ -803,8 +1108,10 @@ def register(app: web.Application) -> None: [ web.get(_GUILD_PATH, require_session(guild_queue)), web.get(_GUILD_STREAM_PATH, require_session(guild_queue_stream)), + web.post(_GUILD_PRIORITY_PATH, require_session(prioritise_guild_queue)), web.get(_STATUS_PATH, require_session(queue_status)), web.get(_STATUS_STREAM_PATH, require_session(queue_status_stream)), + web.post(_PLACE_PATH, require_session(place_session)), web.post(_REQUEUE_PATH, require_session(requeue_session)), web.get(_MODELS_PATH, require_session(known_models)), ] diff --git a/src/sturnus/infrastructure/db/models.py b/src/sturnus/infrastructure/db/models.py index 847368e..90a7015 100644 --- a/src/sturnus/infrastructure/db/models.py +++ b/src/sturnus/infrastructure/db/models.py @@ -463,16 +463,28 @@ class TranscriptionJob(Base): #: an order, and writes what that order implies. #: #: Lower-first is not arbitrary. The claim reads - #: `ORDER BY priority, id`, and mixed directions cannot be served by - #: one btree scan: `priority DESC, id ASC` would need an index with a - #: descending column, where `priority ASC, id ASC` is a plain forward - #: scan of `ix_job_claim_order` that also keeps first-in-first-out - #: within a priority for free. + #: `ORDER BY priority, id`, both ascending, and mixed directions could + #: never be served by a plain btree at all: `priority DESC, id ASC` + #: would need an index with a descending column. `priority ASC, id + #: ASC` also keeps first-in-first-out within a priority for free. #: - #: Nothing reads it yet -- `JobQueue.claim` is unchanged in the - #: migration that adds this. The column and its index land first so - #: that the branch which adds the ordering adds a query and not a - #: schema. + #: **The claim does not, however, get that ordering out of + #: `ix_job_claim_order`, whatever the migration that added it says.** + #: `status` leads that index and a claim matches two values of it, so + #: PostgreSQL sorts. Measured, not assumed -- + #: `sturnus.infrastructure.db.queue.claim_statement` has the plan, the + #: partial index that would fix it, and why the cheaper-looking fix + #: must not be used. + #: + #: One number is written to every outstanding job of a session at + #: once, and only by `sturnus.infrastructure.db.priority` -- the unit + #: an administrator reorders is a meeting, never a speaker. The + #: console never shows the number; it shows an order, and the server + #: writes what that order implies + #: (`sturnus.application.priorities`). It only ever raises one, so + #: nothing reachable from the console can move a guild's work ahead of + #: another guild's ordinary run; a negative number is an operator's to + #: write by hand. priority: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0") #: What the recording is, as its own RIFF header declares it. @@ -513,10 +525,11 @@ class TranscriptionJob(Base): UniqueConstraint("session_id", "discord_user_id", name="uq_job_per_speaker"), Index("ix_job_status", "status"), Index("ix_job_retention", "retention_until"), - # The claim's order, in the claim's order: status narrows to the - # pending jobs, priority orders them, and id breaks ties by age. - # All ascending, which is what keeps it one forward index scan -- - # see `priority`. + # Narrows a claim to the outstanding jobs, which is what keeps a + # poll off this table's history -- it keeps every job the + # deployment has ever run. It does *not* supply the claim's + # ordering, despite its name and despite what the migration that + # added it says: see `priority` above and `claim_statement`. Index("ix_job_claim_order", "status", "priority", "id"), ) diff --git a/src/sturnus/infrastructure/db/priority.py b/src/sturnus/infrastructure/db/priority.py new file mode 100644 index 0000000..09a8778 --- /dev/null +++ b/src/sturnus/infrastructure/db/priority.py @@ -0,0 +1,249 @@ +"""Reading a guild's queue order, and writing a new one. + +The decision this module acts on is not made here. `sturnus.application. +priorities` holds all of it -- what order was asked for, and which +integers express it -- and this module is the transaction that reads the +rows those functions need, hands them over, and writes back what comes +out. The same arrangement `sturnus.infrastructure.db.requeue` has with +`sturnus.application.requeue`, and for the same reason: the rule then has +one definition and can be tested without a database at all. + +**Why the decision is a callable rather than an argument.** `reorder` +takes a function from "the queue as it is right now" to "the order it +should be in". A drag and a quick action are then the same write with two +different decisions -- and, more importantly, the decision is taken +*inside* the lock, from rows this transaction has already locked. An API +that took a finished list of session ids would be taking one computed +from whatever the browser was showing, which is precisely the stale +snapshot two administrators dragging at once produce. + +**The lock, and why it is the guild's rows rather than one session's.** +An order is a statement about a queue, not about a session: deciding +where one meeting goes means reading where all the others are. So the +guild's outstanding jobs are locked before any of them is read, which +makes two concurrent reorders of one guild serialise -- the second sees +what the first wrote and decides against it. Rows are locked in ascending +`id`, identically to `JobQueue.complete` and `apply_requeue`, which is +what stops any two of the three deadlocking: every one of them takes its +locks in the same direction, so no cycle can form. + +While a reorder holds those locks a worker's `claim` cannot take the +guild's jobs -- it runs `FOR UPDATE SKIP LOCKED`, so it walks past them +and claims somebody else's work instead. That is the correct behaviour +and it lasts for one short transaction, but it is worth knowing that a +reorder is momentarily also a hold. + +**`FOR UPDATE OF transcription_job`**, not a bare `FOR UPDATE`. The +statement joins to `session` only to name the guild, and locking a +session row would collide with `apply_requeue`, which locks jobs and then +writes the session. Only the rows that are about to be written are +locked. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from dataclasses import dataclass + +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from sturnus.application.priorities import QueuedSession, priorities_for +from sturnus.infrastructure.db.models import Session, SessionParticipant, TranscriptionJob + +#: Job statuses that mean a worker may still act on this row -- the same +#: set `JobQueue.claim` selects from and `_outstanding_before` counts. +#: These are the only rows a priority means anything on, so they are the +#: only rows this module reads a priority from or writes one to. +OUTSTANDING_STATUSES: tuple[str, str] = ("pending", "running") + +#: What a caller wants done with the queue: given it as it stands, the +#: order it should be in, or `None` to refuse and write nothing. `None` is +#: how a drag says that the session it names, or the session it was +#: dropped beside, is not in this queue any more. +Decision = Callable[[Sequence[QueuedSession]], "tuple[int, ...] | None"] + + +@dataclass(frozen=True) +class QueueOrder: + """A guild's queue after a reorder, and what the reorder wrote. + + `changed` is separate from `sessions` and is not derivable from it: a + reorder that changed nothing and a reorder that changed everything + both return the whole queue, and the difference is exactly what an + administrator's page needs in order to say whether anything happened. + """ + + #: Every outstanding session of the guild, in the order a claim would + #: now reach them, with the priority each one now carries. + sessions: tuple[QueuedSession, ...] + #: The sessions whose priority this call wrote, ascending. Empty when + #: the order asked for was the order that already held. + changed: tuple[int, ...] + + @property + def order(self) -> tuple[int, ...]: + return tuple(session.id for session in self.sessions) + + +async def load_queued_sessions( + session_factory: async_sessionmaker[AsyncSession], guild_id: int +) -> tuple[QueuedSession, ...]: + """This guild's outstanding sessions, in the order a claim would reach them. + + Lock-free, because this is the read behind a page: holding a lock + across the time a human spends looking at a list would block every + worker completing one of that guild's jobs meanwhile. The order that + actually gets written is re-derived inside `reorder`'s lock. + """ + async with session_factory() as db: + return await _queued_sessions(db, guild_id) + + +async def reorder( + session_factory: async_sessionmaker[AsyncSession], + guild_id: int, + decide: Decision, +) -> QueueOrder | None: + """Applies one decision to one guild's queue, in one locked transaction. + + Returns the queue as it now stands, or `None` when `decide` refused -- + which for a drag means the session named, or the session it was + dropped beside, is no longer in this queue. `None` is a refusal to act + and never a partial write: nothing is written on that path at all. + + A session's number goes onto **every one of its outstanding jobs**, and + onto none of its finished ones. The unit an administrator moves is a + meeting; the rows are one per speaker; a write that moved four of five + speakers would leave a queue half-reordered in a way no page renders + and nobody could see. A `done` or `dead` job is not going to be + claimed again, so a queue position on it would be an intention + recorded about work that is over. + """ + async with session_factory() as db: + # Before anything is read. A decision taken outside this lock is a + # decision about a queue that may already have moved -- which is + # exactly what two administrators dragging at once produce. + await db.execute( + select(TranscriptionJob.id) + .select_from(TranscriptionJob) + .join(Session, Session.id == TranscriptionJob.session_id) + .where( + Session.guild_id == guild_id, + TranscriptionJob.status.in_(OUTSTANDING_STATUSES), + ) + .order_by(TranscriptionJob.id) + .with_for_update(of=TranscriptionJob) + ) + queued = await _queued_sessions(db, guild_id) + wanted = decide(queued) + if wanted is None: + await db.rollback() + return None + + changes = priorities_for(queued, wanted) + for session_id, priority in changes.items(): + await db.execute( + update(TranscriptionJob) + .where( + TranscriptionJob.session_id == session_id, + TranscriptionJob.status.in_(OUTSTANDING_STATUSES), + ) + .values(priority=priority) + ) + await db.commit() + + written = {session.id: changes.get(session.id, session.priority) for session in queued} + return QueueOrder( + # Rebuilt from what was decided rather than re-read, so the answer + # describes this transaction's own result. A second read would be + # a different moment, and could show a session that a worker + # finished in between as having left the queue this call just + # ordered. + sessions=tuple( + _at(session, written[session.id]) + for session in sorted(queued, key=lambda row: wanted.index(row.id)) + ), + changed=tuple(sorted(changes)), + ) + + +def _at(session: QueuedSession, priority: int) -> QueuedSession: + if priority == session.priority: + return session + return QueuedSession( + id=session.id, + priority=priority, + participants=session.participants, + audio_seconds=session.audio_seconds, + ) + + +async def _queued_sessions(db: AsyncSession, guild_id: int) -> tuple[QueuedSession, ...]: + """The guild's outstanding sessions, with everything a rule reads. + + Two statements, and each aggregates over a different set of rows on + purpose. + + **Priority is read from the outstanding jobs only**, as the *minimum* + over them. A reorder writes one number to all of them, so the minimum + is that number -- but a session that closed a moment later has jobs + enqueued at the ordinary `0`, and the minimum is then `0`, which is + genuinely where a claim would reach that session next. The alternative + readings would report a place the queue does not actually have it in. + + **Length is read from every job of the session**, outstanding or not. + It is a fact about the recording rather than about the queue, and a + re-queued session keeps the measurements its first pass produced -- + which is the one case where a queued session has a length at all, + since `audio_seconds` is not written until a job completes. Null where + nothing has measured anything, never zero: see + `sturnus.application.priorities`. + """ + outstanding = TranscriptionJob.status.in_(OUTSTANDING_STATUSES) + rows = ( + await db.execute( + select( + TranscriptionJob.session_id, + func.min(TranscriptionJob.priority).filter(outstanding).label("priority"), + func.sum(TranscriptionJob.audio_seconds).label("audio_seconds"), + ) + .select_from(TranscriptionJob) + .join(Session, Session.id == TranscriptionJob.session_id) + .where(Session.guild_id == guild_id) + .group_by(TranscriptionJob.session_id) + # A session with nothing outstanding is not in the queue and + # cannot be given a place in it. Expressed as a `HAVING` over + # the same aggregate the priority comes from, so the two can + # never disagree about which sessions those are. + .having(func.count().filter(outstanding) > 0) + ) + ).all() + if not rows: + return () + + counted = await db.execute( + select(SessionParticipant.session_id, func.count()) + .where(SessionParticipant.session_id.in_([row.session_id for row in rows])) + .group_by(SessionParticipant.session_id) + ) + participants: dict[int, int] = {session_id: int(count) for session_id, count in counted.all()} + return tuple( + sorted( + ( + QueuedSession( + id=row.session_id, + priority=row.priority, + # Zero for a session whose participant rows are gone, + # rather than absent: a job exists for it, so it is in + # the queue and has to be orderable. A rule that reads + # participants ranks it last, which is the truthful + # place for a meeting nobody is recorded as attending. + participants=participants.get(row.session_id, 0), + audio_seconds=None if row.audio_seconds is None else float(row.audio_seconds), + ) + for row in rows + ), + key=lambda session: (session.priority, session.id), + ) + ) diff --git a/src/sturnus/infrastructure/db/queue.py b/src/sturnus/infrastructure/db/queue.py index 666c743..5d435c2 100644 --- a/src/sturnus/infrastructure/db/queue.py +++ b/src/sturnus/infrastructure/db/queue.py @@ -85,7 +85,7 @@ from dataclasses import dataclass from datetime import UTC, datetime, timedelta -from sqlalchemy import Integer, and_, case, cast, func, literal, or_, select +from sqlalchemy import Integer, Select, and_, case, cast, func, literal, or_, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.orm import aliased from sqlalchemy.sql.elements import ColumnElement @@ -200,6 +200,81 @@ def _outstanding_before() -> ColumnElement[int]: ) +def claim_statement(lease_cutoff: datetime) -> Select[tuple[TranscriptionJob]]: + """The one statement a claim runs, built apart from running it. + + Lifted out of `claim` for a single reason: the ordering below makes a + claim about a *plan*, and a claim about a plan can only be settled by + asking PostgreSQL for one. A test that hand-copied this query into an + `EXPLAIN` would settle it about the copy, so the query + `test_no_index_this_schema_has_can_order_a_claim_by_priority` + explains is this one. + + **What the planner does with this, which is not what migration 0013 + predicted.** That migration added + `ix_job_claim_order (status, priority, id)` and said the claim's + `ORDER BY priority, id` would be one forward scan of it. It is not, + and it cannot be. `status` leads that index and this statement matches + *two* values of it -- a `pending` job and a `running` one whose lease + expired -- so a scan of it yields the pending rows in `(priority, id)` + order and then the running ones in `(priority, id)` order: two ordered + runs, not one. PostgreSQL puts a `Sort` on top. Measured on PostgreSQL + 17, with sequential and bitmap scans disabled so that the ordered + index scan was the only plan left: it still sorted. The claim is + therefore no worse off than it was before this ordering existed -- it + sorted by `id` for the same reason -- but the index is earning + nothing, and nobody should read 0013's paragraph and believe + otherwise. + + Two ways to earn it were considered and both rejected. Saying why here + is the point of this paragraph, because both look obvious. + + `ORDER BY status, priority, id` does produce one forward scan with no + sort -- measured, not assumed. It also silently means "no expired + lease is ever reclaimed while any pending job is claimable", because + `pending` sorts before `running`. That is the Defect 4 hazard put + back: a job whose worker was killed would wait for the whole queue to + drain before anybody picked it up, and its session would stay + undocumented for exactly that long. A plan node is not worth that. + + A partial index -- `(priority, id) WHERE status IN ('pending', + 'running')` -- gives the ordering *and* keeps the semantics, because + the status predicate moves into the index's `WHERE` and stops being a + key column. It is the right answer, and it is a migration, which this + branch deliberately is not. Whoever writes the next one should write + it, and may drop `ix_job_claim_order` in the same breath. + """ + return ( + select(TranscriptionJob) + .where( + or_( + TranscriptionJob.status == "pending", + and_( + TranscriptionJob.status == "running", + TranscriptionJob.claimed_at < lease_cutoff, + ), + ), + _outstanding_before() < _parallel_track_limit(), + ) + # **Lower first, then oldest first.** `priority` is what an + # administrator said should happen sooner (see + # `TranscriptionJob.priority` and `sturnus.application.priorities`); + # `id` is the age it has always been ordered by, and it still + # breaks every tie, so a queue nobody has touched claims in exactly + # the first-in-first-out order it did before this clause grew a + # column. + # + # Both ascending. That is what the lower-first convention was + # chosen for and it stays right whatever the plan does with it: + # `priority DESC, id ASC` would need an index with a descending + # column before it could ever be scanned in order, so it forecloses + # the partial index the docstring above recommends. + .order_by(TranscriptionJob.priority, TranscriptionJob.id) + .with_for_update(skip_locked=True) + .limit(1) + ) + + def _claim_is_current(job: TranscriptionJob, lease: datetime | None) -> bool: """Whether the caller may still act on this job. @@ -308,36 +383,29 @@ async def claim(self, now: datetime | None = None) -> ClaimedJob | None: past a session that is at its limit and takes the next session's work. - **Oldest first.** `ORDER BY id` is the ordering guarantee this - method now makes and did not make before: previously there was no - `ORDER BY` at all and a claim took whatever the scan reached - first. Jobs are enqueued when a session closes, so ascending id is - the order the meetings ended in -- and a guild that keeps meeting - gets ids that sort *behind* everything already waiting, which is - what stops it from starving a quieter guild. The cap and the - ordering answer opposite halves of the same question: the cap - stops one session monopolising the pool, the ordering stops the - sessions it makes room for being served out of turn. + **Lower priority first, then oldest first.** `ORDER BY priority, + id` is the ordering guarantee this method makes; see + `claim_statement`, which holds the clause and the argument for its + directions. `id` is the age it has always sorted by: jobs are + enqueued when a session closes, so ascending id is the order the + meetings ended in, and a guild that keeps meeting gets ids that + sort *behind* everything already waiting, which is what stops it + from starving a quieter guild. `priority` in front of it is the + only thing that can override that, it is written by nobody except + `sturnus.infrastructure.db.priority` on an administrator's + instruction, and it defaults to the same value for everybody -- so + a deployment where nobody has ever asked for anything claims in + exactly the order it did before. + + The cap and the ordering answer opposite halves of the same + question: the cap stops one session monopolising the pool, the + ordering stops the sessions it makes room for being served out of + turn. """ now = now if now is not None else datetime.now(UTC) lease_cutoff = now - timedelta(seconds=self._lease_seconds) async with self._session_factory() as session: - job = await session.scalar( - select(TranscriptionJob) - .where( - or_( - TranscriptionJob.status == "pending", - and_( - TranscriptionJob.status == "running", - TranscriptionJob.claimed_at < lease_cutoff, - ), - ), - _outstanding_before() < _parallel_track_limit(), - ) - .order_by(TranscriptionJob.id) - .with_for_update(skip_locked=True) - .limit(1) - ) + job = await session.scalar(claim_statement(lease_cutoff)) if job is None: return None job.status = "running" diff --git a/src/sturnus/infrastructure/db/requeue.py b/src/sturnus/infrastructure/db/requeue.py index 607c8d7..df3f244 100644 --- a/src/sturnus/infrastructure/db/requeue.py +++ b/src/sturnus/infrastructure/db/requeue.py @@ -26,6 +26,7 @@ from sturnus.application.publishing import DOCUMENTED_STATUS from sturnus.application.requeue import TERMINAL_STATUSES, RequeuePlan, plan_requeue from sturnus.infrastructure.db.models import Session, SessionParticipant, TranscriptionJob +from sturnus.infrastructure.db.priority import OUTSTANDING_STATUSES #: Job statuses reported by a queue summary, in lifecycle order rather #: than alphabetically: a reader looks at this to see where work is piling @@ -276,6 +277,15 @@ class ActiveSession: #: One entry per `REPORTED_STATUSES`, zero-filled, so a caller can #: render the lifecycle in order without checking for absent keys. counts: dict[str, int] + #: Where this session sits in its guild's queue -- the priority its + #: outstanding jobs carry, lower first. **`None` when it has none**, + #: which is a meeting that is still recording, or one whose every job + #: has finished and which is only still listed because one of them + #: died. Deliberately not `0`: zero is the ordinary priority and a + #: real place in the queue, and a session with nothing queued does not + #: have a place in it. A page that read the two as the same would + #: offer a drag handle on a row that nothing can be reordered about. + priority: int | None async def load_active_sessions( @@ -330,18 +340,39 @@ async def load_active_sessions( return [], False counted = await db.execute( - select(TranscriptionJob.session_id, TranscriptionJob.status, func.count()) + # The priority comes back from the same grouped read as the + # counts rather than from a statement of its own, because the + # two are read for one page and a second query would be a + # second moment: a row could show a job that the counts say is + # still pending sitting at a priority its reorder has already + # moved on from. + select( + TranscriptionJob.session_id, + TranscriptionJob.status, + func.count(), + func.min(TranscriptionJob.priority), + ) .where(TranscriptionJob.session_id.in_([row.id for row in found])) .group_by(TranscriptionJob.session_id, TranscriptionJob.status) ) per_session: dict[int, dict[str, int]] = { row.id: dict.fromkeys(REPORTED_STATUSES, 0) for row in found } - for session_id, status, count in counted: + priorities: dict[int, int] = {} + for session_id, status, count, priority in counted: # `setdefault` rather than assignment: a status this build does # not know about is still work somebody has to account for, and # dropping it would make the counts silently fail to add up. per_session[session_id][status] = per_session[session_id].get(status, 0) + int(count) + if status in OUTSTANDING_STATUSES: + # The lowest number over the jobs a worker may still take, + # which is where a claim would next reach this session. A + # reorder writes one number to all of them, so this is + # normally that number; it differs only while a session + # that was reordered has since enqueued more speakers, and + # then the lower one is the honest answer. + current = priorities.get(session_id) + priorities[session_id] = priority if current is None else min(current, priority) return [ ActiveSession( @@ -353,6 +384,7 @@ async def load_active_sessions( status=row.status, document_url=row.document_url, counts=per_session[row.id], + priority=priorities.get(row.id), ) for row in found ], truncated diff --git a/src/sturnus/observability/events.py b/src/sturnus/observability/events.py index 23e330a..3af421a 100644 --- a/src/sturnus/observability/events.py +++ b/src/sturnus/observability/events.py @@ -192,6 +192,18 @@ class Event(StrEnum): #: not in a state a redo is safe from. INFO: this is the feature #: working, not failing. CONSOLE_REQUEUE_REFUSED = "console.requeue_refused" + #: An administrator changed the order a guild's transcription work + #: will be done in -- a session dragged, or a quick action applied to + #: the whole queue. **WARNING for the same reason + #: `CONSOLE_REQUEUE_APPLIED` is:** the effect is felt by everybody + #: waiting on a protocol from that guild, and this line is the only + #: record of who asked. Emitted only when something was actually + #: written; a reorder that changed nothing is not an event. + CONSOLE_QUEUE_REORDERED = "console.queue_reordered" + #: A reorder that was asked for and refused, because the session it + #: named -- or the one it was dropped beside -- had left the queue + #: since the page was drawn. INFO: this is the feature working. + CONSOLE_QUEUE_REORDER_REFUSED = "console.queue_reorder_refused" #: A stored recording that this reader cannot make sense of. **A human #: must act:** the object is there, the person is entitled to it, and #: it will not decrypt -- which is either a truncated upload or a diff --git a/src/sturnus/observability/fields.py b/src/sturnus/observability/fields.py index 181ffad..8802975 100644 --- a/src/sturnus/observability/fields.py +++ b/src/sturnus/observability/fields.py @@ -161,6 +161,12 @@ def service_name(component: str) -> str: #: one clicking "withdraw" both leave a perfectly ordinary date #: in the column, and only this field says which act it was. "effective_at_given", + #: Which quick action reordered a guild's queue -- one of the + #: names in `sturnus.application.priorities.KNOWN_RULES`, checked + #: against that registry before anything logs it, and `None` for a + #: drag, which names no rule. Fixed literals of this repository's + #: own source, the same standard `model` is held to. + "rule", #: A key of `sturnus.domain.settings` -- one of the literals in #: `KNOWN_KEYS`, from this repository's own source, and checked #: against the registry before anything logs it. The *value* is @@ -198,6 +204,11 @@ def service_name(component: str) -> str: "speakers", "skipped", "participants", + #: How many sessions a reorder actually moved. A count of rows + #: written, never which ones -- the ids are in the response the + #: administrator got, and a log line naming which meetings a guild + #: has outstanding is a log line about a guild's calendar. + "sessions", "blocks", "packets", "segments", diff --git a/tests/application/test_priorities.py b/tests/application/test_priorities.py new file mode 100644 index 0000000..a610b68 --- /dev/null +++ b/tests/application/test_priorities.py @@ -0,0 +1,285 @@ +"""Turning "this one goes first" into the integers the claim reads. + +Two rules carry every test here. + +**An order is only ever expressed by holding sessions back.** No function +in this module may lower a priority, because the queue is shared with +every other guild: a reorder that could write a smaller number would be a +control by which any one administrator jumps their whole guild ahead of +everybody else's ordinary work. + +**Null is not zero.** A session nothing has measured has no length, and a +rule that ordered by length must not read that absence as "nought +seconds", which would promote every unmeasured session to the front on +the strength of knowing nothing about it. +""" + +from __future__ import annotations + +import pytest + +from sturnus.application.priorities import ( + KNOWN_RULES, + Placement, + QueuedSession, + UnknownPriorityRule, + claim_order, + order_by_rule, + order_with, + priorities_for, + resolve_rule, +) + +# Ids ascending, because the claim breaks a tie by id and every test here +# depends on knowing which way that tie falls. +FIRST, SECOND, THIRD, FOURTH = 10, 20, 30, 40 + + +def session( + session_id: int, + priority: int = 0, + participants: int = 1, + audio_seconds: float | None = None, +) -> QueuedSession: + return QueuedSession( + id=session_id, + priority=priority, + participants=participants, + audio_seconds=audio_seconds, + ) + + +# --------------------------------------------------------------------------- +# The order the claim would take them in +# --------------------------------------------------------------------------- + + +def test_an_untouched_queue_runs_oldest_first() -> None: + """Every session at the ordinary priority is the queue as it shipped.""" + sessions = [session(SECOND), session(FIRST), session(THIRD)] + + assert [row.id for row in claim_order(sessions)] == [FIRST, SECOND, THIRD] + + +def test_a_lower_priority_runs_before_an_older_session() -> None: + """Lower first, which is the `nice(1)` sense the column was given.""" + sessions = [session(FIRST, priority=1), session(SECOND, priority=0)] + + assert [row.id for row in claim_order(sessions)] == [SECOND, FIRST] + + +# --------------------------------------------------------------------------- +# A drag, turned into integers +# --------------------------------------------------------------------------- + + +def test_a_session_dragged_to_the_front_is_first() -> None: + sessions = [session(FIRST), session(SECOND), session(THIRD)] + + order = order_with(sessions, THIRD, Placement("first")) + + assert order == (THIRD, FIRST, SECOND) + + +def test_a_session_dragged_behind_another_sits_directly_after_it() -> None: + sessions = [session(FIRST), session(SECOND), session(THIRD)] + + order = order_with(sessions, FIRST, Placement("after", anchor=SECOND)) + + assert order == (SECOND, FIRST, THIRD) + + +def test_a_session_dragged_in_front_of_another_sits_directly_before_it() -> None: + sessions = [session(FIRST), session(SECOND), session(THIRD)] + + order = order_with(sessions, THIRD, Placement("before", anchor=SECOND)) + + assert order == (FIRST, THIRD, SECOND) + + +def test_a_session_dragged_to_the_end_is_last() -> None: + sessions = [session(FIRST), session(SECOND), session(THIRD)] + + order = order_with(sessions, FIRST, Placement("last")) + + assert order == (SECOND, THIRD, FIRST) + + +def test_a_drag_of_a_session_that_is_no_longer_queued_is_refused() -> None: + """The list the browser was showing has moved on. + + `None` rather than an exception or a silent no-op: the caller turns it + into a refusal that carries the queue as it now is, so the page can + redraw instead of the drag being applied to a queue nobody looked at. + """ + sessions = [session(FIRST), session(SECOND)] + + assert order_with(sessions, THIRD, Placement("first")) is None + + +def test_a_drag_against_an_anchor_that_is_no_longer_queued_is_refused() -> None: + sessions = [session(FIRST), session(SECOND)] + + assert order_with(sessions, FIRST, Placement("after", anchor=THIRD)) is None + + +# --------------------------------------------------------------------------- +# The integers themselves +# --------------------------------------------------------------------------- + + +def test_an_order_that_already_holds_writes_nothing() -> None: + """Idempotence, and it is not a nicety. + + Two administrators who agree, a quick action applied twice, and a page + that re-sends what it is already showing must all cost nothing -- + otherwise every one of them would push the whole queue further back. + """ + sessions = [session(FIRST), session(SECOND), session(THIRD)] + + assert priorities_for(sessions, (FIRST, SECOND, THIRD)) == {} + + +def test_going_first_holds_back_only_what_it_overtakes() -> None: + """The moved session keeps its number; the ones it passed lose theirs. + + THIRD goes to the front of a queue that is entirely ordinary, so + FIRST and SECOND -- the two it overtook -- move to 1, and THIRD stays + at 0. Nothing else in the deployment is touched. + """ + sessions = [session(FIRST), session(SECOND), session(THIRD)] + + assert priorities_for(sessions, (THIRD, FIRST, SECOND)) == {FIRST: 1, SECOND: 1} + + +def test_a_session_that_already_sorts_after_its_predecessor_keeps_its_number() -> None: + """Ascending ids need no new integer at all. + + FIRST at 0 and SECOND at 0 already run in that order, because the + claim breaks the tie by id -- so putting THIRD last writes nothing for + any of them. + """ + sessions = [session(FIRST), session(SECOND), session(THIRD)] + + assert priorities_for(sessions, (FIRST, SECOND, THIRD)) == {} + + +def test_no_reorder_ever_lowers_a_priority() -> None: + """The property that makes this endpoint safe to expose at all. + + A guild expresses an order over its own queue by holding its own + sessions back, never by writing a smaller number -- so no + administrator can move their guild ahead of another guild's ordinary + work, however they drag. + """ + sessions = [session(FIRST, priority=3), session(SECOND, priority=3), session(THIRD)] + before = {row.id: row.priority for row in sessions} + + for order in ((THIRD, FIRST, SECOND), (SECOND, THIRD, FIRST), (FIRST, SECOND, THIRD)): + for session_id, priority in priorities_for(sessions, order).items(): + assert priority >= before[session_id] + + +def test_the_written_numbers_reproduce_the_order_that_was_asked_for() -> None: + """The whole contract, checked by replaying the claim's own sort.""" + sessions = [session(FIRST), session(SECOND), session(THIRD), session(FOURTH)] + wanted = (THIRD, FIRST, FOURTH, SECOND) + + written = priorities_for(sessions, wanted) + applied = [session(row.id, priority=written.get(row.id, row.priority)) for row in sessions] + + assert tuple(row.id for row in claim_order(applied)) == wanted + + +# --------------------------------------------------------------------------- +# The quick actions +# --------------------------------------------------------------------------- + + +def test_the_biggest_meeting_runs_first() -> None: + sessions = [ + session(FIRST, participants=2), + session(SECOND, participants=8), + session(THIRD, participants=5), + ] + + order = order_by_rule(sessions, resolve_rule("many-participants-first")) + + assert order == (SECOND, THIRD, FIRST) + + +def test_meetings_of_the_same_size_keep_the_order_they_had() -> None: + """A quick action reorders; it does not shuffle what it has no opinion on.""" + sessions = [ + session(SECOND, participants=3), + session(FIRST, participants=3), + session(THIRD, participants=9), + ] + + order = order_by_rule(sessions, resolve_rule("many-participants-first")) + + assert order == (THIRD, FIRST, SECOND) + + +def test_the_shortest_recording_runs_first() -> None: + sessions = [ + session(FIRST, audio_seconds=600.0), + session(SECOND, audio_seconds=60.0), + session(THIRD, audio_seconds=300.0), + ] + + order = order_by_rule(sessions, resolve_rule("short-recordings-first")) + + assert order == (SECOND, THIRD, FIRST) + + +def test_a_recording_nobody_has_measured_is_not_treated_as_the_shortest() -> None: + """Null is not zero, and here that distinction decides the whole order. + + A session whose tracks have never been transcribed has no + `audio_seconds` at all. Read as nought it would be the shortest + recording in the queue and would go to the front on the strength of + nothing being known about it, which is the opposite of what the rule + promises. + """ + sessions = [ + session(FIRST, audio_seconds=None), + session(SECOND, audio_seconds=90.0), + session(THIRD, audio_seconds=45.0), + ] + + order = order_by_rule(sessions, resolve_rule("short-recordings-first")) + + assert order == (THIRD, SECOND, FIRST) + + +def test_unmeasured_recordings_keep_the_order_they_had_among_themselves() -> None: + sessions = [ + session(SECOND, audio_seconds=None), + session(FIRST, audio_seconds=None), + session(THIRD, audio_seconds=30.0), + ] + + order = order_by_rule(sessions, resolve_rule("short-recordings-first")) + + assert order == (THIRD, FIRST, SECOND) + + +def test_a_rule_nobody_has_is_refused_by_name() -> None: + """The refusal names both what was asked for and what there is. + + The same trade `transcription_models.resolve` makes, for the same + reason: a rule name is a fixed literal of this repository, so a caller + who mistyped one can be shown the list without disclosing anything. + """ + with pytest.raises(UnknownPriorityRule) as refused: + resolve_rule("longest-first") + + assert "longest-first" in str(refused.value) + assert "many-participants-first" in str(refused.value) + + +def test_every_known_rule_can_be_resolved() -> None: + """The registry and the resolver cannot drift apart.""" + for name in KNOWN_RULES: + assert resolve_rule(name) is not None diff --git a/tests/console/conftest.py b/tests/console/conftest.py index 8336520..adc3518 100644 --- a/tests/console/conftest.py +++ b/tests/console/conftest.py @@ -27,6 +27,7 @@ from aiohttp.test_utils import TestClient, TestServer from sturnus.application.directory_mirror import MirroredGuild +from sturnus.application.priorities import Placement from sturnus.console.app import build_api from sturnus.console.audio import AudioDelivery from sturnus.console.filters import SessionFilter @@ -54,6 +55,7 @@ PreferenceDirectory, ProfileDirectory, QueueControl, + QueueOrder, QueueOverview, QueueSnapshot, RequeueOutcome, @@ -502,10 +504,17 @@ def __init__( self, snapshot: QueueSnapshot | None = None, outcome: RequeueOutcome | None = None, + order: QueueOrder | None = None, ) -> None: self.snapshot = snapshot self.outcome = outcome + self.order = order self.requeued: list[tuple[int, int, str]] = [] + #: Every placement this was asked to write, with who asked. Same + #: reason `requeued` is recorded: a handler that dropped the + #: placement, or that passed an id out of the URL instead of the + #: signed-in one, would still return a perfectly plausible 200. + self.placed: list[tuple[int, int, Placement]] = [] async def status_for(self, session_id: int, *, requested_by: int) -> QueueSnapshot | None: del session_id, requested_by @@ -522,6 +531,12 @@ async def requeue( self.requeued.append((session_id, requested_by, model)) return self.outcome + async def place( + self, session_id: int, *, requested_by: int, placement: Placement + ) -> QueueOrder | None: + self.placed.append((session_id, requested_by, placement)) + return self.order + class FakeTags: """The tag write path, in memory, scoped by the asking user. @@ -567,17 +582,26 @@ class FakeQueueOverview: gets 404s rather than a fake that quietly authorises everything. """ - def __init__(self, queue: GuildQueue | None = None) -> None: + def __init__(self, queue: GuildQueue | None = None, order: QueueOrder | None = None) -> None: self.queue = queue + self.order = order #: Every guild this was asked about, with who asked. The route #: tests assert on it: "the handler passed the signed-in id, not #: one from the URL" cannot be seen in a response body. self.asked: list[tuple[int, int]] = [] + #: Every quick action this was asked to apply, with who asked. + self.reprioritised: list[tuple[int, int, str]] = [] async def for_guild(self, guild_id: int, *, requested_by: int) -> GuildQueue | None: self.asked.append((guild_id, requested_by)) return self.queue + async def reprioritise( + self, guild_id: int, *, requested_by: int, rule: str + ) -> QueueOrder | None: + self.reprioritised.append((guild_id, requested_by, rule)) + return self.order + class FakeConsents: """A consent directory nobody administers, until a test says otherwise. diff --git a/tests/console/test_queue_overview.py b/tests/console/test_queue_overview.py index 31eeaeb..061473c 100644 --- a/tests/console/test_queue_overview.py +++ b/tests/console/test_queue_overview.py @@ -18,6 +18,7 @@ from datetime import UTC, datetime, timedelta import pytest +from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sturnus.console.adapters import ConsoleQueueOverview @@ -441,3 +442,126 @@ async def test_a_guild_with_nothing_outstanding_returns_an_empty_list( assert sessions == [] assert truncated is False + + +# --------------------------------------------------------------------------- +# What runs first +# --------------------------------------------------------------------------- + + +async def test_a_session_reports_where_it_sits_in_the_queue( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await a_session(factory, jobs={BEN: "pending"}) + await hold_back(factory, session_id, 3) + + queue = await overview(factory).for_guild(GUILD, requested_by=ANNA) + + assert queue is not None + assert queue.sessions[0].priority == 3 + + +async def test_a_session_nobody_has_reordered_sits_at_the_ordinary_priority( + factory: async_sessionmaker[AsyncSession], +) -> None: + await a_session(factory, jobs={BEN: "pending"}) + + queue = await overview(factory).for_guild(GUILD, requested_by=ANNA) + + assert queue is not None + assert queue.sessions[0].priority == 0 + + +async def test_a_recording_in_progress_has_no_place_in_the_queue_at_all( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Null, not zero -- there is nothing queued for it to have a place with. + + Zero is the ordinary priority and a real position. Reporting it here + would put a drag handle on a row that nothing can be reordered about. + """ + await a_session(factory, status="open", jobs={}) + + queue = await overview(factory).for_guild(GUILD, requested_by=ANNA) + + assert queue is not None + assert queue.sessions[0].priority is None + + +async def test_a_session_listed_only_for_a_dead_job_has_no_place_either( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Every job terminal means nothing will ever be claimed for it again.""" + await a_session(factory, status="documented", document_url="u", jobs={BEN: "dead"}) + + queue = await overview(factory).for_guild(GUILD, requested_by=ANNA) + + assert queue is not None + assert queue.sessions[0].priority is None + + +async def test_a_place_is_read_from_the_outstanding_jobs_and_not_the_finished_ones( + factory: async_sessionmaker[AsyncSession], +) -> None: + """A finished job's number describes a queue it has already left.""" + session_id = await a_session(factory, jobs={ANNA: "done", BEN: "pending"}) + await hold_back(factory, session_id, 7, status="done") + + queue = await overview(factory).for_guild(GUILD, requested_by=ANNA) + + assert queue is not None + assert queue.sessions[0].priority == 0 + + +# --------------------------------------------------------------------------- +# Reordering a guild's queue by a rule +# --------------------------------------------------------------------------- + + +async def test_an_administrator_can_put_the_biggest_meetings_first( + factory: async_sessionmaker[AsyncSession], +) -> None: + small = await a_session(factory, jobs={BEN: "pending"}) + large = await a_session(factory, jobs={ANNA: "pending", BEN: "pending", CARL: "pending"}) + + order = await overview(factory).reprioritise( + GUILD, requested_by=ANNA, rule="many-participants-first" + ) + + assert order is not None + assert order.accepted is True + assert [position.session_id for position in order.sessions] == [large, small] + assert order.changed == (small,) + + +async def test_somebody_who_does_not_administer_the_guild_cannot_reorder_it( + factory: async_sessionmaker[AsyncSession], +) -> None: + """`None` for "no such guild" and "not yours" alike, as everywhere else.""" + session_id = await a_session(factory, jobs={BEN: "pending"}) + + order = await overview(factory).reprioritise( + GUILD, requested_by=BEN, rule="many-participants-first" + ) + + assert order is None + async with factory() as db: + rows = await db.execute( + select(TranscriptionJob.priority).where(TranscriptionJob.session_id == session_id) + ) + assert [row.priority for row in rows] == [0] + + +async def hold_back( + factory: async_sessionmaker[AsyncSession], + session_id: int, + priority: int, + status: str | None = None, +) -> None: + """Puts a session's jobs at a priority, as a reorder does.""" + async with factory() as db: + statement = update(TranscriptionJob).where(TranscriptionJob.session_id == session_id) + if status is not None: + statement = statement.where(TranscriptionJob.status == status) + await db.execute(statement.values(priority=priority)) + await db.commit() diff --git a/tests/console/test_queue_placement.py b/tests/console/test_queue_placement.py new file mode 100644 index 0000000..0cb354d --- /dev/null +++ b/tests/console/test_queue_placement.py @@ -0,0 +1,187 @@ +"""Dragging one session, through `ConsoleQueueControl`, against the database. + +The arithmetic is `tests/application/test_priorities.py` and the write is +`tests/infrastructure/test_priority.py`. What is left, and what is tested +here, is the join between them: that the authorisation check is the one +`requeue` already uses, that it is made against the session's *own* guild, +and that a person who fails it changes nothing rather than merely being +told no. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from sturnus.application.priorities import Placement +from sturnus.console.adapters import STALE_DRAG, ConsoleQueueControl +from sturnus.infrastructure.db.models import ( + Base, + Session, + SessionParticipant, + TranscriptionJob, +) + +T0 = datetime(2026, 8, 23, 12, 0, 0, tzinfo=UTC) +GUILD, OTHER_GUILD = 4711, 9999 +ANNA, BEN = 100, 200 + + +@pytest.fixture +async def factory(clean_database: str) -> async_sessionmaker[AsyncSession]: + engine = create_async_engine(clean_database) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + return async_sessionmaker(engine, expire_on_commit=False) + + +class Admins: + def __init__(self, by_guild: dict[int, set[int]] | None = None) -> None: + self.by_guild = by_guild if by_guild is not None else {GUILD: {ANNA}} + + async def is_admin_anywhere(self, discord_user_id: int) -> bool: + return any(discord_user_id in members for members in self.by_guild.values()) + + async def administered_guilds(self, discord_user_id: int) -> tuple[int, ...]: + return tuple( + sorted(g for g, members in self.by_guild.items() if discord_user_id in members) + ) + + async def is_admin(self, guild_id: int, discord_user_id: int) -> bool: + return discord_user_id in self.by_guild.get(guild_id, set()) + + +def control( + factory: async_sessionmaker[AsyncSession], admins: Admins | None = None +) -> ConsoleQueueControl: + return ConsoleQueueControl(factory, admins or Admins()) + + +async def a_session( + factory: async_sessionmaker[AsyncSession], + *, + guild_id: int = GUILD, + speakers: tuple[int, ...] = (BEN,), + status: str = "pending", +) -> int: + async with factory() as db: + session = Session( + guild_id=guild_id, + channel_id=555, + channel_name="meeting", + started_at=T0, + ended_at=T0 + timedelta(hours=1), + status="closed", + ) + db.add(session) + await db.flush() + for discord_user_id in speakers: + db.add( + SessionParticipant( + session_id=session.id, + discord_user_id=discord_user_id, + discord_display_name=f"user-{discord_user_id}", + first_seen_at=T0, + ) + ) + db.add( + TranscriptionJob( + session_id=session.id, + discord_user_id=discord_user_id, + s3_key=f"sessions/{session.id}/speakers/{discord_user_id}.enc", + encryption_key_id="k1", + wrapped_data_key=b"wrapped", + retention_until=T0 + timedelta(days=30), + status=status, + ) + ) + await db.commit() + return session.id + + +async def priorities(factory: async_sessionmaker[AsyncSession]) -> dict[int, int]: + async with factory() as db: + rows = await db.execute( + select(TranscriptionJob.session_id, TranscriptionJob.priority).order_by( + TranscriptionJob.id + ) + ) + return {session_id: priority for session_id, priority in rows} + + +async def test_an_administrator_moves_a_session_to_the_front( + factory: async_sessionmaker[AsyncSession], +) -> None: + first = await a_session(factory) + second = await a_session(factory) + + order = await control(factory).place(second, requested_by=ANNA, placement=Placement("first")) + + assert order is not None + assert order.accepted is True + assert [position.session_id for position in order.sessions] == [second, first] + assert order.changed == (first,) + + +async def test_somebody_who_does_not_administer_the_guild_writes_nothing( + factory: async_sessionmaker[AsyncSession], +) -> None: + """`None`, which the route renders as the same 404 a missing session gets. + + And, more importantly than the answer: the queue is untouched. An + authorisation check that refused the *response* while the write had + already happened would be no check at all. + """ + first = await a_session(factory) + second = await a_session(factory) + + order = await control(factory).place(second, requested_by=BEN, placement=Placement("first")) + + assert order is None + assert await priorities(factory) == {first: 0, second: 0} + + +async def test_administering_another_guild_is_not_administering_this_one( + factory: async_sessionmaker[AsyncSession], +) -> None: + admins = Admins({GUILD: {ANNA}, OTHER_GUILD: {BEN}}) + first = await a_session(factory) + second = await a_session(factory) + + order = await control(factory, admins).place( + second, requested_by=BEN, placement=Placement("first") + ) + + assert order is None + assert await priorities(factory) == {first: 0, second: 0} + + +async def test_a_session_that_does_not_exist_is_refused_the_same_way( + factory: async_sessionmaker[AsyncSession], +) -> None: + order = await control(factory).place(9999, requested_by=ANNA, placement=Placement("first")) + + assert order is None + + +async def test_a_drag_the_queue_has_moved_past_is_refused_with_the_queue( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The session finished while the page was open. + + Not `None`: the person may make this request, so the answer is a + refusal rather than a 404 -- and it carries the queue as it now is, so + the page redraws instead of replaying a drag that cannot land. + """ + waiting = await a_session(factory) + finished = await a_session(factory, status="done") + + order = await control(factory).place(finished, requested_by=ANNA, placement=Placement("first")) + + assert order is not None + assert order.accepted is False + assert order.refusal == STALE_DRAG + assert [position.session_id for position in order.sessions] == [waiting] diff --git a/tests/console/test_queue_priority_routes.py b/tests/console/test_queue_priority_routes.py new file mode 100644 index 0000000..c19e828 --- /dev/null +++ b/tests/console/test_queue_priority_routes.py @@ -0,0 +1,387 @@ +"""Who may say what the queue does first, and what they have to say to say it. + +Three things are pinned here and each of them is a way this endpoint could +be plausibly wrong while returning 200. + +**The rule is administrator-of-the-guild, and the refusal is a 404.** +Reordering a queue is an operation on the system -- it changes the order +everybody in that guild waits in -- so it is the re-queue rule and not the +"was this your meeting" rule the audio endpoints use. 403 would confirm +that a session exists and roughly when it ran, to somebody just +established as having no business knowing. + +**The console never sends a number.** A drag says "before that one", and +the arithmetic is the server's. A request shape that accepted an integer +would be an API asking a browser to compute the queue's order from a stale +copy of it. + +**A drag the queue has moved out from under is a 409 that carries the +queue.** Not a 400 -- the request was good when it was made -- and not a +silent success, which is the failure that makes a drag-and-drop list +untrustworthy. +""" + +from __future__ import annotations + +from datetime import timedelta + +import pytest +from aiohttp import web +from aiohttp.test_utils import TestClient + +from sturnus.application.priorities import Placement +from sturnus.console.app import SESSION_COOKIE +from sturnus.console.ports import QueueOrder, QueuePosition +from sturnus.console.session import SessionCookie, SignedSession +from tests.console.conftest import ( + ANNA, + GUILD, + SECRET, + SESSION, + T0, + AiohttpClientFactory, + FakeQueue, + FakeQueueOverview, + build_test_api, +) + +OTHER_SESSION = 512 + + +def token(discord_user_id: int = ANNA) -> str: + return SessionCookie(SECRET, timedelta(hours=12)).issue(SignedSession(discord_user_id), now=T0) + + +async def signed_in( + aiohttp_client: AiohttpClientFactory, app: web.Application, as_user: int = ANNA +) -> TestClient[web.Request, web.Application]: + client = await aiohttp_client(app) + client.session.cookie_jar.update_cookies({SESSION_COOKIE: token(as_user)}) + return client + + +def place_url(session_id: int = SESSION) -> str: + return f"/api/sessions/{session_id}/queue/priority" + + +def guild_url(guild_id: int = GUILD) -> str: + return f"/api/guilds/{guild_id}/queue/priority" + + +def order(**over: object) -> QueueOrder: + base: dict[str, object] = { + "accepted": True, + "refusal": None, + "sessions": ( + QueuePosition(session_id=SESSION, priority=0), + QueuePosition(session_id=OTHER_SESSION, priority=1), + ), + "changed": (OTHER_SESSION,), + } + base.update(over) + return QueueOrder(**base) # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- +# A drag +# --------------------------------------------------------------------------- + + +async def test_an_administrator_can_put_a_session_at_the_front_of_the_queue( + aiohttp_client: AiohttpClientFactory, +) -> None: + queue = FakeQueue(order=order()) + client = await signed_in(aiohttp_client, build_test_api(queue=queue)) + + response = await client.post(place_url(), json={"place": "first"}) + + assert response.status == 200 + body = await response.json() + assert body["accepted"] is True + assert body["order"] == [ + {"session_id": str(SESSION), "priority": 0}, + {"session_id": str(OTHER_SESSION), "priority": 1}, + ] + assert body["changed"] == [str(OTHER_SESSION)] + + +async def test_a_drag_names_a_neighbour_and_the_placement_reaches_the_write( + aiohttp_client: AiohttpClientFactory, +) -> None: + """A handler that dropped the anchor would still return a plausible 200.""" + queue = FakeQueue(order=order()) + client = await signed_in(aiohttp_client, build_test_api(queue=queue)) + + await client.post(place_url(), json={"place": "before", "session": str(OTHER_SESSION)}) + + assert queue.placed == [(SESSION, ANNA, Placement("before", OTHER_SESSION))] + + +async def test_the_write_is_made_on_behalf_of_the_signed_in_person( + aiohttp_client: AiohttpClientFactory, +) -> None: + queue = FakeQueue(order=order()) + client = await signed_in(aiohttp_client, build_test_api(queue=queue)) + + await client.post(place_url(), json={"place": "last"}) + + assert [asked for _, asked, _ in queue.placed] == [ANNA] + + +async def test_somebody_who_does_not_administer_the_guild_is_told_it_does_not_exist( + aiohttp_client: AiohttpClientFactory, +) -> None: + """404, not 403, and the same 404 a session that never existed gets. + + The control answers `None` for both reasons, which is the whole point + of folding them together there. + """ + client = await signed_in(aiohttp_client, build_test_api(queue=FakeQueue())) + + response = await client.post(place_url(), json={"place": "first"}) + + assert response.status == 404 + assert (await response.json())["error"] == "no such session" + + +async def test_a_session_id_that_is_not_a_number_gets_the_same_refusal( + aiohttp_client: AiohttpClientFactory, +) -> None: + client = await signed_in(aiohttp_client, build_test_api(queue=FakeQueue(order=order()))) + + response = await client.post("/api/sessions/nonsense/queue/priority", json={"place": "first"}) + + assert response.status == 404 + + +async def test_a_request_without_a_session_cookie_is_refused( + aiohttp_client: AiohttpClientFactory, +) -> None: + client = await aiohttp_client(build_test_api(queue=FakeQueue(order=order()))) + + response = await client.post(place_url(), json={"place": "first"}) + + assert response.status == 401 + + +async def test_a_drag_the_queue_has_moved_out_from_under_is_a_conflict( + aiohttp_client: AiohttpClientFactory, +) -> None: + """409, and it carries the queue as it now is. + + The request was well formed and the person may make it; what changed + is the state, which is what 409 says. A page told only "no" would + redraw from the list that caused the refusal and offer the same drag + again. + """ + refused = order( + accepted=False, + refusal="that session is no longer in this guild's queue", + changed=(), + ) + client = await signed_in(aiohttp_client, build_test_api(queue=FakeQueue(order=refused))) + + response = await client.post(place_url(), json={"place": "first"}) + + assert response.status == 409 + body = await response.json() + assert body["accepted"] is False + assert body["refusal"] == "that session is no longer in this guild's queue" + assert [row["session_id"] for row in body["order"]] == [str(SESSION), str(OTHER_SESSION)] + + +async def test_a_reorder_that_changed_nothing_says_so_rather_than_claiming_a_write( + aiohttp_client: AiohttpClientFactory, +) -> None: + """An empty `changed` is the difference between "done" and "nothing to do".""" + client = await signed_in( + aiohttp_client, build_test_api(queue=FakeQueue(order=order(changed=()))) + ) + + response = await client.post(place_url(), json={"place": "first"}) + + assert response.status == 200 + assert (await response.json())["changed"] == [] + + +# --------------------------------------------------------------------------- +# What a drag may say +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "body", + [ + pytest.param({}, id="nothing at all"), + pytest.param({"place": "somewhere"}, id="a placement nobody has"), + pytest.param({"place": 1}, id="a number where a placement goes"), + pytest.param({"place": "before"}, id="before nothing in particular"), + pytest.param({"place": "after"}, id="after nothing in particular"), + pytest.param( + {"place": "first", "session": str(OTHER_SESSION)}, id="an anchor that means nothing" + ), + pytest.param({"place": "before", "session": 512}, id="an id that is not a string"), + pytest.param({"place": "before", "session": "later"}, id="an id that is not an id"), + pytest.param({"place": "before", "session": str(SESSION)}, id="beside itself"), + ], +) +async def test_a_drag_that_does_not_say_where_is_refused_rather_than_guessed_at( + aiohttp_client: AiohttpClientFactory, body: dict[str, object] +) -> None: + """Strict, and never coerced -- the rule `_requested_model` already set. + + Every one of these is a client with a bug in it. A server that picked + a sensible interpretation would hide that bug until the day the + sensible interpretation stopped matching what the client meant, and by + then the queue would be in an order nobody chose. + """ + queue = FakeQueue(order=order()) + client = await signed_in(aiohttp_client, build_test_api(queue=queue)) + + response = await client.post(place_url(), json=body) + + assert response.status == 400 + assert queue.placed == [] + + +async def test_a_body_that_is_not_json_is_refused_without_being_echoed_back( + aiohttp_client: AiohttpClientFactory, +) -> None: + queue = FakeQueue(order=order()) + client = await signed_in(aiohttp_client, build_test_api(queue=queue)) + + response = await client.post( + place_url(), data="not json at all", headers={"Content-Type": "application/json"} + ) + + assert response.status == 400 + assert (await response.json())["error"] == "malformed request body" + assert queue.placed == [] + + +async def test_a_drag_with_no_body_at_all_is_refused( + aiohttp_client: AiohttpClientFactory, +) -> None: + """Unlike a re-queue, where an absent body is the button and means "no choice". + + A reorder with no body has not said what to do, and there is no + sensible default for where something goes. + """ + queue = FakeQueue(order=order()) + client = await signed_in(aiohttp_client, build_test_api(queue=queue)) + + response = await client.post(place_url()) + + assert response.status == 400 + assert queue.placed == [] + + +# --------------------------------------------------------------------------- +# The quick actions +# --------------------------------------------------------------------------- + + +async def test_an_administrator_can_reorder_a_whole_guilds_queue_by_a_rule( + aiohttp_client: AiohttpClientFactory, +) -> None: + queues = FakeQueueOverview(order=order()) + client = await signed_in(aiohttp_client, build_test_api(queues=queues)) + + response = await client.post(guild_url(), json={"rule": "many-participants-first"}) + + assert response.status == 200 + assert queues.reprioritised == [(GUILD, ANNA, "many-participants-first")] + assert (await response.json())["order"][0]["session_id"] == str(SESSION) + + +async def test_the_other_quick_action_the_owner_asked_for_is_there_too( + aiohttp_client: AiohttpClientFactory, +) -> None: + queues = FakeQueueOverview(order=order()) + client = await signed_in(aiohttp_client, build_test_api(queues=queues)) + + response = await client.post(guild_url(), json={"rule": "short-recordings-first"}) + + assert response.status == 200 + + +async def test_a_rule_nobody_has_is_refused_before_the_queue_is_touched( + aiohttp_client: AiohttpClientFactory, +) -> None: + """400, and the message names both what was asked for and what there is. + + The registry is closed and its names are literals of this repository, + so echoing the mistyped one back discloses nothing -- and running some + other rule instead would reorder a guild's queue in a way nobody could + tell from the feature working. + """ + queues = FakeQueueOverview(order=order()) + client = await signed_in(aiohttp_client, build_test_api(queues=queues)) + + response = await client.post(guild_url(), json={"rule": "longest-first"}) + + assert response.status == 400 + assert "longest-first" in (await response.json())["error"] + assert queues.reprioritised == [] + + +async def test_a_quick_action_without_a_rule_is_refused_rather_than_defaulted( + aiohttp_client: AiohttpClientFactory, +) -> None: + queues = FakeQueueOverview(order=order()) + client = await signed_in(aiohttp_client, build_test_api(queues=queues)) + + response = await client.post(guild_url(), json={}) + + assert response.status == 400 + assert queues.reprioritised == [] + + +async def test_somebody_who_does_not_administer_the_guild_is_told_it_does_not_exist_either( + aiohttp_client: AiohttpClientFactory, +) -> None: + client = await signed_in(aiohttp_client, build_test_api(queues=FakeQueueOverview())) + + response = await client.post(guild_url(), json={"rule": "many-participants-first"}) + + assert response.status == 404 + assert (await response.json())["error"] == "no such guild" + + +async def test_a_guild_id_that_is_not_a_number_gets_the_same_refusal( + aiohttp_client: AiohttpClientFactory, +) -> None: + client = await signed_in( + aiohttp_client, build_test_api(queues=FakeQueueOverview(order=order())) + ) + + response = await client.post( + "/api/guilds/nonsense/queue/priority", json={"rule": "many-participants-first"} + ) + + assert response.status == 404 + + +async def test_nothing_in_between_may_cache_a_queue_order( + aiohttp_client: AiohttpClientFactory, +) -> None: + """It names which meetings a guild has outstanding, behind a session cookie.""" + client = await signed_in( + aiohttp_client, build_test_api(queues=FakeQueueOverview(order=order())) + ) + + response = await client.post(guild_url(), json={"rule": "short-recordings-first"}) + + assert response.headers["Cache-Control"] == "private, no-store" + + +async def test_every_id_in_an_order_travels_as_a_string( + aiohttp_client: AiohttpClientFactory, +) -> None: + """A JSON number loses a snowflake's last digits; one id shape or none.""" + client = await signed_in(aiohttp_client, build_test_api(queue=FakeQueue(order=order()))) + + body = await (await client.post(place_url(), json={"place": "first"})).json() + + assert all(isinstance(row["session_id"], str) for row in body["order"]) + assert all(isinstance(session_id, str) for session_id in body["changed"]) diff --git a/tests/console/test_queue_routes.py b/tests/console/test_queue_routes.py index 44ce440..c3447ea 100644 --- a/tests/console/test_queue_routes.py +++ b/tests/console/test_queue_routes.py @@ -275,6 +275,7 @@ def queued(**over: object) -> QueuedSession: "running": 1, "done": 0, "dead": 0, + "priority": 0, } base.update(over) return QueuedSession(**base) # type: ignore[arg-type] @@ -315,6 +316,37 @@ async def test_an_administrator_sees_what_their_guild_still_owes( assert body["sessions"][0]["counts"]["pending"] == 2 +async def test_the_overview_says_where_each_session_sits_in_the_queue( + aiohttp_client: AiohttpClientFactory, +) -> None: + """The number an administrator just wrote has to come back, or the page + cannot render the order it set.""" + overview = FakeQueueOverview(queue=guild_queue(sessions=(queued(priority=2),))) + client = await signed_in(aiohttp_client, build_test_api(queues=overview)) + + body = await (await client.get(guild_url())).json() + + assert body["sessions"][0]["priority"] == 2 + + +async def test_a_session_with_nothing_queued_has_no_place_rather_than_the_ordinary_one( + aiohttp_client: AiohttpClientFactory, +) -> None: + """Null, not zero, and the difference is what a page acts on. + + Zero is the ordinary priority and a real place in the queue. A meeting + that is still recording has no jobs at all, so it has no place -- and + a row reported as `0` would be a row offering a drag handle that + nothing can be reordered about. + """ + overview = FakeQueueOverview(queue=guild_queue(sessions=(queued(priority=None),))) + client = await signed_in(aiohttp_client, build_test_api(queues=overview)) + + body = await (await client.get(guild_url())).json() + + assert body["sessions"][0]["priority"] is None + + async def test_the_overview_asks_on_behalf_of_the_signed_in_person( aiohttp_client: AiohttpClientFactory, ) -> None: diff --git a/tests/console/test_queue_stream_routes.py b/tests/console/test_queue_stream_routes.py index e24277e..7851f6e 100644 --- a/tests/console/test_queue_stream_routes.py +++ b/tests/console/test_queue_stream_routes.py @@ -94,6 +94,7 @@ def queued(**over: object) -> QueuedSession: "running": 0, "done": 0, "dead": 0, + "priority": 0, } base.update(over) return QueuedSession(**base) # type: ignore[arg-type] @@ -281,6 +282,32 @@ async def test_a_change_in_the_queue_arrives_as_an_event( assert '"pending":1,"running":2' in sent[1] +async def test_a_reorder_alone_is_a_change_the_stream_sends( + aiohttp_client: AiohttpClientFactory, +) -> None: + """Nothing moved, nothing finished -- only the order changed. + + The stream sends an event when the serialised snapshot differs, and + priority being part of that snapshot is what makes one administrator's + drag arrive on another administrator's open page. Without it the two + would disagree about the order until something else happened to change + the counts. + """ + overview = ScriptedOverview( + guild_queue(sessions=(queued(id=1, priority=0), queued(id=2, priority=0))), + guild_queue(sessions=(queued(id=1, priority=1), queued(id=2, priority=0))), + at_rest(), + ) + client = await signed_in(aiohttp_client, streaming_api(queues=overview)) + + body = await (await client.get(guild_stream_url())).text() + sent = [block.replace(" ", "") for block in data_blocks(body)] + + assert len(sent) == 3 + assert '"priority":0' in sent[0] + assert '"priority":1' in sent[1] + + async def test_a_stream_with_nothing_to_say_keeps_the_connection_alive_with_a_comment( aiohttp_client: AiohttpClientFactory, ) -> None: diff --git a/tests/infrastructure/test_priority.py b/tests/infrastructure/test_priority.py new file mode 100644 index 0000000..922e0d9 --- /dev/null +++ b/tests/infrastructure/test_priority.py @@ -0,0 +1,425 @@ +"""Writing a queue order, against a real PostgreSQL. + +The pure arithmetic is `tests/application/test_priorities.py`. What is +tested here is everything the database has an opinion about: that a +session's jobs move together, that the read sees what the claim would see, +that a guild's write cannot reach another guild's rows, and that two +administrators reordering at the same instant produce one coherent order +rather than a blend of two. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Sequence +from datetime import UTC, datetime, timedelta + +import pytest +from sqlalchemy import delete, select, update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from sturnus.application.priorities import ( + Placement, + QueuedSession, + order_by_rule, + order_with, + resolve_rule, +) +from sturnus.infrastructure.db.models import Base, Session, TranscriptionJob +from sturnus.infrastructure.db.priority import Decision, load_queued_sessions, reorder +from sturnus.infrastructure.db.repositories import JobRepository, SessionRepository + +T0 = datetime(2026, 8, 23, 9, 0, 0, tzinfo=UTC) +GUILD, CHANNEL = 1, 2 +OTHER_GUILD, OTHER_CHANNEL = 11, 12 +ANNA, BEN, CARLA, DORA = 100, 200, 300, 400 + + +@pytest.fixture +async def factory(clean_database: str) -> async_sessionmaker[AsyncSession]: + engine = create_async_engine(clean_database) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + return async_sessionmaker(engine, expire_on_commit=False) + + +async def seed( + factory: async_sessionmaker[AsyncSession], + speakers: list[int], + guild: int = GUILD, + channel: int = CHANNEL, + started_at: datetime = T0, +) -> int: + sessions = SessionRepository(factory) + jobs = JobRepository(factory) + session_id = await sessions.open_session(guild, channel, "meeting-raum", started_at) + for user_id in speakers: + await sessions.add_participant(session_id, user_id, f"user{user_id}", started_at) + await jobs.enqueue( + session_id=session_id, + discord_user_id=user_id, + s3_key=f"sessions/{session_id}/speakers/{user_id}.enc", + encryption_key_id="k1", + wrapped_data_key=b"wrapped", + retention_until=started_at + timedelta(days=30), + ) + await sessions.close_session(session_id, started_at + timedelta(hours=1), "empty") + return session_id + + +async def priorities_of(factory: async_sessionmaker[AsyncSession], session_id: int) -> list[int]: + async with factory() as db: + rows = await db.execute( + select(TranscriptionJob.priority) + .where(TranscriptionJob.session_id == session_id) + .order_by(TranscriptionJob.id) + ) + return [row.priority for row in rows] + + +async def finish( + factory: async_sessionmaker[AsyncSession], + session_id: int, + *, + status: str = "done", + audio_seconds: float | None = None, +) -> None: + """Takes a session's jobs out of the queue, optionally measuring them.""" + values: dict[str, object] = {"status": status} + if audio_seconds is not None: + values["audio_seconds"] = audio_seconds + async with factory() as db: + await db.execute( + update(TranscriptionJob) + .where(TranscriptionJob.session_id == session_id) + .values(**values) + ) + await db.commit() + + +async def measure( + factory: async_sessionmaker[AsyncSession], session_id: int, audio_seconds: float +) -> None: + async with factory() as db: + await db.execute( + update(TranscriptionJob) + .where(TranscriptionJob.session_id == session_id) + .values(audio_seconds=audio_seconds) + ) + await db.commit() + + +def drag(session_id: int, placement: Placement) -> Decision: + """The decision a drag makes, as `reorder` takes it.""" + + def decide(sessions: Sequence[QueuedSession]) -> tuple[int, ...] | None: + return order_with(sessions, session_id, placement) + + return decide + + +def quick_action(name: str) -> Decision: + rule = resolve_rule(name) + + def decide(sessions: Sequence[QueuedSession]) -> tuple[int, ...] | None: + return order_by_rule(sessions, rule) + + return decide + + +# --------------------------------------------------------------------------- +# What the reader sees +# --------------------------------------------------------------------------- + + +async def test_a_queue_nobody_has_touched_reads_as_ordinary_and_oldest_first( + factory: async_sessionmaker[AsyncSession], +) -> None: + first = await seed(factory, [ANNA]) + second = await seed(factory, [BEN]) + + queued = await load_queued_sessions(factory, GUILD) + + assert [row.id for row in queued] == [first, second] + assert [row.priority for row in queued] == [0, 0] + + +async def test_a_session_counts_the_people_who_were_in_it( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await seed(factory, [ANNA, BEN, CARLA]) + + queued = await load_queued_sessions(factory, GUILD) + + assert [row.participants for row in queued] == [3] + assert session_id == queued[0].id + + +async def test_a_session_nothing_has_measured_has_no_length_rather_than_none_of_it( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Null, never zero. The quick action's whole correctness rests on it.""" + await seed(factory, [ANNA]) + + queued = await load_queued_sessions(factory, GUILD) + + assert queued[0].audio_seconds is None + + +async def test_a_sessions_length_is_the_audio_its_tracks_were_measured_at( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Summed across speakers, because that is the work the queue still owes.""" + session_id = await seed(factory, [ANNA, BEN]) + await measure(factory, session_id, 120.0) + + queued = await load_queued_sessions(factory, GUILD) + + assert queued[0].audio_seconds == 240.0 + + +async def test_a_session_with_nothing_outstanding_is_not_in_the_queue( + factory: async_sessionmaker[AsyncSession], +) -> None: + """A documented meeting has no place in the queue and cannot be dragged.""" + finished = await seed(factory, [ANNA]) + waiting = await seed(factory, [BEN]) + await finish(factory, finished) + + queued = await load_queued_sessions(factory, GUILD) + + assert [row.id for row in queued] == [waiting] + + +async def test_the_queue_of_one_guild_never_shows_another_guilds_sessions( + factory: async_sessionmaker[AsyncSession], +) -> None: + mine = await seed(factory, [ANNA], guild=GUILD, channel=CHANNEL) + await seed(factory, [BEN], guild=OTHER_GUILD, channel=OTHER_CHANNEL) + + queued = await load_queued_sessions(factory, GUILD) + + assert [row.id for row in queued] == [mine] + + +# --------------------------------------------------------------------------- +# The write +# --------------------------------------------------------------------------- + + +async def test_dragging_a_session_to_the_front_puts_it_there( + factory: async_sessionmaker[AsyncSession], +) -> None: + first = await seed(factory, [ANNA]) + second = await seed(factory, [BEN]) + third = await seed(factory, [CARLA]) + + result = await reorder(factory, GUILD, drag(third, Placement("first"))) + + assert result is not None + assert result.order == (third, first, second) + assert [row.id for row in await load_queued_sessions(factory, GUILD)] == [ + third, + first, + second, + ] + + +async def test_every_speaker_of_a_meeting_moves_together( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The unit is the session, and this is the reason it has to be. + + The rows are one per speaker. A write that moved some of a meeting's + speakers and not the others would leave a queue that is half + reordered, which no page can render and nobody would ever see. + """ + crowded = await seed(factory, [ANNA, BEN, CARLA, DORA]) + await seed(factory, [ANNA]) + + await reorder(factory, GUILD, drag(crowded, Placement("last"))) + + assert await priorities_of(factory, crowded) == [1, 1, 1, 1] + + +async def test_an_order_that_already_holds_writes_nothing( + factory: async_sessionmaker[AsyncSession], +) -> None: + first = await seed(factory, [ANNA]) + second = await seed(factory, [BEN]) + + result = await reorder(factory, GUILD, drag(second, Placement("after", anchor=first))) + + assert result is not None + assert result.changed == () + assert await priorities_of(factory, first) == [0] + assert await priorities_of(factory, second) == [0] + + +async def test_a_reorder_never_touches_another_guilds_queue( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The `WHERE` that names the guild is the whole of this rule. + + Priority is one number in one column shared by every guild in the + deployment, so a write that forgot its guild would silently reorder + somebody else's meetings. + """ + mine = await seed(factory, [ANNA], guild=GUILD, channel=CHANNEL) + also_mine = await seed(factory, [BEN], guild=GUILD, channel=CHANNEL) + theirs = await seed(factory, [CARLA], guild=OTHER_GUILD, channel=OTHER_CHANNEL) + + await reorder(factory, GUILD, drag(also_mine, Placement("first"))) + + assert await priorities_of(factory, theirs) == [0] + assert await priorities_of(factory, mine) == [1] + + +async def test_a_drag_of_a_session_that_has_left_the_queue_is_refused( + factory: async_sessionmaker[AsyncSession], +) -> None: + """It finished while the page was open. Refused, and nothing is written.""" + gone = await seed(factory, [ANNA]) + still_here = await seed(factory, [BEN]) + await finish(factory, gone) + + result = await reorder(factory, GUILD, drag(gone, Placement("first"))) + + assert result is None + assert await priorities_of(factory, still_here) == [0] + + +async def test_a_drag_against_another_guilds_session_is_refused( + factory: async_sessionmaker[AsyncSession], +) -> None: + """An anchor is looked up in this guild's queue and nowhere else.""" + mine = await seed(factory, [ANNA], guild=GUILD, channel=CHANNEL) + theirs = await seed(factory, [BEN], guild=OTHER_GUILD, channel=OTHER_CHANNEL) + + result = await reorder(factory, GUILD, drag(mine, Placement("after", anchor=theirs))) + + assert result is None + + +async def test_only_the_outstanding_jobs_of_a_session_are_renumbered( + factory: async_sessionmaker[AsyncSession], +) -> None: + """A finished job's priority means nothing and is left alone. + + Its transcript is written; nothing will ever claim it again. Writing a + queue position onto it would be recording an intention about work that + is over. + """ + partly_done = await seed(factory, [ANNA, BEN]) + await seed(factory, [CARLA]) + async with factory() as db: + first_job = await db.scalar( + select(TranscriptionJob.id) + .where(TranscriptionJob.session_id == partly_done) + .order_by(TranscriptionJob.id) + ) + await db.execute( + update(TranscriptionJob).where(TranscriptionJob.id == first_job).values(status="done") + ) + await db.commit() + + await reorder(factory, GUILD, drag(partly_done, Placement("last"))) + + assert await priorities_of(factory, partly_done) == [0, 1] + + +# --------------------------------------------------------------------------- +# The quick actions, end to end +# --------------------------------------------------------------------------- + + +async def test_the_biggest_meeting_is_moved_to_the_front( + factory: async_sessionmaker[AsyncSession], +) -> None: + small = await seed(factory, [ANNA]) + large = await seed(factory, [ANNA, BEN, CARLA]) + + result = await reorder(factory, GUILD, quick_action("many-participants-first")) + + assert result is not None + assert result.order == (large, small) + + +async def test_the_shortest_measured_recording_is_moved_to_the_front( + factory: async_sessionmaker[AsyncSession], +) -> None: + long_one = await seed(factory, [ANNA]) + short_one = await seed(factory, [BEN]) + await measure(factory, long_one, 900.0) + await measure(factory, short_one, 90.0) + + result = await reorder(factory, GUILD, quick_action("short-recordings-first")) + + assert result is not None + assert result.order == (short_one, long_one) + + +async def test_an_unmeasured_recording_is_not_promoted_by_what_nobody_knows( + factory: async_sessionmaker[AsyncSession], +) -> None: + unmeasured = await seed(factory, [ANNA]) + measured = await seed(factory, [BEN]) + await measure(factory, measured, 300.0) + + result = await reorder(factory, GUILD, quick_action("short-recordings-first")) + + assert result is not None + assert result.order == (measured, unmeasured) + + +# --------------------------------------------------------------------------- +# Two administrators at once +# --------------------------------------------------------------------------- + + +async def test_two_reorders_at_the_same_instant_produce_one_coherent_order( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The property this endpoint has to have, and the only way to show it. + + Two administrators drag two different sessions to the front at the + same instant, on two connections, against a real PostgreSQL. Each + drag is decided *inside* the lock, from the queue as it stands at that + moment, so the second is applied to what the first left rather than to + the list its browser was showing. + + The assertion is exact on purpose. There are two ways these two calls + can serialise and each leaves a different set of numbers behind, so + the result matching one of them is what rules out the failure that + matters: two decisions taken from the same snapshot and both written, + which would leave an order neither administrator asked for and which + obeys neither instruction. + + Repeated, because a race that fires once in twenty runs fires in + production. + """ + for _ in range(20): + await wipe(factory) + first = await seed(factory, [ANNA]) + second = await seed(factory, [BEN]) + third = await seed(factory, [CARLA]) + + await asyncio.gather( + reorder(factory, GUILD, drag(third, Placement("first"))), + reorder(factory, GUILD, drag(second, Placement("first"))), + ) + + queued = await load_queued_sessions(factory, GUILD) + written = {row.id: row.priority for row in queued} + assert written in ( + # The third session went first, then the second overtook it. + {second: 1, third: 1, first: 2}, + # The other way round. + {third: 1, second: 2, first: 3}, + ), written + + +async def wipe(factory: async_sessionmaker[AsyncSession]) -> None: + async with factory() as db: + await db.execute(delete(Session)) + await db.commit() diff --git a/tests/infrastructure/test_queue.py b/tests/infrastructure/test_queue.py index ab01162..2a17916 100644 --- a/tests/infrastructure/test_queue.py +++ b/tests/infrastructure/test_queue.py @@ -3,14 +3,14 @@ from typing import Any import pytest -from sqlalchemy import delete, func, select +from sqlalchemy import delete, func, select, text, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sturnus.domain import settings from sturnus.domain.measurements import JobMeasurements, RecordedAudio from sturnus.infrastructure.db.config_store import ConfigStore from sturnus.infrastructure.db.models import Base, GuildConfig, Session, TranscriptionJob -from sturnus.infrastructure.db.queue import JobQueue +from sturnus.infrastructure.db.queue import DEFAULT_LEASE_SECONDS, JobQueue, claim_statement from sturnus.infrastructure.db.repositories import JobRepository, SessionRepository from sturnus.infrastructure.telemetry import JOB_OUTCOME, record @@ -931,3 +931,138 @@ async def test_a_worker_that_lost_its_job_does_not_stamp_the_file_it_measured( assert stored is not None assert stored.sample_rate is None assert stored.stored_bytes is None + + +# --------------------------------------------------------------------------- +# `priority`: what an administrator said should run first +# +# The column and its index landed in migration 0013 with nothing reading +# them. These are the tests that make the claim read them -- lower first, +# ties still broken by id, and the plan still one forward scan of +# `ix_job_claim_order` rather than a sort over the whole queue. +# --------------------------------------------------------------------------- + + +async def set_priority( + factory: async_sessionmaker[AsyncSession], session_id: int, priority: int +) -> None: + """Puts one session's jobs at a priority, the way the console's write does.""" + async with factory() as session: + await session.execute( + update(TranscriptionJob) + .where(TranscriptionJob.session_id == session_id) + .values(priority=priority) + ) + await session.commit() + + +async def test_a_raised_session_is_claimed_before_an_older_ordinary_one( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The whole point of the column, and the story it was asked for. + + The retrospective was enqueued second, so oldest-first alone would put + it behind the recording nobody is waiting on. Lower first: the + ordinary session is held back to `1` and the retrospective, still at + the ordinary `0`, is claimed first. + """ + ordinary = await seed(factory, [ANNA], guild=GUILD, channel=CHANNEL) + retrospective = await seed(factory, [BEN], guild=GUILD, channel=CHANNEL) + await set_priority(factory, ordinary, 1) + queue = JobQueue(factory) + + claimed = [job for _ in range(2) if (job := await queue.claim()) is not None] + + assert [job.session_id for job in claimed] == [retrospective, ordinary] + + +async def test_a_session_held_back_runs_after_a_meeting_that_ended_later( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Holding one back is how going first is expressed; it must actually work. + + The held-back session's jobs are the *oldest* in the queue, which is + the case a claim that merely ordered by id would get wrong. + """ + held_back = await seed(factory, [ANNA, BEN], guild=GUILD, channel=CHANNEL) + await set_priority(factory, held_back, 2) + later = await seed(factory, [CARLA], guild=OTHER_GUILD, channel=OTHER_CHANNEL) + queue = JobQueue(factory) + + claimed = [job for _ in range(3) if (job := await queue.claim()) is not None] + + assert [job.session_id for job in claimed] == [later, held_back, held_back] + + +async def test_two_sessions_at_the_same_priority_still_run_oldest_first( + factory: async_sessionmaker[AsyncSession], +) -> None: + """First-in-first-out within a priority, which the index gives for free.""" + first = await seed(factory, [ANNA], guild=GUILD, channel=CHANNEL) + second = await seed(factory, [BEN], guild=GUILD, channel=CHANNEL) + await set_priority(factory, first, 5) + await set_priority(factory, second, 5) + queue = JobQueue(factory) + + claimed = [job for _ in range(2) if (job := await queue.claim()) is not None] + + assert [job.session_id for job in claimed] == [first, second] + + +async def test_no_index_this_schema_has_can_order_a_claim_by_priority( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The plan, asked rather than assumed -- and it is not the one 0013 said. + + Migration 0013 added `ix_job_claim_order (status, priority, id)` and + said `ORDER BY priority, id` would be one forward scan of it. It is + not: `status` leads that index and a claim matches two values of it, + so the scan yields two ordered runs and PostgreSQL sorts them. The + ordering is correct and costs exactly what ordering by `id` alone + already cost; the index is simply not earning it. + + Sequential *and* bitmap scans are switched off here, which leaves the + ordered index scan as the only plan available -- so a `Sort` under + those conditions is not the planner preferring something on a + twelve-row table, it is the planner having no way to avoid one. + + The assertion is deliberately about the sort and not about which index + is chosen. With a handful of rows and a predicate on `status` alone, + `ix_job_status` and `ix_job_claim_order` are equally good and the + cheaper one wins by size; that choice is noise and a test asserting it + would fail on the next row inserted. The sort is not noise. + + **This test fails the day somebody adds the partial index + `claim_statement` recommends, and that is the point.** Read that + docstring before changing anything here: ordering by `status` first + would also make it pass, and would silently restore Defect 4. + """ + await seed(factory, [ANNA, BEN], guild=GUILD, channel=CHANNEL) + await seed(factory, [CARLA], guild=OTHER_GUILD, channel=OTHER_CHANNEL) + + plan = await explain_a_claim(factory) + + assert "Sort Key: transcription_job.priority, transcription_job.id" in plan, plan + + +async def explain_a_claim(factory: async_sessionmaker[AsyncSession]) -> str: + """PostgreSQL's plan for the statement `claim` runs, as text. + + The statement comes from `JobQueue` itself rather than being rewritten + here, because a plan for a hand-copied query would settle something + about the copy. + """ + statement = claim_statement(T0 - timedelta(seconds=DEFAULT_LEASE_SECONDS)) + async with factory() as session: + connection = await session.connection() + # Compiled against the dialect that will run it, with the values + # written in: `EXPLAIN` takes a statement, not a statement and a + # bag of parameters, and a plan for the wrong dialect's rendering + # of `FOR UPDATE SKIP LOCKED` would be a plan for another query. + compiled = statement.compile( + dialect=connection.dialect, compile_kwargs={"literal_binds": True} + ) + await session.execute(text("SET LOCAL enable_seqscan = off")) + await session.execute(text("SET LOCAL enable_bitmapscan = off")) + rows = await session.execute(text(f"EXPLAIN {compiled}")) + return "\n".join(str(row[0]) for row in rows)