diff --git a/src/StackExchange.Redis/Message.cs b/src/StackExchange.Redis/Message.cs
index c6aceb60b..e6943cbcf 100644
--- a/src/StackExchange.Redis/Message.cs
+++ b/src/StackExchange.Redis/Message.cs
@@ -887,6 +887,13 @@ internal void SetWriteTime()
.TickCount; // note this might be reset if we resend a message, cluster-moved etc; I'm OK with that
}
+ ///
+ /// Sub-commands of an are written inside the lock without passing through
+ /// ; they take the parent's write time, so that they share its timeout budget
+ /// (including any time it spent in the backlog) rather than looking overdue by the process uptime.
+ ///
+ internal void InheritWriteTime(Message parent) => _writeTickCount = parent._writeTickCount;
+
private int _writeTickCount;
public int GetWriteTime() => Volatile.Read(ref _writeTickCount);
diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs
index d1cb8b5b1..3ad0afb28 100644
--- a/src/StackExchange.Redis/PhysicalBridge.cs
+++ b/src/StackExchange.Redis/PhysicalBridge.cs
@@ -865,6 +865,9 @@ private WriteResult WriteMessageInsideLock(PhysicalConnection physical, Message
SelectDatabaseInsideWriteLock(physical, message); // need to switch database *before* the transaction
foreach (var subCommand in multiMessage.GetMessages(physical))
{
+ // sub-commands do not pass through SetEnqueued (only the parent did, before the backlog);
+ // without a write time the heartbeat sees them as overdue by the process uptime (#3292)
+ if (!ReferenceEquals(subCommand, message)) subCommand.InheritWriteTime(message);
result = WriteMessageToServerInsideWriteLock(physical, subCommand);
if (result != WriteResult.Success)
{
@@ -1614,6 +1617,21 @@ private void OnInternalError(Exception exception, [CallerMemberName] string? ori
Multiplexer.OnInternalError(exception, ServerEndPoint.EndPoint, ConnectionType, origin);
}
+ ///
+ /// Writes a fire-and-forget command that is injected ahead of a caller's message (SELECT, READONLY, ASKING, PREPARE).
+ /// These never pass through , so the write time is stamped here; some of them are
+ /// shared static instances, so the stamp can be overwritten by another connection, but only ever with a later time,
+ /// which can make a pending prefix look younger than it is - never older (#3292).
+ ///
+ private void WritePrefixCommandInsideWriteLock(PhysicalConnection connection, Message prefix)
+ {
+ prefix.SetWriteTime();
+ connection.EnqueueInsideWriteLock(prefix);
+ prefix.WriteTo(connection);
+ prefix.SetRequestSent();
+ IncrementOpCount();
+ }
+
private void SelectDatabaseInsideWriteLock(PhysicalConnection connection, Message message)
{
int db = message.Db;
@@ -1622,10 +1640,7 @@ private void SelectDatabaseInsideWriteLock(PhysicalConnection connection, Messag
var sel = connection.GetSelectDatabaseCommand(db, message);
if (sel != null)
{
- connection.EnqueueInsideWriteLock(sel);
- sel.WriteTo(connection);
- sel.SetRequestSent();
- IncrementOpCount();
+ WritePrefixCommandInsideWriteLock(connection, sel);
}
}
}
@@ -1644,10 +1659,7 @@ private void PrepareFieldSetInsideWriteLock(PhysicalConnection connection, Messa
if (connection.TryAddPreparedFieldSet(fieldSet.Id))
{
var prepare = fieldSet.CreatePrepareMessage(message.Db);
- connection.EnqueueInsideWriteLock(prepare);
- prepare.WriteTo(connection);
- prepare.SetRequestSent();
- IncrementOpCount();
+ WritePrefixCommandInsideWriteLock(connection, prepare);
fieldSet.RegisterServer(ServerEndPoint, message.Db);
}
}
@@ -1707,19 +1719,12 @@ private WriteResult WriteMessageToServerInsideWriteLock(PhysicalConnection conne
var readmode = connection.GetReadModeCommand(isPrimaryOnly);
if (readmode != null)
{
- connection.EnqueueInsideWriteLock(readmode);
- readmode.WriteTo(connection);
- readmode.SetRequestSent();
- IncrementOpCount();
+ WritePrefixCommandInsideWriteLock(connection, readmode);
}
}
if (message.IsAsking)
{
- var asking = ReusableAskingCommand;
- connection.EnqueueInsideWriteLock(asking);
- asking.WriteTo(connection);
- asking.SetRequestSent();
- IncrementOpCount();
+ WritePrefixCommandInsideWriteLock(connection, ReusableAskingCommand);
}
}
switch (cmd)
diff --git a/tests/StackExchange.Redis.Tests/MultiMessageWriteTimeTests.cs b/tests/StackExchange.Redis.Tests/MultiMessageWriteTimeTests.cs
new file mode 100644
index 000000000..4a1f0c408
--- /dev/null
+++ b/tests/StackExchange.Redis.Tests/MultiMessageWriteTimeTests.cs
@@ -0,0 +1,79 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Xunit;
+
+namespace StackExchange.Redis.Tests;
+
+///
+/// The sub-commands of a transaction (and the SELECT/READONLY/ASKING prefixes written ahead of a command) are
+/// added to the awaiting-response queue without going through the normal enqueue path. If they carry no write
+/// time, the heartbeat sees them as overdue by the process uptime, and on a connection that has been idle for
+/// a while that is enough to trip the dead-socket heuristic while the reply is still in flight; see #3292.
+///
+[Collection(NonParallelCollection.Name)]
+public class MultiMessageWriteTimeTests(ITestOutputHelper output) : TestBase(output)
+{
+ [Fact]
+ public async Task HeartbeatDuringTransactionOnIdleConnectionDoesNotTearItDown()
+ {
+ const int Timeout = 600;
+ var logs = new CapturingLoggerFactory();
+ using var server = new InProcessTestServer(Output);
+ var config = server.GetClientConfig(defaultOnly: true);
+ config.SyncTimeout = config.AsyncTimeout = Timeout;
+ config.HeartbeatInterval = TimeSpan.FromHours(1); // only the heartbeats this test issues
+ config.KeepAlive = 600; // no pings resetting the last-read clock during the idle period
+ config.LoggerFactory = logs;
+
+ await using var conn = await ConnectionMultiplexer.ConnectAsync(config);
+ var db = conn.GetDatabase(1); // a non-default database, so a SELECT prefix is written ahead of the transaction
+ var key = Me();
+ await db.StringSetAsync(key, "before");
+
+ // the heuristic needs the connection silent for more than 4x the timeout (whole seconds) before the write
+ await Task.Delay((4 * Timeout) + 1_000);
+
+ // replies now take longer than the gap to the heartbeat, but well inside the timeout
+ server.SetLatency(TimeSpan.FromMilliseconds(250));
+ var tran = db.CreateTransaction();
+ tran.AddCondition(Condition.StringEqual(key, "before")); // WATCH + GET, written by the transaction itself
+ _ = tran.StringSetAsync(key, "after");
+ var exec = Task.Run(() => tran.ExecuteAsync());
+
+ await Task.Delay(50); // the sub-commands are on the wire, their replies are not back yet
+ conn.OnHeartbeat();
+
+ var committed = await exec.ContinueWith(t => t.IsCompletedSuccessfully && t.Result, TaskScheduler.Default);
+ Assert.DoesNotContain(logs.Lines, l => l.Contains("Dead socket", StringComparison.Ordinal));
+ Assert.True(committed, "the transaction should have committed");
+ Assert.Equal("after", await db.StringGetAsync(key));
+ }
+
+ private sealed class CapturingLoggerFactory : ILoggerFactory, ILogger
+ {
+ private readonly List _lines = [];
+
+ public IReadOnlyList Lines
+ {
+ get { lock (_lines) return [.. _lines]; }
+ }
+
+ public ILogger CreateLogger(string categoryName) => this;
+
+ public void AddProvider(ILoggerProvider provider) { }
+
+ public IDisposable? BeginScope(TState state) where TState : notnull => null;
+
+ public bool IsEnabled(LogLevel logLevel) => true;
+
+ public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter)
+ {
+ lock (_lines) _lines.Add(formatter(state, exception));
+ }
+
+ public void Dispose() { }
+ }
+}