fix(rover-planning): async reconcile + strict band match in path worker
Two fixes: 1) Form save / rover-site add/remove no longer runs the path-matrix reconcile inline. update_mission and the show LiveView now enqueue a single RoverMissionReconcileWorker job and return — for missions with many rover sites x bands the inline path was firing hundreds of Multi inserts + Oban.inserts on the request. 2) RoverPathProfileWorker.load_path/1 now strictly matches a 4-tuple (mission_id, rover_location_id, station_id, band_mhz) with band_mhz as an integer, plus updates the Oban unique key set to include band_mhz so multi-band missions enqueue per-band jobs (not just one). Legacy/malformed args are logged and dropped instead of crashing the queue with MultipleResultsError.
This commit is contained in:
parent
1d4b2aeedb
commit
fe2024275b
5 changed files with 110 additions and 43 deletions
|
|
@ -20,6 +20,7 @@ defmodule Microwaveprop.RoverPlanning do
|
|||
alias Microwaveprop.RoverPlanning.Mission
|
||||
alias Microwaveprop.RoverPlanning.Path
|
||||
alias Microwaveprop.RoverPlanning.Station
|
||||
alias Microwaveprop.Workers.RoverMissionReconcileWorker
|
||||
alias Microwaveprop.Workers.RoverPathProfileWorker
|
||||
|
||||
@spec list_missions() :: [Mission.t()]
|
||||
|
|
@ -79,7 +80,7 @@ defmodule Microwaveprop.RoverPlanning do
|
|||
|> case do
|
||||
{:ok, updated} ->
|
||||
updated = Repo.preload(updated, [:user, :stations], force: true)
|
||||
:ok = replace_mission_paths(updated)
|
||||
:ok = enqueue_reconcile(updated)
|
||||
{:ok, updated}
|
||||
|
||||
other ->
|
||||
|
|
@ -152,6 +153,22 @@ defmodule Microwaveprop.RoverPlanning do
|
|||
@spec replace_mission_paths(Mission.t()) :: :ok
|
||||
defdelegate replace_mission_paths(mission), to: __MODULE__, as: :reconcile_mission_paths
|
||||
|
||||
@doc """
|
||||
Async entry point for path-matrix reconciliation. Enqueues a
|
||||
`RoverMissionReconcileWorker` job and returns immediately so callers
|
||||
(form save, rover-site add/remove) don't pay the per-tuple insert
|
||||
cost on the request path. Oban inline mode in tests still runs
|
||||
this synchronously, so test expectations don't have to change.
|
||||
"""
|
||||
@spec enqueue_reconcile(Mission.t()) :: :ok
|
||||
def enqueue_reconcile(%Mission{id: mission_id}) do
|
||||
%{"mission_id" => mission_id}
|
||||
|> RoverMissionReconcileWorker.new()
|
||||
|> Oban.insert()
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
@doc """
|
||||
Walks every mission and re-enqueues any (rover_location × station)
|
||||
pair whose `Path` is missing or not `:complete`. Existing complete
|
||||
|
|
|
|||
45
lib/microwaveprop/workers/rover_mission_reconcile_worker.ex
Normal file
45
lib/microwaveprop/workers/rover_mission_reconcile_worker.ex
Normal file
|
|
@ -0,0 +1,45 @@
|
|||
defmodule Microwaveprop.Workers.RoverMissionReconcileWorker do
|
||||
@moduledoc """
|
||||
Async reconciler for one rover-planning mission's path matrix.
|
||||
Form saves (mission edits, add/remove rover site) enqueue this job
|
||||
instead of running the reconcile inline — for a mission with many
|
||||
rover sites × stations × bands the inline path produced a multi-
|
||||
hundred-row Multi transaction + as many `Oban.insert/1` calls,
|
||||
which made the form submit feel sluggish.
|
||||
|
||||
The reconcile itself is idempotent: deleting stale (mission_id,
|
||||
rover_location_id, station_id, band_mhz) rows, leaving :complete
|
||||
rows in place, and enqueueing per-pair compute jobs only for the
|
||||
tuples that aren't already represented.
|
||||
"""
|
||||
|
||||
use Oban.Worker,
|
||||
queue: :terrain,
|
||||
priority: 3,
|
||||
max_attempts: 3,
|
||||
unique: [
|
||||
period: 60,
|
||||
keys: [:mission_id],
|
||||
states: [:available, :scheduled, :executing, :retryable]
|
||||
]
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.RoverPlanning
|
||||
alias Microwaveprop.RoverPlanning.Mission
|
||||
alias Microwaveprop.RoverPlanning.Station
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: %{"mission_id" => mission_id}}) do
|
||||
case Repo.get(Mission, mission_id) do
|
||||
nil ->
|
||||
:ok
|
||||
|
||||
mission ->
|
||||
mission
|
||||
|> Repo.preload([:user, stations: from(s in Station, order_by: s.position)])
|
||||
|> RoverPlanning.reconcile_mission_paths()
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -16,7 +16,7 @@ defmodule Microwaveprop.Workers.RoverPathProfileWorker do
|
|||
max_attempts: 5,
|
||||
unique: [
|
||||
period: 300,
|
||||
keys: [:mission_id, :rover_location_id, :station_id],
|
||||
keys: [:mission_id, :rover_location_id, :station_id, :band_mhz],
|
||||
states: [:available, :scheduled, :executing, :retryable]
|
||||
]
|
||||
|
||||
|
|
@ -57,25 +57,30 @@ defmodule Microwaveprop.Workers.RoverPathProfileWorker do
|
|||
end
|
||||
end
|
||||
|
||||
defp load_path(
|
||||
%{"mission_id" => mission_id, "rover_location_id" => rover_location_id, "station_id" => station_id} = args
|
||||
) do
|
||||
band_mhz = Map.get(args, "band_mhz")
|
||||
|
||||
defp load_path(%{
|
||||
"mission_id" => mission_id,
|
||||
"rover_location_id" => rover_location_id,
|
||||
"station_id" => station_id,
|
||||
"band_mhz" => band_mhz
|
||||
})
|
||||
when is_integer(band_mhz) do
|
||||
Path
|
||||
|> where([p], p.mission_id == ^mission_id)
|
||||
|> where([p], p.rover_location_id == ^rover_location_id)
|
||||
|> where([p], p.station_id == ^station_id)
|
||||
|> maybe_filter_band(band_mhz)
|
||||
|> where([p], p.band_mhz == ^band_mhz)
|
||||
|> preload([:rover_location, :station, :mission])
|
||||
|> Repo.one()
|
||||
end
|
||||
|
||||
# Legacy queued jobs (pre-multi-band) didn't include band_mhz in
|
||||
# their args. The single Path row that existed for that tuple is the
|
||||
# only candidate, so loading without the band filter is unambiguous.
|
||||
defp maybe_filter_band(query, nil), do: query
|
||||
defp maybe_filter_band(query, band) when is_integer(band), do: where(query, [p], p.band_mhz == ^band)
|
||||
# Legacy / malformed args (no band_mhz, or wrong types). Don't crash
|
||||
# the queue — log enough context to debug and drop the job. New
|
||||
# enqueues from `enqueue_paths_for/1` always include band_mhz.
|
||||
defp load_path(args) do
|
||||
Logger.warning("RoverPathProfileWorker: dropping job with unexpected args (no integer band_mhz): #{inspect(args)}")
|
||||
|
||||
nil
|
||||
end
|
||||
|
||||
defp run(%Path{rover_location: rover, station: station, mission: mission} = path) do
|
||||
mark_status(path, :computing)
|
||||
|
|
|
|||
|
|
@ -123,35 +123,6 @@ defmodule MicrowavepropWeb.RoverPlanningLive.Form do
|
|||
{:noreply, assign(socket, form: to_form(cs))}
|
||||
end
|
||||
|
||||
# When the form changeset already carries a "stations" params map
|
||||
# (every keystroke fires phx-change="validate" which builds one),
|
||||
# trust it. Otherwise — e.g. the user clicked add/remove BEFORE
|
||||
# touching any other input on an :edit-mode form — fall back to
|
||||
# rebuilding the params from the persisted mission's stations so
|
||||
# `cast_assoc(on_replace: :delete)` doesn't wipe every existing row.
|
||||
defp stations_params(socket, %{"stations" => stations}) when is_map(stations) and stations != %{}, do: stations
|
||||
|
||||
defp stations_params(socket, _params), do: stations_from_mission(socket.assigns.mission)
|
||||
|
||||
defp stations_from_mission(%Mission{stations: list}) when is_list(list) do
|
||||
list
|
||||
|> Enum.with_index()
|
||||
|> Map.new(fn {s, i} ->
|
||||
{Integer.to_string(i),
|
||||
%{
|
||||
"id" => s.id,
|
||||
"input" => s.input || s.callsign || s.grid,
|
||||
"callsign" => s.callsign,
|
||||
"grid" => s.grid,
|
||||
"lat" => s.lat,
|
||||
"lon" => s.lon,
|
||||
"position" => Integer.to_string(s.position || i)
|
||||
}}
|
||||
end)
|
||||
end
|
||||
|
||||
defp stations_from_mission(_), do: %{}
|
||||
|
||||
def handle_event("save", %{"mission" => params}, socket) do
|
||||
user = socket.assigns.current_user
|
||||
|
||||
|
|
@ -176,6 +147,35 @@ defmodule MicrowavepropWeb.RoverPlanningLive.Form do
|
|||
end
|
||||
end
|
||||
|
||||
# When the form changeset already carries a "stations" params map
|
||||
# (every keystroke fires phx-change="validate" which builds one),
|
||||
# trust it. Otherwise — e.g. the user clicked add/remove BEFORE
|
||||
# touching any other input on an :edit-mode form — fall back to
|
||||
# rebuilding the params from the persisted mission's stations so
|
||||
# `cast_assoc(on_replace: :delete)` doesn't wipe every existing row.
|
||||
defp stations_params(_socket, %{"stations" => stations}) when is_map(stations) and stations != %{}, do: stations
|
||||
|
||||
defp stations_params(socket, _params), do: stations_from_mission(socket.assigns.mission)
|
||||
|
||||
defp stations_from_mission(%Mission{stations: list}) when is_list(list) do
|
||||
list
|
||||
|> Enum.with_index()
|
||||
|> Map.new(fn {s, i} ->
|
||||
{Integer.to_string(i),
|
||||
%{
|
||||
"id" => s.id,
|
||||
"input" => s.input || s.callsign || s.grid,
|
||||
"callsign" => s.callsign,
|
||||
"grid" => s.grid,
|
||||
"lat" => s.lat,
|
||||
"lon" => s.lon,
|
||||
"position" => Integer.to_string(s.position || i)
|
||||
}}
|
||||
end)
|
||||
end
|
||||
|
||||
defp stations_from_mission(_), do: %{}
|
||||
|
||||
defp current_params(socket) do
|
||||
case socket.assigns.form.source do
|
||||
%Ecto.Changeset{params: params} when is_map(params) -> params
|
||||
|
|
|
|||
|
|
@ -97,7 +97,7 @@ defmodule MicrowavepropWeb.RoverPlanningLive.Show do
|
|||
# that depend on the rover-site set.
|
||||
defp refresh_after_site_change(socket, flash_message) do
|
||||
mission = socket.assigns.mission
|
||||
:ok = RoverPlanning.replace_mission_paths(mission)
|
||||
:ok = RoverPlanning.enqueue_reconcile(mission)
|
||||
|
||||
socket
|
||||
|> assign(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue