feat(multi_layer): cross-node layer coherence via pg-tracked coordinators

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

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

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