diff --git a/libs/client/Garnet.client.csproj b/libs/client/Garnet.client.csproj
index 0a436edfd6a..b9b756ba48b 100644
--- a/libs/client/Garnet.client.csproj
+++ b/libs/client/Garnet.client.csproj
@@ -13,4 +13,8 @@
+
+
+
+
\ No newline at end of file
diff --git a/libs/client/NetworkWriter.cs b/libs/client/NetworkWriter.cs
index 9c5ac69e795..6862bc92d33 100644
--- a/libs/client/NetworkWriter.cs
+++ b/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;
diff --git a/test/standalone/Garnet.test/GarnetClientTests.cs b/test/standalone/Garnet.test/GarnetClientTests.cs
index 8f63ddb1ea3..2a892152eee 100644
--- a/test/standalone/Garnet.test/GarnetClientTests.cs
+++ b/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);
}
+
+ ///
+ /// 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).
+ ///
+ [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();
+ }
+ }
+
+ ///
+ /// Network sender stub whose SendResponse always throws, simulating the underlying
+ /// connection being disposed/broken concurrently with a page flush.
+ ///
+ 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;
+ }
}
}
\ No newline at end of file