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, )