fix(upload): harden index cache recovery

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,19 +737,52 @@ 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")
+        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
@@ -751,20 +794,17 @@ class UploadHandler:
             existing_candidates = [path for path in candidates if os.path.exists(path)]
         if not existing_candidates:
             self._index_cache = {}
-            self._index_mtime = 0.0
+            self._index_signature = signature
             return {}
 
         # Check cache validity
-        try:
-            mtime = max(os.path.getmtime(path) for path in existing_candidates)
-            if (
-                not fail_on_error
-                and self._index_cache is not None
-                and mtime <= self._index_mtime
-            ):
-                return self._index_cache
-        except OSError:
-            mtime = 0.0
+        if (
+            not fail_on_error
+            and signature is not None
+            and self._index_cache is not None
+            and signature == self._index_signature
+        ):
+            return self._index_cache
 
         # Try the live file first, fall back to the .bak sibling if the
         # live file is truncated/corrupted.
@@ -774,7 +814,7 @@ class UploadHandler:
                     data = json.load(f)
                 if isinstance(data, dict):
                     self._index_cache = data
-                    self._index_mtime = mtime
+                    self._index_signature = self._upload_index_signature(candidates)
                     return data
             except Exception as e:
                 logger.warning(f"Failed to read uploads database ({candidate}): {e}")
@@ -783,6 +823,7 @@ class UploadHandler:
         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

@@ -59,6 +59,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 +256,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 +266,40 @@ 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
+
+
 # ---------------------------------------------------------------------------
 # Atomicity primitive audit on the production module.
 # ---------------------------------------------------------------------------
@@ -390,10 +431,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."