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..79578a00c 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)); @@ -525,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 } } @@ -842,6 +847,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 +1075,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(