From 24a5915818a46a7f17d685e8008139e902700f06 Mon Sep 17 00:00:00 2001
From: KumamuKuma <157353275+KumamuKuma@users.noreply.github.com>
Date: Mon, 20 Jul 2026 16:54:02 +0800
Subject: [PATCH 1/2] fix(store): make object writes atomic and verifiable

---
 src/rgit/store/objects.py |  83 +++++++++++++++++-
 tests/test_objects.py     | 173 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 252 insertions(+), 4 deletions(-)

diff --git a/src/rgit/store/objects.py b/src/rgit/store/objects.py
index e2af728..10a77a5 100644
--- a/src/rgit/store/objects.py
+++ b/src/rgit/store/objects.py
@@ -1,12 +1,22 @@
 import hashlib
 import json
+import os
+import time
 from pathlib import Path
-from typing import Any
+from typing import Any, Callable, TypeVar
+from uuid import uuid4
+
+
+_T = TypeVar("_T")
 
 
 class ObjectStore:
     """Immutable sha256-addressed blob store under a directory."""
 
+    _DIGEST_LENGTH = hashlib.sha256().digest_size * 2
+    _DIGEST_CHARS = frozenset("0123456789abcdef")
+    _ACCESS_RETRY_DELAYS = (0.001, 0.002, 0.004, 0.008, 0.016, 0.032)
+
     def __init__(self, root: Path, create: bool = True):
         self.root = Path(root)
         if create:
@@ -15,17 +25,82 @@ def __init__(self, root: Path, create: bool = True):
     def path_for(self, digest: str) -> Path:
         """On-disk location of a digest — the single source of truth for the
         store layout, so read-only consumers (doctor) can't drift from it."""
+        if not self.is_valid_digest(digest):
+            raise ValueError(f"invalid sha256 digest: {digest!r}")
         return self.root / digest[:2] / digest[2:]
 
     def _path(self, digest: str) -> Path:
         return self.path_for(digest)
 
+    @classmethod
+    def is_valid_digest(cls, digest: object) -> bool:
+        """Return whether *digest* is a canonical lowercase sha256 hex value."""
+        return (
+            isinstance(digest, str)
+            and len(digest) == cls._DIGEST_LENGTH
+            and all(char in cls._DIGEST_CHARS for char in digest)
+        )
+
+    def verify(self, digest: str) -> bool:
+        """Return whether the stored bytes hash to *digest*.
+
+        Missing objects still raise ``FileNotFoundError`` so callers such as
+        doctor can distinguish an absent object from a corrupt one.
+        """
+        path = self.path_for(digest)
+        with path.open("rb") as blob:
+            actual = hashlib.file_digest(blob, "sha256").hexdigest()
+        return actual == digest
+
+    @classmethod
+    def _retry_permission_error(cls, operation: Callable[[], _T]) -> _T:
+        """Retry brief sharing violations without hiding persistent errors."""
+        for delay in cls._ACCESS_RETRY_DELAYS:
+            try:
+                return operation()
+            except PermissionError:
+                time.sleep(delay)
+        return operation()
+
+    @classmethod
+    def _write_atomic(cls, path: Path, data: bytes) -> None:
+        """Durably write *data* before atomically publishing it at *path*."""
+        # Path.open("x") preserves the mode/umask behavior of the previous
+        # direct write while guaranteeing that a stale temp file is not reused.
+        temp_path = path.with_name(f".{path.name}.{uuid4().hex}.tmp")
+        try:
+            with temp_path.open("xb") as blob:
+                blob.write(data)
+                blob.flush()
+                os.fsync(blob.fileno())
+            cls._retry_permission_error(lambda: os.replace(temp_path, path))
+        finally:
+            try:
+                temp_path.unlink()
+            except OSError:
+                pass
+
     def put(self, data: bytes) -> str:
         digest = hashlib.sha256(data).hexdigest()
         p = self._path(digest)
