Preserve SpaceDevs caches on disk write failures

This commit is contained in:
Edward Betts 2026-09-30 16:45:21 +01:00
parent 3f6ee4f2ec
commit 5dce485216
2 changed files with 129 additions and 50 deletions

View file

@ -6,12 +6,13 @@ import re
import time import time
import os import os
import typing import typing
import tempfile
from datetime import datetime from datetime import datetime
import requests import requests
from .types import StrDict from .types import StrDict
from .utils import filename_timestamp, get_most_recent_file from .utils import filename_timestamp
Launch = dict[str, typing.Any] Launch = dict[str, typing.Any]
Summary = dict[str, typing.Any] Summary = dict[str, typing.Any]
@ -22,6 +23,53 @@ LIMIT = 500
ACTIVE_CREWED_FLIGHTS_CACHE_FILE = "active_crewed_flights.json" ACTIVE_CREWED_FLIGHTS_CACHE_FILE = "active_crewed_flights.json"
def write_json_cache(filename: str, payload: StrDict) -> None:
"""Publish a complete cache atomically, preserving old data on write failure."""
contents = json.dumps(payload)
temporary: str | None = None
try:
with tempfile.NamedTemporaryFile(
mode="w", dir=os.path.dirname(filename), prefix=".cache-", delete=False
) as cache:
temporary = cache.name
cache.write(contents)
cache.flush()
os.fsync(cache.fileno())
os.replace(temporary, filename)
finally:
if temporary is not None and os.path.exists(temporary):
os.unlink(temporary)
def read_json_cache(filename: str) -> StrDict | None:
"""Skip empty, unreadable, or damaged cache files."""
try:
with open(filename) as cache:
if os.fstat(cache.fileno()).st_size == 0:
return None
data = json.load(cache)
except (OSError, ValueError):
return None
return data if isinstance(data, dict) else None
def latest_launches_cache(rocket_dir: str) -> tuple[datetime, StrDict] | None:
"""Find the newest usable timestamped launch cache."""
existing = sorted(
(
entry
for f in os.listdir(rocket_dir)
if (entry := filename_timestamp(f, "json")) is not None
),
reverse=True,
)
for updated, filename in existing:
data = read_json_cache(os.path.join(rocket_dir, filename))
if data is not None and isinstance(data.get("results"), list):
return updated, data
return None
class RateLimitDeferred(Exception): class RateLimitDeferred(Exception):
"""The shared API budget or server cooldown requires waiting.""" """The shared API budget or server cooldown requires waiting."""
@ -35,19 +83,14 @@ def api_get(
with open(filename + ".lock", "a") as lock: with open(filename + ".lock", "a") as lock:
fcntl.flock(lock, fcntl.LOCK_EX) fcntl.flock(lock, fcntl.LOCK_EX)
now = time.time() now = time.time()
try: state = read_json_cache(filename) or {}
with open(filename) as cache:
state = json.load(cache)
except (OSError, ValueError):
state = {}
calls = [float(t) for t in state.get("calls", []) if float(t) > now - 3600] calls = [float(t) for t in state.get("calls", []) if float(t) > now - 3600]
retry_at = float(state.get("retry_at", 0)) retry_at = float(state.get("retry_at", 0))
if now < retry_at or len(calls) >= 12: if now < retry_at or len(calls) >= 12:
raise RateLimitDeferred() raise RateLimitDeferred()
calls.append(now) calls.append(now)
# Count attempts even if the connection fails. # Count attempts even if the connection fails.
with open(filename, "w") as cache: write_json_cache(filename, {"calls": calls, "retry_at": retry_at})
json.dump({"calls": calls, "retry_at": retry_at}, cache)
response = requests.get(url, params=params, timeout=30) response = requests.get(url, params=params, timeout=30)
if response.status_code == 429: if response.status_code == 429:
delay = 3600.0 delay = 3600.0
@ -57,8 +100,7 @@ def api_get(
match = re.search(r"available in (\d+) seconds", response.text) match = re.search(r"available in (\d+) seconds", response.text)
if match: if match:
delay = float(match[1]) delay = float(match[1])
with open(filename, "w") as cache: write_json_cache(filename, {"calls": calls, "retry_at": now + delay + 1})
json.dump({"calls": calls, "retry_at": now + delay + 1}, cache)
raise RateLimitDeferred() raise RateLimitDeferred()
return response return response
@ -78,7 +120,7 @@ def next_launch_api_data(rocket_dir: str, limit: int = LIMIT) -> StrDict | None:
# Only persist valid launch payloads; rate-limit / error responses must not # Only persist valid launch payloads; rate-limit / error responses must not
# overwrite the cache or they become the "most recent" file. # overwrite the cache or they become the "most recent" file.
if isinstance(data.get("results"), list): if isinstance(data.get("results"), list):
open(filename, "w").write(r.text) write_json_cache(filename, data)
return data return data
@ -211,12 +253,8 @@ def load_active_crewed_flights_cache(
) -> tuple[datetime, list[Summary]] | None: ) -> tuple[datetime, list[Summary]] | None:
"""Load active crewed flights cache file.""" """Load active crewed flights cache file."""
filename = get_active_crewed_flights_cache_filename(rocket_dir) filename = get_active_crewed_flights_cache_filename(rocket_dir)
if not os.path.exists(filename): cache_data = read_json_cache(filename)
return None if cache_data is None:
try:
cache_data = json.load(open(filename))
except (json.JSONDecodeError, OSError):
return None return None
updated_str = cache_data.get("updated") updated_str = cache_data.get("updated")
@ -236,8 +274,7 @@ def write_active_crewed_flights_cache(rocket_dir: str, launches: list[Summary])
"updated": datetime.now().isoformat(), "updated": datetime.now().isoformat(),
"results": launches, "results": launches,
} }
with open(filename, "w") as f: write_json_cache(filename, payload)
json.dump(payload, f)
def get_active_crewed_flights( def get_active_crewed_flights(
@ -371,20 +408,14 @@ def summarize_launch(launch: Launch) -> Summary:
def is_launches_cache_fresh(rocket_dir: str) -> bool: def is_launches_cache_fresh(rocket_dir: str) -> bool:
"""Return True if the launches cache is younger than the TTL.""" """Return True if the launches cache is younger than the TTL."""
now = datetime.now() cached = latest_launches_cache(rocket_dir)
existing = [ return cached is not None and (datetime.now() - cached[0]).total_seconds() <= ttl
x for x in (filename_timestamp(f, "json") for f in os.listdir(rocket_dir)) if x
]
if not existing:
return False
existing.sort(reverse=True)
return (now - existing[0][0]).total_seconds() <= ttl
def load_cached_launches(rocket_dir: str) -> StrDict | None: def load_cached_launches(rocket_dir: str) -> StrDict | None:
"""Read the most recent cache of launches.""" """Read the most recent usable cache of launches."""
filename = get_most_recent_file(rocket_dir, "json") cached = latest_launches_cache(rocket_dir)
return typing.cast(StrDict, json.load(open(filename))) if filename else None return cached[1] if cached is not None else None
def read_cached_launches(rocket_dir: str) -> list[Summary]: def read_cached_launches(rocket_dir: str) -> list[Summary]:
@ -399,14 +430,7 @@ def get_launches(
rocket_dir: str, limit: int = LIMIT, refresh: bool = False rocket_dir: str, limit: int = LIMIT, refresh: bool = False
) -> list[Summary] | None: ) -> list[Summary] | None:
"""Get rocket launches with caching.""" """Get rocket launches with caching."""
now = datetime.now() if refresh or not is_launches_cache_fresh(rocket_dir):
existing = [
x for x in (filename_timestamp(f, "json") for f in os.listdir(rocket_dir)) if x
]
existing.sort(reverse=True)
if refresh or not existing or (now - existing[0][0]).total_seconds() > ttl:
try: try:
upcoming = next_launch_api(rocket_dir, limit=limit) upcoming = next_launch_api(rocket_dir, limit=limit)
if upcoming is None: if upcoming is None:
@ -425,18 +449,7 @@ def get_launches(
except Exception: except Exception:
pass # fallback to cached version pass # fallback to cached version
# Find the most recent cache file that contains a valid results list. data = load_cached_launches(rocket_dir)
# Older files without "results" (e.g. stale rate-limit responses) are skipped.
data = None
for _, f in existing:
filename = os.path.join(rocket_dir, f)
try:
candidate = json.load(open(filename))
except (json.JSONDecodeError, OSError):
continue
if isinstance(candidate.get("results"), list):
data = candidate
break
if not data: if not data:
return [] return []
upcoming = [summarize_launch(launch) for launch in data["results"]] upcoming = [summarize_launch(launch) for launch in data["results"]]

View file

@ -0,0 +1,66 @@
"""SpaceDevs caches survive empty files and failed disk writes."""
import errno
import json
from datetime import datetime, timedelta
from pathlib import Path
from unittest.mock import Mock
import pytest
from agenda import thespacedevs
@pytest.mark.parametrize("contents", ["", "{", "[]", '{"detail": "throttled"}'])
def test_launch_cache_falls_back_to_usable_file(tmp_path: Path, contents: str) -> None:
old = datetime.now() - timedelta(days=2)
(tmp_path / old.strftime("%Y-%m-%d_%H:%M:%S.json")).write_text('{"results": []}')
(tmp_path / datetime.now().strftime("%Y-%m-%d_%H:%M:%S.json")).write_text(contents)
assert thespacedevs.load_cached_launches(str(tmp_path)) == {"results": []}
assert thespacedevs.read_cached_launches(str(tmp_path)) == []
assert not thespacedevs.is_launches_cache_fresh(str(tmp_path))
def test_empty_file_is_not_parsed(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
filename = tmp_path / "empty.json"
filename.touch()
parser = Mock(side_effect=AssertionError("empty file must not be parsed"))
monkeypatch.setattr("agenda.thespacedevs.json.load", parser)
assert thespacedevs.read_json_cache(str(filename)) is None
parser.assert_not_called()
@pytest.mark.parametrize("existing", [False, True])
@pytest.mark.parametrize("failure", ["fsync", "replace"])
def test_disk_failure_preserves_cache(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, existing: bool, failure: str
) -> None:
filename = tmp_path / "cache.json"
original = '{"results": ["old"]}'
if existing:
filename.write_text(original)
fail = Mock(side_effect=OSError(errno.ENOSPC, "No space left on device"))
monkeypatch.setattr(f"agenda.thespacedevs.os.{failure}", fail)
with pytest.raises(OSError, match="No space left"):
thespacedevs.write_json_cache(str(filename), {"results": ["new"]})
if existing:
assert filename.read_text() == original
else:
assert not filename.exists()
assert not list(tmp_path.glob(".cache-*"))
def test_successful_write_replaces_empty_file(tmp_path: Path) -> None:
filename = tmp_path / "cache.json"
filename.touch()
thespacedevs.write_json_cache(str(filename), {"results": []})
assert json.loads(filename.read_text()) == {"results": []}
assert not list(tmp_path.glob(".cache-*"))
def test_all_empty_launch_files_return_no_cache(tmp_path: Path) -> None:
(tmp_path / "2026-09-30_12:00:00.json").touch()
assert thespacedevs.load_cached_launches(str(tmp_path)) is None
assert not thespacedevs.is_launches_cache_fresh(str(tmp_path))