From e0615cda47eb72be6e7a8f9c60e17a2ea0c80546 Mon Sep 17 00:00:00 2001 From: RaresKeY <158580472+RaresKeY@users.noreply.github.com> Date: Wed, 12 Aug 2026 03:22:31 +0100 Subject: [PATCH] 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 | 143 ++++++++++++++++++------- tests/test_upload_handler_atomicity.py | 85 +++++++++++++-- 2 files changed, 182 insertions(+), 46 deletions(-) diff --git a/src/upload_handler.py b/src/upload_handler.py index ce0b4b129..e2907699d 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,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]]: diff --git a/tests/test_upload_handler_atomicity.py b/tests/test_upload_handler_atomicity.py index 73cf27917..717bf7f0a 100644 --- a/tests/test_upload_handler_atomicity.py +++ b/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."