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

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

graham commented

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

Fixes from the 2026-09-22 audit (tmp/bugs.md), propagation core group.

NotifyListener was never supervised — ScoreCache invalidation, retain_scores_window, and the propagation:updated PubSub for /map and /status never ran since the Rust cutover. Now supervised with auto_reconnect; retention anchored only on hrrr NOTIFY payloads (new 3-field payload, backward compatible); the NFS sweep runs in a detached task.
score_grid_point KeyError on pgrid profiles missing dewpoint/pressure keys (map click / API crash).
point_detail with nil valid_time returned the furthest-future hour; now defaults to earliest like scores_at/3.
.tmp.* sweep gains a 10-min age check so it can't delete in-flight Rust atomic writes.
grid_tasks reseed no longer requeues rows past the reclaim-attempt ceiling.
PipelineStatus running detection reads grid_tasks instead of the millisecond-fast Oban seeder.
Pgrid.read_point fd leak on bad header; to_profile O(n^2) -> O(n).
Path-averaged wind pairs u/v per profile (was mis-pairing across cells), fixed in both copies.
Recalibrator no longer fabricates wind/sky cover for missing inputs.

Verified: full suite 5000 tests pass on the combined tree; targeted tests green.

Fixes from the 2026-09-22 audit (tmp/bugs.md), propagation core group. - **NotifyListener was never supervised** — ScoreCache invalidation, retain_scores_window, and the propagation:updated PubSub for /map and /status never ran since the Rust cutover. Now supervised with auto_reconnect; retention anchored only on hrrr NOTIFY payloads (new 3-field payload, backward compatible); the NFS sweep runs in a detached task. - score_grid_point KeyError on pgrid profiles missing dewpoint/pressure keys (map click / API crash). - point_detail with nil valid_time returned the furthest-future hour; now defaults to earliest like scores_at/3. - .tmp.* sweep gains a 10-min age check so it can't delete in-flight Rust atomic writes. - grid_tasks reseed no longer requeues rows past the reclaim-attempt ceiling. - PipelineStatus running detection reads grid_tasks instead of the millisecond-fast Oban seeder. - Pgrid.read_point fd leak on bad header; to_profile O(n^2) -> O(n). - Path-averaged wind pairs u/v per profile (was mis-pairing across cells), fixed in both copies. - Recalibrator no longer fabricates wind/sky cover for missing inputs. Verified: full suite 5000 tests pass on the combined tree; targeted tests green.

graham added 1 commit

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

Fix propagation pipeline bugs: supervise NotifyListener, harden score reads

skippy-bot/review Skippy review: 2 blocking findings open — see the PR thread

Details

233850c266

- Supervise NotifyListener (was never started since the Rust cutover) with
  auto_reconnect so ScoreCache invalidation, retain_scores_window, and the
  propagation:updated PubSub actually run; retention now only anchors on
  hrrr NOTIFY payloads and runs in a detached task
- score_grid_point no longer raises KeyError on pgrid profiles missing
  dewpoint/pressure keys
- point_detail with nil valid_time defaults to the earliest (current) hour
  instead of the furthest-future
- .tmp.* sweep only removes files older than 10 min (in-flight writes safe)
- grid_tasks reseed no longer requeues rows past the reclaim-attempt ceiling
- PipelineStatus running detection reads grid_tasks, not the ms-fast seeder
- Pgrid.read_point closes its fd on bad headers; to_profile is O(n) not O(n^2)
- Path-averaged wind pairs u/v per profile instead of across cells
- Recalibrator no longer fabricates wind/sky cover for missing inputs

skippy-bot reviewed

2026-09-22 11:31:18 -05:00

skippy-bot left a comment

🤖 Skippy PR review

2 findings — 2 blocking before merge.

Severity	Location	Issue
🟠 High	`lib/microwaveprop/propagation/notify_listener.ex:88`	The hrrr-only retain gate is inert: Rust never sends the source field
🟠 High	`lib/microwaveprop/propagation/notify_listener.ex:109`	Raw Repo config has no connection details in prod, so the listener never subscribes

_{Reviewed 233850c266df. Comment skippy review to re-run.}

