feat(523): the backup as an uncompressed tar of per-table JSON Lines, ordered so restic dedupes it

export_full_backup orders every query by primary key: an UPDATE moves a
row to the end of the heap, so an unordered export shifted every byte
after an edited row. services/backup_file.py writes the payload as one
dated tar per run - manifest.json plus <section>.jsonl, sorted keys,
fixed member headers - atomically via a dot-prefixed .part file, and
reads it back into the dict restore_full_backup takes, refusing a
missing or short section.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/scribe/services/backup.py

@@ -982,42 +982,53 @@ def _rule_rows(rows) -> list[dict]:
 # ---------------------------------------------------------------------------
 
 async def export_full_backup() -> dict:
-    """Export all data as a version-5 JSON backup."""
+    """Export all data as a full backup payload.
+
+    Every query is ORDERED, by primary key unless a restore needs another
+    order. Postgres returns an unordered select in heap order, and an UPDATE
+    moves a row to the end of the heap: an edited note would jump to the
+    bottom of `notes`, shifting every byte after its old place. The scheduled
+    backup file (services/backup_file.py) is deduplicated by restic between
+    runs, which only works when unchanged rows serialise to the same bytes in
+    the same place.
+    """
     async with async_session() as session:
-        users = (await session.execute(select(User))).scalars().all()
-        projects = (await session.execute(select(Project))).scalars().all()
-        milestones = (await session.execute(select(Milestone))).scalars().all()
-        notes = (await session.execute(select(Note))).scalars().all()
-        task_logs = (await session.execute(select(TaskLog))).scalars().all()
-        note_drafts = (await session.execute(select(NoteDraft))).scalars().all()
+        users =(await session.execute(select(User).order_by(User.id))).scalars().all()
+        projects = (await session.execute(select(Project).order_by(Project.id))).scalars().all()
+        milestones = (await session.execute(select(Milestone).order_by(Milestone.id))).scalars().all()
+        notes = (await session.execute(select(Note).order_by(Note.id))).scalars().all()
+        task_logs = (await session.execute(select(TaskLog).order_by(TaskLog.id))).scalars().all()
+        note_drafts = (await session.execute(select(NoteDraft).order_by(NoteDraft.id))).scalars().all()
         note_versions = (await session.execute(
             select(NoteVersion).order_by(NoteVersion.note_id, NoteVersion.id)
         )).scalars().all()
         rule_versions = (await session.execute(
             select(RuleVersion).order_by(RuleVersion.rule_id, RuleVersion.id)
         )).scalars().all()
-        settings = (await session.execute(select(Setting))).scalars().all()
-        systems = (await session.execute(select(System))).scalars().all()
+        settings = (await session.execute(select(Setting).order_by(Setting.user_id, Setting.key))).scalars().all()
+        systems = (await session.execute(select(System).order_by(System.id))).scalars().all()
         canonical_systems = (await session.execute(
             select(CanonicalSystem).where(CanonicalSystem.deleted_at.is_(None))
-            .order_by(CanonicalSystem.order_index)
+            .order_by(CanonicalSystem.order_index, CanonicalSystem.id)
         )).scalars().all()
         rule_system_rows = (await session.execute(
             select(rule_systems_t.c.rule_id, CanonicalSystem.slug)
             .join(CanonicalSystem, CanonicalSystem.id == rule_systems_t.c.canonical_id)
+            .order_by(rule_systems_t.c.rule_id, CanonicalSystem.slug)
         )).all()
         rule_moment_rows = (await session.execute(
             select(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
+            .order_by(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
         )).all()
         rule_moment_judgments = (await session.execute(
-            select(RuleMomentJudgment)
+            select(RuleMomentJudgment).order_by(RuleMomentJudgment.id)
         )).scalars().all()
-        rule_relations = (await session.execute(select(RuleRelation))).scalars().all()
-        lesson_rule_links = (await session.execute(select(LessonRuleLink))).scalars().all()
-        lesson_no_rule = (await session.execute(select(LessonNoRule))).scalars().all()
-        record_systems = (await session.execute(select(RecordSystem))).scalars().all()
+        rule_relations = (await session.execute(select(RuleRelation).order_by(RuleRelation.id))).scalars().all()
+        lesson_rule_links = (await session.execute(select(LessonRuleLink).order_by(LessonRuleLink.id))).scalars().all()
+        lesson_no_rule = (await session.execute(select(LessonNoRule).order_by(LessonNoRule.lesson_id))).scalars().all()
+        record_systems = (await session.execute(select(RecordSystem).order_by(RecordSystem.id))).scalars().all()
         supersessions = (
-            await session.execute(select(NoteSupersession))
+            await session.execute(select(NoteSupersession).order_by(NoteSupersession.id))
         ).scalars().all()
         # Parent-first, so a restore can resolve parent_id as it goes rather
         # than needing a second pass — the self-FK is the only ordering
@@ -1026,13 +1037,13 @@ async def export_full_backup() -> dict:
             select(DesignSystem).order_by(DesignSystem.parent_id.nullsfirst(),
                                           DesignSystem.id)
         )).scalars().all()
