diff --git a/docs/ServerMaintenanceEvent.md b/docs/ServerMaintenanceEvent.md index 15ef44c5a..13d05eb8b 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 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 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..a44eef52c 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; @@ -132,6 +140,155 @@ 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 . 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 + /// 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() + { + _quiesceTarget = physical; + _quiesceRecycled = false; + _quiescing = true; + } + + internal void EndQuiesce() + { + if (!_quiescing) return; + _quiescing = false; + _quiesceTarget = null; + // 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. + private void KickBacklogProcessor() + { + try + { + _backlogAutoReset.Set(); + } + catch (ObjectDisposedException) + { + return; // a bridge disposed mid-handoff has no processor to wake, and nothing left to write + } + 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) => 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 + /// 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 ). + /// + /// Replace the connection even if caller work is still in flight on it. + /// Why; for the trace. + /// 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. + internal async ValueTask TryRetireQuiesceTargetAsync(bool force, string reason, CancellationToken lockDeadline) + { + 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) + { + gotLock = false; + } + catch (ObjectDisposedException) + { + return false; // the bridge was disposed mid-handoff; there is nothing left to retire + } + if (!gotLock) return null; + try + { + 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[_recycledForHandoffNext++ & 3] = target; // a writer already past the hold check queues for the replacement instead + _quiesceRecycled = true; + target.RecordConnectionFailed(ConnectionFailureType.MaintenanceHandoff); + target.Dispose(); + return true; + } + finally + { + try + { + _singleWriter.Release(); + } + catch (ObjectDisposedException) + { + // disposed while we held it; nobody is waiting on a disposed mutex + } + } + } + public void Dispose() { isDisposed = true; @@ -176,18 +333,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 } } @@ -212,6 +372,7 @@ private WriteResult FailDueToNoConnection(Message message) public WriteResult TryWriteSync(Message message, bool isReplica) { if (isDisposed) throw new ObjectDisposedException(Name); + if (ShouldHoldForHandoff(message)) return QueueOrFailMessage(message, holdForHandoff: true); if (!IsConnected || NeedsReconnect) return QueueOrFailMessage(message); var physical = this.physical; @@ -236,6 +397,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(QueueOrFailMessage(message, holdForHandoff: true)); if ((!IsConnected || NeedsReconnect) && !bypassBacklog) return new ValueTask(QueueOrFailMessage(message)); var physical = this.physical; @@ -349,7 +511,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, @@ -477,8 +639,11 @@ 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 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); } @@ -542,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. @@ -776,6 +946,16 @@ internal bool TryEnqueue(List messages, bool isReplica) if (isDisposed) throw new ObjectDisposedException(Name); + if (HoldActive && ShouldHoldBatch(messages)) + { + // a maintenance handoff is draining the connection: hold the batch for the replacement + foreach (var message in messages) + { + QueueOrFailMessage(message, holdForHandoff: true); + } + return true; + } + if (!IsConnected || NeedsReconnect) { // During reconnection (e.g. MOVED-to-same-endpoint), queue messages to the backlog @@ -928,6 +1108,7 @@ internal WriteResult WriteMessageTakingWriteLockSync(PhysicalConnection physical if (!gotLock) return TimedOutBeforeWrite(message); } + if (TryRequeueForHandoff(physical, message, bypassBacklog: false, out var requeued)) return requeued; var result = WriteMessageInsideLock(physical, message); if ((result == WriteResult.Success) & message.IsFlushRequiredSync) @@ -993,7 +1174,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); @@ -1014,7 +1195,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; @@ -1022,6 +1203,32 @@ 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) + { + bool wasEmpty = !BacklogHasItems; + Interlocked.Increment(ref _backlogCurrentEnqueued); + Interlocked.Increment(ref _backlogTotalEnqueued); + // 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(); + 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(); + } + [MethodImpl(MethodImplOptions.AggressiveInlining)] private void StartBacklogProcessor() { @@ -1064,14 +1271,17 @@ 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 + } } } - /// - /// 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 this bridge owes a *caller* anything, ignoring our own internal traffic. /// @@ -1079,13 +1289,21 @@ 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; } 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; @@ -1099,7 +1317,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. @@ -1108,7 +1326,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 @@ -1163,7 +1381,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 @@ -1260,7 +1478,11 @@ 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 + + // 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(); @@ -1272,7 +1494,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 @@ -1373,7 +1595,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; } @@ -1390,6 +1616,46 @@ private WriteResult TimedOutBeforeWrite(Message message) return WriteResult.TimeoutBeforeWrite; } + /// + /// 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 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 + /// 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 TryRequeueForHandoff(PhysicalConnection physical, Message message, bool bypassBacklog, out WriteResult result) + { + if (bypassBacklog || message.IsInternalCall || ReferenceEquals(physical, this.physical) || !WasRecycledForHandoff(physical)) + { + result = default; + return false; + } + BacklogRequeueFront(message); + result = WriteResult.Success; // queued counts as success + return true; + } + /// /// This writes a message to the output stream. /// @@ -1427,11 +1693,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 (TryRequeueForHandoff(physical, message, bypassBacklog, out var requeued)) return new ValueTask(requeued); var result = WriteMessageInsideLock(physical, message); if ((result == WriteResult.Success) & message.IsFlushRequiredAsync) { @@ -1463,13 +1730,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 (TryRequeueForHandoff(physical, message, bypassBacklog, out var requeued)) return requeued; var result = WriteMessageInsideLock(physical, message); if (result == WriteResult.Success) @@ -1861,33 +2130,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. - 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 53308faa6..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,14 +502,64 @@ 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; + } + /// - /// Lets in-flight caller work finish, then replaces the connections. + /// 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. + /// + /// + /// 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. 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) + { + 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); + + /// + /// 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. 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. + /// + /// + /// 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 /// to one proxy, differing only in handshake, all closed simultaneously, and only the ones that had opted in @@ -487,18 +569,74 @@ 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) + // 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; + var subscriptionBridge = subscription; + interactiveBridge?.BeginQuiesce(); + subscriptionBridge?.BeginQuiesce(); + try + { + using var lockDeadline = new CancellationTokenSource(cap + LockGrace); + 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; + // 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) + { + // 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" + + (forced ? " (still busy at the drain cap)" : " (drained)") + + (interactiveResult is null || subscriptionResult is null ? " (write lock not obtained; left to the server)" : ""), + ToString()); + } + finally { - await Task.Delay(TimeSpan.FromMilliseconds(20)).ForAwait(); + // whatever was not released in the loop (the lock-not-obtained exit, an exception): EndQuiesce is idempotent + Release(interactiveBridge); + Release(subscriptionBridge); } - 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()); + 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 bfdb3a7a5..aa1de859e 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,177 @@ 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 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 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 = 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); + 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 + 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 9s commands in flight"); + + // 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 blocking) + { + 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: 40_000), + "the client should be serving commands on the replacement connection"); + } + } + [Fact] public async Task MovingRecyclesTheConnectionBeforeTheServerCloses() {