graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(weather): GridCache backend pluggable, Valkey for cross-pod cache

Browse source 
Each cached HRRR valid_time is ~32 MiB compressed × cap of 8 = up to
256 MiB per pod. With 4 hot pods + 1 backfill that's ~1.25 GiB of
cluster RAM holding identical content. Replacing the per-pod ETS
replicas with a single shared Valkey copy collapses that to ~256 MiB
total (off-pod) and makes the BEAM heap on each replica meaningfully
smaller — directly addresses the headroom the 2026-05-03 OOM cascade
ate into.

* New `Microwaveprop.Valkey`: single Redix connection (named conn,
  sync_connect: false, exit_on_disconnection: false). `start_link/0`
  returns `:ignore` when VALKEY_URL is unset so dev/test boot without
  Valkey. Helpers wrap GET/MGET/SET-with-TTL/ZADD/ZREVRANGE/SCAN with
  `:erlang.term_to_binary` round-tripping under the safe atom flag.

* `Microwaveprop.Weather.GridCache` now branches on
  `Valkey.configured?/0`. Public API is identical so callers (Weather,
  GridCachePruneWorker, MapLive) don't change. ETS path is preserved
  unchanged for dev/test.

* Storage layout when on Valkey:
    prop:wg:vts                   ZSET of valid_time iso8601 strings
    prop:wg:<vt>:<lat>:<lon>      one chunk per 5°×5° spatial bucket
  fetch_bounds MGETs only the chunks intersecting the viewport, so a
  regional view is 1-4 round-trips instead of pulling the full grid.

* Per-key TTL = 8 h (matches the ETS valid_time cap). Eviction is
  automatic; prune_keep_latest is a no-op on Valkey, prune_older_than
  uses ZRANGEBYSCORE + DEL.

* All Valkey errors fall back to disk (ScalarFile.read_bounds), logged
  as warnings — page works through Valkey outages, no user-visible
  break beyond a one-shot latency bump.

ScoreCache, Microwaveprop.Cache, NexradCache, telemetry/Postgres
protocol/libcluster registry tables remain ETS — they're sub-ms
hot-path lookups where a network hop would be felt.

VALKEY_URL must be present in prop-secrets for production rollout
(applied out-of-band, secret.yaml is git-ignored).
This commit is contained in:
Graham McIntire 2026-05-03 17:00:51 -05:00
parent 803130ee64
commit 9edaee9344
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
6 changed files with 647 additions and 98 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
config/runtime.exs
View file

@ -369,6 +369,15 @@ if config_env() == :prod do
# pool_count: 4,
socket_options: maybe_ipv6
# Cross-pod cache backend (Valkey). Optional — when unset, GridCache
# falls back to the legacy per-pod ETS implementation. Set to a Redis
# URL like `redis://:password@10.0.15.21:6379/0` in prop-secrets.
case System.get_env("VALKEY_URL") do
nil -> :ok
"" -> :ok
url -> config :microwaveprop, :valkey_url, url
end
# Optional secondary repo for read-only access to aprs.me's database.
# Used by APRS-based 144 MHz calibration tooling. When the env var is
# unset OR an empty string (k8s secret resolves to "") we leave the

5
lib/microwaveprop/application.ex
View file

@ -29,6 +29,11 @@ defmodule Microwaveprop.Application do
# downloads, GEFS scoring) do not serialize on a single supervisor PID.
{PartitionSupervisor,
child_spec: Task.Supervisor, name: Microwaveprop.TaskSupervisor, partitions: System.schedulers_online()},
# Valkey connection — must start before GridCache so the cache
# backend selector sees an alive Redix process. Returns :ignore
# when VALKEY_URL is unset (dev/test), and GridCache falls back
# to its ETS implementation.
Microwaveprop.Valkey,
Microwaveprop.Cache,
Microwaveprop.Buildings.Index,
Microwaveprop.Propagation.ScoreCache,

246
lib/microwaveprop/valkey.ex Normal file
View file

