fix(upload): recover backups after same-timestamp corruption (#5860)

* fix(upload): harden index cache recovery

* fix: retry upload index loads across replacement

---------

Co-authored-by: Alexandre Teixeira <111787685+alteixeira20@users.noreply.github.com>

src/upload_handler.py

@@ -35,6 +35,16 @@ import logging
 
 logger = logging.getLogger(__name__)
 
+UploadIndexFileSignature = tuple[
+    str,
+    Optional[int],
+    Optional[int],
+    Optional[int],
+    Optional[int],
+    Optional[int],
+]
+UploadIndexSignature = tuple[UploadIndexFileSignature, ...]
+
 
 class UploadCleanupSafetyError(RuntimeError):
     """Raised when cleanup cannot prove that destructive work is safe."""
@@ -242,7 +252,7 @@ class UploadHandler:
 
         # In-memory index cache to avoid O(N) disk I/O on every request
         self._index_cache: Optional[Dict[str, Any]] = None
-        self._index_mtime: float = 0.0
+        self._index_signature: Optional[UploadIndexSignature] = None
     
     def inside_base_dir(self, path: str) -> bool:
         """Check if path is inside base directory"""
@@ -727,62 +737,119 @@ class UploadHandler:
         # Update cache if this is the main index
         if path.endswith("uploads.json"):
             self._index_cache = data
+            self._index_signature = self._upload_index_signature(
+                (path, path + ".bak")
+            )
+
+    @staticmethod
+    def _upload_index_signature(
+        paths: tuple[str, ...],
+    ) -> Optional[UploadIndexSignature]:
+        """Return file identities strong enough to validate the index cache.
+
+        Modification time alone is insufficient: a torn write can change a
+        file without receiving a strictly newer timestamp on some filesystems.
+        Size, inode, and nanosecond change times make those mutations visible
+        while preserving the cache fast path for unchanged files.
+        """
+        signature: list[UploadIndexFileSignature] = []
+        for candidate in paths:
             try:
-                self._index_mtime = os.path.getmtime(path)
+                stat_result = os.stat(candidate)
+            except FileNotFoundError:
+                signature.append((candidate, None, None, None, None, None))
+                continue
             except OSError:
-                self._index_mtime = time.time()
+                return None
+            signature.append(
+                (
+                    candidate,
+                    stat_result.st_dev,
+                    stat_result.st_ino,
+                    stat_result.st_size,
+                    stat_result.st_mtime_ns,
+                    stat_result.st_ctime_ns,
+                )
+            )
+        return tuple(signature)
 
     def _load_upload_index(self, *, fail_on_error: bool = False) -> Dict[str, Any]:
-        """Load the upload index from disk/cache. Uses mtime-based validation
-        to avoid redundant parsing on hot paths. When ``fail_on_error`` is
-        true, a missing, malformed, or unreadable live index raises so
-        destructive callers cannot mistake corruption for an empty store.
+        """Load the upload index from disk/cache. Uses file-identity validation
+        to avoid redundant parsing on hot paths without missing same-timestamp
+        mutations. When ``fail_on_error`` is true, a missing, malformed, or
+        unreadable live index raises so destructive callers cannot mistake
+        corruption for an empty store.
         """
         uploads_db_path = os.path.join(self.upload_dir, "uploads.json")
         candidates = (uploads_db_path, uploads_db_path + ".bak")
-        if fail_on_error:
-            # A backup is intentionally the previous snapshot. It is useful for
-            # non-destructive reads, but cannot authorize deletion when the live
-            # index is missing or corrupt.
-            if not os.path.exists(uploads_db_path):
-                raise ValueError("live uploads database is missing")
-            existing_candidates = [uploads_db_path]
-        else:
-            existing_candidates = [path for path in candidates if os.path.exists(path)]
-        if not existing_candidates:
-            self._index_cache = {}
-            self._index_mtime = 0.0
-            return {}
+        for _attempt in range(3):
+            signature = self._upload_index_signature(candidates)
+            if fail_on_error:
+                # A backup is intentionally the previous snapshot. It is useful for
+                # non-destructive reads, but cannot authorize deletion when the live
+                # index is missing or corrupt.
+                if not os.path.exists(uploads_db_path):
+                    raise ValueError("live uploads database is missing")
+                existing_candidates = [uploads_db_path]
+            else:
+                existing_candidates = [
+                    path for path in candidates if os.path.exists(path)
+                ]
+            if not existing_candidates:
+                self._index_cache = {}
+                self._index_signature = signature
+                return {}
 
-        # Check cache validity
-        try:
-            mtime = max(os.path.getmtime(path) for path in existing_candidates)
+            # Check cache validity
             if (
                 not fail_on_error
+                and signature is not None
                 and self._index_cache is not None
-                and mtime <= self._index_mtime
+                and signature == self._index_signature
             ):
                 return self._index_cache
-        except OSError:
-            mtime = 0.0
 
-        # Try the live file first, fall back to the .bak sibling if the
-        # live file is truncated/corrupted.
-        for candidate in existing_candidates:
-            try:
-                with open(candidate, "r", encoding="utf-8") as f:
-                    data = json.load(f)
-                if isinstance(data, dict):
-                    self._index_cache = data
-                    self._index_mtime = mtime
-                    return data
-            except Exception as e:
-                logger.warning(f"Failed to read uploads database ({candidate}): {e}")
+            # Try the live file first, fall back to the .bak sibling if the
+            # live file is truncated/corrupted. A candidate parsed from an old
+            # inode is accepted only when the whole index signature stays
+            # stable through the read; otherwise retry so the cache cannot pair
+            # stale data with a fresh replacement signature.
+            index_changed_during_read = False
+            for candidate in existing_candidates:
+                try:
+                    with open(candidate, "r", encoding="utf-8") as f:
+                        data = json.load(f)
+                    verified_signature = self._upload_index_signature(candidates)
+                    if (
+                        signature is not None
+                        and verified_signature is not None
+                        and verified_signature != signature
+                    ):
+                        index_changed_during_read = True
+                        break
+                    if isinstance(data, dict):
+                        self._index_cache = data
+                        self._index_signature = verified_signature
+                        return data
+                except Exception as e:
+                    logger.warning(f"Failed to read uploads database ({candidate}): {e}")
+                    verified_signature = self._upload_index_signature(candidates)
+                    if (
+                        signature is not None
+                        and verified_signature is not None
+                        and verified_signature != signature
+                    ):
+                        index_changed_during_read = True
+                        break
+                    continue
+            if index_changed_during_read:
                 continue
+            break
 
         if fail_on_error:
             raise ValueError("live uploads database is unreadable")
         self._index_cache = {}
+        self._index_signature = self._upload_index_signature(candidates)
         return {}
 
     def get_upload_info(self, upload_id: str) -> Optional[Dict[str, Any]]:

tests/test_upload_handler_atomicity.py

@@ -15,6 +15,7 @@ These tests exercise:
 * Smoke tests: normal upload, duplicate detection, info lookup after
   a backup-recovery scenario.
 """
+import builtins
 import concurrent.futures
 import io
 import json
@@ -59,6 +60,16 @@ def _db_path(handler: UploadHandler) -> str:
     return os.path.join(handler.upload_dir, "uploads.json")
 
 
+def _truncate_without_newer_mtime(path: str) -> None:
+    """Model a filesystem where a torn write shares the cached timestamp."""
+    before = os.stat(path)
+    with open(path, "rb") as f:
+        full = f.read()
+    with open(path, "wb") as f:
+        f.write(full[: max(1, len(full) // 2)])
+    os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns))
+
+
 def _seed_entry(owner: str, file_hash: str, file_id: str) -> dict:
     return {
         "id": file_id,
@@ -246,10 +257,7 @@ def test_partial_write_recovery_via_bak(tmp_path):
         "Production _atomic_write_json must create a .bak sibling on subsequent writes."
     )
 
-    full = open(db_path, "rb").read()
-    truncated_len = max(1, len(full) // 2)
-    with open(db_path, "wb") as f:
-        f.write(full[:truncated_len])
+    _truncate_without_newer_mtime(db_path)
 
     recovered = handler._load_upload_index()
     missing = [k for k in original if k not in recovered]
@@ -259,6 +267,69 @@ def test_partial_write_recovery_via_bak(tmp_path):
     )
 
 
+def test_partial_write_recovery_via_bak_after_restart(tmp_path):
+    """A fresh handler must recover the previous snapshot from ``.bak``."""
+    handler = _make_handler(tmp_path)
+    db_path = _db_path(handler)
+    original = {
+        f"owner:hash_{i}": _seed_entry("owner", f"hash_{i}", f"id_{i}")
+        for i in range(3)
+    }
+    handler._atomic_write_json(db_path, original)
+    handler._atomic_write_json(db_path, {"latest": True})
+    _truncate_without_newer_mtime(db_path)
+
+    restarted_handler = UploadHandler(
+        base_dir=handler.base_dir,
+        upload_dir=handler.upload_dir,
+    )
+
+    assert restarted_handler._load_upload_index() == original
+
+
+def test_unchanged_upload_index_uses_cache(tmp_path, monkeypatch):
+    """The stronger file signature must preserve the unchanged-index fast path."""
+    handler = _make_handler(tmp_path)
+    original = {"owner:hash": _seed_entry("owner", "hash", "id")}
+    handler._atomic_write_json(_db_path(handler), original)
+
+    def fail_if_parsed(_file):
+        raise AssertionError("unchanged upload index should be served from cache")
+
+    monkeypatch.setattr(json, "load", fail_if_parsed)
+
+    assert handler._load_upload_index() == original
+
+
+def test_upload_index_retries_when_replaced_during_read(tmp_path, monkeypatch):
+    """Do not cache old JSON under the signature of a newer atomic replace."""
+    handler = _make_handler(tmp_path)
+    db_path = _db_path(handler)
+    old_index = {"owner:old": _seed_entry("owner", "old", "old_id")}
+    new_index = {"owner:new": _seed_entry("owner", "new", "new_id")}
+    handler._atomic_write_json(db_path, old_index)
+    handler._index_cache = None
+    handler._index_signature = None
+
+    real_open = builtins.open
+    replaced = False
+
+    def racing_open(file, mode="r", *args, **kwargs):
+        nonlocal replaced
+        handle = real_open(file, mode, *args, **kwargs)
+        if os.fspath(file) == db_path and "r" in mode and not replaced:
+            replaced = True
+            replacement = db_path + ".replacement"
+            with real_open(replacement, "w", encoding="utf-8") as out:
+                json.dump(new_index, out)
+            os.replace(replacement, db_path)
+        return handle
+
+    monkeypatch.setattr(builtins, "open", racing_open)
+
+    assert handler._load_upload_index() == new_index
+
+
 # ---------------------------------------------------------------------------
 # Atomicity primitive audit on the production module.
 # ---------------------------------------------------------------------------
@@ -390,10 +461,8 @@ def test_smoke_info_lookup_after_bak_recovery(tmp_path):
     handler._atomic_write_json(db_path, {"sentinel": True})
     assert os.path.exists(db_path + ".bak")
 
-    # Truncate the live file.
-    full = open(db_path, "rb").read()
-    with open(db_path, "wb") as f:
-        f.write(full[: max(1, len(full) // 2)])
+    # Truncate the live file without assuming the filesystem advances mtime.
+    _truncate_without_newer_mtime(db_path)
 
     info = handler.get_upload_info(first["id"])
     assert info is not None, "Info lookup must succeed after .bak recovery."