fix(weather): era-complete GOES coverage — filter, clamps, fallback, scope keys, size cap

.github/workflows/run-weather-backfill.yml

@@ -35,7 +35,7 @@ on:
   workflow_dispatch:
     inputs:
       image_tag:
-        description: "weather-backfill image tag to run (must already be pushed by deploy-weather-ingest)."
+        description: "weather-backfill image tag to run (must already be pushed by deploy-weather-ingest). MUST be built from a commit AFTER the GOES coverage filter (PR #101) when using the multi-satellite default below — an older image cartesian-products all requested satellites (wrong-hemisphere downloads)."
         required: true
         default: "latest"
         type: string
@@ -58,6 +58,11 @@ on:
         required: true
         default: "ABI-L2-ACMC"
         type: string
+      satellites:
+        description: "Comma-separated satellites for pilot/station modes (default: the era-complete GOES set — REQUIRES an image built after the GOES coverage filter, PR #101; older images cartesian-product all four = wrong-hemisphere downloads). Override for non-GOES stations, e.g. himawari8,himawari9 + products=AHI-L2-FLDK-Clouds (Asia/RJTT) or viirs-npp,viirs-n20,viirs-n21 + products=VIIRS-JRR-CloudMask (Europe — Meteosat backfill is not wired yet). mode=full IGNORES this input entirely (roster defaults apply)."
+        required: true
+        default: "goes16,goes17,goes18,goes19"
+        type: string
       year_start:
         description: "First year (inclusive) of the backfill window (mode=pilot and mode=station)."
         required: true
@@ -120,6 +125,7 @@ jobs:
           MODE: ${{ inputs.mode }}
           PILOT_STATION: ${{ inputs.pilot_station }}
           PRODUCTS: ${{ inputs.products }}
+          SATELLITES: ${{ inputs.satellites }}
           YEAR_START: ${{ inputs.year_start }}
           YEAR_END: ${{ inputs.year_end }}
           R2_BUCKET: ${{ vars.R2_BUCKET }}
@@ -226,9 +232,17 @@ jobs:
             # proven prod path (the 2024 pilot processed + uploaded DSRF with
             # threads); numpy/h5py release the GIL for the heavy decode. Re-enable
             # process only after the spawn-context fix is proven in a container.
+            # Era-complete GOES set (goes19-era coverage fix): all four GOES
+            # platforms so East stations keep coverage past the 2025
+            # goes16->goes19 handover and West stations gain the goes17 archive
+            # era. Requires an image built AFTER the hemisphere filter landed
+            # (the CLI enumerates only each station's own side pair; an OLD
+            # image would cartesian-product all four = wrong-hemisphere
+            # downloads). available_since clamps make the off-era platform
+            # nearly free.
             COMMANDS=$(jq -nc --arg st "$PILOT_STATION" --arg ys "$YEAR_START_EFF" --arg ye "$YEAR_END_EFF" \
-              --arg pb "$PROGRESS_BUCKET" --arg rb "$R2_BUCKET" --arg pr "$PRODUCTS" \
-              '["--mirror","gcp","--satellites","goes16","--products",$pr,"--stations",$st,"--year-start",$ys,"--year-end",$ye,"--out","/tmp/derived","--r2-target","--r2-bucket",$rb,"--progress-bucket",$pb,"--max-workers","8"]')
+              --arg pb "$PROGRESS_BUCKET" --arg rb "$R2_BUCKET" --arg pr "$PRODUCTS" --arg sats "$SATELLITES" \
+              '["--mirror","gcp","--satellites",$sats,"--products",$pr,"--stations",$st,"--year-start",$ys,"--year-end",$ye,"--out","/tmp/derived","--r2-target","--r2-bucket",$rb,"--progress-bucket",$pb,"--max-workers","8"]')
           else
             TASK_COUNT=1
             # Pilot machine — UNCHANGED (n2-standard-4, parallelism 16): a single
@@ -242,11 +256,13 @@ jobs:
             YEAR_END_EFF="${YEAR_END:-$(date -u +%Y)}"
             # Pilot: explicit single-station backfill. The CLI's explicit mode
             # (no --roster) requires satellites/products/year-window/out, so pass
