Harden Rust grid worker: NOTIFY source, temp leaks, partial downloads #25

graham · 2026-09-22T11:23:19-05:00

graham commented

2026-09-22 11:23:19 -05:00

Audit Rust crate group.

complete() NOTIFY payload now '<run_time>|<valid_time>|' (Elixir side in fix/propagation-pipeline defaults missing source to hrrr — safe in either merge order).
HRDPS step: temp GRIB leak on error paths fixed via Drop guard; decodes from memory instead of re-reading the temp file (removes 2 of 3 blob copies).
Panicked range-download task now fails the fetch instead of returning a partial blob (fetcher.rs + hrdps_fetcher.rs twin).
Chain steps fail with NoCellsScored instead of completing done with zero cells.
NFS idx cache uses unique tmp names; stale doc comments fixed.

Verified: CFLAGS=-O2 cargo clippy --all-targets -- -D warnings clean; cargo test --release 265 pass.

Audit Rust crate group. - complete() NOTIFY payload now '<run_time>|<valid_time>|<source>' (Elixir side in fix/propagation-pipeline defaults missing source to hrrr — safe in either merge order). - HRDPS step: temp GRIB leak on error paths fixed via Drop guard; decodes from memory instead of re-reading the temp file (removes 2 of 3 blob copies). - Panicked range-download task now fails the fetch instead of returning a partial blob (fetcher.rs + hrdps_fetcher.rs twin). - Chain steps fail with NoCellsScored instead of completing done with zero cells. - NFS idx cache uses unique tmp names; stale doc comments fixed. Verified: CFLAGS=-O2 cargo clippy --all-targets -- -D warnings clean; cargo test --release 265 pass.

graham added 1 commit

2026-09-22 11:23:19 -05:00

Harden Rust grid worker: NOTIFY source, temp leaks, partial downloads

skippy-bot/review Skippy review: 1 blocking finding open — see the PR thread

Details

0e9b801150

- complete() NOTIFY payload now carries '<run_time>|<valid_time>|<source>'
  so the Elixir listener can anchor retention on the right source
- HRDPS chain step no longer leaks its temp GRIB on error paths and
  decodes from the in-memory blob instead of re-reading the temp file
- A panicked range-download task now fails the fetch instead of
  returning a partial GRIB blob (fetcher + hrdps_fetcher)
- Chain steps fail with NoCellsScored instead of completing 'done' with
  zero scored cells
- Shared NFS idx cache uses a unique tmp name instead of a deterministic
  one that could collide across pods
- Stale IDLE_SLEEP/NOTIFY doc comments corrected

skippy-bot reviewed

2026-09-22 11:26:05 -05:00

skippy-bot left a comment

🤖 Skippy PR review

1 finding — 1 blocking before merge.

Severity	Location	Issue
🔴 Critical	`rust/prop_grid_rs/src/db.rs:274`	Third payload field breaks the listener on main if this lands first

_{Reviewed 0e9b801150f2. Comment skippy review to re-run.}

### 🤖 Skippy PR review **1 finding** — 1 blocking before merge. | Severity | Location | Issue | | --- | --- | --- | | 🔴 Critical | `rust/prop_grid_rs/src/db.rs:274` | Third payload field breaks the listener on main if this lands first | Reviewed `0e9b801150f2`. Comment `skippy review` to re-run.

