From 7e87b894a336bb0de3f6bdf6b90f0e3ee6587149 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Fri, 21 Aug 2026 13:41:15 +0200 Subject: [PATCH] fix(postgres): fail loudly instead of silently dropping racing appends Concurrent appends to the same stream could both pass the expected-version check because check_stream read the stream row without a lock, and the loser's events were then silently discarded by ON CONFLICT DO NOTHING in append_events, which still reported success with the winner's version. check_stream now locks the stream row with FOR UPDATE, serialising appends to the same stream, and handles the concurrent stream-creation race with ON CONFLICT DO NOTHING plus a locked re-read. append_events no longer swallows insert conflicts: a stream-position conflict raises WrongExpectedVersion, which the client maps to OptimisticConcurrencyException. Adds concurrency tests to the shared store test base so every store enforces the invariant that a successful append is durable and conflicting concurrent appends fail loudly. Fixes #553 Co-Authored-By: Claude Fable 5 --- .../Store/Append.cs | 44 +++++++++++++++++++ .../Scripts/2_AppendEvents.sql | 13 +++++- .../Scripts/3_CheckStream.sql | 35 +++++++++------ 3 files changed, 76 insertions(+), 16 deletions(-) diff --git a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Append.cs b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Append.cs index 67f0714cb..b7648a0f4 100644 --- a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Append.cs +++ b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Append.cs @@ -123,4 +123,48 @@ public async Task ShouldReturnEmptyResultsForEmptyAppends() { var results = await _fixture.AppendEventsToMultipleStreams([]); await Assert.That(results).HasCount().EqualTo(0); } + + [Test] + [Category("Store")] + public async Task ShouldNotLoseConcurrentAppends(CancellationToken cancellationToken) { + const int writers = 20; + + var stream = Helpers.GetStreamName(); + await _fixture.AppendEvent(stream, Helpers.CreateEvent(), ExpectedStreamVersion.NoStream); + + var results = await Task.WhenAll(Enumerable.Range(0, writers).Select(_ => TryAppend(stream, ExpectedStreamVersion.Any))); + var succeeded = results.Count(x => x); + + var stored = await _fixture.EventStore.ReadEvents(stream, StreamReadPosition.Start, writers * 2, true, cancellationToken); + + await Assert.That(succeeded).IsGreaterThan(0); + await Assert.That(stored.Length).IsEqualTo(1 + succeeded); + } + + [Test] + [Category("Store")] + public async Task ShouldRejectConcurrentAppendsWithSameVersion(CancellationToken cancellationToken) { + const int writers = 20; + + var stream = Helpers.GetStreamName(); + await _fixture.AppendEvent(stream, Helpers.CreateEvent(), ExpectedStreamVersion.NoStream); + + var results = await Task.WhenAll(Enumerable.Range(0, writers).Select(_ => TryAppend(stream, new(0)))); + var succeeded = results.Count(x => x); + + var stored = await _fixture.EventStore.ReadEvents(stream, StreamReadPosition.Start, writers * 2, true, cancellationToken); + + await Assert.That(succeeded).IsEqualTo(1); + await Assert.That(stored.Length).IsEqualTo(1 + succeeded); + } + + async Task TryAppend(StreamName stream, ExpectedStreamVersion version) { + try { + await _fixture.AppendEvent(stream, Helpers.CreateEvent(), version); + + return true; + } catch (AppendToStreamException) { + return false; + } + } } diff --git a/src/Postgres/src/Eventuous.Postgresql/Scripts/2_AppendEvents.sql b/src/Postgres/src/Eventuous.Postgresql/Scripts/2_AppendEvents.sql index aa0e8aa3d..b804acf20 100644 --- a/src/Postgres/src/Eventuous.Postgresql/Scripts/2_AppendEvents.sql +++ b/src/Postgres/src/Eventuous.Postgresql/Scripts/2_AppendEvents.sql @@ -11,6 +11,7 @@ declare _current_version integer; _stream_id integer; _position bigint; + _constraint_name text; begin if _created is null then _created = now() at time zone 'utc'; @@ -24,8 +25,7 @@ begin select m.message_id, m.message_type, _stream_id, _current_version + (row_number() over ()) :: int, m.json_data, m.json_metadata, _created - from unnest(_messages) m - on conflict do nothing; + from unnest(_messages) m; select m.stream_position, m.global_position into _current_version, _position from __schema__.messages m @@ -38,6 +38,15 @@ begin end if; return query select _current_version, _position; +exception + when unique_violation then + get stacked diagnostics _constraint_name = constraint_name; + -- Should not happen since check_stream locks the stream row, but a writer racing us + -- must get a concurrency conflict, never a silent no-op reported as success + if _constraint_name = 'uq_messages_stream_id_and_stream_position' then + raise exception 'WrongExpectedVersion %, concurrent append detected', _expected_version; + end if; + raise; end; $$ language 'plpgsql'; diff --git a/src/Postgres/src/Eventuous.Postgresql/Scripts/3_CheckStream.sql b/src/Postgres/src/Eventuous.Postgresql/Scripts/3_CheckStream.sql index 9d1ba82c9..4fd902718 100644 --- a/src/Postgres/src/Eventuous.Postgresql/Scripts/3_CheckStream.sql +++ b/src/Postgres/src/Eventuous.Postgresql/Scripts/3_CheckStream.sql @@ -8,26 +8,33 @@ declare _current_version integer; _stream_id integer; begin - select s.version, s.stream_id into _current_version, _stream_id + -- The row lock is held until the end of the caller's transaction, + -- serialising concurrent appends to the same stream + select s.version, s.stream_id into _current_version, _stream_id from __schema__.streams s - where s.stream_name = _stream_name; + where s.stream_name = _stream_name + for update; if _stream_id is null then -- Stream doesn't exist - if _expected_version = -2 -- Any - or _expected_version = -1 then -- NoStream - insert into __schema__.streams (stream_name, version) values (_stream_name, -1); - select s.stream_id, s.version into _stream_id, _current_version - from __schema__.streams s - where stream_name = _stream_name; - else + if _expected_version != -2 -- Any + and _expected_version != -1 then -- NoStream raise exception 'StreamNotFound'; end if; - else -- Stream exists - if _expected_version != -2 and _expected_version != _current_version then - raise exception 'WrongExpectedVersion %, current version %', _expected_version, _current_version; - end if; + + -- A concurrent transaction may create the same stream; do nothing then, + -- and the re-read below locks the winner's row and gets its version + insert into __schema__.streams (stream_name, version) values (_stream_name, -1) + on conflict (stream_name) do nothing; + select s.stream_id, s.version into _stream_id, _current_version + from __schema__.streams s + where stream_name = _stream_name + for update; + end if; + + if _expected_version != -2 and _expected_version != _current_version then + raise exception 'WrongExpectedVersion %, current version %', _expected_version, _current_version; end if; return query select _stream_id, _current_version; end; -$$ language 'plpgsql'; \ No newline at end of file +$$ language 'plpgsql';