### 🤖 Skippy PR review **2 findings** — 2 blocking before merge. | Severity | Location | Issue | | --- | --- | --- | | 🟠 High | `lib/microwaveprop/propagation/notify_listener.ex:88` | The hrrr-only retain gate is inert: Rust never sends the source field | | 🟠 High | `lib/microwaveprop/propagation/notify_listener.ex:109` | Raw Repo config has no connection details in prod, so the listener never subscribes | <sub>Reviewed `233850c266df`. Comment `skippy review` to re-run.</sub>

lib/microwaveprop/propagation/notify_listener.ex Outdated

					
				@ -72,3 +86,3 @@

				         {:ok, run_time, _} <- DateTime.from_iso8601(run_time_str),

				         {:ok, valid_time, _} <- DateTime.from_iso8601(valid_time_str) do

				      handle_propagation_ready(run_time, valid_time)

				      source = List.first(rest) || "hrrr"

🟠 High — The hrrr-only retain gate is inert: Rust never sends the source field

rust/prop_grid_rs/src/db.rs complete() still formats the payload as "{}|{}" (run_time, valid_time) with no source, and git diff main...HEAD -- rust/ is empty, so rest is always [], source is always the "hrrr" fallback, and the sweep at line 194 still runs for every HRDPS completion exactly as before. The damage the PR set out to stop is therefore untouched: HrdpsGridWorker.pick_run_time/1 seeds run_time = now - 4h (or now - 10h), so retain_scores_window/1 anchors the 48 h profile/scalar window 4-10 h behind wall clock and deletes the live HRRR f39-f48 files.

The default is also the destructive direction, so this re-arms itself any time the field goes missing (and List.first(rest) || "hrrr" maps a present-but-empty field such as a|b| to "", which skips retention and leaks disk instead). Fix both ends: bind task.source into the payload in db.rs complete() (ClaimedTask already carries it), and until that ships make the absent/unknown case fail safe by skipping the sweep rather than defaulting to delete (the 15 min PropagationPruneWorker still bounds disk).

The new test feeds a 3-field payload the producer never emits, so the suite cannot catch this.

**🟠 High — The hrrr-only retain gate is inert: Rust never sends the source field** `rust/prop_grid_rs/src/db.rs` `complete()` still formats the payload as `"{}|{}"` (run_time, valid_time) with no source, and `git diff main...HEAD -- rust/` is empty, so `rest` is always `[]`, `source` is always the `"hrrr"` fallback, and the sweep at line 194 still runs for every HRDPS completion exactly as before. The damage the PR set out to stop is therefore untouched: `HrdpsGridWorker.pick_run_time/1` seeds `run_time = now - 4h` (or `now - 10h`), so `retain_scores_window/1` anchors the 48 h profile/scalar window 4-10 h behind wall clock and deletes the live HRRR f39-f48 files. The default is also the destructive direction, so this re-arms itself any time the field goes missing (and `List.first(rest) || "hrrr"` maps a present-but-empty field such as `a|b|` to `""`, which skips retention and leaks disk instead). Fix both ends: bind `task.source` into the payload in `db.rs complete()` (`ClaimedTask` already carries it), and until that ships make the absent/unknown case fail safe by skipping the sweep rather than defaulting to delete (the 15 min `PropagationPruneWorker` still bounds disk). The new test feeds a 3-field payload the producer never emits, so the suite cannot catch this.

skippy-bot marked this conversation as resolved

lib/microwaveprop/propagation/notify_listener.ex Outdated

					
				@ -89,0 +106,4 @@

				    # before the :DOWN/retry path below could ever run. On reconnect

				    # Postgrex.Notifications re-issues LISTEN for every registered

				    # channel, so no resubscribe is needed here.

				    opts = Keyword.merge(config, auto_reconnect: true, reconnect_backoff: 5_000)

🟠 High — Raw Repo config has no connection details in prod, so the listener never subscribes

