fix(weather): spawn-context process pool; station mode runs thread executor

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

@@ -214,11 +214,21 @@ jobs:
             MEM_MIB=32768
             # Explicit mode (satellites/products/year-window/out all passed). The
             # CLI decomposes BATCH_TASK_INDEX/COUNT into this task's (station,
-            # year). --executor process for DSRF full-disk; --max-workers 8 matches
-            # the 8-vCPU machine. GOES-East (goes16) covers the US pilot stations.
+            # year). --max-workers 8 matches the 8-vCPU machine. GOES-East
+            # (goes16) covers the US pilot stations.
+            #
+            # DELIBERATELY the default THREAD executor, NOT --executor process:
+            # the GCS progress store (--progress-bucket) starts gRPC threads
+            # before the pool spins up, and ProcessPoolExecutor's Linux default
+            # fork start-method breaks in forked children after gRPC threads
+            # exist -> BrokenProcessPool, tasks exit with "0 slices completed"
+            # (observed live, job 28734758679, 2026-07-05). Threads are the
+            # 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.
             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","--executor","process"]')
+              '["--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"]')
           else
             TASK_COUNT=1
             # Pilot machine — UNCHANGED (n2-standard-4, parallelism 16): a single

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

@@ -52,6 +52,7 @@
 import contextlib
 import json
 import logging
+import multiprocessing
 import os
 import re
 import socket
@@ -775,9 +776,23 @@ def _run_slice(item: _SliceItem) -> ProductBackfillResult:
 
 
 def _make_executor(executor: str, max_workers: int) -> Executor:
-    """Return a Thread or Process pool (D7 — CONUS thread / DSRF process)."""
+    """Return a Thread or Process pool (D7 — CONUS thread / DSRF process).
+
+    The process pool uses the SPAWN start method explicitly: on Linux the
+    default is fork, and by the time the pool spins up the process may already
+    carry live gRPC threads (the ``--progress-bucket`` GcsProgressStore client
+    starts them at CLI startup). fork()-ing a thread-holding process gives
+    children a torn copy of those threads' locks — observed live as an
+    immediate ``BrokenProcessPool`` with "0 slices completed" on every fleet
+    task (job weather-backfill-station-28734758679, 2026-07-05). Spawn starts
+    clean children; ``_run_slice`` is module-level + fully picklable by design
+    (P1-1), so spawn is safe here. macOS already defaults to spawn, which is
+    why this never reproduced locally.
+    """
     if executor == "process":
-        return ProcessPoolExecutor(max_workers=max_workers)
+        return ProcessPoolExecutor(
+            max_workers=max_workers, mp_context=multiprocessing.get_context("spawn")
+        )
     if executor == "thread":
         return ThreadPoolExecutor(max_workers=max_workers)
     raise ValueError(f"executor must be 'thread' or 'process'; got {executor!r}")

packages/weather/tests/test_satellite_backfill.py

@@ -538,6 +538,15 @@ def _spy(executor: str, max_workers: int):
         assert ThreadPoolExecutor in captured
 
     def test_process_executor_for_dsrf(self, tmp_path) -> None:
+        # NOTE (spawn mock-escape): the pool is a REAL ProcessPoolExecutor whose
+        # children re-import _backfill under the spawn start method, so a
+        # parent-side mock.patch of backfill_goes_satellite NEVER reaches them
+        # (under spawn it silently escaped to the real 2024 GOES transport —
+        # macOS hit this even before the spawn pin, spawn being its default).
+        # Instead of mocking, submit a pre-first-light year (goes16 first light
+        # 2017-05-24) so every slice short-circuits child-side via the
+        # available_since clamp with ZERO I/O — the same spawn-safe shape as
+        # TestProcessPoolPicklable.test_process_executor_runs_end_to_end.
         captured: list[type] = []
         orig = _backfill._make_executor
 
@@ -546,33 +555,21 @@ def _spy(executor: str, max_workers: int):
             captured.append(type(ex))
             return ex
 
-        with (
-            mock.patch.object(_backfill, "backfill_goes_satellite") as m_slice,
-            mock.patch.object(_backfill, "_make_executor", _spy),
-        ):
-            m_slice.return_value = _backfill.ProductBackfillResult(
-                station="KNYC",
-                satellite="goes16",
-                product="ABI-L2-DSRF",
-                year=2024,
-                month=1,
-                scans_fetched=0,
-                rows_written=0,
-                duration_s=0.0,
-                errors=(),
-                skipped_pre_availability=False,
-            )
-            _backfill.bulk_backfill(
+        with mock.patch.object(_backfill, "_make_executor", _spy):
+            res = _backfill.bulk_backfill(
                 satellites=["goes16"],
                 products=["ABI-L2-DSRF"],
                 stations=["KNYC"],
-                year_start=2024,
-                year_end=2024,
+                year_start=2016,  # entirely before goes16 first-light -> no child I/O
+                year_end=2016,
                 out=tmp_path,
                 resume=False,
                 executor="process",
+                max_workers=2,
             )
         assert ProcessPoolExecutor in captured
+        # Prove the no-I/O shape held: every slice pre-availability skipped.
+        assert all(r.skipped_pre_availability for r in res.results)
 
     def test_max_workers_threaded_and_default_is_constant(self, tmp_path) -> None:
         seen: dict[str, int] = {}
@@ -1042,6 +1039,23 @@ def test_cli_invalid_mirror_rejected_by_argparse(self, tmp_path) -> None:
 # picklable payload.
 # ---------------------------------------------------------------------------
 class TestProcessPoolPicklable:
+    def test_process_pool_uses_spawn_start_method(self) -> None:
+        """The process pool MUST use spawn, never Linux's default fork.
+
+        Regression for the fleet outage (job weather-backfill-station-28734758679,
+        2026-07-05): the --progress-bucket GcsProgressStore starts gRPC threads at
+        CLI startup, and fork()-ing after that gives pool children a torn copy of
+        those threads' locks -> instant BrokenProcessPool, every task exiting with
+        "0 slices completed". Spawn starts clean children; the worker + items are
+        module-level/picklable by design (P1-1), so spawn is safe. macOS defaults
+        to spawn, which is why the fork bug never reproduced locally.
+        """
+        pool = _backfill._make_executor("process", 2)
+        try:
+            assert pool._mp_context.get_start_method() == "spawn"
+        finally:
+            pool.shutdown(wait=False, cancel_futures=True)
+
     def test_module_level_worker_and_item_round_trip_pickle(self, knyc) -> None:
         # The worker callable submitted into the pool must be importable by
         # qualified name (module scope), i.e. picklable on its own.