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

[alexeyzimarev]

Fixes #553

Problem

CommandService could report success while the events were missing from the stream under concurrent writes to the same aggregate with the Postgres store:

1. check_stream read streams.version without a row lock, so concurrent writers both passed the expected-version check with the same current version.
2. append_events inserted messages with on conflict do nothing, so the losing writer's rows were silently dropped when they hit the (stream_id, stream_position) unique constraint.
3. The function then returned the stream's max position, so the loser saw success with the winner's version.

Reproduced with 20 parallel ExpectedStreamVersion.Any appends: all 20 reported success, 10 events were lost.

Fix

• 3_CheckStream.sql: the stream row is selected FOR UPDATE — the lock is held to the end of the append transaction, serialising concurrent appends to the same stream. The stream-creation race is handled with insert … on conflict (stream_name) do nothing followed by a locked re-read, and the expected-version check runs after the lock is acquired. Concurrent Any appends now all succeed (queued behind the lock); stale expected versions fail with WrongExpectedVersion.
• 2_AppendEvents.sql: removed on conflict do nothing. As defence in depth, a residual stream-position conflict raises WrongExpectedVersion (mapped by PostgresStore.IsConflict to AppendToStreamException/OptimisticConcurrencyException); any other unique violation is re-raised as-is. This mirrors the SQL Server implementation, which already handled the race this way.

Tests

Two new tests in the shared StoreAppendTests base, so all stores enforce the invariant:

• ShouldNotLoseConcurrentAppends — every append that reports success must be durable (fails pre-fix on Postgres: 10 of 20 events lost).
• ShouldRejectConcurrentAppendsWithSameVersion — of N concurrent appends with the same expected version, exactly one succeeds and the rest throw AppendToStreamException.

Verified: full Postgres suite 49/49, Sqlite and KurrentDB append tests green. SQL Server tests are excluded on macOS and will run in CI.

Deployment note

The functions are create or replace, so existing databases pick up the fix only when the schema scripts re-run — users with InitializeDatabase = false need to apply them manually. Worth a line in the release notes.

🤖 Generated with Claude Code

[qodo-free-for-open-source-projects]

PR Summary by Qodo

Fix Postgres concurrent appends to fail loudly (no silent event loss)

🐞 Bug fix 🧪 Tests 🕐 20-40 Minutes

AI Description

• Serialize concurrent appends per stream by locking the stream row in Postgres.
• Stop swallowing insert conflicts; surface concurrency violations as WrongExpectedVersion.
• Add cross-store concurrency tests to ensure successful appends are always durable.

Diagram

graph TD
  A["CommandService"] --> B["PostgresStore"] --> C[["append_events()"]] --> D[["check_stream()"]] --> E[("streams")]
  C --> F[("messages")]
  C --> G{"Unique violation?"} --> H["WrongExpectedVersion"] --> B

  subgraph Legend
    direction LR
    _svc["Service/Module"] ~~~ _fn[["DB function"]] ~~~ _db[("Table")] ~~~ _dec{"Decision"}
  end

Loading

High-Level Assessment

The following are alternative approaches to this PR:

1. Advisory locks per stream

• ➕ Explicit, per-stream serialization without relying on row existence
• ➕ Can avoid extra insert/read path for stream-creation races
• ➖ Additional operational complexity (lock keying, deadlock considerations)
• ➖ Less idiomatic than row-level locking on the stream record

2. Keep ON CONFLICT DO NOTHING but validate inserted row count

• ➕ Avoids raising exceptions on conflicts at the SQL layer
• ➕ Can provide custom error messages based on detected short-write
• ➖ More error-prone: must correctly detect partial inserts across a batch
• ➖ Still risks edge cases if validation logic diverges from constraint semantics

3. Use SERIALIZABLE isolation for append transactions

• ➕ Lets the database detect write skew without manual locks
• ➕ Uniform semantics across multiple tables
• ➖ Higher overhead and more frequent transaction retries under contention
• ➖ Requires careful retry strategy at the application layer

Recommendation: The PR’s approach (row-level locking via SELECT ... FOR UPDATE in check_stream plus removing ON CONFLICT DO NOTHING in append_events) is the most direct and reliable fix. It prevents the write-skew window, ensures conflicts are surfaced as concurrency errors, and aligns Postgres behavior with the existing SQL Server strategy. The added shared tests also guard against regressions across all stores.

Files changed (3) +76 / -16

Bug fix (2) +32 / -16

2_AppendEvents.sqlStop swallowing insert conflicts; map residual stream-position conflicts to WrongExpectedVersion +11/-2

Stop swallowing insert conflicts; map residual stream-position conflicts to WrongExpectedVersion

• Removes "ON CONFLICT DO NOTHING" from message inserts so conflicts are not silently ignored. Adds an exception handler for unique_violation: stream position constraint violations are rethrown as WrongExpectedVersion (defense in depth), while other unique violations are re-raised unchanged.

