From 6d408a8bdbc46f8bc59e318376beac61048dc843 Mon Sep 17 00:00:00 2001 From: jing Date: Mon, 13 Jul 2026 16:45:32 +1000 Subject: [PATCH] =?UTF-8?q?=F0=9F=A4=9D=20world=5Fplanes:=20read=20the=20s?= =?UTF-8?q?hared=20OpenSky=20cache=20(quota-collision=20fix)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reader side of the OpenSky shared-cache v1 contract (mirrored in GODSIGH SPEC2.md §4). Path ~/.cache/godverse/opensky-states.json (env OPENSKY_CACHE_FILE), content the raw global /states/all body, freshness = mtime within 120 s. Each poll cycle now checks the cache first: on a fresh hit it reads the global body, filters the states to our bbox client-side (indices 5=lon, 6=lat), and skips the HTTP call entirely (logs "cache hit"). On 429 or a network failure it falls back to a stale cache rather than nothing. This worker only ever polls a bbox, so it never WRITES the cache (never partial data into a global-contract file) — it is a pure reader, and degrades to its old ≥450 s bbox fetch when no writer is around (i.e. prod). Result: godstrument + GODSIGH running together spend ~one poller's quota instead of colliding and 429-ing each other out. Verified: unit tests for read_cache freshness/stale/missing + bbox filtering, and an integration run where a fresh cache is served (count filtered to bbox, zero HTTP). Co-Authored-By: Claude Opus 4.8 --- workers/world_planes.py | 101 +++++++++++++++++++++++++++++++++------- 1 file changed, 85 insertions(+), 16 deletions(-) diff --git a/workers/world_planes.py b/workers/world_planes.py index 66a1918..b7825a6 100644 --- a/workers/world_planes.py +++ b/workers/world_planes.py @@ -5,6 +5,7 @@ and mean groundspeed (intensity) — the sky's traffic becomes a slow chord. """ import argparse import json +import os import sys import time import urllib.error @@ -23,6 +24,57 @@ POLL_OK = 450.0 # normal poll interval (s) POLL_429 = 600.0 # backoff poll interval after HTTP 429 REEMIT = 10.0 # re-emit last values every 10s +# --- OpenSky shared cache v1 (reader side) -------------------------------- +# Path: ~/.cache/godverse/opensky-states.json, override via OPENSKY_CACHE_FILE. +# Content: the raw, unmodified OpenSky /states/all body (global, no bbox), written +# atomically by WHOEVER fetched it (GODSIGH's dev proxy is the writer). Freshness = +# file mtime. We treat the cache as fresh within 120 s; on a fresh hit we skip our +# own HTTP entirely and just filter the global states to our bbox, so a godstrument +# + GODSIGH running together spend ~one poller's quota instead of two. This worker +# only ever polls a bbox, so it never WRITES the cache (never partial data into a +# global-contract file) — it is a pure reader, and degrades to its old bbox fetch +# when no writer is around (i.e. prod). +CACHE_FILE = os.environ.get("OPENSKY_CACHE_FILE") or \ + os.path.expanduser("~/.cache/godverse/opensky-states.json") +CACHE_FRESH = 120.0 # a cache within this many seconds of its mtime is "fresh" + + +def read_cache(max_age): + """Return (data, age_seconds) from the shared cache, or None. If max_age is not + None, returns None when the cache is older than that (i.e. not fresh).""" + try: + mtime = os.path.getmtime(CACHE_FILE) + except OSError: + return None + age = time.time() - mtime + if max_age is not None and age > max_age: + return None + try: + with open(CACHE_FILE, "r", encoding="utf-8") as f: + return json.load(f), age + except (OSError, json.JSONDecodeError, ValueError): + return None + + +def filter_states_bbox(data, bbox): + """Filter a global states payload to our bbox client-side (the cache is global). + OpenSky state vectors: index 5 = longitude, index 6 = latitude.""" + s, w, n, e = bbox + states = (data or {}).get("states") or [] + out = [] + for row in states: + if not row or len(row) < 7: + continue + lon, lat = row[5], row[6] + if lon is None or lat is None: + continue + try: + if s <= float(lat) <= n and w <= float(lon) <= e: + out.append(row) + except (TypeError, ValueError): + continue + return {"states": out} + def fetch(bbox, timeout=25): """Fetch OpenSky states for the bbox. Returns parsed JSON dict. @@ -95,25 +147,42 @@ def main(): while True: now = time.monotonic() if now >= next_poll: - try: - data = fetch(args.bbox) - last = reduce_states(data) + cached = read_cache(CACHE_FRESH) # a fresh global cache -> skip HTTP entirely + if cached is not None: + data, age = cached + last = reduce_states(filter_states_bbox(data, args.bbox)) if poll != POLL_OK: print(f"[{NAME}] recovered; back to {POLL_OK:.0f}s poll") poll = POLL_OK - except urllib.error.HTTPError as ex: - if ex.code == 429: - retry_after = ex.headers.get("X-Rate-Limit-Retry-After-Seconds") - try: - poll = max(POLL_429, float(retry_after)) - except (TypeError, ValueError): - poll = POLL_429 - print(f"[{NAME}] HTTP 429 rate limited; backing off to {poll:.0f}s") - else: - print(f"[{NAME}] HTTP {ex.code}; retrying in {poll:.0f}s") - except (urllib.error.URLError, TimeoutError, OSError, - json.JSONDecodeError, ValueError) as ex: - print(f"[{NAME}] fetch error: {ex}; retrying in {poll:.0f}s") + print(f"[{NAME}] cache hit ({age:.0f}s old) -> {int(last[0])} aircraft in bbox") + else: + try: + data = fetch(args.bbox) + last = reduce_states(data) + if poll != POLL_OK: + print(f"[{NAME}] recovered; back to {POLL_OK:.0f}s poll") + poll = POLL_OK + except urllib.error.HTTPError as ex: + if ex.code == 429: + retry_after = ex.headers.get("X-Rate-Limit-Retry-After-Seconds") + try: + poll = max(POLL_429, float(retry_after)) + except (TypeError, ValueError): + poll = POLL_429 + print(f"[{NAME}] HTTP 429 rate limited; backing off to {poll:.0f}s") + else: + print(f"[{NAME}] HTTP {ex.code}; retrying in {poll:.0f}s") + _stale = read_cache(None) # 429 -> a stale cache beats nothing + if _stale is not None: + last = reduce_states(filter_states_bbox(_stale[0], args.bbox)) + print(f"[{NAME}] serving stale cache ({_stale[1]:.0f}s old)") + except (urllib.error.URLError, TimeoutError, OSError, + json.JSONDecodeError, ValueError) as ex: + print(f"[{NAME}] fetch error: {ex}; retrying in {poll:.0f}s") + _stale = read_cache(None) # network down -> fall back to a stale cache + if _stale is not None: + last = reduce_states(filter_states_bbox(_stale[0], args.bbox)) + print(f"[{NAME}] serving stale cache ({_stale[1]:.0f}s old)") next_poll = time.monotonic() + poll # Emit (or re-emit) the last known values.