Application.fetch_env!(:microwaveprop, Repo) (line 101) returns the unparsed config. In prod that is only [url: DATABASE_URL, pool_size: 12, socket_options: ...] (config/runtime.exs:465) because Ecto itself turns :url into username/password/database/hostname (Ecto.Repo.Supervisor.init_config/4) and never writes the result back to the app env, while Postgrex has no :url handling at all (the string url does not appear anywhere in postgrex 0.22.4's lib/).

So Postgrex.Notifications gets none of the connection keys and falls back to the process-env defaults from Postgrex.Utils.default_opts/1: localhost:5432, no database, no password. Nothing in k8s/deployment.yaml or the secret bundle sets PGHOST/PGUSER/PGPASSWORD/PGDATABASE, so that connect fails, SimpleConnection defaults sync_connect: true, start_link returns {:error, _}, and this GenServer logs "failed to subscribe" and retries every 30 s forever: the listener still never gets a LISTEN in production, which is the bug this PR exists to fix (and dev only works because config/dev.exs sets username/password/hostname/database directly).

Use Microwaveprop.Repo.config() instead, which is what RepoListener already does to get the URL-parsed config, or pass hostname/port/username/password/database explicitly.

**🟠 High — Raw Repo config has no connection details in prod, so the listener never subscribes** `Application.fetch_env!(:microwaveprop, Repo)` (line 101) returns the *unparsed* config. In prod that is only `[url: DATABASE_URL, pool_size: 12, socket_options: ...]` (config/runtime.exs:465) because Ecto itself turns `:url` into username/password/database/hostname (`Ecto.Repo.Supervisor.init_config/4`) and never writes the result back to the app env, while Postgrex has no `:url` handling at all (the string `url` does not appear anywhere in postgrex 0.22.4's `lib/`). So `Postgrex.Notifications` gets none of the connection keys and falls back to the process-env defaults from `Postgrex.Utils.default_opts/1`: localhost:5432, no database, no password. Nothing in `k8s/deployment.yaml` or the secret bundle sets `PGHOST`/`PGUSER`/`PGPASSWORD`/`PGDATABASE`, so that connect fails, `SimpleConnection` defaults `sync_connect: true`, `start_link` returns `{:error, _}`, and this GenServer logs "failed to subscribe" and retries every 30 s forever: the listener still never gets a LISTEN in production, which is the bug this PR exists to fix (and dev only works because config/dev.exs sets username/password/hostname/database directly). Use `Microwaveprop.Repo.config()` instead, which is what `RepoListener` already does to get the URL-parsed config, or pass hostname/port/username/password/database explicitly.

skippy-bot marked this conversation as resolved

graham commented

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

Both findings addressed in 9336e7e3:

Inert source gate — missing/empty source now skips the retain sweep entirely (fail-safe; PropagationPruneWorker still bounds disk). The emit side (db.rs complete() → run_time|valid_time|source) is in #25; until it merges, retention simply does not run rather than guessing wrong. Tests updated: 2-field payload asserts the file survives, 3-field hrrr asserts the sweep runs.
Raw Repo config — now uses Repo.config() (URL-parsed) via RepoListener.start_notifications/2, which also carries the direct-5432 pgbouncer bypass. extra_opts param added for auto_reconnect.

Both findings addressed in 9336e7e3: 1. **Inert source gate** — missing/empty source now skips the retain sweep entirely (fail-safe; PropagationPruneWorker still bounds disk). The emit side (`db.rs` `complete()` → `run_time|valid_time|source`) is in #25; until it merges, retention simply does not run rather than guessing wrong. Tests updated: 2-field payload asserts the file survives, 3-field `hrrr` asserts the sweep runs. 2. **Raw Repo config** — now uses `Repo.config()` (URL-parsed) via `RepoListener.start_notifications/2`, which also carries the direct-5432 pgbouncer bypass. `extra_opts` param added for `auto_reconnect`.

graham added 1 commit

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

Address review: fail-safe source default, URL-parsed notifications config

skippy-bot/review Skippy review: clean — no open findings

Details

9336e7e3d7

- Missing/empty source field now skips the retain sweep instead of
  defaulting to 'hrrr' — the destructive direction. The 15-min
  PropagationPruneWorker still bounds disk until the Rust side ships
  the source field (fix/rust-worker).
- start_notifications_conn now uses Repo.config() (URL-parsed prod
  config) and shares RepoListener's direct-5432/pgbouncer-bypass
  connect, which also gained an extra_opts param for auto_reconnect.
  Raw app env only carries url: DATABASE_URL in prod, which Postgrex
  cannot read — the listener would have retried localhost:5432 forever.

graham commented

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

skippy review

skippy-bot commented

2026-09-22 11:42:47 -05:00

First-time contributor

Resolved 2 of 2 earlier findings (both on notify_listener.ex):

Absent/empty source no longer defaults to "hrrr" — a missing field now yields nil and skips the retain sweep, so a stale HRDPS run_time can't delete live HRRR f39-f48 files, and a present-but-empty field no longer silently skips into a leak.
The notifications connection now reads Repo.config() (URL-parsed hostname/username/password/database) through the shared RepoListener.start_notifications/2, so prod no longer hands Postgrex a bare url: config.

Nothing new in 233850c266df..9336e7e3: the new source branch, the extra_opts merge in do_start_notifications/4, and the three payload tests all read correctly.

Not blocking, worth knowing: rust/prop_grid_rs/src/db.rs still emits the two-field payload, so the sweep stays dormant (disk stays bounded by the 15-minute PropagationPruneWorker, which prunes the same scores/profiles/scalars trees with a 3 h cutoff) — and the "retains GEFS score files through 168 hours" test now passes vacuously, since it calls handle_propagation_ready/2 with the nil default.

**Resolved 2 of 2 earlier findings** (both on `notify_listener.ex`): - Absent/empty source no longer defaults to `"hrrr"` — a missing field now yields `nil` and skips the retain sweep, so a stale HRDPS `run_time` can't delete live HRRR f39-f48 files, and a present-but-empty field no longer silently skips into a leak. - The notifications connection now reads `Repo.config()` (URL-parsed hostname/username/password/database) through the shared `RepoListener.start_notifications/2`, so prod no longer hands Postgrex a bare `url:` config. Nothing new in `233850c266df..9336e7e3`: the new source branch, the `extra_opts` merge in `do_start_notifications/4`, and the three payload tests all read correctly. Not blocking, worth knowing: `rust/prop_grid_rs/src/db.rs` still emits the two-field payload, so the sweep stays dormant (disk stays bounded by the 15-minute `PropagationPruneWorker`, which prunes the same scores/profiles/scalars trees with a 3 h cutoff) — and the "retains GEFS score files through 168 hours" test now passes vacuously, since it calls `handle_propagation_ready/2` with the nil default.

skippy-bot commented

2026-09-22 11:43:28 -05:00

First-time contributor

Resolved 2 of 2 earlier findings at 9336e7e3:

Raw Repo config (notify_listener.ex:116): now Repo.config(), which Ecto 3.14.2 parses via init_config/4 + parse_url/1 into hostname/username/password/database, so the prod listener gets a real connection instead of Postgrex defaults.
Inert hrrr retain gate (notify_listener.ex:93): absent/empty source now yields nil and skips the sweep, so a stale HRDPS run_time can no longer trim live HRRR f39-f48 files. Disk stays bounded by the 15-minute PropagationPruneWorker.

Nothing new in 233850c2..9336e7e3.

One residual, non-blocking: rust/prop_grid_rs/src/db.rs:complete/2 still formats the payload as "{}|{}", so source is never present and retain_scores_window/1 is now dormant in production (only the past-dated prune runs, leaving GEFS/future files to their producer horizons). Bind task.source into the payload when the Rust side is next touched.

Resolved 2 of 2 earlier findings at `9336e7e3`: - **Raw `Repo` config** (`notify_listener.ex:116`): now `Repo.config()`, which Ecto 3.14.2 parses via `init_config/4` + `parse_url/1` into hostname/username/password/database, so the prod listener gets a real connection instead of Postgrex defaults. - **Inert hrrr retain gate** (`notify_listener.ex:93`): absent/empty source now yields `nil` and skips the sweep, so a stale HRDPS `run_time` can no longer trim live HRRR f39-f48 files. Disk stays bounded by the 15-minute `PropagationPruneWorker`. Nothing new in `233850c2..9336e7e3`. One residual, non-blocking: `rust/prop_grid_rs/src/db.rs:complete/2` still formats the payload as `"{}|{}"`, so `source` is never present and `retain_scores_window/1` is now dormant in production (only the past-dated prune runs, leaving GEFS/future files to their producer horizons). Bind `task.source` into the payload when the Rust side is next touched.

graham merged commit 7d264c49b8 into main

2026-09-22 11:59:16 -05:00

graham deleted branch fix/propagation-pipeline

2026-09-22 11:59:16 -05:00

graham referenced this pull request from a commit

2026-09-22 11:59:17 -05:00

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

graham referenced this pull request

2026-09-22 12:01:17 -05:00

Fix weather client file handling and lookup bugs #19

skippy-bot referenced this pull request

2026-09-22 12:08:30 -05:00