fix(weather): constellation product parity + probe-measured ingest speed

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

@@ -232,10 +232,13 @@ jobs:
             # pilot's n2-standard-4, but one station is far cheaper than the
             # 65-station fleet. Run all year-shards concurrently (parallelism
             # capped at 48 to match the fleet's Spot-availability ceiling).
-            MACHINE_TYPE="n2-standard-8"
+            # n2-standard-16 (compute budget approved 2026-07-05): with the
+            # spawn process pool + the probe-raised rate cap, decode scales
+            # with cores — 16 workers on 16 vCPUs.
+            MACHINE_TYPE="n2-standard-16"
             PARALLELISM=$(( TASK_COUNT < 48 ? TASK_COUNT : 48 ))
-            CPU_MILLI=8000
-            MEM_MIB=32768
+            CPU_MILLI=16000
+            MEM_MIB=65536
             # Explicit mode (satellites/products/year-window/out all passed). The
             # CLI decomposes BATCH_TASK_INDEX/COUNT into this task's (station,
             # year). --max-workers 8 matches the 8-vCPU machine. GOES-East
@@ -267,7 +270,7 @@ jobs:
             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" --arg sats "$SATELLITES" \
               --arg ex "$EXECUTOR" \
-              '["--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","--executor",$ex]')
+              '["--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","16","--executor",$ex]')
           else
             TASK_COUNT=1
             # Pilot machine — UNCHANGED (n2-standard-4, parallelism 16): a single

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

@@ -283,6 +283,33 @@ class ProductVariable:
         description=("Downward Shortwave Radiation at the surface (full-disk product only)"),
         tier="v1",
     ),