-        design_tokens = (await session.execute(select(DesignToken))).scalars().all()
-        usage_events = (await session.execute(select(NoteUsageEvent))).scalars().all()
+        design_tokens = (await session.execute(select(DesignToken).order_by(DesignToken.id))).scalars().all()
+        usage_events = (await session.execute(select(NoteUsageEvent).order_by(NoteUsageEvent.id))).scalars().all()
         rule_usage_events = (
-            await session.execute(select(RuleUsageEvent))
+            await session.execute(select(RuleUsageEvent).order_by(RuleUsageEvent.id))
         ).scalars().all()
         system_usage_events = (
-            await session.execute(select(SystemUsageEvent))
+            await session.execute(select(SystemUsageEvent).order_by(SystemUsageEvent.id))
         ).scalars().all()
         # Oldest first, so a restored history reads in the order the dials
         # actually moved — the sequence IS the argument when a surface has been
@@ -1043,30 +1054,33 @@ async def export_full_backup() -> dict:
         moment_mappings = (await session.execute(
             select(MomentMapping).order_by(MomentMapping.id)
         )).scalars().all()
-        repo_bindings = (await session.execute(select(RepoBinding))).scalars().all()
-        code_shapes = (await session.execute(select(CodeShape))).scalars().all()
+        repo_bindings = (await session.execute(select(RepoBinding).order_by(RepoBinding.id))).scalars().all()
+        code_shapes = (await session.execute(select(CodeShape).order_by(CodeShape.id))).scalars().all()
         code_shape_events = (await session.execute(
             select(CodeShapeEvent).order_by(CodeShapeEvent.at, CodeShapeEvent.id)
         )).scalars().all()
         code_shape_uses = (await session.execute(
-            select(CodeShapeUse).order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id)
+            select(CodeShapeUse).order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id, CodeShapeUse.id)
         )).scalars().all()
-        rulebooks = (await session.execute(select(Rulebook))).scalars().all()
-        topics = (await session.execute(select(RulebookTopic))).scalars().all()
-        rules = (await session.execute(select(Rule))).scalars().all()
+        rulebooks = (await session.execute(select(Rulebook).order_by(Rulebook.id))).scalars().all()
+        topics = (await session.execute(select(RulebookTopic).order_by(RulebookTopic.id))).scalars().all()
+        rules = (await session.execute(select(Rule).order_by(Rule.id))).scalars().all()
         platforms = (await session.execute(
             select(Platform).where(Platform.deleted_at.is_(None))
-            .order_by(Platform.order_index)
+            .order_by(Platform.order_index, Platform.id)
         )).scalars().all()
-        project_platforms = (await session.execute(select(ProjectPlatform))).scalars().all()
-        family_ideas = (await session.execute(select(FamilyIdea))).scalars().all()
+        project_platforms = (await session.execute(select(ProjectPlatform).order_by(
+            ProjectPlatform.project_id, ProjectPlatform.platform_id))).scalars().all()
+        family_ideas = (await session.execute(select(FamilyIdea).order_by(FamilyIdea.note_id))).scalars().all()
         family_idea_platforms = (await session.execute(
-            select(FamilyIdeaPlatform)
+            select(FamilyIdeaPlatform).order_by(
+                FamilyIdeaPlatform.note_id, FamilyIdeaPlatform.platform_id)
         )).scalars().all()
         family_idea_references = (await session.execute(
-            select(FamilyIdeaReference)
+            select(FamilyIdeaReference).order_by(
+                FamilyIdeaReference.idea_id, FamilyIdeaReference.snippet_id)
         )).scalars().all()
