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)