From a16a80d159a4db868f5fe46ec40316cfd04f3eb2 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 20 Oct 2026 19:33:03 -0400 Subject: [PATCH 1/5] chore(test): show subscriptions have no record of which lock owner may write their progress Signed-off-by: Yordis Prieto --- test/list_event_store_migrations_test.exs | 1 + test/storage/subscription_lock_test.exs | 108 ++++++++++++++++++++++ 2 files changed, 109 insertions(+) create mode 100644 test/storage/subscription_lock_test.exs diff --git a/test/list_event_store_migrations_test.exs b/test/list_event_store_migrations_test.exs index b0ac0aee..f45bab7d 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/storage/subscription_lock_test.exs b/test/storage/subscription_lock_test.exs new file mode 100644 index 00000000..948cdbba --- /dev/null +++ b/test/storage/subscription_lock_test.exs @@ -0,0 +1,108 @@ +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 + + 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 From 5eb179e5e595f37191703e3c21b9d2cf4c3a4171 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 20 Oct 2026 19:41:39 -0400 Subject: [PATCH 2/5] feat(storage): record which lock owner may write a subscription's progress 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 | 31 +++++++ 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/migrate_event_store_test.exs | 1 + 11 files changed, 251 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 diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex index b018cf63..1243734c 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 99081d30..18f5a97b 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 6d53011d..ce2a1fdc 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 00000000..798fbaea --- /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 9fd609bc..f6853985 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,35 @@ 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, _error} = 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 00000000..7ecac724 --- /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 00000000..b72f7d6c --- /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 8b3dbc19..a8527611 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 00000000..ebbddb42 --- /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 b0799af3..2f141d56 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() @@ -513,4 +591,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/migrate_event_store_test.exs b/test/migrate_event_store_test.exs index 31588b0c..f958a109 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 From 32a028abf6d8c7b29ec9f509439b20b47a40cfd4 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Wed, 21 Oct 2026 01:04:52 -0400 Subject: [PATCH 3/5] chore(test): show a node on the previous version keeps working against the migrated schema Signed-off-by: Yordis Prieto --- .../subscription_previous_version_test.exs | 91 +++++++++++++++++++ 1 file changed, 91 insertions(+) create mode 100644 test/storage/subscription_previous_version_test.exs diff --git a/test/storage/subscription_previous_version_test.exs b/test/storage/subscription_previous_version_test.exs new file mode 100644 index 00000000..57ad2a74 --- /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 baad3fea1dcacb07ede074a589d2a7a5ae5b84f6 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Wed, 21 Oct 2026 03:58:07 -0400 Subject: [PATCH 4/5] chore(test): show a claim that fails after taking the advisory lock leaves the subscription unclaimable Signed-off-by: Yordis Prieto --- test/storage/subscription_lock_test.exs | 45 +++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/test/storage/subscription_lock_test.exs b/test/storage/subscription_lock_test.exs index 948cdbba..baf2199b 100644 --- a/test/storage/subscription_lock_test.exs +++ b/test/storage/subscription_lock_test.exs @@ -88,6 +88,51 @@ defmodule EventStore.Storage.SubscriptionLockTest do 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) From 8485fadbd5c4e08bedfb884738e268ae52e8ca19 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Wed, 21 Oct 2026 04:11:58 -0400 Subject: [PATCH 5/5] fix(storage): release the advisory lock when a claim fails after taking it Signed-off-by: Yordis Prieto --- lib/event_store/storage/lock.ex | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/lib/event_store/storage/lock.ex b/lib/event_store/storage/lock.ex index f6853985..0c4301a3 100644 --- a/lib/event_store/storage/lock.ex +++ b/lib/event_store/storage/lock.ex @@ -75,7 +75,12 @@ defmodule EventStore.Storage.Lock do {:ok, Lease.new(token, last_seen, Session.new(backend_pid, session_id))} - {:error, _error} = reply -> + {:error, %Postgrex.Error{} = error} -> + _ = unlock(conn, subscription_id, [schema: schema] ++ opts) + + {:error, error} + + {:error, %DBConnection.ConnectionError{}} = reply -> reply end end