diff --git a/CHANGELOG.md b/CHANGELOG.md index f9aa42e6..31a763d8 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/morphobricks_preprocessing/run/description_file.py b/src/morphobricks_preprocessing/run/description_file.py index ec159bc7..bc438242 100644 --- a/src/morphobricks_preprocessing/run/description_file.py +++ b/src/morphobricks_preprocessing/run/description_file.py @@ -8,12 +8,19 @@ 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 @@ -21,10 +28,11 @@ 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")) diff --git a/tests/test_description_file.py b/tests/test_description_file.py index a7b20ad7..8b97ef75 100644 --- a/tests/test_description_file.py +++ b/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") diff --git a/tests/test_spec_encoding.py b/tests/test_spec_encoding.py index 6005ccd8..aafcf087 100644 --- a/tests/test_spec_encoding.py +++ b/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