Persist Global Spot forecast dedupe state

Author: Sam-Bolling

docs/research/new-publisher-source-planning/Met_Office_Global_Spot_Forecast_Publisher_Status_2026-05-26.md

@@ -95,6 +95,7 @@ Oracle live validation completed after host-local key installation:
 - A live Global Spot probe against `/sitespecific/v0/point/hourly` succeeded for London Heathrow Area with 49 candidate forecast records and recognized forecast temperature, humidity, wind speed, precipitation probability, and weather-code fields.
 - The Global Spot bootstrap resources already existed on OSH: 5 virtual forecast systems, 30 forecast datastreams, and the deployment hierarchy.
 - The first live publish attempt inserted 0 observations because OSH rejected an empty `leadTimeHours` decimal field. The publisher now preserves the field shape and uses OSH's supported `NaN` decimal sentinel when the upstream response lacks an issued/model-run time needed to compute lead time.
+- Before installing a persistent service, the publisher was updated to persist recently published forecast dedupe keys in `publishers/met_office_global_spot/state.json`, so service restarts do not repost the same forecast horizon.
 - After the fix, one live Global Spot `--once` cycle published 625 forecast observations with 0 errors and 0 skipped records.
 - CSAPI verification against datastream `06hg2` returned a live forecast observation with forecast type `Met Office Global Spot hourly deterministic forecast`, valid time `2026-05-26T19:00:00Z`, result time `2026-05-26T19:35:34Z`, air temperature `29.9`, and `leadTimeHours=NaN`.
 - Production Explorer reloaded to 905 map features after the Global Spot resources and observations were live. Selecting `Met Office Global Spot Portsmouth / Thorney Island Area` rendered a dedicated Forecast section and did not render Latest readings or Recent trend for forecast datastreams.
@@ -123,7 +124,7 @@ MET_OFFICE_DATAHUB_API_KEY_HEADER=apikey
 1. Commit and push the publisher `leadTimeHours` sentinel fix and focused parser tests.
 2. Commit and push the Explorer UI polish that hides unknown lead time values.
 3. Verify Cloudflare Pages production bundle after deployment and re-check the Global Spot Portsmouth / Thorney Island card.
-4. Install a persistent Oracle systemd service only after the deployed UI smoke check is clean.
+4. Seed the Oracle publisher state from the currently published forecast horizon, then install and start the persistent Oracle systemd service.
 
 ## Explorer Follow-Up
 

publishers/met_office_global_spot/met_office_global_spot_publisher.py

@@ -29,6 +29,8 @@
 DEFAULT_API_BASE = "https://data.hub.api.metoffice.gov.uk/sitespecific/v0"
 DEFAULT_HOURLY_PATH = "/point/hourly"
 FORECAST_TYPE = "Met Office Global Spot hourly deterministic forecast"
+STATE_PATH = Path(__file__).with_name("state.json")
+MAX_SEEN_KEYS = 5000
 
 
 class UpstreamRateLimit(RuntimeError):
@@ -298,6 +300,7 @@ def __init__(self, location_filter: list[str] | None = None):
             self.locations = [s for s in self.locations if s["id"] in wanted]
         self.parameters = _load_parameters()
         self.client = MetOfficeGlobalSpotClient()
+        self.state = self._load_state()
 
         self.osh_address = os.environ.get("OSH_ADDRESS", "")
         self.osh_port = int(os.environ.get("OSH_PORT", "443"))
@@ -318,13 +321,39 @@ def __init__(self, location_filter: list[str] | None = None):
         self._auth = "Basic " + base64.b64encode(
             f"{self.osh_user}:{self.osh_pass}".encode()).decode()
         self._ds_ids: dict[str, dict[str, str]] = {}
