From 33e346a12e53e743974d954719734dd80cd176ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Recep=20=C5=9Een?= Date: Tue, 20 Oct 2026 11:57:07 +0300 Subject: [PATCH 1/5] fix(idempotency): store observe results in the business transaction Observe mode reserved and completed through the detached store, so a Completed record could outlive a business transaction whose commit failed. After the switch to Enforce, a retry replayed a success that never committed. Observe now reserves through the detached store, releases that reservation once the operation finishes, and writes the result through the attached store inside the business or ambient transaction. A failed commit rolls the result back with the business data, so the retry runs again. Release paths stay detached and Enforce is unchanged. The EF Core store guards a result write for a token it did not reserve with a savepoint (the EF savepoint API, or raw SAVEPOINT statements in an ambient PostgreSQL transaction), so a failed result write cannot abort the business transaction and lose its data on commit. The HTTP middleware applies the same flow. --- .../Idempotency/IdempotencyMiddleware.cs | 26 +++-- .../Idempotency/IdempotencyOptions.cs | 4 +- .../EfCoreIdempotencyRecordStore.cs | 98 ++++++++++++++++--- .../Idempotency/IdempotencyRecordSql.cs | 12 ++- .../ITransactionalIdempotencyRecordStore.cs | 13 ++- .../IdempotencyExecutor.cs | 32 +++--- .../IdempotencyMode.cs | 5 +- ...CoreIdempotencyRecordStorePostgresTests.cs | 4 +- .../Idempotency/IdempotencyExecutorTests.cs | 17 +++- README.MD | 2 +- 10 files changed, 162 insertions(+), 51 deletions(-) diff --git a/CSharpEssentials.AspNetCore/Idempotency/IdempotencyMiddleware.cs b/CSharpEssentials.AspNetCore/Idempotency/IdempotencyMiddleware.cs index 01bc4d4..d3143b3 100644 --- a/CSharpEssentials.AspNetCore/Idempotency/IdempotencyMiddleware.cs +++ b/CSharpEssentials.AspNetCore/Idempotency/IdempotencyMiddleware.cs @@ -24,8 +24,11 @@ namespace CSharpEssentials.AspNetCore; /// The options are read once per request from , so a reload applies to the next /// request; the endpoint's mode (, its profile's, else /// ) decides whether the key is ignored, observed or enforced. In Observe an -/// is used through its -/// view, so observing never joins the request's transaction. +/// reserves and releases through its +/// view, so the reservation never joins the request's +/// transaction. A response to store releases that reservation and is then written through the store itself, in the +/// transaction open around UseIdempotency if there is one: the record exists only when that transaction +/// commits, so a retry after the switch to Enforce never replays a response whose work rolled back. /// internal sealed partial class IdempotencyMiddleware( RequestDelegate next, @@ -126,10 +129,11 @@ await WriteProblemAsync(context, StatusCodes.Status400BadRequest, "Idempotency s IdempotencyRecordKey recordKey = new(options.Namespace, scope.Value, key); IIdempotencyRecordStore store; + IIdempotencyRecordStore results; string? fingerprint; try { - store = settings.StoreKey is { } storeKey + store = results = settings.StoreKey is { } storeKey ? context.RequestServices.GetRequiredKeyedService(storeKey) : defaultStore; if (observe && store is ITransactionalIdempotencyRecordStore transactional) @@ -141,7 +145,7 @@ await WriteProblemAsync(context, StatusCodes.Status400BadRequest, "Idempotency s LogObserveResolutionFailed(logger, exception, context.Request.Path); if (context.Request.Body.CanSeek) context.Request.Body.Position = 0; - store = defaultStore; + store = results = defaultStore; fingerprint = null; } @@ -172,7 +176,7 @@ await WriteProblemAsync(context, StatusCodes.Status400BadRequest, "Idempotency s switch (reservation.Status) { case IdempotencyReservationStatus.Reserved: - await ExecuteAsync(context, options, settings, store, recordKey, reservation.Token!, fingerprint, target).ConfigureAwait(false); + await ExecuteAsync(context, options, settings, store, results, recordKey, reservation.Token!, fingerprint, target).ConfigureAwait(false); break; case IdempotencyReservationStatus.Completed when observe: LogObserveWouldReplay(logger, context.Request.Path, new IdempotencyKeyHash(recordKey)); @@ -274,6 +278,7 @@ private async Task ExecuteAsync( IdempotencyOptions options, IdempotencyEndpointSettings settings, IIdempotencyRecordStore store, + IIdempotencyRecordStore results, IdempotencyRecordKey recordKey, string token, string fingerprint, @@ -304,7 +309,7 @@ private async Task ExecuteAsync( try { bool filterInstalled = context.GetEndpoint()?.Metadata.GetMetadata() is not null; - if (!await SettleAsync(context, options, settings, store, recordKey, token, fingerprint, capture, execution.HandlerStarted, filterInstalled, target).ConfigureAwait(false)) + if (!await SettleAsync(context, options, settings, store, results, recordKey, token, fingerprint, capture, execution.HandlerStarted, filterInstalled, target).ConfigureAwait(false)) LogStaleReservation(logger, context.Request.Path, new IdempotencyKeyHash(recordKey)); } catch (Exception exception) @@ -318,7 +323,9 @@ private async Task ExecuteAsync( /// ignore : a client that went away still gets its outcome on retry. /// An endpoint without the execution filter cannot report whether its handler started, so a 2xx from it is /// settled as a started handler's response and every other status releases the key. A release or a stored response - /// is counted in . + /// is counted in . A response is written to + /// ; when that is not the that holds the reservation, the + /// reservation is released first, so the response is written as the record of a key that is gone. /// /// when the reservation was taken over after InFlightTimeout. private async ValueTask SettleAsync( @@ -326,6 +333,7 @@ private async ValueTask SettleAsync( IdempotencyOptions options, IdempotencyEndpointSettings settings, IIdempotencyRecordStore store, + IIdempotencyRecordStore results, IdempotencyRecordKey recordKey, string token, string fingerprint, @@ -370,7 +378,9 @@ private async ValueTask SettleAsync( headers[name] = [.. values.OfType()]; } - bool stored = await store.CompleteAsync( + if (!ReferenceEquals(store, results)) + await TryReleaseAsync(store, recordKey, token, context.Request.Path).ConfigureAwait(false); + bool stored = await results.CompleteAsync( recordKey, token, fingerprint, diff --git a/CSharpEssentials.AspNetCore/Idempotency/IdempotencyOptions.cs b/CSharpEssentials.AspNetCore/Idempotency/IdempotencyOptions.cs index 35ea21e..d558df7 100644 --- a/CSharpEssentials.AspNetCore/Idempotency/IdempotencyOptions.cs +++ b/CSharpEssentials.AspNetCore/Idempotency/IdempotencyOptions.cs @@ -20,7 +20,9 @@ public sealed class IdempotencyOptions /// answers differently from : no replay, no Idempotency-Replayed, no /// 400/409/422; it logs and counts what would have done. /// Observe measures duplicates, it does not prevent them; keep it short for operations that move money. A record - /// stored in Observe is replayed once the mode becomes Enforce. This switch is separate from + /// stored in Observe is replayed once the mode becomes Enforce; an + /// stores it in the transaction open around UseIdempotency, so a response whose transaction rolled back is + /// never replayed. This switch is separate from /// IdempotencyExecutorOptions.Mode, which Mediator and message consumers read. Default: /// . /// diff --git a/CSharpEssentials.EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStore.cs b/CSharpEssentials.EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStore.cs index 87252b9..f1c3abf 100644 --- a/CSharpEssentials.EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStore.cs +++ b/CSharpEssentials.EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStore.cs @@ -36,9 +36,17 @@ namespace CSharpEssentials.EntityFrameworkCore.Idempotency; /// . /// /// -/// In the executor and the HTTP middleware use , which writes -/// from a new of its own scope, outside the business and ambient transactions, so -/// observing neither waits for a concurrent request nor can fail the business transaction. +/// In the executor and the HTTP middleware reserve and release through +/// , which writes from a new of its own scope, outside the business +/// and ambient transactions, so a reservation neither waits for a concurrent request nor can fail the business +/// transaction. A result to store releases that reservation and is written through this store, in the business +/// transaction, so it commits and rolls back with the business data as in . +/// A result for a token this instance did not reserve, as in Observe, is written behind a savepoint of the context's +/// current transaction (on PostgreSQL also of an ambient System.Transactions transaction the connection is +/// enlisted in): a database error rolls back to the savepoint and is rethrown, so the transaction stays usable and the +/// business data commits without the record. Observe therefore stores results only under READ COMMITTED: under a +/// stricter level the write fails with a serialization error, because the release changed the row after the +/// transaction's snapshot was taken. /// /// /// The context that maps . @@ -47,11 +55,13 @@ public sealed class EfCoreIdempotencyRecordStore : ITransactionalIdemp { private const int MaxReserveAttempts = 8; private const string SavepointName = "cse_idempotency"; + private const string ResultSavepointName = "cse_idempotency_result"; private readonly TContext _context; private readonly IServiceScopeFactory _scopeFactory; private readonly TimeProvider _timeProvider; private readonly bool _plainInsert; + private readonly HashSet _reservedTokens = []; private DetachedEfCoreIdempotencyRecordStore? _detached; /// Creates a store on . @@ -110,7 +120,10 @@ public async ValueTask TryReserveAsync( string token = Guid.NewGuid().ToString("N"); if (await TryInsertAsync(key, hash, token, fingerprint, payload: null, expiresAt, cancellationToken).ConfigureAwait(false)) + { + _reservedTokens.Add(token); return IdempotencyRecordReservation.Reserved(token); + } int takenOver = await Records(key, hash) .Where(record => record.ExpiresAt <= now) @@ -123,7 +136,10 @@ public async ValueTask TryReserveAsync( cancellationToken) .ConfigureAwait(false); if (takenOver == 1) + { + _reservedTokens.Add(token); return IdempotencyRecordReservation.Reserved(token); + } var existing = await Records(hash) .Select(record => new { record.Namespace, record.Scope, record.Key, record.Fingerprint, record.Payload, record.ExpiresAt }) @@ -166,6 +182,69 @@ public async ValueTask CompleteAsync( ArgumentException.ThrowIfNullOrEmpty(fingerprint); ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(retention, TimeSpan.Zero); + if (_reservedTokens.Remove(token)) + return await CompleteCoreAsync(key, token, fingerprint, payload, retention, cancellationToken).ConfigureAwait(false); + + IDbContextTransaction? transaction = _context.Database.CurrentTransaction; + bool savepoint = transaction is null + ? System.Transactions.Transaction.Current is not null && IdempotencyRecordSql.For(_context).AmbientSavepoints + : transaction.SupportsSavepoints; + if (!savepoint) + return await CompleteCoreAsync(key, token, fingerprint, payload, retention, cancellationToken).ConfigureAwait(false); + + await CreateResultSavepointAsync(transaction, cancellationToken).ConfigureAwait(false); + bool completed; + try + { + completed = await CompleteCoreAsync(key, token, fingerprint, payload, retention, cancellationToken).ConfigureAwait(false); + } + catch (Exception exception) when (exception is DbException || exception.InnerException is DbException) + { + await RollbackToResultSavepointAsync(transaction).ConfigureAwait(false); + throw; + } + + await ReleaseResultSavepointAsync(transaction, cancellationToken).ConfigureAwait(false); + return completed; + } + + /// + public async ValueTask ReleaseAsync(IdempotencyRecordKey key, string token, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(key); + ArgumentException.ThrowIfNullOrEmpty(token); + + _reservedTokens.Remove(token); + int released = await Records(key, IdempotencyRecord.HashKey(key)) + .Where(record => record.Token == token && record.Payload == null) + .ExecuteDeleteAsync(cancellationToken) + .ConfigureAwait(false); + return released == 1; + } + + private Task CreateResultSavepointAsync(IDbContextTransaction? transaction, CancellationToken cancellationToken) => + transaction is null + ? _context.Database.ExecuteSqlRawAsync("SAVEPOINT " + ResultSavepointName, cancellationToken) + : transaction.CreateSavepointAsync(ResultSavepointName, cancellationToken); + + private Task RollbackToResultSavepointAsync(IDbContextTransaction? transaction) => + transaction is null + ? _context.Database.ExecuteSqlRawAsync("ROLLBACK TO SAVEPOINT " + ResultSavepointName, CancellationToken.None) + : transaction.RollbackToSavepointAsync(ResultSavepointName, CancellationToken.None); + + private Task ReleaseResultSavepointAsync(IDbContextTransaction? transaction, CancellationToken cancellationToken) => + transaction is null + ? _context.Database.ExecuteSqlRawAsync("RELEASE SAVEPOINT " + ResultSavepointName, cancellationToken) + : transaction.ReleaseSavepointAsync(ResultSavepointName, cancellationToken); + + private async ValueTask CompleteCoreAsync( + IdempotencyRecordKey key, + string token, + string fingerprint, + ReadOnlyMemory payload, + TimeSpan retention, + CancellationToken cancellationToken) + { byte[] hash = IdempotencyRecord.HashKey(key); byte[] stored = payload.ToArray(); DateTime now = _timeProvider.GetUtcNow().UtcDateTime; @@ -199,19 +278,6 @@ public async ValueTask CompleteAsync( return await TryInsertAsync(key, hash, token, fingerprint, stored, expiresAt, cancellationToken).ConfigureAwait(false); } - /// - public async ValueTask ReleaseAsync(IdempotencyRecordKey key, string token, CancellationToken cancellationToken = default) - { - ArgumentNullException.ThrowIfNull(key); - ArgumentException.ThrowIfNullOrEmpty(token); - - int released = await Records(key, IdempotencyRecord.HashKey(key)) - .Where(record => record.Token == token && record.Payload == null) - .ExecuteDeleteAsync(cancellationToken) - .ConfigureAwait(false); - return released == 1; - } - /// /// Deletes every record whose lease or retention has ended, in one statement on the context's connection (inside /// its current transaction when there is one). An expired record already counts as absent, so purging changes no diff --git a/CSharpEssentials.EntityFrameworkCore/Idempotency/IdempotencyRecordSql.cs b/CSharpEssentials.EntityFrameworkCore/Idempotency/IdempotencyRecordSql.cs index 7c78259..37925cb 100644 --- a/CSharpEssentials.EntityFrameworkCore/Idempotency/IdempotencyRecordSql.cs +++ b/CSharpEssentials.EntityFrameworkCore/Idempotency/IdempotencyRecordSql.cs @@ -20,8 +20,9 @@ internal sealed class IdempotencyRecordSql private static readonly ConditionalWeakTable Cache = []; - private IdempotencyRecordSql(string insertInFlight, string insertCompleted, string? onConflictDoNothing) + private IdempotencyRecordSql(string insertInFlight, string insertCompleted, string? onConflictDoNothing, bool ambientSavepoints) { + AmbientSavepoints = ambientSavepoints; InsertInFlight = insertInFlight; InsertCompleted = insertCompleted; if (onConflictDoNothing is not null) @@ -46,6 +47,12 @@ private IdempotencyRecordSql(string insertInFlight, string insertCompleted, stri /// with ON CONFLICT (…) DO NOTHING; see . public string? InsertCompletedOrNothing { get; } + /// + /// Whether the store takes a SAVEPOINT with plain SQL inside an ambient System.Transactions + /// transaction: only on PostgreSQL, whose failed statement aborts the whole transaction. + /// + public bool AmbientSavepoints { get; } + public static IdempotencyRecordSql For(DbContext context) { IModel model = context.Model; @@ -79,7 +86,8 @@ string Column(string property) => ? $" ON CONFLICT ({keyColumns}) DO NOTHING" : null; - return new IdempotencyRecordSql(insert + "NULL, {6})", insert + "{6}, {7})", onConflictDoNothing); + return new IdempotencyRecordSql( + insert + "NULL, {6})", insert + "{6}, {7})", onConflictDoNothing, context.Database.ProviderName is PostgreSqlProvider); } /// Doubles the braces of an identifier so ExecuteSqlRaw does not read them as placeholders. diff --git a/CSharpEssentials.Idempotency/ITransactionalIdempotencyRecordStore.cs b/CSharpEssentials.Idempotency/ITransactionalIdempotencyRecordStore.cs index 0b3e0b4..e24d948 100644 --- a/CSharpEssentials.Idempotency/ITransactionalIdempotencyRecordStore.cs +++ b/CSharpEssentials.Idempotency/ITransactionalIdempotencyRecordStore.cs @@ -5,17 +5,20 @@ namespace CSharpEssentials.Idempotency; /// rolls back with the business data written next to it, and a concurrent request for the same key waits until that /// transaction ends. /// -/// In that wait would block the business transaction and a failed write could -/// abort it, while Observe promises to change nothing. Callers in Observe use instead: -/// and the HTTP middleware switch to it on their own. +/// In a reservation that waited would block the business transaction and a +/// failed one could abort it, while Observe promises to change nothing. and the HTTP +/// middleware therefore reserve and release through in Observe. A result to store releases the +/// detached reservation and is written through this store, in the caller's transaction, so Observe records only work +/// that committed, as Enforce does: a retry after the switch to never replays a +/// result whose transaction rolled back. Between that write and the commit, a reservation of the same key waits for +/// the caller's transaction, as in Enforce. /// /// public interface ITransactionalIdempotencyRecordStore : IIdempotencyRecordStore { /// /// The same records, written outside the caller's transaction: every call runs on its own connection and commits - /// on its own, and a reservation never waits for an uncommitted one. A record written through it is not atomic - /// with the business data. + /// on its own. A record written through it is not atomic with the business data. /// IIdempotencyRecordStore Detached { get; } } diff --git a/CSharpEssentials.Idempotency/IdempotencyExecutor.cs b/CSharpEssentials.Idempotency/IdempotencyExecutor.cs index a2c0339..01193c2 100644 --- a/CSharpEssentials.Idempotency/IdempotencyExecutor.cs +++ b/CSharpEssentials.Idempotency/IdempotencyExecutor.cs @@ -10,9 +10,11 @@ namespace CSharpEssentials.Idempotency; /// on every operation, so a mode changed at runtime applies to the next one. /// Logs name the record namespace and a hash of the key, never the key or the scope. Decisions are counted in /// with the record namespace as target. -/// In an is used through its -/// view, so observing never waits on or joins the -/// business transaction. +/// In an reserves and releases +/// through its view, so the reservation never waits on or +/// fails the business transaction. A result to store releases that reservation and is then written through the store +/// itself, in the business transaction: the record exists only when the business work committed, so a retry after the +/// switch to replays committed work and runs again what rolled back. /// public sealed partial class IdempotencyExecutor : IIdempotencyExecutor { @@ -67,7 +69,7 @@ public async ValueTask> ExecuteAsync( return reservation.Status switch { IdempotencyReservationStatus.Reserved => - await RunAsync(store, key, reservation.Token!, fingerprint, settings, operation, codec, cancellationToken).ConfigureAwait(false), + await RunAsync(store, store, key, reservation.Token!, fingerprint, settings, operation, codec, cancellationToken).ConfigureAwait(false), IdempotencyReservationStatus.Completed => Replay(key, reservation.Payload.GetValueOrDefault(), codec), IdempotencyReservationStatus.InFlight => new(IdempotencyExecutionStatus.InFlight, default), IdempotencyReservationStatus.FingerprintMismatch => new(IdempotencyExecutionStatus.KeyReused, default), @@ -98,13 +100,11 @@ private async ValueTask> ObserveAsync( IIdempotencyResultCodec codec, CancellationToken cancellationToken) { - if (store is ITransactionalIdempotencyRecordStore transactional) - store = transactional.Detached; - + IIdempotencyRecordStore reservations = store is ITransactionalIdempotencyRecordStore transactional ? transactional.Detached : store; IdempotencyRecordReservation? reservation = null; try { - reservation = await store.TryReserveAsync(key, fingerprint, settings.Lease, cancellationToken).ConfigureAwait(false); + reservation = await reservations.TryReserveAsync(key, fingerprint, settings.Lease, cancellationToken).ConfigureAwait(false); } catch (Exception exception) when (!cancellationToken.IsCancellationRequested) { @@ -117,7 +117,7 @@ private async ValueTask> ObserveAsync( switch (reservation?.Status) { case IdempotencyReservationStatus.Reserved: - return await RunAsync(store, key, reservation.Token!, fingerprint, settings, operation, codec, cancellationToken).ConfigureAwait(false); + return await RunAsync(reservations, store, key, reservation.Token!, fingerprint, settings, operation, codec, cancellationToken).ConfigureAwait(false); case IdempotencyReservationStatus.Completed: LogWouldReplay(_logger, key.Namespace, new IdempotencyKeyHash(key)); break; @@ -136,6 +136,7 @@ private async ValueTask> ObserveAsync( private async ValueTask> RunAsync( IIdempotencyRecordStore store, + IIdempotencyRecordStore results, IdempotencyRecordKey key, string token, string fingerprint, @@ -161,17 +162,20 @@ private async ValueTask> RunAsync( throw; } - await SettleAsync(store, key, token, fingerprint, settings, result, codec).ConfigureAwait(false); + await SettleAsync(store, results, key, token, fingerprint, settings, result, codec).ConfigureAwait(false); return new(IdempotencyExecutionStatus.Executed, result); } /// /// Stores, stores as a failure or releases the result of an operation that ran. The work is done, so the store /// calls ignore cancellation, and a failure here is logged instead of thrown: the caller gets its result and the - /// key stays in flight until its lease expires. + /// key stays in flight until its lease expires. A result is written to ; when that is not + /// the that holds the reservation, the reservation is released first, so the result is + /// written as the record of a key that is gone. /// private async ValueTask SettleAsync( IIdempotencyRecordStore store, + IIdempotencyRecordStore results, IdempotencyRecordKey key, string token, string fingerprint, @@ -182,12 +186,14 @@ private async ValueTask SettleAsync( try { IdempotencyResultHandling handling = codec.Classify(result); + if (handling != IdempotencyResultHandling.Release && !ReferenceEquals(store, results)) + await TryReleaseAsync(store, key, token).ConfigureAwait(false); bool settled = handling switch { - IdempotencyResultHandling.Store => await store + IdempotencyResultHandling.Store => await results .CompleteAsync(key, token, fingerprint, codec.Serialize(result), settings.Retention, CancellationToken.None) .ConfigureAwait(false), - IdempotencyResultHandling.StoreAsFailure => await store + IdempotencyResultHandling.StoreAsFailure => await results .CompleteAsync(key, token, fingerprint, codec.Serialize(result), settings.FailureRetention, CancellationToken.None) .ConfigureAwait(false), IdempotencyResultHandling.Release => await store.ReleaseAsync(key, token, CancellationToken.None).ConfigureAwait(false), diff --git a/CSharpEssentials.Idempotency/IdempotencyMode.cs b/CSharpEssentials.Idempotency/IdempotencyMode.cs index 6a3c199..9236d3a 100644 --- a/CSharpEssentials.Idempotency/IdempotencyMode.cs +++ b/CSharpEssentials.Idempotency/IdempotencyMode.cs @@ -9,8 +9,9 @@ public enum IdempotencyMode /// /// The key is reserved and the outcome stored as in , but nothing is blocked: a replay, an /// in-flight key or a reused key is logged as the decision would make, and the operation runs - /// anyway. A record written in this mode is used once the mode becomes . It measures duplicate - /// runs; it does not prevent them. + /// anyway. A record written in this mode is used once the mode becomes ; an + /// writes it in the caller's transaction, so only committed work + /// is replayed. It measures duplicate runs; it does not prevent them. /// Observe = 1, diff --git a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStorePostgresTests.cs b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStorePostgresTests.cs index 465978e..7a1ceab 100644 --- a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStorePostgresTests.cs +++ b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStorePostgresTests.cs @@ -216,7 +216,7 @@ public async Task Observe_Should_Commit_The_Business_Transaction_When_The_Bookke } [Fact] - public async Task Observe_Should_Keep_Its_Record_Out_Of_An_Ambient_Transaction() + public async Task Observe_Should_Leave_No_Record_When_The_Ambient_Transaction_Rolls_Back() { IdempotencyRecordKey key = Key("order-1"); @@ -230,7 +230,7 @@ public async Task Observe_Should_Keep_Its_Record_Out_Of_An_Ambient_Transaction() } (await CountPaymentsAsync(EventId("evt-1"))).Should().Be(0); - (await CountRecordsAsync(key)).Should().Be(1); + (await CountRecordsAsync(key)).Should().Be(0); } [Fact] diff --git a/CSharpEssentials.Tests/Idempotency/IdempotencyExecutorTests.cs b/CSharpEssentials.Tests/Idempotency/IdempotencyExecutorTests.cs index cdce0df..cfdcb36 100644 --- a/CSharpEssentials.Tests/Idempotency/IdempotencyExecutorTests.cs +++ b/CSharpEssentials.Tests/Idempotency/IdempotencyExecutorTests.cs @@ -353,7 +353,7 @@ public async Task ExecuteAsync_Should_Run_In_Observe_Mode_When_The_Store_Fails() } [Fact] - public async Task ExecuteAsync_Should_Use_The_Detached_View_Of_A_Transactional_Store_In_Observe_Mode() + public async Task ExecuteAsync_Should_Reserve_Through_The_Detached_View_And_Store_Through_The_Transactional_Store_In_Observe_Mode() { RecordingStore detached = new(new InMemoryIdempotencyRecordStore(_time)); IdempotencyExecutor executor = new(new TransactionalStore(_store, detached), new TestOptionsMonitor(_options)); @@ -361,6 +361,21 @@ public async Task ExecuteAsync_Should_Use_The_Detached_View_Of_A_Transactional_S await executor.ExecuteAsync(Key, "fp", new IdempotencyExecutionOptions { Mode = IdempotencyMode.Observe }, Operation("done"), Codec); detached.Calls.Should().Be(2); + detached.Releases.Should().Be(1); + _store.Calls.Should().Be(1); + _store.Retentions.Should().ContainSingle(); + } + + [Fact] + public async Task ExecuteAsync_Should_Release_Through_The_Detached_View_Only_In_Observe_Mode() + { + RecordingStore detached = new(new InMemoryIdempotencyRecordStore(_time)); + IdempotencyExecutor executor = new(new TransactionalStore(_store, detached), new TestOptionsMonitor(_options)); + + await executor.ExecuteAsync(Key, "fp", new IdempotencyExecutionOptions { Mode = IdempotencyMode.Observe }, Operation("rolled-back"), Codec); + + detached.Calls.Should().Be(2); + detached.Releases.Should().Be(1); _store.Calls.Should().Be(0); } diff --git a/README.MD b/README.MD index a7bb875..0f507e7 100644 --- a/README.MD +++ b/README.MD @@ -6,7 +6,7 @@ [![Build](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml/badge.svg)](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml) -[![Tests](https://img.shields.io/badge/tests-5385%20passing-brightgreen)](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml) +[![Tests](https://img.shields.io/badge/tests-5386%20passing-brightgreen)](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml) [![NuGet](https://img.shields.io/nuget/v/CSharpEssentials.svg)](https://www.nuget.org/packages/CSharpEssentials) [![Downloads](https://img.shields.io/nuget/dt/CSharpEssentials.svg)](https://www.nuget.org/packages/CSharpEssentials) [![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://github.com/senrecep/CSharpEssentials/blob/main/LICENCE) From e00586a069ca97527241bbbdd95f9daa47f8545a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Recep=20=C5=9Een?= Date: Tue, 20 Oct 2026 11:57:13 +0300 Subject: [PATCH 2/5] test(idempotency): cover observe to enforce rollout after a failed commit Observe with an EF Core store, through the mediator inside the transaction and through the HTTP middleware under a unit of work: a failed commit leaves no record, so the retry under Enforce runs again and leaves one business row. Also covers a committed observe being replayed, a failing result write behind the savepoint, and ambient transactions at read committed and serializable. --- ...fCoreIdempotencyHttpObserveRolloutTests.cs | 112 ++++++++++++ .../EfCoreIdempotencyObserveRolloutTests.cs | 170 ++++++++++++++++++ .../Idempotency/FailingCommitInterceptor.cs | 19 ++ .../FailingRecordWriteInterceptor.cs | 32 ++++ README.MD | 2 +- 5 files changed, 334 insertions(+), 1 deletion(-) create mode 100644 CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyHttpObserveRolloutTests.cs create mode 100644 CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyObserveRolloutTests.cs create mode 100644 CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingCommitInterceptor.cs create mode 100644 CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingRecordWriteInterceptor.cs diff --git a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyHttpObserveRolloutTests.cs b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyHttpObserveRolloutTests.cs new file mode 100644 index 0000000..0ef611b --- /dev/null +++ b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyHttpObserveRolloutTests.cs @@ -0,0 +1,112 @@ +using CSharpEssentials.AspNetCore; +using CSharpEssentials.EntityFrameworkCore.Idempotency; +using CSharpEssentials.Idempotency; +using CSharpEssentials.Tests.AspNetCore.Idempotency; + +using FluentAssertions; + +using Microsoft.AspNetCore.Builder; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Storage; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; + +using HttpResults = Microsoft.AspNetCore.Http.Results; + +namespace CSharpEssentials.Tests.EntityFrameworkCore.Idempotency; + +/// +/// A rollout from to of UseIdempotency +/// on the EF Core store, inside a unit of work that begins a transaction before UseIdempotency and commits it after. +/// +[Collection(IdempotencyPostgresCollection.Name)] +public sealed class EfCoreIdempotencyHttpObserveRolloutTests(IdempotencyPostgresFixture postgres) +{ + private readonly string _eventId = "http-" + Guid.NewGuid().ToString("N"); + private readonly string _key = "key-" + Guid.NewGuid().ToString("N"); + private bool _failCommit; + + [Fact] + public async Task A_Retry_After_The_Switch_To_Enforce_Should_Run_Again_When_The_Observed_Commit_Failed() + { + IConfigurationRoot configuration = Configuration(IdempotencyMode.Observe); + await using IdempotencyTestHost host = await StartAsync(configuration); + + _failCommit = true; + IdempotencyResponse? observed = await SendAsync(host); + int paymentsAfterFailure = await CountPaymentsAsync(); + _failCommit = false; + SetMode(configuration, IdempotencyMode.Enforce); + IdempotencyResponse? retried = await SendAsync(host); + + host.Probe.Executions.Should().Be(2); + retried!.Replayed.Should().BeFalse(); + retried.StatusCode.Should().Be(200); + observed?.StatusCode.Should().NotBe(200); + paymentsAfterFailure.Should().Be(0); + (await CountPaymentsAsync()).Should().Be(1); + } + + [Fact] + public async Task A_Retry_After_The_Switch_To_Enforce_Should_Replay_What_Observe_Committed() + { + IConfigurationRoot configuration = Configuration(IdempotencyMode.Observe); + await using IdempotencyTestHost host = await StartAsync(configuration); + + IdempotencyResponse? observed = await SendAsync(host); + SetMode(configuration, IdempotencyMode.Enforce); + IdempotencyResponse? retried = await SendAsync(host); + + observed!.StatusCode.Should().Be(200); + retried!.Replayed.Should().BeTrue(); + retried.Body.Should().Be(observed.Body); + host.Probe.Executions.Should().Be(1); + (await CountPaymentsAsync()).Should().Be(1); + } + + private static IConfigurationRoot Configuration(IdempotencyMode mode) => new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary { ["Mode"] = mode.ToString() }) + .Build(); + + private static void SetMode(IConfigurationRoot configuration, IdempotencyMode mode) + { + configuration["Mode"] = mode.ToString(); + configuration.Reload(); + } + + private Task StartAsync(IConfigurationRoot configuration) => IdempotencyTestHost.StartAsync( + configureServices: services => + { + services.Configure(configuration); + services.AddDbContext(options => options.UseNpgsql(postgres.ConnectionString)); + services.AddEfCoreIdempotencyRecordStore(); + }, + configurePipeline: app => app.Use(async (context, next) => + { + IdempotencyTestDbContext db = context.RequestServices.GetRequiredService(); + await using IDbContextTransaction transaction = await db.Database.BeginTransactionAsync(); + await next(context); + if (_failCommit) + throw new InvalidOperationException("The commit failed."); + await transaction.CommitAsync(); + }), + configureEndpoints: app => app.MapPost("/payments", async (IdempotencyTestDbContext db, IdempotencyProbe probe) => + { + int execution = probe.Execute(); + db.Payments.Add(new IdempotencyPayment { EventId = _eventId, Amount = 10 }); + await db.SaveChangesAsync(); + return HttpResults.Ok(new { execution }); + }).WithIdempotency()); + + private async Task SendAsync(IdempotencyTestHost host) + { + using HttpRequestMessage request = IdempotencyTestHost.CreateRequest("/payments", _key); + return await host.SendSettledAsync(request); + } + + private async Task CountPaymentsAsync() + { + await using IdempotencyTestDbContext context = postgres.CreateContext(); + return await context.Payments.CountAsync(payment => payment.EventId == _eventId); + } +} diff --git a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyObserveRolloutTests.cs b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyObserveRolloutTests.cs new file mode 100644 index 0000000..1f00971 --- /dev/null +++ b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyObserveRolloutTests.cs @@ -0,0 +1,170 @@ +using System.Transactions; + +using CSharpEssentials.EntityFrameworkCore.Idempotency; +using CSharpEssentials.EntityFrameworkCore.Transactions; +using CSharpEssentials.Idempotency; +using CSharpEssentials.Mediator; +using CSharpEssentials.ResultPattern; +using CSharpEssentials.Tests.Idempotency; + +using FluentAssertions; + +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; + +namespace CSharpEssentials.Tests.EntityFrameworkCore.Idempotency; + +/// +/// A rollout from to with the EF Core store +/// and : what Observe records must be exactly what committed. +/// +[Collection(IdempotencyPostgresCollection.Name)] +public sealed class EfCoreIdempotencyObserveRolloutTests(IdempotencyPostgresFixture postgres) : IDisposable +{ + private readonly ServiceProvider _services = postgres.CreateServices(); + private readonly string _account = "acct-" + Guid.NewGuid().ToString("N"); + private readonly TestOptionsMonitor _monitor = new(new IdempotencyExecutorOptions { Mode = IdempotencyMode.Observe }); + private int _handled; + + public void Dispose() => _services.Dispose(); + + [Fact] + public async Task A_Retry_After_The_Switch_To_Enforce_Should_Run_Again_When_The_Observed_Commit_Failed() + { + PaymentWebhook webhook = new(_account + ":evt-1", _account, 25, Decline: false); + + Func observed = () => DeliverAsync(webhook, new FailingCommitInterceptor()); + await observed.Should().ThrowAsync().WithMessage("The commit failed."); + int paymentsAfterFailure = await CountPaymentsAsync(webhook.EventId); + int recordsAfterFailure = await CountRecordsAsync(webhook); + _monitor.CurrentValue = new IdempotencyExecutorOptions { Mode = IdempotencyMode.Enforce }; + Result retried = await DeliverAsync(webhook); + + _handled.Should().Be(2); + retried.IsSuccess.Should().BeTrue(); + paymentsAfterFailure.Should().Be(0); + recordsAfterFailure.Should().Be(0); + (await CountPaymentsAsync(webhook.EventId)).Should().Be(1); + (await CountRecordsAsync(webhook)).Should().Be(1); + } + + [Fact] + public async Task A_Retry_After_The_Switch_To_Enforce_Should_Replay_What_Observe_Committed() + { + PaymentWebhook webhook = new(_account + ":evt-1", _account, 25, Decline: false); + + Result observed = await DeliverAsync(webhook); + _monitor.CurrentValue = new IdempotencyExecutorOptions { Mode = IdempotencyMode.Enforce }; + Result retried = await DeliverAsync(webhook); + + observed.IsSuccess.Should().BeTrue(); + retried.IsSuccess.Should().BeTrue(); + retried.Value.Should().Be(observed.Value); + _handled.Should().Be(1); + (await CountPaymentsAsync(webhook.EventId)).Should().Be(1); + (await CountRecordsAsync(webhook)).Should().Be(1); + } + + [Fact] + public async Task Observe_Should_Commit_The_Business_When_The_Result_Write_Fails() + { + PaymentWebhook webhook = new(_account + ":evt-1", _account, 25, Decline: false); + + Result observed = await DeliverAsync(webhook, new FailingRecordWriteInterceptor()); + + observed.IsSuccess.Should().BeTrue(); + _handled.Should().Be(1); + (await CountPaymentsAsync(webhook.EventId)).Should().Be(1); + (await CountRecordsAsync(webhook)).Should().Be(0); + } + + [Fact] + public async Task Observe_Should_Store_The_Result_In_A_Read_Committed_Ambient_Transaction() + { + PaymentWebhook webhook = new(_account + ":evt-1", _account, 25, Decline: false); + + Result observed = await DeliverAsync(webhook, ambient: IsolationLevel.ReadCommitted); + + observed.IsSuccess.Should().BeTrue(); + (await CountPaymentsAsync(webhook.EventId)).Should().Be(1); + (await CountRecordsAsync(webhook)).Should().Be(1); + } + + [Fact] + public async Task Observe_Should_Commit_The_Ambient_Transaction_When_The_Result_Write_Fails() + { + PaymentWebhook webhook = new(_account + ":evt-1", _account, 25, Decline: false); + + Result observed = await DeliverAsync(webhook, new FailingRecordWriteInterceptor(), IsolationLevel.ReadCommitted); + + observed.IsSuccess.Should().BeTrue(); + _handled.Should().Be(1); + (await CountPaymentsAsync(webhook.EventId)).Should().Be(1); + (await CountRecordsAsync(webhook)).Should().Be(0); + } + + [Fact] + public async Task Observe_Should_Commit_A_Serializable_Ambient_Transaction_Without_The_Result() + { + PaymentWebhook webhook = new(_account + ":evt-1", _account, 25, Decline: false); + + Result observed = await DeliverAsync(webhook, ambient: IsolationLevel.Serializable); + + observed.IsSuccess.Should().BeTrue(); + (await CountPaymentsAsync(webhook.EventId)).Should().Be(1); + (await CountRecordsAsync(webhook)).Should().Be(0); + } + + private async Task> DeliverAsync(PaymentWebhook webhook, IInterceptor? interceptor = null, IsolationLevel? ambient = null) + { + using TransactionScope? scope = ambient is { } isolation + ? new(TransactionScopeOption.Required, new TransactionOptions { IsolationLevel = isolation }, TransactionScopeAsyncFlowOption.Enabled) + : null; + DbContextOptionsBuilder options = new DbContextOptionsBuilder().UseNpgsql(postgres.ConnectionString); + if (interceptor is not null) + options.AddInterceptors(interceptor); + await using IdempotencyTestDbContext context = new(options.Options); + IdempotencyExecutor executor = new( + new EfCoreIdempotencyRecordStore(context, _services.GetRequiredService()), + _monitor); + TransactionBehavior> transaction = new(new EfCoreTransactionRunner(context)); + IdempotencyBehavior> idempotency = new( + executor, + _monitor, + new IdempotencyBehaviorOptions { Placement = IdempotencyPlacement.InsideTransaction }, + NullLogger>>.Instance); + + Result result = scope is not null + ? await idempotency.Handle(webhook, (request, ct) => HandleAsync(context, request, ct), CancellationToken.None) + : await transaction.Handle( + webhook, + (message, token) => idempotency.Handle(message, (request, ct) => HandleAsync(context, request, ct), token), + CancellationToken.None); + scope?.Complete(); + return result; + } + + private async ValueTask> HandleAsync(IdempotencyTestDbContext context, PaymentWebhook webhook, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _handled); + IdempotencyPayment payment = new() { EventId = webhook.EventId, Amount = webhook.Amount }; + context.Payments.Add(payment); + await context.SaveChangesAsync(cancellationToken); + return payment.Id; + } + + private async Task CountPaymentsAsync(string eventId) + { + await using IdempotencyTestDbContext context = postgres.CreateContext(); + return await context.Payments.CountAsync(payment => payment.EventId == eventId); + } + + private async Task CountRecordsAsync(PaymentWebhook webhook) + { + string key = ((IIdempotentRequest)webhook).IdempotencyKey; + await using IdempotencyTestDbContext context = postgres.CreateContext(); + return await context.IdempotencyRecords.CountAsync(record => record.Key == key); + } +} diff --git a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingCommitInterceptor.cs b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingCommitInterceptor.cs new file mode 100644 index 0000000..66f3a16 --- /dev/null +++ b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingCommitInterceptor.cs @@ -0,0 +1,19 @@ +using System.Data.Common; + +using Microsoft.EntityFrameworkCore.Diagnostics; + +namespace CSharpEssentials.Tests.EntityFrameworkCore.Idempotency; + +/// +/// Fails every commit of the context it is added to, as a database that rejects the commit would: nothing is committed +/// and the transaction rolls back when it is disposed. +/// +internal sealed class FailingCommitInterceptor : DbTransactionInterceptor +{ + public override InterceptionResult TransactionCommitting(DbTransaction transaction, TransactionEventData eventData, InterceptionResult result) => + throw new InvalidOperationException("The commit failed."); + + public override ValueTask TransactionCommittingAsync( + DbTransaction transaction, TransactionEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) => + throw new InvalidOperationException("The commit failed."); +} diff --git a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingRecordWriteInterceptor.cs b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingRecordWriteInterceptor.cs new file mode 100644 index 0000000..d8ec68c --- /dev/null +++ b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingRecordWriteInterceptor.cs @@ -0,0 +1,32 @@ +using System.Data.Common; + +using Microsoft.EntityFrameworkCore.Diagnostics; + +namespace CSharpEssentials.Tests.EntityFrameworkCore.Idempotency; + +/// +/// Turns every write of the context it is added to that touches the idempotency records into a statement that fails on +/// the server, as a lock timeout or any other database error would: on PostgreSQL the failure aborts the transaction the +/// write runs in. +/// +internal sealed class FailingRecordWriteInterceptor : DbCommandInterceptor +{ + public override InterceptionResult NonQueryExecuting(DbCommand command, CommandEventData eventData, InterceptionResult result) + { + Fail(command); + return result; + } + + public override ValueTask> NonQueryExecutingAsync( + DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + Fail(command); + return ValueTask.FromResult(result); + } + + private static void Fail(DbCommand command) + { + if (command.CommandText.Contains("IdempotencyRecords", StringComparison.Ordinal)) + command.CommandText = "SELECT 1/0"; + } +} diff --git a/README.MD b/README.MD index 0f507e7..be5738f 100644 --- a/README.MD +++ b/README.MD @@ -6,7 +6,7 @@ [![Build](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml/badge.svg)](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml) -[![Tests](https://img.shields.io/badge/tests-5386%20passing-brightgreen)](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml) +[![Tests](https://img.shields.io/badge/tests-5394%20passing-brightgreen)](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml) [![NuGet](https://img.shields.io/nuget/v/CSharpEssentials.svg)](https://www.nuget.org/packages/CSharpEssentials) [![Downloads](https://img.shields.io/nuget/dt/CSharpEssentials.svg)](https://www.nuget.org/packages/CSharpEssentials) [![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://github.com/senrecep/CSharpEssentials/blob/main/LICENCE) From 8500ce7052d72db404cda7b582ce07cfa9f6939f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Recep=20=C5=9Een?= Date: Tue, 20 Oct 2026 11:57:13 +0300 Subject: [PATCH 3/5] docs(idempotency): describe observe results inside the business transaction --- .../csharpessentials-efcore/SKILL.md | 2 +- CSharpEssentials.AspNetCore/Readme.MD | 4 +- .../Readme.MD | 2 +- CSharpEssentials.Idempotency/Readme.MD | 39 +++++++++++++++---- docs/API_REFERENCE.md | 2 +- 5 files changed, 36 insertions(+), 13 deletions(-) diff --git a/.well-known/agent-skills/csharpessentials-efcore/SKILL.md b/.well-known/agent-skills/csharpessentials-efcore/SKILL.md index d3ae5b5..abc83ca 100644 --- a/.well-known/agent-skills/csharpessentials-efcore/SKILL.md +++ b/.well-known/agent-skills/csharpessentials-efcore/SKILL.md @@ -327,7 +327,7 @@ builder.Services.AddEfCoreIdempotencyRecordStore(); // builder.Services.AddEfCoreIdempotencyRecordStore("payments"); // or: keyed store for profiles with StoreKey = "payments" ``` -Records are written with SQL inside the context's current transaction, so a rolled-back request leaves no record and its retry runs again; a concurrent duplicate waits for the first commit and replays its response. The primary key is `KeyHash`, a `binary(32)` SHA-256 of the storage key, so keys that differ in case or trailing spaces stay separate under any collation; the readable `Namespace`/`Scope`/`Key` columns get the provider's binary collation (`KeyCollation` overrides it, e.g. `utf8mb4_nopad_bin` on MariaDB), which only makes them case-sensitive: SQL Server ignores trailing spaces under every collation, `_BIN2` included, so only the `KeyHash` key separates them there (on PostgreSQL never turn the columns into `char(n)`, which pads). Needs `READ COMMITTED`. In `Observe` mode the executor and the HTTP middleware use `store.Detached` (own context and connection, not atomic with the business data). Expired rows count as absent; call `PurgeExpiredAsync` from a hosted service to delete them. +Records are written with SQL inside the context's current transaction, so a rolled-back request leaves no record and its retry runs again; a concurrent duplicate waits for the first commit and replays its response. The primary key is `KeyHash`, a `binary(32)` SHA-256 of the storage key, so keys that differ in case or trailing spaces stay separate under any collation; the readable `Namespace`/`Scope`/`Key` columns get the provider's binary collation (`KeyCollation` overrides it, e.g. `utf8mb4_nopad_bin` on MariaDB), which only makes them case-sensitive: SQL Server ignores trailing spaces under every collation, `_BIN2` included, so only the `KeyHash` key separates them there (on PostgreSQL never turn the columns into `char(n)`, which pads). Needs `READ COMMITTED`. In `Observe` mode the executor and the HTTP middleware reserve and release through `store.Detached` (own context and connection) and write a result to store through the store itself, in the business transaction, so a failed commit leaves no record and is not replayed after the switch to `Enforce`; a failed result write rolls back to a savepoint and the business data commits without the record (always so under an isolation level stricter than `READ COMMITTED`). Expired rows count as absent; call `PurgeExpiredAsync` from a hosted service to delete them. --- diff --git a/CSharpEssentials.AspNetCore/Readme.MD b/CSharpEssentials.AspNetCore/Readme.MD index aed0b9e..3f66716 100644 --- a/CSharpEssentials.AspNetCore/Readme.MD +++ b/CSharpEssentials.AspNetCore/Readme.MD @@ -272,8 +272,8 @@ app.MapPost("/payments", Pay).WithIdempotency(o => o.Mode = IdempotencyMode.Enfo - The options are read through `IOptionsMonitor` once per request, so a reload of a bound configuration section changes `Mode`, or any other option, for the next request without a restart. Endpoint settings are resolved again after a reload. The store is chosen in `AddIdempotency`; a reload does not change it. A reload that makes the options invalid throws `OptionsValidationException` on every request to an idempotent endpoint until it is fixed. `IdempotencyOptions` resolved from DI is the current value. - The mode comes from the endpoint, else its profile, else `IdempotencyOptions.Mode`. It is separate from `IdempotencyExecutorOptions.Mode`, which the Mediator behavior and message consumers read. -- After a switch from `Observe` to `Enforce`, the responses stored in `Observe` are replayed and a key still in flight gets 409. -- `Observe` measures duplicates; it does not prevent them. Keep it short for endpoints that move money. A record that `Observe` stored for a request whose business commit then failed is replayed after the switch to `Enforce`: the [Idempotency Readme](../CSharpEssentials.Idempotency/Readme.MD#rollout-observe--enforce) has the rollout steps and this residual risk. +- After a switch from `Observe` to `Enforce`, the responses stored in `Observe` are replayed and a key still in flight gets 409. With a store that writes in the request's transaction (the EF Core store) and a transaction open around `UseIdempotency`, a response is stored in that transaction, so a response whose commit failed is never replayed and its retry runs again. +- `Observe` measures duplicates; it does not prevent them. Keep it short for endpoints that move money. A store that cannot join the request's transaction (in-memory, distributed cache, Redis) stores the response before a transaction open around `UseIdempotency` commits, in `Observe` and `Enforce` alike, so a response whose commit then failed is replayed: the [Idempotency Readme](../CSharpEssentials.Idempotency/Readme.MD#rollout-observe--enforce) has the rollout steps and what remains in `Observe`. - In `Observe`, a scope resolver, keyed store or fingerprint that throws is logged as `Idempotency.Observe.ResolutionFailed`, and a store that throws while reserving as `Idempotency.Observe.ReserveFailed`; the request then runs once without idempotency. In `Enforce`, or when the client aborted, the exception propagates and the handler does not run. - Decisions are counted on `idempotency.decisions` (meter `CSharpEssentials.Idempotency`) with `source` `http` and the route template as `target`. Logs carry a key hash, never the key or the scope. See the [Idempotency Readme](../CSharpEssentials.Idempotency/Readme.MD#telemetry) for the tags and the log events. diff --git a/CSharpEssentials.EntityFrameworkCore/Readme.MD b/CSharpEssentials.EntityFrameworkCore/Readme.MD index a536ace..1f054f7 100644 --- a/CSharpEssentials.EntityFrameworkCore/Readme.MD +++ b/CSharpEssentials.EntityFrameworkCore/Readme.MD @@ -614,7 +614,7 @@ Add a migration after `AddIdempotencyRecords`: it maps `IdempotencyRecord` to `I - **Isolation.** The store needs `READ COMMITTED` (the default of PostgreSQL, SQL Server and MySQL). Under `REPEATABLE READ` or `SERIALIZABLE` a concurrent reservation fails with a serialization error. - **Set-based writes.** Records are written with SQL, never through the change tracker, so `SaveChanges`, interceptors and query filters do not see them. - **Hybrid profiles.** `AddEfCoreIdempotencyRecordStore("payments")` registers the store as the keyed store of every profile whose `StoreKey` is `"payments"` and keeps the default store for the rest: `o.AddProfile("payments", p => p.StoreKey = "payments")`. -- **Observe.** In `Observe` the executor and the `Idempotency-Key` middleware write through `Detached`, which resolves a new `TContext` in a scope of its own for every call and suppresses any ambient `TransactionScope`. Observing then never waits for a concurrent request and a failing bookkeeping write cannot fail the business transaction. The records are not atomic with the business data in this mode: a commit that fails after the operation completed leaves a stored response, which a request replays after the switch to `Enforce`. The context must not depend on state that only the request scope has. +- **Observe.** In `Observe` the executor and the `Idempotency-Key` middleware reserve and release through `Detached`, which resolves a new `TContext` in a scope of its own for every call and suppresses any ambient `TransactionScope`. A reservation then never waits for a concurrent request and a failing reservation cannot fail the business transaction. A result to store releases the detached reservation and is written through the store itself, in the business transaction, so it commits and rolls back with the business data as in `Enforce`: a commit that fails leaves no record, and its retry runs again after the switch to `Enforce`. From that write until the commit a reservation of the same key waits, as in `Enforce`. A database error on that write is rolled back to a savepoint (of the context's transaction, or on PostgreSQL of an ambient `TransactionScope`), so the business data commits without the record. Under an isolation level stricter than `READ COMMITTED` (`TransactionScope` defaults to `Serializable`) the write always fails this way and `Observe` records nothing. The context must not depend on state that only the request scope has. Expired records stay in the table until `PurgeExpiredAsync` deletes them in one statement; an expired record already counts as absent, so purging only keeps the table small: diff --git a/CSharpEssentials.Idempotency/Readme.MD b/CSharpEssentials.Idempotency/Readme.MD index cc59f2c..54133fa 100644 --- a/CSharpEssentials.Idempotency/Readme.MD +++ b/CSharpEssentials.Idempotency/Readme.MD @@ -112,7 +112,7 @@ services.Configure(configuration.GetSection("Idempot - The executor reads `IOptionsMonitor` on every call, so a reload of a bound configuration section changes the mode, globally or per profile, without a restart. A reload that makes the options invalid throws `OptionsValidationException` from every call until it is fixed. The store is the one registered at startup. - The mode comes from the operation (`IdempotencyExecutionOptions.Mode`), else its profile, else `IdempotencyExecutorOptions.Mode`. -- After a switch from `Observe` to `Enforce`, a result stored in `Observe` is replayed and a key still in flight is answered `InFlight`. +- After a switch from `Observe` to `Enforce`, a result stored in `Observe` is replayed and a key still in flight is answered `InFlight`. With a store that writes in the business transaction (`ITransactionalIdempotencyRecordStore`, such as the EF Core store) only results whose transaction committed are stored, so a retry of work that rolled back runs again. - `Observe` measures duplicates; it does not prevent them. Keep it short for operations that move money. - In `Observe` a store that throws while reserving is logged (`Idempotency.Observe.ReserveFailed`) and the operation runs once without idempotency, unless the caller cancelled. In `Enforce` the exception propagates. - The `Idempotency-Key` middleware of `CSharpEssentials.AspNetCore` has its own switch, `IdempotencyOptions.Mode`. The Mediator behavior and `IdempotencyInbox` use this one. @@ -123,10 +123,32 @@ To add idempotency to an operation that already has callers: 2. Read the [telemetry](#telemetry). `key_reused` shows callers that reuse a key for different content, `in_flight` shows concurrent duplicates, `replay` shows how often a retry would have been answered from the store, and `unscoped` shows callers that send a key without a resolvable scope (the log events name the endpoint or request type, never the key). 3. Fix what the numbers show (a client that reuses keys, a machine caller that needs `IdempotencyScopes.Client`), then switch to `Enforce` through configuration. -**Residual risk.** `Observe` records are real records. When the business transaction fails after the operation completed -and its record was written, for example because the commit failed, the record stays, and after the switch to `Enforce` a -retry with that key replays a result for work that never committed. With the EF Core store this is the one case where the -record and the data are not atomic: in `Observe` it writes through the `Detached` view (see [EF Core store](#ef-core-store)). +**Transactions in `Observe`.** `Observe` records are real records. With an `ITransactionalIdempotencyRecordStore` (the +EF Core store) the executor and the HTTP middleware reserve the key through the `Detached` view, so the reservation never +waits for or fails the business transaction. When the operation returns a result to store, they release that reservation +and write the result through the store itself, in the business transaction (the Mediator transaction with +`IdempotencyPlacement.InsideTransaction`, or the one open around `UseIdempotency`). The record then commits and rolls back +with the business data, as in `Enforce`: when the commit fails, nothing is recorded and a retry after the switch to +`Enforce` runs the operation again. What remains: + +- A store that cannot join the business transaction (in-memory, distributed cache, Redis) stores the result before the + commit when the transaction wraps the idempotency layer, in `Observe` and `Enforce` alike. A commit that then fails + leaves a record, and a retry replays a result for work that never committed. Use the EF Core store, or put the + idempotency layer outside the transaction, for work whose commit can fail after the operation returned. +- Between the release of the detached reservation and the write in the business transaction the key is absent for one + round trip. A concurrent duplicate that arrives then is reserved: in `Observe` it is counted `reserved` instead of + `in_flight`, and on an instance that already runs `Enforce` (a rolling switch) it runs. +- From that write until the commit, a reservation of the same key waits for the business transaction, as in `Enforce`. + An operation that reuses the key of an operation it already completed in the same transaction waits until its command + times out, then runs without idempotency (`Idempotency.Observe.ReserveFailed`). +- A database error on that write (a lock or statement timeout, for example) is rolled back to a savepoint, so the + business data still commits, without the record, and a retry after the switch to `Enforce` runs again. The EF Core + store takes the savepoint in the context's transaction, and on PostgreSQL also in an ambient `TransactionScope`; on + other providers an error that aborts an ambient transaction takes the business data with it. Under an isolation level + stricter than `READ COMMITTED` (`TransactionScope` defaults to `Serializable`) the write always fails this way, + because the release changed the row after the transaction's snapshot, so `Observe` records nothing: run the business + transaction at `READ COMMITTED`, which the store needs anyway. + Keep `Observe` short for operations that move money. ### Lease and retry budget @@ -405,9 +427,10 @@ services.AddEfCoreIdempotencyRecordStore(); them, for example from a hosted service. - `AddEfCoreIdempotencyRecordStore("payments")` registers it as a keyed store, so only the profiles with `StoreKey = "payments"` use the database and the others keep the default store. -- In `Observe` the executor and the HTTP middleware write through its `Detached` view, in a context and transaction of - their own, so observing does not wait for or fail the business transaction. Those records are not atomic with the - business data. +- In `Observe` the executor and the HTTP middleware reserve and release through its `Detached` view, in a context and + transaction of their own, so a reservation does not wait for or fail the business transaction. A result to store is + written in the business transaction, so it commits and rolls back with the business data as in `Enforce` (see + [Transactions in `Observe`](#rollout-observe--enforce)). The [EF Core Readme](../CSharpEssentials.EntityFrameworkCore/Readme.MD#idempotency-records) has the details and a purge service. diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md index 54a590c..57f9d86 100644 --- a/docs/API_REFERENCE.md +++ b/docs/API_REFERENCE.md @@ -1326,7 +1326,7 @@ Namespace `CSharpEssentials.EntityFrameworkCore.Idempotency`; depends on `CSharp | `IdempotencyRecord.KeyHash` | `binary(32)`: SHA-256 of the UTF-8 length-prefixed storage key. Uniqueness does not depend on the collation, so keys that differ in case or trailing spaces stay separate under a case-insensitive or PAD SPACE collation. The store compares the readable columns ordinally after each read and throws `InvalidOperationException` for a row of another key | | `IdempotencyRecordModelOptions` | `TableName` (`IdempotencyRecords`), `Schema`, `KeyCollation` of the readable columns (`null`: `C` on PostgreSQL, `Latin1_General_100_BIN2` on SQL Server, `utf8mb4_0900_bin` on MySQL, `BINARY` on SQLite, the database default elsewhere; set `utf8mb4_nopad_bin` on MariaDB). It only makes the columns case-sensitive: SQL Server ignores trailing spaces under every collation, `_BIN2` included, so `KeyHash` is what keeps keys that differ by trailing spaces apart. Keep the columns `text` or `varchar` on PostgreSQL (`char(n)` pads) | | `EfCoreIdempotencyRecordStore(context, scopeFactory, timeProvider?)` | `ITransactionalIdempotencyRecordStore` that writes with set-based SQL on the context's connection, inside its current transaction. Reserve: `INSERT … ON CONFLICT DO NOTHING` (PostgreSQL, SQLite) or a plain `INSERT` behind a savepoint, then a read; a concurrent reservation waits for the first transaction. Complete and release are compare-and-set on the token; an expired row is taken over. Needs `READ COMMITTED` | -| `EfCoreIdempotencyRecordStore.Detached` | The same records from a new `TContext` per call in a scope of its own, with the ambient `TransactionScope` suppressed; used in `Observe` mode by the executor and the HTTP middleware | +| `EfCoreIdempotencyRecordStore.Detached` | The same records from a new `TContext` per call in a scope of its own, with the ambient `TransactionScope` suppressed; in `Observe` mode the executor and the HTTP middleware reserve and release through it, and write a result to store through the store itself, in the business transaction, behind a savepoint so a failed write leaves the business data committable | | `EfCoreIdempotencyRecordStore.PurgeExpiredAsync(ct)` | Deletes expired records in one statement and returns the count | | `AddEfCoreIdempotencyRecordStore(serviceKey?)` | Registers the store scoped; without a key it replaces the default `IIdempotencyRecordStore` (in any order with `AddIdempotencyExecutor`), with a key it is the keyed store of the profiles whose `StoreKey` matches. An empty key throws `ArgumentException` | From a8a5d378bc7e72bcbe901c369076f6b3f55ad0de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Recep=20=C5=9Een?= Date: Tue, 20 Oct 2026 12:16:33 +0300 Subject: [PATCH 4/5] fix(efcore): keep the original error and skip the savepoint without a transaction A failing rollback to the result savepoint no longer replaces the database error that caused it, and a PostgreSQL connection that is not enlisted in the ambient transaction writes the result without a savepoint instead of failing. Cover the savepoint path on SQLite, and assert the aborted request explicitly. --- .../EfCoreIdempotencyRecordStore.cs | 46 ++++++++++++---- ...fCoreIdempotencyHttpObserveRolloutTests.cs | 2 +- ...CoreIdempotencyRecordStorePostgresTests.cs | 16 ++++++ ...EfCoreIdempotencyRecordStoreSqliteTests.cs | 53 +++++++++++++++++++ .../FailingRecordWriteInterceptor.cs | 2 +- README.MD | 2 +- 6 files changed, 108 insertions(+), 13 deletions(-) diff --git a/CSharpEssentials.EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStore.cs b/CSharpEssentials.EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStore.cs index f1c3abf..b3a8e8d 100644 --- a/CSharpEssentials.EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStore.cs +++ b/CSharpEssentials.EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStore.cs @@ -43,10 +43,14 @@ namespace CSharpEssentials.EntityFrameworkCore.Idempotency; /// transaction, so it commits and rolls back with the business data as in . /// A result for a token this instance did not reserve, as in Observe, is written behind a savepoint of the context's /// current transaction (on PostgreSQL also of an ambient System.Transactions transaction the connection is -/// enlisted in): a database error rolls back to the savepoint and is rethrown, so the transaction stays usable and the -/// business data commits without the record. Observe therefore stores results only under READ COMMITTED: under a -/// stricter level the write fails with a serialization error, because the release changed the row after the -/// transaction's snapshot was taken. +/// enlisted in; when the connection is not enlisted, the result is written without a savepoint): a database error rolls +/// back to the savepoint and is rethrown, so the transaction stays usable and the business data commits without the +/// record. Two limits remain. On PostgreSQL, Observe stores results only under READ COMMITTED: under a stricter +/// level the write fails with a serialization error, because the release changed the row after the transaction's +/// snapshot was taken; other providers may succeed. On SQL Server a savepoint cannot save a transaction that the server +/// has doomed, which happens when the connection has XACT_ABORT ON or under SNAPSHOT isolation after an +/// update conflict (error 3960): the rollback to the savepoint fails, the transaction is lost with the business data, +/// and the original error is thrown. /// /// /// The context that maps . @@ -192,7 +196,15 @@ public async ValueTask CompleteAsync( if (!savepoint) return await CompleteCoreAsync(key, token, fingerprint, payload, retention, cancellationToken).ConfigureAwait(false); - await CreateResultSavepointAsync(transaction, cancellationToken).ConfigureAwait(false); + try + { + await CreateResultSavepointAsync(transaction, cancellationToken).ConfigureAwait(false); + } + catch (DbException) when (transaction is null) + { + return await CompleteCoreAsync(key, token, fingerprint, payload, retention, cancellationToken).ConfigureAwait(false); + } + bool completed; try { @@ -200,7 +212,7 @@ public async ValueTask CompleteAsync( } catch (Exception exception) when (exception is DbException || exception.InnerException is DbException) { - await RollbackToResultSavepointAsync(transaction).ConfigureAwait(false); + _ = await TryRollbackToResultSavepointAsync(transaction).ConfigureAwait(false); throw; } @@ -227,10 +239,24 @@ transaction is null ? _context.Database.ExecuteSqlRawAsync("SAVEPOINT " + ResultSavepointName, cancellationToken) : transaction.CreateSavepointAsync(ResultSavepointName, cancellationToken); - private Task RollbackToResultSavepointAsync(IDbContextTransaction? transaction) => - transaction is null - ? _context.Database.ExecuteSqlRawAsync("ROLLBACK TO SAVEPOINT " + ResultSavepointName, CancellationToken.None) - : transaction.RollbackToSavepointAsync(ResultSavepointName, CancellationToken.None); + /// + /// Rolls back to the result savepoint; when the rollback itself failed. The failure is not + /// thrown, so that the caller rethrows the database error that made the write fail instead of the rollback's. + /// + private async Task TryRollbackToResultSavepointAsync(IDbContextTransaction? transaction) + { + try + { + await (transaction is null + ? _context.Database.ExecuteSqlRawAsync("ROLLBACK TO SAVEPOINT " + ResultSavepointName, CancellationToken.None) + : transaction.RollbackToSavepointAsync(ResultSavepointName, CancellationToken.None)).ConfigureAwait(false); + return true; + } + catch (Exception rollbackFailure) when (rollbackFailure is DbException or InvalidOperationException) + { + return false; + } + } private Task ReleaseResultSavepointAsync(IDbContextTransaction? transaction, CancellationToken cancellationToken) => transaction is null diff --git a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyHttpObserveRolloutTests.cs b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyHttpObserveRolloutTests.cs index 0ef611b..e1f7ad5 100644 --- a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyHttpObserveRolloutTests.cs +++ b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyHttpObserveRolloutTests.cs @@ -42,7 +42,7 @@ public async Task A_Retry_After_The_Switch_To_Enforce_Should_Run_Again_When_The_ host.Probe.Executions.Should().Be(2); retried!.Replayed.Should().BeFalse(); retried.StatusCode.Should().Be(200); - observed?.StatusCode.Should().NotBe(200); + observed.Should().BeNull(); paymentsAfterFailure.Should().Be(0); (await CountPaymentsAsync()).Should().Be(1); } diff --git a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStorePostgresTests.cs b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStorePostgresTests.cs index 7a1ceab..19170dc 100644 --- a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStorePostgresTests.cs +++ b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStorePostgresTests.cs @@ -233,6 +233,22 @@ public async Task Observe_Should_Leave_No_Record_When_The_Ambient_Transaction_Ro (await CountRecordsAsync(key)).Should().Be(0); } + [Fact] + public async Task CompleteAsync_Should_Write_Without_A_Savepoint_When_The_Connection_Is_Not_Enlisted_In_The_Ambient_Transaction() + { + IdempotencyRecordKey key = Key("order-1"); + string connectionString = new NpgsqlConnectionStringBuilder(postgres.ConnectionString) { Enlist = false }.ConnectionString; + + using (TransactionScope ambient = new(TransactionScopeAsyncFlowOption.Enabled)) + { + await using IdempotencyTestDbContext context = new(new DbContextOptionsBuilder().UseNpgsql(connectionString).Options); + bool completed = await Store(context).CompleteAsync(key, "detached-token", "fp", "done"u8.ToArray(), TimeSpan.FromHours(1)); + completed.Should().BeTrue(); + } + + (await CountRecordsAsync(key)).Should().Be(1); + } + [Fact] public async Task Profiles_Should_Write_To_The_Store_Their_StoreKey_Names() { diff --git a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStoreSqliteTests.cs b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStoreSqliteTests.cs index 63d1d4b..6910b09 100644 --- a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStoreSqliteTests.cs +++ b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/EfCoreIdempotencyRecordStoreSqliteTests.cs @@ -118,6 +118,59 @@ public async Task CompleteAsync_Should_Insert_The_Record_When_It_Was_Purged() replay.Payload!.Value.ToArray().Should().Equal("done"u8.ToArray()); } + [Fact] + public async Task CompleteAsync_Should_Write_A_Result_Of_Another_Token_Without_A_Transaction() + { + await using IdempotencyTestDbContext context = new(_options); + EfCoreIdempotencyRecordStore store = new(context, ScopeFactory); + + bool completed = await store.CompleteAsync(Key, "detached-token", "fp", "done"u8.ToArray(), Retention); + + completed.Should().BeTrue(); + (await CountAsync(records => records.IdempotencyRecords)).Should().Be(1); + } + + [Fact] + public async Task CompleteAsync_Should_Leave_No_Record_When_The_Commit_After_A_Savepoint_Write_Fails() + { + DbContextOptions options = new DbContextOptionsBuilder() + .UseSqlite(_connection) + .AddInterceptors(new FailingCommitInterceptor()) + .Options; + await using IdempotencyTestDbContext context = new(options); + await using IDbContextTransaction transaction = await context.Database.BeginTransactionAsync(); + EfCoreIdempotencyRecordStore store = new(context, ScopeFactory); + + bool completed = await store.CompleteAsync(Key, "detached-token", "fp", "done"u8.ToArray(), Retention); + Func commit = () => transaction.CommitAsync(); + await commit.Should().ThrowAsync().WithMessage("The commit failed."); + await transaction.RollbackAsync(); + + completed.Should().BeTrue(); + (await CountAsync(records => records.IdempotencyRecords)).Should().Be(0); + } + + [Fact] + public async Task CompleteAsync_Should_Roll_Back_To_The_Savepoint_And_Keep_The_Business_Data_When_The_Write_Fails() + { + DbContextOptions options = new DbContextOptionsBuilder() + .UseSqlite(_connection) + .AddInterceptors(new FailingRecordWriteInterceptor()) + .Options; + await using IdempotencyTestDbContext context = new(options); + await using IDbContextTransaction transaction = await context.Database.BeginTransactionAsync(); + EfCoreIdempotencyRecordStore store = new(context, ScopeFactory); + context.Payments.Add(new IdempotencyPayment { EventId = "evt-1", Amount = 10 }); + await context.SaveChangesAsync(); + + Func complete = () => store.CompleteAsync(Key, "detached-token", "fp", "done"u8.ToArray(), Retention).AsTask(); + await complete.Should().ThrowAsync(); + await transaction.CommitAsync(); + + (await CountAsync(records => records.Payments)).Should().Be(1); + (await CountAsync(records => records.IdempotencyRecords)).Should().Be(0); + } + [Fact] public async Task PurgeExpiredAsync_Should_Delete_Only_The_Expired_Records() { diff --git a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingRecordWriteInterceptor.cs b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingRecordWriteInterceptor.cs index d8ec68c..4be8f36 100644 --- a/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingRecordWriteInterceptor.cs +++ b/CSharpEssentials.Tests/EntityFrameworkCore/Idempotency/FailingRecordWriteInterceptor.cs @@ -27,6 +27,6 @@ public override ValueTask> NonQueryExecutingAsync( private static void Fail(DbCommand command) { if (command.CommandText.Contains("IdempotencyRecords", StringComparison.Ordinal)) - command.CommandText = "SELECT 1/0"; + command.CommandText = "SELECT * FROM cse_missing_table"; } } diff --git a/README.MD b/README.MD index be5738f..65a1d35 100644 --- a/README.MD +++ b/README.MD @@ -6,7 +6,7 @@ [![Build](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml/badge.svg)](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml) -[![Tests](https://img.shields.io/badge/tests-5394%20passing-brightgreen)](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml) +[![Tests](https://img.shields.io/badge/tests-5398%20passing-brightgreen)](https://github.com/senrecep/CSharpEssentials/actions/workflows/build.yml) [![NuGet](https://img.shields.io/nuget/v/CSharpEssentials.svg)](https://www.nuget.org/packages/CSharpEssentials) [![Downloads](https://img.shields.io/nuget/dt/CSharpEssentials.svg)](https://www.nuget.org/packages/CSharpEssentials) [![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://github.com/senrecep/CSharpEssentials/blob/main/LICENCE) From 64d0f9b6ed17df76fb6d7635da5f4c3bf6185d5c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Recep=20=C5=9Een?= Date: Tue, 20 Oct 2026 12:16:33 +0300 Subject: [PATCH 5/5] docs(idempotency): document observe result write limits on sql server and postgresql --- .../csharpessentials-efcore/SKILL.md | 2 +- .../Readme.MD | 2 +- .../ITransactionalIdempotencyRecordStore.cs | 8 +++++- CSharpEssentials.Idempotency/Readme.MD | 27 +++++++++++++------ 4 files changed, 28 insertions(+), 11 deletions(-) diff --git a/.well-known/agent-skills/csharpessentials-efcore/SKILL.md b/.well-known/agent-skills/csharpessentials-efcore/SKILL.md index abc83ca..c18f1ea 100644 --- a/.well-known/agent-skills/csharpessentials-efcore/SKILL.md +++ b/.well-known/agent-skills/csharpessentials-efcore/SKILL.md @@ -327,7 +327,7 @@ builder.Services.AddEfCoreIdempotencyRecordStore(); // builder.Services.AddEfCoreIdempotencyRecordStore("payments"); // or: keyed store for profiles with StoreKey = "payments" ``` -Records are written with SQL inside the context's current transaction, so a rolled-back request leaves no record and its retry runs again; a concurrent duplicate waits for the first commit and replays its response. The primary key is `KeyHash`, a `binary(32)` SHA-256 of the storage key, so keys that differ in case or trailing spaces stay separate under any collation; the readable `Namespace`/`Scope`/`Key` columns get the provider's binary collation (`KeyCollation` overrides it, e.g. `utf8mb4_nopad_bin` on MariaDB), which only makes them case-sensitive: SQL Server ignores trailing spaces under every collation, `_BIN2` included, so only the `KeyHash` key separates them there (on PostgreSQL never turn the columns into `char(n)`, which pads). Needs `READ COMMITTED`. In `Observe` mode the executor and the HTTP middleware reserve and release through `store.Detached` (own context and connection) and write a result to store through the store itself, in the business transaction, so a failed commit leaves no record and is not replayed after the switch to `Enforce`; a failed result write rolls back to a savepoint and the business data commits without the record (always so under an isolation level stricter than `READ COMMITTED`). Expired rows count as absent; call `PurgeExpiredAsync` from a hosted service to delete them. +Records are written with SQL inside the context's current transaction, so a rolled-back request leaves no record and its retry runs again; a concurrent duplicate waits for the first commit and replays its response. The primary key is `KeyHash`, a `binary(32)` SHA-256 of the storage key, so keys that differ in case or trailing spaces stay separate under any collation; the readable `Namespace`/`Scope`/`Key` columns get the provider's binary collation (`KeyCollation` overrides it, e.g. `utf8mb4_nopad_bin` on MariaDB), which only makes them case-sensitive: SQL Server ignores trailing spaces under every collation, `_BIN2` included, so only the `KeyHash` key separates them there (on PostgreSQL never turn the columns into `char(n)`, which pads). Needs `READ COMMITTED`. In `Observe` mode the executor and the HTTP middleware reserve and release through `store.Detached` (own context and connection) and write a result to store through the store itself, in the business transaction, so a failed commit leaves no record and is not replayed after the switch to `Enforce`; a failed result write rolls back to a savepoint and the business data commits without the record (on PostgreSQL always so under an isolation level stricter than `READ COMMITTED`; on SQL Server a transaction doomed by `XACT_ABORT ON` or a `SNAPSHOT` update conflict, error 3960, is lost with the business data, because a savepoint cannot save it). Expired rows count as absent; call `PurgeExpiredAsync` from a hosted service to delete them. --- diff --git a/CSharpEssentials.EntityFrameworkCore/Readme.MD b/CSharpEssentials.EntityFrameworkCore/Readme.MD index 1f054f7..cd7429d 100644 --- a/CSharpEssentials.EntityFrameworkCore/Readme.MD +++ b/CSharpEssentials.EntityFrameworkCore/Readme.MD @@ -614,7 +614,7 @@ Add a migration after `AddIdempotencyRecords`: it maps `IdempotencyRecord` to `I - **Isolation.** The store needs `READ COMMITTED` (the default of PostgreSQL, SQL Server and MySQL). Under `REPEATABLE READ` or `SERIALIZABLE` a concurrent reservation fails with a serialization error. - **Set-based writes.** Records are written with SQL, never through the change tracker, so `SaveChanges`, interceptors and query filters do not see them. - **Hybrid profiles.** `AddEfCoreIdempotencyRecordStore("payments")` registers the store as the keyed store of every profile whose `StoreKey` is `"payments"` and keeps the default store for the rest: `o.AddProfile("payments", p => p.StoreKey = "payments")`. -- **Observe.** In `Observe` the executor and the `Idempotency-Key` middleware reserve and release through `Detached`, which resolves a new `TContext` in a scope of its own for every call and suppresses any ambient `TransactionScope`. A reservation then never waits for a concurrent request and a failing reservation cannot fail the business transaction. A result to store releases the detached reservation and is written through the store itself, in the business transaction, so it commits and rolls back with the business data as in `Enforce`: a commit that fails leaves no record, and its retry runs again after the switch to `Enforce`. From that write until the commit a reservation of the same key waits, as in `Enforce`. A database error on that write is rolled back to a savepoint (of the context's transaction, or on PostgreSQL of an ambient `TransactionScope`), so the business data commits without the record. Under an isolation level stricter than `READ COMMITTED` (`TransactionScope` defaults to `Serializable`) the write always fails this way and `Observe` records nothing. The context must not depend on state that only the request scope has. +- **Observe.** In `Observe` the executor and the `Idempotency-Key` middleware reserve and release through `Detached`, which resolves a new `TContext` in a scope of its own for every call and suppresses any ambient `TransactionScope`. A reservation then never waits for a concurrent request and a failing reservation cannot fail the business transaction. A result to store releases the detached reservation and is written through the store itself, in the business transaction, so it commits and rolls back with the business data as in `Enforce`: a commit that fails leaves no record, and its retry runs again after the switch to `Enforce`. From that write until the commit a reservation of the same key waits; unlike in `Enforce`, that wait can be a cycle across the request's two connections that the database cannot detect, ended only by the command timeout. A database error on that write is rolled back to a savepoint (of the context's transaction, or on PostgreSQL of an ambient `TransactionScope` the connection is enlisted in), so the business data commits without the record. On PostgreSQL, under an isolation level stricter than `READ COMMITTED` (`TransactionScope` defaults to `Serializable`) the write always fails this way and `Observe` records nothing; other providers may succeed. On SQL Server a transaction the server has doomed (`XACT_ABORT ON`, or an update conflict under `SNAPSHOT` isolation, error 3960) cannot be saved by the savepoint: the business data is lost with it and the original error is thrown. The context must not depend on state that only the request scope has. Expired records stay in the table until `PurgeExpiredAsync` deletes them in one statement; an expired record already counts as absent, so purging only keeps the table small: diff --git a/CSharpEssentials.Idempotency/ITransactionalIdempotencyRecordStore.cs b/CSharpEssentials.Idempotency/ITransactionalIdempotencyRecordStore.cs index e24d948..b427ca1 100644 --- a/CSharpEssentials.Idempotency/ITransactionalIdempotencyRecordStore.cs +++ b/CSharpEssentials.Idempotency/ITransactionalIdempotencyRecordStore.cs @@ -11,7 +11,13 @@ namespace CSharpEssentials.Idempotency; /// detached reservation and is written through this store, in the caller's transaction, so Observe records only work /// that committed, as Enforce does: a retry after the switch to never replays a /// result whose transaction rolled back. Between that write and the commit, a reservation of the same key waits for -/// the caller's transaction, as in Enforce. +/// the caller's transaction; in Observe that wait can be a cycle across the caller's two connections that the database +/// cannot detect and only a command timeout ends. +/// +/// +/// In Observe, of this store is therefore called with a token whose +/// reservation has already released. An implementation must write the result anyway and succeed +/// when the entry is gone, as requires. /// /// public interface ITransactionalIdempotencyRecordStore : IIdempotencyRecordStore diff --git a/CSharpEssentials.Idempotency/Readme.MD b/CSharpEssentials.Idempotency/Readme.MD index 54133fa..711a2a9 100644 --- a/CSharpEssentials.Idempotency/Readme.MD +++ b/CSharpEssentials.Idempotency/Readme.MD @@ -138,16 +138,27 @@ with the business data, as in `Enforce`: when the commit fails, nothing is recor - Between the release of the detached reservation and the write in the business transaction the key is absent for one round trip. A concurrent duplicate that arrives then is reserved: in `Observe` it is counted `reserved` instead of `in_flight`, and on an instance that already runs `Enforce` (a rolling switch) it runs. -- From that write until the commit, a reservation of the same key waits for the business transaction, as in `Enforce`. - An operation that reuses the key of an operation it already completed in the same transaction waits until its command - times out, then runs without idempotency (`Idempotency.Observe.ReserveFailed`). +- From that write until the commit, a reservation of the same key waits for the business transaction. In `Enforce` that + wait is a plain lock wait the database can see; in `Observe` it can be a cycle the database cannot detect: request B + holds its business transaction while its detached reservation, on a second connection, waits for the uncommitted + result row of request A, and one edge of the cycle lives in the application across B's two connections. Only the + command timeout ends it, and B then runs without idempotency (`Idempotency.Observe.ReserveFailed`). An operation that + reuses the key of an operation it already completed in the same transaction waits the same way. - A database error on that write (a lock or statement timeout, for example) is rolled back to a savepoint, so the business data still commits, without the record, and a retry after the switch to `Enforce` runs again. The EF Core - store takes the savepoint in the context's transaction, and on PostgreSQL also in an ambient `TransactionScope`; on - other providers an error that aborts an ambient transaction takes the business data with it. Under an isolation level - stricter than `READ COMMITTED` (`TransactionScope` defaults to `Serializable`) the write always fails this way, - because the release changed the row after the transaction's snapshot, so `Observe` records nothing: run the business - transaction at `READ COMMITTED`, which the store needs anyway. + store takes the savepoint in the context's transaction, and on PostgreSQL also in an ambient `TransactionScope` whose + transaction the connection is enlisted in; on other providers an error that aborts an ambient transaction takes the + business data with it. On PostgreSQL, under an isolation level stricter than `READ COMMITTED` (`TransactionScope` + defaults to `Serializable`) the write always fails this way, because the release changed the row after the + transaction's snapshot, so `Observe` records nothing: run the business transaction at `READ COMMITTED`, which the + store needs anyway. Other providers may succeed under a stricter level; that is not verified. +- On SQL Server a savepoint cannot save a transaction the server has doomed: with `XACT_ABORT ON` on the connection, or + under `SNAPSHOT` isolation after an update conflict (error 3960), the failed write dooms the whole transaction, the + rollback to the savepoint fails, and the business data is lost with it. The original error is thrown. Keep + `XACT_ABORT` off and avoid `SNAPSHOT` for transactions that run `Observe`, or use `Enforce`. +- If the detached release fails, the write in the business transaction turns the committed in-flight row to `Completed` + inside that transaction. When that commit then fails, the row stays in flight, and after the switch to `Enforce` a + retry is answered `InFlight` (409) until the lease ends; it is not replayed. Keep `Observe` short for operations that move money.