src/Postgres/src/Eventuous.Postgresql/Scripts/2_AppendEvents.sql

3_CheckStream.sqlLock stream row FOR UPDATE and handle stream-creation races safely +21/-14

Lock stream row FOR UPDATE and handle stream-creation races safely

• Changes the stream lookup to SELECT ... FOR UPDATE so concurrent appends to the same stream serialize behind the row lock. Handles concurrent stream creation by inserting with ON CONFLICT DO NOTHING and then re-reading the stream row with a lock, and moves the expected-version check to run after the lock is acquired.

src/Postgres/src/Eventuous.Postgresql/Scripts/3_CheckStream.sql

Tests (1) +44 / -0

Append.csAdd cross-store concurrent append durability and conflict tests +44/-0

Add cross-store concurrent append durability and conflict tests

• Adds two new concurrency tests to the shared store append test suite. One ensures that every successful concurrent append is durable (no silent loss), and the other asserts that concurrent appends using the same expected version result in exactly one success and the rest failing with AppendToStreamException.

src/Core/test/Eventuous.Tests.Persistence.Base/Store/Append.cs

[qodo-free-for-open-source-projects]

Code Review by Qodo

🐞 Bugs (0) 📘 Rule violations (0) 📎 Requirement gaps (0)

Great, no issues found!

Qodo reviewed your code and found no material issues that require review

Tip of the day

💡 Did you know, you can tweak Display preferences with a live preview to see your comment before it ships

More tips ↗ | Customize Qodo ↗ | Qodo docs ↗

[github-actions]

Test Results

   44 files  + 21     44 suites  +21   12m 27s ⏱️ - 1m 15s
  555 tests  -   9    555 ✅  -   8  0 💤 ±0  0 ❌  - 1 
1 110 runs  +535  1 110 ✅ +536  0 💤 ±0  0 ❌  - 1 

Results for commit 7e87b89. ± Comparison against base commit a178a37.

This pull request removes 26 and adds 17 tests. Note that renamed tests count towards both.

Eventuous.Tests.Azure.ServiceBus.IsSerialisableByServiceBus ‑ Passes(08/21/2026 11:25:08 +00:00)
Eventuous.Tests.Azure.ServiceBus.IsSerialisableByServiceBus ‑ Passes(08/21/2026 11:25:08)
Eventuous.Tests.Azure.ServiceBus.IsSerialisableByServiceBus ‑ Passes(e7c7ecc7-b9a8-4dff-8071-39f696a285ed)
Eventuous.Tests.Azure.Storage.Blobs.BlobStorageProjectorTests ‑ AsyncContextAwareHandler_ExistingBlob_ShouldUpdateStateAndContext
Eventuous.Tests.Azure.Storage.Blobs.BlobStorageProjectorTests ‑ AsyncContextAwareHandler_NewBlob_ShouldUseContextAndStoreState
Eventuous.Tests.Azure.Storage.Blobs.BlobStorageProjectorTests ‑ AsyncStateHandler_ExistingBlob_ShouldUpdateState
Eventuous.Tests.Azure.Storage.Blobs.BlobStorageProjectorTests ‑ AsyncStateHandler_NewBlob_ShouldCreateAndStoreState
Eventuous.Tests.Azure.Storage.Blobs.BlobStorageProjectorTests ‑ ConcurrentAdditionOfNewBlob_ShouldReturnFailure
Eventuous.Tests.Azure.Storage.Blobs.BlobStorageProjectorTests ‑ ConcurrentModificationOfExistingBlob_ShouldReturnFailure
Eventuous.Tests.Azure.Storage.Blobs.BlobStorageProjectorTests ‑ CustomBlobId_ExistingBlob_ShouldUpdateWithEventId
…

Eventuous.Tests.Azure.ServiceBus.IsSerialisableByServiceBus ‑ Passes(08/21/2026 11:46:40 +00:00)
Eventuous.Tests.Azure.ServiceBus.IsSerialisableByServiceBus ‑ Passes(08/21/2026 11:46:40)
Eventuous.Tests.Azure.ServiceBus.IsSerialisableByServiceBus ‑ Passes(383bd357-2886-4cc9-b577-1b99c0a43b8d)
Eventuous.Tests.KurrentDB.Store.Append ‑ ShouldNotLoseConcurrentAppends
Eventuous.Tests.KurrentDB.Store.Append ‑ ShouldRejectConcurrentAppendsWithSameVersion
Eventuous.Tests.Postgres.Store.Append ‑ ShouldNotLoseConcurrentAppends
Eventuous.Tests.Postgres.Store.Append ‑ ShouldRejectConcurrentAppendsWithSameVersion
Eventuous.Tests.SqlServer.Store.Append ‑ ShouldNotLoseConcurrentAppends
Eventuous.Tests.SqlServer.Store.Append ‑ ShouldRejectConcurrentAppendsWithSameVersion
Eventuous.Tests.Sqlite.Store.Append ‑ ShouldNotLoseConcurrentAppends
…