-        if not p.exists():
-            p.parent.mkdir(parents=True, exist_ok=True)
-            p.write_bytes(data)
+        try:
+            if self._retry_permission_error(lambda: self.verify(digest)):
+                return digest
+        except FileNotFoundError:
+            pass
+
+        p.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            self._write_atomic(p, data)
+        except OSError:
+            # Another writer may have won the race. This is success only when
+            # it published the exact object we were trying to store.
+            try:
+                if self._retry_permission_error(lambda: self.verify(digest)):
+                    return digest
+            except OSError:
+                pass
+            raise
         return digest
 
     def get(self, digest: str) -> bytes:
diff --git a/tests/test_objects.py b/tests/test_objects.py
index ebec502..f6d16c9 100644
--- a/tests/test_objects.py
+++ b/tests/test_objects.py
@@ -1,3 +1,10 @@
+from concurrent.futures import ThreadPoolExecutor
+import hashlib
+import os
+
+import pytest
+
+import rgit.store.objects as objects_module
 from rgit.store.objects import ObjectStore
 
 
@@ -14,3 +21,169 @@ def test_put_json_roundtrips(tmp_path):
     store = ObjectStore(tmp_path / "objects")
     h = store.put_json({"b": 2, "a": 1})
     assert store.get_json(h) == {"b": 2, "a": 1}
+
+
+def test_verify_accepts_matching_object_and_rejects_changed_bytes(tmp_path):
+    store = ObjectStore(tmp_path / "objects")
+    digest = store.put(b"complete payload")
+
+    assert store.verify(digest) is True
+
+    store.path_for(digest).write_bytes(b"corrupt payload")
+    assert store.verify(digest) is False
+
+
+def test_verify_distinguishes_missing_object(tmp_path):
+    store = ObjectStore(tmp_path / "objects")
+    digest = "a" * 64
+
+    with pytest.raises(FileNotFoundError):
+        store.verify(digest)
+
+
+@pytest.mark.parametrize("digest", ["", "abc", "A" * 64, "g" * 64, "../escape"])
+def test_object_paths_reject_noncanonical_digests(tmp_path, digest):
+    store = ObjectStore(tmp_path / "objects")
+
+    assert store.is_valid_digest(digest) is False
+    with pytest.raises(ValueError, match="invalid sha256 digest"):
+        store.path_for(digest)
+
+
+def test_put_fsyncs_and_publishes_from_same_directory(tmp_path, monkeypatch):
+    store = ObjectStore(tmp_path / "objects")
+    data = b"atomic payload"
+    digest = hashlib.sha256(data).hexdigest()
+    target = store.path_for(digest)
+    real_fsync = os.fsync
+    real_replace = os.replace
+    fsync_calls = []
+    replace_calls = []
+
+    def recording_fsync(fd):
+        fsync_calls.append(fd)
+        return real_fsync(fd)
+
+    def recording_replace(source, destination):
+        source = source if hasattr(source, "parent") else type(target)(source)
+        destination = (
+            destination if hasattr(destination, "parent") else type(target)(destination)
+        )
+        assert source.parent == target.parent
+        assert destination == target
+        assert source.read_bytes() == data
+        assert not target.exists()
+        replace_calls.append((source, destination))
+        return real_replace(source, destination)
+
+    monkeypatch.setattr(objects_module.os, "fsync", recording_fsync)
+    monkeypatch.setattr(objects_module.os, "replace", recording_replace)
+
+    assert store.put(data) == digest
+    assert fsync_calls
+    assert len(replace_calls) == 1
+    assert target.read_bytes() == data
+
+
+def test_failed_publish_leaves_no_object_or_temp_file(tmp_path, monkeypatch):
+    store = ObjectStore(tmp_path / "objects")
+    data = b"interrupted payload"
+    digest = hashlib.sha256(data).hexdigest()
+    target = store.path_for(digest)
+
+    def fail_replace(source, destination):
+        raise OSError("simulated publish failure")
+
+    monkeypatch.setattr(objects_module.os, "replace", fail_replace)
+
+    with pytest.raises(OSError, match="simulated publish failure"):
+        store.put(data)
+
+    assert not target.exists()
+    assert list(target.parent.iterdir()) == []
+
+
+def test_put_does_not_replace_valid_existing_object(tmp_path, monkeypatch):
+    store = ObjectStore(tmp_path / "objects")
+    data = b"existing payload"
+    digest = store.put(data)
+
+    def unexpected_replace(source, destination):
+        pytest.fail("valid object should not be replaced")
+
+    monkeypatch.setattr(objects_module.os, "replace", unexpected_replace)
+
+    assert store.put(data) == digest
+    assert store.get(digest) == data
+
+
+def test_put_repairs_corrupt_existing_object(tmp_path):
+    store = ObjectStore(tmp_path / "objects")
+    data = b"complete payload"
+    digest = store.put(data)
+    target = store.path_for(digest)
+    target.write_bytes(b"partial")
+
+    assert store.verify(digest) is False
+    assert store.put(data) == digest
+    assert store.get(digest) == data
+    assert store.verify(digest) is True
+
+
+def test_put_accepts_valid_object_published_by_concurrent_writer(
+    tmp_path, monkeypatch
+):
+    store = ObjectStore(tmp_path / "objects")
+    data = b"racing payload"
+    digest = hashlib.sha256(data).hexdigest()
+    target = store.path_for(digest)
+
+    def concurrent_publish_then_fail(source, destination):
+        target.write_bytes(data)
+        raise PermissionError("target was published by another writer")
+
+    monkeypatch.setattr(
+        objects_module.os,
+        "replace",
+        concurrent_publish_then_fail,
+    )
+
+    assert store.put(data) == digest
+    assert store.verify(digest) is True
+    assert not list(target.parent.glob("*.tmp"))
+
+
+def test_put_retries_transient_verify_permission_error(tmp_path, monkeypatch):
+    store = ObjectStore(tmp_path / "objects")
+    data = b"temporarily locked payload"
+    digest = store.put(data)
+    real_verify = store.verify
+    attempts = 0
+    delays = []
+
+    def locked_once(candidate):
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise PermissionError("simulated Windows sharing violation")
+        return real_verify(candidate)
+
+    monkeypatch.setattr(store, "verify", locked_once)
+    monkeypatch.setattr(objects_module.time, "sleep", delays.append)
+
+    assert store.put(data) == digest
+    assert attempts == 2
+    assert delays == [store._ACCESS_RETRY_DELAYS[0]]
+
+
+def test_concurrent_puts_publish_one_valid_object(tmp_path):
+    store = ObjectStore(tmp_path / "objects")
+    data = b"concurrent payload" * 1024
+    expected = hashlib.sha256(data).hexdigest()
+
+    with ThreadPoolExecutor(max_workers=16) as pool:
+        digests = list(pool.map(lambda _: store.put(data), range(64)))
+
+    assert set(digests) == {expected}
+    assert store.verify(expected) is True
+    assert store.get(expected) == data

