Elixir by Example

Replicate behind a breaker

Builds on Circuit BreakerIdempotency and Retries

Edit any cell and run it in your browser Download source

KV copies every write to a replica on another node. When that node is down, Erlang notices quickly and the call fails at once. When it hangs (a long garbage collection, a stuck disk, a network that drops packets silently) the call waits for its whole timeout. Then every put on the primary is slow, because of a replica nobody is even reading from.

This step puts the replication behind the circuit breaker from Circuit Breaker. After three timeouts the breaker opens, and writes stop waiting for the replica. The keys it missed are remembered, and once a trial write gets through, the primary sends the replica their current values.

To keep this runnable anywhere, the "remote" replica is a local process, and a hang is simulated with :sys.suspend/1: the process stops handling messages, exactly what a hung node looks like from the outside.

defmodule Breaker do
  @behaviour :gen_statem

  def start_link(opts), do: :gen_statem.start_link({:local, opts[:name]}, __MODULE__, opts, [])
  def state(breaker), do: :gen_statem.call(breaker, :state)

  def run(breaker, fun) do
    case :gen_statem.call(breaker, :ask) do
      :go ->
        result = fun.()
        :gen_statem.cast(breaker, {:report, result})
        result

      :reject ->
        {:error, :circuit_open}
    end
  end

  @impl true
  def callback_mode, do: [:handle_event_function, :state_enter]

  @impl true
  def init(opts), do: {:ok, :closed, %{failures: 0, opts: opts, trial: false}}

  # Enter calls set each state up: open starts the
  # cooldown, closed and half-open clear the counters.
  @impl true
  def handle_event(:enter, _, :open, data),
    do: {:keep_state_and_data, [{:state_timeout, data.opts[:cooldown], :retry}]}

  def handle_event(:enter, _, _, data), do: {:keep_state, %{data | failures: 0, trial: false}}

  def handle_event({:call, from}, :state, state, _),
    do: {:keep_state_and_data, [{:reply, from, state}]}

  def handle_event({:call, from}, :ask, :closed, _),
    do: {:keep_state_and_data, [{:reply, from, :go}]}

  def handle_event({:call, from}, :ask, :open, _),
    do: {:keep_state_and_data, [{:reply, from, :reject}]}

  def handle_event({:call, from}, :ask, :half_open, %{trial: false} = data),
    do: {:keep_state, %{data | trial: true}, [{:reply, from, :go}]}

  def handle_event({:call, from}, :ask, :half_open, _),
    do: {:keep_state_and_data, [{:reply, from, :reject}]}

  def handle_event(:cast, {:report, {:ok, _}}, :closed, data),
    do: {:keep_state, %{data | failures: 0}}

  def handle_event(:cast, {:report, {:ok, _}}, :half_open, data), do: {:next_state, :closed, data}

  def handle_event(:cast, {:report, {:error, _}}, :closed, data) do
    if data.failures + 1 >= data.opts[:threshold],
      do: {:next_state, :open, data},
      else: {:keep_state, %{data | failures: data.failures + 1}}
  end

  def handle_event(:cast, {:report, {:error, _}}, :half_open, data),
    do: {:next_state, :open, data}

  def handle_event(:cast, {:report, _}, :open, _), do: :keep_state_and_data
  def handle_event(:state_timeout, :retry, :open, data), do: {:next_state, :half_open, data}
end

The replica: a map in a GenServer. Calls to it time out after 50 ms.

defmodule KV.Replica do
  use GenServer

  def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
  def dump, do: GenServer.call(__MODULE__, :dump)

  def put(key, value) do
    {:ok, GenServer.call(__MODULE__, {:put, key, value}, 50)}
  catch
    :exit, {:timeout, _} -> {:error, :timeout}
  end

  @impl true
  def init(map), do: {:ok, map}

  @impl true
  def handle_call({:put, key, value}, _from, map), do: {:reply, :ok, Map.put(map, key, value)}
  def handle_call(:dump, _from, map), do: {:reply, map, map}
end

The primary: an ETS table for the data and another for the keys the replica missed.

defmodule KV do
  def start do
    :ets.new(:kv, [:set, :public, :named_table])
    :ets.new(:kv_missed, [:set, :public, :named_table])
  end

  def put(key, value) do
    :ets.insert(:kv, {key, value})

    case Breaker.run(KV.Breaker, fn -> KV.Replica.put(key, value) end) do
      {:ok, :ok} ->
        catch_up()
        :replicated

      {:error, reason} ->
        :ets.insert(:kv_missed, {key})
        reason
    end
  end

  # Send the replica the *current* value of every missed
  # key. Writing a key's latest value again is harmless,
  # so this can safely repeat.
  def catch_up do
    for {key} <- :ets.tab2list(:kv_missed) do
      [{^key, value}] = :ets.lookup(:kv, key)
      {:ok, :ok} = KV.Replica.put(key, value)
      :ets.delete(:kv_missed, key)
    end
  end

  def dump, do: Map.new(:ets.tab2list(:kv))
end

defmodule Wait do
  def until(check) do
    if check.(), do: :ok, else: Process.sleep(1) && until(check)
  end
end

KV.start()
{:ok, _} = KV.Replica.start_link(nil)
{:ok, _} = Breaker.start_link(name: KV.Breaker, threshold: 3, cooldown: 100)

IO.inspect(KV.put("a", 1), label: "put a")
put a: :replicated

The replica hangs. The first three writes each wait the full 50 ms and time out; then the breaker opens, and the next writes don't wait at all.

:sys.suspend(KV.Replica)

for key <- ["b", "c", "d", "e", "f"] do
  IO.inspect(KV.put(key, 2), label: "put #{key}")
end

IO.inspect(Breaker.state(KV.Breaker), label: "breaker")
put b: :timeout
put c: :timeout
put d: :timeout
put e: :circuit_open
put f: :circuit_open
breaker: :open

The replica recovers. While it was hung, the three timed-out calls stayed in its mailbox; it handles them now, and nobody is waiting for the replies any more.

:sys.resume(KV.Replica)

After the cooldown, the next write is the trial. It gets through, the breaker closes, and the primary replays every missed key.

Wait.until(fn -> Breaker.state(KV.Breaker) == :half_open end)
IO.inspect(KV.put("g", 3), label: "put g")
IO.inspect(Breaker.state(KV.Breaker), label: "breaker")
IO.inspect(KV.Replica.dump() == KV.dump(), label: "replica matches primary")
IO.inspect(KV.Replica.dump(), label: "replica")
put g: :replicated
breaker: :closed
replica matches primary: true
replica: %{"a" => 1, "b" => 2, "c" => 2, "d" => 2, "e" => 2, "f" => 2, "g" => 3}

KV now limits greedy clients, bounds its memory under a burst of writes, and keeps working when its replica doesn't. What it doesn't do yet is remember the missed keys across a restart of the primary: :kv_missed is in memory. Storing it next to the data, in the same write (the idea behind the outbox), closes that gap.

Improve this page on GitHub