diff --git a/.github/workflows/run-weather-backfill.yml b/.github/workflows/run-weather-backfill.yml index 1e8a8e9..723f134 100644 --- a/.github/workflows/run-weather-backfill.yml +++ b/.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 diff --git a/packages/weather/src/mostlyright/weather/satellite/_backfill.py b/packages/weather/src/mostlyright/weather/satellite/_backfill.py index 2f77e71..b498c1a 100644 --- a/packages/weather/src/mostlyright/weather/satellite/_backfill.py +++ b/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}") diff --git a/packages/weather/tests/test_satellite_backfill.py b/packages/weather/tests/test_satellite_backfill.py index df7f564..9198bbb 100644 --- a/packages/weather/tests/test_satellite_backfill.py +++ b/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.