Sync path must wait for completion, not merely for a fault

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -2485,21 +2485,42 @@ internal static void ThrowFailed<T>(TaskCompletionSource<T>? source, Exception u
 #pragma warning disable CS0618 // Type or member is obsolete
                     result = TryPushMessageToBridgeSync(message, processor, source, ref server);
 #pragma warning restore CS0618
-                    if (!source.IsFaulted) // if we faulted while writing, we don't need to wait
+                    // Note this tests IsCompleted, *not* IsFaulted: the write path can fault and complete the
+                    // message inline on this thread (the lock is ours, so its PulseAll had no waiter and is
+                    // gone) - in which case there is nothing to wait for. But a fault published by *another*
+                    // thread always still owes us a pulse, and leaving early on that would let the pulse
+                    // arrive after we have recycled the box, landing on the next operation to borrow it.
+                    if (!source.IsCompleted)
                     {
                         if (result != WriteResult.Success)
                         {
                             throw GetException(result, message, server);
                         }
 
-                        if (Monitor.Wait(source, TimeoutMilliseconds))
+                        // wait for the *completion*, not merely for a pulse; a pulse that we did not cause
+                        // must not be allowed to shorten this wait, so keep the original deadline
+                        var remaining = TimeoutMilliseconds;
+                        var startedAt = Environment.TickCount;
+                        while (true)
                         {
-                            Trace("Timely response to " + message);
-                        }
-                        else
-                        {
-                            Trace("Timeout performing " + message);
-                            timeout = true;
+                            if (!Monitor.Wait(source, remaining))
+                            {
+                                Trace("Timeout performing " + message);
+                                timeout = true;
+                                break;
+                            }
+                            if (source.IsCompleted)
+                            {
+                                Trace("Timely response to " + message);
+                                break;
+                            }
+                            remaining = TimeoutMilliseconds - unchecked(Environment.TickCount - startedAt);
+                            if (remaining <= 0)
+                            {
+                                Trace("Timeout performing " + message);
+                                timeout = true;
+                                break;
+                            }
                         }
                     }
                 }

src/StackExchange.Redis/ResultBox.cs

@@ -9,6 +9,20 @@ internal interface IResultBox
         Exception? Fault { get; }
         bool IsAsync { get; }
         bool IsFaulted { get; }
+
+        /// <summary>
+        /// Indicates that <see cref="ActivateContinuations"/> has completed, i.e. the completing thread
+        /// has finished with this box.
+        /// </summary>
+        /// <remarks>
+        /// This - not <see cref="IsFaulted"/> - is the signal a synchronous waiter must use. The fault is
+        /// published (unlocked) by <c>ResultProcessor.SetException</c>, but the matching pulse only happens
+        /// later, in <c>Message.Complete</c>; a waiter that stops on <see cref="IsFaulted"/> can therefore
+        /// recycle the box while a pulse is still inbound, and that pulse then lands on whatever operation
+        /// next borrows the box from the thread-static pool.
+        /// </remarks>
+        bool IsCompleted { get; }
+
         void SetException(Exception ex);
         void ActivateContinuations();
         void Cancel();
@@ -22,9 +36,11 @@ internal interface IResultBox<T> : IResultBox
     internal abstract class SimpleResultBox : IResultBox
     {
         private volatile Exception? _exception;
+        private volatile bool _completed;
 
         bool IResultBox.IsAsync => false;
         bool IResultBox.IsFaulted => _exception != null;
+        bool IResultBox.IsCompleted => _completed;
         Exception? IResultBox.Fault => _exception;
         void IResultBox.SetException(Exception exception) => _exception = exception ?? CancelledException;
         void IResultBox.Cancel() => _exception = CancelledException;
@@ -33,22 +49,28 @@ void IResultBox.ActivateContinuations()
         {
             lock (this)
             { // tell the waiting thread that we're done
+                // note this is set *inside* the lock, so a waiter that only leaves on _completed cannot
+                // exit (and recycle the box) while this pulse is still inbound
+                _completed = true;
                 Monitor.PulseAll(this);
             }
             ConnectionMultiplexer.TraceWithoutContext("Pulsed", "Result");
         }
 
+        /// <summary>Clears the per-operation state so the box can be handed to another operation.</summary>
+        protected void ResetForRecycle()
+        {
+            _exception = null;
+            _completed = false;
+        }
+
         // in theory nobody should directly observe this; the only things
         // that call Cancel are transactions etc - TCS-based, and we detect
         // that and use TrySetCanceled instead
         // about any confusion in stack-trace
         internal static readonly Exception CancelledException = new TaskCanceledException();
 
-        protected Exception? Exception
-        {
-            get => _exception;
-            set => _exception = value;
-        }
+        protected Exception? Exception => _exception;
     }
 
     internal sealed class SimpleResultBox<T> : SimpleResultBox, IResultBox<T>
@@ -62,8 +84,10 @@ private SimpleResultBox() { }
         public static IResultBox<T> Create() => new SimpleResultBox<T>();
         public static IResultBox<T> Get() // includes recycled boxes; used from sync, so makes re-use easy
         {
-            var obj = _perThreadInstance ?? new SimpleResultBox<T>();
+            var obj = _perThreadInstance;
+            if (obj is null) return new SimpleResultBox<T>();
             _perThreadInstance = null; // in case of oddness; only set back when recycled
+            obj.ResetForRecycle(); // belt and braces; the recycle already did this
             return obj;
         }
 