-        family_adoptions = (await session.execute(select(FamilyAdoption))).scalars().all()
+        family_adoptions = (await session.execute(select(FamilyAdoption).order_by(FamilyAdoption.id))).scalars().all()
         # Oldest first: a precedent is always an earlier decision, so a restore
         # in this order has every precedent mapped before anything cites it.
         family_decisions = (await session.execute(

src/scribe/services/backup_file.py

@@ -0,0 +1,214 @@
+"""The full backup as a file on disk: one uncompressed tar per run.
+
+The scheduled backup lands somewhere that keeps a history of it — a folder a
+backup service watches, an S3 bucket — and from there it is usually copied
+off-site by a deduplicating tool such as restic. Everything about the layout
+serves that copy: restic cuts a file into chunks at boundaries chosen by the
+content, so a chunk it has seen before costs nothing to store again. The file
+is therefore laid out so that what did not change between two runs is the
+same bytes, in the same place, both times.
+
+- **Never compressed.** Compression rewrites every byte after the first
+  change, so no chunk would ever match. restic compresses its own repository
+  after deduplicating.
+- **One member per payload section**, `<section>.jsonl`, one row per line.
+  Rows arrive in primary-key order (export_full_backup orders every query)
+  and each is written with sorted keys and no spacing, so an unchanged row is
+  identical bytes, an edit changes one line, and rows appended to a growing
+  table land at the end of its member. A busy table never shifts the bytes of
+  a quiet one.
+- **The parts that change every run live in `manifest.json` only**: the
+  export time, the app version and the row counts. The member headers are
+  fixed (mtime 0, owner 0, mode 0644) and the members come in one order: the
+  manifest, then the sections by name.
+- **One file per run, under a new dated name**, so a service that keeps one
+  file per backup can thin the history on its own schedule.
+
+The payload inside is exactly `export_full_backup()`'s, so the coverage
+guards on that payload (tests/test_services_backup.py) cover this file too,
+and `read_archive` hands `restore_full_backup` the same dict it always took.
+"""
+from __future__ import annotations
+
+import io
+import json
+import os
+import tarfile
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import IO
+
+ARCHIVE_FORMAT = "scribe-backup-tar"
+ARCHIVE_FORMAT_VERSION = 1
+MANIFEST_NAME = "manifest.json"
+SECTION_SUFFIX = ".jsonl"
+NAME_PREFIX = "scribe-"
+NAME_SUFFIX = ".tar"
+
+
+def archive_name(at: datetime) -> str:
+    """`scribe-20261015T030000Z.tar` — UTC, sortable, one per run."""
+    return f"{NAME_PREFIX}{at.astimezone(timezone.utc):%Y%m%dT%H%M%SZ}{NAME_SUFFIX}"
+
+
+def _encode(value) -> str:
+    # default=str matches the admin download's json.dumps, so a datetime that
+    # reaches the payload unconverted reads back the same way on restore.
+    return json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str,
+    )
+
+
+def _split(payload: dict) -> tuple[dict, dict[str, list]]:
+    """Separate the row sections from the header (version, scope, notices).
+
+    A section is a list of rows under a plain name. `_not_included` is a list
+    too, but of table names: it is a notice, so it stays in the header.
+    """
+    header: dict = {}
+    sections: dict[str, list] = {}
+    for key, value in payload.items():
+        if isinstance(value, list) and not key.startswith("_"):
+            sections[key] = value
+        else:
+            header[key] = value
+    return header, sections
+
+
+def _member(name: str, data: bytes) -> tuple[tarfile.TarInfo, io.BytesIO]:
+    info = tarfile.TarInfo(name)
+    info.size = len(data)
+    info.mtime = 0
+    info.mode = 0o644
+    info.uid = info.gid = 0
+    info.uname = info.gname = ""
+    return info, io.BytesIO(data)
+
+
+def write_archive_to(payload: dict, out: IO[bytes], *, app_version: str) -> dict:
+    """Write `payload` as the backup tar to an open binary file.
+
+    Returns the manifest, so the caller can report row counts without
+    reading the file back.
+    """
+    header, sections = _split(payload)
+    names = sorted(sections)
+    manifest = {
+        "format": ARCHIVE_FORMAT,
+        "format_version": ARCHIVE_FORMAT_VERSION,
+        "app_version": app_version,
+        "header": header,
+        "sections": {name: len(sections[name]) for name in names},
+    }
+    with tarfile.open(fileobj=out, mode="w", format=tarfile.USTAR_FORMAT) as tar:
+        tar.addfile(*_member(
+            MANIFEST_NAME, (json.dumps(manifest, sort_keys=True, indent=2,
+                                       default=str) + "\n").encode(),
+        ))
+        for name in names:
+            data = b"".join(
+                _encode(row).encode() + b"\n" for row in sections[name]
+            )
+            tar.addfile(*_member(name + SECTION_SUFFIX, data))
+    return manifest
+
+
+def write_archive(
+    payload: dict, directory: Path, at: datetime, *, app_version: str,
+) -> Path:
+    """Write the backup tar into `directory` under its dated name, atomically.
+
+    The bytes go to a dot-prefixed `.part` file first, are fsync'd, and only
+    then renamed to the final name. A watcher of the folder (or anything else
+    listing it) never sees half a backup under a backup's name, and a run that
+    fails leaves nothing behind but the error.
+    """
+    directory.mkdir(parents=True, exist_ok=True)
+    final = directory / archive_name(at)
+    part = directory / f".{final.name}.part"
+    try:
+        with open(part, "wb") as f:
+            write_archive_to(payload, f, app_version=app_version)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(part, final)
+    except BaseException:
+        part.unlink(missing_ok=True)
+        raise
+    _fsync_dir(directory)
+    return final
+
+
+def _fsync_dir(directory: Path) -> None:
+    """Make the rename itself durable. Some filesystems (and NFS mounts)
+    refuse a directory fsync; the file's own fsync already happened, so that
+    refusal is not a failure of the backup."""
+    try:
+        fd = os.open(directory, os.O_RDONLY)
+    except OSError:
+        return
+    try:
+        os.fsync(fd)
+    except OSError:
+        pass
+    finally:
+        os.close(fd)
+
+
+class ArchiveError(ValueError):
+    """The file is not a backup tar this version can read, or is damaged."""
+
+
+def is_archive(fileobj: IO[bytes]) -> bool:
+    """True when `fileobj` (seekable, at its start) holds a tar. Leaves the
+    position where it was, so the caller can fall back to reading JSON."""
+    pos = fileobj.tell()
+    try:
+        head = fileobj.read(tarfile.BLOCKSIZE)
+    finally:
+        fileobj.seek(pos)
+    return len(head) == tarfile.BLOCKSIZE and head[257:262] == b"ustar"
+
+
+def read_archive(fileobj: IO[bytes]) -> dict:
+    """Read a backup tar back into the payload `restore_full_backup` takes.
+
+    Every section the manifest names must be present with the row count it
+    names: a member that went missing or was cut short is an error here, not
+    a restore that reports success with less than it was given.
+    """
+    try:
+        tar = tarfile.open(fileobj=fileobj, mode="r:")
+    except tarfile.TarError as exc:
+        raise ArchiveError(f"not a backup archive: {exc}") from exc
+    with tar:
+        manifest = json.loads(_read_member(tar, MANIFEST_NAME))
+        if manifest.get("format") != ARCHIVE_FORMAT:
+            raise ArchiveError("not a Scribe backup archive (no Scribe manifest)")
+        if manifest.get("format_version", 0) > ARCHIVE_FORMAT_VERSION:
+            raise ArchiveError(
+                f"archive format {manifest['format_version']} is newer than "
+                f"this Scribe reads ({ARCHIVE_FORMAT_VERSION}); upgrade first"
+            )
+        payload = dict(manifest.get("header") or {})
+        for name, count in (manifest.get("sections") or {}).items():
+            raw = _read_member(tar, name + SECTION_SUFFIX)
+            rows = [json.loads(line) for line in raw.splitlines() if line]
+            if len(rows) != count:
+                raise ArchiveError(
+                    f"{name}{SECTION_SUFFIX} holds {len(rows)} rows; "
+                    f"the manifest says {count}"
+                )
+            payload[name] = rows
+    return payload
+
+
+def _read_member(tar: tarfile.TarFile, name: str) -> bytes:
+    try:
+        member = tar.getmember(name)
+    except KeyError as exc:
+        raise ArchiveError(f"the archive has no {name}") from exc
+    f = tar.extractfile(member)
+    if f is None:
+        raise ArchiveError(f"{name} in the archive is not a file")
+    return f.read()