MOVING handoff: hold new work back while draining, so the recycle happens within a round trip instead of at the deadline

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:

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -2736,29 +2736,34 @@ internal static void ThrowFailed<T>(TaskCompletionSource<T>? 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;
-                            }
                         }
                     }
                 }

src/StackExchange.Redis/Maintenance/TickDeadline.cs

@@ -0,0 +1,105 @@
+using System;
+using System.Threading;
+
+namespace StackExchange.Redis.Maintenance;
+
+/// <summary>
+/// A deadline expressed in <see cref="Environment.TickCount"/> 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 (<see cref="MaxMilliseconds"/>) 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 <see cref="Ticks"/>,
+/// so it also records <em>when</em> 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.
+/// </summary>
+internal struct TickDeadline
+{
+    private int _ticks;
+
+    /// <summary>
+    /// 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 <c>int</c> 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.
+    /// </summary>
+    private const int MaxMilliseconds = int.MaxValue / 4;
+
+    /// <summary>The raw deadline, passed or not, or zero when unset (or so long passed that it has been cleared).</summary>
+    public int Ticks => Volatile.Read(ref _ticks);
+
+    public void SetIn(TimeSpan span) => Volatile.Write(ref _ticks, NudgeFromZero(unchecked(Environment.TickCount + Clamp(span))));
+
+    /// <summary>A span as whole milliseconds, never negative and never beyond <see cref="MaxMilliseconds"/>.</summary>
+    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);
+
+    /// <summary>
+    /// Moves the deadline out to now plus <paramref name="span"/> unless it is already later than that.
+    /// </summary>
+    /// <param name="span">How far from now the deadline should be, at least.</param>
+    /// <param name="deadline">The deadline now in force, whether or not this call set it.</param>
+    /// <returns>Whether this call extended it.</returns>
+    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;
+    }
+
+    /// <summary>
+    /// Moves the deadline out to <paramref name="ticks"/> (a tick count, as <see cref="Ticks"/> reads) unless it is
+    /// already later than that; zero means "no deadline" and extends nothing.
+    /// </summary>
+    /// <returns>Whether this call extended it.</returns>
+    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;
+        }
+    }
+
+    /// <summary>
+    /// Milliseconds until the deadline, or zero if unset or passed. A passed deadline is kept (see <see cref="Ticks"/>)
+    /// until it is so long past that a wrapping tick count could make it look future again, and then cleared.
+    /// </summary>
+    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;
+
+    /// <summary>Zero means "unset" (or "never"), so a tick count that lands on it moves by one.</summary>
+    public static int NudgeFromZero(int ticks) => ticks == 0 ? 1 : ticks;
+}

src/StackExchange.Redis/Message.cs

@@ -892,7 +892,45 @@ internal void SetWriteTime()
         /// <see cref="SetEnqueued"/>; 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.
         /// </summary>
-        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);
+        }
+
+        /// <summary>
+        /// 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 <see cref="SetEnqueued"/>, which restarts it.
+        /// </summary>
+        internal void SetRequeued()
+        {
+            _enqueuedTo = null;
+            _queuedStampSent = _queuedStampReceived = -1;
+        }
+
+        /// <summary>
+        /// 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.
+        /// </summary>
+        internal bool HeldForHandoff => _holdCover.Ticks != 0;
+
+        /// <summary>
+        /// The handoff's deadline, in <see cref="Environment.TickCount"/> 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.
+        /// </summary>
+        internal int HoldCoverDeadlineTicks => _holdCover.Ticks;
+
+        /// <summary>Holds the message for a handoff, covered until <paramref name="deadlineTicks"/>; a later deadline replaces an earlier one, never the reverse (and zero, no deadline, covers nothing).</summary>
+        internal void MarkHeldForHandoff(int deadlineTicks) => _holdCover.ExtendTo(deadlineTicks);
+
+        private Maintenance.TickDeadline _holdCover;
 
         private int _writeTickCount;