Fix NetworkWriter.AsyncFlushPages hanging FlushEvent waiters on send failure

libs/client/Garnet.client.csproj

@@ -13,4 +13,8 @@
     <ProjectReference Include="..\storage\Tsavorite\cs\src\core\Tsavorite.core.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <InternalsVisibleTo Include="Garnet.test" Key="0024000004800000940000000602000000240000525341310004000001000100011b1661238d3d3c76232193c8aa2de8c05b8930d6dfe8cd88797a8f5624fdf14a1643141f31da05c0f67961b0e3a64c7120001d2f8579f01ac788b0ff545790d44854abe02f42bfe36a056166a75c6a694db8c5b6609cff8a2dbb429855a1d9f79d4d8ec3e145c74bfdd903274b7344beea93eab86b422652f8dd8eecf530d2" />
+  </ItemGroup>
+
 </Project>

libs/client/NetworkWriter.cs

@@ -333,6 +333,9 @@ public void AsyncFlushPages(long fromAddress, long untilAddress)
                     {
                         logger?.LogError(ex, "Exception calling networkSender.SendResponse in AsyncFlushPages");
                         networkHandler.Dispose();
+                        // Still signal completion for this page so FlushEvent waiters (e.g. GarnetClient
+                        // callers blocked in InternalExecuteNoResponse) are not left hanging indefinitely.
+                        AsyncFlushPageCallback(asyncResult);
                     }
                 }
                 if (flushPage == endPage) break;

test/standalone/Garnet.test/GarnetClientTests.cs

@@ -4,10 +4,13 @@
 using System.Linq;
 using System.Net;
 using System.Net.Sockets;
+using System.Reflection;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Garnet.client;
 using Garnet.common;
+using Garnet.networking;
 using NUnit.Framework;
 using NUnit.Framework.Legacy;
 
@@ -568,5 +571,96 @@ public async Task MultipleSocketPing([Values] bool useTls)
             result = await db.ExecuteForStringResultAsync("PING").ConfigureAwait(false);
             ClassicAssert.AreEqual("PONG", result);
         }
+
+        /// <summary>
+        /// Regression test for: NetworkWriter.AsyncFlushPages swallows send failures without
+        /// releasing the flush-completion event, hanging callers blocked on FlushEvent (e.g.
+        /// GarnetClient.InternalExecuteNoResponse, used by cluster PUBLISH forwarding). A
+        /// thread is parked in flushEvent.Wait (mirroring a caller already blocked when the
+        /// connection dies) before the failing flush is triggered; if AsyncFlushPageCallback
+        /// is not invoked when networkSender.SendResponse throws, that thread is never
+        /// released and only unblocks via our cancellation-token timeout (which the
+        /// assertions below turn into a test failure instead of a hang).
+        /// </summary>
+        [Test]
+        public void AsyncFlushPagesSignalsFlushEventOnSendFailure()
+        {
+            using var server = TestUtils.CreateGarnetServer(TestUtils.MethodTestDir);
+            server.Start();
+
+            using var db = new GarnetClient(TestUtils.EndPoint, sendPageSize: 1 << 12);
+            db.Connect();
+
+            var networkWriter = (NetworkWriter)typeof(GarnetClient)
+                .GetField("networkWriter", BindingFlags.NonPublic | BindingFlags.Instance)
+                .GetValue(db);
+
+            networkWriter.epoch.Resume();
+            try
+            {
+                // Reserve some space in the send buffer so AsyncFlushPages below has a real
+                // (non-empty) range to send, exercising the networkSender.SendResponse call
+                // rather than the empty-range fast path.
+                var (_, address) = networkWriter.TryAllocate(64, out var flushEvent);
+                ClassicAssert.GreaterOrEqual(address, 0);
+                var untilAddress = networkWriter.GetTailAddress();
+
+                // Simulate the connection being disposed/broken concurrently with the flush,
+                // as happens in production when a gossip timeout tears down the connection:
+                // make the send throw.
+                typeof(NetworkWriter)
+                    .GetField("networkSender", BindingFlags.NonPublic | BindingFlags.Instance)
+                    .SetValue(networkWriter, new ThrowingNetworkSender());
+
+                using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3));
+                // WaitAsync registers with the underlying SemaphoreSlim synchronously, before
+                // returning - unlike the previous Wait()-on-a-background-thread approach, there
+                // is no window between "registered as a waiter" and "flush triggered" for a
+                // racing Set()/Dispose() to be missed, so no separate thread or readiness
+                // rendezvous is needed at all.
+                var waitTask = flushEvent.WaitAsync(cts.Token);
+
+                networkWriter.AsyncFlushPages(0, untilAddress);
+
+                var completed = waitTask.Wait(TimeSpan.FromSeconds(5));
+                ClassicAssert.IsTrue(completed,
+                    "FlushEvent was not signaled after a failed flush (regression)");
+                ClassicAssert.IsFalse(waitTask.IsFaulted, $"Wait faulted: {waitTask.Exception}");
+            }
+            finally
+            {
+                networkWriter.epoch.Suspend();
+            }
+        }
+
+        /// <summary>
+        /// Network sender stub whose SendResponse always throws, simulating the underlying
+        /// connection being disposed/broken concurrently with a page flush.
+        /// </summary>
+        sealed unsafe class ThrowingNetworkSender : INetworkSender
+        {
+            readonly MaxSizeSettings maxSizeSettings = new();
+
+            public MaxSizeSettings GetMaxSizeSettings => maxSizeSettings;
+            public string RemoteEndpointName => "";
+            public string LocalEndpointName => "";
+            public bool IsLocalConnection() => true;
+            public void Dispose() { }
+            public void DisposeNetworkSender(bool waitForSendCompletion) { }
+            public void Enter() { }
+            public void EnterAndGetResponseObject(out byte* head, out byte* tail) => throw new NotSupportedException();
+            public void Exit() { }
+            public void ExitAndReturnResponseObject() { }
+            public void GetResponseObject() { }
+            public byte* GetResponseObjectHead() => throw new NotSupportedException();
+            public byte* GetResponseObjectTail() => throw new NotSupportedException();
+            public void ReturnResponseObject() { }
+            public void SendCallback(object context) { }
+            public bool SendResponse(int offset, int size) => throw new NotSupportedException();
+            public void SendResponse(byte[] buffer, int offset, int count, object context)
+                => throw new Exception("Simulated send failure (connection disposed)");
+            public void Throttle() { }
+            public bool TryClose() => false;
+        }
     }
 }