rust/prop_grid_rs/src/db.rs Outdated

					
				@ -269,3 +272,3 @@

				    let payload = format!(

				        "{}|{}",

				        "{}|{}|{}",

🔴 Critical — Third payload field breaks the listener on main if this lands first

The new run|valid|source payload is only safe once the Elixir side knows about it. At this head lib/microwaveprop/propagation/notify_listener.ex:71 still does String.split(payload, "|", parts: 2), so "<run>|<valid>|hrrr" splits to ["<run>", "<valid>|hrrr"]; the pattern matches, DateTime.from_iso8601("<valid>|hrrr") returns :error, and the with else branch only logs malformed propagation_ready payload. Net effect on the Rust-first merge order: handle_propagation_ready/2 never runs, so no ScoreCache.broadcast_invalidate/1, no retain_scores_window/1 (score files on NFS stop being pruned), no scalar materialization, no PubSub broadcast — Rust reports every task done while /map silently keeps serving the last pre-deploy cycle.

The PR says this is "safe in either merge order"; that holds in the Elixir-first order only (fix/propagation-pipeline parses [run_time_str, valid_time_str | rest] and defaults source to "hrrr", so it accepts both shapes) — the Rust-first order is the broken one. Land fix/propagation-pipeline first, or gate the new payload behind the listener change. Also stale on the old contract: docs/runbook_propagation_pipeline.md:21 and the runbook table in AGENTS.md both still document '<run_time>|<valid_time>'.

**🔴 Critical — Third payload field breaks the listener on main if this lands first** The new `run|valid|source` payload is only safe once the Elixir side knows about it. At this head `lib/microwaveprop/propagation/notify_listener.ex:71` still does `String.split(payload, "|", parts: 2)`, so `"<run>|<valid>|hrrr"` splits to `["<run>", "<valid>|hrrr"]`; the pattern matches, `DateTime.from_iso8601("<valid>|hrrr")` returns `:error`, and the `with` else branch only logs `malformed propagation_ready payload`. Net effect on the Rust-first merge order: `handle_propagation_ready/2` never runs, so no `ScoreCache.broadcast_invalidate/1`, no `retain_scores_window/1` (score files on NFS stop being pruned), no scalar materialization, no PubSub broadcast — Rust reports every task `done` while `/map` silently keeps serving the last pre-deploy cycle. The PR says this is "safe in either merge order"; that holds in the Elixir-first order only (`fix/propagation-pipeline` parses `[run_time_str, valid_time_str | rest]` and defaults source to `"hrrr"`, so it accepts both shapes) — the Rust-first order is the broken one. Land `fix/propagation-pipeline` first, or gate the new payload behind the listener change. Also stale on the old contract: `docs/runbook_propagation_pipeline.md:21` and the runbook table in `AGENTS.md` both still document `'<run_time>|<valid_time>'`.

skippy-bot marked this conversation as resolved

graham referenced this pull request

2026-09-22 11:41:00 -05:00

Fix propagation pipeline bugs: supervise NotifyListener, harden score reads #18

graham force-pushed fix/rust-worker from 0e9b801150

skippy-bot/review Skippy review: 1 blocking finding open — see the PR thread

Details

to 1bfe12b83e

skippy-bot/review Skippy review: 1 blocking finding open — see the PR thread

Details

2026-09-22 12:04:18 -05:00

Compare

skippy-bot reviewed

2026-09-22 12:06:54 -05:00

skippy-bot left a comment

🤖 Skippy PR review

1 finding — 1 blocking before merge.

Severity	Location	Issue
🟡 Warning	`rust/prop_grid_rs/src/pipeline.rs:330`	TempGrib guard is built after the write, so a failed write still leaks the temp file

Resolved 1 earlier finding: the 3-field payload is safe now that main's NotifyListener parses [run_time, valid_time | rest] and treats a missing source as nil (no retain sweep). Still stale on the old 2-field contract: docs/runbook_propagation_pipeline.md:21 and the AGENTS.md payload line. Branch was force-pushed, so this pass reviewed the full PR diff.

_{Reviewed 1bfe12b83ef9. Comment skippy review to re-run.}

### 🤖 Skippy PR review **1 finding** — 1 blocking before merge. | Severity | Location | Issue | | --- | --- | --- | | 🟡 Warning | `rust/prop_grid_rs/src/pipeline.rs:330` | TempGrib guard is built after the write, so a failed write still leaks the temp file | Resolved 1 earlier finding: the 3-field payload is safe now that main's NotifyListener parses [run_time, valid_time | rest] and treats a missing source as nil (no retain sweep). Still stale on the old 2-field contract: docs/runbook_propagation_pipeline.md:21 and the AGENTS.md payload line. Branch was force-pushed, so this pass reviewed the full PR diff. Reviewed `1bfe12b83ef9`. Comment `skippy review` to re-run.

rust/prop_grid_rs/src/pipeline.rs Outdated

					
				@ -311,3 +329,2 @@

				            let path = tmp_dir.join(format!("hrdps_native_{nanos}_{pid}.grib2"));

				            std::fs::write(&path, &blob_for_decode)?;

				            path

				            std::fs::write(&path, &blob)?;

🟡 Warning — TempGrib guard is built after the write, so a failed write still leaks the temp file

std::fs::write(&path, &blob)? runs before TempGrib(path) exists, and fs::write creates the file before it writes it: if the ~100 MB write dies partway (ENOSPC on the pod's /tmp), the ? returns with no guard in scope and the partial file is stranded. That is the exact leak class this change fixes, and the stranded bytes make the next HRDPS write more likely to fail the same way.

Construct the guard first, then write through it:

let guard = TempGrib(tmp_dir.join(format!("hrdps_native_{nanos}_{pid}.grib2")));
std::fs::write(&guard.0, &blob)?;
guard

(decoder::extract_grid handles the same shape correctly by removing both temp paths after the closure regardless of its result.)

**🟡 Warning — TempGrib guard is built after the write, so a failed write still leaks the temp file** `std::fs::write(&path, &blob)?` runs before `TempGrib(path)` exists, and `fs::write` creates the file before it writes it: if the ~100 MB write dies partway (ENOSPC on the pod's `/tmp`), the `?` returns with no guard in scope and the partial file is stranded. That is the exact leak class this change fixes, and the stranded bytes make the next HRDPS write more likely to fail the same way. Construct the guard first, then write through it: ```rust let guard = TempGrib(tmp_dir.join(format!("hrdps_native_{nanos}_{pid}.grib2"))); std::fs::write(&guard.0, &blob)?; guard ``` (`decoder::extract_grid` handles the same shape correctly by removing both temp paths after the closure regardless of its result.)

graham force-pushed fix/rust-worker from 1bfe12b83e

skippy-bot/review Skippy review: 1 blocking finding open — see the PR thread

Details

to 945209fc42

skippy-bot/review Skippy review: 1 blocking finding open — see the PR thread

Details

2026-09-22 12:54:40 -05:00

Compare

graham commented

2026-09-22 12:54:49 -05:00

Addressed in 945209fc — the critical finding is stale by merge order: fix/propagation-pipeline (#18) merged first, and main's NotifyListener already parses [run_time_str, valid_time_str | rest] with a missing source defaulting to "hrrr" (notify_listener.ex:89-99), so the 3-field payload is safe on current main. The genuinely stale part — docs/runbook_propagation_pipeline.md and AGENTS.md documenting the 2-field contract — is now updated to describe the optional third field.

Addressed in 945209fc — the critical finding is stale by merge order: `fix/propagation-pipeline` (#18) merged first, and main's `NotifyListener` already parses `[run_time_str, valid_time_str | rest]` with a missing source defaulting to `"hrrr"` (notify_listener.ex:89-99), so the 3-field payload is safe on current main. The genuinely stale part — `docs/runbook_propagation_pipeline.md` and `AGENTS.md` documenting the 2-field contract — is now updated to describe the optional third field.

graham merged commit af2ff3b30e into main

2026-09-22 12:55:04 -05:00

graham referenced this pull request from a commit

2026-09-22 12:55:05 -05:00

Merge pull request 'Harden Rust grid worker: NOTIFY source, temp leaks, partial downloads' (#25) from fix/rust-worker into main

skippy-bot reviewed

2026-09-22 12:57:55 -05:00

skippy-bot left a comment

🤖 Skippy PR review

2 findings — none blocking.

Severity	Location	Issue
🔵 Suggestion	`AGENTS.md:146`	Doc says a missing source means `hrrr`; the listener on main does the opposite
🔵 Suggestion	`rust/prop_grid_rs/src/db.rs:277`	This is the change that switches the on-disk retention sweep back on

Resolved 0 of 1 earlier finding: pipeline.rs:330 still runs std::fs::write(&path, &blob)? before TempGrib(path) exists, so a mid-write failure still strands the temp file (that thread stays open). Branch was force-pushed, so this pass reviewed the full PR diff.

_{Reviewed 945209fc4245. Comment skippy review to re-run.}

### 🤖 Skippy PR review **2 findings** — none blocking. | Severity | Location | Issue | | --- | --- | --- | | 🔵 Suggestion | `AGENTS.md:146` | Doc says a missing source means `hrrr`; the listener on main does the opposite | | 🔵 Suggestion | `rust/prop_grid_rs/src/db.rs:277` | This is the change that switches the on-disk retention sweep back on | Resolved 0 of 1 earlier finding: pipeline.rs:330 still runs std::fs::write(&path, &blob)? before TempGrib(path) exists, so a mid-write failure still strands the temp file (that thread stays open). Branch was force-pushed, so this pass reviewed the full PR diff. Reviewed `945209fc4245`. Comment `skippy review` to re-run.

AGENTS.md

					
				@ -144,3 +144,3 @@

				- `GridTaskEnqueuer` inserts are idempotent via the `(run_time, forecast_hour, kind, source)` unique index. Re-seeding a cycle is a no-op, which is why the crons fire multiple times per hour.

				- **Elixir has no fallback compute path.** If Rust is dead, `grid_tasks` backs up and `/map` serves the last good run. See `docs/runbook_propagation_pipeline.md` for the full failure-mode table (FM1–FM4).

				- The NOTIFY payload is `"<run_time_iso>|<valid_time_iso>"` — pipe-delimited, run_time first, because `retain_scores_window/1` anchors retention on run_time.

				- The NOTIFY payload is `"<run_time_iso>|<valid_time_iso>|<source>"` — pipe-delimited, run_time first, because `retain_scores_window/1` anchors retention on run_time. The trailing source (`hrrr`/`hrdps`/`gefs`) is optional on the wire; a missing field is treated as `hrrr` for mixed-version deploys.

🔵 Suggestion — Doc says a missing source means hrrr; the listener on main does the opposite

This line now reads "a missing field is treated as hrrr for mixed-version deploys", but main's listener does the reverse. In lib/microwaveprop/propagation/notify_listener.ex (as merged in #18 on 2026-09-22) source becomes nil when the third field is absent or empty, and kickoff_post_notify_work/3 only calls Propagation.retain_scores_window/1 when source == "hrrr"; its moduledoc states that keeping a missing source as the skip case is the fail-safe direction, because guessing hrrr would let an HRDPS run_time (4-10 h stale) anchor the 48 h window and trim live HRRR f39-f48 files. docs/runbook_propagation_pipeline.md:24-26 repeats the same claim. Either drop the fallback sentence or write it as "a missing source skips the sweep", otherwise the next agent that implements this file's contract reintroduces exactly the deletion hazard the gate was added to prevent.

**🔵 Suggestion — Doc says a missing source means `hrrr`; the listener on main does the opposite** This line now reads "a missing field is treated as `hrrr` for mixed-version deploys", but main's listener does the reverse. In `lib/microwaveprop/propagation/notify_listener.ex` (as merged in #18 on 2026-09-22) `source` becomes `nil` when the third field is absent or empty, and `kickoff_post_notify_work/3` only calls `Propagation.retain_scores_window/1` when `source == "hrrr"`; its moduledoc states that keeping a missing source as the skip case is the fail-safe direction, because guessing `hrrr` would let an HRDPS run_time (4-10 h stale) anchor the 48 h window and trim live HRRR f39-f48 files. `docs/runbook_propagation_pipeline.md:24-26` repeats the same claim. Either drop the fallback sentence or write it as "a missing source skips the sweep", otherwise the next agent that implements this file's contract reintroduces exactly the deletion hazard the gate was added to prevent.

rust/prop_grid_rs/src/db.rs

					
				@ -272,2 +275,3 @@

				        task.run_time.format("%Y-%m-%dT%H:%M:%SZ"),

				        task.valid_time.format("%Y-%m-%dT%H:%M:%SZ")

				        task.valid_time.format("%Y-%m-%dT%H:%M:%SZ"),

				        task.source.as_str()

🔵 Suggestion — This is the change that switches the on-disk retention sweep back on

Adding task.source.as_str() here is not inert on main: NotifyListener gates Propagation.retain_scores_window/1 on an explicit hrrr, so while Rust sent two fields (every Rust build merged to date, including #18 this morning) that sweep never ran. After this merges, every completed HRRR task fires it (~29 per cycle) and it walks three NFS trees, deleting profiles/scalars outside [run_time, run_time+48h] and score files outside 168 h. That matches the intended design, but it is a prod behaviour flip that neither the description nor the CHANGELOG mentions, and the description's "safe in either merge order" only covers payload parsing. Add a line to the description/CHANGELOG so the first post-deploy cycle gets watched instead of assumed, or gate the new field until then.

**🔵 Suggestion — This is the change that switches the on-disk retention sweep back on** Adding `task.source.as_str()` here is not inert on main: `NotifyListener` gates `Propagation.retain_scores_window/1` on an explicit `hrrr`, so while Rust sent two fields (every Rust build merged to date, including #18 this morning) that sweep never ran. After this merges, every completed HRRR task fires it (~29 per cycle) and it walks three NFS trees, deleting profiles/scalars outside `[run_time, run_time+48h]` and score files outside 168 h. That matches the intended design, but it is a prod behaviour flip that neither the description nor the CHANGELOG mentions, and the description's "safe in either merge order" only covers payload parsing. Add a line to the description/CHANGELOG so the first post-deploy cycle gets watched instead of assumed, or gate the new field until then.

Sign in to join this conversation.

No reviewers

No labels

No milestone

No project

No assignees

2 participants

Notifications

Due date

The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set

Reference

graham/prop!25

Rows
Columns