From bb99b988b7b101a313f2f1713abb02fcb1d5c244 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Fri, 23 Oct 2026 09:48:44 +0100 Subject: [PATCH] Give multi-message sub-commands and injected prefix commands a write time (#3292) Sub-commands of an IMultiMessage (WATCH, condition reads, MULTI, the queued operations; also SCRIPT LOAD and PTTL) are written inside the lock without passing through SetEnqueued, as are the SELECT/READONLY/ASKING/PREPARE prefixes injected ahead of a command. With a zero write time the heartbeat sees them as overdue by the process uptime; since #2965 such no-box messages count as sync timeouts, and together with an idle connection that satisfies the dead-socket heuristic while the reply is still in flight. Sub-commands now inherit the parent's write time (sharing its budget, including backlog time); prefix commands are stamped when written, via a single helper replacing four identical blocks. --- src/StackExchange.Redis/Message.cs | 7 ++ src/StackExchange.Redis/PhysicalBridge.cs | 39 +++++---- .../MultiMessageWriteTimeTests.cs | 79 +++++++++++++++++++ 3 files changed, 108 insertions(+), 17 deletions(-) create mode 100644 tests/StackExchange.Redis.Tests/MultiMessageWriteTimeTests.cs 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() { } + } +}