+    ("ABI-L2-DSRC", "DSR"): ProductVariable(
+        product_code="ABI-L2-DSRC",
+        variable="DSR",
+        units_expected="W m-2",
+        # goes16 CONUS sector grid, verified on a LIVE sample 2026-07-05
+        # (OR_ABI-L2-DSRC-M6_G16_s20231671801177). goes17/goes18 publish the
+        # PACUS sector at (157, 380) — the per-satellite split lives in
+        # _goes_s3._DSRC_SECTOR_SHAPES (same pattern as the DSRF grid split).
+        # goes19 publishes NO DSRC (full-disk DSRF only). All eras ship on
+        # goes_lat_lon_projection (the coarse lat/lon extractor branch).
+        # goes16 archive spans 2019-12 -> 2024 day-108 (the same NOAA
+        # product-overhaul cutover as the DSRF coarse->full-disk transition);
+        # outside that window listings are simply empty (hourly cadence makes
+        # the empty-listing cost negligible).
+        grid_shape_expected=(166, 394),
+        cadence_s=3600,
+        bounds=(0.0, 1400.0),
+        dqf_good=frozenset({0}),
+        dqf_variable="DQF",
+        is_3d_profile=False,
+        is_categorical=False,
+        description=(
+            "Downward Shortwave Radiation at the surface (CONUS/PACUS sector, "
+            "hourly ~150 KB files — the cheap DSR complement to full-disk DSRF)"
+        ),
+        tier="v1",
+    ),
     # ==================== v2 ====================
     ("ABI-L2-ACHAC", "HT"): ProductVariable(
         product_code="ABI-L2-ACHAC",
@@ -865,8 +892,10 @@ def _extract_from_dataset(
         if var_name not in ds.variables:
             # Skip missing variables instead of aborting the entire file.
             # Real-world GOES-16 files from different time periods have
-            # different variable sets (e.g. ACMC files from 2017 have only
-            # BCM+DQF, while 2024 files have BCM+ACM+Cloud_Probabilities+DQF).
+            # different variable sets. Parity audit (2026-07-05): the 2-variable
+            # BCM+DQF ACMC era runs 2017 through ~Dec 2021 on goes16 (and
+            # through ~mid-2022 on goes17) — 3-var files only after that; AODC
+            # gained AE1/AE2 mid-2020. goes18/goes19 are full-var from birth.
             # Raising GoesDataCorruptError here would abort extraction for
             # ALL registered variables in the product, losing even the
             # variables that DO exist. Verified 2026-04-12 during sprint2i

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

@@ -164,9 +164,50 @@
     "goes17": date(2023, 1, 10),
 }
 
+# Per-PRODUCT availability floors (parity audit P2, live-listed 2026-07-05).
+# Several enterprise L2 products rolled out YEARS after a platform's first
+# light — e.g. goes16 TPWC/ACHAC/DSRC/LVM/LVT first objects are Dec 2019 —
+# so the satellite-level floor alone enumerates ~2.5 years of guaranteed-empty
+# hourly listings per product. Floors are set conservatively EARLY (a too-early
+# floor only wastes cheap listings; a too-late one silently loses data):
+# Dec-2019 products floor at 2019-12-01; CTPC/DSIC (first objects in 2019,
+# exact day unverified) floor at 2019-01-01. The Dec-2019 rollout was a
+# product-wide NOAA enterprise update, so goes17 carries the same floors.
+# Consumed via _available_since_for() (max of satellite + product floors).
+_PRODUCT_AVAILABLE_SINCE: dict[tuple[str, str], date] = {
+    ("goes16", "ABI-L2-TPWC"): date(2019, 12, 1),
+    ("goes16", "ABI-L2-ACHAC"): date(2019, 12, 1),
+    ("goes16", "ABI-L2-DSRC"): date(2019, 12, 1),
+    ("goes16", "ABI-L2-LVMPC"): date(2019, 12, 1),
+    ("goes16", "ABI-L2-LVTPC"): date(2019, 12, 1),
+    ("goes16", "ABI-L2-CTPC"): date(2019, 1, 1),
+    ("goes16", "ABI-L2-DSIC"): date(2019, 1, 1),
+    ("goes17", "ABI-L2-TPWC"): date(2019, 12, 1),
+    ("goes17", "ABI-L2-ACHAC"): date(2019, 12, 1),
+    ("goes17", "ABI-L2-DSRC"): date(2019, 12, 1),
+    ("goes17", "ABI-L2-LVMPC"): date(2019, 12, 1),
+    ("goes17", "ABI-L2-LVTPC"): date(2019, 12, 1),
+    ("goes17", "ABI-L2-CTPC"): date(2019, 1, 1),
+    ("goes17", "ABI-L2-DSIC"): date(2019, 1, 1),
+}
+
+
+def _available_since_for(satellite: str, product: str) -> date | None:
+    """The effective availability floor for a (satellite, product) pair.
+
+    max() of the satellite first-light floor and the per-product rollout floor
+    — the backfill clamp skips any month wholly before this with NO I/O.
+    """
+    sat = _AVAILABLE_SINCE.get(satellite)
+    prod = _PRODUCT_AVAILABLE_SINCE.get((satellite, product))
+    if sat is None:
+        return prod
+    return max(sat, prod) if prod is not None else sat
+
+
 # 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
+_GOES_S3_RATE_HZ = 100.0
 
 _MAX_S3_RETRIES = 3
 _S3_BACKOFF_BASE_S = 1.0
@@ -190,8 +231,16 @@
 # 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
+# Products needing a cap above the CONUS-class default: DSRF full-disk
+# (~50 MB) and the 3D profile pair — pre-Feb-2024 LVM/LVT files are ~29.4 MiB
+# (live-measured 30,830,368 B; NOAA switched them to a compressed ~7 MiB layout
+# at 2024 day-038). The 16 MiB default silently zeroed the entire pre-2024
+# profile era "before read" (parity audit P1 — the same failure family as the
+# old 2 MiB ACMC cap).
+_LARGE_CAP_PRODUCTS: frozenset[str] = frozenset({"ABI-L2-DSRF", "ABI-L2-LVMPC", "ABI-L2-LVTPC"})
 _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
+    p: (64 * 1024 * 1024 if p in _LARGE_CAP_PRODUCTS else _DEFAULT_SIZE_CAP_BYTES)
+    for (p, _v) in PRODUCTS
 }
 
 # The coarse DSRF grid shape (D6 §A.6). The registry carries the full-disk ABI
@@ -203,6 +252,31 @@
 # too.
 _DSRF_GOES16_COARSE_SHAPE: tuple[int, int] = (326, 326)
 
