From 2ac4f10d823df7e7d265d233feb85db00b2f62c3 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Mon, 3 Aug 2026 10:51:12 +0100 Subject: [PATCH] WithRetry().CreateTransaction(...) should : ITransaction, for legacy type testing --- .../Availability/RetryTransaction.cs | 86 ++++++++- .../RetryTests/RetryEndToEndTests.cs | 175 ++++++++++++++++++ 2 files changed, 259 insertions(+), 2 deletions(-) diff --git a/src/StackExchange.Redis/Availability/RetryTransaction.cs b/src/StackExchange.Redis/Availability/RetryTransaction.cs index 0d87cd47d..2ae8dacaf 100644 --- a/src/StackExchange.Redis/Availability/RetryTransaction.cs +++ b/src/StackExchange.Redis/Availability/RetryTransaction.cs @@ -14,10 +14,16 @@ namespace StackExchange.Redis.Availability; // so a retry after failover lands on the new member), replays every recorded operation and constraint onto // it, and awaits it. On a clean execution the per-attempt results are forwarded onto the durable proxies; on // a transient fault the whole attempt is discarded and replayed; on terminal failure the proxies are faulted. +// +// ITransaction (rather than just ITransactionAsync) so that callers written against the long-standing +// interface can move to a retrying database without rewriting their transaction code; the synchronous +// Execute is sync-over-async (see below). Note that ITransaction : IBatch : IDatabaseAsync, so the only +// members this adds over ITransactionAsync are the two Execute overloads. [AutoDatabase] -internal sealed partial class RetryTransaction : IDatabaseAsync, ITransactionAsync +internal sealed partial class RetryTransaction : IDatabaseAsync, ITransaction { - // Note: async-only, exactly like RetryDatabase - retrying is inherently delay-ish. + // Note: the *command* surface is async-only, exactly like RetryDatabase - retrying is inherently + // delay-ish; only the terminal Execute is offered synchronously. private readonly IDatabaseAsync _source; private readonly RetryController _controller; @@ -156,6 +162,32 @@ public async Task ExecuteAsync(CommandFlags flags = CommandFlags.None) } } + // ---- the synchronous ITransaction surface ------------------------------------------------------- + // Both are explicit implementations: they exist purely so that callers written against ITransaction can + // move onto a retrying database, so they should be reachable only *through* that interface - internal + // code (which has ExecuteAsync) has no business blocking, and this also keeps Execute() and + // Execute(flags) from competing in overload resolution on the concrete type. + + /// + // The IBatch shape: enqueue, and don't ask about the outcome. Mirrors RedisTransaction by mapping onto + // fire-and-forget - which for a *retrying* transaction means there is nothing for the retry machinery to + // observe (no reply is requested, so no fault is ever reported) and it collapses to a single attempt: + // the trade-off fire-and-forget always makes, here extending to the retries as well. + void IBatch.Execute() => ((ITransaction)this).Execute(CommandFlags.FireAndForget); + + /// + // Sync-over-async, deliberately: retrying is inherently delay-ish, which is why the *command* surface + // here is async-only - but a caller written against ITransaction has no async alternative for the + // terminal Execute, and refusing it would mean such code cannot adopt a retrying database at all. + // Blocking here is safe against a captured synchronization context: every await in ExecuteAsync (and in + // RetryController) is ConfigureAwait(false), and the per-operation proxies complete asynchronously. + // + // The wait is *not* bounded by the multiplexer's SyncTimeout - the whole point of a retry is to outlast + // the fault, so the delays between attempts are expected to exceed it. Each individual attempt is still + // bounded by the inner connection's own timeout, so the total remains bounded by the attempt budget. + // GetResult (rather than Wait) so the original exception surfaces, not an AggregateException wrapping it. + bool ITransaction.Execute(CommandFlags flags) => ExecuteAsync(flags).GetAwaiter().GetResult(); + private void CheckNotExecuted() { if (Volatile.Read(ref _executed) != 0) @@ -168,6 +200,13 @@ private void CheckNotExecuted() private interface IRecordedOp { void Replay(IDatabaseAsync inner); + + // Copy this attempt's outcome onto the durable proxy. A *replied* EXEC populates every queued + // operation's result before its own task completes (TransactionProcessor does both while processing + // that one reply), so the outcome is normally available inline. A fire-and-forget EXEC is the + // exception: it completes as soon as it has been written, with the replies still in flight - so + // forwarding must be deferred rather than blocking for a round-trip the caller explicitly declined + // (which is what F+F means, and what a plain RedisTransaction does in the same situation). void ForwardSuccess(); void Fault(Exception ex); void Observe(); @@ -194,6 +233,26 @@ public RecordedOp(in TState state, AutoDatabaseAsyncOperation o public void ForwardSuccess() { var attempt = _attempt!; + if (attempt.IsCompleted) + { + Forward(attempt); + } + else + { + // fire-and-forget EXEC: see the note on ForwardLater + ForwardLater(attempt); + } + } + + private void ForwardLater(Task attempt) => attempt.ContinueWith( + static (completed, state) => ((RecordedOp)state!).Forward(completed), + this, + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + + private void Forward(Task attempt) + { if (attempt.IsCanceled) _proxy.TrySetCanceled(); else if (attempt.IsFaulted) _proxy.TrySetException(attempt.Exception!.InnerExceptions); else _proxy.TrySetResult(attempt.GetAwaiter().GetResult()); @@ -231,6 +290,26 @@ public RecordedVoidOp(in TState state, AutoDatabaseAsyncOperation operat public void ForwardSuccess() { var attempt = _attempt!; + if (attempt.IsCompleted) + { + Forward(attempt); + } + else + { + // fire-and-forget EXEC: see the note on ForwardLater + ForwardLater(attempt); + } + } + + private void ForwardLater(Task attempt) => attempt.ContinueWith( + static (completed, state) => ((RecordedVoidOp)state!).Forward(completed), + this, + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + + private void Forward(Task attempt) + { if (attempt.IsCanceled) _proxy.TrySetCanceled(); else if (attempt.IsFaulted) _proxy.TrySetException(attempt.Exception!.InnerExceptions); else _proxy.TrySetResult(true); @@ -263,6 +342,9 @@ public RecordedCondition(Condition condition) public void Replay(ITransactionAsync inner) => _attempt = inner.AddCondition(_condition); + // Unlike the operations, this needs no deferral for a fire-and-forget EXEC: a conditional transaction + // has to know whether its constraints held before it can decide between EXEC and DISCARD, so the + // condition replies are resolved as part of writing it - even when the EXEC itself wants no reply. public void ForwardSuccess() { if (_attempt is not null) Result.SetSatisfied(_attempt.WasSatisfied); diff --git a/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs b/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs index 8f2b6b623..44ce533f1 100644 --- a/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs +++ b/tests/StackExchange.Redis.Tests/RetryTests/RetryEndToEndTests.cs @@ -745,6 +745,170 @@ public async Task WithRetry_Transaction_WithoutConditions_IgnoresWatchBudget() Assert.Equal(1, server.ExecOpsReceived); } + // ---- the synchronous ITransaction surface -------------------------------------------------------- + // A transaction built from a retrying database is an ITransaction, not just an ITransactionAsync, so code + // written against the long-standing interface - including its *synchronous* Execute - can move onto a + // retrying database unchanged. The static type of CreateTransaction is still ITransactionAsync (that is + // what IDatabaseAsync declares), so a down-level caller reaches it with a cast. + [Fact] + public async Task WithRetry_Transaction_IsAnITransaction() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:sync"; + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + ITransaction tran = Assert.IsAssignableFrom(db.WithRetry(policy).CreateTransaction()); + + var set = tran.StringSetAsync(key, "committed"); + var get = tran.StringGetAsync(key); + + Assert.True(tran.Execute()); // sync-over-async; note this binds to ITransaction's bool overload + + Assert.True(await set); // the durable proxies resolve exactly as on the async path + Assert.Equal("committed", await get); + Assert.Equal("committed", await db.StringGetAsync(key)); + Assert.Equal(1, server.ExecOpsReceived); + } + + // The point of the exercise: retries (and therefore the delays between them) happen *inside* the blocking + // Execute, so a down-level caller gets the retry behaviour without touching their code. + [Fact] + public async Task WithRetry_Transaction_SyncExecute_RidesOutTransientExec() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:sync:transient"; + Assert.True(await db.StringSetAsync(key, "seed")); + + server.FailExecOps = 1; // fail the first EXEC; the second should commit + + // a real (if small) delay between attempts, so the blocking wait genuinely has to outlast it + RetryPolicy policy = new RetryPolicy.Builder + { + MaxAttempts = 3, + RetryDelay = TimeSpan.FromMilliseconds(50), + JitterMax = TimeSpan.Zero, + }; + ITransaction tran = (ITransaction)db.WithRetry(policy).CreateTransaction(); + var cond = tran.AddCondition(Condition.StringEqual(key, "seed")); + var set = tran.StringSetAsync(key, "committed"); + + Assert.True(tran.Execute()); + + Assert.True(cond.WasSatisfied); // the WATCH constraint was replayed onto the winning attempt + Assert.True(await set); + Assert.Equal("committed", await db.StringGetAsync(key)); + Assert.Equal(2, server.ExecOpsReceived); // 1 x LOADING + 1 x commit, all within Execute() + } + + // A transaction that does not commit reports false from the sync Execute, just as it does from + // ExecuteAsync - and WasWatchConflict still distinguishes an elective abort from a lost race. + [Fact] + public async Task WithRetry_Transaction_SyncExecute_UnsatisfiedCondition_ReturnsFalse() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:sync:cond"; + Assert.True(await db.StringSetAsync(key, "seed")); + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + ITransaction tran = (ITransaction)db.WithRetry(policy).CreateTransaction(); + var cond = tran.AddCondition(Condition.StringEqual(key, "different")); // NOT satisfied + var set = tran.StringSetAsync(key, "committed"); + + Assert.False(tran.Execute()); + + Assert.False(cond.WasSatisfied); + Assert.False(tran.WasWatchConflict); // an elective abort, not a lost race + await Assert.ThrowsAnyAsync(async () => await set); + Assert.Equal("seed", await db.StringGetAsync(key)); + Assert.Equal(0, server.ExecOpsReceived); // never EXEC'd + } + + // When the attempts run out, the sync Execute surfaces the *original* server fault - not an + // AggregateException wrapping it (which is what a naive Task.Wait would have produced). + [Fact] + public async Task WithRetry_Transaction_SyncExecute_WhenAttemptsExhausted_ThrowsOriginalFault() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:sync:exhaust"; + + server.FailExecOps = 100; // never succeeds + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 2, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + ITransaction tran = (ITransaction)db.WithRetry(policy).CreateTransaction(); + var set = tran.StringSetAsync(key, "never"); + + var fault = Assert.Throws(() => tran.Execute()); + Assert.Equal(RedisErrorKind.Loading, fault.Kind); + + Assert.Same(fault, await Assert.ThrowsAsync(async () => await set)); // proxy faulted too + Assert.Equal(2, server.ExecOpsReceived); // exactly MaxAttempts + } + + // IBatch.Execute() is the "enqueue and don't ask" shape, which RedisTransaction maps onto fire-and-forget; + // a retrying transaction does the same. Fire-and-forget requests no reply for the EXEC, so there is + // nothing for the retry machinery to observe and it collapses to a single attempt - but the queued + // operations were not themselves fire-and-forget, so their proxies still resolve when the replies land. + [Fact] + public async Task WithRetry_Transaction_BatchExecute_IsFireAndForget() + { + using var server = new ExecFailServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:sync:ff"; + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + var tran = (ITransaction)db.WithRetry(policy).CreateTransaction(); + var set = tran.StringSetAsync(key, "committed"); + + ((IBatch)tran).Execute(); // the void overload: no outcome is reported + + Assert.True(await set); // ...but the per-operation proxy still resolves + Assert.Equal("committed", await db.StringGetAsync(key)); + Assert.Equal(1, server.ExecOpsReceived); + } + + // ...and it must not *wait* for the outcome either. A replied EXEC populates the queued operations' + // results before its own task completes, so the retry machinery can normally read each attempt's outcome + // inline; a fire-and-forget EXEC completes as soon as it has been written, with the replies still in + // flight. Forwarding must therefore be deferred rather than blocking for a round-trip the caller + // explicitly declined - exactly what a plain transaction does. The server here holds the EXEC reply back, + // so "did we wait for it?" is directly observable. + [Fact] + public async Task WithRetry_Transaction_BatchExecute_DoesNotWaitForReplies() + { + using var server = new SlowExecServer(Output); + await using var conn = await server.ConnectAsync(log: Writer); + + var db = conn.GetDatabase(); + RedisKey key = "retry:tran:sync:ff:nowait"; + + RetryPolicy policy = new RetryPolicy.Builder { MaxAttempts = 3, RetryDelay = TimeSpan.Zero, JitterMax = TimeSpan.Zero }; + var tran = (ITransaction)db.WithRetry(policy).CreateTransaction(); + var set = tran.StringSetAsync(key, "committed"); + + ((IBatch)tran).Execute(); + + // the server is still sitting on the EXEC reply, so this proves we returned without waiting for it + Assert.False(set.IsCompleted); + + // ...and the proxy is still resolved once the reply does land + Assert.True(await set); + Assert.Equal("committed", await db.StringGetAsync(key)); + } + // An in-proc server that *applies* each INCR and then reports OOM: the write happened, but the client // has no way to know that. Counts the applications so a test can tell "the client gave up" from "the // client repeated a write it could not account for". @@ -801,6 +965,17 @@ protected override TypedRedisValue Exec(RedisClient client, in RedisRequest requ } } + // An in-proc server that holds each EXEC reply back for a while, so a test can tell whether the client + // waited for it. (The server core is single-threaded, so sleeping here delays only this connection.) + private sealed class SlowExecServer(ITestOutputHelper? log, EndPoint? endpoint = null) : InProcessTestServer(log, endpoint) + { + protected override TypedRedisValue Exec(RedisClient client, in RedisRequest request) + { + Thread.Sleep(500); + return base.Exec(client, in request); + } + } + // An in-proc server that fails the first LoadingOps GET operations with a transient LOADING error // (decrementing the counter each time), then serves normally. Every GET bumps GetOpsReceived so the // test can confirm how many attempts actually reached the server.