From 87669f625e90dbee0e26ccb13e58f3026a4e6a30 Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 11 Nov 2026 10:39:16 +0000 Subject: [PATCH 1/2] MOVING handoff: hold new work in the backlog while the in-flight commands drain, so the recycle happens within a round trip instead of at the window end The drain waited for "no caller work", which under steady traffic never comes: there is always a command in flight, so the recycle waited out the whole announced window and failed whatever was in flight at that instant. Measured live (endpoint rebind, single proxy, ~900 ops/s): recycle at +20.0s against a MOVING at +6.0s, 16 failed commands, missed-deadline warning. While a handoff drains, the bridge holds a caller's work in the backlog instead of writing it to the connection about to be replaced; the backlog processor does not flush while the hold stands; the drain waits only for what is already written, capped at half the relaxed timeout (bounded by half the remaining window, floor one second); the connection is then recycled as before, the hold cleared, and the held work flushed to the replacement by the ordinary reconnect path. Timeouts are relaxed for the hold and the reconnect. A FailFast policy does not abandon held work for the handoff's own disconnect. Same scenario on this branch: recycle at +6.2s against a MOVING at +5.9s, 0 failed of 97k. Deliberately a flag and nothing more: no per-message timeout cover (with relaxation off and a timeout shorter than the hold, held commands can time out in the backlog, as on any reconnect), and no retire under the write lock (a writer that reaches the lock after the recycle gets a write fault, as before). Also: the drain budget is recomputed after the DNS poll, a second MOVING during a handoff updates the successor pin rather than being dropped, and the pin lasts the window so the second bridge dials the same successor. Tests: steady load (with and without relaxation), slow command inside the cap, blocking command past the cap; plus a fault-injector tier measurement that records the live timeline. --- docs/ServerMaintenanceEvent.md | 4 +- src/StackExchange.Redis/PhysicalBridge.cs | 90 +++++++- .../ServerEndPoint.Maintenance.cs | 126 +++++++++--- src/StackExchange.Redis/ServerEndPoint.cs | 3 - .../MovingUnderLoadScenarioTests.cs | 186 +++++++++++++++++ .../MaintenanceNotificationTests.cs | 193 +++++++++++++++++- 6 files changed, 563 insertions(+), 39 deletions(-) create mode 100644 tests/StackExchange.Redis.FaultInjector.Tests/MovingUnderLoadScenarioTests.cs diff --git a/docs/ServerMaintenanceEvent.md b/docs/ServerMaintenanceEvent.md index 3e61ecf33..6a372b8d1 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 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 a `MOVING` 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 that connection is live and the command would otherwise have been written to it; and a `FailFast` policy does not abandon the 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 more than half of what is left of the announced window, and never less than a second. Timeouts are relaxed for the hold and the reconnect, so with relaxation on nothing is timed out by the hold itself; an application that has turned relaxation off (`maintRelaxedTimeout=0`) and runs timeouts shorter than the hold can see held commands time out in the backlog, as they would in the backlog of any other reconnect. A command still in flight when the hold runs out - a long blocking read - is failed by the replacement, as the server's close would have failed it, so that it 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 ee3a7bfb2..03bc289de 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -132,6 +132,70 @@ public enum State : byte /// internal void MarkNeedsReconnect() => Volatile.Write(ref _needsReconnect, true); + /// + /// A maintenance handoff is draining the current connection: a caller's work is held in the backlog for the + /// replacement rather than written to a socket the server is about to close. 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. + /// + /// + /// Deliberately only a flag, and only while the connection is live: once it has been replaced, the ordinary + /// disconnected rules and the backlog policy apply, and the held work is flushed by the same path that + /// flushes any other backlog when the replacement establishes. The hold is not raced against the write lock, + /// so a writer that passed this gate before the hold began and reached the lock after the recycle gets a + /// write fault, exactly as it would have before; and a held command is covered by the relaxed window rather + /// than by anything per-message, so with relaxation turned off and a timeout shorter than the hold it can + /// time out in the backlog - as it would in the backlog of any other reconnect. + /// + private volatile bool _holdForHandoff; + + internal void BeginHandoffHold() => _holdForHandoff = true; + + /// Ends the hold; the connection it was draining has been replaced (or there was none). + internal void EndHandoffHold() + { + _holdForHandoff = false; + // the usual case is that the connection has just been recycled, and the reconnect's establishment flushes + // the backlog; this covers the one where nothing was recycled, so that nothing else would + if (BacklogHasItems) StartBacklogProcessor(); + } + + /// + /// Whether a message is held for the handoff rather than written: a caller's work (internal traffic such as + /// keep-alives still goes to the old connection, and QUIT is never queued), on a live connection. + /// + private bool ShouldHoldForHandoff(Message message) + => _holdForHandoff && !message.IsInternalCall && message.Command != RedisCommand.QUIT && IsConnected && !NeedsReconnect; + + /// Queues a message the hold diverted; queued counts as success, as it does for a disconnected backlog. + private WriteResult HoldForHandoff(Message message) + { + message.SetEnqueued(null); + BacklogEnqueue(message); + // the hold may have ended between the check and the enqueue; the processor itself does nothing while it + // stands, so starting it is always safe, and starting it here means nothing is stranded in that race + StartBacklogProcessor(); + return WriteResult.Success; + } + + /// + /// 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 |= !message.IsInternalCall; + } + return any; + } + + /// Whether a caller is waiting for a reply on the current connection; the backlog is not counted. + internal bool HasCallerMessagesInFlight() => physical?.HasCallerMessagesAwaitingResponse() == true; + public void Dispose() { isDisposed = true; @@ -212,6 +276,7 @@ private WriteResult FailDueToNoConnection(Message message) public WriteResult TryWriteSync(Message message, bool isReplica) { if (isDisposed) throw new ObjectDisposedException(Name); + if (ShouldHoldForHandoff(message)) return HoldForHandoff(message); if (!IsConnected || NeedsReconnect) return QueueOrFailMessage(message); var physical = this.physical; @@ -236,6 +301,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 (!bypassBacklog && ShouldHoldForHandoff(message)) return new ValueTask(HoldForHandoff(message)); if ((!IsConnected || NeedsReconnect) && !bypassBacklog) return new ValueTask(QueueOrFailMessage(message)); var physical = this.physical; @@ -477,8 +543,10 @@ 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: that + // is a replacement we chose, and the backlog then holds work diverted off a live connection, which would + // have been written and answered; it is for the replacement to flush + if (Multiplexer.RawConfig.BacklogPolicy?.AbortPendingOnConnectionFailure == true && failureType != ConnectionFailureType.MaintenanceHandoff) { AbandonPendingBacklog(innerException, connection); } @@ -776,6 +844,18 @@ internal bool TryEnqueue(List messages, bool isReplica) if (isDisposed) throw new ObjectDisposedException(Name); + if (_holdForHandoff && IsConnected && !NeedsReconnect && ShouldHoldBatch(messages)) + { + // a maintenance handoff is draining the connection: hold the batch, as a unit, for the replacement + foreach (var message in messages) + { + message.SetEnqueued(null); + BacklogEnqueue(message); + } + StartBacklogProcessor(); + return true; + } + if (!IsConnected || NeedsReconnect) { // During reconnection (e.g. MOVED-to-same-endpoint), queue messages to the backlog @@ -1262,6 +1342,10 @@ private void ProcessBridgeBacklog() // check whether the backlog is empty *before* even trying to get the lock if (_backlog.IsEmpty) return; // nothing to do + // a maintenance handoff is holding work back from the connection about to be replaced: nothing + // would be written; the recycle and the reconnect that follows it kick this again + if (_holdForHandoff) return; + // try and get the lock; if unsuccessful, retry gotLock = _singleWriter.TryTakeSync(); if (gotLock) break; // got the lock; now go do something with it @@ -1272,7 +1356,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 && !_holdForHandoff) // not while a handoff is holding work back from the connection about to be replaced { // 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..efdd33933 100644 --- a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs +++ b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs @@ -343,15 +343,6 @@ internal void SetHandoffTarget(EndPoint target, TimeSpan window) _handoffTarget = target; } - /// - /// Forgets any handoff target, once a connection has been established. - /// - /// - /// Called on full establishment rather than on the connect attempt: if the attempt fails we want the next - /// one to try the target again, within its window. Once a connection is up, normal resolution resumes - - /// by then DNS has usually caught up anyway. - /// - internal void ClearHandoffTarget() => _handoffTarget = null; private volatile string? _lastHandoffOutcome; /// @@ -384,18 +375,21 @@ internal void SetHandoffTarget(EndPoint target, TimeSpan window) /// internal void OnMovingAnnounced(TimeSpan? window, EndPoint? successor, PhysicalConnection connection) { - // One at a time per server. A rolling operation delivers one MOVING per connection, so a second one - // arriving while a handoff is in flight is a repeat or a much later event; either way, starting a - // second poll against the same endpoint achieves nothing. + var budget = window is { } value && value > TimeSpan.Zero + ? value + : Multiplexer.RawConfig.MaintenanceRelaxedTimeout; // no window given: use the relaxation floor + + // One at a time per server: two drains on the same bridges would end each other's holds. A second MOVING + // arriving while one is in flight (which can be most of a window: the jitter, the DNS poll, the drain) is + // not acted on separately - the handoff in flight replaces the connections - but if it names a successor, + // that is where the reconnect should go, so the pin is updated and the next dial takes it. if (Interlocked.CompareExchange(ref _handoffInFlight, 1, 0) != 0) { - Multiplexer.Trace("MOVING: a handoff is already in flight", ToString()); + if (successor is not null) SetHandoffTarget(successor, budget); + Multiplexer.Trace(successor is null ? "MOVING: a handoff is already in flight" : "MOVING: a handoff is already in flight; successor noted", ToString()); return; } - var budget = window is { } value && value > TimeSpan.Zero - ? value - : Multiplexer.RawConfig.MaintenanceRelaxedTimeout; // no window given: use the relaxation floor var current = (connection.VolatileSocket?.RemoteEndPoint as IPEndPoint)?.Address; _ = HandoffAsync(budget, successor, current); @@ -422,6 +416,11 @@ 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 = window - TimeSpan.FromMilliseconds(watch.ElapsedMilliseconds); + if (remaining < TimeSpan.Zero) remaining = TimeSpan.Zero; + Multiplexer.Trace($"MOVING: {decision}", ToString()); _lastHandoffOutcome = decision.ToString(); @@ -471,36 +470,103 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? } /// - /// Lets in-flight caller work finish, then replaces the connections. + /// How long the handoff holds new work and waits for in-flight commands before replacing the connections + /// regardless: half the relaxed timeout (five seconds by default), never more than half of what is left of the + /// window, never less than , and never more than the window itself. + /// + /// + /// 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 covers the held work, and follows the one setting an application already + /// uses to say how patient it is prepared to be during maintenance. Half of what is left of the window is a + /// ceiling whatever the relaxed timeout: a generous one must not turn the hold into a full-window stall that + /// recycles at the deadline with work in flight, which is the failure this exists to remove. + /// + private TimeSpan DrainCap(TimeSpan budget) + { + var cap = TimeSpan.FromTicks(Multiplexer.RawConfig.MaintenanceRelaxedTimeout.Ticks / 2); + var halfWindow = TimeSpan.FromTicks(budget.Ticks / 2); + if (cap > halfWindow) cap = halfWindow; + if (cap < DrainFloor) cap = DrainFloor; // a relaxed timeout of zero disables relaxation; it must not disable the drain + return cap > budget ? budget : cap; + } + + /// 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); + + /// + /// 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" + /// 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. With new work held in the backlog, the + /// written commands complete within a round trip, the recycle follows, and the held work is flushed to the + /// replacement by the same path that flushes any backlog when a connection establishes. + /// + /// + /// The wait is bounded by , and timeouts are relaxed for the hold and the reconnect so + /// that the held work is not 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. + /// /// /// 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. /// /// + /// What is left of the announced window. + /// Why; for the trace. private async Task DrainThenRecycleAsync(TimeSpan budget, string reason) { var watch = ValueStopwatch.StartNew(); - while (HasCallerWork() && watch.ElapsedMilliseconds < budget.TotalMilliseconds) + var cap = DrainCap(budget); + + // The held work waits out the hold, the replacement's handshake, then its own turn; relax timeouts for + // that span so the hold itself times nothing out. The reconnect can take until the end of the window plus + // a connect attempt, which the relaxed timeout comfortably covers at its default. + var relax = cap + Multiplexer.RawConfig.MaintenanceRelaxedTimeout; + var ceiling = Multiplexer.RawConfig.MaintenanceRelaxedWindowMax; + ExtendRelaxedWindow(relax > ceiling ? ceiling : relax, "MOVING handoff"); + + // 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?.BeginHandoffHold(); + subscriptionBridge?.BeginHandoffHold(); + bool drained; + try { - await Task.Delay(TimeSpan.FromMilliseconds(20)).ForAwait(); - } + while (!(drained = !HasCallerMessagesInFlight(interactiveBridge, subscriptionBridge)) + && watch.ElapsedMilliseconds < cap.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 recycled = (interactiveBridge?.RecycleConnection(reason) == true) | (subscriptionBridge?.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 at the drain cap)"), + ToString()); + } + finally + { + // after the recycle, so there is no instant at which the hold is down and the old connection live + interactiveBridge?.EndHandoffHold(); + subscriptionBridge?.EndHandoffHold(); + } } + private static bool HasCallerMessagesInFlight(PhysicalBridge? interactive, PhysicalBridge? subscription) + => interactive?.HasCallerMessagesInFlight() == true || subscription?.HasCallerMessagesInFlight() == true; + /// /// Warns when a replacement connection is not fully established by the time the announced window ends. /// diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index e203139c6..d4da7173c 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -1126,9 +1126,6 @@ internal void OnFullyEstablished(PhysicalConnection connection, string source) // Clear the unselectable flag ASAP since we are open for business ClearUnselectable(UnselectableFlags.DidNotRespond); - // whatever a handoff pointed us at, we are connected now: resume normal resolution - ClearHandoffTarget(); - // is *this specific* connection using RESP3? (without reference to config preferences) bool isResp3 = connection?.Protocol is >= RedisProtocol.Resp3; // and if so, is pub/sub sharing it? (if not, subscriptions live on a dedicated connection) diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/MovingUnderLoadScenarioTests.cs b/tests/StackExchange.Redis.FaultInjector.Tests/MovingUnderLoadScenarioTests.cs new file mode 100644 index 000000000..8adf18163 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/MovingUnderLoadScenarioTests.cs @@ -0,0 +1,186 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using StackExchange.Redis.Maintenance; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// The MOVING handoff under steady traffic, against a real deployment: how long after the notification the +/// connections are replaced, how long until the client is serving again, and how many commands failed meanwhile. +/// +/// +/// A drain that waits for "no caller work" never finishes under load - there is always a command in flight - so it +/// recycles at the end of the announced window and fails whatever is in flight at that instant. Holding new work +/// back while the written commands complete is what lets the recycle happen within a round trip. This records the +/// timeline rather than asserting a bound, so that a run against any branch yields a comparable row. +/// +[Trait("tier", "fault-injector")] +[Trait("scenario", "moving-handoff")] +public class MovingUnderLoadScenarioTests(ExistingDatabaseFixture fixture, ITestOutputHelper log) + : IClassFixture +{ + [Theory] + [InlineData(MaintenanceEndpointType.Auto)] + public async Task HandoffUnderSteadyLoad(MaintenanceEndpointType endpointType) + { + fixture.RequireAvailable(); + var cancellationToken = TestContext.Current.CancellationToken; + + await using var scenario = await ScenarioRun.SetupAsync( + fixture.Injector, "topology-change-standalone", "conn_drop", "endpoint_rebind", log.WriteLine, + cancellationToken: cancellationToken); + + var database = scenario.Database; + Assert.NotNull(database); + + var clock = Stopwatch.StartNew(); + var timeline = new List(); + void Note(string what) + { + var entry = $" +{clock.Elapsed.TotalSeconds,6:0.0}s {what}"; + lock (timeline) timeline.Add(entry); + log.WriteLine(entry); + } + + var logs = new CapturingLogger(); + var config = database.GetClientConfig(fixture.Environment); + config.MaintenanceMovingEndpointType = endpointType; + config.LoggerFactory = logs; + log.WriteLine($"moving-endpoint-type: {endpointType}"); + + await using var conn = await ConnectionMultiplexer.ConnectAsync(config); + var muxer = (IInternalConnectionMultiplexer)conn; + var endpoint = muxer.GetServerEndPoint(conn.GetEndPoints()[0]); + + TimeSpan? movingAt = null, recycledAt = null, restoredAt = null; + var failures = new List<(TimeSpan At, string What)>(); + conn.ConnectionFailed += (_, e) => Note($"disconnect: {e.FailureType} ({e.ConnectionType})"); + conn.ConnectionRestored += (_, e) => + { + restoredAt ??= clock.Elapsed; + Note($"reconnected ({e.ConnectionType})"); + }; + conn.ServerMaintenanceEvent += (_, e) => + { + if (e is not PushMaintenanceEvent push) return; + Note($"{push.NotificationType} seq={push.SequenceId} time={push.Time?.TotalSeconds.ToString() ?? "-"}"); + if (push.NotificationType == MaintenanceNotificationType.Moving) movingAt ??= clock.Elapsed; + }; + + // steady load: sixteen workers, each issuing its next PING the moment the previous one is answered, so there + // is never an instant with nothing in flight and a drain that waits for an idle connection cannot finish (a + // first cut paused 10ms between commands, and the old drain slipped through the gaps) + long completed = 0; + using var stop = new CancellationTokenSource(); + var db = conn.GetDatabase(); + var workload = new MultiThreadedWorkload( + async (_, _) => + { + try + { + await db.PingAsync().ForAwait(); + Interlocked.Increment(ref completed); + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + lock (failures) failures.Add((clock.Elapsed, ex.GetType().Name + ": " + FirstLine(ex.Message))); + } + }, + workerCount: 16); + workload.Start(stop.Token); + await Task.Delay(3000, cancellationToken); + var warmup = Interlocked.Read(ref completed); + log.WriteLine($"load: {warmup / 3.0:0} ops/s before the scenario fires"); + + clock.Restart(); + // the injector call blocks until the scenario has run its course (about 20s), so watch for the recycle + // alongside it rather than after it; then cover the announced window plus the measured overshoot + var fire = scenario.FireAsync(cancellationToken); + var deadline = TimeSpan.FromSeconds(100); + while (clock.Elapsed < deadline) + { + if (recycledAt is null && endpoint.HandoffRecycles > 0) + { + recycledAt = clock.Elapsed; + Note($"connections replaced (recycles={endpoint.HandoffRecycles}); handoff: {endpoint.LastHandoffOutcome}"); + } + if (fire is { IsCompleted: true }) + { + await fire; + fire = null; + Note("injector reports the scenario finished"); + } + await Task.Delay(100, cancellationToken); + } + if (fire is not null) await fire; + + stop.Cancel(); + try + { + await workload.Completion; + } + catch (OperationCanceledException) + { + } + + var total = Interlocked.Read(ref completed); + int failed; + lock (failures) + { + failed = failures.Count; + foreach (var (at, what) in failures.Take(20)) log.WriteLine($" failed at +{at.TotalSeconds,6:0.0}s: {what}"); + if (failures.Count > 20) log.WriteLine($" ... and {failures.Count - 20} more"); + } + var missed = logs.Lines.Where(l => l.Contains("did not establish a replacement", StringComparison.Ordinal)).ToList(); + foreach (var line in missed) log.WriteLine(" " + line); + + log.WriteLine(string.Empty); + log.WriteLine("| event | at |"); + log.WriteLine("|---|---|"); + log.WriteLine($"| MOVING received | {Format(movingAt)} |"); + log.WriteLine($"| connections replaced | {Format(recycledAt)} |"); + log.WriteLine($"| ConnectionRestored | {Format(restoredAt)} |"); + log.WriteLine($"| completed commands | {total} |"); + log.WriteLine($"| failed commands | {failed} |"); + log.WriteLine($"| missed-deadline warning | {(missed.Count > 0 ? "yes" : "no")} |"); + log.WriteLine($"| handoff outcome | {endpoint.LastHandoffOutcome ?? "(none)"} |"); + + Assert.NotNull(movingAt); + Assert.True(endpoint.HandoffRecycles > 0, "the handoff should have replaced the connections"); + + static string Format(TimeSpan? at) => at is { } t ? $"+{t.TotalSeconds:0.0}s" : "(never)"; + static string FirstLine(string s) + { + var i = s.IndexOf('\n'); + return i < 0 ? s : s.Substring(0, i); + } + } + + private sealed class CapturingLogger : ILoggerFactory, ILogger + { + private readonly List _lines = new(); + public IReadOnlyList Lines + { + get + { + lock (_lines) return _lines.ToArray(); + } + } + + public void AddProvider(ILoggerProvider provider) { } + public ILogger CreateLogger(string categoryName) => this; + public void Dispose() { } + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + { + lock (_lines) _lines.Add($"{logLevel}: {formatter(state, exception)}"); + } + } +} diff --git a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs index bfdb3a7a5..f6107af1c 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; @@ -26,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)); @@ -842,6 +845,192 @@ await Poll.UntilAsync( } } + [Fact] + 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 + // 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, configure); + 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 workerTasks = 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(); + try + { + Assert.True(await Poll.UntilAsync(() => Volatile.Read(ref completed) > 20, timeoutMilliseconds: 30_000), "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(5), $"recycled after {recycledAfter}: that is waiting for the window, not draining"); + + // the replacement carries the held work and whatever follows. Note the in-process transport never dials the + // successor: its tunnel derives the socket target from the configured endpoint, so the reconnect goes + // straight back to this server; the timing here is the hold and the handshake, not a refused address. + var resumeWatch = System.Diagnostics.Stopwatch.StartNew(); + var before = Volatile.Read(ref completed); + Assert.True( + await Poll.UntilAsync(() => Volatile.Read(ref completed) > before + 20, timeoutMilliseconds: 20_000), + "traffic should resume on the replacement connection"); + log.WriteLine($"traffic resumed {resumeWatch.ElapsedMilliseconds}ms after the recycle was observed"); + + stop.Cancel(); + await Task.WhenAll(workerTasks); + 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)); + } + finally + { + stop.Cancel(); // on any exit: a failed assertion must not leave four workers hot-looping on a disposed multiplexer + } + } + } + + [Fact] + public async Task MovingWaitsForSlowCommandsThatFitInsideTheDrainCap() + { + // 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)); // the in-process server answers one request at a time, so one slow command, not several + var db = conn.GetDatabase(); + var slow = db.PingAsync(); + Assert.True(await Poll.UntilAsync(() => conn.GetCounters().Interactive.SentItemsAwaitingResponse > 0), "the slow command should be in flight before MOVING arrives"); + + 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: 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 (had it not, this await would throw) + Assert.True(await slow >= TimeSpan.Zero); + // 3s for the command, up to 1s of jitter, poll granularity, and slack; the cap would be 5s + jitter + Assert.True(recycledAfter < TimeSpan.FromSeconds(7), $"recycled after {recycledAfter}: later than the command 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(12)); // well past the 5s cap and the slack below; the relaxed window covers it until the recycle + var db = conn.GetDatabase(); + // one, not several: the test server answers one request at a time, so each extra 12s command would hold the + // replacement's handshake behind it as well + var blocking = db.PingAsync(); + Assert.True(await Poll.UntilAsync(() => conn.GetCounters().Interactive.SentItemsAwaitingResponse > 0), "the blocking command should be in flight before MOVING arrives"); + + 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: 14_000), + "the handoff should have replaced the connection"); + var recycledAfter = watch.Elapsed; + // Only now: the server reads its latency when it *executes* a request, and zeroing it any earlier races the + // one command this test needs to stay slow. From here the replacement's handshake waits behind at most + // that one 12s reply. + server.SetLatency(TimeSpan.Zero); + log.WriteLine($"recycled after {recycledAfter.TotalMilliseconds:0}ms with a 12s command in flight"); + + // jitter (at most a tenth of the window, capped at 1s) + the 5s cap + poll granularity is the mechanism's + // own worst case, about 6.5s; the bug being guarded against is a recycle at the 15s deadline, so 12s + // discriminates with room on both sides + Assert.True(recycledAfter < TimeSpan.FromSeconds(12), $"recycled after {recycledAfter}: the drain waited for the blocking command instead of capping"); + + // the in-flight command was failed by the recycle, which is what the server's close would have done + await Assert.ThrowsAnyAsync(async () => await blocking); + + 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: 40_000), + "the client should be serving commands on the replacement connection"); + } + } + [Fact] public async Task MovingRecyclesTheConnectionBeforeTheServerCloses() { @@ -884,7 +1073,7 @@ public async Task MovingRecyclesTheConnectionBeforeTheServerCloses() Assert.True( await Poll.UntilAsync(() => endpoint.HandoffTarget is not null || endpoint.HandoffRecycles > 0, timeoutMilliseconds: 10_000), "the handoff should have taken the named successor"); - log.WriteLine($"handoff target: {endpoint.HandoffTarget?.ToString() ?? "(cleared after reconnect)"}; recycles={endpoint.HandoffRecycles}"); + log.WriteLine($"handoff target: {endpoint.HandoffTarget?.ToString() ?? "(expired with the window)"}; recycles={endpoint.HandoffRecycles}"); // a fresh connection re-sends the opt-in, which is how a recycle is visible from the server's side Assert.True( From aec25119eff50e58dad154519ec55e764928ab8c Mon Sep 17 00:00:00 2001 From: Marc Gravell Date: Wed, 11 Nov 2026 13:40:10 +0000 Subject: [PATCH 2/2] OneLogicalEventFromEveryNodeRaisesOneEvent: wait for the one event rather than counting at the instant the second node relaxed; the callback is dispatched after the relaxation, so the count could be read as zero (flaked on main at ~1 in 6) --- .../MaintenanceNotificationTests.cs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs index f6107af1c..79578a00c 100644 --- a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs +++ b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs @@ -528,11 +528,13 @@ await Poll.UntilAsync(() => await Poll.UntilAsync(() => conn.GetEndPoints().All(ep => muxer.GetServerEndPoint(ep).IsMaintenanceRelaxed)), "every node that was told should have relaxed its own timeouts"); - // ...and the consumer sees it once - Assert.Equal(1, events.Count); + // ...and the consumer sees it once: the callback is dispatched after the relaxation, not before, so wait + // for the one event rather than counting at the instant the second node relaxed, then prove the other + // delivery did not become a second event var evt = await events.NextAsync(); Assert.Equal(5, evt.SequenceId); - log.WriteLine($"{events.Count} event(s) from {delivered} deliveries: {evt.RawMessage}"); + await events.AssertNoneAsync(); + log.WriteLine($"1 event from {delivered} deliveries: {evt.RawMessage}"); Assert.Contains(evt.EndPoint, conn.GetEndPoints()); // whichever told us first } }