+# ABI-L2-DSRC per-satellite SECTOR grids (live-verified 2026-07-05): goes16
+# publishes the CONUS sector (166, 394) — the registry baseline — while
+# goes17/goes18 publish the PACUS sector (157, 380). goes19 publishes no DSRC.
+# Same accept-only-your-own-variant discipline as the DSRF split: a swapped
+# sector still fails loudly.
+_DSRC_SECTOR_SHAPES: dict[str, tuple[int, int]] = {
+    "goes16": (166, 394),
+    "goes17": (157, 380),
+    "goes18": (157, 380),
+}
+
+# DSRF grid VARIANT per satellite (parity audit, live-verified 2026-07-05):
+#   - "bimodal": coarse (326,326) lat/lon before the 2024 day-108 NOAA product
+#     overhaul, full-disk (5424,5424) ABI fixed grid after — keyed per-file on
+#     the projection variable. goes16 AND goes18 (goes18 cutover confirmed at
+#     the SAME day: last coarse 2024/107, first full-disk 2024/108).
+#   - "coarse": coarse for the platform's ENTIRE life. goes17 (2019-2023) died
+#     before the overhaul — its every DSRF file is (326,326) lat/lon.
+#   - absent (goes19+): registry full-disk only (born after the overhaul).
+_DSRF_VARIANTS: dict[str, str] = {
+    "goes16": "bimodal",
+    "goes17": "coarse",
+    "goes18": "bimodal",
+}
+
 
 # ---------------------------------------------------------------------------
 # Rate limiter (identical to iem_mos._RateLimiter / 2i 475-492)
@@ -224,7 +298,28 @@ def acquire(self) -> None:
             self._last = time.monotonic()
 
 
-_DEFAULT_LIMITER = _RateLimiter(1.0 / _GOES_S3_RATE_HZ)
+def _effective_rate_hz() -> float:
+    """The rate this PROCESS may use: the aggregate cap, or its worker share.
+
+    _GOES_S3_RATE_HZ is the measured PER-TASK AGGREGATE cap (SOURCE-LIMITS,
+    probe 2026-07-05). Under ``executor=process`` every spawn worker re-imports
+    this module — an unscaled limiter would multiply the cap by the worker
+    count (Codex P2: 16 workers ~= 16x the documented limit). The parent
+    divides the cap across workers via this env override BEFORE creating the
+    pool; spawn children inherit env and each takes its share at import.
+    """
+    import os
+
+    raw = os.environ.get("_MOSTLYRIGHT_GOES_RATE_HZ_OVERRIDE")
+    if raw:
+        try:
+            return float(raw)
+        except ValueError:
+            log.warning("ignoring malformed _MOSTLYRIGHT_GOES_RATE_HZ_OVERRIDE=%r", raw)
+    return _GOES_S3_RATE_HZ
+
+
+_DEFAULT_LIMITER = _RateLimiter(1.0 / _effective_rate_hz())
 
 
 # ---------------------------------------------------------------------------
@@ -468,27 +563,30 @@ def _validate_dataset_shape(ds: Any, product: str, satellite: str) -> None:
             continue
 
         accepted = {expected}
+        if product == "ABI-L2-DSRC":
+            # Per-satellite sector split (goes16 CONUS vs goes17/18 PACUS).
+            # Unknown satellites (goes19 publishes no DSRC) keep the registry
+            # baseline so an unexpected file still fails loudly.
+            accepted = {_DSRC_SECTOR_SHAPES.get(satellite, expected)}
         if product == "ABI-L2-DSRF":
