diff --git a/config/runtime.exs b/config/runtime.exs index 4f73b6be..a443e55e 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -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 diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 7781f48d..94ce33cd 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -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, diff --git a/lib/microwaveprop/valkey.ex b/lib/microwaveprop/valkey.ex new file mode 100644 index 00000000..a01d4526 --- /dev/null +++ b/lib/microwaveprop/valkey.ex @@ -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 diff --git a/lib/microwaveprop/weather/grid_cache.ex b/lib/microwaveprop/weather/grid_cache.ex index 32215e79..5631cbde 100644 --- a/lib/microwaveprop/weather/grid_cache.ex +++ b/lib/microwaveprop/weather/grid_cache.ex @@ -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:::` — 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 -> diff --git a/mix.exs b/mix.exs index 390fb6d3..1f5991b6 100644 --- a/mix.exs +++ b/mix.exs @@ -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"}, diff --git a/mix.lock b/mix.lock index c4ea8aa5..3bc9f305 100644 --- a/mix.lock +++ b/mix.lock @@ -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"},