Fix worker error handling, queues, and stale task reclaim #20
Loading…
Reference in a new issue
No description provided.
Delete branch "fix/workers"
Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?
Audit workers/contexts group.
Verified: 169 targeted tests pass; full suite green.
- WeatherFetchWorker returns {:error, reason} so Oban retries failed sounding stores instead of recording success - HrrrNativeGridWorker targets :weather (the :hrrr queue never existed, so native backfill jobs sat available forever) - RepoListener actually stops the broken LISTEN connection on reconnect (Process.exit(pid, :normal) on a non-self pid was a no-op leak) - HrrrPointEnqueuer reclaims hrrr_fetch_tasks rows stuck 'running' after a worker dies mid-batch (15-min stale window, 5-attempt cap) - CanadianSoundingFetchWorker/StationElevationWorker/radar workers no longer mark work complete on failed upserts; failures are logged - GefsFetchWorker logs seed_extended_horizon insert failures - IemreFetchWorker uniqueness covers all incomplete states - UserHomeQthLookupWorker moves to :admin so it runs on hot pods (:weather is paused there)🤖 Skippy PR review
2 findings — 1 blocking before merge.
lib/microwaveprop/weather/hrrr_point_enqueuer.ex:156failed, which the backfill reads as "give up on these contacts"lib/microwaveprop/workers/station_elevation_worker.ex:47Reviewed
b01ddb955bce. Commentskippy reviewto re-run.@ -109,0 +153,4 @@)|> Repo.update_all(set: [status: "failed",🟠 High — Reclaim writes
failed, which the backfill reads as "give up on these contacts"BackfillEnqueueWorker.reconcile_failed_hrrr_task_contacts/1(lib/microwaveprop/workers/backfill_enqueue_worker.ex:168) treats everyhrrr_fetch_tasksrow withstatus = 'failed'as a permanently unreachablevalid_time: it flips every contact in that hour fromhrrr_status: :queuedto:unavailableand logsmarked N hrrr-queued contacts as unavailable (upstream archive gap). Until nowfailedwas only written by Rust'sfail_hrrr_task/3, i.e. after the worker itself reported a terminal batch error (wgrib2 decode,idx404). This new flip writes the same value for a row whose only fault is that hrrr-point-rs died after 5 claims (OOM, node drain, rolling deploy), which is precisely the transient case the reclaim exists to recover, so a crash-looping hrrr-point-rs now turns recoverable enrichments into "no data" for other users' contacts and waits out the 24 h@unavailable_reprobe_cooldown_secondsbefore trying again. It also does not retire a row the way the docstring claims: the nextenqueuefor thatvalid_timehits theon_conflictCASE (status IN ('done','failed') THEN 'queued',attempt THEN 0), so the row is resurrected with a fresh 5-attempt budget while the contacts it feeds sit:unavailableand no longer select for re-enqueue. Suggest not overloadingfailed- make the reconcile skip reclaim rows (t.error NOT LIKE 'reclaim-orphan:%'), or give the reclaim its own state sofailedkeeps meaning "the Rust worker gave up".@ -45,2 +45,3 @@{:error, _} ->{:error, reason} ->Logger.warning(🔵 Suggestion — Failed elevation upsert still reports job success
The PR description lists StationElevation among the workers that "no longer mark complete on failed upserts", but the upsert failure branch above this hunk (lines 42-44) still returns
:ok-Logger.warning/1returns:ok, so a failedRepo.updatereaches Oban as a successful job, is never retried, and the station silently keepselevation_m: nil. The added SRTM log line is a genuine improvement; this branch is the missing half of the stated fix. Return{:error, changeset}there (the SRTM lookup branch can stay:ok, which the moduledoc already documents as non-fatal).b01ddb955bcfb90af535- reconcile_failed_hrrr_task_contacts now excludes rows whose error is 'reclaim-orphan:*' — a stale-running reclaim after the point worker dies mid-batch is transient, not an unreachable valid_time, so its contacts stay :queued instead of being flipped :unavailable - StationElevationWorker returns {:error, changeset} on a failed elevation upsert so Oban retries instead of recording successBoth findings addressed in
62d45a20:Reclaim writes
failed—reconcile_failed_hrrr_task_contactsnow excludes rows whose error isreclaim-orphan:*(NULL error still counts as a real failure). A worker death is transient; the next enqueue for that valid_time resets the row toqueued, so contacts stay:queuedinstead of being flipped:unavailable. New test covers the exclusion.Failed elevation upsert —
StationElevationWorkernow returns{:error, changeset}on theRepo.updatefailure branch so Oban retries; the SRTM-lookup branch stays:okper the moduledoc.skippy review
Resolved 2 of 2 earlier findings —
reclaim_stale_running/1now tags its rowsreclaim-orphan:andreconcile_failed_hrrr_task_contacts/1excludes them (so a dead point worker no longer reads as an archive gap);StationElevationWorkerreturns{:error, changeset}instead of an:okfrom the logger.Nothing new in cfb90af..62d45a2.
62d45a20c360f1a68f93🤖 Skippy PR review
2 findings — 1 blocking before merge.
lib/microwaveprop/weather/hrrr_point_enqueuer.ex:55lib/microwaveprop/weather/hrrr_point_enqueuer.ex:38claimed_at, which is never refreshed during a batchBranch was force-pushed (
b01ddb95is no longer in the PR history), so this pass re-read the full diff. Resolved both earlier findings:reclaim-orphan:rows are now excluded from the backfill reconcile, and failed elevation upserts are returned as errors. Nothing from the previous pass is still open.Reviewed
62d45a20c3ad. Commentskippy reviewto re-run.@ -30,0 +35,4 @@# and the unique index on `valid_time` blocks any replacement.# Mirrors `Propagation.GridTaskEnqueuer`'s cutoff: 15 min comfortably# exceeds a healthy batch's wall time.@stale_running_cutoff_seconds 15 * 60🔵 Suggestion — Cutoff keys off
claimed_at, which is never refreshed during a batchclaim_next_hrrr_tasksetsclaimed_at = NOW()exactly once (rust/prop_grid_rs/src/db.rs:376) and nothing touches it again until the terminal update, so a batch whose wall time exceeds 15 minutes (busy hour with a large point union, loaded node, slow S3) is stillrunningwith an ancientclaimed_atwhen the next sweep reclaims it. A second worker then claims the same row via SKIP LOCKED and re-downloads/decodes that hour concurrently with the first, and each reclaim cycle pushesattempttoward@max_reclaim_attempts. Either have the Rust side touchclaimed_atafter each extracted point (a cheap heartbeat), or set the cutoff above any realistic batch wall time so the two claims cannot overlap.@ -36,6 +52,8 @@ defmodule Microwaveprop.Weather.HrrrPointEnqueuer do"""@spec enqueue(%{DateTime.t() => [{float(), float()}]}) :: {:ok, non_neg_integer()} | {:error, term()}def enqueue(groups) when is_map(groups) do_ = reclaim_stale_running()🟡 Warning — Reclaim runs once per contact enqueue, not once per sweep
enqueue/1now callsreclaim_stale_running/1, but the per-contact path reachesenqueue/1once per contact:ContactWeatherEnqueueWorker.enqueue_for_contact/2(line 147) callsHrrrPointEnqueuer.enqueue_for_contacts([contact]), andBackfillEnqueueWorkerdrives that withEnum.each(contacts, ...)under the*/30 * * * *cron atlimit: 2000. A sweep over a fresh backlog therefore fires up to 2000 reclaim transactions and 4000UPDATE hrrr_fetch_tasksstatements, each filtered onstatus = 'running' and claimed_at < ?- and the only status index cannot serve that predicate, becausehrrr_fetch_tasks_queued_idxis partial (WHERE status = 'queued'). Every one of those statements seq-scans a table with no prune path that grows one row per enriched hour.GridTaskEnqueuercalls its reclaim once per seed cycle (PropagationGridWorker,HrdpsGridWorker), not per row, which is the pattern worth keeping here. Fix: hoist the call out ofenqueue/1to the sweep entry points, or addcreate index(:hrrr_fetch_tasks, [:claimed_at], where: "status = 'running'")if per-contact reclaim is deliberate.🤖 Skippy PR review
4 findings — 1 blocking before merge.
lib/microwaveprop/weather/hrrr_point_enqueuer.ex:55lib/microwaveprop/weather/hrrr_point_enqueuer.ex:38claimed_at, which is never refreshed during a batchlib/microwaveprop/workers/gefs_fetch_worker.ex:88lib/microwaveprop/workers/hrrr_native_grid_worker.ex:20Branch was force-pushed (prev head
62d45a2is no longer in history), so this re-read the full diff. Both earlier findings are resolved at60f1a68(reclaim rows tagged reclaim-orphan: and excluded from the archive-gap reconcile; StationElevationWorker returns {:error, changeset}) and were not reposted.Reviewed
60f1a68f93f5. Commentskippy reviewto re-run.@ -30,0 +35,4 @@# and the unique index on `valid_time` blocks any replacement.# Mirrors `Propagation.GridTaskEnqueuer`'s cutoff: 15 min comfortably# exceeds a healthy batch's wall time.@stale_running_cutoff_seconds 15 * 60🔵 Suggestion — Cutoff keys off
claimed_at, which is never refreshed during a batchclaim_next_hrrr_taskstampsclaimed_at = NOW()exactly once (rust/prop_grid_rs/src/db.rs:376) and nothing touches it again until the terminal update, so a batch whose wall time exceeds this 15 minute cutoff (large point union, loaded node, slow S3) is stillrunningwith an ancientclaimed_atwhen the next sweep reclaims it. Another replica can then claim the same row through SKIP LOCKED and re-download plus re-decode that hour concurrently with the first, and every reclaim cycle also pushesattempttoward@max_reclaim_attempts. Either have the Rust side touchclaimed_atafter each extracted point (a cheap heartbeat), or set the cutoff above the worst realistic batch wall time.@ -36,6 +52,8 @@ defmodule Microwaveprop.Weather.HrrrPointEnqueuer do"""@spec enqueue(%{DateTime.t() => [{float(), float()}]}) :: {:ok, non_neg_integer()} | {:error, term()}def enqueue(groups) when is_map(groups) do_ = reclaim_stale_running()🟡 Warning — Reclaim sweeps the table once per contact, not once per pass
enqueue/1is also the per-contact entry point, so this new sweep runs thousands of times per pass, not once:ContactWeatherEnqueueWorker.enqueue_for_contact/2callsHrrrPointEnqueuer.enqueue_for_contacts([contact])per contact (line 147), andBackfillEnqueueWorkerdrives that withEnum.each/2atlimit: 2000under the*/30 * * * *cron. A fresh backlog sweep is therefore up to 2000 reclaim transactions and 4000UPDATE hrrr_fetch_tasksstatements. Neither update can use an index, because the only status index is partial (hrrr_fetch_tasks_queued_idx,where: "status = 'queued'", migration 20260419231502), sostatus = 'running' and claimed_at < ?seq-scans a table nothing prunes (it grows a row per enriched hour; thecompleted_atindex has no deleting job behind it). This repo already has the right shape next door:GridTaskEnqueuer.reclaim_stale_running/1is called once per seed cycle (PropagationGridWorker:46,HrdpsGridWorker:56). Hoist the call to the pass entry points (BackfillEnqueueWorker, the PSKR sampler, the contact retry button), or addcreate index(:hrrr_fetch_tasks, [:claimed_at], where: "status = 'running'")if per-call reclaim is deliberate.@ -87,0 +85,4 @@new(%{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => fh})end)inserted = length(Oban.insert_all(jobs))🔵 Suggestion — Seed-failure guard can never fire: insert_all returns conflicts too
inserted < length(hours)is dead code. On the Smart engineOban.insert_all/1returns one entry per input changeset - a job rejected by the unique key comes back withconflict?: true(Oban.Pro.Engines.Smart.apply_conflicts/1returnsold_jobs ++ new_jobs; onlyon_conflict: :skipdrops conflicting entries from the result). Soinserted == length(hours)on every call and a partial seed stays as silent as it was before this change, which is the one signal this hunk was meant to add. Count the real inserts instead:Oban.insert_all(jobs, on_conflict: :skip)(the result then holds only newly inserted jobs) orEnum.count(result, &(not &1.conflict?)).@ -17,3 +18,3 @@use Oban.Pro.Worker,queue: :hrrr,queue: :weather,🔵 Suggestion — Retarget does not revive hours already enqueued onto :hrrr
Nothing serves the retired
:hrrrqueue, so jobs inserted before this change sit:availableforever - and because this worker is unique on{year, month, day, hour}withperiod: :infinityandstates: :incomplete, those rows keep holding the unique key.mix hrrr_native_backfillandrelease.native_backfillwill therefore reportskippedfor exactly those hours instead of re-inserting them, so after deploy only hours never enqueued before will actually run. Worth a one-off check (oban_jobswherequeue = 'hrrr'andstate in ('available','scheduled')) and a cancel, so the retarget can backfill the hours it was meant to unstick.