From 6dd86f672d5e604d320f11024d81cd96c1175962 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 19 Oct 2026 15:08:19 -0400 Subject: [PATCH 01/13] chore(test): show lock queries hang when the lock pool goes down Signed-off-by: Yordis Prieto --- test/advisory_locks_test.exs | 40 +++++++++++++++++++++++++++++++++--- 1 file changed, 37 insertions(+), 3 deletions(-) diff --git a/test/advisory_locks_test.exs b/test/advisory_locks_test.exs index 7292fb5..d5d946d 100644 --- a/test/advisory_locks_test.exs +++ b/test/advisory_locks_test.exs @@ -220,6 +220,36 @@ defmodule EventStore.AdvisoryLocksTest do end end + describe "lock pool down" do + test "should reply to a lock query in flight when the lock pool goes down" do + reply_to = self() + + holder = hold_only_connection() + + try do + spawn(fn -> + send(reply_to, {:lock, AdvisoryLocks.try_advisory_lock(@locks, 1, 15_000)}) + end) + + wait_until_waiting_for_connection() + + Process.exit(lock_pool_pid(), :kill) + + assert_receive {:lock, {:error, _error}}, 1_000 + after + send(holder, :release) + end + end + + test "should not wait for the lock pool to restart" do + Process.exit(lock_pool_pid(), :kill) + + Wait.until(fn -> assert is_nil(lock_pool_pid()) end) + + assert {:error, _error} = AdvisoryLocks.try_advisory_lock(@locks, 1, 1_000) + end + end + describe "graceful shutdown" do test "should complete an in-flight lock query before terminating" do reply_to = self() @@ -391,7 +421,7 @@ defmodule EventStore.AdvisoryLocksTest do end defp hold_only_connection do - %{pid: conn} = Process.whereis(@conn) |> :sys.get_state() + conn = lock_pool_pid() reply_to = self() @@ -485,8 +515,12 @@ defmodule EventStore.AdvisoryLocksTest do end defp shutdown_database_connection do - %{pid: pid} = Process.whereis(@conn) |> :sys.get_state() + EventStore.ProcessHelper.shutdown(lock_pool_pid()) + end + + defp lock_pool_pid do + %{pid: pid} = :sys.get_state(@conn) - EventStore.ProcessHelper.shutdown(pid) + pid end end From 3925c24ec0601fc008f6f12e921dbfb862036e62 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 19 Oct 2026 15:13:23 -0400 Subject: [PATCH 02/13] fix(subscriptions): stop lock queries from waiting forever on a lock pool that went down Signed-off-by: Yordis Prieto --- lib/event_store/advisory_locks.ex | 41 ++++++++++++++++++++++------- lib/event_store/monitored_server.ex | 11 ++++++++ 2 files changed, 43 insertions(+), 9 deletions(-) diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex index 97b284d..2a4af16 100644 --- a/lib/event_store/advisory_locks.ex +++ b/lib/event_store/advisory_locks.ex @@ -76,6 +76,7 @@ defmodule EventStore.AdvisoryLocks do def current?(%Connection{}, _pid), do: false end + alias DBConnection.ConnectionError alias EventStore.AdvisoryLocks.{Connection, Lock, State} alias EventStore.Storage alias EventStore.Storage.Session @@ -141,13 +142,15 @@ defmodule EventStore.AdvisoryLocks do 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 + with_pool(state, fn pool -> + Storage.Lock.try_acquire_exclusive_session_lock(pool, key, + schema: schema, + timeout: timeout + ) + end) do {:ok, Lock.new(key, owner, session)} end end @@ -246,9 +249,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 with_pool(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) @@ -301,8 +304,28 @@ defmodule EventStore.AdvisoryLocks do 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 - Storage.Lock.unlock(conn, key, schema: schema, timeout: query_timeout) + with_pool(state, &Storage.Lock.unlock(&1, key, schema: schema, timeout: query_timeout)) + end + + # Querying through the monitored server would wait on a pool that may never answer: + # it queues requests while the pool is down and forgets those a dying pool received. + defp with_pool(%State{conn: conn}, query) do + case MonitoredServer.whereis(conn) do + nil -> + {:error, ConnectionError.exception("advisory lock connection pool is down")} + + pool -> + try do + query.(pool) + catch + :exit, reason -> + {:error, + ConnectionError.exception( + "advisory lock connection pool exited: " <> inspect(reason) + )} + end + end end end diff --git a/lib/event_store/monitored_server.ex b/lib/event_store/monitored_server.ex index 6aca05c..e4d58ff 100644 --- a/lib/event_store/monitored_server.ex +++ b/lib/event_store/monitored_server.ex @@ -84,6 +84,13 @@ defmodule EventStore.MonitoredServer do GenServer.call(name, {__MODULE__, :monitors}) end + @doc """ + The pid of the started process, or `nil` while it is down. + """ + def whereis(name) do + GenServer.call(name, {__MODULE__, :whereis}) + end + def init(%State{} = state) do Process.flag(:trap_exit, true) @@ -107,6 +114,10 @@ defmodule EventStore.MonitoredServer do {:reply, {:ok, monitors}, state} end + def handle_call({__MODULE__, :whereis}, _from, %State{pid: pid} = state) do + {:reply, pid, state} + end + def handle_call(msg, from, %State{pid: nil} = state) do {:noreply, enqueue({:call, msg, from}, state)} end From 1597086f8e0be7fa00fc77a68c310bd180e9e8a6 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 19 Oct 2026 15:29:08 -0400 Subject: [PATCH 03/13] chore(test): show the lock server crashes when the lock pool monitor does not answer Signed-off-by: Yordis Prieto --- test/advisory_locks_test.exs | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/test/advisory_locks_test.exs b/test/advisory_locks_test.exs index d5d946d..a7b9509 100644 --- a/test/advisory_locks_test.exs +++ b/test/advisory_locks_test.exs @@ -248,6 +248,20 @@ defmodule EventStore.AdvisoryLocksTest do assert {:error, _error} = AdvisoryLocks.try_advisory_lock(@locks, 1, 1_000) end + + test "should keep running when the lock pool monitor does not answer" do + locks = Process.whereis(@locks) + + :ok = :sys.suspend(@conn) + + try do + assert {:error, _error} = AdvisoryLocks.try_advisory_lock(@locks, 1, 10_000) + after + :ok = :sys.resume(@conn) + end + + assert Process.whereis(@locks) == locks + end end describe "graceful shutdown" do From 67692bfaeaf732f29a441434344e8e42a877e3f8 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 19 Oct 2026 15:29:08 -0400 Subject: [PATCH 04/13] fix(subscriptions): keep the lock server running when the lock pool monitor does not answer Signed-off-by: Yordis Prieto --- lib/event_store/advisory_locks.ex | 19 ++++++------------- 1 file changed, 6 insertions(+), 13 deletions(-) diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex index 2a4af16..73a1709 100644 --- a/lib/event_store/advisory_locks.ex +++ b/lib/event_store/advisory_locks.ex @@ -313,19 +313,12 @@ defmodule EventStore.AdvisoryLocks do # it queues requests while the pool is down and forgets those a dying pool received. defp with_pool(%State{conn: conn}, query) do case MonitoredServer.whereis(conn) do - nil -> - {:error, ConnectionError.exception("advisory lock connection pool is down")} - - pool -> - try do - query.(pool) - catch - :exit, reason -> - {:error, - ConnectionError.exception( - "advisory lock connection pool exited: " <> inspect(reason) - )} - end + nil -> {:error, ConnectionError.exception("advisory lock connection pool is down")} + pool -> query.(pool) end + catch + :exit, reason -> + {:error, + ConnectionError.exception("advisory lock connection pool exited: " <> inspect(reason))} end end From f1c5d29c4871d857ad83622ae4c70da7430ff63a Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 20 Oct 2026 14:09:58 -0400 Subject: [PATCH 05/13] refactor(subscriptions): query the lock pool by its own name Signed-off-by: Yordis Prieto --- lib/event_store/advisory_locks.ex | 25 +++++++++++-------------- lib/event_store/monitored_server.ex | 11 ----------- lib/event_store/supervisor.ex | 9 +++++++-- test/advisory_locks_test.exs | 27 +++++++++++++++------------ 4 files changed, 33 insertions(+), 39 deletions(-) diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex index 73a1709..4202715 100644 --- a/lib/event_store/advisory_locks.ex +++ b/lib/event_store/advisory_locks.ex @@ -18,6 +18,7 @@ defmodule EventStore.AdvisoryLocks do defstruct [ :conn, :connection, + :monitor, :query_timeout, :ref, :schema, @@ -28,10 +29,11 @@ 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 @@ -93,9 +95,9 @@ 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. @@ -145,8 +147,8 @@ defmodule EventStore.AdvisoryLocks do %State{schema: schema} = state with {:ok, %Session{} = session} <- - with_pool(state, fn pool -> - Storage.Lock.try_acquire_exclusive_session_lock(pool, key, + query(state, fn conn -> + Storage.Lock.try_acquire_exclusive_session_lock(conn, key, schema: schema, timeout: timeout ) @@ -251,7 +253,7 @@ defmodule EventStore.AdvisoryLocks do defp verify_sessions(%State{} = state) do %State{locks: locks, query_timeout: query_timeout, released: released} = state - case with_pool(state, &Session.current(&1, 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) @@ -306,16 +308,11 @@ defmodule EventStore.AdvisoryLocks do defp release_lock(key, %State{} = state) do %State{query_timeout: query_timeout, schema: schema} = state - with_pool(state, &Storage.Lock.unlock(&1, key, schema: schema, timeout: query_timeout)) + query(state, &Storage.Lock.unlock(&1, key, schema: schema, timeout: query_timeout)) end - # Querying through the monitored server would wait on a pool that may never answer: - # it queues requests while the pool is down and forgets those a dying pool received. - defp with_pool(%State{conn: conn}, query) do - case MonitoredServer.whereis(conn) do - nil -> {:error, ConnectionError.exception("advisory lock connection pool is down")} - pool -> query.(pool) - end + defp query(%State{conn: conn}, fun) do + fun.(conn) catch :exit, reason -> {:error, diff --git a/lib/event_store/monitored_server.ex b/lib/event_store/monitored_server.ex index e4d58ff..6aca05c 100644 --- a/lib/event_store/monitored_server.ex +++ b/lib/event_store/monitored_server.ex @@ -84,13 +84,6 @@ defmodule EventStore.MonitoredServer do GenServer.call(name, {__MODULE__, :monitors}) end - @doc """ - The pid of the started process, or `nil` while it is down. - """ - def whereis(name) do - GenServer.call(name, {__MODULE__, :whereis}) - end - def init(%State{} = state) do Process.flag(:trap_exit, true) @@ -114,10 +107,6 @@ defmodule EventStore.MonitoredServer do {:reply, {:ok, monitors}, state} end - def handle_call({__MODULE__, :whereis}, _from, %State{pid: pid} = state) do - {:reply, pid, state} - end - def handle_call(msg, from, %State{pid: nil} = state) do {:noreply, enqueue({:call, msg, from}, state)} end diff --git a/lib/event_store/supervisor.ex b/lib/event_store/supervisor.ex index d2da6da..369a73f 100644 --- a/lib/event_store/supervisor.ex +++ b/lib/event_store/supervisor.ex @@ -59,9 +59,13 @@ defmodule EventStore.Supervisor do advisory_locks_name = Module.concat([name, AdvisoryLocks]) advisory_locks_postgrex_conn = Module.concat([advisory_locks_name, Postgrex]) + advisory_locks_postgrex_monitor = + Module.concat([advisory_locks_name, Postgrex, MonitoredServer]) + advisory_locks_config = config |> Config.advisory_locks_postgrex_opts() + |> Keyword.put(:name, advisory_locks_postgrex_conn) |> Config.add_connection_listener(advisory_locks_name) subscriptions_name = Module.concat([name, Subscriptions.Supervisor]) @@ -81,12 +85,13 @@ defmodule EventStore.Supervisor do Supervisor.child_spec( {MonitoredServer, mfa: {Postgrex, :start_link, [advisory_locks_config]}, - name: advisory_locks_postgrex_conn, + name: advisory_locks_postgrex_monitor, backoff_min: 30_000}, - id: Module.concat([advisory_locks_postgrex_conn, MonitoredServer]) + id: advisory_locks_postgrex_monitor ), {AdvisoryLocks, conn: advisory_locks_postgrex_conn, + monitor: advisory_locks_postgrex_monitor, query_timeout: query_timeout, schema: schema, name: advisory_locks_name} diff --git a/test/advisory_locks_test.exs b/test/advisory_locks_test.exs index a7b9509..5922e93 100644 --- a/test/advisory_locks_test.exs +++ b/test/advisory_locks_test.exs @@ -8,6 +8,7 @@ defmodule EventStore.AdvisoryLocksTest do @locks TestEventStore.EventStore.AdvisoryLocks @conn TestEventStore.EventStore.AdvisoryLocks.Postgrex + @monitor TestEventStore.EventStore.AdvisoryLocks.Postgrex.EventStore.MonitoredServer setup do postgrex_config = Config.parsed(TestEventStore, :eventstore) |> Config.default_postgrex_opts() @@ -252,12 +253,12 @@ defmodule EventStore.AdvisoryLocksTest do test "should keep running when the lock pool monitor does not answer" do locks = Process.whereis(@locks) - :ok = :sys.suspend(@conn) + :ok = :sys.suspend(@monitor) try do - assert {:error, _error} = AdvisoryLocks.try_advisory_lock(@locks, 1, 10_000) + assert {:ok, _lock} = AdvisoryLocks.try_advisory_lock(@locks, 1, 10_000) after - :ok = :sys.resume(@conn) + :ok = :sys.resume(@monitor) end assert Process.whereis(@locks) == locks @@ -360,6 +361,7 @@ defmodule EventStore.AdvisoryLocksTest do test "should acquire lock and keep running" do locks_name = Module.concat([__MODULE__, SwitchedRole, AdvisoryLocks]) conn_name = Module.concat([locks_name, Postgrex]) + monitor_name = Module.concat([conn_name, MonitoredServer]) config = Config.parsed(TestEventStore, :eventstore) @@ -367,18 +369,20 @@ defmodule EventStore.AdvisoryLocksTest do config |> Config.advisory_locks_postgrex_opts() |> Config.add_connection_listener(locks_name) + |> Keyword.put(:name, conn_name) |> Keyword.put(:after_connect, {Postgrex, :query!, ["SET ROLE pg_signal_backend", []]}) start_supervised!( {MonitoredServer, - mfa: {Postgrex, :start_link, [postgrex_config]}, name: conn_name, backoff_min: 30_000}, - id: conn_name + mfa: {Postgrex, :start_link, [postgrex_config]}, name: monitor_name, backoff_min: 30_000}, + id: monitor_name ) locks = start_supervised!( {AdvisoryLocks, conn: conn_name, + monitor: monitor_name, query_timeout: Keyword.fetch!(config, :timeout), schema: "public", name: locks_name} @@ -394,6 +398,7 @@ defmodule EventStore.AdvisoryLocksTest do test "should send `lock_released` message when the lock connection crashes" do locks_name = Module.concat([__MODULE__, ConnectedFirst, AdvisoryLocks]) conn_name = Module.concat([locks_name, Postgrex]) + monitor_name = Module.concat([conn_name, MonitoredServer]) config = Config.parsed(TestEventStore, :eventstore) @@ -401,11 +406,12 @@ defmodule EventStore.AdvisoryLocksTest do config |> Config.advisory_locks_postgrex_opts() |> Config.add_connection_listener(locks_name) + |> Keyword.put(:name, conn_name) start_supervised!( {MonitoredServer, - mfa: {Postgrex, :start_link, [postgrex_config]}, name: conn_name, backoff_min: 30_000}, - id: conn_name + mfa: {Postgrex, :start_link, [postgrex_config]}, name: monitor_name, backoff_min: 30_000}, + id: monitor_name ) # The notice for this connection is sent to a lock server that does not exist yet. @@ -414,6 +420,7 @@ defmodule EventStore.AdvisoryLocksTest do start_supervised!( {AdvisoryLocks, conn: conn_name, + monitor: monitor_name, query_timeout: Keyword.fetch!(config, :timeout), schema: "public", name: locks_name} @@ -532,9 +539,5 @@ defmodule EventStore.AdvisoryLocksTest do EventStore.ProcessHelper.shutdown(lock_pool_pid()) end - defp lock_pool_pid do - %{pid: pid} = :sys.get_state(@conn) - - pid - end + defp lock_pool_pid, do: Process.whereis(@conn) end From a0593701ec816712a4ff32155dc657b1715820ad Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 16:41:15 -0400 Subject: [PATCH 06/13] fix(subscriptions): ignore lock notices for a lock the subscription no longer holds (#28) Signed-off-by: Yordis Prieto --- .../subscriptions/subscription_fsm.ex | 2 + .../subscription_locking_test.exs | 76 ++++++++++++++++++- 2 files changed, 77 insertions(+), 1 deletion(-) diff --git a/lib/event_store/subscriptions/subscription_fsm.ex b/lib/event_store/subscriptions/subscription_fsm.ex index b3941d4..079e1bc 100644 --- a/lib/event_store/subscriptions/subscription_fsm.ex +++ b/lib/event_store/subscriptions/subscription_fsm.ex @@ -313,6 +313,8 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do %__MODULE__{fsm | state: :disconnected, data: data} end + def disconnect(%__MODULE__{} = fsm, _lock_ref), do: fsm + def unsubscribe( %__MODULE__{state: :unsubscribed, data: %SubscriptionState{}} = fsm, _subscriber diff --git a/test/subscriptions/subscription_locking_test.exs b/test/subscriptions/subscription_locking_test.exs index fbf69d3..6eb59ae 100644 --- a/test/subscriptions/subscription_locking_test.exs +++ b/test/subscriptions/subscription_locking_test.exs @@ -1,7 +1,7 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do use EventStore.StorageCase - alias EventStore.{Config, EventFactory, Storage, UUID} + alias EventStore.{Config, EventFactory, Storage, UUID, Wait} alias EventStore.Subscriptions.Subscription @conn TestEventStore.Postgrex @@ -163,6 +163,62 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do end end + describe "late lock notice" do + setup [:create_subscription] + + test "should keep running when the lock was already released", %{ + event_store: event_store, + subscription: subscription + } do + assert_receive {:subscribed, ^subscription} + + Process.flag(:trap_exit, true) + monitor_ref = Process.monitor(subscription) + + lock_ref = lock_ref(subscription) + + :ok = release_lock(subscription, lock_ref) + :sys.get_state(subscription) + + :ok = release_lock(subscription, lock_ref) + + refute_receive {:DOWN, ^monitor_ref, :process, ^subscription, _reason}, 200 + assert Process.alive?(subscription) + + :ok = reconnect(subscription) + + append_events_to_stream(event_store, 3) + + assert_receive_events([1, 2, 3]) + end + + test "should ignore a notice for a lock that was replaced by a new one", %{ + event_store: event_store, + subscription: subscription + } do + assert_receive {:subscribed, ^subscription} + + Process.flag(:trap_exit, true) + monitor_ref = Process.monitor(subscription) + + old_lock_ref = lock_ref(subscription) + + :ok = release_lock(subscription, old_lock_ref) + :ok = reconnect(subscription) + + _new_lock_ref = wait_for_new_lock_ref(subscription, old_lock_ref) + + :ok = release_lock(subscription, old_lock_ref) + + refute_receive {:DOWN, ^monitor_ref, :process, ^subscription, _reason}, 200 + assert Process.alive?(subscription) + + append_events_to_stream(event_store, 3) + + assert_receive_events([1, 2, 3]) + end + end + defp lock_subscription(context) do %{config: config, schema: schema} = context @@ -221,8 +277,26 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do end defp disconnect(subscription) do + release_lock(subscription, lock_ref(subscription)) + end + + defp lock_ref(subscription) do %Subscription{subscription: %{data: %{lock_ref: lock_ref}}} = :sys.get_state(subscription) + lock_ref + end + + defp wait_for_new_lock_ref(subscription, old_lock_ref) do + Wait.until(2_000, fn -> + lock_ref = lock_ref(subscription) + + assert is_reference(lock_ref) and lock_ref != old_lock_ref + + lock_ref + end) + end + + defp release_lock(subscription, lock_ref) do Process.send( subscription, {EventStore.AdvisoryLocks, :lock_released, lock_ref, :shutdown}, From 60e98e1bc4e7e562504fd25eec3211b47bc3c975 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 16:41:15 -0400 Subject: [PATCH 07/13] feat(storage): record which lock owner may write a subscription's progress (#29) Signed-off-by: Yordis Prieto --- lib/event_store/advisory_locks.ex | 70 +++++++- lib/event_store/sql/init.ex | 5 +- lib/event_store/sql/statements.ex | 1 + .../claim_subscription_lock.sql.eex | 16 ++ lib/event_store/storage/lock.ex | 36 +++++ lib/event_store/storage/lock/lease.ex | 19 +++ .../subscriptions/fencing_token.ex | 16 ++ lib/event_store/tasks/migrations.ex | 3 +- priv/event_store/migrations/v1.5.0.sql | 8 + test/advisory_locks_test.exs | 85 ++++++++++ test/list_event_store_migrations_test.exs | 1 + test/migrate_event_store_test.exs | 1 + test/storage/subscription_lock_test.exs | 153 ++++++++++++++++++ .../subscription_previous_version_test.exs | 91 +++++++++++ 14 files changed, 501 insertions(+), 4 deletions(-) create mode 100644 lib/event_store/sql/statements/claim_subscription_lock.sql.eex create mode 100644 lib/event_store/storage/lock/lease.ex create mode 100644 lib/event_store/subscriptions/fencing_token.ex create mode 100644 priv/event_store/migrations/v1.5.0.sql create mode 100644 test/storage/subscription_lock_test.exs create mode 100644 test/storage/subscription_previous_version_test.exs diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex index 4202715..fd1c0c0 100644 --- a/lib/event_store/advisory_locks.ex +++ b/lib/event_store/advisory_locks.ex @@ -59,6 +59,28 @@ defmodule EventStore.AdvisoryLocks do 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 @@ -79,7 +101,7 @@ defmodule EventStore.AdvisoryLocks do end alias DBConnection.ConnectionError - alias EventStore.AdvisoryLocks.{Connection, Lock, State} + alias EventStore.AdvisoryLocks.{Connection, Lease, Lock, State} alias EventStore.Storage alias EventStore.Storage.Session @@ -131,6 +153,52 @@ defmodule EventStore.AdvisoryLocks do GenServer.call(server, {:try_advisory_lock, key, self(), timeout}, timeout) end + @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) + end + + def handle_call( + {:try_subscription_lock, subscription_id, owner, timeout}, + _from, + %State{} = state + ) do + %State{schema: schema} = state + + claim = + query(state, fn conn -> + Storage.Lock.claim_subscription_lock(conn, subscription_id, + schema: schema, + timeout: timeout + ) + end) + + case claim do + {:ok, %Storage.Lock.Lease{session: session} = lease} -> + lock = Lock.new(subscription_id, owner, session) + state = monitor_acquired_lock(lock, state) + + {:reply, {:ok, Lease.new(Lock.ref(lock), lease)}, state} + + reply -> + {:reply, reply, state} + end + 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} -> diff --git a/lib/event_store/sql/init.ex b/lib/event_store/sql/init.ex index 99081d3..18f5a97 100644 --- a/lib/event_store/sql/init.ex +++ b/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 diff --git a/lib/event_store/sql/statements.ex b/lib/event_store/sql/statements.ex index 6d53011..ce2a1fd 100644 --- a/lib/event_store/sql/statements.ex +++ b/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]}, diff --git a/lib/event_store/sql/statements/claim_subscription_lock.sql.eex b/lib/event_store/sql/statements/claim_subscription_lock.sql.eex new file mode 100644 index 0000000..798fbae --- /dev/null +++ b/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; diff --git a/lib/event_store/storage/lock.ex b/lib/event_store/storage/lock.ex index 9fd609b..0c4301a 100644 --- a/lib/event_store/storage/lock.ex +++ b/lib/event_store/storage/lock.ex @@ -2,7 +2,9 @@ defmodule EventStore.Storage.Lock do @moduledoc false alias EventStore.Sql.Statements + alias EventStore.Storage.Lock.Lease alias EventStore.Storage.Session + alias EventStore.Subscriptions.FencingToken @doc """ Use two 32-bit key values for advisory locks where the first key acts as the @@ -49,6 +51,40 @@ defmodule EventStore.Storage.Lock do end end + @doc """ + Takes the advisory lock of a subscription and bumps its `lock_generation` in + the same statement, returning the lease that fences the subscription's + progress writes to this lock owner. + """ + def claim_subscription_lock(conn, subscription_id, opts) do + {schema, opts} = Keyword.pop(opts, :schema) + + query = Statements.claim_subscription_lock(schema) + + case Postgrex.query(conn, query, [subscription_id], opts) do + {:ok, %Postgrex.Result{rows: [[false, _backend_pid, _session_id, _generation, _last_seen]]}} -> + {:error, :lock_already_taken} + + {:ok, %Postgrex.Result{rows: [[true, _backend_pid, _session_id, nil, nil]]}} -> + with :ok <- unlock(conn, subscription_id, [schema: schema] ++ opts) do + {:error, :subscription_not_found} + end + + {:ok, %Postgrex.Result{rows: [[true, backend_pid, session_id, generation, last_seen]]}} -> + token = FencingToken.new(subscription_id, generation) + + {:ok, Lease.new(token, last_seen, Session.new(backend_pid, session_id))} + + {:error, %Postgrex.Error{} = error} -> + _ = unlock(conn, subscription_id, [schema: schema] ++ opts) + + {:error, error} + + {:error, %DBConnection.ConnectionError{}} = reply -> + reply + end + end + def unlock(conn, key, opts) do {schema, opts} = Keyword.pop(opts, :schema) diff --git a/lib/event_store/storage/lock/lease.ex b/lib/event_store/storage/lock/lease.ex new file mode 100644 index 0000000..7ecac72 --- /dev/null +++ b/lib/event_store/storage/lock/lease.ex @@ -0,0 +1,19 @@ +defmodule EventStore.Storage.Lock.Lease do + @moduledoc false + + alias EventStore.Storage.Session + alias EventStore.Subscriptions.FencingToken + + @type t :: %__MODULE__{ + token: FencingToken.t(), + last_seen: non_neg_integer() | nil, + session: Session.t() + } + + @enforce_keys [:token, :last_seen, :session] + defstruct [:token, :last_seen, :session] + + def new(%FencingToken{} = token, last_seen, %Session{} = session) do + %__MODULE__{token: token, last_seen: last_seen, session: session} + end +end diff --git a/lib/event_store/subscriptions/fencing_token.ex b/lib/event_store/subscriptions/fencing_token.ex new file mode 100644 index 0000000..b72f7d6 --- /dev/null +++ b/lib/event_store/subscriptions/fencing_token.ex @@ -0,0 +1,16 @@ +defmodule EventStore.Subscriptions.FencingToken do + @moduledoc false + + @type t :: %__MODULE__{ + subscription_id: pos_integer(), + generation: pos_integer() + } + + @enforce_keys [:subscription_id, :generation] + defstruct [:subscription_id, :generation] + + def new(subscription_id, generation) + when is_integer(subscription_id) and is_integer(generation) and generation > 0 do + %__MODULE__{subscription_id: subscription_id, generation: generation} + end +end diff --git a/lib/event_store/tasks/migrations.ex b/lib/event_store/tasks/migrations.ex index 8b3dbc1..a852761 100644 --- a/lib/event_store/tasks/migrations.ex +++ b/lib/event_store/tasks/migrations.ex @@ -19,7 +19,8 @@ defmodule EventStore.Tasks.Migrations do "1.1.0", "1.2.0", "1.3.0", - "1.3.2" + "1.3.2", + "1.5.0" ] @dialyzer {:no_return, exec: 2, handle_response: 1} diff --git a/priv/event_store/migrations/v1.5.0.sql b/priv/event_store/migrations/v1.5.0.sql new file mode 100644 index 0000000..ebbddb4 --- /dev/null +++ b/priv/event_store/migrations/v1.5.0.sql @@ -0,0 +1,8 @@ +DO $migration$ +BEGIN + ALTER TABLE subscriptions ADD COLUMN lock_generation bigint NOT NULL DEFAULT 0; + + -- Record schema migration + INSERT INTO schema_migrations (major_version, minor_version, patch_version) VALUES (1, 5, 0); +END; +$migration$ LANGUAGE plpgsql; diff --git a/test/advisory_locks_test.exs b/test/advisory_locks_test.exs index 5922e93..b337f79 100644 --- a/test/advisory_locks_test.exs +++ b/test/advisory_locks_test.exs @@ -42,6 +42,84 @@ defmodule EventStore.AdvisoryLocksTest do end end + describe "acquire subscription lock" do + test "should acquire the lock and record the lock owner", context do + %{subscription_id: subscription_id} = create_subscription(context) + + assert {:ok, %AdvisoryLocks.Lease{} = lease} = + AdvisoryLocks.try_subscription_lock(@locks, subscription_id) + + assert is_reference(lease.ref) + assert lease.token == EventStore.Subscriptions.FencingToken.new(subscription_id, 1) + assert lease.last_seen == nil + end + + test "should record a new lock owner each time the lock is acquired", context do + %{subscription_id: subscription_id} = create_subscription(context) + + assert {:ok, %{token: %{generation: 1}}} = + AdvisoryLocks.try_subscription_lock(@locks, subscription_id) + + assert {:ok, %{token: %{generation: 2}}} = + AdvisoryLocks.try_subscription_lock(@locks, subscription_id) + end + + test "should fail to acquire the lock when another session holds it", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + assert {:ok, _lease} = + Storage.Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + assert {:error, :lock_already_taken} = + AdvisoryLocks.try_subscription_lock(@locks, subscription_id) + end + + test "should fail to acquire the lock of a subscription that does not exist" do + assert {:error, :subscription_not_found} = + AdvisoryLocks.try_subscription_lock(@locks, 999_999) + end + + test "should send `lock_released` message with the lease reference", context do + %{subscription_id: subscription_id} = create_subscription(context) + + {:ok, %{ref: ref}} = AdvisoryLocks.try_subscription_lock(@locks, subscription_id) + + shutdown_database_connection() + + assert_receive({AdvisoryLocks, :lock_released, ^ref, :shutdown}) + end + + test "should release the lock when the owner terminates", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + reply_to = self() + + pid = + spawn_link(fn -> + {:ok, _lease} = AdvisoryLocks.try_subscription_lock(@locks, subscription_id) + + send(reply_to, :lock_acquired) + + receive do + :shutdown -> :ok + end + end) + + assert_receive :lock_acquired + + assert {:error, :lock_already_taken} = + Storage.Lock.try_acquire_exclusive_lock(conn, subscription_id, schema: schema) + + send(pid, :shutdown) + + Wait.until(fn -> + assert :ok = + Storage.Lock.try_acquire_exclusive_lock(conn, subscription_id, schema: schema) + end) + end + end + describe "release lock" do test "should release lock when process terminates", %{conn: conn, schema: schema} do reply_to = self() @@ -540,4 +618,11 @@ defmodule EventStore.AdvisoryLocksTest do end defp lock_pool_pid, do: Process.whereis(@conn) + + defp create_subscription(%{conn: conn, schema: schema}) do + {:ok, %Storage.Subscription{} = subscription} = + Storage.subscribe_to_stream(conn, "$all", "test_subscription", schema: schema) + + subscription + end end diff --git a/test/list_event_store_migrations_test.exs b/test/list_event_store_migrations_test.exs index b0ac0ae..f45bab7 100644 --- a/test/list_event_store_migrations_test.exs +++ b/test/list_event_store_migrations_test.exs @@ -26,6 +26,7 @@ defmodule EventStore.ListEventStoreMigrationsTest do 1.2.0 \tpending 1.3.0 \tpending 1.3.2 \tpending + 1.5.0 \tpending """ diff --git a/test/migrate_event_store_test.exs b/test/migrate_event_store_test.exs index 31588b0..f958a10 100644 --- a/test/migrate_event_store_test.exs +++ b/test/migrate_event_store_test.exs @@ -88,6 +88,7 @@ defmodule EventStore.MigrateEventStoreTest do "1\t1\t0\tNOW" -> true "1\t2\t0\tNOW" -> true "1\t3\t0\tNOW" -> true + "1\t3\t2\tNOW" -> true _line -> false end) end diff --git a/test/storage/subscription_lock_test.exs b/test/storage/subscription_lock_test.exs new file mode 100644 index 0000000..baf2199 --- /dev/null +++ b/test/storage/subscription_lock_test.exs @@ -0,0 +1,153 @@ +defmodule EventStore.Storage.SubscriptionLockTest do + use EventStore.StorageCase + + alias EventStore.ProcessHelper + alias EventStore.Storage + alias EventStore.Storage.Lock + alias EventStore.Storage.Session + alias EventStore.Subscriptions.FencingToken + + @all_stream "$all" + @subscription_name "test_subscription" + + setup %{postgrex_config: postgrex_config} do + [conn: start_supervised!({Postgrex, postgrex_config})] + end + + test "a new subscription has never had its lock claimed", context do + %{subscription_id: subscription_id} = create_subscription(context) + + assert 0 == lock_generation(context, subscription_id) + end + + test "claim the lock of a new subscription", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + assert {:ok, lease} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + assert is_struct(lease, Lock.Lease) + assert lease.token == FencingToken.new(subscription_id, 1) + assert lease.last_seen == nil + assert %Session{} = lease.session + assert 1 == lock_generation(context, subscription_id) + end + + test "claim the lock resumes from the committed position", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + :ok = Storage.ack_last_seen_event(conn, subscription_id, 3, schema: schema) + + assert {:ok, %{last_seen: 3}} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + end + + test "claim the lock again from the same connection", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + assert {:ok, %{token: %{generation: 1}}} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + assert {:ok, %{token: %{generation: 2}}} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + end + + test "claim the lock while another connection holds it", context do + %{postgrex_config: postgrex_config, conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + {:ok, owner} = Postgrex.start_link(postgrex_config) + + assert {:ok, _lease} = + Lock.claim_subscription_lock(owner, subscription_id, schema: schema) + + assert {:error, :lock_already_taken} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + assert 1 == lock_generation(context, subscription_id) + + ProcessHelper.shutdown(owner) + end + + test "claim the lock of a subscription that no longer exists", context do + %{postgrex_config: postgrex_config, conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + :ok = Storage.delete_subscription(conn, @all_stream, @subscription_name, schema: schema) + + assert {:error, :subscription_not_found} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + {:ok, other} = Postgrex.start_link(postgrex_config) + + assert :ok = Lock.try_acquire_exclusive_lock(other, subscription_id, schema: schema) + + ProcessHelper.shutdown(other) + end + + test "a claim that fails after taking the advisory lock leaves the subscription claimable", + context do + %{postgrex_config: postgrex_config, conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + claiming_conn = + start_supervised!( + {Postgrex, + Keyword.merge(postgrex_config, parameters: [lock_timeout: "200ms"], pool_size: 1)}, + id: :claiming_conn + ) + + test_pid = self() + + holder = + Task.async(fn -> + Postgrex.transaction(conn, fn holder_conn -> + Postgrex.query!( + holder_conn, + "SELECT subscription_id FROM \"#{schema}\".subscriptions WHERE subscription_id = $1 FOR UPDATE", + [subscription_id] + ) + + send(test_pid, :row_locked) + + receive do + :release -> :ok + end + end) + end) + + assert_receive :row_locked + + assert {:error, %Postgrex.Error{postgres: %{code: :lock_not_available}}} = + Lock.claim_subscription_lock(claiming_conn, subscription_id, schema: schema) + + send(holder.pid, :release) + assert {:ok, :ok} = Task.await(holder) + + assert {:ok, %Lock.Lease{}} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + assert :ok = Lock.unlock(conn, subscription_id, schema: schema) + end + + defp create_subscription(%{conn: conn, schema: schema}) do + {:ok, %Storage.Subscription{} = subscription} = + Storage.subscribe_to_stream(conn, @all_stream, @subscription_name, schema: schema) + + subscription + end + + defp lock_generation(%{conn: conn, schema: schema}, subscription_id) do + %Postgrex.Result{rows: [[generation]]} = + Postgrex.query!( + conn, + "SELECT lock_generation FROM \"#{schema}\".subscriptions WHERE subscription_id = $1", + [subscription_id] + ) + + generation + end +end diff --git a/test/storage/subscription_previous_version_test.exs b/test/storage/subscription_previous_version_test.exs new file mode 100644 index 0000000..57ad2a7 --- /dev/null +++ b/test/storage/subscription_previous_version_test.exs @@ -0,0 +1,91 @@ +defmodule EventStore.Storage.SubscriptionPreviousVersionTest do + use EventStore.StorageCase + + alias EventStore.ProcessHelper + alias EventStore.Storage.Lock + alias EventStore.Subscriptions.FencingToken + + @insert_subscription """ + INSERT INTO "public".subscriptions + ( + stream_uuid, + subscription_name, + last_seen + ) + VALUES ($1, $2, $3) + RETURNING subscription_id, stream_uuid, subscription_name, last_seen, created_at; + """ + + @try_advisory_lock """ + SELECT pg_try_advisory_lock( + '"public".subscriptions'::regclass::oid::int, + (CASE WHEN $1 > 2147483647 THEN mod($1, 2147483647) ELSE $1 END)::int + ); + """ + + @subscription_ack """ + UPDATE "public".subscriptions + SET last_seen = $2 + WHERE subscription_id = $1; + """ + + @query_subscription """ + SELECT + subscription_id, + stream_uuid, + subscription_name, + last_seen, + created_at + FROM "public".subscriptions + WHERE stream_uuid = $1 AND subscription_name = $2; + """ + + setup %{postgrex_config: postgrex_config} do + {:ok, previous_node} = Postgrex.start_link(postgrex_config) + + on_exit(fn -> ProcessHelper.shutdown(previous_node) end) + + [previous_node: previous_node] + end + + test "a node on the previous version keeps working against the migrated schema", context do + %{conn: conn, previous_node: previous_node, schema: schema} = context + + %Postgrex.Result{rows: [[subscription_id | _columns]]} = + Postgrex.query!(previous_node, @insert_subscription, ["$all", "previous_version", nil]) + + assert 0 == lock_generation(conn, subscription_id) + + assert %Postgrex.Result{rows: [[true]]} = + Postgrex.query!(previous_node, @try_advisory_lock, [subscription_id]) + + Postgrex.query!(previous_node, @subscription_ack, [subscription_id, 7]) + + assert 0 == lock_generation(conn, subscription_id) + + assert %Postgrex.Result{ + rows: [[^subscription_id, "$all", "previous_version", 7, _created_at]] + } = + Postgrex.query!(previous_node, @query_subscription, ["$all", "previous_version"]) + + assert {:error, :lock_already_taken} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + :ok = Lock.unlock(previous_node, subscription_id, schema: schema) + + assert {:ok, lease} = Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + assert lease.token == FencingToken.new(subscription_id, 1) + assert lease.last_seen == 7 + end + + defp lock_generation(conn, subscription_id) do + %Postgrex.Result{rows: [[generation]]} = + Postgrex.query!( + conn, + "SELECT lock_generation FROM public.subscriptions WHERE subscription_id = $1", + [subscription_id] + ) + + generation + end +end From a9d120921b0134406823e8981ccce53dc1b74a91 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 16:41:16 -0400 Subject: [PATCH 08/13] fix(subscriptions): stop a subscription from writing progress once another node owns it (#30) Signed-off-by: Yordis Prieto --- lib/event_store/advisory_locks.ex | 128 ++++- .../sql/statements/subscription_ack.sql.eex | 11 +- lib/event_store/storage.ex | 2 +- lib/event_store/storage/subscription.ex | 26 +- lib/event_store/subscriptions/subscription.ex | 23 +- .../subscriptions/subscription_fsm.ex | 36 +- .../subscriptions/subscription_state.ex | 1 + mix.exs | 3 +- mix.lock | 1 + test/advisory_locks_test.exs | 199 +++++++- test/migrated_event_store_test.exs | 55 +++ .../subscription_fencing_property_test.exs | 171 +++++++ test/storage/subscription_lock_test.exs | 59 ++- .../storage/subscription_persistence_test.exs | 58 ++- .../subscription_churn_property_test.exs | 332 +++++++++++++ .../subscription_fencing_test.exs | 442 ++++++++++++++++++ test/subscriptions/subscription_fsm_test.exs | 139 +++++- .../subscription_locking_test.exs | 140 +++++- .../subscription_unmigrated_schema_test.exs | 55 +++ .../subscription_zombie_lock_test.exs | 175 +++++++ test/support/tcp_proxy.ex | 181 +++++++ 21 files changed, 2157 insertions(+), 80 deletions(-) create mode 100644 test/storage/subscription_fencing_property_test.exs create mode 100644 test/subscriptions/subscription_churn_property_test.exs create mode 100644 test/subscriptions/subscription_fencing_test.exs create mode 100644 test/subscriptions/subscription_unmigrated_schema_test.exs create mode 100644 test/subscriptions/subscription_zombie_lock_test.exs create mode 100644 test/support/tcp_proxy.ex diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex index fd1c0c0..35c2250 100644 --- a/lib/event_store/advisory_locks.ex +++ b/lib/event_store/advisory_locks.ex @@ -40,18 +40,20 @@ defmodule EventStore.AdvisoryLocks do 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 @@ -151,6 +153,8 @@ 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 @doc """ @@ -170,6 +174,40 @@ defmodule EventStore.AdvisoryLocks do 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( @@ -177,40 +215,66 @@ defmodule EventStore.AdvisoryLocks do _from, %State{} = state ) do - %State{schema: schema} = state - - claim = - query(state, fn conn -> - Storage.Lock.claim_subscription_lock(conn, subscription_id, - schema: schema, - timeout: timeout - ) - end) - - case claim do - {:ok, %Storage.Lock.Lease{session: session} = lease} -> - lock = Lock.new(subscription_id, owner, session) + 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, Lease.new(Lock.ref(lock), lease)}, state} - reply -> + {{owner_ref, %Lock{} = held}, {:ok, %Storage.Lock.Lease{session: session} = lease}} -> + %Lock{key: key, ref: ref} = held + %State{locks: locks} = state + + 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 try_acquire_exclusive_lock(key, owner, timeout, state) do - {:ok, %Lock{} = lock} -> - state = monitor_acquired_lock(lock, state) - + case held_lock(state, key, owner) do + {_owner_ref, %Lock{} = lock} -> {:reply, {:ok, Lock.ref(lock)}, state} - reply -> - {:reply, reply, 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{schema: schema} = state @@ -373,6 +437,22 @@ 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{key: key} = lock, %State{} = state) do + %State{query_timeout: query_timeout, released: released} = state + + case query(state, &Session.current(&1, timeout: query_timeout)) do + {:ok, %Session{} = session} -> + if Lock.held_by?(lock, session), do: release_lock(key, state) + + state + + {:error, _error} -> + %State{state | released: [lock | released]} + end + end + defp release_lock(key, %State{} = state) do %State{query_timeout: query_timeout, schema: schema} = state diff --git a/lib/event_store/sql/statements/subscription_ack.sql.eex b/lib/event_store/sql/statements/subscription_ack.sql.eex index c6da30f..ee3f951 100644 --- a/lib/event_store/sql/statements/subscription_ack.sql.eex +++ b/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; diff --git a/lib/event_store/storage.ex b/lib/event_store/storage.ex index 668d88c..7e95d43 100644 --- a/lib/event_store/storage.ex +++ b/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. diff --git a/lib/event_store/storage/subscription.ex b/lib/event_store/storage/subscription.ex index c609a86..2d41f1b 100644 --- a/lib/event_store/storage/subscription.ex +++ b/lib/event_store/storage/subscription.ex @@ -5,6 +5,7 @@ defmodule EventStore.Storage.Subscription do alias EventStore.Sql.Statements alias EventStore.Storage.Subscription + alias EventStore.Subscriptions.FencingToken alias EventStore.Storage.Subscription.{ CreateSubscription, @@ -41,8 +42,8 @@ defmodule EventStore.Storage.Subscription do end end - def ack_last_seen_event(conn, subscription_id, last_seen, opts) do - Subscription.Ack.execute(conn, subscription_id, last_seen, opts) + def ack_last_seen_event(conn, %FencingToken{} = token, last_seen, opts) do + Subscription.Ack.execute(conn, token, last_seen, opts) end def delete_subscription(conn, stream_uuid, subscription_name, opts), @@ -136,25 +137,34 @@ defmodule EventStore.Storage.Subscription do defmodule Ack do @moduledoc false - def execute(conn, subscription_id, last_seen, opts) do + def execute(conn, %FencingToken{} = token, last_seen, opts) do + %FencingToken{subscription_id: subscription_id, generation: generation} = token + {schema, opts} = Keyword.pop(opts, :schema) query = Statements.subscription_ack(schema) - case Postgrex.query(conn, query, [subscription_id, last_seen], opts) do + case Postgrex.query(conn, query, [subscription_id, last_seen, generation], opts) do # Naming the subscription by its identifier, rather than by the stream and name a later # subscription can reuse, is what keeps this from writing a checkpoint onto whichever row # holds that name now. - {:ok, %Postgrex.Result{num_rows: 0}} -> + {:ok, %Postgrex.Result{rows: [[true, _present]]}} -> + :ok + + {:ok, %Postgrex.Result{rows: [[false, true]]}} -> + Logger.warning( + "Failed to ack last seen event for subscription #{subscription_id} because another lock owner has claimed it" + ) + + {:error, :fenced} + + {:ok, %Postgrex.Result{rows: [[false, false]]}} -> Logger.warning( "Failed to ack last seen event for subscription #{subscription_id} because it no longer exists" ) {:error, :subscription_not_found} - {:ok, _result} -> - :ok - {:error, error} = reply -> Logger.warning( "Failed to ack last seen event for subscription #{subscription_id} due to: " <> diff --git a/lib/event_store/subscriptions/subscription.ex b/lib/event_store/subscriptions/subscription.ex index 08497ea..61497a3 100644 --- a/lib/event_store/subscriptions/subscription.ex +++ b/lib/event_store/subscriptions/subscription.ex @@ -11,7 +11,7 @@ defmodule EventStore.Subscriptions.Subscription do require Logger - alias EventStore.RecordedEvent + alias EventStore.{AdvisoryLocks, RecordedEvent} alias EventStore.Subscriptions.{SubscriptionFsm, Subscription, SubscriptionState} # Mirrors the storage `:timeout` a subscription gives its own queries, for the internal callers @@ -169,6 +169,27 @@ defmodule EventStore.Subscriptions.Subscription do {:noreply, state} end + @impl GenServer + def handle_info({:fenced, lock_ref}, %Subscription{} = state) do + %Subscription{ + subscription: %SubscriptionFsm{data: %SubscriptionState{} = data} = subscription + } = + state + + %SubscriptionState{event_store: event_store} = data + + Logger.warning(describe(state) <> " lost its lock to another owner, disconnecting") + + :ok = AdvisoryLocks.release(Module.concat(event_store, AdvisoryLocks), lock_ref) + + state = + subscription + |> SubscriptionFsm.disconnect(lock_ref) + |> apply_subscription_to_state(state) + + {:noreply, state} + end + @impl GenServer def handle_info({:DOWN, _ref, :process, pid, reason}, %Subscription{} = state) do %Subscription{subscription: subscription} = state diff --git a/lib/event_store/subscriptions/subscription_fsm.ex b/lib/event_store/subscriptions/subscription_fsm.ex index 079e1bc..6490506 100644 --- a/lib/event_store/subscriptions/subscription_fsm.ex +++ b/lib/event_store/subscriptions/subscription_fsm.ex @@ -72,10 +72,11 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do data = SubscriptionState.reset_event_tracking(data) with {:ok, subscription} <- create_subscription(data), - {:ok, lock_ref} <- try_acquire_exclusive_lock(data, subscription), + {:ok, %AdvisoryLocks.Lease{} = lease} <- + try_acquire_subscription_lock(data, subscription), :ok <- subscribe_to_events(data) do - %Storage.Subscription{subscription_id: subscription_id, last_seen: last_seen} = - subscription + %Storage.Subscription{subscription_id: subscription_id} = subscription + %AdvisoryLocks.Lease{ref: lock_ref, token: token, last_seen: last_seen} = lease last_seen = last_seen || 0 @@ -83,6 +84,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do data | subscription_id: subscription_id, lock_ref: lock_ref, + fencing_token: token, last_received: last_seen, last_sent: last_seen, last_ack: last_seen @@ -101,18 +103,20 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do # Attempt to subscribe def subscribe(%__MODULE__{state: :disconnected, data: %SubscriptionState{} = data} = fsm) do with {:ok, subscription} <- create_subscription(data), - {:ok, lock_ref} <- try_acquire_exclusive_lock(data, subscription) do - %Storage.Subscription{ - subscription_id: subscription_id, - last_seen: last_seen - } = subscription + {:ok, %AdvisoryLocks.Lease{} = lease} <- + try_acquire_subscription_lock(data, subscription) do + %Storage.Subscription{subscription_id: subscription_id} = subscription + %AdvisoryLocks.Lease{ref: lock_ref, token: token, last_seen: last_seen} = lease last_ack = last_seen || 0 + %SubscriptionState{last_received: last_received} = data data = %SubscriptionState{ data | subscription_id: subscription_id, lock_ref: lock_ref, + fencing_token: token, + last_received: max(last_received, last_ack), last_sent: last_ack, last_ack: last_ack } @@ -306,7 +310,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do lock_ref ) do data = - %SubscriptionState{data | lock_ref: nil} + %SubscriptionState{data | lock_ref: nil, fencing_token: nil} |> SubscriptionState.reset_event_tracking() |> purge_in_flight_events() @@ -357,7 +361,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do ) end - defp try_acquire_exclusive_lock( + defp try_acquire_subscription_lock( %SubscriptionState{} = data, %Storage.Subscription{} = subscription ) do @@ -366,7 +370,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do server = Module.concat(event_store, AdvisoryLocks) - AdvisoryLocks.try_advisory_lock(server, subscription_id, timeout) + AdvisoryLocks.try_subscription_lock(server, subscription_id, timeout) end defp subscribe_to_events(%SubscriptionState{} = data) do @@ -765,14 +769,15 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do schema: schema, stream_uuid: stream_uuid, subscription_name: subscription_name, - subscription_id: subscription_id, + fencing_token: fencing_token, + lock_ref: lock_ref, last_ack: last_ack, query_timeout: query_timeout, checkpoints_pending: checkpoints_pending, telemetry_metadata: telemetry_metadata } = data - if checkpoints_pending > 0 do + if checkpoints_pending > 0 and fencing_token != nil do metadata = Map.merge(telemetry_metadata, %{ stream_uuid: stream_uuid, @@ -782,7 +787,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do result = Telemetry.span(:subscription_checkpoint, metadata, fn -> - Storage.Subscription.ack_last_seen_event(conn, subscription_id, last_ack, + Storage.Subscription.ack_last_seen_event(conn, fencing_token, last_ack, schema: schema, timeout: query_timeout ) @@ -795,6 +800,9 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do {:error, :subscription_not_found} -> send(self(), {:checkpoint_failed, :subscription_not_found}) + {:error, :fenced} -> + send(self(), {:fenced, lock_ref}) + _ -> :ok end diff --git a/lib/event_store/subscriptions/subscription_state.ex b/lib/event_store/subscriptions/subscription_state.ex index 20ae04e..7a0a5c7 100644 --- a/lib/event_store/subscriptions/subscription_state.ex +++ b/lib/event_store/subscriptions/subscription_state.ex @@ -19,6 +19,7 @@ defmodule EventStore.Subscriptions.SubscriptionState do :partition_by, :query_timeout, :lock_ref, + :fencing_token, telemetry_metadata: %{}, correlation_id_type: "uuid", causation_id_type: "uuid", diff --git a/mix.exs b/mix.exs index d390d60..20a5ab0 100644 --- a/mix.exs +++ b/mix.exs @@ -50,7 +50,8 @@ defmodule EventStore.Mixfile do # Development and test tooling {:benchfella, "~> 0.3", only: :bench}, {:dialyxir, "~> 1.4", only: :dev, runtime: false}, - {:ex_doc, ">= 0.0.0", only: :dev} + {:ex_doc, ">= 0.0.0", only: :dev}, + {:stream_data, "~> 1.1", only: [:test, :jsonb, :text_ids, :migration]} ] end diff --git a/mix.lock b/mix.lock index 85f2ddc..12283e0 100644 --- a/mix.lock +++ b/mix.lock @@ -15,5 +15,6 @@ "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"}, "poolboy": {:hex, :poolboy, "1.5.2", "392b007a1693a64540cead79830443abf5762f5d30cf50bc95cb2c1aaafa006b", [:rebar3], [], "hexpm", "dad79704ce5440f3d5a3681c8590b9dc25d1a561e8f5a9c995281012860901e3"}, "postgrex": {:hex, :postgrex, "0.22.4", "d271f595dfd25230b6398354e19d17bb5e2d20130fd2d9bdca7e15f125d43552", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "4aae45a2d60e35b04eea2602440be152fae332901f1fc7a60fc7cb7f0f9a9c5a"}, + "stream_data": {:hex, :stream_data, "1.4.0", "026f929db613aabea6208012ae9b8970d3fd5f88b3bdf26831bc536f98c42036", [:mix], [], "hexpm", "2b0ee3a340dcce1c8cf6302a763ee757d1e01c54d6e16d9069062509d68b1dc9"}, "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, } diff --git a/test/advisory_locks_test.exs b/test/advisory_locks_test.exs index b337f79..0aa9383 100644 --- a/test/advisory_locks_test.exs +++ b/test/advisory_locks_test.exs @@ -25,14 +25,22 @@ defmodule EventStore.AdvisoryLocksTest do assert is_reference(lock) end - test "should acquire lock when same process already has lock" do + test "should hand the owner the lock it already holds and free it with one release", + %{conn: conn, schema: schema} do assert {:ok, lock1} = AdvisoryLocks.try_advisory_lock(@locks, 1) assert {:ok, lock2} = AdvisoryLocks.try_advisory_lock(@locks, 1) assert {:ok, lock3} = AdvisoryLocks.try_advisory_lock(@locks, 1) assert is_reference(lock1) - assert is_reference(lock2) - assert is_reference(lock3) + assert lock2 == lock1 + assert lock3 == lock1 + + assert :ok = AdvisoryLocks.release(@locks, lock1) + + _state = :sys.get_state(@locks) + + assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema) + assert :ok = Storage.Lock.unlock(conn, 1, schema: schema) end test "should fail to acquire lock when already taken", %{conn: conn, schema: schema} do @@ -64,6 +72,66 @@ defmodule EventStore.AdvisoryLocksTest do AdvisoryLocks.try_subscription_lock(@locks, subscription_id) end + test "should keep one hold when the owner claims the lock again", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + assert {:ok, %AdvisoryLocks.Lease{} = first} = + AdvisoryLocks.try_subscription_lock(@locks, subscription_id) + + Postgrex.query!( + conn, + "UPDATE \"#{schema}\".subscriptions SET last_seen = 7 WHERE subscription_id = $1", + [subscription_id] + ) + + assert {:ok, %AdvisoryLocks.Lease{} = second} = + AdvisoryLocks.try_subscription_lock(@locks, subscription_id) + + assert second.ref == first.ref + assert second.token.generation == first.token.generation + 1 + assert second.last_seen == 7 + + assert :ok = AdvisoryLocks.release(@locks, first.ref) + + _state = :sys.get_state(@locks) + + assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, subscription_id, schema: schema) + assert :ok = Storage.Lock.unlock(conn, subscription_id, schema: schema) + end + + test "should free the lock with one release after a claim outlived its caller", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + {:ok, busy_ref} = AdvisoryLocks.try_advisory_lock(@locks, 2) + + holder = hold_only_connection() + + try do + :ok = AdvisoryLocks.release(@locks, busy_ref) + + wait_until_waiting_for_connection() + + assert {:error, :timeout} = + AdvisoryLocks.try_subscription_lock(@locks, subscription_id, 300) + after + send(holder, :release) + end + + _state = :sys.get_state(@locks, 5_000) + + assert {:ok, %AdvisoryLocks.Lease{ref: ref}} = + AdvisoryLocks.try_subscription_lock(@locks, subscription_id) + + assert :ok = AdvisoryLocks.release(@locks, ref) + + _state = :sys.get_state(@locks, 5_000) + + assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, subscription_id, schema: schema) + assert :ok = Storage.Lock.unlock(conn, subscription_id, schema: schema) + end + test "should fail to acquire the lock when another session holds it", context do %{conn: conn, schema: schema} = context %{subscription_id: subscription_id} = create_subscription(context) @@ -120,6 +188,127 @@ defmodule EventStore.AdvisoryLocksTest do end end + describe "release a lock by reference" do + test "should free the key for another session", %{conn: conn, schema: schema} do + {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1) + + assert {:error, :lock_already_taken} = + Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema) + + assert :ok = AdvisoryLocks.release(@locks, lock) + + Wait.until(fn -> + assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema) + end) + end + + test "should free a subscription lock for another session", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + {:ok, %AdvisoryLocks.Lease{ref: ref}} = + AdvisoryLocks.try_subscription_lock(@locks, subscription_id) + + assert {:error, :lock_already_taken} = + Storage.Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + assert :ok = AdvisoryLocks.release(@locks, ref) + + _ = :sys.get_state(@locks) + + Wait.until(fn -> + assert {:ok, %{token: %{generation: 2}}} = + Storage.Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + end) + end + + test "should keep the key held for another owner of the same key", + %{conn: conn, schema: schema} do + reply_to = self() + + other_owner = + spawn_link(fn -> + {:ok, _lock} = AdvisoryLocks.try_advisory_lock(@locks, 1) + + send(reply_to, :lock_acquired) + + receive do + :shutdown -> :ok + end + end) + + assert_receive :lock_acquired + + {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1) + + assert :ok = AdvisoryLocks.release(@locks, lock) + + _state = :sys.get_state(@locks) + + assert {:error, :lock_already_taken} = + Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema) + + send(other_owner, :shutdown) + + Wait.until(fn -> + assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema) + end) + end + + test "should ignore a reference that is not held" do + assert :ok = AdvisoryLocks.release(@locks, make_ref()) + end + + test "should not notify the owner that the lock was released" do + {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1) + + assert :ok = AdvisoryLocks.release(@locks, lock) + + refute_receive {AdvisoryLocks, :lock_released, ^lock, _reason}, 200 + end + + test "should not wait for the lock server to answer", %{conn: conn, schema: schema} do + {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1) + + :ok = :sys.suspend(@locks) + + try do + {elapsed, :ok} = :timer.tc(fn -> AdvisoryLocks.release(@locks, lock) end, :millisecond) + + assert elapsed < 1_000 + after + :ok = :sys.resume(@locks) + end + + _state = :sys.get_state(@locks) + + assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema) + end + + test "should not crash a claim the busy lock server cannot answer in time", context do + %{subscription_id: held_id} = create_subscription(context) + + {:ok, %AdvisoryLocks.Lease{ref: ref}} = AdvisoryLocks.try_subscription_lock(@locks, held_id) + + other = create_subscription(context, "other_subscription") + + holder = hold_only_connection() + + try do + :ok = AdvisoryLocks.release(@locks, ref) + + wait_until_waiting_for_connection() + + assert {:error, :timeout} = + AdvisoryLocks.try_subscription_lock(@locks, other.subscription_id, 300) + after + send(holder, :release) + end + + _state = :sys.get_state(@locks, 5_000) + end + end + describe "release lock" do test "should release lock when process terminates", %{conn: conn, schema: schema} do reply_to = self() @@ -619,9 +808,9 @@ defmodule EventStore.AdvisoryLocksTest do defp lock_pool_pid, do: Process.whereis(@conn) - defp create_subscription(%{conn: conn, schema: schema}) do + defp create_subscription(%{conn: conn, schema: schema}, name \\ "test_subscription") do {:ok, %Storage.Subscription{} = subscription} = - Storage.subscribe_to_stream(conn, "$all", "test_subscription", schema: schema) + Storage.subscribe_to_stream(conn, "$all", name, schema: schema) subscription end diff --git a/test/migrated_event_store_test.exs b/test/migrated_event_store_test.exs index fa7c7a4..45081c6 100644 --- a/test/migrated_event_store_test.exs +++ b/test/migrated_event_store_test.exs @@ -124,3 +124,58 @@ defmodule EventStore.MigratedEventStoreTest do EventStore.Tasks.Migrate.exec(config, quiet: true) end end + +defmodule EventStore.MigratedSubscriptionLockTest do + use ExUnit.Case + + alias EventStore.Config + alias EventStore.Storage + alias EventStore.Subscriptions.FencingToken + + @moduletag :migration + + setup_all do + config = TestEventStore.config() + + EventStore.Storage.Database.drop(config) + :ok = EventStore.Storage.Database.create(config) + {_, 0} = EventStore.Storage.Database.restore(config, "test/fixture/eventstore_seed.dump") + EventStore.Tasks.Migrate.exec(config, quiet: true) + + postgrex_config = + TestEventStore |> Config.parsed(:eventstore) |> Config.default_postgrex_opts() + + [conn: start_supervised!({Postgrex, postgrex_config})] + end + + test "claims the lock of a subscription that existed before the migration", %{conn: conn} do + assert [0, 0, 0, 0] == lock_generations(conn) + + assert {:ok, lease} = Storage.Lock.claim_subscription_lock(conn, 2, schema: "public") + + assert lease.last_seen == 100 + assert lease.token == FencingToken.new(2, 1) + assert [0, 1, 0, 0] == lock_generations(conn) + + assert :ok = + Storage.Subscription.ack_last_seen_event(conn, lease.token, 101, schema: "public") + + assert %Postgrex.Result{rows: [[101]]} = + Postgrex.query!( + conn, + "SELECT last_seen FROM public.subscriptions WHERE subscription_id = 2", + [] + ) + end + + defp lock_generations(conn) do + %Postgrex.Result{rows: rows} = + Postgrex.query!( + conn, + "SELECT lock_generation FROM public.subscriptions ORDER BY subscription_id", + [] + ) + + List.flatten(rows) + end +end diff --git a/test/storage/subscription_fencing_property_test.exs b/test/storage/subscription_fencing_property_test.exs new file mode 100644 index 0000000..36dbe94 --- /dev/null +++ b/test/storage/subscription_fencing_property_test.exs @@ -0,0 +1,171 @@ +defmodule EventStore.Storage.SubscriptionFencingPropertyTest do + use EventStore.StorageCase + use ExUnitProperties + + alias EventStore.{Storage, UUID, Wait} + alias EventStore.Storage.Lock + + @sessions [:one, :two, :three] + + defmodule Model do + defstruct owner: nil, generation: 0, last_seen: nil, tokens: %{}, connections: %{} + end + + property "progress only moves forward and only for the latest lock owner", context do + check all(commands <- list_of(command(), max_length: 12), max_runs: 30) do + subscription_id = create_subscription(context) + model = start_sessions(%Model{}, context) + + try do + Enum.reduce(commands, model, fn command, model -> + model = execute(command, model, subscription_id, context) + + assert {model.generation, model.last_seen} == read(subscription_id, context) + + model + end) + after + delete_subscription(subscription_id, context) + stop_sessions(model) + end + end + end + + defp command do + session = member_of(@sessions) + + one_of([ + tuple({constant(:claim), session}), + tuple({constant(:ack), session, integer(0..50)}), + tuple({constant(:drop), session}), + tuple({constant(:take_over), session}) + ]) + end + + defp execute({:take_over, session}, %Model{owner: owner} = model, subscription_id, context) + when owner not in [nil, session] do + model = execute({:drop, owner}, model, subscription_id, context) + + execute({:claim, session}, model, subscription_id, context) + end + + defp execute({:take_over, session}, %Model{} = model, subscription_id, context) do + execute({:claim, session}, model, subscription_id, context) + end + + defp execute({:claim, session}, %Model{owner: owner} = model, subscription_id, context) + when owner not in [nil, session] do + assert {:error, :lock_already_taken} = claim(model, session, subscription_id, context) + + model + end + + defp execute({:claim, session}, %Model{} = model, subscription_id, context) do + assert {:ok, lease} = claim(model, session, subscription_id, context) + + assert lease.last_seen == model.last_seen + assert lease.token.generation == model.generation + 1 + + %Model{ + model + | owner: session, + generation: model.generation + 1, + tokens: Map.put(model.tokens, session, lease.token) + } + end + + defp execute({:ack, session, position}, %Model{} = model, _subscription_id, context) do + case Map.get(model.tokens, session) do + nil -> + model + + %{generation: generation} = token when generation == model.generation -> + assert :ok = ack(model, session, token, position, context) + + %Model{model | last_seen: max(model.last_seen || position, position)} + + token -> + assert {:error, :fenced} = ack(model, session, token, position, context) + + model + end + end + + defp execute({:drop, session}, %Model{} = model, _subscription_id, context) do + model = end_session(model, session, context) + + %Model{model | owner: if(model.owner == session, do: nil, else: model.owner)} + end + + defp claim(model, session, subscription_id, %{schema: schema}) do + Lock.claim_subscription_lock(connection(model, session), subscription_id, schema: schema) + end + + defp ack(model, session, token, position, %{schema: schema}) do + Storage.ack_last_seen_event(connection(model, session), token, position, schema: schema) + end + + defp stop_sessions(%Model{connections: connections}) do + Enum.each(connections, fn {_session, connection} -> GenServer.stop(connection) end) + end + + defp connection(%Model{connections: connections}, session), do: Map.fetch!(connections, session) + + defp start_sessions(%Model{} = model, %{postgrex_config: postgrex_config}) do + connections = + Map.new(@sessions, fn session -> + {:ok, connection} = Postgrex.start_link(postgrex_config) + + {session, connection} + end) + + %Model{model | connections: connections} + end + + defp end_session(%Model{} = model, session, %{conn: conn}) do + connection = connection(model, session) + + %Postgrex.Result{rows: [[backend_pid]]} = + Postgrex.query!(connection, "SELECT pg_backend_pid()", []) + + %Postgrex.Result{rows: [[true]]} = + Postgrex.query!(conn, "SELECT pg_terminate_backend($1)", [backend_pid]) + + Wait.until(fn -> + assert %Postgrex.Result{rows: [[0]]} = + Postgrex.query!( + conn, + "SELECT count(*)::int FROM pg_stat_activity WHERE pid = $1", + [backend_pid] + ) + end) + + Wait.until(fn -> assert {:ok, _result} = Postgrex.query(connection, "SELECT 1", []) end) + + model + end + + defp create_subscription(%{conn: conn, schema: schema}) do + {:ok, %Storage.Subscription{subscription_id: subscription_id}} = + Storage.subscribe_to_stream(conn, "$all", UUID.uuid4(), schema: schema) + + subscription_id + end + + defp delete_subscription(subscription_id, %{conn: conn}) do + Postgrex.query!(conn, "DELETE FROM public.subscriptions WHERE subscription_id = $1", [ + subscription_id + ]) + end + + defp read(subscription_id, %{conn: conn}) do + %Postgrex.Result{rows: [[generation, last_seen]]} = + Postgrex.query!( + conn, + "SELECT lock_generation, last_seen FROM public.subscriptions WHERE subscription_id = $1", + [subscription_id] + ) + + {generation, last_seen} + end +end diff --git a/test/storage/subscription_lock_test.exs b/test/storage/subscription_lock_test.exs index baf2199..36c78d7 100644 --- a/test/storage/subscription_lock_test.exs +++ b/test/storage/subscription_lock_test.exs @@ -38,7 +38,9 @@ defmodule EventStore.Storage.SubscriptionLockTest do %{conn: conn, schema: schema} = context %{subscription_id: subscription_id} = create_subscription(context) - :ok = Storage.ack_last_seen_event(conn, subscription_id, 3, schema: schema) + {:ok, %{token: token}} = Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + :ok = Storage.ack_last_seen_event(conn, token, 3, schema: schema) assert {:ok, %{last_seen: 3}} = Lock.claim_subscription_lock(conn, subscription_id, schema: schema) @@ -133,6 +135,50 @@ defmodule EventStore.Storage.SubscriptionLockTest do assert :ok = Lock.unlock(conn, subscription_id, schema: schema) end + describe "ack with a fencing token" do + test "a stale generation is fenced and leaves the position alone", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + {:ok, %{token: stale}} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + {:ok, %{token: current}} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + assert :ok = Storage.ack_last_seen_event(conn, current, 3, schema: schema) + + assert {:error, :fenced} = Storage.ack_last_seen_event(conn, stale, 7, schema: schema) + + assert 3 == last_seen(context, subscription_id) + end + + test "the current generation is accepted", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + {:ok, %{token: token}} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + assert :ok = Storage.ack_last_seen_event(conn, token, 4, schema: schema) + + assert 4 == last_seen(context, subscription_id) + end + + test "a subscription that no longer exists is not found rather than fenced", context do + %{conn: conn, schema: schema} = context + %{subscription_id: subscription_id} = create_subscription(context) + + {:ok, %{token: token}} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + :ok = Storage.delete_subscription(conn, @all_stream, @subscription_name, schema: schema) + + assert {:error, :subscription_not_found} = + Storage.ack_last_seen_event(conn, token, 1, schema: schema) + end + end + defp create_subscription(%{conn: conn, schema: schema}) do {:ok, %Storage.Subscription{} = subscription} = Storage.subscribe_to_stream(conn, @all_stream, @subscription_name, schema: schema) @@ -150,4 +196,15 @@ defmodule EventStore.Storage.SubscriptionLockTest do generation end + + defp last_seen(%{conn: conn, schema: schema}, subscription_id) do + %Postgrex.Result{rows: [[last_seen]]} = + Postgrex.query!( + conn, + "SELECT last_seen FROM \"#{schema}\".subscriptions WHERE subscription_id = $1", + [subscription_id] + ) + + last_seen + end end diff --git a/test/storage/subscription_persistence_test.exs b/test/storage/subscription_persistence_test.exs index fd7ce5f..5577491 100644 --- a/test/storage/subscription_persistence_test.exs +++ b/test/storage/subscription_persistence_test.exs @@ -2,6 +2,8 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do use EventStore.StorageCase alias EventStore.Storage + alias EventStore.Storage.Lock + alias EventStore.Subscriptions.FencingToken @all_stream "$all" @subscription_name "test_subscription" @@ -74,10 +76,12 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) + token = claim_token(context, subscription_id) + :ok = delete_subscription(context) assert {:error, :subscription_not_found} = - Storage.ack_last_seen_event(conn, subscription_id, 1, schema: schema) + Storage.ack_last_seen_event(conn, token, 1, schema: schema) end test "ack last seen event only moves the subscription it names", context do @@ -88,7 +92,7 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do {:ok, %Storage.Subscription{subscription_id: untouched}} = Storage.subscribe_to_stream(conn, @all_stream, "another_subscription", schema: schema) - :ok = Storage.ack_last_seen_event(conn, acked, 1, schema: schema) + :ok = Storage.ack_last_seen_event(conn, claim_token(context, acked), 1, schema: schema) assert {:ok, %Storage.Subscription{subscription_id: ^acked, last_seen: 1}} = Storage.Subscription.subscription(conn, @all_stream, @subscription_name, @@ -106,6 +110,8 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do {:ok, %Storage.Subscription{subscription_id: deleted}} = subscribe_to_stream(context) + deleted_token = claim_token(context, deleted) + :ok = delete_subscription(context) {:ok, %Storage.Subscription{subscription_id: recreated}} = subscribe_to_stream(context) @@ -113,7 +119,7 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do assert recreated != deleted assert {:error, :subscription_not_found} = - Storage.ack_last_seen_event(conn, deleted, 1, schema: schema) + Storage.ack_last_seen_event(conn, deleted_token, 1, schema: schema) # The stream and name are the same as the deleted subscription's, so anything keyed by those # would have credited this subscription with a position it never acknowledged. @@ -130,7 +136,9 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) assert {:error, %Postgrex.Error{}} = - Storage.ack_last_seen_event(conn, subscription_id, 1, schema: "no_such_schema") + Storage.ack_last_seen_event(conn, claim_token(context, subscription_id), 1, + schema: "no_such_schema" + ) end test "ack last seen event for a subscription that never existed", context do @@ -139,7 +147,9 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) assert {:error, :subscription_not_found} = - Storage.ack_last_seen_event(conn, subscription_id + 1_000, 1, schema: schema) + Storage.ack_last_seen_event(conn, FencingToken.new(subscription_id + 1_000, 1), 1, + schema: schema + ) assert {:ok, %Storage.Subscription{last_seen: nil}} = read_subscription(context) end @@ -149,11 +159,13 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) - assert :ok = Storage.ack_last_seen_event(conn, subscription_id, 3, schema: schema) + token = claim_token(context, subscription_id) + + assert :ok = Storage.ack_last_seen_event(conn, token, 3, schema: schema) # Postgres counts the rows it matched rather than the rows it changed, so an acknowledgement # that moves nothing is still an acknowledgement its row accepted. - assert :ok = Storage.ack_last_seen_event(conn, subscription_id, 3, schema: schema) + assert :ok = Storage.ack_last_seen_event(conn, token, 3, schema: schema) assert {:ok, %Storage.Subscription{last_seen: 3}} = read_subscription(context) end @@ -163,7 +175,9 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) - assert :ok = Storage.ack_last_seen_event(conn, subscription_id, 0, schema: schema) + token = claim_token(context, subscription_id) + + assert :ok = Storage.ack_last_seen_event(conn, token, 0, schema: schema) # Zero is a position that was acknowledged, where null is a subscription that never has. assert {:ok, %Storage.Subscription{last_seen: 0}} = read_subscription(context) @@ -174,25 +188,26 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do {:ok, %Storage.Subscription{} = before} = subscribe_to_stream(context) - :ok = Storage.ack_last_seen_event(conn, before.subscription_id, 3, schema: schema) + token = claim_token(context, before.subscription_id) + + :ok = Storage.ack_last_seen_event(conn, token, 3, schema: schema) assert {:ok, %Storage.Subscription{} = after_ack} = read_subscription(context) assert after_ack == %Storage.Subscription{before | last_seen: 3} end - test "ack last seen event does not guard against a position going backwards", context do + test "ack last seen event late does not roll the position back", context do %{conn: conn, schema: schema} = context {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) - :ok = Storage.ack_last_seen_event(conn, subscription_id, 5, schema: schema) - :ok = Storage.ack_last_seen_event(conn, subscription_id, 2, schema: schema) + token = claim_token(context, subscription_id) + + :ok = Storage.ack_last_seen_event(conn, token, 10, schema: schema) + :ok = Storage.ack_last_seen_event(conn, token, 5, schema: schema) - # Storage takes the position it is given. Nothing reaches here out of order, because the one - # process that owns a row acknowledges in order, and refusing the write would be indistinguish - # able from the row being gone. - assert {:ok, %Storage.Subscription{last_seen: 2}} = read_subscription(context) + assert {:ok, %Storage.Subscription{last_seen: 10}} = read_subscription(context) end defp read_subscription(context) do @@ -207,7 +222,16 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do {:ok, %Storage.Subscription{subscription_id: subscription_id}} = Storage.Subscription.subscription(conn, @all_stream, @subscription_name, schema: schema) - Storage.ack_last_seen_event(conn, subscription_id, last_seen, schema: schema) + token = claim_token(context, subscription_id) + + Storage.ack_last_seen_event(conn, token, last_seen, schema: schema) + end + + defp claim_token(%{conn: conn, schema: schema}, subscription_id) do + {:ok, %Lock.Lease{token: token}} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + token end defp subscribe_to_stream(context) do diff --git a/test/subscriptions/subscription_churn_property_test.exs b/test/subscriptions/subscription_churn_property_test.exs new file mode 100644 index 0000000..c45f81b --- /dev/null +++ b/test/subscriptions/subscription_churn_property_test.exs @@ -0,0 +1,332 @@ +defmodule EventStore.Subscriptions.SubscriptionChurnPropertyTest do + use EventStore.StorageCase + use ExUnitProperties + + alias EventStore.{EventFactory, UUID, Wait} + alias EventStore.Subscriptions.Subscription + + @moduletag :slow + @moduletag timeout: 600_000 + + @nodes [1, 2, 3] + + defmodule Subscriber do + @moduledoc false + + use GenServer + + def start(), do: GenServer.start(__MODULE__, nil) + + def seen(subscriber), do: GenServer.call(subscriber, :seen) + def subscription(subscriber), do: GenServer.call(subscriber, :subscription) + + def ack(subscriber, event_number), + do: GenServer.call(subscriber, {:ack, event_number}, 20_000) + + @impl GenServer + def init(_), do: {:ok, %{subscription: nil, seen: MapSet.new()}} + + @impl GenServer + def handle_info({:subscribed, subscription}, state), + do: {:noreply, %{state | subscription: subscription}} + + def handle_info({:events, events}, state) do + seen = Enum.reduce(events, state.seen, &MapSet.put(&2, &1.event_number)) + + {:noreply, %{state | seen: seen}} + end + + @impl GenServer + def handle_call(:seen, _from, state), + do: {:reply, state.seen |> MapSet.to_list() |> Enum.sort(:desc), state} + + def handle_call(:subscription, _from, state), do: {:reply, state.subscription, state} + + def handle_call({:ack, _event_number}, _from, %{subscription: nil} = state), + do: {:reply, {:error, :not_subscribed}, state} + + def handle_call({:ack, event_number}, _from, state), + do: {:reply, Subscription.ack(state.subscription, event_number, self()), state} + end + + defmodule Model do + @moduledoc false + + defstruct subscription_name: nil, + subscribers: %{}, + generation: 0, + last_seen: 0, + legit_max: 0 + end + + property "progress only follows the owner of the current generation across node churn", + %{conn: conn, schema: schema} do + check all( + commands <- resize(list_of(command(), max_length: 10), 10), + max_runs: 10, + max_shrinking_steps: 20 + ) do + model = %Model{subscription_name: UUID.uuid4()} + + try do + model = + Enum.reduce(commands, model, fn command, model -> + model + |> apply_command(command, conn, schema) + |> check_invariants(model, conn, schema) + end) + + finish(model, conn, schema) + after + for node <- @nodes, do: stop_node(%Model{}, node) + :timer.sleep(100) + end + end + end + + defp command do + node = member_of(@nodes) + + one_of([ + tuple({constant(:start), node}), + tuple({constant(:stop), node}), + tuple({constant(:kill_lock_session), node}), + tuple({constant(:append), integer(1..5)}), + tuple({constant(:ack), node, integer(1..5)}), + constant(:settle) + ]) + end + + defp store_name(node), do: :"churn_node_#{node}" + + defp apply_command(%Model{} = model, {:start, node}, _conn, schema) do + if Map.has_key?(model.subscribers, node), do: model, else: start_node(model, node, schema) + end + + defp apply_command(%Model{} = model, {:stop, node}, _conn, _schema), do: stop_node(model, node) + + defp apply_command(%Model{} = model, {:kill_lock_session, node}, conn, _schema) do + with true <- Map.has_key?(model.subscribers, node), + {:ok, backend_pid} <- lock_session_backend_pid(node) do + Postgrex.query!(conn, "SELECT pg_terminate_backend($1)", [backend_pid]) + end + + model + end + + defp apply_command(%Model{} = model, {:append, count}, _conn, _schema) do + append(count) + + model + end + + defp apply_command(%Model{} = model, {:ack, node, k}, conn, schema) do + case Map.fetch(model.subscribers, node) do + {:ok, subscriber} -> + case Enum.at(Subscriber.seen(subscriber), k - 1) do + nil -> model + event_number -> ack(model, node, event_number, conn, schema) + end + + :error -> + model + end + end + + defp apply_command(%Model{} = model, :settle, conn, schema), do: settle(model, conn, schema) + + # A node whose lock session was already killed has no lock connection to + # ask until its pool comes back, so a repeated kill has nothing to do. + defp lock_session_backend_pid(node) do + lock_pool = Module.concat([store_name(node), EventStore.AdvisoryLocks, Postgrex]) + + case Postgrex.query(lock_pool, "SELECT pg_backend_pid()", []) do + {:ok, %Postgrex.Result{rows: [[backend_pid]]}} -> {:ok, backend_pid} + {:error, _reason} -> :error + end + catch + :exit, _reason -> :error + end + + defp start_node(%Model{} = model, node, schema) do + start_supervised!({TestEventStore, name: store_name(node), schema: schema}, + id: store_name(node) + ) + + {:ok, subscriber} = Subscriber.start() + + {:ok, _subscription} = + TestEventStore.subscribe_to_all_streams(model.subscription_name, subscriber, + name: store_name(node), + buffer_size: 10 + ) + + %Model{model | subscribers: Map.put(model.subscribers, node, subscriber)} + end + + defp stop_node(%Model{} = model, node) do + _ = stop_supervised(store_name(node)) + + case Map.pop(model.subscribers, node) do + {nil, _subscribers} -> + model + + {subscriber, subscribers} -> + Process.exit(subscriber, :kill) + %Model{model | subscribers: subscribers} + end + end + + defp append(count) do + :ok = TestEventStore.append_to_stream(UUID.uuid4(), 0, EventFactory.create_events(count)) + end + + # An ack counts as made by the owner when the acking node holds the + # generation the database holds around the time it acks. A takeover that + # completes while the ack is in flight is read as the owner's ack rather + # than failing the run. + defp ack(%Model{} = model, node, event_number, conn, schema) do + subscriber = Map.fetch!(model.subscribers, node) + + legit_before? = holds_current_generation?(subscriber, model, conn, schema) + + _ = Subscriber.ack(subscriber, event_number) + + if legit_before? or holds_current_generation?(subscriber, model, conn, schema), + do: %Model{model | legit_max: max(model.legit_max, event_number)}, + else: model + end + + defp holds_current_generation?(subscriber, %Model{} = model, conn, schema) do + case subscription_row(conn, schema, model.subscription_name) do + {generation, _last_seen} -> token_generation(subscriber) == generation + nil -> false + end + end + + defp token_generation(subscriber) do + with subscription when is_pid(subscription) <- Subscriber.subscription(subscriber), + %Subscription{subscription: %{data: %{fencing_token: token}}} + when token != nil <- :sys.get_state(subscription, 5_000) do + token.generation + else + _ -> nil + end + catch + :exit, _ -> nil + end + + defp subscription_state(subscriber) do + with subscription when is_pid(subscription) <- Subscriber.subscription(subscriber), + %Subscription{subscription: %{state: state}} <- :sys.get_state(subscription, 5_000) do + state + else + _ -> nil + end + catch + :exit, _ -> nil + end + + defp settle(%Model{subscribers: subscribers} = model, _conn, _schema) + when map_size(subscribers) == 0, + do: model + + defp settle(%Model{} = model, conn, schema) do + max_event = max_event_number(conn) + + Wait.until(15_000, fn -> + for {node, subscriber} <- model.subscribers, do: ack_all_seen(model, node, subscriber) + + owners = + for {_node, subscriber} <- model.subscribers, + subscription_state(subscriber) == :subscribed, + do: subscriber + + assert [_owner] = owners + + {_generation, last_seen} = subscription_row(conn, schema, model.subscription_name) + assert last_seen >= max_event + end) + + {generation, _last_seen} = subscription_row(conn, schema, model.subscription_name) + + owner_seen = + for {_node, subscriber} <- model.subscribers, + token_generation(subscriber) == generation, + event_number <- Subscriber.seen(subscriber), + do: event_number + + %Model{model | legit_max: Enum.max([model.legit_max | owner_seen])} + end + + defp ack_all_seen(_model, _node, subscriber) do + for event_number <- subscriber |> Subscriber.seen() |> Enum.reverse() do + Subscriber.ack(subscriber, event_number) + end + end + + defp check_invariants(%Model{} = model, previous, conn, schema) do + case subscription_row(conn, schema, model.subscription_name) do + nil -> + model + + {generation, last_seen} -> + assert generation >= previous.generation, + "lock_generation went from #{previous.generation} to #{generation}" + + assert last_seen >= previous.last_seen, + "last_seen went from #{previous.last_seen} to #{last_seen}" + + assert last_seen <= model.legit_max, + "last_seen #{last_seen} exceeds the highest event acked by a holder of the " <> + "current generation (#{model.legit_max})" + + %Model{model | generation: generation, last_seen: last_seen} + end + end + + defp finish(%Model{} = model, conn, schema) do + model = Enum.reduce(@nodes, model, &apply_command(&2, {:start, &1}, conn, schema)) + model = settle(model, conn, schema) + + append(3) + final = max_event_number(conn) + + Wait.until(15_000, fn -> + for {node, subscriber} <- model.subscribers do + ack_all_seen(model, node, subscriber) + end + + {_generation, last_seen} = subscription_row(conn, schema, model.subscription_name) + assert last_seen == final + end) + + delivered = + for {node, subscriber} <- model.subscribers, + Enum.any?(Subscriber.seen(subscriber), &(&1 > final - 3)), + do: node + + assert [_one_node] = delivered + end + + defp max_event_number(conn) do + %Postgrex.Result{rows: [[max]]} = + Postgrex.query!(conn, "SELECT stream_version FROM streams WHERE stream_uuid = '$all'", []) + + max + end + + defp subscription_row(conn, schema, subscription_name) do + %Postgrex.Result{rows: rows} = + Postgrex.query!( + conn, + "SELECT lock_generation, COALESCE(last_seen, 0) FROM \"#{schema}\".subscriptions WHERE subscription_name = $1", + [subscription_name] + ) + + case rows do + [[generation, last_seen]] -> {generation, last_seen} + [] -> nil + end + end +end diff --git a/test/subscriptions/subscription_fencing_test.exs b/test/subscriptions/subscription_fencing_test.exs new file mode 100644 index 0000000..68d5000 --- /dev/null +++ b/test/subscriptions/subscription_fencing_test.exs @@ -0,0 +1,442 @@ +defmodule EventStore.Subscriptions.SubscriptionFencingTest do + use EventStore.StorageCase + + alias EventStore.{EventFactory, TcpProxy, UUID, Wait} + alias EventStore.Subscriptions.{FencingToken, Subscription} + + @store_a :fenced_store_a + @store_b :fenced_store_b + + setup %{config: config, schema: schema} do + proxy = + start_supervised!( + {TcpProxy, {Keyword.fetch!(config, :hostname), Keyword.get(config, :port, 5432)}} + ) + + start_supervised!( + {TestEventStore, + name: @store_a, + schema: schema, + session_mode_pool: [port: TcpProxy.port(proxy), timeout: 60_000, idle_interval: 100]}, + id: @store_a + ) + + start_supervised!({TestEventStore, name: @store_b, schema: schema}, id: @store_b) + + [proxy: proxy, subscription_name: UUID.uuid4()] + end + + test "a subscription that lost its lock does not overwrite the progress of the new owner", + %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do + {:ok, subscription_a} = subscribe_a(subscription_name, self()) + + fence_a(subscription_a, subscription_name, conn, schema, proxy) + + :ok = TestEventStore.ack(subscription_a, 7) + + assert 5 == last_seen(conn, schema, subscription_name) + + Wait.until(fn -> assert :disconnected == subscription_state(subscription_a) end) + + assert 5 == last_seen(conn, schema, subscription_name) + + refute_receive {:events, _events}, 500 + end + + test "a fenced subscription resumes from the committed position when it takes the lock back", + %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do + {:ok, subscription_a} = subscribe_a(subscription_name, self()) + + fence_a(subscription_a, subscription_name, conn, schema, proxy) + + :ok = TestEventStore.ack(subscription_a, 7) + + Wait.until(fn -> assert :disconnected == subscription_state(subscription_a) end) + + assert 5 == last_seen(conn, schema, subscription_name) + + stop_supervised!(@store_b) + :ok = TcpProxy.forward(proxy) + + assert_receive {:events, [%{event_number: 6} | _]}, 5_000 + + assert %FencingToken{generation: 3} = fencing_token(subscription_a) + assert 3 == lock_generation(conn, schema, subscription_name) + + :ok = TestEventStore.ack(subscription_a, 7) + + Wait.until(fn -> assert 7 == last_seen(conn, schema, subscription_name) end) + end + + test "a fenced subscription that keeps retrying does not disturb the new owner", + %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do + {:ok, subscription_a} = subscribe_a(subscription_name, self()) + + {subscription_b, subscriber_b} = + fence_a(subscription_a, subscription_name, conn, schema, proxy) + + :ok = TestEventStore.ack(subscription_a, 7) + + Wait.until(fn -> assert :disconnected == subscription_state(subscription_a) end) + + :timer.sleep(3_500) + + # While its lock pool is unreachable the subscription may sit inside a + # bounded lock claim, so only the database is probed here. + assert 2 == lock_generation(conn, schema, subscription_name) + assert 5 == last_seen(conn, schema, subscription_name) + refute_received {:events, _events} + + :ok = TcpProxy.forward(proxy) + + :timer.sleep(3_500) + + Wait.until(10_000, fn -> assert :disconnected == subscription_state(subscription_a) end) + assert 2 == lock_generation(conn, schema, subscription_name) + assert 5 == last_seen(conn, schema, subscription_name) + refute_received {:events, _events} + + send(subscriber_b, {:ack, subscription_b, 6}) + + Wait.until(fn -> assert 6 == last_seen(conn, schema, subscription_name) end) + + assert :subscribed == subscription_state(subscription_b) + end + + test "a subscription fenced while catching up disconnects and the new owner resumes from the committed position", + %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do + for _ <- 1..4, do: append_to_a(50) + + {:ok, subscription_a} = subscribe_a(subscription_name, self(), buffer_size: 1) + + assert_receive {:subscribed, ^subscription_a} + + for event_number <- 1..3 do + assert_receive {:events, [%{event_number: ^event_number}]}, 5_000 + + :ok = TestEventStore.ack(subscription_a, event_number) + end + + assert_receive {:events, [%{event_number: 4}]}, 5_000 + + Wait.until(fn -> assert 3 == last_seen(conn, schema, subscription_name) end) + + refute :subscribed == subscription_state(subscription_a) + + monitor = Process.monitor(subscription_a) + + {subscription_b, subscriber_b} = take_over(subscription_name, conn, schema, proxy) + + :ok = TestEventStore.ack(subscription_a, 4) + + Wait.until(fn -> assert :disconnected == subscription_state(subscription_a) end) + + assert 5 == last_seen(conn, schema, subscription_name) + + flush_events() + + for event_number <- 6..12 do + send(subscriber_b, {:ack, subscription_b, event_number}) + end + + assert_receive_b_through(15) + + refute_receive {:DOWN, ^monitor, :process, _pid, _reason}, 500 + refute_receive {:events, _events}, 500 + + Wait.until(fn -> assert 12 == last_seen(conn, schema, subscription_name) end) + end + + defp assert_receive_b_through(last_event_number) do + assert_receive {:b, :events, events}, 5_000 + + unless List.last(events).event_number >= last_event_number do + assert_receive_b_through(last_event_number) + end + end + + defp subscribe_a(subscription_name, subscriber, opts \\ []) do + TestEventStore.subscribe_to_all_streams( + subscription_name, + subscriber, + Keyword.merge([name: @store_a, buffer_size: 10], opts) + ) + end + + defp fence_a(subscription_a, subscription_name, conn, schema, proxy) do + assert_receive {:subscribed, ^subscription_a} + + append_to_a(7) + + assert_receive_through(7) + + :ok = TestEventStore.ack(subscription_a, 3) + + Wait.until(fn -> assert 3 == last_seen(conn, schema, subscription_name) end) + + take_over(subscription_name, conn, schema, proxy) + end + + defp append_to_a(count) do + :ok = + TestEventStore.append_to_stream(UUID.uuid4(), 0, EventFactory.create_events(count), + name: @store_a + ) + end + + # Takes A's lock session away and has B own the subscription at event 5. + defp take_over(subscription_name, conn, schema, proxy) do + end_lock_session(conn, proxy) + + subscriber_b = start_forwarding_subscriber(:b) + + {:ok, subscription_b} = + TestEventStore.subscribe_to_all_streams(subscription_name, subscriber_b, + name: @store_b, + buffer_size: 10 + ) + + assert_receive {:b, :subscribed, ^subscription_b}, 5_000 + assert_receive {:b, :events, [%{event_number: 4} | _]}, 5_000 + + send(subscriber_b, {:ack, subscription_b, 5}) + + Wait.until(fn -> assert 5 == last_seen(conn, schema, subscription_name) end) + + flush_events() + + {subscription_b, subscriber_b} + end + + defp end_lock_session(conn, proxy) do + lock_pool = Module.concat([@store_a, EventStore.AdvisoryLocks, Postgrex]) + + %Postgrex.Result{rows: [[backend_pid]]} = + Postgrex.query!(lock_pool, "SELECT pg_backend_pid()", []) + + :ok = TcpProxy.blackhole(proxy) + + %Postgrex.Result{rows: [[true]]} = + Postgrex.query!(conn, "SELECT pg_terminate_backend($1)", [backend_pid]) + + Wait.until(fn -> + assert %Postgrex.Result{rows: [[0]]} = + Postgrex.query!( + conn, + "SELECT count(*)::int FROM pg_stat_activity WHERE pid = $1", + [backend_pid] + ) + end) + end + + defp take_over_all(tagged_names, conn, schema, proxy) do + end_lock_session(conn, proxy) + + owners = + for {tag, subscription_name} <- tagged_names do + subscriber = start_forwarding_subscriber(tag) + + {:ok, subscription} = + TestEventStore.subscribe_to_all_streams(subscription_name, subscriber, + name: @store_b, + buffer_size: 10 + ) + + assert_receive {^tag, :subscribed, ^subscription}, 5_000 + assert_receive {^tag, :events, [%{event_number: 4} | _]}, 5_000 + + send(subscriber, {:ack, subscription, 5}) + + Wait.until(fn -> assert 5 == last_seen(conn, schema, subscription_name) end) + + {tag, subscription} + end + + flush_tagged_events([:a1, :a2, :b1, :b2]) + + owners + end + + defp flush_tagged_events(tags), do: Enum.each(tags, &flush_tagged/1) + + defp flush_tagged(tag) do + receive do + {^tag, :events, _events} -> flush_tagged(tag) + after + 0 -> :ok + end + end + + test "every subscription on a node whose lock session ends is fenced and recovers", + %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do + other_name = UUID.uuid4() + subscriber_1 = start_forwarding_subscriber(:a1) + subscriber_2 = start_forwarding_subscriber(:a2) + + {:ok, subscription_1} = subscribe_a(subscription_name, subscriber_1) + {:ok, subscription_2} = subscribe_a(other_name, subscriber_2) + + assert_receive {:a1, :subscribed, ^subscription_1} + assert_receive {:a2, :subscribed, ^subscription_2} + + append_to_a(7) + + for tag <- [:a1, :a2], do: assert_receive_tagged_through(tag, 7) + + send(subscriber_1, {:ack, subscription_1, 3}) + send(subscriber_2, {:ack, subscription_2, 3}) + + for name <- [subscription_name, other_name] do + Wait.until(fn -> assert 3 == last_seen(conn, schema, name) end) + end + + take_over_all([b1: subscription_name, b2: other_name], conn, schema, proxy) + + send(subscriber_1, {:ack, subscription_1, 7}) + send(subscriber_2, {:ack, subscription_2, 7}) + + for subscription <- [subscription_1, subscription_2] do + Wait.until(fn -> assert :disconnected == subscription_state(subscription) end) + end + + for name <- [subscription_name, other_name] do + assert 5 == last_seen(conn, schema, name) + end + + flush_tagged_events([:a1, :a2, :b1, :b2]) + + stop_supervised!(@store_b) + :ok = TcpProxy.forward(proxy) + + assert_receive {:a1, :events, [%{event_number: 6} | _]}, 5_000 + assert_receive {:a2, :events, [%{event_number: 6} | _]}, 5_000 + + for name <- [subscription_name, other_name] do + assert 3 == lock_generation(conn, schema, name) + end + end + + defp assert_receive_tagged_through(tag, last_event_number) do + assert_receive {^tag, :events, events} + + unless List.last(events).event_number >= last_event_number do + assert_receive_tagged_through(tag, last_event_number) + end + end + + test "every subscriber of a fenced subscription stops receiving events", + %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do + subscriber_1 = start_forwarding_subscriber(:a1) + subscriber_2 = start_forwarding_subscriber(:a2) + + {:ok, subscription_a} = subscribe_a(subscription_name, subscriber_1, concurrency_limit: 2) + {:ok, ^subscription_a} = subscribe_a(subscription_name, subscriber_2, concurrency_limit: 2) + + assert_receive {:a1, :subscribed, ^subscription_a} + assert_receive {:a2, :subscribed, ^subscription_a} + + append_to_a(7) + + owners = receive_owners(%{a1: subscriber_1, a2: subscriber_2}, 7) + + ack_in_order(owners, subscription_a, 1..3) + + Wait.until(fn -> assert 3 == last_seen(conn, schema, subscription_name) end) + + take_over(subscription_name, conn, schema, proxy) + + ack_in_order(owners, subscription_a, 4..7) + + Wait.until(fn -> assert :disconnected == subscription_state(subscription_a) end) + + assert 5 == last_seen(conn, schema, subscription_name) + + refute_receive {:a1, :events, _events}, 500 + refute_receive {:a2, :events, _events}, 0 + end + + defp ack_in_order(owners, subscription, event_numbers) do + for event_number <- event_numbers do + send(Map.fetch!(owners, event_number), {:ack, subscription, event_number}) + end + end + + defp receive_owners(subscribers, last, acc \\ %{}) do + receive do + {tag, :events, events} when tag in [:a1, :a2] -> + acc = + Enum.reduce(events, acc, &Map.put(&2, &1.event_number, Map.fetch!(subscribers, tag))) + + if map_size(acc) >= last, do: acc, else: receive_owners(subscribers, last, acc) + after + 2_000 -> acc + end + end + + defp assert_receive_through(last_event_number) do + assert_receive {:events, events} + + unless List.last(events).event_number >= last_event_number do + assert_receive_through(last_event_number) + end + end + + defp flush_events do + receive do + {:events, _events} -> flush_events() + after + 0 -> :ok + end + end + + defp start_forwarding_subscriber(tag) do + reply_to = self() + + spawn_link(fn -> forward(tag, reply_to) end) + end + + defp forward(tag, reply_to) do + receive do + {:subscribed, subscription} -> send(reply_to, {tag, :subscribed, subscription}) + {:events, events} -> send(reply_to, {tag, :events, events}) + {:ack, subscription, event_number} -> TestEventStore.ack(subscription, event_number) + end + + forward(tag, reply_to) + end + + defp subscription_state(subscription) do + %Subscription{subscription: %{state: state}} = :sys.get_state(subscription) + + state + end + + defp fencing_token(subscription) do + %Subscription{subscription: %{data: %{fencing_token: fencing_token}}} = + :sys.get_state(subscription) + + fencing_token + end + + defp lock_generation(conn, schema, subscription_name) do + %Postgrex.Result{rows: [[generation]]} = + Postgrex.query!( + conn, + "SELECT lock_generation FROM \"#{schema}\".subscriptions WHERE subscription_name = $1", + [subscription_name] + ) + + generation + end + + defp last_seen(conn, schema, subscription_name) do + %Postgrex.Result{rows: [[last_seen]]} = + Postgrex.query!( + conn, + "SELECT last_seen FROM \"#{schema}\".subscriptions WHERE subscription_name = $1", + [subscription_name] + ) + + last_seen + end +end diff --git a/test/subscriptions/subscription_fsm_test.exs b/test/subscriptions/subscription_fsm_test.exs index 4088c1a..93066c4 100644 --- a/test/subscriptions/subscription_fsm_test.exs +++ b/test/subscriptions/subscription_fsm_test.exs @@ -1,7 +1,8 @@ defmodule EventStore.Subscriptions.SubscriptionFsmTest do use EventStore.StorageCase - alias EventStore.{Storage, UUID} + alias EventStore.{EventFactory, Storage, UUID} + alias EventStore.Storage.Lock alias EventStore.Subscriptions.{SubscriptionFsm, SubscriptionState} # The states that own a stored position, and those that do not. @@ -17,10 +18,15 @@ defmodule EventStore.Subscriptions.SubscriptionFsmTest do schema: context.schema ) + fencing_token = claim_token(context, subscription_id) + + on_exit(fn -> :ok = Lock.unlock(context.conn, subscription_id, schema: context.schema) end) + [ stream_uuid: stream_uuid, subscription_name: subscription_name, - subscription_id: subscription_id + subscription_id: subscription_id, + fencing_token: fencing_token ] end @@ -155,6 +161,31 @@ defmodule EventStore.Subscriptions.SubscriptionFsmTest do assert_received {:checkpoint_failed, :subscription_not_found} end + test "tells the subscription it is fenced when the probe that classifies the rejection fails", + context do + %{conn: conn, schema: schema, subscription_id: subscription_id} = context + + probe_conn = start_supervised!({Postgrex, context.postgrex_config}, id: :probe_conn) + lock_ref = make_ref() + + fsm = fsm(context, checkpoints_pending: 1, last_ack: 3, conn: probe_conn) + fsm = %SubscriptionFsm{fsm | data: %SubscriptionState{fsm.data | lock_ref: lock_ref}} + + {:ok, _result} = + Postgrex.query( + conn, + ~s|UPDATE "#{schema}".subscriptions SET lock_generation = lock_generation + 1 WHERE subscription_id = $1|, + [subscription_id] + ) + + drop_privileges_after_update(context) + + assert %SubscriptionFsm{data: %SubscriptionState{checkpoints_pending: 0}} = + SubscriptionFsm.checkpoint(fsm) + + assert_received {:fenced, ^lock_ref} + end + test "does not write the position onto a subscription that reused the name", context do %{conn: conn, schema: schema, stream_uuid: stream_uuid} = context %{subscription_name: subscription_name, subscription_id: subscription_id} = context @@ -305,26 +336,128 @@ defmodule EventStore.Subscriptions.SubscriptionFsmTest do end end + describe "subscribe/1 after another owner advanced the stored position" do + test "catches up to the stored position and subscribes instead of spinning on empty reads", + context do + reclaim_after_other_owner_acked(context, appended: 15, stored: 15) + + fsm = fsm(context, state: :disconnected, last_received: 10, last_sent: 5, last_ack: 5) + + assert %SubscriptionFsm{state: :request_catch_up, data: data} = + SubscriptionFsm.subscribe(fsm) + + assert data.last_ack == 15 + assert data.last_sent == 15 + + assert %SubscriptionFsm{state: :subscribed, data: data} = + SubscriptionFsm.catch_up(%SubscriptionFsm{state: :request_catch_up, data: data}) + + assert data.last_received == 15 + end + + test "does not move last_received backwards when events past the stored position were seen", + context do + reclaim_after_other_owner_acked(context, appended: 20, stored: 15) + + fsm = fsm(context, state: :disconnected, last_received: 20, last_sent: 5, last_ack: 5) + + assert %SubscriptionFsm{state: :request_catch_up, data: data} = + SubscriptionFsm.subscribe(fsm) + + assert data.last_received == 20 + + assert %SubscriptionFsm{state: :catching_up, data: data} = + SubscriptionFsm.catch_up(%SubscriptionFsm{state: :request_catch_up, data: data}) + + assert data.queue_size == 5 + assert data.last_received == 20 + end + end + + defp reclaim_after_other_owner_acked(context, appended: appended, stored: stored) do + %{conn: conn, schema: schema, subscription_id: subscription_id} = context + + :ok = + TestEventStore.append_to_stream( + context.stream_uuid, + 0, + EventFactory.create_events(appended) + ) + + {:ok, _result} = + Postgrex.query( + conn, + ~s|UPDATE "#{schema}".subscriptions SET last_seen = $2 WHERE subscription_id = $1|, + [subscription_id, stored] + ) + + :ok = Lock.unlock(conn, subscription_id, schema: schema) + end + + defp claim_token(%{conn: conn, schema: schema}, subscription_id) do + {:ok, %Lock.Lease{token: token}} = + Lock.claim_subscription_lock(conn, subscription_id, schema: schema) + + token + end + defp fsm(context, overrides \\ []) do %{conn: conn, schema: schema} = context %{stream_uuid: stream_uuid, subscription_name: subscription_name} = context %{subscription_id: subscription_id} = context data = %SubscriptionState{ - conn: conn, + conn: Keyword.get(overrides, :conn, conn), schema: Keyword.get(overrides, :schema, schema), stream_uuid: stream_uuid, subscription_name: subscription_name, subscription_id: subscription_id, + fencing_token: context.fencing_token, last_ack: Keyword.get(overrides, :last_ack, 0), checkpoints_pending: Keyword.get(overrides, :checkpoints_pending, 0), query_timeout: 15_000, + event_store: TestEventStore, + serializer: context.serializer, + correlation_id_type: context.correlation_id_type, + causation_id_type: context.causation_id_type, + max_size: 1_000, + last_received: Keyword.get(overrides, :last_received, 0), + last_sent: Keyword.get(overrides, :last_sent, 0), transient: Keyword.get(overrides, :transient, false) } %SubscriptionFsm{state: Keyword.get(overrides, :state, :subscribed), data: data} end + defp drop_privileges_after_update(%{conn: conn, schema: schema}) do + {:ok, _result} = + Postgrex.query( + conn, + """ + CREATE FUNCTION fsm_test_drop_privileges() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN PERFORM set_config('role', 'pg_monitor', false); RETURN NULL; END $$ + """, + [] + ) + + {:ok, _result} = + Postgrex.query( + conn, + ~s|CREATE TRIGGER fsm_test_drop_privileges AFTER UPDATE ON "#{schema}".subscriptions FOR EACH STATEMENT EXECUTE FUNCTION fsm_test_drop_privileges()|, + [] + ) + + on_exit(fn -> + Postgrex.query!( + conn, + ~s|DROP TRIGGER fsm_test_drop_privileges ON "#{schema}".subscriptions|, + [] + ) + + Postgrex.query!(conn, "DROP FUNCTION fsm_test_drop_privileges()", []) + end) + end + defp reset_position(context) do %{conn: conn, schema: schema, subscription_id: subscription_id} = context diff --git a/test/subscriptions/subscription_locking_test.exs b/test/subscriptions/subscription_locking_test.exs index 6eb59ae..168aa69 100644 --- a/test/subscriptions/subscription_locking_test.exs +++ b/test/subscriptions/subscription_locking_test.exs @@ -1,10 +1,13 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do use EventStore.StorageCase + import ExUnit.CaptureLog + alias EventStore.{Config, EventFactory, Storage, UUID, Wait} - alias EventStore.Subscriptions.Subscription + alias EventStore.Subscriptions.{FencingToken, Subscription} @conn TestEventStore.Postgrex + @advisory_locks TestEventStore.EventStore.AdvisoryLocks @advisory_locks_conn TestEventStore.EventStore.AdvisoryLocks.Postgrex setup do @@ -81,7 +84,9 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do {:ok, %Storage.Subscription{subscription_id: subscription_id}} = Storage.Subscription.subscription(@conn, "$all", subscription_name, schema: schema) - :ok = Storage.Subscription.ack_last_seen_event(@conn, subscription_id, 2, schema: schema) + token = FencingToken.new(subscription_id, lock_generation(subscription_id, schema)) + + :ok = Storage.Subscription.ack_last_seen_event(@conn, token, 2, schema: schema) :ok = reconnect(subscription) @@ -219,6 +224,117 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do end end + describe "late fenced notice" do + setup [:create_subscription] + + test "should keep running when the lock was already released", %{ + event_store: event_store, + subscription: subscription + } do + assert_receive {:subscribed, ^subscription} + + monitor_ref = Process.monitor(subscription) + + lock_ref = lock_ref(subscription) + + :ok = disconnect(subscription) + :sys.get_state(subscription) + + send(subscription, {:fenced, lock_ref}) + + refute_receive {:DOWN, ^monitor_ref, :process, ^subscription, _reason}, 200 + assert Process.alive?(subscription) + + :ok = reconnect(subscription) + + append_events_to_stream(event_store, 3) + + assert_receive_events([1, 2, 3]) + end + + test "should ignore a notice for a lock that was replaced by a new one", %{ + event_store: event_store, + subscription: subscription + } do + assert_receive {:subscribed, ^subscription} + + monitor_ref = Process.monitor(subscription) + + old_lock_ref = lock_ref(subscription) + + :ok = disconnect(subscription) + :ok = reconnect(subscription) + + new_lock_ref = wait_for_new_lock_ref(subscription, old_lock_ref) + + Wait.until(fn -> assert :subscribed == subscription_state(subscription) end) + + send(subscription, {:fenced, old_lock_ref}) + + refute_receive {:DOWN, ^monitor_ref, :process, ^subscription, _reason}, 200 + assert Process.alive?(subscription) + + assert :subscribed == subscription_state(subscription) + assert new_lock_ref == lock_ref(subscription) + + append_events_to_stream(event_store, 3) + + assert_receive_events([1, 2, 3]) + end + end + + describe "subscription stopped after another owner took over" do + test "should leave the progress of the new owner untouched", %{ + conn: conn, + event_store: event_store, + schema: schema, + subscription_name: subscription_name + } do + {:ok, subscription} = + event_store.subscribe_to_all_streams(subscription_name, self(), + buffer_size: 3, + checkpoint_threshold: 100 + ) + + assert_receive {:subscribed, ^subscription} + + append_events_to_stream(event_store, 3) + + assert_receive_events([1, 2, 3]) + + Postgrex.query!( + conn, + "UPDATE \"#{schema}\".subscriptions SET lock_generation = lock_generation + 1, last_seen = 2 WHERE subscription_name = $1", + [subscription_name] + ) + + :ok = Subscription.ack(subscription, 3) + + monitor_ref = Process.monitor(subscription) + + log = + capture_log(fn -> + assert :ok = GenServer.stop(subscription) + end) + + assert_receive {:DOWN, ^monitor_ref, :process, ^subscription, :normal} + refute log =~ "terminating" + + assert 2 == last_seen(conn, schema, subscription_name) + end + end + + defp last_seen(conn, schema, subscription_name) do + %Postgrex.Result{rows: [[last_seen]]} = + Postgrex.query!( + conn, + "SELECT last_seen FROM \"#{schema}\".subscriptions WHERE subscription_name = $1", + [subscription_name] + ) + + last_seen + end + defp lock_subscription(context) do %{config: config, schema: schema} = context @@ -272,6 +388,17 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do :ok end + defp lock_generation(subscription_id, schema) do + %Postgrex.Result{rows: [[generation]]} = + Postgrex.query!( + @conn, + "SELECT lock_generation FROM \"#{schema}\".subscriptions WHERE subscription_id = $1", + [subscription_id] + ) + + generation + end + defp reconnect(subscription) do Process.send(subscription, :subscribe_to_stream, []) end @@ -286,6 +413,12 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do lock_ref end + defp subscription_state(subscription) do + %Subscription{subscription: %{state: state}} = :sys.get_state(subscription) + + state + end + defp wait_for_new_lock_ref(subscription, old_lock_ref) do Wait.until(2_000, fn -> lock_ref = lock_ref(subscription) @@ -297,6 +430,9 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do end defp release_lock(subscription, lock_ref) do + :ok = EventStore.AdvisoryLocks.release(@advisory_locks, lock_ref) + _state = :sys.get_state(@advisory_locks) + Process.send( subscription, {EventStore.AdvisoryLocks, :lock_released, lock_ref, :shutdown}, diff --git a/test/subscriptions/subscription_unmigrated_schema_test.exs b/test/subscriptions/subscription_unmigrated_schema_test.exs new file mode 100644 index 0000000..c66a85c --- /dev/null +++ b/test/subscriptions/subscription_unmigrated_schema_test.exs @@ -0,0 +1,55 @@ +defmodule EventStore.Subscriptions.SubscriptionUnmigratedSchemaTest do + use EventStore.StorageCase + + alias EventStore.{EventFactory, UUID, Wait} + alias EventStore.Subscriptions.Subscription + + @advisory_locks TestEventStore.EventStore.AdvisoryLocks + @advisory_locks_conn TestEventStore.EventStore.AdvisoryLocks.Postgrex + + test "a database without the lock generation keeps a new node retrying instead of crashing", + %{conn: conn} do + on_exit(fn -> add_lock_generation(conn) end) + + Postgrex.query!(conn, "ALTER TABLE subscriptions DROP COLUMN lock_generation", []) + + advisory_locks = Process.whereis(@advisory_locks) + advisory_locks_conn = Process.whereis(@advisory_locks_conn) + + {:ok, subscription} = + TestEventStore.subscribe_to_all_streams(UUID.uuid4(), self(), buffer_size: 10) + + ref = Process.monitor(subscription) + + refute_receive {:subscribed, _subscription}, 1_500 + refute_receive {:DOWN, ^ref, :process, _pid, _reason} + + assert :initial == subscription_state(subscription) + assert advisory_locks == Process.whereis(@advisory_locks) + assert advisory_locks_conn == Process.whereis(@advisory_locks_conn) + + add_lock_generation(conn) + + assert_receive {:subscribed, ^subscription}, 3_000 + + :ok = TestEventStore.append_to_stream(UUID.uuid4(), 0, EventFactory.create_events(1)) + + assert_receive {:events, [%{event_number: 1}]} + + Wait.until(fn -> assert :subscribed == subscription_state(subscription) end) + end + + defp add_lock_generation(conn) do + Postgrex.query!( + conn, + "ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS lock_generation bigint NOT NULL DEFAULT 0", + [] + ) + end + + defp subscription_state(subscription) do + %Subscription{subscription: %{state: state}} = :sys.get_state(subscription) + + state + end +end diff --git a/test/subscriptions/subscription_zombie_lock_test.exs b/test/subscriptions/subscription_zombie_lock_test.exs new file mode 100644 index 0000000..9ccd841 --- /dev/null +++ b/test/subscriptions/subscription_zombie_lock_test.exs @@ -0,0 +1,175 @@ +defmodule EventStore.Subscriptions.SubscriptionZombieLockTest do + use EventStore.StorageCase + + alias EventStore.{EventFactory, TcpProxy, UUID, Wait} + alias EventStore.Subscriptions.Subscription + + @store_a :zombie_store_a + @store_b :zombie_store_b + + setup %{config: config, schema: schema} do + proxy = + start_supervised!( + {TcpProxy, {Keyword.fetch!(config, :hostname), Keyword.get(config, :port, 5432)}} + ) + + start_supervised!( + {TestEventStore, + name: @store_a, + schema: schema, + port: TcpProxy.port(proxy), + session_mode_pool: [timeout: 60_000]}, + id: @store_a + ) + + start_supervised!({TestEventStore, name: @store_b, schema: schema}, id: @store_b) + + [proxy: proxy, subscription_name: UUID.uuid4()] + end + + test "a node that dies dirty cannot move the progress of the owner that replaced it", + %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do + {:ok, subscription_a} = + TestEventStore.subscribe_to_all_streams(subscription_name, self(), + name: @store_a, + buffer_size: 10 + ) + + assert_receive {:subscribed, ^subscription_a} + + :ok = + TestEventStore.append_to_stream(UUID.uuid4(), 0, EventFactory.create_events(7), + name: @store_a + ) + + assert_receive_through(7) + + :ok = TestEventStore.ack(subscription_a, 3) + + Wait.until(fn -> assert 3 == last_seen(conn, schema, subscription_name) end) + assert 1 == lock_generation(conn, schema, subscription_name) + + :ok = TcpProxy.freeze(proxy) + + test_pid = self() + + pending_ack = + Task.async(fn -> + try do + Subscription.ack(subscription_a, 7, test_pid) + catch + :exit, reason -> {:exit, reason} + end + end) + + :timer.sleep(300) + + subscriber_b = start_forwarding_subscriber(:b) + + {:ok, subscription_b} = + TestEventStore.subscribe_to_all_streams(subscription_name, subscriber_b, + name: @store_b, + buffer_size: 10 + ) + + refute_receive {:b, :subscribed, _subscription}, 2_500 + assert 1 == lock_generation(conn, schema, subscription_name) + + terminate_lock_holder(conn) + + assert_receive {:b, :subscribed, ^subscription_b}, 5_000 + assert_receive {:b, :events, [%{event_number: 4} | _]}, 5_000 + + assert 2 == lock_generation(conn, schema, subscription_name) + + send(subscriber_b, {:ack, subscription_b, 5}) + + Wait.until(fn -> assert 5 == last_seen(conn, schema, subscription_name) end) + + :ok = TcpProxy.forward(proxy) + + Task.await(pending_ack, 20_000) + + assert 5 == last_seen(conn, schema, subscription_name) + assert 2 == lock_generation(conn, schema, subscription_name) + + Wait.until(10_000, fn -> assert :disconnected == subscription_state(subscription_a) end) + + assert 5 == last_seen(conn, schema, subscription_name) + + flush_events() + refute_receive {:events, _events}, 1_500 + + send(subscriber_b, {:ack, subscription_b, 6}) + + Wait.until(fn -> assert 6 == last_seen(conn, schema, subscription_name) end) + assert 2 == lock_generation(conn, schema, subscription_name) + end + + defp terminate_lock_holder(conn) do + %Postgrex.Result{rows: [[backend_pid]]} = + Postgrex.query!( + conn, + "SELECT pid FROM pg_locks WHERE locktype = 'advisory' AND granted", + [] + ) + + %Postgrex.Result{rows: [[true]]} = + Postgrex.query!(conn, "SELECT pg_terminate_backend($1)", [backend_pid]) + end + + defp assert_receive_through(last_event_number) do + assert_receive {:events, events} + + unless List.last(events).event_number >= last_event_number do + assert_receive_through(last_event_number) + end + end + + defp flush_events do + receive do + {:events, _events} -> flush_events() + after + 0 -> :ok + end + end + + defp start_forwarding_subscriber(tag) do + reply_to = self() + + spawn_link(fn -> forward(tag, reply_to) end) + end + + defp forward(tag, reply_to) do + receive do + {:subscribed, subscription} -> send(reply_to, {tag, :subscribed, subscription}) + {:events, events} -> send(reply_to, {tag, :events, events}) + {:ack, subscription, event_number} -> TestEventStore.ack(subscription, event_number) + end + + forward(tag, reply_to) + end + + defp subscription_state(subscription) do + %Subscription{subscription: %{state: state}} = :sys.get_state(subscription) + + state + end + + defp lock_generation(conn, schema, subscription_name), + do: subscription_column(conn, schema, "lock_generation", subscription_name) + + defp last_seen(conn, schema, subscription_name), + do: subscription_column(conn, schema, "last_seen", subscription_name) + + defp subscription_column(conn, schema, column, subscription_name) do + %Postgrex.Result{rows: [[value]]} = + Postgrex.query!( + conn, + "SELECT #{column} FROM \"#{schema}\".subscriptions WHERE subscription_name = $1", + [subscription_name] + ) + + value + end +end diff --git a/test/support/tcp_proxy.ex b/test/support/tcp_proxy.ex new file mode 100644 index 0000000..1439228 --- /dev/null +++ b/test/support/tcp_proxy.ex @@ -0,0 +1,181 @@ +defmodule EventStore.TcpProxy do + @moduledoc false + + use GenServer + + @grace 500 + + def start_link(upstream), do: GenServer.start_link(__MODULE__, upstream) + + def port(proxy), do: GenServer.call(proxy, :port) + + @doc """ + Keeps the connections already piped and holds every new connection open + without ever answering it. + """ + def blackhole(proxy), do: GenServer.call(proxy, :blackhole) + + @doc """ + Pipes new connections again and closes the ones held open by `blackhole/1` + so their clients fail fast and reconnect. + """ + def forward(proxy), do: GenServer.call(proxy, :forward) + + @doc """ + Parks every byte and every close on the connections already piped without + closing any socket, and holds new connections like `blackhole/1`. Both ends + keep believing the connection is alive. `forward/1` delivers what was parked + and keeps relaying briefly before resetting those connections, the way late + packets still arrive when a partition heals. + """ + def freeze(proxy), do: GenServer.call(proxy, :freeze) + + def stop(proxy), do: GenServer.stop(proxy) + + @impl GenServer + def init({host, port}) do + {:ok, listen} = + :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true, ip: {127, 0, 0, 1}]) + + {:ok, listen_port} = :inet.port(listen) + proxy = self() + + acceptor = + spawn_link(fn -> + Process.put(:proxy, proxy) + accept(listen, proxy, to_charlist(host), port) + end) + + {:ok, + %{ + acceptor: acceptor, + listen: listen, + port: listen_port, + mode: :forward, + held: [], + relays: [] + }} + end + + @impl GenServer + def handle_call(:port, _from, state), do: {:reply, state.port, state} + def handle_call(:mode, _from, state), do: {:reply, state.mode, state} + def handle_call(:blackhole, _from, state), do: {:reply, :ok, %{state | mode: :blackhole}} + + def handle_call({:hold, client}, _from, state), + do: {:reply, :ok, %{state | held: [client | state.held]}} + + def handle_call(:freeze, _from, state), do: {:reply, :ok, %{state | mode: :frozen}} + + def handle_call({:register, relay}, _from, state), + do: {:reply, :ok, %{state | relays: [relay | state.relays]}} + + def handle_call(:forward, _from, state) do + Enum.each(state.held, &:gen_tcp.close/1) + + if state.mode == :frozen, do: Enum.each(state.relays, &send(&1, :release)) + + {:reply, :ok, %{state | mode: :forward, held: [], relays: []}} + end + + @impl GenServer + def terminate(_reason, %{acceptor: acceptor, listen: listen}) do + Process.unlink(acceptor) + Process.exit(acceptor, :kill) + :gen_tcp.close(listen) + end + + defp accept(listen, proxy, host, port) do + case :gen_tcp.accept(listen) do + {:ok, client} -> + case GenServer.call(proxy, :mode) do + :forward -> pipe(client, host, port) + mode when mode in [:blackhole, :frozen] -> hold(client, proxy) + end + + accept(listen, proxy, host, port) + + {:error, _reason} -> + :ok + end + end + + defp hold(client, proxy) do + :ok = :gen_tcp.controlling_process(client, proxy) + :ok = GenServer.call(proxy, {:hold, client}) + end + + defp pipe(client, host, port) do + {:ok, upstream} = :gen_tcp.connect(host, port, [:binary, active: false]) + + proxy = self_proxy() + + for {from, to} <- [{client, upstream}, {upstream, client}] do + pid = spawn_link(fn -> receive do: (:go -> relay(from, to, proxy, :infinity)) end) + :ok = :gen_tcp.controlling_process(from, pid) + :ok = GenServer.call(proxy, {:register, pid}) + send(pid, :go) + end + end + + defp self_proxy, do: Process.get(:proxy) + + defp relay(from, to, proxy, deadline) do + receive do + :release -> release(from, to, []) + after + 0 -> relay_once(from, to, proxy, deadline) + end + end + + defp relay_once(from, to, proxy, deadline) do + if expired?(deadline) do + close(from, to) + else + case :gen_tcp.recv(from, 0, 50) do + {:error, :timeout} -> + relay(from, to, proxy, deadline) + + {:ok, data} -> + if deadline == :infinity and GenServer.call(proxy, :mode) == :frozen do + park(from, to, [data]) + else + forward_data(from, to, proxy, deadline, data) + end + + {:error, _reason} -> + if deadline == :infinity and GenServer.call(proxy, :mode) == :frozen do + park(from, to, []) + else + close(from, to) + end + end + end + end + + defp expired?(:infinity), do: false + defp expired?(deadline), do: System.monotonic_time(:millisecond) > deadline + + defp forward_data(from, to, proxy, deadline, data) do + case :gen_tcp.send(to, data) do + :ok -> relay(from, to, proxy, deadline) + {:error, _reason} -> close(from, to) + end + end + + defp park(from, to, parked) do + receive do + :release -> release(from, to, parked) + end + end + + defp release(from, to, parked) do + Enum.each(Enum.reverse(parked), &:gen_tcp.send(to, &1)) + relay(from, to, nil, System.monotonic_time(:millisecond) + @grace) + end + + defp close(from, to) do + :gen_tcp.close(from) + :gen_tcp.close(to) + end +end From d6de2e21428d14785fb5d478a55f02157556a15d Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 16:41:16 -0400 Subject: [PATCH 09/13] chore(guides): describe what the subscription lock guarantees once progress is fenced (#31) Signed-off-by: Yordis Prieto --- guides/Architecture.md | 2 ++ guides/Cluster.md | 4 ++++ guides/Upgrades.md | 4 ++++ 3 files changed, 10 insertions(+) diff --git a/guides/Architecture.md b/guides/Architecture.md index 4982463..d4fda6a 100644 --- a/guides/Architecture.md +++ b/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. diff --git a/guides/Cluster.md b/guides/Cluster.md index 4656980..dc42e6b 100644 --- a/guides/Cluster.md +++ b/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. diff --git a/guides/Upgrades.md b/guides/Upgrades.md index dd8ade2..bc0675c 100644 --- a/guides/Upgrades.md +++ b/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: From 087771f1437ba337eb1452e595379cc81c7438bb Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 20:37:42 -0400 Subject: [PATCH 10/13] chore(test): show the lock server fails to start when its pool is down Signed-off-by: Yordis Prieto --- test/advisory_locks_test.exs | 42 ++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/test/advisory_locks_test.exs b/test/advisory_locks_test.exs index 0aa9383..1d51522 100644 --- a/test/advisory_locks_test.exs +++ b/test/advisory_locks_test.exs @@ -708,6 +708,48 @@ defmodule EventStore.AdvisoryLocksTest do end end + describe "lock pool down when the lock server starts" do + @tag capture_log: true + test "should start even when the lock pool is down" do + locks_name = Module.concat([__MODULE__, PoolDownAtStart, AdvisoryLocks]) + conn_name = Module.concat([locks_name, Postgrex]) + monitor_name = Module.concat([conn_name, MonitoredServer]) + + config = Config.parsed(TestEventStore, :eventstore) + + postgrex_config = + config + |> Config.advisory_locks_postgrex_opts() + |> Config.add_connection_listener(locks_name) + |> Keyword.put(:name, conn_name) + + start_supervised!( + {MonitoredServer, + mfa: {Postgrex, :start_link, [postgrex_config]}, name: monitor_name, backoff_min: 30_000}, + id: monitor_name + ) + + Wait.until(fn -> assert is_pid(Process.whereis(conn_name)) end) + + # The pool is down and its long backoff keeps the name unregistered while the + # lock server starts, so the startup disconnect has nothing to call. + Process.exit(Process.whereis(conn_name), :kill) + Wait.until(fn -> assert is_nil(Process.whereis(conn_name)) end) + + assert {:ok, locks} = + start_supervised( + {AdvisoryLocks, + conn: conn_name, + monitor: monitor_name, + query_timeout: Keyword.fetch!(config, :timeout), + schema: "public", + name: locks_name} + ) + + assert Process.alive?(locks) + end + end + defp hold_only_connection do conn = lock_pool_pid() From 245ba33db3a652a8252fd26203d1f446ecd54592 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 20:37:43 -0400 Subject: [PATCH 11/13] fix(subscriptions): keep the lock server starting when the lock pool is down Signed-off-by: Yordis Prieto --- lib/event_store/advisory_locks.ex | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex index 35c2250..501e7d2 100644 --- a/lib/event_store/advisory_locks.ex +++ b/lib/event_store/advisory_locks.ex @@ -124,8 +124,9 @@ defmodule EventStore.AdvisoryLocks do {: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 @@ -466,4 +467,11 @@ defmodule EventStore.AdvisoryLocks do {: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 From 50f89514c347a16fc36d57de46fb746af1b4fbb9 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 21:35:32 -0400 Subject: [PATCH 12/13] chore(test): show an explicit release leaks its lock when the session cannot be verified Signed-off-by: Yordis Prieto --- test/advisory_locks_test.exs | 64 ++++++++++++++++++++++++++++++++++-- 1 file changed, 62 insertions(+), 2 deletions(-) diff --git a/test/advisory_locks_test.exs b/test/advisory_locks_test.exs index 1d51522..f2e978b 100644 --- a/test/advisory_locks_test.exs +++ b/test/advisory_locks_test.exs @@ -750,9 +750,69 @@ defmodule EventStore.AdvisoryLocksTest do end end - defp hold_only_connection do - conn = lock_pool_pid() + describe "release a lock while the lock connection cannot answer" do + @tag capture_log: true + test "should retry the unlock until the lock is released", %{conn: conn, schema: schema} do + alias EventStore.Storage.Session + + locks_name = Module.concat([__MODULE__, ReleaseWithoutSession, AdvisoryLocks]) + conn_name = Module.concat([locks_name, Postgrex]) + monitor_name = Module.concat([conn_name, MonitoredServer]) + + config = Config.parsed(TestEventStore, :eventstore) + + postgrex_config = + config + |> Config.advisory_locks_postgrex_opts() + |> Config.add_connection_listener(locks_name) + |> Keyword.put(:name, conn_name) + |> Keyword.merge(queue_target: 1, queue_interval: 1) + + start_supervised!( + {MonitoredServer, + mfa: {Postgrex, :start_link, [postgrex_config]}, name: monitor_name, backoff_min: 30_000}, + id: monitor_name + ) + + start_supervised!( + {AdvisoryLocks, + conn: conn_name, + monitor: monitor_name, + query_timeout: 100, + schema: schema, + name: locks_name} + ) + + # The startup replacement of the pre-registration connection settles once a + # session answers, so the lock rides a connection that survives. + Wait.until(5_000, fn -> + assert {:ok, _session} = Session.current(conn_name, timeout: 100) + end) + + {:ok, lock} = AdvisoryLocks.try_advisory_lock(locks_name, 1) + + assert {:error, :lock_already_taken} = + Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema) + + # The only pooled connection is busy, so the release cannot confirm its + # session and parks the unlock instead of completing it. + holder = hold_only_connection(conn_name) + + try do + :ok = AdvisoryLocks.release(locks_name, lock) + + Wait.until(fn -> refute Enum.empty?(:sys.get_state(locks_name).released) end) + after + send(holder, :release) + end + + Wait.until(5_000, fn -> + assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema) + end) + end + end + defp hold_only_connection(conn \\ lock_pool_pid()) do reply_to = self() holder = From 566dabf382023abebedf30055261dfbf16edc080 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Fri, 23 Oct 2026 21:35:32 -0400 Subject: [PATCH 13/13] fix(subscriptions): drain an explicit release through the session reconciliation path Signed-off-by: Yordis Prieto --- lib/event_store/advisory_locks.ex | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex index 501e7d2..e5bfea6 100644 --- a/lib/event_store/advisory_locks.ex +++ b/lib/event_store/advisory_locks.ex @@ -440,18 +440,8 @@ defmodule EventStore.AdvisoryLocks do # 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{key: key} = lock, %State{} = state) do - %State{query_timeout: query_timeout, released: released} = state - - case query(state, &Session.current(&1, timeout: query_timeout)) do - {:ok, %Session{} = session} -> - if Lock.held_by?(lock, session), do: release_lock(key, state) - - state - - {:error, _error} -> - %State{state | released: [lock | released]} - end + 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