-            # them all. GOES-East ACMC covers the default KNYC pilot; an
-            # international pilot station needs --satellites overridden.
+            # them all. Era-complete GOES set — the CLI's hemisphere filter
+            # keeps the station on its own side pair (requires an image built
+            # after the filter landed); an international pilot station still
+            # needs --satellites overridden (Himawari/Meteosat).
             COMMANDS=$(jq -nc --arg st "$PILOT_STATION" --arg ys "$YEAR_START_EFF" --arg ye "$YEAR_END_EFF" \
-              --arg pb "$PROGRESS_BUCKET" --arg rb "$R2_BUCKET" --arg pr "$PRODUCTS" \
-              '["--mirror","gcp","--satellites","goes16","--products",$pr,"--stations",$st,"--year-start",$ys,"--year-end",$ye,"--out","/tmp/derived","--r2-target","--r2-bucket",$rb,"--progress-bucket",$pb]')
+              --arg pb "$PROGRESS_BUCKET" --arg rb "$R2_BUCKET" --arg pr "$PRODUCTS" --arg sats "$SATELLITES" \
+              '["--mirror","gcp","--satellites",$sats,"--products",$pr,"--stations",$st,"--year-start",$ys,"--year-end",$ye,"--out","/tmp/derived","--r2-target","--r2-bucket",$rb,"--progress-bucket",$pb]')
           fi
           jq -n \
             --arg img "${IMAGE}" \

packages/weather/src/mostlyright/weather/_fetchers/_goes_s3.py

@@ -120,6 +120,50 @@
     "goes19": date(2024, 11, 15),
 }
 
+# Operational-era END clamps for RETIRED platforms (goes19-era coverage fix,
+# Codex P2) — the symmetric counterpart of _AVAILABLE_SINCE, consumed by the
+# backfill's ENUMERATION filter so post-retirement (satellite, month) slices are
+# never even submitted (each otherwise burns ~a task-minute of empty per-hour
+# listing calls, growing every future fleet year).
+#
+# SAFETY over savings: a too-early date here would SILENTLY skip real
+# settlement-adjacent data, while a too-late date only wastes cheap empty
+# listings — so every entry carries ~1 YEAR of margin past the public
+# retirement date, skips are logged loudly at enumeration, and enumeration-skip
+# writes NO progress marker (removing/raising a date later backfills those
+# months normally). Amend on any NOAA reactivation (standby platforms have
+# historically been reactivated during operational anomalies).
+#
+#   - goes17: operational GOES-West ended 2023-01-10 (goes18 took over
+#     2023-01-04; goes17 to storage). Clamp 2024-01-01 = ~1 yr margin.
+#   - goes16: goes19 became the operational GOES-East 2025-04-07 (goes16 to
+#     standby). Clamp 2026-01-01 = ~9 mo margin, AND 2026 Jan-Jun was measured
+#     ALL-EMPTY live (KNYC station backfill, 2026-07-05) — the clamp is
+#     observation-backed, not date-arithmetic alone.
+#   - goes18 / goes19: OPERATIONAL — deliberately absent (no end clamp).
+_AVAILABLE_UNTIL: dict[str, date] = {
+    "goes16": date(2026, 1, 1),
+    "goes17": date(2024, 1, 1),
+}
+
+# UN-MARGINED actual last-data dates for retired platforms — used ONLY for
+# POSITIVE coverage claims (the East-for-West fallback clip asks "does this
+# native platform REALLY cover this month?"), NEVER for fetch clamps. The two
+# maps deliberately lean opposite ways so both stay safe: _AVAILABLE_UNTIL is
+# margined LATE (fetch side: a too-late date only wastes cheap empty listings),
+# while claiming coverage from a margined date would suppress the East fallback
+# for months with NO real native data (verified live: s3://noaa-goes17/ has no
+# objects past 2023-01-10). Widening an _AVAILABLE_UNTIL margin must therefore
+# never touch this map.
+#   - goes17: last data 2023-01-10 (went to storage; goes18 took over 2023-01-04).
+#   - goes16: goes19 became operational GOES-East 2025-04-07 (kept for
+#     documentation; East platforms are never consulted as NATIVE coverage —
+#     no fallback direction exists toward the East side).
+_GOES_ACTUAL_DATA_END: dict[str, date] = {
+    "goes16": date(2025, 4, 7),
+    "goes17": date(2023, 1, 10),
+}
+
 # Conservative top-level S3/GCS call-rate cap (2i ``_GOES_S3_RATE_HZ``). The
 # empirical concurrency probe (25-05 D10) may later derive a tuned value.
 _GOES_S3_RATE_HZ = 20.0