@ -0,0 +1,246 @@
defmodule Microwaveprop.Valkey do
@moduledoc """
Thin wrapper around a single Redix connection to the cluster's
Valkey instance. Used by `Microwaveprop.Weather.GridCache` to back
the cross-pod weather-grid cache so the per-pod ETS replica
(~768 MiB worst case × N pods) collapses to a single shared copy.
Connection URL is read from `:microwaveprop, :valkey_url` at boot
(set in `runtime.exs` from the `VALKEY_URL` env var). When unset the
child returns `:ignore` and `configured?/0` is `false`, so callers
can fall back to ETS / disk in dev and test without Valkey running.
Public read/write helpers term-encode their values with
`:erlang.term_to_binary/1` so atom-keyed maps and DateTimes round-trip
cleanly. Decode uses the `:safe` flag — only existing atoms are
resolvable, so a malicious Valkey can't blow the atom table.
"""
require Logger
@conn __MODULE__.Conn
@spec child_spec(keyword()) :: Supervisor.child_spec()
def child_spec(_opts) do
%{
id: __MODULE__,
start: {__MODULE__, :start_link, []},
type: :worker,
restart: :permanent
}
end
@spec start_link() :: GenServer.on_start() | :ignore
def start_link do
case url() do
nil ->
Logger.info("Microwaveprop.Valkey not configured (VALKEY_URL unset) — GridCache will use ETS fallback.")
:ignore
url ->
# sync_connect: false lets the app boot even when Valkey is
# briefly unreachable; reads/writes during the gap return
# {:error, _} and the GridCache caller falls back to disk.
Redix.start_link(url,
name: @conn,
sync_connect: false,
exit_on_disconnection: false
)
end
end
@doc "True when VALKEY_URL is configured AND the Redix process is alive."
@spec configured?() :: boolean()
def configured? do
case Process.whereis(@conn) do
nil -> false
pid when is_pid(pid) -> Process.alive?(pid)
end
end
@doc """
GET a single key. Returns `{:ok, term}` on hit, `:miss` when the key
doesn't exist or is expired, `{:error, reason}` on connection / decode
failure. The error tuple lets callers fall back to disk without
crashing the request.
"""
@spec get(String.t()) :: {:ok, term()} | :miss | {:error, term()}
def get(key) when is_binary(key) do
case command(["GET", key]) do
{:ok, nil} -> :miss
{:ok, bin} when is_binary(bin) -> safe_decode(bin)
{:error, _} = err -> err
end
end
@doc """
MGET a list of keys. Returns `{:ok, [term_or_nil]}` preserving the
input order, where each element is the decoded term or `nil` on miss.
Errors propagate as `{:error, _}` so the caller falls back to disk
for the entire bounds query.
"""
@spec mget([String.t()]) :: {:ok, [term() | nil]} | {:error, term()}
def mget([]), do: {:ok, []}
def mget(keys) when is_list(keys) do
case command(["MGET" | keys]) do
{:ok, vals} ->
decoded =
Enum.map(vals, fn
nil -> nil
bin -> bin |> safe_decode() |> elem_or_nil()
end)
{:ok, decoded}
{:error, _} = err ->
err
end
end
@doc """
SET key with the term-encoded value and an EX TTL in seconds. Used
by the cache fill path; on connection failure returns `{:error, _}`
and the caller logs + continues (cache miss next time, refilled then).
"""
@spec set(String.t(), term(), pos_integer()) :: :ok | {:error, term()}
def set(key, term, ttl_seconds) when is_binary(key) and is_integer(ttl_seconds) and ttl_seconds > 0 do
bin = :erlang.term_to_binary(term)
case command(["SET", key, bin, "EX", Integer.to_string(ttl_seconds)]) do
{:ok, "OK"} -> :ok
{:error, _} = err -> err
other -> {:error, other}
end
end
@doc """
Pipeline N SETs with TTL in a single round-trip. Returns `:ok` if
every command succeeded, `{:error, _}` otherwise.
"""
@spec mset_with_ttl([{String.t(), term()}], pos_integer()) :: :ok | {:error, term()}
def mset_with_ttl([], _ttl), do: :ok
def mset_with_ttl(pairs, ttl_seconds) when is_list(pairs) and is_integer(ttl_seconds) do
cmds =
Enum.map(pairs, fn {key, term} ->
["SET", key, :erlang.term_to_binary(term), "EX", Integer.to_string(ttl_seconds)]
end)
case pipeline(cmds) do
{:ok, results} ->
if Enum.all?(results, &(&1 == "OK")), do: :ok, else: {:error, {:partial, results}}
{:error, _} = err ->
err
end
end
@doc """
Sorted-set add. Used by `GridCache` to maintain a `prop:wg:vts` index
of cached valid_times so `latest_valid_time/0` can `ZREVRANGE` instead
of scanning all keys.
"""
@spec zadd(String.t(), number(), String.t()) :: :ok | {:error, term()}
def zadd(key, score, member) do
case command(["ZADD", key, to_string(score), member]) do
{:ok, _} -> :ok
{:error, _} = err -> err
end
end
@spec zrevrange(String.t(), integer(), integer()) :: {:ok, [String.t()]} | {:error, term()}
def zrevrange(key, start, stop) do
case command(["ZREVRANGE", key, to_string(start), to_string(stop)]) do
{:ok, members} -> {:ok, members}
{:error, _} = err -> err
end
end
@spec zrangebyscore(String.t(), String.t(), String.t()) :: {:ok, [String.t()]} | {:error, term()}
def zrangebyscore(key, min, max) do
case command(["ZRANGEBYSCORE", key, min, max]) do
{:ok, members} -> {:ok, members}
{:error, _} = err -> err
end
end
@spec zrem(String.t(), [String.t()]) :: :ok | {:error, term()}
def zrem(_key, []), do: :ok
def zrem(key, members) when is_list(members) do
case command(["ZREM", key | members]) do
{:ok, _} -> :ok
{:error, _} = err -> err
end
end
@spec del([String.t()]) :: :ok | {:error, term()}
def del([]), do: :ok
def del(keys) when is_list(keys) do
case command(["DEL" | keys]) do
{:ok, _} -> :ok
{:error, _} = err -> err
end
end
@doc """
Run a single SCAN MATCH pass and return all matching keys. Used only
by maintenance / test paths — the read path uses ZSET indexes instead.
"""
@spec scan_match(String.t()) :: {:ok, [String.t()]} | {:error, term()}
def scan_match(pattern) do
do_scan_match(pattern, "0", [])
end
defp do_scan_match(pattern, cursor, acc) do
case command(["SCAN", cursor, "MATCH", pattern, "COUNT", "500"]) do
{:ok, [next_cursor, keys]} ->
new_acc = acc ++ keys
case next_cursor do
"0" -> {:ok, new_acc}
other -> do_scan_match(pattern, other, new_acc)
end
{:error, _} = err ->
err
end
end
defp command(args) do
if configured?() do
try do
Redix.command(@conn, args, timeout: 1500)
catch
:exit, reason -> {:error, {:exit, reason}}
end
else
{:error, :not_configured}
end
end
defp pipeline(cmds) do
if configured?() do
try do
Redix.pipeline(@conn, cmds, timeout: 3000)
catch
:exit, reason -> {:error, {:exit, reason}}
end
else
{:error, :not_configured}
end
end
defp safe_decode(bin) do
{:ok, :erlang.binary_to_term(bin, [:safe])}
rescue
e -> {:error, {:decode, Exception.message(e)}}
end
defp elem_or_nil({:ok, term}), do: term
defp elem_or_nil(_), do: nil
defp url, do: Application.get_env(:microwaveprop, :valkey_url)
end

