From 5dce485216e43fb1ce641873e8685d04a6c7959c Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Wed, 30 Sep 2026 16:45:21 +0100 Subject: [PATCH] Preserve SpaceDevs caches on disk write failures --- agenda/thespacedevs.py | 113 +++++++++++++++++++--------------- tests/test_spacedevs_cache.py | 66 ++++++++++++++++++++ 2 files changed, 129 insertions(+), 50 deletions(-) create mode 100644 tests/test_spacedevs_cache.py diff --git a/agenda/thespacedevs.py b/agenda/thespacedevs.py index 5b9e6db..0066b73 100644 --- a/agenda/thespacedevs.py +++ b/agenda/thespacedevs.py @@ -6,12 +6,13 @@ import re import time import os import typing +import tempfile from datetime import datetime import requests from .types import StrDict -from .utils import filename_timestamp, get_most_recent_file +from .utils import filename_timestamp Launch = dict[str, typing.Any] Summary = dict[str, typing.Any] @@ -22,6 +23,53 @@ LIMIT = 500 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): """The shared API budget or server cooldown requires waiting.""" @@ -35,19 +83,14 @@ def api_get( with open(filename + ".lock", "a") as lock: fcntl.flock(lock, fcntl.LOCK_EX) now = time.time() - try: - with open(filename) as cache: - state = json.load(cache) - except (OSError, ValueError): - state = {} + state = read_json_cache(filename) or {} calls = [float(t) for t in state.get("calls", []) if float(t) > now - 3600] retry_at = float(state.get("retry_at", 0)) if now < retry_at or len(calls) >= 12: raise RateLimitDeferred() calls.append(now) # Count attempts even if the connection fails. - with open(filename, "w") as cache: - json.dump({"calls": calls, "retry_at": retry_at}, cache) + write_json_cache(filename, {"calls": calls, "retry_at": retry_at}) response = requests.get(url, params=params, timeout=30) if response.status_code == 429: delay = 3600.0 @@ -57,8 +100,7 @@ def api_get( match = re.search(r"available in (\d+) seconds", response.text) if match: delay = float(match[1]) - with open(filename, "w") as cache: - json.dump({"calls": calls, "retry_at": now + delay + 1}, cache) + write_json_cache(filename, {"calls": calls, "retry_at": now + delay + 1}) raise RateLimitDeferred() 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 # overwrite the cache or they become the "most recent" file. if isinstance(data.get("results"), list): - open(filename, "w").write(r.text) + write_json_cache(filename, data) return data @@ -211,12 +253,8 @@ def load_active_crewed_flights_cache( ) -> tuple[datetime, list[Summary]] | None: """Load active crewed flights cache file.""" filename = get_active_crewed_flights_cache_filename(rocket_dir) - if not os.path.exists(filename): - return None - - try: - cache_data = json.load(open(filename)) - except (json.JSONDecodeError, OSError): + cache_data = read_json_cache(filename) + if cache_data is None: return None 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(), "results": launches, } - with open(filename, "w") as f: - json.dump(payload, f) + write_json_cache(filename, payload) def get_active_crewed_flights( @@ -371,20 +408,14 @@ def summarize_launch(launch: Launch) -> Summary: def is_launches_cache_fresh(rocket_dir: str) -> bool: """Return True if the launches cache is younger than the TTL.""" - now = datetime.now() - existing = [ - 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 + cached = latest_launches_cache(rocket_dir) + return cached is not None and (datetime.now() - cached[0]).total_seconds() <= ttl def load_cached_launches(rocket_dir: str) -> StrDict | None: - """Read the most recent cache of launches.""" - filename = get_most_recent_file(rocket_dir, "json") - return typing.cast(StrDict, json.load(open(filename))) if filename else None + """Read the most recent usable cache of launches.""" + cached = latest_launches_cache(rocket_dir) + return cached[1] if cached is not None else None 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 ) -> list[Summary] | None: """Get rocket launches with caching.""" - now = datetime.now() - 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: + if refresh or not is_launches_cache_fresh(rocket_dir): try: upcoming = next_launch_api(rocket_dir, limit=limit) if upcoming is None: @@ -425,18 +449,7 @@ def get_launches( except Exception: pass # fallback to cached version - # Find the most recent cache file that contains a valid results list. - # 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 + data = load_cached_launches(rocket_dir) if not data: return [] upcoming = [summarize_launch(launch) for launch in data["results"]] diff --git a/tests/test_spacedevs_cache.py b/tests/test_spacedevs_cache.py new file mode 100644 index 0000000..0e9ec0d --- /dev/null +++ b/tests/test_spacedevs_cache.py @@ -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))