@@ -134,10 +178,18 @@
 _S3_FAIL_FAST_CODES = frozenset({"404", "NoSuchKey", "NoSuchBucket", "AccessDenied"})
 
 # Per-product size cap (P2-d). Rejects an over-cap object BEFORE the full-object
-# read so the in-memory BytesIO buffer is bounded. CONUS files run ~0.3-1.5 MB;
-# DSRF full-disk runs ~50 MB. Caps are generous over the documented sizes.
-# Default cap (CONUS-class) applies to any product not listed explicitly.
-_DEFAULT_SIZE_CAP_BYTES = 2 * 1024 * 1024  # ~2 MB
+# read so the in-memory BytesIO buffer is bounded. Default cap (CONUS-class)
+# applies to any product not listed explicitly; DSRF full-disk (~50 MB) carries
+# its own larger cap.
+#
+# 16 MiB, NOT the original 2 MiB: the 2 MiB cap was calibrated on 2017-era
+# 2-variable ACMC files (~0.3-1.5 MB), but NOAA later added ACM +
+# Cloud_Probabilities to the product and modern ACMC files run ~4.3-4.6 MB —
+# observed LIVE 2026-07-05 (KNYC station backfill): the old cap rejected EVERY
+# 2022+ ACMC file "before read", zeroing ACMC coverage for the modern era
+# fleet-wide. 16 MiB = >3x headroom over the observed modern sizes while still
+# bounding the in-memory buffer well below the DSRF cap.
+_DEFAULT_SIZE_CAP_BYTES = 16 * 1024 * 1024  # 16 MiB
 _SIZE_CAP_BYTES: dict[str, int] = {
     p: (64 * 1024 * 1024 if p == "ABI-L2-DSRF" else _DEFAULT_SIZE_CAP_BYTES) for (p, _v) in PRODUCTS
 }

packages/weather/src/mostlyright/weather/satellite/__main__.py

@@ -50,11 +50,17 @@
 
 #: Default satellites for roster/incremental mode. batch.tf passes NO
 #: ``--satellites``, so roster mode must supply the canonical native-ring
-#: default. The ACMC-only MVP (project memory) runs on the GOES East/West pair —
-#: the two operational GOES platforms — so the fleet covers the Americas span the
-#: settlement stations concentrate in. (Explicit-args mode still requires
+#: default. ERA-COMPLETE GOES set (goes19-era coverage fix): the East pair
+#: goes16 (operational 2017 -> standby 2025) + goes19 (GOES-East since the 2025
+#: handover) AND the West pair goes17 (archive, -> storage 2023) + goes18. The
+#: old ("goes16", "goes18") pair silently lost East coverage after the 2025
+#: handover (observed live: KNYC 2026 backfilled ALL-EMPTY) and never had
+#: goes17-era West coverage (2018-2022). Requesting all four is cheap: the
+#: backfill hemisphere filter (_backfill._enumerate_slices) enumerates only the
+#: station's own side pair, and the ``available_since`` clamp skips
+#: pre-first-light slices with no I/O. (Explicit-args mode still requires
 #: ``--satellites``; this default ONLY applies when ``--roster`` is given.)
-_DEFAULT_ROSTER_SATELLITES: tuple[str, ...] = ("goes16", "goes18")
+_DEFAULT_ROSTER_SATELLITES: tuple[str, ...] = ("goes16", "goes17", "goes18", "goes19")
 
 #: Default products for roster/incremental mode: the ACMC-only MVP cloud mask
 #: (``ABI-L2-ACMC``, the cheap CONUS product — matches the ``_sources`` GOES
@@ -96,9 +102,10 @@ def _build_parser() -> argparse.ArgumentParser:
         "--satellites",
         type=_split_csv,
         default=None,
