diff --git a/CHANGELOG.md b/CHANGELOG.md index 914e61c9..d3875334 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,11 +8,40 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ### Added +- **Maintenance notifications from either source.** `RedisPlannedMaintenance` recognised only + `AzureMaintenanceEvent`, the pub/sub notifications Azure Cache for Redis publishes, and returned on anything else — + so the RESP3 push notifications Redis Enterprise and Redis Cloud send as `PushMaintenanceEvent` were discarded — + as will Azure Managed Redis be, once its rollout lands and the option is set for it. Both are handled now, and a source neither of them models is recorded from + the base event rather than dropped, so a provider added upstream is visible rather than silent. + `RedisConnectionOptions.MaintenanceNotifications` opts in, defaulting to the client's own behaviour. It takes + `RedisMaintenanceNotifications`, a neutral enum of this library's own, because the StackExchange type is marked + experimental and would otherwise raise `SER010` in every consumer that set the option. A value outside the + enum is refused with an `InvalidOperationException` — it arrives from configuration, not as an argument. A notification is recorded + whichever connection delivered it, the one carrying commands reaching this through the new + `IRedisConnector.ServerMaintenance`, and it can be delivered more than once -- Azure's is a broadcast every + connection receives, and a push frame is replayed to a connection that reconnects. A copy matching one recorded in + the last 30 seconds is therefore dropped, on the notification's own identity rather than on which connection ought + to have had it. They are recorded rather than acted on — the client relaxes timeouts and hands the connection off itself, and probing would force a reconnect + against that — so `InProgress` is still driven by the Azure route alone. + +- `IRedisConnector.ServerMaintenance`, the maintenance the server announced on the connection carrying commands. + Defaulted to never raising, so an existing implementer is unaffected. It exists because the connector is what + rebuilds that connection, so one subscription here survives a `ForceReconnect` where subscribing to the + multiplexer directly would not. - `IRedisConnector.GetPrimaries()`, the connected primaries a server-scoped command such as `SCAN` has to be sent to one by one. Defaulted to an empty sequence, so an existing implementer neither breaks nor changes behaviour. ### Changed +- **A maintenance handoff is no longer reported as a connection failure.** When the client moves off an endpoint the + server said is going away, it raises `ConnectionFailed` with `ConnectionFailureType.MaintenanceHandoff`. That was + tracked as `Redis.ConnectionFailed` — by `RedisConnector` and again by `ConnectionStateMonitor` — which would + alert on exactly the event advance notice exists to make uneventful. Both now track it as + `Redis.MaintenanceHandoff`. The event is still raised to subscribers — the connection did drop — and only the + telemetry name distinguishes them. +- `RedisHealthCheck` reports "Redis maintenance in progress" rather than naming Azure Cache for Redis. The wording + is provider-neutral in readiness for the push route; the state behind it is not yet, since only the Azure route + opens it here. - Bumped `StackExchange.Redis` from 3.2.1 to 3.3.0. No public API change here, and nothing in this repository calls an API 3.2.15 or 3.3.0 altered. Three things in the range are worth knowing: - `SwitchPrimary` now retires the servers its rebuild drops (upstream #3225). They previously stayed in the server @@ -21,15 +50,78 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) longer part of the service, and occasionally the same address twice. - `IServer.Execute` supplies the configured default database rather than refusing a database-specific command (upstream #3237). The maintainer passes the database explicitly regardless, so it does not depend on this. - - Server-native maintenance notifications arrive as opt-in (`maintNotifications=Auto`, or - `ConfigurationOptions.MaintenanceNotifications`). Left off: `RedisPlannedMaintenance` continues to drive itself - from `ServerMaintenanceEvent`/`AzureMaintenanceEvent`, which is unchanged, and adopting the native path is a - behavioural decision rather than part of a version bump. + - Server-native maintenance notifications arrive as opt-in, through `maintNotifications` on the connection string + or `ConfigurationOptions.MaintenanceNotifications`. Surfacing them is a behavioural decision rather than part of + a version bump, and is covered by the `MaintenanceNotifications` entry under **Added** above. - Dropped the `SER007` suppression in `RedisStreamSubjectWriterTests`. `RedisErrorKind` is no longer marked `[Experimental]` in 3.3.0, so the pragma suppressed a diagnostic that is no longer raised. ### Fixed +- **A throwing subscriber cost the remaining ones their notification.** `RedisConnector` and + `ConnectionStateMonitor` raised their connection events with a plain multicast invoke, which stops at the first + handler that throws; `ForceReconnect` caught the exception, but around the whole invocation list rather than around + each handler, so the outcome was the same. Reconnection is how the pub/sub writer, the stream notify channel and the + stream subject writer re-subscribe, and the pub/sub writer's handler can throw `ObjectDisposedException` when a + dispose races the notification — one such throw left every subscriber after it in the list detached until the next + reconnect. Handlers are now invoked one at a time with per-handler tracking, on every event of both types. This also + covers the new `ServerMaintenance`, which the client raises on its own dispatch thread and which had no guard at + all, and `ConnectionStateMonitor`, whose own re-multicast to `IConnectionState` subscribers sits inside the + connector's — so a throw from one of those aborted the connector's list as well. `RedisPlannedMaintenance`'s own + maintenance connection is guarded too: its handler is attached straight to the client rather than reaching this + class through the connector, so nothing else would have kept a throw off that dispatch path. Reporting the + caught failure is itself guarded, since a sink that cannot take the report would otherwise put the exception + back on the path the boundary exists to keep clear. + +- **A notification whose recording threw was suppressed for the next 30 seconds.** `RedisPlannedMaintenance` + claimed a notification's identity in its deduplication set before handling it, so a handler that threw -- + starting the Azure probe loop, or emitting the `Redis.Maintenance` event -- recorded nothing while every + matching copy was collapsed into a record that did not exist. The claim is still taken up front, so a + concurrent copy from the other route still collapses, but it is committed to the retention queue only once + the recording succeeded and released otherwise. It is timestamped on commit rather than on claim, so the + queue stays ordered by expiry when two routes record concurrently. Cancelling the service is guarded the same + way: `CancellationTokenSource.Cancel` runs its registrations inline, so it raises what a caller's callback + threw, and that left `StopReacting` before the maintenance connection had been let go of. A probe run now marks + itself started only once `Redis.MaintenanceStarted` has been emitted, so a run that could not announce its + start no longer emits `Redis.MaintenanceEnded` against nothing, and clears `InProgress` ahead of the rest of + its cleanup, since leaving it set would make the guard refuse every later run. Reporting a failed probe no + longer precedes the `ForceReconnect` it must not cost, and neither interval announcement can end the run: + a sink that refused `Redis.MaintenanceStarted` used to kill the worker before it reached the probe loop, + so the Azure route stopped recovering at all. + +- **A rejected connection candidate stayed the newest multiplexer.** `RedisConnector` marks each candidate as + the incoming generation before its handlers go on, so a `MOVING` arriving before the swap is still taken. + A candidate the swap then rejected — a disposal or a newer reconnect won the race — was disposed without + clearing that marker, leaving a disposed multiplexer referenced until the next reconnect replaced it. The + marker is now compare-and-cleared as that instance is disposed, before its handlers come off. The maintenance + handler also closes over the connection it was attached to rather than reading the event's sender, since a + composite multiplexer can attach it to its children and raise with a child, which belongs to no generation + the connector knows about — a `MOVING` for the connection carrying commands would have been dropped. The + raiser is still used, for the opposite question: a group attaches the handler to every member, so a `MOVING` + is taken only when it came from the member currently carrying commands. `IConnectionGroup.ActiveMember` is + public but `ConnectionGroupMember.Multiplexer` is not, so that link is read reflectively and the check is + skipped whenever it cannot be made — a spare notice costs a record, a lost one costs the handoff. + +- **A refused telemetry report could defeat the catch that made it.** Every `catch` in `RedisConnector` and + `RedisPlannedMaintenance` exists so the surrounding work carries on, and each reported through the sink + directly — so a sink that threw took the recovery with it. They report through the guarded helper now. The + worst of them was `GetVersion`, the factory behind a `Lazy`: a `Lazy` caches what its factory threw, + so `IRedisConnector.Version` would have thrown for the life of the process instead of falling back once. + The same holds for recording an event where recovery follows it: a refused `Redis.ForcedReconnect` skipped + both the `OnReconnected` multicast and disposal of the retired connection, and a refused + `Redis.StaleEndpointDetected` or `Redis.HangDetected` skipped the `ForceReconnect` it exists to announce. + Every record raised on StackExchange.Redis's own dispatch thread is guarded for the same reason, and + `ConnectionStateMonitor` guards its records in one place, since each precedes both a state reset and a + multicast to its own subscribers. + +- **Azure notifications the client could not parse shared one identity.** Such a payload leaves every field at + its default, `RawMessage` included, so the deduplication key was the same for all of them and two unrelated + ones arriving within the retention window collapsed into one record. They are recorded individually now, the + way a push frame whose sequence could not be read already was. A notification whose *type* is unrecognised + but whose other fields parsed still has an identity and still collapses. `Redis.Maintenance` also records + `ReceivedTimeUtc` and `StartTimeUtc` in the round-trip format on every route, rather than the general + invariant pattern on two of the three, so a query over the field does not have to guess which route wrote it. + - **A consumer group with a stale last-delivered-id skipped its quarantine.** The maintainer quarantines such a group by recording the instant in a hash field and deleting it once `MaintainerQuarantineInterval` has passed — which is what `CheckEmptyStreamGroupAsync` does for a group with no consumers. The stale-last-delivered-id path wrote the diff --git a/docs/how-to/resilience.md b/docs/how-to/resilience.md index 0cf2f0ad..cf28ad03 100644 --- a/docs/how-to/resilience.md +++ b/docs/how-to/resilience.md @@ -423,6 +423,7 @@ old one (`Redis.ForcedReconnect` event, `OnReconnected` raised). |---|---|---| | Hang detection | More than 100 commands awaiting a reply on the primary with no read or write for `LastWrite/ReadIntervalThresholdMilliseconds` | `EnableHangDetection`, `HangDetectionDueTime`, `HangDetectionPeriod` | | Planned maintenance | `NodeMaintenanceStarting` on the `AzureRedisEvents` channel; probes with a write every second for 10 minutes and reconnects on failure | `PlannedMaintenanceEnabled` | +| Announced maintenance | RESP3 push notifications on the command connection, from Redis Enterprise and Redis Cloud (Azure Managed Redis once its rollout lands); recorded, while the client relaxes timeouts and hands the connection off | `PlannedMaintenanceEnabled`, `MaintenanceNotifications` | | Stale endpoint detection | A topology-discovered node has been disconnected for `StaleEndpointThreshold` and is no longer in the cluster topology the client refreshes | `EnableStaleEndpointDetection`, `StaleEndpointThreshold`, `StaleEndpointScanInterval` | **Stale endpoints** are the clustered-cache failure mode. StackExchange.Redis discovers the @@ -456,12 +457,41 @@ refreshes, so that a single lost topology reply is retried instead, the scan emi `Redis.StaleEndpointScanDisabled` event and stops for the lifetime of the connector, rather than reporting the same failure every interval. -**Which Azure offering sends maintenance events.** The `AzureRedisEvents` channel exists on -Azure Cache for Redis Basic, Standard and Premium only. Azure Managed Redis (`*.redis.azure.net`) -does not publish it, so on that service `PlannedMaintenanceEnabled` never fires and the -planned-maintenance state never reports in-progress; the stale-endpoint scan and hang detection -are what recover a connection there. Microsoft's own guidance for Azure Managed Redis is the same -ForceReconnect pattern: recreate the multiplexer when errors persist past a threshold. +**Which offering sends maintenance events, and how.** There are two routes, and the difference +decides what this library does about them. + +Azure Cache for Redis Basic, Standard and Premium publish on the `AzureRedisEvents` pub/sub +channel. The server announces that a node is going away but hands nothing off, so +`NodeMaintenanceStarting` starts the probe loop above: write every second for ten minutes, and +force a reconnect when a write fails. + +Redis Enterprise and Redis Cloud instead send RESP3 push notifications on the connection carrying +your commands, and none of them publish `AzureRedisEvents`. The client acts on these itself — +relaxing timeouts, re-reading topology, moving off a departing endpoint — so this library records +them and leaves the recovery alone: probing force-reconnects on a failed write, which would fight +the handoff. Azure Managed Redis (`*.redis.azure.net`) is recognised as a provider but nothing +turns the request on for it, so `MaintenanceNotifications` below is what asks. Reporting the +disruption through `InProgress` is a separate change. + +A notification can arrive more than once: Azure's is a broadcast every connection receives, and a +push frame is replayed to a connection that reconnects, which the client collapses only within the +multiplexer that received it. So both routes record, and a copy matching one seen in the last 30 +seconds is dropped — keyed on the notification's own identity (the fields parsed from Azure's +payload, or a push frame's type and sequence id) and measured on timestamps rather than the wall +clock. Two kinds are never collapsed, a duplicate costing less than a loss: a frame whose sequence +could not be read, reported as zero and told apart from a genuine zero by the `seq=?` in its +description; and a source this library does not model, whose payload carries no uniqueness +contract. + +Two asymmetries remain. A push frame on the planned-maintenance connection is ignored, since that +connection carries no commands. And only a `MOVING` is tied to a connection generation — the one +carrying commands or the one about to, since a rebuild subscribes the replacement before +publishing it and the server never replays a `MOVING`. + +A handoff surfaces as a `ConnectionFailed` event with +`ConnectionFailureType.MaintenanceHandoff`. It is tracked as `Redis.MaintenanceHandoff` rather +than `Redis.ConnectionFailed`, so planned maintenance does not raise a failure alert, but it is +still raised to subscribers of `OnConnectionFailed` — the connection did drop. ### Don't roll your own diff --git a/docs/recipes/redis-health-check.md b/docs/recipes/redis-health-check.md index ee0d01cf..9db67e13 100644 --- a/docs/recipes/redis-health-check.md +++ b/docs/recipes/redis-health-check.md @@ -36,7 +36,7 @@ A 3-second timeout is the conventional value — longer than a typical ping (sub The healthy result carries the multiplexer's `IsConnected`, `IsConnecting`, `OperationCount`, `Status` and `DisconnectedEndPoints` (a `;`-joined `host:port` list) in its data, so a probe that is green can still show a node the multiplexer cannot reach. A discovered node that stays in that list after a cluster patch is what [stale endpoint detection](../how-to/resilience.md#redis-connection-self-healing) removes by rebuilding the connection; the health check does not need to fail for that to happen. -`IRedisPlannedMaintenance.InProgress` only ever becomes `true` on Azure Cache for Redis Basic/Standard/Premium, the tiers that publish the `AzureRedisEvents` channel. On Azure Managed Redis the check behaves as if no maintenance tracker were registered. +`IRedisPlannedMaintenance.InProgress` becomes `true` on one route only. Azure Cache for Redis Basic/Standard/Premium publish the `AzureRedisEvents` channel, and a node going away there starts the probe loop that holds the state open. Redis Enterprise and Redis Cloud instead send RESP3 push notifications on the connection carrying commands; those are recorded but do not yet move this state, so on that route the check behaves as if no maintenance tracker were registered. Azure Managed Redis is recognised as a push provider but its servers do not emit these yet either. ## When not to use diff --git a/docs/reference/settings.md b/docs/reference/settings.md index 902b95fc..eb9a19ee 100644 --- a/docs/reference/settings.md +++ b/docs/reference/settings.md @@ -55,6 +55,7 @@ Every binding-visible property on every shipped options class, with shipped defa | `HeartbeatInterval` | `TimeSpan?` | `null` | App-wide | `null` = StackExchange.Redis default; TimeSpan override for the heartbeat period. | | `ProfilerFeatureFlagKey` | `string` | `"RedisProfiler.Enabled"` | App-wide | Feature-flag key consulted before enabling the StackExchange.Redis command profiler. | | `PlannedMaintenanceEnabled` | `bool` | `true` | App-wide | Tolerate planned-maintenance disconnects gracefully instead of faulting. | +| `MaintenanceNotifications` | `RedisMaintenanceNotifications?` | `null` | App-wide | Ask the server for advance notice of maintenance. `null` leaves the client's own default, so provider enlistment applies as it lands upstream. `Auto` asks and connects normally if the server does not offer them. `Required` refuses a connection that will not deliver them — useful to prove the feature is live. Delivered as RESP3 push notifications by Redis Enterprise and Redis Cloud; Azure Managed Redis is recognised as that kind of provider but its servers do not emit them yet, so the opt-in is preparatory there. Azure Cache for Redis uses the `AzureRedisEvents` channel instead and needs no opt-in. On a server that does not emit them — Redis OSS, Valkey, AMR for now — `Auto` is refused and the connection carries on, while `Required` rejects it. One exception, and it is the client's: inside a multi-group (geo-redundant) connection the feature is not activated at all, so `Required` warns and connects rather than rejecting — failing there would leave a group with no way to opt in, and the restriction is expected to be lifted upstream. Requires RESP3, which is negotiated by default. | | `PlannedMaintenanceConnectionRetryCount` | `int` | `5` | App-wide | Attempts to establish the planned-maintenance subscription before backing off to quiet retries; failures are logged as warnings, never faulting startup. | | `PlannedMaintenanceConnectionRetryDelay` | `TimeSpan` | `00:00:05` | App-wide | Delay between planned-maintenance subscription attempts (negative/zero is clamped to 1s). | | `LogConnectionFailedEvents` | `bool` | `true` | App-wide | Log `ConnectionFailed` events from the multiplexer. | diff --git a/samples/UiPath.Caching.Sample/appsettings.all.json b/samples/UiPath.Caching.Sample/appsettings.all.json index 76debe11..22c44e3b 100644 --- a/samples/UiPath.Caching.Sample/appsettings.all.json +++ b/samples/UiPath.Caching.Sample/appsettings.all.json @@ -54,6 +54,13 @@ "ProfilerFeatureFlagKey": "RedisProfiler.Enabled", // PlannedMaintenanceEnabled: tolerate planned-maintenance disconnects gracefully "PlannedMaintenanceEnabled": true, + // MaintenanceNotifications: ask the server for advance notice of maintenance. + // null (omitted) leaves the client's default; "Auto" asks and connects anyway if + // unsupported; "Required" rejects a connection that will not deliver them, so use it + // only where the server is known to emit them (Redis Enterprise, Redis Cloud). + // A multi-group (geo-redundant) connection is the exception: the client does not + // activate the feature there, so "Required" warns and connects rather than rejecting. + "MaintenanceNotifications": null, // LogConnectionFailedEvents: log ConnectionFailed events from the multiplexer "LogConnectionFailedEvents": true, // LogConnectionRestoredEvents: log ConnectionRestored events from the multiplexer diff --git a/src/UiPath.Caching/PublicAPI.Unshipped.txt b/src/UiPath.Caching/PublicAPI.Unshipped.txt index c92ba29c..acb709d3 100644 --- a/src/UiPath.Caching/PublicAPI.Unshipped.txt +++ b/src/UiPath.Caching/PublicAPI.Unshipped.txt @@ -1,3 +1,12 @@ #nullable enable UiPath.Caching.Redis.IRedisConnector.GetPrimaries() -> System.Collections.Generic.IEnumerable! +UiPath.Caching.Redis.IRedisConnector.ServerMaintenance -> System.EventHandler? +UiPath.Caching.Redis.RedisConnectionOptions.MaintenanceNotifications.get -> UiPath.Caching.Redis.RedisMaintenanceNotifications? +UiPath.Caching.Redis.RedisConnectionOptions.MaintenanceNotifications.set -> void UiPath.Caching.Redis.RedisConnector.GetPrimaries() -> System.Collections.Generic.IEnumerable! +UiPath.Caching.Redis.RedisConnector.ServerMaintenance -> System.EventHandler? +UiPath.Caching.Redis.RedisMaintenanceNotifications +UiPath.Caching.Redis.RedisMaintenanceNotifications.Auto = 1 -> UiPath.Caching.Redis.RedisMaintenanceNotifications +UiPath.Caching.Redis.RedisMaintenanceNotifications.Disabled = 0 -> UiPath.Caching.Redis.RedisMaintenanceNotifications +UiPath.Caching.Redis.RedisMaintenanceNotifications.Required = 2 -> UiPath.Caching.Redis.RedisMaintenanceNotifications +UiPath.Caching.Redis.RedisPlannedMaintenance.RedisPlannedMaintenance(UiPath.Caching.Telemetry.ICachingTelemetryProvider! telemetryProvider, UiPath.Caching.Redis.IRedisConnector! redisConnector, UiPath.Caching.Redis.IRedisConfigurationOptionsProvider! redisConfigurationOptionsProvider, UiPath.Caching.Redis.IConnectionMultiplexerFactory! connectionMultiplexerFactory, Microsoft.Extensions.Logging.ILogger! logger, Microsoft.Extensions.Options.IOptions! options, System.Collections.Generic.IEnumerable? configurators, System.TimeProvider! clock) -> void diff --git a/src/UiPath.Caching/Redis/ConnectionStateMonitor.cs b/src/UiPath.Caching/Redis/ConnectionStateMonitor.cs index 2643c73f..7ad3d0c5 100644 --- a/src/UiPath.Caching/Redis/ConnectionStateMonitor.cs +++ b/src/UiPath.Caching/Redis/ConnectionStateMonitor.cs @@ -7,6 +7,7 @@ public sealed class ConnectionStateMonitor : IConnectionState, IDisposable { private const string EventConnectionRestored = "Redis.ConnectionRestored"; private const string EventConnectionFailed = "Redis.ConnectionFailed"; + private const string EventMaintenanceHandoff = "Redis.MaintenanceHandoff"; private const string EventReconnected = "Redis.Reconnected"; private const string EventEvaluateConnected = "Redis.EvaluateConnected"; private const string PropNow = "Now"; @@ -58,21 +59,24 @@ private void InternalOnConnectionRestored(object? sender, EventArgs e) { TrackEvent(EventConnectionRestored); ResetIsConnected(); - OnConnectionRestored?.Invoke(this, EventArgs.Empty); + OnConnectionRestored.TryRaise(_telemetryProvider, handler => handler(this, EventArgs.Empty)); } private void InternalOnConnectionFailed(object? sender, EventArgs e) { - TrackEvent(EventConnectionFailed); + // Classified again here: the monitor sees the same args on the way to subscribers. + TrackEvent(e is ConnectionFailedEventArgs { FailureType: ConnectionFailureType.MaintenanceHandoff } + ? EventMaintenanceHandoff + : EventConnectionFailed); ResetIsConnected(); - OnConnectionFailed?.Invoke(sender, e); + OnConnectionFailed.TryRaise(_telemetryProvider, handler => handler(sender, e)); } private void InternalOnReconnected(object? sender, EventArgs e) { TrackEvent(EventReconnected); ResetIsConnected(); - OnReconnected?.Invoke(sender, e); + OnReconnected.TryRaise(_telemetryProvider, handler => handler(sender, e)); } private void ResetIsConnected(bool addTimer = true) { @@ -106,6 +110,8 @@ private void TrackEvent(string eventName, params KeyValuePair[] var properties = new KeyValuePair[data.Length + 1]; properties[0] = new(PropNow, Environment.TickCount.ToString(CultureInfo.InvariantCulture)); Array.Copy(data, 0, properties, 1, data.Length); - _telemetryProvider.TrackEvent(eventName, properties); + // Guarded here rather than at each call site: every one of them is followed by state to reset and + // subscribers to notify, and one of them runs inside the connector's own multicast. + _telemetryProvider.TryTrackEvent(eventName, properties); } } diff --git a/src/UiPath.Caching/Redis/IRedisConnector.cs b/src/UiPath.Caching/Redis/IRedisConnector.cs index 47a7b49c..e0327f72 100644 --- a/src/UiPath.Caching/Redis/IRedisConnector.cs +++ b/src/UiPath.Caching/Redis/IRedisConnector.cs @@ -1,9 +1,18 @@ using System.Net; +using StackExchange.Redis.Maintenance; namespace UiPath.Caching.Redis; public interface IRedisConnector : IConnectionState, IDisposable { + /// Maintenance the server announced on the connection carrying commands. + event EventHandler? ServerMaintenance + { + // Defaulted to never raising, so an existing implementer neither breaks nor starts forwarding. + add => _ = value; + remove => _ = value; + } + Version Version { get; } IDatabase Database { get; } diff --git a/src/UiPath.Caching/Redis/RedisConfigurationOptionsProvider.cs b/src/UiPath.Caching/Redis/RedisConfigurationOptionsProvider.cs index 2d0e93bb..a07d88ae 100644 --- a/src/UiPath.Caching/Redis/RedisConfigurationOptionsProvider.cs +++ b/src/UiPath.Caching/Redis/RedisConfigurationOptionsProvider.cs @@ -21,10 +21,14 @@ public ConfigurationOptions GetConfiguration() if (sb.Length == 0) { - return new ConfigurationOptions + // With no connection string these options are what a supplied ConnectionFactory is handed, so the + // mapping still has to run. + var supplied = new ConfigurationOptions { LoggerFactory = loggerFactory, }; + ApplyMaintenanceNotifications(supplied); + return supplied; } var config = ConfigurationOptions.Parse(sb.ToString()); @@ -40,6 +44,8 @@ public ConfigurationOptions GetConfiguration() config.ReconnectRetryPolicy = new ExponentialRetry(_options.BackOffMilliseconds); } + ApplyMaintenanceNotifications(config); + if (_options.HeartbeatConsistencyChecks.HasValue) { config.HeartbeatConsistencyChecks = _options.HeartbeatConsistencyChecks.Value; @@ -57,4 +63,24 @@ public ConfigurationOptions GetConfiguration() return config; } + private void ApplyMaintenanceNotifications(ConfigurationOptions config) + { + if (_options.MaintenanceNotifications is not { } notifications) + { + return; + } + + // The suppression stops here: SER010 rides on the type, so exposing StackExchange's enum would raise it + // in every consumer that sets the option. +#pragma warning disable SER010 // Server-native maintenance notifications are for evaluation purposes only + config.MaintenanceNotifications = notifications switch + { + RedisMaintenanceNotifications.Disabled => MaintenanceNotificationMode.Disabled, + RedisMaintenanceNotifications.Auto => MaintenanceNotificationMode.Auto, + RedisMaintenanceNotifications.Required => MaintenanceNotificationMode.Enabled, + _ => throw new InvalidOperationException($"Unsupported {nameof(RedisMaintenanceNotifications)} value '{notifications}'."), + }; +#pragma warning restore SER010 + } + } diff --git a/src/UiPath.Caching/Redis/RedisConnectionOptions.cs b/src/UiPath.Caching/Redis/RedisConnectionOptions.cs index c5be5f08..a8666c4d 100644 --- a/src/UiPath.Caching/Redis/RedisConnectionOptions.cs +++ b/src/UiPath.Caching/Redis/RedisConnectionOptions.cs @@ -22,6 +22,9 @@ public class RedisConnectionOptions public bool PlannedMaintenanceEnabled { get; set; } = true; + /// Ask the server for advance notice of maintenance; null leaves the client's own default. + public RedisMaintenanceNotifications? MaintenanceNotifications { get; set; } + public int PlannedMaintenanceConnectionRetryCount { get; set; } = 5; public TimeSpan PlannedMaintenanceConnectionRetryDelay { get; set; } = TimeSpan.FromSeconds(5); diff --git a/src/UiPath.Caching/Redis/RedisConnector.cs b/src/UiPath.Caching/Redis/RedisConnector.cs index 9d30301e..9c2f4fa1 100644 --- a/src/UiPath.Caching/Redis/RedisConnector.cs +++ b/src/UiPath.Caching/Redis/RedisConnector.cs @@ -1,6 +1,9 @@ using System.Globalization; using System.Net; using System.Reflection; +using System.Runtime.CompilerServices; +using StackExchange.Redis.Availability; +using StackExchange.Redis.Maintenance; using UiPath.Caching.Telemetry; namespace UiPath.Caching.Redis; @@ -13,8 +16,11 @@ public sealed class RedisConnector : IRedisConnector /// Most scan intervals skipped after a failing refresh: an hour at the default interval. internal const int MaxScanBackoff = 120; + private static PropertyInfo? _memberConnection; + private readonly RedisConnectionOptions _redisOptions; private readonly ICachingTelemetryProvider _telemetryProvider; + private readonly IRedisConfigurationOptionsProvider _redisConfigurationOptionsProvider; private readonly IConnectionMultiplexerFactory _connectionMultiplexerFactory; private readonly IEnumerable? _configurators; @@ -28,7 +34,16 @@ public sealed class RedisConnector : IRedisConnector private readonly Lazy _version; private readonly object _swapLock = new(); + // The handler closes over the multiplexer it was attached to, and unsubscribing needs that same delegate. + private readonly ConditionalWeakTable> _maintenanceHandlers = []; + private volatile Lazy> _lazyCacheConnectionMultiplexer; + private IConnectionMultiplexer? _newestMultiplexer; + + // Swapped in tests: the reflective default cannot be driven from outside, because a member's connection is + // a sealed ConnectionMultiplexer that cannot be substituted. + private Func _activeMemberConnection = ResolveActiveMemberConnection; + private volatile bool _disposed; private volatile bool _staleScanDisabled; private int _reconnecting; @@ -93,6 +108,8 @@ internal RedisConnector(ICachingTelemetryProvider telemetryProvider, public event EventHandler? OnConnectionFailed; + public event EventHandler? ServerMaintenance; + public event EventHandler? OnConnectionRestored; public event EventHandler? OnReconnected; @@ -206,7 +223,7 @@ internal async Task ScanStaleEndpointsAsync() return; } - _telemetryProvider.TrackEvent( + _telemetryProvider.TryTrackEvent( "Redis.StaleEndpointDetected", [ new("EndPoints", string.Join(";", stale.Select(FormatEndPoint))), @@ -217,7 +234,7 @@ internal async Task ScanStaleEndpointsAsync() catch (Exception ex) { _scansToSkip = Math.Min(1 << Math.Min(++_scanFailures, 7), MaxScanBackoff); // a handshake that never completes fails every refresh; back off instead of tracking it every interval - _telemetryProvider.TrackException(ex); + _telemetryProvider.TryTrackException(ex); } finally { @@ -277,6 +294,9 @@ internal async Task RefreshClusterMembershipAsync(IConnection return ++_nullTopologyRefreshes >= NullTopologyRefreshLimit ? new(Conclusive: true, Members: null) : ClusterMembership.Inconclusive; } + /// Test seam: a member's connection is a sealed type that cannot be substituted. + internal void SetActiveMemberConnectionResolver(Func resolver) => _activeMemberConnection = resolver; + #pragma warning disable IDE0079 // Remove unnecessary suppression [SuppressMessage("SonarQube", "S3011:Reflection should not be used to create instances of types", Justification = "By design")] #pragma warning restore IDE0079 // Remove unnecessary suppression @@ -311,12 +331,29 @@ internal async Task RefreshClusterMembershipAsync(IConnection } catch (Exception ex) { - _telemetryProvider.TrackException(ex); + _telemetryProvider.TryTrackException(ex); return null; } } [ExcludeFromCodeCoverage(Justification = "Only called from the excluded OnInternalConnection* event handlers.")] +#pragma warning disable IDE0079 // Remove unnecessary suppression + [SuppressMessage("SonarQube", "S3011:Reflection should not be used to create instances of types", Justification = "By design")] +#pragma warning restore IDE0079 // Remove unnecessary suppression + private static object? ResolveActiveMemberConnection(IConnectionGroup group) + { + if (group.ActiveMember is not { } member) + { + return null; + } + + // ConnectionGroupMember.Multiplexer is internal upstream; raised with StackExchange.Redis. + var accessor = _memberConnection ??= member.GetType().GetProperty( + "Multiplexer", + BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic); + return accessor?.GetValue(member); + } + private static KeyValuePair[] GetEventProperties(ConnectionFailedEventArgs e) => [ new(nameof(e.EndPoint), e.EndPoint?.ToString() ?? string.Empty), @@ -389,7 +426,7 @@ private void ForceReconnect(Lazy> current) } catch (Exception ex) { - _telemetryProvider.TrackException(ex); + _telemetryProvider.TryTrackException(ex); return; } @@ -408,16 +445,10 @@ private void ForceReconnect(Lazy> current) _lazyCacheConnectionMultiplexer = swapped; } - _telemetryProvider.TrackEvent("Redis.ForcedReconnect"); + // A refused record must not cost the reconnect: the multicast and the retired connection follow. + _telemetryProvider.TryTrackEvent("Redis.ForcedReconnect"); - try - { - OnReconnected?.Invoke(this, EventArgs.Empty); - } - catch (Exception ex) - { - _telemetryProvider.TrackException(ex); - } + OnReconnected.TryRaise(_telemetryProvider, handler => handler(this, EventArgs.Empty)); await CloseAndDisposeAsync(previousTask).ConfigureAwait(false); } @@ -437,7 +468,7 @@ private async Task CloseAndDisposeAsync(Task multiplexer } catch (Exception ex) { - _telemetryProvider.TrackException(ex); + _telemetryProvider.TryTrackException(ex); return; } @@ -447,7 +478,7 @@ private async Task CloseAndDisposeAsync(Task multiplexer } catch (Exception ex) { - _telemetryProvider.TrackException(ex); + _telemetryProvider.TryTrackException(ex); } finally { @@ -496,7 +527,7 @@ Version DefaultVersion() } catch (Exception ex) { - _telemetryProvider.TrackException(ex); + _telemetryProvider.TryTrackException(ex); return DefaultVersion(); } } @@ -650,7 +681,7 @@ private void TryDisposeMultiplexer(IConnectionMultiplexer multiplexer) } catch (Exception ex) { - _telemetryProvider.TrackException(ex); + _telemetryProvider.TryTrackException(ex); } } @@ -706,7 +737,7 @@ private void OnHangScan() _redisOptions.LastWriteIntervalThresholdMilliseconds, _redisOptions.LastReadIntervalThresholdMilliseconds)) { - _telemetryProvider.TrackEvent( + _telemetryProvider.TryTrackEvent( "Redis.HangDetected", [ new("Now", now.ToString(CultureInfo.InvariantCulture)), @@ -723,8 +754,16 @@ private void OnHangScan() [ExcludeFromCodeCoverage(Justification = "Wires multiplexer event handlers (ConnectionFailed/Restored/InternalError/ErrorMessage) — fires only from real StackExchange.Redis multiplexer events.")] private IConnectionMultiplexer ConfigureMultiplexerEvents(IConnectionMultiplexer multiplexer) { + // Before the handlers go on, not after the swap: a MOVING arriving while the retired one is still + // current says where this connection is going, and the server never replays it. + Volatile.Write(ref _newestMultiplexer, multiplexer); multiplexer.ConnectionFailed += OnInternalConnectionFailed; multiplexer.ConnectionRestored += OnInternalConnectionRestored; + // Closed over the multiplexer rather than reading the sender: a composite connection can attach this + // to its children and raise with a child, which belongs to no generation this knows about. + var maintenanceHandler = new EventHandler((raisedBy, e) => OnInternalServerMaintenance(multiplexer, raisedBy, e)); + _maintenanceHandlers.AddOrUpdate(multiplexer, maintenanceHandler); + multiplexer.ServerMaintenanceEvent += maintenanceHandler; if (_redisOptions.LogConnectionFailedEvents) { @@ -738,8 +777,16 @@ private IConnectionMultiplexer ConfigureMultiplexerEvents(IConnectionMultiplexer [ExcludeFromCodeCoverage(Justification = "Unwires multiplexer event handlers and disposes — only reached from ForceReconnect against a live multiplexer.")] private void DisposeMultiplexer(IConnectionMultiplexer multiplexer) { + // Before the handlers come off, so a candidate the swap rejected stops counting as incoming rather than + // staying the marker until the next one replaces it. Compare-and-clear: a newer candidate may own it. + Interlocked.CompareExchange(ref _newestMultiplexer, null, multiplexer); multiplexer.ConnectionFailed -= OnInternalConnectionFailed; multiplexer.ConnectionRestored -= OnInternalConnectionRestored; + if (_maintenanceHandlers.TryGetValue(multiplexer, out var maintenanceHandler)) + { + multiplexer.ServerMaintenanceEvent -= maintenanceHandler; + _maintenanceHandlers.Remove(multiplexer); + } if (_redisOptions.LogConnectionFailedEvents) { @@ -753,7 +800,7 @@ private void DisposeMultiplexer(IConnectionMultiplexer multiplexer) [ExcludeFromCodeCoverage(Justification = "Handler for IConnectionMultiplexer.InternalError — fires only from real Redis transport errors.")] private void OnInternalError(object? send, InternalErrorEventArgs e) { - _telemetryProvider.TrackEvent( + _telemetryProvider.TryTrackEvent( "Redis.InternalError", [ new("Endpoint", e.EndPoint?.ToString() ?? string.Empty), @@ -766,7 +813,7 @@ private void OnInternalError(object? send, InternalErrorEventArgs e) [ExcludeFromCodeCoverage(Justification = "Handler for IConnectionMultiplexer.ErrorMessage — fires only from real Redis-side error replies.")] private void OnInternalErrorMessage(object? send, RedisErrorEventArgs e) { - _telemetryProvider.TrackEvent( + _telemetryProvider.TryTrackEvent( "Redis.ErrorMessage", [ new("Endpoint", e.EndPoint?.ToString() ?? string.Empty), @@ -777,20 +824,77 @@ private void OnInternalErrorMessage(object? send, RedisErrorEventArgs e) [ExcludeFromCodeCoverage(Justification = "Handler for IConnectionMultiplexer.ConnectionRestored — fires only from a real reconnect event.")] private void OnInternalConnectionRestored(object? sender, ConnectionFailedEventArgs e) { - OnConnectionRestored?.Invoke(sender, e); + OnConnectionRestored.TryRaise(_telemetryProvider, handler => handler(sender, e)); if (_redisOptions.LogConnectionRestoredEvents) { - _telemetryProvider.TrackEvent("Redis.ConnectionRestored", GetEventProperties(e)); + _telemetryProvider.TryTrackEvent("Redis.ConnectionRestored", GetEventProperties(e)); } } + private void OnInternalServerMaintenance(IConnectionMultiplexer owner, object? raisedBy, ServerMaintenanceEvent e) + { + // Only a MOVING is scoped to the connection it arrived on, naming that connection's replacement. The + // rest are broadcast to every node, and a retired connection may be the only one that observed one -- + // so filtering those by generation could lose it, where letting them through costs at worst a duplicate. +#pragma warning disable SER010 // Server-native maintenance notifications are for evaluation purposes only + var connectionScoped = e is PushMaintenanceEvent { NotificationType: MaintenanceNotificationType.Moving }; +#pragma warning restore SER010 + if (connectionScoped && (!IsCurrentOrIncoming(owner) || !IsFromTheActiveGroupMember(owner, raisedBy))) + { + return; + } + + ServerMaintenance.TryRaise(_telemetryProvider, handler => handler(owner, e)); + } + + // A group hands our handler to each of its members, so a MOVING can arrive from one the cache is not using. + // Which member is active is public; the connection behind it is not, so it is read reflectively and the + // check is skipped whenever that fails -- a spare notice costs a record, a lost one costs the handoff. + private bool IsFromTheActiveGroupMember(IConnectionMultiplexer owner, object? raisedBy) + { + if (owner is not IConnectionGroup group) + { + return true; + } + + try + { + var active = _activeMemberConnection(group); + return active is null || ReferenceEquals(active, raisedBy); + } + catch (Exception ex) + { + _telemetryProvider.TryTrackException(ex); + return true; + } + } + + + // The generation carrying commands, or the one about to: the replacement is subscribed before it is published. + private bool IsCurrentOrIncoming(object? sender) + { + if (ReferenceEquals(Volatile.Read(ref _newestMultiplexer), sender)) + { + return true; + } + + var lazy = _lazyCacheConnectionMultiplexer; + return lazy.IsValueCreated + && lazy.Value.IsCompletedSuccessfully + && ReferenceEquals(lazy.Value.Result, sender); + } + [ExcludeFromCodeCoverage(Justification = "Handler for IConnectionMultiplexer.ConnectionFailed — fires only from a real connection-drop event.")] private void OnInternalConnectionFailed(object? sender, ConnectionFailedEventArgs e) { - OnConnectionFailed?.Invoke(sender, e); + // Still raised, since the connection did drop; only the telemetry separates a handoff from a failure, + // so planned maintenance does not raise the alert it exists to avoid. + OnConnectionFailed.TryRaise(_telemetryProvider, handler => handler(sender, e)); if (_redisOptions.LogConnectionFailedEvents) { - _telemetryProvider.TrackEvent("Redis.ConnectionFailed", GetEventProperties(e)); + _telemetryProvider.TryTrackEvent( + e.FailureType == ConnectionFailureType.MaintenanceHandoff ? "Redis.MaintenanceHandoff" : "Redis.ConnectionFailed", + GetEventProperties(e)); } } diff --git a/src/UiPath.Caching/Redis/RedisHealthCheck.cs b/src/UiPath.Caching/Redis/RedisHealthCheck.cs index ce4bfb0c..d659404a 100644 --- a/src/UiPath.Caching/Redis/RedisHealthCheck.cs +++ b/src/UiPath.Caching/Redis/RedisHealthCheck.cs @@ -19,7 +19,7 @@ public async Task CheckHealthAsync(HealthCheckContext context { if (_redisPlannedMaintenance?.InProgress ?? false) { - return new HealthCheckResult(HealthStatus.Healthy, "Azure Cache for Redis maintenance in progress"); + return new HealthCheckResult(HealthStatus.Healthy, "Redis maintenance in progress"); } var latency = await _redisConnector.Database.PingAsync(); diff --git a/src/UiPath.Caching/Redis/RedisMaintenanceNotifications.cs b/src/UiPath.Caching/Redis/RedisMaintenanceNotifications.cs new file mode 100644 index 00000000..03e88ae7 --- /dev/null +++ b/src/UiPath.Caching/Redis/RedisMaintenanceNotifications.cs @@ -0,0 +1,15 @@ +namespace UiPath.Caching.Redis; + +/// Whether to ask the server for advance notice of maintenance, where the server offers it. +public enum RedisMaintenanceNotifications +{ + /// Never ask. + Disabled = 0, + + /// Ask, and connect normally if the server does not offer them. + Auto = 1, + + /// Ask, and refuse the connection if the server will not deliver them. + /// Except inside a multi-group (geo-redundant) connection, where the client warns and connects. + Required = 2, +} diff --git a/src/UiPath.Caching/Redis/RedisPlannedMaintenance.cs b/src/UiPath.Caching/Redis/RedisPlannedMaintenance.cs index 5b551beb..5f07ff8a 100644 --- a/src/UiPath.Caching/Redis/RedisPlannedMaintenance.cs +++ b/src/UiPath.Caching/Redis/RedisPlannedMaintenance.cs @@ -8,6 +8,9 @@ namespace UiPath.Caching.Redis; [ExcludeFromCodeCoverage(Justification = "Wires up StackExchange.Redis ServerMaintenanceEvent — exercised only by real Azure Cache for Redis planned-maintenance notifications.")] public sealed class RedisPlannedMaintenance : IRedisPlannedMaintenance, IHostedService { + /// How long a notification stays recognisable as one already recorded, and so what bounds the set. + private static readonly TimeSpan SeenRetention = TimeSpan.FromSeconds(30); + private readonly ICachingTelemetryProvider _telemetryProvider; private readonly IRedisConnector _redisConnector; private readonly IRedisConfigurationOptionsProvider _redisConfigurationOptionsProvider; @@ -20,9 +23,16 @@ public sealed class RedisPlannedMaintenance : IRedisPlannedMaintenance, IHostedS private readonly TimeSpan _probeInterval = TimeSpan.FromSeconds(1); private readonly TimeSpan _hangingTime = TimeSpan.FromSeconds(10); private readonly object _lock = new(); + private readonly object _stateLock = new(); + private readonly object _seenLock = new(); + private readonly object _cancelLock = new(); + private readonly Queue<(string Raw, long At)> _seen = new(); + private readonly HashSet _seenIdentities = new(StringComparer.Ordinal); + private readonly TimeProvider _clock; private readonly CancellationTokenSource _cancellationTokenSource = new(); private IConnectionMultiplexer? _multiplexer; private volatile bool _disposed; + private bool _cancellationDisposed; private int _stopped; private long _maintenanceInProgress; @@ -34,7 +44,21 @@ public RedisPlannedMaintenance( ILogger logger, IOptions options, IEnumerable? configurators = null) + : this(telemetryProvider, redisConnector, redisConfigurationOptionsProvider, connectionMultiplexerFactory, logger, options, configurators, TimeProvider.System) + { + } + + public RedisPlannedMaintenance( + ICachingTelemetryProvider telemetryProvider, + IRedisConnector redisConnector, + IRedisConfigurationOptionsProvider redisConfigurationOptionsProvider, + IConnectionMultiplexerFactory connectionMultiplexerFactory, + ILogger logger, + IOptions options, + IEnumerable? configurators, + TimeProvider clock) { + _clock = clock; _telemetryProvider = telemetryProvider; _redisConnector = redisConnector; _redisConfigurationOptionsProvider = redisConfigurationOptionsProvider; @@ -54,13 +78,15 @@ public bool InProgress public Task StartAsync(CancellationToken cancellationToken) { + // Via the connector, so one subscription survives a ForceReconnect; the multiplexer's own would not. + _redisConnector.ServerMaintenance += OnServerMaintenance; _ = Task.Run(() => InitializeAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token); return Task.CompletedTask; } public Task StopAsync(CancellationToken cancellationToken) { - Cancel(); + StopReacting(); return Task.CompletedTask; } @@ -79,8 +105,9 @@ public void Dispose() _multiplexer = null; } - Cancel(); - _cancellationTokenSource.Dispose(); + // Dispose can be reached without StopAsync, so both paths share this rather than drifting apart. + StopReacting(multiplexer); + DisposeCancellation(); if (multiplexer is not null) { @@ -88,11 +115,90 @@ public void Dispose() } } + /// Stops the service reacting further; both the stop and the dispose path run it. + private void StopReacting(IConnectionMultiplexer? multiplexer = null) + { + _redisConnector.ServerMaintenance -= OnServerMaintenance; + + // Cancelled first, so _stopped is set before anything still in flight reaches a guard. + Cancel(); + + if (multiplexer is null) + { + lock (_lock) + { + multiplexer = _multiplexer; + } + } + + if (multiplexer is not null) + { + multiplexer.ServerMaintenanceEvent -= OnMaintenanceConnectionEvent; + } + + } + private void Cancel() { - if (Interlocked.Exchange(ref _stopped, 1) == 0) + bool first; + lock (_stateLock) + { + first = Interlocked.Exchange(ref _stopped, 1) == 0; + } + + if (!first) + { + return; + } + + // Outside _stateLock, since cancellation callbacks run inline and one taking that lock would deadlock. + // Under _cancelLock, since Dispose frees the source there: otherwise it could free it before this runs. + lock (_cancelLock) + { + if (_cancellationDisposed) + { + return; + } + + try + { + _cancellationTokenSource.Cancel(); + } + catch (Exception ex) + { + // Runs the registrations inline, so it throws what a caller did; stopping is not optional. + _telemetryProvider.TryTrackException(ex); + } + } + } + + /// + /// Cancels and frees the source as one step, so a concurrent cannot still be inside it. + /// Cancelling here too covers the case where this path won: the loser returns without cancelling. + /// + private void DisposeCancellation() + { + lock (_cancelLock) { - _cancellationTokenSource.Cancel(); + if (_cancellationDisposed) + { + return; + } + + _cancellationDisposed = true; + try + { + _cancellationTokenSource.Cancel(); + } + catch (Exception ex) + { + // Unreachable while Cancel guards its own: a second Cancel on a cancelled source runs nothing. + _telemetryProvider.TryTrackException(ex); + } + finally + { + _cancellationTokenSource.Dispose(); + } } } @@ -139,20 +245,24 @@ private async Task TryConnectAsync(CancellationToken cancellationToken) { var configuration = _redisConfigurationOptionsProvider.GetConfiguration(); await RedisConnectionConfigurators.ApplyAsync(configuration, _configurators, cancellationToken).ConfigureAwait(false); + var multiplexer = await _connectionMultiplexerFactory.CreateAsync(configuration, cancellationToken).ConfigureAwait(false); - multiplexer.ServerMaintenanceEvent += OnServerMaintenance; - bool disposed; + bool stopped; lock (_lock) { - disposed = _disposed; - if (!disposed) + // _stopped as well as _disposed: StopAsync can finish while CreateAsync is in flight, having + // already taken its snapshot, so publishing here would leave a subscribed connection behind it. + stopped = _disposed || Volatile.Read(ref _stopped) == 1; + if (!stopped) { + // Subscribed under the lock that publishes the field, so the two are never separately visible. + multiplexer.ServerMaintenanceEvent += OnMaintenanceConnectionEvent; _multiplexer = multiplexer; } } - if (disposed) + if (stopped) { TryDisposeMultiplexer(multiplexer); } @@ -162,22 +272,173 @@ private void TryDisposeMultiplexer(IConnectionMultiplexer multiplexer) { try { - multiplexer.ServerMaintenanceEvent -= OnServerMaintenance; + multiplexer.ServerMaintenanceEvent -= OnMaintenanceConnectionEvent; multiplexer.Dispose(); } catch (Exception ex) { - _telemetryProvider.TrackException(ex); + _telemetryProvider.TryTrackException(ex); + } + } + + // Push frames are ignored here: this connection carries no commands, so a MOVING on it names a replacement + // for a connection nothing is using. + private void OnMaintenanceConnectionEvent(object? sender, ServerMaintenanceEvent e) + { +#pragma warning disable SER010 // Server-native maintenance notifications are for evaluation purposes only + if (e is PushMaintenanceEvent) + { + return; + } +#pragma warning restore SER010 + + try + { + OnServerMaintenance(sender, e); + } + catch (Exception ex) + { + // Attached straight to the client, unlike the command route, so nothing else keeps throws off its dispatch. + _telemetryProvider.TryTrackException(ex); + } + } + + // Either route can deliver a copy of the same notification -- Azure's is a broadcast, and a push frame is + // replayed on reconnect, which the client collapses only within the multiplexer that received it. So the + // once-only claim is made on the notification's own identity, not on which connection ought to have had it. + // Claimed before the handler runs so a concurrent copy still collapses, committed only on success -- see Settle. + private bool TryClaim(ServerMaintenanceEvent e, out string? claim) + { + // Not RawMessage: the client parses Azure's payload into properties and leaves that null. Sequences are + // shared across types, so the type is part of a push frame's key. Nothing stamped per copy is in either. + static string? IdentityOf(ServerMaintenanceEvent e) => e switch + { + // A payload the client could not parse leaves every field at its default, RawMessage included, so two + // unrelated ones would share an identity. Recorded individually, like a frame with no sequence. The + // type string is part of that: the client keeps one it does not recognise, and that alone identifies. + AzureMaintenanceEvent { NotificationType: AzureNotificationType.Unknown, StartTimeUtc: null, IPAddress: null, SslPort: 0, NonSslPort: 0 } unparsed + when unparsed.NotificationTypeString == nameof(AzureNotificationType.Unknown) => null, + AzureMaintenanceEvent azureEvent => string.Create( + CultureInfo.InvariantCulture, + $"{azureEvent.NotificationTypeString}|{azureEvent.StartTimeUtc:O}|{azureEvent.IsReplica}|{azureEvent.IPAddress}|{azureEvent.SslPort}|{azureEvent.NonSslPort}"), +#pragma warning disable SER010 // Server-native maintenance notifications are for evaluation purposes only + // An unreadable sequence also surfaces as zero, and the client declines to collapse those -- but zero + // is a legitimate sequence too, and only the description separates them. If that wording ever changes + // these fall back to being keyed like any other, which is the milder way to be wrong. + PushMaintenanceEvent { RawMessage: { } description } + when description.Contains(" seq=?", StringComparison.Ordinal) => null, + PushMaintenanceEvent pushEvent => string.Create( + CultureInfo.InvariantCulture, + $"{pushEvent.NotificationType}|{pushEvent.SequenceId}"), +#pragma warning restore SER010 + // Nothing else has an identity to key on: RawMessage carries no uniqueness contract, and the + // fallback exists to keep an unknown source visible. + _ => null, + }; + + // Nothing to key on, so nothing to claim: always handled, never collapsed. + if (IdentityOf(e) is not { Length: > 0 } raw) + { + claim = null; + return true; + } + + lock (_seenLock) + { + // Timestamps rather than UtcNow, so a host clock correction neither holds entries past the retention + // nor drops the protection; read under the lock, since expiring from the head assumes that order. + var now = _clock.GetTimestamp(); + + // Nothing is enqueued twice, so an entry leaving the queue is the last of its identity. + while (_seen.Count > 0 && _clock.GetElapsedTime(_seen.Peek().At, now) > SeenRetention) + { + _seenIdentities.Remove(_seen.Dequeue().Raw); + } + + if (!_seenIdentities.Add(raw)) + { + claim = null; + return false; + } + + claim = raw; + return true; + } + } + + // A handler that threw recorded nothing, so holding its claim would collapse later copies into a record that + // does not exist. Timestamped on commit, not on claim, so the queue stays ordered when two routes race. + // A copy arriving while the first is still in Record is lost if that one then fails: holding it back means + // waiting on the client's dispatch thread, across a foreign telemetry call, for a retry that would fail too. + private void Settle(string? claim, bool recorded) + { + if (claim is null) + { + return; + } + + lock (_seenLock) + { + if (recorded) + { + _seen.Enqueue((claim, _clock.GetTimestamp())); + } + else + { + _seenIdentities.Remove(claim); + } } } private void OnServerMaintenance(object? sender, ServerMaintenanceEvent e) { - if (e is not AzureMaintenanceEvent azureEvent) + if (!TryClaim(e, out var claim)) { return; } + var recorded = false; + try + { + Record(e); + recorded = true; + } + finally + { + Settle(claim, recorded); + } + } + + private void Record(ServerMaintenanceEvent e) + { +#pragma warning disable SER010 // Server-native maintenance notifications are for evaluation purposes only + switch (e) + { + case AzureMaintenanceEvent azureEvent: + OnAzureMaintenance(azureEvent); + break; + case PushMaintenanceEvent pushEvent: + OnPushMaintenance(pushEvent); + break; + default: + // Recording the base properties keeps a source we do not model visible, rather than silent. + _telemetryProvider.TrackEvent( + "Redis.Maintenance", + [ + new("Source", e.GetType().Name), + new("ReceivedTimeUtc", e.ReceivedTimeUtc.ToString("O", CultureInfo.InvariantCulture)), + new("StartTimeUtc", e.StartTimeUtc?.ToString("O", CultureInfo.InvariantCulture) ?? string.Empty), + new("RawMessage", e.RawMessage ?? string.Empty), + ]); + break; + } +#pragma warning restore SER010 + } + + private void OnAzureMaintenance(AzureMaintenanceEvent azureEvent) + { + // Azure Cache for Redis announces the node going away but hands nothing off, so the connection has to be + // probed back into health. if (azureEvent.NotificationType == AzureNotificationType.NodeMaintenanceStarting) { StartConnectionProbing(); @@ -186,26 +447,55 @@ private void OnServerMaintenance(object? sender, ServerMaintenanceEvent e) _telemetryProvider.TrackEvent( "Redis.Maintenance", [ + new("Source", nameof(AzureMaintenanceEvent)), new("IPAddress", azureEvent.IPAddress?.ToString() ?? string.Empty), new("NotificationTypeString", azureEvent.NotificationTypeString), new("SslPort", azureEvent.SslPort.ToString(CultureInfo.InvariantCulture)), - new("ReceivedTimeUtc", azureEvent.ReceivedTimeUtc.ToString(CultureInfo.InvariantCulture)), - new("StartTimeUtc", azureEvent.StartTimeUtc?.ToString(CultureInfo.InvariantCulture) ?? string.Empty), + new("ReceivedTimeUtc", azureEvent.ReceivedTimeUtc.ToString("O", CultureInfo.InvariantCulture)), + new("StartTimeUtc", azureEvent.StartTimeUtc?.ToString("O", CultureInfo.InvariantCulture) ?? string.Empty), new("IsReplica", azureEvent.IsReplica.ToString(CultureInfo.InvariantCulture)), new("RawMessage", azureEvent.RawMessage ?? string.Empty), ]); } +#pragma warning disable SER010 // Server-native maintenance notifications are for evaluation purposes only + private void OnPushMaintenance(PushMaintenanceEvent pushEvent) + { + // Recorded, not acted on: the client handles the handoff itself, and probing force-reconnects on a + // failed write, which would fight it. + _telemetryProvider.TrackEvent( + "Redis.Maintenance", + [ + new("Source", nameof(PushMaintenanceEvent)), + new("NotificationTypeString", pushEvent.NotificationType.ToString()), + new("SequenceId", pushEvent.SequenceId.ToString(CultureInfo.InvariantCulture)), + new("EndPoint", pushEvent.EndPoint?.ToString() ?? string.Empty), + new("NewEndPoint", pushEvent.NewEndPoint?.ToString() ?? string.Empty), + new("SlotMigrations", pushEvent.SlotMigrations.Count.ToString(CultureInfo.InvariantCulture)), + new("ReceivedTimeUtc", pushEvent.ReceivedTimeUtc.ToString("O", CultureInfo.InvariantCulture)), + new("StartTimeUtc", pushEvent.StartTimeUtc?.ToString("O", CultureInfo.InvariantCulture) ?? string.Empty), + new("RawMessage", pushEvent.RawMessage ?? string.Empty), + ]); + } +#pragma warning restore SER010 + /// Announces an interval boundary, reporting a sink that refuses rather than ending the run. + private bool TryAnnounce(string eventName) => _telemetryProvider.TryTrackEvent(eventName); + private void StartConnectionProbing() { - if (_disposed) + // Under the lock Cancel takes, so the check and the transition are one step: separately, a caller could + // pass the check and then schedule a probe on a service that has since stopped. + lock (_stateLock) { - return; - } + if (_disposed || Volatile.Read(ref _stopped) == 1) + { + return; + } - if (Interlocked.CompareExchange(ref _maintenanceInProgress, 1, 0) != 0) - { - return; + if (Interlocked.CompareExchange(ref _maintenanceInProgress, 1, 0) != 0) + { + return; + } } CancellationTokenSource tokenSource; @@ -222,49 +512,80 @@ private void StartConnectionProbing() tokenSource.CancelAfter(_probingTime); var token = tokenSource.Token; - _ = Task.Run( - async () => + // Never skip the delegate: its finally disposes the linked source and clears InProgress. + _ = Task.Run(() => ProbeUntilCancelledAsync(tokenSource, token), CancellationToken.None); + } + + private async Task ProbeUntilCancelledAsync(CancellationTokenSource tokenSource, CancellationToken token) + { + var started = false; + try + { + // Queued after the lock was released, so StopAsync can have completed in between; a worker that + // lost that race announces nothing. + lock (_stateLock) { - try + if (_disposed || Volatile.Read(ref _stopped) == 1 || token.IsCancellationRequested) { - _telemetryProvider.TrackEvent("Redis.MaintenanceStarted"); - - while (!token.IsCancellationRequested) - { - try - { - var probeTask = _redisConnector.Database.StringSetAsync("probeRedis_" + Environment.MachineName, DateTime.UtcNow.ToString(CultureInfo.InvariantCulture), expiry: TimeSpan.FromDays(1)); - - await probeTask.WaitAsync(_hangingTime, token); - } - catch (OperationCanceledException) when (token.IsCancellationRequested) - { - break; - } - catch (Exception ex) - { - _telemetryProvider.TrackException(ex); - _redisConnector.ForceReconnect(); - } - - try - { - await Task.Delay(_probeInterval, token); - } - catch (OperationCanceledException) when (token.IsCancellationRequested) - { - break; - } - } + return; } - finally + + // Under the same lock that cleared the guard: outside it, a shutdown landing between the two + // would let this worker announce an interval that starts after the service stopped. A refused + // announcement must not end the run -- probing is what it is for -- and leaves started false, + // so there is no end to announce either. + started = TryAnnounce("Redis.MaintenanceStarted"); + } + + while (!token.IsCancellationRequested) + { + if (!await ProbeOnceAsync(token).ConfigureAwait(false)) { - tokenSource.Dispose(); - InProgress = false; - _telemetryProvider.TrackEvent("Redis.MaintenanceEnded"); + break; } - }, - // Never skip the delegate: its finally disposes the linked source and clears InProgress. - CancellationToken.None); + } + } + finally + { + // Cleared first: left set, the CompareExchange guard would refuse every later probe run. + InProgress = false; + tokenSource.Dispose(); + if (started) + { + _ = TryAnnounce("Redis.MaintenanceEnded"); + } + } + } + + /// One probe and the wait after it; false once the run should stop. + private async Task ProbeOnceAsync(CancellationToken token) + { + try + { + var probeTask = _redisConnector.Database.StringSetAsync("probeRedis_" + Environment.MachineName, DateTime.UtcNow.ToString(CultureInfo.InvariantCulture), expiry: TimeSpan.FromDays(1)); + + await probeTask.WaitAsync(_hangingTime, token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + return false; + } + catch (Exception ex) + { + // Reporting must not cost the reconnect: that call is the whole point of probing. + _telemetryProvider.TryTrackException(ex); + _redisConnector.ForceReconnect(); + } + + try + { + await Task.Delay(_probeInterval, token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + return false; + } + + return true; } } diff --git a/src/UiPath.Caching/TelemetrySafeguards.cs b/src/UiPath.Caching/TelemetrySafeguards.cs new file mode 100644 index 00000000..eb0cee46 --- /dev/null +++ b/src/UiPath.Caching/TelemetrySafeguards.cs @@ -0,0 +1,58 @@ +using UiPath.Caching.Telemetry; + +namespace UiPath.Caching; + +/// Telemetry calls that never throw at the caller, so the work around them carries on. +internal static class TelemetrySafeguards +{ + /// Invokes each subscriber separately, so one that throws does not cost the rest theirs. + public static void TryRaise(this THandler? handlers, ICachingTelemetryProvider telemetryProvider, Action raise) + where THandler : Delegate + { + if (handlers is null) + { + return; + } + + foreach (var handler in handlers.GetInvocationList()) + { + try + { + raise((THandler)handler); + } + catch (Exception ex) + { + // Subscribers re-subscribe from these handlers, so stopping at the first throw detaches the rest. + telemetryProvider.TryTrackException(ex); + } + } + } + + /// Records an event, reporting a sink that refuses rather than abandoning what follows. + public static bool TryTrackEvent(this ICachingTelemetryProvider telemetryProvider, string eventName, ReadOnlySpan> properties = default) + { + try + { + telemetryProvider.TrackEvent(eventName, properties); + return true; + } + catch (Exception ex) + { + telemetryProvider.TryTrackException(ex); + return false; + } + } + + /// Reports a caught failure; a sink that refuses it leaves nowhere else to put it. + public static void TryTrackException(this ICachingTelemetryProvider telemetryProvider, Exception ex) + { + try + { + telemetryProvider.TrackException(ex); + } + catch (Exception) + { + // Rethrowing would put it back on the path the caller's catch exists to keep clear, and there is no second sink. + } + } +} diff --git a/tests/UiPath.Caching.Tests/Redis/ConnectionStateMonitorHandoffTests.cs b/tests/UiPath.Caching.Tests/Redis/ConnectionStateMonitorHandoffTests.cs new file mode 100644 index 00000000..6b7cea8e --- /dev/null +++ b/tests/UiPath.Caching.Tests/Redis/ConnectionStateMonitorHandoffTests.cs @@ -0,0 +1,66 @@ +using System.Net; +using System.Reflection; +using StackExchange.Redis; +using UiPath.Caching.Redis; +using UiPath.Caching.Tests.Telemetry; + +namespace UiPath.Caching.Tests.Redis; + +public class ConnectionStateMonitorHandoffTests +{ + private readonly RecordingTelemetryProvider _telemetry = new(); + private readonly FakeConnectionState _source = new(); + + [Fact] + public void A_handoff_is_forwarded_without_the_failure_event() + { + // The monitor sees the same args again on the way to subscribers, so it has to classify them too. + using var sut = new ConnectionStateMonitor(_telemetry, Timeout.InfiniteTimeSpan, _source); + var forwarded = 0; + sut.OnConnectionFailed += (_, _) => forwarded++; + + _source.RaiseConnectionFailed(FailedArgs(ConnectionFailureType.MaintenanceHandoff)); + + forwarded.Should().Be(1, "the connection did drop, so subscribers still need to hear about it"); + _telemetry.Events.Should().NotContain(e => e.Name == "Redis.ConnectionFailed"); + _telemetry.Events.Should().Contain(e => e.Name == "Redis.MaintenanceHandoff"); + } + + [Fact] + public void A_real_failure_still_reports_as_one() + { + using var sut = new ConnectionStateMonitor(_telemetry, Timeout.InfiniteTimeSpan, _source); + + _source.RaiseConnectionFailed(FailedArgs(ConnectionFailureType.SocketFailure)); + + _telemetry.Events.Should().Contain(e => e.Name == "Redis.ConnectionFailed"); + _telemetry.Events.Should().NotContain(e => e.Name == "Redis.MaintenanceHandoff"); + } + + private static ConnectionFailedEventArgs FailedArgs(ConnectionFailureType failureType) => + (ConnectionFailedEventArgs)Activator.CreateInstance( + typeof(ConnectionFailedEventArgs), + BindingFlags.Instance | BindingFlags.NonPublic, + null, + [null, null, new DnsEndPoint("node", 6379), ConnectionType.Interactive, failureType, null, null], + null)!; + + private sealed class FakeConnectionState : IConnectionState + { + public event EventHandler? OnConnectionFailed; + + public event EventHandler? OnConnectionRestored; + + public event EventHandler? OnReconnected; + + public bool IsConnected => true; + + public void RaiseConnectionFailed(EventArgs e) => OnConnectionFailed?.Invoke(this, e); + + public void RaiseUnused() + { + OnConnectionRestored?.Invoke(this, EventArgs.Empty); + OnReconnected?.Invoke(this, EventArgs.Empty); + } + } +} diff --git a/tests/UiPath.Caching.Tests/Redis/ConnectionStateMonitorMulticastTests.cs b/tests/UiPath.Caching.Tests/Redis/ConnectionStateMonitorMulticastTests.cs new file mode 100644 index 00000000..0516f746 --- /dev/null +++ b/tests/UiPath.Caching.Tests/Redis/ConnectionStateMonitorMulticastTests.cs @@ -0,0 +1,98 @@ +using UiPath.Caching.Redis; +using UiPath.Caching.Telemetry; +using UiPath.Caching.Tests.Telemetry; + +namespace UiPath.Caching.Tests.Redis; + +/// The monitor's re-multicast sits inside the connector's, so a throw here aborts that list too. +public class ConnectionStateMonitorMulticastTests +{ + private readonly RecordingTelemetryProvider _telemetry = new(); + private readonly FakeConnectionState _source = new(); + private readonly InvalidOperationException _boom = new("subscriber boom"); + + [Fact] + public void OnReconnected_reaches_every_subscriber_when_one_throws() + { + using var sut = new ConnectionStateMonitor(_telemetry, Timeout.InfiniteTimeSpan, _source); + var reached = 0; + sut.OnReconnected += (_, _) => throw _boom; + sut.OnReconnected += (_, _) => reached++; + + var raise = () => _source.RaiseReconnected(); + + raise.Should().NotThrow("the connector's multicast continues past this monitor"); + reached.Should().Be(1, "a throwing subscriber must not cost the rest their notification"); + _telemetry.Exceptions.Should().ContainSingle().Which.Exception.Should().BeSameAs(_boom); + } + + [Fact] + public void OnConnectionFailed_reaches_every_subscriber_when_one_throws() + { + using var sut = new ConnectionStateMonitor(_telemetry, Timeout.InfiniteTimeSpan, _source); + var reached = 0; + sut.OnConnectionFailed += (_, _) => throw _boom; + sut.OnConnectionFailed += (_, _) => reached++; + + var raise = () => _source.RaiseConnectionFailed(); + + raise.Should().NotThrow("the connector's multicast continues past this monitor"); + reached.Should().Be(1, "a throwing subscriber must not cost the rest their notification"); + _telemetry.Exceptions.Should().ContainSingle().Which.Exception.Should().BeSameAs(_boom); + } + + [Fact] + public void OnConnectionRestored_reaches_every_subscriber_when_one_throws() + { + using var sut = new ConnectionStateMonitor(_telemetry, Timeout.InfiniteTimeSpan, _source); + var reached = 0; + sut.OnConnectionRestored += (_, _) => throw _boom; + sut.OnConnectionRestored += (_, _) => reached++; + + var raise = () => _source.RaiseConnectionRestored(); + + raise.Should().NotThrow("the connector's multicast continues past this monitor"); + reached.Should().Be(1, "a throwing subscriber must not cost the rest their notification"); + _telemetry.Exceptions.Should().ContainSingle().Which.Exception.Should().BeSameAs(_boom); + } + + [Fact] + public void A_refused_record_costs_neither_the_state_reset_nor_the_subscribers() + { + // The record runs ahead of both, and this monitor sits inside the connector's own multicast -- the + // connector can isolate the monitor, but it cannot hand the monitor's subscribers their notification. + var telemetry = new RefusingTelemetryProvider(); + using var sut = new ConnectionStateMonitor(telemetry, Timeout.InfiniteTimeSpan, _source); + var reached = 0; + sut.OnConnectionFailed += (_, _) => reached++; + + var raise = () => _source.RaiseConnectionFailed(); + + raise.Should().NotThrow(); + reached.Should().Be(1, "a sink that refuses the record must not swallow the notification"); + } + + /// Refuses every record, the way a saturated sink would. + private sealed class RefusingTelemetryProvider : ICachingTelemetryProvider + { + public void TrackEvent(string eventName, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) => + throw new InvalidOperationException("sink boom"); + } + + private sealed class FakeConnectionState : IConnectionState + { + public event EventHandler? OnConnectionFailed; + + public event EventHandler? OnConnectionRestored; + + public event EventHandler? OnReconnected; + + public bool IsConnected => true; + + public void RaiseConnectionFailed() => OnConnectionFailed?.Invoke(this, EventArgs.Empty); + + public void RaiseConnectionRestored() => OnConnectionRestored?.Invoke(this, EventArgs.Empty); + + public void RaiseReconnected() => OnReconnected?.Invoke(this, EventArgs.Empty); + } +} diff --git a/tests/UiPath.Caching.Tests/Redis/RedisConnectorLifecycleTests.cs b/tests/UiPath.Caching.Tests/Redis/RedisConnectorLifecycleTests.cs index c7a3d81a..bea99992 100644 --- a/tests/UiPath.Caching.Tests/Redis/RedisConnectorLifecycleTests.cs +++ b/tests/UiPath.Caching.Tests/Redis/RedisConnectorLifecycleTests.cs @@ -1,7 +1,12 @@ +using System.Net; +using System.Reflection; using Microsoft.Extensions.Logging.Abstractions; using StackExchange.Redis; +using StackExchange.Redis.Availability; +using StackExchange.Redis.Maintenance; using UiPath.Caching.Redis; using UiPath.Caching.Telemetry; +using UiPath.Caching.Tests.Telemetry; namespace UiPath.Caching.Tests.Redis; @@ -264,6 +269,215 @@ public async Task ForceReconnect_SwallowsOnReconnectedHandlerException() connector.Dispose(); } + [Fact] + public async Task ForceReconnect_ReachesEveryOnReconnectedHandler_WhenOneThrows() + { + // Subscribers re-subscribe from this handler, so a multicast that stops at the first throw detaches the rest. + var telemetry = new RecordingTelemetryProvider(); + var oldMultiplexer = Substitute.For(); + var newMultiplexer = Substitute.For(); + oldMultiplexer.CloseAsync(Arg.Any()).Returns(Task.CompletedTask); + var disposed = new TaskCompletionSource(); + oldMultiplexer.When(m => m.Dispose()).Do(_ => disposed.TrySetResult()); + var connector = NewConnector(new SequenceFactory(oldMultiplexer, newMultiplexer), telemetry); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + var boom = new InvalidOperationException("handler boom"); + var reached = 0; + connector.OnReconnected += (_, _) => throw boom; + connector.OnReconnected += (_, _) => Interlocked.Increment(ref reached); + + connector.ForceReconnect(); + await disposed.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + + Volatile.Read(ref reached).Should().Be(1, "a throwing subscriber must not cost the rest their notification"); + telemetry.Exceptions.Should().ContainSingle().Which.Exception.Should().BeSameAs(boom); + connector.Dispose(); + } + + [Fact] + public async Task ServerMaintenance_ReachesEveryHandler_WhenOneThrows() + { + var (connector, multiplexer, telemetry) = await ConnectedAsync(); + var boom = new InvalidOperationException("handler boom"); + var reached = 0; + connector.ServerMaintenance += (_, _) => throw boom; + connector.ServerMaintenance += (_, _) => reached++; + + var raise = () => multiplexer.ServerMaintenanceEvent += Raise.Event>(multiplexer, MaintenanceEvent()); + + raise.Should().NotThrow("this is raised on the client's own dispatch, which must not see our subscribers throw"); + reached.Should().Be(1, "a throwing subscriber must not cost the rest their notification"); + telemetry.Exceptions.Should().ContainSingle().Which.Exception.Should().BeSameAs(boom); + connector.Dispose(); + } + + [Fact] + public async Task OnConnectionFailed_ReachesEveryHandler_WhenOneThrows() + { + var (connector, multiplexer, telemetry) = await ConnectedAsync(); + var boom = new InvalidOperationException("handler boom"); + var reached = 0; + connector.OnConnectionFailed += (_, _) => throw boom; + connector.OnConnectionFailed += (_, _) => reached++; + + var raise = () => multiplexer.ConnectionFailed += Raise.EventWith(multiplexer, FailedArgs(ConnectionFailureType.SocketFailure)); + + raise.Should().NotThrow("this is raised on the client's own dispatch, which must not see our subscribers throw"); + reached.Should().Be(1, "a throwing subscriber must not cost the rest their notification"); + telemetry.Exceptions.Should().ContainSingle().Which.Exception.Should().BeSameAs(boom); + connector.Dispose(); + } + + [Fact] + public async Task OnConnectionRestored_ReachesEveryHandler_WhenOneThrows() + { + var (connector, multiplexer, telemetry) = await ConnectedAsync(); + var boom = new InvalidOperationException("handler boom"); + var reached = 0; + connector.OnConnectionRestored += (_, _) => throw boom; + connector.OnConnectionRestored += (_, _) => reached++; + + var raise = () => multiplexer.ConnectionRestored += Raise.EventWith(multiplexer, FailedArgs(ConnectionFailureType.SocketFailure)); + + raise.Should().NotThrow("this is raised on the client's own dispatch, which must not see our subscribers throw"); + reached.Should().Be(1, "a throwing subscriber must not cost the rest their notification"); + telemetry.Exceptions.Should().ContainSingle().Which.Exception.Should().BeSameAs(boom); + connector.Dispose(); + } + + [Fact] + public async Task Version_FallsBack_WhenReadingItAndReportingTheFailureBothThrow() + { + // _version is a Lazy, and a Lazy caches a factory exception for the life of the process -- a sink that + // refuses the report would make Version throw for good rather than fall back once. + var multiplexer = Substitute.For(); + multiplexer.GetEndPoints(Arg.Any()).Returns(_ => throw ConnectFailure()); + var connector = NewConnector(new SequenceFactory(multiplexer), new ThrowingSinkTelemetryProvider()); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + + var read = () => connector.Version; + + read.Should().NotThrow(); + read.Should().NotThrow("a Lazy caches what its factory threw, so the second read would throw too"); + connector.Dispose(); + } + + [Fact] + public async Task ServerMaintenance_IsRoutedByTheSubscribedMultiplexer_NotTheSender() + { + // A composite multiplexer can attach our handler to its children and raise with the child as sender, so + // the generation has to be the connection we subscribed to, not whatever arrives in the argument. + var (connector, multiplexer, _) = await ConnectedAsync(); + var seen = new List(); + connector.ServerMaintenance += (_, e) => seen.Add(e); + + var moving = PushEvent(); + multiplexer.ServerMaintenanceEvent += Raise.Event>(new object(), moving); + + seen.Should().ContainSingle("a MOVING on the current connection is not dropped because a child raised it") + .Which.Should().BeSameAs(moving); + connector.Dispose(); + } + + [Theory] + [InlineData(true, 1)] + [InlineData(false, 0)] + public async Task AMoving_FromAGroup_IsTakenOnlyFromTheActiveMember(bool fromActive, int expected) + { + // A group attaches our handler to every member, so the raiser tells us which one saw it -- and only the + // member carrying commands has a replacement the cache cares about. + var group = Substitute.For(); + var activeMember = new object(); + var connector = NewConnector(new SequenceFactory(group)); + connector.SetActiveMemberConnectionResolver(_ => activeMember); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + var seen = new List(); + connector.ServerMaintenance += (_, e) => seen.Add(e); + + group.ServerMaintenanceEvent += Raise.Event>(fromActive ? activeMember : new object(), PushEvent()); + + seen.Should().HaveCount(expected); + connector.Dispose(); + } + + [Fact] + public async Task AMoving_FromAGroup_IsTaken_WhenTheActiveMemberCannotBeRead() + { + // The member's connection is read reflectively, so an upstream rename has to cost a spare record rather + // than the handoff notice. + var telemetry = new RecordingTelemetryProvider(); + var group = Substitute.For(); + var connector = NewConnector(new SequenceFactory(group), telemetry); + var boom = new InvalidOperationException("no such property"); + connector.SetActiveMemberConnectionResolver(_ => throw boom); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + var seen = new List(); + connector.ServerMaintenance += (_, e) => seen.Add(e); + + group.ServerMaintenanceEvent += Raise.Event>(new object(), PushEvent()); + + seen.Should().ContainSingle("a check that cannot be made must not drop the notice"); + telemetry.Exceptions.Should().ContainSingle().Which.Exception.Should().BeSameAs(boom); + connector.Dispose(); + } + + [Fact] + public async Task ABroadcast_FromAGroup_IsTaken_WhicheverMemberSawIt() + { + // Only a MOVING is scoped to a member; a broadcast may have been seen by just one of them. + var group = Substitute.For(); + var connector = NewConnector(new SequenceFactory(group)); + connector.SetActiveMemberConnectionResolver(_ => new object()); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + var seen = new List(); + connector.ServerMaintenance += (_, e) => seen.Add(e); + + group.ServerMaintenanceEvent += Raise.Event>(new object(), BroadcastPushEvent()); + + seen.Should().ContainSingle("a broadcast is not scoped to the member that received it"); + connector.Dispose(); + } + + [Fact] + public async Task ForceReconnect_Completes_WhenRecordingTheEventThrows() + { + // The record sits between the swap and the two things that finish the reconnect, so a refusing sink + // would leave every subscriber unnotified and the retired connection undisposed. + var oldMultiplexer = Substitute.For(); + var newMultiplexer = Substitute.For(); + oldMultiplexer.CloseAsync(Arg.Any()).Returns(Task.CompletedTask); + var disposed = new TaskCompletionSource(); + oldMultiplexer.When(m => m.Dispose()).Do(_ => disposed.TrySetResult()); + var connector = NewConnector(new SequenceFactory(oldMultiplexer, newMultiplexer), new ThrowOnEventTelemetryProvider("Redis.ForcedReconnect")); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + var reconnected = 0; + connector.OnReconnected += (_, _) => Interlocked.Increment(ref reconnected); + + connector.ForceReconnect(); + await disposed.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + + Volatile.Read(ref reconnected).Should().Be(1, "subscribers re-subscribe on this event"); + oldMultiplexer.Received(1).Dispose(); + connector.Dispose(); + } + + [Fact] + public async Task AHandoff_DoesNotFaultTheClientDispatch_WhenRecordingItThrows() + { + // Raised by StackExchange.Redis on its own thread, so nothing here may let a refused record reach it. + var multiplexer = Substitute.For(); + var connector = NewConnector(new SequenceFactory(multiplexer), new ThrowOnEventTelemetryProvider("Redis.MaintenanceHandoff")); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + var forwarded = 0; + connector.OnConnectionFailed += (_, _) => forwarded++; + + var raise = () => multiplexer.ConnectionFailed += Raise.EventWith(multiplexer, FailedArgs(ConnectionFailureType.MaintenanceHandoff)); + + raise.Should().NotThrow(); + forwarded.Should().Be(1, "the connection did drop, so subscribers still hear about it"); + connector.Dispose(); + } + [Fact] public async Task ForceReconnect_DisposesOld_WhenCloseAsyncThrows() { @@ -331,6 +545,130 @@ public async Task Dispose_Swallows_WhenMultiplexerDisposeThrows() multiplexer.Received(1).Dispose(); } + [Fact] + public async Task ServerMaintenance_IsForwarded_FromTheCurrentMultiplexer_AcrossAReconnect() + { + // The routing tests substitute IRedisConnector, so nothing else covers the wiring: without this they + // would all pass while production received nothing. + var first = Substitute.For(); + var replacement = Substitute.For(); + var connector = NewConnector(new SequenceFactory(first, replacement)); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + + var seen = new List(); + connector.ServerMaintenance += (_, e) => seen.Add(e); + + var onFirst = MaintenanceEvent(); + first.ServerMaintenanceEvent += Raise.Event>(first, onFirst); + seen.Should().ContainSingle().Which.Should().BeSameAs(onFirst); + + // Closing the retired multiplexer detaches its handler, and OnReconnected fires just before that -- so + // without this gate the assertions below race teardown and the first would pass for the wrong reason. + var closing = new TaskCompletionSource(); + first.CloseAsync(Arg.Any()).Returns(closing.Task); + + var reconnected = new TaskCompletionSource(); + connector.OnReconnected += (_, _) => reconnected.TrySetResult(); + connector.ForceReconnect(); + await reconnected.Task.WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken); + + var onReplacement = MaintenanceEvent(); + replacement.ServerMaintenanceEvent += Raise.Event>(replacement, onReplacement); + + seen.Should().HaveCount(2, "the replacement multiplexer must be wired too"); + seen[1].Should().BeSameAs(onReplacement); + + // The positive side of the filter. + var onCurrent = PushEvent(); + replacement.ServerMaintenanceEvent += Raise.Event>(replacement, onCurrent); + seen.Should().HaveCount(3, "a push frame from the current connection describes the endpoint in use"); + seen[2].Should().BeSameAs(onCurrent); + + first.DidNotReceive().Dispose(); + + // Still subscribed, so these reach the filter rather than an absent handler. A retired generation's + // MOVING names an endpoint the cache has left. + first.ServerMaintenanceEvent += Raise.Event>(first, PushEvent()); + seen.Should().HaveCount(3, "a retired connection's MOVING names an endpoint the cache has left"); + + // Broadcast kinds stay: the retired connection may be the only one that observed this. + var migrating = BroadcastPushEvent(); + first.ServerMaintenanceEvent += Raise.Event>(first, migrating); + seen.Should().HaveCount(4, "a broadcast push frame is not scoped to the connection that received it"); + seen[3].Should().BeSameAs(migrating); + + // Azure's is pub/sub, which the client does not collapse, and the retired connection may be the only + // one that was subscribed when it went out. + var broadcast = AzureEvent(); + first.ServerMaintenanceEvent += Raise.Event>(first, broadcast); + seen.Should().HaveCount(5, "an Azure broadcast is true whichever connection received it"); + seen[4].Should().BeSameAs(broadcast); + + closing.SetResult(); + connector.Dispose(); + } + + [Fact] + public async Task AMoving_IsForwarded_FromTheReplacementBeforeItBecomesCurrent() + { + // The replacement is subscribed before it is published, and the server never replays a MOVING -- so one + // arriving in between has to be taken. Raised from inside the subscription, which is exactly that interval. + var first = Substitute.For(); + var replacement = Substitute.For(); + var moving = PushEvent(); + var raised = false; + replacement.When(m => m.ServerMaintenanceEvent += Arg.Any>()) + .Do(call => + { + if (raised) + { + return; + } + + raised = true; + call.Arg>().Invoke(replacement, moving); + }); + + var connector = NewConnector(new SequenceFactory(first, replacement)); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + + var seen = new List(); + connector.ServerMaintenance += (_, e) => seen.Add(e); + + var reconnected = new TaskCompletionSource(); + connector.OnReconnected += (_, _) => reconnected.TrySetResult(); + connector.ForceReconnect(); + await reconnected.Task.WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken); + + seen.Should().ContainSingle("the generation about to carry commands is the one a MOVING describes") + .Which.Should().BeSameAs(moving); + + connector.Dispose(); + } + + [Fact] + public async Task AMaintenanceHandoff_IsNotReportedAsAConnectionFailure() + { + // ConnectionStateMonitorHandoffTests raises on a fake source, so it never reaches this branch. + var telemetry = new RecordingTelemetryProvider(); + var multiplexer = Substitute.For(); + var connector = NewConnector(new SequenceFactory(multiplexer), telemetry); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + var forwarded = 0; + connector.OnConnectionFailed += (_, _) => forwarded++; + + multiplexer.ConnectionFailed += Raise.EventWith(multiplexer, FailedArgs(ConnectionFailureType.MaintenanceHandoff)); + + forwarded.Should().Be(1, "the connection did drop, so subscribers still need to hear about it"); + telemetry.Events.Should().Contain(e => e.Name == "Redis.MaintenanceHandoff"); + telemetry.Events.Should().NotContain(e => e.Name == "Redis.ConnectionFailed"); + + multiplexer.ConnectionFailed += Raise.EventWith(multiplexer, FailedArgs(ConnectionFailureType.SocketFailure)); + telemetry.Events.Should().Contain(e => e.Name == "Redis.ConnectionFailed"); + + connector.Dispose(); + } + [Fact] public void GetPrimaries_IsEmpty_BeforeConnect_DoesNotTriggerConnect() { @@ -386,6 +724,48 @@ private static IServer StubServer(bool isReplica, bool isConnected) return server; } + private static ConnectionFailedEventArgs FailedArgs(ConnectionFailureType failureType) => + (ConnectionFailedEventArgs)Activator.CreateInstance( + typeof(ConnectionFailedEventArgs), + BindingFlags.Instance | BindingFlags.NonPublic, + null, + [null, null, new System.Net.DnsEndPoint("node", 6379), ConnectionType.Interactive, failureType, null, null], + null)!; + +#pragma warning disable SER010 // Server-native maintenance notifications are for evaluation purposes only + /// A push frame every node broadcasts, rather than a MOVING. + private static PushMaintenanceEvent BroadcastPushEvent() => PushEvent(MaintenanceNotificationType.Migrating); + + private static PushMaintenanceEvent PushEvent(MaintenanceNotificationType type = MaintenanceNotificationType.Moving) => + (PushMaintenanceEvent)Activator.CreateInstance( + typeof(PushMaintenanceEvent), + BindingFlags.Instance | BindingFlags.NonPublic, + null, + [type, 1L, (EndPoint)new DnsEndPoint("node", 6379), (TimeSpan?)null, (EndPoint?)null, "payload", $">{type} 1 payload", Array.Empty()], + null)!; +#pragma warning restore SER010 + + private static AzureMaintenanceEvent AzureEvent() => + (AzureMaintenanceEvent)Activator.CreateInstance( + typeof(AzureMaintenanceEvent), + BindingFlags.Instance | BindingFlags.NonPublic, + null, + ["NotificationType|NodeMaintenanceStarting|StartTimeInUTC|2026-09-19T00:00:00|IsReplica|False|IPAddress|127.0.0.1|SSLPort|6380|NonSSLPort|6379"], + null)!; + + private static ServerMaintenanceEvent MaintenanceEvent() => + (ServerMaintenanceEvent)Activator.CreateInstance( + typeof(ServerMaintenanceEvent), BindingFlags.Instance | BindingFlags.NonPublic, null, null, null)!; + + private static async Task<(RedisConnector Connector, IConnectionMultiplexer Multiplexer, RecordingTelemetryProvider Telemetry)> ConnectedAsync() + { + var telemetry = new RecordingTelemetryProvider(); + var multiplexer = Substitute.For(); + var connector = NewConnector(new SequenceFactory(multiplexer), telemetry); + await connector.ConnectAsync(TestContext.Current.CancellationToken); + return (connector, multiplexer, telemetry); + } + private static RedisConnector NewConnector(IConnectionMultiplexerFactory factory, ICachingTelemetryProvider? telemetry = null) { var options = Options.Create(new RedisConnectionOptions { ConnectionString = "localhost:6379", EnableHangDetection = false }); @@ -394,6 +774,25 @@ private static RedisConnector NewConnector(IConnectionMultiplexerFactory factory } private static RedisConnectionException ConnectFailure() => new(ConnectionFailureType.UnableToConnect, CommandFlags.None, "boom"); + /// A sink that refuses one named event and takes everything else. + private sealed class ThrowOnEventTelemetryProvider(string failingEvent) : ICachingTelemetryProvider + { + public void TrackEvent(string eventName, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) + { + if (eventName == failingEvent) + { + throw new InvalidOperationException("sink boom"); + } + } + } + + /// A sink that refuses every report, leaving a catch with nowhere to put what it caught. + private sealed class ThrowingSinkTelemetryProvider : ICachingTelemetryProvider + { + public void TrackException(Exception ex, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) => + throw new InvalidOperationException("sink boom"); + } + private sealed class SequenceFactory : IConnectionMultiplexerFactory { private readonly Queue _multiplexers; diff --git a/tests/UiPath.Caching.Tests/Redis/RedisConnectorTests.cs b/tests/UiPath.Caching.Tests/Redis/RedisConnectorTests.cs index 396a4847..4cb94366 100644 --- a/tests/UiPath.Caching.Tests/Redis/RedisConnectorTests.cs +++ b/tests/UiPath.Caching.Tests/Redis/RedisConnectorTests.cs @@ -59,6 +59,63 @@ public void ConnectionStringExtraParamsX(string connectionString, string extraPa cnn.Should().Be(expected); } +#pragma warning disable SER010 // Server-native maintenance notifications are for evaluation purposes only + [Theory] + [InlineData(RedisMaintenanceNotifications.Auto, MaintenanceNotificationMode.Auto)] + [InlineData(RedisMaintenanceNotifications.Required, MaintenanceNotificationMode.Enabled)] + [InlineData(RedisMaintenanceNotifications.Disabled, MaintenanceNotificationMode.Disabled)] + public void MaintenanceNotifications_MapsOntoTheClientMode(RedisMaintenanceNotifications configured, MaintenanceNotificationMode expected) + { + // Required maps to Enabled, not a same-named member, so a reversed arm would silently do nothing. + var opt = new RedisConnectionOptions + { + ConnectionString = "localhost:6379", + MaintenanceNotifications = configured, + }; + var sut = new RedisConfigurationOptionsProvider(NullLoggerFactory.Instance, Options.Create(opt)); + + sut.GetConfiguration().MaintenanceNotifications.Should().Be(expected); + } + + [Fact] + public void MaintenanceNotifications_AppliesWithoutAConnectionString() + { + // With no connection string these options are what a supplied ConnectionFactory is handed. + var opt = new RedisConnectionOptions { MaintenanceNotifications = RedisMaintenanceNotifications.Auto }; + var sut = new RedisConfigurationOptionsProvider(NullLoggerFactory.Instance, Options.Create(opt)); + + sut.GetConfiguration().MaintenanceNotifications.Should().Be(MaintenanceNotificationMode.Auto); + } + + [Fact] + public void MaintenanceNotifications_RejectsAnUnsupportedValue() + { + // Configuration binds enums from numbers, and folding an unknown one into Disabled would be + // indistinguishable from asking and being refused. + var opt = new RedisConnectionOptions + { + ConnectionString = "localhost:6379", + MaintenanceNotifications = (RedisMaintenanceNotifications)3, + }; + var sut = new RedisConfigurationOptionsProvider(NullLoggerFactory.Instance, Options.Create(opt)); + + // Invalid configuration rather than a bad argument: the value reaches this from the options, not + // from a parameter of the method that rejects it. + sut.Invoking(p => p.GetConfiguration()).Should().Throw() + .WithMessage("*RedisMaintenanceNotifications*"); + } + + [Fact] + public void MaintenanceNotifications_LeavesTheClientDefault_WhenUnset() + { + var opt = new RedisConnectionOptions { ConnectionString = "localhost:6379" }; + var sut = new RedisConfigurationOptionsProvider(NullLoggerFactory.Instance, Options.Create(opt)); + + sut.GetConfiguration().MaintenanceNotifications + .Should().Be(new ConfigurationOptions().MaintenanceNotifications, "null must not overwrite what the client itself decides"); + } +#pragma warning restore SER010 + public ValueTask DisposeAsync() { return ValueTask.CompletedTask; diff --git a/tests/UiPath.Caching.Tests/Redis/RedisPlannedMaintenanceRoutingTests.cs b/tests/UiPath.Caching.Tests/Redis/RedisPlannedMaintenanceRoutingTests.cs new file mode 100644 index 00000000..6d2a53ca --- /dev/null +++ b/tests/UiPath.Caching.Tests/Redis/RedisPlannedMaintenanceRoutingTests.cs @@ -0,0 +1,685 @@ +using System.Globalization; +using System.Net; +using System.Reflection; +using Microsoft.Extensions.Logging.Abstractions; +using StackExchange.Redis; +using StackExchange.Redis.Maintenance; +using UiPath.Caching.Redis; +using UiPath.Caching.Telemetry; +using UiPath.Caching.Tests.Telemetry; + +namespace UiPath.Caching.Tests.Redis; + +#pragma warning disable SER010 // Server-native maintenance notifications are for evaluation purposes only + +public class RedisPlannedMaintenanceRoutingTests : IDisposable +{ + private readonly List _started = []; + private readonly MovableClock _clock = new(); + + private readonly RecordingTelemetryProvider _telemetry = new(); + private readonly IConnectionMultiplexer _multiplexer = Substitute.For(); + private readonly IRedisConnector _connector = Substitute.For(); + + [Fact] + public async Task An_announced_disruption_is_recorded_without_reconnecting() + { + // The client hands the connection off itself, and probing force-reconnects on a failed write. + var sut = await StartedAsync(); + + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.Migrating)); + + _connector.DidNotReceive().ForceReconnect(); + sut.InProgress.Should().BeFalse(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task An_Azure_notification_is_recorded_whichever_route_it_arrives_on(bool onCommandConnection) + { + // Which connection received the broadcast is not knowable after the fact, so both routes record. + await StartedAsync(); + var azure = AzureEvent(); + + if (onCommandConnection) + { + RaiseOnCommandConnection(azure); + } + else + { + RaiseMaintenance(azure); + } + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(1); + } + + [Fact] + public async Task An_Azure_notification_falls_back_to_the_command_connection_when_there_is_no_maintenance_one() + { + // The maintenance connection is established in the background and gives up after its retries, so the + // command connection's copy can be the only one there is. + await StartedWithoutMaintenanceConnectionAsync(); + + RaiseOnCommandConnection(AzureEvent()); + + _telemetry.Events.Should().ContainSingle(e => e.Name == "Redis.Maintenance") + .Which.Properties!["Source"].Should().Be(nameof(AzureMaintenanceEvent)); + } + + [Fact] + public async Task An_Azure_notification_is_recorded_when_only_the_command_connection_received_it() + { + // The channel can go live between publication and delivery, and pub/sub does not replay -- so asking + // whether the other route is subscribed now would drop the only copy there was. + await StartedAsync(); + + RaiseOnCommandConnection(AzureEvent()); + + _telemetry.Events.Should().ContainSingle(e => e.Name == "Redis.Maintenance") + .Which.Properties!["Source"].Should().Be(nameof(AzureMaintenanceEvent)); + } + + [Fact] + public async Task A_push_frame_replayed_to_a_rebuilt_connection_is_recorded_once() + { + // The client collapses copies only within one multiplexer, and the server replays on reconnect. + await StartedAsync(); + + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.SlotMigrating, sequenceId: 16)); + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.SlotMigrating, sequenceId: 16)); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(1); + } + + [Fact] + public async Task Notifications_whose_sequence_could_not_be_read_are_each_recorded() + { + // The client reports an unreadable sequence as zero and declines to collapse those itself. + await StartedAsync(); + + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.Migrating, sequenceId: null)); + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.Migrating, sequenceId: null)); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(2); + } + + [Fact] + public async Task Sources_that_are_not_modelled_are_each_recorded_even_when_identical() + { + // RawMessage carries no uniqueness contract, and the fallback exists to keep an unknown source visible. + await StartedAsync(); + + RaiseOnCommandConnection(UnmodelledEvent("something we do not model yet")); + RaiseOnCommandConnection(UnmodelledEvent("something we do not model yet")); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(2); + } + + [Fact] + public async Task A_replayed_notification_numbered_zero_is_recorded_once() + { + // Zero is a legitimate sequence too, so it must not be read as the absent one. + await StartedAsync(); + + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.Migrating, sequenceId: 0)); + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.Migrating, sequenceId: 0)); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(1); + } + + [Fact] + public async Task A_completion_is_not_collapsed_into_the_starter_it_follows() + { + // One sequence rather than adjacent ones: on 16 and 17 a key of the sequence alone would pass too. + await StartedAsync(); + + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.SlotMigrating, sequenceId: 16)); + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.SlotMigrated, sequenceId: 16)); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(2); + } + + [Fact] + public async Task A_push_frame_on_the_maintenance_connection_is_ignored() + { + // Asserted on the telemetry: nothing here moves health state, so only the record distinguishes the routes. + await StartedAsync(); + + RaiseMaintenance(PushEvent(MaintenanceNotificationType.Moving)); + + _telemetry.Events.Should().NotContain(e => e.Name == "Redis.Maintenance"); + } + + [Fact] + public async Task Every_push_notification_is_recorded_with_its_source() + { + await StartedAsync(); + + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.SlotMigrating, sequenceId: 16)); + + var recorded = _telemetry.Events.Should().ContainSingle(e => e.Name == "Redis.Maintenance").Which; + recorded.Properties!["Source"].Should().Be(nameof(PushMaintenanceEvent)); + recorded.Properties["NotificationTypeString"].Should().Be(nameof(MaintenanceNotificationType.SlotMigrating)); + recorded.Properties["SequenceId"].Should().Be("16"); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_source_that_is_not_modelled_is_recorded_rather_than_dropped(bool onCommandConnection) + { + // Both routers filter, so both have to be held to this or one can regress alone. + await StartedAsync(); + var unmodelled = (ServerMaintenanceEvent)Activator.CreateInstance( + typeof(ServerMaintenanceEvent), BindingFlags.Instance | BindingFlags.NonPublic, null, null, null)!; + + if (onCommandConnection) + { + RaiseOnCommandConnection(unmodelled); + } + else + { + RaiseMaintenance(unmodelled); + } + + _telemetry.Events.Should().ContainSingle(e => e.Name == "Redis.Maintenance") + .Which.Properties!["Source"].Should().Be(nameof(ServerMaintenanceEvent)); + } + + [Fact] + public async Task A_broadcast_taken_on_both_routes_is_recorded_once() + { + // Both connections are subscribed to the channel, so both receive the broadcast and both routes record. + await StartedAsync(); + RaiseOnCommandConnection(AzureEvent()); + RaiseMaintenance(AzureEvent()); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(1); + } + + [Fact] + public async Task A_broadcast_forwarded_by_two_connection_generations_is_recorded_once() + { + // A rebuild leaves both the retired and the replacement connection subscribed, deliberately. + await StartedWithoutMaintenanceConnectionAsync(); + + RaiseOnCommandConnection(AzureEvent()); + RaiseOnCommandConnection(AzureEvent()); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(1); + } + + [Fact] + public async Task A_different_broadcast_is_recorded_after_a_suppressed_copy() + { + await StartedWithoutMaintenanceConnectionAsync(); + + RaiseOnCommandConnection(AzureEvent()); + RaiseOnCommandConnection(AzureEvent()); + RaiseOnCommandConnection(AzureEvent("2026-09-19T01:00:00")); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(2); + } + + [Fact] + public async Task A_copy_is_still_recognised_behind_a_run_of_other_notifications() + { + // A set sized by count would drop an identity still inside the window once enough others arrived. + await StartedWithoutMaintenanceConnectionAsync(); + + RaiseOnCommandConnection(AzureEvent()); + for (var i = 0; i < 64; i++) + { + RaiseOnCommandConnection(AzureEvent($"2026-09-19T{i / 60:D2}:{i % 60:D2}:30")); + } + + RaiseOnCommandConnection(AzureEvent()); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(65); + } + + [Fact] + public async Task The_retention_window_survives_a_backward_clock_correction() + { + // Measured on timestamps, so a host clock correction neither holds entries nor drops the protection. + await StartedWithoutMaintenanceConnectionAsync(); + + RaiseOnCommandConnection(AzureEvent()); + _clock.CorrectWallClock(TimeSpan.FromMinutes(-5)); + _clock.Advance(TimeSpan.FromMinutes(1)); + RaiseOnCommandConnection(AzureEvent()); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(2); + } + + [Fact] + public async Task A_broadcast_repeated_beyond_the_retention_window_is_recorded_again() + { + // Only the copies of one broadcast are collapsed; a later announcement is news however it is worded. + await StartedWithoutMaintenanceConnectionAsync(); + + RaiseOnCommandConnection(AzureEvent()); + _clock.Advance(TimeSpan.FromMinutes(1)); + RaiseOnCommandConnection(AzureEvent()); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(2); + } + + [Fact] + public async Task Disposing_cancels_the_token_a_connection_attempt_was_given() + { + // InitializeAsync catches the cancellation and returns, so the wait ends with the service. It is the + // token that ends here, not the connect: ConnectionMultiplexerFactory checks it once and then awaits + // ConnectionMultiplexer.ConnectAsync, which takes none -- a real attempt already in flight runs on. + // Handed over through a completion source: the worker that creates it and the thread that reads it are + // different, and a ValueTask is several fields, so polling a shared one can see it half-written. + var captured = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); + var options = Options.Create(new RedisConnectionOptions { ConnectionString = "localhost:6379" }); + var factory = Substitute.For(); + // CA2012: arranging a ValueTask-returning member with NSubstitute means calling it and handing the + // result to Returns. There is no shape of this that awaits it, and nothing ever consumes it. +#pragma warning disable CA2012 + factory.CreateAsync(Arg.Any(), Arg.Any()) + .Returns(call => + { + var attempt = NeverConnects(call.Arg()); + captured.TrySetResult(attempt); + return new ValueTask(attempt); + }); +#pragma warning restore CA2012 + var sut = new RedisPlannedMaintenance(_telemetry, _connector, new RedisConfigurationOptionsProvider(NullLoggerFactory.Instance, options), factory, NullLogger.Instance, options, null, _clock); + await sut.StartAsync(TestContext.Current.CancellationToken); + + var pending = await captured.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + pending.IsCompleted.Should().BeFalse("the maintenance connection never arrives"); + sut.Dispose(); + var finished = await Task.WhenAny(pending, Task.Delay(5000, TestContext.Current.CancellationToken)); + finished.Should().BeSameAs(pending, "disposing must end the wait rather than leave it suspended"); + } + + [Fact] + public async Task A_throw_while_handling_a_notification_does_not_escape_the_maintenance_connection() + { + // Attached straight to the client, unlike the command route, so nothing else keeps throws off its dispatch. + var telemetry = new ThrowOnMaintenanceEventProvider(); + await StartedAsync(telemetry); + + var raise = () => RaiseMaintenance(AzureEvent()); + + raise.Should().NotThrow("StackExchange.Redis is raising this, and it did not subscribe to our bookkeeping"); + telemetry.Exceptions.Should().ContainSingle().Which.Should().BeSameAs(ThrowOnMaintenanceEventProvider.Failure); + } + + [Fact] + public async Task A_report_that_fails_too_does_not_escape_the_maintenance_connection() + { + // A sink that cannot take the report leaves nowhere to put it, and it still must not reach the dispatch thread. + var telemetry = new ThrowOnMaintenanceEventProvider { ReportingThrows = true }; + await StartedAsync(telemetry); + + var raise = () => RaiseMaintenance(AzureEvent()); + + raise.Should().NotThrow(); + } + + [Fact] + public async Task A_notification_whose_recording_threw_is_recorded_when_it_arrives_again() + { + // A handler that threw recorded nothing, so holding its claim would lose the notification for the window. + var telemetry = new ThrowOnMaintenanceEventProvider(failures: 1); + await StartedAsync(telemetry); + var azure = AzureEvent(); + + RaiseMaintenance(azure); + RaiseMaintenance(azure); + + telemetry.Exceptions.Should().Contain(ThrowOnMaintenanceEventProvider.Failure); + telemetry.Events.Count(e => e == "Redis.Maintenance").Should().Be(1, "the retry recorded what the failed attempt did not"); + } + + [Fact] + public async Task Disposing_reaches_the_multiplexer_when_cancelling_and_reporting_both_throw() + { + // Cancel runs the probe loop's registrations, so it can throw what a caller did; the rest of Dispose + // still has a subscribed connection to let go of. + var telemetry = new ThrowOnMaintenanceEventProvider { ReportingThrows = true }; + var sut = await StartedAsync(telemetry); + var source = (CancellationTokenSource)typeof(RedisPlannedMaintenance) + .GetField("_cancellationTokenSource", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(sut)!; + source.Token.Register(() => throw new InvalidOperationException("callback boom")); + + var dispose = () => sut.Dispose(); + + dispose.Should().NotThrow(); + _multiplexer.Received(1).Dispose(); + } + + [Fact] + public async Task A_probe_run_whose_start_could_not_be_announced_keeps_probing() + { + // Announcing is not what the run is for: losing the event must not cost the probing and the ForceReconnect + // that are the Azure route's whole recovery. + var telemetry = new ThrowOnMaintenanceEventProvider { FailingEvent = "Redis.MaintenanceStarted" }; + var sut = await StartedAsync(telemetry); + + RaiseMaintenance(AzureEvent()); + await WaitForRefusalAsync(telemetry); + + // Long enough for the run to have died here, had the refusal ended it. + await Task.Delay(200, TestContext.Current.CancellationToken); + + sut.InProgress.Should().BeTrue("a refused announcement must not end the probe run"); + telemetry.Events.Should().NotContain("Redis.MaintenanceEnded", "a start that was never announced has no end to announce"); + } + + [Fact] + public async Task A_probe_run_whose_end_could_not_be_announced_reports_that_rather_than_faulting() + { + var telemetry = new ThrowOnMaintenanceEventProvider { FailingEvent = "Redis.MaintenanceEnded" }; + var sut = await StartedAsync(telemetry); + + RaiseMaintenance(AzureEvent()); + for (var i = 0; i < 500 && !telemetry.Events.Contains("Redis.MaintenanceStarted"); i++) + { + await Task.Delay(10, TestContext.Current.CancellationToken); + } + + telemetry.Events.Should().Contain("Redis.MaintenanceStarted", "the run announced its start, so it has an end to announce"); + + sut.Dispose(); + await WaitForRefusalAsync(telemetry); + + telemetry.Exceptions.Should().Contain(ThrowOnMaintenanceEventProvider.Failure, "the refusal is reported rather than left as an unobserved fault"); + } + + [Fact] + public async Task Azure_notifications_that_could_not_be_parsed_are_each_recorded() + { + // A payload the client cannot parse leaves every field at its default, RawMessage included, so keying on + // them would collapse two unrelated notifications into one. + await StartedAsync(); + + RaiseMaintenance(AzureEventFrom("garbage")); + RaiseMaintenance(AzureEventFrom("nonsense")); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(2, "neither carries anything to tell it apart by"); + } + + [Fact] + public async Task An_Azure_notification_with_an_unrecognised_type_is_still_collapsed() + { + // Unknown type but the rest parsed: it has fields to be told apart by, so the claim still holds. + await StartedAsync(); + const string Payload = "NotificationType|SomethingNew|StartTimeInUTC|2026-09-19T00:00:00|IsReplica|False|IPAddress|127.0.0.1|SSLPort|6380|NonSSLPort|6379"; + + RaiseMaintenance(AzureEventFrom(Payload)); + RaiseMaintenance(AzureEventFrom(Payload)); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(1, "an unrecognised type is not an unparsed payload"); + } + + [Fact] + public async Task An_Azure_notification_carrying_only_a_new_type_is_still_collapsed() + { + // The client keeps the type string it did not recognise, so this parsed -- it is identified by that + // string alone, where a payload it could not read at all keeps the default. + await StartedAsync(); + const string Payload = "NotificationType|SomethingNew"; + + RaiseMaintenance(AzureEventFrom(Payload)); + RaiseMaintenance(AzureEventFrom(Payload)); + + _telemetry.Events.Count(e => e.Name == "Redis.Maintenance").Should().Be(1, "the type string is an identity"); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task Recorded_timestamps_round_trip_whichever_route_carried_them(bool push) + { + // One field, one format: a query over Redis.Maintenance should not have to guess which route wrote it. + await StartedAsync(); + + if (push) + { + // Push frames are ignored on the maintenance connection, so this route carries them. + RaiseOnCommandConnection(PushEvent(MaintenanceNotificationType.Migrating, announced: TimeSpan.FromMinutes(5))); + } + else + { + RaiseMaintenance(AzureEvent()); + } + + var recorded = _telemetry.Events.Should().ContainSingle(e => e.Name == "Redis.Maintenance").Which.Properties!; + foreach (var field in new[] { "ReceivedTimeUtc", "StartTimeUtc" }) + { + var value = recorded[field]; + value.Should().NotBeEmpty(); + var parse = () => DateTimeOffset.ParseExact(value, "O", CultureInfo.InvariantCulture); + parse.Should().NotThrow($"{field} must be round-trippable"); + } + } + + /// Disposes every service started here; a probe loop left running writes through later tests. + public void Dispose() + { + foreach (var started in _started) + { + started.Dispose(); + } + + GC.SuppressFinalize(this); + } + + /// An Azure notification built from an arbitrary payload, parsed or not. + private static AzureMaintenanceEvent AzureEventFrom(string payload) => + (AzureMaintenanceEvent)Activator.CreateInstance( + typeof(AzureMaintenanceEvent), BindingFlags.Instance | BindingFlags.NonPublic, null, [payload], null)!; + + private static AzureMaintenanceEvent AzureEvent(string startTime = "2026-09-19T00:00:00") => + (AzureMaintenanceEvent)Activator.CreateInstance( + typeof(AzureMaintenanceEvent), + BindingFlags.Instance | BindingFlags.NonPublic, + null, + [$"NotificationType|NodeMaintenanceStarting|StartTimeInUTC|{startTime}|IsReplica|False|IPAddress|127.0.0.1|SSLPort|6380|NonSSLPort|6379"], + null)!; + + /// A connection attempt that never succeeds, but that ends when the service is disposed. + private static Task NeverConnects(CancellationToken cancellationToken) + { + var pending = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var registration = cancellationToken.Register(() => pending.TrySetCanceled(cancellationToken)); + _ = pending.Task.ContinueWith(_ => registration.Dispose(), TaskScheduler.Default); + return pending.Task; + } + + /// A notification from a source this library does not model, carrying the given payload. + private static ServerMaintenanceEvent UnmodelledEvent(string rawMessage) + { + var unmodelled = (ServerMaintenanceEvent)Activator.CreateInstance( + typeof(ServerMaintenanceEvent), BindingFlags.Instance | BindingFlags.NonPublic, null, null, null)!; + typeof(ServerMaintenanceEvent).GetProperty(nameof(ServerMaintenanceEvent.RawMessage))! + .SetValue(unmodelled, rawMessage); + return unmodelled; + } + + /// Null is a frame whose sequence could not be read: zero, described as unknown. + private static PushMaintenanceEvent PushEvent(MaintenanceNotificationType type, long? sequenceId = 1, TimeSpan? announced = null) => + (PushMaintenanceEvent)Activator.CreateInstance( + typeof(PushMaintenanceEvent), + BindingFlags.Instance | BindingFlags.NonPublic, + null, + [ + type, + sequenceId ?? 0L, + (EndPoint)new DnsEndPoint("node", 6379), + announced, + (EndPoint?)null, + "payload", + $"{type.ToString().ToUpperInvariant()} seq={sequenceId?.ToString(CultureInfo.InvariantCulture) ?? "?"} payload", + Array.Empty(), + ], + null)!; + + private static async Task WaitForRefusalAsync(ThrowOnMaintenanceEventProvider telemetry) + { + for (var i = 0; i < 500 && telemetry.RefusedAttempts == 0; i++) + { + await Task.Delay(10, TestContext.Current.CancellationToken); + } + + telemetry.RefusedAttempts.Should().Be(1, "the announcement was attempted and refused"); + } + + /// As the maintenance connection sees it — where Azure's pub/sub copy arrives. + private void RaiseMaintenance(ServerMaintenanceEvent e) => + _multiplexer.ServerMaintenanceEvent += Raise.Event>(_multiplexer, e); + + /// As the connection carrying commands sees it — the route push frames are taken from. + private void RaiseOnCommandConnection(ServerMaintenanceEvent e) => + _connector.ServerMaintenance += Raise.Event>(_connector, e); + + private async Task StartedWithoutMaintenanceConnectionAsync() + { + var options = Options.Create(new RedisConnectionOptions { ConnectionString = "localhost:6379" }); + var factory = Substitute.For(); + // CA2012: arranging a ValueTask-returning member with NSubstitute means calling it and handing the + // result to Returns. There is no shape of this that awaits it, and nothing ever consumes it. +#pragma warning disable CA2012 + factory.CreateAsync(Arg.Any(), Arg.Any()) + .Returns(call => new ValueTask(NeverConnects(call.Arg()))); +#pragma warning restore CA2012 + + var sut = new RedisPlannedMaintenance( + _telemetry, + _connector, + new RedisConfigurationOptionsProvider(NullLoggerFactory.Instance, options), + factory, + NullLogger.Instance, + options, + null, + _clock); + + _started.Add(sut); + await sut.StartAsync(TestContext.Current.CancellationToken); + return sut; + } + + private async Task StartedAsync(ICachingTelemetryProvider? telemetry = null) + { + var options = Options.Create(new RedisConnectionOptions { ConnectionString = "localhost:6379" }); + var factory = Substitute.For(); + // CA2012: arranging a ValueTask-returning member with NSubstitute means calling it and handing the + // result to Returns. There is no shape of this that awaits it, and nothing ever consumes it. +#pragma warning disable CA2012 + factory.CreateAsync(Arg.Any(), Arg.Any()) + .Returns(_ => new ValueTask(_multiplexer)); +#pragma warning restore CA2012 + + var sut = new RedisPlannedMaintenance( + telemetry ?? _telemetry, + _connector, + new RedisConfigurationOptionsProvider(NullLoggerFactory.Instance, options), + factory, + NullLogger.Instance, + options, + null, + _clock); + + _started.Add(sut); + await sut.StartAsync(TestContext.Current.CancellationToken); + + // StartAsync subscribes on a background task. The multiplexer field is assigned after the handler is + // attached, so seeing it set means an event raised now will be delivered. + var field = typeof(RedisPlannedMaintenance).GetField("_multiplexer", BindingFlags.Instance | BindingFlags.NonPublic)!; + for (var i = 0; i < 200 && field.GetValue(sut) is null; i++) + { + await Task.Delay(10, TestContext.Current.CancellationToken); + } + + field.GetValue(sut).Should().NotBeNull("the maintenance subscription must be established before the test raises an event"); + return sut; + } + + /// Fails the recording step, the way a telemetry sink under load would. + private sealed class ThrowOnMaintenanceEventProvider(int failures = int.MaxValue) : ICachingTelemetryProvider + { + public static readonly InvalidOperationException Failure = new("telemetry boom"); + + private readonly object _gate = new(); + private readonly List _exceptions = []; + private readonly List _events = []; + private int _remaining = failures; + private int _refused; + + /// Leaves the boundary with nowhere to report. + public bool ReportingThrows { get; init; } + + public string FailingEvent { get; init; } = "Redis.Maintenance"; + + /// Counts the attempts that were refused, so a test can wait for one without racing it. + public int RefusedAttempts => Volatile.Read(ref _refused); + + public IReadOnlyList Exceptions => Snapshot(_exceptions); + + public IReadOnlyList Events => Snapshot(_events); + + public void TrackEvent(string eventName, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) + { + if (eventName == FailingEvent && Interlocked.Decrement(ref _remaining) >= 0) + { + Interlocked.Increment(ref _refused); + throw Failure; + } + + lock (_gate) + { + _events.Add(eventName); + } + } + + public void TrackException(Exception ex, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) + { + if (ReportingThrows) + { + throw new InvalidOperationException("sink boom"); + } + + lock (_gate) + { + _exceptions.Add(ex); + } + } + + private T[] Snapshot(List source) + { + lock (_gate) + { + return source.ToArray(); + } + } + } + + /// A clock whose timestamps move only when a test moves them, and whose wall reading can disagree. + private sealed class MovableClock : TimeProvider + { + private long _timestamp = 1_000_000; + private TimeSpan _wallOffset; + + public override long TimestampFrequency => TimeSpan.TicksPerSecond; + + public override DateTimeOffset GetUtcNow() => + new DateTimeOffset(2026, 9, 19, 0, 0, 0, TimeSpan.Zero).AddTicks(_timestamp) + _wallOffset; + + public override long GetTimestamp() => _timestamp; + + public void Advance(TimeSpan by) => _timestamp += by.Ticks; + + /// What an NTP correction does to the wall clock and not to the timestamps. + public void CorrectWallClock(TimeSpan by) => _wallOffset += by; + } +} + +#pragma warning restore SER010