From 846c677d4b88bc86f6371298cc0fb9e7adfae8f0 Mon Sep 17 00:00:00 2001 From: TheMeinerLP Date: Sun, 23 Aug 2026 15:57:30 +0200 Subject: [PATCH] feat(worker): compute a spectrogram once, and delete it when the audio goes A spectrogram is drawn per request today: a full streamed decrypt of the recording plus 600 transforms, every time somebody opens a track, keeping nothing. `spectrograms_by_default` (per guild, default false) moves that work to the worker, which draws the picture at job completion out of the plaintext WAV it already has on disk and stores it beside the audio, sealed under the same session data key. Storing it is only defensible with one rule attached, and the rule is the design: a stored spectrogram is deleted when its audio is deleted. The retention sweep now deletes both objects in the same pass and stamps `audio_deleted_at` only once both are gone, because a picture of when somebody spoke and for how long that outlived their recording's retention window would be the setting quietly creating exactly what the window exists to end. After a sweep the track is neither playable nor visualisable, as before, and that is pinned by a test. The read path prefers the artefact and falls back to drawing the track -- for a job from before the setting was on, for a guild that never turned it on, and for an artefact that is missing or unreadable -- so the endpoint's contract does not depend on which answered. Authorisation is untouched and still decided per request, before anything looks in the bucket: a cache of the payload must not become a cache of the permission. A failure to draw or store never fails the job. The transcription is the valuable part, it has already succeeded by then, and the console can draw the track itself. The FFT moves to `sturnus.application.spectrogram` so that two processes share one definition of the picture rather than growing a second one, and `CorruptRecording` moves to `sturnus.domain.errors` because two layers now raise it and `application` may not import the console. --- docs/operations.md | 106 +++- src/sturnus/application/recording.py | 24 + src/sturnus/application/retention.py | 68 ++- src/sturnus/application/spectrogram.py | 461 ++++++++++++++++++ src/sturnus/application/worker.py | 190 +++++++- src/sturnus/console/adapters.py | 6 + src/sturnus/console/audio.py | 18 +- src/sturnus/console/ports.py | 13 + src/sturnus/console/routes_audio.py | 76 ++- src/sturnus/console/spectrogram.py | 432 ++++------------ src/sturnus/domain/errors.py | 41 +- src/sturnus/domain/settings.py | 34 ++ src/sturnus/entrypoints/worker.py | 62 ++- src/sturnus/infrastructure/db/repositories.py | 34 +- src/sturnus/observability/events.py | 19 + tests/application/test_retention.py | 61 +++ tests/application/test_spectrogram.py | 166 +++++++ tests/application/test_worker.py | 262 +++++++++- tests/console/conftest.py | 6 + tests/console/test_audio.py | 42 +- tests/console/test_audio_routes.py | 193 ++++++++ tests/console/test_spectrogram.py | 7 +- tests/infrastructure/test_repositories.py | 54 ++ tests/infrastructure/test_traced_ports.py | 4 + tests/observability/test_no_payload_leaks.py | 3 + 25 files changed, 1975 insertions(+), 407 deletions(-) create mode 100644 src/sturnus/application/spectrogram.py create mode 100644 tests/application/test_spectrogram.py diff --git a/docs/operations.md b/docs/operations.md index 0dfaf0f..d00fb17 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -700,6 +700,14 @@ in — from the console. Turning it on asserts that the document at 6.2.10 before setting it**: unlike `video_consent_offered`, this one grants access rather than withholding it. +`spectrograms_by_default` is `true` or `false`, defaults to `false`, and +is not about consent at all: it decides whether the worker draws each +track's spectrogram once and stores it beside the recording, instead of +the console drawing one per view. It stores something new about every +recording, so it starts off; what it stores is deleted with the audio it +was drawn from. **Read section 6.2.13 before setting it** — it has a +storage cost and a rule attached. + ### 4.0 `voice_channel_ids`, and the key it replaced `voice_channel_ids` names every voice channel Sturnus is allowed to record @@ -778,8 +786,8 @@ when it detects this. invocation, per API request, and by the consent cache with a five-second TTL), and `transcription_language`, `transcription_prompt`, `document_target`, `document_provider`, -`merge_gap_seconds` (read per job by the *worker* process, not the bot at -all), and `max_parallel_tracks` (read by the worker *inside* the claim +`merge_gap_seconds`, `spectrograms_by_default` (read per job by the +*worker* process, not the bot at all), and `max_parallel_tracks` (read by the worker *inside* the claim query itself, so it governs the very next claim — see 4.2). The two transcription keys apply to the next job the worker claims, which means a session already recording is still transcribed with the new @@ -1280,6 +1288,16 @@ needs to: a bucket lifecycle rule is a second, independent line of defence, never a substitute for that database record, and never a replacement for this sweep. +**The sweep deletes the recording's spectrogram in the same pass**, when +the guild asked for stored ones (`spectrograms_by_default`, §6.2.13) and +there is one. A spectrogram shows when somebody spoke and for how long; +leaving it behind would make this setting a way for something about a +person's voice to outlive the window their recording was subject to, and +this sweep is the only thing in the system that ends a recording's life. +`audio_deleted_at` is stamped only once both objects are gone, so a +partial failure is retried on the next sweep rather than recorded as +done. + Because recordings outlive their transcription by weeks, not minutes, the retention period is not merely an implementation detail — **it belongs in the privacy policy shown to participants** (the `policy_url` document), @@ -2233,6 +2251,90 @@ clearing its secret, each emit `console.oauth_client_changed` at `registered`, `secret_set`, `secret_cleared` or `removed`. Neither half of the credential is on the line — not the secret, and not the client id. +### 6.2.13 `spectrograms_by_default`: drawing each track's picture once + +The console can show a track as a spectrogram — where the speech is, and +whether the capture worked at all (§6.3 explains what the picture is for). +Drawing one costs a full streamed decrypt of the recording plus 600 +transforms, and by default that happens on **every view**: nothing is +kept. + +`spectrograms_by_default` moves the work to the worker. When it is `true`, +a job that finishes transcribing draws its own track's picture — from the +plaintext WAV it already has on disk, which exists nowhere else and for no +longer than that job — and stores it beside the recording in the same +bucket, encrypted under the same session data key. A view then costs two +small object reads and no arithmetic. + +#### What it stores, and for how long + +A picture is a fixed 600 × 128 cells, one byte each: **76.8 kB of matrix, +about 100 kB stored** once it is base64'd into its small self-describing +envelope. Fixed means fixed — a three-hour workshop costs exactly what a +two-minute stand-up does, because the number of columns does not depend on +the length of the recording. + +Per meeting that is about 100 kB × the number of people who spoke: a +six-person meeting is 0.6 MB. A guild holding four such meetings every +working day accumulates roughly 1.2 GB over a year — *if* nothing were +ever deleted. Nothing here is kept that long: an artefact lives exactly as +long as the recording it was drawn from, so the steady-state cost is a +`audio_retention_days` window of them, not a year of them. At the default +thirty days that is about 100 MB for the guild above, against the tens of +gigabytes its audio occupies over the same window. The picture is a +rounding error beside the recording, in storage. It is not a rounding +error in what it says. + +#### The rule this setting is subject to + +> **A stored spectrogram is deleted when its audio is deleted.** + +The retention sweep (§6) deletes both objects in the same pass, and stamps +`audio_deleted_at` only once both are gone. This is not an optimisation +and it is not optional: a spectrogram shows when a person spoke and for +how long, and a picture left behind by a sweep that erased the recording +would be a record of somebody's voice activity outliving the retention +window their recording — and their consent — was subject to. After a +sweep the track is neither playable nor visualisable, exactly as it was +before this setting existed. + +#### Turning it on + +``` +/config set key:spectrograms_by_default value:true +``` + +or the console's settings page. Values are `true` and `false` and nothing +else, for the reason §6.2.10 gives. `/config clear` restores the default, +which is `false`. It takes effect **immediately**: the worker reads it per +job, and nothing caches it. + +**Nothing is backfilled.** Jobs already transcribed have no artefact and +will not acquire one by this setting being switched on; the console goes +on drawing those tracks on demand, at the old cost, indefinitely. A +session that is re-queued (§6.2.9) is transcribed again by a worker that +reads the setting again, so a redo does produce one. If you want pictures +for a specific old session, re-queueing it is the way, and it costs a full +re-transcription — which is rarely worth it for a picture. + +#### Turning it off + +Existing artefacts **stay**, and this is deliberate. Turning the setting +off says "stop drawing new ones", not "destroy what has been drawn": the +pictures already stored are still governed by the rule above, so each one +is still deleted with its own recording and none of them outlives the +window it was created under. Deleting them on a config change would be a +second, unreviewable deletion path for the same objects — and one that +runs at the moment somebody is editing a form, which is the wrong moment +for a bulk delete of anything. + +What changes immediately is the cost: jobs finishing from now on store +nothing, and views of tracks that already have a picture go on being +answered from it until retention takes both away. If you need the stored +pictures gone sooner than their recordings, shorten +`audio_retention_days` — that moves the recording and the picture +together, which is the only way this system moves them. + ### 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/recording.py b/src/sturnus/application/recording.py index 2658536..53b69f5 100644 --- a/src/sturnus/application/recording.py +++ b/src/sturnus/application/recording.py @@ -46,6 +46,30 @@ def audio_key(session_id: int, discord_user_id: int) -> str: return f"sessions/{session_id}/speakers/{discord_user_id}.enc" +def spectrogram_key(session_id: int, discord_user_id: int) -> str: + """Object key for the stored picture of one speaker's recording. + + Beside the audio and derived from the same two ids, so an operator + looking at the bucket can see at a glance which recording an artefact + belongs to -- and so that a prefix listing of a session shows + everything that session put in the bucket, which is what makes "is + anything left?" a question the bucket can answer after a retention + sweep. + + Still `.enc`, because it is: the picture is sealed under the same + session data key the recording is. What is stored is a rendering of + somebody's voice activity, and an object in this bucket that is + readable without the master key would be the only one. + + The key the worker actually used is nonetheless written down on the + job (`transcription_job.spectrogram_key`) rather than recomputed from + this rule at deletion time. The rule can change; a bucket full of + objects written under the old one cannot, and the retention sweep must + delete what was written rather than what would be written today. + """ + return f"sessions/{session_id}/speakers/{discord_user_id}.spectrogram.enc" + + class SessionRecorder(Protocol): """What `RecordingService` needs to persist about a session's rows.""" diff --git a/src/sturnus/application/retention.py b/src/sturnus/application/retention.py index 4c12ab9..b920e96 100644 --- a/src/sturnus/application/retention.py +++ b/src/sturnus/application/retention.py @@ -8,8 +8,13 @@ `sweep_expired_audio` is the periodic sweep that actually calls it: it reads candidates through `JobStore`, filters them with `expired_jobs`, then -- -for each -- deletes the object through `AudioDeleter` and stamps -`audio_deleted_at` as the durable evidence that the deletion happened. The +for each -- deletes the recording *and its stored spectrogram* through +`AudioDeleter` and stamps `audio_deleted_at` as the durable evidence that +the deletion happened. Both objects, in one pass, because a spectrogram +that outlived the recording it was drawn from would be a rendering of +somebody's voice activity surviving the retention window that voice was +subject to -- see `sweep_expired_audio` and +`sturnus.domain.settings.SPECTROGRAMS_BY_DEFAULT`. The bucket lifecycle rule (Spec 12.2) is a second line of defence, never a substitute for that record. `JobStore`/`AudioDeleter` are narrow local `Protocol`s rather than concrete types, the same pattern @@ -63,28 +68,61 @@ async def candidates_for_retention(self) -> list[dict[str, object]]: of it anywhere in the codebase, the same reasoning `sturnus.application.publishing.sessions_to_announce`'s caller follows for `status`/`announced_at`/`document_url`. + + Every candidate carries `spectrogram_key` -- `None` for the jobs + that have no stored picture, which is most of them and all of them + for a guild that never switched `spectrograms_by_default` on. It + is selected rather than derived from the job's ids because the + naming rule may change and the objects already in the bucket + cannot: this sweep must delete what was written. """ ... - async def mark_audio_deleted(self, job_id: int, now: datetime) -> None: ... + async def mark_audio_deleted(self, job_id: int, now: datetime) -> None: + """Stamps the deletion, and forgets where the picture was. + + `spectrogram_key` is cleared in the same statement. The column + says where this job's artefact is, and once the sweep has deleted + it there is no artefact for it to point at; leaving the key behind + would make every later sweep re-delete an object that is already + gone and would leave the row claiming a picture exists. + """ + ... class AudioDeleter(Protocol): - """Where an expired recording's object is actually removed.""" + """Where an expired recording's objects are actually removed. + + One method for both, because both are objects in the same bucket + under the same credentials, and a port with a second method for the + artefact would invite a caller to delete one kind and not the other. + """ async def delete(self, key: str) -> None: ... async def sweep_expired_audio(jobs: JobStore, store: AudioDeleter, now: datetime) -> None: - """Deletes every expired job's audio object and stamps `audio_deleted_at`. + """Deletes every expired job's audio, and its spectrogram, and stamps the row. + + **A stored spectrogram is deleted when its audio is deleted, in the + same pass.** That rule is the whole reason storing one is defensible. + A spectrogram is a rendering of when somebody spoke and for how long; + it is less than the audio and it is not nothing, and this sweep is the + only thing in the system that ends a recording's life. If it deleted + the object and left the picture, `spectrograms_by_default` would be a + switch that quietly makes something about a person's voice outlive the + retention window that person's recording was subject to -- which is + what the window is for. So the picture goes with it, here, rather than + in a second sweep that could be forgotten, disabled, or fail on its + own. Survives its own errors per job: a failure deleting one job's audio - (or stamping it afterwards) is logged and does not stop the sweep from - handling the rest -- one unreachable object must not block every other - job's retention from being enforced. + (or its picture, or stamping it afterwards) is logged and does not + stop the sweep from handling the rest -- one unreachable object must + not block every other job's retention from being enforced. - `audio_deleted_at` is stamped only after `store.delete` actually - succeeds, so a failed deletion is retried on the next sweep instead of + `audio_deleted_at` is stamped only after both deletions actually + succeed, so a failed deletion is retried on the next sweep instead of being silently recorded as done. The reverse order (stamp then delete) would risk exactly the outcome `audio_deleted_at` exists to rule out -- a stamp claiming deletion happened when it did not. The cost of the @@ -92,11 +130,21 @@ async def sweep_expired_audio(jobs: JobStore, store: AudioDeleter, now: datetime being retried once more next sweep; `store.delete` on an already-missing key is idempotent (an S3 `DELETE` on a missing object still succeeds), so that retry costs nothing. + + The audio goes first and the picture second, which matters only in the + one case where the sweep is interrupted between them: what is left + behind is then a picture whose audio is gone, and a row still asking + to be swept. The console refuses a track whose object has been erased + before it looks for a picture at all, so that interval is invisible + from outside and ends on the next sweep. """ for job in expired_jobs(await jobs.candidates_for_retention(), now): job_id = cast(int, job["id"]) try: await store.delete(cast(str, job["s3_key"])) + picture = cast("str | None", job["spectrogram_key"]) + if picture is not None: + await store.delete(picture) await jobs.mark_audio_deleted(job_id, now) except Exception as exc: log_exception( diff --git a/src/sturnus/application/spectrogram.py b/src/sturnus/application/spectrogram.py new file mode 100644 index 0000000..c1f2aa6 --- /dev/null +++ b/src/sturnus/application/spectrogram.py @@ -0,0 +1,461 @@ +"""Turning one track into a picture of itself, and the form that picture is stored in. + +A listener who opens a recording wants to know where the speech is before +they decide what to play. A waveform answers that badly -- a constant +noise floor and a spoken sentence reach a similar peak -- and a +spectrogram answers it well, because speech has a shape nothing else in a +voice channel has: harmonic stacks under about 4 kHz, moving at syllable +rate. + +It also answers a question the operator scripts were invented for. The +defect that made every recording sound like noise (see +`sturnus.console.audio`, fact 1) is *visible* here: audio played at six +times speed puts its energy in the wrong bands, and a track that is +genuinely empty is flat. Being able to see that in the console is the +difference between "the capture is broken" and "nobody spoke", which cost +a production investigation to tell apart by ear. + +**Why this lives in `application` rather than beside the console route it +was written for.** Two processes draw the same picture now. The console +computes one per request out of the ciphertext in S3 +(`sturnus.console.spectrogram`); the worker computes one at job completion +out of the plaintext WAV it already has on disk +(`sturnus.application.worker`), for a guild that asked for +`spectrograms_by_default`. A second FFT path would be a second answer to +"where is the speech in this track" -- rendered identically, drifting +silently -- so there is one, below both callers, and it takes a stream of +plaintext WAV bytes because that is all either caller can promise. + +**Computed in one pass with bounded memory, never buffered whole.** An +hour of one speaker is about 115 MB of 16 kHz mono, and a handful of +concurrent viewers holding that in a pod would be an outage. The output +size is fixed up front -- `COLUMNS` by `BINS` -- so the arithmetic runs +the other way: the hop between windows is derived from the track's length, +each window is read as it streams past, and everything behind it is +dropped. Peak memory is one piece of the stream plus one window. + +**Nothing here touches the disk**, in either direction. The console's +copy of that promise is a static test +(`tests/console/test_audio.py::test_nothing_on_the_serving_path_can_write_plaintext_to_disk`), +which lists this module alongside the console's own: decrypted voice on +its way to a socket must exist only in memory, and a module shared by the +serving path is on the serving path. + +The response is deliberately a picture rather than numbers to interpret: +one byte per cell, 0 for the noise floor and 255 for the loudest cell in +*this* track, base64 over the whole matrix. That normalisation is a +decision. Absolute dBFS would render every quiet-but-fine recording as an +empty rectangle, and the question this view exists to answer is "where is +the speech in this track", not "how loud was it". +""" + +from __future__ import annotations + +import base64 +import json +import struct +from collections.abc import AsyncGenerator, AsyncIterator +from dataclasses import dataclass + +import numpy as np + +from sturnus.domain.errors import CorruptRecording + +#: Time slices across the whole track. Fixed rather than proportional, so +#: the response is the same size for a two-minute stand-up and a +#: three-hour workshop, and so the client can size a canvas before it has +#: the data. +COLUMNS = 600 + +#: Samples per FFT window. At 16 kHz this is 64 ms -- long enough to +#: resolve a voiced fundamental (about 85 Hz for a low voice, and 1024 +#: samples is twelve of its cycles), short enough that a syllable is not +#: smeared across the window. +WINDOW = 1024 + +#: Frequency rows per slice. **`WINDOW // 2` must be an exact multiple of +#: this**, and that is the whole reason for the number: 512 usable FFT +#: points over 128 rows is exactly 4 points each, so every row spans the +#: same width in Hz and row `r` starts at `r * sample_rate / WINDOW * 4`. +#: An axis a client can label with arithmetic instead of a lookup table. +#: +#: An earlier 96 divided 512 into a mix of five- and six-point rows, which +#: renders identically and makes the frequency axis a lie by up to half a +#: row -- the sort of quiet inaccuracy this file exists to make visible in +#: other people's data, so it does not get to have one of its own. +BINS = 128 + +#: The floor, in dB below this track's loudest cell. Everything at or +#: under it renders as 0. Sixty dB is the range a spectrogram is +#: conventionally drawn over: below that is the dither and the room, and +#: including it turns the picture grey. +DYNAMIC_RANGE_DB = 60.0 + +#: The magnitude a full-scale sine produces in one window: half the 16-bit +#: range, times the window length, over the Hann window's coherent gain of +#: one half. +_FULL_SCALE = 32768.0 * WINDOW / 4.0 + +#: The quietest peak still normalised against itself. Below this the track +#: is normalised against *this* value instead, so it renders as the empty +#: picture it is. +#: +#: Without it, "brightest cell becomes 255" has one catastrophic failure +#: mode: a track carrying nothing but resampler dither at -90 dBFS has a +#: brightest cell too, and stretching it over the full range draws a +#: convincing picture of a meeting that never happened. That is precisely +#: the wrong answer for the question this view exists to answer, and it is +#: how "the capture is broken" and "nobody spoke" became indistinguishable +#: in the first place. +#: +#: Set 60 dB below full scale, which is where `sturnus.domain.silence` also +#: puts the line (`SILENCE_PEAK_AMPLITUDE = 32`, about -60 dBFS): the bot +#: and the console then agree on what counts as a silent recording. +_SILENCE_FLOOR = _FULL_SCALE * 10.0 ** (-DYNAMIC_RANGE_DB / 20.0) + +#: Below this, a cell is drawn as empty no matter what the rest of the +#: track looks like. The relative floor above is not enough on its own: +#: it moves the *reference*, and a cell can still sit within +#: `DYNAMIC_RANGE_DB` of a reference that is itself the floor. +#: +#: Measured, not guessed. `to_mono_16k` turns digital silence into +#: ±1 LSB of resampler dither -- audible to nothing, but it reaches about +#: -111 dBFS in a window, which is only 51 dB under the relative floor and +#: therefore *visible* without this. A -40 dBFS tone, far quieter than any +#: speech worth keeping, sits at -41 dBFS. Eighty decibels is comfortably +#: between the two: thirty above the dither, forty below the quietest +#: thing anybody meant to record. +_NOISE_MAGNITUDE = _FULL_SCALE * 10.0 ** (-80.0 / 20.0) + +#: The canonical header `SpeakerWriter` writes, and the smallest prefix +#: that can describe a track at all. +_MIN_HEADER = 44 + +#: What a stored artefact says it is. Bumped when the *meaning* of the +#: fields changes rather than when `COLUMNS` or `BINS` do -- those two +#: are checked on their own by `decode_artefact`, which can say exactly +#: what it disagreed with. +ARTEFACT_VERSION = 1 + + +@dataclass(frozen=True) +class TrackFormat: + """What a track's own RIFF header says it is. + + Read rather than assumed, and that is the whole lesson of the format + defect this module was written after: `sturnus.console.audio` used to + *state* the sample rate and was wrong by a factor of three. A file + that describes itself is only useful to a reader that asks. + """ + + sample_rate: int + channels: int + sample_width: int + data_offset: int + data_bytes: int + + @property + def frame_bytes(self) -> int: + return self.channels * self.sample_width + + @property + def frames(self) -> int: + return self.data_bytes // self.frame_bytes + + @property + def duration_seconds(self) -> float: + return self.frames / self.sample_rate if self.sample_rate else 0.0 + + +@dataclass(frozen=True) +class Spectrogram: + """One track as a picture, plus the axes needed to label it.""" + + columns: int + bins: int + sample_rate: int + duration_seconds: float + #: Row-major, `bins` rows of `columns` bytes, row 0 the lowest + #: frequency. Base64 because the client's destination is an + #: `ImageData` buffer, and a JSON array of 76 800 numbers is several + #: times the bytes to say the same thing. + magnitudes: str + + @property + def hz_per_bin(self) -> float: + """The width of one row, which is what labels the frequency axis.""" + return self.sample_rate / 2 / self.bins + + +def parse_track_format(head: bytes) -> TrackFormat: + """Reads the RIFF header a track begins with. + + Walks the chunk list rather than trusting the canonical 44-byte + layout. `SpeakerWriter` writes exactly that layout today, but a reader + that hardcodes an offset is how this system got a six-times-speed + playback bug in the first place, and walking costs a dozen lines. + """ + if len(head) < _MIN_HEADER or head[:4] != b"RIFF" or head[8:12] != b"WAVE": + raise CorruptRecording("track does not begin with a RIFF/WAVE header") + + offset = 12 + fmt: tuple[int, int, int] | None = None + while offset + 8 <= len(head): + chunk_id = head[offset : offset + 4] + (size,) = struct.unpack_from(" Spectrogram: + """Streams one plaintext WAV past an FFT and returns the picture. + + The track is read once, forwards, and never held: `_windows` yields + one window per output column as the bytes go by, so peak memory does + not grow with the length of the meeting. + + `pieces` is consumed, never closed. Whoever opened the stream -- an + S3 body, a file on the worker's scratch disk -- is the only party that + knows what closing it costs, and a reader that closed its caller's + generator would take that decision away from them. + """ + columns: list[np.ndarray] = [] + fmt: TrackFormat | None = None + async for window, track_format in _windows(pieces): + fmt = track_format + columns.append(_column(window)) + + if fmt is None: + raise CorruptRecording("track ended before its header was complete") + return _render(columns, fmt) + + +def encode_artefact(picture: Spectrogram) -> bytes: + """One picture as the bytes a stored spectrogram consists of. + + JSON rather than a bespoke binary layout, and self-describing rather + than positional, because the artefact outlives the process that wrote + it: a stored picture whose shape can only be recovered by knowing + which release drew it is a picture that cannot be safely read after + the next change to `COLUMNS`. `decode_artefact` checks what it finds + against what this build draws and refuses a mismatch, which is only + possible because the numbers are written down here. + + The base64 the endpoint answers with is stored as-is rather than + decoded and re-encoded on every read. It costs a third more bytes than + the raw matrix and saves the read path from doing arithmetic on + somebody's voice to hand back exactly what it was given. + """ + return json.dumps( + { + "version": ARTEFACT_VERSION, + "columns": picture.columns, + "bins": picture.bins, + "sample_rate": picture.sample_rate, + "duration_seconds": picture.duration_seconds, + "magnitudes": picture.magnitudes, + }, + separators=(",", ":"), + ).encode("utf-8") + + +def decode_artefact(raw: bytes) -> Spectrogram: + """A stored artefact as a picture, or `CorruptRecording`. + + Every disagreement is the same refusal, because every disagreement has + the same remedy: the caller draws the track again. That is what makes + this strict rather than forgiving -- an artefact drawn by a build with + a different `COLUMNS` still parses, still renders, and describes the + track wrongly by however much the shape moved, and the client sizes a + canvas from numbers it was handed rather than from what it received. + + An artefact is never *repaired* here and never partially trusted. The + audio it was drawn from is still in the bucket at this point (the read + path checks that before it looks for a picture at all), so falling + back costs a recomputation and loses nothing. + """ + try: + document = json.loads(raw) + except ValueError as exc: + raise CorruptRecording("stored spectrogram is not the document it should be") from exc + if not isinstance(document, dict): + raise CorruptRecording("stored spectrogram is not the document it should be") + if document.get("version") != ARTEFACT_VERSION: + raise CorruptRecording("stored spectrogram was written by a different artefact version") + if document.get("columns") != COLUMNS or document.get("bins") != BINS: + raise CorruptRecording("stored spectrogram has a shape this build does not draw") + + magnitudes = document.get("magnitudes") + sample_rate = document.get("sample_rate") + duration = document.get("duration_seconds") + if not isinstance(magnitudes, str) or not isinstance(sample_rate, int): + raise CorruptRecording("stored spectrogram is missing what a picture is made of") + if not isinstance(duration, int | float): + raise CorruptRecording("stored spectrogram is missing what a picture is made of") + try: + cells = base64.b64decode(magnitudes, validate=True) + except ValueError as exc: + raise CorruptRecording("stored spectrogram's matrix is not base64") from exc + if len(cells) != COLUMNS * BINS: + raise CorruptRecording("stored spectrogram holds the wrong number of cells") + + return Spectrogram( + columns=COLUMNS, + bins=BINS, + sample_rate=sample_rate, + duration_seconds=float(duration), + magnitudes=magnitudes, + ) + + +async def _windows( + pieces: AsyncIterator[bytes], +) -> AsyncGenerator[tuple[np.ndarray, TrackFormat], None]: + """Yields `(window, format)` once per output column, in file order. + + The hop is derived from the declared data length, so the columns span + the whole track regardless of how long it is. A track shorter than one + window yields a single, zero-padded column rather than nothing -- a + two-second recording still has a picture, and an empty rectangle would + read as a failure it is not. + """ + head = bytearray() + fmt: TrackFormat | None = None + consumed = 0 # frames of audio already passed + buffer = np.empty(0, dtype=np.int16) + wanted = 0 # index of the next column + hop = 0 + total = 0 + + async for piece in pieces: + if fmt is None: + head += piece + if len(head) < _MIN_HEADER: + continue + fmt = parse_track_format(bytes(head)) + if fmt.sample_width != 2: + raise CorruptRecording("only 16-bit tracks can be drawn") + total = fmt.frames + hop = max(1, (total - WINDOW) // max(1, COLUMNS - 1)) if total > WINDOW else WINDOW + body = bytes(head[fmt.data_offset :]) + head = bytearray() + else: + body = piece + + if not body: + continue + buffer = np.concatenate([buffer, _mono(body, fmt)]) + + # Emit every column whose window is now fully inside the buffer. + while wanted < COLUMNS: + start = wanted * hop + if start + WINDOW > consumed + len(buffer): + break + if start >= total: + wanted = COLUMNS + break + local = start - consumed + yield buffer[local : local + WINDOW].astype(np.float32), fmt + wanted += 1 + + # Drop everything no future column can reach. + keep_from = max(0, wanted * hop - consumed) + if keep_from > 0: + buffer = buffer[keep_from:] + consumed += keep_from + if wanted >= COLUMNS: + break + + if fmt is None: + return + # A track shorter than one window, or a last column that ran past the + # end: pad rather than drop, so short recordings still draw. + while wanted < COLUMNS and wanted * hop < max(total, 1): + start = wanted * hop + local = max(0, start - consumed) + tail = buffer[local : local + WINDOW] + if len(tail) == 0: + break + padded = np.zeros(WINDOW, dtype=np.float32) + padded[: len(tail)] = tail + yield padded, fmt + wanted += 1 + + +def _mono(body: bytes, fmt: TrackFormat) -> np.ndarray: + """The samples in `body` as one channel, whatever the track has.""" + usable = len(body) - (len(body) % fmt.frame_bytes) + if usable <= 0: + return np.empty(0, dtype=np.int16) + samples = np.frombuffer(body[:usable], dtype=" np.ndarray: + """One time slice as `BINS` magnitudes, lowest frequency first. + + Hann-windowed, because a rectangular window on a voiced frame spreads + its harmonics across the whole picture and the harmonic stack is + precisely what makes speech recognisable here. + + The frequency axis is linear rather than mel or log. A mel axis is + better for showing a *word*; a linear one is better for showing the + fault this view exists to expose, because a track played at the wrong + rate is a stack that has moved bodily up the axis, and only a linear + axis makes that a translation rather than a distortion. + """ + spectrum = np.abs(np.fft.rfft(window * np.hanning(len(window)))) + # `rfft` returns `WINDOW/2 + 1` points; dropping DC leaves exactly + # `WINDOW/2`, which `BINS` divides evenly (see its comment). Each row + # is the *peak* of its group rather than the mean, which keeps a + # narrow harmonic visible instead of averaging it into the floor + # beside it -- and the harmonic stack is what identifies speech here. + group = WINDOW // 2 // BINS + return np.maximum.reduceat( + spectrum[1 : 1 + WINDOW // 2], np.arange(0, WINDOW // 2, group) + ).astype(np.float32) + + +def _render(columns: list[np.ndarray], fmt: TrackFormat) -> Spectrogram: + """Normalises the collected slices into one byte per cell.""" + matrix = np.stack(columns, axis=1) if columns else np.zeros((BINS, 1), dtype=np.float32) + + # Normalised against the loudest cell *or* the silence floor, whichever + # is louder. For any real recording the peak wins and nothing changes; + # for a track with no signal in it the floor wins, every cell lands + # more than `DYNAMIC_RANGE_DB` below it, and the picture is empty -- + # which is the truth about that track. + reference = max(float(matrix.max()), _SILENCE_FLOOR) + decibels = 20.0 * np.log10(np.maximum(matrix, 1e-12) / reference) + clipped = np.clip(decibels, -DYNAMIC_RANGE_DB, 0.0) + scaled = ((clipped + DYNAMIC_RANGE_DB) / DYNAMIC_RANGE_DB * 255.0).astype(np.uint8) + # The absolute cut, applied last so it wins over the relative scale. + scaled[matrix < _NOISE_MAGNITUDE] = 0 + + # Padded to the promised width so the client can treat the payload as + # a fixed-size image. A short track is short because it is short, and + # the columns it does not have are floor. + if scaled.shape[1] < COLUMNS: + scaled = np.pad(scaled, ((0, 0), (0, COLUMNS - scaled.shape[1]))) + + return Spectrogram( + columns=COLUMNS, + bins=BINS, + sample_rate=fmt.sample_rate, + duration_seconds=round(fmt.duration_seconds, 3), + magnitudes=base64.b64encode(scaled.tobytes()).decode("ascii"), + ) diff --git a/src/sturnus/application/worker.py b/src/sturnus/application/worker.py index 7ede646..ddd788b 100644 --- a/src/sturnus/application/worker.py +++ b/src/sturnus/application/worker.py @@ -13,6 +13,23 @@ transcription can be redone from the original audio; that deletion belongs to the retention sweep (`sturnus.application.retention`), not to this job. +**The spectrogram, for a guild that asked for one.** `spectrograms_by_default` +makes this function draw each track's picture once, at completion, instead +of leaving the console to draw it again out of S3 on every view. It is done +here for one reason: this is the only moment in the system where the +plaintext WAV is free. It exists on disk for the length of one job and then +the `finally` removes it, and anybody who wants the same picture afterwards +has to fetch and decrypt the whole recording to get it. + +That stored picture comes with a rule, which is not this module's to enforce +but is this module's to understand, because it is why the artefact's key is +written onto the job rather than derived when it is needed: **a stored +spectrogram is deleted when its audio is deleted**. The retention sweep +deletes what `transcription_job.spectrogram_key` names, in the same pass as +the recording. Nothing else would ever delete it, and a picture of when +somebody spoke and for how long that outlives their recording's retention +window is exactly the thing the window exists to end. + Language (Spec 7, Spec 11). Two things can decide what language a recording is transcribed as, and the order between them is the whole point. `transcription_language` is per-guild configuration and wins @@ -98,6 +115,7 @@ import time import uuid import wave +from collections.abc import AsyncGenerator from datetime import UTC, datetime, tzinfo from pathlib import Path from typing import Protocol, cast @@ -119,6 +137,8 @@ destinations_for, publish_session, ) +from sturnus.application.recording import spectrogram_key +from sturnus.application.spectrogram import draw, encode_artefact from sturnus.application.transcription import TranscriptionEngine from sturnus.domain import settings as domain_settings from sturnus.domain.measurements import JobMeasurements, RecordedAudio @@ -202,6 +222,51 @@ class Decryptor(Protocol): def decrypt_to(self, source: Path, target: Path, wrapped: bytes, key_id: str) -> None: ... +class SpectrogramStore(Protocol): + """Where a track's stored picture is written down and put. + + Two methods rather than one, and **an order between them that is the + point of the port**: `record` first, `put` second. The reverse order + has one failure mode this design cannot afford -- an object in the + bucket that no row names. The retention sweep deletes the artefact + `transcription_job.spectrogram_key` points at, so an artefact nothing + points at is an artefact nothing will ever delete: a rendering of + somebody's voice activity outliving the recording it was drawn from, + which is precisely what the rule attached to this feature forbids. + + Written in this order the worst case is the harmless one. A `record` + that succeeds and a `put` that fails leaves a job naming an object + that is not there, the read path finds nothing and draws the track + itself (`sturnus.console.routes_audio.track_spectrogram`), and the + sweep asks the store to delete a key that was never written -- which + an S3 `DELETE` answers successfully, exactly as it does for the audio + object the same sweep may have deleted twice. + + Two backends behind one port because the two writes are one act. A + port offering only half of it is a port that will eventually be used + to do half of it. + """ + + async def record(self, job_id: int, key: str) -> None: + """Writes the artefact's key onto the job, before the object exists.""" + ... + + async def put_sealed(self, key: str, source: Path, wrapped: bytes, key_id: str) -> None: + """Seals `source` under the job's own data key and stores it. + + Sealed rather than stored as it is: the artefact is a rendering of + somebody's voice activity, which is why the console puts it behind + the same authorisation rule as the audio, and an object in this + bucket readable by anybody holding the bucket would be the single + exception to envelope encryption in the whole system. + + `wrapped`/`key_id` are the job's own, so the picture and the + recording are locked with the same key and a master-key rotation + reaches both or neither. + """ + ... + + class ConfigReader(Protocol): """Where per-guild runtime configuration is read from (Spec 11). @@ -540,6 +605,98 @@ async def _create_session_document( ) +#: How much of the plaintext WAV is handed to the FFT at a time. The +#: number is not about memory -- `draw` keeps one window regardless -- it +#: is about giving the event loop the chance to run between pieces. +#: Drawing an hour-long track is on the order of a second of arithmetic, +#: and a worker that did all of it without an `await` in the middle would +#: stop answering its own health probe while it did. +_WAV_PIECE_BYTES = 256 * 1024 + + +async def _wav_pieces(path: Path) -> AsyncGenerator[bytes, None]: + """Yields the decrypted WAV off the scratch disk, a piece at a time. + + Each read is a thread hop for the same reason every other blocking + call in this process is (`sturnus.infrastructure.objectstore`): the + file is on a pod's ephemeral disk, and a synchronous read of it inside + the loop is a stall nothing can preempt. + """ + with path.open("rb") as handle: + while piece := await asyncio.to_thread(handle.read, _WAV_PIECE_BYTES): + yield piece + + +async def _store_spectrogram( + spectrograms: SpectrogramStore, + config: ConfigReader, + job: _ClaimedJobShape, + guild: int, + wav_path: Path, +) -> None: + """Draws this track's picture and stores it, if the guild asked for one. + + **Never raises.** A transcription that succeeded and could not be + drawn has done the valuable part: the transcript is stored, the job is + `done`, and the session's document will be written from it. Letting a + failed artefact reach `process_one`'s handler would return an + already-transcribed job to the queue and transcribe it again -- minutes + of GPU-less inference -- to retry a picture the console can draw for + itself in a second. So this logs and moves on, and the read path falls + back to computing. + + Drawn here rather than by a later sweep because *here* is the only + place the plaintext is free. `process_one`'s `finally` deletes the + decrypted WAV within milliseconds of this returning, and every other + place in the system that wants this picture has to fetch and decrypt + the whole object again to get it. + """ + try: + offered = await config.get(guild, domain_settings.SPECTROGRAMS_BY_DEFAULT) + if not domain_settings.is_true(offered): + return + + pieces = _wav_pieces(wav_path) + try: + picture = await draw(pieces) + finally: + await pieces.aclose() + + # Beside the WAV in the job's scratch directory, so the `finally` + # that removes decrypted speech removes this too -- an artefact is + # a rendering of the same voice and gets the same treatment. + artefact = wav_path.with_name("spectrogram.json") + body = encode_artefact(picture) + await asyncio.to_thread(artefact.write_bytes, body) + + key = spectrogram_key(job.session_id, job.discord_user_id) + # Recorded before it is stored. See `SpectrogramStore`: an object + # no row names is an object the retention sweep cannot delete. + await spectrograms.record(job.id, key) + await spectrograms.put_sealed(key, artefact, job.wrapped_data_key, job.encryption_key_id) + log_event( + log, + logging.INFO, + Event.SPECTROGRAM_STORED, + "Stored a spectrogram beside a recording", + job_id=job.id, + session_id=job.session_id, + discord_user_id=job.discord_user_id, + bytes=len(body), + ) + except Exception as exc: + log_exception( + log, + logging.WARNING, + Event.SPECTROGRAM_FAILED, + "Could not store a spectrogram; the console will draw this track on demand", + exc, + job_id=job.id, + session_id=job.session_id, + stage="spectrogram", + ) + + async def process_one( queue: Queue, engine: TranscriptionEngine, @@ -550,6 +707,7 @@ async def process_one( jobs: JobReader, links: LinkRepository, config: ConfigReader, + spectrograms: SpectrogramStore, work_dir: Path, max_attempts: int, template_source: str = _FALLBACK_TEMPLATE, @@ -568,17 +726,23 @@ async def process_one( guild asked for it (Spec 7; see the module docstring for the order and why it is that way round). 5. Store the transcript on the job; ask whether it was the session's last. - 6. If it was: assemble every participant's stored transcript into one - transcript (`_create_session_document`, `sturnus.application. - assembly.assemble`), publish it to every destination the guild has - enabled (`sturnus.application.exporting.publish_session`), and mark - the session documented from the primary one. - 7. Every temporary file made in steps 2-3 is removed in a `finally`, so + 6. If the guild asked for it (`spectrograms_by_default`): draw this + track's spectrogram from the WAV that is still on disk and store it + beside the audio (`_store_spectrogram`). Between step 5 and step 7 + because that is the only window in which the plaintext exists and + the transcript is already safe. + 7. If it was the session's last: assemble every participant's stored + transcript into one transcript (`_create_session_document`, + `sturnus.application.assembly.assemble`), publish it to every + destination the guild has enabled + (`sturnus.application.exporting.publish_session`), and mark the + session documented from the primary one. + 8. Every temporary file made in steps 2-3 is removed in a `finally`, so a failure anywhere above never leaves decrypted speech on disk. The audio object in S3 is left alone deliberately -- see the module docstring. - `jobs` and `links` are only ever read from in step 6, but are accepted + `jobs` and `links` are only ever read from in step 7, but are accepted as parameters up front (rather than constructed lazily) so every collaborator `process_one` needs is visible in its signature, matching `sessions`/`exports`/`queue` and the rest. @@ -600,7 +764,7 @@ async def process_one( selects `pending` jobs -- see `sturnus.infrastructure.db.queue.JobQueue` for the lease that also reclaims a job stranded this way). - Step 6 (publication) is deliberately handled by a *separate* + Step 7 (publication) is deliberately handled by a *separate* `try`/`except`, outside the one above: by the time it runs, `queue. complete` has already succeeded and the job is `done` -- calling `queue.fail` on it would incorrectly return an already-transcribed job @@ -741,6 +905,16 @@ async def process_one( lease=job.claimed_at, audio=_recorded_audio(wav_path, encrypted_path), ) + + # After the transcript is safely stored, and before the + # `finally` deletes the plaintext it needs. Both halves of + # that sentence are load-bearing: drawn any earlier, a + # failure here would travel to the handler below and re-queue + # a job that has already been transcribed; drawn any later, + # there is nothing left on disk to draw. `_store_spectrogram` + # swallows its own failures, which is what makes the first + # half true. + await _store_spectrogram(spectrograms, config, job, guild, wav_path) except Exception as exc: # Everything other than the transcription failure already # handled above: a failed download, a decrypt error, a diff --git a/src/sturnus/console/adapters.py b/src/sturnus/console/adapters.py index d78fa8e..008147f 100644 --- a/src/sturnus/console/adapters.py +++ b/src/sturnus/console/adapters.py @@ -482,6 +482,7 @@ async def track_for( TranscriptionJob.s3_key, TranscriptionJob.encryption_key_id, TranscriptionJob.wrapped_data_key, + TranscriptionJob.spectrogram_key, ).where( TranscriptionJob.session_id == session_id, TranscriptionJob.discord_user_id == speaker_id, @@ -501,6 +502,11 @@ async def track_for( s3_key=row.s3_key, encryption_key_id=row.encryption_key_id, wrapped_data_key=row.wrapped_data_key, + # Selected in the same statement, under the same + # `audio_deleted_at IS NULL`. A swept job offers neither + # its recording nor its picture, which is the whole point + # of deleting them together. + spectrogram_key=row.spectrogram_key, ) async def downloadable_track_for( diff --git a/src/sturnus/console/audio.py b/src/sturnus/console/audio.py index f1108b7..1fca79c 100644 --- a/src/sturnus/console/audio.py +++ b/src/sturnus/console/audio.py @@ -59,6 +59,7 @@ from cryptography.hazmat.primitives.ciphers.aead import AESGCM from sturnus.console.ports import EncryptedAudioSource, KeyUnwrapper, TrackDirectory +from sturnus.domain.errors import CorruptRecording from sturnus.infrastructure.crypto import ( CHUNK_SIZE, FILE_PREFIX_BYTES, @@ -76,15 +77,6 @@ _RANGE = re.compile(r"^\s*bytes\s*=\s*(?:(\d+)\s*-\s*(\d*)|-\s*(\d+))\s*$") -class CorruptRecording(Exception): - """The object in the bucket is not a recording this reader understands. - - Raised before any plaintext is produced -- a wrong magic, a truncated - upload, an object too short to be a recording at all. The alternative - to refusing is plausible-looking noise on somebody's speakers. - """ - - class UnsatisfiableRange(Exception): """The `Range` was understood and cannot be answered: 416, not 400.""" @@ -197,7 +189,13 @@ async def stream_wav( data_key: bytes, span: ByteRange, ) -> AsyncGenerator[bytes, None]: - """Yields `span` of one encrypted recording, decrypted and otherwise untouched. + """Yields `span` of one encrypted object, decrypted and otherwise untouched. + + Named for the recording it was written for, and true of any object + this system seals: `sturnus.console.spectrogram.stored_spectrogram` + reads a stored picture through it, because a picture beside a + recording in this bucket is sealed exactly as the recording is and + nothing here inspects what it is decrypting. `span` is an offset into the stored WAV file itself, which is both the plaintext and the served resource: there is no header to add and diff --git a/src/sturnus/console/ports.py b/src/sturnus/console/ports.py index 5cdcfa8..e4f1e50 100644 --- a/src/sturnus/console/ports.py +++ b/src/sturnus/console/ports.py @@ -505,6 +505,19 @@ class Track: s3_key: str encryption_key_id: str wrapped_data_key: bytes + #: Where this track's stored spectrogram is, or `None` for the tracks + #: that have none -- every job transcribed before its guild switched + #: `spectrograms_by_default` on, and every job of a guild that never + #: did. Optional here rather than a second lookup because the picture + #: is reached through the same authorisation as the audio and under + #: the same data key: a caller entitled to one is entitled to the + #: other, and a second query would be a second place to get that + #: wrong. + #: + #: A key here is a claim about the bucket, not a promise: the object + #: may be missing, and the reader draws the track itself when it is + #: (`sturnus.console.routes_audio.track_spectrogram`). + spectrogram_key: str | None = None @dataclass(frozen=True) diff --git a/src/sturnus/console/routes_audio.py b/src/sturnus/console/routes_audio.py index 39dda6b..5fb305f 100644 --- a/src/sturnus/console/routes_audio.py +++ b/src/sturnus/console/routes_audio.py @@ -66,17 +66,18 @@ from aiohttp import web +from sturnus.application.spectrogram import Spectrogram from sturnus.console.audio import ( AudioDelivery, ByteRange, - CorruptRecording, UnsatisfiableRange, parse_range, stored_length, stream_wav, ) from sturnus.console.ports import Track -from sturnus.console.spectrogram import spectrogram +from sturnus.console.spectrogram import spectrogram, stored_spectrogram +from sturnus.domain.errors import CorruptRecording from sturnus.observability.events import Event, log_event, log_exception log = logging.getLogger(__name__) @@ -287,6 +288,43 @@ async def _resolve(request: web.Request, rule: _Rule) -> _AuthorisedTrack | web. ) +async def _stored_picture( + delivery: AudioDelivery, resolved: _AuthorisedTrack, data_key: bytes +) -> Spectrogram | None: + """The artefact the worker drew, or `None` to draw the track instead. + + `None` covers three cases the caller must treat identically: this job + has no stored picture, the object it names is not in the bucket, and + the object is not a picture this build can read. All three have the + same remedy and the same cost, and none of them is a reason to refuse + somebody a view they are entitled to. + + Deliberately not `CorruptRecording`-transparent. A stored artefact + that will not decode is *not* the same event as a stored recording + that will not decrypt: the recording is the only copy of what somebody + said and its loss needs a human, while the artefact is a derived + convenience the next line of this handler recreates. Letting the + exception out would have answered 500 for a request that can be + answered perfectly well. + """ + key = resolved.track.spectrogram_key + if key is None: + return None + try: + return await stored_spectrogram(delivery.source, key, data_key) + except (KeyError, CorruptRecording) as exc: + log_exception( + log, + logging.INFO, + Event.CONSOLE_SPECTROGRAM_REDRAWN, + "A stored spectrogram could not be used; drawing this track instead", + exc, + session_id=resolved.session_id, + discord_user_id=resolved.speaker_id, + ) + return None + + async def track_spectrogram(request: web.Request) -> web.StreamResponse: """One track as a picture of where its speech is. @@ -295,6 +333,26 @@ async def track_spectrogram(request: web.Request) -> web.StreamResponse: they spoke and for how long. It is less than the audio; it is not nothing, and the rule that governs the audio is the right one for it. + **The gate is in front of the artefact too, and stays there.** A guild + that switched `spectrograms_by_default` on has a picture the worker + already drew, and answering from it makes this endpoint cheap; it must + not make it *open*. `_authorised_track` runs first and runs on every + request -- it is the same `session_participant` query, decided again, + never cached -- and only then does anything look for a stored picture. + A cache of the payload must never quietly become a cache of the + permission, and the ordering here is what keeps those two apart. + + It also runs first for a second reason: `_authorised_track` is what + refuses a track whose recording the retention sweep has erased. A + swept job offers no picture either, because the sweep deleted both -- + and because the row it would have come from no longer names one. + + **Either source answers the same thing.** A job from before the + setting was enabled, a guild that never enabled it, an artefact that + is missing or that a later build cannot read: all of them fall through + to drawing the track, which is what every view cost before artefacts + existed. The contract does not depend on which happened. + Answered as a whole small JSON body rather than streamed. The payload is a fixed 600 by 128 bytes whatever the meeting's length, so there is nothing to page through -- the streaming happens on the way *in*, past @@ -307,12 +365,14 @@ async def track_spectrogram(request: web.Request) -> web.StreamResponse: delivery = request.app[AUDIO_DELIVERY] data_key = delivery.keys.unwrap(resolved.track.wrapped_data_key) try: - picture = await spectrogram( - delivery.source, - resolved.track.s3_key, - data_key, - stored_length(resolved.ciphertext_bytes), - ) + picture = await _stored_picture(delivery, resolved, data_key) + if picture is None: + picture = await spectrogram( + delivery.source, + resolved.track.s3_key, + data_key, + stored_length(resolved.ciphertext_bytes), + ) except CorruptRecording as exc: log_exception( log, diff --git a/src/sturnus/console/spectrogram.py b/src/sturnus/console/spectrogram.py index 7dfc856..c02b620 100644 --- a/src/sturnus/console/spectrogram.py +++ b/src/sturnus/console/spectrogram.py @@ -1,203 +1,61 @@ -"""Turning one encrypted track into a picture of itself, without a file. - -A listener who opens a recording wants to know where the speech is before -they decide what to play. A waveform answers that badly -- a constant -noise floor and a spoken sentence reach a similar peak -- and a -spectrogram answers it well, because speech has a shape nothing else in a -voice channel has: harmonic stacks under about 4 kHz, moving at syllable -rate. - -It also answers a question the operator scripts were invented for. The -defect that made every recording sound like noise (see -`sturnus.console.audio`, fact 1) is *visible* here: audio played at six -times speed puts its energy in the wrong bands, and a track that is -genuinely empty is flat. Being able to see that in the console is the -difference between "the capture is broken" and "nobody spoke", which cost -a production investigation to tell apart by ear. - -**Computed in one pass with bounded memory, never buffered whole.** An -hour of one speaker is about 115 MB of 16 kHz mono, and a handful of -concurrent viewers holding that in a pod would be an outage. The output -size is fixed up front -- `COLUMNS` by `BINS` -- so the arithmetic runs -the other way: the hop between windows is derived from the track's length, -each window is read as it streams past, and everything behind it is -dropped. Peak memory is one chunk of ciphertext plus one window. +"""Getting a picture of a track out of the bucket, one way or the other. + +What the picture *is* -- the windows, the frequency rows, the +normalisation, the stored form -- lives in +`sturnus.application.spectrogram`, because the worker draws the same +picture from the same definition. What lives here is the pair of ways the +console can obtain one, and the difference between them is entirely a +matter of cost: + +**`spectrogram`** streams the encrypted track past the FFT and draws it +now. A full decrypt of the object plus `COLUMNS` transforms, per view. +That is what every view cost before a stored artefact existed, and it is +still what a guild that has not switched `spectrograms_by_default` on +pays -- along with every job transcribed before it did. + +**`stored_spectrogram`** reads the artefact the worker already drew. Two +object-store requests for about a hundred kilobytes, and no FFT at all. + +Both come back through the same `Spectrogram`, so the endpoint's response +does not depend on which one answered, and neither of them decides +*whether* the caller may have it: authorisation happens before either is +called and is re-decided on every request (`sturnus.console.routes_audio`). +A cache of the payload must never become a cache of the permission. + +**The artefact is read exactly like the audio is** -- same envelope, same +data key, same `stream_wav`. It is sealed because it is a rendering of +somebody's voice activity: less than the audio and not nothing, by the +same argument that puts it behind the same authorisation rule. An object +in this bucket that anybody holding the bucket could read would be the one +exception, and the one exception is the whole exposure. **Nothing here writes plaintext to disk**, for the same reason `sturnus.console.audio` does not, and pinned by the same static test in -`tests/console/test_audio.py`, which lists this module on the serving -path. - -The response is deliberately a picture rather than numbers to interpret: -one byte per cell, 0 for the noise floor and 255 for the loudest cell in -*this* track, base64 over the whole matrix. That normalisation is a -decision. Absolute dBFS would render every quiet-but-fine recording as an -empty rectangle, and the question this view exists to answer is "where is -the speech in this track", not "how loud was it". +`tests/console/test_audio.py`, which lists this module on the serving path. """ from __future__ import annotations -import base64 -import struct -from collections.abc import AsyncGenerator, AsyncIterator -from dataclasses import dataclass +from cryptography.exceptions import InvalidTag -import numpy as np - -from sturnus.console.audio import ByteRange, CorruptRecording, stream_wav +from sturnus.application.spectrogram import Spectrogram, decode_artefact, draw +from sturnus.console.audio import ByteRange, stored_length, stream_wav from sturnus.console.ports import EncryptedAudioSource +from sturnus.domain.errors import CorruptRecording -#: Time slices across the whole track. Fixed rather than proportional, so -#: the response is the same size for a two-minute stand-up and a -#: three-hour workshop, and so the client can size a canvas before it has -#: the data. -COLUMNS = 600 - -#: Samples per FFT window. At 16 kHz this is 64 ms -- long enough to -#: resolve a voiced fundamental (about 85 Hz for a low voice, and 1024 -#: samples is twelve of its cycles), short enough that a syllable is not -#: smeared across the window. -WINDOW = 1024 - -#: Frequency rows per slice. **`WINDOW // 2` must be an exact multiple of -#: this**, and that is the whole reason for the number: 512 usable FFT -#: points over 128 rows is exactly 4 points each, so every row spans the -#: same width in Hz and row `r` starts at `r * sample_rate / WINDOW * 4`. -#: An axis a client can label with arithmetic instead of a lookup table. -#: -#: An earlier 96 divided 512 into a mix of five- and six-point rows, which -#: renders identically and makes the frequency axis a lie by up to half a -#: row -- the sort of quiet inaccuracy this file exists to make visible in -#: other people's data, so it does not get to have one of its own. -BINS = 128 - -#: The floor, in dB below this track's loudest cell. Everything at or -#: under it renders as 0. Sixty dB is the range a spectrogram is -#: conventionally drawn over: below that is the dither and the room, and -#: including it turns the picture grey. -DYNAMIC_RANGE_DB = 60.0 - -#: The magnitude a full-scale sine produces in one window: half the 16-bit -#: range, times the window length, over the Hann window's coherent gain of -#: one half. -_FULL_SCALE = 32768.0 * WINDOW / 4.0 - -#: The quietest peak still normalised against itself. Below this the track -#: is normalised against *this* value instead, so it renders as the empty -#: picture it is. -#: -#: Without it, "brightest cell becomes 255" has one catastrophic failure -#: mode: a track carrying nothing but resampler dither at -90 dBFS has a -#: brightest cell too, and stretching it over the full range draws a -#: convincing picture of a meeting that never happened. That is precisely -#: the wrong answer for the question this view exists to answer, and it is -#: how "the capture is broken" and "nobody spoke" became indistinguishable -#: in the first place. -#: -#: Set 60 dB below full scale, which is where `sturnus.domain.silence` also -#: puts the line (`SILENCE_PEAK_AMPLITUDE = 32`, about -60 dBFS): the bot -#: and the console then agree on what counts as a silent recording. -_SILENCE_FLOOR = _FULL_SCALE * 10.0 ** (-DYNAMIC_RANGE_DB / 20.0) - -#: Below this, a cell is drawn as empty no matter what the rest of the -#: track looks like. The relative floor above is not enough on its own: -#: it moves the *reference*, and a cell can still sit within -#: `DYNAMIC_RANGE_DB` of a reference that is itself the floor. -#: -#: Measured, not guessed. `to_mono_16k` turns digital silence into -#: ±1 LSB of resampler dither -- audible to nothing, but it reaches about -#: -111 dBFS in a window, which is only 51 dB under the relative floor and -#: therefore *visible* without this. A -40 dBFS tone, far quieter than any -#: speech worth keeping, sits at -41 dBFS. Eighty decibels is comfortably -#: between the two: thirty above the dither, forty below the quietest -#: thing anybody meant to record. -_NOISE_MAGNITUDE = _FULL_SCALE * 10.0 ** (-80.0 / 20.0) - -#: The canonical header `SpeakerWriter` writes, and the smallest prefix -#: that can describe a track at all. -_MIN_HEADER = 44 - - -@dataclass(frozen=True) -class TrackFormat: - """What a track's own RIFF header says it is. +#: The smallest object that could hold a WAV header at all. Checked before +#: a byte is fetched, so an object that cannot be a track is refused +#: rather than half-read. +_MIN_TRACK_BYTES = 44 - Read rather than assumed, and that is the whole lesson of the format - defect this module was written after: `sturnus.console.audio` used to - *state* the sample rate and was wrong by a factor of three. A file - that describes itself is only useful to a reader that asks. - """ - - sample_rate: int - channels: int - sample_width: int - data_offset: int - data_bytes: int - - @property - def frame_bytes(self) -> int: - return self.channels * self.sample_width - - @property - def frames(self) -> int: - return self.data_bytes // self.frame_bytes - - @property - def duration_seconds(self) -> float: - return self.frames / self.sample_rate if self.sample_rate else 0.0 - - -@dataclass(frozen=True) -class Spectrogram: - """One track as a picture, plus the axes needed to label it.""" - - columns: int - bins: int - sample_rate: int - duration_seconds: float - #: Row-major, `bins` rows of `columns` bytes, row 0 the lowest - #: frequency. Base64 because the client's destination is an - #: `ImageData` buffer, and a JSON array of 76 800 numbers is several - #: times the bytes to say the same thing. - magnitudes: str - - @property - def hz_per_bin(self) -> float: - """The width of one row, which is what labels the frequency axis.""" - return self.sample_rate / 2 / self.bins - - -def parse_track_format(head: bytes) -> TrackFormat: - """Reads the RIFF header a track begins with. - - Walks the chunk list rather than trusting the canonical 44-byte - layout. `SpeakerWriter` writes exactly that layout today, but a reader - that hardcodes an offset is how this system got a six-times-speed - playback bug in the first place, and walking costs a dozen lines. - """ - if len(head) < _MIN_HEADER or head[:4] != b"RIFF" or head[8:12] != b"WAVE": - raise CorruptRecording("track does not begin with a RIFF/WAVE header") - - offset = 12 - fmt: tuple[int, int, int] | None = None - while offset + 8 <= len(head): - chunk_id = head[offset : offset + 4] - (size,) = struct.unpack_from(" Spectrogram: - """Streams one encrypted track past an FFT and returns the picture. + """Draws one encrypted track, now, without ever holding it. - The track is read once, forwards, and never held: `_windows` yields - one window per output column as the bytes go by, so peak memory does - not grow with the length of the meeting. + `stored_bytes` is the plaintext length the caller already derived from + the object's size (`stored_length`), so this does not go back to the + store to ask. """ - if stored_bytes < _MIN_HEADER: + if stored_bytes < _MIN_TRACK_BYTES: raise CorruptRecording("object is too short to hold a track") pieces = stream_wav(source, key, data_key, ByteRange(0, stored_bytes - 1)) - columns: list[np.ndarray] = [] - fmt: TrackFormat | None = None try: - async for window, track_format in _windows(pieces): - fmt = track_format - columns.append(_column(window, track_format.sample_rate)) + return await draw(pieces) finally: + # Closed here rather than in `draw`, which is shared with a caller + # whose stream is a file: a listener who navigated away should stop + # the transfer out of S3 in the same breath, and a suspended + # generator nobody closed holds that connection open until + # something else gets round to it. await pieces.aclose() - if fmt is None: - raise CorruptRecording("track ended before its header was complete") - return _render(columns, fmt) - - -async def _windows( - pieces: AsyncIterator[bytes], -) -> AsyncGenerator[tuple[np.ndarray, TrackFormat], None]: - """Yields `(window, format)` once per output column, in file order. - - The hop is derived from the declared data length, so the columns span - the whole track regardless of how long it is. A track shorter than one - window yields a single, zero-padded column rather than nothing -- a - two-second recording still has a picture, and an empty rectangle would - read as a failure it is not. - """ - head = bytearray() - fmt: TrackFormat | None = None - consumed = 0 # frames of audio already passed - buffer = np.empty(0, dtype=np.int16) - wanted = 0 # index of the next column - hop = 0 - total = 0 - - async for piece in pieces: - if fmt is None: - head += piece - if len(head) < _MIN_HEADER: - continue - fmt = parse_track_format(bytes(head)) - if fmt.sample_width != 2: - raise CorruptRecording("only 16-bit tracks can be drawn") - total = fmt.frames - hop = max(1, (total - WINDOW) // max(1, COLUMNS - 1)) if total > WINDOW else WINDOW - body = bytes(head[fmt.data_offset :]) - head = bytearray() - else: - body = piece - - if not body: - continue - buffer = np.concatenate([buffer, _mono(body, fmt)]) - - # Emit every column whose window is now fully inside the buffer. - while wanted < COLUMNS: - start = wanted * hop - if start + WINDOW > consumed + len(buffer): - break - if start >= total: - wanted = COLUMNS - break - local = start - consumed - yield buffer[local : local + WINDOW].astype(np.float32), fmt - wanted += 1 - - # Drop everything no future column can reach. - keep_from = max(0, wanted * hop - consumed) - if keep_from > 0: - buffer = buffer[keep_from:] - consumed += keep_from - if wanted >= COLUMNS: - break - - if fmt is None: - return - # A track shorter than one window, or a last column that ran past the - # end: pad rather than drop, so short recordings still draw. - while wanted < COLUMNS and wanted * hop < max(total, 1): - start = wanted * hop - local = max(0, start - consumed) - tail = buffer[local : local + WINDOW] - if len(tail) == 0: - break - padded = np.zeros(WINDOW, dtype=np.float32) - padded[: len(tail)] = tail - yield padded, fmt - wanted += 1 - - -def _mono(body: bytes, fmt: TrackFormat) -> np.ndarray: - """The samples in `body` as one channel, whatever the track has.""" - usable = len(body) - (len(body) % fmt.frame_bytes) - if usable <= 0: - return np.empty(0, dtype=np.int16) - samples = np.frombuffer(body[:usable], dtype=" np.ndarray: - """One time slice as `BINS` magnitudes, lowest frequency first. - - Hann-windowed, because a rectangular window on a voiced frame spreads - its harmonics across the whole picture and the harmonic stack is - precisely what makes speech recognisable here. - - The frequency axis is linear rather than mel or log. A mel axis is - better for showing a *word*; a linear one is better for showing the - fault this view exists to expose, because a track played at the wrong - rate is a stack that has moved bodily up the axis, and only a linear - axis makes that a translation rather than a distortion. +async def stored_spectrogram( + source: EncryptedAudioSource, + key: str, + data_key: bytes, +) -> Spectrogram: + """Reads the artefact the worker drew when the job finished. + + Raises `KeyError` when the object is not there and `CorruptRecording` + when it is not an artefact this build can read. The caller treats both + the same way -- draw the track instead -- because both have the same + remedy and the person waiting is owed a picture either way. + + A failed authentication tag is folded into the same refusal rather + than allowed out. On the audio path an `InvalidTag` means a recording + somebody is entitled to will not decrypt, which is worth an error; + here it means an object sealed under a key this job does not name, + which is not an artefact this reader can use, full stop. The + distinction between the two ways of being unreadable is not one the + caller has any different answer for. """ - del sample_rate - spectrum = np.abs(np.fft.rfft(window * np.hanning(len(window)))) - # `rfft` returns `WINDOW/2 + 1` points; dropping DC leaves exactly - # `WINDOW/2`, which `BINS` divides evenly (see its comment). Each row - # is the *peak* of its group rather than the mean, which keeps a - # narrow harmonic visible instead of averaging it into the floor - # beside it -- and the harmonic stack is what identifies speech here. - group = WINDOW // 2 // BINS - return np.maximum.reduceat( - spectrum[1 : 1 + WINDOW // 2], np.arange(0, WINDOW // 2, group) - ).astype(np.float32) - + ciphertext_bytes = await source.size(key) + plaintext_bytes = stored_length(ciphertext_bytes) + if plaintext_bytes <= 0 or plaintext_bytes > _MAX_ARTEFACT_BYTES: + raise CorruptRecording("stored spectrogram is not the size an artefact is") -def _render(columns: list[np.ndarray], fmt: TrackFormat) -> Spectrogram: - """Normalises the collected slices into one byte per cell.""" - matrix = np.stack(columns, axis=1) if columns else np.zeros((BINS, 1), dtype=np.float32) - - # Normalised against the loudest cell *or* the silence floor, whichever - # is louder. For any real recording the peak wins and nothing changes; - # for a track with no signal in it the floor wins, every cell lands - # more than `DYNAMIC_RANGE_DB` below it, and the picture is empty -- - # which is the truth about that track. - reference = max(float(matrix.max()), _SILENCE_FLOOR) - decibels = 20.0 * np.log10(np.maximum(matrix, 1e-12) / reference) - clipped = np.clip(decibels, -DYNAMIC_RANGE_DB, 0.0) - scaled = ((clipped + DYNAMIC_RANGE_DB) / DYNAMIC_RANGE_DB * 255.0).astype(np.uint8) - # The absolute cut, applied last so it wins over the relative scale. - scaled[matrix < _NOISE_MAGNITUDE] = 0 - - # Padded to the promised width so the client can treat the payload as - # a fixed-size image. A short track is short because it is short, and - # the columns it does not have are floor. - if scaled.shape[1] < COLUMNS: - scaled = np.pad(scaled, ((0, 0), (0, COLUMNS - scaled.shape[1]))) - - return Spectrogram( - columns=COLUMNS, - bins=BINS, - sample_rate=fmt.sample_rate, - duration_seconds=round(fmt.duration_seconds, 3), - magnitudes=base64.b64encode(scaled.tobytes()).decode("ascii"), - ) + pieces = stream_wav(source, key, data_key, ByteRange(0, plaintext_bytes - 1)) + body = bytearray() + try: + async for piece in pieces: + body += piece + except InvalidTag as exc: + raise CorruptRecording("stored spectrogram is sealed under another key") from exc + finally: + await pieces.aclose() + return decode_artefact(bytes(body)) diff --git a/src/sturnus/domain/errors.py b/src/sturnus/domain/errors.py index daa2bfe..b10dc38 100644 --- a/src/sturnus/domain/errors.py +++ b/src/sturnus/domain/errors.py @@ -1,4 +1,14 @@ -"""The one marker that decides whether an exception message may leave the pod. +"""The exception types no single layer gets to own. + +`DiagnosticSafeError` is the marker that decides whether an exception +message may leave the pod. `CorruptRecording` is the refusal two readers +of the stored audio format both raise. Neither belongs to one layer: +the first is checked by `sturnus.infrastructure.observability` and raised +anywhere, the second is raised by `sturnus.application.spectrogram` and by +`sturnus.console.audio`, and `application` may not import the console +(tests/test_architecture.py). `domain` is the only place below both, and +both are stdlib-only by construction so that every layer, `domain` +included, can raise them. Sturnus records people talking. Spec 12.4 -- "Neither audio data nor transcript content appears in logs" -- is the standard the pod logs are held @@ -11,9 +21,6 @@ is checked by a human at review time -- there is no way to verify it mechanically, which is exactly why it lives in a marker class rather than in a regex. - -Stdlib-only by construction so that every layer, including `domain`, can -raise it (see `tests/test_architecture.py`). """ from __future__ import annotations @@ -52,3 +59,29 @@ class DiagnosticSafeError(Exception): follow-up for someone who has actually read what those bodies contain, not a tidy-up. """ + + +class CorruptRecording(Exception): + """The stored object is not a recording this reader understands. + + Raised before any plaintext is produced -- a wrong magic, a truncated + upload, a track whose own header says nothing usable, a stored + spectrogram of a shape this build does not draw. The alternative to + refusing is plausible-looking noise on somebody's speakers, or a + picture of a meeting that never happened. + + **Here because two layers name it.** It began in + `sturnus.console.audio`, where the only reader of the on-disk format + lived. There are two now: that one, and + `sturnus.application.spectrogram`, which the worker also draws with -- + and `application` may not import the console (tests/test_architecture.py), + so an exception both of them raise has to live below both. `domain` is + the only place that is, and this class needs nothing but the standard + library to be what it is. + + Deliberately *not* a `DiagnosticSafeError` yet, even though every + message it can carry today is a literal from this repository. Marking + it is a claim about every future message too, and the class is raised + from two modules now; see that class's docstring for what the claim + costs to make honestly. + """ diff --git a/src/sturnus/domain/settings.py b/src/sturnus/domain/settings.py index 9309fe3..c7cbc1b 100644 --- a/src/sturnus/domain/settings.py +++ b/src/sturnus/domain/settings.py @@ -121,6 +121,33 @@ #: recorded -- see `docs/operations.md` §6.2.9. ADMIN_AUDIO_DOWNLOAD_OFFERED = "admin_audio_download_offered" +#: Whether this guild's worker draws each track's spectrogram once, at job +#: completion, and stores it beside the audio. +#: +#: Off is exactly today's behaviour: the console draws a picture per +#: request out of the ciphertext, which costs a full streamed decrypt and +#: 600 transforms every time somebody opens a recording and buys nothing +#: that survives the response. On, the worker draws it while it still has +#: the plaintext WAV on disk -- work it is already doing, on bytes it +#: already has -- and a view becomes two object-store reads. +#: +#: **What turning it on actually costs is storage, and the storage is a +#: rendering of somebody's voice activity.** That is why it carries a rule +#: rather than only a price: *a stored spectrogram is deleted when its +#: audio is deleted*. The retention sweep +#: (`sturnus.application.retention.sweep_expired_audio`) deletes both in +#: one pass, because a sweep that deleted the recording and left the +#: picture would leave behind a record of when somebody spoke and for how +#: long that outlives the window their recording was subject to. A +#: spectrogram is less than the audio; it is not nothing, and it must not +#: become the thing that survives. +#: +#: Default `false`, and unlike the two switches above the reason is not a +#: claim about anybody's policy document: it is that this stores something +#: new about every recording, and a key that starts storing on every guild +#: the day it ships is a key that decided that on their behalf. +SPECTROGRAMS_BY_DEFAULT = "spectrograms_by_default" + #: The one value of `TRANSCRIPTION_LANGUAGE` that is not a language: it #: asks the engine to detect one per speaker and pin what it found for the #: rest of the session, which is what the worker did unconditionally before @@ -158,6 +185,12 @@ # would hand every administrator of every guild a copy of every # recording the moment this key shipped. ADMIN_AUDIO_DOWNLOAD_OFFERED: FALSE, + # Off, so that a guild's recordings acquire nothing new until + # somebody asks for it. The artefact is deleted with the audio it was + # drawn from, so `true` is defensible -- but it is a decision about + # what is kept about people who spoke, and shipping it as the default + # would be taking that decision for every guild at once. + SPECTROGRAMS_BY_DEFAULT: FALSE, # Protocols are read by the people who were in the room, so the # times in them are theirs, not the cluster's. A wrong offset is # not obviously wrong to a reader -- 15:08 looks like a plausible @@ -248,6 +281,7 @@ { VIDEO_CONSENT_OFFERED, ADMIN_AUDIO_DOWNLOAD_OFFERED, + SPECTROGRAMS_BY_DEFAULT, } ) diff --git a/src/sturnus/entrypoints/worker.py b/src/sturnus/entrypoints/worker.py index 3641823..f5c24ba 100644 --- a/src/sturnus/entrypoints/worker.py +++ b/src/sturnus/entrypoints/worker.py @@ -41,6 +41,13 @@ transcripts_for` and `AccountLinkRepository.external_identity`, likewise already exist and are wired in directly below, with no adapter needed. +`_SpectrogramArtefacts` is the third, and it is here for a different +reason: it is not filling a gap in a repository but joining two of them. +Storing a spectrogram is one act made of a database write and an upload, +with an order between them that the application layer decides and this +file carries out -- plus the sealing, which is `encrypt_file` and +therefore infrastructure by definition. + **Deployment note on the working directory (see the brief's dispatch).** `process_one`'s scratch directory for the downloaded/decrypted audio defaults to `/tmp`, overridable with `STURNUS_WORK_DIR`. This matches what @@ -80,7 +87,7 @@ from sturnus.application.retention import sweep_expired_audio from sturnus.application.worker import process_one, retry_pending_documents from sturnus.config import StrictSettings -from sturnus.infrastructure.crypto import KeyWrapper, decrypt_file +from sturnus.infrastructure.crypto import KeyWrapper, decrypt_file, encrypt_file from sturnus.infrastructure.db.config_store import ConfigStore from sturnus.infrastructure.db.export_targets import ExportTargetStore from sturnus.infrastructure.db.models import Session, SessionParticipant, TranscriptionJob @@ -289,6 +296,55 @@ def decrypt_to(self, source: Path, target: Path, wrapped: bytes, key_id: str) -> decrypt_file(source, target, data_key) +class _SpectrogramArtefacts: + """Adapts persistence and the bucket to `worker.SpectrogramStore`. + + Two collaborators behind one port because the port is one act: write + down where the picture goes, then put it there. See that protocol for + why the order is not negotiable -- an object no row names is an object + the retention sweep will never delete, and this feature's whole + justification is that a stored spectrogram dies with its audio. + + The sealing happens here rather than in `sturnus.application.worker` + for the reason `_KeyWrapperDecryptor` exists: that module may not + import `sturnus.infrastructure` (tests/test_architecture.py), and + `encrypt_file` is the file format itself. The *decision* to seal is + the application's and is written into the port's name; the bytes are + this file's problem. + + `keys` is the process's one `KeyWrapper`, handed in rather than built + from the master key a third time: this composition root already + decodes that key for `_KeyWrapperDecryptor` and for `ExportTargetStore`, + and a third decode is a third place a rotation has to be threaded + through. + """ + + def __init__(self, jobs: JobRepository, store: S3AudioStore, keys: KeyWrapper) -> None: + self._jobs = jobs + self._store = store + self._keys = keys + + async def record(self, job_id: int, key: str) -> None: + await self._jobs.record_spectrogram(job_id, key) + + async def put_sealed(self, key: str, source: Path, wrapped: bytes, _key_id: str) -> None: + """Seals the artefact under the job's own data key and uploads it. + + The same single-master-key assumption `_KeyWrapperDecryptor` + documents, which is why `_key_id` is accepted and unused: there is + one live master key to unwrap with, and a job naming another is + already reported, loudly, where that job's audio was decrypted a + moment ago. A second line here would be a second report of one + misconfiguration. + + Sealed beside the plaintext in the job's own scratch directory, so + `process_one`'s `finally` removes both. + """ + target = source.with_name(source.name + ".enc") + await asyncio.to_thread(encrypt_file, source, target, self._keys.unwrap(wrapped)) + await self._store.put(key, target) + + class _WorkerSessionStore: """Adapts persistence to `sturnus.application.worker.SessionStore`. @@ -602,6 +658,9 @@ async def _run() -> None: # `sturnus.application.worker._create_session_document` -- rather than # assumed to be "outline" for every guild this one process serves. links = AccountLinkRepository(session_factory) + # The same `keys` binding the export targets use: one decode of the + # master key in this process, not one per collaborator that needs it. + spectrograms = _SpectrogramArtefacts(jobs, store, keys) config_store = ConfigStore(session_factory) template_source = _load_template() @@ -720,6 +779,7 @@ async def database_ping() -> bool: jobs=jobs, links=links, config=config_store, + spectrograms=spectrograms, work_dir=settings.work_dir, max_attempts=settings.max_job_attempts, template_source=template_source, diff --git a/src/sturnus/infrastructure/db/repositories.py b/src/sturnus/infrastructure/db/repositories.py index 6efbeab..3bc6d24 100644 --- a/src/sturnus/infrastructure/db/repositories.py +++ b/src/sturnus/infrastructure/db/repositories.py @@ -588,12 +588,19 @@ async def candidates_for_retention(self) -> list[dict[str, object]]: Filters only by `audio_deleted_at`; the `retention_until` boundary is left for that pure function to check, so there is exactly one definition of it. + + `spectrogram_key` rides along because the sweep deletes the + recording and its stored picture in one pass -- see + `sweep_expired_audio`. It is `None` for every job whose guild + never asked for spectrograms, which the sweep reads as "there is + no second object here". """ async with self._session_factory() as session: rows = await session.execute( select( TranscriptionJob.id, TranscriptionJob.s3_key, + TranscriptionJob.spectrogram_key, TranscriptionJob.retention_until, TranscriptionJob.audio_deleted_at, ).where(TranscriptionJob.audio_deleted_at.is_(None)) @@ -602,6 +609,7 @@ async def candidates_for_retention(self) -> list[dict[str, object]]: { "id": row.id, "s3_key": row.s3_key, + "spectrogram_key": row.spectrogram_key, "retention_until": row.retention_until, "audio_deleted_at": row.audio_deleted_at, } @@ -609,11 +617,35 @@ async def candidates_for_retention(self) -> list[dict[str, object]]: ] async def mark_audio_deleted(self, job_id: int, now: datetime) -> None: + """Stamps the deletion and clears the artefact's key in one statement. + + The stamp is what makes a swept job unofferable; clearing + `spectrogram_key` is what stops the row from going on naming an + object the same sweep has just deleted. One `UPDATE`, because they + are one fact: this job's recording, and everything drawn from it, + is gone. + """ + async with self._session_factory() as session: + await session.execute( + update(TranscriptionJob) + .where(TranscriptionJob.id == job_id) + .values(audio_deleted_at=now, spectrogram_key=None) + ) + await session.commit() + + async def record_spectrogram(self, job_id: int, key: str) -> None: + """Writes down where this job's stored picture is going to be. + + Called by the worker *before* the object is written, which is the + order `sturnus.application.worker.SpectrogramStore` explains: the + retention sweep deletes what this column names, so an artefact + this column does not name is an artefact nothing will ever delete. + """ async with self._session_factory() as session: await session.execute( update(TranscriptionJob) .where(TranscriptionJob.id == job_id) - .values(audio_deleted_at=now) + .values(spectrogram_key=key) ) await session.commit() diff --git a/src/sturnus/observability/events.py b/src/sturnus/observability/events.py index 72408d4..7b2d821 100644 --- a/src/sturnus/observability/events.py +++ b/src/sturnus/observability/events.py @@ -145,6 +145,18 @@ class Event(StrEnum): SESSION_EXPORT_SKIPPED = "session.export_skipped" RETENTION_SWEPT = "retention.swept" RETENTION_FAILED = "retention.failed" + #: A guild that asked for `spectrograms_by_default` got one: the + #: worker drew this track's picture while it still held the plaintext + #: and stored it beside the recording. INFO rather than DEBUG because + #: it is the only line that says the bucket gained an object, and + #: "how much is this setting costing us" is an operator question with + #: no other answer. + SPECTROGRAM_STORED = "spectrogram.stored" + #: The picture could not be drawn or could not be stored. **WARNING, + #: not ERROR:** the transcription succeeded and is the valuable part, + #: and the console draws this track on demand exactly as it did before + #: the setting existed. What is lost is a saving, not a recording. + SPECTROGRAM_FAILED = "spectrogram.failed" # -- link --------------------------------------------------------------- LINK_STARTED = "link.started" @@ -209,6 +221,13 @@ class Event(StrEnum): #: it will not decrypt -- which is either a truncated upload or a #: format that has drifted from its reader. CONSOLE_TRACK_UNREADABLE = "console.track_unreadable" + #: A stored spectrogram was named by the job and could not be used -- + #: the object is gone, or it is not a picture this build draws. INFO, + #: because the request is *answered*: the reader falls back to drawing + #: the track itself, which is what every view cost before artefacts + #: existed. It is worth a line only because a burst of them means a + #: guild is paying for storage it is not getting a saving from. + CONSOLE_SPECTROGRAM_REDRAWN = "console.spectrogram_redrawn" #: An administrator changed a guild's runtime configuration from the #: web console. The only trace such a change leaves: a slash command diff --git a/tests/application/test_retention.py b/tests/application/test_retention.py index 105f476..f837dfa 100644 --- a/tests/application/test_retention.py +++ b/tests/application/test_retention.py @@ -10,12 +10,20 @@ def job( until: datetime, deleted: datetime | None = None, s3_key: str | None = None, + spectrogram_key: str | None = None, ) -> dict[str, object]: + """One row as `candidates_for_retention` shapes it. + + `spectrogram_key` defaults to `None`, which is what the column holds + for every job whose guild never asked for spectrograms -- so the + sweep's ordinary case stays the one most of these tests exercise. + """ return { "id": job_id, "retention_until": until, "audio_deleted_at": deleted, "s3_key": s3_key or f"sessions/1/speakers/{job_id}.enc", + "spectrogram_key": spectrogram_key, } @@ -105,3 +113,56 @@ async def test_sweep_expired_audio_survives_one_jobs_failure() -> None: await sweep_expired_audio(jobs, store, T0) # must not raise assert store.deleted == ["succeeds"] assert jobs.deleted == [2] + + +# --------------------------------------------------------------------------- +# The rule a stored spectrogram exists under +# +# A spectrogram is a rendering of when somebody spoke and for how long. It +# is less than the audio and it is not nothing, and this sweep is the only +# thing that ends a recording's life -- so it has to end the picture's too, +# or `spectrograms_by_default` becomes a switch that quietly makes +# something about a person's voice outlive their recording's retention. +# --------------------------------------------------------------------------- + + +async def test_a_swept_job_loses_its_spectrogram_in_the_same_pass() -> None: + """Both objects, one pass. Not a second sweep that could be forgotten.""" + jobs = FakeJobs([job(1, T0 - timedelta(seconds=1), s3_key="a", spectrogram_key="a.spec")]) + store = FakeStore() + + await sweep_expired_audio(jobs, store, T0) + + assert store.deleted == ["a", "a.spec"] + assert jobs.deleted == [1] + + +async def test_a_job_with_no_spectrogram_is_swept_exactly_as_before() -> None: + """Most jobs have no picture, and the sweep must not invent a key for + them: a derived key would send a `DELETE` for an object that was never + written, on every job, for ever.""" + jobs = FakeJobs([job(1, T0 - timedelta(seconds=1), s3_key="a")]) + store = FakeStore() + + await sweep_expired_audio(jobs, store, T0) + + assert store.deleted == ["a"] + assert jobs.deleted == [1] + + +async def test_a_spectrogram_that_would_not_delete_leaves_the_job_unstamped() -> None: + """The stamp is the claim that this recording is gone -- all of it. + + A stamp written while the picture is still in the bucket would end the + job's candidacy for ever and leave the artefact behind with nothing + left that knows to delete it. Leaving the row unstamped costs one + repeated `DELETE` of an object already gone, which S3 answers + successfully. + """ + jobs = FakeJobs([job(1, T0 - timedelta(seconds=1), s3_key="a", spectrogram_key="fails")]) + store = FakeStore(fail_keys={"fails"}) + + await sweep_expired_audio(jobs, store, T0) # must not raise + + assert store.deleted == ["a"] + assert jobs.deleted == [] diff --git a/tests/application/test_spectrogram.py b/tests/application/test_spectrogram.py new file mode 100644 index 0000000..20f95f4 --- /dev/null +++ b/tests/application/test_spectrogram.py @@ -0,0 +1,166 @@ +"""The stored form of a picture, and everything a reader must refuse. + +`tests/console/test_spectrogram.py` covers what the picture *says* -- that +a tone lands in the row it belongs to and that silence draws empty. This +file covers the other half, which only exists because a spectrogram is +stored now: the bytes that go into the bucket, and the reading of them. + +The theme throughout is that a stored artefact outlives the process that +wrote it. It survives a deployment, a `COLUMNS` change, a partially +written object and a key that belongs to another recording, and in every +one of those cases the honest answer is to refuse it and draw the track +again -- which the caller can always do, because the audio it was drawn +from is still there. A picture that renders and is wrong is the one +outcome worth failing to prevent. +""" + +from __future__ import annotations + +import base64 +import json + +import pytest + +from sturnus.application.spectrogram import ( + ARTEFACT_VERSION, + BINS, + COLUMNS, + Spectrogram, + decode_artefact, + draw, + encode_artefact, +) +from sturnus.domain.errors import CorruptRecording + + +def picture(sample_rate: int = 16_000, duration: float = 12.5) -> Spectrogram: + """A picture of the shape this build draws, with recognisable cells.""" + cells = bytes(range(256)) * (COLUMNS * BINS // 256) + return Spectrogram( + columns=COLUMNS, + bins=BINS, + sample_rate=sample_rate, + duration_seconds=duration, + magnitudes=base64.b64encode(cells).decode("ascii"), + ) + + +def artefact(**overrides: object) -> bytes: + """A stored artefact with fields replaced, for the refusal tests.""" + document = json.loads(encode_artefact(picture())) + document.update(overrides) + return json.dumps(document).encode("utf-8") + + +async def _pieces(*chunks: bytes) -> object: + for chunk in chunks: + yield chunk + + +# --------------------------------------------------------------------------- +# The round trip +# --------------------------------------------------------------------------- + + +def test_a_stored_picture_reads_back_as_the_picture_that_was_stored() -> None: + """The console answers from the artefact, so the artefact has to carry + everything the response has: the axes as well as the cells.""" + original = picture(sample_rate=16_000, duration=3.25) + + restored = decode_artefact(encode_artefact(original)) + + assert restored == original + + +def test_the_frequency_axis_survives_the_round_trip() -> None: + """`hz_per_bin` is derived from the sample rate, and the sample rate is + read from the track's own header. A picture that lost it would be + labelled with whatever the client assumed.""" + restored = decode_artefact(encode_artefact(picture(sample_rate=8_000))) + + assert restored.hz_per_bin == pytest.approx(8_000 / 2 / BINS) + + +def test_an_artefact_is_about_a_hundred_kilobytes_whatever_the_meeting_was() -> None: + """The number an operator sizes a bucket with. + + Fixed by construction -- `COLUMNS` by `BINS` cells, base64'd, in a + small envelope -- so a three-hour workshop costs exactly what a + two-minute stand-up does. If this ever became proportional to the + recording, the storage estimate this feature was accepted on would + stop being true and nothing else would say so. + """ + stored = encode_artefact(picture()) + + assert COLUMNS * BINS == 76_800 + assert len(stored) < 110_000 + + +# --------------------------------------------------------------------------- +# What a reader refuses, and why refusing is cheap +# --------------------------------------------------------------------------- + + +def test_a_picture_of_another_shape_is_refused() -> None: + """The failure this check exists for is silent otherwise. + + An artefact drawn when `COLUMNS` was a different number still parses + and still renders; it describes the track wrongly by however much the + shape moved, and the client sizes its canvas from the numbers it was + handed rather than from what arrived. + """ + with pytest.raises(CorruptRecording): + decode_artefact(artefact(columns=COLUMNS // 2)) + + with pytest.raises(CorruptRecording): + decode_artefact(artefact(bins=BINS + 8)) + + +def test_a_picture_from_a_future_version_is_refused() -> None: + with pytest.raises(CorruptRecording): + decode_artefact(artefact(version=ARTEFACT_VERSION + 1)) + + +def test_a_matrix_with_the_wrong_number_of_cells_is_refused() -> None: + """A truncated upload is the ordinary cause, and it renders as a + picture of a meeting that stopped early.""" + short = base64.b64encode(b"\x00" * (COLUMNS * BINS - 1)).decode("ascii") + + with pytest.raises(CorruptRecording): + decode_artefact(artefact(magnitudes=short)) + + +def test_something_that_is_not_an_artefact_at_all_is_refused() -> None: + with pytest.raises(CorruptRecording): + decode_artefact(b"not json") + + with pytest.raises(CorruptRecording): + decode_artefact(b'["a list is not a document"]') + + +def test_an_artefact_missing_its_axes_is_refused() -> None: + """Rather than defaulted. A sample rate this reader guessed is exactly + the six-times-speed defect that made a spectrogram worth having.""" + with pytest.raises(CorruptRecording): + decode_artefact(artefact(sample_rate=None)) + + with pytest.raises(CorruptRecording): + decode_artefact(artefact(magnitudes=None)) + + +# --------------------------------------------------------------------------- +# Drawing, over a stream that is not S3 +# --------------------------------------------------------------------------- + + +async def test_a_stream_that_never_carried_a_header_is_refused() -> None: + """The worker's stream is a file rather than an object body, and an + empty one is a decrypt that produced nothing -- which must not draw an + empty picture and call it a recording.""" + with pytest.raises(CorruptRecording): + await draw(_pieces()) # type: ignore[arg-type] + + +async def test_a_stream_of_something_that_is_not_a_track_is_refused() -> None: + with pytest.raises(CorruptRecording): + await draw(_pieces(b"RIFF" + b"\x00" * 200)) # type: ignore[arg-type] diff --git a/tests/application/test_worker.py b/tests/application/test_worker.py index 5a53572..1129466 100644 --- a/tests/application/test_worker.py +++ b/tests/application/test_worker.py @@ -16,8 +16,12 @@ from pathlib import Path from typing import Any +import pytest + from sturnus.application.documents import CreatedDocument, DocumentSink from sturnus.application.exporting import Destination, ExportPorts +from sturnus.application.recording import spectrogram_key +from sturnus.application.spectrogram import decode_artefact from sturnus.application.transcription import TranscribedSegment, TranscriptionResult from sturnus.application.worker import process_one, retry_pending_documents from sturnus.domain import settings as domain_settings @@ -153,7 +157,13 @@ async def delete(self, key: str) -> None: class FakeCrypto: - """Stands in for unwrap-and-decrypt; writes a recognisable file.""" + """Stands in for unwrap-and-decrypt; writes a recognisable file. + + The eleven bytes it writes are deliberately *not* a track anything can + read: a test that needs a real one reaches for `WritesARealWav` below, + and a test that does not is exercising some reader's refusal path + whether it meant to or not. + """ def __init__(self) -> None: self.decrypted: list[Path] = [] @@ -163,6 +173,44 @@ def decrypt_to(self, _source: Path, target: Path, _wrapped: bytes, _key_id: str) self.decrypted.append(target) +class FakeSpectrograms: + """Satisfies `sturnus.application.worker.SpectrogramStore`. + + Records the two writes in one list, in the order they were made, so a + test can assert the order the protocol insists on -- the key written + down before the object exists -- rather than only that both happened. + """ + + def __init__(self, fail_on: str | None = None) -> None: + #: `("record", key)` / `("put", key)`, in call order. + self.calls: list[tuple[str, str]] = [] + #: The artefact per key, as it reached the store. Plaintext here: + #: the sealing belongs to the adapter that satisfies this port + #: (`sturnus.entrypoints.worker._SpectrogramArtefacts`), and what + #: `process_one` is responsible for is handing over the right + #: envelope material to seal it with. + self.stored: dict[str, bytes] = {} + #: The `(wrapped_data_key, key_id)` each artefact was handed to be + #: sealed under, per key. + self.sealed_with: dict[str, tuple[bytes, str]] = {} + #: Which method blows up, for the "a failure must not fail the + #: job" tests. `None` means neither. + self._fail_on = fail_on + + async def record(self, job_id: int, key: str) -> None: + del job_id + self.calls.append(("record", key)) + if self._fail_on == "record": + raise RuntimeError("the database is briefly unreachable") + + async def put_sealed(self, key: str, source: Path, wrapped: bytes, key_id: str) -> None: + self.calls.append(("put", key)) + if self._fail_on == "put": + raise RuntimeError("S3 is briefly unreachable") + self.stored[key] = source.read_bytes() + self.sealed_with[key] = (wrapped, key_id) + + class FakeDocuments: def __init__(self, permanent_error: bool = False) -> None: self.created: list[tuple[str, str]] = [] @@ -401,6 +449,7 @@ def run(tmp_path: Path, **kw: Any) -> dict[str, Any]: "jobs": kw.get("jobs") or FakeJobs(), "links": kw.get("links") or FakeLinks(), "config": kw.get("config") or FakeConfig(), + "spectrograms": kw.get("spectrograms") or FakeSpectrograms(), "work_dir": tmp_path, "max_attempts": 3, } @@ -1067,6 +1116,7 @@ async def test_a_finished_job_records_what_its_recording_is(tmp_path: Path) -> N jobs=FakeJobs(), links=FakeLinks(), config=FakeConfig(), + spectrograms=FakeSpectrograms(), work_dir=tmp_path, max_attempts=3, ) @@ -1094,6 +1144,7 @@ async def test_the_size_recorded_is_the_stored_object_and_not_the_plaintext(tmp_ jobs=FakeJobs(), links=FakeLinks(), config=FakeConfig(), + spectrograms=FakeSpectrograms(), work_dir=tmp_path, max_attempts=3, ) @@ -1125,6 +1176,7 @@ async def test_a_recording_whose_header_cannot_be_read_records_nothing(tmp_path: jobs=FakeJobs(), links=FakeLinks(), config=FakeConfig(), + spectrograms=FakeSpectrograms(), work_dir=tmp_path, max_attempts=3, ) @@ -1149,6 +1201,7 @@ async def test_a_stereo_recording_is_recorded_as_stereo(tmp_path: Path) -> None: jobs=FakeJobs(), links=FakeLinks(), config=FakeConfig(), + spectrograms=FakeSpectrograms(), work_dir=tmp_path, max_attempts=3, ) @@ -1173,6 +1226,7 @@ async def test_a_failed_transcription_records_nothing_about_the_file(tmp_path: P jobs=FakeJobs(), links=FakeLinks(), config=FakeConfig(), + spectrograms=FakeSpectrograms(), work_dir=tmp_path, max_attempts=3, ) @@ -1321,3 +1375,209 @@ async def test_a_session_in_both_candidate_sets_is_published_once() -> None: exports(documents), sessions, FakeJobs(), FakeLinks(), FakeConfig() ) assert len(documents.created) == 1 + + +# --------------------------------------------------------------------------- +# The spectrogram a guild asked to have drawn once +# +# Off, this changes nothing: the console goes on drawing a picture per +# request out of S3. On, the worker draws it here, where the plaintext is +# free, and stores it beside the recording -- under the rule that the +# retention sweep deletes both (`sturnus.application.retention`). +# --------------------------------------------------------------------------- + + +def spectrograms_on() -> FakeConfig: + """A guild that switched `spectrograms_by_default` on.""" + return guild_config({domain_settings.SPECTROGRAMS_BY_DEFAULT: domain_settings.TRUE}) + + +async def test_a_guild_that_never_asked_for_spectrograms_stores_none(tmp_path: Path) -> None: + """The default is off, and off has to mean exactly today's behaviour. + + Nothing is drawn, nothing is uploaded, and no column is written -- + because storing a rendering of somebody's voice activity for a guild + that did not ask for one is the thing the default exists to prevent. + """ + spectrograms = FakeSpectrograms() + + await process_one(**run(tmp_path, spectrograms=spectrograms, crypto=WritesARealWav())) + + assert spectrograms.calls == [] + + +async def test_a_guild_that_asked_gets_a_picture_stored_beside_its_recording( + tmp_path: Path, +) -> None: + """Drawn once, at completion, from the plaintext already on disk. + + The key is the audio's key with a different suffix, so a prefix + listing of a session shows everything that session put in the bucket + -- which is what makes "did the sweep leave anything behind?" a + question the bucket can answer. + """ + spectrograms = FakeSpectrograms() + + await process_one( + **run( + tmp_path, + config=spectrograms_on(), + spectrograms=spectrograms, + crypto=WritesARealWav(), + ) + ) + + key = spectrogram_key(1, 100) + assert key in spectrograms.stored + assert key.startswith("sessions/1/speakers/100.") + + +async def test_the_stored_picture_describes_the_track_it_was_drawn_from( + tmp_path: Path, +) -> None: + """A stored artefact that renders is not the same as one that is true. + + It is read back through the same reader the console uses, and asked + the one question the console will ask of it: what track is this? A + picture whose sample rate or length disagreed with the recording would + label somebody's meeting with somebody else's axes. + """ + spectrograms = FakeSpectrograms() + + await process_one( + **run( + tmp_path, + config=spectrograms_on(), + spectrograms=spectrograms, + crypto=WritesARealWav(sample_rate=16_000, frames=32_000), + ) + ) + + picture = decode_artefact(spectrograms.stored[spectrogram_key(1, 100)]) + assert picture.sample_rate == 16_000 + assert picture.duration_seconds == pytest.approx(2.0, abs=0.01) + + +async def test_the_key_is_written_down_before_the_object_is_put(tmp_path: Path) -> None: + """The order that keeps the retention rule enforceable. + + The sweep deletes what `transcription_job.spectrogram_key` names, so + an object stored before the row names it is an object that -- if the + write in between failed -- nothing would ever delete: a rendering of + somebody's voice outliving the recording's retention window, which is + the one outcome this feature is not allowed to produce. + """ + spectrograms = FakeSpectrograms() + + await process_one( + **run( + tmp_path, + config=spectrograms_on(), + spectrograms=spectrograms, + crypto=WritesARealWav(), + ) + ) + + assert [call for call, _ in spectrograms.calls] == ["record", "put"] + + +async def test_the_picture_is_sealed_under_the_same_key_as_the_recording( + tmp_path: Path, +) -> None: + """The job's own envelope, handed over rather than reinvented. + + The artefact is a rendering of somebody's voice activity, so it is + encrypted like the voice is; using the job's own wrapped data key + means a master-key rotation reaches the picture and the recording + together, and that a reader entitled to one can already open the + other. + """ + spectrograms = FakeSpectrograms() + claimed = job() + + await process_one( + **run( + tmp_path, + queue=FakeQueue([claimed]), + config=spectrograms_on(), + spectrograms=spectrograms, + crypto=WritesARealWav(), + ) + ) + + sealed = spectrograms.sealed_with[spectrogram_key(1, 100)] + assert sealed == (claimed.wrapped_data_key, claimed.encryption_key_id) + + +@pytest.mark.parametrize("fails", ["record", "put"]) +async def test_a_spectrogram_that_cannot_be_stored_does_not_fail_the_job( + tmp_path: Path, fails: str +) -> None: + """The transcription is the valuable part and it already succeeded. + + Re-queueing the job would spend minutes of inference again to retry a + picture the console draws for itself in a second -- and the read path + does exactly that when there is no artefact. So the failure is logged + and the job stays done. + """ + queue = FakeQueue([job()]) + + done = await process_one( + **run( + tmp_path, + queue=queue, + config=spectrograms_on(), + spectrograms=FakeSpectrograms(fail_on=fails), + crypto=WritesARealWav(), + ) + ) + + assert done is True + assert len(queue.completed) == 1 + assert queue.failed == [] + + +async def test_a_track_that_cannot_be_drawn_does_not_fail_the_job(tmp_path: Path) -> None: + """Same argument, one step earlier: the drawing itself can refuse. + + `FakeCrypto`'s default plaintext is not a track any reader can parse, + which is precisely the shape of a truncated or drifted recording. + """ + queue = FakeQueue([job()]) + spectrograms = FakeSpectrograms() + + done = await process_one( + **run(tmp_path, queue=queue, config=spectrograms_on(), spectrograms=spectrograms) + ) + + assert done is True + assert len(queue.completed) == 1 + assert spectrograms.stored == {} + + +async def test_the_decrypted_track_is_still_deleted_after_it_has_been_drawn( + tmp_path: Path, +) -> None: + """Drawing a picture is not a licence to keep the plaintext around. + + The artefact is written into the same scratch directory the WAV is in, + so the one `finally` that has always removed decrypted speech removes + the rendering of it too. Asserted on the whole of `work_dir` rather + than on the two filenames: what must be gone is everything the job + put there, and naming the files would miss the next one somebody adds. + """ + spectrograms = FakeSpectrograms() + + await process_one( + **run( + tmp_path, + config=spectrograms_on(), + spectrograms=spectrograms, + crypto=WritesARealWav(), + ) + ) + + # Without this the assertion below would pass on a run where nothing + # was ever decrypted or drawn. + assert spectrograms.stored + assert list(tmp_path.iterdir()) == [] diff --git a/tests/console/conftest.py b/tests/console/conftest.py index f398ee9..ce47d70 100644 --- a/tests/console/conftest.py +++ b/tests/console/conftest.py @@ -507,6 +507,11 @@ def __init__(self, objects: dict[str, bytes] | None = None) -> None: self.objects = objects if objects is not None else {} self.reads: list[tuple[int, int]] = [] self.streamed_from: list[int] = [] + #: Which objects were opened, in order. The bucket holds more than + #: recordings now -- a stored spectrogram sits beside each one -- + #: and "was the recording read at all?" is a question about the + #: key, which the offsets alone cannot answer. + self.streamed_keys: list[str] = [] self.streamed_bytes = 0 async def size(self, key: str) -> int: @@ -523,6 +528,7 @@ async def read(self, key: str, start: int, length: int) -> bytes: async def stream(self, key: str, start: int) -> AsyncGenerator[bytes, None]: if key not in self.objects: raise KeyError(key) + self.streamed_keys.append(key) self.streamed_from.append(start) body = self.objects[key][start:] for offset in range(0, len(body), self.PIECE): diff --git a/tests/console/test_audio.py b/tests/console/test_audio.py index 582b4bd..66bf64f 100644 --- a/tests/console/test_audio.py +++ b/tests/console/test_audio.py @@ -28,12 +28,12 @@ from sturnus.console.audio import ( ByteRange, - CorruptRecording, UnsatisfiableRange, parse_range, stored_length, stream_wav, ) +from sturnus.domain.errors import CorruptRecording from sturnus.infrastructure.audio import SOURCE_RATE, TARGET_RATE from sturnus.infrastructure.crypto import CHUNK_SIZE from sturnus.infrastructure.recording_adapters import FileAudioWriterFactory @@ -250,9 +250,15 @@ async def test_a_truncated_recording_is_refused(track: bytes, tmp_path: Path) -> # The property the encryption scheme exists for # --------------------------------------------------------------------------- +#: Every module a decrypted recording passes through on its way to a +#: socket. `application/spectrogram.py` is on this list although it is not +#: in the console package: it holds the transform the console's reader +#: feeds plaintext into, and a module on the serving path is on the +#: serving path wherever it happens to live. +_SRC = Path(__file__).parent.parent.parent / "src" / "sturnus" _SERVING_PATH = ( - Path(__file__).parent.parent.parent / "src" / "sturnus" / "console", - ("audio.py", "routes_audio.py", "spectrogram.py"), + (_SRC / "console", ("audio.py", "routes_audio.py", "spectrogram.py")), + (_SRC / "application", ("spectrogram.py",)), ) _WRITES_TO_DISK = frozenset( @@ -279,22 +285,22 @@ def test_nothing_on_the_serving_path_can_write_plaintext_to_disk() -> None: served, and it would do so invisibly. Reviewing for it once is not the same as being unable to do it. """ - directory, names = _SERVING_PATH offenders: list[str] = [] - for name in names: - tree = ast.parse((directory / name).read_text(encoding="utf-8")) - for node in ast.walk(tree): - if not isinstance(node, ast.Call): - continue - called = ( - node.func.id - if isinstance(node.func, ast.Name) - else node.func.attr - if isinstance(node.func, ast.Attribute) - else None - ) - if called in _WRITES_TO_DISK: - offenders.append(f"{name}:{node.lineno}: calls {called}()") + for directory, names in _SERVING_PATH: + for name in names: + tree = ast.parse((directory / name).read_text(encoding="utf-8")) + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + called = ( + node.func.id + if isinstance(node.func, ast.Name) + else node.func.attr + if isinstance(node.func, ast.Attribute) + else None + ) + if called in _WRITES_TO_DISK: + offenders.append(f"{directory.name}/{name}:{node.lineno}: calls {called}()") assert not offenders, "\n".join(offenders) diff --git a/tests/console/test_audio_routes.py b/tests/console/test_audio_routes.py index e724843..8a017ff 100644 --- a/tests/console/test_audio_routes.py +++ b/tests/console/test_audio_routes.py @@ -26,6 +26,7 @@ from aiohttp.test_utils import TestClient from moto import mock_aws +from sturnus.application.spectrogram import BINS, COLUMNS, Spectrogram, encode_artefact from sturnus.console.audio import AudioDelivery from sturnus.console.ports import ( EncryptedAudioSource, @@ -62,6 +63,11 @@ SESSION_COOKIE = "sturnus_session" WRAPPED = b"wrapped-data-key" +#: Where a stored spectrogram lives: beside the recording it was drawn +#: from. Spelled out rather than derived from `S3_KEY`, so a change to the +#: naming rule shows up here as a test somebody has to read. +SPECTROGRAM_KEY = "sessions/4711/speakers/1.spectrogram.enc" + #: Two chunks and a bit, so a range can start inside the third chunk and #: the "did not download the beginning" assertion has something to measure. TRACK = bytes(range(241)) * 40_000 @@ -464,3 +470,190 @@ async def test_a_spectrogram_is_never_cached_by_anything_in_between( client = await signed_in(aiohttp_client, build(tracks, source)) response = await client.get(spectrogram_url()) assert response.headers["Cache-Control"] == "private, no-store" + + +# --------------------------------------------------------------------------- +# The picture the worker already drew +# +# A guild that switched `spectrograms_by_default` on has an artefact in the +# bucket, and this endpoint prefers it. What must not change is anything +# else: the same rule decides who may see it, decided again on this +# request, and a track whose audio is gone has no picture either. +# --------------------------------------------------------------------------- + + +def _stored_artefact(sample_rate: int, tmp_path: Path) -> bytes: + """One artefact in the real on-disk format, sealed like the audio.""" + cells = bytes(range(256)) * (COLUMNS * BINS // 256) + return sealed( + encode_artefact( + Spectrogram( + columns=COLUMNS, + bins=BINS, + sample_rate=sample_rate, + duration_seconds=61.5, + magnitudes=base64.b64encode(cells).decode("ascii"), + ) + ), + tmp_path, + ) + + +@pytest.fixture +def drawn() -> FakeTracks: + """Anna's track, with a spectrogram the worker stored beside it.""" + return FakeTracks( + tracks={(SESSION, ANNA): Track(S3_KEY, KEY_ID, WRAPPED, spectrogram_key=SPECTROGRAM_KEY)}, + participants={SESSION: {ANNA, BEN}}, + ) + + +async def test_a_stored_picture_is_answered_without_reading_the_recording( + aiohttp_client: AiohttpClientFactory, drawn: FakeTracks, tmp_path: Path +) -> None: + """The whole point of storing one: a view stops costing a full decrypt. + + Asserted on which object was opened rather than on how long it took: + the saving is that the recording's body is never streamed at all, and + on a three-hour workshop that body is the entire cost. A reader that + drew the track anyway would name the recording's key here. + """ + source = FakeAudioSource( + { + S3_KEY: sealed(_real_track(tmp_path), tmp_path), + SPECTROGRAM_KEY: _stored_artefact(TARGET_RATE, tmp_path), + } + ) + client = await signed_in(aiohttp_client, build(drawn, source)) + + response = await client.get(spectrogram_url()) + + assert response.status == 200 + body = await response.json() + assert body["duration_seconds"] == 61.5 + assert body["sample_rate"] == TARGET_RATE + assert len(base64.b64decode(body["magnitudes"])) == BINS * COLUMNS + assert source.streamed_keys == [SPECTROGRAM_KEY] + + +async def test_a_track_drawn_before_the_setting_was_on_is_still_answered( + aiohttp_client: AiohttpClientFactory, tracks: FakeTracks, tmp_path: Path +) -> None: + """No backfill, and therefore no gap in the interface. + + Every job transcribed before a guild switched the setting on has no + artefact and never will unless it is re-queued. The endpoint's + contract does not depend on that: it draws the track, exactly as it + did before artefacts existed. + """ + source = FakeAudioSource({S3_KEY: sealed(_real_track(tmp_path), tmp_path)}) + client = await signed_in(aiohttp_client, build(tracks, source)) + + response = await client.get(spectrogram_url()) + + assert response.status == 200 + assert (await response.json())["duration_seconds"] == pytest.approx(3.0, abs=0.05) + + +async def test_a_picture_that_is_missing_falls_back_to_drawing_the_track( + aiohttp_client: AiohttpClientFactory, drawn: FakeTracks, tmp_path: Path +) -> None: + """A row naming an object that is not there is a possible state. + + The worker writes the key down before it writes the object, on purpose + -- an artefact nothing names is one the retention sweep can never + delete -- so a failed upload leaves exactly this. It must cost a + recomputation and not a refusal. + """ + source = FakeAudioSource({S3_KEY: sealed(_real_track(tmp_path), tmp_path)}) + client = await signed_in(aiohttp_client, build(drawn, source)) + + response = await client.get(spectrogram_url()) + + assert response.status == 200 + assert (await response.json())["duration_seconds"] == pytest.approx(3.0, abs=0.05) + + +async def test_a_picture_this_build_cannot_read_falls_back_to_drawing_it( + aiohttp_client: AiohttpClientFactory, drawn: FakeTracks, tmp_path: Path +) -> None: + """A stored artefact is not a recording: losing one loses nothing. + + So an artefact that will not decode is answered by redrawing, where a + *recording* that will not decrypt is answered with an error -- the + recording is the only copy of what somebody said, and the picture is a + convenience the next line of the handler recreates. + """ + source = FakeAudioSource( + { + S3_KEY: sealed(_real_track(tmp_path), tmp_path), + SPECTROGRAM_KEY: sealed(b'{"version": 1, "columns": 3}', tmp_path), + } + ) + client = await signed_in(aiohttp_client, build(drawn, source)) + + response = await client.get(spectrogram_url()) + + assert response.status == 200 + assert (await response.json())["duration_seconds"] == pytest.approx(3.0, abs=0.05) + + +async def test_a_stranger_cannot_see_a_stored_picture_either( + aiohttp_client: AiohttpClientFactory, drawn: FakeTracks, tmp_path: Path +) -> None: + """The saving is in the payload and must never reach the permission. + + A cheap answer is exactly the kind of answer that grows a cache in + front of it, and the rule this endpoint enforces -- participants of + the session, nobody else -- is decided on every request against + `session_participant`, before anything looks in the bucket. + """ + source = FakeAudioSource( + { + S3_KEY: sealed(_real_track(tmp_path), tmp_path), + SPECTROGRAM_KEY: _stored_artefact(TARGET_RATE, tmp_path), + } + ) + client = await signed_in(aiohttp_client, build(drawn, source), as_user=999) + + response = await client.get(spectrogram_url()) + + assert response.status == 404 + + +async def test_a_stored_picture_is_never_cached_by_anything_in_between( + aiohttp_client: AiohttpClientFactory, drawn: FakeTracks, tmp_path: Path +) -> None: + """Same header whichever source answered. A shared cache holding this + one would hand somebody's voice activity to the next person through + the same proxy.""" + source = FakeAudioSource( + { + S3_KEY: sealed(_real_track(tmp_path), tmp_path), + SPECTROGRAM_KEY: _stored_artefact(TARGET_RATE, tmp_path), + } + ) + client = await signed_in(aiohttp_client, build(drawn, source)) + + response = await client.get(spectrogram_url()) + + assert response.headers["Cache-Control"] == "private, no-store" + + +async def test_a_recording_that_is_gone_has_no_picture_either( + aiohttp_client: AiohttpClientFactory, drawn: FakeTracks, tmp_path: Path +) -> None: + """After the sweep, the track is neither playable nor visualisable. + + The sweep deletes both objects in one pass, so this state should not + outlive a sweep -- but it is reachable for as long as one runs, and it + is the state the whole retention rule is about. An artefact that + answered here would be a rendering of somebody's voice surviving the + deletion of the recording it was drawn from, reachable through the + console, which is precisely what storing one is not allowed to create. + """ + source = FakeAudioSource({SPECTROGRAM_KEY: _stored_artefact(TARGET_RATE, tmp_path)}) + client = await signed_in(aiohttp_client, build(drawn, source)) + + assert (await client.get(spectrogram_url())).status == 404 + assert (await client.get(url())).status == 404 diff --git a/tests/console/test_spectrogram.py b/tests/console/test_spectrogram.py index 462f999..e8baf3c 100644 --- a/tests/console/test_spectrogram.py +++ b/tests/console/test_spectrogram.py @@ -22,14 +22,15 @@ import numpy as np import pytest -from sturnus.console.audio import CorruptRecording, stored_length -from sturnus.console.spectrogram import ( +from sturnus.application.spectrogram import ( BINS, COLUMNS, WINDOW, parse_track_format, - spectrogram, ) +from sturnus.console.audio import stored_length +from sturnus.console.spectrogram import spectrogram +from sturnus.domain.errors import CorruptRecording from sturnus.infrastructure.audio import SOURCE_RATE, TARGET_RATE from sturnus.infrastructure.recording_adapters import FileAudioWriterFactory from tests.console.conftest import ( diff --git a/tests/infrastructure/test_repositories.py b/tests/infrastructure/test_repositories.py index 1006788..0f916e1 100644 --- a/tests/infrastructure/test_repositories.py +++ b/tests/infrastructure/test_repositories.py @@ -16,6 +16,7 @@ Consent, Session, SessionParticipant, + TranscriptionJob, ) from sturnus.infrastructure.db.queue import JobQueue from sturnus.infrastructure.db.repositories import ( @@ -977,3 +978,56 @@ async def test_mark_audio_deleted_excludes_the_job_from_later_candidates( await jobs.mark_audio_deleted(job_id, T0 + timedelta(days=31)) assert await jobs.candidates_for_retention() == [] + + +async def test_a_candidate_carries_the_key_of_its_stored_spectrogram( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The sweep deletes the recording and its picture in one pass, so the + picture's key has to arrive with the candidate rather than be derived + from the job's ids: the naming rule can change and the objects already + in the bucket cannot.""" + sessions = SessionRepository(factory) + jobs = JobRepository(factory) + session_id = await sessions.open_session(GUILD, CHANNEL, "meeting-raum", T0) + job_id = await _enqueue_job(sessions, jobs, session_id, ANNA) + await jobs.record_spectrogram(job_id, "sessions/1/speakers/1.spectrogram.enc") + + candidates = await jobs.candidates_for_retention() + + assert candidates[0]["spectrogram_key"] == "sessions/1/speakers/1.spectrogram.enc" + + +async def test_a_job_with_no_stored_spectrogram_says_so( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Which is most of them: every job whose guild never asked for one.""" + sessions = SessionRepository(factory) + jobs = JobRepository(factory) + session_id = await sessions.open_session(GUILD, CHANNEL, "meeting-raum", T0) + await _enqueue_job(sessions, jobs, session_id, ANNA) + + assert (await jobs.candidates_for_retention())[0]["spectrogram_key"] is None + + +async def test_stamping_the_deletion_forgets_where_the_picture_was( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The column says where this job's artefact is. Once the sweep has + deleted it there is none, and a row that went on naming it would claim + a picture exists and send every later sweep to delete it again.""" + sessions = SessionRepository(factory) + jobs = JobRepository(factory) + session_id = await sessions.open_session(GUILD, CHANNEL, "meeting-raum", T0) + job_id = await _enqueue_job(sessions, jobs, session_id, ANNA) + await jobs.record_spectrogram(job_id, "sessions/1/speakers/1.spectrogram.enc") + + await jobs.mark_audio_deleted(job_id, T0 + timedelta(days=31)) + + async with factory() as session: + assert ( + await session.scalar( + select(TranscriptionJob.spectrogram_key).where(TranscriptionJob.id == job_id) + ) + is None + ) diff --git a/tests/infrastructure/test_traced_ports.py b/tests/infrastructure/test_traced_ports.py index f9406b0..ef536ab 100644 --- a/tests/infrastructure/test_traced_ports.py +++ b/tests/infrastructure/test_traced_ports.py @@ -44,6 +44,7 @@ FakeLinks, FakeQueue, FakeSessions, + FakeSpectrograms, FakeStore, exports, job, @@ -83,6 +84,7 @@ async def _run_one(tmp_path: Path, *, is_last: bool = True) -> tuple[FakeQueue, jobs=FakeJobs(), links=FakeLinks(), config=FakeConfig(), + spectrograms=FakeSpectrograms(), work_dir=tmp_path, max_attempts=3, ) @@ -181,6 +183,7 @@ async def test_a_failing_stage_marks_its_span_without_a_message( jobs=FakeJobs(), links=FakeLinks(), config=FakeConfig(), + spectrograms=FakeSpectrograms(), work_dir=tmp_path, max_attempts=3, ) @@ -218,6 +221,7 @@ async def test_the_root_span_of_a_failed_job_does_not_say_it_was_done( jobs=FakeJobs(), links=FakeLinks(), config=FakeConfig(), + spectrograms=FakeSpectrograms(), work_dir=tmp_path, max_attempts=3, ) diff --git a/tests/observability/test_no_payload_leaks.py b/tests/observability/test_no_payload_leaks.py index 6a020ee..99c6f77 100644 --- a/tests/observability/test_no_payload_leaks.py +++ b/tests/observability/test_no_payload_leaks.py @@ -41,6 +41,7 @@ FakeLinks, FakeQueue, FakeSessions, + FakeSpectrograms, FakeStore, exports, job, @@ -107,6 +108,7 @@ async def test_the_worker_pipeline_logs_no_payload(tmp_path: Path, captured: io. jobs=FakeJobs(), links=FakeLinks(), config=FakeConfig(), + spectrograms=FakeSpectrograms(), work_dir=tmp_path, max_attempts=3, ) @@ -147,6 +149,7 @@ async def transcribe( jobs=FakeJobs(), links=FakeLinks(), config=FakeConfig(), + spectrograms=FakeSpectrograms(), work_dir=tmp_path, max_attempts=3, )