From 66d17a6aca770404808b099282da4b77259b0eeb Mon Sep 17 00:00:00 2001 From: helloiamvu Date: Sun, 5 Jul 2026 10:35:03 +0200 Subject: [PATCH 1/2] fix(weather): spawn-context process pool; station mode uses thread executor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live outage on the first station-mode fleet run (job weather-backfill-station-28734758679, 2026-07-05): every task died with BrokenProcessPool and exited "0 slices completed, 0 rows written" in 0.2s — 5 tasks reported SUCCEEDED having done nothing. Root cause: --progress-bucket starts the GcsProgressStore gRPC client threads at CLI startup; ProcessPoolExecutor then forks (Linux default start method) and the children inherit a torn copy of those threads' locks -> instant pool breakage. Never reproduced locally (macOS defaults to spawn) or in CI (the process-pool test runs without a progress bucket, so no gRPC threads exist pre-fork). - _backfill._make_executor: pin mp_context=spawn for the process pool. _run_slice + its item tuple are module-level/picklable by design (P1-1), so spawn is safe. Regression test pins the start method and the existing end-to-end process test still passes under spawn. - run-weather-backfill.yml station mode: drop --executor process; run the default THREAD executor — 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 path is proven in a container. Co-Authored-By: Claude Opus 4.8 --- .github/workflows/run-weather-backfill.yml | 16 +++++++++++++--- .../weather/satellite/_backfill.py | 19 +++++++++++++++++-- .../weather/tests/test_satellite_backfill.py | 17 +++++++++++++++++ 3 files changed, 47 insertions(+), 5 deletions(-) diff --git a/.github/workflows/run-weather-backfill.yml b/.github/workflows/run-weather-backfill.yml index 1e8a8e97..723f134b 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 2f77e71c..b498c1a8 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 df7f5642..9fe422e9 100644 --- a/packages/weather/tests/test_satellite_backfill.py +++ b/packages/weather/tests/test_satellite_backfill.py @@ -1042,6 +1042,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. From 5e4cc3eea30bad3c658c76929a5d9a1423dfb14f Mon Sep 17 00:00:00 2001 From: helloiamvu Date: Sun, 5 Jul 2026 10:46:56 +0200 Subject: [PATCH 2/2] =?UTF-8?q?test(weather):=20spawn-safe=20process-execu?= =?UTF-8?q?tor=20test=20=E2=80=94=20no=20mock=20across=20fork=20boundary?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex P2 on the spawn-context change: with spawn, pool children re-import _backfill fresh, so test_process_executor_for_dsrf's parent-side mock.patch of backfill_goes_satellite never reached them — the "network-free" test silently ran the real 2024 GOES DSRF transport (macOS hit this even before the spawn pin, spawn being its default; CI's fork start method masked it). Rewrite to the same spawn-safe shape as test_process_executor_runs_end_to_end: submit a pre-first-light year (2016 < goes16 first light 2017-05-24) so every slice short-circuits child-side via the available_since clamp with zero I/O, and assert the skip actually held. Runs in ~2s with no network. Co-Authored-By: Claude Opus 4.8 --- .../weather/tests/test_satellite_backfill.py | 35 +++++++++---------- 1 file changed, 16 insertions(+), 19 deletions(-) diff --git a/packages/weather/tests/test_satellite_backfill.py b/packages/weather/tests/test_satellite_backfill.py index 9fe422e9..9198bbb0 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] = {}