WithRetry().CreateTransaction(...) should : ITransaction

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<bool> 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.
+
+    /// <inheritdoc/>
+    // 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);
+
+    /// <inheritdoc/>
+    // 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<TState, TResult> 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<TResult> attempt) => attempt.ContinueWith(
+            static (completed, state) => ((RecordedOp<TState, TResult>)state!).Forward(completed),
+            this,
+            CancellationToken.None,
+            TaskContinuationOptions.ExecuteSynchronously,
+            TaskScheduler.Default);
+
+        private void Forward(Task<TResult> 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<TState> 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<TState>)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);

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<ITransaction>(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<OperationCanceledException>(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<RedisServerException>(() => tran.Execute());
+        Assert.Equal(RedisErrorKind.Loading, fault.Kind);
+
+        Assert.Same(fault, await Assert.ThrowsAsync<RedisServerException>(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.