-        help="Comma-separated satellites, e.g. goes16,goes19 (GOES-East) or goes18 "
-        "(GOES-West). REQUIRED in explicit mode; in --roster mode defaults to the "
-        "native-ring default (goes16,goes18 — the operational GOES pair).",
+        help="Comma-separated satellites, e.g. goes16,goes19 (GOES-East) or "
+        "goes17,goes18 (GOES-West). REQUIRED in explicit mode; in --roster mode "
+        "defaults to the era-complete GOES set (goes16,goes17,goes18,goes19 — "
+        "the hemisphere filter keeps each station on its own side pair).",
     )
     bf.add_argument(
         "--products",
@@ -464,14 +471,52 @@ def _run_backfill(args: argparse.Namespace) -> int:
     # crash-safe resume. Wire it to the GCS-backed progress store so a preempted
     # Spot slice (or a replacement VM) rehydrates its markers from GCS and SKIPS
     # already-uploaded partitions instead of reprocessing under a fresh local disk.
-    # Each shard writes a DISJOINT marker object (shard-index in the key) so array
-    # tasks never clobber each other's progress.
+    #
+    # SCOPE-keyed marker object (28-29 multi-station fix; was shard-index-keyed).
+    # The key is the RESOLVED (stations, year-window) scope, which is:
+    #   - stable across Spot retries (the shard->(station,year) mapping is pure),
+    #   - disjoint across array tasks (each owns a distinct station-year),
+    #   - disjoint across CONCURRENT jobs for DIFFERENT stations — shard-index
+    #     keys collided there: two station-mode jobs both wrote
+    #     progress/shard-00000/, a read-modify-write clobber on the same object,
+    #   - resume-stable across re-submissions of the same scope.
+    # Old shard-NNNNN objects are simply orphaned (one idempotent re-fetch for
+    # any scope whose markers lived there; slice keys inside files are already
+    # full-identity so no wrong-skip is possible either way).
     if args.progress_bucket is not None:
+        import hashlib
+
         from ._progress import GcsProgressStore
 
-        shard_index, _shard_count = _resolve_shard_index_count(args)
         bucket = args.progress_bucket.rstrip("/")
-        gcs_uri = f"gs://{bucket}/progress/shard-{shard_index:05d}/progress.json"
+        # CANONICAL station identity (Codex P2, round 8): the CLI accepts NWS
+        # aliases (NYC -> KNYC) which bulk_backfill resolves via
+        # _resolve_station_infos — so the scope must be built from the SAME
+        # resolved, deduped, sorted ICAOs or an alias submission and a
+        # canonical submission of identical work would get different progress
+        # objects (resume misses; concurrent double-work).
+        from ._resolve import _resolve_station_infos
+
+        scope_icaos = sorted({i.icao for i in _resolve_station_infos(stations)})
+        if len(scope_icaos) <= 3:
+            scope_stations = "-".join(scope_icaos)
+        else:
+            # Whole-roster scopes (e.g. the incremental Cloud Run Job) would make
+            # an unwieldy object name — use a stable digest of the station tuple.
+            digest = hashlib.sha1("_".join(scope_icaos).encode()).hexdigest()[:10]
+            scope_stations = f"multi{len(scope_icaos)}-{digest}"
+        # The scope must ALSO carry the satellites+products axes: an ACMC job
+        # and a DSRF job for the same station+window are separate concurrent
+        # work-scopes, and GcsProgressStore's mark_complete PUTs the full
+        # in-memory map — same-object concurrent writers are last-write-wins.
+        # SORTED so goes16,goes19 vs goes19,goes16 map to the same scope.
+        sp_digest = hashlib.sha1(
+            (",".join(sorted(satellites)) + "#" + ",".join(sorted(products))).encode()
+        ).hexdigest()[:8]
+        gcs_uri = (
+            f"gs://{bucket}/progress/"
+            f"{scope_stations}_{year_start}-{year_end}_{sp_digest}/progress.json"
+        )
         kwargs["progress_store"] = GcsProgressStore(gcs_uri)
         print(f"using durable GCS progress store for crash-safe resume: {gcs_uri}")