From d33e44be58db60295f2c3c8000bfbea8a17c0e52 Mon Sep 17 00:00:00 2001
From: KumamuKuma <157353275+KumamuKuma@users.noreply.github.com>
Date: Mon, 20 Jul 2026 16:54:08 +0800
Subject: [PATCH 2/2] feat(doctor): report object integrity failures

---
 src/rgit/doctor.py   | 108 ++++++++++++++++++++++++++++++---------
 tests/test_doctor.py | 117 +++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 200 insertions(+), 25 deletions(-)

diff --git a/src/rgit/doctor.py b/src/rgit/doctor.py
index 0fa0f6d..ef09f20 100644
--- a/src/rgit/doctor.py
+++ b/src/rgit/doctor.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import json
 from pathlib import Path
 from typing import Any
@@ -145,17 +146,20 @@ def _check_feature_payloads(store, findings: list[dict[str, Any]]) -> None:
                 fid,
             )
             continue
-        try:
-            payload = _load_json_object(store, digest)
-        except FileNotFoundError:
-            _add(
-                findings,
-                "error",
-                "missing_feature_payload_object",
-                "feature payload_hash does not resolve to an object",
-                fid,
-            )
+        payload_ok, payload_bytes = _check_object_reference(
+            store,
+            findings,
+            digest,
+            fid,
+            kind="feature_payload",
+            reference="feature payload_hash",
+            load_bytes=True,
+        )
+        if not payload_ok:
             continue
