fix(postgres): fail loudly instead of silently dropping racing appends

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<bool> TryAppend(StreamName stream, ExpectedStreamVersion version) {
+        try {
+            await _fixture.AppendEvent(stream, Helpers.CreateEvent(), version);
+
+            return true;
+        } catch (AppendToStreamException) {
+            return false;
+        }
+    }
 }

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';

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';
+$$ language 'plpgsql';