480
lib/microwaveprop/weather/grid_cache.ex
View file

@ -1,176 +1,179 @@
defmodule Microwaveprop.Weather.GridCache do
@moduledoc """
Node-local ETS cache of derived HRRR grid rows keyed by `valid_time`. Mirrors
`Microwaveprop.Propagation.ScoreCache` but for the `/weather` map.
Cross-pod cache of derived HRRR grid rows keyed by `valid_time`. When
`Microwaveprop.Valkey` is configured, the chunked map for each
valid_time is held in Valkey (one key per 5°×5° spatial chunk plus a
ZSET index of cached valid_times). Otherwise the cache stays
pod-local in ETS — preserved so dev / test work without Valkey.
The Weather map LiveView calls `latest_weather_grid/1` on mount and every
pan/zoom. Each call otherwise hits the 42M-row partitioned `hrrr_profiles`
table, runs per-row `derive_and_clean` transforms, and returns 3-10k rows.
With this cache those calls become in-memory map iterations.
Each valid_time worth of CONUS HRRR is ~92k cells × 22 fields ≈
32 MiB compressed. Holding that in pod-local ETS meant N replicas
paid the full memory cost per pod. With the Valkey backend a single
cluster-wide copy serves every reader.
## Storage layout
## Storage layout (Valkey)
Each cache entry is bucketed into 5°×5° spatial chunks matching
`Microwaveprop.Weather.ScalarFile`'s on-disk layout. The ETS value for
`valid_time` is `%{{lat_band, lon_band} => %{{lat, lon} => row}}`. That
way `fetch_bounds/2` only walks the chunks that intersect the requested
viewport instead of the full 92k-cell CONUS map. `fetch_point/3` reads
exactly one chunk.
* `prop:wg:vts` — ZSET of cached valid_time iso8601 strings, score
= unix epoch seconds. `latest_valid_time/0` does ZREVRANGE 0 0.
* `prop:wg:<vt_iso>:<lat_band>:<lon_band>` — string key per chunk,
value is `:erlang.term_to_binary(%{{lat, lon} => row, ...})`.
`fetch_bounds/2` MGETs only the chunks intersecting the viewport.
Populated by `Microwaveprop.Weather.warm_grid_cache/1` after the hourly
worker upserts new HRRR data, fanned out across the cluster via the
`"weather:cache"` PubSub topic so every node stays in sync.
Per-key TTL is `@ttl_seconds`, so cache eviction is automatic — no
prune worker needed when running on Valkey. The ETS path keeps the
legacy `prune_keep_latest`/`prune_older_than` helpers since they're
still called by `Microwaveprop.Weather.fill_grid_cache_from_scalar/1`
on every fill.
## Fill coordination
Cache fills run `ScalarFile.read_bounds/2` which gunzips and decodes
per-chunk files (~15 s for a full CONUS valid_time). The
`:weather_grid_fill_locks` ETS table (kept regardless of backend)
stops N concurrent /weather mounts on a single pod from each firing
the slow read. Multi-pod duplicate fills are tolerated: the second
pod's writes overwrite identical bytes in Valkey, no data corruption.
"""
use GenServer
alias Microwaveprop.Valkey
alias Phoenix.PubSub
require Logger
@table :weather_grid_cache
@lock_table :weather_grid_fill_locks
@topic "weather:cache"
@pubsub Microwaveprop.PubSub
@chunk_step 5
# Per-chunk TTL in Valkey. 8 hours covers analysis + 7 forecast hours
# (matches the ETS valid_time cap). Beyond 8 h, fall back to disk.
@ttl_seconds 8 * 3600
@valkey_index_key "prop:wg:vts"
@valkey_chunk_prefix "prop:wg:"
@type row :: %{required(:lat) => float(), required(:lon) => float(), optional(atom()) => any()}
@type bounds :: %{optional(String.t()) => float()}
@spec start_link(keyword()) :: GenServer.on_start()
def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
# ---------- Public API ----------
@spec fetch(DateTime.t()) :: {:ok, [row()]} | :miss
def fetch(valid_time) do
case :ets.lookup(@table, valid_time) do
[{_, chunked}] ->
emit_lookup(true)
{:ok, chunks_to_list(chunked)}
[] ->
emit_lookup(false)
:miss
if valkey_backend?() do
valkey_fetch(valid_time)
else
ets_fetch(valid_time)
end
end
@spec fetch_bounds(DateTime.t(), bounds() | nil) :: {:ok, [row()]} | :miss
def fetch_bounds(valid_time, bounds) do
case :ets.lookup(@table, valid_time) do
[{_, chunked}] ->
emit_lookup(true)
{:ok, chunks_filtered_to_list(chunked, bounds)}
[] ->
emit_lookup(false)
:miss
if valkey_backend?() do
valkey_fetch_bounds(valid_time, bounds)
else
ets_fetch_bounds(valid_time, bounds)
end
end
@spec fetch_point(DateTime.t(), float(), float()) :: {:ok, row()} | :miss
def fetch_point(valid_time, lat, lon) do
case :ets.lookup(@table, valid_time) do
[{_, chunked}] ->
chunk_key = chunk_key_for(lat, lon)
case chunked |> Map.get(chunk_key, %{}) |> Map.get({lat, lon}) do
nil ->
emit_lookup(false)
:miss
row ->
emit_lookup(true)
{:ok, row}
end
[] ->
emit_lookup(false)
:miss
if valkey_backend?() do
valkey_fetch_point(valid_time, lat, lon)
else
ets_fetch_point(valid_time, lat, lon)
end
end
defp emit_lookup(hit) do
:telemetry.execute([:microwaveprop, :weather, :grid_cache, :lookup], %{}, %{hit: hit})
end
@spec put(DateTime.t(), [row()]) :: :ok
def put(valid_time, rows) do
chunked = list_to_chunks(rows)
:ets.insert(@table, {valid_time, chunked})
:ok
if valkey_backend?() do
valkey_put(valid_time, chunked)
else
:ets.insert(@table, {valid_time, chunked})
:ok
end
end
@doc "Insert locally AND broadcast to peer nodes via PubSub."
@doc """
Insert (Valkey path) or insert + broadcast (ETS path). Kept as a single
public name so legacy callers don't have to branch.
"""
@spec broadcast_put(DateTime.t(), [row()]) :: :ok
def broadcast_put(valid_time, rows) do
_ = PubSub.broadcast(@pubsub, @topic, {:weather_cache_refresh, valid_time, rows})
:ok
if valkey_backend?() do
# Valkey is the single source of truth — peers will read the same
# data on their next fetch. Broadcasting becomes redundant.
put(valid_time, rows)
else
_ = PubSub.broadcast(@pubsub, @topic, {:weather_cache_refresh, valid_time, rows})
:ok
end
end
@spec latest_valid_time() :: DateTime.t() | nil
def latest_valid_time do
match_spec = [{{:"$1", :_}, [], [:"$1"]}]
case :ets.select(@table, match_spec) do
[] -> nil
times -> Enum.max(times, DateTime)
if valkey_backend?() do
valkey_latest_valid_time()
else
ets_latest_valid_time()
end
end
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
def prune_older_than(cutoff) do
match_spec = [{{:"$1", :_}, [{:<, :"$1", {:const, cutoff}}], [true]}]
:ets.select_delete(@table, match_spec)
if valkey_backend?() do
valkey_prune_older_than(cutoff)
else
match_spec = [{{:"$1", :_}, [{:<, :"$1", {:const, cutoff}}], [true]}]
:ets.select_delete(@table, match_spec)
end
end
@doc """
Keep only the `keep` most-recent valid_times, dropping the rest. Returns
the number of entries removed. Used to bound memory when forecast hours
are cached on demand.
Keep only the `keep` most-recent valid_times, dropping the rest. On
Valkey this is a no-op because per-key TTL handles eviction; the ETS
path is unchanged.
"""
@spec prune_keep_latest(non_neg_integer()) :: non_neg_integer()
def prune_keep_latest(keep) when is_integer(keep) and keep >= 0 do
case :ets.select(@table, [{{:"$1", :_}, [], [:"$1"]}]) do
[] ->
0
times ->
sorted = Enum.sort(times, {:desc, DateTime})
to_drop = Enum.drop(sorted, keep)
Enum.each(to_drop, &:ets.delete(@table, &1))
length(to_drop)
if valkey_backend?() do
0
else
ets_prune_keep_latest(keep)
end
end
@spec clear() :: :ok
def clear do
GenServer.call(__MODULE__, :clear)
end
def clear, do: GenServer.call(__MODULE__, :clear)
@doc """
Atomically claim the right to fill the cache for `valid_time`. Returns
`true` if this caller won the claim and should run the fill; `false` if
another caller is already filling. Prevents N concurrent /weather mounts
after a pod restart from each firing the 15-second cold-fill read and
starving the Postgres connection pool.
The GenServer `Process.monitor`s the caller: if the caller crashes
before calling `release_fill/1`, the lock is released automatically.
"""
@spec claim_fill(term()) :: boolean()
def claim_fill(valid_time) do
GenServer.call(__MODULE__, {:claim_fill, valid_time, self()})
end
@doc "Release a fill lock claimed via `claim_fill/1`."
@spec release_fill(term()) :: :ok
def release_fill(valid_time) do
GenServer.call(__MODULE__, {:release_fill, valid_time})
end
@spec sync() :: :ok
def sync do
GenServer.call(__MODULE__, :sync)
end
def sync, do: GenServer.call(__MODULE__, :sync)
# ---------- GenServer ----------
@impl true
def init(_opts) do
# ETS table is created unconditionally so the legacy backend (and
# tests that bypass Valkey) work without separate setup. The lock
# table is also unconditional — fill coordination still happens
# per-pod regardless of where the data lives.
_ = :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true])
_ = :ets.new(@lock_table, [:set, :named_table, :protected])
:ok = PubSub.subscribe(@pubsub, @topic)
@ -181,9 +184,8 @@ defmodule Microwaveprop.Weather.GridCache do
def handle_call(:sync, _from, state), do: {:reply, :ok, state}
def handle_call(:clear, _from, state) do
:ets.delete_all_objects(@table)
if valkey_backend?(), do: valkey_clear(), else: :ets.delete_all_objects(@table)
# Demonitor all tracked callers and drop every lock so tests start clean.
for {ref, _vt} <- state.monitors, do: Process.demonitor(ref, [:flush])
:ets.delete_all_objects(@lock_table)
@ -218,7 +220,9 @@ defmodule Microwaveprop.Weather.GridCache do
end
def handle_info({:weather_cache_refresh, valid_time, rows}, state) do
put(valid_time, rows)
# Only meaningful on the ETS backend. With Valkey, broadcasting was
# disabled at the source and this handler will simply not fire.
if !valkey_backend?(), do: put(valid_time, rows)
{:noreply, state}
end
@ -235,7 +239,287 @@ defmodule Microwaveprop.Weather.GridCache do
end
end
# ---------- Chunk helpers ----------
# ---------- Backend selector ----------
defp valkey_backend?, do: Valkey.configured?()
# ---------- ETS backend ----------
defp ets_fetch(valid_time) do
case :ets.lookup(@table, valid_time) do
[{_, chunked}] ->
emit_lookup(true)
{:ok, chunks_to_list(chunked)}
[] ->
emit_lookup(false)
:miss
end
end
defp ets_fetch_bounds(valid_time, bounds) do
case :ets.lookup(@table, valid_time) do
[{_, chunked}] ->
emit_lookup(true)
{:ok, chunks_filtered_to_list(chunked, bounds)}
[] ->
emit_lookup(false)
:miss
end
end
defp ets_fetch_point(valid_time, lat, lon) do
case :ets.lookup(@table, valid_time) do
[{_, chunked}] ->
chunk_key = chunk_key_for(lat, lon)
case chunked |> Map.get(chunk_key, %{}) |> Map.get({lat, lon}) do
nil ->
emit_lookup(false)
:miss
row ->
emit_lookup(true)
{:ok, row}
end
[] ->
emit_lookup(false)
:miss
end
end
defp ets_latest_valid_time do
match_spec = [{{:"$1", :_}, [], [:"$1"]}]
case :ets.select(@table, match_spec) do
[] -> nil
times -> Enum.max(times, DateTime)
end
end
defp ets_prune_keep_latest(keep) do
case :ets.select(@table, [{{:"$1", :_}, [], [:"$1"]}]) do
[] ->
0
times ->
sorted = Enum.sort(times, {:desc, DateTime})
to_drop = Enum.drop(sorted, keep)
Enum.each(to_drop, &:ets.delete(@table, &1))
length(to_drop)
end
end
# ---------- Valkey backend ----------
defp valkey_fetch(valid_time) do
case Valkey.scan_match(chunk_key_pattern(valid_time)) do
{:ok, []} ->
emit_lookup(false)
:miss
{:ok, keys} ->
merge_fetched_chunks(keys)
{:error, reason} ->
log_valkey_error(:scan, reason)
:miss
end
end
defp merge_fetched_chunks(keys) do
case Valkey.mget(keys) do
{:ok, vals} ->
chunks =
vals
|> Enum.reject(&is_nil/1)
|> Enum.reduce(%{}, &Map.merge(&2, decode_chunk_value(&1)))
if map_size(chunks) == 0 do
emit_lookup(false)
:miss
else
emit_lookup(true)
{:ok, chunks_to_list(chunks)}
end
{:error, reason} ->
log_valkey_error(:fetch, reason)
:miss
end
end
defp valkey_fetch_bounds(valid_time, nil), do: valkey_fetch(valid_time)
defp valkey_fetch_bounds(valid_time, %{"south" => _, "north" => _, "west" => _, "east" => _} = bounds) do
keys = chunk_keys_in_bounds(valid_time, bounds)
case Valkey.mget(keys) do
{:ok, vals} ->
rows = collect_rows_in_bounds(vals, bounds)
if rows == [] do
emit_lookup(false)
:miss
else
emit_lookup(true)
{:ok, rows}
end
{:error, reason} ->
log_valkey_error(:mget, reason)
:miss
end
end
defp collect_rows_in_bounds(vals, bounds) do
vals
|> Enum.reject(&is_nil/1)
|> Enum.flat_map(fn cells ->
cells |> decode_chunk_value() |> Enum.flat_map(&cells_for_chunk_in_bounds(&1, bounds))
end)
end
defp cells_for_chunk_in_bounds({_chunk_key, cell_map}, bounds) do
cells_in_bounds(cell_map, bounds)
end
defp valkey_fetch_point(valid_time, lat, lon) do
chunk_key = chunk_key_for(lat, lon)
redis_key = chunk_redis_key(valid_time, chunk_key)
case Valkey.get(redis_key) do
{:ok, %{} = chunk_map} ->
case Map.get(chunk_map, {lat, lon}) do
nil ->
emit_lookup(false)
:miss
row ->
emit_lookup(true)
{:ok, row}
end
:miss ->
emit_lookup(false)
:miss
{:error, reason} ->
log_valkey_error(:get, reason)
:miss
end
end
defp valkey_put(valid_time, chunked) do
pairs =
Enum.map(chunked, fn {chunk_key, cells} ->
{chunk_redis_key(valid_time, chunk_key), %{chunk_key => cells}}
end)
case Valkey.mset_with_ttl(pairs, @ttl_seconds) do
:ok ->
_ = Valkey.zadd(@valkey_index_key, DateTime.to_unix(valid_time), DateTime.to_iso8601(valid_time))
:ok
{:error, reason} ->
log_valkey_error(:mset, reason)
:ok
end
end
defp valkey_latest_valid_time do
case Valkey.zrevrange(@valkey_index_key, 0, 0) do
{:ok, [iso | _]} ->
case DateTime.from_iso8601(iso) do
{:ok, dt, _} -> dt
_ -> nil
end
_ ->
nil
end
end
defp valkey_prune_older_than(cutoff) do
cutoff_score = DateTime.to_unix(cutoff)
case Valkey.zrangebyscore(@valkey_index_key, "-inf", "(#{cutoff_score}") do
{:ok, []} ->
0
{:ok, members} ->
keys = Enum.flat_map(members, &chunk_keys_for_iso/1)
_ = Valkey.del(keys)
_ = Valkey.zrem(@valkey_index_key, members)
length(members)
_ ->
0
end
end
defp chunk_keys_for_iso(iso) do
case DateTime.from_iso8601(iso) do
{:ok, dt, _} ->
case Valkey.scan_match(chunk_key_pattern(dt)) do
{:ok, ks} -> ks
_ -> []
end
_ ->
[]
end
end
defp valkey_clear do
case Valkey.scan_match("#{@valkey_chunk_prefix}*") do
{:ok, keys} -> Valkey.del(keys)
_ -> :ok
end
Valkey.del([@valkey_index_key])
end
defp chunk_key_pattern(valid_time) do
"#{@valkey_chunk_prefix}#{DateTime.to_iso8601(valid_time)}:*"
end
defp chunk_redis_key(valid_time, {lat_band, lon_band}) do
"#{@valkey_chunk_prefix}#{DateTime.to_iso8601(valid_time)}:#{lat_band}:#{lon_band}"
end
defp chunk_keys_in_bounds(valid_time, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
s_band = chunk_band(s * 1.0)
n_band = chunk_band(n * 1.0)
w_band = chunk_band(w * 1.0)
e_band = chunk_band(e * 1.0)
for lat <- s_band..n_band, lon <- w_band..e_band do
chunk_redis_key(valid_time, {lat, lon})
end
end
defp cells_in_bounds(cell_map, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
for {{lat, lon}, row} <- cell_map, lat >= s, lat <= n, lon >= w, lon <= e, do: row
end
# The chunk value persisted to Valkey is `%{chunk_key => cells_map}`
# so the decoder can return the cells to a uniform shape regardless of
# whether we read 1 chunk (fetch_point/fetch_bounds) or many (fetch).
defp decode_chunk_value(%{} = single_chunk_map), do: single_chunk_map
defp decode_chunk_value(_), do: %{}
defp log_valkey_error(op, reason) do
Logger.warning("GridCache Valkey #{op} failed (#{inspect(reason)}); falling back to disk.")
end
defp emit_lookup(hit) do
:telemetry.execute([:microwaveprop, :weather, :grid_cache, :lookup], %{}, %{hit: hit})
end
# ---------- Chunk helpers (shared) ----------
defp list_to_chunks(rows) do
Enum.reduce(rows, %{}, fn %{lat: lat, lon: lon} = row, acc ->

4
mix.exs
View file

@ -103,6 +103,10 @@ defmodule Microwaveprop.MixProject do
{:polaris, "~> 0.1", only: [:dev, :test]},
{:tidewave, "~> 0.5", only: :dev},
{:libcluster, "~> 3.4"},
# Cross-pod cache backend for Microwaveprop.Weather.GridCache.
# Talks to the Valkey/Redis instance at VALKEY_URL; falls back to
# the legacy ETS implementation when unset (dev/test).
{:redix, "~> 1.5"},
{:ecto_psql_extras, "~> 0.8"},
{:dialyxir, "~> 1.4", only: [:dev, :test], runtime: false},
{:live_table, "~> 0.4.1"},

1
mix.lock
View file

@ -59,6 +59,7 @@
"postgrex": {:hex, :postgrex, "0.22.0", "fb027b58b6eab1f6de5396a2abcdaaeb168f9ed4eccbb594e6ac393b02078cbd", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "a68c4261e299597909e03e6f8ff5a13876f5caadaddd0d23af0d0a61afcc5d84"},
"prom_ex": {:hex, :prom_ex, "1.11.0", "1f6d67f2dead92224cb4f59beb3e4d319257c5728d9638b4a5e8ceb51a4f9c7e", [:mix], [{:absinthe, ">= 1.7.0", [hex: :absinthe, repo: "hexpm", optional: true]}, {:broadway, ">= 1.1.0", [hex: :broadway, repo: "hexpm", optional: true]}, {:ecto, ">= 3.11.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:finch, "~> 0.18", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:oban, ">= 2.10.0", [hex: :oban, repo: "hexpm", optional: true]}, {:octo_fetch, "~> 0.4", [hex: :octo_fetch, repo: "hexpm", optional: false]}, {:peep, "~> 3.0", [hex: :peep, repo: "hexpm", optional: false]}, {:phoenix, ">= 1.7.0", [hex: :phoenix, repo: "hexpm", optional: true]}, {:phoenix_live_view, ">= 0.20.0", [hex: :phoenix_live_view, repo: "hexpm", optional: true]}, {:plug, ">= 1.16.0", [hex: :plug, repo: "hexpm", optional: true]}, {:plug_cowboy, ">= 2.6.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:telemetry, ">= 1.0.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}, {:telemetry_metrics_prometheus_core, "~> 1.2", [hex: :telemetry_metrics_prometheus_core, repo: "hexpm", optional: false]}, {:telemetry_poller, "~> 1.1", [hex: :telemetry_poller, repo: "hexpm", optional: false]}], "hexpm", "76b074bc3730f0802978a7eb5c7091a65473eaaf07e99ec9e933138dcc327805"},
"ranch": {:hex, :ranch, "2.2.0", "25528f82bc8d7c6152c57666ca99ec716510fe0925cb188172f41ce93117b1b0", [:make, :rebar3], [], "hexpm", "fa0b99a1780c80218a4197a59ea8d3bdae32fbff7e88527d7d8a4787eff4f8e7"},
"redix": {:hex, :redix, "1.5.3", "4eaae29c75e3285c0ff9957046b7c209aa7f72a023a17f0a9ea51c2a50ab5b0f", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:nimble_options, "~> 0.5.0 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7b06fb5246373af41f5826b03334dfa3f636347d4d5d98b4d455b699d425ae7e"},
"req": {:hex, :req, "0.5.17", "0096ddd5b0ed6f576a03dde4b158a0c727215b15d2795e59e0916c6971066ede", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0b8bc6ffdfebbc07968e59d3ff96d52f2202d0536f10fef4dc11dc02a2a43e39"},
"rewrite": {:hex, :rewrite, "1.3.0", "67448ba7975690b35ba7e7f35717efcce317dbd5963cb0577aa7325c1923121a", [:mix], [{:glob_ex, "~> 0.1", [hex: :glob_ex, repo: "hexpm", optional: false]}, {:sourceror, "~> 1.0", [hex: :sourceror, repo: "hexpm", optional: false]}, {:text_diff, "~> 0.1", [hex: :text_diff, repo: "hexpm", optional: false]}], "hexpm", "d111ac7ff3a58a802ef4f193bbd1831e00a9c57b33276e5068e8390a212714a5"},
"sourceror": {:hex, :sourceror, "1.12.0", "da354c5f35aad3cc1132f5d5b0d8437d865e2661c263260480bab51b5eedb437", [:mix], [], "hexpm", "755703683bd014ebcd5de9acc24b68fb874a660a568d1d63f8f98cd8a6ef9cd0"},