diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.Events.cs b/src/StackExchange.Redis/ConnectionMultiplexer.Events.cs index b2910fb8c..fd5bd7c7f 100644 --- a/src/StackExchange.Redis/ConnectionMultiplexer.Events.cs +++ b/src/StackExchange.Redis/ConnectionMultiplexer.Events.cs @@ -89,6 +89,17 @@ private void OnEndpointChanged(EndPoint endpoint, EventHandler public event EventHandler? ServerMaintenanceEvent; + /// + /// How host names are resolved during a maintenance handoff. + /// + /// + /// A seam rather than a call to directly, because no in-process fake can move a + /// DNS record: the handoff's whole decision turns on the answer *changing*, and the only way to test that + /// deterministically is to supply the answers. Defaults to real DNS. + /// + internal Func> AddressResolver { get; set; } + = Maintenance.AdvertisedAddressProbe.DefaultResolveAsync; + // recently-raised (type, sequence) pairs, so one logical event raises one event however many nodes told // us. Small and fixed: the copies arrive within milliseconds of each other, so a handful of slots covers // any realistic proxy count even with other notifications interleaved. diff --git a/src/StackExchange.Redis/Enums/ConnectionFailureType.cs b/src/StackExchange.Redis/Enums/ConnectionFailureType.cs index 9f0df5ba0..3fd7847b3 100644 --- a/src/StackExchange.Redis/Enums/ConnectionFailureType.cs +++ b/src/StackExchange.Redis/Enums/ConnectionFailureType.cs @@ -68,5 +68,22 @@ public enum ConnectionFailureType /// [Experimental(Experiments.GeoRedundantFailover, UrlFormat = Experiments.UrlFormat)] CircuitBreaker, + + /// + /// The connection was replaced deliberately, to move off an endpoint the server announced it is + /// retiring. + /// + /// + /// Not a fault: the connection was working, and we chose to replace it while a replacement address was + /// known rather than wait to be disconnected. Reported here because the alternative is worse - the + /// replacement raises , so without this a consumer + /// tracking connection state sees a restore with no matching failure, and no reason for the churn. + /// + /// Consumers alerting on should filter this out: + /// it is expected during planned maintenance and says nothing is wrong. is + /// the precedent for a deliberate client action being reported this way. + /// + /// + MaintenanceHandoff, } } diff --git a/src/StackExchange.Redis/Maintenance/AdvertisedAddressProbe.cs b/src/StackExchange.Redis/Maintenance/AdvertisedAddressProbe.cs new file mode 100644 index 000000000..6170e4dc1 --- /dev/null +++ b/src/StackExchange.Redis/Maintenance/AdvertisedAddressProbe.cs @@ -0,0 +1,227 @@ +using System; +using System.Net; +using System.Threading; +using System.Threading.Tasks; + +namespace StackExchange.Redis.Maintenance; + +/// +/// Asks DNS the two questions a handoff needs: what replaces the address we are leaving, and are we still +/// advertised at all. +/// +/// +/// A poll rather than a lookup, and that is the whole point. Measured on Redis Enterprise (2026-08-28), with +/// times relative to the notification: the server-side endpoint moves at +8.6s, DNS follows at +9.7s and +4.4s +/// across two runs, and the sockets close at +18.4s and +16.6s - against a declared 15s grace and a 5s record +/// TTL. So resolving immediately returns the address being retired, in every run observed, and a +/// client that treats the first answer as authoritative hands off to the node it was told to leave. +/// +/// DNS is not guaranteed to win the race. On a second cluster the same notification with the same 15s +/// grace saw the socket close at +15.7s and DNS update at +18.7s - three seconds after the close. So +/// there are three outcomes here, not two, and the third is a normal one: the record may still be stale when +/// the window runs out, and the only thing left is to reconnect after the close and resolve then. Do not +/// "simplify" the null return away. +/// +/// +/// There is no way to observe the intermediate state from outside: the endpoint has moved server-side well +/// before DNS reflects it, and nothing tells a client which of those has happened. Probing until the answer +/// changes is the only mechanism available, and the short TTL is what makes it work - several attempts fit +/// inside the window. +/// +/// +/// The rule is "take any address that is not the one being retired". Note it is deliberately *not* "prefer an +/// address that has newly appeared", even though a MOVING implies one exists - it is emitted when the +/// connection's own proxy *leaves* the endpoint's address set **and** the set gains a member (established +/// across thirteen observations: a pure reduction takes the proxy away without announcing anything, and a pure +/// widening adds addresses while leaving the connection's proxy in place, which is equally silent). A live +/// sibling is at least as good a target as a newly joined node and is available *now*, whereas the new node +/// only becomes visible when the record updates - which can be after the socket has already closed. Preferring +/// the newcomer would mean waiting for it, and waiting is the thing to avoid. The one case where it would pay +/// is a rolling operation, where the sibling we step to may take its own turn later; that costs one further +/// handoff, bounded at one per connection per operation, which is cheaper than a lost window. +/// +/// +/// This rule is also doing more work than it looks. A Redis Cloud hostname usually carries several A records - measured 2026-08-28: 2 for +/// all-nodes, 3 for all-master-shards, 1 for single, all on a 5s TTL - and the count +/// follows actual proxy *placement* rather than the policy name, so a multi-proxy database whose shards happen +/// to share a node still resolves to one address. With several records the first resolution already names a +/// live sibling proxy, so this returns immediately and steps sideways rather than waiting: any proxy of the +/// same database serves the same data, so a sibling now beats the replacement in nine seconds. The poll only +/// engages when the record names nothing but the address being retired - the single-address case, which is +/// exactly where waiting is the only option available. +/// +/// +/// Deliberately free of jitter, clocks-by-configuration and connection state, so that it can be tested +/// exhaustively without a server: the caller supplies the resolver, the interval and the budget. Jitter +/// belongs at the call site, where the existing refresh jitter lives. +/// +/// +internal static class AdvertisedAddressProbe +{ + /// + /// Ordinary DNS, which is what everything but a test uses. + /// + internal static Task DefaultResolveAsync(string host, CancellationToken cancellationToken) => +#if NET + Dns.GetHostAddressesAsync(host, cancellationToken); +#else + Dns.GetHostAddressesAsync(host); +#endif + + /// + /// Polls DNS until it stops naming , or until the window runs out. + /// + /// + /// The replacement endpoint, or null if the window expired without DNS moving - a measured outcome + /// rather than a failure (see the type remarks: DNS has been seen updating three seconds after the socket + /// closed). The caller should then do nothing proactive: the server closes the socket, the reconnect that + /// follows re-resolves anyway because the endpoint is a hostname, and the relaxed timeout window covers + /// the gap. Guessing an address here would be strictly worse. + /// + internal static async Task ProbeAsync( + DnsEndPoint endpoint, + IPAddress retiring, + TimeSpan window, + TimeSpan pollInterval, + Func> resolve, + Action? log = null, + CancellationToken cancellationToken = default) + { + if (endpoint is null) throw new ArgumentNullException(nameof(endpoint)); + if (retiring is null) throw new ArgumentNullException(nameof(retiring)); + if (resolve is null) throw new ArgumentNullException(nameof(resolve)); + + // A non-positive window is not an error: a notification can arrive with nothing left of its budget + // (the shard notifications legitimately carry zero or negative times), and "act now" means one attempt + // rather than none. + var deadline = Environment.TickCount + (int)Math.Max(window.TotalMilliseconds, 0); + int attempt = 0; + + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + attempt++; + + IPAddress[]? addresses = null; + try + { + addresses = await resolve(endpoint.Host, cancellationToken).ForAwait(); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + // a DNS blip mid-handoff is exactly when we can least afford to give up; keep trying until + // the window says otherwise + log?.Invoke($"MOVING: resolve attempt {attempt} for {endpoint.Host} failed: {ex.Message}"); + } + + if (addresses is not null) + { + IPAddress? candidate = null; + bool retiringStillAdvertised = false; + foreach (var address in addresses) + { + if (address.Equals(retiring)) retiringStillAdvertised = true; + else candidate ??= address; + } + + if (candidate is not null) + { + // Two operationally different outcomes, worth distinguishing in a log somebody reads while + // debugging a handoff: we either stepped sideways to a proxy that was already there, or the + // record itself has moved on to the replacement. + log?.Invoke(retiringStillAdvertised + ? $"MOVING: {endpoint.Host} still advertises {retiring}; moving to sibling {candidate} (attempt {attempt})" + : $"MOVING: {endpoint.Host} now resolves to {candidate} after {attempt} attempt(s)"); + return new IPEndPoint(candidate, endpoint.Port); + } + + log?.Invoke($"MOVING: {endpoint.Host} still resolves only to {retiring} (attempt {attempt}); not yet updated"); + } + + var remaining = unchecked(deadline - Environment.TickCount); + if (remaining <= 0) + { + log?.Invoke($"MOVING: {endpoint.Host} never stopped resolving to {retiring} within the window"); + return null; + } + + // never sleep past the deadline: the last attempt should land inside the window, not after it + var delay = (int)Math.Min(pollInterval.TotalMilliseconds, remaining); + if (delay > 0) await Task.Delay(delay, cancellationToken).ForAwait(); + } + } + + /// + /// Whether the address we are connected on is still one of the addresses the hostname advertises. + /// + /// + /// true if still advertised, false if the record no longer names it, and null if DNS + /// could not be asked - which is *not* the same as "no": a resolution failure is no reason to give up a + /// working connection. + /// + /// + /// The trigger that needs no notification, and the reason it matters is a measured gap. On a multi-proxy + /// database, taking a node out for maintenance announced only the data-movement pair - MIGRATING at + /// +4.3s and MIGRATED at +16.6s, to every proxy - and then dropped the victim from DNS at +21.4s and + /// closed its socket *silently* at +34.7s, while sibling connections stayed up past +90s. No + /// MOVING was ever sent. + /// + /// So for thirteen seconds the condition was plainly visible to anybody who asked - our address is no + /// longer advertised - and the client's only other signal was the socket dying with no explanation. Asking + /// this question when a MIGRATED arrives converts that into a controlled handoff. Note + /// MIGRATED is also the notification the server *retains* and replays on connect, so a client that + /// arrives mid-operation gets the same prompt. + /// + /// + /// The same condition is what a support case turned on: a client kept dialling endpoints that no longer + /// existed for 37 hours, because nothing it could observe told it to stop. Two unrelated failures, one + /// detection rule. + /// + /// + internal static async Task IsStillAdvertisedAsync( + DnsEndPoint endpoint, + IPAddress current, + Func> resolve, + Action? log = null, + CancellationToken cancellationToken = default) + { + if (endpoint is null) throw new ArgumentNullException(nameof(endpoint)); + if (current is null) throw new ArgumentNullException(nameof(current)); + if (resolve is null) throw new ArgumentNullException(nameof(resolve)); + + IPAddress[] addresses; + try + { + addresses = await resolve(endpoint.Host, cancellationToken).ForAwait(); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + log?.Invoke($"{endpoint.Host}: cannot tell whether {current} is still advertised: {ex.Message}"); + return null; + } + + // an empty answer is "cannot tell" rather than "no": a record that momentarily resolves to nothing is + // a DNS problem, not an instruction to abandon a connection that is working + if (addresses is null || addresses.Length == 0) + { + log?.Invoke($"{endpoint.Host}: resolved to nothing; treating {current} as still advertised"); + return null; + } + + foreach (var address in addresses) + { + if (address.Equals(current)) return true; + } + + log?.Invoke($"{endpoint.Host}: no longer advertises {current}; it is being taken out of service"); + return false; + } +} diff --git a/src/StackExchange.Redis/Maintenance/MaintenanceHandoff.cs b/src/StackExchange.Redis/Maintenance/MaintenanceHandoff.cs new file mode 100644 index 000000000..38df5c310 --- /dev/null +++ b/src/StackExchange.Redis/Maintenance/MaintenanceHandoff.cs @@ -0,0 +1,133 @@ +using System; +using System.Net; +using System.Threading; +using System.Threading.Tasks; + +namespace StackExchange.Redis.Maintenance; + +/// +/// What to do about a MOVING notification. +/// +internal enum HandoffAction +{ + /// Nothing useful is available; let the server close the socket and reconnect then. + None, + + /// Drop our connections so they re-establish against the replacement address. + Recycle, + + /// The replacement is a different endpoint; the topology has to be re-read to find it. + Reconfigure, +} + +/// +/// The outcome of deciding how to hand off, and why - the reason is logged either way. +/// +internal readonly struct HandoffDecision(HandoffAction action, EndPoint? target, string reason) +{ + public HandoffAction Action { get; } = action; + + public EndPoint? Target { get; } = target; + + public string Reason { get; } = reason; + + public override string ToString() => Target is null ? $"{Action}: {Reason}" : $"{Action} -> {Target}: {Reason}"; +} + +/// +/// Decides what a MOVING notification means for this connection, and where to go. +/// +/// +/// Separated from the acting so that it can be tested exhaustively without a server: the caller supplies the +/// endpoint, the address it is currently on, the announced window and a resolver, and gets back an action. +/// +/// The dispatch turns on the *form* of the endpoint, which is the thing that decides whether a handoff is even +/// possible - and the answer corrects an earlier assumption that MOVING should reuse the endpoint +/// retirement path: +/// +/// +/// +/// A hostname endpoint with no named successor - the case every observed MOVING has been - keeps its +/// : only the address behind the name moves. Retiring the endpoint would delete our +/// only way of reaching the deployment. So the action is to recycle the connections once DNS has moved, which +/// re-resolves them. +/// +/// +/// An address endpoint with a named successor is genuinely a different endpoint, so the topology has to be +/// re-read. Never observed: eleven routes have all carried an explicit null, including cases where the server +/// had already chosen the replacement node. Treat this branch as code that must exist rather than code that is +/// exercised. +/// +/// +/// An address endpoint with no successor has nothing to re-resolve and nowhere named to go. Doing nothing is +/// correct: the socket closes, the reconnect happens, and the relaxed window covers it. +/// +/// +/// +internal static class MaintenanceHandoff +{ + internal static async Task DecideAsync( + EndPoint endpoint, + EndPoint? successor, + IPAddress? currentAddress, + TimeSpan window, + TimeSpan pollInterval, + Func> resolve, + Action? log = null, + CancellationToken cancellationToken = default) + { + if (successor is not null) + { + // Nothing is asserted about *which* endpoint: a successor may be an address we have never seen, so + // finding it is a topology question rather than a DNS one. + return new HandoffDecision(HandoffAction.Reconfigure, successor, "the server named a replacement endpoint"); + } + + if (endpoint is not DnsEndPoint dns) + { + return new HandoffDecision( + HandoffAction.None, + null, + $"{Format.ToString(endpoint)} is an address, not a name, and no replacement was named: nothing to re-resolve"); + } + + if (currentAddress is null) + { + // Without knowing where we are, "has it moved" is unanswerable - and guessing would mean recycling + // onto whatever DNS says right now, which for the first several seconds is the address being retired. + return new HandoffDecision(HandoffAction.None, null, "the address of the current connection is unknown"); + } + + var replacement = await AdvertisedAddressProbe.ProbeAsync( + dns, currentAddress, window, pollInterval, resolve, log, cancellationToken).ForAwait(); + + return replacement is null + ? new HandoffDecision( + HandoffAction.None, + null, + $"{dns.Host} still resolved only to {currentAddress} when the window expired") + : new HandoffDecision( + HandoffAction.Recycle, + replacement, + $"{dns.Host} now resolves to {replacement}"); + } + + /// + /// How long to wait before probing, so a fleet does not resolve in lockstep. + /// + /// + /// A *fraction* of the announced window rather than a fixed delay, which is the difference from the refresh + /// jitter elsewhere. Windows are not always generous - the shard notifications have been measured announcing + /// two seconds - so a flat one-second jitter could spend half of one, and on a short window the right amount + /// of jitter is almost none. Capped as well as scaled, because a long window does not justify a long wait + /// when DNS has been seen moving after four seconds. + /// + internal static TimeSpan GetJitter(TimeSpan window, Random random) + { + if (window <= TimeSpan.Zero) return TimeSpan.Zero; + + var tenth = window.TotalMilliseconds / 10; + var capped = Math.Min(tenth, 1000); + return TimeSpan.FromMilliseconds(random.Next(0, (int)Math.Max(capped, 1))); + } +} diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs index eac58bd01..a739e7df8 100644 --- a/src/StackExchange.Redis/PhysicalBridge.cs +++ b/src/StackExchange.Redis/PhysicalBridge.cs @@ -1794,6 +1794,33 @@ 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.Maintenance.cs b/src/StackExchange.Redis/PhysicalConnection.Maintenance.cs index 36ebefe09..03e74f875 100644 --- a/src/StackExchange.Redis/PhysicalConnection.Maintenance.cs +++ b/src/StackExchange.Redis/PhysicalConnection.Maintenance.cs @@ -130,7 +130,21 @@ private OutOfBandResult OnMaintenanceNotification(ConnectionMultiplexer muxer, P { if (IsWindowOpening(type)) { - server.OnMaintenanceWindowOpened(type, sequenceId, time); + var isNew = server.OnMaintenanceWindowOpened(type, sequenceId, time); + + // ...and MOVING alone means "this endpoint is going away", which is worth acting on rather than + // waiting to be disconnected. + // + // Only when the notification is *new*, and that is load-bearing rather than tidy. A server + // re-sends MOVING to a connection that opts in while the window is still open - measured, and + // the handoff replaces the connection, so acting on the repeat is a feedback loop: recycle, + // reconnect, get told again, recycle. It produced twelve recycles from one event on a real + // deployment before this guard. The per-server sequence dedup already knew it was a repeat; the + // handoff simply was not asking. + if (isNew && type == MaintenanceNotificationType.Moving) + { + server.OnMovingAnnounced(time, newEndPoint, this); + } } else if (IsWindowClosing(type)) { diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt index 77e76b4e8..4b87978f5 100644 --- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt @@ -175,3 +175,4 @@ override StackExchange.Redis.Configuration.DefaultOptionsProvider.ToString() -> [SER010]StackExchange.Redis.Maintenance.ClusterSlotMigration.Target.get -> System.Net.EndPoint? [SER010]StackExchange.Redis.Maintenance.PushMaintenanceEvent.SlotMigrations.get -> System.Collections.Generic.IReadOnlyList! [SER007]StackExchange.Redis.Availability.HealthCheckContext.ProbeFlags.get -> StackExchange.Redis.CommandFlags +StackExchange.Redis.ConnectionFailureType.MaintenanceHandoff = 12 -> StackExchange.Redis.ConnectionFailureType diff --git a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs index ad71d28d5..28f00e1b0 100644 --- a/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs +++ b/src/StackExchange.Redis/ServerEndPoint.Maintenance.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Diagnostics.CodeAnalysis; +using System.Net; using System.Threading; using System.Threading.Tasks; using RESPite; @@ -160,9 +161,13 @@ private int GetRelaxedRemaining() /// /// An announced disruption has started (or is still running): open or extend the relaxed window. /// - internal void OnMaintenanceWindowOpened(MaintenanceNotificationType type, long? sequenceId, TimeSpan? time) + /// + /// Whether this notification was new. A replay extends nothing and, importantly, must not be *acted* on - + /// see the handoff, where re-acting on a repeat is a feedback loop rather than merely wasted work. + /// + internal bool OnMaintenanceWindowOpened(MaintenanceNotificationType type, long? sequenceId, TimeSpan? time) { - if (!TryClaimSequenceId(type, sequenceId)) return; + if (!TryClaimSequenceId(type, sequenceId)) return false; var config = Multiplexer.RawConfig; var floor = config.MaintenanceRelaxedTimeout; @@ -176,6 +181,7 @@ internal void OnMaintenanceWindowOpened(MaintenanceNotificationType type, long? Volatile.Write(ref _relaxedType, (int)type); ExtendRelaxedWindow(duration, $"{type} for {duration.TotalSeconds}s"); + return true; } /// @@ -204,6 +210,145 @@ internal void OnMaintenanceWindowClosed(MaintenanceNotificationType type, long? Multiplexer.Trace($"{type}: relaxation continues for {tail.TotalSeconds}s (post-event)", ToString()); } + private int _handoffInFlight, _handoffRecycles; + private volatile string? _lastHandoffOutcome; + + /// + /// What the last MOVING handoff decided, and why. + /// + /// + /// Recorded rather than only traced because Multiplexer.Trace is [Conditional("VERBOSE")] - it + /// compiles away in any normal build, so a handoff that misbehaves in production would leave no trace at + /// all. This is the minimum that survives: what was decided, readable afterwards. + /// + internal string? LastHandoffOutcome => _lastHandoffOutcome; + + /// How many times a handoff has replaced this server's connections. + internal int HandoffRecycles => Volatile.Read(ref _handoffRecycles); + + /// + /// Acts on a MOVING: find where to go, then get off this connection before we are pushed. + /// + /// + /// The value here is entirely in the *timing*. Without it we already survive a MOVING - the socket + /// closes and we reconnect - but the reconnect happens when the server decides, and re-resolves to whatever + /// DNS says at that moment, which has been measured as still naming the node being retired. Measured on a + /// real deployment: MOVING arrives about six seconds in, DNS moves somewhere between four and + /// nineteen seconds later, and the socket closes seventeen to nineteen seconds after the notification. So + /// the window exists to be *used*, and using it means waiting for DNS to move and then choosing the moment. + /// + /// Fire-and-forget by design: this runs while the connection it concerns is still serving commands, and the + /// notification is delivered on the read loop, which must not wait for a DNS poll. + /// + /// + 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. + if (Interlocked.CompareExchange(ref _handoffInFlight, 1, 0) != 0) + { + Multiplexer.Trace("MOVING: a handoff is already in flight", ToString()); + return; + } + + var budget = window is { } value && value > TimeSpan.Zero + ? value + : Multiplexer.RawConfig.MaintenanceRelaxedTimeout; // no window given: use the relaxation floor + var current = (connection.VolatileSocket?.RemoteEndPoint as IPEndPoint)?.Address; + + _ = HandoffAsync(budget, successor, current); + } + + private async Task HandoffAsync(TimeSpan window, EndPoint? successor, IPAddress? currentAddress) + { + try + { + // Spread the fleet, but only by a fraction of the window - see MaintenanceHandoff.GetJitter for why + // a flat delay is wrong here. + var jitter = MaintenanceHandoff.GetJitter(window, RandomFor(this)); + if (jitter > TimeSpan.Zero) await Task.Delay(jitter).ForAwait(); + + var remaining = window - jitter; + var decision = await MaintenanceHandoff.DecideAsync( + EndPoint, + successor, + currentAddress, + 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(); + + Multiplexer.Trace($"MOVING: {decision}", ToString()); + _lastHandoffOutcome = decision.ToString(); + switch (decision.Action) + { + case HandoffAction.Recycle: + await DrainThenRecycleAsync(remaining, decision.Reason).ForAwait(); + break; + case HandoffAction.Reconfigure: + // A named successor is a different endpoint, so this is a topology question. Note no + // observed deployment has ever named one, so this path has never run outside a test. + Multiplexer.ReconfigureIfNeeded(EndPoint, fromBroadcast: false, "moving names a successor"); + await DrainThenRecycleAsync(remaining, decision.Reason).ForAwait(); + break; + default: + // Nothing to do is a legitimate outcome, not a failure: the server closes the socket, the + // reconnect re-resolves, and the relaxed window covers the gap. + break; + } + } + catch (Exception ex) + { + Multiplexer.OnInternalError(ex, EndPoint); + } + finally + { + Volatile.Write(ref _handoffInFlight, 0); + } + } + + /// + /// Lets in-flight caller work 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. + /// + /// 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. + /// + /// + 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); + if (recycled) Interlocked.Increment(ref _handoffRecycles); + Multiplexer.Trace( + $"MOVING: {(recycled ? "recycled" : "nothing to recycle")} after {watch.ElapsedMilliseconds}ms" + + (drained ? " (drained)" : " (still busy; the window ran out)"), + ToString()); + } + + [ThreadStatic] + private static Random? _random; + + /// + /// A per-thread , seeded so that two processes handing off at once do not pick the same + /// jitter. + /// + private static Random RandomFor(ServerEndPoint server) => + _random ??= new Random(Environment.TickCount ^ server.GetHashCode()); + private void ExtendRelaxedWindow(TimeSpan duration, string cause) { var candidate = NudgeFromZero(unchecked(Environment.TickCount + (int)duration.TotalMilliseconds)); diff --git a/src/StackExchange.Redis/StackExchange.Redis.csproj b/src/StackExchange.Redis/StackExchange.Redis.csproj index d0e16977d..e799f510a 100644 --- a/src/StackExchange.Redis/StackExchange.Redis.csproj +++ b/src/StackExchange.Redis/StackExchange.Redis.csproj @@ -60,6 +60,7 @@ + diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/ClusterFamilyScenarioTests.cs b/tests/StackExchange.Redis.FaultInjector.Tests/ClusterFamilyScenarioTests.cs new file mode 100644 index 000000000..b904632e2 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/ClusterFamilyScenarioTests.cs @@ -0,0 +1,218 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using StackExchange.Redis.Maintenance; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// The OSS cluster notification family against a real cluster: SMIGRATING, SMIGRATED, and what +/// the client is supposed to do about them. +/// +/// +/// Every /slot-migrate trigger requires oss_cluster: true with all-master-shards, so this +/// scenario family is the cluster half of the feature - the half whose reactions (re-reading the slot map, +/// re-subscribing stranded sharded channels) have until now only been exercised against the in-process fake. +/// +[Trait("tier", "fault-injector")] +[Trait("scenario", "cluster-family")] +public class ClusterFamilyScenarioTests(ExistingDatabaseFixture fixture, ITestOutputHelper log) + : IClassFixture +{ + /// + /// Slot migrations, in the four shapes the injector can produce. + /// + /// + /// The effects differ in what happens to the *endpoint list*, which is the thing a client has to keep up + /// with: remove-add retires a node and introduces one, remove only retires, add only + /// introduces, and slot-shuffle moves slots between nodes that both stay. So they cover the four + /// ways a slot map can go stale, and only one of them (shuffle) leaves the endpoint set alone. + /// + /// + /// Only remove here. The other effects need a database the setup leg will not provision, and get + /// their own fixture below rather than skipping: add and slot-shuffle need a node holding + /// several shards, and remove-add is not in /slot-migrate/setup's effect enum at all. + /// + [Theory] + [InlineData("remove", "migrate")] + public async Task SlotMigrationIsAnnouncedAndActedOn(string effect, string trigger) + { + fixture.RequireAvailable(); + var cancellationToken = TestContext.Current.CancellationToken; + + await using var scenario = await ScenarioRun.SetupAsync( + fixture.Injector, "slot-migrate", effect, trigger, log.WriteLine, + setupTrigger: "reshard", cancellationToken: cancellationToken); + + var database = scenario.Database; + Assert.NotNull(database); + ScenarioSupport.RequireEffectIsAchievable(scenario, effect); + + var clock = Stopwatch.StartNew(); + var events = new List(); + + await using var conn = await ConnectionMultiplexer.ConnectAsync(database.GetClientConfig(fixture.Environment)); + conn.ServerMaintenanceEvent += (_, e) => + { + if (e is PushMaintenanceEvent push) + { + lock (events) events.Add(push); + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s {push.NotificationType} seq={push.SequenceId} {push.RawMessage}"); + foreach (var migration in push.SlotMigrations) + { + log.WriteLine($" slots {migration.RawSlots}: {migration.Source} -> {migration.Target}"); + } + } + }; + + var endpointsBefore = conn.GetEndPoints().Length; + log.WriteLine($"connected as {conn.GetServer(conn.GetEndPoints()[0]).ServerType} across {endpointsBefore} endpoint(s)"); + + // keys spread across slots, so a migration of any shard moves something we are actually using + var db = conn.GetDatabase(); + var keys = Enumerable.Range(0, 32).Select(i => (RedisKey)$"fi-{effect}-{i}").ToArray(); + foreach (var key in keys) await db.StringSetAsync(key, "before"); + + clock.Restart(); + await ScenarioSupport.FireOrSkipAsync(scenario, effect, cancellationToken); + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s injector reports finished"); + + // Keep using the connection while things move: a slot map that has gone stale shows up as MOVED + // redirections, and the point of the feature is that we learn the new one rather than eating them. + var deadline = clock.Elapsed + TimeSpan.FromSeconds(45); + int reads = 0, failures = 0; + while (clock.Elapsed < deadline) + { + foreach (var key in keys) + { + try + { + await db.StringGetAsync(key); + reads++; + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + failures++; + if (failures <= 3) log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s read failed: {ex.Message}"); + } + } + + await Task.Delay(500, cancellationToken); + } + + lock (events) + { + log.WriteLine($" {events.Count} notification(s); {reads} reads, {failures} failures; " + + $"endpoints {endpointsBefore} -> {conn.GetEndPoints().Length}"); + + // A slot migration is announced to every proxy, so any connection should see it. What is asserted is + // only that we were told and understood it - the *reaction* is asserted separately below, because a + // migration that moves no slot we hold would legitimately change nothing here. + Assert.NotEmpty(events); + Assert.All(events, e => Assert.NotEqual(MaintenanceNotificationType.None, e.NotificationType)); + Assert.Contains(events, e => e.NotificationType is MaintenanceNotificationType.SlotMigrating or MaintenanceNotificationType.SlotMigrated); + + // Any SMIGRATED that carried a payload must have parsed into something usable; an empty list would + // mean we saw the frame and threw the contents away. + foreach (var migrated in events.Where(e => e.NotificationType == MaintenanceNotificationType.SlotMigrated)) + { + if (!string.IsNullOrEmpty(migrated.Payload) || migrated.SlotMigrations.Count > 0) + { + Assert.NotEmpty(migrated.SlotMigrations); + Assert.All(migrated.SlotMigrations, m => Assert.NotNull(m.Source)); + } + } + } + + // and the client is still serving keys across the new topology + Assert.True(reads > 0, "no reads succeeded at all during the migration"); + foreach (var key in keys) Assert.Equal("before", await db.StringGetAsync(key)); + } + + [Fact] + public async Task ShardedSubscriptionSurvivesASlotMigration() + { + // D5's resubscription, against a real cluster. A sharded channel is bound to the node owning its slot, + // so a migration strands the subscription: the server sends an unsolicited SUNSUBSCRIBE and stops + // delivering. Recovering that is the client's job, and until now only the fake has tested it. + fixture.RequireAvailable(); + var cancellationToken = TestContext.Current.CancellationToken; + + await using var scenario = await ScenarioRun.SetupAsync( + fixture.Injector, "slot-migrate", "remove", "migrate", log.WriteLine, + setupTrigger: "reshard", cancellationToken: cancellationToken); + Assert.NotNull(scenario.Database); + + await using var conn = await ConnectionMultiplexer.ConnectAsync(scenario.Database.GetClientConfig(fixture.Environment)); + var subscriber = conn.GetSubscriber(); + + // several channels, because the migration moves the slots of one node: with one channel the test would + // usually be asserting that nothing happened to it + var channels = Enumerable.Range(0, 16).Select(i => RedisChannel.Sharded($"fi-shard-{i}")).ToArray(); + var received = new int[channels.Length]; + for (int i = 0; i < channels.Length; i++) + { + var index = i; + await subscriber.SubscribeAsync(channels[i], (_, _) => Interlocked.Increment(ref received[index])); + } + + Assert.True(await DeliversEverywhereAsync(subscriber, channels, received, cancellationToken), + "every sharded channel should deliver before the migration, or the test proves nothing"); + log.WriteLine($"all {channels.Length} sharded channels delivering before the migration"); + + await ScenarioSupport.FireOrSkipAsync(scenario, "remove", cancellationToken); + + // The recovery is deliberately allowed to be slow: it is driven by a jittered topology refresh and, as a + // fallback, by a delayed resubscription sweep. What matters is that it happens without the caller doing + // anything, not that it is instant. + var recovered = await DeliversEverywhereAsync(subscriber, channels, received, cancellationToken, timeoutSeconds: 90); + log.WriteLine($"delivery after migration: {(recovered ? "all channels" : "INCOMPLETE")}"); + Assert.True(recovered, "sharded subscriptions should deliver again once the topology settles"); + } + + /// + /// Publishes to every channel and waits until each one has delivered at least one more message. + /// + /// + /// Publishing repeatedly rather than once: the publish itself routes by slot, so during a migration an + /// individual publish can land on a node that no longer owns the slot. Retrying is what a real caller would + /// do, and it keeps the test measuring *delivery* rather than the timing of one message. + /// + private async Task DeliversEverywhereAsync( + ISubscriber subscriber, + RedisChannel[] channels, + int[] received, + CancellationToken cancellationToken, + int timeoutSeconds = 20) + { + var baseline = channels.Select((_, i) => Volatile.Read(ref received[i])).ToArray(); + var deadline = DateTime.UtcNow.AddSeconds(timeoutSeconds); + + while (DateTime.UtcNow < deadline) + { + for (int i = 0; i < channels.Length; i++) + { + if (Volatile.Read(ref received[i]) > baseline[i]) continue; + try + { + await subscriber.PublishAsync(channels[i], "ping"); + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + // expected mid-migration; the retry is the point + } + } + + await Task.Delay(500, cancellationToken); + if (channels.Select((_, i) => Volatile.Read(ref received[i]) > baseline[i]).All(x => x)) return true; + } + + var missing = channels.Where((_, i) => Volatile.Read(ref received[i]) <= baseline[i]).Select(c => c.ToString()); + log.WriteLine($" not delivering: {string.Join(", ", missing)}"); + return false; + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/DenseClusterScenarioTests.cs b/tests/StackExchange.Redis.FaultInjector.Tests/DenseClusterScenarioTests.cs new file mode 100644 index 000000000..2146492a5 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/DenseClusterScenarioTests.cs @@ -0,0 +1,109 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading.Tasks; +using StackExchange.Redis.Maintenance; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// An OSS cluster with packed shards, so add and slot-shuffle have something to move. +/// +public sealed class DenseClusterFixture() : FaultInjectorFixture(DatabaseShape.OssClusterDense); + +/// +/// The slot migrations the scenario setup legs cannot provision for. +/// +/// +/// These looked like an environment limitation - the injector refuses them with "No node with multiple shards +/// found" - and they are not: the setup leg provisions one shard per node, so there is never a node to take a +/// shard *from*. Six shards with dense placement over three nodes gives two per node, and both effects +/// then run. The lesson generalises: a scenario that setup cannot arrange is still reachable by provisioning +/// the database ourselves and driving the run leg by bdb_id. +/// +/// No scenario teardown here, deliberately. Teardown deletes the database, and this database belongs to the +/// fixture, which deletes it itself; and migrate excludes no nodes, so there is nothing to restore. +/// +/// +[Trait("tier", "fault-injector")] +[Trait("scenario", "cluster-family")] +public class DenseClusterScenarioTests(DenseClusterFixture fixture, ITestOutputHelper log) + : IClassFixture +{ + [Theory] + [InlineData("add")] + [InlineData("slot-shuffle")] + [InlineData("remove-add")] + public async Task SlotMigrationIsAnnouncedAndActedOn(string effect) + { + fixture.RequireAvailable(); + var cancellationToken = TestContext.Current.CancellationToken; + var database = fixture.Database; + Assert.NotNull(database); + log.WriteLine($"{effect} against {database}"); + + var clock = Stopwatch.StartNew(); + var events = new List(); + + await using var conn = await ConnectionMultiplexer.ConnectAsync(database.GetClientConfig()); + conn.ServerMaintenanceEvent += (_, e) => + { + if (e is PushMaintenanceEvent push) + { + lock (events) events.Add(push); + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s {push.NotificationType} seq={push.SequenceId} {push.RawMessage}"); + foreach (var migration in push.SlotMigrations) + { + log.WriteLine($" slots {migration.RawSlots}: {migration.Source} -> {migration.Target}"); + } + } + }; + + var db = conn.GetDatabase(); + var keys = Enumerable.Range(0, 32).Select(i => (RedisKey)$"fi-dense-{effect}-{i}").ToArray(); + foreach (var key in keys) await db.StringSetAsync(key, "before"); + + clock.Restart(); + var query = new Dictionary + { + ["effect"] = effect, + ["trigger"] = "migrate", + ["bdb_id"] = database.BdbId.ToString(), + }; + await fixture.Injector.PostScenarioAsync("slot-migrate", leg: null, query, cancellationToken: cancellationToken); + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s injector reports finished"); + + int reads = 0, failures = 0; + var deadline = clock.Elapsed + TimeSpan.FromSeconds(20); + while (clock.Elapsed < deadline) + { + foreach (var key in keys) + { + try + { + await db.StringGetAsync(key); + reads++; + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + failures++; + } + } + + await Task.Delay(500, cancellationToken); + } + + lock (events) + { + log.WriteLine($" {events.Count} notification(s); {reads} reads, {failures} failures"); + Assert.NotEmpty(events); + Assert.Contains(events, e => e.NotificationType is MaintenanceNotificationType.SlotMigrating + or MaintenanceNotificationType.SlotMigrated); + } + + Assert.True(reads > 0, "no reads succeeded during the migration"); + foreach (var key in keys) Assert.Equal("before", await db.StringGetAsync(key)); + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/Environment/CertificateSanity.cs b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/CertificateSanity.cs new file mode 100644 index 000000000..de33a1d44 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/CertificateSanity.cs @@ -0,0 +1,94 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.IO; +using System.Security.Cryptography.X509Certificates; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// Checks that the environment's TLS material belongs to the cluster we are about to talk to. +/// +/// +/// Environments get re-provisioned, and the certificates do not always follow. Observed 2026-09-01: the folder +/// held a server certificate for *.marcgravell-test-46be1d08... while the live cluster was +/// marcgravell-test-e21cd75d... - a leftover from a previous provision, three days older than the +/// env_output.json beside it. +/// +/// Without this check, that presents as a TLS handshake failure, which reads like a client bug and is not one: +/// the certificate genuinely does not cover the name being dialled, and refusing it is correct +/// (TrustIssuer tolerates chain errors only, so a name mismatch fails - as it should). Detecting it here +/// turns half an hour of certificate archaeology into a skip that names both clusters, and it costs nothing +/// because it happens before a database is provisioned. +/// +/// +internal static class CertificateSanity +{ + /// + /// Skips when the environment's certificates were issued for a different cluster. + /// + public static void RequireCertificatesMatchThisCluster(FaultInjectorEnvironment environment, Action log) + { + var clusterName = environment.Cluster?.ClusterName; + if (clusterName is null) return; // nothing to compare against; let the connect speak for itself + + // the server certificate the environment generated, if it left one behind + var leafPath = Path.Combine(environment.ConfigDirectory.FullName, "redis.crt"); + if (!File.Exists(leafPath)) return; + + var leaf = X509CertificateLoader.LoadCertificateFromFile(leafPath); + var names = ReadDnsNames(leaf); + log($"environment server certificate covers [{string.Join(", ", names)}]; cluster is '{clusterName}'"); + + if (!names.Any(name => CoversClusterEndpoints(name, clusterName))) + { + Assert.Skip( + $"the environment's TLS certificates cover [{string.Join(", ", names)}] but this cluster is " + + $"'{clusterName}' - they are left over from an earlier provision, so a TLS test here would " + + "only be measuring the mismatch. Re-provision with certificate generation enabled to run it."); + } + } + + /// + /// Every DNS name a certificate carries, not just the first. + /// + /// + /// GetNameInfo returns one name, which is not enough: these certificates carry both a wildcard and + /// the bare cluster name, and which one comes back decides the answer. Reading the SAN extension properly is + /// the difference between a check that works and one that skips a perfectly good environment - as the first + /// version of this did. + /// + private static List ReadDnsNames(X509Certificate2 certificate) + { + foreach (var extension in certificate.Extensions) + { + if (extension is X509SubjectAlternativeNameExtension san) + { + return [.. san.EnumerateDnsNames()]; + } + } + + var subject = certificate.GetNameInfo(X509NameType.DnsName, forIssuer: false); + return subject is null ? [] : [subject]; + } + + /// + /// Whether a certificate name covers the endpoints of a given cluster. + /// + /// + /// The hosts actually dialled are *database* endpoints - redis-13500.<cluster> - so a wildcard + /// whose parent is the cluster name is exactly right, even though (correctly) it does not match the bare + /// cluster name itself. An exact SAN for the cluster name also counts, since that is what the REST API is + /// reached by. + /// + private static bool CoversClusterEndpoints(string certificateName, string clusterName) + { + if (certificateName.StartsWith("*.", StringComparison.Ordinal)) + { + return string.Equals(certificateName[2..], clusterName, StringComparison.OrdinalIgnoreCase); + } + + return string.Equals(certificateName, clusterName, StringComparison.OrdinalIgnoreCase); + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ClusterRestClient.cs b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ClusterRestClient.cs new file mode 100644 index 000000000..c9565bb79 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ClusterRestClient.cs @@ -0,0 +1,71 @@ +using System; +using System.Collections.Generic; +using System.Net.Http; +using System.Net.Http.Headers; +using System.Security.Cryptography.X509Certificates; +using System.Text; +using System.Text.Json; +using System.Threading.Tasks; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// The cluster's own REST API on port 9443, for the few facts the fault injector does not expose. +/// +/// +/// Reads only. Anything that *changes* state goes through the injector so it is recorded as a job with an +/// action id - which is also how the console works, and it means a scenario can be reconstructed afterwards +/// from the injector's history rather than from somebody's memory. +/// +public sealed class ClusterRestClient : IDisposable +{ + private readonly HttpClient _http; + + public ClusterRestClient(FaultInjectorEnvironment.ClusterCredentials credentials, string? certificateAuthorityPath) + { + var handler = new HttpClientHandler(); + + if (certificateAuthorityPath is not null) + { + // The cluster's management certificate is self-signed per environment, like the proxy ones. Pin to + // the CA in the config directory rather than accepting anything: this channel carries credentials. + var issuer = X509CertificateLoader.LoadCertificateFromFile(certificateAuthorityPath); + handler.ServerCertificateCustomValidationCallback = (_, certificate, chain, _) => + { + if (certificate is null || chain is null) return false; + chain.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust; + chain.ChainPolicy.CustomTrustStore.Add(issuer); + chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck; + return chain.Build(certificate); + }; + } + + _http = new HttpClient(handler) { BaseAddress = credentials.RestUrl, Timeout = TimeSpan.FromSeconds(30) }; + var basic = Convert.ToBase64String(Encoding.UTF8.GetBytes($"{credentials.Username}:{credentials.Password}")); + _http.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Basic", basic); + } + + public void Dispose() => _http.Dispose(); + + /// + /// Every database on the cluster, as (bdb_id, name). + /// + public async Task> ListDatabasesAsync() + { + using var response = await _http.GetAsync("/v1/bdbs?fields=uid,name"); + response.EnsureSuccessStatusCode(); + + using var document = JsonDocument.Parse(await response.Content.ReadAsStringAsync()); + var results = new List<(int, string)>(); + foreach (var bdb in document.RootElement.EnumerateArray()) + { + if (bdb.TryGetProperty("uid", out var uid) && bdb.TryGetProperty("name", out var name) + && uid.TryGetInt32(out var id) && name.GetString() is { } text) + { + results.Add((id, text)); + } + } + + return results; + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/Environment/DatabaseShape.cs b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/DatabaseShape.cs new file mode 100644 index 000000000..b9778e213 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/DatabaseShape.cs @@ -0,0 +1,129 @@ +using System; +using System.Collections.Generic; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// The database shapes this suite provisions, described declaratively. +/// +/// +/// Shape is the axis that matters, and it is not cosmetic: measurements on 2026-08-28 showed the number of A +/// records a hostname carries follows actual proxy *placement*, giving 2 for all-nodes, 3 for +/// all-master-shards and 1 for single - and the handoff takes a different branch depending on +/// whether a live sibling address exists. So the shapes below are the reason to provision databases from inside +/// the tests at all: they turn that into a matrix axis instead of a manual sweep. +/// +public sealed record DatabaseShape( + string Label, + bool OssCluster = false, + string? ProxyPolicy = null, + bool Tls = false, + bool Replication = false, + int ShardCount = 1, + string? EndpointType = null, + string ShardsPlacement = "sparse", + string IpType = "external") +{ + /// A proxied standalone database: the shape MOVING actually fires on. + public static readonly DatabaseShape ProxiedStandalone = new("proxied-standalone", ProxyPolicy: "single"); + + /// Multiple proxies, so the hostname carries several A records and a sibling always exists. + public static readonly DatabaseShape MultiProxy = new("multi-proxy", ProxyPolicy: "all-master-shards", ShardCount: 2); + + /// OSS cluster API: the family that emits SMIGRATING/SMIGRATED instead. + public static readonly DatabaseShape OssClusterApi = new("oss-cluster", OssCluster: true, ProxyPolicy: "all-master-shards", ShardCount: 2); + + /// + /// An OSS cluster whose shards are packed, so some node holds more than one. + /// + /// + /// Exists to reach the add and slot-shuffle migrations, which need a node holding several + /// shards to have something to move. The scenario setup legs cannot produce that - they provision one shard + /// per node - so those effects looked like an environment limitation until this was tried: six shards with + /// dense placement over three nodes gives two per node, and both effects then run. + /// + /// Note it also demonstrates the placement-versus-policy point from the other direction: packed masters mean + /// all-master-shards advertises a *single* address, because the count follows placement. + /// + /// + public static readonly DatabaseShape OssClusterDense = new( + "oss-cluster-dense", OssCluster: true, ProxyPolicy: "all-master-shards", ShardCount: 6, ShardsPlacement: "dense"); + + /// + /// TLS with hostname-advertised endpoints - the documented coverage gap. + /// + /// + /// endpoint_type: ip is the interesting counterpart: with addresses advertised, a verifying client + /// cannot check identity on the targets it is told to use, because the proxy certificate carries DNS names + /// and no IP SAN. That pairing is the thing no in-process harness can honestly reproduce. + /// + public static readonly DatabaseShape TlsWithHostnames = new("tls-hostnames", OssCluster: true, Tls: true, ShardCount: 2, EndpointType: "hostname"); + + /// + /// The create_database parameters for this shape. + /// + /// + /// Goes inside a database_config wrapper when it reaches the injector - a flat payload is rejected + /// with "Invalid parameter 'database_config': got None". + /// + /// Built as data rather than as a typed record, because these are the injector's wire names and its schema + /// declares parameters as untyped. The names here are no longer guesses: they match the environment's + /// own bdb_config.json and the dbconfig the injector itself publishes as a trigger requirement + /// (GET /topology-change-standalone?effect=...), which is the authoritative list. + /// + /// + public Dictionary ToCreateParameters(string name, int port) + { + var parameters = new Dictionary + { + ["name"] = name, + ["port"] = port, + ["memory_size"] = 134_217_728, // 128MB, matching what the injector asks for in its own requirements + ["eviction_policy"] = "volatile-lru", + ["replication"] = Replication, + ["sharding"] = ShardCount > 1, + ["shards_count"] = ShardCount, + ["shards_placement"] = ShardsPlacement, + ["oss_cluster"] = OssCluster, + }; + + if (ShardCount > 1) + { + // Required whenever sharding is on: Redis Enterprise rejects the database outright with + // "Invalid sharding configuration: missing shard_key_regex". These two patterns are the standard + // pair - an explicit hash tag if present, otherwise the whole key - and are what the environment's + // own bdb_config.json uses. + parameters["shard_key_regex"] = new[] + { + new Dictionary { ["regex"] = ".*\\{(?.*)\\}.*" }, + new Dictionary { ["regex"] = "(?.*)" }, + }; + } + + // only when asked for: a database with no tls_mode serves plaintext, and setting it here without + // certificates in place produces a database nothing can connect to + if (Tls) parameters["tls_mode"] = "enabled"; + + if (OssCluster) + { + // Which addresses CLUSTER SLOTS advertises, and it is not cosmetic: the default is *internal*, so a + // cluster database created without this advertises VPC-private addresses. A client outside the VPC + // then connects to the seed, discovers nodes at 10.x, and cannot reach any of them - which is + // exactly how the first dense-cluster run failed. Distinct from EndpointType (ip versus hostname), + // which is about identity rather than routing. + parameters["oss_cluster_api_preferred_ip_type"] = IpType; + } + + if (OssCluster && EndpointType is not null) + { + // Two distinct axes, easily conflated. "endpoint_type" is ip-versus-hostname - what CLUSTER SLOTS + // advertises, and therefore whether a verifying TLS client can check identity on the targets it is + // given. "ip_type" is internal-versus-external, which is about routing rather than identity. + parameters["oss_cluster_api_preferred_endpoint_type"] = EndpointType; + } + + if (ProxyPolicy is not null) parameters["proxy_policy"] = ProxyPolicy; + + return parameters; + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ExistingDatabase.cs b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ExistingDatabase.cs new file mode 100644 index 000000000..3040ea891 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ExistingDatabase.cs @@ -0,0 +1,151 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Text.Json; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// A database the environment already provisioned, read from endpoints.json. +/// +/// +/// The counterpart to creating our own. Provisioning gives control over the shape, which is what the matrix +/// needs; this gives a run against whatever the environment template made, which is what you want when the +/// question is "does any of this work at all" rather than "does it work for shape X". +/// +/// Note endpoints.json carries more than expected: each entry's raw_endpoints includes +/// proxy_policy, oss_cluster_api_preferred_endpoint_type and the address list behind the DNS +/// name. So the facts that decide client behaviour are mostly here, and the cluster REST API is only needed for +/// the explicit oss_cluster flag and the client-certificate settings. +/// +/// +public sealed record ExistingDatabase( + string Key, + int BdbId, + string Host, + int Port, + bool Tls, + string? Username, + string? Password, + string? ProxyPolicy, + string? EndpointType, + IReadOnlyList Addresses) +{ + /// + /// How many addresses the hostname is expected to resolve to. + /// + /// + /// The measured driver of handoff behaviour: with more than one address a live sibling always exists, so a + /// handoff steps sideways immediately; with one, it has to wait for the record to move. Tests that care + /// should assert on this rather than on the policy name, because the count follows actual proxy + /// *placement* - an all-master-shards database whose shards share a node advertises one address. + /// + public int AdvertisedAddressCount => Addresses.Count; + + public override string ToString() => $"{Key} ({Host}:{Port}, bdb {BdbId}, {ProxyPolicy ?? "?"}, {AdvertisedAddressCount} addr)"; + + /// + /// Reads every database in the environment's endpoints.json, keyed as that file keys them. + /// + public static Dictionary ReadAll(FaultInjectorEnvironment environment) + { + var results = new Dictionary(StringComparer.OrdinalIgnoreCase); + var path = Path.Combine(environment.ConfigDirectory.FullName, "endpoints.json"); + if (!File.Exists(path)) return results; + + using var document = JsonDocument.Parse(File.ReadAllText(path)); + foreach (var entry in document.RootElement.EnumerateObject()) + { + if (entry.Value.ValueKind != JsonValueKind.Object) continue; + if (TryRead(entry.Name, entry.Value, out var database)) results[entry.Name] = database; + } + + return results; + } + + private static bool TryRead(string key, JsonElement element, out ExistingDatabase database) + { + database = null!; + if (!element.TryGetProperty("bdb_id", out var bdbId) || !bdbId.TryGetInt32(out var id)) return false; + + // "endpoints" holds "host:port" or a redis:// URI depending on how the environment was templated + string? host = null; + int port = 0; + if (element.TryGetProperty("endpoints", out var endpoints) && endpoints.ValueKind == JsonValueKind.Array + && endpoints.GetArrayLength() > 0 && endpoints[0].GetString() is { } first) + { + var text = first; + var scheme = text.IndexOf("://", StringComparison.Ordinal); + if (scheme >= 0) text = text[(scheme + 3)..]; + var colon = text.LastIndexOf(':'); + if (colon > 0 && int.TryParse(text[(colon + 1)..], out port)) host = text[..colon]; + } + + string? proxyPolicy = null, endpointType = null; + var addresses = new List(); + if (element.TryGetProperty("raw_endpoints", out var raw) && raw.ValueKind == JsonValueKind.Array + && raw.GetArrayLength() > 0) + { + var head = raw[0]; + proxyPolicy = ReadString(head, "proxy_policy"); + endpointType = ReadString(head, "oss_cluster_api_preferred_endpoint_type"); + host ??= ReadString(head, "dns_name"); + if (port == 0 && head.TryGetProperty("port", out var rawPort)) rawPort.TryGetInt32(out port); + + if (head.TryGetProperty("addr", out var addr) && addr.ValueKind == JsonValueKind.Array) + { + foreach (var item in addr.EnumerateArray()) + { + if (item.GetString() is { } address) addresses.Add(address); + } + } + } + + if (host is null || port == 0) return false; + + database = new ExistingDatabase( + key, + id, + host, + port, + element.TryGetProperty("tls", out var tls) && tls.ValueKind == JsonValueKind.True, + ReadString(element, "username"), + ReadString(element, "password"), + proxyPolicy, + endpointType, + addresses); + return true; + } + + private static string? ReadString(JsonElement element, string name) + => element.TryGetProperty(name, out var value) && value.ValueKind == JsonValueKind.String ? value.GetString() : null; + + /// + /// Connection options for this database, with maintenance notifications required. + /// + public ConfigurationOptions GetClientConfig(FaultInjectorEnvironment environment, MaintenanceNotificationMode mode = MaintenanceNotificationMode.Enabled) + { + var options = new ConfigurationOptions + { + EndPoints = { { Host, Port } }, + User = string.Equals(Username, "default", StringComparison.OrdinalIgnoreCase) ? null : Username, + Password = Password, + Protocol = RedisProtocol.Resp3, + MaintenanceNotifications = mode, + AbortOnConnectFail = false, + ConnectTimeout = 15_000, + SyncTimeout = 15_000, + }; + + if (Tls) + { + options.Ssl = true; + options.SslHost = Host; + var caPath = environment.CertificateAuthorityPath + ?? throw new InvalidOperationException($"{Key} uses TLS but no CA certificate was found in {environment.ConfigDirectory.FullName}"); + options.TrustIssuer(caPath); + } + + return options; + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ExistingDatabaseFixture.cs b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ExistingDatabaseFixture.cs new file mode 100644 index 000000000..c6ebe4453 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ExistingDatabaseFixture.cs @@ -0,0 +1,138 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// Runs against the databases the environment template already created, rather than provisioning new ones. +/// +/// +/// The cheap tier, and the right one to start from: it needs none of the create_database schema, so it +/// answers "does the client work against a real deployment" before anything depends on parameter names that are +/// documented only as prose. +/// +public class ExistingDatabaseFixture : IAsyncLifetime +{ + private FaultInjectorClient? _injector; + private string? _skipReason; + + public IReadOnlyDictionary Databases { get; private set; } + = new Dictionary(); + + public FaultInjectorEnvironment Environment => + FaultInjectorEnvironment.Current ?? throw new InvalidOperationException("no environment; call RequireAvailable first"); + + public FaultInjectorClient Injector => + _injector ?? throw new InvalidOperationException("no injector; call RequireAvailable first"); + + /// Skips the calling test when there is nothing to talk to. + public void RequireAvailable() + { + if (_skipReason is not null) Assert.Skip(_skipReason); + } + + /// + /// The named database, or a skip if this environment does not have one. + /// + /// + /// A skip rather than a failure, deliberately: which databases exist is a property of the environment + /// template somebody chose, so a template without an oss_cluster database is a reason not to run the + /// cluster tests, not evidence of a bug. + /// + public ExistingDatabase Require(string key) + { + RequireAvailable(); + if (!Databases.TryGetValue(key, out var database)) + { + Assert.Skip($"this environment has no '{key}' database (found: {string.Join(", ", Databases.Keys)})"); + } + + return database!; + } + + /// + /// Where fixture-level diagnostics go. Deliberately collected rather than written to + /// : a fixture has no test output helper, and anything it prints to the console is + /// easily lost - which is how a silently failing setup call survived a day of use. + /// + public List SetupLog { get; } = []; + + private void log(string message) + { + SetupLog.Add(message); + Console.WriteLine(message); + } + + public async ValueTask InitializeAsync() + { + if (FaultInjectorEnvironment.Current is null) + { + _skipReason = FaultInjectorEnvironment.UnavailableReason ?? "no fault-injector environment configured"; + return; + } + + if (!FaultInjectorEnvironment.IsEnabled) + { + _skipReason = "set E2E_SCENARIO_TESTS=true to run against a real deployment"; + return; + } + + _injector = new FaultInjectorClient(FaultInjectorEnvironment.Current.InjectorUrl); + Databases = ExistingDatabase.ReadAll(FaultInjectorEnvironment.Current); + await EnsureMaintenanceNotificationsEnabledAsync(); + } + + public ValueTask DisposeAsync() + { + _injector?.Dispose(); + return default; + } + + /// + /// Turns on the cluster-level maintenance-notification flags. + /// + /// + /// A different thing from the per-connection opt-in, and both are required: the cluster flag decides whether + /// CLIENT MAINT_NOTIFICATIONS exists at all, and the command opts one connection in. Without this a + /// run fails at connect - correctly, since the tests ask for + /// - but for a reason that has nothing to do with the + /// client, so it is worth setting rather than diagnosing. + /// + /// Through the injector rather than a direct REST PUT, so the change leaves an action id behind and + /// shows up in the injector's history like everything else. Best-effort: if the action type is not + /// available on this build, the connect failure that follows says so clearly enough. + /// + /// + private async Task EnsureMaintenanceNotificationsEnabledAsync() + { + // nested under "config", not flattened - the same shape create_database wants for "database_config". + // A flat payload is rejected with "Invalid parameter 'config': got None", and because this call is + // best-effort it failed *silently* for a whole day of testing without anybody noticing. The impact was + // small: the environment templates enable these flags at provision time, so this is a safety net rather + // than the mechanism, and the tests were passing on their own merits. It matters for an environment + // provisioned without them, where the alternative is every test failing at connect and blaming the + // client for a server-side setting. + var parameters = new Dictionary + { + ["config"] = new Dictionary + { + ["client_maint_notifications"] = true, // proxy-routed databases + ["oss_cluster_client_maint_notifications"] = true, // oss_cluster databases + }, + }; + + try + { + await Injector.RunActionAsync("update_cluster_config", parameters, timeout: TimeSpan.FromMinutes(2)); + log("cluster maintenance-notification flags enabled"); + } + catch (Exception ex) + { + // Still best-effort - a cluster may not support the flags at all - but no longer silent: it goes to + // the test output, where the connect failure that follows can be attributed to it. + log($"could not set cluster maintenance-notification flags: {ScenarioSupport.Summarize(ex.Message)}"); + } + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/Environment/FaultInjectorEnvironment.cs b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/FaultInjectorEnvironment.cs new file mode 100644 index 000000000..6b1b18b5a --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/FaultInjectorEnvironment.cs @@ -0,0 +1,179 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Text.Json; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// Everything this suite needs to reach a real deployment, discovered from one directory. +/// +/// +/// One path is the whole configuration, deliberately. That directory is the one mounted into the fault +/// injector as /app/config and the one docker compose up is run from, so it already holds the +/// compose file, the cluster credentials (env_output.json), the CA certificate, and - once databases +/// exist - endpoints.json. Asking for anything else would mean the person who provisioned the +/// environment has to hand-carry facts into the test run, which is the error-prone part. +/// +/// Point at it with SER_FI_CONFIG_DIR; FI_CONSOLE_CONFIG_DIR is honoured as a fallback so this +/// runs against the same directory as the fault-injector console with no extra setup. +/// +/// +public sealed class FaultInjectorEnvironment +{ + private const string ConfigDirVariable = "SER_FI_CONFIG_DIR"; + private const string ConsoleConfigDirVariable = "FI_CONSOLE_CONFIG_DIR"; + private const string InjectorUrlVariable = "FAULT_INJECTION_API_URL"; + private const string EnabledVariable = "E2E_SCENARIO_TESTS"; + private const string DefaultInjectorUrl = "http://127.0.0.1:20324"; + + /// + /// The environment for this run, or null when none is configured. + /// + public static FaultInjectorEnvironment? Current { get; } = Discover(); + + /// + /// Why there is no environment, for a skip message that says something useful. + /// + public static string? UnavailableReason { get; private set; } + + private FaultInjectorEnvironment(DirectoryInfo configDirectory, Uri injectorUrl) + { + ConfigDirectory = configDirectory; + InjectorUrl = injectorUrl; + } + + public DirectoryInfo ConfigDirectory { get; } + + public Uri InjectorUrl { get; } + + /// + /// The CA certificate that signs the proxy certificates, for . + /// + /// + /// The certificates are self-signed per environment, so trusting the issuer is how a TLS test validates + /// anything at all. Note what this is *not*: a switch that disables validation. A test that cannot find the + /// CA fails rather than falling back to trusting everything, because a TLS test that silently stops + /// checking identity is worse than no TLS test - it reports success for the one thing it exists to catch. + /// + public string? CertificateAuthorityPath { get; private set; } + + /// + /// Cluster credentials for the REST API on port 9443, when env_output.json carries them. + /// + /// + /// Needed for the facts endpoints.json does not carry - notably oss_cluster and the + /// advertised endpoint type - which matter because they decide which notification family a database can + /// even emit. Tests that provision their own database know what they asked for and need none of this. + /// + public ClusterCredentials? Cluster { get; private set; } + + private static FaultInjectorEnvironment? Discover() + { + var path = Environment.GetEnvironmentVariable(ConfigDirVariable) + ?? Environment.GetEnvironmentVariable(ConsoleConfigDirVariable); + + if (string.IsNullOrWhiteSpace(path)) + { + UnavailableReason = $"no fault-injector environment: set {ConfigDirVariable} to the directory holding env_output.json"; + return null; + } + + var directory = new DirectoryInfo(path); + if (!directory.Exists) + { + // configured but wrong is a mistake worth reporting, not a reason to quietly do nothing + UnavailableReason = $"{ConfigDirVariable} points at '{path}', which does not exist"; + return null; + } + + var url = Environment.GetEnvironmentVariable(InjectorUrlVariable); + var injectorUrl = Uri.TryCreate(url, UriKind.Absolute, out var parsed) ? parsed : new Uri(DefaultInjectorUrl); + + var result = new FaultInjectorEnvironment(directory, injectorUrl) + { + CertificateAuthorityPath = FindCertificateAuthority(directory), + }; + result.Cluster = ReadClusterCredentials(directory); + return result; + } + + /// + /// Whether the caller actually meant to run destructive tests against a real deployment. + /// + /// + /// Two gates rather than one, and they mean different things. Without a config directory there is nothing + /// to talk to, which is the ordinary state for everybody else and skips. With a directory but without + /// E2E_SCENARIO_TESTS=true we also skip, because these tests create and delete databases and nobody + /// should trip that by running the full traversal. What must *not* happen is a third state where the + /// environment is configured, meant, and broken, yet the run still reports success - see + /// . + /// + public static bool IsEnabled => + string.Equals(Environment.GetEnvironmentVariable(EnabledVariable), "true", StringComparison.OrdinalIgnoreCase); + + private static string? FindCertificateAuthority(DirectoryInfo directory) + { + // names seen across the environment templates, most specific first + foreach (var candidate in new[] { "ca.pem", "ca.crt", "proxy_cert.pem", "redislabs_ca.pem" }) + { + var path = Path.Combine(directory.FullName, candidate); + if (File.Exists(path)) return path; + } + + return null; + } + + private static ClusterCredentials? ReadClusterCredentials(DirectoryInfo directory) + { + var path = Path.Combine(directory.FullName, "env_output.json"); + if (!File.Exists(path)) return null; + + try + { + using var document = JsonDocument.Parse(File.ReadAllText(path)); + + // two schemas in the wild: single-cluster templates put outputs at the top level, and the AWS + // multi-cluster template nests them under .clusters.value[N]. Both are handled because which one + // you get is a property of the template somebody chose, not of anything a test can control. + var root = document.RootElement; + if (root.TryGetProperty("clusters", out var clusters) + && clusters.TryGetProperty("value", out var values) + && values.ValueKind == JsonValueKind.Array + && values.GetArrayLength() > 0) + { + root = values[0]; + } + + var name = ReadValue(root, "cluster_name") ?? ReadValue(root, "name"); + var user = ReadValue(root, "username") ?? ReadValue(root, "cluster_username"); + var password = ReadValue(root, "password") ?? ReadValue(root, "cluster_password"); + + return name is null || user is null || password is null ? null : new ClusterCredentials(name, user, password); + } + catch (Exception) + { + // a malformed env_output.json is not fatal here: only the tests that need REST enrichment care, + // and they report it themselves rather than failing every test in the suite at discovery time + return null; + } + } + + /// + /// Reads a property that may be a bare value or a terraform-style { "value": ... } wrapper. + /// + private static string? ReadValue(JsonElement element, string property) + { + if (!element.TryGetProperty(property, out var value)) return null; + if (value.ValueKind == JsonValueKind.Object && value.TryGetProperty("value", out var inner)) value = inner; + return value.ValueKind == JsonValueKind.String ? value.GetString() : null; + } + + public sealed record ClusterCredentials(string ClusterName, string Username, string Password) + { + public Uri RestUrl => new($"https://{ClusterName}:9443"); + + /// Never let the password reach a log; the endpoints are real and reachable. + public override string ToString() => $"{Username}@{ClusterName}"; + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/Environment/FaultInjectorFixture.cs b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/FaultInjectorFixture.cs new file mode 100644 index 000000000..6406b8b88 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/FaultInjectorFixture.cs @@ -0,0 +1,186 @@ +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// Provisions one database shape for the tests that need it, and takes it away afterwards. +/// +/// +/// Per *shape*, not per test: creating a database on a real cluster is slow, so several test classes share one. +/// +/// The three-state behaviour matters more than the provisioning. No environment configured, or the +/// E2E_SCENARIO_TESTS gate not set, means every test skips - which is the ordinary case for anybody +/// running the full traversal, and these tests create and delete real databases. But an environment that *is* +/// configured and meant, and then does not work, must **fail**: a suite that skips on a broken environment +/// reports success for tests that never ran, and you will trust it exactly when you should not. So the absent +/// case skips from the test body, and the broken case throws from here. +/// +/// +public abstract class FaultInjectorFixture(DatabaseShape shape) : IAsyncLifetime +{ + /// + /// Shared by every database this suite creates, in every run. + /// + /// + /// The sweep that cleans up leaks from earlier runs matches on this and nothing else, so it can never touch + /// a database somebody created by hand. Worth being strict about: the alternative is a test suite that + /// deletes production-shaped things on a shared cluster. + /// + public const string NamePrefix = "sertest-"; + + private static readonly string RunId = Guid.NewGuid().ToString("n")[..6]; + + private FaultInjectorClient? _injector; + private string? _skipReason; + + public DatabaseShape Shape { get; } = shape; + + /// The database created for this fixture, once has run. + public ProvisionedDatabase? Database { get; private set; } + + public FaultInjectorEnvironment Environment => + FaultInjectorEnvironment.Current ?? throw new InvalidOperationException("no environment; call RequireAvailable first"); + + public FaultInjectorClient Injector => + _injector ?? throw new InvalidOperationException("no injector; call RequireAvailable first"); + + /// + /// Skips the calling test when there is no environment to talk to. + /// + public void RequireAvailable() + { + if (_skipReason is not null) Assert.Skip(_skipReason); + } + + public async ValueTask InitializeAsync() + { + if (FaultInjectorEnvironment.Current is null) + { + _skipReason = FaultInjectorEnvironment.UnavailableReason ?? "no fault-injector environment configured"; + return; + } + + if (!FaultInjectorEnvironment.IsEnabled) + { + _skipReason = "set E2E_SCENARIO_TESTS=true to run against a real deployment (these tests create and delete databases)"; + return; + } + + // configured and meant: from here on, problems are failures + _injector = new FaultInjectorClient(FaultInjectorEnvironment.Current.InjectorUrl); + await SweepOrphansAsync(); + Database = await CreateDatabaseAsync(); + } + + public async ValueTask DisposeAsync() + { + // unconditionally, and before anything else can throw: a database left behind holds a port and a slice + // of cluster memory, and the next run collides with it + if (Database is { } database && _injector is { } injector) + { + try + { + await injector.RunActionAsync("delete_database", new Dictionary { ["bdb_id"] = database.BdbId }); + } + catch (Exception ex) + { + // never mask a test failure with a cleanup failure; the sweep will get it next time + Console.WriteLine($"failed to delete {database.Name} (bdb {database.BdbId}): {ex.Message}"); + } + } + + _injector?.Dispose(); + } + + /// + /// Creates the database for this shape, working around port collisions the way go-redis has to. + /// + /// + /// Port collisions are common enough on a shared cluster that go-redis carries a dedicated + /// CreateDatabaseWithPortRetry helper; this is the same idea. The base port is deliberately high and + /// the walk is upward, so a collision costs one attempt rather than a failed run. + /// + private async Task CreateDatabaseAsync() + { + const int BasePort = 13500, Attempts = 8; + var name = $"{NamePrefix}{Shape.Label}-{RunId}"; + Exception? last = null; + + for (int attempt = 0; attempt < Attempts; attempt++) + { + var port = BasePort + (attempt * 3); + try + { + // nested under database_config, not flattened: the injector answers a flat payload with + // "Invalid parameter 'database_config': got None" + var result = await Injector.RunActionAsync( + "create_database", + new Dictionary { ["database_config"] = Shape.ToCreateParameters(name, port) }); + return ProvisionedDatabase.FromCreateResult(name, port, Shape, result, Environment); + } + catch (Exception ex) when (IsPortCollision(ex)) + { + last = ex; + Console.WriteLine($"create_database on port {port} collided, retrying higher"); + } + } + + throw new InvalidOperationException($"could not create '{name}' after {Attempts} attempts", last); + } + + /// + /// Whether a create failure is worth trying a different port for. + /// + /// + /// Only port collisions are: everything else - a malformed config, a cluster with no capacity - fails + /// identically on every port, and retrying eight times turns one clear error message into eight and delays + /// the report by half a minute. The first version of this retried everything, and buried + /// "missing shard_key_regex" under eight identical tracebacks. + /// + private static bool IsPortCollision(Exception ex) => + ex.Message.Contains("port", StringComparison.OrdinalIgnoreCase) + && (ex.Message.Contains("in use", StringComparison.OrdinalIgnoreCase) + || ex.Message.Contains("already", StringComparison.OrdinalIgnoreCase) + || ex.Message.Contains("taken", StringComparison.OrdinalIgnoreCase) + || ex.Message.Contains("conflict", StringComparison.OrdinalIgnoreCase)); + + /// + /// Best-effort removal of databases left behind by earlier runs. + /// + /// + /// Best-effort on purpose: listing databases needs the cluster REST API, which needs credentials this + /// directory may not carry, and being unable to tidy up is not a reason to fail a run that could otherwise + /// proceed. The per-fixture delete in is the reliable path; this only catches + /// leaks from a run that was killed. + /// + private async Task SweepOrphansAsync() + { + var cluster = Environment.Cluster; + if (cluster is null) + { + Console.WriteLine("no cluster credentials in env_output.json; skipping orphan sweep"); + return; + } + + try + { + using var rest = new ClusterRestClient(cluster, Environment.CertificateAuthorityPath); + foreach (var (bdbId, name) in await rest.ListDatabasesAsync()) + { + if (!name.StartsWith(NamePrefix, StringComparison.Ordinal)) continue; // never anything but ours + + Console.WriteLine($"sweeping orphaned test database {name} (bdb {bdbId})"); + await Injector.RunActionAsync("delete_database", new Dictionary { ["bdb_id"] = bdbId }); + } + } + catch (Exception ex) + { + Console.WriteLine($"orphan sweep skipped: {ex.Message}"); + } + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ProvisionedDatabase.cs b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ProvisionedDatabase.cs new file mode 100644 index 000000000..a6268eb87 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/Environment/ProvisionedDatabase.cs @@ -0,0 +1,170 @@ +using System; +using System.Collections.Generic; +using System.Text.Json; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// A database this suite created, and how to connect to it. +/// +/// +/// Note what this type does *not* need to do: read endpoints.json to discover whether the database is +/// oss_cluster, or what endpoint type it advertises. A test that provisioned the database knows what it +/// asked for, which is the real argument for provisioning from inside the suite - the alternative is carrying +/// those facts from whoever set the environment up into the run, by hand. +/// +public sealed class ProvisionedDatabase +{ + private ProvisionedDatabase(string name, int bdbId, string host, int port, DatabaseShape shape, FaultInjectorEnvironment environment) + { + Name = name; + BdbId = bdbId; + Host = host; + Port = port; + Shape = shape; + Environment = environment; + } + + public string Name { get; } + + public int BdbId { get; } + + public string Host { get; } + + public int Port { get; } + + public DatabaseShape Shape { get; } + + public FaultInjectorEnvironment Environment { get; } + + public string? Password { get; private init; } + + /// + /// Connection options for this database, with maintenance notifications requested. + /// + /// + /// rather than Auto: in a test, a server that + /// silently declines the opt-in should fail the connection loudly rather than produce a run that passes + /// while observing nothing. That is the opposite of the right default for production, and exactly right + /// here. + /// + public ConfigurationOptions GetClientConfig() + { + var options = new ConfigurationOptions + { + EndPoints = { { Host, Port } }, + Password = Password, + Protocol = RedisProtocol.Resp3, + MaintenanceNotifications = MaintenanceNotificationMode.Enabled, + AbortOnConnectFail = false, + // real network, real cluster: connect and command budgets have to tolerate a WAN round trip, and a + // cluster that is mid-scenario is slower still + ConnectTimeout = 15_000, + SyncTimeout = 15_000, + }; + + if (Shape.Tls) + { + options.Ssl = true; + options.SslHost = Host; + + // The certificates are self-signed per environment, so trusting the issuer is what makes a TLS test + // mean anything. If the CA is missing we fail here rather than disabling validation: a TLS test that + // quietly stops checking identity reports success for the one thing it exists to catch. + var caPath = Environment.CertificateAuthorityPath + ?? throw new InvalidOperationException( + $"{Shape.Label} needs TLS, but no CA certificate was found in {Environment.ConfigDirectory.FullName}; " + + "validation is not disabled for tests"); + options.TrustIssuer(caPath); + } + + return options; + } + + /// Masked, because these endpoints are real and reachable from the internet. + public override string ToString() => $"{Name} ({Host}:{Port}, bdb {BdbId}, {Shape.Label})"; + + /// + /// Reads what the injector reported back after create_database. + /// + /// + /// Tolerant by design: the response shape is documented as prose, and the fields worth having may arrive at + /// the top level or nested under an output/result object. What cannot be guessed is the bdb_id - + /// without it there is nothing to delete afterwards - so that one is required and its absence is loud. + /// + public static ProvisionedDatabase FromCreateResult( + string name, + int requestedPort, + DatabaseShape shape, + JsonElement result, + FaultInjectorEnvironment environment) + { + var bdbId = FindInt(result, "bdb_id") + ?? throw new InvalidOperationException($"create_database for '{name}' returned no bdb_id: {result}"); + + var host = FindString(result, "endpoint") ?? FindString(result, "dns_name") ?? environment.Cluster?.ClusterName + ?? throw new InvalidOperationException($"create_database for '{name}' returned no endpoint, and env_output.json names no cluster: {result}"); + + // an endpoint may arrive as "host:port" or as a bare host + var port = requestedPort; + var colon = host.LastIndexOf(':'); + if (colon > 0 && int.TryParse(host[(colon + 1)..], out var parsedPort)) + { + port = parsedPort; + host = host[..colon]; + } + + return new ProvisionedDatabase(name, bdbId, host, port, shape, environment) + { + Password = FindString(result, "password"), + }; + } + + private static int? FindInt(JsonElement element, string name) + { + foreach (var candidate in Walk(element, name)) + { + if (candidate.ValueKind == JsonValueKind.Number && candidate.TryGetInt32(out var value)) return value; + if (candidate.ValueKind == JsonValueKind.String && int.TryParse(candidate.GetString(), out var parsed)) return parsed; + } + + return null; + } + + private static string? FindString(JsonElement element, string name) + { + foreach (var candidate in Walk(element, name)) + { + if (candidate.ValueKind == JsonValueKind.String) return candidate.GetString(); + if (candidate.ValueKind == JsonValueKind.Array && candidate.GetArrayLength() > 0 && candidate[0].ValueKind == JsonValueKind.String) + { + return candidate[0].GetString(); // endpoints arrive as a list often enough + } + } + + return null; + } + + /// + /// Yields every value for a property name, at any depth - shallowest first. + /// + private static IEnumerable Walk(JsonElement element, string name) + { + if (element.ValueKind == JsonValueKind.Object) + { + if (element.TryGetProperty(name, out var direct)) yield return direct; + + foreach (var property in element.EnumerateObject()) + { + foreach (var nested in Walk(property.Value, name)) yield return nested; + } + } + else if (element.ValueKind == JsonValueKind.Array) + { + foreach (var item in element.EnumerateArray()) + { + foreach (var nested in Walk(item, name)) yield return nested; + } + } + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/FaultInjector/FaultInjectorClient.cs b/tests/StackExchange.Redis.FaultInjector.Tests/FaultInjector/FaultInjectorClient.cs new file mode 100644 index 000000000..3bbc96e30 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/FaultInjector/FaultInjectorClient.cs @@ -0,0 +1,204 @@ +using System; +using System.Collections.Generic; +using System.Net.Http; +using System.Net.Http.Json; +using System.Text.Json; +using System.Text.Json.Serialization; +using System.Threading; +using System.Threading.Tasks; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// The fault injector's HTTP surface, as much of it as this suite needs. +/// +/// +/// Deliberately the same shape go-redis, redis-py and node-redis wrap: POST /action returning an id and +/// GET /action/{id} to poll. They integrated independently and converged, so the contract is stable +/// enough to depend on, and their scenario documentation transfers to this client. +/// +public sealed class FaultInjectorClient(Uri baseAddress) : IDisposable +{ + private readonly HttpClient _http = new() { BaseAddress = baseAddress, Timeout = TimeSpan.FromMinutes(2) }; + + /// + /// Statuses that mean "still going". Both of them, which is the point. + /// + /// + /// A job passes through pending *and* running, so a loop that waits only on pending + /// returns while the work is still in flight - and then the test asserts against a cluster that has not + /// finished changing. Documented as a trap in the console's notes; encoded here so it cannot be + /// rediscovered. + /// + private static readonly HashSet PendingStatuses = new(StringComparer.OrdinalIgnoreCase) { "pending", "running", "in_progress" }; + + private static readonly HashSet FailedStatuses = new(StringComparer.OrdinalIgnoreCase) { "failed", "cancelled", "error" }; + + public void Dispose() => _http.Dispose(); + + /// + /// Fires an action and returns its id, without waiting. + /// + public async Task StartActionAsync(string type, object? parameters = null, CancellationToken cancellationToken = default) + { + var payload = new ActionRequest(type, parameters); + using var response = await _http.PostAsJsonAsync("/action", payload, cancellationToken); + await EnsureSuccessAsync(response, $"POST /action ({type})", cancellationToken); + + using var document = JsonDocument.Parse(await response.Content.ReadAsStringAsync(cancellationToken)); + return ReadActionId(document.RootElement) + ?? throw new InvalidOperationException($"the injector accepted '{type}' but returned no action id: {document.RootElement}"); + } + + /// + /// Fires an action and waits for it to finish, returning its final payload. + /// + public async Task RunActionAsync( + string type, + object? parameters = null, + TimeSpan? timeout = null, + CancellationToken cancellationToken = default) + { + var id = await StartActionAsync(type, parameters, cancellationToken); + return await WaitForActionAsync(id, timeout, cancellationToken); + } + + /// + /// Polls an action to completion. + /// + /// + /// The timeout is generous by default because these actions move data: a slot migration or a node coming + /// out of maintenance mode is minutes, not seconds. A tight default here would produce failures that look + /// like product bugs. + /// + public async Task WaitForActionAsync( + string id, + TimeSpan? timeout = null, + CancellationToken cancellationToken = default) + { + var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromMinutes(10)); + string? lastStatus = null; + + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + + using var response = await _http.GetAsync($"/action/{id}", cancellationToken); + await EnsureSuccessAsync(response, $"GET /action/{id}", cancellationToken); + + var body = await response.Content.ReadAsStringAsync(cancellationToken); + using var document = JsonDocument.Parse(body); + var status = ReadStatus(document.RootElement); + lastStatus = status ?? lastStatus; + + if (status is not null && !PendingStatuses.Contains(status)) + { + if (FailedStatuses.Contains(status)) + { + throw new InvalidOperationException($"fault-injector action {id} ended as '{status}': {body}"); + } + + // clone: the JsonDocument is disposed with this scope, and callers keep the result + return document.RootElement.Clone(); + } + + if (DateTime.UtcNow > deadline) + { + throw new TimeoutException($"fault-injector action {id} was still '{lastStatus ?? "unknown"}' after {(timeout ?? TimeSpan.FromMinutes(10)).TotalSeconds:0}s"); + } + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + } + } + + /// + /// Discovers which triggers are valid for an effect. + /// + /// + /// Worth asking rather than hardcoding: the effect/trigger matrix is sparse - maintenance_mode only + /// supports remove-add and remove, failover needs replication enabled - and some + /// scenarios do not enumerate their triggers in the schema at all. + /// + public async Task GetValidTriggersAsync(string scenario, string effect, int clusterIndex = 0, CancellationToken cancellationToken = default) + { + using var response = await _http.GetAsync($"/{scenario}?effect={Uri.EscapeDataString(effect)}&cluster_index={clusterIndex}", cancellationToken); + await EnsureSuccessAsync(response, $"GET /{scenario}?effect={effect}", cancellationToken); + using var document = JsonDocument.Parse(await response.Content.ReadAsStringAsync(cancellationToken)); + return document.RootElement.Clone(); + } + + /// + /// Runs a scenario's setup / run / teardown triple's individual legs. + /// + public async Task PostScenarioAsync( + string scenario, + string? leg, + IReadOnlyDictionary query, + TimeSpan? timeout = null, + CancellationToken cancellationToken = default) + { + var path = leg is null ? $"/{scenario}" : $"/{scenario}/{leg}"; + var separator = '?'; + foreach (var pair in query) + { + if (pair.Value is null) continue; + path += $"{separator}{pair.Key}={Uri.EscapeDataString(pair.Value)}"; + separator = '&'; + } + + using var response = await _http.PostAsync(path, content: null, cancellationToken); + await EnsureSuccessAsync(response, $"POST {path}", cancellationToken); + + var body = await response.Content.ReadAsStringAsync(cancellationToken); + using var document = JsonDocument.Parse(body); + + // the scenario legs only *enqueue*; a caller that returns here is racing the work it asked for + var id = ReadActionId(document.RootElement); + return id is null ? document.RootElement.Clone() : await WaitForActionAsync(id, timeout, cancellationToken); + } + + private static async Task EnsureSuccessAsync(HttpResponseMessage response, string what, CancellationToken cancellationToken) + { + if (response.IsSuccessStatusCode) return; + + // include the body: the injector explains refusals there, and "400 Bad Request" alone has cost people + // hours on the effect/trigger matrix + var body = await response.Content.ReadAsStringAsync(cancellationToken); + throw new InvalidOperationException($"{what} failed: {(int)response.StatusCode} {response.ReasonPhrase}. {body}"); + } + + /// + /// The id of a pollable action, if this response describes one. + /// + /// + /// Deliberately not setup_id: a scenario's setup returns a *handle* to state held in the injector, + /// which is passed back on the run and teardown legs, and is not something /action/{id} knows about. + /// Treating it as an action id polls a URL that does not exist. + /// + private static string? ReadActionId(JsonElement element) + { + foreach (var name in new[] { "action_id", "id" }) + { + if (element.TryGetProperty(name, out var value)) + { + return value.ValueKind switch + { + JsonValueKind.String => value.GetString(), + JsonValueKind.Number => value.ToString(), + _ => null, + }; + } + } + + return null; + } + + private static string? ReadStatus(JsonElement element) + => element.TryGetProperty("status", out var status) && status.ValueKind == JsonValueKind.String + ? status.GetString() + : null; + + private sealed record ActionRequest( + [property: JsonPropertyName("type")] string Type, + [property: JsonPropertyName("parameters")] object? Parameters); +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/FaultInjector/ScenarioRun.cs b/tests/StackExchange.Redis.FaultInjector.Tests/FaultInjector/ScenarioRun.cs new file mode 100644 index 000000000..6c90a2f6b --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/FaultInjector/ScenarioRun.cs @@ -0,0 +1,257 @@ +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// One run of a fault-injector scenario: setup, fire, tear down. +/// +/// +/// The setup leg provisions its own database, which is the important part: every trigger publishes the +/// dbconfig it needs (GET /topology-change-standalone?effect=...), and all of them want +/// proxy_policy: single - a shape the environment templates do not create. So a scenario cannot be +/// pointed at an existing database, and there is no need to hand-roll create_database for one either. +/// +/// Teardown is the whole reason this is a type rather than three calls. A scenario left set up holds cluster +/// state - the flags it enabled, the database it made, nodes it excluded - and poisons every run after it. +/// +/// +public sealed class ScenarioRun : IAsyncDisposable +{ + private readonly FaultInjectorClient _injector; + private readonly string _scenario; + private readonly Action _log; + + private ScenarioRun(FaultInjectorClient injector, string scenario, string effect, string trigger, Action log) + { + _injector = injector; + _scenario = scenario; + Effect = effect; + Trigger = trigger; + _log = log; + } + + public string Effect { get; } + + public string Trigger { get; } + + /// The setup handle, passed back on the run and teardown legs. + public string? SetupId { get; private set; } + + /// The database the setup leg created, when it says which. + public int? BdbId { get; private set; } + + /// Whatever setup returned, for a test that wants to look at more than we model. + public JsonElement SetupResult { get; private set; } + + /// The database setup created, ready to connect to. + /// + /// Setup hands back everything needed - endpoint, password, TLS - so a scenario test never has to look in + /// endpoints.json or ask the cluster REST API. Verified against a live run: the response carries + /// setup_id, bdb_id, db_name, endpoints, password, tls, + /// mtls_files and config (the proxy policy it chose to satisfy the trigger). + /// + public ScenarioDatabase? Database { get; private set; } + + /// The database a scenario provisioned for itself. + public sealed record ScenarioDatabase(string Name, int BdbId, string Host, int Port, bool Tls, string? Password, string? ProxyPolicy) + { + public override string ToString() => $"{Name} ({Host}:{Port}, bdb {BdbId}, policy {ProxyPolicy ?? "?"})"; + + public ConfigurationOptions GetClientConfig( + FaultInjectorEnvironment environment, + MaintenanceNotificationMode mode = MaintenanceNotificationMode.Enabled) + { + var options = new ConfigurationOptions + { + EndPoints = { { Host, Port } }, + Password = Password, + Protocol = RedisProtocol.Resp3, + MaintenanceNotifications = mode, + AbortOnConnectFail = false, + ConnectTimeout = 15_000, + SyncTimeout = 15_000, + }; + + if (Tls) + { + options.Ssl = true; + options.SslHost = Host; + options.TrustIssuer(environment.CertificateAuthorityPath + ?? throw new InvalidOperationException($"{Name} uses TLS but no CA certificate was found in {environment.ConfigDirectory.FullName}")); + } + + return options; + } + } + + /// + /// Sets a scenario up, ready to fire. + /// + /// + /// The trigger for the *setup* leg, when it differs from the one being fired. + /// + /// + /// The two triggers are different questions, which the naming hides. The setup leg's trigger says how to + /// *provision* - /slot-migrate/setup accepts only reshard, because provisioning a cluster + /// database is all it does - while the run leg's trigger says how to cause the effect (migrate, + /// maintenance_mode, failover). Passing the run trigger to setup earns a + /// "Trigger 'migrate' is not supported by slot-migrate/setup (only 'reshard')". + /// + public static async Task SetupAsync( + FaultInjectorClient injector, + string scenario, + string effect, + string trigger, + Action log, + IReadOnlyDictionary? extra = null, + string? setupTrigger = null, + CancellationToken cancellationToken = default) + { + var run = new ScenarioRun(injector, scenario, effect, trigger, log); + var query = new Dictionary { ["effect"] = effect, ["trigger"] = setupTrigger ?? trigger }; + if (extra is not null) + { + foreach (var pair in extra) query[pair.Key] = pair.Value; + } + + log($"setup {scenario}: effect={effect} setup-trigger={setupTrigger ?? trigger} (firing '{trigger}')"); + run.SetupResult = await injector.PostScenarioAsync(scenario, "setup", query, cancellationToken: cancellationToken); + run.SetupId = FindString(run.SetupResult, "setup_id"); + run.BdbId = FindInt(run.SetupResult, "bdb_id"); + run.Database = ReadDatabase(run.SetupResult, run.BdbId); + log($"setup complete: setup_id={run.SetupId ?? "(none)"} database={run.Database?.ToString() ?? "(none)"}"); + return run; + } + + /// + /// Fires the scenario and waits for the injector to finish its work. + /// + /// + /// Note "finished" here means the injector has done what it was asked, not that the deployment has settled: + /// the notifications, the DNS change and the socket close all trail it by seconds. Callers should watch for + /// what they expect rather than assuming completion means arrival. + /// + public async Task FireAsync(CancellationToken cancellationToken = default) + { + var query = new Dictionary + { + ["effect"] = Effect, + ["trigger"] = Trigger, + ["setup_id"] = SetupId, + ["bdb_id"] = BdbId?.ToString(), + }; + + _log($"firing {_scenario}"); + var result = await _injector.PostScenarioAsync(_scenario, leg: null, query, cancellationToken: cancellationToken); + _log($"fired: {result}"); + return result; + } + + public async ValueTask DisposeAsync() + { + var query = new Dictionary + { + // both, because setup_id lives in the injector's memory and is lost if it restarts, at which point + // bdb_id is the only handle left + ["setup_id"] = SetupId, + ["bdb_id"] = BdbId?.ToString(), + ["restore_nodes"] = "true", // put back anything the scenario excluded, or the next run starts degraded + }; + + try + { + // Its own budget rather than the caller's token: teardown has to happen even when the test was + // cancelled or timed out, which is exactly when the caller's token is already dead. + using var timeout = new CancellationTokenSource(TimeSpan.FromMinutes(5)); + await _injector.PostScenarioAsync(_scenario, "teardown", query, cancellationToken: timeout.Token); + _log("teardown complete"); + } + catch (Exception ex) + { + // loud, but not an exception: a teardown failure must not replace the test's own verdict + _log($"TEARDOWN FAILED ({ex.Message}) - the cluster may be left with a scenario set up, and " + + $"setup_id={SetupId ?? "(none)"} bdb_id={BdbId?.ToString() ?? "(none)"} is what to clean up by hand"); + } + } + + private static ScenarioDatabase? ReadDatabase(JsonElement setup, int? bdbId) + { + if (bdbId is not { } id) return null; + + var endpoint = FindString(setup, "endpoints"); + if (endpoint is null) return null; + + var text = endpoint; + var scheme = text.IndexOf("://", StringComparison.Ordinal); + if (scheme >= 0) text = text[(scheme + 3)..]; + var colon = text.LastIndexOf(':'); + if (colon <= 0 || !int.TryParse(text[(colon + 1)..], out var port)) return null; + + return new ScenarioDatabase( + FindString(setup, "db_name") ?? $"bdb-{id}", + id, + text[..colon], + port, + setup.TryGetProperty("tls", out var tls) && tls.ValueKind == JsonValueKind.True, + FindString(setup, "password"), + FindString(setup, "config")); + } + + private static string? FindString(JsonElement element, string name) + { + foreach (var candidate in Walk(element, name)) + { + if (candidate.ValueKind == JsonValueKind.String) return candidate.GetString(); + if (candidate.ValueKind == JsonValueKind.Number) return candidate.ToString(); + if (candidate.ValueKind == JsonValueKind.Array && candidate.GetArrayLength() > 0 + && candidate[0].ValueKind == JsonValueKind.String) + { + return candidate[0].GetString(); // "endpoints" is a list even when there is one + } + } + + return null; + } + + private static int? FindInt(JsonElement element, string name) + { + foreach (var candidate in Walk(element, name)) + { + if (candidate.ValueKind == JsonValueKind.Number && candidate.TryGetInt32(out var value)) return value; + if (candidate.ValueKind == JsonValueKind.String && int.TryParse(candidate.GetString(), out var parsed)) return parsed; + } + + return null; + } + + /// + /// Yields every value for a property name at any depth, shallowest first. + /// + /// + /// Tolerant on purpose: the scenario responses are documented as prose and nest differently between legs, so + /// searching beats asserting a shape - and a test that cannot find setup_id still has a logged + /// payload to work from rather than a deserialization error. + /// + private static IEnumerable Walk(JsonElement element, string name) + { + if (element.ValueKind == JsonValueKind.Object) + { + if (element.TryGetProperty(name, out var direct)) yield return direct; + foreach (var property in element.EnumerateObject()) + { + foreach (var nested in Walk(property.Value, name)) yield return nested; + } + } + else if (element.ValueKind == JsonValueKind.Array) + { + foreach (var item in element.EnumerateArray()) + { + foreach (var nested in Walk(item, name)) yield return nested; + } + } + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/FaultInjector/ScenarioSupport.cs b/tests/StackExchange.Redis.FaultInjector.Tests/FaultInjector/ScenarioSupport.cs new file mode 100644 index 000000000..9886ee68b --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/FaultInjector/ScenarioSupport.cs @@ -0,0 +1,95 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// Telling "this deployment cannot do that" apart from "the client got it wrong". +/// +/// +/// The effect/trigger matrix is sparse in ways that depend on the *cluster*, not just the database: an +/// add migration needs a node holding more than one shard, and a three-node cluster with sparse shard +/// placement gives every node exactly one. The injector reports that as a failed action with a Python +/// traceback, which is indistinguishable from a real fault unless you read the message. +/// +/// So these skip, narrowly and by message. A broad catch here would hide genuine injector failures, which is +/// the thing this tier exists to surface - hence matching on the specific condition rather than on the +/// exception type. +/// +/// +internal static class ScenarioSupport +{ + private static readonly string[] PlacementLimitations = + [ + "No node with multiple shards found", + "not enough nodes", + "no empty node", + ]; + + /// + /// Skips when the setup could not produce a database the effect can act on. + /// + public static void RequireEffectIsAchievable(ScenarioRun scenario, string effect) + { + if (scenario.Database is null) + { + Assert.Skip($"the injector did not provision a database for '{effect}'"); + } + } + + /// + /// Fires a scenario, skipping rather than failing when the cluster's shape cannot produce the effect. + /// + public static async Task FireOrSkipAsync(ScenarioRun scenario, string effect, CancellationToken cancellationToken) + { + try + { + await scenario.FireAsync(cancellationToken); + } + catch (Exception ex) when (IsPlacementLimitation(ex)) + { + Assert.Skip($"this cluster cannot produce '{effect}': {Summarize(ex.Message)}"); + } + } + + private static bool IsPlacementLimitation(Exception ex) + { + foreach (var limitation in PlacementLimitations) + { + if (ex.Message.Contains(limitation, StringComparison.OrdinalIgnoreCase)) return true; + } + + return false; + } + + /// + /// The part of a Python traceback that says what actually went wrong. + /// + /// + /// Note the traceback arrives inside a JSON string, so its line breaks are the two characters + /// \n rather than real newlines - splitting on '\n' alone matches nothing and returns + /// the whole wall of text, which is how the first version of this behaved. The injector helpfully ends with + /// "Caused by: ...", which is the only part worth putting in a skip message. + /// + internal static string Summarize(string message) + { + var caused = message.LastIndexOf("Caused by:", StringComparison.Ordinal); + if (caused >= 0) + { + var tail = message[caused..]; + var end = tail.IndexOf("\\n", StringComparison.Ordinal); + return end > 0 ? tail[..end] : tail; + } + + var lines = message.Replace("\\n", "\n", StringComparison.Ordinal) + .Split('\n', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); + for (int i = lines.Length - 1; i >= 0; i--) + { + if (lines[i].Contains("Exception:", StringComparison.Ordinal)) return lines[i]; + } + + return lines.Length > 0 ? lines[^1] : message; + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/MovingHandoffScenarioTests.cs b/tests/StackExchange.Redis.FaultInjector.Tests/MovingHandoffScenarioTests.cs new file mode 100644 index 000000000..14fe17dbd --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/MovingHandoffScenarioTests.cs @@ -0,0 +1,131 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading.Tasks; +using StackExchange.Redis.Maintenance; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// D6 against a real deployment: do we get off the connection before the server takes it away? +/// +/// +/// The one thing the in-process harness structurally cannot test. The handoff's hostname branch needs two +/// things a fake cannot supply - an endpoint that is a *name*, and a real socket whose remote address tells us +/// where we currently are - and its decision turns on DNS changing underneath us. +/// +/// The discriminator is the failure type of the disconnect. ConnectionDisposed means *we* replaced the +/// connection; SocketClosed means the server did. Both outcomes are legitimate, because DNS is not +/// guaranteed to win the race - measured on one cluster at +18.7s against a socket closing at +15.7s - so this +/// records which happened rather than demanding the good one. +/// +/// +[Trait("tier", "fault-injector")] +[Trait("scenario", "moving-handoff")] +public class MovingHandoffScenarioTests(ExistingDatabaseFixture fixture, ITestOutputHelper log) + : IClassFixture +{ + [Theory] + [InlineData("conn_drop", "endpoint_rebind")] + [InlineData("data_movement_conn_drop", "maintenance_mode")] + public async Task HandoffBeatsTheServerToTheClose(string effect, string trigger) + { + fixture.RequireAvailable(); + var cancellationToken = TestContext.Current.CancellationToken; + + await using var scenario = await ScenarioRun.SetupAsync( + fixture.Injector, "topology-change-standalone", effect, trigger, log.WriteLine, + cancellationToken: cancellationToken); + + var database = scenario.Database; + Assert.NotNull(database); + + var clock = Stopwatch.StartNew(); + var timeline = new List(); + void Note(string what) + { + var entry = $" +{clock.Elapsed.TotalSeconds,6:0.0}s {what}"; + lock (timeline) timeline.Add(entry); + log.WriteLine(entry); + } + + await using var conn = await ConnectionMultiplexer.ConnectAsync(database.GetClientConfig(fixture.Environment)); + var muxer = (IInternalConnectionMultiplexer)conn; + var endpoint = muxer.GetServerEndPoint(conn.GetEndPoints()[0]); + + // The precondition the fake could not meet: a name to resolve, and a socket that says where we are. + Assert.IsType(conn.GetEndPoints()[0]); + log.WriteLine($"dialled {conn.GetEndPoints()[0]} (a name, so the probe can engage)"); + + var failures = new List(); + conn.ConnectionFailed += (_, e) => + { + lock (failures) failures.Add(e.FailureType); + Note($"disconnect: {e.FailureType}"); + }; + conn.ConnectionRestored += (_, _) => Note("reconnected"); + conn.ServerMaintenanceEvent += (_, e) => + { + if (e is PushMaintenanceEvent push) Note($"{push.NotificationType} seq={push.SequenceId} time={push.Time?.TotalSeconds.ToString() ?? "-"}"); + }; + + clock.Restart(); + await scenario.FireAsync(cancellationToken); + Note("injector reports the scenario finished"); + + // long enough to cover the whole announced window plus the observed overshoot + var deadline = clock.Elapsed + TimeSpan.FromSeconds(75); + while (clock.Elapsed < deadline) + { + try + { + await conn.GetDatabase().PingAsync(); + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + Note($"ping failed: {ex.GetType().Name}"); + } + + await Task.Delay(1000, cancellationToken); + } + + Note($"handoff outcome: {endpoint.LastHandoffOutcome ?? "(none recorded)"}; recycles={endpoint.HandoffRecycles}"); + + lock (failures) + { + var order = string.Join(" -> ", failures); + log.WriteLine($" disconnect sequence: {(order.Length == 0 ? "(none)" : order)}"); + + // Note what is *not* asserted: that a ConnectionDisposed appears here. Our own recycle does not + // raise ConnectionFailed - disposal is not reported as a failure - so from the outside a handoff is + // invisible, which is worth knowing in its own right and is why HandoffRecycles exists. + } + + // Exactly one handoff per notification. More than one is the feedback loop this test found on its first + // live run: a server re-sends MOVING to a connection that opts in while the window is still open, and + // since the handoff replaces the connection, acting on the repeat produces another one - twelve + // recycles from a single event. The per-server sequence dedup now gates it, and this is the assertion + // that would catch a regression. + Assert.Equal(1, endpoint.HandoffRecycles); + Assert.NotNull(endpoint.LastHandoffOutcome); + Assert.Contains("Recycle", endpoint.LastHandoffOutcome); + + 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: 30_000), + "the client should be serving commands after the handoff"); + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/Poll.cs b/tests/StackExchange.Redis.FaultInjector.Tests/Poll.cs new file mode 100644 index 000000000..0d93b2db2 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/Poll.cs @@ -0,0 +1,21 @@ +using System; +using System.Threading.Tasks; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// Waits for a condition that a real deployment reaches on its own schedule. +/// +internal static class Poll +{ + public static async Task UntilAsync(Func condition, int timeoutMilliseconds = 10_000, int pollMilliseconds = 250) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMilliseconds); + while (true) + { + if (condition()) return true; + if (DateTime.UtcNow > deadline) return false; + await Task.Delay(pollMilliseconds); + } + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/ProxyAndFailoverScenarioTests.cs b/tests/StackExchange.Redis.FaultInjector.Tests/ProxyAndFailoverScenarioTests.cs new file mode 100644 index 000000000..dd2135c74 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/ProxyAndFailoverScenarioTests.cs @@ -0,0 +1,169 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading.Tasks; +using StackExchange.Redis.Maintenance; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// A replicated database, provisioned by us - the only way to reach the failover family. +/// +/// +/// The environment templates create databases with replication: false, and a failover needs a replica to +/// promote, so this is the one case where create_database earns its keep rather than being a +/// less-convenient alternative to the scenario setup legs. +/// +public sealed class ReplicatedDatabaseFixture() + : FaultInjectorFixture(new DatabaseShape("replicated", ProxyPolicy: "single", Replication: true, ShardCount: 2)); + +/// +/// The proxy and failover families: FAILING_OVER/FAILED_OVER, and a proxy restarting underneath us. +/// +[Trait("tier", "fault-injector")] +[Trait("scenario", "failover")] +public class FailoverScenarioTests(ReplicatedDatabaseFixture fixture, ITestOutputHelper log) + : IClassFixture +{ + [Fact] + public async Task FailoverIsAnnouncedAndSurvived() + { + fixture.RequireAvailable(); + var cancellationToken = TestContext.Current.CancellationToken; + var database = fixture.Database; + Assert.NotNull(database); + log.WriteLine($"provisioned {database}"); + + var clock = Stopwatch.StartNew(); + var events = new List(); + + await using var conn = await ConnectionMultiplexer.ConnectAsync(database.GetClientConfig()); + conn.ServerMaintenanceEvent += (_, e) => + { + if (e is PushMaintenanceEvent push) + { + lock (events) events.Add(push); + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s {push.NotificationType} seq={push.SequenceId} {push.RawMessage}"); + } + }; + + var db = conn.GetDatabase(); + await db.StringSetAsync("fi-failover", "before"); + + clock.Restart(); + try + { + await fixture.Injector.RunActionAsync( + "failover", + new Dictionary { ["bdb_id"] = database.BdbId.ToString() }, + cancellationToken: cancellationToken); + } + catch (Exception ex) + { + // The failover action's parameters are untyped in the schema, so a rejection here is a harness + // problem rather than a client finding; say so plainly instead of reporting it as a product failure. + Assert.Skip($"the injector would not run 'failover' against bdb {database.BdbId}: {ScenarioSupport.Summarize(ex.Message)}"); + } + + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s injector reports the failover finished"); + + var deadline = clock.Elapsed + TimeSpan.FromSeconds(45); + while (clock.Elapsed < deadline) + { + try + { + await db.PingAsync(); + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s ping failed: {ex.GetType().Name}"); + } + + await Task.Delay(1000, cancellationToken); + } + + lock (events) + { + log.WriteLine($" {events.Count} notification(s): {string.Join(", ", events.Select(e => e.NotificationType))}"); + + // A failover is the one event whose *pair* we have never observed end to end from a client: Marc + // captured the frames by hand, but nothing has watched SE.Redis receive them. + Assert.NotEmpty(events); + Assert.Contains(events, e => e.NotificationType is MaintenanceNotificationType.FailingOver + or MaintenanceNotificationType.FailedOver + or MaintenanceNotificationType.Migrating // a failover on a proxied database moves shards too + or MaintenanceNotificationType.Migrated + or MaintenanceNotificationType.Moving); + } + + // the data survived, which is the point of replication + Assert.Equal("before", await db.StringGetAsync("fi-failover")); + } +} + +/// +/// The proxy process restarting: no topology change, no notification, just the socket going away. +/// +/// +/// Included because it is the one disruption that is *purely* a connection event - nothing moves, nothing is +/// announced, and recovery is entirely the client's ordinary reconnect path. A useful control: if this fails, +/// the failures in the announced scenarios are not about notifications at all. +/// +[Trait("tier", "fault-injector")] +[Trait("scenario", "proxy-restart")] +public class ProxyRestartScenarioTests(ExistingDatabaseFixture fixture, ITestOutputHelper log) + : IClassFixture +{ + [Theory] + [InlineData("standalone")] + [InlineData("cluster")] + public async Task ProxyRestartIsSurvived(string key) + { + var database = fixture.Require(key); + var cancellationToken = TestContext.Current.CancellationToken; + + var clock = Stopwatch.StartNew(); + await using var conn = await ConnectionMultiplexer.ConnectAsync(database.GetClientConfig(fixture.Environment)); + var db = conn.GetDatabase(); + await db.StringSetAsync($"fi-dmc-{key}", "before"); + + int drops = 0; + conn.ConnectionFailed += (_, e) => + { + drops++; + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s connection failed: {e.FailureType}"); + }; + conn.ConnectionRestored += (_, _) => log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s connection restored"); + + clock.Restart(); + // the one action with a typed parameter object in the injector's schema: RestartDmcParams { bdb_id } + await fixture.Injector.RunActionAsync( + "dmc_restart", + new Dictionary { ["bdb_id"] = database.BdbId.ToString() }, + cancellationToken: cancellationToken); + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s DMC restart reported finished"); + + // Recovery is asserted by polling rather than by waiting a fixed time, because a proxy restart is quick + // and the interesting failure is "never comes back", not "takes a moment". + Assert.True( + await Poll.UntilAsync(() => TryRead(db, $"fi-dmc-{key}"), timeoutMilliseconds: 60_000), + "the client should recover from the proxy restarting"); + + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s recovered after {drops} drop(s)"); + Assert.Equal("before", await db.StringGetAsync($"fi-dmc-{key}")); + } + + private static bool TryRead(IDatabase db, string key) + { + try + { + return db.StringGet(key) == "before"; + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + return false; + } + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/README.md b/tests/StackExchange.Redis.FaultInjector.Tests/README.md new file mode 100644 index 000000000..2be42b3a7 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/README.md @@ -0,0 +1,60 @@ +# StackExchange.Redis.FaultInjector.Tests + +Scenario tests that drive a **real Redis Enterprise deployment** through the fault injector, and watch how +SE.Redis reacts. These are the tier that can observe what no in-process fake can: real DNS, real TLS identity, +real timing. + +## Running them + +```bash +cd # holds docker-compose.yml, env_output.json, ... +docker compose up -d # 10-15 minutes for the cluster to come up + +export SER_FI_CONFIG_DIR=$PWD # or FI_CONSOLE_CONFIG_DIR, which is also honoured +export E2E_SCENARIO_TESTS=true # explicit opt-in: these create and delete databases +dotnet test tests/StackExchange.Redis.FaultInjector.Tests +``` + +One path is the whole configuration. That directory is the one mounted into the injector as `/app/config`, so +it already holds the cluster credentials (`env_output.json`), the CA certificate, and the compose file; nothing +has to be hand-carried into the test run. `FAULT_INJECTION_API_URL` overrides the injector URL +(default `http://127.0.0.1:20324`). + +## Three states, deliberately distinct + +| state | behaviour | +|---|---| +| no `SER_FI_CONFIG_DIR` | every test **skips** - the ordinary case, including `build.ps1`'s full traversal | +| directory set, no `E2E_SCENARIO_TESTS=true` | every test **skips** - nobody should create databases by accident | +| configured and enabled, but broken | every test **fails** | + +The third row is the important one. A suite that skips when the environment is broken reports success for tests +that never ran, and it will be trusted at exactly the wrong moment. + +## Databases are created by the tests, not by you + +Each *shape* (`DatabaseShape`) is a fixture shared by the classes that need it, because creating a database on a +real cluster is slow. The shapes exist because they change client behaviour rather than for coverage's sake: the +number of A records a hostname carries follows proxy placement, and the handoff takes a different branch +depending on whether a live sibling address exists. + +Every database is named `sertest--`. Cleanup is per fixture and unconditional; a sweep at startup +removes leaks from runs that were killed, matching on the `sertest-` prefix and nothing else, so it can never +touch a database created by hand. + +## TLS + +Certificates are self-signed per environment, so tests call `ConfigurationOptions.TrustIssuer(caPath)` with the +CA found in the config directory. If the CA is missing, TLS tests **fail** rather than disabling validation - a +TLS test that quietly stops checking identity reports success for the one thing it exists to catch. + +## Traits + +`tier=fault-injector` on everything, so the whole tier can be excluded in one filter; `scenario=` for +subsets. + +## Unverified + +The `create_database` parameter names in `DatabaseShape.ToCreateParameters` are the injector's wire schema, +which is documented only as prose. They are gathered in one place so a real run can correct them; go-redis's +`DatabaseConfig` is the closest reference implementation. Treat them as unconfirmed until a run accepts them. diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/RealDeploymentSmokeTests.cs b/tests/StackExchange.Redis.FaultInjector.Tests/RealDeploymentSmokeTests.cs new file mode 100644 index 000000000..a5f37b8b6 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/RealDeploymentSmokeTests.cs @@ -0,0 +1,64 @@ +using System; +using System.Threading.Tasks; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// Does the client work against a real deployment at all: connect, negotiate RESP3, opt in, run a command. +/// +/// +/// The floor for everything else. Worth having as its own class because when a scenario test fails, the first +/// question is whether the deployment was reachable in the first place, and a green smoke test answers it +/// without reading logs. +/// +[Trait("tier", "fault-injector")] +[Trait("scenario", "smoke")] +public class RealDeploymentSmokeTests(ExistingDatabaseFixture fixture, ITestOutputHelper log) + : IClassFixture +{ + [Theory] + [InlineData("standalone")] + [InlineData("cluster")] + public async Task OptInIsAcceptedAndTheConnectionWorks(string key) + { + var database = fixture.Require(key); + log.WriteLine($"{database}"); + log.WriteLine($" advertised addresses: {string.Join(", ", database.Addresses)}"); + log.WriteLine($" endpoint type: {database.EndpointType ?? "(unset)"}"); + + // Enabled refuses the connection if the server will not give us notifications, so reaching the + // assertions at all is the opt-in having been accepted - there is nothing weaker to check. A stub +OK + // would pass this, which is what the scenario tests are for. + await using var conn = await ConnectionMultiplexer.ConnectAsync(database.GetClientConfig(fixture.Environment)); + Assert.True(conn.IsConnected); + + var server = conn.GetServer(conn.GetEndPoints()[0]); + log.WriteLine($" connected: {server.Version}, {server.ServerType}, protocol {server.Protocol}"); + Assert.Equal(RedisProtocol.Resp3, server.Protocol); + + var rtt = await conn.GetDatabase().PingAsync(); + log.WriteLine($" ping: {rtt.TotalMilliseconds:0.0}ms"); + Assert.True(rtt > TimeSpan.Zero); + + // and the feature is actually live on this connection, not merely requested + var endpoint = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(conn.GetEndPoints()[0]); + Assert.True(endpoint.MaintenanceNotificationsActive, "maintenance notifications should be active"); + } + + [Fact] + public async Task AdvertisedAddressCountMatchesTheProxyPolicy() + { + // Not a client assertion: a check that the environment is the shape the handoff tests assume. The count + // follows proxy *placement* rather than the policy name, so this records what this deployment actually + // is - and a handoff test that expects a sibling to step to needs more than one. + var standalone = fixture.Require("standalone"); + log.WriteLine($"{standalone.Key}: policy {standalone.ProxyPolicy}, {standalone.AdvertisedAddressCount} address(es)"); + + Assert.NotEmpty(standalone.Addresses); + if (standalone.ProxyPolicy is "single") + { + Assert.Equal(1, standalone.AdvertisedAddressCount); + } + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/StackExchange.Redis.FaultInjector.Tests.csproj b/tests/StackExchange.Redis.FaultInjector.Tests/StackExchange.Redis.FaultInjector.Tests.csproj new file mode 100644 index 000000000..052d85e11 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/StackExchange.Redis.FaultInjector.Tests.csproj @@ -0,0 +1,25 @@ + + + + net10.0 + Exe + StackExchange.Redis.FaultInjector.Tests + true + true + full + enable + true + + + + + + + + + + + + diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/TlsScenarioTests.cs b/tests/StackExchange.Redis.FaultInjector.Tests/TlsScenarioTests.cs new file mode 100644 index 000000000..a99be3bc2 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/TlsScenarioTests.cs @@ -0,0 +1,161 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.IO; +using System.Threading.Tasks; +using StackExchange.Redis.Maintenance; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// The same feature over TLS, against certificates a real cluster generated. +/// +/// +/// A gap that has been open since the identity work: nothing had exercised a real TLS deployment, only the +/// in-process harness with a certificate it made itself. That matters because the interesting failures are +/// about *identity* - which name the certificate carries, and whether the endpoint we are told to use is +/// covered by it - and a fake that issues its own certificate cannot be wrong about that in the way a real +/// deployment can. +/// +/// Certificates here are self-signed per environment, so the client pins the issuer with +/// . Note what is deliberately *not* done: validation is +/// never disabled. A TLS test that stops checking identity reports success for precisely the thing it exists to +/// catch, so a missing CA fails the run instead. +/// +/// +[Trait("tier", "fault-injector")] +[Trait("scenario", "tls")] +public class TlsScenarioTests(ExistingDatabaseFixture fixture, ITestOutputHelper log) + : IClassFixture +{ + [Fact] + public async Task NotificationsArriveOverTlsAndIdentityIsVerified() + { + fixture.RequireAvailable(); + var cancellationToken = TestContext.Current.CancellationToken; + + if (fixture.Environment.CertificateAuthorityPath is null) + { + // a skip rather than a failure: whether the environment generated certificates is a provisioning + // choice, and running this without them would prove nothing + Assert.Skip($"no CA certificate in {fixture.Environment.ConfigDirectory.FullName}; provision with TLS to run this"); + } + + log.WriteLine($"trusting issuer {fixture.Environment.CertificateAuthorityPath}"); + CertificateSanity.RequireCertificatesMatchThisCluster(fixture.Environment, log.WriteLine); + + await using var scenario = await ScenarioRun.SetupAsync( + fixture.Injector, + "topology-change-standalone", + "conn_drop", + "endpoint_rebind", + log.WriteLine, + // include_tls does not *request* TLS, it widens the list of variants the setup may choose from; + // variant_index is what picks one. Confirmed by asking the discovery endpoint: with no flags a + // trigger offers one variant ("single"), with include_tls two ("single", "single_tls"), and with + // include_mtls a third ("mtls"). Passing include_tls alone provisions variant 0 and yields a + // plaintext database, which is how this test first came to skip itself. + extra: new Dictionary + { + ["include_tls"] = "true", + ["variant_index"] = "1", + }, + cancellationToken: cancellationToken); + + var database = scenario.Database; + Assert.NotNull(database); + log.WriteLine($"provisioned {database} (tls={database.Tls})"); + + if (!database.Tls) + { + Assert.Skip("the injector provisioned a plaintext database despite include_tls=true; nothing to test here"); + } + + var clock = Stopwatch.StartNew(); + var events = new List(); + + var config = database.GetClientConfig(fixture.Environment); + + // AbortOnConnectFail=true *for this test only*: everywhere else tolerating a slow start is right, but a + // TLS failure has to surface its reason. With it false, a certificate problem is indistinguishable from + // a slow cluster - ConnectAsync succeeds, IsConnected is false, and the cause is gone. + config.AbortOnConnectFail = true; + + // If the certificate does not cover the name we dialled, this throws - which is the point: the + // assertion that identity was verified is the connect succeeding with validation on. The log goes to + // test output so a handshake failure says *which* check failed. + var connectLog = new StringWriter(); + ConnectionMultiplexer conn; + try + { + conn = await ConnectionMultiplexer.ConnectAsync(config, connectLog); + } + catch (Exception ex) + { + log.WriteLine(connectLog.ToString()); + log.WriteLine($"TLS connect failed: {ex.GetType().Name}: {ex.Message}"); + throw; + } + + await using (conn) + { + Assert.True(conn.IsConnected); + + var endpoint = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(conn.GetEndPoints()[0]); + Assert.True(endpoint.MaintenanceNotificationsActive, "the opt-in should be live over TLS too"); + log.WriteLine($"connected over TLS; opt-in active; ping {(await conn.GetDatabase().PingAsync()).TotalMilliseconds:0.0}ms"); + + conn.ServerMaintenanceEvent += (_, e) => + { + if (e is PushMaintenanceEvent push) + { + lock (events) events.Add(push); + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s {push.NotificationType} seq={push.SequenceId} {push.RawMessage}"); + } + }; + + clock.Restart(); + await scenario.FireAsync(cancellationToken); + + var deadline = clock.Elapsed + TimeSpan.FromSeconds(45); + while (clock.Elapsed < deadline) + { + try + { + await conn.GetDatabase().PingAsync(); + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + log.WriteLine($" +{clock.Elapsed.TotalSeconds,6:0.0}s ping failed: {ex.GetType().Name}"); + } + + await Task.Delay(1000, cancellationToken); + } + + lock (events) + { + log.WriteLine($" {events.Count} notification(s) over TLS"); + Assert.NotEmpty(events); + } + + // and the TLS handshake succeeds again on the *replacement* connection, which is the part a + // certificate-name problem would break rather than the first connect + 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: 30_000), + "the client should re-establish TLS after the endpoint moves"); + } + } +} diff --git a/tests/StackExchange.Redis.FaultInjector.Tests/TopologyChangeScenarioTests.cs b/tests/StackExchange.Redis.FaultInjector.Tests/TopologyChangeScenarioTests.cs new file mode 100644 index 000000000..4c79f1e01 --- /dev/null +++ b/tests/StackExchange.Redis.FaultInjector.Tests/TopologyChangeScenarioTests.cs @@ -0,0 +1,164 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading.Tasks; +using StackExchange.Redis.Maintenance; +using Xunit; + +namespace StackExchange.Redis.FaultInjector.Tests; + +/// +/// Real topology changes on a real deployment, watched by a real client. +/// +/// +/// Each scenario provisions its own database, because every trigger publishes the dbconfig it requires +/// and all of them want proxy_policy: single - a shape the environment templates do not create. +/// +/// What these assert is deliberately modest: that the notifications arrive, parse, and open the relaxation +/// window. Timings are *recorded* rather than asserted, because the measured spread across clusters is wide +/// enough that a bound tight enough to be interesting would be flaky - DNS has been seen updating 4.4s after +/// MOVING on one cluster and 3s after the socket closed on another. The log is the deliverable for +/// timing; the assertions cover behaviour. +/// +/// +[Trait("tier", "fault-injector")] +[Trait("scenario", "topology-change")] +public class TopologyChangeScenarioTests(ExistingDatabaseFixture fixture, ITestOutputHelper log) + : IClassFixture +{ + /// + /// Records what arrived and when, relative to the moment the scenario was fired. + /// + private sealed class Timeline(Stopwatch clock, ITestOutputHelper log) + { + private readonly List _entries = []; + + public List Events { get; } = []; + + public void Note(string what) + { + var entry = $" +{clock.Elapsed.TotalSeconds,6:0.0}s {what}"; + lock (_entries) + { + _entries.Add(entry); + } + + log.WriteLine(entry); + } + + public void Add(PushMaintenanceEvent evt) + { + lock (_entries) + { + Events.Add(evt); + } + + Note($"{evt.NotificationType} seq={evt.SequenceId} time={evt.Time?.TotalSeconds.ToString() ?? "-"} {evt.RawMessage}"); + } + + public int Count + { + get { lock (_entries) { return Events.Count; } } + } + } + + /// + /// Whether each scenario announces itself, and why - measured 2026-09-01 against RS 8.0.22. + /// + /// + /// The expectations encode the rule the measurements produced, which is narrower than either half of it + /// looks: a connection is told MOVING when its own proxy leaves the endpoint's address set *and* + /// the set gains a member. Both conditions are needed - a pure reduction takes the proxy away without + /// announcing it, and a pure widening adds addresses while leaving the connection's proxy in place, which is + /// also silent. The data-movement pair (MIGRATING/MIGRATED) is separate and fires whenever + /// shards move, whether or not any endpoint changes. + /// + [Theory] + [InlineData("conn_drop", "endpoint_rebind", true)] + [InlineData("dns_resolution_change", "endpoint_rebind", false)] + [InlineData("data_movement_conn_drop", "maintenance_mode", true)] + [InlineData("data_movement_no_conn_drop", "migrate", true)] + public async Task ScenarioProducesNotificationsWeUnderstand(string effect, string trigger, bool expectNotifications) + { + fixture.RequireAvailable(); + var cancellationToken = TestContext.Current.CancellationToken; + + await using var scenario = await ScenarioRun.SetupAsync( + fixture.Injector, "topology-change-standalone", effect, trigger, log.WriteLine, cancellationToken: cancellationToken); + + var database = scenario.Database; + Assert.NotNull(database); + + var clock = Stopwatch.StartNew(); + var timeline = new Timeline(clock, log); + + await using var conn = await ConnectionMultiplexer.ConnectAsync(database.GetClientConfig(fixture.Environment)); + var endpoint = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(conn.GetEndPoints()[0]); + Assert.True(endpoint.MaintenanceNotificationsActive, "the opt-in must be live, or this test proves nothing"); + + conn.ServerMaintenanceEvent += (_, e) => + { + if (e is PushMaintenanceEvent push) timeline.Add(push); + }; + conn.ConnectionFailed += (_, e) => timeline.Note($"connection failed: {e.FailureType} {e.Exception?.Message}"); + conn.ConnectionRestored += (_, e) => timeline.Note("connection restored"); + + clock.Restart(); + timeline.Note($"firing {effect}/{trigger} against {database}"); + await scenario.FireAsync(cancellationToken); + timeline.Note("injector reports the scenario finished"); + + // The injector finishing is not the deployment settling: notifications, the DNS change and the socket + // close all trail it. Keep watching, and keep the client busy so a broken connection actually surfaces + // rather than sitting idle. + var deadline = clock.Elapsed + TimeSpan.FromSeconds(60); + while (clock.Elapsed < deadline) + { + try + { + await conn.GetDatabase().PingAsync(); + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + timeline.Note($"ping failed: {ex.GetType().Name}"); + } + + await Task.Delay(1000, cancellationToken); + } + + timeline.Note($"finished with {timeline.Count} notification(s); relaxed={endpoint.IsMaintenanceRelaxed}"); + + if (expectNotifications) + { + Assert.NotEmpty(timeline.Events); + Assert.All(timeline.Events, e => Assert.NotEqual(MaintenanceNotificationType.None, e.NotificationType)); + } + else + { + // dns_resolution_change widens the policy (single -> all-master-shards), so addresses are *added* + // and the connection's own proxy stays where it is - there is nothing to tell this connection to + // move, and the proxy restarting to apply the change closes the socket with no warning at all. + // Asserting the silence deliberately: it is the case with no signal, so if a future build starts + // announcing it, that is a behaviour change we want to be told about rather than to absorb quietly. + Assert.Empty(timeline.Events); + } + + // and the client is usable afterwards, which is the point of the whole feature + Assert.True( + await Poll.UntilAsync(() => TryPing(conn), timeoutMilliseconds: 30_000), + "the client should be serving commands again after the topology change"); + } + + private static bool TryPing(IConnectionMultiplexer conn) + { + try + { + return conn.IsConnected && conn.GetDatabase().Ping() >= TimeSpan.Zero; + } + catch (Exception ex) when (ex is RedisException or TimeoutException) + { + return false; + } + } +} diff --git a/tests/StackExchange.Redis.Tests/AdvertisedAddressProbeTests.cs b/tests/StackExchange.Redis.Tests/AdvertisedAddressProbeTests.cs new file mode 100644 index 000000000..302c880f1 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/AdvertisedAddressProbeTests.cs @@ -0,0 +1,216 @@ +using System; +using System.Collections.Generic; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using StackExchange.Redis.Maintenance; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// The DNS probe behind the MOVING handoff. Tested against an injected resolver rather than real DNS, +/// which is the only way to exercise the case that actually happens: the first answer naming the address we +/// were just told to leave. +/// +public class AdvertisedAddressProbeTests(ITestOutputHelper log) +{ + private static readonly IPAddress Retiring = IPAddress.Parse("10.129.228.140"); + private static readonly IPAddress Replacement = IPAddress.Parse("10.252.90.18"); + private static readonly DnsEndPoint Endpoint = new("db.example.cloud.redislabs.com", 13486); + + /// + /// Resolves from a script: one entry per call, the last repeating forever. + /// + private sealed class ScriptedResolver(params IPAddress[][] answers) + { + private int _calls; + public int Calls => Volatile.Read(ref _calls); + + public Task ResolveAsync(string host, CancellationToken cancellationToken) + { + var index = Interlocked.Increment(ref _calls) - 1; + return Task.FromResult(answers[Math.Min(index, answers.Length - 1)]); + } + } + + [Fact] + public async Task DnsTrailingTheNotificationIsPolledThrough() + { + // The measured case: DNS named the retiring address for the first 4.4-9.7 seconds after MOVING. A + // client that accepted the first answer would hand off to the node it was told to leave. + var resolver = new ScriptedResolver( + [Retiring], + [Retiring], + [Retiring], + [Replacement]); + + var result = await AdvertisedAddressProbe.ProbeAsync( + Endpoint, Retiring, window: TimeSpan.FromSeconds(5), pollInterval: TimeSpan.FromMilliseconds(10), + resolve: resolver.ResolveAsync, log: log.WriteLine); + + Assert.Equal(new IPEndPoint(Replacement, 13486), result); + Assert.Equal(4, resolver.Calls); + } + + [Fact] + public async Task ReplacementOnTheFirstAnswerIsTakenImmediately() + { + var resolver = new ScriptedResolver([Replacement]); + + var result = await AdvertisedAddressProbe.ProbeAsync( + Endpoint, Retiring, window: TimeSpan.FromSeconds(5), pollInterval: TimeSpan.FromSeconds(1), + resolve: resolver.ResolveAsync, log: log.WriteLine); + + Assert.Equal(new IPEndPoint(Replacement, 13486), result); + Assert.Equal(1, resolver.Calls); + } + + [Fact] + public async Task WindowExpiringWithoutAMoveGivesUpRatherThanGuessing() + { + // Not a failure to report: the server closes the socket regardless, and the relaxed window covers the + // reconnect. Guessing an address here would be worse than doing nothing. + var resolver = new ScriptedResolver([Retiring]); + + var result = await AdvertisedAddressProbe.ProbeAsync( + Endpoint, Retiring, window: TimeSpan.FromMilliseconds(120), pollInterval: TimeSpan.FromMilliseconds(20), + resolve: resolver.ResolveAsync, log: log.WriteLine); + + Assert.Null(result); + Assert.True(resolver.Calls > 1, $"should have retried within the window, but resolved {resolver.Calls} time(s)"); + } + + [Fact] + public async Task ZeroWindowStillGetsOneAttempt() + { + // "act now" rather than "do nothing": the notifications legitimately carry zero or negative times for + // a connection that arrived mid-window + var resolver = new ScriptedResolver([Replacement]); + + var result = await AdvertisedAddressProbe.ProbeAsync( + Endpoint, Retiring, window: TimeSpan.Zero, pollInterval: TimeSpan.FromSeconds(1), + resolve: resolver.ResolveAsync, log: log.WriteLine); + + Assert.Equal(new IPEndPoint(Replacement, 13486), result); + Assert.Equal(1, resolver.Calls); + } + + [Fact] + public async Task ResolutionFailureIsRetriedNotFatal() + { + // a DNS blip mid-handoff is when we can least afford to give up + int calls = 0; + Task Resolve(string host, CancellationToken cancellationToken) + { + calls++; + return calls switch + { + 1 => throw new System.Net.Sockets.SocketException(11001), // host not found + 2 => Task.FromResult([Retiring]), + _ => Task.FromResult([Replacement]), + }; + } + + var result = await AdvertisedAddressProbe.ProbeAsync( + Endpoint, Retiring, window: TimeSpan.FromSeconds(5), pollInterval: TimeSpan.FromMilliseconds(10), + resolve: Resolve, log: log.WriteLine); + + Assert.Equal(new IPEndPoint(Replacement, 13486), result); + Assert.Equal(3, calls); + } + + [Fact] + public async Task MultipleAddressesTakeTheOneThatIsNotRetiring() + { + // a round-robin record can name both nodes at once mid-move + var resolver = new ScriptedResolver([Retiring, Replacement]); + + var result = await AdvertisedAddressProbe.ProbeAsync( + Endpoint, Retiring, window: TimeSpan.FromSeconds(5), pollInterval: TimeSpan.FromSeconds(1), + resolve: resolver.ResolveAsync, log: log.WriteLine); + + Assert.Equal(new IPEndPoint(Replacement, 13486), result); + } + + [Fact] + public async Task PlacementNotPolicyDecidesWhetherWeWait() + { + // The A-record count follows actual proxy placement, not the policy name: an all-master-shards + // database whose shards share a node resolves to one address, and then there is no sibling to step to + // and the wait is the only option. Same code path as `single`, which is the point - nothing here reads + // the policy. + var resolver = new ScriptedResolver([Retiring], [Retiring], [Replacement]); + + var result = await AdvertisedAddressProbe.ProbeAsync( + Endpoint, Retiring, window: TimeSpan.FromSeconds(5), pollInterval: TimeSpan.FromMilliseconds(10), + resolve: resolver.ResolveAsync, log: log.WriteLine); + + Assert.Equal(new IPEndPoint(Replacement, 13486), result); + Assert.Equal(3, resolver.Calls); // it waited, because there was nothing else advertised + } + + [Fact] + public async Task SiblingIsTakenWithoutWaitingForTheRecordToMove() + { + // The common case: several A records, so the first resolution already names a live sibling proxy while + // the retiring address is *still* advertised. Stepping sideways immediately is correct - any proxy of + // the same database serves the same data - and it means the poll usually never engages. + var sibling = IPAddress.Parse("10.246.250.155"); + var resolver = new ScriptedResolver([Retiring, sibling]); + + var result = await AdvertisedAddressProbe.ProbeAsync( + Endpoint, Retiring, window: TimeSpan.FromSeconds(5), pollInterval: TimeSpan.FromSeconds(1), + resolve: resolver.ResolveAsync, log: log.WriteLine); + + Assert.Equal(new IPEndPoint(sibling, 13486), result); + Assert.Equal(1, resolver.Calls); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task StillAdvertisedAnswersTheUnannouncedCase(bool present) + { + // The measured gap: a multi-proxy node taken out for maintenance announced only MIGRATING/MIGRATED, + // dropped the victim from DNS at +21.4s, and closed its socket silently at +34.7s. For thirteen + // seconds the condition was visible to anyone who asked, and no notification said so. + var sibling = IPAddress.Parse("10.246.250.155"); + var resolver = new ScriptedResolver(present ? [Retiring, sibling] : [sibling, Replacement]); + + var result = await AdvertisedAddressProbe.IsStillAdvertisedAsync( + Endpoint, Retiring, resolve: resolver.ResolveAsync, log: log.WriteLine); + + Assert.Equal(present, result); + } + + [Fact] + public async Task ResolutionFailureIsNotAReasonToAbandonAConnection() + { + // null rather than false, deliberately: "cannot tell" must not become "give it up", or a DNS blip + // recycles every healthy connection at once + Task Throws(string host, CancellationToken cancellationToken) + => throw new System.Net.Sockets.SocketException(11001); + + Assert.Null(await AdvertisedAddressProbe.IsStillAdvertisedAsync( + Endpoint, Retiring, resolve: Throws, log: log.WriteLine)); + + // and the same for a record that momentarily resolves to nothing at all + Assert.Null(await AdvertisedAddressProbe.IsStillAdvertisedAsync( + Endpoint, Retiring, resolve: (_, _) => Task.FromResult([]), log: log.WriteLine)); + } + + [Fact] + public async Task CancellationStopsThePoll() + { + using var cts = new CancellationTokenSource(); + var resolver = new ScriptedResolver([Retiring]); + + var probe = AdvertisedAddressProbe.ProbeAsync( + Endpoint, Retiring, window: TimeSpan.FromMinutes(1), pollInterval: TimeSpan.FromMilliseconds(10), + resolve: resolver.ResolveAsync, log: log.WriteLine, cancellationToken: cts.Token); + + cts.Cancel(); + await Assert.ThrowsAnyAsync(() => probe); + } +} diff --git a/tests/StackExchange.Redis.Tests/MaintenanceHandoffTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceHandoffTests.cs new file mode 100644 index 000000000..33ee7aeb6 --- /dev/null +++ b/tests/StackExchange.Redis.Tests/MaintenanceHandoffTests.cs @@ -0,0 +1,113 @@ +using System; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using StackExchange.Redis.Maintenance; +using Xunit; + +namespace StackExchange.Redis.Tests; + +/// +/// Deciding what a MOVING means: where to go, or that there is nowhere to go. +/// +/// +/// Tested as a pure decision, with the resolver supplied, because the whole thing turns on DNS *changing* - +/// which no in-process fake can arrange. +/// +public class MaintenanceHandoffTests(ITestOutputHelper log) +{ + private static readonly IPAddress Retiring = IPAddress.Parse("10.129.228.140"); + private static readonly IPAddress Replacement = IPAddress.Parse("10.252.90.18"); + private static readonly DnsEndPoint Hostname = new("db.example.cloud.redislabs.com", 13486); + + private static Func> Resolves(params IPAddress[] addresses) + => (_, _) => Task.FromResult(addresses); + + [Fact] + public async Task HostnameThatHasMovedIsRecycled() + { + var decision = await MaintenanceHandoff.DecideAsync( + Hostname, successor: null, currentAddress: Retiring, + window: TimeSpan.FromSeconds(5), pollInterval: TimeSpan.FromMilliseconds(10), + resolve: Resolves(Replacement), log: log.WriteLine); + + log.WriteLine(decision.ToString()); + Assert.Equal(HandoffAction.Recycle, decision.Action); + Assert.Equal(new IPEndPoint(Replacement, 13486), decision.Target); + } + + [Fact] + public async Task HostnameThatNeverMovesDoesNothing() + { + // Not a failure: the server closes the socket regardless, the reconnect re-resolves, and the relaxed + // window covers the gap. Measured on a real cluster - DNS updated three seconds *after* the close - so + // this is a normal outcome rather than a defensive branch. + var decision = await MaintenanceHandoff.DecideAsync( + Hostname, successor: null, currentAddress: Retiring, + window: TimeSpan.FromMilliseconds(100), pollInterval: TimeSpan.FromMilliseconds(20), + resolve: Resolves(Retiring), log: log.WriteLine); + + log.WriteLine(decision.ToString()); + Assert.Equal(HandoffAction.None, decision.Action); + Assert.Null(decision.Target); + } + + [Fact] + public async Task AddressEndpointWithNoSuccessorHasNothingToDo() + { + // An address cannot be re-resolved, and nothing was named: there is no handoff to make. This is the + // case a cluster deployment would hit, and it is why MOVING must not simply reuse endpoint retirement - + // there would be nothing to retire *to*. + var decision = await MaintenanceHandoff.DecideAsync( + new IPEndPoint(Retiring, 13486), successor: null, currentAddress: Retiring, + window: TimeSpan.FromSeconds(5), pollInterval: TimeSpan.FromMilliseconds(10), + resolve: Resolves(Replacement), log: log.WriteLine); + + log.WriteLine(decision.ToString()); + Assert.Equal(HandoffAction.None, decision.Action); + } + + [Fact] + public async Task NamedSuccessorAsksForAReconfigure() + { + // Never observed on any real deployment - eleven routes, all explicit nulls - so this exists because the + // contract has it, not because it fires. + var successor = new IPEndPoint(Replacement, 13486); + var decision = await MaintenanceHandoff.DecideAsync( + Hostname, successor, currentAddress: Retiring, + window: TimeSpan.FromSeconds(5), pollInterval: TimeSpan.FromMilliseconds(10), + resolve: Resolves(Retiring), log: log.WriteLine); + + log.WriteLine(decision.ToString()); + Assert.Equal(HandoffAction.Reconfigure, decision.Action); + Assert.Equal(successor, decision.Target); + } + + [Fact] + public async Task UnknownCurrentAddressDoesNothingRatherThanGuessing() + { + // Recycling here would mean accepting whatever DNS says *now*, which for the first several seconds is + // the address being retired - so we would hand off to the node we were told to leave. + var decision = await MaintenanceHandoff.DecideAsync( + Hostname, successor: null, currentAddress: null, + window: TimeSpan.FromSeconds(5), pollInterval: TimeSpan.FromMilliseconds(10), + resolve: Resolves(Replacement), log: log.WriteLine); + + log.WriteLine(decision.ToString()); + Assert.Equal(HandoffAction.None, decision.Action); + } + + [Theory] + [InlineData(15, 0, 1000)] // a generous window: capped at a second, not a tenth of fifteen + [InlineData(2, 0, 200)] // a short one: a tenth, so a 2s window never spends more than 200ms + [InlineData(0, 0, 0)] // "act now" + public void JitterScalesWithTheWindowAndIsCapped(int windowSeconds, int minMilliseconds, int maxMilliseconds) + { + var random = new Random(12345); + for (int i = 0; i < 200; i++) + { + var jitter = MaintenanceHandoff.GetJitter(TimeSpan.FromSeconds(windowSeconds), random); + Assert.InRange(jitter.TotalMilliseconds, minMilliseconds, maxMilliseconds); + } + } +} diff --git a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs index fa59a6411..0fad1947f 100644 --- a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs +++ b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs @@ -602,6 +602,132 @@ public async Task RetentionCanBeTurnedOff() } } + [Theory] + [InlineData(null)] // the measured shape: the close arrives *after* the announced window + [InlineData(300)] // a less generous proxy: the close beats the window it announced + public async Task MovingClosesTheConnectionAndWeRecover(int? closeDelayMilliseconds) + { + // The defining half of MOVING, which no test covered until now: we are told to move, and then the + // socket goes away. Measured on RS (2026-08-28): the close landed at +18.4s and +16.6s against a + // declared 15s window, so the window is a floor with slack rather than a deadline - which is why the + // default case here lets the fake overshoot, and why nothing below asserts on the window expiring. + // The short case is the defensive one: a client that treats the announced window as guaranteed fails + // it. What must hold either way is that the connection comes back and the relaxed window opened by the + // notification is what covers the reconnect. + var (server, conn, events) = await ConnectAsync(log); + using (server) + await using (conn) + { + server.MovingClosesConnection = true; + server.MovingCloseDelay = closeDelayMilliseconds is { } ms ? TimeSpan.FromMilliseconds(ms) : null; + server.SendMoving(null, timeSeconds: 1, newEndpoint: null, sequenceId: 0); + + var moving = await events.NextAsync(); + Assert.Equal(MaintenanceNotificationType.Moving, moving.NotificationType); + + var endpoint = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(server.DefaultEndPoint); + Assert.True(endpoint.IsMaintenanceRelaxed, "the window has to be open before the socket dies"); + + // the client must reconnect on its own; nothing here asks it to. Note the probe has to tolerate + // throwing rather than returning false: a command issued in the window between the socket dying + // and the reconnect completing raises, and that is the state being polled *out* of + 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: 15_000), + "the multiplexer should have re-established itself after the close"); + + log.WriteLine($"close delay {closeDelayMilliseconds?.ToString() ?? "announced+slack"}: reconnected, relaxed = {endpoint.IsMaintenanceRelaxed}"); + } + } + + [Fact] + public async Task MovingRecyclesTheConnectionBeforeTheServerCloses() + { + // The point of D6: today a MOVING is survivable because the socket eventually closes and we reconnect, + // but the announced window goes unused. Here nothing closes the connection - MovingClosesConnection is + // deliberately left off - so the *only* thing that can replace it is our own handoff. + // + // The named-successor branch is what is exercised, because the in-process transport has no socket and so + // no current address for the DNS branch to compare against; the deciding half of that branch is covered + // by MaintenanceHandoffTests against a supplied resolver. What this proves is the acting half: drain, + // drop, and come back. + var (server, conn, events) = await ConnectAsync(log); + using (server) + await using (conn) + { + var optInsBefore = server.TotalMaintenanceOptIns; + Assert.Equal(1, optInsBefore); // the handshake's own opt-in, so the increment below means something + + // A handoff has to be *visible*: the replacement raises ConnectionRestored, so reporting nothing for + // the drop would leave a consumer tracking connection state with an unpaired restore. + var failures = new List(); + conn.ConnectionFailed += (_, e) => + { + lock (failures) failures.Add(e.FailureType); + }; + + server.SendMoving(null, timeSeconds: 2, newEndpoint: server.DefaultEndPoint, sequenceId: 0); + + var moving = await events.NextAsync(); + Assert.Equal(MaintenanceNotificationType.Moving, moving.NotificationType); + Assert.Equal(server.DefaultEndPoint, moving.NewEndPoint); + + // a fresh connection re-sends the opt-in, which is how a recycle is visible from the server's side + Assert.True( + await Poll.UntilAsync(() => server.TotalMaintenanceOptIns > optInsBefore, timeoutMilliseconds: 15_000), + "the handoff should have replaced the connection without the server closing it"); + + log.WriteLine($"opt-ins: {optInsBefore} -> {server.TotalMaintenanceOptIns}"); + + Assert.True( + await Poll.UntilAsync( + () => + { + lock (failures) return failures.Contains(ConnectionFailureType.MaintenanceHandoff); + }, + timeoutMilliseconds: 5_000), + "the recycle should be reported as a MaintenanceHandoff rather than silently"); + + lock (failures) + { + log.WriteLine($"reported: {string.Join(", ", failures)}"); + + // and *only* as that: reporting it as a socket failure would put planned maintenance into + // everybody's fault dashboards + Assert.DoesNotContain(ConnectionFailureType.SocketFailure, failures); + Assert.DoesNotContain(ConnectionFailureType.SocketClosed, failures); + } + + // ...and the replacement is usable, which is the only outcome a caller cares about + 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: 15_000), + "the client should be serving commands on the replacement connection"); + } + } + [Fact] public async Task MalformedTripletIsSkippedNotFatal() { diff --git a/tests/StackExchange.Redis.Tests/MaintenanceOptInClientTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceOptInClientTests.cs index afcda452e..7da72d32d 100644 --- a/tests/StackExchange.Redis.Tests/MaintenanceOptInClientTests.cs +++ b/tests/StackExchange.Redis.Tests/MaintenanceOptInClientTests.cs @@ -226,7 +226,12 @@ public async Task OptInIsReArmedOnReconnect() log.WriteLine($"opt-ins: {before} -> {server.TotalMaintenanceOptIns}"); Assert.True(server.TotalMaintenanceOptIns > before, "the opt-in should be sent again on the new connection"); - Assert.True(IsActive(conn, server)); + + // ...and then wait for *our* side of it. The count above is the server having processed the opt-in, + // whereas IsActive is us having processed its reply - a beat later, so asserting it directly is a race + // that only shows up when the runner is starved of cores. + await UntilCondition(() => IsActive(conn, server)); + Assert.True(IsActive(conn, server), "the feature should be live again on the replacement connection"); } private static async Task UntilCondition(System.Func condition, int timeoutMilliseconds = 5000) diff --git a/toys/StackExchange.Redis.Server/RedisServer.Maintenance.cs b/toys/StackExchange.Redis.Server/RedisServer.Maintenance.cs index c070203bc..51530f5a9 100644 --- a/toys/StackExchange.Redis.Server/RedisServer.Maintenance.cs +++ b/toys/StackExchange.Redis.Server/RedisServer.Maintenance.cs @@ -1,6 +1,8 @@ using System; +using System.Collections.Generic; using System.Net; using System.Threading; +using System.Threading.Tasks; using RESPite; using RESPite.Messages; @@ -160,7 +162,84 @@ private void AnnounceMigration(int hashSlot, Node from, Node to) /// /// The number of clients the notification was sent to. public int SendMoving(RedisClient client, int timeSeconds, EndPoint newEndpoint, int? sequenceId = null) - => Send(client, MaintenanceNotificationKind.Moving, timeSeconds, sequenceId, newEndpoint, null); + { + var recipients = MovingClosesConnection ? new List() : null; + Action onSent = recipients is null ? null : recipients.Add; + var count = Send(client, MaintenanceNotificationKind.Moving, timeSeconds, sequenceId, newEndpoint, null, onSent); + + if (recipients is { Count: > 0 }) + { + // The half of MOVING that defines it: the socket goes away - and it takes every other + // connection to the same node with it (see MovingClosesConnection). + var delay = MovingCloseDelay ?? TimeSpan.FromSeconds(Math.Max(timeSeconds, 0)) + MeasuredCloseSlack; + _ = CloseAfterAsync(CollectSiblings(recipients), delay); + } + + return count; + } + + /// + /// Whether then closes the affected connections, as a real proxy does. + /// + /// + /// Note the blast radius is the *node*, not the connection. Measured on RS (2026-08-28) with four + /// connections to one node differing only in handshake - two opted in, one RESP3 without the opt-in, + /// one RESP2 - all four closed simultaneously, and only the two that had opted in were warned. So this + /// is endpoint retirement rather than socket recycling, and a connection that did not opt in gets no + /// warning at all before it dies: an argument for opting in on every connection rather than one. + /// + public bool MovingClosesConnection { get; set; } + + /// + /// How much later than the announced window the close actually arrives, by default. + /// + /// + /// Measured at +3.4s and +1.6s against a declared 15s grace, i.e. the announced window behaves as a + /// floor with slack rather than a deadline. Tests should assert that a client acts *within* the window, + /// never that the socket survives to the end of it. + /// + public TimeSpan MeasuredCloseSlack { get; set; } = TimeSpan.FromSeconds(2); + + /// + /// How long after MOVING the connections are closed; defaults to the announced window plus + /// , which is what a real proxy was measured doing. + /// + /// + /// Set it shorter than the announced window to exercise a proxy less generous than the one measured - + /// a client that treats the window as guaranteed rather than as a budget fails that case. It cannot + /// usefully be set to zero: the close would race the delivery of the notification itself, which is not + /// something any real timing produces. + /// + public TimeSpan? MovingCloseDelay { get; set; } + + /// + /// Expands the notified connections to every connection sharing a node with them. + /// + private List CollectSiblings(List notified) + { + var nodes = new HashSet(); + foreach (var client in notified) nodes.Add(client.Node); + + var all = new List(); + ForAllClients( + all, + (client, list) => + { + if (nodes.Contains(client.Node)) list.Add(client); + return 0; + }); + return all; + } + + private async Task CloseAfterAsync(List clients, TimeSpan delay) + { + if (delay > TimeSpan.Zero) await Task.Delay(delay).ConfigureAwait(false); + foreach (var client in clients) + { + Log($"[{client}] closing connection after MOVING"); + client.Kill(); + } + } /// /// Sends one of the shard-scoped notifications. is a remaining-time @@ -252,7 +331,8 @@ private int Send( int? timeSeconds, int? sequenceId, EndPoint newEndpoint, - string extra) + string extra, + Action onSent = null) { // [type, seqID, ...] - the sequence id is an integer, which is precisely why these frames could // not be treated as pub/sub: element 1 is not a channel name @@ -262,7 +342,7 @@ private int Send( // records what was *sent*, so a test arranges it by sending the completion, not by poking state if (IsRetainedCompletion(kind)) _retainedCompletion = (kind, seq, extra); - return Dispatch(client, Build, requireOptIn: true); + return Dispatch(client, Build, requireOptIn: true, onSent); TypedRedisValue Build() => BuildShardNotification(kind, timeSeconds, seq, newEndpoint, extra); } @@ -307,11 +387,12 @@ private static TypedRedisValue BuildShardNotification( /// broadcast therefore used to reach exactly one client, which is invisible in a single-node test and /// silently made multi-node fan-out untestable. /// - private int Dispatch(RedisClient client, Func frameFactory, bool requireOptIn) + private int Dispatch(RedisClient client, Func frameFactory, bool requireOptIn, Action onSent = null) { if (client is not null) { client.AddOutbound(frameFactory()); + onSent?.Invoke(client); return 1; } @@ -324,6 +405,7 @@ private int Dispatch(RedisClient client, Func frameFactory, boo { if (gated && !target.MaintenanceNotifications) return 0; target.AddOutbound(frameFactory()); + onSent?.Invoke(target); return 1; }); }