-        self._seen: set[str] = set()
+        self._seen: set[str] = set(self.state.get("publishedKeys", []))
         self._request_delay = float(os.environ.get("MET_OFFICE_GLOBAL_SPOT_REQUEST_DELAY", "1.0"))
         self._rate_limit_backoff = float(os.environ.get("MET_OFFICE_GLOBAL_SPOT_429_BACKOFF", "3600"))
         self._forecast_hours = float(os.environ.get("MET_OFFICE_GLOBAL_SPOT_FORECAST_HOURS", "24"))
         self._cooldown_until = 0.0
         self.stats = {"published": 0, "errors": 0, "reconnects": 0, "skipped": 0}
 
+    def _load_state(self) -> dict:
+        if not STATE_PATH.exists():
+            return {"publishedKeys": []}
+        try:
+            state = json.loads(STATE_PATH.read_text(encoding="utf-8"))
+        except Exception:
+            return {"publishedKeys": []}
+        if not isinstance(state.get("publishedKeys"), list):
+            state["publishedKeys"] = []
+        return state
+
+    def _save_state(self):
+        keys = list(dict.fromkeys(self.state.get("publishedKeys", [])))[-MAX_SEEN_KEYS:]
+        self.state["publishedKeys"] = keys
+        STATE_PATH.write_text(json.dumps(self.state, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+    def _remember_seen(self, key: str, *, persist: bool):
+        self._seen.add(key)
+        if not persist:
+            return
+        keys = self.state.setdefault("publishedKeys", [])
+        keys.append(key)
+        if len(keys) > MAX_SEEN_KEYS:
+            del keys[:-MAX_SEEN_KEYS]
+        self._save_state()
+
     def _system_uid(self, location_id: str) -> str:
         return f"urn:os4csapi:system:met-office-datahub-global-spot:{_uid_token(location_id)}:v1"
 
@@ -534,13 +563,13 @@ def publish_cycle(self, dry_run: bool = False) -> int:
                 value_label = f"{parameter['label']}={forecast['value']} {parameter['unit']} valid {forecast['phenomenonTime']}"
                 if dry_run:
                     print(f"  [{ts_label}] {location_id}/{output_name}: [DRY] {value_label}")
-                    self._seen.add(forecast["dedupeKey"])
+                    self._remember_seen(forecast["dedupeKey"], persist=False)
                 else:
                     try:
                         self._post_observation(ds_id, obs)
                         self.stats["published"] += 1
                         published += 1
-                        self._seen.add(forecast["dedupeKey"])
+                        self._remember_seen(forecast["dedupeKey"], persist=True)
                         print(f"  [{ts_label}] {location_id}/{output_name}: OK  {value_label}")
                     except Exception as exc:
                         self.stats["errors"] += 1

tests/test_met_office_global_spot_parser.py

@@ -1,5 +1,6 @@
 from datetime import datetime, timedelta, timezone
 
+import publishers.met_office_global_spot.met_office_global_spot_publisher as global_spot
 from publishers.met_office_global_spot.met_office_global_spot_publisher import (
     MetOfficeGlobalSpotPublisher,
     _candidate_records,
@@ -86,3 +87,19 @@ def hourly_forecast(self, _location):
 
     assert forecasts[0]["result"]["issuedTime"] == ""
     assert forecasts[0]["result"]["leadTimeHours"] == "NaN"
+
+
+def test_seen_forecast_keys_persist_for_service_restarts(tmp_path, monkeypatch):
+    monkeypatch.setattr(global_spot, "STATE_PATH", tmp_path / "state.json")
+
+    publisher = MetOfficeGlobalSpotPublisher.__new__(MetOfficeGlobalSpotPublisher)
+    publisher.state = {"publishedKeys": []}
+    publisher._seen = set()
+
+    publisher._remember_seen("location|air_temperature_forecast|valid|12.3", persist=True)
+
+    restarted = MetOfficeGlobalSpotPublisher.__new__(MetOfficeGlobalSpotPublisher)
+    restarted.state = restarted._load_state()
+    restarted._seen = set(restarted.state.get("publishedKeys", []))
+
+    assert "location|air_temperature_forecast|valid|12.3" in restarted._seen