diff --git a/docs/ServerMaintenanceEvent.md b/docs/ServerMaintenanceEvent.md index 15ef44c5a..64c7af3da 100644 --- a/docs/ServerMaintenanceEvent.md +++ b/docs/ServerMaintenanceEvent.md @@ -111,10 +111,14 @@ 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, approximately in the order they were issued (a command that was racing the replacement itself may be reordered against the ones issued just after it). While the old connection is being drained 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, or when an attempt to reach the replacement fails, since a handoff is a replacement the client chose rather than a failure. Between the old connection being replaced and the replacement completing its handshake, a `FailFast` policy fails *new* commands immediately, as it does on any reconnect; the held ones wait for the replacement. 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 (a generous relaxed timeout must not become a full-window stall), never less than a second (which wins over the half-window bound when less than two seconds remain), and never longer than the window itself, since the server closes the socket when that ends regardless; with less than a second of window left there is no drain, and the connections are replaced at once (commands are held only for the instant of the replacement itself). The commands the hold diverted keep a longer timeout until they are answered - the hold, the rest of the window and one connect attempt, bounded by `maintRelaxedWindowMax` but never less than the hold plus one connect attempt, since that much is what the held commands need - whatever `maintRelaxedTimeout` is set to, so nothing is timed out by the hold itself; every other command on that server keeps its configured timeout throughout, including under an application that has turned relaxation off. 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. + +**What is guaranteed, and what is best effort.** One handoff at a time per endpoint is fully covered: commands already in flight drain (up to the cap above), commands issued during the hold are never failed by the hold itself, and they are written to the replacement or fail within the bound above (the hold, the rest of the window and one connect attempt). The replacement is whatever connection fully establishes once the old one has been retired; the client does not try to tell from a socket's address which node it reached, so a reconnect that lands back on the retiring node (no successor named, a stale resolver) is treated as the replacement and is closed by the server at the end of the window as before, with the relaxed window covering the gap. Back-to-back `MOVING`s inside one window are handled on a best-effort basis: a second notification while a handoff is in progress updates where the reconnect goes (or drops a successor that may itself be retiring) rather than starting a second drain, held commands are never lost or abandoned contrary to the `backlogPolicy`, but they may wait longer than the single-handoff bound, and the client resolves the hold conservatively, later rather than wrong. + ### 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/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs index 49a5682e5..fdc337cef 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs @@ -2736,29 +2736,34 @@ internal static void ThrowFailed(TaskCompletionSource? source, Exception u // sync caller in flight when a MIGRATING arrives times out at the strict timeout while // its async neighbour is relaxed. It can shrink back too: when a window closes the // effective timeout drops to the configured value, and we stop at that. + // + // Waiting in slices is what makes that re-read reachable: a single wait for the whole budget + // that nothing pulses (a genuinely slow command) would time out at the original deadline and + // never see a relaxation, or a handoff's cover, that began after we parked. Only where the + // budget can change at all, though: with maintenance notifications disabled for the server + // nothing ever relaxes or covers it, and one wait for the whole budget is enough. var startedAt = Environment.TickCount; - var remaining = server?.GetEffectiveTimeoutMilliseconds(TimeoutMilliseconds) ?? TimeoutMilliseconds; while (true) { - if (!Monitor.Wait(source, remaining)) + // per wait, not once: a window that opens after we park puts us on slices (so we notice it + // close again), and one that closes takes us off them + var sliceMilliseconds = server is not null && server.MaintenanceMayChangeTimeouts ? 1000 : int.MaxValue; + var budget = server is null + ? TimeoutMilliseconds + : ServerEndPoint.GetEffectiveTimeoutMilliseconds(server.GetEffectiveTimeoutMilliseconds(TimeoutMilliseconds), message); + var remaining = budget - unchecked(Environment.TickCount - startedAt); + if (remaining <= 0) { Trace("Timeout performing " + message); timeout = true; break; } + Monitor.Wait(source, remaining < sliceMilliseconds ? remaining : sliceMilliseconds); if (source.IsCompleted) { Trace("Timely response to " + message); break; } - var budget = server?.GetEffectiveTimeoutMilliseconds(TimeoutMilliseconds) ?? TimeoutMilliseconds; - remaining = budget - unchecked(Environment.TickCount - startedAt); - if (remaining <= 0) - { - Trace("Timeout performing " + message); - timeout = true; - break; - } } } } diff --git a/src/StackExchange.Redis/Maintenance/TickDeadline.cs b/src/StackExchange.Redis/Maintenance/TickDeadline.cs new file mode 100644 index 000000000..42e0e6f7e --- /dev/null +++ b/src/StackExchange.Redis/Maintenance/TickDeadline.cs @@ -0,0 +1,105 @@ +using System; +using System.Threading; + +namespace StackExchange.Redis.Maintenance; + +/// +/// A deadline expressed in terms, where zero means "unset". One place for the +/// three invariants every such field needs: a genuine zero tick count is nudged to one so it cannot read as unset; +/// a deadline is in force only within one span () of now, so the residue of one that +/// passed long ago cannot come back to life when the tick count wraps ~24.8 days later, nor win a comparison against +/// a fresh extension; and extension is monotonic, never shortening a deadline that is already later. A deadline that has merely passed is kept, readable through , +/// so it also records when it passed: fault attribution asks how recently the relaxed window closed, and a +/// held command's cover is stamped from the handoff deadline whether or not the handoff is still running. Used for +/// the relaxed-timeout window, the successor pin, the per-bridge handoff deadline and each held message's cover, which +/// all had their own copies of this before. +/// +internal struct TickDeadline +{ + private int _ticks; + + /// + /// The most a deadline can be set into the future: about six days. A tick count wraps in about 24.8 days, so a + /// span approaching that would be indistinguishable from the past; and a span that overflows an int of + /// milliseconds (a buggy or hostile announced window) would otherwise land the deadline in the past outright, + /// raising a protection that is silently inactive. No maintenance window is days long; clamping is always right. + /// + private const int MaxMilliseconds = int.MaxValue / 4; + + /// The raw deadline, passed or not, or zero when unset (or so long passed that it has been cleared). + public int Ticks => Volatile.Read(ref _ticks); + + public void SetIn(TimeSpan span) => Volatile.Write(ref _ticks, NudgeFromZero(unchecked(Environment.TickCount + Clamp(span)))); + + /// A span as whole milliseconds, never negative and never beyond . + public static int Clamp(TimeSpan span) + { + if (span <= TimeSpan.Zero) return 0; + var ms = span.TotalMilliseconds; + return ms >= MaxMilliseconds ? MaxMilliseconds : (int)ms; + } + + public void Clear() => Volatile.Write(ref _ticks, 0); + + /// + /// Moves the deadline out to now plus unless it is already later than that. + /// + /// How far from now the deadline should be, at least. + /// The deadline now in force, whether or not this call set it. + /// Whether this call extended it. + public bool TryExtendBy(TimeSpan span, out int deadline) + { + var candidate = NudgeFromZero(unchecked(Environment.TickCount + Clamp(span))); + if (ExtendTo(candidate)) + { + deadline = candidate; + return true; + } + deadline = Volatile.Read(ref _ticks); + return false; + } + + /// + /// Moves the deadline out to (a tick count, as reads) unless it is + /// already later than that; zero means "no deadline" and extends nothing. + /// + /// Whether this call extended it. + public bool ExtendTo(int ticks) + { + if (ticks == 0) return false; + var now = Environment.TickCount; + while (true) + { + var current = Volatile.Read(ref _ticks); + // a deadline in force is within one span of now, either way; anything else is the residue of a deadline + // that passed so long ago that the tick count has wrapped past it, and must not win a comparison + var ahead = unchecked(current - now); + var inForce = current != 0 && ahead > -MaxMilliseconds && ahead <= MaxMilliseconds; + if (inForce && unchecked(ticks - current) <= 0) return false; + if (Interlocked.CompareExchange(ref _ticks, ticks, current) == current) return true; + } + } + + /// + /// Milliseconds until the deadline, or zero if unset or passed. A passed deadline is kept (see ) + /// until it is so long past that a wrapping tick count could make it look future again, and then cleared. + /// + public int RemainingMilliseconds() + { + var deadline = Volatile.Read(ref _ticks); + if (deadline == 0) return 0; + var remaining = unchecked(deadline - Environment.TickCount); + if (remaining > 0 && remaining <= MaxMilliseconds) return remaining; + + // Long passed, or further ahead than any span this type sets - which is what a deadline that passed more + // than half a wrap ago looks like once the tick count has come round: clear it, but only if nobody has moved + // it on in the meantime. + if (remaining < -MaxMilliseconds || remaining > MaxMilliseconds) Interlocked.CompareExchange(ref _ticks, 0, deadline); + return 0; + } + + public bool IsActive => RemainingMilliseconds() > 0; + + /// Zero means "unset" (or "never"), so a tick count that lands on it moves by one. + public static int NudgeFromZero(int ticks) => ticks == 0 ? 1 : ticks; +} diff --git a/src/StackExchange.Redis/Message.cs b/src/StackExchange.Redis/Message.cs index e6943cbcf..bcdbc6acb 100644 --- a/src/StackExchange.Redis/Message.cs +++ b/src/StackExchange.Redis/Message.cs @@ -892,7 +892,45 @@ internal void SetWriteTime() /// ; they take the parent's write time, so that they share its timeout budget /// (including any time it spent in the backlog) rather than looking overdue by the process uptime. /// - internal void InheritWriteTime(Message parent) => _writeTickCount = parent._writeTickCount; + internal void InheritWriteTime(Message parent) + { + _writeTickCount = parent._writeTickCount; + // and the cover that goes with that budget: a held transaction's inner commands reach the replacement with + // a clock that includes the whole hold, and without the parent's cover the next sweep would fail them + _holdCover.ExtendTo(parent._holdCover.Ticks); + } + + /// + /// The message is going back to the backlog after the connection it was headed for was retired by a maintenance + /// handoff: it no longer points at that connection, but it keeps its write time, since the budget it has already + /// spent (including waiting for the write lock) still counts. Unlike , which restarts it. + /// + internal void SetRequeued() + { + _enqueuedTo = null; + _queuedStampSent = _queuedStampReceived = -1; + } + + /// + /// Whether this message entered a backlog during a maintenance handoff - diverted off a live connection by the + /// hold, requeued after the connection it was headed for was retired, or queued while the bridge reconnected to + /// the replacement. Such a message is waiting on the client's own choice, not on the server, so the handoff's + /// timeout cover applies to it (and only to it: an ordinary command on the same server keeps its configured + /// timeout) for as long as it lives, including after it has been written to the replacement. + /// + internal bool HeldForHandoff => _holdCover.Ticks != 0; + + /// + /// The handoff's deadline, in terms, until which this message cannot time + /// out: the cover is a property of the message, stamped when it is held, so nothing that happens to the handoff + /// afterwards can withdraw it. Zero if the message was never held. + /// + internal int HoldCoverDeadlineTicks => _holdCover.Ticks; + + /// Holds the message for a handoff, covered until ; a later deadline replaces an earlier one, never the reverse (and zero, no deadline, covers nothing). + internal void MarkHeldForHandoff(int deadlineTicks) => _holdCover.ExtendTo(deadlineTicks); + + private Maintenance.TickDeadline _holdCover; private int _writeTickCount; diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index ee3a7bfb2..5c7b596f1 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -34,6 +34,7 @@ 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; private int _backlogProcessorIsRunning = 0; private int _backlogCurrentEnqueued = 0; @@ -132,9 +133,508 @@ public enum State : byte /// internal void MarkNeedsReconnect() => Volatile.Write(ref _needsReconnect, true); + /// + /// Where a maintenance handoff stands on this bridge. One variable, advanced in one direction, and the only + /// thing every handoff question is answered from: whether new work is held, whether the backlog may flush, + /// whether a failure is the handoff's own, and whether work enqueued now is covered. (Parallel flags for the + /// same facts kept disagreeing with each other in the races between them.) + /// + private enum HandoffPhase + { + /// + /// The connection current when the hold began (the target) is being drained: new caller work is held in + /// the backlog, the backlog does not flush, and the drain is waiting for the target's in-flight work. + /// + Holding, + + /// + /// The target is gone - retired by the drain, or it went away by itself - and whatever is current is not + /// yet known to be a bona fide replacement. New work is still held and the backlog still does not flush, + /// because writing to a connection the server is about to close would only hand it work to fail. + /// + AwaitingReplacement, + + /// + /// A bona fide replacement is current and established, and the held work is being written to it. The + /// handoff's protection stands until the last held message has been written: a failure before that would + /// still abandon them under a FailFast policy. + /// + Flushing, + + /// Nothing left to protect; or the deadline passed; or the bridge was disposed. + Done, + } + + /// + /// One maintenance handoff on this bridge: its phase, its target, and its deadline. A new object per handoff, + /// so that a transition requested against an earlier handoff (a backlog pass finishing an earlier flush, say) + /// cannot land on a later one. + /// + private sealed class Handoff + { + private int _phase; // HandoffPhase + private int _retiring; // a retire of the target, outside the drain loop, is under way + private PhysicalConnection? _target; + + /// + /// Until when this handoff stands at the latest: the hold, the rest of the window and one connect attempt, + /// from when the hold began. This handoff's own, not shared with an earlier one: held work is stamped with + /// it, so the time a command can be in limbo is bounded by the handoff that held it, never by a longer one + /// that happened to come before. It is read on every question put to a live handoff, so a passed deadline + /// is noticed while it is still fresh. + /// + public Maintenance.TickDeadline Deadline; + + /// The connection the hold began on; null if the bridge had none. Let go of once the handoff is done. + public PhysicalConnection? Target => Volatile.Read(ref _target); + + /// Whether this handoff retired a connection (rather than finding it gone). + public volatile bool Retired; + + public Handoff(PhysicalConnection? target, TimeSpan protection) + { + _target = target; + Deadline.SetIn(protection); + } + + /// Claims the one-at-a-time retire of the target outside the drain loop; see . + public bool TryBeginRetire() => Interlocked.CompareExchange(ref _retiring, 1, 0) == 0; + + public void EndRetire() => Volatile.Write(ref _retiring, 0); + + public HandoffPhase Phase => (HandoffPhase)Volatile.Read(ref _phase); + + /// Advances the phase, if it is still where the caller saw it. Done lets go of the target: a finished handoff must not root a retired connection. + public bool TryAdvance(HandoffPhase from, HandoffPhase to) + { + if (Interlocked.CompareExchange(ref _phase, (int)to, (int)from) != (int)from) return false; + if (to == HandoffPhase.Done) Volatile.Write(ref _target, null); + return true; + } + + /// Ends the handoff whatever its phase: the deadline passed, or the bridge was disposed. + public void Finish() + { + Volatile.Write(ref _phase, (int)HandoffPhase.Done); + Volatile.Write(ref _target, null); + } + } + + private volatile Handoff? _handoff; + + /// The handoff in progress, if there is one and its deadline has not passed (a passed deadline ends it). + private bool TryGetHandoff([NotNullWhen(true)] out Handoff? handoff) + { + handoff = _handoff; + if (handoff is null || handoff.Phase == HandoffPhase.Done) return false; + if (handoff.Deadline.IsActive) return true; + handoff.Finish(); + return false; + } + + private HandoffPhase Phase => TryGetHandoff(out var handoff) ? handoff.Phase : HandoffPhase.Done; + + /// + /// A handoff's hold is in force: whatever is current is the target, or is not known to be a bona fide + /// replacement. New work is queued and the backlog does not flush; the moment a replacement is current and + /// established (or the deadline passes), the hold is over and the backlog flushes to it. + /// + private bool HoldUnresolved => Phase is HandoffPhase.Holding or HandoffPhase.AwaitingReplacement; + + /// + /// The last resort when the write lock could not be had for the whole of the drain's deadline: retires the + /// target without it. A writer wedged on the lock is the only thing this can race, and what it risks is that + /// one writer's message failing with a write fault; what it buys is the replacement starting now rather than + /// at the server's window-end close, which would fail everything in flight and leave the bridge dark for the + /// reconnect. Before handoffs existed the recycle was always lockless, so this is the old behaviour kept as + /// the fallback rather than the rule. + /// + internal bool RetireTargetWithoutLock(string reason) + { + if (!TryGetHandoff(out var handoff) || handoff.Target is not { } target || !ReferenceEquals(physical, target)) return false; + if (!target.TryMarkRetiredForHandoff()) return false; // a retire under the lock got there first; its teardown follows + handoff.Retired = true; + handoff.TryAdvance(HandoffPhase.Holding, HandoffPhase.AwaitingReplacement); + RetireConnection(target, reason + " (without the write lock: last resort)"); + return true; + } + + /// + /// The rules the handoff keeps, stated once so that the next edit is checked against them rather than + /// against the last edit: + /// + /// (a) protection stands down only when the held work is safe - written to the replacement, + /// or past the deadline that also bounds its cover - never merely because a replacement exists; + /// (b) the replacement is whatever fully establishes once the target is gone, and nothing else: the + /// client does not try to tell from a socket's address which node it reached. Address evidence (is this + /// reconnect back on the node being retired?) was tried, and cost eight review rounds of special cases - a + /// tunnel, a fixed front address, a successor at the same address, a node with several addresses, an address + /// that could not be read - for one corner it would have served: a reconnect to the retiring node after the + /// retire, with no successor named and a stale resolver. There the server closes that reconnect at the end of + /// the window and the client reconnects again, as it always did; the held work rides the relaxed window and + /// its own cover meanwhile. Less machinery, resolved later rather than wrong; + /// (c) what a connection is - the target, established - is recorded on the connection when it + /// happens and read from there; the handoff's phase says only where the flow stands. + /// + /// + private enum Replacement + { + /// Established once the target was gone: the replacement. + Yes, + + /// The target: the server will close it. + Doomed, + + /// A different connection current while the target is still being drained: judged again once the drain has let go. + Unknown, + } + + /// Classifies a connection against a handoff; see rule (b). + private static Replacement Classify(Handoff handoff, PhysicalConnection connection) + { + if (connection.WasHandoffTarget) return Replacement.Doomed; + return handoff.Phase == HandoffPhase.Holding ? Replacement.Unknown : Replacement.Yes; + } + + /// + /// Whether the backlog still holds work a handoff diverted, including a message the processor has dequeued but + /// not yet written; the protection stands until it has been written (rule a). Under the backlog lock, which is + /// where the processor moves a message from the queue to its hand, so the message is seen in one place or the + /// other. Only the termination path asks this, never the per-message path. + /// + private bool AnyHeldQueued() + { + lock (_backlog) + { + if (_backlogInHand is { HeldForHandoff: true }) return true; + foreach (var message in _backlog) + { + if (message.HeldForHandoff) return true; + } + return false; + } + } + + /// The message the backlog processor has taken off the queue and not yet finished writing; see . + private volatile Message? _backlogInHand; + + /// + /// Whether a connection failure belongs to a maintenance handoff in progress: the target (retired by the + /// drain, closed by the server at the end of the window, or dropped by itself), a reconnect that landed back + /// on the node being retired, or an attempt to bring up the replacement that never completed its handshake, + /// whatever the failure type. Only a replacement that was fully established and then failed is an ordinary + /// failure, and a FailFast policy applies to it as usual, even while the handoff is still flushing. + /// + private bool IsHandoffOwnFailure(PhysicalConnection connection) + => TryGetHandoff(out var handoff) + && (!connection.EverFullyEstablished || Classify(handoff, connection) != Replacement.Yes); + + /// + /// 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. + /// + /// For how long at most the handoff stands - the FailFast shield here and the timeout + /// cover on the held commands - if no replacement establishes and flushes first. + internal void BeginHold(TimeSpan protection) + { + var target = physical; + target?.MarkHandoffTarget(); + // Publish the new handoff, then finish the previous one - in that order, so that there is no instant at + // which neither hold is in force: a wake racing this would otherwise flush the previous hold's work to the + // very connection this hold is about to drain. The previous handoff (still flushing, say) is finished + // explicitly rather than left mid-phase: its replacement is the connection this one retires, so its flush + // cannot complete, and its held work - which belongs to the bridge's backlog, not to a handoff - keeps its + // stamped cover and is flushed with this handoff's. + var previous = _handoff; + var handoff = new Handoff(target, protection); + _handoff = handoff; + previous?.Finish(); + + // Work already in the backlog when the hold begins is delayed by it exactly as diverted work is (the + // processor does not flush while the hold is unresolved), so it gets the same cover: without this, a + // queue that predates the MOVING would be timed out by the hold itself under short timeouts with + // relaxation off. Store the handoff, fence, then walk the queue - paired with BacklogEnqueue, which + // enqueues, fences, then re-reads the handoff - so a message enqueued concurrently is stamped by one side + // or the other. + Interlocked.MemoryBarrier(); + foreach (var queued in _backlog) + { + if (!queued.IsInternalCall) queued.MarkHeldForHandoff(handoff.Deadline.Ticks); + } + } + + /// + /// The drain loop has ended, whatever it achieved. A target it could not retire (the lock never came) is left + /// to the server's close, and the hold stays until a replacement is current; a connection that established + /// while the drain was still judging the target is judged again now that the target is gone. + /// + internal void EndDrain() + { + if (!TryGetHandoff(out var handoff)) return; + handoff.TryAdvance(HandoffPhase.Holding, HandoffPhase.AwaitingReplacement); + ReevaluateHandoff(); + } + + /// + /// Judges the current connection against the handoff in progress, if it is established. Run at establishment, + /// when the drain loop ends, and from every heartbeat: the drain loop and the establishment callback each see + /// one instant, and a phase must not be wedged by what that instant happened to miss - a poll landing in the + /// gap between a disconnect and the reconnect, or an address that could not be read at the moment of + /// establishment (rule b: unknown defers to a later pass, and the heartbeat is that pass). + /// + private void ReevaluateHandoff() + { + if (!TryGetHandoff(out var handoff)) return; + if (physical is { EverFullyEstablished: true, RetiredForHandoff: false } current) Judge(handoff, current); + } + + /// + /// An established connection, judged: the replacement resolves the hold; the target, still current once the + /// drain loop is over (its retire never got the lock, or faulted), is retired again - one attempt at a time, + /// idempotent under the lock, so a heartbeat that finds the previous attempt still waiting does not pile on + /// and one that finds it failed tries again. While the target is still being drained, the drain loop owns + /// the judgement. + /// + private void Judge(Handoff handoff, PhysicalConnection connection) + { + switch (Classify(handoff, connection)) + { + case Replacement.Yes: + OnReplacementEstablished(handoff); + break; + case Replacement.Doomed when handoff.Phase != HandoffPhase.Holding && ReferenceEquals(connection, handoff.Target): + if (handoff.TryBeginRetire()) _ = RetireTargetAsync(handoff); + break; + } + } + + /// Whether a handoff is in progress on this bridge, in any phase but Done; see . + internal bool HandoffLive => TryGetHandoff(out _); + + /// + /// The retire outside the drain loop. Forced: the loop's own forced pass has come and gone by the time this + /// runs, and nothing a caller is waiting on can be in flight on a reconnect that the hold has diverted caller + /// work away from since before it was connected. + /// + private async Task RetireTargetAsync(Handoff handoff) + { + try + { + using var lockDeadline = new CancellationTokenSource(TimeSpan.FromSeconds(1)); + await TryRetireTargetAsync(force: true, "still on the node being retired after the drain", lockDeadline.Token).ForAwait(); + } + catch (Exception ex) + { + OnInternalError(ex); + } + finally + { + handoff.EndRetire(); + } + } + + /// + /// A bona fide replacement is current and established: the hold is over and the backlog flushes to it. The + /// handoff stands until the held work has been written (rule a); if nothing is held, it is done now. + /// + private void OnReplacementEstablished(Handoff handoff) + { + if (!(handoff.TryAdvance(HandoffPhase.Holding, HandoffPhase.Flushing) + || handoff.TryAdvance(HandoffPhase.AwaitingReplacement, HandoffPhase.Flushing))) + { + return; // already flushing, or done + } + TryFinishFlush(handoff); + // Store the phase, fence, then load the queue - the mirror image of WakeIfHoldReleased, which every held + // enqueue (single, batch, requeue) goes through. With a full fence on both sides at least one of them + // sees the other's store, so a message held concurrently with the release is not stranded; neither side + // starts a processor for an empty backlog, and StartBacklogProcessor wakes one that is already running. + Interlocked.MemoryBarrier(); + if (BacklogHasItems) StartBacklogProcessor(); + } + + /// + /// The held work has all been written: nothing left to protect (rule a). A message held between the check and + /// the transition (a requeue by a writer that reached the lock late) keeps its cover - that is stamped on the + /// message - and is flushed by the processor like any other; only the FailFast shield is down for it. + /// + private void TryFinishFlush(Handoff handoff) + { + if (handoff.Phase == HandoffPhase.Flushing && !AnyHeldQueued()) handoff.TryAdvance(HandoffPhase.Flushing, HandoffPhase.Done); + } + + /// + /// 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's hold is unresolved, and the current connection is live. + private bool HoldActive => HoldUnresolved && 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. If that + /// connection has gone by itself, whatever is current is the replacement (rule b) and is left alone. 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 retire; such a writer finds the connection retired 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 handoff retired a + /// connection. + internal async ValueTask TryRetireTargetAsync(bool force, string reason, CancellationToken lockDeadline) + { + if (!TryGetHandoff(out var handoff)) return false; + if (handoff.Phase is HandoffPhase.Flushing or HandoffPhase.Done) return handoff.Retired; + + // Lock-free first: the drain polls this every 20ms, and only a pass that is going to retire needs the write + // lock (held across the check and the marking). The in-flight question needs only the connection's own + // lock; and with the target gone there is nothing to retire, so nothing to lock for. The locked pass below + // repeats the judgement on what is current then, which is the one that counts. + var current = physical; + if (handoff.Target is { } drainTarget && ReferenceEquals(current, drainTarget)) + { + if (!force && drainTarget.HasCallerMessagesAwaitingResponse()) return null; + } + else + { + return TargetGone(handoff, current); + } + + PhysicalConnection toRetire; + bool gotLock; + try + { + gotLock = await _singleWriter.TryTakeAsync(lockDeadline).ForAwait(); + } + catch (OperationCanceledException) + { + gotLock = false; + } + catch (ObjectDisposedException) + { + // the bridge was disposed mid-handoff: nothing left to retire, and no replacement is coming + handoff.Finish(); + return false; + } + if (!gotLock) return null; + try + { + if (handoff.Phase is HandoffPhase.Flushing or HandoffPhase.Done) return handoff.Retired; + current = physical; + if (handoff.Target is not { } target || !ReferenceEquals(current, target)) return TargetGone(handoff, current); + toRetire = target; + if (toRetire.RetiredForHandoff) return handoff.Retired; // already retired by another path; its teardown follows that path's lock release + if (!force && toRetire.HasCallerMessagesAwaitingResponse()) return null; + + // Under the lock only the marking: from here no writer that takes the lock writes to the connection (see + // TryRequeueForHandoff), and the backlog processor skips it too, so the teardown itself can happen + // after the release rather than stalling every writer behind event handlers and socket shutdown. The + // marking is a compare-exchange, so of this and a lockless last-resort retire racing it, exactly one + // tears the connection down. + if (!toRetire.TryMarkRetiredForHandoff()) return handoff.Retired; + handoff.Retired = true; + handoff.TryAdvance(HandoffPhase.Holding, HandoffPhase.AwaitingReplacement); + } + finally + { + try + { + _singleWriter.Release(); + } + catch (ObjectDisposedException) + { + // disposed while we held it; nobody is waiting on a disposed mutex + } + } + + RetireConnection(toRetire, reason); + return true; + } + + /// + /// The connection the hold began on is gone - it went away by itself (the server closed it, or it faulted), or + /// there was none when the hold began because the bridge was already disconnected. The phase says so first, + /// so that whatever is current is judged as of a target that is gone: an established connection is the + /// replacement and the hold is over now; one still connecting is judged when it establishes; none means the + /// reconnect is coming, and whatever the hold diverted stays protected for it. Nothing to retire, so the + /// drain loop ends. + /// + private bool TargetGone(Handoff handoff, PhysicalConnection? current) + { + handoff.TryAdvance(HandoffPhase.Holding, HandoffPhase.AwaitingReplacement); + if (current is { EverFullyEstablished: true } && Classify(handoff, current) == Replacement.Yes) OnReplacementEstablished(handoff); + return false; + } + + /// + /// Tears down a connection a handoff has marked retired. 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 is cleanup. + /// A user handler that throws must not leak the connection, nor make the retire look as if it had not happened. + /// + private void RetireConnection(PhysicalConnection connection, string reason) + { + Trace($"recycling {ConnectionType} connection: {reason}"); + try + { + connection.RecordConnectionFailed(ConnectionFailureType.MaintenanceHandoff); + } + catch (Exception ex) + { + OnInternalError(ex); + } + finally + { + try + { + connection.Dispose(); + } + catch (Exception ex) + { + OnInternalError(ex); + } + } + } + public void Dispose() { isDisposed = true; + _handoff?.Finish(); // nothing left to protect, and no replacement is coming // If there's anything in the backlog and we're being torn down - exfil it immediately (e.g. so all awaitables complete) AbandonPendingBacklog(new ObjectDisposedException("Connection is being disposed"), null); try @@ -155,7 +655,7 @@ public void Dispose() isDisposed = true; // make damn sure we don't true to resurrect // If there's anything in the backlog and we're being torn down - exfil it immediately (e.g. so all awaitables complete) - AbandonPendingBacklog(new ObjectDisposedException("Connection is being finalized"), null); + AbandonPendingBacklog(new ObjectDisposedException("Connection is being finalized"), null, locked: false); // shouldn't *really* touch managed objects // in a finalizer, but we need to kill that socket, @@ -212,6 +712,11 @@ private WriteResult FailDueToNoConnection(Message message) public WriteResult TryWriteSync(Message message, bool isReplica) { if (isDisposed) throw new ObjectDisposedException(Name); + if (ShouldHoldForHandoff(message)) + { + EnqueueHeldAndWakeIfReleased(message); + return WriteResult.Success; // queued counts as success + } if (!IsConnected || NeedsReconnect) return QueueOrFailMessage(message); var physical = this.physical; @@ -236,6 +741,11 @@ 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)) + { + EnqueueHeldAndWakeIfReleased(message); + return new ValueTask(WriteResult.Success); // queued counts as success + } if ((!IsConnected || NeedsReconnect) && !bypassBacklog) return new ValueTask(QueueOrFailMessage(message)); var physical = this.physical; @@ -477,8 +987,13 @@ 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 for a failure that is a handoff's own: the backlog then holds work + // diverted off a live connection, which would have been written and answered, so a death of the old + // connection (the window expiring, a blip) or a failed attempt to reach the replacement must leave it for + // the replacement to flush + if (Multiplexer.RawConfig.BacklogPolicy?.AbortPendingOnConnectionFailure == true && failureType != ConnectionFailureType.MaintenanceHandoff + && !IsHandoffOwnFailure(connection)) { AbandonPendingBacklog(innerException, connection); } @@ -540,10 +1055,26 @@ internal void OnDisconnected(ConnectionFailureType failureType, PhysicalConnecti } } - private void AbandonPendingBacklog(Exception ex, PhysicalConnection? connection) + /// Why. + /// The connection, if any, to report against. + /// Take the dequeuers' lock, as every head-changing party must while anything else can run; + /// the finalizer passes false, since it must not block and nothing else is running by then. + private void AbandonPendingBacklog(Exception ex, PhysicalConnection? connection, bool locked = true) { - while (BacklogTryDequeue(out Message? next)) + while (true) { + Message? next; + if (locked) + { + lock (_backlog) // the timeout sweep peeks, then dequeues and compares under this lock; a concurrent dequeue here would trip it + { + if (!BacklogTryDequeue(out next)) break; + } + } + else 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. @@ -558,11 +1089,16 @@ internal void OnFullyEstablished(PhysicalConnection connection, string source) Trace("OnFullyEstablished"); // the attempt itself succeeded (handshake complete), even if this bridge has since moved on and discards it below connection.ReportAttemptOutcome(isSuccess: true, ConnectionFailureType.None, null); + connection.MarkFullyEstablished(); connection.SetIdle(); if (physical == connection && !isDisposed && ChangeState(State.ConnectedEstablishing, State.ConnectedEstablished)) { reportNextFailure = reconfigureNextFailure = true; LastException = null; + // a replacement establishing ends the hold for this bridge - not the very connection the hold is + // draining completing a handshake it was in the middle of when the hold began (rule c: the connection + // says what it is) + if (TryGetHandoff(out var handoff)) Judge(handoff, connection); Interlocked.Exchange(ref failConnectCount, 0); ServerEndPoint.OnFullyEstablished(connection, source); @@ -603,6 +1139,9 @@ internal void OnHeartbeat(bool ifConnectedOnly) StartBacklogProcessor(); } + // a handoff in progress is judged again on every beat, so that no phase is wedged by one missed instant + ReevaluateHandoff(); + runThisTime = !isDisposed && Interlocked.CompareExchange(ref beating, 1, 0) == 0; if (!runThisTime) return; @@ -776,6 +1315,13 @@ 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 + EnqueueHeldBatchAndWakeIfReleased(messages); + return true; + } + if (!IsConnected || NeedsReconnect) { // During reconnection (e.g. MOVED-to-same-endpoint), queue messages to the backlog @@ -928,6 +1474,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) @@ -990,15 +1537,50 @@ internal void AcceptRerouted(Message message) StartBacklogProcessor(); } + /// + /// The one place a message enters the backlog, and so the one place the hold's cover is applied: whatever + /// enters during a handoff is held - diverted off the live connection, queued while the bridge reconnects to + /// the replacement, or parked behind held work being flushed - and gets the same cover, to the same deadline. + /// The shared deadline also keeps the timeout sweeps' head-of-line rule sound for everything enqueued fresh: a + /// message enqueued later does not time out before one enqueued earlier, because an earlier held message + /// expires at the same deadline or sooner, and an ordinary message can only be enqueued after the handoff is + /// done, which is when nothing held is left or that deadline has passed. The one exception is a message + /// requeued by a writer that reached the lock after the retire: it keeps its older write time (the budget it + /// spent counts) behind heads with newer ones, so where its own timeout rather than the shared deadline is the + /// later of the two it can be reported late - by at most the heads' remaining budget - never early, and never + /// lost. + /// + /// Internal traffic (keep-alives, probes) is not held: a stale probe has nothing to gain from lingering to the + /// handoff's deadline, and it is never diverted on its own. An internal message behind held heads is at worst + /// reported late, which costs nothing. The exception is an internal member of a held batch, which the batch + /// helper stamps itself: the batch is held as a unit, and the sweep must not fail one member of it. + /// + /// + /// A message diverted by the hold whose handoff has gone by the time it is enqueued is ordinary work: it is + /// flushed with the rest, and a stamp from a finished handoff would mark it held with no cover behind it. + /// + /// [MethodImpl(MethodImplOptions.AggressiveInlining)] private void BacklogEnqueue(Message message) { - bool wasEmpty = _backlog.IsEmpty; + // the same handoff decides the gate and the stamp: a message gated on one handoff must not carry the + // deadline of the one before it (a diverted message after the hold resolved takes the last handoff's) + var live = TryGetHandoff(out var handoff); + var holdable = !message.IsInternalCall; + if (live && holdable) message.MarkHeldForHandoff(handoff!.Deadline.Ticks); // live: TryGetHandoff said so + bool wasEmpty = !BacklogHasItems; // important that this *precedes* enqueue, to play well with HasPendingCallerFacingItems Interlocked.Increment(ref _backlogCurrentEnqueued); - Interlocked.Increment(ref _backlogTotalEnqueued); _backlog.Enqueue(message); message.SetBacklogged(); + // The total is counted after the enqueue, deliberately: an interlocked increment is a full fence, and the + // re-read below needs one after the enqueue's store. Enqueue, fence, re-read: a hold that began between + // the read above and the enqueue has walked the queue without this message, or is about to walk it with + // it; either way it ends up stamped - with the later handoff's deadline if the two differ, since the stamp + // only ever extends and a same-handoff double stamp is a no-op. A bridge that has never seen a handoff + // pays one null read for this. + Interlocked.Increment(ref _backlogTotalEnqueued); + if (holdable && _handoff is not null && TryGetHandoff(out var late)) message.MarkHeldForHandoff(late.Deadline.Ticks); if (wasEmpty) { @@ -1022,6 +1604,68 @@ private bool BacklogTryDequeue([NotNullWhen(true)] out Message? message) return false; } + /// Whether the backlog holds a caller's message; snapshot enumeration, which a concurrent queue allows. + private bool AnyCallerFacingQueued() + { + foreach (var message in _backlog) + { + if (message.IsCallerFacing) return true; + } + return false; + } + + /// + /// Queues work held for a maintenance handoff - diverted at the gate, or requeued by a writer that reached the + /// lock to find its connection retired - and, through , makes sure it is not + /// stranded if the hold is released concurrently: the queue is stored, then a full fence, then the phase is + /// loaded - the mirror image of , which stores the phase, fences, then + /// loads the queue. With a full fence on both sides at least one of them sees the other's store, so one side or + /// the other starts the processor; neither starts one for an empty backlog. + /// + /// + /// A requeued message is appended, not inserted at the head: a head insert would buy only a best effort at the + /// order the writer would have had (a writer starved past the start of the flush lands behind flushed work + /// anyway), at the price of a second queue and a head-mutation discipline across every backlog operation. The + /// write path never promised more than this in the same races. + /// + /// The message to queue. + /// The message had already been enqueued (to the connection that has since been retired): + /// it keeps its write time, since the budget it has spent still counts, and only forgets that connection. + private void EnqueueHeldAndWakeIfReleased(Message message, bool requeue = false) + { + if (requeue) message.SetRequeued(); + else message.SetEnqueued(null); + BacklogEnqueue(message); + WakeIfHoldReleased(); + } + + /// + /// A batch held as a unit: every member queued, then one fence and one check for the batch. An internal member + /// rides along with the rest (see ), so it is covered with the rest: left to the + /// ordinary timeout it would be failed by the sweep while the batch waits, rather than merely reported late. + /// + private void EnqueueHeldBatchAndWakeIfReleased(List messages) + { + var cover = TryGetHandoff(out var handoff) ? handoff.Deadline.Ticks : 0; + foreach (var message in messages) + { + message.SetEnqueued(null); + if (message.IsInternalCall) message.MarkHeldForHandoff(cover); // zero (no live handoff) covers nothing + BacklogEnqueue(message); + } + WakeIfHoldReleased(); + } + + /// + /// The enqueue side of the pairing with , and the only one: the queue + /// has been stored (by whichever helper called this), then a full fence here, then the phase is loaded. + /// + private void WakeIfHoldReleased() + { + Interlocked.MemoryBarrier(); + if (!HoldUnresolved) StartBacklogProcessor(); + } + [MethodImpl(MethodImplOptions.AggressiveInlining)] private void StartBacklogProcessor() { @@ -1064,14 +1708,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 +1726,13 @@ internal bool HasCallerWork() { if (physical?.HasCallerMessagesAwaitingResponse() == true) return true; - foreach (var message in _backlog) // snapshot enumeration; a concurrent queue is fine to walk - { - if (message.IsCallerFacing) return true; - } - return false; + return AnyCallerFacingQueued(); } + /// + /// 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,17 +1746,21 @@ 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. + // a command a maintenance handoff held is covered until the handoff's deadline; an ordinary command keeps + // the timeout above. The head-of-line rule below still holds across the two: see BacklogEnqueue + static int TimeoutFor(Message m, int timeout) => m.HeldForHandoff ? ServerEndPoint.GetEffectiveTimeoutMilliseconds(timeout, m) : timeout; + while (_backlog.TryPeek(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. - if (!message.HasTimedOut(now, timeout, out var _)) break; // not a timeout - we can stop looking + if (!message.HasTimedOut(now, TimeoutFor(message, timeout), out var _)) break; // not a timeout - we can stop looking lock (_backlog) { // 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 (!message.HasTimedOut(now, timeout, out var _)) break; + if (!message.HasTimedOut(now, TimeoutFor(message, timeout), out var _)) break; if (!BacklogTryDequeue(out var message2) || (message != message2)) // consume it for real { @@ -1163,7 +1814,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 +1911,12 @@ 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 from the connection about to be retired: 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 (HoldUnresolved) return; // try and get the lock; if unsuccessful, retry gotLock = _singleWriter.TryTakeSync(); @@ -1272,12 +1928,12 @@ 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 && !HoldUnresolved) // not while a handoff is holding work back from the connection about to be retired { // 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 // null to a write path that requires a connection. See #3167. - if (this.physical is not { HasOutputPipe: true } physical) break; + if (this.physical is not { HasOutputPipe: true, RetiredForHandoff: false } physical) break; // retired by a handoff: its teardown follows the lock release Message? message; _backlogStatus = BacklogStatus.CheckingForWork; @@ -1290,6 +1946,7 @@ private void ProcessBridgeBacklog() { break; } + _backlogInHand = message; // still "queued" as far as a handoff's flush is concerned, until written } try @@ -1315,6 +1972,7 @@ private void ProcessBridgeBacklog() } finally { + _backlogInHand = null; UnmarkActiveMessage(message); } } @@ -1326,6 +1984,9 @@ private void ProcessBridgeBacklog() physical?.Flush(); } + // the held work has reached the replacement: nothing left to protect (rule a) + if (_handoff is { } handoff) TryFinishFlush(handoff); + _backlogStatus = BacklogStatus.SettingIdle; physical?.SetIdle(); _backlogStatus = BacklogStatus.Inactive; @@ -1371,13 +2032,7 @@ internal void TryFlushStagedWrites(PhysicalConnection physical) public bool HasPendingCallerFacingItems() { - if (BacklogHasItems) - { - foreach (var item in _backlog) // non-consuming, thread-safe, etc - { - if (item.IsCallerFacing) return true; - } - } + if (BacklogHasItems && AnyCallerFacingQueued()) return true; return physical?.HasPendingCallerFacingItems() ?? false; } @@ -1390,6 +2045,35 @@ private WriteResult TimedOutBeforeWrite(Message message) return WriteResult.TimeoutBeforeWrite; } + /// + /// For a writer that has just taken the write lock: if the connection it was headed for was retired by a + /// maintenance handoff while it waited, nothing is written to it - its teardown follows the lock release. A + /// caller's message is queued for the replacement with the rest of the held work (appended, see + /// ); internal traffic and a message bypassing the backlog are failed + /// as having no connection, as they would be on a disconnected bridge. The connection itself says whether a + /// handoff retired it (), so a writer starved across any + /// number of later handoffs still recognises its own. Only the handoff case: a write that loses its connection + /// to an ordinary closure is still reported as a closure (#3167). + /// + private bool TryRequeueForHandoff(PhysicalConnection physical, Message message, bool bypassBacklog, out WriteResult result) + { + if (!physical.RetiredForHandoff) + { + result = default; + return false; + } + if (bypassBacklog || message.IsInternalCall) + { + // not queued, but not written to a connection whose teardown is under way either: a keep-alive or a + // backlog-bypassing message that reached the lock after the retire simply has no connection + result = FailDueToNoConnection(message); + return true; + } + EnqueueHeldAndWakeIfReleased(message, requeue: true); + result = WriteResult.Success; // queued counts as success + return true; + } + /// /// This writes a message to the output stream. /// @@ -1427,11 +2111,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 +2148,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 +2548,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/PhysicalConnection.cs b/src/StackExchange.Redis/PhysicalConnection.cs index 9e5eab6a1..081e13508 100644 --- a/src/StackExchange.Redis/PhysicalConnection.cs +++ b/src/StackExchange.Redis/PhysicalConnection.cs @@ -977,6 +977,67 @@ internal void GetStormLog(StringBuilder sb) } } + /// + /// Whether a maintenance handoff retired this connection, so that a writer which was headed for it and only + /// reached the write lock afterwards can tell a planned replacement from an ordinary closure. Set under the + /// bridge's write lock; the report and dispose follow once the lock is released. Never reset. + /// + internal bool RetiredForHandoff => Volatile.Read(ref _retiredForHandoff) != 0; + + /// Marks the connection retired, once: only the caller that marked it tears it down. + internal bool TryMarkRetiredForHandoff() => Interlocked.CompareExchange(ref _retiredForHandoff, 1, 0) == 0; + + private int _retiredForHandoff; + + // volatile: each is written on one thread and read on others as part of the handoff's identity test, in the + // very races that test exists for; a plain field could be read stale exactly then + private volatile bool _wasHandoffTarget, _everFullyEstablished; + + /// + /// Whether this connection was ever the one a maintenance handoff held work back from. A durable fact about the + /// connection rather than about where the handoff's flow stands: whatever happens to it afterwards - retired by + /// the drain, closed by the server at the end of the window, dropped by itself - its failure is the handoff's + /// own, and work diverted off it must not be abandoned on that account. Never reset. + /// + internal bool WasHandoffTarget => _wasHandoffTarget; + + internal void MarkHandoffTarget() => _wasHandoffTarget = true; + + /// + /// The socket's remote address and port, or null if the socket is not connected yet, is already gone, or + /// there is no socket. Read when a MOVING arrives, to tell the handoff's DNS poll where we are now. + /// + internal System.Net.IPEndPoint? TryGetRemoteEndPoint() + { + var cached = _remoteEndPoint; + if (cached is not null) return cached; + try + { + cached = VolatileSocket?.RemoteEndPoint as System.Net.IPEndPoint; + } + catch (Exception ex) when (ex is System.Net.Sockets.SocketException or ObjectDisposedException or InvalidOperationException) + { + return null; + } + if (cached is not null) _remoteEndPoint = cached; + return cached; + } + + private volatile System.Net.IPEndPoint? _remoteEndPoint; + + /// + /// Whether this connection ever completed its handshake. A failure on a connection that never did is a failed + /// attempt to bring a connection up, which is how a maintenance handoff tells "the replacement is not up yet" + /// from "the replacement was up and then failed" without enumerating failure types. + /// + internal bool EverFullyEstablished => _everFullyEstablished; + + internal void MarkFullyEstablished() + { + _everFullyEstablished = true; + _ = TryGetRemoteEndPoint(); // cache it now, while the socket is certainly connected + } + /// /// Whether any message awaiting a response was issued by a *caller* rather than by us. /// @@ -1027,14 +1088,17 @@ internal void OnBridgeHeartbeat(out int asyncTimeoutDetected, out int syncTimeou { // We only handle async timeouts here, synchronous timeouts are handled upstream. // Those sync timeouts happen in ConnectionMultiplexer.ExecuteSyncImpl() via Monitor.Wait. - if (msg.HasTimedOut(now, timeout, out var elapsed)) + // A command a maintenance handoff held, now written to the replacement, is covered by the + // handoff for as long as that stands; every other command keeps the timeout above. + var effective = msg.HeldForHandoff ? ServerEndPoint.GetEffectiveTimeoutMilliseconds(timeout, msg) : timeout; + if (msg.HasTimedOut(now, effective, out var elapsed)) { if (msg.ResultBoxIsAsync) { bool haveDeltas = msg.TryGetPhysicalState(out _, out _, out long sentDelta, out var receivedDelta) && sentDelta >= 0 && receivedDelta >= 0; var baseErrorMessage = haveDeltas - ? $"Timeout awaiting response (outbound={sentDelta >> 10}KiB, inbound={receivedDelta >> 10}KiB, {elapsed}ms elapsed, timeout is {timeout}ms)" - : $"Timeout awaiting response ({elapsed}ms elapsed, timeout is {timeout}ms)"; + ? $"Timeout awaiting response (outbound={sentDelta >> 10}KiB, inbound={receivedDelta >> 10}KiB, {elapsed}ms elapsed, timeout is {effective}ms)" + : $"Timeout awaiting response ({elapsed}ms elapsed, timeout is {effective}ms)"; var timeoutEx = ExceptionFactory.Timeout(multiplexer, baseErrorMessage, msg, server); multiplexer.OnMessageFaulted(msg, timeoutEx); msg.SetExceptionAndComplete(timeoutEx, this); // tell the message that it is doomed @@ -1055,6 +1119,12 @@ internal void OnBridgeHeartbeat(out int asyncTimeoutDetected, out int syncTimeou } } } + else if (msg.HeldForHandoff) + { + // a command a handoff held carries a longer cover than the ordinary ones written behind it on + // the replacement, so it says nothing about them: look past it + continue; + } else { // This is a head-of-line queue, which means the first thing we hit that *hasn't* timed out means no more will timeout diff --git a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs index 53308faa6..e99a4b898 100644 --- a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs +++ b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs @@ -46,6 +46,24 @@ private MaintenanceNotificationMode MaintenanceMode ? MaintenanceNotificationMode.Disabled : Multiplexer.RawConfig.MaintenanceNotifications; + /// + /// Whether a command's timeout against this server can shrink while a sync caller waits, so that the + /// caller knows whether to wake and re-read its budget in slices or take it in one wait: a relaxed window is + /// open (it can close, and the budget drops back to the configured value), or a handoff is in flight. An + /// extension needs no slicing - the budget is re-read after the wait either way, and a wait that ended under a + /// budget that has since grown simply continues - so a server with nothing in progress costs a sync caller + /// nothing, however long its calls. + /// + internal bool MaintenanceMayChangeTimeouts => GetRelaxedRemaining() > 0 || HandoffLive; + + /// + /// Whether a handoff is in progress on either bridge, in any phase but done - from the hold beginning until the + /// held work has been written to the replacement or the deadline passed. This, not the drain's one-at-a-time + /// claim, is what "a handoff exists" means to anything outside the drain: the claim covers the decision and the + /// drain, which is a shorter span than the handoff, and the two had been confused for each other. + /// + internal bool HandoffLive => interactive?.HandoffLive == true || subscription?.HandoffLive == true; + /// /// Whether to ask this server for maintenance notifications during handshake. /// @@ -159,7 +177,7 @@ private void ReconcileMaintenanceNotifications(PhysicalConnection connection) // can be read without a lock from the heartbeat sweeps. Zero means "no window"; a deadline that computes // to zero is nudged by a tick rather than complicating the sentinel. Comparisons are unchecked // subtraction, as elsewhere in this type, so the ~49-day wrap is a non-event. - private int _relaxedDeadlineTicks; + private TickDeadline _relaxedDeadline; // the notification that last touched the window, reported on faults that happen inside it private int _relaxedType; @@ -198,28 +216,46 @@ 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 - see the overload taking + // the message, which applies the hold's cover to them and to them alone + return result; } /// - /// Milliseconds of relaxation left, or zero if none; clears an expired window as a side-effect. + /// The effective timeout for a particular command: as , and + /// for a command a maintenance handoff held () no sooner than the handoff's + /// deadline - the hold, the rest of the window and one connect attempt, whatever the relaxation setting: 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. The deadline is + /// stamped on the command when it is held, so it holds for as long as the command lives and nothing that happens + /// to the handoff afterwards can withdraw it. Only such a command: an ordinary command on the same server, + /// including one issued by an application that has turned relaxation off and runs short timeouts, keeps its + /// configured timeout. /// - private int GetRelaxedRemaining() + internal static int GetEffectiveTimeoutMilliseconds(int effectiveMilliseconds, Message message) { - var deadline = Volatile.Read(ref _relaxedDeadlineTicks); - if (deadline == 0) return 0; - - var remaining = unchecked(deadline - Environment.TickCount); - if (remaining > 0) return remaining; - - // expired; clear it, but only if nobody has moved it on in the meantime - Interlocked.CompareExchange(ref _relaxedDeadlineTicks, 0, deadline); - return 0; + var deadline = message.HoldCoverDeadlineTicks; + if (deadline != 0) + { + var floor = unchecked(deadline - message.GetWriteTime()); + if (floor > effectiveMilliseconds) effectiveMilliseconds = floor; + } + return effectiveMilliseconds; } + /// + /// Milliseconds of relaxation left, or zero if none; clears an expired window as a side-effect. + /// + private int GetRelaxedRemaining() => _relaxedDeadline.RemainingMilliseconds(); + /// /// An announced disruption has started (or is still running): open or extend the relaxed window. /// @@ -233,13 +269,12 @@ internal bool OnMaintenanceWindowOpened(MaintenanceNotificationType type, long? var config = Multiplexer.RawConfig; var floor = config.MaintenanceRelaxedTimeout; - var cap = config.MaintenanceRelaxedWindowMax; // no time, or a time at or below zero, means "act now" rather than "ignore this": a connection that // arrives mid-window is told what is left of it, and that can legitimately be negative var duration = time is { } value && value > TimeSpan.Zero ? value : floor; if (duration < floor) duration = floor; - if (duration > cap) duration = cap; + duration = ClampToRelaxationCeiling(duration, TimeSpan.Zero); Volatile.Write(ref _relaxedType, (int)type); ExtendRelaxedWindow(duration, $"{type} for {duration.TotalSeconds}s"); @@ -280,24 +315,47 @@ internal void OnMaintenanceWindowClosed(MaintenanceNotificationType type, long? return; } - Volatile.Write(ref _relaxedType, (int)type); var tail = Multiplexer.RawConfig.MaintenancePostEventRelaxedDuration; + + // While a handoff is live - from the hold beginning until the held work has reached the replacement - the + // drain and the reconnect depend on the relaxation the handoff extended: a completion for some other + // operation arriving meanwhile (a MIGRATED for a shard move) must not pull the deadline back under it and + // have the sweep fail the very commands the handoff is carrying across. So during a handoff a close can + // lengthen the window but not shorten it; the replace semantics above apply the rest of the time. + // The type reported on faults follows the window it describes: it changes only when this completion is what + // the window now stands for - not when a handoff's window is preserved past it, or a timeout the handoff + // covered would be attributed to an operation that had finished. + var handoffInFlight = HandoffLive; if (tail <= TimeSpan.Zero) { - Volatile.Write(ref _relaxedDeadlineTicks, 0); - Volatile.Write(ref _relaxedEndedTicks, NudgeFromZero(Environment.TickCount)); + if (!handoffInFlight) + { + _relaxedDeadline.SetIn(TimeSpan.Zero); // ended now, and recorded as such for fault attribution + Volatile.Write(ref _relaxedType, (int)type); + } Multiplexer.Trace($"{type}: relaxation ended", ToString()); return; } - var deadline = NudgeFromZero(unchecked(Environment.TickCount + (int)tail.TotalMilliseconds)); - Volatile.Write(ref _relaxedDeadlineTicks, deadline); - Volatile.Write(ref _relaxedEndedTicks, deadline); + if (!handoffInFlight) + { + _relaxedDeadline.SetIn(tail); + Volatile.Write(ref _relaxedType, (int)type); + } + else if (_relaxedDeadline.TryExtendBy(tail, out _)) + { + Volatile.Write(ref _relaxedType, (int)type); + } Multiplexer.Trace($"{type}: relaxation continues for {tail.TotalSeconds}s (post-event)", ToString()); } - private volatile EndPoint? _handoffTarget; - private int _handoffTargetExpiryTicks; + // The pin: the target, its expiry and the generation of the MOVING that set it, kept consistent under + // _maintenanceSync (every touch is rare: a MOVING, a connect, a heartbeat), which is what lets the heartbeat's + // expiry check and a fresh SetHandoffTarget coexist without one nulling the other's pin. + private EndPoint? _handoffTarget; + private TickDeadline _handoffTargetExpiry; + private int _handoffTargetGeneration; + private int _movingGeneration; // one per MOVING acted on; a later MOVING's pin outranks an earlier one's decision private int _handoffInFlight, _handoffRecycles; /// @@ -314,44 +372,67 @@ internal void OnMaintenanceWindowClosed(MaintenanceNotificationType type, long? /// Expires, and that is not decoration. Without it, a server that named an address which turns out to be /// unreachable would pin this endpoint to it for the lifetime of the multiplexer, because every reconnect /// would keep trying the same dead target. On expiry we fall back to resolving the endpoint normally, which - /// is what we would have done anyway. + /// is what we would have done anyway. Until then it stays, whatever establishes: both bridges reconnect + /// through it, and the second one's reconnect must not be sent back to DNS - still naming the node being + /// retired - because the first one's replacement happened to establish first. /// /// internal EndPoint? HandoffTarget { get { - var target = _handoffTarget; - if (target is null) return null; - - if (unchecked(Environment.TickCount - Volatile.Read(ref _handoffTargetExpiryTicks)) >= 0) + lock (_maintenanceSync) { - _handoffTarget = null; // expired; resolve the endpoint the usual way from here on - return null; + if (_handoffTarget is null) return null; + if (!_handoffTargetExpiry.IsActive) + { + _handoffTarget = null; // expired; resolve the endpoint the usual way from here on + return null; + } + return _handoffTarget; } + } + } - return target; + /// + /// Points the next connection attempt at a named replacement, for as long as the announced window lasts - unless + /// a later MOVING has pinned something since (a handoff's decision can resolve, after a DNS poll, well + /// after the next notification arrived; the later one knows more). + /// + private void SetHandoffTarget(EndPoint target, TimeSpan window, int generation) + { + lock (_maintenanceSync) + { + if (generation < _handoffTargetGeneration) return; // whether or not that later pin has since expired: it knew more + _handoffTargetExpiry.SetIn(window < TimeSpan.FromSeconds(1) ? TimeSpan.FromSeconds(1) : window); + _handoffTarget = target; + _handoffTargetGeneration = generation; } } /// - /// Points the next connection attempt at a named replacement, for as long as the announced window lasts. + /// Drops the pin when a MOVING names no successor: the pinned node is then, as likely as not, the very one + /// now being retired, and a reconnect dialled through the stale pin would land on it. Only a pin this or an earlier + /// MOVING set: a later one's stands. /// - internal void SetHandoffTarget(EndPoint target, TimeSpan window) + private void ClearHandoffTarget(int generation) { - Volatile.Write(ref _handoffTargetExpiryTicks, unchecked(Environment.TickCount + (int)Math.Max(window.TotalMilliseconds, 1000))); - _handoffTarget = target; + lock (_maintenanceSync) + { + if (generation < _handoffTargetGeneration) return; + _handoffTarget = null; + _handoffTargetGeneration = generation; + } } /// - /// Forgets any handoff target, once a connection has been established. + /// Reads the pin on every heartbeat so that its expiry is noticed while it is fresh. Expiry is lazy, on read, + /// and the connect path is the only other reader: a server that stays connected for weeks after a handoff would + /// otherwise carry a months-old expiry into its next reconnect, where a wrapped tick count could read it as in + /// force again. /// - /// - /// 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; + internal void ExpireHandoffTargetIfDue() => _ = HandoffTarget; + private volatile string? _lastHandoffOutcome; /// @@ -384,27 +465,33 @@ 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 + if (budget > MaxHandoffWindow) budget = MaxHandoffWindow; // an absurd announced window must not overflow a delay or a deadline + + // One decision-and-drain at a time per server: two drains on the same bridges would end each other's holds. + // A MOVING arriving while one is in flight (which can be most of a window: the jitter, the DNS poll, the + // drain) is not dropped, though - the handoff in flight will replace the connections, and what this one adds + // is where the reconnect should go, if it says: the pin is updated, and the next dial takes it. + var generation = Interlocked.Increment(ref _movingGeneration); 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, generation); + else ClearHandoffTarget(generation); // as on the direct path: no successor named means any pinned node may itself be retiring + Multiplexer.Trace(successor is null ? "MOVING: a handoff is already in flight; pin dropped" : "MOVING: a handoff is already in flight; successor noted", ToString()); return; } + var current = connection.TryGetRemoteEndPoint(); // null if the socket is gone or not connected: nothing to compare against - 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); + _ = HandoffAsync(budget, successor, current, generation); } - private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? currentAddress) + private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPEndPoint? currentEndPoint, int generation) { var watch = ValueStopwatch.StartNew(); var replaced = false; + var claimed = true; // the one-at-a-time claim taken by OnMovingAnnounced; released once, by whoever holds it try { // Spread the fleet, but only by a fraction of the window - see MaintenanceHandoff.GetJitter for why @@ -416,12 +503,16 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? var decision = await MaintenanceHandoff.DecideAsync( EndPoint, successor, - currentAddress, + currentEndPoint?.Address, remaining, pollInterval: TimeSpan.FromSeconds(1), // records carry a 5s TTL, so this is several looks per record 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(); @@ -432,6 +523,9 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? switch (decision.Action) { case HandoffAction.Recycle: + // DNS has moved to a new address, so a reconnect reporting the old one is back on the retiring + // node; a pin left by an earlier handoff is stale now and must not dial it + ClearHandoffTarget(generation); await DrainThenRecycleAsync(remaining, decision.Reason).ForAwait(); replaced = true; break; @@ -440,7 +534,9 @@ 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(); + // nothing to re-resolve, or nothing known to compare against: no second address has been seen + ClearHandoffTarget(generation); + await DrainThenRecycleAsync(Remaining(window, watch), decision.Reason).ForAwait(); replaced = true; break; case HandoffAction.MoveTo when decision.Target is { } target: @@ -448,7 +544,7 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? // this only re-read the topology and recycled, which measurably did not work: we recycled // at +6.2s, landed back on the node being retired because DNS had not moved yet, and were // closed at +21.6s anyway - exactly the outcome the handoff exists to avoid. - SetHandoffTarget(target, remaining); + SetHandoffTarget(target, remaining, generation); await DrainThenRecycleAsync(remaining, decision.Reason).ForAwait(); replaced = true; break; @@ -458,6 +554,11 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? break; } + // The one-at-a-time guard covers the decision and the drain, not the warning: that is observability, and + // holding the guard through its wait would drop a genuine second MOVING inside the first window outright - + // no pin, no hold - which is the pre-handoff failure for the second window. + claimed = false; + Volatile.Write(ref _handoffInFlight, 0); if (replaced) await WarnIfNotEstablishedInTimeAsync(watch, window).ForAwait(); } catch (Exception ex) @@ -466,41 +567,219 @@ private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? } finally { - Volatile.Write(ref _handoffInFlight, 0); + // only if still ours: after the early release the claim may belong to a second handoff, which must not + // find its guard gone and a third MOVING racing it + if (claimed) Volatile.Write(ref _handoffInFlight, 0); } } + /// What is left of an announced window; never negative. + private static TimeSpan Remaining(TimeSpan window, ValueStopwatch watch) + { + var left = window - TimeSpan.FromMilliseconds(watch.ElapsedMilliseconds); + return left > TimeSpan.Zero ? left : TimeSpan.Zero; + } + + /// + /// How long a handoff holds new work and waits for in-flight commands before replacing the connections + /// regardless: half the relaxed timeout (five seconds by default), never more than half of what is left of the + /// window, never less than , never more than the relaxation ceiling (or the floor, if + /// that is higher), and never more than 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. Half of what is left of the window is an absolute ceiling as well, + /// whatever the relaxed timeout: a generous relaxed timeout must not turn the hold into a full-window stall that + /// recycles at the deadline with work in flight, which is the failure this whole mechanism 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 + // the relaxation the drain depends on is itself bounded by the user's ceiling: holding for longer than the + // sweep will relax would pay the hold's latency and still fail the command. The floor still applies, as for a + // relaxed timeout of zero: a sub-second ceiling says "do not relax", not "do not drain". + cap = ClampToRelaxationCeiling(cap, DrainFloor); + return cap > budget ? budget : cap; + } + /// - /// Lets in-flight caller work finish, then replaces the connections. + /// The one statement of what bounds: a span of + /// relaxation, or of anything that depends on relaxation, never exceeds it - except that a span may carry its own + /// floor, below which the ceiling does not cut (the drain's floor, the cover's drain-plus-connect). + /// + private TimeSpan ClampToRelaxationCeiling(TimeSpan span, TimeSpan floor) + { + var ceiling = Multiplexer.RawConfig.MaintenanceRelaxedWindowMax; + if (ceiling < floor) ceiling = floor; + return span > ceiling ? ceiling : span; + } + + /// 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); + + /// + /// The longest MOVING window acted on. Measured windows are 15 seconds; an hour is already absurd, and anything + /// larger (a buggy or hostile announcement) would overflow the delays, cancellation sources and tick deadlines + /// downstream, or hold traffic for days. + /// + private static readonly TimeSpan MaxHandoffWindow = TimeSpan.FromHours(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 /// 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) - { - await Task.Delay(TimeSpan.FromMilliseconds(20)).ForAwait(); - } - - var drained = !HasCallerWork(); - var recycled = (interactive?.RecycleConnection(reason) == true) | (subscription?.RecycleConnection(reason) == true); + // With less than the floor left of the window there is no drain 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. The hold still begins, because the retire is built on it - the target, the + // requeue of a writer that reaches the lock late, the cover for what the reconnect carries - but it lasts + // only the instant of the replacement. 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 handoff's + // deadline below covers it regardless, stamped on each held command. The reconnect can take until the end + // of the window plus one connect attempt: a MoveTo pins the socket target to the named successor for the + // rest of the window, and a successor that does not answer is tried until that expires before the endpoint + // is resolved the ordinary way. + ExtendRelaxedWindow(ClampToRelaxationCeiling(cap + Multiplexer.RawConfig.MaintenanceRelaxedTimeout, TimeSpan.Zero), "MOVING handoff"); + var connect = TimeSpan.FromMilliseconds(Multiplexer.RawConfig.ConnectTimeout); + var pinned = cap + LockGrace > budget ? cap + LockGrace : budget; + // the user's ceiling on relaxation bounds the cover too - a huge announced window must not floor the held + // commands' timeouts for its whole length - but the cover always spans at least the drain itself and one + // connect attempt, which is what the held commands need; it applies to them alone, so an application that + // has turned relaxation off keeps its short timeouts for everything else + var cover = ClampToRelaxationCeiling(pinned + connect, cap + LockGrace + connect); + + // 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?.BeginHold(cover); + subscriptionBridge?.BeginHold(cover); + + // One loop per bridge, each with its own lock deadline, run side by side: a writer wedged on one bridge's + // lock must not spend the other's time, and the other's idle connection must still be retired before the + // server closes it. Each bridge is released the moment its own connection is settled. + var interactiveTask = RetireBridgeAsync(interactiveBridge, watch, cap, reason); + var subscriptionTask = RetireBridgeAsync(subscriptionBridge, watch, cap, reason); + var interactiveResult = await interactiveTask.ForAwait(); + var subscriptionResult = await subscriptionTask.ForAwait(); + + var recycled = interactiveResult.Recycled == true || subscriptionResult.Recycled == 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)"), + + (interactiveResult.Forced || subscriptionResult.Forced ? " (still busy at the drain cap)" : " (drained)") + + (interactiveResult.Recycled is null || subscriptionResult.Recycled is null ? " (write lock not obtained; left to the server)" : ""), ToString()); } + /// + /// What did: whether the connection was recycled (null: the write lock + /// could not be had, and the connection was left to the server's close), and whether the cap was reached with + /// work still in flight. A struct rather than a tuple: the library does not reference System.ValueTuple. + /// + private readonly struct BridgeRetirement(bool? recycled, bool forced) + { + public bool? Recycled { get; } = recycled; + public bool Forced { get; } = forced; + } + + /// + /// Retires one bridge's connection once drained, or at the cap regardless, and releases its hold; see + /// . + /// + private async Task RetireBridgeAsync(PhysicalBridge? bridge, ValueStopwatch watch, TimeSpan cap, string reason) + { + if (bridge is null) return new(false, false); + bool forced = false; + try + { + using var lockDeadline = new CancellationTokenSource(cap + LockGrace); + while (true) + { + var force = watch.ElapsedMilliseconds >= cap.TotalMilliseconds; + forced |= force; + var result = await bridge.TryRetireTargetAsync(force, reason, lockDeadline.Token).ForAwait(); + if (result is { } done) + { + // whether we replaced the connection or it went away by itself, a replacement is now on its way, + // and the protection stays up for it until it establishes or the span runs out + return new(done, forced); + } + if (lockDeadline.IsCancellationRequested) + { + // The write lock could not be had even for the forced pass: something is wedged on it. Leaving the + // connection to the server's window-end close would fail everything in flight and leave the bridge + // dark for the reconnect - the whole window's traffic, to protect at most one wedged writer's + // message from a write fault. So, as a last resort, retire without the lock, which is what every + // recycle did before handoffs existed. The protection stays up for the diverted work either way. + return new(bridge.RetireTargetWithoutLock(reason) ? true : null, forced); + } + await Task.Delay(TimeSpan.FromMilliseconds(20)).ForAwait(); + } + } + catch (Exception ex) + { + // as the give-up above: stop holding new work, keep protecting what was already diverted + Multiplexer.OnInternalError(ex, EndPoint); + return new(null, forced); + } + finally + { + try + { + bridge.EndDrain(); // idempotent; one bridge failing to release must not affect the other + } + catch (Exception ex) + { + Multiplexer.OnInternalError(ex, EndPoint); + } + } + } + /// /// Warns when a replacement connection is not fully established by the time the announced window ends. /// @@ -546,33 +825,14 @@ private static Random RandomFor(ServerEndPoint server) => private void ExtendRelaxedWindow(TimeSpan duration, string cause) { - var candidate = NudgeFromZero(unchecked(Environment.TickCount + (int)duration.TotalMilliseconds)); - while (true) + // a new notification never shortens an existing window: a FAILING_OVER arriving inside a longer + // MIGRATING window must not cut it short + if (_relaxedDeadline.TryExtendBy(duration, out _)) { - var current = Volatile.Read(ref _relaxedDeadlineTicks); - - // a new notification never shortens an existing window: a FAILING_OVER arriving inside a longer - // MIGRATING window must not cut it short - if (current != 0 && unchecked(candidate - current) <= 0) return; - if (Interlocked.CompareExchange(ref _relaxedDeadlineTicks, candidate, current) == current) - { - Volatile.Write(ref _relaxedEndedTicks, candidate); - Multiplexer.Trace($"timeouts relaxed: {cause}", ToString()); - return; - } + Multiplexer.Trace($"timeouts relaxed: {cause}", ToString()); } } - /// - /// When the most recent window was due to end, whether or not it has; zero if there has never been one. - /// - /// - /// A mirror of the deadline rather than a record of when it was cleared, which is what makes it cheap: - /// expiry is lazy (it happens on the next read), so there is no single moment at which to stamp "the - /// window closed", and the deadline already *is* that moment. - /// - private int _relaxedEndedTicks; - /// How long after a window closes a fault may still be attributed to it, at minimum. /// /// The bridge heartbeat raises timeouts on roughly a one-second cadence rather than at the deadline, so a @@ -594,7 +854,7 @@ private void ExtendRelaxedWindow(TimeSpan duration, string cause) /// So a window that closed no longer ago than the command could have been waiting still counts. That is /// the tightest bound that catches every genuine case, and it is deliberately a bound rather than a /// certainty: without per-message state - which the timeout sweeps rule out, see - /// - a client cannot know whether *this* command overlapped + /// - a client cannot know whether *this* command overlapped /// the window, only whether it could have. /// /// @@ -604,7 +864,9 @@ internal MaintenanceNotificationType GetMaintenanceTypeForFault(int timeoutMilli var active = ActiveMaintenanceType; if (active != MaintenanceNotificationType.None) return active; - var ended = Volatile.Read(ref _relaxedEndedTicks); + // the deadline is kept after it passes (see TickDeadline), so it is also the record of when the most recent + // window ended: there is no single moment at which to stamp "the window closed", and the deadline is it + var ended = _relaxedDeadline.Ticks; if (ended == 0) return MaintenanceNotificationType.None; // never had a window at all var since = unchecked(Environment.TickCount - ended); @@ -614,11 +876,6 @@ internal MaintenanceNotificationType GetMaintenanceTypeForFault(int timeoutMilli : MaintenanceNotificationType.None; } - /// - /// Zero is the "no window" sentinel, so a deadline that lands on it moves by a tick. - /// - private static int NudgeFromZero(int ticks) => ticks == 0 ? 1 : ticks; - /// /// Whether this notification is new, per the conservative dedup described on . /// diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs index e203139c6..4f092af03 100644 --- a/src/StackExchange.Redis/ServerEndPoint.cs +++ b/src/StackExchange.Redis/ServerEndPoint.cs @@ -1084,15 +1084,12 @@ internal void OnRepeatedConnectFailure(int consecutiveFailures) var last = Volatile.Read(ref _lastConnectFailureRefreshTicks); if (last != 0 && unchecked(now - last) < interval) return; - if (Interlocked.CompareExchange(ref _lastConnectFailureRefreshTicks, NudgeFromZeroTicks(now), last) != last) return; + if (Interlocked.CompareExchange(ref _lastConnectFailureRefreshTicks, Maintenance.TickDeadline.NudgeFromZero(now), last) != last) return; Multiplexer.Logger?.LogInformationRefreshingAfterConnectFailures(new(this), consecutiveFailures); Multiplexer.ReconfigureIfNeeded(EndPoint, fromBroadcast: false, $"{consecutiveFailures} consecutive connect failures"); } - /// Zero means "never", so a tick count that lands on it moves by one. - private static int NudgeFromZeroTicks(int ticks) => ticks == 0 ? 1 : ticks; - /// /// Subscribes to the configuration-change broadcast on a RESP3 interactive connection that is shared with /// pub/sub (). Then there is no separate @@ -1126,9 +1123,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) @@ -1241,6 +1235,7 @@ internal void OnHeartbeat() { try { + ExpireHandoffTargetIfDue(); interactive?.OnHeartbeat(false); subscription?.OnHeartbeat(false); } diff --git a/tests/StackExchange.Redis.Tests/InProcessTestServer.cs b/tests/StackExchange.Redis.Tests/InProcessTestServer.cs index 726eb0bbb..d8af405f0 100644 --- a/tests/StackExchange.Redis.Tests/InProcessTestServer.cs +++ b/tests/StackExchange.Redis.Tests/InProcessTestServer.cs @@ -501,14 +501,27 @@ protected override void Dispose(bool disposing) */ // written by the test thread, read by the server's own thread(s): store as ticks so the read/write can // be made explicitly visible (a plain TimeSpan field can be missed indefinitely) - public void SetLatency(TimeSpan latency) => Volatile.Write(ref _latencyTicks, latency.Ticks); + public void SetLatency(TimeSpan latency) => SetLatency(latency, only: null); + + /// + /// Hold every reply by , or only the replies to one command: a test that wants the + /// workload slow while a reconnecting client's handshake stays fast (so that a short timeout bites the workload + /// and not the handshake) names the workload's command here. + /// + internal void SetLatency(TimeSpan latency, RedisCommand? only) + { + Volatile.Write(ref _latencyOnly, only is { } command ? (int)command : -1); + Volatile.Write(ref _latencyTicks, latency.Ticks); + } private long _latencyTicks; + private int _latencyOnly = -1; protected override ValueTask ClientPauseAsync(RedisClient client, in RedisRequest request) { var latency = TimeSpan.FromTicks(Volatile.Read(ref _latencyTicks)); - if (latency > TimeSpan.Zero & request.KnownCommand != RedisCommand.QUIT) + var only = Volatile.Read(ref _latencyOnly); + if (latency > TimeSpan.Zero & request.KnownCommand != RedisCommand.QUIT & (only < 0 || (int)request.KnownCommand == only)) { Log($"[{client}] holding {request.Command} response by {latency.TotalMilliseconds}ms"); return new(Task.Delay(latency)); diff --git a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs index bfdb3a7a5..9631837da 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,324 @@ 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); + + [Fact] + public Task MovingUnderSteadyLoadWithShortTimeoutsAndRelaxationOffKeepsHeldWork() + // The sharp case for the hold's cover: relaxation off, and a timeout shorter than the time a held command spends + // waiting (the drain of the in-flight ones, the replacement's handshake, then its own turn). The held commands + // must keep their cover until they are answered on the replacement - not only until it establishes - or the + // sweep fails them in the gap; everything else keeps the short timeout. Only the workload is slow: the test + // server answers one request at a time, so slowing the handshake too would push *it* past the timeout and + // nothing could ever resume, which is a harness limit and not the behaviour under test. + => SteadyLoadHandoffAsync( + configure: c => + { + c.MaintenanceRelaxedTimeout = TimeSpan.Zero; + c.AsyncTimeout = 1000; + c.SyncTimeout = 1000; + }, + serverLatencyMilliseconds: 300, + latencyOnlyFor: RedisCommand.PING, + workers: 2); // two in flight at most: serially served, their ordinary wait stays inside the timeout + + /// When, in the first handoff's life, the second MOVING arrives. + public enum SecondMovingAt + { + /// At once: the first handoff is still in its jitter, deciding. + Decision, + + /// Inside the first drain, kept open by a slow command in flight. + Drain, + + /// After the first recycle, while the held work reaches the replacement. + Flush, + } + + [Theory] + [InlineData(true, SecondMovingAt.Decision)] + [InlineData(false, SecondMovingAt.Decision)] + [InlineData(true, SecondMovingAt.Drain)] + [InlineData(false, SecondMovingAt.Drain)] + [InlineData(true, SecondMovingAt.Flush)] + [InlineData(false, SecondMovingAt.Flush)] + public async Task MovingTwiceInOneWindowLosesNothing(bool secondNamesSuccessor, SecondMovingAt at) + { + // The contract for back-to-back MOVINGs inside one window is best effort (see docs/ServerMaintenanceEvent.md): + // nothing is lost or abandoned contrary to the backlog policy, and traffic resumes within the handoff's own + // bound. Which successor the pin ends up naming, and whether the second notification starts a drain of its own + // or only redirects the first's reconnect, are not asserted: they are the mechanism, not the promise. + var logs = new CapturingLoggerFactory(); + var (server, conn, events) = await ConnectAsync(log, logs); + using (server) + await using (conn) + { + 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 first = new IPEndPoint(IPAddress.Parse("127.0.0.9"), 6380); + var second = secondNamesSuccessor ? new IPEndPoint(IPAddress.Parse("127.0.0.10"), 6380) : null; + + if (at == SecondMovingAt.Drain) + { + // slow replies keep the first drain open (3s, inside the 5s cap) for the second MOVING to land in. Every + // command, not one: a per-command latency lifts the throttle from the workers' PINGs, and four + // unthrottled workers for ten seconds starve the timing-sensitive tests running alongside on a small + // CI runner (observed: three different neighbours failing once each across two runs) + server.SetLatency(TimeSpan.FromSeconds(3)); + Assert.True(await Poll.UntilAsync(() => conn.GetCounters().Interactive.SentItemsAwaitingResponse > 0), "slow commands should be in flight before MOVING arrives"); + } + + var watch = System.Diagnostics.Stopwatch.StartNew(); + server.SendMoving(null, timeSeconds: 10, newEndpoint: first, sequenceId: 0); + Assert.Equal(MaintenanceNotificationType.Moving, (await events.NextAsync()).NotificationType); + + switch (at) + { + case SecondMovingAt.Drain: + await Task.Delay(1500); // past the jitter (at most a tenth of the window), inside the drain the slow command holds open + break; + case SecondMovingAt.Flush: + Assert.True(await Poll.UntilAsync(() => endpoint.HandoffRecycles > 0, timeoutMilliseconds: 8_000), "the first handoff should have replaced the connection"); + break; + } + server.SendMoving(null, timeSeconds: 10, newEndpoint: second, sequenceId: 1); + Assert.Equal(MaintenanceNotificationType.Moving, (await events.NextAsync()).NotificationType); + if (at == SecondMovingAt.Drain) server.SetLatency(TimeSpan.FromMilliseconds(40)); // the replies already being held keep their 3s; what follows flows + log.WriteLine($"second MOVING ({(second is null ? "no successor" : "successor named")}) sent at +{watch.ElapsedMilliseconds}ms, during {at}; pin now {endpoint.HandoffTarget?.ToString() ?? "(none)"}"); + + var before = Volatile.Read(ref completed); + Assert.True( + await Poll.UntilAsync(() => Volatile.Read(ref completed) > before + 20, timeoutMilliseconds: 20_000), + "traffic should resume after the second MOVING"); + log.WriteLine($"traffic flowing again at +{watch.ElapsedMilliseconds}ms; recycles={endpoint.HandoffRecycles}; pin {endpoint.HandoffTarget?.ToString() ?? "(none)"}"); + + 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)); + } + } + finally + { + stop.Cancel(); + } + } + } + + private async Task SteadyLoadHandoffAsync(Action? configure, int serverLatencyMilliseconds = 40, RedisCommand? latencyOnlyFor = null, int workers = 4) + { + // 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(serverLatencyMilliseconds), latencyOnlyFor); + var db = conn.GetDatabase(); + using var stop = new CancellationTokenSource(); + long completed = 0; + var failures = new List(); + var workerTasks = Enumerable.Range(0, workers).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 + { + // 20 completions can take a while when the workload is deliberately slow and served one at a time + 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 = new[] { 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, the recycle would have + // failed the command, and this await would throw) + foreach (var pending in slow) Assert.True(await pending >= TimeSpan.Zero); + // 3s for the command, up to 1s of jitter, lock and 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 = new[] { 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 (it was observed answered at once, and then nothing was in + // flight to cap on). 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 + the lock grace (1s) + poll granularity + // is the mechanism's own worst case, about 7.5s; the bug being guarded against is a recycle at the 15s + // deadline, so 12s discriminates with room on both sides, where 10s left two to three seconds over the + // worst case on a slow machine + 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 + 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() { @@ -884,7 +1205,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(