From 4901148d3ebea0cb85f6192ad8d450589a9836d3 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 4 Nov 2026 14:03:22 +0000 Subject: [PATCH 1/5] MOVING handoff: hold new work back while draining, so the recycle happens within a round trip instead of at the deadline The MoveTo/Recycle paths drained by waiting for "no caller work" before replacing the connections. Under steady traffic that never happens: there is always a command in flight, so the drain waited out the entire announced window and recycled at the deadline, failing whatever was in flight at that instant and leaving no time to establish the replacement. Measured live on a Redis Enterprise endpoint rebind with ~450 ops/s on the connection: MOVING at +5.8s, MoveTo decided at +6.6s, connections replaced at +20.9s against a 15s window, 8 failed commands, and the "did not establish a replacement within the announced window" warning. The bridge now has a quiesce mode for the duration of the drain: new caller messages (single, batch, sync and async) go to the backlog instead of the socket, the backlog processor does not write while quiescing, and the drain waits only for commands already written. Held work is written to the replacement once it is established, as across any reconnect; a FailFast backlog policy no longer abandons it for a MaintenanceHandoff, which is a replacement we chose rather than a failure. Internal traffic (keep-alives) is not held. Same scenario with this change: MOVING +5.8s, handoff +6.7s, restored +6.8s, 0 failed, nothing outstanding at the recycle, no warning. The new test runs a steady load against the in-process server and asserts that the recycle follows the notification within a fraction of the window and that no command fails; it fails on main (recycle at ~10s, the whole window) and passes here. --- src/StackExchange.Redis/PhysicalBridge.cs | 62 +++++++++++++++- .../ServerEndPoint.Maintenance.cs | 42 ++++++++--- src/StackExchange.Redis/ServerEndPoint.cs | 7 ++ .../MaintenanceNotificationTests.cs | 70 +++++++++++++++++++ 4 files changed, 167 insertions(+), 14 deletions(-) diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index ee3a7bfb2..c82d066bc 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -132,6 +132,42 @@ public enum State : byte /// internal void MarkNeedsReconnect() => Volatile.Write(ref _needsReconnect, true); + private volatile bool _quiescing; + + /// + /// Starts holding caller work back from the current connection while a maintenance handoff lets what has + /// already been written complete; see . + /// + /// + /// Held work goes to the backlog rather than the socket, and is written to the replacement connection once + /// that is established. Without this the drain could never finish under steady traffic: there was always + /// something in flight, so the recycle waited for the end of the announced window and then failed whatever + /// was in flight at that instant. + /// + internal void BeginQuiesce() => _quiescing = true; + + internal void EndQuiesce() + { + _quiescing = false; + if (BacklogHasItems) + { + // the processor may be lingering between passes, or may have exited; either way, get the held work moving + _backlogAutoReset.Set(); + StartBacklogProcessor(); + } + } + + /// + /// Queue a caller's message for the replacement connection during a handoff, regardless of the backlog policy: + /// the connection is live and the message would have been written, so failing it would be a regression. + /// + private WriteResult HoldForHandoff(Message message) + { + message.SetEnqueued(null); + BacklogEnqueue(message); + return WriteResult.Success; + } + public void Dispose() { isDisposed = true; @@ -212,6 +248,7 @@ private WriteResult FailDueToNoConnection(Message message) public WriteResult TryWriteSync(Message message, bool isReplica) { if (isDisposed) throw new ObjectDisposedException(Name); + if (_quiescing && !message.IsInternalCall) return HoldForHandoff(message); if (!IsConnected || NeedsReconnect) return QueueOrFailMessage(message); var physical = this.physical; @@ -236,6 +273,7 @@ public WriteResult TryWriteSync(Message message, bool isReplica) public ValueTask TryWriteAsync(Message message, bool isReplica, bool bypassBacklog = false) { if (isDisposed) throw new ObjectDisposedException(Name); + if (_quiescing && !bypassBacklog && !message.IsInternalCall) return new ValueTask(HoldForHandoff(message)); if ((!IsConnected || NeedsReconnect) && !bypassBacklog) return new ValueTask(QueueOrFailMessage(message)); var physical = this.physical; @@ -477,8 +515,9 @@ internal void OnConnectionFailed(PhysicalConnection connection, ConnectionFailur Multiplexer.Logger?.LogErrorConnectionIssue(innerException, innerException.Message); } Trace($"OnConnectionFailed: {connection}"); - // If we're configured to, fail all pending backlogged messages - if (Multiplexer.RawConfig.BacklogPolicy?.AbortPendingOnConnectionFailure == true) + // If we're configured to, fail all pending backlogged messages - but not for a maintenance handoff, which + // is a replacement we chose, and whose backlog is work deliberately held for the replacement connection + if (Multiplexer.RawConfig.BacklogPolicy?.AbortPendingOnConnectionFailure == true && failureType != ConnectionFailureType.MaintenanceHandoff) { AbandonPendingBacklog(innerException, connection); } @@ -776,6 +815,16 @@ internal bool TryEnqueue(List messages, bool isReplica) if (isDisposed) throw new ObjectDisposedException(Name); + if (_quiescing) + { + // a maintenance handoff is draining the connection: hold the batch for the replacement + foreach (var message in messages) + { + HoldForHandoff(message); + } + return true; + } + if (!IsConnected || NeedsReconnect) { // During reconnection (e.g. MOVED-to-same-endpoint), queue messages to the backlog @@ -1072,6 +1121,13 @@ private void StartBacklogProcessor() /// Crawls from the head of the backlog queue, consuming anything that should have timed out /// and pruning it accordingly (these messages will get timeout exceptions). /// + /// + /// Whether a caller is waiting on something that has actually been written to the current connection, as + /// opposed to queued in the backlog. This is what a handoff drains: backlogged work is held for the + /// replacement rather than abandoned, so it must not keep the old connection alive. + /// + internal bool HasCallerWorkInFlight() => physical?.HasCallerMessagesAwaitingResponse() == true; + /// /// Whether this bridge owes a *caller* anything, ignoring our own internal traffic. /// @@ -1272,7 +1328,7 @@ private void ProcessBridgeBacklog() // Timeouts are handled above, so we're exclusively into backlog items eligible to write at this point. // If we can't write them, abort and wait for the next heartbeat or activation to try this again. bool flush = false; - while (IsConnected) + while (IsConnected && !_quiescing) // not while a handoff is holding work back for the replacement { // Snapshot the connection for the whole of this message: OnDisconnected nulls the field // without waiting for the write lock, so re-reading it (as the loop guard used to) can hand diff --git a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs index 53308faa6..d900c3440 100644 --- a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs +++ b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs @@ -471,7 +471,7 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? } /// - /// Lets in-flight caller work finish, then replaces the connections. + /// Holds new caller work back, lets what has already been written finish, then replaces the connections. /// /// /// Draining first because the socket is still working: anything already written may still be answered, and @@ -479,6 +479,14 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? /// because the server closes the socket at the end of it regardless - so anything not drained by then was /// going to fail either way, and draining strictly dominates. /// + /// Holding new work back is what makes the drain finish. Measured live on a connection carrying about 450 + /// operations a second: there was always something in flight, so a drain that waited for "no caller work" + /// waited out the entire 15s window and then recycled at the deadline - failing the eight commands in flight + /// at that instant, with no time left to establish the replacement. That is the exact outcome the handoff + /// exists to avoid. With new work held in the backlog, the written commands complete within a round trip, the + /// recycle follows, and the held work is written to the replacement once it is established. + /// + /// /// Both bridges, not just the one that was told. The measured blast radius is the *node*: four connections /// to one proxy, differing only in handshake, all closed simultaneously, and only the ones that had opted in /// were warned. @@ -487,18 +495,30 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? private async Task DrainThenRecycleAsync(TimeSpan budget, string reason) { var watch = ValueStopwatch.StartNew(); - while (HasCallerWork() && watch.ElapsedMilliseconds < budget.TotalMilliseconds) + interactive?.BeginQuiesce(); + subscription?.BeginQuiesce(); + try { - await Task.Delay(TimeSpan.FromMilliseconds(20)).ForAwait(); - } + while (HasCallerWorkInFlight() && watch.ElapsedMilliseconds < budget.TotalMilliseconds) + { + await Task.Delay(TimeSpan.FromMilliseconds(20)).ForAwait(); + } - var drained = !HasCallerWork(); - var recycled = (interactive?.RecycleConnection(reason) == true) | (subscription?.RecycleConnection(reason) == true); - if (recycled) Interlocked.Increment(ref _handoffRecycles); - Multiplexer.Trace( - $"MOVING: {(recycled ? "recycled" : "nothing to recycle")} after {watch.ElapsedMilliseconds}ms" - + (drained ? " (drained)" : " (still busy; the window ran out)"), - ToString()); + var drained = !HasCallerWorkInFlight(); + var recycled = (interactive?.RecycleConnection(reason) == true) | (subscription?.RecycleConnection(reason) == true); + if (recycled) Interlocked.Increment(ref _handoffRecycles); + Multiplexer.Trace( + $"MOVING: {(recycled ? "recycled" : "nothing to recycle")} after {watch.ElapsedMilliseconds}ms" + + (drained ? " (drained)" : " (still busy; the window ran out)"), + ToString()); + } + finally + { + // the recycle has already nulled the old connection, so releasing the hold cannot write to it; the + // held work waits in the backlog for the replacement, as it would across any reconnect + interactive?.EndQuiesce(); + subscription?.EndQuiesce(); + } } /// diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index e203139c6..f7da7a062 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -392,6 +392,13 @@ internal bool IsIdle() internal bool HasCallerWork() => interactive?.HasCallerWork() == true || subscription?.HasCallerWork() == true; + /// + /// Whether a caller is waiting on something already written to a current connection; backlogged work does + /// not count. See . + /// + internal bool HasCallerWorkInFlight() + => interactive?.HasCallerWorkInFlight() == true || subscription?.HasCallerWorkInFlight() == true; + /// /// Work this server still owes an answer on: written-and-awaiting-response, plus anything queued in /// the backlog. Zero means a retirement can complete without abandoning anyone. diff --git a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs index bfdb3a7a5..a14a4ede8 100644 --- a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs +++ b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs @@ -3,6 +3,7 @@ using System.Collections.Generic; using System.Linq; using System.Net; +using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; using StackExchange.Redis.Maintenance; @@ -842,6 +843,75 @@ await Poll.UntilAsync( } } + [Fact] + public async Task MovingUnderSteadyLoadRecyclesPromptlyAndFailsNothing() + { + // The drain used to wait for "no caller work", which under steady traffic never comes: measured live at + // about 450 ops/s, the recycle arrived at the end of the 15s window and failed the eight commands in + // flight at that instant, with no time left to establish the replacement. Holding new work back lets the + // written commands finish, so the recycle follows within a round trip and nothing fails: the held work is + // written to the replacement. + var logs = new CapturingLoggerFactory(); + var (server, conn, events) = await ConnectAsync(log, logs); + using (server) + await using (conn) + { + // every reply is held briefly, so with several workers something is always in flight + server.SetLatency(TimeSpan.FromMilliseconds(40)); + var db = conn.GetDatabase(); + using var stop = new CancellationTokenSource(); + long completed = 0; + var failures = new List(); + var workers = Enumerable.Range(0, 4).Select(_ => Task.Run(async () => + { + while (!stop.IsCancellationRequested) + { + try + { + await db.PingAsync(); + Interlocked.Increment(ref completed); + } + catch (Exception ex) + { + lock (failures) failures.Add(ex); + } + } + })).ToArray(); + Assert.True(await Poll.UntilAsync(() => Volatile.Read(ref completed) > 20), "the load should be flowing before the handoff"); + + var endpoint = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(server.DefaultEndPoint); + var successor = new IPEndPoint(IPAddress.Parse("127.0.0.9"), 6380); + var watch = System.Diagnostics.Stopwatch.StartNew(); + server.SendMoving(null, timeSeconds: 10, newEndpoint: successor, sequenceId: 0); + var moving = await events.NextAsync(); + Assert.Equal(MaintenanceNotificationType.Moving, moving.NotificationType); + + Assert.True( + await Poll.UntilAsync(() => endpoint.HandoffRecycles > 0, timeoutMilliseconds: 8_000), + "the handoff should have replaced the connection"); + var recycledAfter = watch.Elapsed; + log.WriteLine($"recycled after {recycledAfter.TotalMilliseconds:0}ms of a 10s window"); + + // jitter is at most a tenth of the window and the drain is one round trip; recycling at the deadline is the bug + Assert.True(recycledAfter < TimeSpan.FromSeconds(4), $"recycled after {recycledAfter}: that is waiting for the window, not draining"); + + // the replacement carries the held work and whatever follows + var before = Volatile.Read(ref completed); + Assert.True( + await Poll.UntilAsync(() => Volatile.Read(ref completed) > before + 20, timeoutMilliseconds: 10_000), + "traffic should resume on the replacement connection"); + + stop.Cancel(); + await Task.WhenAll(workers); + lock (failures) + { + log.WriteLine($"completed: {Volatile.Read(ref completed)}; failed: {failures.Count}"); + Assert.Empty(failures.Select(f => f.Message)); + } + Assert.DoesNotContain(logs.Lines, l => l.Contains("did not establish a replacement", StringComparison.Ordinal)); + } + } + [Fact] public async Task MovingRecyclesTheConnectionBeforeTheServerCloses() { From bc523608db46fba61c8b332618f993dfae7ffec5 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 4 Nov 2026 15:38:29 +0000 Subject: [PATCH 2/5] Handoff drain: cap the wait, recycle under the write lock, hold only live caller work Review follow-ups on the quiesce-then-drain change: - The drain is capped at one second as well as by the window. A blocking or stuck command in flight would otherwise hold everyone else's work in the backlog for the rest of the window and still be failed at the end of it; past the cap the connections are replaced regardless, and whatever was still in flight fails as the server's close would have failed it. - The recycle takes the bridge's write lock and re-checks for in-flight work under it, so a writer that passed the hold check before it was set cannot slip onto the old connection between the check and the recycle. A writer that reaches the lock after a handoff recycled its connection queues for the replacement instead of writing to a dead connection; only for the handoff (an ordinary closure is still reported as one, #3167), never for a message bypassing the backlog, and never for internal traffic. - Held work goes through QueueOrFailMessage's queue branch, so QUIT is never queued and the duplication is gone; QUIT and internal traffic are not held at all, in the batch path as well as the single-message paths. - Work is held only while the connection is still live; once it has dropped, the ordinary disconnected rules and the backlog policy apply. - A held enqueue that lands just after the hold is released kicks the backlog processor itself rather than waiting for the heartbeat. - The orphaned summary is back on CheckBacklogForTimeouts, and docs/ServerMaintenanceEvent.md describes the hold. New test: slow commands in flight do not hold the drain past the cap, and traffic resumes on the replacement. --- docs/ServerMaintenanceEvent.md | 4 +- src/StackExchange.Redis/PhysicalBridge.cs | 108 +++++++++++++----- .../ServerEndPoint.Maintenance.cs | 61 ++++++++-- .../MaintenanceNotificationTests.cs | 57 +++++++++ 4 files changed, 194 insertions(+), 36 deletions(-) diff --git a/docs/ServerMaintenanceEvent.md b/docs/ServerMaintenanceEvent.md index 15ef44c5a..530465c32 100644 --- a/docs/ServerMaintenanceEvent.md +++ b/docs/ServerMaintenanceEvent.md @@ -111,10 +111,12 @@ or `ServerDefault` to ask for nothing at all, which is what earlier versions did | `MIGRATING`, `FAILING_OVER`, `SMIGRATING` | relaxes command timeouts for that server while the disruption lasts | | `MIGRATED`, `FAILED_OVER` | ends the window, but keeps timeouts relaxed for a short tail while things settle | | `SMIGRATED` | as above, and re-reads the cluster topology, and re-subscribes any sharded channels whose slots moved | -| `MOVING` | works out the replacement address, lets in-flight work finish, then replaces the connections | +| `MOVING` | works out the replacement address, holds new commands back for up to a second while in-flight ones finish, then replaces the connections and sends the held commands on the replacement | So an application that does nothing at all still benefits: commands that would have timed out during a migration are given more room, a moved slot is learned without waiting to be redirected, and a `MOVING` is acted on before the server closes the socket. +During that handoff, commands the application issues are not written to the connection being retired: they wait in the backlog, for up to about a second, and are sent on the replacement once it is established. This happens whatever the `backlogPolicy`, because the connection is live and the command would otherwise have been written; and a `FailFast` policy does not abandon that held work when the old connection is closed, since a handoff is a replacement the client chose rather than a failure. A command still in flight when the second is up, such as a blocking read, is failed by the replacement, as the server's close would have failed it; the hold is bounded so that one such command cannot stall everything else for the rest of the window. + ### Notifications that arrive as you connect Redis Enterprise **retains the most recent completion** - `MIGRATED` or `FAILED_OVER` - and replays it to each connection that opts in, so a client that connects after a disruption still learns that it happened. Measured behaviour, worth knowing if you handle these events yourself: diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index c82d066bc..ee752a780 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -149,23 +149,49 @@ public enum State : byte internal void EndQuiesce() { _quiescing = false; - if (BacklogHasItems) - { - // the processor may be lingering between passes, or may have exited; either way, get the held work moving - _backlogAutoReset.Set(); - StartBacklogProcessor(); - } + if (BacklogHasItems) KickBacklogProcessor(); } + /// Gets held or queued work moving: the processor may be lingering between passes, or may have exited. + private void KickBacklogProcessor() + { + _backlogAutoReset.Set(); + StartBacklogProcessor(); + } + + /// + /// Whether a message should be held in the backlog for the replacement connection rather than written: only + /// while a handoff is draining a connection that is still live (once it has dropped, the ordinary disconnected + /// rules and the backlog policy apply), and only for a caller's work. Internal traffic (keep-alives, probes) + /// still goes to the old connection, and so does QUIT, which is never queued. + /// + private bool ShouldHoldForHandoff(Message message) + => _quiescing && IsConnected && !NeedsReconnect && !message.IsInternalCall && message.Command != RedisCommand.QUIT; + /// - /// Queue a caller's message for the replacement connection during a handoff, regardless of the backlog policy: - /// the connection is live and the message would have been written, so failing it would be a regression. + /// Replaces the connection once nothing a caller is waiting on is in flight, under the write lock: a writer that + /// passed the hold check before it was set, but has not yet written, cannot slip onto the old connection between + /// the check and the recycle. Such a writer finds the connection gone when it gets the lock, and queues for the + /// replacement instead (see ). /// - private WriteResult HoldForHandoff(Message message) + /// null if caller work was still in flight, so the drain should keep waiting; otherwise whether + /// there was a connection to recycle. + internal async ValueTask TryRecycleWhenDrainedAsync(string reason) { - message.SetEnqueued(null); - BacklogEnqueue(message); - return WriteResult.Success; + if (!await _singleWriter.TryTakeAsync().ForAwait()) + { + // could not get the lock inside the write timeout: something is badly stuck, and the window is finite + return RecycleConnection(reason); + } + try + { + if (HasCallerWorkInFlight()) return null; + return RecycleConnection(reason); + } + finally + { + _singleWriter.Release(); + } } public void Dispose() @@ -212,18 +238,21 @@ public void ReportNextFailure() public override string ToString() => ConnectionType + "/" + Format.ToString(ServerEndPoint.EndPoint); - private WriteResult QueueOrFailMessage(Message message) + private WriteResult QueueOrFailMessage(Message message, bool holdForHandoff = false) { - // If it's an internal call that's not a QUIT - // or we're allowed to queue in general, then queue - if (message.IsInternalCall || Multiplexer.RawConfig.BacklogPolicy.QueueWhileDisconnected) + // If it's an internal call that's not a QUIT, or we're allowed to queue in general, then queue. Work held + // for a maintenance handoff is queued whatever the policy: the connection is live and the message would + // otherwise have been written, so failing it would be a regression (see ShouldHoldForHandoff). + if (holdForHandoff || message.IsInternalCall || Multiplexer.RawConfig.BacklogPolicy.QueueWhileDisconnected) { // Let's just never ever queue a QUIT message if (message.Command != RedisCommand.QUIT) { message.SetEnqueued(null); BacklogEnqueue(message); - // Note: we don't start a worker on each message here + // Note: we don't start a worker on each message here - except for held work that lands just after + // the hold was released, which nothing else would pick up until the heartbeat + if (holdForHandoff && !_quiescing) KickBacklogProcessor(); return WriteResult.Success; // Successfully queued, so indicate success } } @@ -248,7 +277,7 @@ private WriteResult FailDueToNoConnection(Message message) public WriteResult TryWriteSync(Message message, bool isReplica) { if (isDisposed) throw new ObjectDisposedException(Name); - if (_quiescing && !message.IsInternalCall) return HoldForHandoff(message); + if (ShouldHoldForHandoff(message)) return QueueOrFailMessage(message, holdForHandoff: true); if (!IsConnected || NeedsReconnect) return QueueOrFailMessage(message); var physical = this.physical; @@ -273,7 +302,7 @@ public WriteResult TryWriteSync(Message message, bool isReplica) public ValueTask TryWriteAsync(Message message, bool isReplica, bool bypassBacklog = false) { if (isDisposed) throw new ObjectDisposedException(Name); - if (_quiescing && !bypassBacklog && !message.IsInternalCall) return new ValueTask(HoldForHandoff(message)); + if (!bypassBacklog && ShouldHoldForHandoff(message)) return new ValueTask(QueueOrFailMessage(message, holdForHandoff: true)); if ((!IsConnected || NeedsReconnect) && !bypassBacklog) return new ValueTask(QueueOrFailMessage(message)); var physical = this.physical; @@ -602,6 +631,7 @@ internal void OnFullyEstablished(PhysicalConnection connection, string source) { reportNextFailure = reconfigureNextFailure = true; LastException = null; + _recycledForHandoff = null; Interlocked.Exchange(ref failConnectCount, 0); ServerEndPoint.OnFullyEstablished(connection, source); @@ -815,12 +845,13 @@ internal bool TryEnqueue(List messages, bool isReplica) if (isDisposed) throw new ObjectDisposedException(Name); - if (_quiescing) + if (_quiescing && IsConnected && !NeedsReconnect && messages.Exists(m => !m.IsInternalCall && m.Command != RedisCommand.QUIT)) { - // a maintenance handoff is draining the connection: hold the batch for the replacement + // a maintenance handoff is draining the connection: hold the batch for the replacement. A batch is + // written as a unit, so it is held as a unit; a batch of purely internal traffic is not held at all foreach (var message in messages) { - HoldForHandoff(message); + QueueOrFailMessage(message, holdForHandoff: true); } return true; } @@ -977,6 +1008,7 @@ internal WriteResult WriteMessageTakingWriteLockSync(PhysicalConnection physical if (!gotLock) return TimedOutBeforeWrite(message); } + if (ConnectionReplacedWhileWaiting(physical, message)) return QueueOrFailMessage(message); var result = WriteMessageInsideLock(physical, message); if ((result == WriteResult.Success) & message.IsFlushRequiredSync) @@ -1117,10 +1149,6 @@ private void StartBacklogProcessor() } } - /// - /// Crawls from the head of the backlog queue, consuming anything that should have timed out - /// and pruning it accordingly (these messages will get timeout exceptions). - /// /// /// Whether a caller is waiting on something that has actually been written to the current connection, as /// opposed to queued in the backlog. This is what a handoff drains: backlogged work is held for the @@ -1142,6 +1170,10 @@ internal bool HasCallerWork() return false; } + /// + /// Crawls from the head of the backlog queue, consuming anything that should have timed out + /// and pruning it accordingly (these messages will get timeout exceptions). + /// private void CheckBacklogForTimeouts() { var now = Environment.TickCount; @@ -1446,6 +1478,22 @@ private WriteResult TimedOutBeforeWrite(Message message) return WriteResult.TimeoutBeforeWrite; } + /// + /// The connection most recently replaced by a maintenance handoff, so that a writer which was headed for it + /// can tell a planned replacement from an ordinary closure. + /// + private PhysicalConnection? _recycledForHandoff; + + /// + /// Whether the connection a caller's message was headed for was replaced by a maintenance handoff while the + /// writer waited for the lock. The message has not been written, so it is queued for the replacement rather + /// than written to a connection that is gone. Only the handoff case: a write that loses its connection to an + /// ordinary closure is still reported as a closure (#3167), and a message bypassing the backlog is never + /// queued. Internal traffic is exempt as well: the handshake writes to a connection before it is current. + /// + private bool ConnectionReplacedWhileWaiting(PhysicalConnection physical, Message message, bool bypassBacklog = false) + => !bypassBacklog && !message.IsInternalCall && !ReferenceEquals(physical, this.physical) && ReferenceEquals(physical, _recycledForHandoff); + /// /// This writes a message to the output stream. /// @@ -1483,11 +1531,12 @@ internal ValueTask WriteMessageTakingWriteLockAsync(PhysicalConnect // if we *still* can't get it: that counts as // an actual timeout var pending = _singleWriter.TryTakeAsync(); - if (!pending.IsCompletedSuccessfully) return WriteMessageTakingWriteLockAsync_Awaited(pending, physical, message); + if (!pending.IsCompletedSuccessfully) return WriteMessageTakingWriteLockAsync_Awaited(pending, physical, message, bypassBacklog); gotLock = pending.GetAwaiter().GetResult(); // fine since we know we got a result if (!gotLock) return new ValueTask(TimedOutBeforeWrite(message)); } + if (ConnectionReplacedWhileWaiting(physical, message, bypassBacklog)) return new ValueTask(QueueOrFailMessage(message)); var result = WriteMessageInsideLock(physical, message); if ((result == WriteResult.Success) & message.IsFlushRequiredAsync) { @@ -1519,13 +1568,15 @@ internal ValueTask WriteMessageTakingWriteLockAsync(PhysicalConnect private async ValueTask WriteMessageTakingWriteLockAsync_Awaited( ValueTask pending, PhysicalConnection physical, - Message message) + Message message, + bool bypassBacklog) { bool gotLock = false; try { gotLock = await pending.ForAwait(); if (!gotLock) return TimedOutBeforeWrite(message); + if (ConnectionReplacedWhileWaiting(physical, message, bypassBacklog)) return QueueOrFailMessage(message); var result = WriteMessageInsideLock(physical, message); if (result == WriteResult.Success) @@ -1939,6 +1990,7 @@ internal bool RecycleConnection(string reason) // the public event while the pipe is still live ("if *we* didn't burn the pipe: flag it"), and // Dispose runs Shutdown first - which is exactly why an ordinary dispose is silent. Recording first // also performs the disconnect, so the Dispose below is cleanup. + _recycledForHandoff = current; // a writer already past the hold check queues for the replacement instead current.RecordConnectionFailed(ConnectionFailureType.MaintenanceHandoff); current.Dispose(); return true; diff --git a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs index d900c3440..b65f7893a 100644 --- a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs +++ b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs @@ -470,14 +470,19 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? } } + /// + /// How long a handoff waits for in-flight commands before replacing the connections regardless. Ordinary + /// commands clear in a round trip; anything still in flight after this is a blocking or stuck command that will + /// not finish inside the window either, and holding everyone else's work for it would be a regression. + /// + private static readonly TimeSpan DrainCap = TimeSpan.FromSeconds(1); + /// /// Holds new caller work back, lets what has already been written finish, then replaces the connections. /// /// /// Draining first because the socket is still working: anything already written may still be answered, and - /// dropping it would fail commands that were about to succeed. Bounded by what is left of the window, - /// because the server closes the socket at the end of it regardless - so anything not drained by then was - /// going to fail either way, and draining strictly dominates. + /// dropping it would fail commands that were about to succeed. /// /// Holding new work back is what makes the drain finish. Measured live on a connection carrying about 450 /// operations a second: there was always something in flight, so a drain that waited for "no caller work" @@ -487,6 +492,12 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? /// recycle follows, and the held work is written to the replacement once it is established. /// /// + /// The wait is bounded by as well as by the window: a blocking read in flight would + /// otherwise hold every other caller's work for the whole window and still be failed at the end of it. Past + /// the cap the connections are replaced regardless, and whatever was still in flight fails as the server's + /// close would have failed it. + /// + /// /// Both bridges, not just the one that was told. The measured blast radius is the *node*: four connections /// to one proxy, differing only in handshake, all closed simultaneously, and only the ones that had opted in /// were warned. @@ -495,21 +506,34 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? private async Task DrainThenRecycleAsync(TimeSpan budget, string reason) { var watch = ValueStopwatch.StartNew(); + var cap = budget < DrainCap ? budget : DrainCap; interactive?.BeginQuiesce(); subscription?.BeginQuiesce(); try { - while (HasCallerWorkInFlight() && watch.ElapsedMilliseconds < budget.TotalMilliseconds) + bool drained, recycled; + while (true) { + var attempt = await TryRecycleWhenDrainedAsync(reason).ForAwait(); + if (attempt is { } done) + { + drained = true; + recycled = done; + break; + } + if (watch.ElapsedMilliseconds >= cap.TotalMilliseconds) + { + drained = false; + recycled = (interactive?.RecycleConnection(reason) == true) | (subscription?.RecycleConnection(reason) == true); + break; + } await Task.Delay(TimeSpan.FromMilliseconds(20)).ForAwait(); } - var drained = !HasCallerWorkInFlight(); - var recycled = (interactive?.RecycleConnection(reason) == true) | (subscription?.RecycleConnection(reason) == true); if (recycled) Interlocked.Increment(ref _handoffRecycles); Multiplexer.Trace( $"MOVING: {(recycled ? "recycled" : "nothing to recycle")} after {watch.ElapsedMilliseconds}ms" - + (drained ? " (drained)" : " (still busy; the window ran out)"), + + (drained ? " (drained)" : " (still busy at the drain cap)"), ToString()); } finally @@ -521,6 +545,29 @@ private async Task DrainThenRecycleAsync(TimeSpan budget, string reason) } } + /// + /// Each bridge in turn, under its write lock: null as soon as one still has caller work in flight, + /// otherwise whether anything was recycled. A bridge recycled on an earlier pass has nothing left to recycle, + /// so a partial pass is safe to repeat. + /// + private async ValueTask TryRecycleWhenDrainedAsync(string reason) + { + bool recycled = false; + if (interactive is { } i) + { + var result = await i.TryRecycleWhenDrainedAsync(reason).ForAwait(); + if (result is null) return null; + recycled |= result.Value; + } + if (subscription is { } s) + { + var result = await s.TryRecycleWhenDrainedAsync(reason).ForAwait(); + if (result is null) return null; + recycled |= result.Value; + } + return recycled; + } + /// /// Warns when a replacement connection is not fully established by the time the announced window ends. /// diff --git a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs index a14a4ede8..1c8fe4319 100644 --- a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs +++ b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs @@ -912,6 +912,63 @@ await Poll.UntilAsync(() => Volatile.Read(ref completed) > before + 20, timeoutM } } + [Fact] + public async Task MovingWithSlowCommandsInFlightRecyclesAtTheDrainCap() + { + // A blocking or very slow command must not hold everyone else's work in the backlog for the whole window: + // the drain waits a bounded time for in-flight commands and then recycles regardless. The slow commands are + // failed by the recycle, as the server's close would have failed them; everything issued afterwards flows. + var logs = new CapturingLoggerFactory(); + var (server, conn, events) = await ConnectAsync(log, logs); + using (server) + await using (conn) + { + server.SetLatency(TimeSpan.FromSeconds(3)); // every command now outlasts the drain cap + var db = conn.GetDatabase(); + var slow = Enumerable.Range(0, 3).Select(_ => db.PingAsync()).ToArray(); + await Task.Delay(250); // let them be written + + var endpoint = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(server.DefaultEndPoint); + var successor = new IPEndPoint(IPAddress.Parse("127.0.0.9"), 6380); + var watch = System.Diagnostics.Stopwatch.StartNew(); + server.SendMoving(null, timeSeconds: 10, newEndpoint: successor, sequenceId: 0); + var moving = await events.NextAsync(); + Assert.Equal(MaintenanceNotificationType.Moving, moving.NotificationType); + + Assert.True( + await Poll.UntilAsync(() => endpoint.HandoffRecycles > 0, timeoutMilliseconds: 8_000), + "the handoff should have replaced the connection"); + var recycledAfter = watch.Elapsed; + server.SetLatency(TimeSpan.Zero); // so the replacement's handshake is not slowed too + log.WriteLine($"recycled after {recycledAfter.TotalMilliseconds:0}ms with slow commands in flight"); + + // jitter is at most a tenth of the window, the cap one second: well short of the 3s the commands need + Assert.True(recycledAfter < TimeSpan.FromSeconds(3), $"recycled after {recycledAfter}: the drain waited for the slow commands instead of capping"); + + // the in-flight commands were failed by the recycle, which is what the server's close would have done + foreach (var pending in slow) + { + await Assert.ThrowsAnyAsync(async () => await pending); + } + + Assert.True( + await Poll.UntilAsync( + () => + { + try + { + return conn.IsConnected && conn.GetDatabase().Ping() >= TimeSpan.Zero; + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + return false; + } + }, + timeoutMilliseconds: 30_000), + "the client should be serving commands on the replacement connection"); + } + } + [Fact] public async Task MovingRecyclesTheConnectionBeforeTheServerCloses() { From 9a0f4abde639cce8dd2d59471378be575015a0c8 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 4 Nov 2026 16:15:07 +0000 Subject: [PATCH 3/5] Handoff drain: retire only the connection the hold began on, under the lock, with a bounded wait; requeue raced writers at the head Second round of review follow-ups: - Each bridge records the connection current when the hold begins and only ever retires that instance (TryRetireQuiesceTargetAsync). A replacement established meanwhile is never touched however often the drain loop retries, which the previous pass could do while the other bridge was still draining; and if the target went away by itself there is nothing to retire. - Every retirement happens under the write lock: the forced pass at the cap as well. The lock wait is bounded by a cancellation token the drain owns (the cap plus a second of grace) rather than the mutex timeout; if the lock cannot be had even then, the connection is left to the server's close rather than recycled around a writer. - A writer that lost its connection to the handoff while waiting for the lock is requeued as held work (never failed under FailFast) at the head of the backlog, via a front queue every backlog operation drains first, so it keeps its place ahead of what the hold queued after it. - The cap is half the relaxed timeout (five seconds by default) bounded by the window, and timeouts are relaxed for the hold, so a blocking command with a timeout up to the cap completes and nothing held is timed out by the hold itself. A fixed second failed commands the server's close would have let finish; the whole window held everyone else's work for one of them. - Each bridge is released in its own try/finally and the processor wake is guarded, so a bridge disposed mid-handoff cannot leave the other quiesced. - The two unused in-flight predicates are gone; the requeue check is one helper at the three lock sites. Tests: a slow command that fits inside the cap is answered and the recycle follows it; a longer one is failed at the cap and traffic resumes on the replacement; the steady-load test is unchanged. --- docs/ServerMaintenanceEvent.md | 4 +- src/StackExchange.Redis/PhysicalBridge.cs | 202 ++++++++++++------ .../ServerEndPoint.Maintenance.cs | 99 ++++----- src/StackExchange.Redis/ServerEndPoint.cs | 7 - .../MaintenanceNotificationTests.cs | 63 ++++-- 5 files changed, 234 insertions(+), 141 deletions(-) diff --git a/docs/ServerMaintenanceEvent.md b/docs/ServerMaintenanceEvent.md index 530465c32..7fcb8201b 100644 --- a/docs/ServerMaintenanceEvent.md +++ b/docs/ServerMaintenanceEvent.md @@ -111,11 +111,11 @@ or `ServerDefault` to ask for nothing at all, which is what earlier versions did | `MIGRATING`, `FAILING_OVER`, `SMIGRATING` | relaxes command timeouts for that server while the disruption lasts | | `MIGRATED`, `FAILED_OVER` | ends the window, but keeps timeouts relaxed for a short tail while things settle | | `SMIGRATED` | as above, and re-reads the cluster topology, and re-subscribes any sharded channels whose slots moved | -| `MOVING` | works out the replacement address, holds new commands back for up to a second while in-flight ones finish, then replaces the connections and sends the held commands on the replacement | +| `MOVING` | works out the replacement address, holds new commands back while in-flight ones finish (for up to half the relaxed timeout, with timeouts relaxed meanwhile), then replaces the connections and sends the held commands on the replacement | So an application that does nothing at all still benefits: commands that would have timed out during a migration are given more room, a moved slot is learned without waiting to be redirected, and a `MOVING` is acted on before the server closes the socket. -During that handoff, commands the application issues are not written to the connection being retired: they wait in the backlog, for up to about a second, and are sent on the replacement once it is established. This happens whatever the `backlogPolicy`, because the connection is live and the command would otherwise have been written; and a `FailFast` policy does not abandon that held work when the old connection is closed, since a handoff is a replacement the client chose rather than a failure. A command still in flight when the second is up, such as a blocking read, is failed by the replacement, as the server's close would have failed it; the hold is bounded so that one such command cannot stall everything else for the rest of the window. +During that handoff, commands the application issues are not written to the connection being retired: they wait in the backlog and are sent on the replacement once it is established, in the order they were issued. This happens whatever the `backlogPolicy`, because the connection is live and the command would otherwise have been written; and a `FailFast` policy does not abandon that held work when the old connection is closed, since a handoff is a replacement the client chose rather than a failure. The hold lasts until the commands already in flight have been answered, and at most half of `maintRelaxedTimeout` (five seconds by default); timeouts are relaxed for its duration, so nothing is timed out by the hold itself, and a blocking command with a timeout up to that completes normally. A command still in flight when the hold runs out is failed by the replacement, as the server's close would have failed it, so that one blocking read cannot stall everything else for the rest of the window. ### Notifications that arrive as you connect diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index ee752a780..9603b3958 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -34,7 +34,15 @@ internal sealed class PhysicalBridge : IDisposable /// For now, simplicity in: queue it all, replay or timeout it all. /// private readonly ConcurrentQueue _backlog = new(); - private bool BacklogHasItems => !_backlog.IsEmpty; + + /// + /// Messages that go out before anything in : a writer that lost its connection to a + /// maintenance handoff while waiting for the write lock was issued before anything the hold has queued + /// since, so it is requeued here rather than appended. Every backlog operation drains this first. + /// + private readonly ConcurrentQueue _backlogFront = new(); + + private bool BacklogHasItems => !_backlogFront.IsEmpty || !_backlog.IsEmpty; private int _backlogProcessorIsRunning = 0; private int _backlogCurrentEnqueued = 0; private long _backlogTotalEnqueued = 0; @@ -133,10 +141,14 @@ public enum State : byte internal void MarkNeedsReconnect() => Volatile.Write(ref _needsReconnect, true); private volatile bool _quiescing; + private PhysicalConnection? _quiesceTarget; + private volatile bool _quiesceRecycled; /// /// Starts holding caller work back from the current connection while a maintenance handoff lets what has - /// already been written complete; see . + /// already been written complete; see . The connection + /// current at this moment is the one the handoff will retire, and the only one it will ever retire: a + /// replacement established meanwhile is left alone. /// /// /// Held work goes to the backlog rather than the socket, and is written to the replacement connection once @@ -144,18 +156,31 @@ public enum State : byte /// something in flight, so the recycle waited for the end of the announced window and then failed whatever /// was in flight at that instant. /// - internal void BeginQuiesce() => _quiescing = true; + internal void BeginQuiesce() + { + _quiesceTarget = physical; + _quiesceRecycled = false; + _quiescing = true; + } internal void EndQuiesce() { _quiescing = false; + _quiesceTarget = null; if (BacklogHasItems) KickBacklogProcessor(); } /// Gets held or queued work moving: the processor may be lingering between passes, or may have exited. private void KickBacklogProcessor() { - _backlogAutoReset.Set(); + try + { + _backlogAutoReset.Set(); + } + catch (ObjectDisposedException) + { + return; // a bridge disposed mid-handoff has no processor to wake, and nothing left to write + } StartBacklogProcessor(); } @@ -169,24 +194,58 @@ private bool ShouldHoldForHandoff(Message message) => _quiescing && IsConnected && !NeedsReconnect && !message.IsInternalCall && message.Command != RedisCommand.QUIT; /// - /// Replaces the connection once nothing a caller is waiting on is in flight, under the write lock: a writer that - /// passed the hold check before it was set, but has not yet written, cannot slip onto the old connection between - /// the check and the recycle. Such a writer finds the connection gone when it gets the lock, and queues for the - /// replacement instead (see ). + /// Retires the connection the hold began on, under the write lock, once nothing a caller is waiting on is in + /// flight on it - or regardless, when says the drain has waited long enough. Only + /// that connection: one that replaced it meanwhile is never touched, however often this is called. The lock + /// is what keeps a writer that passed the hold check before it was set from slipping onto the old connection + /// between the check and the recycle; such a writer finds the connection replaced when it gets the lock, and + /// queues for the replacement instead (see ). /// - /// null if caller work was still in flight, so the drain should keep waiting; otherwise whether - /// there was a connection to recycle. - internal async ValueTask TryRecycleWhenDrainedAsync(string reason) + /// Replace the connection even if caller work is still in flight on it. + /// Why; for the trace. + /// Bounds the wait for the write lock: the drain decides how long, not the mutex timeout. + /// null if the connection could not be retired yet - caller work in flight, or the lock not + /// obtained inside the deadline - so the drain should try again; otherwise whether this call, or an earlier + /// one, replaced it. + internal async ValueTask TryRetireQuiesceTargetAsync(bool force, string reason, CancellationToken lockDeadline) { - if (!await _singleWriter.TryTakeAsync().ForAwait()) + var target = _quiesceTarget; + if (target is null) return false; // nothing was connected when the hold began + if (_quiesceRecycled) return true; + + bool gotLock; + try + { + gotLock = await _singleWriter.TryTakeAsync(lockDeadline).ForAwait(); + } + catch (OperationCanceledException) { - // could not get the lock inside the write timeout: something is badly stuck, and the window is finite - return RecycleConnection(reason); + gotLock = false; } + if (!gotLock) return null; try { - if (HasCallerWorkInFlight()) return null; - return RecycleConnection(reason); + if (_quiesceRecycled) return true; + if (!ReferenceEquals(physical, target)) + { + // it went away by itself (the server closed it, or it faulted) and the ordinary reconnect owns + // what happens next; there is nothing left for the handoff to retire + _quiesceRecycled = true; + return false; + } + if (!force && target.HasCallerMessagesAwaitingResponse()) return null; + + Trace($"recycling {ConnectionType} connection: {reason}"); + + // Report *before* tearing down, and in this order for a reason: RecordConnectionFailed only raises + // the public event while the pipe is still live ("if *we* didn't burn the pipe: flag it"), and + // Dispose runs Shutdown first - which is exactly why an ordinary dispose is silent. Recording first + // also performs the disconnect, so the Dispose below is cleanup. + _recycledForHandoff = target; // a writer already past the hold check queues for the replacement instead + _quiesceRecycled = true; + target.RecordConnectionFailed(ConnectionFailureType.MaintenanceHandoff); + target.Dispose(); + return true; } finally { @@ -416,7 +475,7 @@ public override string ToString() => MessagesSinceLastHeartbeat = (int)(Volatile.Read(ref operationCount) - Volatile.Read(ref profileLastLog)), ConnectedAt = ConnectedAt, IsWriterActive = !_singleWriter.IsAvailable, - BacklogMessagesPending = _backlog.Count, + BacklogMessagesPending = _backlogFront.Count + _backlog.Count, BacklogMessagesPendingCounter = Volatile.Read(ref _backlogCurrentEnqueued), BacklogStatus = _backlogStatus, TotalBacklogMessagesQueued = _backlogTotalEnqueued, @@ -1008,7 +1067,7 @@ internal WriteResult WriteMessageTakingWriteLockSync(PhysicalConnection physical if (!gotLock) return TimedOutBeforeWrite(message); } - if (ConnectionReplacedWhileWaiting(physical, message)) return QueueOrFailMessage(message); + if (TryRequeueForHandoff(physical, message, bypassBacklog: false, out var requeued)) return requeued; var result = WriteMessageInsideLock(physical, message); if ((result == WriteResult.Success) & message.IsFlushRequiredSync) @@ -1074,7 +1133,7 @@ internal void AcceptRerouted(Message message) [MethodImpl(MethodImplOptions.AggressiveInlining)] private void BacklogEnqueue(Message message) { - bool wasEmpty = _backlog.IsEmpty; + bool wasEmpty = !BacklogHasItems; // important that this *precedes* enqueue, to play well with HasPendingCallerFacingItems Interlocked.Increment(ref _backlogCurrentEnqueued); Interlocked.Increment(ref _backlogTotalEnqueued); @@ -1095,7 +1154,7 @@ private void BacklogEnqueue(Message message) [MethodImpl(MethodImplOptions.AggressiveInlining)] private bool BacklogTryDequeue([NotNullWhen(true)] out Message? message) { - if (_backlog.TryDequeue(out message)) + if (_backlogFront.TryDequeue(out message) || _backlog.TryDequeue(out message)) { Interlocked.Decrement(ref _backlogCurrentEnqueued); return true; @@ -1103,6 +1162,24 @@ private bool BacklogTryDequeue([NotNullWhen(true)] out Message? message) return false; } + private bool BacklogTryPeek([NotNullWhen(true)] out Message? message) + => _backlogFront.TryPeek(out message) || _backlog.TryPeek(out message); + + /// + /// Requeues, ahead of everything else, a caller's message that lost its connection to a maintenance handoff + /// while waiting for the write lock: it was issued before anything the hold has queued since, and has not + /// been written. See . + /// + private void BacklogRequeueFront(Message message) + { + Interlocked.Increment(ref _backlogCurrentEnqueued); + Interlocked.Increment(ref _backlogTotalEnqueued); + _backlogFront.Enqueue(message); + message.SetBacklogged(); + // the hold may already have been released, in which case nothing else would pick this up before the heartbeat + if (!_quiescing) KickBacklogProcessor(); + } + [MethodImpl(MethodImplOptions.AggressiveInlining)] private void StartBacklogProcessor() { @@ -1149,13 +1226,6 @@ private void StartBacklogProcessor() } } - /// - /// Whether a caller is waiting on something that has actually been written to the current connection, as - /// opposed to queued in the backlog. This is what a handoff drains: backlogged work is held for the - /// replacement rather than abandoned, so it must not keep the old connection alive. - /// - internal bool HasCallerWorkInFlight() => physical?.HasCallerMessagesAwaitingResponse() == true; - /// /// Whether this bridge owes a *caller* anything, ignoring our own internal traffic. /// @@ -1163,7 +1233,11 @@ internal bool HasCallerWork() { if (physical?.HasCallerMessagesAwaitingResponse() == true) return true; - foreach (var message in _backlog) // snapshot enumeration; a concurrent queue is fine to walk + foreach (var message in _backlogFront) // snapshot enumeration; a concurrent queue is fine to walk + { + if (message.IsCallerFacing) return true; + } + foreach (var message in _backlog) { if (message.IsCallerFacing) return true; } @@ -1187,7 +1261,7 @@ private void CheckBacklogForTimeouts() // Because peeking at the backlog, checking message and then dequeuing, is not thread-safe, we do have to use // a lock here, for mutual exclusion of backlog DEQUEUERS. Unfortunately. // But we reduce contention by only locking if we see something that looks timed out. - while (_backlog.TryPeek(out Message? message)) + while (BacklogTryPeek(out Message? message)) { // See if the message has pass our async timeout threshold // Note: All timed out messages must be dequeued, even when no completion is needed, to be able to dequeue and complete other timed out messages. @@ -1196,7 +1270,7 @@ private void CheckBacklogForTimeouts() { // Peek again since we didn't have lock before... // and rerun the exact same checks as above, note that it may be a different message now - if (!_backlog.TryPeek(out message)) break; + if (!BacklogTryPeek(out message)) break; if (!message.HasTimedOut(now, timeout, out var _)) break; if (!BacklogTryDequeue(out var message2) || (message != message2)) // consume it for real @@ -1251,7 +1325,7 @@ private void ProcessBacklog() { while (!isDisposed) { - if (!_backlog.IsEmpty) + if (BacklogHasItems) { // TODO: vNext handoff this backlog to another primary ("can handle everything") connection // and remove any per-server commands. This means we need to track a bit of whether something @@ -1348,7 +1422,7 @@ private void ProcessBridgeBacklog() while (IsConnected) { // check whether the backlog is empty *before* even trying to get the lock - if (_backlog.IsEmpty) return; // nothing to do + if (!BacklogHasItems) return; // nothing to do // try and get the lock; if unsuccessful, retry gotLock = _singleWriter.TryTakeSync(); @@ -1461,7 +1535,11 @@ public bool HasPendingCallerFacingItems() { if (BacklogHasItems) { - foreach (var item in _backlog) // non-consuming, thread-safe, etc + foreach (var item in _backlogFront) // non-consuming, thread-safe, etc + { + if (item.IsCallerFacing) return true; + } + foreach (var item in _backlog) { if (item.IsCallerFacing) return true; } @@ -1480,19 +1558,29 @@ private WriteResult TimedOutBeforeWrite(Message message) /// /// The connection most recently replaced by a maintenance handoff, so that a writer which was headed for it - /// can tell a planned replacement from an ordinary closure. + /// can tell a planned replacement from an ordinary closure. Written and read under the write lock. /// private PhysicalConnection? _recycledForHandoff; /// - /// Whether the connection a caller's message was headed for was replaced by a maintenance handoff while the - /// writer waited for the lock. The message has not been written, so it is queued for the replacement rather - /// than written to a connection that is gone. Only the handoff case: a write that loses its connection to an - /// ordinary closure is still reported as a closure (#3167), and a message bypassing the backlog is never - /// queued. Internal traffic is exempt as well: the handshake writes to a connection before it is current. + /// For a writer that has just taken the write lock: if the connection it was headed for was replaced by a + /// maintenance handoff while it waited, the message - not yet written - is queued for the replacement at the + /// head of the backlog, since it was issued before anything the hold has queued since. Only the handoff case: + /// a write that loses its connection to an ordinary closure is still reported as a closure (#3167), a message + /// bypassing the backlog is never queued, and internal traffic is exempt because the handshake writes to a + /// connection before it is current. /// - private bool ConnectionReplacedWhileWaiting(PhysicalConnection physical, Message message, bool bypassBacklog = false) - => !bypassBacklog && !message.IsInternalCall && !ReferenceEquals(physical, this.physical) && ReferenceEquals(physical, _recycledForHandoff); + private bool TryRequeueForHandoff(PhysicalConnection physical, Message message, bool bypassBacklog, out WriteResult result) + { + if (bypassBacklog || message.IsInternalCall || ReferenceEquals(physical, this.physical) || !ReferenceEquals(physical, _recycledForHandoff)) + { + result = default; + return false; + } + BacklogRequeueFront(message); + result = WriteResult.Success; // queued counts as success + return true; + } /// /// This writes a message to the output stream. @@ -1536,7 +1624,7 @@ internal ValueTask WriteMessageTakingWriteLockAsync(PhysicalConnect gotLock = pending.GetAwaiter().GetResult(); // fine since we know we got a result if (!gotLock) return new ValueTask(TimedOutBeforeWrite(message)); } - if (ConnectionReplacedWhileWaiting(physical, message, bypassBacklog)) return new ValueTask(QueueOrFailMessage(message)); + if (TryRequeueForHandoff(physical, message, bypassBacklog, out var requeued)) return new ValueTask(requeued); var result = WriteMessageInsideLock(physical, message); if ((result == WriteResult.Success) & message.IsFlushRequiredAsync) { @@ -1576,7 +1664,7 @@ private async ValueTask WriteMessageTakingWriteLockAsync_Awaited( { gotLock = await pending.ForAwait(); if (!gotLock) return TimedOutBeforeWrite(message); - if (ConnectionReplacedWhileWaiting(physical, message, bypassBacklog)) return QueueOrFailMessage(message); + if (TryRequeueForHandoff(physical, message, bypassBacklog, out var requeued)) return requeued; var result = WriteMessageInsideLock(physical, message); if (result == WriteResult.Success) @@ -1968,34 +2056,6 @@ internal void SimulateConnectionFailure(SimulatedFailureType failureType) physical?.SimulateConnectionFailure(failureType); } - /// - /// Drops the current connection so that a fresh one is established. - /// - /// - /// Deliberately just a dispose: that routes through RecordConnectionFailed to - /// , which reconnects immediately by itself, so there is no new lifecycle - /// here to get wrong. Used by the MOVING handoff, where the point is to choose *when* the - /// connection is replaced - after the replacement address is known - rather than waiting for the server - /// to close it and re-resolving to whatever DNS says at that moment, which has been measured as still - /// naming the node being retired. - /// - internal bool RecycleConnection(string reason) - { - var current = physical; - if (current is null) return false; - - Multiplexer.Trace($"recycling {ConnectionType} connection: {reason}", ToString()); - - // Report *before* tearing down, and in this order for a reason: RecordConnectionFailed only raises - // the public event while the pipe is still live ("if *we* didn't burn the pipe: flag it"), and - // Dispose runs Shutdown first - which is exactly why an ordinary dispose is silent. Recording first - // also performs the disconnect, so the Dispose below is cleanup. - _recycledForHandoff = current; // a writer already past the hold check queues for the replacement instead - current.RecordConnectionFailed(ConnectionFailureType.MaintenanceHandoff); - current.Dispose(); - return true; - } - internal RedisCommand? GetActiveMessage() => Volatile.Read(ref _activeMessage)?.Command; } } diff --git a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs index b65f7893a..b5ec5c401 100644 --- a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs +++ b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs @@ -471,11 +471,28 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? } /// - /// How long a handoff waits for in-flight commands before replacing the connections regardless. Ordinary - /// commands clear in a round trip; anything still in flight after this is a blocking or stuck command that will - /// not finish inside the window either, and holding everyone else's work for it would be a regression. + /// How long a handoff holds new work and waits for in-flight commands before replacing the connections + /// regardless: half the relaxed timeout (five seconds by default), bounded by what is left of the window. /// - private static readonly TimeSpan DrainCap = TimeSpan.FromSeconds(1); + /// + /// Two things pull in opposite directions. A blocking command in flight - BLPOP, XREAD BLOCK, WAIT - is failed + /// by the replacement if the wait is too short, when the server's close would have let it finish; but every + /// command the application issues during the wait is held, so a long wait is a latency spike for everyone. + /// Half the relaxed timeout lets a blocking command with a timeout up to that complete, keeps the hold well + /// inside the relaxed timeout that applies to the held work for the duration (so nothing held times out + /// because of the hold), and follows the one setting an application already uses to say how patient it is + /// prepared to be during maintenance. A cap derived from the window alone would be wrong in both directions: a + /// 15s window is not a reason to hold traffic for 15s, and a 2s window is not a reason to fail a 3s BLPOP that + /// the server would have answered. + /// + private TimeSpan DrainCap(TimeSpan budget) + { + var half = TimeSpan.FromTicks(Multiplexer.RawConfig.MaintenanceRelaxedTimeout.Ticks / 2); + return budget < half ? budget : half; + } + + /// How much longer than the cap the forced pass may wait for the write lock before giving the connection up to the server's close. + private static readonly TimeSpan LockGrace = TimeSpan.FromSeconds(1); /// /// Holds new caller work back, lets what has already been written finish, then replaces the connections. @@ -492,10 +509,12 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? /// recycle follows, and the held work is written to the replacement once it is established. /// /// - /// The wait is bounded by as well as by the window: a blocking read in flight would - /// otherwise hold every other caller's work for the whole window and still be failed at the end of it. Past - /// the cap the connections are replaced regardless, and whatever was still in flight fails as the server's - /// close would have failed it. + /// The wait is bounded by , and timeouts are relaxed for the hold so that neither the + /// held work nor a blocking command being waited for is timed out by the hold itself. Past the cap the + /// connections are replaced regardless, and whatever was still in flight fails as the server's close would + /// have failed it. Every replacement happens under the bridge's write lock, and only ever to the connection the + /// hold began on: a replacement established meanwhile is never touched, and if the lock cannot be had at all + /// the connection is left to the server's close rather than recycled around a writer. /// /// /// Both bridges, not just the one that was told. The measured blast radius is the *node*: four connections @@ -506,66 +525,50 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? private async Task DrainThenRecycleAsync(TimeSpan budget, string reason) { var watch = ValueStopwatch.StartNew(); - var cap = budget < DrainCap ? budget : DrainCap; + var cap = DrainCap(budget); + + // the held work, and any blocking command being waited for, must not be timed out by the hold itself + ExtendRelaxedWindow(cap + Multiplexer.RawConfig.MaintenanceRelaxedTimeout, "MOVING handoff"); + interactive?.BeginQuiesce(); subscription?.BeginQuiesce(); try { - bool drained, recycled; + using var lockDeadline = new CancellationTokenSource(cap + LockGrace); + bool? interactiveResult = interactive is null ? false : null; + bool? subscriptionResult = subscription is null ? false : null; + bool forced = false; while (true) { - var attempt = await TryRecycleWhenDrainedAsync(reason).ForAwait(); - if (attempt is { } done) - { - drained = true; - recycled = done; - break; - } - if (watch.ElapsedMilliseconds >= cap.TotalMilliseconds) + var force = watch.ElapsedMilliseconds >= cap.TotalMilliseconds; + forced |= force; + if (interactiveResult is null) interactiveResult = await interactive!.TryRetireQuiesceTargetAsync(force, reason, lockDeadline.Token).ForAwait(); + if (subscriptionResult is null) subscriptionResult = await subscription!.TryRetireQuiesceTargetAsync(force, reason, lockDeadline.Token).ForAwait(); + if (interactiveResult is not null && subscriptionResult is not null) break; + if (lockDeadline.IsCancellationRequested) { - drained = false; - recycled = (interactive?.RecycleConnection(reason) == true) | (subscription?.RecycleConnection(reason) == true); + // the write lock could not be had even for the forced pass: something is stuck on it. Recycling + // without the lock would reopen the mid-write loss the lock exists to prevent, so the connection + // is left to the server's close, which is where things stood before handoffs existed break; } await Task.Delay(TimeSpan.FromMilliseconds(20)).ForAwait(); } + var recycled = interactiveResult == true || subscriptionResult == true; if (recycled) Interlocked.Increment(ref _handoffRecycles); Multiplexer.Trace( $"MOVING: {(recycled ? "recycled" : "nothing to recycle")} after {watch.ElapsedMilliseconds}ms" - + (drained ? " (drained)" : " (still busy at the drain cap)"), + + (forced ? " (still busy at the drain cap)" : " (drained)") + + (interactiveResult is null || subscriptionResult is null ? " (write lock not obtained; left to the server)" : ""), ToString()); } finally { - // the recycle has already nulled the old connection, so releasing the hold cannot write to it; the - // held work waits in the backlog for the replacement, as it would across any reconnect - interactive?.EndQuiesce(); - subscription?.EndQuiesce(); - } - } - - /// - /// Each bridge in turn, under its write lock: null as soon as one still has caller work in flight, - /// otherwise whether anything was recycled. A bridge recycled on an earlier pass has nothing left to recycle, - /// so a partial pass is safe to repeat. - /// - private async ValueTask TryRecycleWhenDrainedAsync(string reason) - { - bool recycled = false; - if (interactive is { } i) - { - var result = await i.TryRecycleWhenDrainedAsync(reason).ForAwait(); - if (result is null) return null; - recycled |= result.Value; - } - if (subscription is { } s) - { - var result = await s.TryRecycleWhenDrainedAsync(reason).ForAwait(); - if (result is null) return null; - recycled |= result.Value; + // each bridge is released on its own: a failure releasing one must not leave the other holding work + try { interactive?.EndQuiesce(); } catch (Exception ex) { Multiplexer.OnInternalError(ex, EndPoint); } + try { subscription?.EndQuiesce(); } catch (Exception ex) { Multiplexer.OnInternalError(ex, EndPoint); } } - return recycled; } /// diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index f7da7a062..e203139c6 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -392,13 +392,6 @@ internal bool IsIdle() internal bool HasCallerWork() => interactive?.HasCallerWork() == true || subscription?.HasCallerWork() == true; - /// - /// Whether a caller is waiting on something already written to a current connection; backlogged work does - /// not count. See . - /// - internal bool HasCallerWorkInFlight() - => interactive?.HasCallerWorkInFlight() == true || subscription?.HasCallerWorkInFlight() == true; - /// /// Work this server still owes an answer on: written-and-awaiting-response, plus anything queued in /// the backlog. Zero means a retirement can complete without abandoning anyone. diff --git a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs index 1c8fe4319..f503c0b90 100644 --- a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs +++ b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs @@ -913,20 +913,20 @@ await Poll.UntilAsync(() => Volatile.Read(ref completed) > before + 20, timeoutM } [Fact] - public async Task MovingWithSlowCommandsInFlightRecyclesAtTheDrainCap() + public async Task MovingWaitsForSlowCommandsThatFitInsideTheDrainCap() { - // A blocking or very slow command must not hold everyone else's work in the backlog for the whole window: - // the drain waits a bounded time for in-flight commands and then recycles regardless. The slow commands are - // failed by the recycle, as the server's close would have failed them; everything issued afterwards flows. + // A slow or blocking command that will finish inside the drain cap (half the relaxed timeout, 5s by default) + // is allowed to: the hold waits for it, timeouts are relaxed meanwhile, and the recycle follows its reply. + // The server's close would have let it finish too, so failing it would be a regression. var logs = new CapturingLoggerFactory(); var (server, conn, events) = await ConnectAsync(log, logs); using (server) await using (conn) { - server.SetLatency(TimeSpan.FromSeconds(3)); // every command now outlasts the drain cap + server.SetLatency(TimeSpan.FromSeconds(3)); // the in-process server answers one request at a time, so one slow command, not several var db = conn.GetDatabase(); - var slow = Enumerable.Range(0, 3).Select(_ => db.PingAsync()).ToArray(); - await Task.Delay(250); // let them be written + var slow = new[] { db.PingAsync() }; + await Task.Delay(250); // let it be written var endpoint = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(server.DefaultEndPoint); var successor = new IPEndPoint(IPAddress.Parse("127.0.0.9"), 6380); @@ -936,17 +936,54 @@ public async Task MovingWithSlowCommandsInFlightRecyclesAtTheDrainCap() Assert.Equal(MaintenanceNotificationType.Moving, moving.NotificationType); Assert.True( - await Poll.UntilAsync(() => endpoint.HandoffRecycles > 0, timeoutMilliseconds: 8_000), + await Poll.UntilAsync(() => endpoint.HandoffRecycles > 0, timeoutMilliseconds: 10_000), + "the handoff should have replaced the connection"); + var recycledAfter = watch.Elapsed; + server.SetLatency(TimeSpan.Zero); + log.WriteLine($"recycled after {recycledAfter.TotalMilliseconds:0}ms with a 3s command in flight"); + + // the slow command was answered, not failed: the recycle waited for it + foreach (var pending in slow) Assert.True(await pending >= TimeSpan.Zero); + Assert.True(recycledAfter >= TimeSpan.FromSeconds(2), $"recycled after {recycledAfter}: before the in-flight command could have been answered"); + Assert.True(recycledAfter < TimeSpan.FromSeconds(6), $"recycled after {recycledAfter}: later than the cap plus jitter"); + } + } + + [Fact] + public async Task MovingWithBlockingCommandsInFlightRecyclesAtTheDrainCap() + { + // A command that will not finish inside the cap - a long blocking read - must not hold everyone else's work + // in the backlog for the whole window: at the cap the connection is replaced regardless. The blocking command + // is failed by the recycle, as the server's close would have failed it; everything issued afterwards flows. + var logs = new CapturingLoggerFactory(); + var (server, conn, events) = await ConnectAsync(log, logs); + using (server) + await using (conn) + { + server.SetLatency(TimeSpan.FromSeconds(9)); // longer than the 5s cap; shorter than the relaxed 10s timeout + var db = conn.GetDatabase(); + var blocking = Enumerable.Range(0, 3).Select(_ => db.PingAsync()).ToArray(); + await Task.Delay(250); // let them be written + + var endpoint = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(server.DefaultEndPoint); + var successor = new IPEndPoint(IPAddress.Parse("127.0.0.9"), 6380); + var watch = System.Diagnostics.Stopwatch.StartNew(); + server.SendMoving(null, timeSeconds: 15, newEndpoint: successor, sequenceId: 0); + var moving = await events.NextAsync(); + Assert.Equal(MaintenanceNotificationType.Moving, moving.NotificationType); + + Assert.True( + await Poll.UntilAsync(() => endpoint.HandoffRecycles > 0, timeoutMilliseconds: 12_000), "the handoff should have replaced the connection"); var recycledAfter = watch.Elapsed; server.SetLatency(TimeSpan.Zero); // so the replacement's handshake is not slowed too - log.WriteLine($"recycled after {recycledAfter.TotalMilliseconds:0}ms with slow commands in flight"); + log.WriteLine($"recycled after {recycledAfter.TotalMilliseconds:0}ms with 9s commands in flight"); - // jitter is at most a tenth of the window, the cap one second: well short of the 3s the commands need - Assert.True(recycledAfter < TimeSpan.FromSeconds(3), $"recycled after {recycledAfter}: the drain waited for the slow commands instead of capping"); + // the cap (5s) plus at most a second of jitter: well short of the 9s the commands need + Assert.True(recycledAfter < TimeSpan.FromSeconds(7), $"recycled after {recycledAfter}: the drain waited for the blocking commands instead of capping"); // the in-flight commands were failed by the recycle, which is what the server's close would have done - foreach (var pending in slow) + foreach (var pending in blocking) { await Assert.ThrowsAnyAsync(async () => await pending); } @@ -964,7 +1001,7 @@ await Poll.UntilAsync( return false; } }, - timeoutMilliseconds: 30_000), + timeoutMilliseconds: 40_000), "the client should be serving commands on the replacement connection"); } } From bd786f3bbec5d21901daf419b4a824ab5ea054ba Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 4 Nov 2026 16:38:04 +0000 Subject: [PATCH 4/5] Handoff drain: head inserts under the backlog lock, marker cleared only at the next hold, per-bridge release, cap floor Third round of review follow-ups: - BacklogRequeueFront takes lock(_backlog) around the head insert. The timeout sweep peeks, then dequeues and compares under that lock; an unlocked insert between the two handed it a different message, which it treats as a threading bug, and the dequeued message was dropped uncompleted. - _recycledForHandoff is cleared when the next hold begins, not when the replacement establishes: the write lock is not fair, so a writer parked on it can be bypassed by the backlog processor long enough for the replacement to come up, and clearing then made it write to the disposed connection and tear down the healthy replacement. - DrainCap has a one-second floor, so a relaxed timeout of zero (relaxation disabled) does not force the recycle with commands still in flight. - Each bridge is released as its own retirement resolves, not when both have; the interactive bridge usually drains in a round trip and must not hold work while the subscription bridge takes longer. EndQuiesce is idempotent and kicks the processor unconditionally on the transition, closing the store-load race with a concurrent enqueuer. - The drain works on snapshots of the bridge fields, and the mutex take and release tolerate disposal mid-handoff. - A batch carrying a QUIT is not held at all, so the hold decision and the per-message action agree. - The lockDeadline comment no longer claims the mutex timeout plays no part. New test: the steady-load handoff with maintRelaxedTimeout=0. --- docs/ServerMaintenanceEvent.md | 2 +- src/StackExchange.Redis/PhysicalBridge.cs | 46 +++++++++++++++---- .../ServerEndPoint.Maintenance.cs | 40 ++++++++++++---- .../MaintenanceNotificationTests.cs | 16 +++++-- 4 files changed, 81 insertions(+), 23 deletions(-) diff --git a/docs/ServerMaintenanceEvent.md b/docs/ServerMaintenanceEvent.md index 7fcb8201b..4579d9e20 100644 --- a/docs/ServerMaintenanceEvent.md +++ b/docs/ServerMaintenanceEvent.md @@ -115,7 +115,7 @@ or `ServerDefault` to ask for nothing at all, which is what earlier versions did So an application that does nothing at all still benefits: commands that would have timed out during a migration are given more room, a moved slot is learned without waiting to be redirected, and a `MOVING` is acted on before the server closes the socket. -During that handoff, commands the application issues are not written to the connection being retired: they wait in the backlog and are sent on the replacement once it is established, in the order they were issued. This happens whatever the `backlogPolicy`, because the connection is live and the command would otherwise have been written; and a `FailFast` policy does not abandon that held work when the old connection is closed, since a handoff is a replacement the client chose rather than a failure. The hold lasts until the commands already in flight have been answered, and at most half of `maintRelaxedTimeout` (five seconds by default); timeouts are relaxed for its duration, so nothing is timed out by the hold itself, and a blocking command with a timeout up to that completes normally. A command still in flight when the hold runs out is failed by the replacement, as the server's close would have failed it, so that one blocking read cannot stall everything else for the rest of the window. +During that handoff, commands the application issues are not written to the connection being retired: they wait in the backlog and are sent on the replacement once it is established, in the order they were issued. This happens whatever the `backlogPolicy`, because the connection is live and the command would otherwise have been written; and a `FailFast` policy does not abandon that held work when the old connection is closed, since a handoff is a replacement the client chose rather than a failure. The hold lasts until the commands already in flight have been answered, and at most half of `maintRelaxedTimeout` (five seconds by default, and never less than a second); timeouts are relaxed for its duration, so nothing is timed out by the hold itself, and a blocking command with a timeout up to that completes normally. A command still in flight when the hold runs out is failed by the replacement, as the server's close would have failed it, so that one blocking read cannot stall everything else for the rest of the window. ### Notifications that arrive as you connect diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 9603b3958..2f4d1b773 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -158,6 +158,7 @@ public enum State : byte /// internal void BeginQuiesce() { + _recycledForHandoff = null; // the previous handoff's marker; any writer parked across two handoffs is on its own _quiesceTarget = physical; _quiesceRecycled = false; _quiescing = true; @@ -165,9 +166,13 @@ internal void BeginQuiesce() internal void EndQuiesce() { + if (!_quiescing) return; _quiescing = false; _quiesceTarget = null; - if (BacklogHasItems) KickBacklogProcessor(); + // Unconditionally, not only when the backlog has items: an enqueuer that read the flag just before it + // flipped did not kick, and neither side's volatile access orders its store before the other's load. A + // kick with nothing queued costs one idle pass of the processor. + KickBacklogProcessor(); } /// Gets held or queued work moving: the processor may be lingering between passes, or may have exited. @@ -203,7 +208,8 @@ private bool ShouldHoldForHandoff(Message message) /// /// Replace the connection even if caller work is still in flight on it. /// Why; for the trace. - /// Bounds the wait for the write lock: the drain decides how long, not the mutex timeout. + /// Bounds the wait for the write lock (together with the mutex's own timeout, whichever + /// ends first); the drain re-waits until this deadline, so it is the deadline that decides how long overall. /// null if the connection could not be retired yet - caller work in flight, or the lock not /// obtained inside the deadline - so the drain should try again; otherwise whether this call, or an earlier /// one, replaced it. @@ -222,6 +228,10 @@ private bool ShouldHoldForHandoff(Message message) { gotLock = false; } + catch (ObjectDisposedException) + { + return false; // the bridge was disposed mid-handoff; there is nothing left to retire + } if (!gotLock) return null; try { @@ -249,7 +259,14 @@ private bool ShouldHoldForHandoff(Message message) } finally { - _singleWriter.Release(); + try + { + _singleWriter.Release(); + } + catch (ObjectDisposedException) + { + // disposed while we held it; nobody is waiting on a disposed mutex + } } } @@ -690,7 +707,6 @@ internal void OnFullyEstablished(PhysicalConnection connection, string source) { reportNextFailure = reconfigureNextFailure = true; LastException = null; - _recycledForHandoff = null; Interlocked.Exchange(ref failConnectCount, 0); ServerEndPoint.OnFullyEstablished(connection, source); @@ -904,10 +920,11 @@ internal bool TryEnqueue(List messages, bool isReplica) if (isDisposed) throw new ObjectDisposedException(Name); - if (_quiescing && IsConnected && !NeedsReconnect && messages.Exists(m => !m.IsInternalCall && m.Command != RedisCommand.QUIT)) + if (_quiescing && IsConnected && !NeedsReconnect && messages.Exists(m => !m.IsInternalCall) && !messages.Exists(m => m.Command == RedisCommand.QUIT)) { // a maintenance handoff is draining the connection: hold the batch for the replacement. A batch is - // written as a unit, so it is held as a unit; a batch of purely internal traffic is not held at all + // written as a unit, so it is held as a unit; a batch of purely internal traffic is not held at all, + // and nor is one carrying a QUIT, which is never queued - the whole batch goes to the live connection foreach (var message in messages) { QueueOrFailMessage(message, holdForHandoff: true); @@ -1174,10 +1191,16 @@ private void BacklogRequeueFront(Message message) { Interlocked.Increment(ref _backlogCurrentEnqueued); Interlocked.Increment(ref _backlogTotalEnqueued); - _backlogFront.Enqueue(message); + // Under the dequeuers' lock: CheckBacklogForTimeouts peeks, then dequeues and compares under this lock, + // and an insert at the head between the two would hand it a different message - which it treats as a + // threading bug, and the dequeued message would be dropped uncompleted. Only locked parties change the head. + lock (_backlog) + { + _backlogFront.Enqueue(message); + } message.SetBacklogged(); - // the hold may already have been released, in which case nothing else would pick this up before the heartbeat - if (!_quiescing) KickBacklogProcessor(); + // unconditionally: the hold may be released concurrently, and a kick with the hold still on costs one idle pass + KickBacklogProcessor(); } [MethodImpl(MethodImplOptions.AggressiveInlining)] @@ -1558,7 +1581,10 @@ private WriteResult TimedOutBeforeWrite(Message message) /// /// The connection most recently replaced by a maintenance handoff, so that a writer which was headed for it - /// can tell a planned replacement from an ordinary closure. Written and read under the write lock. + /// can tell a planned replacement from an ordinary closure. Set under the write lock when the connection is + /// retired, read under it by the writer, and cleared only when the next hold begins: the writer may have + /// been waiting for the lock long enough for the replacement to be fully established, since the lock is not + /// fair and the backlog processor may keep taking it to flush held work. /// private PhysicalConnection? _recycledForHandoff; diff --git a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs index b5ec5c401..6894ebc82 100644 --- a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs +++ b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs @@ -488,9 +488,13 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? private TimeSpan DrainCap(TimeSpan budget) { var half = TimeSpan.FromTicks(Multiplexer.RawConfig.MaintenanceRelaxedTimeout.Ticks / 2); + if (half < DrainFloor) half = DrainFloor; // a relaxed timeout of zero disables relaxation; it must not disable the drain return budget < half ? budget : half; } + /// The least the drain waits for in-flight commands, whatever the relaxed timeout: one round trip on any network, with room to spare. + private static readonly TimeSpan DrainFloor = TimeSpan.FromSeconds(1); + /// How much longer than the cap the forced pass may wait for the write lock before giving the connection up to the server's close. private static readonly TimeSpan LockGrace = TimeSpan.FromSeconds(1); @@ -530,20 +534,25 @@ private async Task DrainThenRecycleAsync(TimeSpan budget, string reason) // the held work, and any blocking command being waited for, must not be timed out by the hold itself ExtendRelaxedWindow(cap + Multiplexer.RawConfig.MaintenanceRelaxedTimeout, "MOVING handoff"); - interactive?.BeginQuiesce(); - subscription?.BeginQuiesce(); + // snapshots: Dispose nulls the fields, and a handoff in progress must not fault on a server going away + var interactiveBridge = interactive; + var subscriptionBridge = subscription; + interactiveBridge?.BeginQuiesce(); + subscriptionBridge?.BeginQuiesce(); try { using var lockDeadline = new CancellationTokenSource(cap + LockGrace); - bool? interactiveResult = interactive is null ? false : null; - bool? subscriptionResult = subscription is null ? false : null; + bool? interactiveResult = interactiveBridge is null ? false : null; + bool? subscriptionResult = subscriptionBridge is null ? false : null; bool forced = false; while (true) { var force = watch.ElapsedMilliseconds >= cap.TotalMilliseconds; forced |= force; - if (interactiveResult is null) interactiveResult = await interactive!.TryRetireQuiesceTargetAsync(force, reason, lockDeadline.Token).ForAwait(); - if (subscriptionResult is null) subscriptionResult = await subscription!.TryRetireQuiesceTargetAsync(force, reason, lockDeadline.Token).ForAwait(); + // each bridge is released the moment its own connection is settled: the interactive bridge usually + // drains in a round trip, and must not keep holding work while the subscription bridge takes longer + if (interactiveResult is null && (interactiveResult = await interactiveBridge!.TryRetireQuiesceTargetAsync(force, reason, lockDeadline.Token).ForAwait()) is not null) Release(interactiveBridge); + if (subscriptionResult is null && (subscriptionResult = await subscriptionBridge!.TryRetireQuiesceTargetAsync(force, reason, lockDeadline.Token).ForAwait()) is not null) Release(subscriptionBridge); if (interactiveResult is not null && subscriptionResult is not null) break; if (lockDeadline.IsCancellationRequested) { @@ -565,9 +574,22 @@ private async Task DrainThenRecycleAsync(TimeSpan budget, string reason) } finally { - // each bridge is released on its own: a failure releasing one must not leave the other holding work - try { interactive?.EndQuiesce(); } catch (Exception ex) { Multiplexer.OnInternalError(ex, EndPoint); } - try { subscription?.EndQuiesce(); } catch (Exception ex) { Multiplexer.OnInternalError(ex, EndPoint); } + // whatever was not released in the loop (the lock-not-obtained exit, an exception): EndQuiesce is idempotent + Release(interactiveBridge); + Release(subscriptionBridge); + } + + void Release(PhysicalBridge? bridge) + { + if (bridge is null) return; + try + { + bridge.EndQuiesce(); + } + catch (Exception ex) + { + Multiplexer.OnInternalError(ex, EndPoint); // one bridge failing to release must not leave the other holding work + } } } diff --git a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs index f503c0b90..aa1de859e 100644 --- a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs +++ b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs @@ -27,13 +27,15 @@ public class MaintenanceNotificationTests(ITestOutputHelper log) /// private static async Task<(InProcessTestServer Server, ConnectionMultiplexer Connection, EventCollector Events)> ConnectAsync( ITestOutputHelper log, - ILoggerFactory? loggerFactory = null) + ILoggerFactory? loggerFactory = null, + Action? configure = null) { var server = new InProcessTestServer(log); var config = server.GetClientConfig(defaultOnly: true); config.Protocol = RedisProtocol.Resp3; config.MaintenanceNotifications = MaintenanceNotificationMode.Enabled; // must be live, or the test is vacuous config.LoggerFactory = loggerFactory; + configure?.Invoke(config); var conn = await ConnectionMultiplexer.ConnectAsync(config); return (server, conn, new EventCollector(conn)); @@ -844,7 +846,15 @@ await Poll.UntilAsync( } [Fact] - public async Task MovingUnderSteadyLoadRecyclesPromptlyAndFailsNothing() + public Task MovingUnderSteadyLoadRecyclesPromptlyAndFailsNothing() => SteadyLoadHandoffAsync(configure: null); + + [Fact] + public Task MovingUnderSteadyLoadWithRelaxationDisabledStillDrains() + // a relaxed timeout of zero is the documented way to turn relaxation off; the drain cap is derived from it, + // and must keep a floor rather than recycle at once with commands still in flight + => SteadyLoadHandoffAsync(configure: c => c.MaintenanceRelaxedTimeout = TimeSpan.Zero); + + private async Task SteadyLoadHandoffAsync(Action? configure) { // The drain used to wait for "no caller work", which under steady traffic never comes: measured live at // about 450 ops/s, the recycle arrived at the end of the 15s window and failed the eight commands in @@ -852,7 +862,7 @@ public async Task MovingUnderSteadyLoadRecyclesPromptlyAndFailsNothing() // written commands finish, so the recycle follows within a round trip and nothing fails: the held work is // written to the replacement. var logs = new CapturingLoggerFactory(); - var (server, conn, events) = await ConnectAsync(log, logs); + var (server, conn, events) = await ConnectAsync(log, logs, configure); using (server) await using (conn) { From a518499eca06070bf8a03991a2d783cf739e059a Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 4 Nov 2026 17:16:08 +0000 Subject: [PATCH 5/5] Handoff drain: survive the old connection dying mid-hold, cover held work's timeout, budget from what is left Fourth round of review follow-ups: - A connection failure during the hold no longer abandons the held work under a FailFast backlog policy: that backlog is work diverted off a live connection, which would have been written and answered, so it is left for the replacement to flush. Previously only the handoff's own recycle was exempt. - The hold imposes its own floor on the effective timeout for its duration plus the reconnect (the cap, the lock grace and the connect timeout), independent of relaxation: with maintRelaxedTimeout=0 and a sub-second configured timeout, the one-second drain floor would otherwise have timed out exactly the work the hold protects. - The front requeue roots the multiplexer on the empty-to-non-empty transition, as the ordinary enqueue does. - The drain budget is what is left of the window after the decision has spent its share polling DNS, not the window minus jitter; with less than the floor left there is no hold, and the connections are replaced at once (still under the lock) so the replacement gets what remains. - The handoff marker is a short ring of the last four retired connections, never cleared, so a writer starved across back-to-back MOVINGs still recognises its connection. - Abandoning the backlog dequeues under the dequeuers' lock; a quiesced backlog pass returns before contending for the write lock; the processor start's wake tolerates disposal. - One hold predicate, shared by the single-message gates and the batch gate, which makes one pass over the batch. - The docs and the DrainCap remarks now agree with the clamp: the window bounds the hold because the server closes the socket when it ends. --- docs/ServerMaintenanceEvent.md | 2 +- src/StackExchange.Redis/PhysicalBridge.cs | 86 +++++++++++++++---- .../ServerEndPoint.Maintenance.cs | 66 +++++++++++--- 3 files changed, 124 insertions(+), 30 deletions(-) diff --git a/docs/ServerMaintenanceEvent.md b/docs/ServerMaintenanceEvent.md index 4579d9e20..13d05eb8b 100644 --- a/docs/ServerMaintenanceEvent.md +++ b/docs/ServerMaintenanceEvent.md @@ -115,7 +115,7 @@ or `ServerDefault` to ask for nothing at all, which is what earlier versions did So an application that does nothing at all still benefits: commands that would have timed out during a migration are given more room, a moved slot is learned without waiting to be redirected, and a `MOVING` is acted on before the server closes the socket. -During that handoff, commands the application issues are not written to the connection being retired: they wait in the backlog and are sent on the replacement once it is established, in the order they were issued. This happens whatever the `backlogPolicy`, because the connection is live and the command would otherwise have been written; and a `FailFast` policy does not abandon that held work when the old connection is closed, since a handoff is a replacement the client chose rather than a failure. The hold lasts until the commands already in flight have been answered, and at most half of `maintRelaxedTimeout` (five seconds by default, and never less than a second); timeouts are relaxed for its duration, so nothing is timed out by the hold itself, and a blocking command with a timeout up to that completes normally. A command still in flight when the hold runs out is failed by the replacement, as the server's close would have failed it, so that one blocking read cannot stall everything else for the rest of the window. +During that handoff, commands the application issues are not written to the connection being retired: they wait in the backlog and are sent on the replacement once it is established, in the order they were issued. This happens whatever the `backlogPolicy`, because the connection is live and the command would otherwise have been written; and a `FailFast` policy does not abandon that held work when the old connection is closed, since a handoff is a replacement the client chose rather than a failure. The hold lasts until the commands already in flight have been answered, and at most half of `maintRelaxedTimeout` (five seconds by default, never less than a second, and never longer than what is left of the announced window, since the server closes the socket when that ends regardless). For the duration of the hold and the reconnect after it, commands against that server time out no sooner than that whole span, whatever `maintRelaxedTimeout` is set to, so nothing is timed out by the hold itself. A blocking command that was already in flight when the hold began and whose timeout fits inside the hold completes normally; one issued during the hold is held like everything else and runs on the replacement. A command still in flight when the hold runs out is failed by the replacement, as the server's close would have failed it, so that one blocking read cannot stall everything else for the rest of the window. ### Notifications that arrive as you connect diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index 2f4d1b773..a44eef52c 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -158,7 +158,6 @@ public enum State : byte /// internal void BeginQuiesce() { - _recycledForHandoff = null; // the previous handoff's marker; any writer parked across two handoffs is on its own _quiesceTarget = physical; _quiesceRecycled = false; _quiescing = true; @@ -195,8 +194,28 @@ private void KickBacklogProcessor() /// rules and the backlog policy apply), and only for a caller's work. Internal traffic (keep-alives, probes) /// still goes to the old connection, and so does QUIT, which is never queued. /// - private bool ShouldHoldForHandoff(Message message) - => _quiescing && IsConnected && !NeedsReconnect && !message.IsInternalCall && message.Command != RedisCommand.QUIT; + private bool ShouldHoldForHandoff(Message message) => HoldActive && IsHoldable(message); + + /// A handoff is draining a connection that is still live. + private bool HoldActive => _quiescing && IsConnected && !NeedsReconnect; + + /// A caller's work, and not QUIT, which is never queued. + private static bool IsHoldable(Message message) => !message.IsInternalCall && message.Command != RedisCommand.QUIT; + + /// + /// A batch is written as a unit, so it is held as a unit: when it carries any holdable message and no QUIT. + /// An internal member rides along with the rest, because splitting the batch would be worse than holding it. + /// + private static bool ShouldHoldBatch(List messages) + { + bool any = false; + foreach (var message in messages) + { + if (message.Command == RedisCommand.QUIT) return false; + any |= IsHoldable(message); + } + return any; + } /// /// Retires the connection the hold began on, under the write lock, once nothing a caller is waiting on is in @@ -251,7 +270,7 @@ private bool ShouldHoldForHandoff(Message message) // the public event while the pipe is still live ("if *we* didn't burn the pipe: flag it"), and // Dispose runs Shutdown first - which is exactly why an ordinary dispose is silent. Recording first // also performs the disconnect, so the Dispose below is cleanup. - _recycledForHandoff = target; // a writer already past the hold check queues for the replacement instead + _recycledForHandoff[_recycledForHandoffNext++ & 3] = target; // a writer already past the hold check queues for the replacement instead _quiesceRecycled = true; target.RecordConnectionFailed(ConnectionFailureType.MaintenanceHandoff); target.Dispose(); @@ -621,8 +640,10 @@ internal void OnConnectionFailed(PhysicalConnection connection, ConnectionFailur } Trace($"OnConnectionFailed: {connection}"); // If we're configured to, fail all pending backlogged messages - but not for a maintenance handoff, which - // is a replacement we chose, and whose backlog is work deliberately held for the replacement connection - if (Multiplexer.RawConfig.BacklogPolicy?.AbortPendingOnConnectionFailure == true && failureType != ConnectionFailureType.MaintenanceHandoff) + // is a replacement we chose, and not while a handoff is holding work back: that backlog is work diverted + // off a live connection, which would have been written and answered, so a death of the old connection + // during the hold (the window expiring, a blip) must leave it for the replacement to flush + if (Multiplexer.RawConfig.BacklogPolicy?.AbortPendingOnConnectionFailure == true && failureType != ConnectionFailureType.MaintenanceHandoff && !_quiescing) { AbandonPendingBacklog(innerException, connection); } @@ -686,8 +707,13 @@ internal void OnDisconnected(ConnectionFailureType failureType, PhysicalConnecti private void AbandonPendingBacklog(Exception ex, PhysicalConnection? connection) { - while (BacklogTryDequeue(out Message? next)) + while (true) { + Message? next; + lock (_backlog) // only locked parties change the head: the timeout sweep peeks, then dequeues and compares under this lock + { + if (!BacklogTryDequeue(out next)) break; + } // as in PhysicalConnection.RecordMessageFailed: don't hand the same exception to every // message. A backlogged message has provably never been written, which is exactly what the // retry machinery needs to know, and it is lost if we share the connection-level instance. @@ -920,11 +946,9 @@ internal bool TryEnqueue(List messages, bool isReplica) if (isDisposed) throw new ObjectDisposedException(Name); - if (_quiescing && IsConnected && !NeedsReconnect && messages.Exists(m => !m.IsInternalCall) && !messages.Exists(m => m.Command == RedisCommand.QUIT)) + if (HoldActive && ShouldHoldBatch(messages)) { - // a maintenance handoff is draining the connection: hold the batch for the replacement. A batch is - // written as a unit, so it is held as a unit; a batch of purely internal traffic is not held at all, - // and nor is one carrying a QUIT, which is never queued - the whole batch goes to the live connection + // a maintenance handoff is draining the connection: hold the batch for the replacement foreach (var message in messages) { QueueOrFailMessage(message, holdForHandoff: true); @@ -1189,6 +1213,7 @@ private bool BacklogTryPeek([NotNullWhen(true)] out Message? message) /// private void BacklogRequeueFront(Message message) { + bool wasEmpty = !BacklogHasItems; Interlocked.Increment(ref _backlogCurrentEnqueued); Interlocked.Increment(ref _backlogTotalEnqueued); // Under the dequeuers' lock: CheckBacklogForTimeouts peeks, then dequeues and compares under this lock, @@ -1199,6 +1224,7 @@ private void BacklogRequeueFront(Message message) _backlogFront.Enqueue(message); } message.SetBacklogged(); + if (wasEmpty) Multiplexer.Root(); // as BacklogEnqueue: a backlog with a caller's task in it must pin the multiplexer // unconditionally: the hold may be released concurrently, and a kick with the hold still on costs one idle pass KickBacklogProcessor(); } @@ -1245,7 +1271,14 @@ private void StartBacklogProcessor() } else { - _backlogAutoReset.Set(); + try + { + _backlogAutoReset.Set(); + } + catch (ObjectDisposedException) + { + // disposed between the check and the wake; the processor that was running is winding down anyway + } } } @@ -1447,6 +1480,10 @@ private void ProcessBridgeBacklog() // check whether the backlog is empty *before* even trying to get the lock if (!BacklogHasItems) return; // nothing to do + // a maintenance handoff is holding work back: nothing would be written, and the lock is better + // left to the drain and the writers it is waiting on; the release kicks this again + if (_quiescing) return; + // try and get the lock; if unsuccessful, retry gotLock = _singleWriter.TryTakeSync(); if (gotLock) break; // got the lock; now go do something with it @@ -1580,13 +1617,24 @@ private WriteResult TimedOutBeforeWrite(Message message) } /// - /// The connection most recently replaced by a maintenance handoff, so that a writer which was headed for it - /// can tell a planned replacement from an ordinary closure. Set under the write lock when the connection is - /// retired, read under it by the writer, and cleared only when the next hold begins: the writer may have - /// been waiting for the lock long enough for the replacement to be fully established, since the lock is not - /// fair and the backlog processor may keep taking it to flush held work. + /// The last few connections replaced by maintenance handoffs, so that a writer which was headed for one of + /// them can tell a planned replacement from an ordinary closure. Written and read under the write lock, and + /// never cleared: the lock is not fair, so a writer parked on it can be starved past the replacement being + /// established, and rolling maintenance is a series of MOVINGs, so past the *next* handoff as well. Four is + /// more handoffs than a writer can realistically be starved across; older entries are simply overwritten. /// - private PhysicalConnection? _recycledForHandoff; + private readonly PhysicalConnection?[] _recycledForHandoff = new PhysicalConnection?[4]; + private int _recycledForHandoffNext; + + /// Whether a maintenance handoff retired this connection; under the write lock. + private bool WasRecycledForHandoff(PhysicalConnection connection) + { + foreach (var recycled in _recycledForHandoff) + { + if (ReferenceEquals(connection, recycled)) return true; + } + return false; + } /// /// For a writer that has just taken the write lock: if the connection it was headed for was replaced by a @@ -1598,7 +1646,7 @@ private WriteResult TimedOutBeforeWrite(Message message) /// private bool TryRequeueForHandoff(PhysicalConnection physical, Message message, bool bypassBacklog, out WriteResult result) { - if (bypassBacklog || message.IsInternalCall || ReferenceEquals(physical, this.physical) || !ReferenceEquals(physical, _recycledForHandoff)) + if (bypassBacklog || message.IsInternalCall || ReferenceEquals(physical, this.physical) || !WasRecycledForHandoff(physical)) { result = default; return false; diff --git a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs index 6894ebc82..953903688 100644 --- a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs +++ b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs @@ -198,10 +198,38 @@ internal MaintenanceNotificationType ActiveMaintenanceType /// internal int GetEffectiveTimeoutMilliseconds(int configuredMilliseconds) { - if (GetRelaxedRemaining() <= 0) return configuredMilliseconds; + var result = configuredMilliseconds; + if (GetRelaxedRemaining() > 0) + { + var relaxed = (int)Multiplexer.RawConfig.MaintenanceRelaxedTimeout.TotalMilliseconds; + if (relaxed > result) result = relaxed; + } - var relaxed = (int)Multiplexer.RawConfig.MaintenanceRelaxedTimeout.TotalMilliseconds; - return relaxed > configuredMilliseconds ? relaxed : configuredMilliseconds; + // a handoff hold diverts commands off a live connection into the backlog; whatever the relaxation setting + // (zero disables it), those commands must not be timed out by the hold itself, so the hold imposes its + // own floor on the effective timeout for as long as it, and the reconnect after it, can take + var cover = Volatile.Read(ref _holdCoverDeadlineTicks); + if (cover != 0 && unchecked(cover - Environment.TickCount) > 0) + { + var floor = Volatile.Read(ref _holdCoverMilliseconds); + if (floor > result) result = floor; + } + return result; + } + + private int _holdCoverDeadlineTicks, _holdCoverMilliseconds; + + /// + /// For the duration of a handoff hold plus the reconnect that follows it, commands against this server time out + /// no sooner than that whole span. Independent of relaxation: a relaxed timeout of zero is the documented way to + /// turn relaxation off, and a sub-second configured timeout with the hold's one-second floor would otherwise + /// time out exactly the work the hold is protecting. + /// + private void SetHoldCover(TimeSpan span) + { + var ms = (int)span.TotalMilliseconds; + Volatile.Write(ref _holdCoverMilliseconds, ms); + Volatile.Write(ref _holdCoverDeadlineTicks, NudgeFromZero(unchecked(Environment.TickCount + ms))); } /// @@ -422,6 +450,10 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? resolve: Multiplexer.AddressResolver, log: message => Multiplexer.Trace(message, ToString())).ForAwait(); + // the decision may have spent a good part of the window polling DNS (4 to 19 seconds of a 15 second + // window has been measured); what the drain has to work with is what is actually left + remaining = Remaining(window, watch); + Multiplexer.Trace($"MOVING: {decision}", ToString()); _lastHandoffOutcome = decision.ToString(); @@ -440,7 +472,7 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? // go on. Half of the *announced* window, less whatever the jitter already spent. var half = TimeSpan.FromTicks(window.Ticks / 2) - jitter; if (half > TimeSpan.Zero) await Task.Delay(half).ForAwait(); - await DrainThenRecycleAsync(window - jitter - (half > TimeSpan.Zero ? half : TimeSpan.Zero), decision.Reason).ForAwait(); + await DrainThenRecycleAsync(Remaining(window, watch), decision.Reason).ForAwait(); replaced = true; break; case HandoffAction.MoveTo when decision.Target is { } target: @@ -470,6 +502,13 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? } } + /// What is left of an announced window; never negative. + private static TimeSpan Remaining(TimeSpan window, ValueStopwatch watch) + { + var left = window - TimeSpan.FromMilliseconds(watch.ElapsedMilliseconds); + return left > TimeSpan.Zero ? left : TimeSpan.Zero; + } + /// /// How long a handoff holds new work and waits for in-flight commands before replacing the connections /// regardless: half the relaxed timeout (five seconds by default), bounded by what is left of the window. @@ -481,9 +520,10 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? /// Half the relaxed timeout lets a blocking command with a timeout up to that complete, keeps the hold well /// inside the relaxed timeout that applies to the held work for the duration (so nothing held times out /// because of the hold), and follows the one setting an application already uses to say how patient it is - /// prepared to be during maintenance. A cap derived from the window alone would be wrong in both directions: a - /// 15s window is not a reason to hold traffic for 15s, and a 2s window is not a reason to fail a 3s BLPOP that - /// the server would have answered. + /// prepared to be during maintenance. The window still bounds it, because the server closes the socket when + /// the window ends whatever we do: a command that cannot finish inside the window fails either way, and holding + /// past it would only delay everything else. A cap derived from the window alone would be wrong the other way: + /// a 15s window is not a reason to hold traffic for 15s. /// private TimeSpan DrainCap(TimeSpan budget) { @@ -529,10 +569,16 @@ private TimeSpan DrainCap(TimeSpan budget) private async Task DrainThenRecycleAsync(TimeSpan budget, string reason) { var watch = ValueStopwatch.StartNew(); - var cap = DrainCap(budget); - - // the held work, and any blocking command being waited for, must not be timed out by the hold itself + // With less than the floor left of the window there is no hold worth having: the server is about to close + // the socket regardless, so the connections are replaced at once (still under the lock) and the replacement + // gets what little time remains. Otherwise the hold lasts up to the cap. + var cap = budget < DrainFloor ? TimeSpan.Zero : DrainCap(budget); + + // the held work, and any blocking command being waited for, must not be timed out by the hold itself, nor + // by the reconnect that follows it; the relaxed window covers it where relaxation is on, and the cover below + // covers it regardless ExtendRelaxedWindow(cap + Multiplexer.RawConfig.MaintenanceRelaxedTimeout, "MOVING handoff"); + SetHoldCover(cap + LockGrace + TimeSpan.FromMilliseconds(Multiplexer.RawConfig.ConnectTimeout)); // snapshots: Dispose nulls the fields, and a handoff in progress must not fault on a server going away var interactiveBridge = interactive;