@@ -75,7 +99,8 @@ private SimpleResultBox() { }
             ex = Exception;
             if (canRecycle)
             {
-                Exception = null;
+                // only ever reached once the box is IsCompleted, so no other thread is still touching it
+                ResetForRecycle();
                 _value = default!;
                 _perThreadInstance = this;
             }
@@ -97,6 +122,7 @@ private TaskResultBox(object? asyncState, TaskCreationOptions creationOptions) :
         bool IResultBox.IsAsync => true;
 
         bool IResultBox.IsFaulted => _exception != null;
+        bool IResultBox.IsCompleted => Task.IsCompleted;
         Exception? IResultBox.Fault => _exception;
 
         void IResultBox.Cancel() => _exception = SimpleResultBox.CancelledException;

tests/StackExchange.Redis.Tests/ResultBoxCompletionTests.cs

@@ -0,0 +1,116 @@
+using System;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace StackExchange.Redis.Tests;
+
+/// <summary>
+/// <see cref="SimpleResultBox{T}"/> boxes are pooled per-thread, so a synchronous caller must not stop
+/// waiting on one until the completing thread has finished with it. Faulting the box
+/// (<c>ResultProcessor.SetException</c>) and pulsing it (<c>Message.Complete</c> -&gt;
+/// <c>ActivateContinuations</c>) are separate, unsynchronized steps: a waiter that leaves on the fault
+/// alone can recycle the box while the pulse is still inbound, and that pulse then wakes whichever
+/// operation next borrows the box - which returns with neither result nor exception, shifting every
+/// later reply on that thread by one. Reported against Garnet CI as microsoft/garnet#2110.
+/// </summary>
+public class ResultBoxCompletionTests(ITestOutputHelper log)
+{
+    /// <summary>
+    /// A fault published by another thread must not be mistaken for completion: that thread still owes
+    /// us the pulse, and it is only safe to recycle the box once the pulse has been delivered.
+    /// </summary>
+    [Fact]
+    public void FaultAloneIsNotCompletion()
+    {
+        var box = SimpleResultBox<int>.Get();
+        using var faulted = new ManualResetEventSlim();
+        var pulsedAt = 0L;
+
+        var completer = Task.Run(() =>
+        {
+            box.SetException(new InvalidOperationException("boom"));
+            faulted.Set();
+            Thread.Sleep(200); // the real gap between ResultProcessor.SetException and Message.Complete
+            Volatile.Write(ref pulsedAt, Stopwatch.GetTimestamp());
+            box.ActivateContinuations();
+        });
+
+        lock (box)
+        {
+            Assert.True(faulted.Wait(5000));
+
+            // this is the state the old code bailed out on - and doing so is what recycles a box that
+            // is still owed a pulse
+            Assert.True(box.IsFaulted);
+            Assert.False(box.IsCompleted);
+
+            while (!box.IsCompleted)
+            {
+                Assert.True(Monitor.Wait(box, 5000), "timed out waiting for completion");
+            }
+        }
+
+        var leftAt = Stopwatch.GetTimestamp();
+        completer.Wait(5000);
+        log.WriteLine($"left the wait {(leftAt - Volatile.Read(ref pulsedAt)) * 1000.0 / Stopwatch.Frequency:F2}ms after the pulse");
+        Assert.True(leftAt >= Volatile.Read(ref pulsedAt), "waiter left before the completing thread pulsed");
+
+        var value = box.GetResult(out var ex, canRecycle: true);
+        Assert.Equal(0, value);
+        Assert.IsType<InvalidOperationException>(ex);
+    }
+
+    /// <summary>Recycling must hand back a box with no completion or fault left on it.</summary>
+    [Fact]
+    public void RecyclingResetsCompletionState()
+    {
+        var box = SimpleResultBox<int>.Get();
+        box.SetException(new InvalidOperationException("boom"));
+        box.ActivateContinuations();
+        Assert.True(box.IsCompleted);
+        Assert.True(box.IsFaulted);
+
+        box.GetResult(out _, canRecycle: true);
+
+        var reused = SimpleResultBox<int>.Get();
+        Assert.Same(box, reused); // same thread, so we get the pooled instance back
+        Assert.False(reused.IsCompleted);
+        Assert.False(reused.IsFaulted);
+    }
+
+    /// <summary>
+    /// A pulse the waiter did not cause must not shorten the wait - the sync path treats any wake as
+    /// "the reply is here", so a stale pulse would otherwise be reported as a successful (empty) result.
+    /// </summary>
+    [Fact]
+    public void StrayPulseDoesNotEndTheWait()
+    {
+        var box = SimpleResultBox<int>.Get();
+        using var strayed = new ManualResetEventSlim();
+
+        _ = Task.Run(() =>
+        {
+            Thread.Sleep(100);
+            lock (box) { Monitor.PulseAll(box); } // a pulse with no completion behind it
+            strayed.Set();
+            Thread.Sleep(100);
+            box.SetResult(42);
+            box.ActivateContinuations();
+        });
+
+        lock (box)
+        {
+            while (!box.IsCompleted)
+            {
+                Assert.True(Monitor.Wait(box, 5000), "timed out waiting for completion");
+            }
+        }
+
+        Assert.True(strayed.IsSet);
+        var value = box.GetResult(out var ex, canRecycle: true);
+        Assert.Null(ex);
+        Assert.Equal(42, value);
+    }
+}