Subscription keeps processing events without its advisory lock after the lock connection reconnects #330

# An EventStore subscription keeps running after its advisory lock session ends, while another
# instance takes the lock, so both process every event.
#
# NodeA and NodeB are two Commanded applications in one BEAM standing in for two nodes. Each runs
# on its own instance of one EventStore module, both instances on the same database, and each
# starts the same event handler (Recorder) and process manager (IncrementOnPing). NodeA subscribes
# first and holds the advisory locks. The script terminates NodeA's lock session with
# pg_terminate_backend, dispatches two Ping commands and prints what happened. Both instances use
# subscription_retry_interval: 1_000, so NodeB retries the lock every second instead of every 60 s.
#
#   elixir repro.exs           # terminate NodeA's lock session
#   elixir repro.exs control   # the same steps without the termination
#
# Connection settings come from PGHOST, PGPORT, PGUSER and PGPASSWORD (defaults localhost, 5432,
# postgres, postgres). REPRO_DATABASE names the database (default eventstore_lock_loss_repro);
# the script drops and recreates it on every run.

Mix.install([
  {:commanded, "1.4.11"},
  {:commanded_eventstore_adapter, "1.4.2"},
  {:eventstore, "1.4.10"},
  {:postgrex, "0.22.4"},
  {:db_connection, "2.10.2"}
])

Logger.configure(level: :warning)

defmodule Ping do
  defstruct [:ping_id, :counter_id]
end

defmodule Pinged do
  defstruct [:ping_id, :counter_id]
end

defmodule Increment do
  defstruct [:counter_id]
end

defmodule Incremented do
  defstruct [:counter_id]
end

defmodule Pinger do
  defstruct [:ping_id]

  def execute(%Pinger{}, %Ping{ping_id: ping_id, counter_id: counter_id}), do: %Pinged{ping_id: ping_id, counter_id: counter_id}

  def apply(%Pinger{} = pinger, %Pinged{ping_id: ping_id}), do: %Pinger{pinger | ping_id: ping_id}
end

defmodule Counter do
  @moduledoc "Every Increment command appends one Incremented event: the command is not idempotent."
  defstruct count: 0

  def execute(%Counter{}, %Increment{counter_id: counter_id}), do: %Incremented{counter_id: counter_id}

  def apply(%Counter{count: count} = counter, %Incremented{}), do: %Counter{counter | count: count + 1}
end

defmodule TagDispatchingNode do
  @moduledoc "Writes the dispatching application into the metadata of every event a command produces."
  @behaviour Commanded.Middleware

  alias Commanded.Middleware.Pipeline

  def before_dispatch(%Pipeline{application: application} = pipeline) do
    Pipeline.assign_metadata(pipeline, "dispatched_by", inspect(application))
  end

  def after_dispatch(pipeline), do: pipeline
  def after_failure(pipeline), do: pipeline
end

defmodule Router do
  use Commanded.Commands.Router

  middleware(TagDispatchingNode)

  dispatch(Ping, to: Pinger, identity: :ping_id)
  dispatch(Increment, to: Counter, identity: :counter_id)
end

defmodule Store do
  use EventStore, otp_app: :repro
end

defmodule NodeA do
  use Commanded.Application, otp_app: :repro
  router(Router)
end

defmodule NodeB do
  use Commanded.Application, otp_app: :repro
  router(Router)
end

defmodule Journal do
  @moduledoc "What the handlers report, with the time in milliseconds since the script started."

  def start_link, do: Agent.start_link(fn -> [] end, name: __MODULE__)

  def record(entry) do
    elapsed_ms = elapsed_ms()
    Agent.update(__MODULE__, &[{elapsed_ms, entry} | &1])
  end

  def elapsed_ms, do: System.monotonic_time(:millisecond) - :persistent_term.get(:repro_started_at)

  def entries do
    __MODULE__
    |> Agent.get(& &1)
    |> Enum.reverse()
  end
end

defmodule Recorder do
  @moduledoc "Event handler that records when its subscription starts and which node handled which event."
  use Commanded.Event.Handler, name: "Recorder"

  def after_start(%{node: node} = state) do
    Journal.record({:subscribed, node})
    {:ok, state}
  end

  def handle(event, %{application: node, event_number: event_number}) do
    Journal.record({:handled, node, event_number, event.__struct__})
    :ok
  end
end

defmodule IncrementOnPing do
  @moduledoc "Process manager: every Pinged event dispatches one Increment command for its counter."
  use Commanded.ProcessManagers.ProcessManager, name: "IncrementOnPing"

  defstruct []

  def interested?(%Pinged{ping_id: ping_id}), do: {:start, ping_id}
  def interested?(_event), do: false

  def handle(%IncrementOnPing{}, %Pinged{counter_id: counter_id}), do: %Increment{counter_id: counter_id}
end

defmodule Database do
  @moduledoc "Database setup and the pg_locks, pg_stat_activity and pg_terminate_backend queries."

  def settings do
    [
      hostname: System.get_env("PGHOST", "localhost"),
      port: String.to_integer(System.get_env("PGPORT", "5432")),
      username: System.get_env("PGUSER", "postgres"),
      password: System.get_env("PGPASSWORD", "postgres")
    ]
  end

  def connect(database) do
    {:ok, conn} = Postgrex.start_link([database: database] ++ settings())
    conn
  end

  def recreate(admin, database) do
    Postgrex.query!(admin, ~s[DROP DATABASE IF EXISTS "#{database}" WITH (FORCE)], [])
    Postgrex.query!(admin, ~s[CREATE DATABASE "#{database}"], [])
    config = Store.config([database: database] ++ settings())
    :ok = EventStore.Tasks.Init.exec(config, quiet: true)
  end

  def server_version(admin) do
    %Postgrex.Result{rows: [[version]]} =
      Postgrex.query!(admin, "SELECT split_part(current_setting('server_version'), ' ', 1)", [])

    version
  end

  @doc "Advisory locks held in the database, as `{application_name, backend_pid, lock_count}`."
  def advisory_locks(admin, database) do
    sql = """
    SELECT a.application_name, l.pid, count(*)
    FROM pg_locks l JOIN pg_stat_activity a ON a.pid = l.pid
    WHERE l.locktype = 'advisory' AND l.granted AND a.datname = $1
    GROUP BY a.application_name, l.pid
    ORDER BY a.application_name, l.pid
    """

    %Postgrex.Result{rows: rows} = Postgrex.query!(admin, sql, [database])
    Enum.map(rows, &List.to_tuple/1)
  end

  def terminate_backend(admin, database, backend_pid) do
    sql = "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE pid = $1 AND datname = $2"
    %Postgrex.Result{rows: [[terminated?]]} = Postgrex.query!(admin, sql, [backend_pid, database])
    terminated?
  end
end

defmodule Repro do
  @versioned_apps [:eventstore, :commanded, :commanded_eventstore_adapter, :postgrex, :db_connection]
  @lock_count_per_node 2

  def run(mode) do
    :persistent_term.put(:repro_started_at, System.monotonic_time(:millisecond))
    database = System.get_env("REPRO_DATABASE", "eventstore_lock_loss_repro")
    admin = Database.connect("postgres")
    Database.recreate(admin, database)
    {:ok, _journal} = Journal.start_link()
    print_versions(admin)

    start_node(NodeA, :node_a, database)
    wait_for_locks(admin, database, "node_a", deadline_in(20_000))
    log("NodeA started. Advisory locks: #{describe_locks(admin, database)}")

    start_node(NodeB, :node_b, database)
    Process.sleep(3_000)
    log("NodeB started 3 s ago. Advisory locks: #{describe_locks(admin, database)}")
    log("NodeA lock connection: #{describe_lock_connection(:node_a)}")

    end_lock_session(mode, admin, database)
    Process.sleep(3_000)
    log("3 s later. Advisory locks: #{describe_locks(admin, database)}")
    log("NodeA lock connection: #{describe_lock_connection(:node_a)}")

    dispatch_ping(NodeA, "ping-1")
    Process.sleep(2_000)
    dispatch_ping(NodeA, "ping-2")
    Process.sleep(2_000)
    log("End. Advisory locks: #{describe_locks(admin, database)}")

    print_subscriptions()
    print_handled_events()
    print_counter_stream()
  end

  defp start_node(application, store_name, database) do
    event_store =
      [
        adapter: Commanded.EventStore.Adapters.EventStore,
        event_store: Store,
        name: store_name,
        serializer: EventStore.TermSerializer,
        database: database,
        subscription_retry_interval: 1_000,
        parameters: [application_name: Atom.to_string(store_name)]
      ] ++ Database.settings()

    {:ok, _application} = application.start_link(event_store: event_store)
    {:ok, _handler} = Recorder.start_link(application: application, state: %{node: application})
    {:ok, _process_manager} = IncrementOnPing.start_link(application: application)
  end

  defp end_lock_session(:control, _admin, _database), do: log("Control run: NodeA's lock session is left alone.")

  defp end_lock_session(:terminate, admin, database) do
    [{"node_a", lock_backend, @lock_count_per_node}] = Database.advisory_locks(admin, database)
    terminated? = Database.terminate_backend(admin, database, lock_backend)
    log("pg_terminate_backend(#{lock_backend}): #{terminated?}")
  end

  defp wait_for_locks(admin, database, application_name, deadline) do
    locks = Database.advisory_locks(admin, database)

    cond do
      Enum.any?(locks, &match?({^application_name, _backend, @lock_count_per_node}, &1)) -> :ok
      System.monotonic_time(:millisecond) > deadline -> raise "#{application_name} did not take its advisory locks"
      true -> retry_wait_for_locks(admin, database, application_name, deadline)
    end
  end

  defp retry_wait_for_locks(admin, database, application_name, deadline) do
    Process.sleep(100)
    wait_for_locks(admin, database, application_name, deadline)
  end

  defp deadline_in(timeout_ms), do: System.monotonic_time(:millisecond) + timeout_ms

  defp describe_locks(admin, database) do
    case Database.advisory_locks(admin, database) do
      [] -> "none"
      locks -> Enum.map_join(locks, ", ", &describe_lock_holder/1)
    end
  end

  defp describe_lock_holder({application_name, backend, count}), do: "#{application_name} backend #{backend} holds #{count}"

  # The pool process EventStore watches to detect a lost lock, and the database session behind it.
  defp describe_lock_connection(store_name) do
    pool_name = Module.concat([store_name, EventStore.AdvisoryLocks, Postgrex])
    %{pid: pool} = :sys.get_state(pool_name)
    session_backend = fetch_session_backend(pool_name, deadline_in(15_000))
    "pool process #{inspect(pool)}, session backend #{session_backend}"
  end

  defp fetch_session_backend(pool_name, deadline) do
    case Postgrex.query(pool_name, "SELECT pg_backend_pid()", []) do
      {:ok, %Postgrex.Result{rows: [[backend]]}} -> backend
      {:error, _reason} -> retry_fetch_session_backend(pool_name, deadline)
    end
  end

  defp retry_fetch_session_backend(pool_name, deadline) do
    if System.monotonic_time(:millisecond) > deadline,
      do: raise("the lock connection did not reconnect"),
      else: Process.sleep(250)

    fetch_session_backend(pool_name, deadline)
  end

  defp dispatch_ping(application, ping_id) do
    result = application.dispatch(%Ping{ping_id: ping_id, counter_id: "counter-1"})
    log("Dispatched Ping #{ping_id} through #{inspect(application)}: #{inspect(result)}")
  end

  defp print_versions(admin) do
    versions = Enum.map_join(@versioned_apps, ", ", &"#{&1} #{Application.spec(&1, :vsn)}")
    log("#{versions}, PostgreSQL #{Database.server_version(admin)}, Elixir #{System.version()}, OTP #{System.otp_release()}")
  end

  defp print_subscriptions do
    IO.puts("\nRecorder subscribed (after_start):")
    for {elapsed_ms, {:subscribed, node}} <- Journal.entries(), do: IO.puts("  #{inspect(node)} at #{format_seconds(elapsed_ms)}")
  end

  defp print_handled_events do
    IO.puts("\nEvents handled by Recorder:")

    handled =
      for {_elapsed_ms, {:handled, node, event_number, event_type}} <- Journal.entries(),
          do: {{event_number, event_type}, node}

    handled_by = Enum.group_by(handled, &elem(&1, 0), &elem(&1, 1))

    for {{event_number, event_type}, nodes} <- Enum.sort(handled_by) do
      node_names =
        nodes
        |> Enum.sort()
        |> Enum.map_join(", ", &inspect/1)

      IO.puts("  ##{event_number} #{inspect(event_type)}: #{node_names}")
    end
  end

  defp print_counter_stream do
    {:ok, all_events} = Store.read_all_streams_forward(0, 1_000, name: :node_a)

    ping_by_event_id =
      for %{data: %Pinged{ping_id: ping_id}, event_id: event_id} <- all_events, into: %{}, do: {event_id, ping_id}

    {:ok, counter_events} = Store.read_stream_forward("counter-1", 0, 1_000, name: :node_a)
    IO.puts("\nStream counter-1:")

    for event <- counter_events do
      ping = Map.fetch!(ping_by_event_id, event.causation_id)
      dispatched_by = event.metadata["dispatched_by"]

      IO.puts(
        "  version #{event.stream_version} #{inspect(event.data.__struct__)}, causation_id #{event.causation_id} (Pinged #{ping}), dispatched by #{dispatched_by}"
      )
    end
  end

  defp log(text), do: IO.puts("[#{format_seconds(Journal.elapsed_ms())}] #{text}")

  defp format_seconds(elapsed_ms), do: :erlang.float_to_binary(elapsed_ms / 1000, decimals: 2) <> "s"
end

mode =
  case System.argv() do
    [] -> :terminate
    ["control"] -> :control
  end

Repro.run(mode)