-            if satellite == "goes16":
-                # goes16 DSRF is BIMODAL: it shipped on the coarse 326x326 lat/lon
-                # grid (goes_lat_lon_projection) through early 2024, then switched
-                # to the full-disk ABI fixed grid (goes_imager_projection,
-                # 5424x5424) from ~2024 day-108. Key the accepted shape on the
-                # file's OWN projection variable — the SAME per-file branch
-                # _extract_from_dataset uses — so each grid must match its own
-                # expected shape and a mislabeled/swapped grid still fails loudly
-                # (no silent accept). Before this, goes16 DSRF was hard-coded to
-                # the coarse shape and every post-cutover full-disk file was
-                # rejected "before pixel read", stalling the DSRF >= mid-2024
-                # backfill.
+            # Variant per satellite (_DSRF_VARIANTS, parity audit 2026-07-05):
+            # bimodal platforms (goes16/goes18) key the accepted shape on the
+            # file's OWN projection variable — the SAME per-file branch
+            # _extract_from_dataset uses — so each grid era must match its own
+            # expected shape and a mislabeled/swapped grid still fails loudly.
+            # goes17 was coarse for its ENTIRE life (died before the 2024
+            # day-108 overhaul): coarse is its ONLY accepted shape, and a
+            # "full-disk goes17" file is mislabeled data. goes19+ (absent from
+            # the map) is full-disk only.
+            variant = _DSRF_VARIANTS.get(satellite)
+            if variant == "bimodal":
                 if "goes_lat_lon_projection" in ds.variables:
                     accepted = {_DSRF_GOES16_COARSE_SHAPE}
                 else:
                     accepted = {expected}
+            elif variant == "coarse":
+                accepted = {_DSRF_GOES16_COARSE_SHAPE}
             else:
-                # Other satellites (goes19) ship DSRF ONLY on the full-disk ABI
-                # grid — a coarse/lat-lon file here is swapped/corrupt, so keep the
-                # registry full-disk shape as the sole accepted variant.
                 accepted = {expected}
         if actual not in accepted:
             raise GoesDataCorruptError(

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

@@ -67,10 +67,14 @@
 #: (the operational launch dates of each JPSS platform: Suomi-NPP 2011-10,
 #: NOAA-20 2017-11, NOAA-21 2022-11; clamped to the start of the launch year so
 #: a pre-launch day returns [] without any I/O).
+# JRR-CloudMask floors on the NODD buckets, LIVE-LISTED 2026-07-05 (parity
+# audit): the platforms flew years earlier, but the JRR product objects only
+# begin on these dates — the old platform-launch floors created phantom windows
+# (viirs-npp: ~11.5 years of guaranteed-empty listings).
 _AVAILABLE_SINCE: dict[str, date] = {
-    "viirs-npp": date(2012, 1, 1),
-    "viirs-n20": date(2018, 1, 1),
-    "viirs-n21": date(2023, 1, 1),
+    "viirs-npp": date(2023, 6, 20),
+    "viirs-n20": date(2023, 2, 19),
+    "viirs-n21": date(2023, 11, 13),
 }
 
 # Conservative top-level S3 call-rate cap (shared discipline with GOES/Himawari).

packages/weather/src/mostlyright/weather/satellite/SOURCE-LIMITS-satellite.md

@@ -1,6 +1,10 @@
 # SOURCE-LIMITS.md — satellite (GOES ABI L2, NOAA NODD)
 
-**Status:** CONSERVATIVE-PENDING — no live probe has been run yet.
+**Status:** MEASURED — live in-region probe run 2026-07-05 (Cloud Batch job
+`probe-goes-ondemand-121450`, us-central1 n2-standard-8 STANDARD, gcp mirror,
+sweep 1/4/8/16/32/64, wall 20 s). Knee at 32 workers; derived aggregate rate
+cap 118.0 Hz. Shipped constants carry margin below the record
+(`_GOES_S3_RATE_HZ = 100.0`, `_DEFAULT_MAX_WORKERS = 32`).
 **Re-run command:** `python -m mostlyright.weather.satellite probe --mirror aws --out .planning/research`
 **Valid until:** the first live throughput probe (D10 SAT-25-11) overwrites this
 with the measured anonymous-throttle / diminishing-returns knee.
@@ -37,8 +41,8 @@ conservative) is always safe.
 
 **Conservative-pending derived defaults** (the values the shipped constants equal):
 
-- derive_rate_cap: **20.0** Hz
-- derive_max_workers: **8**
+- derive_rate_cap: **118.0** Hz
+- derive_max_workers: **32**
 
 ## Inputs to other plans
 

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

@@ -73,6 +73,7 @@
     _AVAILABLE_SINCE,
     _AVAILABLE_UNTIL,
     _GOES_ACTUAL_DATA_END,
