diff --git a/AGENTS.md b/AGENTS.md index e523146b..7d840af6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -89,6 +89,7 @@ Apply to every role, every session. 9. **Announce actions before running them.** No silent long running or integration test runs, scripts, or background commands. 10. **No em dashes in user-facing prose.** Use hyphens or sentence breaks. 11. **Put data in through the front door.** Seed and exercise a stacklet the way a family does, never by writing to the service behind it. Documents go through `tools/family-docs/ingest.py` (Matrix room -> archivist -> OCR -> classify -> mirror), never a direct `POST /api/documents/post_document/`. The back door skips the pipeline, so what lands is not what users get: no tags, no correspondent, no document type, no rewritten title, no summary note, no vault entry. Any conclusion drawn from that data is about a system famstack does not ship. +12. **Comments explain the code, not sell it.** Literate but neutral: state the constraint and the reason, without rhetoric or drama. Write `# Tags are mutable, so only the digest identifies the image this dump came from.` rather than `# Tags lie over time -- the digest is the load-bearing field that still names this exact image in five years.` ## Deeper docs (load on demand) diff --git a/docs/admin-guide.md b/docs/admin-guide.md index ce2ea8f6..b93cf697 100644 --- a/docs/admin-guide.md +++ b/docs/admin-guide.md @@ -683,7 +683,7 @@ All commands output JSON when piped. Use `--json` to force it, `--pretty` to for ## Backups -This is the part everyone skips and regrets. famstack ships an opt-in backup stacklet for the irreplaceable file-level data (photo originals, scanned documents). It does not yet cover stacklet databases or your config files; for those, layer it with Time Machine or a periodic tar. +This is the part everyone skips and regrets. famstack ships an opt-in backup stacklet that covers the irreplaceable data: photo originals, scanned documents, and everything in your family chat including the voice messages. Immich's and Paperless's databases are not covered yet, so layer it with Time Machine or a periodic tar if you want full coverage today. ### The backup stacklet @@ -695,15 +695,75 @@ Run `stack up backup`. You need an APFS-formatted external drive plugged in. The |---|---| | Immich photo originals | `/Volumes//data/photos-library/` | | Paperless archived PDFs | `/Volumes//data/docs-media/` | +| Matrix uploads: voice messages, photos, files | `/Volumes//data/messages-media/` | +| Matrix timeline, as dated snapshots | `/Volumes//data/messages-synapse/` | -Postgres databases for both are not backed up yet. You get your files back but lose albums, tags, custom fields, and saved views. Pg-dump snapshots will ship as `[[backup.snapshot]]` in a later release. +Immich's and Paperless's Postgres databases are still not covered. You get those files back but lose albums, tags, custom fields and saved views. They use the same snapshot mechanism the chat server already uses, so wiring them up is a small change rather than a new design. **How the protection works** Every file written to the archive gets the kernel `uchg` flag. macOS refuses to modify or delete uchg files, even with `sudo`. `rsync --ignore-existing` means files already in the archive are skipped on every run, so the backup is append-only by design and accidental `rm -rf` on your main system cannot propagate. +A second check watches the sources themselves. Before each run the engine compares every source against **the number of files it held on the previous run**, recorded in its own history. A source that is suddenly empty, or that has lost more than half its files, aborts the whole run so you look before anything else happens. There is nothing to configure: the baseline is the source's own past, so it means the same thing whether you have fifty files or five hundred thousand. A source that has never had any data, like the chat media store before anyone sends a photo, is simply skipped and reported. + A **canary file** is a tripwire (named after the canary miners used to take underground to detect bad air). famstack plants a small file with known contents inside `~/famstack-data`; before every sync the engine reads it and refuses to proceed if the contents have changed. If something has been encrypting or modifying files under the data directory, the canary will not match what was planted and the sync aborts before opening the archive, so the corrupted state cannot propagate. +**Database snapshots** + +Files that never change are easy: rsync copies them and the kernel locks them. A database is the opposite. Its files change under you continuously, and a copy taken mid-write will not restore. So databases are dumped instead. + +Before each sync the backup stacklet runs `pg_dump` against every database that declares itself, and packs the dump into one dated `.tar.gz`. `pg_dump` takes its own consistent view, so nothing stops and nobody gets logged out while it runs. The tarballs then ride to the archive disk as an ordinary append-only source: each run adds one, no run ever touches an older one. + +The chat server is the first thing wired up, and it is the one that matters most. The media store holds your family's actual voice recordings; the database is what makes them messages. Without it you would restore a folder of anonymous audio blobs with no sender, no room and no date. + +A Synapse snapshot is around 200 KB and contains: + +| File | Why | +|---|---| +| `synapse.sql` | The whole timeline: rooms, messages, who said what when | +| `homeserver.yaml` | Config, and the secret that keeps existing logins valid | +| `*.signing.key` | The server's identity | +| `MANIFEST.json` | What this file is, what produced it, and how to put it back | + +The manifest also records the exact image versions and digests that were running when the snapshot was taken. A dump only restores into something compatible with what wrote it, and the failure is not subtle: a Paperless 3.x database will not boot under 2.x, and there is no downgrade. The digest matters more than the tag, because `latest` names a different image every month and nothing identifiable in five years. Nothing reads this yet. It is recorded now because it is the one part of a snapshot that cannot be added afterwards. + +Those config files carry live secrets, including the database password. That is deliberate, because a dump without them restores a server nobody can log into, and it is one more reason the archive disk is a physical object you keep somewhere safe. + +The last seven tarballs are kept on the internal disk; the archive disk keeps every one ever made. Pruning locally is safe precisely because the archive never deletes. + +**Restoring the chat server** + +Recover the snapshot and the media together, and do the database first. Media the database does not know about is harmless; a message whose recording is missing is broken. + +```bash +# 1. Take the snapshot off the archive disk and unpack it +sudo chflags nouchg /Volumes//data/messages-synapse/synapse-.tar.gz +cp /Volumes//data/messages-synapse/synapse-.tar.gz ~/ +tar xzf ~/synapse-.tar.gz -C ~/restore/ + +# 2. Stop the homeserver so nothing writes while you work +./stack down messages + +# 3. Put the database back +docker start stack-messages-db +docker exec stack-messages-db dropdb -U synapse --if-exists synapse +docker exec stack-messages-db createdb -U synapse synapse +docker exec -i stack-messages-db psql -U synapse -d synapse < ~/restore/synapse.sql + +# 4. Put the config and identity back +cp ~/restore/homeserver.yaml ~/restore/*.signing.key \ + ~/famstack-data/messages/synapse/ + +# 5. Put the recordings back +sudo chflags -R nouchg /Volumes//data/messages-media/ +cp -R /Volumes//data/messages-media/* \ + ~/famstack-data/messages/synapse/media_store/local_content/ + +./stack up messages +``` + +`MANIFEST.json` inside every tarball repeats the essentials, so this works even years from now with no famstack and no documentation to hand. + **Daily operation** ```bash @@ -788,7 +848,7 @@ git pull ./stack up docs ``` -The backup stacklet does not cover this. It archives files, not the Postgres database, so a `stack backup sync` alone will not get you back to 2.x. +The backup stacklet does not cover this yet. It archives Paperless's files but not its Postgres database, so a `stack backup sync` alone will not get you back to 2.x. (Database snapshots exist and are wired up for the chat server; Paperless is next.) **If Watchtower already moved you to 3.x.** Older famstack releases tracked the `:latest` Paperless tag, and Watchtower's nightly pull rolled some instances from 2.x straight to 3.0 without asking. If that happened to you, your database is already migrated and pinning the image back to 2.20.15 will not start. You need a backup taken before the 3.x boot. Without one, staying on 3.x is the only option, which this release makes the supported path. Paperless is now pinned to an exact tag, so Watchtower can still deliver 3.0.x patches but can no longer jump a major version on its own. diff --git a/docs/stack-reference.md b/docs/stack-reference.md index cb803c95..00fa3b0f 100644 --- a/docs/stack-reference.md +++ b/docs/stack-reference.md @@ -375,16 +375,23 @@ job. ```toml # stacklets/photos/stacklet.toml [[backup.archive]] -name = "library" -path = "{data_dir}/photos/library/library" -min_files = 10 +name = "library" +path = "{data_dir}/photos/library/library" ``` | Field | Description | |---|---| | `name` | Short slug for this source. Combined with the stacklet id, this becomes the global source id (`photos/library`). Used in `stack backup status` output and (future) `--source=` selection. | | `path` | Filesystem path to sync. Template variables from the rendered environment are available (`{data_dir}`, etc.). | -| `min_files` | Coarse ransomware smoke test. The engine counts files at `path` before syncing and refuses if the count is below this. The canary file is the precise tripwire; this is the dumb-and-cheap secondary check. Keep low enough that fresh installs don't trip it. | + +There is no threshold to declare. The engine judges each source against +the number of files it held on the previous run, recorded in its own +history, and refuses to sync one that is suddenly empty or has lost more +than half its files. This replaced a `min_files` constant, which could +not work: a number written at authoring time cannot know the scale of the +household it guards, so a library of 50,000 photos reduced to 11 passed +`min_files = 10` without complaint. A source that has never held anything +is skipped rather than failing the run. **`[[backup.archive]]`** declares an append-only store: files are added, never modified, never deleted. The engine commits to kernel-enforced @@ -393,10 +400,41 @@ genuinely append-only (photo originals, archived PDFs), this is the right section. (Storage-industry vocabulary calls this WORM — Write Once Read Many.) -**`[[backup.snapshot]]`** is reserved for time-stamped point-in-time -captures of mutable state (Postgres dumps, Docker volume tarballs). Not -yet implemented — declare an `archive` section today; a `snapshot` -section will be added later when DB-restore semantics ship. +**`[[backup.snapshot]]`** declares mutable state that cannot be rsynced. +A database changes under you continuously and a copy taken mid-write will +not restore, so it is dumped instead: one `pg_dump` per run, packed with +whatever small files must travel beside it, into a dated `.tar.gz`. Each +run adds a tarball and never touches an older one, which gives mutable +state the same append-only shape an archive has. + +```toml +# stacklets/messages/stacklet.toml +[[backup.snapshot]] +name = "synapse" +postgres = { container = "stack-messages-db", database = "synapse", user = "synapse" } +include = ["{data_dir}/messages/synapse/homeserver.yaml", + "{data_dir}/messages/synapse/*.signing.key"] +``` + +| Field | Description | +|---|---| +| `name` | Short slug, as for an archive. Becomes the source id (`messages/synapse`). | +| `postgres` | How to capture the state, namespaced by what captures it. `container` is where the database runs (the dump goes through `docker exec`), `database` and `user` are what to dump and as whom. | +| `include` | Files that must travel with the dump for a restore to be possible. Globs allowed; a path this install never created is skipped rather than failing the snapshot. | + +The capture key is namespaced so a stacklet keeping state somewhere other +than Postgres can declare a snapshot later without the contract having to +pretend every database looks like this one. Driving a containerised +Postgres lives in `stack.postgres`, shared with whatever eventually +restores one, rather than inside the backup coordinator where only it +could reach it. + +Snapshots run before the sync, so a dump is never newer than the files it +references. `pg_dump` takes its own consistent view, so nothing stops and +nobody is logged out while it runs. The tarball records the image +versions and digests that produced it; nothing reads that yet, but it is +the one part of a snapshot that cannot be added afterwards, and a restore +has to be able to refuse an incompatible target. A stacklet may declare zero, one, or several entries of each kind. Sources flow to every configured target whose engine supports the declared diff --git a/lib/stack/postgres.py b/lib/stack/postgres.py new file mode 100644 index 00000000..1e86c3c3 --- /dev/null +++ b/lib/stack/postgres.py @@ -0,0 +1,96 @@ +"""pg_dump and psql against a Postgres running in a container. + +Three stacklets keep their state in Postgres: messages (the Matrix +timeline), docs (Paperless) and photos (Immich). Each needs the same two +operations, extracting a consistent dump and loading one back, so the +invocations live here rather than inside whichever component needed them +first. The backup coordinator calls `dump` today and a restore path will +call `restore`. + +Command construction is separated from execution so the exact arguments +can be asserted in tests without a running database. +""" + +from __future__ import annotations + +import subprocess +from typing import List + + +class PostgresError(RuntimeError): + """A pg_dump or psql invocation that exited non-zero.""" + + +# ── Commands ─────────────────────────────────────────────────────────────── + +def dump_command(container: str, database: str, user: str) -> List[str]: + """Arguments for a plain-SQL dump of `database`. + + Plain SQL rather than pg_dump's custom format, because the output is + read back by `psql`, which exists in every Postgres image. A custom + dump would additionally require a `pg_restore` of compatible version. + """ + return ["docker", "exec", container, + "pg_dump", "-U", user, "-d", database] + + +def restore_command(container: str, database: str, user: str) -> List[str]: + """Arguments for loading a plain-SQL dump from stdin. + + `-i` keeps the container's stdin connected. Without it `psql` + receives no input and blocks indefinitely. + """ + return ["docker", "exec", "-i", container, + "psql", "-U", user, "-d", database] + + +def version_command(container: str, database: str, user: str) -> List[str]: + """Arguments for reading the server version. + + `-t` drops the column header and `-A` the alignment padding, leaving + the bare value on stdout. + """ + return ["docker", "exec", container, + "psql", "-U", user, "-d", database, + "-tAc", "show server_version;"] + + +# ── Execution ────────────────────────────────────────────────────────────── + +def dump(container: str, database: str, user: str) -> bytes: + """Return a consistent SQL dump of `database`. + + `pg_dump` reads from a single MVCC snapshot, so the service keeps + running and accepting writes for the duration. + + Raises `PostgresError` carrying the beginning of stderr, which is + where Postgres reports an unknown database or a failed authentication. + """ + proc = subprocess.run( + dump_command(container, database, user), capture_output=True, + ) + if proc.returncode != 0: + detail = proc.stderr.decode(errors="replace").strip()[:400] + raise PostgresError( + f"pg_dump failed for {database} in {container}: {detail}" + ) + return proc.stdout + + +def server_version(container: str, database: str, user: str) -> str: + """Return the server version, or an empty string if it cannot be read. + + This is recorded as metadata beside a dump rather than used for any + decision, so an unreachable or stopped container is reported as + unknown instead of raising. + """ + try: + proc = subprocess.run( + version_command(container, database, user), + capture_output=True, timeout=30, + ) + except (OSError, subprocess.SubprocessError): + return "" + if proc.returncode != 0: + return "" + return proc.stdout.decode(errors="replace").strip() diff --git a/stacklets/backup/README.md b/stacklets/backup/README.md index 3aac2709..3e0e6542 100644 --- a/stacklets/backup/README.md +++ b/stacklets/backup/README.md @@ -20,21 +20,55 @@ with explicit guarantees: | Engine | Status | What it does | |---|---|---| -| `external-disk` | scaffolded, port pending | rsync + chflags uchg on attached APFS disk | +| `external-disk` | shipped | rsync + chflags uchg on attached APFS disk | | `restic` | planned | encrypted, deduplicated, snapshotted offsite (S3/B2) | -Sources are discovered from other stacklets via a manifest contract. -Every stacklet that declares `[[backup.archive]]` (an append-only store) -in its `stacklet.toml` contributes one source path to the next sync: +Sources are discovered from other stacklets via a manifest contract, in +two kinds. + +**`[[backup.archive]]`** is an append-only store: files are added, never +modified, never deleted. It syncs incrementally, so nothing is ever +re-copied. ```toml # stacklets/photos/stacklet.toml [[backup.archive]] -name = "library" -path = "{data_dir}/photos/library/library" -min_files = 10 +name = "library" +path = "{data_dir}/photos/library/library" +``` + +**`[[backup.snapshot]]`** is for state that cannot be rsynced. A database +changes under you continuously and a copy taken mid-write will not +restore, so it is dumped instead: one consistent `pg_dump` per run, +packed with whatever small files must travel beside it, into a dated +`.tar.gz`. Each run adds a tarball and never touches an older one, which +turns mutable state into the same append-only shape the vault keeps. + +```toml +# stacklets/messages/stacklet.toml +[[backup.snapshot]] +name = "synapse" +postgres = { container = "stack-messages-db", database = "synapse", user = "synapse" } +include = ["{data_dir}/messages/synapse/homeserver.yaml", + "{data_dir}/messages/synapse/*.signing.key"] ``` +The capture key is namespaced by what does the capturing, so a stacklet +on something other than Postgres can declare a snapshot without the +contract assuming every database is this one. The invocation itself lives +in `stack.postgres`, next to the restore half, rather than inside this +coordinator. + +Snapshots run before the sync, so a dump is never newer than the media it +references, and their output directory then joins the source list as an +ordinary append-only source. `pg_dump` takes its own consistent view, so +nothing stops while it runs. + +Each snapshot records the image versions and digests that produced it. +Nothing reads that yet; it is captured because it is the one part of a +snapshot that cannot be added afterwards, and a restore has to be able to +refuse an incompatible target. + Targets are configured in `stack.toml`. Today the only target is the attached disk: @@ -46,7 +80,7 @@ engine = "external-disk" disk = "backup-vault" ``` -Routing: every `[[backup.archive]]` source flows to every target whose +Routing: every source, archive or snapshot, flows to every target whose engine supports append-only semantics. Adding a second target later (offsite restic) is purely additive — no manifest change on photos/docs. @@ -62,6 +96,37 @@ restore (`stack backup restore --source=…`) are intentionally not in v1 — they'll layer on once the engine port lands and the manifest contract has been exercised on at least one production sync. +## Guarding the sources + +Two checks run before anything is written. + +The **canary** is a file with known contents planted under the data +directory. If it does not read back as planted, something is modifying +files in place and the sync aborts before the archive disk is opened. + +The **shrink check** compares every source against the number of files it +held on the previous run, recorded in the engine's own history. A source +that is suddenly empty, or has lost more than half its files, aborts the +run. + +That baseline used to be `min_files`, a constant each stacklet declared. +It could not work: a threshold written at authoring time cannot know the +scale of the household it guards, so a library of 50,000 photos reduced +to 11 passed `min_files = 10` without complaint. A source's own previous +count means the same thing at any size and needs no declaration. + +A source that has never held anything is skipped rather than failing the +run, because a stacklet with no data yet must not cost the household +every other backup. Nothing is at risk in that case: the engine syncs +with `--ignore-existing` and never `--delete`, so an empty source copies +nothing and the vault keeps everything it already had. **That property is +load-bearing for the skip.** An engine that ever gains `--delete` has to +revisit this. + +Snapshot staging directories are marked `rolling` by the coordinator and +exempt from the shrink check: they are pruned to a fixed window on +purpose, so shrinking is their normal operation. + ## Destroy semantics `stack destroy backup` removes the backup *tooling* — never the @@ -111,9 +176,18 @@ is the documented path. ## Status -This stacklet is currently **scaffold only**. The hooks and CLI files -raise `NotImplementedError`. The next step is porting `vault-sync.sh` -from `family-server/backup/` into `engines/external-disk/`, with two -adaptations: source discovery via the manifest contract, and Matrix -notifications via the local `stacker-bot` instead of the legacy -`kit-control-bot`. +The `external-disk` engine is ported and in use. Archives and snapshots +both sync; Matrix notifications go through the local `stacker-bot`. + +Backed up today: Immich photo originals, Paperless archived PDFs, Matrix +uploads (voice messages, photos, files), and the Matrix timeline as +snapshots. + +Not yet: Immich's and Paperless's databases. Both use the same +`[[backup.snapshot]]` contract the chat server already uses, so each is a +declaration plus a restore check rather than new design work. Without +them you get those files back but lose albums, tags, custom fields and +saved views. + +Restore is still manual, by design for now. `stack backup restore` and +`on_restore` hooks are the planned shape. diff --git a/stacklets/backup/cli/_orchestrator.py b/stacklets/backup/cli/_orchestrator.py index b151c79c..c86097ae 100644 --- a/stacklets/backup/cli/_orchestrator.py +++ b/stacklets/backup/cli/_orchestrator.py @@ -50,7 +50,10 @@ class SourceRecord: display: str # Human-readable, e.g. "Photos" src_path: Path # Absolute path on internal SSD (post-rendering) vault_subdir: str # Relative path under /Volumes// - min_files: int # Coarse ransomware guard threshold + # Pruned on purpose (a snapshot staging area keeps a fixed window), so + # the engine must not read its shrinking as data loss. Set here, never + # by a manifest. + rolling: bool = False @dataclass @@ -116,17 +119,11 @@ def discover_archive_sources( # will surface the problem with a useful error. rendered_path = raw_path - try: - min_files = int(archive.get("min_files", 1)) - except (TypeError, ValueError): - min_files = 1 - sources.append(SourceRecord( id=f"{stacklet_id}/{name}", display=stacklet_display, src_path=Path(rendered_path), vault_subdir=f"data/{stacklet_id}-{name}", - min_files=min_files, )) return sources @@ -180,10 +177,10 @@ def serialize_sources_env(sources: List[SourceRecord]) -> str: """Format SourceRecords for the engine's ``$SOURCES`` env var. The engine's :func:`parse_sources` expects newline-separated, - pipe-delimited records: ``id|display|src_path|vault_subdir|min_files``. + pipe-delimited records: ``id|display|src_path|vault_subdir|rolling``. """ return "\n".join( - f"{s.id}|{s.display}|{s.src_path}|{s.vault_subdir}|{s.min_files}" + f"{s.id}|{s.display}|{s.src_path}|{s.vault_subdir}|{1 if s.rolling else 0}" for s in sources ) diff --git a/stacklets/backup/cli/_snapshot.py b/stacklets/backup/cli/_snapshot.py new file mode 100644 index 00000000..d1ba294b --- /dev/null +++ b/stacklets/backup/cli/_snapshot.py @@ -0,0 +1,361 @@ +"""Snapshots: capturing state that cannot be rsynced. + +An archive source is a directory whose files are written once and never +changed, which rsync can copy incrementally. A database is not: its files +are rewritten continuously, and a copy taken while it is running does not +restore. Such state is dumped instead. + +A snapshot is one dump per run, packed with the files that must accompany +it, into a dated tarball. Runs add tarballs and never modify an existing +one, which gives mutable state the same append-only shape the vault +already stores. + +The tarballs on the internal disk are a staging area; the copy on the +vault is the backup. `prune_snapshots` can therefore delete local +tarballs freely, because the engine syncs with `--ignore-existing` and +never passes `--delete`. +""" + +from __future__ import annotations + +import glob +import io +import json +import os +import subprocess +import sys +import tarfile +import tempfile +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Callable, List, Optional + +try: + import tomllib +except ModuleNotFoundError: # pragma: no cover — py < 3.11 fallback + from stack._vendor import tomli as tomllib # type: ignore + +from stack import postgres + +from _orchestrator import SourceRecord + + +# Tarballs retained on the internal disk. The vault retains all of them, +# so this bounds local disk use only. +DEFAULT_KEEP = 7 + + +@dataclass +class SnapshotSpec: + """One ``[[backup.snapshot]]`` entry, after template rendering.""" + + id: str # "{stacklet_id}/{name}", e.g. "messages/synapse" + display: str # Human-readable, e.g. "Messages" + name: str # "synapse" + # Capture parameters, keyed by the mechanism that reads them. The + # namespace lets a stacklet storing state elsewhere declare a + # snapshot without the contract assuming Postgres. + postgres: dict = field(default_factory=dict) + include: List[str] = field(default_factory=list) # extra files (globs ok) + + @property + def container(self) -> str: + return self.postgres.get("container", "") + + @property + def database(self) -> str: + return self.postgres.get("database", "") + + @property + def user(self) -> str: + return self.postgres.get("user", "") + + @property + def stacklet_id(self) -> str: + return self.id.split("/", 1)[0] + + @property + def subdir(self) -> str: + """Directory holding this snapshot's tarballs, used both on the + internal disk and under the vault's `data/`. Qualified by stacklet + so two stacklets choosing the same `name` do not collide.""" + return f"{self.stacklet_id}-{self.name}" + + +# ── Discovery ────────────────────────────────────────────────────────────── + +def discover_snapshots( + repo_root: Path, instance_dir: Path, data_dir: Path, +) -> List[SnapshotSpec]: + """Return the ``[[backup.snapshot]]`` entries of enabled stacklets. + + Follows the same rules as ``discover_archive_sources``: a stacklet + counts as enabled when `.stack/{id}.setup-done` exists, and + `{data_dir}` is the only template variable rendered into paths. + """ + stacklets_dir = repo_root / "stacklets" + if not stacklets_dir.is_dir(): + return [] + + template_vars = {"data_dir": str(data_dir)} + specs: List[SnapshotSpec] = [] + + for manifest_path in sorted(stacklets_dir.glob("*/stacklet.toml")): + stacklet_id = manifest_path.parent.name + if not (instance_dir / ".stack" / f"{stacklet_id}.setup-done").exists(): + continue + + try: + with open(manifest_path, "rb") as f: + manifest = tomllib.load(f) + except (OSError, tomllib.TOMLDecodeError): + continue + + entries = manifest.get("backup", {}).get("snapshot", []) + display = manifest.get("name", stacklet_id) + + for entry in entries: + name = entry.get("name", "default") + includes = [] + for raw in entry.get("include", []): + try: + includes.append(raw.format(**template_vars)) + except (KeyError, IndexError): + # An unrecognised variable is kept verbatim, so the + # problem appears later as a file that did not match + # rather than as an exception during discovery. + includes.append(raw) + specs.append(SnapshotSpec( + id=f"{stacklet_id}/{name}", + display=display, + name=name, + postgres=entry.get("postgres", {}) or {}, + include=includes, + )) + + return specs + + +# ── Taking one ───────────────────────────────────────────────────────────── + +def pg_dump(spec: SnapshotSpec) -> bytes: + """Default capture for a spec declaring `postgres` parameters.""" + return postgres.dump(spec.container, spec.database, spec.user) + + +def container_versions(spec: SnapshotSpec) -> dict: + """Return the images and Postgres version present at snapshot time. + + A dump loads only into a compatible version of the application that + produced it, so a restore needs to know what that was. Nothing reads + this yet; it is recorded now because it describes a moment that has + passed by the time anything wants it. + + Both the tag and the digest are kept. Tags are mutable, so a + reference like `synapse:latest` resolves to different images over + time and only the digest identifies the exact one. + + Containers are found through the compose project label rather than a + list in the manifest, so a stacklet that gains a service does not + also have to remember to declare it here. + """ + project = f"stack-{spec.stacklet_id}" + names = _docker( + "ps", "--filter", f"label=com.docker.compose.project={project}", + "--format", "{{.Names}}", + ).split() + + containers = {} + for name in names: + image = _docker("inspect", name, "--format", "{{.Config.Image}}") + version = _docker( + "inspect", name, "--format", + '{{index .Config.Labels "org.opencontainers.image.version"}}', + ) + digest = _docker( + "image", "inspect", image, "--format", + "{{if .RepoDigests}}{{index .RepoDigests 0}}{{end}}", + ) + entry = {"image": image} + if version and version != "": + entry["version"] = version + if "@" in digest: + entry["digest"] = digest.split("@", 1)[1] + containers[name] = entry + + versions: dict = {"containers": containers} + if spec.container: + pg = postgres.server_version(spec.container, spec.database, spec.user) + if pg: + versions["postgres"] = pg + return versions + + +def _docker(*args: str) -> str: + """Run one docker command and return its stripped stdout. + + Any failure yields an empty string. Callers use this for metadata + only, where an absent value is preferable to an aborted snapshot. + """ + try: + proc = subprocess.run(["docker", *args], capture_output=True, timeout=30) + except (OSError, subprocess.SubprocessError): + return "" + if proc.returncode != 0: + return "" + return proc.stdout.decode(errors="replace").strip() + + +def take_snapshot( + spec: SnapshotSpec, + out_root: Path, + *, + dump: Optional[Callable[[SnapshotSpec], bytes]] = None, + versions: Optional[Callable[[SnapshotSpec], dict]] = None, + now: Optional[time.struct_time] = None, +) -> Path: + """Write one dated tarball for `spec` and return its path. + + The tarball is assembled under a temporary name and moved into place + only once complete. A partially written file left in the output + directory would be picked up by the next sync and locked immutable on + the vault, where it could not be replaced. + """ + dump = dump or pg_dump + versions = versions or container_versions + stamp = time.strftime("%Y%m%dT%H%M%SZ", now or time.gmtime()) + + # Taken first, so a failure propagates before any file exists. + sql = dump(spec) + + out_dir = out_root / spec.subdir + out_dir.mkdir(parents=True, exist_ok=True) + # The timestamp has second resolution, so two runs within the same + # second would otherwise produce the same name. A suffix keeps them + # distinct rather than letting the second overwrite the first. + target = out_dir / f"{spec.name}-{stamp}.tar.gz" + attempt = 2 + while target.exists(): + target = out_dir / f"{spec.name}-{stamp}-{attempt}.tar.gz" + attempt += 1 + + dump_name = f"{spec.database or spec.name}.sql" + included: List[str] = [] + + # Recorded as an empty mapping rather than omitted, which + # distinguishes a snapshot whose versions could not be read from one + # written before versions were recorded at all. + try: + recorded = versions(spec) + except Exception as e: + logger_warn(f"could not record versions for {spec.id}: {e}") + recorded = {} + + fd, tmp_path = tempfile.mkstemp(dir=out_dir, suffix=".tar.gz.partial") + os.close(fd) + try: + with tarfile.open(tmp_path, "w:gz") as tar: + _add_bytes(tar, dump_name, sql) + for pattern in spec.include: + # Patterns that match nothing are skipped. An install may + # legitimately lack a file another one has, and that is + # not a reason to discard the dump. + for path in sorted(glob.glob(pattern)): + p = Path(path) + if p.is_file(): + tar.add(p, arcname=p.name) + included.append(p.name) + _add_bytes(tar, "MANIFEST.json", json.dumps( + _manifest(spec, dump_name, stamp, included, recorded), + indent=2, + ).encode()) + os.replace(tmp_path, target) + except BaseException: + Path(tmp_path).unlink(missing_ok=True) + raise + + return target + + +def _add_bytes(tar: tarfile.TarFile, name: str, payload: bytes) -> None: + info = tarfile.TarInfo(name) + info.size = len(payload) + info.mtime = int(time.time()) + tar.addfile(info, io.BytesIO(payload)) + + +def logger_warn(msg: str) -> None: + print(f" warning: {msg}", file=sys.stderr) + + +def _manifest( + spec: SnapshotSpec, dump_name: str, stamp: str, included: List[str], + versions: dict, +) -> dict: + """Describe the tarball for a reader who does not have this code. + + `restore` holds a literal command rather than a reference to + documentation, so the tarball remains self-describing if it is opened + somewhere the project is not available. + """ + return { + "famstack_snapshot": 1, + "stacklet": spec.stacklet_id, + "name": spec.name, + "database": spec.database, + "dump_file": dump_name, + "included_files": included, + "versions": versions, + "taken_at": time.strftime( + "%Y-%m-%dT%H:%M:%SZ", time.strptime(stamp, "%Y%m%dT%H%M%SZ"), + ), + "restore": ( + f"createdb -U {spec.user} {spec.database} && " + f"psql -U {spec.user} -d {spec.database} -f {dump_name}" + ), + "note": ( + "Restore the database first, then the media files from the " + "matching archive. Media without a database row is harmless; " + "a database row without its media is a broken message." + ), + } + + +# ── Feeding the vault ────────────────────────────────────────────────────── + +def snapshot_source(spec: SnapshotSpec, out_root: Path) -> SourceRecord: + """Return the engine source carrying this snapshot's tarballs. + + Marked `rolling`, because `prune_snapshots` keeps the directory at a + fixed size. Without that flag the engine's shrink check would read + routine pruning as data loss. + """ + return SourceRecord( + id=spec.id, + display=spec.display, + src_path=out_root / spec.subdir, + vault_subdir=f"data/{spec.subdir}", + rolling=True, + ) + + +# ── Keeping the internal disk honest ─────────────────────────────────────── + +def prune_snapshots(directory: Path, keep: int = DEFAULT_KEEP) -> List[Path]: + """Delete all but the newest `keep` tarballs; return those removed. + + Only the internal disk is affected. The engine syncs with + `--ignore-existing` and never `--delete`, so tarballs already on the + vault are unaffected by anything removed here. + + Ordering by name is ordering by time, because the timestamp is + fixed-width and UTC. + """ + if not directory.is_dir(): + return [] + tarballs = sorted(directory.glob("*.tar.gz")) + doomed = tarballs[:-keep] if keep > 0 else tarballs + for path in doomed: + path.unlink(missing_ok=True) + return doomed diff --git a/stacklets/backup/cli/sync.py b/stacklets/backup/cli/sync.py index 6328216a..40999e24 100644 --- a/stacklets/backup/cli/sync.py +++ b/stacklets/backup/cli/sync.py @@ -2,7 +2,15 @@ Discovers ``[[backup.archive]]`` sources from every enabled stacklet (append-only stores) and routes them through every configured -``[backup.targets.*]`` engine. Each engine writes a structured result +``[backup.targets.*]`` engine. + +``[[backup.snapshot]]`` declarations run first. Each writes one dated +tarball (a database dump plus the small files that must travel with it) +and then joins the source list, so a mutable database reaches the vault +in the same append-only shape as everything else. Snapshots run *before* +the sync so a dump is never newer than the media it references: media +without a database row is a harmless orphan, a row without its media is +a broken message. Each engine writes a structured result to ``$BACKUP_DATA_DIR/logs/history.jsonl``; the orchestrator reads that file, formats a per-target summary, and posts it to the ``#famstack`` room as ``stacker-bot``. @@ -35,6 +43,12 @@ invoke_engine, read_latest_run, ) +from _snapshot import ( + discover_snapshots, + prune_snapshots, + snapshot_source, + take_snapshot, +) # MatrixClient lives in the messages stacklet — the canonical Matrix # interface for any CLI plugin that needs to post. Cross-stacklet import @@ -44,6 +58,46 @@ sys.path.insert(0, str(_messages_cli)) +def _take_snapshots( + repo_root: Path, instance_dir: Path, data_dir: Path, + backup_data_dir: Path, *, dry_run: bool, +) -> tuple[list, bool]: + """Take every declared snapshot; return its sources and whether any failed. + + A stacklet whose database will not dump must not cost the household + its photos, so a failure here is reported and the run continues with + the archives. The caller turns that into a non-zero exit, because a + backup that silently captured less than it was asked to is the exact + failure mode backups are supposed to protect against. + """ + specs = discover_snapshots(repo_root, instance_dir, data_dir) + if not specs: + return [], False + + out_root = backup_data_dir / "snapshots" + sources, failed = [], False + + print("\n Snapshots") + for spec in specs: + if dry_run: + print(f" {spec.display}: would snapshot {spec.database}") + # No tarball was written, so the directory may not exist and + # the engine's own --dry-run would trip over a missing source. + continue + try: + path = take_snapshot(spec, out_root) + except Exception as e: + print(f" {spec.display}: snapshot FAILED — {e}", file=sys.stderr) + failed = True + continue + size_kb = max(1, path.stat().st_size // 1024) + print(f" {spec.display}: {path.name} ({size_kb} KB)") + prune_snapshots(out_root / spec.subdir) + sources.append(snapshot_source(spec, out_root)) + + return sources, failed + + def _parse_args(argv: list) -> argparse.Namespace: parser = argparse.ArgumentParser(prog="stack backup sync", description=HELP) parser.add_argument("--dry-run", action="store_true", @@ -167,7 +221,13 @@ def run(args, stacklet, config): "block to stack.toml (see stack.example.toml)." } + snapshot_sources, snapshot_failed = _take_snapshots( + repo_root, instance_dir, data_dir, backup_data_dir, + dry_run=parsed.dry_run, + ) + sources = discover_archive_sources(repo_root, instance_dir, data_dir) + sources = snapshot_sources + sources if not sources: return { "error": "No backup sources discovered. No enabled stacklet declares " @@ -209,6 +269,12 @@ def run(args, stacklet, config): print(f" [{target.name}] notification skipped: {notify_error}", file=sys.stderr) + if snapshot_failed: + # The archives still synced, which is the right call: a database + # that would not dump must not cost you the photos. But the run + # did not capture everything it was asked to, and saying "ok" + # here is how a backup quietly stops being one. + return {"error": "One or more snapshots failed; the archives synced"} if any_failed: return {"error": "One or more targets failed; see history.jsonl for details"} return { diff --git a/stacklets/backup/engines/external-disk/sync.py b/stacklets/backup/engines/external-disk/sync.py index 3f6e959d..9f26f1b3 100755 --- a/stacklets/backup/engines/external-disk/sync.py +++ b/stacklets/backup/engines/external-disk/sync.py @@ -38,7 +38,7 @@ mount point is ``/Volumes/``. ``SOURCES`` Required. Newline-separated records, pipe-delimited:: - |||| + |||| ``TZ`` Optional. Affects log timestamps. ================== ========================================================== @@ -79,6 +79,14 @@ NETWORK_FILESYSTEMS = frozenset({"smbfs", "nfs", "afpfs"}) REMOVABLE_FILESYSTEMS = frozenset({"msdos", "exfat", "ntfs"}) +# How much of last run's file count a source must still hold. Deleting +# things is normal household behaviour, so the guard is deliberately loose +# — it is looking for catastrophe (a wipe, an encryption run, a mount that +# came up empty), not policing somebody's tidying. Judged against the +# source's own previous count, so it means the same thing whether the +# household has fifty files or five hundred thousand. +SHRINK_FLOOR = 0.5 + # rsync exit codes that aren't real failures for incremental backups: # 0 = success, 23 = partial transfer (vanished files during sync), # 24 = source files vanished. All acceptable in our context. @@ -95,7 +103,10 @@ class Source: display: str # human-readable label, e.g. "Photos" src_path: Path # absolute path on the internal SSD vault_subdir: str # relative path under /Volumes// - min_files: int # coarse ransomware guard threshold + # A rolling source is pruned on purpose (a snapshot staging area + # keeps a fixed window), so shrinking is its normal operation rather + # than a loss. Set by the orchestrator, never by a manifest. + rolling: bool = False @dataclass @@ -107,6 +118,11 @@ class SourceResult: status: str # "ok" | "FAILED" | "skipped" total_files: int # files on vault after this run new_files: int # files added this run + # What the *source* held this run. This is the baseline the next run + # judges itself against, and the reason no manifest has to guess a + # threshold. Cumulative vault counts cannot serve: a household deletes + # things over years, so source and vault drift apart legitimately. + source_files: int = 0 @dataclass @@ -190,7 +206,10 @@ def parse_sources(sources_env: str) -> List[Source]: Records are newline-separated, fields pipe-delimited:: - |||| + |||| + + ``rolling`` is "1" for a source that is pruned on purpose (a snapshot + staging area) and "0" otherwise. Pipe over colon because paths can (rarely) contain colons on macOS but never pipes. Empty input or malformed records raise @@ -208,17 +227,13 @@ def parse_sources(sources_env: str) -> List[Source]: f"Malformed source record: {line!r} " f"(expected 5 pipe-delimited fields, got {len(parts)})" ) - id_, display, src_path, vault_subdir, min_files = parts - try: - min_files_int = int(min_files) - except ValueError: - raise SyncAborted(f"min_files must be an integer in {line!r}") + id_, display, src_path, vault_subdir, rolling = parts records.append(Source( id=id_, display=display, src_path=Path(src_path), vault_subdir=vault_subdir, - min_files=min_files_int, + rolling=rolling.strip() == "1", )) if not records: raise SyncAborted("No sources provided — $SOURCES is empty.") @@ -295,39 +310,107 @@ def verify_canary(canary_file: Path) -> None: # ── Preflight ────────────────────────────────────────────────────────────── -def preflight_check_sources(sources: List[Source]) -> None: - """Each source must exist and contain at least ``min_files`` entries. +def previous_source_counts(latest_run: Optional[dict]) -> dict: + """What each source held on the previous run, by source id. - The canary catches "every file got encrypted in place"; this catches - "the directory got ``rm -rf``'d." Together they're a layered smoke - test that refuses to propagate a broken source to the vault. + Only a source that synced contributes one. A skipped source had no + data, and a failed source did not finish, so neither describes a + count the next run should be measured against. + """ + if not latest_run: + return {} + counts = {} + for entry in latest_run.get("sources", []): + if entry.get("status") != "ok": + continue + sid = entry.get("id") + count = entry.get("source_files") + if sid and isinstance(count, int): + counts[sid] = count + return counts + + +def preflight_check_sources( + sources: List[Source], previous: Optional[dict] = None, +) -> List[Source]: + """Return the sources worth syncing; abort on one that lost data. + + Each source is judged against **what it held on the previous run**, + read from the engine's own history. That baseline is the whole point. + The threshold used to be `min_files`, a constant a developer wrote in + a manifest while guessing at a household they would never see, and it + could only ever catch "dropped to almost exactly zero". A library of + 50,000 photos reduced to 11 sailed straight past `min_files = 10`, + which is precisely the disaster the check existed for. + + Against its own previous count, that same library is unmissable, at + any scale, with nothing for anyone to configure. + + Three outcomes: + + *Nothing there yet.* Empty or not created, and no baseline. The + stacklet simply has no data — Synapse does not create its media store + until the first upload — so it is skipped and the run continues. + Nothing is at risk: the engine syncs with ``--ignore-existing`` and + never ``--delete``, so an empty source copies nothing and the vault + keeps everything it held. + + *Lost most of it.* Below :data:`SHRINK_FLOOR` of last run's count, or + empty when it used to have files. The run aborts so a human looks + before anything else happens. + + *Fine.* Everything else, including growth and the ordinary deleting + people do. A rolling source (a snapshot staging area, pruned to a + fixed window on purpose) is never judged to have shrunk. + + The canary remains the separate, precise tripwire for "encrypted in + place", which no count can detect. """ header("Preflight checks") + previous = previous or {} + syncable: List[Source] = [] failures: List[str] = [] + for src in sources: - if not src.src_path.is_dir(): - error(f"{src.display}: source directory not found ({src.src_path})") - failures.append(src.display) + exists = src.src_path.is_dir() + count = count_files(src.src_path) if exists else 0 + was = previous.get(src.id) + + if count == 0 and not was: + warn( + f"{src.display}: nothing to back up yet " + f"({'empty' if exists else 'not created'}) — skipping" + ) continue - count = count_files(src.src_path) - if count < src.min_files: + + if count == 0: error( - f"{src.display}: only {count} files " - f"(minimum: {src.min_files}) — refusing to sync" + f"{src.display}: empty, but held {format_number(was)} files " + f"last run — refusing to sync" ) failures.append(src.display) - else: - info( - f"{src.display}: {format_number(count)} files " - f"(minimum: {src.min_files}) — ok" + continue + + if was and not src.rolling and count < was * SHRINK_FLOOR: + error( + f"{src.display}: {format_number(count)} files, down from " + f"{format_number(was)} last run — refusing to sync" ) + failures.append(src.display) + continue + + seen = f" (was {format_number(was)})" if was else "" + info(f"{src.display}: {format_number(count)} files{seen} — ok") + syncable.append(src) if failures: raise SyncAborted( - "Preflight failed — source directories missing or too few files" + "Preflight failed — a source lost files it had last run" ) + return syncable + # ── Mount vault ──────────────────────────────────────────────────────────── @@ -725,6 +808,7 @@ def sync_data( status="ok", total_files=after_count, new_files=new_count, + source_files=count_files(src.src_path), )) if dry_run: info( @@ -985,14 +1069,25 @@ def run_sync( print(f" {YELLOW}DRY RUN — no changes will be made{NC}") verify_canary(canary_file) - preflight_check_sources(sources) + # Each source is judged against what it held on the previous run, + # read from our own history. Sources with nothing in them are + # dropped rather than failing the run; they are still reported + # below, so an empty source is visible rather than silently absent. + previous = previous_source_counts(read_latest_run(history_path)) + syncable = preflight_check_sources(sources, previous) + skipped = [s for s in sources if s not in syncable] mount_vault(vault_disk, mount_point, args.dry_run) if not args.dry_run: probe_filesystem(mount_point) check_vault_space(mount_point) result.sources = sync_data( - sources, mount_point, log_path, args.dry_run, args.verbose + syncable, mount_point, log_path, args.dry_run, args.verbose + ) + result.sources.extend( + SourceResult(id=s.id, display=s.display, status="skipped", + total_files=0, new_files=0) + for s in skipped ) if any(r.status == "FAILED" for r in result.sources): result.success = False @@ -1002,7 +1097,7 @@ def run_sync( result.vault_size = measure_vault_size(mount_point) if args.verify: - verify_sync(sources, mount_point) + verify_sync(syncable, mount_point) eject_vault(vault_disk, args.dry_run, args.no_eject) diff --git a/stacklets/core/bot-runner/microbot.py b/stacklets/core/bot-runner/microbot.py index ff5e2ed2..c5052183 100644 --- a/stacklets/core/bot-runner/microbot.py +++ b/stacklets/core/bot-runner/microbot.py @@ -129,10 +129,9 @@ def __init__(self, homeserver: str, user_id: str, password: str, session_dir: st # via sync. The SyncResponse callback drains this set and fires # `on_room_joined` once nio has populated the room. self._pending_room_joins: set[str] = set() - # Speech-to-text for the decode in `_dispatch`. Built here rather - # than on demand because a missing whisper is a normal state, not - # an error: without it, audio simply stays audio and reaches no - # handler, the same as a message in a language we cannot read. + # Built at construction rather than on first use, because an + # absent whisper is an ordinary configuration, not a fault. + # Without it audio is simply never decoded and reaches no handler. try: self._transcriber = Transcriber.from_env(namespace=self.name) except LLMUnavailableError as e: @@ -141,8 +140,7 @@ def __init__(self, homeserver: str, user_id: str, password: str, session_dir: st "as audio", self.name, e, ) self._transcriber = None - # Optional polish on raw whisper output (punctuation, sentence - # breaks). Absent, the transcript still lands, just rougher. + # Optional. Without it the raw transcript is used unchanged. try: self._transcript_cleanup = LLM.from_env(namespace=self.name) except LLMUnavailableError: @@ -525,11 +523,10 @@ async def _dispatch(self, room_id: str, event) -> None: decoding = voice.is_voice(event) if decoding: - # Transcription is the one part of handling a message that can - # run for minutes, and it happens before any handler gets to - # signal it is working. A matched handler's wrap clears the - # indicator in its `finally`; we clear it ourselves on the two - # paths where no handler ever runs. + # Transcription can run for minutes and happens before any + # handler could raise the indicator itself. A handler that + # matches clears it in the wrap's `finally`; the two paths + # below clear it where no handler runs at all. await self._set_typing(room_id, on=True) event = await self._decode_voice(room_id, event) if event is None: @@ -545,14 +542,14 @@ async def _dispatch(self, room_id: str, event) -> None: await self._set_typing(room_id, on=False) async def _decode_voice(self, room_id: str, event): - """Turn a voice message into the text event it is, or None. - - None means the words could not be recovered — whisper is absent, - unreachable, or heard nothing. That is dispatched to nobody - rather than answered with an apology: the family can see their - own voice message sitting in the room, and a bot volunteering - "I could not hear that" in every room it is in, for audio nobody - was addressing to it, is noise. + """Return the text event this voice message decodes to, or None. + + None means the words could not be recovered, because whisper is + absent, unreachable, or found no speech. The event is then + dispatched to no handler rather than answered with an error: the + recording remains visible in the room, and every bot present + reporting the same failure for audio not addressed to it would + add noise without adding information. """ if self._transcriber is None: logger.debug( @@ -570,10 +567,10 @@ async def run() -> dict: if not audio: raise LLMError(f"could not download {url}") raw = await self._transcriber.transcribe(audio, filename=filename) - # whisper emits one unbroken lowercase run of words; the polish - # pass puts the sentences back without changing them. Both are - # kept: polishing again with a better model is cheap, and - # transcribing again is not. + # whisper returns an unpunctuated run of words; the polish + # pass restores sentence boundaries. Both forms are stored, + # since re-polishing later is cheap and re-transcribing is + # not. text = raw if raw.strip() and self._transcript_cleanup is not None: text = await Transcriber.polish(raw, self._transcript_cleanup) diff --git a/stacklets/core/bot-runner/voice.py b/stacklets/core/bot-runner/voice.py index d8cf9854..8a085311 100644 --- a/stacklets/core/bot-runner/voice.py +++ b/stacklets/core/bot-runner/voice.py @@ -1,28 +1,21 @@ -"""Voice messages are messages. The transport decodes them, not the bots. - -Someone holding the mic button and someone typing are doing the same -thing: putting words in the room. Only the encoding differs, so only the -transport should know about it. By the time a handler runs, an `m.audio` -event has already become an ordinary text event — same sender, same event -id, same thread — and every gate a typed message passes through (room -mode, thread ownership, mention, routing) applies to it unchanged. - -This is deliberately not a capability bots reach for. That was the earlier -design, and it grew an audio branch in every consumer, two whisper clients -in one process, and three places that could transcribe the same bytes. -Decoding belongs beside decryption: below everyone, done once. - -What survives the decode is provenance. Whisper is lossy in a way a -keyboard is not, so the text event carries a `dev.famstack.transcript` -block naming the audio it came from. Routing ignores it; the bot that -acts on the words uses it to echo them back, which is the only way a -mistranscription is visible without opening the vault. The audio itself -stays on the timeline as the reproducibility anchor (ADR-010) — we never -need to keep a second copy of the bytes. - -The module owns the pure half: recognising a voice event and rewriting its -raw source dict. `MicroBot._dispatch` owns the I/O half (download, -whisper, share). +"""Decoding voice messages into text before handlers see them. + +Speech and typing differ only in encoding, so only the transport needs to +know which one arrived. By the time a handler runs, an `m.audio` event has +become an ordinary text event carrying the same sender, event id and +thread relation, and every gate a typed message passes through applies to +it unchanged. + +This is deliberately not a capability that bots invoke. The previous +design made transcription a per-bot concern, which produced an audio +branch in each consumer and three code paths that could transcribe the +same recording. Decoding sits alongside decryption instead: below every +handler, performed once. + +The decoded event carries a `dev.famstack.transcript` block recording the +audio it came from. Routing ignores that block; the reply layer reads it +to quote the words back for checking. The audio itself stays on the +timeline and remains the reproducibility anchor (ADR-010). """ from __future__ import annotations @@ -39,28 +32,26 @@ from loguru import logger -# Provenance on a decoded message: which audio event these words came -# from. Its presence is the answer to "were these words guessed at by a -# machine?" — a question the reply layer asks and the routing layer must not. +# Marks a text event as decoded from audio, and names the recording. TRANSCRIPT_KEY = "dev.famstack.transcript" -# Fields of the audio event that describe the payload rather than the -# message. They move into the provenance block; leaving them on a text -# event would let a consumer treat it as an upload and re-file the bytes. +# Fields describing the audio payload rather than the message. They move +# into the provenance block; left in place, a consumer would read the +# decoded event as an upload and file the bytes a second time. _PAYLOAD_FIELDS = ("url", "file", "info", "filename") def is_voice(event) -> bool: - """Whether this timeline event is speech we can decode. + """Whether `event` is audio this module can decode. - Any `m.audio` with a plain mxc payload counts. We do not try to tell - a voice memo from a music file first: whisper answering "no words - here" is a better detector than a guess at the sender's intent, and - the clients families use mark both the same way. + Any `m.audio` carrying a plain mxc payload qualifies. No attempt is + made to distinguish a voice memo from a music file first, because + whisper returning no speech is a more reliable answer than inferring + the sender's intent from metadata. - Encrypted media (`file` rather than `url`) is deliberately not - claimed. The framework already tells encrypted rooms it cannot read - them, so saying yes here would only produce a download that fails. + Encrypted media, which carries `file` instead of `url`, is excluded. + The framework already declines to read encrypted rooms, so claiming + it here would only produce downloads that fail. """ content = (getattr(event, "source", None) or {}).get("content") or {} if content.get("msgtype") != "m.audio": @@ -71,22 +62,20 @@ def is_voice(event) -> bool: def was_transcribed(content: dict) -> bool: """Whether a message's words were transcribed rather than typed. - Reads the framework contract, so it works for any consumer without - knowing which component decoded the audio. Consumers ask this to - decide whether to show the words back for checking, never to decide - routing — routing is the whole thing that must not care. + Read by the reply layer to decide whether to quote the words back for + checking. Routing does not consult it: handling speech and typing + identically is the point of decoding in the transport. """ return isinstance(content.get(TRANSCRIPT_KEY), dict) def transcribed_source(source: dict, transcript: str) -> dict: - """Rewrite a voice event's raw source dict as the text event it is. + """Return `source` rewritten as the text event it decodes to. - Identity is preserved wholesale — event id, sender, timestamp, - thread relation — because this is the same message, read aloud - rather than typed. Bots reply to it, react on it and claim thread - ownership of it by that identity, so a synthesised id would detach - every one of those from the message the family can actually see. + Event id, sender, timestamp and thread relation are preserved + unchanged. Replies, reactions and thread ownership are all keyed on + that identity, so a synthesised id would detach each of them from the + message visible in the room. """ out = copy.deepcopy(source) content = out.setdefault("content", {}) @@ -114,28 +103,17 @@ def transcribed_source(source: dict, transcript: str) -> dict: class TranscriptStore: """Transcripts on disk, keyed by Matrix event id. - Transcription is the most expensive thing the stack does per message: - minutes of GPU for a long memo. Three callers want the same answer and - must not each pay for it. - - * Every bot in the room drains the same timeline in the same - process, so each reaches the same audio independently. - * The drain is at-least-once, so a handler that dies mid-flight - brings its event back around. - * A backfill walks a room's whole history in a *separate* process. - The memories room holds years of recordings; re-transcribing that - because the answer was only ever in RAM is not acceptable once, - let alone every time something wants to read it. - - So the store is durable and shared, and the in-flight map on top of it - collapses concurrent askers onto one run. One file per voice message, - written atomically, because the backfill and the bot runner are - different processes writing the same directory. - - Each record keeps the raw whisper output next to the polished text. - Polishing is cheap and improves with better models; whisper is not and - does not. Keeping both means years of recordings can be re-polished - without touching the audio again. + Transcription costs minutes of GPU for a long recording, and three + callers arrive at the same message independently: every bot in a room + drains the same timeline, the drain is at-least-once so a failed + handler brings its event back, and a backfill walks history in a + separate process. The store is therefore durable and shared, with one + file per message written atomically. + + Each record holds the raw whisper output beside the polished text. + Polishing is cheap and improves with better models; transcription is + neither, so keeping both allows a later re-polish without returning + to the audio. """ def __init__(self, path: str | Path | None = None): @@ -163,8 +141,12 @@ def read(self, event_id: str) -> dict | None: return None def write(self, event_id: str, record: dict) -> None: - """Persist a record. A store we cannot write is not fatal — the - transcript still reaches the handler, it just costs again later.""" + """Write a record, replacing any existing one atomically. + + A store that cannot be written is logged and ignored. The + transcript still reaches the handler; only the saving is lost, at + the cost of transcribing again later. + """ target = self._file(event_id) try: self.path.mkdir(parents=True, exist_ok=True) @@ -179,11 +161,12 @@ def write(self, event_id: str, record: dict) -> None: async def run( self, event_id: str, produce: Callable[[], Awaitable[dict]], ) -> dict: - """The record for `event_id`, produced at most once across callers. + """Return the record for `event_id`, producing it at most once. - A failure is never remembered. The drain is at-least-once and - whisper outages are transient, so caching an error would turn a - restartable service into a permanently silent message. + A stored record short-circuits; concurrent callers await the + first one's result. Failures are not retained: whisper outages + are transient and the drain retries, so caching an error would + make a message permanently undecodable. """ if (stored := self.read(event_id)) is not None: return stored @@ -203,8 +186,9 @@ async def run( future.exception() raise record.setdefault("at", time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())) - # Publish before releasing the slot, so a caller arriving in - # between finds the answer rather than starting a second run. + # Written and published before the in-flight slot is released, so + # a caller arriving in between finds the result rather than + # starting a second transcription. self.write(event_id, record) if not future.done(): future.set_result(record) diff --git a/stacklets/docs/stacklet.toml b/stacklets/docs/stacklet.toml index a88d5a82..11b90502 100644 --- a/stacklets/docs/stacklet.toml +++ b/stacklets/docs/stacklet.toml @@ -65,4 +65,3 @@ expect = "200" [[backup.archive]] name = "media" path = "{data_dir}/docs/paperless/media" -min_files = 10 diff --git a/stacklets/messages/bot/scribe.py b/stacklets/messages/bot/scribe.py index 3dc022be..00e02c72 100644 --- a/stacklets/messages/bot/scribe.py +++ b/stacklets/messages/bot/scribe.py @@ -1,21 +1,13 @@ -"""Scribe — retired, and saying so. +"""Scribe, retired. -Transcribing voice messages used to be this bot's whole job. It is the -transport's job now: a voice message is decoded before any handler sees -it, so every bot in every room gets the words without asking, and nothing -needs Scribe to be present. +Transcribing voice messages was this bot's job. The transport does it +now, so nothing requires Scribe to be present in a room. -This shell exists for one release only, because the framework has no way -to deprovision a bot that goes away. Deleting the declaration stops the -runner from launching it, but the Matrix account survives — still joined -to whatever rooms it was invited to, listed as a member, answering -nothing, forever. Somebody would eventually go looking for why. - -So instead of vanishing it explains itself once per room and leaves. The -people this reaches are the only ones it can have affected: Scribe -declared no room of its own, so it was never in a room unless a person -went and invited it by hand. Those are exactly the people who would -notice it go quiet. +It ships for one more release because the framework has no way to +deprovision a bot. Removing the declaration stops the runner launching +it, but the Matrix account remains, still joined to whatever rooms it was +invited to and answering nothing. Rather than going silent it explains +the change once per room and leaves. Delete this file and its bot.toml one release after it ships. """ @@ -29,8 +21,8 @@ from microbot import MicroBot -# Kept inline rather than in a message catalogue: this is two strings with -# a known expiry, and a catalogue would outlive the bot that uses it. +# Kept inline rather than in the message catalogue: two strings with a +# known expiry, which a catalogue entry would outlive. _GOODBYE = { "en": ( "**Voice messages are transcribed automatically now.**\n\n" @@ -57,22 +49,22 @@ class ScribeBot(MicroBot): name = "scribe-bot" def register_callbacks(self, client: AsyncClient) -> None: - """Register nothing, and start the retirement sweep. - - No message handlers at all: this bot answers nothing. The sweep - runs on every launch rather than once, so a room it could not - leave (homeserver hiccup, lost network) is retried next boot - instead of keeping a silent member forever. + """Register no handlers and start the retirement sweep. - Scheduled as a task because `register_callbacks` is sync and runs - inside `start()`'s event loop, after the initial sync has - populated `client.rooms`. + The sweep runs on every launch rather than once, so a room it + failed to leave is retried instead of keeping a silent member + indefinitely. It is scheduled as a task because this method is + synchronous and runs inside `start()`'s event loop, after the + initial sync has populated `client.rooms`. """ asyncio.create_task(self.retire_everywhere()) async def on_room_joined(self, room_id: str) -> None: - """Someone followed an older guide and invited it. Same answer, - so an invite never leaves a silent member behind.""" + """Handle an invite from someone following older documentation. + + Same response as the boot sweep, so an invite does not leave a + silent member behind either. + """ await self._retire_from(room_id) async def retire_everywhere(self) -> None: @@ -86,8 +78,11 @@ async def retire_everywhere(self) -> None: await self._retire_from(room_id) async def _retire_from(self, room_id: str) -> None: - """Explain, then leave. A failure to leave is logged and dropped: - the goodbye still landed, and the next launch tries again.""" + """Post the notice, then leave the room. + + A failure to leave is logged and otherwise ignored: the notice + has landed, and the next launch tries again. + """ lang = os.environ.get("LANGUAGE", "en") await self._send( room_id, _GOODBYE.get(lang, _GOODBYE["en"]), msgtype="m.notice", diff --git a/stacklets/messages/stacklet.toml b/stacklets/messages/stacklet.toml index eafc4c3e..4ecbafe4 100644 --- a/stacklets/messages/stacklet.toml +++ b/stacklets/messages/stacklet.toml @@ -58,3 +58,44 @@ DB_NAME = "synapse" [health] url = "http://localhost:42031/_matrix/client/versions" expect = "200" + +# Backup snapshot — the Matrix timeline, dumped consistently. +# +# The media store holds the family's actual voice recordings, but the +# database is what makes them messages: who spoke, in which room, when, +# and which media id belongs to which event. Without it the recordings +# are anonymous blobs. +# +# Postgres cannot be rsynced (its files change under you and a mid-write +# copy will not restore), so the backup stacklet dumps it instead and +# packs the dump with the files a homeserver cannot come back without: +# the signing key, which is its identity, and homeserver.yaml, whose +# macaroon secret keeps every existing login valid. +# +# pg_dump takes its own consistent view, so this runs against a live +# server with no downtime. A dump of this database compresses to a few +# hundred KB; the recordings themselves stay out of the tarball and ride +# the incremental archive path instead, so they are never re-copied and +# never subject to snapshot retention. +[[backup.snapshot]] +name = "synapse" +postgres = { container = "stack-messages-db", database = "synapse", user = "synapse" } +include = [ + "{data_dir}/messages/synapse/homeserver.yaml", + "{data_dir}/messages/synapse/*.signing.key", + "{data_dir}/messages/synapse/*.log.config", +] + +# The recordings themselves. Append-only by construction: Synapse writes a +# media file once and never edits it, and famstack ships with both message +# and media retention disabled, so nothing here is ever purged. +# +# Only local_content — the originals. local_thumbnails regenerates from +# them, and url_cache is a cache of other people's web pages. +# +# Synapse does not create this directory until somebody sends the first +# photo or voice message, so on a new install the backup reports it as +# "nothing to back up yet" and carries on with everything else. +[[backup.archive]] +name = "media" +path = "{data_dir}/messages/synapse/media_store/local_content" diff --git a/stacklets/photos/stacklet.toml b/stacklets/photos/stacklet.toml index 5f62772d..899bfd95 100644 --- a/stacklets/photos/stacklet.toml +++ b/stacklets/photos/stacklet.toml @@ -71,4 +71,3 @@ expect = "pong" [[backup.archive]] name = "library" path = "{data_dir}/photos/library/library" -min_files = 10 diff --git a/tests/framework/test_postgres_helper.py b/tests/framework/test_postgres_helper.py new file mode 100644 index 00000000..f561692e --- /dev/null +++ b/tests/framework/test_postgres_helper.py @@ -0,0 +1,64 @@ +"""Talking to a Postgres running in a container. + +Three stacklets keep their state in Postgres and all three need the same +two operations: get a consistent dump out, put one back. This is that, +in one place, so the knowledge of how to drive `pg_dump` does not end up +copied into a backup coordinator, a restore hook, and whatever comes +after. + +The command construction is pure and tested directly; the subprocess call +around it is a thin wrapper with nothing to assert. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent / "lib")) + +from stack.postgres import dump_command, restore_command, version_command # noqa: E402 + + +class TestDumpCommand: + + def test_it_dumps_through_the_container(self): + assert dump_command("stack-messages-db", "synapse", "synapse") == [ + "docker", "exec", "stack-messages-db", + "pg_dump", "-U", "synapse", "-d", "synapse", + ] + + def test_the_role_and_database_are_separate(self): + """Paperless runs its database under a differently named role, so + these cannot be collapsed into one argument.""" + cmd = dump_command("stack-docs-db", "paperless", "paperless_user") + assert "-U" in cmd and cmd[cmd.index("-U") + 1] == "paperless_user" + assert "-d" in cmd and cmd[cmd.index("-d") + 1] == "paperless" + + +class TestRestoreCommand: + """Nothing calls this yet. It lives here so that whatever eventually + restores a snapshot does not have to reinvent the invocation, and so + the two halves stay next to each other where they can be kept in + step.""" + + def test_it_loads_a_dump_into_a_named_database(self): + assert restore_command("stack-messages-db", "synapse", "synapse") == [ + "docker", "exec", "-i", "stack-messages-db", + "psql", "-U", "synapse", "-d", "synapse", + ] + + def test_it_reads_the_dump_from_stdin(self): + """`-i` is the difference between a restore and a hang: without it + docker gives psql no stdin and it waits forever.""" + assert "-i" in restore_command("c", "d", "u") + + +class TestVersionCommand: + + def test_it_asks_the_server_what_it_is(self): + cmd = version_command("stack-messages-db", "synapse", "synapse") + assert cmd[:3] == ["docker", "exec", "stack-messages-db"] + assert "show server_version;" in cmd + # -tA: no header, no alignment. The caller wants a bare value. + assert "-tAc" in cmd diff --git a/tests/stacklets/test_backup_e2e.py b/tests/stacklets/test_backup_e2e.py index 43474b8e..fb3b84b1 100644 --- a/tests/stacklets/test_backup_e2e.py +++ b/tests/stacklets/test_backup_e2e.py @@ -132,14 +132,39 @@ def fake_sources(tmp_path): # ── Helpers ──────────────────────────────────────────────────────────────── -def _sources_env(fake_sources: dict, *, photos_min: int = 10, docs_min: int = 5) -> str: - """Build the ``$SOURCES`` env string the engine expects.""" +def _sources_env(fake_sources: dict, *, rolling: bool = False) -> str: + """Build the ``$SOURCES`` env string the engine expects. + + Final field is the rolling flag: 1 for a source pruned on purpose + (a snapshot staging area), 0 for an ordinary append-only archive. + """ + flag = 1 if rolling else 0 return "\n".join([ - f"photos/library|Photos|{fake_sources['photos']}|data/photos-library|{photos_min}", - f"docs/media|Documents|{fake_sources['docs']}|data/docs-media|{docs_min}", + f"photos/library|Photos|{fake_sources['photos']}|data/photos-library|{flag}", + f"docs/media|Documents|{fake_sources['docs']}|data/docs-media|{flag}", ]) +def _seed_history(backup_data_dir: Path, counts: dict) -> None: + """Record a previous run, so the next one has a baseline to be judged + against. This is what replaced a hand-written `min_files`.""" + history = backup_data_dir / "logs" / "history.jsonl" + history.parent.mkdir(parents=True, exist_ok=True) + history.write_text(json.dumps({ + "engine": "external-disk", + "success": True, + "sources": [ + {"id": sid, "display": sid, "status": "ok", + "total_files": n, "new_files": 0, "source_files": n} + for sid, n in counts.items() + ], + }) + "\n") + + +def _count_files(path: Path) -> int: + return sum(1 for p in Path(path).rglob("*") if p.is_file()) + + def _run_engine(backup_data_dir: Path, vault_name: str, sources_env: str, *, args=None): env = os.environ.copy() @@ -314,15 +339,16 @@ def test_canary_tamper_aborts_sync( assert data["success"] is False assert "canary" in (data["failure_reason"] or "").lower() - def test_refuses_when_source_under_minimum( + def test_refuses_when_a_source_lost_files_since_last_run( self, vault_image, backup_data_dir, fake_sources ): - """Preflight is the coarse ransomware guard — refuses to sync - a source that's been wiped to fewer files than the declared - minimum. Critically, no vault writes happen.""" + """Preflight judges a source against its own previous count, so + the guard means the same thing at any scale and nothing has to be + configured. Critically, no vault writes happen.""" name, mount = vault_image - # photos has 15 files; bump min to 100 so preflight fails - sources = _sources_env(fake_sources, photos_min=100) + # photos has 15 files on disk; last run it held 5,000. + _seed_history(backup_data_dir, {"photos/library": 5000}) + sources = _sources_env(fake_sources) result = _run_engine(backup_data_dir, name, sources, args=["--no-eject"]) assert result.returncode != 0 @@ -495,3 +521,106 @@ def test_stack_backup_sync_runs_engine_end_to_end( # Spot-check the uchg flag on a few files assert _has_uchg(photos[0]) assert _has_uchg(docs[0]) + + +class TestUpgradingAnExistingVault: + """An instance that has been backing up photos and documents for + months, upgraded to a release that adds snapshots and a new archive. + + Two things could go wrong. The shrink check reads each source's count + from the previous run, and every run recorded before the upgrade + predates that field, so an absent baseline must not read as loss. And + the new sources must be added to the vault without disturbing what is + already on it, which is immutable and cannot be rewritten. + """ + + def _existing_vault(self, mount: Path) -> dict: + """Photos and documents already synced and locked, as a vault in + use would be.""" + before = {} + for subdir, count in (("photos-library", 6), ("docs-media", 4)): + d = mount / "data" / subdir + d.mkdir(parents=True) + for i in range(count): + (d / f"old-{i}.txt").write_text("x") + # Files only, as the engine locks them. Locking the directory + # would stop rsync writing into it, which it never does. + subprocess.run( + ["find", str(d), "-type", "f", "-exec", "chflags", "uchg", + "{}", "+"], check=False, + ) + before[subdir] = sorted(p.name for p in d.iterdir()) + return before + + def _old_format_history(self, backup_data_dir: Path) -> None: + """A run record as written before `source_files` existed.""" + history = backup_data_dir / "logs" / "history.jsonl" + history.parent.mkdir(parents=True, exist_ok=True) + history.write_text(json.dumps({ + "engine": "external-disk", + "success": True, + "sources": [ + {"id": "photos/library", "display": "Photos", "status": "ok", + "total_files": 6, "new_files": 0}, + {"id": "docs/media", "display": "Documents", "status": "ok", + "total_files": 4, "new_files": 0}, + ], + }) + "\n") + + def test_the_first_upgraded_sync_adds_without_disturbing( + self, vault_image, backup_data_dir, fake_sources, tmp_path, + ): + name, mount = vault_image + before = self._existing_vault(mount) + self._old_format_history(backup_data_dir) + + # The new sources: Matrix media, and a snapshot staging directory. + media = tmp_path / "data" / "messages" / "media_store" / "local_content" + media.mkdir(parents=True) + for i in range(3): + (media / f"voice-{i}.ogg").write_text("audio") + snaps = tmp_path / "data" / "snapshots" / "messages-synapse" + snaps.mkdir(parents=True) + (snaps / "synapse-20260911T000000Z.tar.gz").write_text("dump") + + sources = "\n".join([ + _sources_env(fake_sources), + f"messages/media|Messages|{media}|data/messages-media|0", + f"messages/synapse|Messages|{snaps}|data/messages-synapse|1", + ]) + result = _run_engine(backup_data_dir, name, sources, args=["--no-eject"]) + assert result.returncode == 0 + + data = _read_result(backup_data_dir) + assert data["success"] is True + + # Everything already on the vault is still there. The sync adds + # the source's files alongside; it never removes or rewrites. + for subdir, names in before.items(): + now = {p.name for p in (mount / "data" / subdir).iterdir()} + assert set(names) <= now + + # And the new sources arrived. + assert len(list((mount / "data" / "messages-media").iterdir())) == 3 + assert len(list((mount / "data" / "messages-synapse").iterdir())) == 1 + + def test_it_records_a_baseline_the_next_run_can_use( + self, vault_image, backup_data_dir, fake_sources, + ): + """The upgraded run has no baseline to check against, so it also + has to leave one behind or the guard never arms.""" + name, mount = vault_image + self._existing_vault(mount) + self._old_format_history(backup_data_dir) + + _run_engine(backup_data_dir, name, _sources_env(fake_sources), + args=["--no-eject"]) + + counts = { + s["id"]: s.get("source_files") + for s in _read_result(backup_data_dir)["sources"] + } + # Counted from the sources themselves, so the assertion states the + # rule rather than a number that drifts with the fixture. + assert counts["photos/library"] == _count_files(fake_sources["photos"]) + assert counts["docs/media"] == _count_files(fake_sources["docs"]) diff --git a/tests/stacklets/test_backup_engine.py b/tests/stacklets/test_backup_engine.py index 7bbc2616..3bb553c6 100644 --- a/tests/stacklets/test_backup_engine.py +++ b/tests/stacklets/test_backup_engine.py @@ -30,6 +30,7 @@ format_number, parse_sources, preflight_check_sources, + previous_source_counts, probe_filesystem, read_latest_run, verify_canary, @@ -41,7 +42,7 @@ class TestParseSources: def test_single_record(self): sources = parse_sources( - "photos/library|Photos|/data/photos/library|data/photos-library|10" + "photos/library|Photos|/data/photos/library|data/photos-library|0" ) assert len(sources) == 1 s = sources[0] @@ -49,22 +50,24 @@ def test_single_record(self): assert s.display == "Photos" assert s.src_path == Path("/data/photos/library") assert s.vault_subdir == "data/photos-library" - assert s.min_files == 10 + assert s.rolling is False def test_multiple_records_separated_by_newlines(self): sources = parse_sources( - "photos/library|Photos|/a|data/p|10\n" - "docs/media|Documents|/b|data/d|5" + "photos/library|Photos|/a|data/p|0\n" + "messages/synapse|Messages|/b|data/d|1" ) - assert [s.id for s in sources] == ["photos/library", "docs/media"] - assert sources[1].min_files == 5 + assert [s.id for s in sources] == ["photos/library", "messages/synapse"] + # A snapshot staging area: pruned on purpose, so shrinking is not + # a loss and the engine must be told so. + assert sources[1].rolling is True def test_blank_lines_ignored(self): sources = parse_sources( "\n" - "photos/library|Photos|/a|data/p|10\n" + "photos/library|Photos|/a|data/p|0\n" " \n" - "docs/media|Documents|/b|data/d|5\n" + "docs/media|Documents|/b|data/d|0\n" ) assert len(sources) == 2 @@ -84,11 +87,7 @@ def test_too_few_fields_aborts(self): def test_too_many_fields_aborts(self): with pytest.raises(SyncAborted, match="Malformed source record"): - parse_sources("a|b|c|d|10|extra") - - def test_non_integer_min_files_aborts(self): - with pytest.raises(SyncAborted, match="min_files must be an integer"): - parse_sources("a|b|c|d|many") + parse_sources("a|b|c|d|0|extra") # ── Canary ───────────────────────────────────────────────────────────────── @@ -122,7 +121,7 @@ def test_trailing_whitespace_in_canary_is_tolerated(self, tmp_path, capsys): # ── Preflight ────────────────────────────────────────────────────────────── class TestPreflightCheckSources: - def _make_source(self, tmp_path: Path, name: str, file_count: int, min_files: int) -> Source: + def _make_source(self, tmp_path: Path, name: str, file_count: int) -> Source: src_dir = tmp_path / name src_dir.mkdir() for i in range(file_count): @@ -132,39 +131,107 @@ def _make_source(self, tmp_path: Path, name: str, file_count: int, min_files: in display=name.title(), src_path=src_dir, vault_subdir=f"data/test-{name}", - min_files=min_files, ) - def test_passes_when_each_source_meets_min(self, tmp_path, capsys): - sources = [ - self._make_source(tmp_path, "a", file_count=20, min_files=10), - self._make_source(tmp_path, "b", file_count=15, min_files=10), - ] - preflight_check_sources(sources) # should not raise - - def test_aborts_when_any_source_under_min(self, tmp_path, capsys): - sources = [ - self._make_source(tmp_path, "ok", file_count=20, min_files=10), - self._make_source(tmp_path, "low", file_count=2, min_files=10), - ] - with pytest.raises(SyncAborted, match="Preflight failed"): - preflight_check_sources(sources) - - def test_aborts_when_source_dir_missing(self, tmp_path, capsys): - source = Source( - id="test/missing", - display="Missing", + def test_a_first_ever_run_syncs_whatever_is_there(self, tmp_path, capsys): + """No history means no baseline, so nothing can be judged a loss.""" + sources = [self._make_source(tmp_path, "a", file_count=20)] + assert len(preflight_check_sources(sources, {})) == 1 + + def test_growth_is_normal(self, tmp_path, capsys): + sources = [self._make_source(tmp_path, "a", file_count=20)] + assert len(preflight_check_sources(sources, {"test/a": 15})) == 1 + + def test_a_source_with_no_data_yet_is_skipped_not_fatal(self, tmp_path, capsys): + """A stacklet that has never had data must not cost the household + every other backup. + + Matrix media forced this: Synapse does not create the media store + until somebody sends the first photo, so a fresh install would + have failed every backup until then, photos included. Nothing is + at risk — the engine syncs with `--ignore-existing` and never + `--delete`, so an empty source copies nothing and the vault keeps + what it had. + """ + missing = Source( + id="test/missing", display="Missing", src_path=tmp_path / "does-not-exist", vault_subdir="data/test-missing", - min_files=1, ) + present = self._make_source(tmp_path, "ok", file_count=20) + + syncable = preflight_check_sources([missing, present], {}) + + assert [s.id for s in syncable] == ["test/ok"] + + def test_a_source_that_vanished_aborts(self, tmp_path, capsys): + """Empty now, but it had files last run. That is the disaster.""" + gone = self._make_source(tmp_path, "gone", file_count=0) with pytest.raises(SyncAborted, match="Preflight failed"): - preflight_check_sources([source]) + preflight_check_sources([gone], {"test/gone": 4000}) - def test_exact_min_count_passes(self, tmp_path, capsys): - # Edge: file_count == min_files should pass (not "strictly greater than"). - sources = [self._make_source(tmp_path, "exact", file_count=10, min_files=10)] - preflight_check_sources(sources) # should not raise + def test_a_source_that_lost_most_of_its_files_aborts(self, tmp_path, capsys): + """The case a hand-written `min_files` could never catch. A library + of 50,000 photos reduced to 11 sails past `min_files = 10`; against + its own previous count it is unmissable.""" + raided = self._make_source(tmp_path, "photos", file_count=11) + with pytest.raises(SyncAborted, match="Preflight failed"): + preflight_check_sources([raided], {"test/photos": 50_000}) + + def test_a_modest_loss_is_allowed(self, tmp_path, capsys): + """People do delete things. The guard is for catastrophe, not for + policing a household's own housekeeping.""" + tidied = self._make_source(tmp_path, "photos", file_count=18) + assert len(preflight_check_sources([tidied], {"test/photos": 20})) == 1 + + def test_a_rolling_source_may_shrink_to_its_window(self, tmp_path, capsys): + """Snapshot staging directories are pruned on purpose. Shrinking is + their normal operation, not a loss.""" + rolling = self._make_source(tmp_path, "snaps", file_count=7) + rolling.rolling = True + assert len(preflight_check_sources([rolling], {"test/snaps": 40})) == 1 + + +class TestSourceCountsFromHistory: + """The baseline each run is judged against: what the source itself + held last time. Self-calibrating, so no manifest has to guess.""" + + def test_it_reads_the_previous_runs_counts(self): + run = {"sources": [ + {"id": "photos/library", "status": "ok", "source_files": 4021}, + {"id": "docs/media", "status": "ok", "source_files": 57}, + ]} + assert previous_source_counts(run) == { + "photos/library": 4021, "docs/media": 57, + } + + def test_a_run_recorded_before_this_field_existed_yields_nothing(self): + """Upgrade path. Runs written by the previous engine carry + `total_files` but no `source_files`, so the first run after an + upgrade has no baseline and must not read that as loss.""" + run = {"sources": [ + {"id": "photos/library", "status": "ok", + "total_files": 4021, "new_files": 3}, + ]} + assert previous_source_counts(run) == {} + + def test_a_failed_source_contributes_no_baseline(self): + """It did not finish, so its count does not describe what the + source held.""" + run = {"sources": [ + {"id": "photos/library", "status": "FAILED", "source_files": 0}, + ]} + assert previous_source_counts(run) == {} + + def test_no_previous_run_means_no_baseline(self): + assert previous_source_counts(None) == {} + + def test_a_skipped_source_contributes_no_baseline(self): + """A source skipped for having nothing must not become a baseline + of zero that makes the next run look like growth from nothing.""" + run = {"sources": [{"id": "messages/media", "status": "skipped", + "source_files": 0}]} + assert previous_source_counts(run) == {} # ── _stat_fs_type (mocked mount output) ──────────────────────────────────── @@ -460,8 +527,7 @@ def _source(self, tmp_path: Path) -> Source: src.mkdir() (src / "a.jpg").write_text("x") return Source(id="photos/library", display="Photos", - src_path=src, vault_subdir="data/photos-library", - min_files=1) + src_path=src, vault_subdir="data/photos-library") def _fake_run(self, chflags_returncode: int): from types import SimpleNamespace diff --git a/tests/stacklets/test_backup_orchestrator.py b/tests/stacklets/test_backup_orchestrator.py index ab3ae296..59a91834 100644 --- a/tests/stacklets/test_backup_orchestrator.py +++ b/tests/stacklets/test_backup_orchestrator.py @@ -54,8 +54,6 @@ def _make_fake_stacklet( lines.append("[[backup.archive]]") lines.append(f'name = "{archive["name"]}"') lines.append(f'path = "{archive["path"]}"') - if "min_files" in archive: - lines.append(f'min_files = {archive["min_files"]}') (stacklets_dir / "stacklet.toml").write_text("\n".join(lines) + "\n") @@ -71,7 +69,7 @@ class TestDiscoverArchiveSources: def test_finds_archive_entries_from_enabled_stacklets(self, tmp_path): _make_fake_stacklet( tmp_path, "photos", - [{"name": "library", "path": "{data_dir}/photos/library/library", "min_files": 10}], + [{"name": "library", "path": "{data_dir}/photos/library/library"}], name="Photos", ) sources = discover_archive_sources( @@ -83,18 +81,19 @@ def test_finds_archive_entries_from_enabled_stacklets(self, tmp_path): assert s.display == "Photos" assert s.src_path == Path("/var/famstack-data/photos/library/library") assert s.vault_subdir == "data/photos-library" - assert s.min_files == 10 + # Replaced by a baseline the engine derives itself. + assert s.rolling is False def test_skips_unenabled_stacklets(self, tmp_path): # Enabled photos contributes; disabled docs does not. _make_fake_stacklet( tmp_path, "photos", - [{"name": "library", "path": "{data_dir}/photos", "min_files": 1}], + [{"name": "library", "path": "{data_dir}/photos"}], enabled=True, ) _make_fake_stacklet( tmp_path, "docs", - [{"name": "media", "path": "{data_dir}/docs", "min_files": 1}], + [{"name": "media", "path": "{data_dir}/docs"}], enabled=False, ) sources = discover_archive_sources(tmp_path, tmp_path, Path("/d")) @@ -116,8 +115,8 @@ def test_multiple_archives_per_stacklet(self, tmp_path): _make_fake_stacklet( tmp_path, "photos", [ - {"name": "library", "path": "{data_dir}/a", "min_files": 1}, - {"name": "shared", "path": "{data_dir}/b", "min_files": 1}, + {"name": "library", "path": "{data_dir}/a"}, + {"name": "shared", "path": "{data_dir}/b"}, ], ) sources = discover_archive_sources(tmp_path, tmp_path, Path("/d")) @@ -128,7 +127,7 @@ def test_template_variable_renders(self, tmp_path): # {data_dir} must expand to whatever the orchestrator was given. _make_fake_stacklet( tmp_path, "photos", - [{"name": "library", "path": "{data_dir}/photos/library", "min_files": 1}], + [{"name": "library", "path": "{data_dir}/photos/library"}], ) sources = discover_archive_sources( tmp_path, tmp_path, Path("/totally/custom/data") @@ -141,7 +140,7 @@ def test_unknown_template_variable_kept_literal(self, tmp_path): # error pointing at the broken path. _make_fake_stacklet( tmp_path, "photos", - [{"name": "library", "path": "{nonexistent_var}/photos", "min_files": 1}], + [{"name": "library", "path": "{nonexistent_var}/photos"}], ) sources = discover_archive_sources(tmp_path, tmp_path, Path("/d")) # The format() call raises KeyError, we fall back to the raw string. @@ -156,7 +155,7 @@ def test_malformed_manifest_skipped_not_fatal(self, tmp_path): # of all the others. _make_fake_stacklet( tmp_path, "photos", - [{"name": "library", "path": "{data_dir}/p", "min_files": 1}], + [{"name": "library", "path": "{data_dir}/p"}], ) broken = tmp_path / "stacklets" / "broken" broken.mkdir(parents=True) @@ -230,18 +229,27 @@ def test_single_record_formatted(self): sources = [SourceRecord( id="photos/library", display="Photos", src_path=Path("/var/famstack-data/photos/library/library"), - vault_subdir="data/photos-library", min_files=10, + vault_subdir="data/photos-library", )] env = serialize_sources_env(sources) assert env == ( "photos/library|Photos|/var/famstack-data/photos/library/library|" - "data/photos-library|10" + "data/photos-library|0" ) + def test_a_rolling_source_is_flagged_for_the_engine(self): + """Snapshot staging areas are pruned on purpose, and the engine + has to know so it does not read that as data loss.""" + sources = [SourceRecord( + id="messages/synapse", display="Messages", src_path=Path("/a"), + vault_subdir="data/messages-synapse", rolling=True, + )] + assert serialize_sources_env(sources).endswith("|1") + def test_multiple_records_newline_joined(self): sources = [ - SourceRecord("photos/library", "Photos", Path("/a"), "data/p", 10), - SourceRecord("docs/media", "Documents", Path("/b"), "data/d", 5), + SourceRecord("photos/library", "Photos", Path("/a"), "data/p"), + SourceRecord("docs/media", "Documents", Path("/b"), "data/d"), ] env = serialize_sources_env(sources) lines = env.split("\n") diff --git a/tests/stacklets/test_backup_snapshot.py b/tests/stacklets/test_backup_snapshot.py new file mode 100644 index 00000000..b40056c2 --- /dev/null +++ b/tests/stacklets/test_backup_snapshot.py @@ -0,0 +1,358 @@ +"""Snapshots: capturing state that rsync cannot copy. + +An archive source is a directory whose files are written once and never +changed. A database is not, so it is dumped instead: one consistent dump +per run, packed with the files that must accompany it, into a dated +tarball that later runs add to but never modify. + +Synapse is the first stacklet wired up. Its media store holds the +recordings themselves, while the database holds the sender, room and +date that make each one a message. +""" + +from __future__ import annotations + +import json +import sys +import tarfile +from pathlib import Path + +import pytest + +_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(_ROOT / "stacklets" / "backup" / "cli")) + +from _snapshot import ( # noqa: E402 + SnapshotSpec, + discover_snapshots, + prune_snapshots, + snapshot_source, + take_snapshot, +) + + +def _spec(tmp_path, **kw) -> SnapshotSpec: + defaults = dict( + id="messages/synapse", + display="Messages", + name="synapse", + postgres={"container": "stack-messages-db", + "database": "synapse", "user": "synapse"}, + include=[], + ) + defaults.update(kw) + return SnapshotSpec(**defaults) + + +def _fake_dump(text: str = "-- pg_dump output\nCREATE TABLE events();\n"): + """Stand-in for `docker exec ... pg_dump`, so tests need no Postgres.""" + calls: list[dict] = [] + + def run(spec: SnapshotSpec) -> bytes: + calls.append({"container": spec.container, "database": spec.database, + "user": spec.user}) + return text.encode() + + run.calls = calls # type: ignore[attr-defined] + return run + + +# ── Taking one ─────────────────────────────────────────────────────────── + +class TestTakeSnapshot: + + def test_it_writes_one_dated_tarball(self, tmp_path): + out = tmp_path / "snapshots" + path = take_snapshot(_spec(tmp_path), out, dump=_fake_dump()) + + # Namespaced per stacklet+name so two stacklets cannot collide. + assert path.parent == out / "messages-synapse" + assert path.suffixes[-2:] == [".tar", ".gz"] + assert path.name.startswith("synapse-") + assert path.exists() + + def test_the_dump_is_inside(self, tmp_path): + path = take_snapshot( + _spec(tmp_path), tmp_path / "s", + dump=_fake_dump("-- the whole database\n"), + ) + with tarfile.open(path) as tar: + names = tar.getnames() + assert "synapse.sql" in names + body = tar.extractfile("synapse.sql").read().decode() + assert "the whole database" in body + + def test_named_files_travel_with_the_dump(self, tmp_path): + """A dump alone does not restore a homeserver. The signing key is + the server's identity, and homeserver.yaml holds the macaroon + secret that keeps existing device logins valid.""" + cfg = tmp_path / "synapse" + cfg.mkdir() + (cfg / "homeserver.yaml").write_text("{}") + (cfg / "simpson.signing.key").write_text("key") + + path = take_snapshot( + _spec(tmp_path, include=[str(cfg / "homeserver.yaml"), + str(cfg / "*.signing.key")]), + tmp_path / "s", dump=_fake_dump(), + ) + with tarfile.open(path) as tar: + names = tar.getnames() + assert "homeserver.yaml" in names + assert "simpson.signing.key" in names + + def test_a_missing_include_is_skipped_not_fatal(self, tmp_path): + """Installs differ in which optional config files exist. A + pattern matching nothing is skipped, because the dump is the part + that cannot be reproduced from elsewhere.""" + path = take_snapshot( + _spec(tmp_path, include=[str(tmp_path / "nope.yaml")]), + tmp_path / "s", dump=_fake_dump(), + ) + with tarfile.open(path) as tar: + assert "synapse.sql" in tar.getnames() + + def test_it_carries_a_manifest_describing_itself(self, tmp_path): + """The tarball is self-describing, so it can be interpreted + without this code available.""" + path = take_snapshot(_spec(tmp_path), tmp_path / "s", + dump=_fake_dump()) + with tarfile.open(path) as tar: + manifest = json.loads(tar.extractfile("MANIFEST.json").read()) + + assert manifest["stacklet"] == "messages" + assert manifest["database"] == "synapse" + assert manifest["dump_file"] == "synapse.sql" + assert manifest["taken_at"].endswith("Z") + assert "psql" in manifest["restore"] + + def test_each_run_adds_a_tarball_and_keeps_the_old_one(self, tmp_path): + """Runs add tarballs and never modify an existing one, so a + snapshot verified earlier stays as it was verified.""" + out = tmp_path / "s" + first = take_snapshot(_spec(tmp_path), out, dump=_fake_dump("one")) + second = take_snapshot(_spec(tmp_path), out, dump=_fake_dump("two")) + + assert first != second + assert first.exists() and second.exists() + + def test_a_failed_dump_leaves_no_tarball(self, tmp_path): + """A partial tarball would sync to the vault and be locked + immutable there, so nothing is written unless the dump + succeeds.""" + def boom(spec): + raise RuntimeError("postgres is down") + + out = tmp_path / "s" + with pytest.raises(RuntimeError): + take_snapshot(_spec(tmp_path), out, dump=boom) + + assert list(out.glob("*.tar.gz")) == [] + + +# ── Feeding the vault ──────────────────────────────────────────────────── + +class TestSnapshotSource: + """Snapshots reach the vault through the ordinary engine. The + output directory is registered as another append-only source.""" + + def test_the_output_directory_becomes_a_source(self, tmp_path): + src = snapshot_source(_spec(tmp_path), tmp_path / "snapshots") + assert src.id == "messages/synapse" + assert src.src_path == tmp_path / "snapshots" / "messages-synapse" + assert src.vault_subdir == "data/messages-synapse" + + def test_it_is_marked_rolling(self, tmp_path): + """The directory is pruned to a fixed size, so the engine's + shrink check must not read that as data loss.""" + assert snapshot_source(_spec(tmp_path), tmp_path / "s").rolling is True + + +# ── Keeping the internal disk honest ───────────────────────────────────── + +class TestPrune: + """Local tarballs are a staging area; the vault copy is the backup. + The engine syncs with `--ignore-existing` and never `--delete`, so + pruning here does not affect the vault.""" + + def _fill(self, d: Path, n: int) -> list[Path]: + d.mkdir(parents=True, exist_ok=True) + made = [] + for i in range(n): + p = d / f"synapse-2026091{i}T000000Z.tar.gz" + p.write_bytes(b"x") + made.append(p) + return made + + def test_it_keeps_the_newest_and_drops_the_rest(self, tmp_path): + made = self._fill(tmp_path / "s", 5) + prune_snapshots(tmp_path / "s", keep=2) + left = sorted(p.name for p in (tmp_path / "s").glob("*.tar.gz")) + assert left == sorted(p.name for p in made[-2:]) + + def test_it_does_nothing_when_under_the_limit(self, tmp_path): + self._fill(tmp_path / "s", 2) + prune_snapshots(tmp_path / "s", keep=7) + assert len(list((tmp_path / "s").glob("*.tar.gz"))) == 2 + + def test_it_ignores_anything_that_is_not_a_snapshot(self, tmp_path): + d = tmp_path / "s" + self._fill(d, 3) + (d / "README.txt").write_text("do not delete me") + prune_snapshots(d, keep=1) + assert (d / "README.txt").exists() + + +# ── Discovery ──────────────────────────────────────────────────────────── + +class TestDiscovery: + """Same shape as archive discovery: walk enabled stacklets, read their + manifests, render `{data_dir}`.""" + + def _stacklet(self, root: Path, sid: str, body: str) -> None: + d = root / "stacklets" / sid + d.mkdir(parents=True, exist_ok=True) + (d / "stacklet.toml").write_text(body) + + def _enable(self, instance: Path, sid: str) -> None: + (instance / ".stack").mkdir(parents=True, exist_ok=True) + (instance / ".stack" / f"{sid}.setup-done").touch() + + MANIFEST = ''' +id = "messages" +name = "Messages" + +[[backup.snapshot]] +name = "synapse" +postgres = { container = "stack-messages-db", database = "synapse", user = "synapse" } +include = ["{data_dir}/messages/synapse/homeserver.yaml"] +''' + + def test_it_finds_an_enabled_stacklets_snapshot(self, tmp_path): + self._stacklet(tmp_path, "messages", self.MANIFEST) + self._enable(tmp_path, "messages") + + found = discover_snapshots(tmp_path, tmp_path, Path("/data")) + + assert len(found) == 1 + spec = found[0] + assert spec.id == "messages/synapse" + # Namespaced in the manifest, read back through the accessor so + # callers never touch the raw dict. + assert spec.container == "stack-messages-db" + assert spec.database == "synapse" + assert spec.include == ["/data/messages/synapse/homeserver.yaml"] + + def test_a_disabled_stacklet_contributes_nothing(self, tmp_path): + self._stacklet(tmp_path, "messages", self.MANIFEST) + # no setup-done marker + assert discover_snapshots(tmp_path, tmp_path, Path("/data")) == [] + + def test_a_stacklet_without_snapshots_is_skipped(self, tmp_path): + self._stacklet(tmp_path, "photos", 'id = "photos"\nname = "Photos"\n') + self._enable(tmp_path, "photos") + assert discover_snapshots(tmp_path, tmp_path, Path("/data")) == [] + + +class TestRecordedVersions: + """A dump loads only into a compatible version of the application + that wrote it. Paperless is the concrete case: a 3.x database will not + boot under 2.x, and there is no downgrade path. + + The snapshot therefore records what produced it. This describes a + moment that has passed by the time a restore wants it, so it cannot + be reconstructed later. + """ + + def _versions(self, payload=None, error=None): + def probe(spec): + if error is not None: + raise error + return payload if payload is not None else { + "containers": { + "stack-messages-synapse": { + "image": "matrixdotorg/synapse:latest", + "version": "1.160.0", + "digest": "sha256:1231c84d", + }, + }, + "postgres": "16.15", + } + return probe + + def test_the_manifest_records_what_produced_it(self, tmp_path): + path = take_snapshot(_spec(tmp_path), tmp_path / "s", + dump=_fake_dump(), versions=self._versions()) + with tarfile.open(path) as tar: + manifest = json.loads(tar.extractfile("MANIFEST.json").read()) + + synapse = manifest["versions"]["containers"]["stack-messages-synapse"] + assert synapse["version"] == "1.160.0" + assert manifest["versions"]["postgres"] == "16.15" + + def test_the_digest_is_kept_because_a_tag_is_not_a_version(self, tmp_path): + """Tags are mutable, so a reference like `synapse:latest` + resolves to different images over time. The digest identifies the + exact one.""" + path = take_snapshot(_spec(tmp_path), tmp_path / "s", + dump=_fake_dump(), versions=self._versions()) + with tarfile.open(path) as tar: + manifest = json.loads(tar.extractfile("MANIFEST.json").read()) + + synapse = manifest["versions"]["containers"]["stack-messages-synapse"] + assert synapse["digest"].startswith("sha256:") + + def test_unavailable_versions_do_not_cost_the_dump(self, tmp_path): + """Version lookup can fail for reasons unrelated to the data: + docker unreachable, a container stopped, an image pruned. The dump + still proceeds.""" + path = take_snapshot( + _spec(tmp_path), tmp_path / "s", dump=_fake_dump(), + versions=self._versions(error=RuntimeError("docker is not running")), + ) + with tarfile.open(path) as tar: + manifest = json.loads(tar.extractfile("MANIFEST.json").read()) + assert "synapse.sql" in tar.getnames() + + # Recorded as unknown rather than omitted, so a reader can tell + # "we could not look" from "this predates version recording". + assert manifest["versions"] == {} + + +class TestTheShippedManifest: + """Parses the real `stacklets/messages/stacklet.toml` rather than a + fixture. + + The other discovery tests write their own manifest, so they verify the + parser against an example the test also wrote. They stay green if the + declaration that actually ships is malformed or renames a key, which + is a failure only a live snapshot would otherwise reveal. + """ + + def _enabled(self, tmp_path: Path, stacklet_id: str) -> Path: + (tmp_path / ".stack").mkdir(parents=True, exist_ok=True) + (tmp_path / ".stack" / f"{stacklet_id}.setup-done").touch() + return tmp_path + + def test_the_messages_snapshot_declaration_parses(self, tmp_path): + specs = discover_snapshots( + _ROOT, self._enabled(tmp_path, "messages"), Path("/data"), + ) + + assert [s.id for s in specs] == ["messages/synapse"] + spec = specs[0] + assert spec.container == "stack-messages-db" + assert spec.database == "synapse" + assert spec.user == "synapse" + + def test_it_carries_the_files_a_homeserver_cannot_restore_without( + self, tmp_path, + ): + spec = discover_snapshots( + _ROOT, self._enabled(tmp_path, "messages"), Path("/data"), + )[0] + + assert any(p.endswith("homeserver.yaml") for p in spec.include) + assert any(p.endswith("*.signing.key") for p in spec.include) + # `{data_dir}` is rendered, not passed through to the glob. + assert all(p.startswith("/data/") for p in spec.include) diff --git a/tests/stacklets/test_microbot_voice.py b/tests/stacklets/test_microbot_voice.py index 99fef183..3fdea56f 100644 --- a/tests/stacklets/test_microbot_voice.py +++ b/tests/stacklets/test_microbot_voice.py @@ -1,13 +1,11 @@ -"""Voice messages arrive at handlers as text. +"""Voice messages reach handlers as text. -Holding the mic button and typing are the same act: putting words in the -room. Only the encoding differs, so only the transport should know about -it. These tests pin that contract from the caller's side — a bot registers -an ordinary text handler and receives spoken words through it, with the -speaker still the sender, having passed every gate a typed message passes. +Speech and typing differ only in encoding, so the decode happens in the +transport and handlers see an ordinary text event: same sender, same +event id, same thread, having passed the same gates. -The whisper call itself is stubbed; what is under test is the framework -wiring, not the speech model. +The whisper call is stubbed throughout. What is under test is the +framework wiring, not the speech model. """ from __future__ import annotations @@ -167,9 +165,9 @@ async def test_a_text_handler_receives_the_spoken_words(self, tmp_path): @pytest.mark.asyncio async def test_the_speaker_remains_the_sender(self, tmp_path): - """Replies, reactions and thread ownership all key off the sender - and event id. If the framework re-attributed the words to itself, - a bot would answer its own message and corrections would break.""" + """Replies, reactions and thread ownership are keyed on the + sender and event id. Re-attributing the words to the framework + would detach all three from the visible message.""" bot = _bot(tmp_path, transcriber=_StubTranscriber()) seen = _collect(bot) @@ -196,9 +194,9 @@ async def test_a_memo_sent_into_a_thread_stays_in_it(self, tmp_path): @pytest.mark.asyncio async def test_the_words_are_marked_as_transcribed(self, tmp_path): - """Whisper is lossy in a way a keyboard is not, so provenance has - to survive: what was said, and the audio it came from, so a bad - transcript is visible and correctable.""" + """Transcription can be wrong in ways typing cannot, so the + decoded event records the audio it came from and the reply layer + can quote the words back.""" bot = _bot(tmp_path, transcriber=_StubTranscriber()) seen = _collect(bot) @@ -216,8 +214,8 @@ async def test_the_words_are_marked_as_transcribed(self, tmp_path): @pytest.mark.asyncio async def test_audio_handlers_no_longer_see_it(self, tmp_path): - """The decode is total, not a fan-out: nothing downstream is left - holding raw audio, so there is no second place to transcribe.""" + """The decode replaces the event rather than duplicating it, so + no handler is left holding raw audio to transcribe again.""" bot = _bot(tmp_path, transcriber=_StubTranscriber()) audio_seen = _collect(bot, RoomMessageAudio) text_seen = _collect(bot, RoomMessageText) @@ -240,9 +238,8 @@ async def test_typed_messages_pass_through_untouched(self, tmp_path): class TestWhenTheWordsCannotBeRecovered: - """Undecodable speech is dispatched to nobody. It is not an error to - answer: the family sees their own voice message sitting in the room, - the same as a message in a language the stack cannot read.""" + """Speech that cannot be decoded is dispatched to no handler and + draws no reply. The recording stays visible in the room.""" @pytest.mark.asyncio async def test_without_whisper_nothing_is_dispatched(self, tmp_path): @@ -289,9 +286,8 @@ async def _no_bytes(mxc_url): class TestOneWhisperRunPerMessage: - """Every bot in a room drains the same timeline in the same process. - Without sharing, a five-minute memo would be sent to whisper once per - bot listening.""" + """Bots in a room drain the same timeline in one process, so without + sharing a long recording would be sent to whisper once per bot.""" @pytest.mark.asyncio async def test_two_bots_transcribe_the_same_memo_once(self, tmp_path): @@ -311,8 +307,8 @@ async def test_two_bots_transcribe_the_same_memo_once(self, tmp_path): @pytest.mark.asyncio async def test_a_stored_transcript_is_not_produced_again(self, tmp_path): - """The backfill and the drain are different processes over one - store. Whatever already cost GPU time must never cost it twice.""" + """The drain and a backfill are separate processes over one + store, so a transcript already produced is reused.""" first = _StubTranscriber("buy milk") bot = _bot(tmp_path, transcriber=first) _collect(bot) @@ -330,8 +326,8 @@ async def test_a_stored_transcript_is_not_produced_again(self, tmp_path): @pytest.mark.asyncio async def test_a_failure_is_not_remembered(self, tmp_path): - """A whisper outage must not poison the message forever — the - drain is at-least-once, so the retry has to be able to succeed.""" + """Outages are transient and the drain retries, so a failure is + not retained.""" from stack.ai.client import LLMError broken = _StubTranscriber(error=LLMError("whisper down")) @@ -393,10 +389,9 @@ def test_audio_without_a_payload_is_not_decodable(self): class TestThePolishPass: - """whisper.cpp emits one unbroken lowercase run of words. The polish - pass puts the sentences back. It is kept deliberately word-preserving: - the memories room holds things people said to their children, and a - model "improving" those is not a transcript any more.""" + """whisper returns an unpunctuated run of words and the polish pass + restores sentence boundaries. The prompt forbids changing any word, + so the result stays verbatim.""" @pytest.mark.asyncio async def test_handlers_receive_the_polished_text(self, tmp_path): @@ -415,9 +410,8 @@ async def test_handlers_receive_the_polished_text(self, tmp_path): async def test_the_raw_transcript_is_kept_alongside_the_polished_one( self, tmp_path, ): - """Polishing gets better with better models and costs almost - nothing; whisper does not and is not. Keeping the raw text means - years of recordings can be re-polished without the audio.""" + """Re-polishing is cheap and re-transcribing is not, so the raw + output is kept alongside the polished text.""" bot = _bot( tmp_path, transcriber=_StubTranscriber("buy milk on the way home"), @@ -434,7 +428,7 @@ async def test_the_raw_transcript_is_kept_alongside_the_polished_one( @pytest.mark.asyncio async def test_without_an_llm_the_raw_transcript_still_lands(self, tmp_path): - """A rough transcript beats no transcript. Polish is never a gate.""" + """Polish is optional; without an LLM the raw output is used.""" bot = _bot(tmp_path, transcriber=_StubTranscriber("buy milk"), cleanup=None) seen = _collect(bot) @@ -446,8 +440,8 @@ async def test_without_an_llm_the_raw_transcript_still_lands(self, tmp_path): class TestTranscriptStore: - """Durable because a backfill over the memories room is years of - recordings, and a separate process from the bot that reads them.""" + """Durable and shared, because a backfill runs in a separate process + from the bots and covers a room's whole history.""" def test_a_record_survives_a_new_store_over_the_same_directory(self, tmp_path): store = voice.TranscriptStore(tmp_path / "t") @@ -468,8 +462,8 @@ def test_event_ids_with_awkward_characters_round_trip(self, tmp_path): assert store.read(awkward)["text"] == "fine" def test_an_unwritable_store_does_not_break_the_message(self, tmp_path): - """A store we cannot write costs us the transcript again later. - It must never cost the family the message now.""" + """An unwritable store costs a later re-transcription, not the + message itself.""" blocked = tmp_path / "afile" blocked.write_text("not a directory") store = voice.TranscriptStore(blocked / "t") @@ -478,10 +472,9 @@ def test_an_unwritable_store_does_not_break_the_message(self, tmp_path): class TestTheWorkingIndicator: - """Whisper runs before any handler can say it is working, so the - framework raises the indicator itself. Whoever raises it owns putting - it down — a bot left "typing" at a room it is not going to answer - sits there for the full five-minute timeout.""" + """Transcription runs before any handler could raise the typing + indicator, so the framework raises it. It must also clear it on the + paths where no handler runs, or it persists for the full timeout.""" @pytest.mark.asyncio async def test_typing_stops_when_no_handler_wants_the_message(self, tmp_path): @@ -508,9 +501,8 @@ async def test_typing_stops_when_the_words_cannot_be_recovered(self, tmp_path): @pytest.mark.asyncio async def test_a_typed_message_never_raises_it(self, tmp_path): - """Only the decode needs the indicator; ordinary text is fast and - the handler decides for itself. (The handler wrap still clears it - on the way out, as it does for every message.)""" + """Only the decode raises it. The handler wrap still clears it on + the way out, as it does for every message.""" bot = _bot(tmp_path, transcriber=_StubTranscriber()) _collect(bot) diff --git a/tests/stacklets/test_scribe.py b/tests/stacklets/test_scribe.py index 292e3e3d..edc9b7b7 100644 --- a/tests/stacklets/test_scribe.py +++ b/tests/stacklets/test_scribe.py @@ -1,15 +1,12 @@ """Scribe retires itself. -Transcription moved into the transport, so Scribe has no job left. It -ships for one more release as a shell that explains itself and leaves, -because the framework has no way to deprovision a removed bot: delete the -declaration and the Matrix account simply survives, still joined to -whatever room someone invited it to, answering nothing forever. - -Scribe declared no room of its own, so the only installs affected are the -ones where a person went looking for it and invited it by hand. Those are -exactly the people who would notice it going quiet, which is why it says -goodbye rather than just stopping. +Transcription moved into the transport, leaving this bot without a job. +It ships for one more release because the framework cannot deprovision a +bot: removing the declaration leaves the Matrix account joined to rooms +and answering nothing. + +Scribe declared no room of its own, so the only affected installs are +those where someone invited it by hand. """ from __future__ import annotations @@ -75,8 +72,8 @@ async def test_it_says_goodbye_and_leaves_every_room(self, tmp_path): @pytest.mark.asyncio async def test_the_goodbye_explains_itself(self, tmp_path): - """A member that vanishes without a word is a mystery to debug. - It has to say what replaced it and that nothing is lost.""" + """The notice names what replaced the bot and confirms nothing + needs setting up, so its departure is self-explanatory.""" client = _FakeClient("!kitchen:simpson") bot = _bot(tmp_path, client) @@ -90,8 +87,8 @@ async def test_the_goodbye_explains_itself(self, tmp_path): @pytest.mark.asyncio async def test_it_leaves_a_room_it_is_freshly_invited_to(self, tmp_path): - """Someone following an older guide invites it. Same answer, so - the invite does not leave a silent member behind.""" + """An invite gets the same response as the boot sweep, so it + does not leave a silent member behind.""" client = _FakeClient("!new:simpson") bot = _bot(tmp_path, client) @@ -104,8 +101,8 @@ async def test_it_leaves_a_room_it_is_freshly_invited_to(self, tmp_path): async def test_a_room_it_cannot_leave_does_not_stop_the_others( self, tmp_path, ): - """The sweep runs on every launch, so a failure is retried next - boot. It must not strand the rooms behind it in the meantime.""" + """A room that cannot be left is retried on the next launch, and + must not block the rooms after it in this one.""" client = _FakeClient("!stuck:simpson", "!fine:simpson", leave_error=RuntimeError("homeserver said no")) bot = _bot(tmp_path, client) @@ -126,9 +123,9 @@ async def test_it_speaks_the_household_language(self, tmp_path, monkeypatch): class TestScribeAnswersNothing: - """It transcribes nothing and replies to nothing. The framework does - the transcribing now, and a second transcriber in the same process is - the exact thing this release removed.""" + """The bot registers no handlers and builds no transcriber. The + framework performs the decode, and a second transcriber in the same + process is what this release removed.""" @pytest.mark.asyncio async def test_it_registers_no_message_handlers(self, tmp_path): @@ -141,8 +138,8 @@ async def test_it_registers_no_message_handlers(self, tmp_path): assert bot._handlers == [] def test_it_builds_no_transcriber_of_its_own(self, tmp_path, monkeypatch): - """MicroBot gives every bot one for the transport decode; Scribe - must not reach for it. Nothing here should ever call whisper.""" + """MicroBot builds one for the transport decode. Scribe does not + use it and should never reach whisper.""" monkeypatch.setenv("WHISPER_URL", "http://localhost:42062/v1") bot = _bot(tmp_path, _FakeClient()) assert not hasattr(bot, "_scribe_transcriber")