+        assert payload_bytes is not None
+        try:
+            payload = json.loads(payload_bytes)
         except (UnicodeDecodeError, json.JSONDecodeError):
             _add(
                 findings,
@@ -203,14 +207,14 @@ def _check_run_artifacts(store, findings: list[dict[str, Any]]) -> None:
                 rid,
             )
             continue
-        if not store.objects.path_for(digest).exists():
-            _add(
-                findings,
-                "error",
-                "missing_run_artifact_object",
-                "run artifact_hash does not resolve to an object",
-                rid,
-            )
+        _check_object_reference(
+            store,
+            findings,
+            digest,
+            rid,
+            kind="run_artifact",
+            reference="run artifact_hash",
+        )
 
 
 def _check_proposals(store, findings: list[dict[str, Any]]) -> None:
@@ -225,13 +229,14 @@ def _check_proposals(store, findings: list[dict[str, Any]]) -> None:
                 "proposal has no diff_ref",
                 pid,
             )
-        elif not store.objects.path_for(diff_ref).exists():
-            _add(
+        else:
+            _check_object_reference(
+                store,
                 findings,
-                "error",
-                "missing_proposal_diff_object",
-                "proposal diff_ref does not resolve to an object",
+                diff_ref,
                 pid,
+                kind="proposal_diff",
+                reference="proposal diff_ref",
             )
         try:
             candidates = json.loads(row["candidates"])
@@ -343,8 +348,61 @@ def _expected_endpoint(edge_type: str, side: str) -> str:
     return "known"
 
 
-def _load_json_object(store, digest: str) -> Any:
-    return json.loads(store.objects.get(digest))
+def _check_object_reference(
+    store,
+    findings: list[dict[str, Any]],
+    digest: str,
+    subject: str,
+    *,
+    kind: str,
+    reference: str,
+    load_bytes: bool = False,
+) -> tuple[bool, bytes | None]:
+    """Return validity and, when requested, the exact bytes that were hashed."""
+    if not store.objects.is_valid_digest(digest):
+        _add(
+            findings,
+            "error",
+            f"invalid_{kind}_reference",
+            f"{reference} is not a canonical lowercase sha256 digest",
+            subject,
+        )
+        return False, None
+    try:
+        if load_bytes:
+            data = store.objects.get(digest)
+            matches = hashlib.sha256(data).hexdigest() == digest
+        else:
+            data = None
+            matches = store.objects.verify(digest)
+    except FileNotFoundError:
+        _add(
+            findings,
+            "error",
+            f"missing_{kind}_object",
+            f"{reference} does not resolve to an object",
+            subject,
+        )
+        return False, None
+    except OSError:
+        _add(
+            findings,
+            "error",
+            f"unreadable_{kind}_object",
+            f"{kind.replace('_', ' ')} object cannot be read",
+            subject,
+        )
+        return False, None
+    if not matches:
+        _add(
+            findings,
+            "error",
+            f"corrupt_{kind}_object",
+            f"{kind.replace('_', ' ')} object does not match sha256 {digest}",
+            subject,
+        )
+        return False, None
+    return True, data
 
 
 def _add(
diff --git a/tests/test_doctor.py b/tests/test_doctor.py
index 074cb3e..bb68cef 100644
--- a/tests/test_doctor.py
+++ b/tests/test_doctor.py
@@ -103,6 +103,65 @@ def test_doctor_reports_missing_feature_payload_object(git_repo):
     assert "missing_feature_payload_object" in _codes(report, level="error")
 
 
+def test_doctor_reports_corrupt_feature_payload_object(git_repo):
+    from rgit.doctor import run_doctor
+
+    store = Store.init(git_repo)
+    fid = store.add_feature(_cap())
+    payload_hash = store.conn.execute(
+        "SELECT payload_hash FROM features WHERE id=?", (fid,)
+    ).fetchone()["payload_hash"]
+    _object_path(store, payload_hash).write_bytes(b"[]")
+
+    report = run_doctor(store)
+
+    assert report["ok"] is False
+    assert "corrupt_feature_payload_object" in _codes(report, level="error")
+    assert "malformed_feature_payload_json" not in _codes(report)
+
+
+def test_doctor_hashes_the_same_feature_payload_bytes_it_parses(
+    git_repo, monkeypatch
+):
+    from rgit.doctor import run_doctor
+
+    store = Store.init(git_repo)
+    fid = store.add_feature(_cap())
+    payload_hash = store.conn.execute(
+        "SELECT payload_hash FROM features WHERE id=?", (fid,)
+    ).fetchone()["payload_hash"]
+    real_get = store.objects.get
+
+    def tamper_before_read(digest):
+        _object_path(store, payload_hash).write_bytes(b"[]")
+        return real_get(digest)
+
+    monkeypatch.setattr(store.objects, "get", tamper_before_read)
+
+    report = run_doctor(store)
+
+    assert report["ok"] is False
+    assert "corrupt_feature_payload_object" in _codes(report, level="error")
+    assert "malformed_feature_payload_json" not in _codes(report)
+
+
+def test_doctor_reports_feature_payload_read_race(git_repo, monkeypatch):
+    from rgit.doctor import run_doctor
+
+    store = Store.init(git_repo)
+    store.add_feature(_cap())
+
+    def deny_read(digest):
+        raise PermissionError("simulated read race")
+
+    monkeypatch.setattr(store.objects, "get", deny_read)
+
+    report = run_doctor(store)
+
+    assert report["ok"] is False
+    assert "unreadable_feature_payload_object" in _codes(report, level="error")
+
+
 def test_doctor_reports_missing_run_artifact_object(git_repo):
     from rgit.doctor import run_doctor
 
@@ -116,6 +175,20 @@ def test_doctor_reports_missing_run_artifact_object(git_repo):
     assert "missing_run_artifact_object" in _codes(report, level="error")
 
 
+def test_doctor_reports_corrupt_run_artifact_object(git_repo):
+    from rgit.doctor import run_doctor
+
+    store = Store.init(git_repo)
+    artifact_hash = store.objects.put(b"artifact")
+    _run(store, artifact=artifact_hash)
+    _object_path(store, artifact_hash).write_bytes(b"tampered")
+
+    report = run_doctor(store)
+
+    assert report["ok"] is False
+    assert "corrupt_run_artifact_object" in _codes(report, level="error")
+
+
 def test_doctor_reports_missing_proposal_diff_object(git_repo):
     from rgit.doctor import run_doctor
 
@@ -129,6 +202,50 @@ def test_doctor_reports_missing_proposal_diff_object(git_repo):
     assert "missing_proposal_diff_object" in _codes(report, level="error")
 
 
+def test_doctor_reports_corrupt_proposal_diff_object(git_repo):
+    from rgit.doctor import run_doctor
+
+    store = Store.init(git_repo)
+    diff_ref = store.objects.put(b"diff")
+    _proposal(store, diff=diff_ref)
+    _object_path(store, diff_ref).write_bytes(b"tampered")
+
+    report = run_doctor(store)
+
+    assert report["ok"] is False
+    assert "corrupt_proposal_diff_object" in _codes(report, level="error")
+
+
+def test_doctor_reports_invalid_object_reference_without_path_lookup(git_repo):
+    from rgit.doctor import run_doctor
+
+    store = Store.init(git_repo)
+    _run(store, artifact="../outside")
+
+    report = run_doctor(store)
+
+    assert report["ok"] is False
+    assert "invalid_run_artifact_reference" in _codes(report, level="error")
+
+
+def test_doctor_reports_unreadable_object(git_repo, monkeypatch):
+    from rgit.doctor import run_doctor
+
+    store = Store.init(git_repo)
+    artifact_hash = store.objects.put(b"artifact")
+    _run(store, artifact=artifact_hash)
+
+    def deny_read(digest):
+        raise PermissionError("simulated unreadable object")
+
+    monkeypatch.setattr(store.objects, "verify", deny_read)
+
+    report = run_doctor(store)
+
+    assert report["ok"] is False
+    assert "unreadable_run_artifact_object" in _codes(report, level="error")
+
+
 def test_doctor_reports_malformed_proposal_candidates_json(git_repo):
     from rgit.doctor import run_doctor