Decouple async read loop's socket fill from parsing

.github/workflows/release.yml

@@ -68,6 +68,8 @@ jobs:
       with:
         name: packages
         path: .nupkgs/*.nupkg
+        include-hidden-files: true # .nupkgs is a dot-folder, which upload-artifact skips by default
+        if-no-files-found: error
     - name: NuGet login (OIDC to temp API key)
       if: github.event_name == 'release'
       uses: NuGet/login@v1

docs/SyncOverAsync.md

@@ -123,9 +123,10 @@ service while the real fix is made. It is not a licence to keep the blocking cal
 
 Two caveats worth knowing before you enable it:
 
-- it costs a reader and a writer thread **for each node you connect to** (not RESP2 pub/sub connections, which
-  stay on the thread-pool; RESP3 does not use separate pub/sub connections), so think about it before enabling
-  it against a very wide cluster, where that scales with the number of shards;
+- it costs three threads **for each node you connect to** — two readers (one pulling bytes off the socket, one
+  parsing and dispatching them) and a writer (not RESP2 pub/sub connections, which stay on the thread-pool; RESP3
+  does not use separate pub/sub connections), so think about it before enabling it against a very wide cluster,
+  where that scales with the number of shards;
 - it is deliberately opt-in, and set process-wide at startup rather than per-connection.
 
 Neither is meant to be permanent, and the first one especially. Work is in progress on dedicated readers built

src/RESPite/Buffers/CycleBuffer.cs

@@ -571,6 +571,15 @@ public void Recycle()
             Memory = default;
             RunningIndex = 0;
             _flags = Flags.None;
+
+            // This object becomes available - via the *shared, static* _spare slot below - to any
+            // CycleBuffer instance in the process, handed out by Segment.Create() as a pristine segment.
+            // StartTrimCount must not survive that: some callers reach Recycle() without having gone
+            // through Untrim() first (e.g. AppendOrRecycle's search-exhausted path), and Init() (called
+            // from Create()) never touches it either. A stale nonzero value here previously surfaced as
+            // Debug.Assert(leasedStart == 0, "should be zero for a new segment") failing in
+            // GetUncommittedMemory for what looked like a brand new segment.
+            StartTrimCount = 0;
             Interlocked.Exchange(ref _spare, this);
             DebugAssertValidChain();
         }

src/StackExchange.Redis/ConnectionMultiplexer.FeatureFlags.cs

@@ -25,8 +25,9 @@ private enum FeatureFlags
         /// out of that queue. It does not *fix* the thread-pool, and nothing here can: it means only that redis
         /// traffic keeps flowing while the real problem is found. See docs/SyncOverAsync.md.
         /// <para>
-        /// Costs a reader and a writer thread per connection, so it is worth thinking about before enabling it
-        /// against a very wide cluster, where connection counts scale with the number of shards.
+        /// Costs three threads per connection - two readers (one pulling bytes off the socket, one parsing and
+        /// dispatching them) and a writer - so it is worth thinking about before enabling it against a very wide
+        /// cluster, where connection counts scale with the number of shards.
         /// </para>
         /// </remarks>
         DedicatedThreads = 2,

src/StackExchange.Redis/PhysicalConnection.Write.cs

@@ -73,6 +73,12 @@ internal static BufferedStreamWriter.WriteMode ResolveWriteMode(
     /// </remarks>
     internal bool IsSyncWriter => _output is { IsSync: true };
 
+    /// <summary>
+    /// Asks a switchable writer to leave sync mode; the reader follows (see <c>ReadAllSync</c>'s hand-off to
+    /// <c>ReadAllAsync</c>). Returns <c>false</c> if the writer is not in sync mode, or cannot switch.
+    /// </summary>
+    internal bool TransitionToAsync() => _output?.TransitionToAsync() ?? false;
+
     private void InitOutput(Stream? stream)
     {
         if (stream is null) return;

tests/RESPite.Tests/CycleBufferTests.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Buffers;
 using System.Linq;
+using System.Threading;
 using RESPite.Buffers;
 using Xunit;
 
@@ -128,4 +129,143 @@ public void CanDiscardSafely(Timing timing)
 
         Assert.Equal(0, buffer.GetCommittedLength());
     }
+
+    /// <summary>
+    /// Stress test for the filler/parser split used by <c>PhysicalConnection.ReadAllAsync</c>: one real
+    /// thread keeps writing (GetUncommittedMemory+Commit) while another concurrently reads and discards
+    /// (TryGetCommitted/GetAllCommitted+DiscardCommitted), synchronized only by an external lock - exactly
+    /// the pattern the production code relies on, since CycleBuffer itself has no internal thread-safety.
+    /// A running byte pattern makes any loss, duplication, reordering, or corruption immediately visible.
+    /// </summary>
+    [Fact]
+    public void ConcurrentFillAndParse_PreservesDataUnderStress()
+    {
+        var buffer = CycleBuffer.Create();
+        var bufferLock = new object();
+        const long TargetBytes = 500_000; // several dozen 8KB segments' worth
+        long totalWritten = 0, totalVerified = 0;
+        Exception? fillerError = null, parserError = null;
+
+        var filler = new Thread(() =>
+        {
+            try
+            {
+                var rng = new Random(12345);
+                long written = 0;
+                while (written < TargetBytes)
+                {
+                    Memory<byte> mem;
+                    lock (bufferLock)
+                    {
+                        mem = buffer.GetUncommittedMemory();
+                    }
+
+                    var chunkLen = Math.Min(Math.Min(mem.Length, rng.Next(1, 4001)), (int)Math.Min(TargetBytes - written, int.MaxValue));
+                    var span = mem.Span.Slice(0, chunkLen);
+                    for (int i = 0; i < chunkLen; i++)
+                    {
+                        span[i] = unchecked((byte)(written + i));
+                    }
+
+                    lock (bufferLock)
+                    {
+                        buffer.Commit(chunkLen);
+                    }
+                    written += chunkLen;
+                    Volatile.Write(ref totalWritten, written);
+                }
+            }
+            catch (Exception ex)
+            {
+                fillerError = ex;
+            }
+        });
+
+        var parser = new Thread(() =>
+        {
+            try
+            {
+                long verified = 0;
+                while (verified < TargetBytes)
+                {
+                    bool single;
+                    ReadOnlySpan<byte> span = default;
+                    ReadOnlySequence<byte> seq = default;
+                    lock (bufferLock)
+                    {
+                        single = buffer.TryGetCommitted(out span);
+                        if (!single) seq = buffer.GetAllCommitted();
+                    }
+
+                    long consumed;
+                    if (single)
+                    {
+                        consumed = VerifyAndCount(span, verified);
+                    }
+                    else
+                    {
+                        consumed = 0;
+                        foreach (var segment in seq)
+                        {
+                            consumed += VerifyAndCount(segment.Span, verified + consumed);
+                        }
+                    }
+
+                    if (consumed > 0)
+                    {
+                        lock (bufferLock)
+                        {
+                            buffer.DiscardCommitted(consumed);
+                        }
+                        verified += consumed;
+                        Volatile.Write(ref totalVerified, verified);
+                    }
+                    else
+                    {
+                        Thread.Sleep(0); // nothing new yet; yield rather than hot-spin
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                parserError = ex;
+            }
+
+            static long VerifyAndCount(ReadOnlySpan<byte> span, long expectedStart)
+            {
+                for (int i = 0; i < span.Length; i++)
+                {
+                    var expectedByte = unchecked((byte)(expectedStart + i));
+                    if (span[i] != expectedByte)
+                    {
+                        throw new InvalidOperationException(
+                            $"Data mismatch at offset {expectedStart + i}: expected {expectedByte}, got {span[i]}");
+                    }
+                }
+                return span.Length;
+            }
+        });
+
+        filler.Start();
+        parser.Start();
+        var fillerDone = filler.Join(TimeSpan.FromSeconds(10));
+        var parserDone = parser.Join(TimeSpan.FromSeconds(10));
+        if (!fillerDone || !parserDone)
+        {
+            long committedNow;
+            lock (bufferLock)
+            {
+                committedNow = buffer.GetCommittedLength();
+            }
+            Assert.Fail(
+                $"stalled: fillerDone={fillerDone} parserDone={parserDone} totalWritten={Volatile.Read(ref totalWritten)} " +
+                $"totalVerified={Volatile.Read(ref totalVerified)} committedLength={committedNow} " +
+                $"fillerError={fillerError} parserError={parserError}");
+        }
+
+        Assert.Null(fillerError);
+        Assert.Null(parserError);
+        Assert.Equal(TargetBytes, totalWritten);
+        Assert.Equal(TargetBytes, totalVerified);
+    }
 }

tests/StackExchange.Redis.Tests/ReadBufferPoolTests.cs

@@ -0,0 +1,96 @@
+using System;
+using System.Buffers;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace StackExchange.Redis.Tests;
+
+/// <summary>
+/// The reader's <c>CycleBuffer</c> rents its segments from <see cref="ConfigurationOptions.ResponseBufferPool"/>;
+/// this checks that they go back when a connection is torn down, whichever of the parse loop and the
+/// background filler happens to stop last (see <c>PhysicalConnection.OnParseLoopStopped</c>).
+/// </summary>
+[Collection(NonParallelCollection.Name)]
+public class ReadBufferPoolTests(ITestOutputHelper output) : TestBase(output)
+{
+    protected override string GetConfiguration() => TestConfig.Current.PrimaryServerAndPort + "," + TestConfig.Current.ReplicaServerAndPort;
+
+    [Fact]
+    [Trait(TestCategories.Category, TestCategories.SimulatedConnectionFailure)]
+    public async Task ReadBuffersAreReturnedOnReconnectAndDispose()
+    {
+        var pool = new TrackingMemoryPool();
+        var config = ConfigurationOptions.Parse(GetConfiguration());
+        config.ResponseBufferPool = pool;
+        config.AllowAdmin = true;
+        config.AllowSimulateConnectionFailure = true;
+        config.KeepAlive = 1;
+        config.ReconnectRetryPolicy = new LinearRetry(200);
+
+        try
+        {
+            var conn = await ConnectionMultiplexer.ConnectAsync(config, Writer);
+            await using (conn)
+            {
+                var db = conn.GetDatabase();
+                await db.PingAsync();
+                var steady = pool.Outstanding;
+                Log($"outstanding after connect: {steady} (rented {pool.Rented})");
+                Assert.True(steady > 0, "the reader should be renting from the configured pool");
+
+                var server = conn.GetServer(conn.GetEndPoints()[0]);
+                Assert.SkipUnless(server.CanSimulateConnectionFailure(), "server cannot simulate connection failure");
+
+                // tear the connection down under the reader: its buffers must come back, and the replacement
+                // connection's rentals must not stack on top of leaked ones
+                server.SimulateConnectionFailure(SimulatedFailureType.All);
+                await UntilConditionAsync(TimeSpan.FromSeconds(5), () => server.IsConnected).ForAwait();
+                Assert.True(server.IsConnected, "expected reconnect");
+                await db.PingAsync();
+
+                await UntilConditionAsync(TimeSpan.FromSeconds(5), () => pool.Outstanding <= steady).ForAwait();
+                Log($"outstanding after reconnect: {pool.Outstanding} (rented {pool.Rented})");
+                Assert.True(pool.Outstanding <= steady, $"read buffers leaked across reconnect: {steady} -> {pool.Outstanding}");
+            }
+
+            // and once everything is disposed, nothing should still be out
+            await UntilConditionAsync(TimeSpan.FromSeconds(5), () => pool.Outstanding == 0).ForAwait();
+            Log($"outstanding after dispose: {pool.Outstanding} (rented {pool.Rented})");
+            Assert.Equal(0, pool.Outstanding);
+        }
+        finally
+        {
+            ClearAmbientFailures();
+        }
+    }
+
+    private sealed class TrackingMemoryPool : MemoryPool<byte>
+    {
+        private int _rented, _outstanding;
+        public int Rented => Volatile.Read(ref _rented);
+        public int Outstanding => Volatile.Read(ref _outstanding);
+        public override int MaxBufferSize => Shared.MaxBufferSize;
+        public override IMemoryOwner<byte> Rent(int minBufferSize = -1)
+        {
+            Interlocked.Increment(ref _rented);
+            Interlocked.Increment(ref _outstanding);
+            return new Lease(this, Shared.Rent(minBufferSize));
+        }
+        protected override void Dispose(bool disposing) { }
+
+        private sealed class Lease(TrackingMemoryPool owner, IMemoryOwner<byte> inner) : IMemoryOwner<byte>
+        {
+            private IMemoryOwner<byte>? _inner = inner;
+            public Memory<byte> Memory => _inner?.Memory ?? default;
+            public void Dispose()
+            {
+                if (Interlocked.Exchange(ref _inner, null) is { } tmp)
+                {
+                    tmp.Dispose();
+                    Interlocked.Decrement(ref owner._outstanding);
+                }
+            }
+        }
+    }
+}

tests/StackExchange.Redis.Tests/RoundTripUnitTests/ReaderLifecycleRoundTrip.cs

@@ -0,0 +1,72 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace StackExchange.Redis.Tests.RoundTripUnitTests;
+
+/// <summary>
+/// Exercises the filler/parser split in <c>PhysicalConnection.ReadAllAsync</c>/<c>ReadAllSync</c> through
+/// the normal request/response machinery, for behaviours that a simple one-shot round trip does not cover.
+/// </summary>
+public class ReaderLifecycleRoundTrip(ITestOutputHelper log)
+{
+    private static readonly TimeSpan Patience = TimeSpan.FromSeconds(5);
+
+    [Theory(Timeout = 15_000)]
+    [InlineData(WriteMode.Default)]
+    [InlineData(WriteMode.Sync)]
+    public async Task ReceivedBytesAreCounted(WriteMode mode)
+    {
+        // the received-byte counter feeds the "in:"/"recd=" diagnostics in timeout and connection-failure
+        // messages; it is maintained by the filler, so it must keep working in both reader modes
+        using var conn = new TestConnection(startReading: true, writeMode: mode, log: log);
+        Assert.Equal(0, conn.GetBytesReceived());
+
+        Assert.True(await Bounded(conn.PingAsync()));
+        Assert.Equal("+PONG\r\n".Length, conn.GetBytesReceived());
+
+        Assert.True(await Bounded(conn.PingAsync()));
+        Assert.Equal(2 * "+PONG\r\n".Length, conn.GetBytesReceived());
+    }
+
+    [Fact(Timeout = 30_000)]
+    public async Task SyncToAsyncHandoffLosesNoReplies()
+    {
+        using var conn = new TestConnection(startReading: true, writeMode: WriteMode.Sync, log: log);
+        Assert.True(conn.IsSyncWriter);
+        Assert.True(conn.IsSyncReader);
+        Assert.True(await Bounded(conn.PingAsync(), "sync round trip"));
+
+        // flip the writer; the reader follows on its next wake, i.e. after it has parsed the *next* reply
+        Assert.True(conn.TransitionToAsync(), "writer should accept the transition");
+        SpinUntil(() => !conn.IsSyncWriter, "writer should leave sync mode");
+        Assert.True(await Bounded(conn.PingAsync(), "reply that triggers the transition"));
+
+        // The sync parse loop has now returned and is waiting for its filler thread to stop - and that filler
+        // is blocked in a read, with nothing to read. The next reply lands in *that* read: it must be
+        // committed and parsed, not dropped, or this command never completes and every later reply on the
+        // connection pairs with the wrong command.
+        SpinUntil(() => conn.GetReadStatus() == PhysicalConnection.ReadStatus.TransitioningToAsync, "reader should be handing off");
+        Assert.True(await Bounded(conn.PingAsync(), "reply that arrives mid-handoff"));
+
+        // and from here on the async reader owns the connection
+        SpinUntil(() => conn.GetReadStatus() == PhysicalConnection.ReadStatus.ReadAsync, "async reader should take over");
+        Assert.False(conn.IsSyncReader);
+        Assert.True(await Bounded(conn.PingAsync(), "async round trip"));
+        Assert.True(await Bounded(conn.PingAsync(), "second async round trip"));
+        Assert.Equal(5 * "+PONG\r\n".Length, conn.GetBytesReceived());
+    }
+
+    private static void SpinUntil(Func<bool> condition, string what)
+        => Assert.True(SpinWait.SpinUntil(condition, Patience), $"timed out: {what}");
+
+    private static async Task<T> Bounded<T>(Task<T> task, string? what = null)
+    {
+        // Task.WaitAsync is not available on every test TFM, so: race against a delay, but fail with a
+        // message rather than just hanging until the test-level timeout fires
+        var completed = await Task.WhenAny(task, Task.Delay(Patience));
+        Assert.True(ReferenceEquals(completed, task), $"timed out waiting for reply{(what is null ? "" : ": " + what)}");
+        return await task;
+    }
+}