mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-08-05 02:45:28 +00:00
Merge 2c703bef5f into 20e7fc0164
This commit is contained in:
commit
b215234614
2 changed files with 107 additions and 27 deletions
|
|
@ -35,6 +35,16 @@ import logging
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
UploadIndexFileSignature = tuple[
|
||||||
|
str,
|
||||||
|
Optional[int],
|
||||||
|
Optional[int],
|
||||||
|
Optional[int],
|
||||||
|
Optional[int],
|
||||||
|
Optional[int],
|
||||||
|
]
|
||||||
|
UploadIndexSignature = tuple[UploadIndexFileSignature, ...]
|
||||||
|
|
||||||
|
|
||||||
class UploadCleanupSafetyError(RuntimeError):
|
class UploadCleanupSafetyError(RuntimeError):
|
||||||
"""Raised when cleanup cannot prove that destructive work is safe."""
|
"""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
|
# In-memory index cache to avoid O(N) disk I/O on every request
|
||||||
self._index_cache: Optional[Dict[str, Any]] = None
|
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:
|
def inside_base_dir(self, path: str) -> bool:
|
||||||
"""Check if path is inside base directory"""
|
"""Check if path is inside base directory"""
|
||||||
|
|
@ -727,19 +737,52 @@ class UploadHandler:
|
||||||
# Update cache if this is the main index
|
# Update cache if this is the main index
|
||||||
if path.endswith("uploads.json"):
|
if path.endswith("uploads.json"):
|
||||||
self._index_cache = data
|
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:
|
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:
|
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]:
|
def _load_upload_index(self, *, fail_on_error: bool = False) -> Dict[str, Any]:
|
||||||
"""Load the upload index from disk/cache. Uses mtime-based validation
|
"""Load the upload index from disk/cache. Uses file-identity validation
|
||||||
to avoid redundant parsing on hot paths. When ``fail_on_error`` is
|
to avoid redundant parsing on hot paths without missing same-timestamp
|
||||||
true, a missing, malformed, or unreadable live index raises so
|
mutations. When ``fail_on_error`` is true, a missing, malformed, or
|
||||||
destructive callers cannot mistake corruption for an empty store.
|
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")
|
uploads_db_path = os.path.join(self.upload_dir, "uploads.json")
|
||||||
candidates = (uploads_db_path, uploads_db_path + ".bak")
|
candidates = (uploads_db_path, uploads_db_path + ".bak")
|
||||||
|
signature = self._upload_index_signature(candidates)
|
||||||
if fail_on_error:
|
if fail_on_error:
|
||||||
# A backup is intentionally the previous snapshot. It is useful for
|
# A backup is intentionally the previous snapshot. It is useful for
|
||||||
# non-destructive reads, but cannot authorize deletion when the live
|
# 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)]
|
existing_candidates = [path for path in candidates if os.path.exists(path)]
|
||||||
if not existing_candidates:
|
if not existing_candidates:
|
||||||
self._index_cache = {}
|
self._index_cache = {}
|
||||||
self._index_mtime = 0.0
|
self._index_signature = signature
|
||||||
return {}
|
return {}
|
||||||
|
|
||||||
# Check cache validity
|
# Check cache validity
|
||||||
try:
|
if (
|
||||||
mtime = max(os.path.getmtime(path) for path in existing_candidates)
|
not fail_on_error
|
||||||
if (
|
and signature is not None
|
||||||
not fail_on_error
|
and self._index_cache is not None
|
||||||
and self._index_cache is not None
|
and signature == self._index_signature
|
||||||
and mtime <= self._index_mtime
|
):
|
||||||
):
|
return self._index_cache
|
||||||
return self._index_cache
|
|
||||||
except OSError:
|
|
||||||
mtime = 0.0
|
|
||||||
|
|
||||||
# Try the live file first, fall back to the .bak sibling if the
|
# Try the live file first, fall back to the .bak sibling if the
|
||||||
# live file is truncated/corrupted.
|
# live file is truncated/corrupted.
|
||||||
|
|
@ -774,7 +814,7 @@ class UploadHandler:
|
||||||
data = json.load(f)
|
data = json.load(f)
|
||||||
if isinstance(data, dict):
|
if isinstance(data, dict):
|
||||||
self._index_cache = data
|
self._index_cache = data
|
||||||
self._index_mtime = mtime
|
self._index_signature = self._upload_index_signature(candidates)
|
||||||
return data
|
return data
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"Failed to read uploads database ({candidate}): {e}")
|
logger.warning(f"Failed to read uploads database ({candidate}): {e}")
|
||||||
|
|
@ -783,6 +823,7 @@ class UploadHandler:
|
||||||
if fail_on_error:
|
if fail_on_error:
|
||||||
raise ValueError("live uploads database is unreadable")
|
raise ValueError("live uploads database is unreadable")
|
||||||
self._index_cache = {}
|
self._index_cache = {}
|
||||||
|
self._index_signature = self._upload_index_signature(candidates)
|
||||||
return {}
|
return {}
|
||||||
|
|
||||||
def get_upload_info(self, upload_id: str) -> Optional[Dict[str, Any]]:
|
def get_upload_info(self, upload_id: str) -> Optional[Dict[str, Any]]:
|
||||||
|
|
|
||||||
|
|
@ -59,6 +59,16 @@ def _db_path(handler: UploadHandler) -> str:
|
||||||
return os.path.join(handler.upload_dir, "uploads.json")
|
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:
|
def _seed_entry(owner: str, file_hash: str, file_id: str) -> dict:
|
||||||
return {
|
return {
|
||||||
"id": file_id,
|
"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."
|
"Production _atomic_write_json must create a .bak sibling on subsequent writes."
|
||||||
)
|
)
|
||||||
|
|
||||||
full = open(db_path, "rb").read()
|
_truncate_without_newer_mtime(db_path)
|
||||||
truncated_len = max(1, len(full) // 2)
|
|
||||||
with open(db_path, "wb") as f:
|
|
||||||
f.write(full[:truncated_len])
|
|
||||||
|
|
||||||
recovered = handler._load_upload_index()
|
recovered = handler._load_upload_index()
|
||||||
missing = [k for k in original if k not in recovered]
|
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.
|
# 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})
|
handler._atomic_write_json(db_path, {"sentinel": True})
|
||||||
assert os.path.exists(db_path + ".bak")
|
assert os.path.exists(db_path + ".bak")
|
||||||
|
|
||||||
# Truncate the live file.
|
# Truncate the live file without assuming the filesystem advances mtime.
|
||||||
full = open(db_path, "rb").read()
|
_truncate_without_newer_mtime(db_path)
|
||||||
with open(db_path, "wb") as f:
|
|
||||||
f.write(full[: max(1, len(full) // 2)])
|
|
||||||
|
|
||||||
info = handler.get_upload_info(first["id"])
|
info = handler.get_upload_info(first["id"])
|
||||||
assert info is not None, "Info lookup must succeed after .bak recovery."
|
assert info is not None, "Info lookup must succeed after .bak recovery."
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue