From a6496c2fcaed6c6bf5f62e4b57a27e19908486b2 Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Mon, 21 Sep 2026 22:40:41 +0300 Subject: [PATCH] feat(cache): take maintenance notices from push frames as well as pub/sub RedisPlannedMaintenance matched AzureMaintenanceEvent and returned on anything else. That subtype is the pub/sub notification Azure Cache for Redis publishes on AzureRedisEvents. Redis Enterprise and Redis Cloud send RESP3 push frames instead, which surface as PushMaintenanceEvent, and those fell through the type check and were lost. resilience.md already carried the consequence as a known limitation. Azure Managed Redis counts as that kind of provider too, though its servers do not emit the frames yet. Both are handled now, and a source neither subtype models is recorded from the base event rather than dropped -- matching one subtype and returning is precisely what hid the push kind, and upstream invites other vendors to add their own. Each route is handled on the connection where it carries meaning. Azure's is a pub/sub broadcast every connection receives, so it is taken on the planned-maintenance connection and the command connection's copy ignored, which keeps a single notification from counting twice. Push frames are per connection -- a MOVING names the replacement for the connection it arrived on -- so they are taken from the one carrying commands, reached through the new IRedisConnector.ServerMaintenance. The connector re-wires that on every multiplexer it builds, so one subscription survives a ForceReconnect where subscribing to the multiplexer directly would not. They are recorded and not acted on. The client relaxes timeouts, re-reads topology and moves off a departing endpoint by itself, so probing -- which force-reconnects on a failed write -- would fight the very handoff it exists to survive. Reporting an announced disruption through IRedisPlannedMaintenance.InProgress needs a window whose lifecycle has to agree with the Azure probe loop it runs alongside, which is a change of its own; InProgress stays driven by the Azure route here. RedisConnectionOptions.MaintenanceNotifications is the opt-in, since 3.3.0 ships the feature off unless asked and recognising a hostname does not amount to asking. It takes RedisMaintenanceNotifications, an enum of our own: SER010 rides on StackExchange's type, so exposing theirs would raise that diagnostic in every consumer that sets the option -- in a public library whose consumers may be on AWS, Valkey or self-hosted Redis and never see one of these notifications at all. An unsupported value is rejected rather than folded into Disabled, because silently not asking is indistinguishable from asking and being refused. A handoff raises ConnectionFailed with FailureType MaintenanceHandoff, which the connector tracked as Redis.ConnectionFailed and ConnectionStateMonitor tracked again -- an alert on the event that advance notice exists to make uneventful. Both track it as Redis.MaintenanceHandoff now, and it is still raised to subscribers, because the connection did drop. A connection event raised by this change, and the ones already here, now reach every subscriber: RedisConnector and ConnectionStateMonitor invoked their handlers as one multicast, which stops at the first that throws. ForceReconnect caught it, but around the whole invocation list rather than 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 can throw ObjectDisposedException when a dispose races the notification -- one such throw left every subscriber after it detached until the next reconnect. IsolationExtensions.RaiseIsolated walks the invocation list and tracks each exception on its own; TrackIsolationFailure guards the report itself, since a sink that cannot take it would put the exception back on the path the boundary exists to keep clear. Every catch in RedisConnector and RedisPlannedMaintenance that exists so the surrounding work carries on now reports through it -- including GetVersion, whose Lazy would otherwise have cached the failure for the life of the process. RedisPlannedMaintenance also claimed a notification identity before recording it, so a handler that threw suppressed every copy for 30 seconds while recording nothing; the claim is now released unless the recording succeeded. A probe run marks itself started only once it has announced that, so a refused announcement no longer kills the worker before the probe loop. An Azure payload the client could not parse leaves every field at its default, so those are recorded individually rather than sharing one identity. And a connection candidate the swap rejected no longer stays the newest multiplexer. Signed-off-by: Cosmin Staicu --- CHANGELOG.md | 100 ++- docs/how-to/resilience.md | 42 +- docs/recipes/redis-health-check.md | 2 +- docs/reference/settings.md | 1 + .../appsettings.all.json | 7 + src/UiPath.Caching/PublicAPI.Unshipped.txt | 9 + .../Redis/ConnectionStateMonitor.cs | 16 +- src/UiPath.Caching/Redis/IRedisConnector.cs | 9 + .../RedisConfigurationOptionsProvider.cs | 28 +- .../Redis/RedisConnectionOptions.cs | 3 + src/UiPath.Caching/Redis/RedisConnector.cs | 152 +++- src/UiPath.Caching/Redis/RedisHealthCheck.cs | 2 +- .../Redis/RedisMaintenanceNotifications.cs | 15 + .../Redis/RedisPlannedMaintenance.cs | 441 +++++++++-- src/UiPath.Caching/TelemetrySafeguards.cs | 58 ++ .../ConnectionStateMonitorHandoffTests.cs | 66 ++ .../ConnectionStateMonitorMulticastTests.cs | 98 +++ .../Redis/RedisConnectorLifecycleTests.cs | 399 ++++++++++ .../Redis/RedisConnectorTests.cs | 57 ++ .../RedisPlannedMaintenanceRoutingTests.cs | 685 ++++++++++++++++++ 20 files changed, 2088 insertions(+), 102 deletions(-) create mode 100644 src/UiPath.Caching/Redis/RedisMaintenanceNotifications.cs create mode 100644 src/UiPath.Caching/TelemetrySafeguards.cs create mode 100644 tests/UiPath.Caching.Tests/Redis/ConnectionStateMonitorHandoffTests.cs create mode 100644 tests/UiPath.Caching.Tests/Redis/ConnectionStateMonitorMulticastTests.cs create mode 100644 tests/UiPath.Caching.Tests/Redis/RedisPlannedMaintenanceRoutingTests.cs 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