refactor(weather): make rebatch logic callable from release eval
Mix.Task.run isn't available in a compiled release, so the in-prod `bin/microwaveprop eval 'Mix.Tasks.Weather.RebatchAsos.run(...)'` dispatch crashes. Moves the real logic into a plain module (`Microwaveprop.Weather.RebatchAsos`) that both the Mix task and a release eval can call. Uses IO.puts over Mix.shell for the same reason.
This commit is contained in:
parent
8c75ca640a
commit
e4668790a4
3 changed files with 122 additions and 97 deletions
111
lib/microwaveprop/weather/rebatch_asos.ex
Normal file
111
lib/microwaveprop/weather/rebatch_asos.ex
Normal file
|
|
@ -0,0 +1,111 @@
|
|||
defmodule Microwaveprop.Weather.RebatchAsos do
|
||||
@moduledoc """
|
||||
Converts pending single-station `WeatherFetchWorker` ASOS jobs into
|
||||
the batched `asos_batch` shape — one job per unique
|
||||
`(start_dt, end_dt)` window covering every station that had a
|
||||
pending job for that window.
|
||||
|
||||
Callable from both a Mix task (`mix weather.rebatch_asos`) and a
|
||||
release `eval` shell (`bin/microwaveprop eval
|
||||
'Microwaveprop.Weather.RebatchAsos.run()'`). In prod the Mix API
|
||||
isn't available, so this module is Mix-free.
|
||||
|
||||
Idempotent: a second run with no pending "asos" jobs is a no-op.
|
||||
"""
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Workers.WeatherFetchWorker
|
||||
|
||||
@doc """
|
||||
Run the rebatch. Options:
|
||||
* `:dry_run` (boolean) — log the plan but don't mutate the queue.
|
||||
|
||||
Returns a summary map: `%{found: n, windows: n, inserted: n, cancelled: n}`.
|
||||
"""
|
||||
@spec run(keyword()) :: %{
|
||||
found: non_neg_integer(),
|
||||
windows: non_neg_integer(),
|
||||
inserted: non_neg_integer(),
|
||||
cancelled: non_neg_integer()
|
||||
}
|
||||
def run(opts \\ []) do
|
||||
dry_run? = Keyword.get(opts, :dry_run, false)
|
||||
jobs = load_pending_asos_jobs()
|
||||
|
||||
if jobs == [] do
|
||||
IO.puts("No pending `asos` jobs. Nothing to do.")
|
||||
%{found: 0, windows: 0, inserted: 0, cancelled: 0}
|
||||
else
|
||||
do_rebatch(jobs, dry_run?)
|
||||
end
|
||||
end
|
||||
|
||||
defp load_pending_asos_jobs do
|
||||
query =
|
||||
from(j in Oban.Job,
|
||||
where:
|
||||
j.worker == "Microwaveprop.Workers.WeatherFetchWorker" and
|
||||
j.state in ["available", "scheduled", "retryable"] and
|
||||
fragment("?->>'fetch_type' = ?", j.args, "asos"),
|
||||
select: %{id: j.id, args: j.args}
|
||||
)
|
||||
|
||||
Repo.all(query)
|
||||
end
|
||||
|
||||
defp do_rebatch(jobs, dry_run?) do
|
||||
groups =
|
||||
jobs
|
||||
|> Enum.group_by(fn %{args: a} ->
|
||||
{Map.get(a, "start_dt"), Map.get(a, "end_dt")}
|
||||
end)
|
||||
|> Enum.reject(fn {{s, e}, _} -> is_nil(s) or is_nil(e) end)
|
||||
|
||||
IO.puts("Found #{length(jobs)} pending `asos` jobs across #{length(groups)} time windows.")
|
||||
|
||||
batch_args =
|
||||
Enum.map(groups, fn {{start_dt, end_dt}, members} ->
|
||||
sorted = Enum.sort_by(members, fn %{args: a} -> a["station_code"] end)
|
||||
|
||||
%{
|
||||
"fetch_type" => "asos_batch",
|
||||
"station_ids" => Enum.map(sorted, fn %{args: a} -> a["station_id"] end),
|
||||
"station_codes" => Enum.map(sorted, fn %{args: a} -> a["station_code"] end),
|
||||
"start_dt" => start_dt,
|
||||
"end_dt" => end_dt,
|
||||
"_source_ids" => Enum.map(sorted, & &1.id)
|
||||
}
|
||||
end)
|
||||
|
||||
if dry_run? do
|
||||
Enum.each(batch_args, fn a ->
|
||||
IO.puts("[dry-run] #{length(a["station_codes"])} stations for #{a["start_dt"]}..#{a["end_dt"]}")
|
||||
end)
|
||||
|
||||
%{found: length(jobs), windows: length(groups), inserted: 0, cancelled: 0}
|
||||
else
|
||||
apply_rebatch(batch_args, length(jobs))
|
||||
end
|
||||
end
|
||||
|
||||
defp apply_rebatch(batch_args, found) do
|
||||
{inserted, cancelled} =
|
||||
Enum.reduce(batch_args, {0, 0}, fn a, {ins, can} ->
|
||||
source_ids = a["_source_ids"]
|
||||
job_args = Map.delete(a, "_source_ids")
|
||||
|
||||
Repo.transaction(fn ->
|
||||
_ = job_args |> WeatherFetchWorker.new() |> Oban.insert!()
|
||||
{:ok, _n} = Oban.cancel_all_jobs(from(j in Oban.Job, where: j.id in ^source_ids))
|
||||
end)
|
||||
|
||||
{ins + 1, can + length(source_ids)}
|
||||
end)
|
||||
|
||||
IO.puts("Rebatched: inserted #{inserted} batch jobs, cancelled #{cancelled} single-station jobs.")
|
||||
|
||||
%{found: found, windows: length(batch_args), inserted: inserted, cancelled: cancelled}
|
||||
end
|
||||
end
|
||||
|
|
@ -2,111 +2,25 @@ defmodule Mix.Tasks.Weather.RebatchAsos do
|
|||
@shortdoc "Collapse pending single-station ASOS jobs into batched jobs"
|
||||
|
||||
@moduledoc """
|
||||
Converts every pending single-station `WeatherFetchWorker` ASOS job
|
||||
into the batched (`asos_batch`) shape — one job per unique
|
||||
`(start_dt, end_dt)` window covering every station that had a
|
||||
pending job for that window.
|
||||
Thin Mix-task wrapper over `Microwaveprop.Weather.RebatchAsos`.
|
||||
The real logic lives in the plain module so it's callable from a
|
||||
release shell too.
|
||||
|
||||
Why: the worker still accepts the old `"asos"` fetch_type for
|
||||
backward compat, but every such job pays the full `IemRateLimiter`
|
||||
gap + IEM 429-retry tail for a single station's rows. Collapsing
|
||||
them into batched jobs gets the queue through the backlog at the
|
||||
new per-request efficiency.
|
||||
|
||||
Usage (idempotent — running it twice is a no-op):
|
||||
|
||||
mix weather.rebatch_asos # rewrites everything
|
||||
mix weather.rebatch_asos --dry-run # report only
|
||||
Usage:
|
||||
|
||||
mix weather.rebatch_asos # rewrite
|
||||
mix weather.rebatch_asos --dry-run # report only
|
||||
"""
|
||||
|
||||
use Mix.Task
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Workers.WeatherFetchWorker
|
||||
|
||||
require Logger
|
||||
alias Microwaveprop.Weather.RebatchAsos
|
||||
|
||||
@impl Mix.Task
|
||||
def run(args) do
|
||||
{opts, _, _} = OptionParser.parse(args, strict: [dry_run: :boolean])
|
||||
Mix.Task.run("app.start")
|
||||
dry_run? = Keyword.get(opts, :dry_run, false)
|
||||
|
||||
jobs = load_pending_asos_jobs()
|
||||
|
||||
if jobs == [] do
|
||||
Mix.shell().info("No pending `asos` jobs. Nothing to do.")
|
||||
else
|
||||
do_rebatch(jobs, dry_run?)
|
||||
end
|
||||
end
|
||||
|
||||
@doc false
|
||||
def load_pending_asos_jobs do
|
||||
query =
|
||||
from(j in Oban.Job,
|
||||
where:
|
||||
j.worker == "Microwaveprop.Workers.WeatherFetchWorker" and
|
||||
j.state in ["available", "scheduled", "retryable"] and
|
||||
fragment("?->>'fetch_type' = ?", j.args, "asos"),
|
||||
select: %{id: j.id, args: j.args}
|
||||
)
|
||||
|
||||
Repo.all(query)
|
||||
end
|
||||
|
||||
defp do_rebatch(jobs, dry_run?) do
|
||||
groups =
|
||||
jobs
|
||||
|> Enum.group_by(fn %{args: a} ->
|
||||
{Map.get(a, "start_dt"), Map.get(a, "end_dt")}
|
||||
end)
|
||||
|> Enum.reject(fn {{s, e}, _} -> is_nil(s) or is_nil(e) end)
|
||||
|
||||
Mix.shell().info("Found #{length(jobs)} pending `asos` jobs across #{length(groups)} time windows.")
|
||||
|
||||
batch_args =
|
||||
Enum.map(groups, fn {{start_dt, end_dt}, members} ->
|
||||
# Deterministic sort (matches the enqueuer's ordering) so a
|
||||
# re-run with identical membership hits Oban's unique dedup.
|
||||
sorted = Enum.sort_by(members, fn %{args: a} -> a["station_code"] end)
|
||||
|
||||
%{
|
||||
"fetch_type" => "asos_batch",
|
||||
"station_ids" => Enum.map(sorted, fn %{args: a} -> a["station_id"] end),
|
||||
"station_codes" => Enum.map(sorted, fn %{args: a} -> a["station_code"] end),
|
||||
"start_dt" => start_dt,
|
||||
"end_dt" => end_dt,
|
||||
"_source_ids" => Enum.map(sorted, & &1.id)
|
||||
}
|
||||
end)
|
||||
|
||||
if dry_run? do
|
||||
Enum.each(batch_args, fn a ->
|
||||
Mix.shell().info("[dry-run] #{length(a["station_codes"])} stations for #{a["start_dt"]}..#{a["end_dt"]}")
|
||||
end)
|
||||
else
|
||||
apply_rebatch(batch_args)
|
||||
end
|
||||
end
|
||||
|
||||
defp apply_rebatch(batch_args) do
|
||||
{inserted, cancelled} =
|
||||
Enum.reduce(batch_args, {0, 0}, fn a, {ins, can} ->
|
||||
source_ids = a["_source_ids"]
|
||||
job_args = Map.delete(a, "_source_ids")
|
||||
|
||||
Repo.transaction(fn ->
|
||||
_ = job_args |> WeatherFetchWorker.new() |> Oban.insert!()
|
||||
{:ok, _n} = Oban.cancel_all_jobs(from(j in Oban.Job, where: j.id in ^source_ids))
|
||||
end)
|
||||
|
||||
{ins + 1, can + length(source_ids)}
|
||||
end)
|
||||
|
||||
Mix.shell().info("Rebatched: inserted #{inserted} batch jobs, cancelled #{cancelled} single-station jobs.")
|
||||
_ = RebatchAsos.run(opts)
|
||||
:ok
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -2,8 +2,8 @@ defmodule Mix.Tasks.Weather.RebatchAsosTest do
|
|||
use Microwaveprop.DataCase, async: false
|
||||
|
||||
alias Microwaveprop.Weather.IemClient
|
||||
alias Microwaveprop.Weather.RebatchAsos
|
||||
alias Microwaveprop.Workers.WeatherFetchWorker
|
||||
alias Mix.Tasks.Weather.RebatchAsos
|
||||
|
||||
setup do
|
||||
# Oban runs inline in tests, so the batch job the task inserts
|
||||
|
|
@ -75,7 +75,7 @@ defmodule Mix.Tasks.Weather.RebatchAsosTest do
|
|||
test "dry-run does not mutate the queue" do
|
||||
insert_asos_job(Ecto.UUID.generate(), "KDFW", "2026-03-28T16:00:00Z", "2026-03-28T20:00:00Z")
|
||||
|
||||
ExUnit.CaptureIO.capture_io(fn -> RebatchAsos.run(["--dry-run"]) end)
|
||||
ExUnit.CaptureIO.capture_io(fn -> RebatchAsos.run(dry_run: true) end)
|
||||
|
||||
counts = pending_count_by_type()
|
||||
assert counts["asos"] == 1
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue