diff --git a/lib/cache/multi_layer.ex b/lib/cache/multi_layer.ex index 013ec7f..b15b184 100644 --- a/lib/cache/multi_layer.ex +++ b/lib/cache/multi_layer.ex @@ -69,6 +69,33 @@ defmodule Cache.MultiLayer do end ``` + ## Cross-Node Coherence (Optional) + + Node-local fast layers (e.g. `Cache.ETS`) go stale on every node except the + writer. Setting `broadcast_mode` keeps them coherent: after a successful + `put`/`delete`, every other node running this cache (tracked via `:pg` — + see `Cache.MultiLayer.Coordinator`) is notified and applies the change to + its own `broadcast_layers`. + + ```elixir + defmodule MyApp.LayeredCache do + use Cache, + adapter: {Cache.MultiLayer, [MyApp.EtsLayer, MyApp.RedisCache]}, + name: :layered_cache, + opts: [ + backfill_ttl: :timer.seconds(30), + broadcast_mode: :invalidate, + broadcast_layers: [MyApp.EtsLayer] + ] + end + ``` + + `:invalidate` sends only the key — remote nodes drop their local entry and + lazily re-read through the shared layer. `:replicate` ships the value so + remote local layers are updated immediately; use it only for small values. + Delivery is best-effort: always keep `backfill_ttl` (and layer TTLs) as the + correctness floor for members that miss a message. + ## Options #{NimbleOptions.docs([ @@ -79,6 +106,14 @@ defmodule Cache.MultiLayer do backfill_ttl: [ type: {:or, [:pos_integer, nil]}, doc: "TTL in milliseconds to use when backfilling layers on a hit from a slower layer. Defaults to nil (no expiry)." + ], + broadcast_mode: [ + type: {:in, [:invalidate, :replicate]}, + doc: "Cross-node coherence for writes: `:invalidate` deletes the key from other nodes' `broadcast_layers`; `:replicate` pushes the written value to them. Best-effort delivery." + ], + broadcast_layers: [ + type: {:list, :atom}, + doc: "Node-local layer modules the broadcast applies to on other nodes. Required with `broadcast_mode`; must not include the shared (slowest) layer." ] ])} """ @@ -93,6 +128,16 @@ defmodule Cache.MultiLayer do backfill_ttl: [ type: {:or, [:pos_integer, nil]}, doc: "TTL for backfilled entries." + ], + broadcast_mode: [ + type: {:in, [:invalidate, :replicate]}, + doc: + "Cross-node coherence for writes: :invalidate deletes the key from other nodes' broadcast_layers (next read falls through and backfills fresh); :replicate pushes the written value to them. Best-effort delivery — keep TTLs as the correctness floor." + ], + broadcast_layers: [ + type: {:list, :atom}, + doc: + "Node-local layer modules the broadcast applies to on the other nodes. Required when broadcast_mode is set; must not include the shared (slowest) layer." ] ] @@ -101,10 +146,7 @@ defmodule Cache.MultiLayer do @impl Cache.Strategy def child_spec({cache_name, _layers, _adapter_opts}) do - %{ - id: :"#{cache_name}_multi_layer", - start: {Agent, :start_link, [fn -> :ok end, [name: :"#{cache_name}_multi_layer"]]} - } + Cache.MultiLayer.Coordinator.child_spec(cache_name) end @impl Cache.Strategy @@ -124,17 +166,67 @@ defmodule Cache.MultiLayer do @impl Cache.Strategy def put(cache_name, key, ttl, value, layers, adapter_opts) do reversed = Enum.reverse(layers) - put_to_layers(cache_name, key, ttl, value, reversed, adapter_opts) + + with :ok <- put_to_layers(cache_name, key, ttl, value, reversed, adapter_opts) do + broadcast_write(cache_name, key, ttl, value, adapter_opts) + :ok + end end @impl Cache.Strategy - def delete(cache_name, key, layers, _adapter_opts) do - Enum.reduce_while(layers, :ok, fn layer, _acc -> - case layer_delete(cache_name, key, layer) do - :ok -> {:cont, :ok} - {:error, _} = error -> {:halt, error} - end - end) + def delete(cache_name, key, layers, adapter_opts) do + result = + Enum.reduce_while(layers, :ok, fn layer, _acc -> + case layer_delete(cache_name, key, layer) do + :ok -> {:cont, :ok} + {:error, _} = error -> {:halt, error} + end + end) + + with :ok <- result do + broadcast_delete(cache_name, key, adapter_opts) + :ok + end + end + + # Cross-node coherence (see Cache.MultiLayer.Coordinator). Broadcast only + # after the local write succeeded — writes go slowest-first, so by the time + # a remote node reacts (delete + lazy re-read, or replicated put) the shared + # layer already holds the new value. + defp broadcast_write(cache_name, key, ttl, value, adapter_opts) do + case adapter_opts[:broadcast_mode] do + nil -> + :ok + + :invalidate -> + Cache.MultiLayer.Coordinator.broadcast( + cache_name, + {:multi_layer_invalidate, key, broadcast_layers!(adapter_opts)} + ) + + :replicate -> + Cache.MultiLayer.Coordinator.broadcast( + cache_name, + {:multi_layer_replicate, key, ttl, value, broadcast_layers!(adapter_opts)} + ) + end + end + + defp broadcast_delete(cache_name, key, adapter_opts) do + if is_nil(adapter_opts[:broadcast_mode]) do + :ok + else + Cache.MultiLayer.Coordinator.broadcast( + cache_name, + {:multi_layer_invalidate, key, broadcast_layers!(adapter_opts)} + ) + end + end + + defp broadcast_layers!(adapter_opts) do + adapter_opts[:broadcast_layers] || + raise ArgumentError, + "broadcast_mode is set but broadcast_layers is missing — list the node-local layer modules the broadcast should apply to" end defp get_from_layers(_cache_name, _key, [], _adapter_opts, _visited), do: :miss diff --git a/lib/cache/multi_layer/coordinator.ex b/lib/cache/multi_layer/coordinator.ex new file mode 100644 index 0000000..5ea3757 --- /dev/null +++ b/lib/cache/multi_layer/coordinator.ex @@ -0,0 +1,109 @@ +defmodule Cache.MultiLayer.Coordinator do + @moduledoc """ + Per-node coordinator for cross-node layer coherence in `Cache.MultiLayer`. + + One coordinator runs per MultiLayer cache per node. Each joins a `:pg` + group named after the cache, so group membership doubles as a registry of + which nodes currently run the cache. When a cache is configured with + `broadcast_mode`, writes on one node notify every other member, which then + applies the change to its own node-local layers (`broadcast_layers`): + + - `:invalidate` — remote nodes delete the key from their local layers; the + next read falls through to the shared slower layer and backfills fresh. + Message cost is the key only. Prefer this for large values or many-node + clusters. + - `:replicate` — remote nodes write the new value into their local layers + immediately. Costs a full value copy per member; prefer only for small + values whose next read must not pay a fallthrough. + + Delivery is best-effort (`send/2` to pg members, no acks). A member that + misses a message (netsplit, restart races) serves its stale local entry + until that entry's TTL expires — configure `backfill_ttl` (and layer TTLs) + as the correctness floor; the broadcast is only the fast path. + """ + + use GenServer + + @pg_scope :cache_multi_layer_coordinator + + def start_link(cache_name) do + GenServer.start_link(__MODULE__, cache_name, name: name(cache_name)) + end + + def child_spec(cache_name) do + %{ + id: :"#{cache_name}_multi_layer", + start: {__MODULE__, :start_link, [cache_name]} + } + end + + @doc "pg-tracked coordinator pids for this cache across the cluster." + @spec members(atom()) :: [pid()] + def members(cache_name) do + :pg.get_members(@pg_scope, cache_name) + end + + @doc """ + Notify every other node's coordinator for `cache_name`. + + Excludes `self()`'s node's coordinator by pid so the writing node never + re-applies its own (already fresh) write. + """ + @spec broadcast(atom(), tuple()) :: :ok + def broadcast(cache_name, message) do + local = Process.whereis(name(cache_name)) + + Enum.each(members(cache_name), fn member -> + if member !== local, do: send(member, message) + end) + end + + @impl GenServer + def init(cache_name) do + {:ok, join_and_monitor_scope(cache_name)} + end + + @impl GenServer + def handle_info({:multi_layer_invalidate, key, layers}, state) do + Enum.each(layers, &(&1.delete(key))) + {:noreply, state} + end + + def handle_info({:multi_layer_replicate, key, ttl, value, layers}, state) do + Enum.each(layers, &(&1.put(key, ttl, value))) + {:noreply, state} + end + + # The pg scope died (it is unlinked/unsupervised — this library has no + # application tree to own it). A restarted scope comes back with empty + # membership, so every coordinator must re-join or broadcasts silently + # stop reaching this node. + def handle_info( + {:DOWN, scope_ref, :process, _pid, _reason}, + %{scope_ref: scope_ref, cache_name: cache_name} + ) do + {:noreply, join_and_monitor_scope(cache_name)} + end + + def handle_info(_message, state) do + {:noreply, state} + end + + defp name(cache_name), do: :"#{cache_name}_multi_layer_coordinator" + + defp join_and_monitor_scope(cache_name) do + scope_pid = ensure_pg_scope() + :ok = :pg.join(@pg_scope, cache_name, self()) + %{cache_name: cache_name, scope_ref: Process.monitor(scope_pid)} + end + + # :pg.start/1 (not start_link) — linking would tie the scope's life to + # whichever coordinator happened to start it first, killing membership for + # every cache on the node when that one coordinator dies. + defp ensure_pg_scope do + case :pg.start(@pg_scope) do + {:ok, pid} -> pid + {:error, {:already_started, pid}} -> pid + end + end +end diff --git a/test/cache/multi_layer_broadcast_test.exs b/test/cache/multi_layer_broadcast_test.exs new file mode 100644 index 0000000..6b1116a --- /dev/null +++ b/test/cache/multi_layer_broadcast_test.exs @@ -0,0 +1,129 @@ +defmodule Cache.MultiLayerBroadcastTest do + use ExUnit.Case + + alias Cache.MultiLayer.Coordinator + + defmodule LocalLayer do + use Cache, + adapter: Cache.ETS, + name: :mlb_local_layer, + opts: [] + end + + defmodule SharedLayer do + use Cache, + adapter: Cache.ETS, + name: :mlb_shared_layer, + opts: [] + end + + defmodule InvalidatingCache do + use Cache, + adapter: {Cache.MultiLayer, [LocalLayer, SharedLayer]}, + name: :mlb_invalidating_cache, + opts: [broadcast_mode: :invalidate, broadcast_layers: [LocalLayer]] + end + + defmodule ReplicatingCache do + use Cache, + adapter: {Cache.MultiLayer, [LocalLayer, SharedLayer]}, + name: :mlb_replicating_cache, + opts: [broadcast_mode: :replicate, broadcast_layers: [LocalLayer]] + end + + setup do + start_supervised!(%{ + id: :mlb_sup, + type: :supervisor, + start: + {Cache, :start_link, + [[LocalLayer, SharedLayer, InvalidatingCache, ReplicatingCache], [name: :mlb_sup]]} + }) + + :ok + end + + describe "coordinator pg membership" do + test "joins a pg group named after the cache" do + assert Coordinator.members(:mlb_invalidating_cache) !== [] + end + end + + describe "coordinator message handling" do + test "invalidate message deletes the key from the given layers only" do + :ok = LocalLayer.put("inv_key", "stale") + :ok = SharedLayer.put("inv_key", "fresh") + + [coordinator | _rest] = Coordinator.members(:mlb_invalidating_cache) + send(coordinator, {:multi_layer_invalidate, "inv_key", [LocalLayer]}) + + # handle_info is async — sync on the coordinator's mailbox draining. + _synced = :sys.get_state(coordinator) + + assert {:ok, nil} === LocalLayer.get("inv_key") + assert {:ok, "fresh"} === SharedLayer.get("inv_key") + end + + test "replicate message writes the value into the given layers only" do + [coordinator | _rest] = Coordinator.members(:mlb_replicating_cache) + send(coordinator, {:multi_layer_replicate, "rep_key", nil, "pushed", [LocalLayer]}) + + _synced = :sys.get_state(coordinator) + + assert {:ok, "pushed"} === LocalLayer.get("rep_key") + assert {:ok, nil} === SharedLayer.get("rep_key") + end + + test "unknown messages are ignored" do + [coordinator | _rest] = Coordinator.members(:mlb_invalidating_cache) + send(coordinator, :unexpected) + + assert :sys.get_state(coordinator) + end + end + + describe "broadcast/2 member targeting" do + test "reaches other pg members but never the cache's own coordinator" do + # Join the test process as a fake remote member alongside the real + # coordinator; broadcast must reach it while the real coordinator's + # local layer entry (freshly written) stays untouched. + :ok = :pg.join(:cache_multi_layer_coordinator, :mlb_invalidating_cache, self()) + + Coordinator.broadcast(:mlb_invalidating_cache, {:multi_layer_invalidate, "bk", [LocalLayer]}) + + assert_receive {:multi_layer_invalidate, "bk", [LocalLayer]} + end + end + + describe "put/delete broadcast integration" do + test "put with broadcast_mode: :invalidate notifies members with the key" do + :ok = :pg.join(:cache_multi_layer_coordinator, :mlb_invalidating_cache, self()) + + assert :ok === InvalidatingCache.put("put_key", "value") + + assert_receive {:multi_layer_invalidate, "put_key", [LocalLayer]} + + # The writing node's own layers hold the fresh value untouched. + assert {:ok, "value"} === LocalLayer.get("put_key") + assert {:ok, "value"} === SharedLayer.get("put_key") + end + + test "put with broadcast_mode: :replicate ships the value" do + :ok = :pg.join(:cache_multi_layer_coordinator, :mlb_replicating_cache, self()) + + assert :ok === ReplicatingCache.put("rep_put", "rep_value") + + assert_receive {:multi_layer_replicate, "rep_put", nil, "rep_value", [LocalLayer]} + end + + test "delete broadcasts an invalidate regardless of mode" do + :ok = :pg.join(:cache_multi_layer_coordinator, :mlb_replicating_cache, self()) + + :ok = ReplicatingCache.put("del_key", "value") + assert_receive {:multi_layer_replicate, "del_key", _ttl, _value, _layers} + + assert :ok === ReplicatingCache.delete("del_key") + assert_receive {:multi_layer_invalidate, "del_key", [LocalLayer]} + end + end +end