Make the records append rollback-safe and validate record shapes

CHANGELOG.md

@@ -8,6 +8,16 @@ are in [docs/plan.md](docs/plan.md) §12. Upgrading from v0.3: see the
 
 ## Unreleased
 
+- **The records append is rolled back with the description (K4 of external review batch 26).**
+  `_append_records` appended straight to the live `morphobricks.records.jsonl`; a `write`,
+  `flush` or `fsync` that failed after bytes were written left a torn line, or records naming a
+  hash of a description that was rolled back, because only the TOML was restored. The new file
+  (old bytes plus new lines) is now written in full to a synced temporary file and replaced
+  atomically, and a failed publication puts back both the description and the records bytes.
+  `read_records` now also refuses any line that is not one of the three Spec record types
+  (`reconfirmation`, `image_count_changed`, `removed`) with exactly that type's keys (`{}`
+  was accepted before). A record's hashes are history: a reader never compares them with the
+  published file.
 - **The Spec test gate holds module-level skips and missing modules (#203).** A `test_spec_*.py`
   module skipped while it is collected (`pytest.skip(..., allow_module_level=True)`, a module-level
   `importorskip`) is now a failed collection, and a run over the whole tests folder fails when no

src/morphobricks_preprocessing/run/description_file.py

@@ -8,23 +8,31 @@
 replaces the file atomically. Every refusal is reported as a set and writes nothing.
 
 The user's re-confirmations that the candidate needed are appended, in the same publication, to
-the append-only ``morphobricks.records.jsonl`` next to the file (one JSON object per line: old
-ID, new ID, date and the SHA-256 of the file before and after). The file is never rewritten or
-truncated; a later run copies it unchanged into its manifest. If the append fails after the file
-was replaced, the previous file is put back, so the publication is whole or absent (a process
-killed in the instant between the two steps can leave a published file without its records, and
-the records file is then the one that cannot be trusted to be complete).
+the append-only ``morphobricks.records.jsonl`` next to the file (one JSON object per line, of one
+of the three Spec record types, each with its fixed key set). The records are never edited or
+dropped: the new file is the old bytes plus the new lines, written in full to a temporary file and
+put in place with one atomic replace, so the live records file is never half written. A later run
+copies it unchanged into its manifest. If any step after the description was replaced fails, the
+previous description and the previous records file are put back, so the publication is whole or
+absent (a process killed in the instant between the two replaces can leave a published file
+without its records, and the records file is then the one that cannot be trusted to be complete).
+
+The hashes in a record are history, not a check on the file now published: later publications
+change the file without a record, so an older record rightly names a hash that is no longer the
+published one. A reader never compares them with the published file and never drops or trusts a
+record because they match or differ; it reads the records in order, unchanged.
 """
 
 from __future__ import annotations
 
 import json
 import os
 import re
+import tempfile
 from dataclasses import dataclass
 from pathlib import Path
 
-from .._io import write_text_atomic
+from .._io import give_output_mode, write_text_atomic
 from ..description import Description, DescriptionError, description_from_bytes
 from ..description_file import (
     DESCRIPTION_FILENAME,
@@ -59,6 +67,40 @@ def __init__(self, issues: tuple[DescriptionIssue, ...]) -> None:
         super().__init__(f"the description was not published: {shown}")
 
 
+RECORD_KEYS: dict[str, frozenset[str]] = {
+    "reconfirmation": frozenset(
+        {"type", "id_kind", "old_id", "new_id", "confirmed_on", "old_sha256", "new_sha256"}
+    ),
+    "image_count_changed": frozenset(
+        {
+            "type",
+            "condition_id",
+            "condition_source_folder",
+            "repeat_id",
+            "source_repeat",
+            "technical_repeats",
+            "old_image_count",
+            "new_image_count",
+            "recorded_on",
+        }
+    ),
+    "removed": frozenset(
+        {
+            "type",
+            "entity",
+            "condition_id",
+            "condition_source_folder",
+            "repeat_id",
+            "source_repeat",
+            "technical_repeat_id",
+            "technical_repeat_source_folder",
+            "confirmed_on",
+        }
+    ),
+}
+"""The key set of each record type (Spec experiment-v1 4.6): a line carries exactly these keys."""
+
+
 class RecordsFileError(OSError):
     """Raised when ``morphobricks.records.jsonl`` cannot be appended to as it stands."""
 
@@ -118,7 +160,8 @@ def read_records(output_folder: str | os.PathLike) -> tuple[dict, ...]:
     DescriptionError
         ``R_FILE_SYMLINK`` when the records file is a symbolic link.
     RecordsFileError
-        When a line is not a JSON object or the last line is torn (no final newline).
+        When a line is not a JSON object, is not one of the three Spec record types with exactly
+        that type's keys (4.6), or the last line is torn (no final newline).
     """
     data = read_file_bytes(Path(output_folder) / RECORDS_FILENAME)
     if not data:
@@ -132,21 +175,47 @@ def read_records(output_folder: str | os.PathLike) -> tuple[dict, ...]:
         raise RecordsFileError(f"{RECORDS_FILENAME} holds a line that is not JSON") from error
     if not all(isinstance(row, dict) for row in rows):
         raise RecordsFileError(f"{RECORDS_FILENAME} holds a line that is not an object")
+    for number, row in enumerate(rows, start=1):
+        keys = RECORD_KEYS.get(row.get("type")) if isinstance(row.get("type"), str) else None
+        if keys is None:
+            raise RecordsFileError(f"{RECORDS_FILENAME} line {number}: no known record type")
+        if set(row) != keys:
+            odd = sorted(set(row) ^ keys)
+            raise RecordsFileError(
+                f"{RECORDS_FILENAME} line {number}: a {row['type']} record has exactly the keys "
+                f"{sorted(keys)}; these differ: {odd}"
+            )
     return rows
 
 
+def _replace_file_durably(path: Path, data: bytes) -> None:
+    """Replace ``path`` with ``data``: a synced temporary file, then one atomic replace."""
+    handle, temporary = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
+    try:
+        with os.fdopen(handle, "wb") as stream:
+            stream.write(data)
+            stream.flush()
+            os.fsync(stream.fileno())
+        give_output_mode(temporary, path)
+        os.replace(temporary, path)
+    except BaseException:
+        Path(temporary).unlink(missing_ok=True)
+        raise
+
+
 def _append_records(folder: Path, records: tuple[dict, ...]) -> None:
-    """Append the records as lines; the file is created if absent and never rewritten."""
+    """Add the records as lines after the existing ones, in one atomic replace of the file.
+
+    The live file is replaced only once the old bytes and the new lines are all on disk in a
+    temporary file, so a failure leaves it exactly as it was.
+    """
     if not records:
         return
     read_records(folder)  # refuses a symbolic link and a torn or foreign file before writing
+    path = folder / RECORDS_FILENAME
+    current = read_file_bytes(path) or b""
     lines = "".join(json.dumps(r, ensure_ascii=False, sort_keys=True) + "\n" for r in records)
-    flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0)
-    handle = os.open(folder / RECORDS_FILENAME, flags, 0o644)
-    with os.fdopen(handle, "ab") as stream:
-        stream.write(lines.encode("utf-8"))
-        stream.flush()
-        os.fsync(stream.fileno())
+    _replace_file_durably(path, current + lines.encode("utf-8"))
 
 
 def _compare(
@@ -210,6 +279,7 @@ def publish_description(
         try:
             published = read_file_bytes(path)
             read_records(folder)  # a symbolic or torn records file refuses before anything
+            records_before = read_file_bytes(folder / RECORDS_FILENAME)
         except DescriptionError as error:
             raise PublicationRefused(error.issues) from error
         base = _check_base(base_sha256, published)
@@ -223,13 +293,14 @@ def publish_description(
             _append_records(folder, records)
         except BaseException:
             _restore(path, published)
+            _restore(folder / RECORDS_FILENAME, records_before)
             raise
         return Publication(path, new_sha, base, records)
 
 
-def _restore(path: Path, published: bytes | None) -> None:
-    """Put the previous published file back (or remove the new one when there was none)."""
-    if published is None:
+def _restore(path: Path, before: bytes | None) -> None:
+    """Put the previous bytes of ``path`` back (or remove the file when there was none)."""
+    if before is None:
         path.unlink(missing_ok=True)
     else:
-        write_text_atomic(path, published.decode("utf-8"))
+        write_text_atomic(path, before.decode("utf-8"))

tests/test_description_file.py

@@ -154,8 +154,41 @@ def test_a_torn_records_file_refuses_before_anything_is_written(tmp_path):
     assert _snapshot(tmp_path) == before
 
 
-@pytest.mark.parametrize("content", [b"not json\n", b"[1, 2]\n"])
-def test_a_records_file_that_is_not_lines_of_objects_is_refused(tmp_path, content):
+SEEDED = (
+    b'{"confirmed_on": "2026-01-01", "id_kind": "repeats", "new_id": "b", "new_sha256": "'
+    + b"b" * 64
+    + b'", "old_id": "a", "old_sha256": "'
+    + b"a" * 64
+    + b'", "type": "reconfirmation"}\n'
+)
+
+
+@pytest.mark.parametrize(
+    "content",
+    [
+        b"not json\n",
+        b"[1, 2]\n",
+        b"{}\n",  # an object, but no record
+        b'{"type": "reconfirmation"}\n',  # the right type, none of its keys
+        b'{"type": "unknown", "old_id": "a"}\n',  # a type the Spec does not define
+        b'{"old_id": "a", "new_id": "b"}\n',  # no type
+        SEEDED.replace(b'"type"', b'"extra": 1, "type"'),  # a key of no type
+        SEEDED.replace(b'"old_sha256"', b'"old_sha"'),  # one key renamed
+        b'{"type": "removed", "entity": "unit", "confirmed_on": "2026-01-01"}\n',
+    ],
+    ids=[
+        "not_json",
+        "array",
+        "empty_object",
+        "type_only",
+        "unknown_type",
+        "no_type",
+        "extra_key",
+        "wrong_key",
+        "removed_partial",
+    ],
+)
+def test_a_records_file_that_is_not_lines_of_spec_records_is_refused(tmp_path, content):
     (tmp_path / RECORDS_FILENAME).write_bytes(content)
     with pytest.raises(RecordsFileError):
         read_records(tmp_path)
@@ -193,6 +226,95 @@ def broken(folder, records):
     assert _snapshot(tmp_path) == before and not (tmp_path / LOCK_FILENAME).exists()
 
 
+def _fail_binary_writes(monkeypatch, *, after_full_bytes: bool) -> None:
+    """Make a binary write stream opened with ``os.fdopen`` raise once it is written to.
+
+    The records file is the only binary write stream a publication opens (the description goes
+    through a text stream). With ``after_full_bytes`` false, only half of the bytes are written.
+    """
+    real = os.fdopen
+
+    class Failing:
+        def __init__(self, stream):
+            self.stream = stream
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            self.stream.close()
+
+        def fileno(self):
+            return self.stream.fileno()
+
+        def flush(self):
+            self.stream.flush()
+
+        def write(self, data):
+            self.stream.write(data if after_full_bytes else data[: len(data) // 2])
+            self.stream.flush()
+            raise OSError("disk full")
+
+    def fdopen(fd, mode="r", *args, **kwargs):
+        stream = real(fd, mode, *args, **kwargs)
+        return Failing(stream) if "b" in mode and ("w" in mode or "a" in mode) else stream
+
+    monkeypatch.setattr(publication.os, "fdopen", fdopen)
+
+
+@pytest.mark.parametrize("after_full_bytes", [False, True], ids=["partial_bytes", "full_bytes"])
+@pytest.mark.parametrize("seeded", [False, True], ids=["no_records_file", "existing_records"])
+def test_a_write_that_fails_inside_the_append_leaves_file_and_records_as_before(
+    tmp_path, monkeypatch, after_full_bytes, seeded
+):
+    base = _published(tmp_path)
+    if seeded:
+        (tmp_path / RECORDS_FILENAME).write_bytes(SEEDED)
+    before = _snapshot(tmp_path)
+    _fail_binary_writes(monkeypatch, after_full_bytes=after_full_bytes)
+    with pytest.raises(OSError, match="disk full"):
+        publish_description(
+            tmp_path,
+            description_from_bytes(BY_NAME["t01"].data),
+            base_sha256=base,
+            confirmations=RENAMES,
+        )
+    assert _snapshot(tmp_path) == before  # file, records, no temporary file, no lock
+
+
+def test_a_failed_fsync_of_the_records_leaves_file_and_records_as_before(tmp_path, monkeypatch):
+    base = _published(tmp_path)
+    (tmp_path / RECORDS_FILENAME).write_bytes(SEEDED)
+    before = _snapshot(tmp_path)
+
+    def broken(fd):
+        raise OSError("fsync failed")
+
+    monkeypatch.setattr(publication.os, "fsync", broken)
+    with pytest.raises(OSError, match="fsync failed"):
+        publish_description(
+            tmp_path,
+            description_from_bytes(BY_NAME["t01"].data),
+            base_sha256=base,
+            confirmations=RENAMES,
+        )
+    assert _snapshot(tmp_path) == before
+
+
+def test_a_record_whose_hash_is_not_the_published_file_is_history_not_an_error(tmp_path):
+    base = _published(tmp_path)
+    (tmp_path / RECORDS_FILENAME).write_bytes(SEEDED)  # names two hashes that were never published
+    assert read_records(tmp_path)[0]["new_sha256"] == "b" * 64
+    done = publish_description(
+        tmp_path,
+        description_from_bytes(BY_NAME["t01"].data),
+        base_sha256=base,
+        confirmations=RENAMES,
+    )
+    assert read_records(tmp_path)[1:] == done.records  # appended after it, the first unchanged
+    assert (tmp_path / RECORDS_FILENAME).read_bytes().startswith(SEEDED)
+
+
 def test_a_failed_append_removes_a_first_publication(tmp_path, monkeypatch):
     def broken(folder, records):
         raise OSError("disk full")

tests/test_spec_encoding.py

@@ -63,7 +63,15 @@ def test_the_records_file_bytes_are_those_of_section_4_6(tmp_path):
 
 def test_non_ascii_text_is_written_unescaped_and_read_back_whole(tmp_path):
     # U+2028 and U+0085 are line breaks to str.splitlines but not to the file format.
-    record = {"type": "reconfirmation", "old_id": "café
x\u0085y", "new_id": None}
+    record = {
+        "type": "reconfirmation",
+        "id_kind": "repeats",
+        "old_id": "café
x\u0085y",
+        "new_id": None,
+        "confirmed_on": "2026-11-07",
+        "old_sha256": "absent",
+        "new_sha256": "0" * 64,
+    }
     _append_records(tmp_path, (record,))
     data = (tmp_path / RECORDS_FILENAME).read_bytes()
     assert "café
x\u0085y".encode() in data and b"\\u" not in data