+    _available_since_for,
     extract_pixel,
     list_product_keys,
 )
@@ -112,16 +113,20 @@
 #: ``_GOES_S3_RATE_HZ = 20.0``. Its value is the probe-derived ``derive_rate_cap``
 #: result recorded in the SOURCE-LIMITS.md satellite entry (provenance:
 #: SOURCE-LIMITS-satellite.md; conservative-pending the first live probe run).
-#: The 25-03 transport's ``_RateLimiter`` consumes this cap.
-_GOES_S3_RATE_HZ: float = 20.0
+#: The 25-03 transport's ``_RateLimiter`` consumes this cap. MEASURED 2026-07-05
+#: by the live in-region probe (Batch job probe-goes-ondemand-121450,
+#: us-central1 n2-standard-8, gcp mirror, sweep 1..64): knee at 32 workers,
+#: derive_rate_cap=118.0 Hz. Shipped at 100.0 (margin below the knee); the
+#: provenance-lock test asserts shipped <= the SOURCE-LIMITS record.
+_GOES_S3_RATE_HZ: float = 100.0
 
 #: Default fan-out width for ``bulk_backfill``. Its value is the probe-derived
 #: ``derive_max_workers`` result (the knee of the diminishing-returns curve)
 #: recorded in the SOURCE-LIMITS.md satellite entry (provenance:
 #: SOURCE-LIMITS-satellite.md). Re-measure with
 #: ``python -m mostlyright.weather.satellite probe``. Conservative-pending the
 #: first live probe run; the provenance-lock test asserts shipped <= recorded.
-_DEFAULT_MAX_WORKERS: int = 8
+_DEFAULT_MAX_WORKERS: int = 32
 
 # ---------------------------------------------------------------------------
 # Resume-layer constants (ported from satellite_backfill_plan_2i.md:40-56).
@@ -288,7 +293,10 @@ def backfill_goes_satellite(
         # available_since clamp: skip a whole slice before the satellite's
         # first-light date with no I/O (2i 1320, collapsed to the month grain).
         if source == _GOES_SOURCE:
-            available_since = _AVAILABLE_SINCE.get(satellite)
+            # max(satellite first-light, per-product rollout floor) — several
+            # enterprise products rolled out years after first light (parity
+            # audit P2); the product floor kills those guaranteed-empty months.
+            available_since = _available_since_for(satellite, product)
         else:
             available_since = _anon_available_since(source, satellite)
         last_day_of_month = _last_day_of_month(year, month)
@@ -793,6 +801,14 @@ def _make_executor(executor: str, max_workers: int) -> Executor:
     why this never reproduced locally.
     """
     if executor == "process":
+        # Divide the PER-TASK aggregate GOES rate cap across the workers: each
+        # spawn child re-imports _goes_s3 and builds its limiter from the env
+        # override at import (Codex P2 — unscaled, N workers ran N x the
+        # documented cap). Floor 1 Hz so a huge worker count cannot starve
+        # children. Set BEFORE pool creation so every child inherits it.
+        os.environ["_MOSTLYRIGHT_GOES_RATE_HZ_OVERRIDE"] = str(
+            max(_GOES_S3_RATE_HZ / max_workers, 1.0)
+        )
         return ProcessPoolExecutor(
             max_workers=max_workers, mp_context=multiprocessing.get_context("spawn")
         )

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

@@ -72,6 +72,13 @@ def _goes_products() -> frozenset[str]:
                 "ABI-L2-AODC",
                 "ABI-L2-CTPC",
                 "ABI-L2-DSIC",
+                # Keep in sync with _goes_extract.PRODUCTS — the no-extra
+                # drift-guard test (test_satellite_sources.py) asserts this
+                # fallback equals the heavy registry's product set, so a new
+                # registration that forgets this list fails CI loudly (Codex
+                # P2 on the DSRC addition: a base-install/hosted request for a
+                # registered product was rejected as unknown).
+                "ABI-L2-DSRC",
                 "ABI-L2-DSRF",
                 "ABI-L2-LSTC",
                 "ABI-L2-LVMPC",