fix(workers): propagate transient errors so Oban retries + counters fire
CommonVolumeRadarWorker and MechanismClassifyWorker were both
returning :ok on upstream failures, so Prometheus job-failure counters
stayed at zero during multi-hour NEXRAD outages and classifier bugs
went unnoticed.
- CommonVolumeRadarWorker: distinguish permanent (4xx) from transient
errors using the same pattern as RadarFrameWorker. Permanent errors
still mark :unavailable + :ok so the backfill stops re-queueing them.
Transient errors return {:error, reason} and leave radar_status alone
so the next retry can succeed.
- MechanismClassifyWorker: keep the row-status :failed update so
operators can see which contacts blew up, but return
{:error, inspect(e)} from the rescue instead of swallowing it as :ok.
This commit is contained in:
parent
f9e0331a2c
commit
7ab426c01d
4 changed files with 76 additions and 5 deletions
|
|
@ -86,12 +86,33 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorker do
|
||||||
:ok
|
:ok
|
||||||
|
|
||||||
{:error, reason} ->
|
{:error, reason} ->
|
||||||
Logger.info("CommonVolumeRadarWorker: no frame for #{contact.id}: #{inspect(reason)}")
|
if permanent_error?(reason) do
|
||||||
mark_unavailable(contact)
|
Logger.info("CommonVolumeRadarWorker: no frame for #{contact.id}: #{inspect(reason)}")
|
||||||
:ok
|
mark_unavailable(contact)
|
||||||
|
:ok
|
||||||
|
else
|
||||||
|
# Transient (5xx, timeout, connrefused) — return {:error, _} so
|
||||||
|
# Oban retries and the failure is reflected in the job-failure
|
||||||
|
# counters. Do NOT pin the contact :unavailable; it must remain
|
||||||
|
# eligible for the next attempt.
|
||||||
|
Logger.warning("CommonVolumeRadarWorker: transient error for #{contact.id}: #{inspect(reason)}")
|
||||||
|
{:error, reason}
|
||||||
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
# Permanent: 4xx means the IEM archive genuinely has no frame at this
|
||||||
|
# timestamp, so retrying is pointless. Everything else (5xx, transport
|
||||||
|
# errors, decode failures) is treated as transient.
|
||||||
|
defp permanent_error?("NEXRAD n0q HTTP " <> code) do
|
||||||
|
case Integer.parse(code) do
|
||||||
|
{status, _} when status in 400..499 -> true
|
||||||
|
_ -> false
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp permanent_error?(_), do: false
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Aggregate n0q pixel statistics over the common volume of the two
|
Aggregate n0q pixel statistics over the common volume of the two
|
||||||
endpoints. Pure function — accepts raw pixel buffer + width so it
|
endpoints. Pure function — accepts raw pixel buffer + width so it
|
||||||
|
|
|
||||||
|
|
@ -66,8 +66,11 @@ defmodule Microwaveprop.Workers.MechanismClassifyWorker do
|
||||||
e ->
|
e ->
|
||||||
Logger.warning("MechanismClassifyWorker: failed for contact #{contact.id}: #{inspect(e)}")
|
Logger.warning("MechanismClassifyWorker: failed for contact #{contact.id}: #{inspect(e)}")
|
||||||
|
|
||||||
|
# Persist row state so operators can see which contacts blew up, but
|
||||||
|
# return {:error, _} to Oban so the failure registers in job-failure
|
||||||
|
# counters and the job is retried rather than silently swallowed.
|
||||||
mark_status(contact, :failed, nil, nil)
|
mark_status(contact, :failed, nil, nil)
|
||||||
:ok
|
{:error, inspect(e)}
|
||||||
end
|
end
|
||||||
|
|
||||||
defp build_inputs(%Contact{} = contact) do
|
defp build_inputs(%Contact{} = contact) do
|
||||||
|
|
|
||||||
|
|
@ -5,6 +5,7 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorkerTest do
|
||||||
alias Microwaveprop.Radio.Contact
|
alias Microwaveprop.Radio.Contact
|
||||||
alias Microwaveprop.Radio.ContactCommonVolumeRadar
|
alias Microwaveprop.Radio.ContactCommonVolumeRadar
|
||||||
alias Microwaveprop.Repo
|
alias Microwaveprop.Repo
|
||||||
|
alias Microwaveprop.Weather.NexradClient
|
||||||
alias Microwaveprop.Workers.CommonVolumeRadarWorker
|
alias Microwaveprop.Workers.CommonVolumeRadarWorker
|
||||||
|
|
||||||
setup do
|
setup do
|
||||||
|
|
@ -38,7 +39,7 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorkerTest do
|
||||||
test "marks radar_status :unavailable and does not insert a row when n0q frame is 404" do
|
test "marks radar_status :unavailable and does not insert a row when n0q frame is 404" do
|
||||||
contact = insert_contact()
|
contact = insert_contact()
|
||||||
|
|
||||||
Req.Test.stub(Microwaveprop.Weather.NexradClient, fn conn ->
|
Req.Test.stub(NexradClient, fn conn ->
|
||||||
Plug.Conn.send_resp(conn, 404, "not found")
|
Plug.Conn.send_resp(conn, 404, "not found")
|
||||||
end)
|
end)
|
||||||
|
|
||||||
|
|
@ -59,6 +60,36 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorkerTest do
|
||||||
assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, contact.id)
|
assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, contact.id)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
test "returns {:error, _} on transient 5xx so Oban retries and status stays :queued" do
|
||||||
|
contact = insert_contact()
|
||||||
|
|
||||||
|
Req.Test.stub(NexradClient, fn conn ->
|
||||||
|
Plug.Conn.send_resp(conn, 503, "service unavailable")
|
||||||
|
end)
|
||||||
|
|
||||||
|
assert {:error, _reason} =
|
||||||
|
perform_job(CommonVolumeRadarWorker, %{"contact_id" => contact.id})
|
||||||
|
|
||||||
|
refute Repo.get_by(ContactCommonVolumeRadar, contact_id: contact.id)
|
||||||
|
# Transient errors must not pin the contact as :unavailable — it needs
|
||||||
|
# to remain eligible for a retry.
|
||||||
|
refute Repo.get!(Contact, contact.id).radar_status == :unavailable
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns {:error, _} on transport-level failures (connrefused, timeout)" do
|
||||||
|
contact = insert_contact()
|
||||||
|
|
||||||
|
Req.Test.stub(NexradClient, fn conn ->
|
||||||
|
Req.Test.transport_error(conn, :econnrefused)
|
||||||
|
end)
|
||||||
|
|
||||||
|
assert {:error, _reason} =
|
||||||
|
perform_job(CommonVolumeRadarWorker, %{"contact_id" => contact.id})
|
||||||
|
|
||||||
|
refute Repo.get_by(ContactCommonVolumeRadar, contact_id: contact.id)
|
||||||
|
refute Repo.get!(Contact, contact.id).radar_status == :unavailable
|
||||||
|
end
|
||||||
|
|
||||||
test "skips contacts beyond 2R so we don't waste a fetch" do
|
test "skips contacts beyond 2R so we don't waste a fetch" do
|
||||||
# 2R = 800 km. Put the stations 2000 km apart.
|
# 2R = 800 km. Put the stations 2000 km apart.
|
||||||
contact =
|
contact =
|
||||||
|
|
|
||||||
|
|
@ -138,6 +138,22 @@ defmodule Microwaveprop.Workers.MechanismClassifyWorkerTest do
|
||||||
})
|
})
|
||||||
end
|
end
|
||||||
|
|
||||||
|
test "propagates errors from classifier so Oban counts failures + retries, but still marks row :failed" do
|
||||||
|
# Force the classifier to blow up by using a non-integer band —
|
||||||
|
# `Decimal.to_integer/1` in build_inputs raises on fractional values.
|
||||||
|
# This models the "unexpected classifier explosion" that silently
|
||||||
|
# dropped failure counts in prod.
|
||||||
|
contact = create_contact(%{band: Decimal.new("10.5")})
|
||||||
|
|
||||||
|
assert {:error, _reason} =
|
||||||
|
MechanismClassifyWorker.perform(%Oban.Job{
|
||||||
|
args: %{"contact_id" => contact.id}
|
||||||
|
})
|
||||||
|
|
||||||
|
reloaded = Repo.get!(Contact, contact.id)
|
||||||
|
assert reloaded.mechanism_status == :failed
|
||||||
|
end
|
||||||
|
|
||||||
test "meteor-shower calendar matches across year boundaries" do
|
test "meteor-shower calendar matches across year boundaries" do
|
||||||
# Quadrantids peak Jan 4, ±3 day window. A 6 m 900 km contact on
|
# Quadrantids peak Jan 4, ±3 day window. A 6 m 900 km contact on
|
||||||
# Dec 31 2025 is 4 days before the Jan 4 2026 peak — should still
|
# Dec 31 2025 is 4 days before the Jan 4 2026 peak — should still
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue