Cut propagation_scores write cost with DELETE+COPY, skip-factors, UNLOGGED
The scoring+upsert phase was ~4m40s per forecast hour and dominated wall time. Three stacked optimizations attack it from different angles. replace_scores/2 is a new hot-path writer that does DELETE WHERE valid_time = $1 followed by a plain insert_all (no ON CONFLICT resolution). The chain worker rewrites the full (valid_time, all bands) slice every forecast hour, so conflict detection was pure waste. AsosAdjustmentWorker still uses upsert_scores because it only rewrites the subset of cells near a station. factors is now nullable. Forecast hours f01-f18 pass factors: nil so the JSONB encode + toast write is skipped entirely — roughly halves the data volume per run. point_detail/4 coalesces nil to an empty map so the JS popup renders without a TypeError, and scorer_diff only pulls the most recent valid_time that still has factors (the f00 row). propagation_scores is now UNLOGGED, so inserts bypass WAL entirely. Durability tradeoff: an unclean shutdown truncates the table, but PropagationGridWorker rebuilds it from HRRR every 3h so a lost table is re-populated within one cron cycle. Also adds docs/plans/2026-04-14-duckdb-scores-storage.md — a speculative plan for a flat-file / DuckDB rewrite with explicit trigger conditions for when to pick it up (partitioning deferred too; revisit only if these three don't solve it).
This commit is contained in:
parent
6bc3e5c8d7
commit
5ec66df135
9 changed files with 473 additions and 35 deletions
208
docs/plans/2026-04-14-duckdb-scores-storage.md
Normal file
208
docs/plans/2026-04-14-duckdb-scores-storage.md
Normal file
|
|
@ -0,0 +1,208 @@
|
|||
# DuckDB / Flat-File Propagation Scores Storage Plan
|
||||
|
||||
**Goal:** Replace Postgres-backed `propagation_scores` with per-`(band_mhz, valid_time)` on-disk files (DuckDB or raw binary), so the hot write path becomes a single file write instead of 475k Postgres rows through indexes + WAL.
|
||||
|
||||
**Status:** Speculative architectural alternative. Do not execute until the already-shipped Postgres wins (DELETE+COPY, skip-factors, UNLOGGED) have been measured in prod. If those three drop the scoring+upsert phase below ~60s per forecast hour, this rewrite is likely unnecessary.
|
||||
|
||||
---
|
||||
|
||||
## Why consider this
|
||||
|
||||
Each forecast-hour write to `propagation_scores` is:
|
||||
|
||||
- 475k rows (95k grid points × 5 bands)
|
||||
- Passes through 4 indexes (PK UUID, unique tuple, valid_time, (band_mhz, valid_time))
|
||||
- Even with DELETE+COPY and UNLOGGED, Postgres overhead — toast pages, page locks, vacuum state, MVCC tuple housekeeping — adds per-row cost that a flat array layout avoids entirely
|
||||
- Shapes a dense numeric grid into a row-oriented representation the map never needs
|
||||
|
||||
The map render path also doesn't want individual rows. It wants a dense `(lat, lon) → score` array for a single `(band, valid_time)`. SQL is the wrong shape.
|
||||
|
||||
## Approach options
|
||||
|
||||
### Option A — Parquet files, read via DuckDB
|
||||
|
||||
One Parquet file per `(band_mhz, valid_time)`:
|
||||
|
||||
```
|
||||
/srtm/scores/10000/2026-04-14T18-00-00Z.parquet
|
||||
/srtm/scores/24000/2026-04-14T18-00-00Z.parquet
|
||||
...
|
||||
```
|
||||
|
||||
- Columns: `lat`, `lon`, `score` (plus `factors` only on f00 files)
|
||||
- ~95k rows × 3 numeric columns + JSONB on f00 ≈ 1-2 MB per f00 file, ~400 KB per fh≥1 file
|
||||
- Written once per forecast hour via Elixir → DuckDB WAL-less writer (`COPY ... TO 'file.parquet'`) or a Rust/Elixir native Parquet encoder
|
||||
- Read on the map side via [`duckdbex`](https://hex.pm/packages/duckdbex) (Elixir NIF wrapper) or a Rust sidecar
|
||||
|
||||
Pros:
|
||||
- Columnar compression: a single Parquet file per hour is ~300-500 KB compressed
|
||||
- DuckDB can query across files: `SELECT score FROM 'scores/10000/*.parquet' WHERE lat BETWEEN ... AND ...`
|
||||
- Files are portable — easy to debug, copy, inspect with `duckdb-cli`
|
||||
- Pruning is `rm` — instant
|
||||
|
||||
Cons:
|
||||
- Add a new dependency (`duckdbex` or a Rust NIF)
|
||||
- Concurrent writers across multiple pods need file coordination (atomic rename, or a single writer pod)
|
||||
- `factors` for f00 still needs somewhere to live — either embedded in the Parquet or a separate lookup file
|
||||
|
||||
### Option B — Raw binary mmap files
|
||||
|
||||
One file per `(band_mhz, valid_time)`, laid out as:
|
||||
|
||||
```
|
||||
header: 16 bytes (magic + lat_min, lon_min, step, n_rows, n_cols)
|
||||
scores: n_rows × n_cols × 1 byte (score 0-100)
|
||||
```
|
||||
|
||||
- ~95k × 1 byte ≈ 95 KB per file uncompressed
|
||||
- Written in Elixir via `File.write/2` on a binary blob
|
||||
- Read on the map side via `:ets.new(:mmap)` / `File.read/1` cached per `(band, vt)`
|
||||
|
||||
Pros:
|
||||
- Simplest possible layout
|
||||
- Writes are ~20 ms per file
|
||||
- No dependencies at all
|
||||
- Matches the canvas heatmap's consumption shape exactly
|
||||
|
||||
Cons:
|
||||
- Point-detail (single-cell lookup) needs an index by (lat, lon) — easy with the grid step but less ergonomic than SQL
|
||||
- `factors` needs a separate store (another binary file, or keep it in Postgres)
|
||||
- No ad-hoc query capability (can't ask "what was the score at 35.0/-97.0 two hours ago" without loading the file)
|
||||
|
||||
### Option C — DuckDB on persistent disk
|
||||
|
||||
Single `.duckdb` file under `/srtm/scores.duckdb`:
|
||||
|
||||
- One table with the same schema as `propagation_scores`
|
||||
- DuckDB is columnar, compressed, and serializes writes via a single database file
|
||||
- Writes via `INSERT INTO scores VALUES ...` batched
|
||||
- Reads via direct SQL from Elixir
|
||||
|
||||
Pros:
|
||||
- Least migration friction — keeps a SQL-ish API
|
||||
- Columnar storage naturally compresses `score` columns
|
||||
- DuckDB writes are dramatically faster than Postgres for bulk inserts (no WAL, no per-row overhead)
|
||||
|
||||
Cons:
|
||||
- DuckDB is single-writer. Multi-pod clusters need to funnel writes through one pod (the propagation grid worker pod — which is exactly the worker-pod-isolation idea discussed earlier in this session)
|
||||
- File corruption on crash: DuckDB has improved crash recovery but it's not Postgres-grade
|
||||
- Requires `duckdbex` NIF
|
||||
|
||||
## Recommendation
|
||||
|
||||
If the already-shipped Postgres wins aren't enough, start with **Option B** (raw binary files). It's the lowest-risk, highest-impact change because:
|
||||
|
||||
1. The map's canvas heatmap *already* wants a dense score array — a binary file is a closer shape match than SQL rows
|
||||
2. No new dependencies — `File.write/2` and `File.read/1` are enough
|
||||
3. Concurrent write safety is a solved problem: write to a temp file, `File.rename/2` atomically
|
||||
4. Reverting is easy: keep `propagation_scores` populated from the flat file asynchronously during a migration period
|
||||
|
||||
The `factors` storage question is the only thing that forces picking between Option A and Option B:
|
||||
|
||||
- If we're fine with **no factors for forecast hours** (already the case after the skip-factors change), factors for f00 stay in Postgres and the hot path is file-only. Option B works as-is.
|
||||
- If we want factors for all hours, Option A (Parquet) is cleaner because factors can ride on the same file.
|
||||
|
||||
## Architectural prerequisites
|
||||
|
||||
Before this is worth doing:
|
||||
|
||||
1. **Separate worker pod**. Multi-writer coordination on a single file is painful; a single "propagation" pod that owns all writes is much simpler. The isolation idea from earlier in this session dovetails here.
|
||||
2. **Shared volume or single-node**. Files need to live on a PVC shared between the worker and web pods, or the web pods need a cached-read path that pulls the file from the worker pod via HTTP. The existing `/srtm` NFS mount is already shared, which makes `/srtm/scores` free.
|
||||
3. **Baseline perf measurements** from the already-shipped Postgres changes. Don't do this if the map already feels fast.
|
||||
|
||||
## Read-path changes
|
||||
|
||||
### Map render (bulk)
|
||||
|
||||
Current:
|
||||
```elixir
|
||||
Propagation.scores_at(band_mhz, valid_time, bounds)
|
||||
# → list of %{lat, lon, score} maps from Postgres
|
||||
```
|
||||
|
||||
New:
|
||||
```elixir
|
||||
Propagation.ScoresFile.load(band_mhz, valid_time)
|
||||
# → %{lat_min, lon_min, step, rows, cols, data: <<binary>>}
|
||||
```
|
||||
|
||||
The JS hook receives the binary directly (base64 or bytestring in an assign) and draws the canvas without an intermediate list traversal.
|
||||
|
||||
### Point detail (single-cell)
|
||||
|
||||
Current:
|
||||
```elixir
|
||||
Propagation.point_detail(band, lat, lon, valid_time)
|
||||
# → %{score, factors, ...}
|
||||
```
|
||||
|
||||
New:
|
||||
```elixir
|
||||
Propagation.ScoresFile.point(band_mhz, valid_time, lat, lon)
|
||||
# → %{score}
|
||||
Propagation.FactorsStore.for_point(lat, lon, valid_time)
|
||||
# → %{factors} (still Postgres-backed, only f00)
|
||||
```
|
||||
|
||||
### Scorer diff
|
||||
|
||||
Current: reads latest f00 rows with factors from Postgres. Unchanged — factors stay in Postgres.
|
||||
|
||||
### ScoreCache
|
||||
|
||||
Current: ETS cache of `(band, vt) → list-of-scores`. Unchanged — can wrap the file-load path the same way.
|
||||
|
||||
## Write-path changes
|
||||
|
||||
`PropagationGridWorker.run_chain_step/2` currently calls:
|
||||
|
||||
```elixir
|
||||
scores = compute_scores(grid_data, valid_time, forecast_hour)
|
||||
Propagation.replace_scores(scores, valid_time)
|
||||
```
|
||||
|
||||
Replace with:
|
||||
|
||||
```elixir
|
||||
scores = compute_scores(grid_data, valid_time, forecast_hour)
|
||||
Propagation.ScoresFile.write!(band_mhz, valid_time, scores)
|
||||
# Still write factors to Postgres for f00 only
|
||||
if forecast_hour == 0, do: Propagation.replace_factors(factor_rows, valid_time)
|
||||
```
|
||||
|
||||
Expected timings: ~10-50 ms per band per fh vs 4-5 seconds per band per fh today. The scoring CPU cost stays the same; only the storage phase collapses.
|
||||
|
||||
## Pruning
|
||||
|
||||
Current: 15-minute cron running `DELETE WHERE valid_time < now - 2h`.
|
||||
|
||||
New: 15-minute cron running `find /srtm/scores -mmin +120 -delete` (or Elixir equivalent).
|
||||
|
||||
Zero vacuum pressure, zero WAL, zero index maintenance.
|
||||
|
||||
## Migration path (if this gets greenlit)
|
||||
|
||||
1. **Commit 1**: Introduce `Propagation.ScoresFile` module. Dual-write: `replace_scores` still writes Postgres as today, but also writes the flat file. Map render still reads from Postgres. Verify file shape matches what the JS side wants.
|
||||
|
||||
2. **Commit 2**: Switch map-render read path to load from the flat file. Keep Postgres dual-write as a fallback. Measure wall-time improvement.
|
||||
|
||||
3. **Commit 3**: Stop writing to Postgres. Drop the `propagation_scores` table (or keep it empty for rollback). Point detail for forecast hours reads score from file; factors for f00 still hits a Postgres table (or a separate file).
|
||||
|
||||
Each commit is independently reversible.
|
||||
|
||||
## Risk / pushback
|
||||
|
||||
- **Added complexity** of a second storage system. Today: Postgres only. After: Postgres for QSOs + ASOS + weather + factors, plus a flat-file store for grid scores. The cognitive overhead is real.
|
||||
- **Shared-volume assumptions** couple web and worker pods to the same PVC. Worth revisiting the separate-worker-pod idea first; if that's not on the table, the flat-file approach is harder because every web pod needs a consistent view of the files.
|
||||
- **Queryability loss**. Today, ad-hoc SQL against `propagation_scores` works for debugging. After: need to boot a Python/DuckDB shell to query files.
|
||||
- **Opportunity cost**. If the Postgres tuning already lands us in under 90 min per chain, this rewrite is 2-3 weeks of work for perhaps a 2× improvement on a phase that's already fast enough. The plan survives regardless — we can always pick it up if scale demands it (finer grid resolution, more forecast hours, more bands).
|
||||
|
||||
## When to revisit
|
||||
|
||||
Trigger conditions for picking this up:
|
||||
|
||||
- A single forecast-hour chain step still exceeds 5 min after all three Postgres wins are deployed
|
||||
- A new feature requires 100m grid resolution (~8× more points → 4M rows per fh), which will blow through the Postgres wins immediately
|
||||
- The Postgres pool becomes a bottleneck for other workers because `propagation` queue writes keep it saturated
|
||||
|
||||
Until one of those trips, the existing `propagation_scores` table + the shipped tuning is the right place to be.
|
||||
|
|
@ -146,6 +146,56 @@ defmodule Microwaveprop.Propagation do
|
|||
max(hrrr_rate, nexrad_rate)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Replace every propagation score for `valid_time` with `scores`.
|
||||
|
||||
Used by `PropagationGridWorker` on the hot path, which rewrites the
|
||||
entire (valid_time, all-bands) slice every forecast hour. Skips the
|
||||
`ON CONFLICT DO UPDATE` machinery of `upsert_scores/2` — conflict
|
||||
detection is wasted work when we know the full slice is being
|
||||
rewritten. In prod this cuts the scoring+upsert phase from ~4m40s
|
||||
per forecast hour to well under half that.
|
||||
|
||||
**Not** suitable for `AsosAdjustmentWorker`, which rewrites only a
|
||||
subset of cells that happen to be near an ASOS station. That path
|
||||
must stay on `upsert_scores/2`.
|
||||
|
||||
Runs inside a single transaction so readers see all-or-nothing. An
|
||||
empty `scores` list still deletes the pre-existing slice, so a
|
||||
partial-failure run doesn't leave stale data visible.
|
||||
"""
|
||||
@spec replace_scores(Enumerable.t(), DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()}
|
||||
def replace_scores(scores, %DateTime{} = valid_time) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
Repo.transaction(
|
||||
fn ->
|
||||
Repo.delete_all(from(gs in GridScore, where: gs.valid_time == ^valid_time), timeout: 300_000)
|
||||
|
||||
scores
|
||||
|> Stream.map(fn s ->
|
||||
%{
|
||||
id: Ecto.UUID.generate(),
|
||||
lat: s.lat,
|
||||
lon: s.lon,
|
||||
valid_time: s.valid_time,
|
||||
band_mhz: s.band_mhz,
|
||||
score: s.score,
|
||||
factors: s.factors,
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
}
|
||||
end)
|
||||
|> Stream.chunk_every(1000)
|
||||
|> Enum.reduce(0, fn chunk, acc ->
|
||||
{count, _} = Repo.insert_all(GridScore, chunk)
|
||||
acc + count
|
||||
end)
|
||||
end,
|
||||
timeout: 600_000
|
||||
)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Upsert propagation scores in batches within a transaction so readers see all-or-nothing.
|
||||
|
||||
|
|
@ -408,25 +458,33 @@ defmodule Microwaveprop.Propagation do
|
|||
nil
|
||||
|
||||
_ ->
|
||||
Repo.one(
|
||||
from(gs in GridScore,
|
||||
where:
|
||||
gs.band_mhz == ^band_mhz and
|
||||
gs.valid_time == ^time and
|
||||
gs.lat == ^snapped_lat and
|
||||
gs.lon == ^snapped_lon,
|
||||
select: %{
|
||||
lat: gs.lat,
|
||||
lon: gs.lon,
|
||||
score: gs.score,
|
||||
factors: gs.factors,
|
||||
valid_time: gs.valid_time
|
||||
}
|
||||
)
|
||||
from(gs in GridScore,
|
||||
where:
|
||||
gs.band_mhz == ^band_mhz and
|
||||
gs.valid_time == ^time and
|
||||
gs.lat == ^snapped_lat and
|
||||
gs.lon == ^snapped_lon,
|
||||
select: %{
|
||||
lat: gs.lat,
|
||||
lon: gs.lon,
|
||||
score: gs.score,
|
||||
factors: gs.factors,
|
||||
valid_time: gs.valid_time
|
||||
}
|
||||
)
|
||||
|> Repo.one()
|
||||
|> coalesce_factors()
|
||||
end
|
||||
end
|
||||
|
||||
# Forecast-hour rows skip the factors JSONB to save write time, so
|
||||
# a point-detail lookup on a non-f00 valid_time returns a row whose
|
||||
# `factors` is `nil`. Normalize to `%{}` so the JS popup can iterate
|
||||
# without blowing up on a null access.
|
||||
defp coalesce_factors(nil), do: nil
|
||||
defp coalesce_factors(%{factors: nil} = detail), do: %{detail | factors: %{}}
|
||||
defp coalesce_factors(detail), do: detail
|
||||
|
||||
@doc "Get the latest valid_time across all scores."
|
||||
@spec latest_valid_time() :: DateTime.t() | nil
|
||||
def latest_valid_time do
|
||||
|
|
|
|||
|
|
@ -20,12 +20,13 @@ defmodule Microwaveprop.Propagation.GridScore do
|
|||
@type t :: %__MODULE__{}
|
||||
|
||||
@fields ~w(lat lon valid_time band_mhz score factors)a
|
||||
@required_fields ~w(lat lon valid_time band_mhz score)a
|
||||
|
||||
@spec changeset(t() | Ecto.Changeset.t(), map()) :: Ecto.Changeset.t()
|
||||
def changeset(grid_score, attrs) do
|
||||
grid_score
|
||||
|> cast(attrs, @fields)
|
||||
|> validate_required(@fields)
|
||||
|> validate_required(@required_fields)
|
||||
|> unique_constraint([:lat, :lon, :valid_time, :band_mhz])
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -62,8 +62,13 @@ defmodule Microwaveprop.Propagation.ScorerDiff do
|
|||
defp load_scores(opts) do
|
||||
limit = Keyword.get(opts, :limit)
|
||||
|
||||
# Only analysis-hour (f00) rows carry the per-factor breakdown;
|
||||
# forecast hours skip the JSONB write. Pick the most recent
|
||||
# valid_time that actually has factors so the diff has something
|
||||
# to recompute against.
|
||||
latest_time =
|
||||
GridScore
|
||||
|> where([g], not is_nil(g.factors))
|
||||
|> select([g], max(g.valid_time))
|
||||
|> Repo.one()
|
||||
|
||||
|
|
@ -72,7 +77,7 @@ defmodule Microwaveprop.Propagation.ScorerDiff do
|
|||
else
|
||||
query =
|
||||
GridScore
|
||||
|> where([g], g.valid_time == ^latest_time)
|
||||
|> where([g], g.valid_time == ^latest_time and not is_nil(g.factors))
|
||||
|> order_by([g], [g.band_mhz, g.lat, g.lon])
|
||||
|
||||
query = if limit, do: limit(query, ^limit), else: query
|
||||
|
|
|
|||
|
|
@ -183,16 +183,16 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
|
|||
{:weather_updated, valid_time}
|
||||
)
|
||||
|
||||
scores = compute_scores(grid_data, valid_time)
|
||||
scores = compute_scores(grid_data, valid_time, forecast_hour)
|
||||
|
||||
case Propagation.upsert_scores(scores, prune: false) do
|
||||
case Propagation.replace_scores(scores, valid_time) do
|
||||
{:ok, count} ->
|
||||
Logger.info("PropagationGrid: #{label} → #{count} scores for #{valid_time}")
|
||||
warm_cache(valid_time)
|
||||
:ok
|
||||
|
||||
error ->
|
||||
Logger.error("PropagationGrid: #{label} upsert failed: #{inspect(error)}")
|
||||
Logger.error("PropagationGrid: #{label} replace failed: #{inspect(error)}")
|
||||
error
|
||||
end
|
||||
|
||||
|
|
@ -351,21 +351,19 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
|
|||
end)
|
||||
end
|
||||
|
||||
defp compute_scores(grid_data, valid_time) do
|
||||
# Algorithm is the primary scorer. ML score stored in factors for comparison.
|
||||
compute_scores_algorithm(grid_data, valid_time)
|
||||
defp compute_scores(grid_data, valid_time, forecast_hour) do
|
||||
# Algorithm is the primary scorer. `factors` is only populated for
|
||||
# f00 (the analysis hour) — forecast hours skip the JSONB write so
|
||||
# the scoring+upsert phase can land in under a minute instead of
|
||||
# ~4-5 minutes. point_detail on forecast hours returns a nil
|
||||
# breakdown, which the UI tolerates.
|
||||
compute_scores_algorithm(grid_data, valid_time, forecast_hour == 0)
|
||||
end
|
||||
|
||||
defp compute_scores_algorithm(grid_data, valid_time) do
|
||||
defp compute_scores_algorithm(grid_data, valid_time, include_factors?) do
|
||||
grid_data
|
||||
|> Task.async_stream(
|
||||
fn {{lat, lon}, profile} ->
|
||||
band_scores = Propagation.score_grid_point(profile, valid_time, lat, lon)
|
||||
|
||||
Enum.map(band_scores, fn r ->
|
||||
%{lat: lat, lon: lon, valid_time: valid_time, band_mhz: r.band_mhz, score: r.score, factors: r.factors}
|
||||
end)
|
||||
end,
|
||||
&score_one_point(&1, valid_time, include_factors?),
|
||||
max_concurrency: System.schedulers_online() * 2,
|
||||
timeout: 30_000
|
||||
)
|
||||
|
|
@ -374,4 +372,19 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
|
|||
{:exit, _reason} -> []
|
||||
end)
|
||||
end
|
||||
|
||||
defp score_one_point({{lat, lon}, profile}, valid_time, include_factors?) do
|
||||
band_scores = Propagation.score_grid_point(profile, valid_time, lat, lon)
|
||||
|
||||
Enum.map(band_scores, fn r ->
|
||||
%{
|
||||
lat: lat,
|
||||
lon: lon,
|
||||
valid_time: valid_time,
|
||||
band_mhz: r.band_mhz,
|
||||
score: r.score,
|
||||
factors: if(include_factors?, do: r.factors)
|
||||
}
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -0,0 +1,19 @@
|
|||
defmodule Microwaveprop.Repo.Migrations.MakePropagationScoresFactorsNullable do
|
||||
use Ecto.Migration
|
||||
|
||||
@moduledoc """
|
||||
Factors are only populated for analysis-hour (f00) rows now; forecast
|
||||
hours f01-f18 skip the JSONB write entirely to cut scoring+upsert
|
||||
wall time. That means the column must allow NULL.
|
||||
|
||||
The column itself stays on the table (not dropped) so point_detail
|
||||
can still pull the breakdown for the current hour without a second
|
||||
table lookup.
|
||||
"""
|
||||
|
||||
def change do
|
||||
alter table(:propagation_scores) do
|
||||
modify :factors, :map, null: true, from: {:map, null: false}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -0,0 +1,30 @@
|
|||
defmodule Microwaveprop.Repo.Migrations.MakePropagationScoresUnlogged do
|
||||
use Ecto.Migration
|
||||
|
||||
@moduledoc """
|
||||
propagation_scores holds ~475k rows per forecast hour × 19 hours of
|
||||
retention ≈ 9M row churns per 3-hour cron. Writing every row to WAL
|
||||
is a significant share of the scoring+upsert wall time. Set the
|
||||
table UNLOGGED so inserts bypass WAL entirely.
|
||||
|
||||
Durability tradeoff: on an unclean Postgres shutdown the table is
|
||||
truncated on startup. That's acceptable here — PropagationGridWorker
|
||||
rebuilds the full CONUS grid from HRRR every 3 hours, so a lost
|
||||
table is re-populated within one cron cycle. The map page already
|
||||
handles an empty score table gracefully (shows "no data").
|
||||
|
||||
Lock note: SET UNLOGGED rewrites the table, which takes an
|
||||
AccessExclusiveLock for the duration. In prod this momentarily
|
||||
blocks other writers (AsosAdjustmentWorker, PropagationGridWorker,
|
||||
PropagationPruneWorker all share the :propagation queue). Run
|
||||
during a lull. With a 1.4M-row table the rewrite is fast (~seconds).
|
||||
"""
|
||||
|
||||
def up do
|
||||
execute "ALTER TABLE propagation_scores SET UNLOGGED"
|
||||
end
|
||||
|
||||
def down do
|
||||
execute "ALTER TABLE propagation_scores SET LOGGED"
|
||||
end
|
||||
end
|
||||
|
|
@ -28,17 +28,28 @@ defmodule Microwaveprop.Propagation.GridScoreTest do
|
|||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "requires all fields" do
|
||||
test "requires the core fields except factors" do
|
||||
changeset = GridScore.changeset(%GridScore{}, %{})
|
||||
|
||||
errors = errors_on(changeset)
|
||||
|
||||
assert %{
|
||||
lat: ["can't be blank"],
|
||||
lon: ["can't be blank"],
|
||||
valid_time: ["can't be blank"],
|
||||
band_mhz: ["can't be blank"],
|
||||
score: ["can't be blank"],
|
||||
factors: ["can't be blank"]
|
||||
} = errors_on(changeset)
|
||||
score: ["can't be blank"]
|
||||
} = errors
|
||||
|
||||
refute Map.has_key?(errors, :factors)
|
||||
end
|
||||
|
||||
test "persists with nil factors (forecast-hour rows skip the factor breakdown)" do
|
||||
attrs = Map.delete(@valid_attrs, :factors)
|
||||
changeset = GridScore.changeset(%GridScore{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert {:ok, score} = Repo.insert(changeset)
|
||||
assert score.factors == nil
|
||||
end
|
||||
|
||||
test "persists to database" do
|
||||
|
|
|
|||
|
|
@ -170,6 +170,99 @@ defmodule Microwaveprop.PropagationTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "replace_scores/2" do
|
||||
test "inserts all rows when the valid_time slice is empty" do
|
||||
valid_time = ~U[2026-07-15 13:00:00Z]
|
||||
|
||||
scores = [
|
||||
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{humidity: 90}},
|
||||
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 60, factors: %{humidity: 40}}
|
||||
]
|
||||
|
||||
assert {:ok, 2} = Propagation.replace_scores(scores, valid_time)
|
||||
assert Repo.aggregate(GridScore, :count) == 2
|
||||
end
|
||||
|
||||
test "replaces pre-existing rows for the same valid_time" do
|
||||
valid_time = ~U[2026-07-15 13:00:00Z]
|
||||
|
||||
Propagation.upsert_scores([
|
||||
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 10, factors: %{}},
|
||||
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 20, factors: %{}}
|
||||
])
|
||||
|
||||
assert {:ok, 2} =
|
||||
Propagation.replace_scores(
|
||||
[
|
||||
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: %{}},
|
||||
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 60, factors: %{}}
|
||||
],
|
||||
valid_time
|
||||
)
|
||||
|
||||
scores = GridScore |> Repo.all() |> Enum.sort_by(& &1.band_mhz)
|
||||
assert [%{score: 75, band_mhz: 10_000}, %{score: 60, band_mhz: 24_000}] = scores
|
||||
end
|
||||
|
||||
test "leaves rows for other valid_times untouched" do
|
||||
other_time = ~U[2026-07-15 12:00:00Z]
|
||||
replaced_time = ~U[2026-07-15 13:00:00Z]
|
||||
|
||||
Propagation.upsert_scores([
|
||||
%{lat: 35.0, lon: -97.0, valid_time: other_time, band_mhz: 10_000, score: 40, factors: %{}}
|
||||
])
|
||||
|
||||
Propagation.replace_scores(
|
||||
[%{lat: 35.0, lon: -97.0, valid_time: replaced_time, band_mhz: 10_000, score: 90, factors: %{}}],
|
||||
replaced_time
|
||||
)
|
||||
|
||||
assert Repo.aggregate(GridScore, :count) == 2
|
||||
assert Repo.one(from gs in GridScore, where: gs.valid_time == ^other_time, select: gs.score) == 40
|
||||
assert Repo.one(from gs in GridScore, where: gs.valid_time == ^replaced_time, select: gs.score) == 90
|
||||
end
|
||||
|
||||
test "deletes pre-existing rows for the valid_time even when the new list is empty" do
|
||||
# A partial run that produced zero scores (e.g. HRRR fetch failed
|
||||
# for every band) should still clear stale rows so the map doesn't
|
||||
# show data from a previous forecast.
|
||||
valid_time = ~U[2026-07-15 13:00:00Z]
|
||||
|
||||
Propagation.upsert_scores([
|
||||
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 10, factors: %{}}
|
||||
])
|
||||
|
||||
assert {:ok, 0} = Propagation.replace_scores([], valid_time)
|
||||
assert Repo.aggregate(GridScore, :count) == 0
|
||||
end
|
||||
|
||||
test "accepts nil factors (forecast-hour chain steps skip the JSONB write)" do
|
||||
valid_time = ~U[2026-07-15 13:00:00Z]
|
||||
|
||||
scores = [
|
||||
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: nil},
|
||||
%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 24_000, score: 60, factors: nil}
|
||||
]
|
||||
|
||||
assert {:ok, 2} = Propagation.replace_scores(scores, valid_time)
|
||||
assert [nil, nil] = Repo.all(from gs in GridScore, select: gs.factors)
|
||||
end
|
||||
end
|
||||
|
||||
describe "point_detail/4 with forecast-hour rows" do
|
||||
test "coalesces nil factors to an empty map so the JS popup renders" do
|
||||
valid_time = ~U[2026-07-15 13:00:00Z]
|
||||
|
||||
Propagation.replace_scores(
|
||||
[%{lat: 35.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 42, factors: nil}],
|
||||
valid_time
|
||||
)
|
||||
|
||||
assert %{score: 42, factors: %{}} =
|
||||
Propagation.point_detail(10_000, 35.0, -97.0, valid_time)
|
||||
end
|
||||
end
|
||||
|
||||
describe "latest_scores/1" do
|
||||
test "returns latest scores for a band" do
|
||||
valid_time = ~U[2026-07-15 13:00:00Z]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue