fix(subscriptions): stop lock queries from waiting forever on a lock pool that went down

guides/Architecture.md

@@ -13,3 +13,5 @@ The pooled connection uses `:exp` back-off. However the other connections use `:
 #### Why are these connections stopped on error?
 
 Related processes need to be notified when the connection exits or is restarted. The [postgrex](https://hexdocs.pm/postgrex/) library doesn't provide a way of being notified when the connection terminates. As an example, EventStore uses [Postgres' advisory locks](https://www.postgresql.org/docs/current/static/explicit-locking.html#ADVISORY-LOCKS) to guarantee only one instance of a subscription runs, regardless of how many nodes are running (or whether they are clustered). These advisory locks are tied to a database connection and are released when the connection terminates. When this happens, EventStore attempts to reacquire the locks. The `EventStore.MonitoredServer` module provides this process monitoring and `after_exit` and `after_restart` callback functions.
+
+Losing the lock is detected after the fact, so progress writes are also guarded by a per-subscription lock generation recorded in the `subscriptions` table. The generation moves on with each lock acquisition and a checkpoint from a previous generation is rejected, which is how a node that no longer holds the lock finds out.

guides/Cluster.md

@@ -10,6 +10,10 @@ PostgreSQL's `LISTEN` / `NOTIFY` is used to pub/sub event notifications. A liste
 
 PostgreSQL's [advisory locks](https://www.postgresql.org/docs/current/static/explicit-locking.html#ADVISORY-LOCKS) are used to limit each uniquely named subscription to run at most once. This prevents multiple instances of a subscription from running on different nodes. Advisory locks are faster than table locks, are stored in memory to avoid table bloat, and are automatically cleaned up by the server at the end of the session.
 
+The lock decides which node may record the subscription's progress, not which node may run its handler. A node only learns that its lock session ended after the fact, so for a short time two nodes can both process the same events. Each time the lock is acquired the subscription row records a new lock generation, and every checkpoint is written against the generation its owner holds. A node whose lock was taken over has its checkpoints rejected, drops the events it still had in flight, and tries to subscribe again later. The node that took over resumes from the last checkpoint committed before the hand-off, and a checkpoint can never move the subscription's position backwards.
+
+Subscribers must therefore be idempotent, as they are for any at-least-once delivery. This protection takes effect once every node runs a version that records lock generations; until then, nodes on an older version behave as before.
+
 ## Automatic cluster formation
 
 You can use [libcluster](https://github.com/bitwalker/libcluster) to automatically form clusters of Erlang nodes, with either static or dynamic node membership.

guides/Upgrades.md

@@ -2,6 +2,10 @@
 
 The [CHANGELOG](https://github.com/commanded/eventstore/blob/master/CHANGELOG.md) is used to indicate when a schema migration is required for a given version of the EventStore.
 
+## Migrate before rolling nodes
+
+Run the migration before starting nodes on the new version. A node on a version that expects a newer schema cannot acquire subscription locks against an unmigrated database and keeps retrying until the migration has run. Nodes on the previous version keep working against the migrated schema, so a rolling upgrade is safe once the migration is in place.
+
 ## Migrating a database with `mix`
 
 You can upgrade an existing EventStore database using the following mix task:

lib/event_store/advisory_locks.ex

@@ -18,6 +18,7 @@ defmodule EventStore.AdvisoryLocks do
     defstruct [
       :conn,
       :connection,
+      :monitor,
       :query_timeout,
       :ref,
       :schema,
@@ -28,35 +29,60 @@ defmodule EventStore.AdvisoryLocks do
 
     def new(opts) do
       conn = Keyword.fetch!(opts, :conn)
+      monitor = Keyword.fetch!(opts, :monitor)
       query_timeout = Keyword.fetch!(opts, :query_timeout)
       schema = Keyword.fetch!(opts, :schema)
 
-      %State{conn: conn, query_timeout: query_timeout, schema: schema}
+      %State{conn: conn, monitor: monitor, query_timeout: query_timeout, schema: schema}
     end
   end
 
   defmodule Lock do
     @moduledoc false
 
+    alias EventStore.Storage.Lock.Lease, as: StorageLease
     alias EventStore.Storage.Session
 
     @type t :: %Lock{
             key: non_neg_integer(),
             owner: pid(),
             ref: reference(),
-            session: Session.t()
+            session: Session.t(),
+            lease: StorageLease.t() | nil
           }
-    defstruct [:key, :owner, :ref, :session]
+    defstruct [:key, :owner, :ref, :session, :lease]
 
-    def new(key, owner, %Session{} = session) do
-      %Lock{key: key, owner: owner, ref: make_ref(), session: session}
+    def new(key, owner, %Session{} = session, lease \\ nil) do
+      %Lock{key: key, owner: owner, ref: make_ref(), session: session, lease: lease}
     end
 
     def ref(%Lock{ref: ref}), do: ref
 
     def held_by?(%Lock{session: session}, %Session{} = current), do: session == current
   end
 
+  defmodule Lease do
+    @moduledoc false
+
+    alias EventStore.Storage.Lock.Lease, as: StorageLease
+    alias EventStore.Storage.Session
+    alias EventStore.Subscriptions.FencingToken
+
+    @type t :: %Lease{
+            ref: reference(),
+            token: FencingToken.t(),
+            last_seen: non_neg_integer() | nil,
+            session: Session.t()
+          }
+    defstruct [:ref, :token, :last_seen, :session]
+
+    def new(ref, %StorageLease{} = lease) do
+      %StorageLease{token: token, last_seen: last_seen, session: session} = lease
+
+      %Lease{ref: ref, token: token, last_seen: last_seen, session: session}
+    end
+  end
+
   defmodule Connection do
     @moduledoc false
 
@@ -76,7 +102,8 @@ defmodule EventStore.AdvisoryLocks do
     def current?(%Connection{}, _pid), do: false
   end
 
-  alias EventStore.AdvisoryLocks.{Connection, Lock, State}
+  alias DBConnection.ConnectionError
+  alias EventStore.AdvisoryLocks.{Connection, Lease, Lock, State}
   alias EventStore.Storage
   alias EventStore.Storage.Session
 
@@ -92,13 +119,14 @@ defmodule EventStore.AdvisoryLocks do
   def init(%State{} = state) do
     Process.flag(:trap_exit, true)
 
-    %State{conn: conn} = state
+    %State{conn: conn, monitor: monitor} = state
 
-    {:ok, ref} = MonitoredServer.monitor(conn)
+    {:ok, ref} = MonitoredServer.monitor(monitor)
 
     # A connection made before this server was registered announced itself to
-    # nobody, so it is replaced by one whose notices can be received.
-    :ok = DBConnection.disconnect_all(conn, 0)
+    # nobody, so it is replaced by one whose notices can be received. A pool that
+    # is down has none, so the replacement is skipped instead of crashing startup.
+    replace_unannounced_connection(conn)
 
     {:ok, %State{state | ref: ref}}
   end
@@ -126,28 +154,138 @@ defmodule EventStore.AdvisoryLocks do
           {:ok, reference()} | {:error, :lock_already_taken} | {:error, term}
   def try_advisory_lock(server, key, timeout \\ 5_000) when is_integer(key) do
     GenServer.call(server, {:try_advisory_lock, key, self(), timeout}, timeout)
+  catch
+    :exit, {:timeout, _call} -> {:error, :timeout}
   end
 
-  def handle_call({:try_advisory_lock, key, owner, timeout}, _from, %State{} = state) do
-    case try_acquire_exclusive_lock(key, owner, timeout, state) do
-      {:ok, %Lock{} = lock} ->
+  @doc """
+  Attempt to obtain the advisory lock of a subscription, recording the new lock
+  owner on the subscription row.
+
+  Returns `{:ok, lease}` when the lock is acquired, `{:error, :lock_already_taken}`
+  if it cannot be acquired immediately, or `{:error, :subscription_not_found}`
+  when the subscription no longer exists. The lock is released and lost the same
+  way as the one returned by `try_advisory_lock/3`, identified by `lease.ref`.
+  """
+  @spec try_subscription_lock(server :: GenServer.server(), pos_integer(), timeout()) ::
+          {:ok, Lease.t()}
+          | {:error, :lock_already_taken}
+          | {:error, :subscription_not_found}
+          | {:error, term}
+  def try_subscription_lock(server, subscription_id, timeout \\ 5_000)
+      when is_integer(subscription_id) do
+    GenServer.call(server, {:try_subscription_lock, subscription_id, self(), timeout}, timeout)
+  catch
+    :exit, {:timeout, _call} -> {:error, :timeout}
+  end
+
+  @doc """
+  Release a lock acquired by `try_advisory_lock/3` or `try_subscription_lock/3`
+  before its owner terminates.
+
+  The owner is not sent a `lock_released` message. Releasing a reference that is
+  no longer held succeeds.
+
+  Returns as soon as the request is queued so a caller is never held up by a
+  lock connection that has stopped answering.
+  """
+  @spec release(server :: GenServer.server(), reference()) :: :ok
+  def release(server, ref) when is_reference(ref) do
+    GenServer.cast(server, {:release, ref})
+  end
+
+  def handle_cast({:release, ref}, %State{} = state) do
+    %State{locks: locks} = state
+
+    state =
+      case Enum.find(locks, fn {_owner_ref, lock} -> Lock.ref(lock) == ref end) do
+        nil ->
+          state
+
+        {owner_ref, %Lock{} = lock} ->
+          Process.demonitor(owner_ref, [:flush])
+
+          release_held_lock(lock, %State{state | locks: Map.delete(locks, owner_ref)})
+      end
+
+    {:noreply, state}
+  end
+
+  def handle_call(
+        {:try_subscription_lock, subscription_id, owner, timeout},
+        _from,
+        %State{} = state
+      ) do
+    case {held_lock(state, subscription_id, owner),
+          claim_subscription_lock(subscription_id, timeout, state)} do
+      {nil, {:ok, %Storage.Lock.Lease{session: session} = lease}} ->
+        lock = Lock.new(subscription_id, owner, session, lease)
         state = monitor_acquired_lock(lock, state)
 
-        {:reply, {:ok, Lock.ref(lock)}, state}
+        {:reply, {:ok, Lease.new(Lock.ref(lock), lease)}, state}
+
+      {{owner_ref, %Lock{} = held}, {:ok, %Storage.Lock.Lease{session: session} = lease}} ->
+        %Lock{key: key, ref: ref} = held
+        %State{locks: locks} = state
 
-      reply ->
+        if Lock.held_by?(held, session), do: release_lock(key, state)
+
+        lock = %Lock{held | session: session, lease: lease}
+
+        {:reply, {:ok, Lease.new(ref, lease)},
+         %State{state | locks: Map.put(locks, owner_ref, lock)}}
+
+      {_held, reply} ->
         {:reply, reply, state}
     end
   end
 
+  def handle_call({:try_advisory_lock, key, owner, timeout}, _from, %State{} = state) do
+    case held_lock(state, key, owner) do
+      {_owner_ref, %Lock{} = lock} ->
+        {:reply, {:ok, Lock.ref(lock)}, state}
+
+      nil ->
+        case try_acquire_exclusive_lock(key, owner, timeout, state) do
+          {:ok, %Lock{} = lock} ->
+            state = monitor_acquired_lock(lock, state)
+
+            {:reply, {:ok, Lock.ref(lock)}, state}
+
+          reply ->
+            {:reply, reply, state}
+        end
+    end
+  end
+
+  defp claim_subscription_lock(subscription_id, timeout, %State{} = state) do
+    %State{schema: schema} = state
+
+    query(state, fn conn ->
+      Storage.Lock.claim_subscription_lock(conn, subscription_id,
+        schema: schema,
+        timeout: timeout
+      )
+    end)
+  end
+
+  defp held_lock(%State{locks: locks}, key, owner) do
+    Enum.find(locks, fn
+      {_owner_ref, %Lock{key: ^key, owner: ^owner}} -> true
+      _other -> false
+    end)
+  end
+
   defp try_acquire_exclusive_lock(key, owner, timeout, %State{} = state) do
-    %State{conn: conn, schema: schema} = state
+    %State{schema: schema} = state
 
     with {:ok, %Session{} = session} <-
-           Storage.Lock.try_acquire_exclusive_session_lock(conn, key,
-             schema: schema,
-             timeout: timeout
-           ) do
+           query(state, fn conn ->
+             Storage.Lock.try_acquire_exclusive_session_lock(conn, key,
+               schema: schema,
+               timeout: timeout
+             )
+           end) do
       {:ok, Lock.new(key, owner, session)}
     end
   end
@@ -246,9 +384,9 @@ defmodule EventStore.AdvisoryLocks do
     do: state
 
   defp verify_sessions(%State{} = state) do
-    %State{conn: conn, locks: locks, query_timeout: query_timeout, released: released} = state
+    %State{locks: locks, query_timeout: query_timeout, released: released} = state
 
-    case Session.current(conn, timeout: query_timeout) do
+    case query(state, &Session.current(&1, timeout: query_timeout)) do
       {:ok, %Session{} = session} ->
         {held, lost} = Map.split_with(locks, fn {_ref, lock} -> Lock.held_by?(lock, session) end)
 
@@ -300,9 +438,30 @@ defmodule EventStore.AdvisoryLocks do
     :ok
   end
 
+  # An owner holds a key once, so releasing it unlocks that owner's hold and leaves
+  # the holds of every other owner in place.
+  defp release_held_lock(%Lock{} = lock, %State{released: released} = state) do
+    verify_sessions(%State{state | released: [lock | released]})
+  end
+
   defp release_lock(key, %State{} = state) do
-    %State{conn: conn, query_timeout: query_timeout, schema: schema} = state
+    %State{query_timeout: query_timeout, schema: schema} = state
+
+    query(state, &Storage.Lock.unlock(&1, key, schema: schema, timeout: query_timeout))
+  end
 
-    Storage.Lock.unlock(conn, key, schema: schema, timeout: query_timeout)
+  defp query(%State{conn: conn}, fun) do
+    fun.(conn)
+  catch
+    :exit, reason ->
+      {:error,
+       ConnectionError.exception("advisory lock connection pool exited: " <> inspect(reason))}
+  end
+
+  defp replace_unannounced_connection(conn) do
+    DBConnection.disconnect_all(conn, 0)
+    :ok
+  catch
+    :exit, _reason -> :ok
   end
 end

lib/event_store/sql/init.ex

@@ -233,7 +233,8 @@ defmodule EventStore.Sql.Init do
         stream_uuid text NOT NULL,
         subscription_name text NOT NULL,
         last_seen bigint NULL,
-        created_at timestamp with time zone DEFAULT NOW() NOT NULL
+        created_at timestamp with time zone DEFAULT NOW() NOT NULL,
+        lock_generation bigint DEFAULT 0 NOT NULL
     );
     """
   end
@@ -276,7 +277,7 @@ defmodule EventStore.Sql.Init do
   defp record_event_store_schema_version do
     """
     INSERT INTO schema_migrations (major_version, minor_version, patch_version)
-    VALUES (1, 3, 2);
+    VALUES (1, 5, 0);
     """
   end
 end

lib/event_store/sql/statements.ex

@@ -21,6 +21,7 @@ defmodule EventStore.Sql.Statements do
         {:try_advisory_lock, [:schema]},
         {:try_advisory_session_lock, [:schema]},
         {:advisory_unlock, [:schema]},
+        {:claim_subscription_lock, [:schema]},
         {:subscription_ack, [:schema]},
         {:insert_snapshot, [:schema]},
         {:delete_snapshot, [:schema]},

lib/event_store/sql/statements/claim_subscription_lock.sql.eex

@@ -0,0 +1,16 @@
+WITH lock AS (
+  SELECT pg_try_advisory_lock(
+    '"<%= schema %>".subscriptions'::regclass::oid::int,
+    (CASE WHEN $1 > 2147483647 THEN mod($1, 2147483647) ELSE $1 END)::int
+  ) AS acquired, pg_backend_pid() AS pid, COALESCE(
+    NULLIF(current_setting('eventstore.session_id', true), ''),
+    set_config('eventstore.session_id', md5(random()::text || clock_timestamp()::text), false)
+  ) AS session_id
+), lease AS (
+  UPDATE "<%= schema %>".subscriptions
+  SET lock_generation = lock_generation + 1
+  WHERE subscription_id = $1 AND (SELECT acquired FROM lock)
+  RETURNING lock_generation, last_seen
+)
+SELECT lock.acquired, lock.pid, lock.session_id, lease.lock_generation, lease.last_seen
+FROM lock LEFT JOIN lease ON true;

lib/event_store/sql/statements/subscription_ack.sql.eex

@@ -1,3 +1,8 @@
-UPDATE "<%= schema %>".subscriptions
-SET last_seen = $2
-WHERE subscription_id = $1;
+WITH progress AS (
+  UPDATE "<%= schema %>".subscriptions
+  SET last_seen = GREATEST(last_seen, $2)
+  WHERE subscription_id = $1 AND lock_generation = $3
+  RETURNING subscription_id
+)
+SELECT EXISTS (SELECT 1 FROM progress) AS acknowledged,
+       EXISTS (SELECT 1 FROM "<%= schema %>".subscriptions WHERE subscription_id = $1) AS present;

lib/event_store/storage.ex

@@ -74,7 +74,7 @@ defmodule EventStore.Storage do
   @doc """
   Acknowledge receipt of an event by its number, for a single subscription.
   """
-  defdelegate ack_last_seen_event(conn, subscription_id, last_seen, opts), to: Subscription
+  defdelegate ack_last_seen_event(conn, token, last_seen, opts), to: Subscription
 
   @doc """
   Delete an existing named subscription to a stream.