From 2c703bef5f1a0cfee8334384caa39fec546db22d Mon Sep 17 00:00:00 2001 From: RaresKeY <158580472+RaresKeY@users.noreply.github.com> Date: Fri, 31 Jul 2026 19:09:50 +0000 Subject: [PATCH] fix(upload): harden index cache recovery --- src/upload_handler.py | 79 +++++++++++++++++++------- tests/test_upload_handler_atomicity.py | 55 +++++++++++++++--- 2 files changed, 107 insertions(+), 27 deletions(-) diff --git a/src/upload_handler.py b/src/upload_handler.py index ce0b4b129..56be5750e 100644 --- a/src/upload_handler.py +++ b/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]]: diff --git a/tests/test_upload_handler_atomicity.py b/tests/test_upload_handler_atomicity.py index 73cf27917..94d9ea09f 100644 --- a/tests/test_upload_handler_atomicity.py +++ b/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."