Atomic score upserts and auto-prune old propagation data
- Wrap upsert_scores in Repo.transaction for all-or-nothing visibility - Prune scores older than the 2 most recent valid_times after each upsert - Add band-specific latest_valid_time/1 to eliminate N+1 query - Add require Logger to Propagation module
This commit is contained in:
parent
1aebd7f1cc
commit
3874fc173d
1 changed files with 44 additions and 12 deletions
|
|
@ -10,6 +10,8 @@ defmodule Microwaveprop.Propagation do
|
||||||
alias Microwaveprop.Repo
|
alias Microwaveprop.Repo
|
||||||
alias Microwaveprop.Weather.SoundingParams
|
alias Microwaveprop.Weather.SoundingParams
|
||||||
|
|
||||||
|
require Logger
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Score a single grid point across all bands using HRRR profile data.
|
Score a single grid point across all bands using HRRR profile data.
|
||||||
Returns a list of %{band_mhz, score, factors} maps.
|
Returns a list of %{band_mhz, score, factors} maps.
|
||||||
|
|
@ -54,7 +56,7 @@ defmodule Microwaveprop.Propagation do
|
||||||
end)
|
end)
|
||||||
end
|
end
|
||||||
|
|
||||||
@doc "Upsert propagation scores in batches."
|
@doc "Upsert propagation scores in batches within a transaction so readers see all-or-nothing."
|
||||||
def upsert_scores(scores) do
|
def upsert_scores(scores) do
|
||||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||||
|
|
||||||
|
|
@ -73,20 +75,50 @@ defmodule Microwaveprop.Propagation do
|
||||||
}
|
}
|
||||||
end)
|
end)
|
||||||
|
|
||||||
total =
|
result =
|
||||||
entries
|
Repo.transaction(fn ->
|
||||||
|> Enum.chunk_every(500)
|
entries
|
||||||
|> Enum.reduce(0, fn chunk, acc ->
|
|> Enum.chunk_every(500)
|
||||||
{count, _} =
|
|> Enum.reduce(0, fn chunk, acc ->
|
||||||
Repo.insert_all(GridScore, chunk,
|
{count, _} =
|
||||||
on_conflict: {:replace, [:score, :factors, :updated_at]},
|
Repo.insert_all(GridScore, chunk,
|
||||||
conflict_target: [:lat, :lon, :valid_time, :band_mhz]
|
on_conflict: {:replace, [:score, :factors, :updated_at]},
|
||||||
)
|
conflict_target: [:lat, :lon, :valid_time, :band_mhz]
|
||||||
|
)
|
||||||
|
|
||||||
acc + count
|
acc + count
|
||||||
|
end)
|
||||||
end)
|
end)
|
||||||
|
|
||||||
{:ok, total}
|
case result do
|
||||||
|
{:ok, _count} -> prune_old_scores()
|
||||||
|
_ -> :ok
|
||||||
|
end
|
||||||
|
|
||||||
|
result
|
||||||
|
end
|
||||||
|
|
||||||
|
defp prune_old_scores do
|
||||||
|
keep_times =
|
||||||
|
from(gs in GridScore,
|
||||||
|
select: gs.valid_time,
|
||||||
|
distinct: gs.valid_time,
|
||||||
|
order_by: [desc: gs.valid_time],
|
||||||
|
limit: 2
|
||||||
|
)
|
||||||
|
|> Repo.all()
|
||||||
|
|
||||||
|
if length(keep_times) == 2 do
|
||||||
|
oldest_kept = List.last(keep_times)
|
||||||
|
|
||||||
|
{deleted, _} =
|
||||||
|
from(gs in GridScore, where: gs.valid_time < ^oldest_kept)
|
||||||
|
|> Repo.delete_all()
|
||||||
|
|
||||||
|
if deleted > 0 do
|
||||||
|
Logger.info("PropagationScores: pruned #{deleted} old scores, keeping 2 most recent valid_times")
|
||||||
|
end
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
@doc "Get the latest scores for a band, optionally within a bounding box. Excludes factors for performance."
|
@doc "Get the latest scores for a band, optionally within a bounding box. Excludes factors for performance."
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue