Give multi-message sub-commands and injected prefix commands a write time

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
         }
 
+        /// <summary>
+        /// Sub-commands of an <see cref="IMultiMessage"/> are written inside the lock without passing through
+        /// <see cref="SetEnqueued"/>; 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.
+        /// </summary>
+        internal void InheritWriteTime(Message parent) => _writeTickCount = parent._writeTickCount;
+
         private int _writeTickCount;
 
         public int GetWriteTime() => Volatile.Read(ref _writeTickCount);

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);
         }
 
+        /// <summary>
+        /// Writes a fire-and-forget command that is injected ahead of a caller's message (SELECT, READONLY, ASKING, PREPARE).
+        /// These never pass through <see cref="Message.SetEnqueued"/>, 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).
+        /// </summary>
+        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)

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+[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<string> _lines = [];
+
+        public IReadOnlyList<string> Lines
+        {
+            get { lock (_lines) return [.. _lines]; }
+        }
+
+        public ILogger CreateLogger(string categoryName) => this;
+
+        public void AddProvider(ILoggerProvider provider) { }
+
+        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+
+        public bool IsEnabled(LogLevel logLevel) => true;
+
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
+        {
+            lock (_lines) _lines.Add(formatter(state, exception));
+        }
+
+        public void Dispose() { }
+    }
+}