From 90306f0ccb50f8a300191414a80eaaa55dcb2d24 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:21:33 -0400 Subject: [PATCH] fix(communication): streaming consumers re-resolve site gRPC addresses on every reopen; spec matches (review 2026-10-21 03/R-2, 03/D-2) --- docs/requirements/Component-Communication.md | 4 +- .../Actors/DebugStreamBridgeActor.cs | 122 ++++++++++++++- .../Actors/SiteAlarmAggregatorActor.cs | 139 +++++++++++++++++- .../DebugStreamService.cs | 35 +++-- .../Grpc/SiteStreamEndpoints.cs | 49 ++++++ .../SiteAlarmLiveCacheService.cs | 39 +++-- .../Grpc/DebugStreamBridgeActorTests.cs | 27 ++++ .../Grpc/SiteAlarmAggregatorActorTests.cs | 57 +++++++ 8 files changed, 438 insertions(+), 34 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamEndpoints.cs diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index c6bb605d..2bf52562 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -100,7 +100,7 @@ Delivered 2026-07-10 (`docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.m - **DebugStreamService**: Singleton service that manages debug stream sessions. Resolves instance ID to unique name and site, creates and tears down `DebugStreamBridgeActor` instances, and provides a clean API for both Blazor components and the SignalR hub. Injects `SiteStreamGrpcClientFactory` for gRPC stream creation. - **DebugStreamBridgeActor**: One per active debug session. Opens a gRPC streaming subscription via `SiteStreamGrpcClient` and receives real-time events via callback. Also receives the initial `DebugViewSnapshot` over gRPC command/control (`SiteCommandService`). Forwards all events to the consumer via callbacks. Handles gRPC stream errors with reconnection logic: tries the other site node endpoint, retries with backoff (max 3 retries), terminates the session if all retries fail. - **SiteStreamGrpcClient**: Per-site gRPC client that manages `GrpcChannel` instances and streaming subscriptions. Reads from the gRPC response stream in a background task, unpacks each frame via `ForEachEvent` (a plain frame passes through; a batch frame is flattened in order), converts protobuf messages to domain events, and invokes the `onEvent` callback. -- **SiteStreamGrpcClientFactory**: Caches per-site `SiteStreamGrpcClient` instances. Reads `GrpcNodeAAddress` / `GrpcNodeBAddress` from the `Site` entity (loaded by `CentralCommunicationActor`). Falls back to NodeB if NodeA connection fails. Disposes clients on site removal or address change. +- **SiteStreamGrpcClientFactory**: Caches `SiteStreamGrpcClient` instances keyed by `(siteId, endpoint)`. It does not read addresses itself: each streaming consumer (`SiteAlarmLiveCacheService` → `SiteAlarmAggregatorActor`, `DebugStreamService` → `DebugStreamBridgeActor`) resolves `GrpcNodeAAddress` / `GrpcNodeBAddress` from the `Site` row in its own DI scope (a single configured address fills both slots), not from `CentralCommunicationActor`'s cache, and flips NodeA↔NodeB itself on stream error. Before **every reopen** (fault retry, graceful-completion reopen, reconcile-tick self-heal) the consumer re-reads the site row through a `SiteStreamEndpointResolver`, so an address edited while a session or aggregator is running is dialled on its next reconnect; a lookup that fails, times out (10 s) or finds no address reopens against the endpoints already known. A client for an edited-away endpoint stays cached until site removal or process exit (documented trade-off). - **DebugStreamHub**: SignalR hub at `/hubs/debug-stream` for external consumers (e.g., CLI). Authenticates via Basic Auth + LDAP and requires the **Deployment** role. Server-to-client methods: `OnSnapshot`, `OnAttributeChanged`, `OnAlarmChanged`, `OnStreamTerminated`. #### gRPC Proto Definition @@ -358,7 +358,7 @@ flowchart LR Central discovers site addresses through the **configuration database**, not runtime registration: - Each site record in the Sites table includes optional **GrpcNodeAAddress** and **GrpcNodeBAddress** fields containing the site nodes' gRPC endpoints (e.g., `http://host:8083`), used for **both** central→site command/control (`SiteCommandService`) and site→central streaming (`SiteStreamService`). The legacy **NodeAAddress** / **NodeBAddress** Akka fields are retained but no longer drive cross-cluster dialling — Akka remoting is now intra-cluster only. -- The **CentralCommunicationActor** loads all site addresses from the database at startup and builds a per-site gRPC-endpoint cache. `GrpcSiteTransport` uses it (via a `SitePairChannelProvider`) to dial each site's `SiteCommandService` as a **per-site NodeA→NodeB failover channel pair**; the **SiteStreamGrpcClientFactory** uses the same `GrpcNodeAAddress` / `GrpcNodeBAddress` for per-site streaming channels. +- The **CentralCommunicationActor** loads all site addresses from the database at startup and builds a per-site gRPC-endpoint cache. `GrpcSiteTransport` uses it (via a `SitePairChannelProvider`) to dial each site's `SiteCommandService` as a **per-site NodeA→NodeB failover channel pair**; the streaming consumers (Alarm Summary aggregator, Debug View bridge) read the same `GrpcNodeAAddress` / `GrpcNodeBAddress` columns directly from the repository at start and again before every stream reopen (see *SiteStreamGrpcClientFactory* above), independent of this cache. - The address cache is **refreshed every 60 seconds** and **on-demand** when site records are added, edited, or deleted via the Central UI or CLI. Command-plane channels are rebuilt when a site's gRPC endpoints change. At most **one** DB load is in flight: a tick or on-demand refresh that arrives mid-load is coalesced into a single re-run after it, each load is bounded by a deadline (`QueryTimeout`, default 30 s) so a wedged read cannot pin the refresh, and every result carries a monotonic sequence so an older result arriving after a newer one is dropped rather than re-applying stale addresses. - A **malformed** `GrpcNodeAAddress`/`GrpcNodeBAddress` (not an absolute `http://`/`https://` URL with a host, or one the gRPC channel cannot be built from) is **rejected per node at refresh time**: `SitePairChannelProvider.UpdateSite` builds every replacement channel *before* parking the old one, so a rejected side keeps its previous channel and endpoint (the other side's valid change still applies), and the rejection is logged at **Error** naming the site and column (once per distinct bad value). `GrpcSiteTransport.ReconcileSites` reconciles each site in its own try/catch and `CentralCommunicationActor` guards the reconcile call, so one bad row can neither stop the other sites in the tick from being refreshed nor restart the actor. Write-path validation (Central UI/CLI/bundle import) is the first line of defence; this is the runtime backstop. - When routing a command to a site, central calls `SiteCommandService` on the site's active gRPC endpoint; on error it flips to the other node endpoint. The failover is an explicit NodeA→NodeB channel-pair flip, not internal transport routing. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs index 908156c2..c4ced0d8 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs @@ -40,8 +40,25 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers private readonly Action _onEvent; private readonly Action _onTerminated; private readonly SiteStreamGrpcClientFactory _grpcFactory; - private readonly string _grpcNodeAAddress; - private readonly string _grpcNodeBAddress; + // Mutable: refreshed from _endpointResolver before each reconnect (review 2026-10-21 03/R-2). + private string _grpcNodeAAddress; + private string _grpcNodeBAddress; + + /// + /// Re-reads the site's current gRPC endpoints before every reconnect, so an address edited + /// mid-session is picked up rather than retried against the start-time address. Null (tests, + /// legacy ctor) keeps the ctor-supplied addresses for the session's life. + /// + private readonly SiteStreamEndpointResolver? _endpointResolver; + + /// True while a pre-reconnect endpoint lookup is running. Actor-thread only. + private bool _resolveInFlight; + + /// The endpoint the current/last stream was opened against (for its unsubscribe). Actor-thread only. + private string? _streamEndpoint; + + /// Upper bound on one pre-reconnect endpoint lookup; on expiry the reconnect uses the known endpoints. + internal static readonly TimeSpan EndpointResolveTimeout = TimeSpan.FromSeconds(10); private const int MaxRetries = 3; private const string ReconnectTimerKey = "grpc-reconnect"; @@ -178,7 +195,38 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers SiteStreamGrpcClientFactory grpcFactory, string grpcNodeAAddress, string grpcNodeBAddress) + : this(siteIdentifier, instanceUniqueName, correlationId, centralCommunicationActor, onEvent, + onTerminated, grpcFactory, grpcNodeAAddress, grpcNodeBAddress, endpointResolver: null) { + } + + /// + /// Initializes the bridge with an endpoint resolver consulted before every reconnect + /// (review 2026-10-21 03/R-2). The address arguments seed the first open. + /// + /// Site identifier for targeting site-addressed messages and logging. + /// Unique name of the instance whose debug stream is being bridged. + /// Correlation id for the debug session. + /// Actor used to forward site-addressed messages to the site. + /// Callback invoked on each received debug event. + /// Callback invoked when the stream terminates. + /// Factory for creating gRPC streaming clients. + /// gRPC address of the site's node A at session start. + /// gRPC address of the site's node B at session start. + /// Re-reads the site's endpoints before each reconnect; null keeps the start-time ones. + public DebugStreamBridgeActor( + string siteIdentifier, + string instanceUniqueName, + string correlationId, + IActorRef centralCommunicationActor, + Action onEvent, + Action onTerminated, + SiteStreamGrpcClientFactory grpcFactory, + string grpcNodeAAddress, + string grpcNodeBAddress, + SiteStreamEndpointResolver? endpointResolver) + { + _endpointResolver = endpointResolver; _siteIdentifier = siteIdentifier; _instanceUniqueName = instanceUniqueName; _correlationId = correlationId; @@ -318,7 +366,16 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers }); // Scheduled reconnection - Receive(_ => OpenGrpcStream()); + Receive(_ => ReopenGrpcStream()); + + // Pre-reconnect endpoint lookup finished (review 2026-10-21 03/R-2). + Receive(msg => + { + _resolveInFlight = false; + if (_stopped) return; + ApplyResolvedEndpoints(msg); + OpenGrpcStream(); + }); // Consumer requests stop Receive(_ => @@ -574,11 +631,64 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers base.PostStop(); } + /// + /// Reconnects the stream, first re-reading the site's endpoints when a resolver is configured + /// (review 2026-10-21 03/R-2), so a session whose site address was edited reconnects to the + /// new address instead of burning its retries on the old one. A lookup that fails, times out + /// or finds no address reconnects against the endpoints already known. + /// + private void ReopenGrpcStream() + { + if (_stopped) return; + if (_endpointResolver is null) + { + OpenGrpcStream(); + return; + } + if (_resolveInFlight) return; // the pending lookup opens when it completes + + _resolveInFlight = true; + var resolver = _endpointResolver; + Task.Run(async () => + { + using var cts = new CancellationTokenSource(EndpointResolveTimeout); + try + { + return new SiteStreamEndpointsResolved(await resolver(cts.Token).ConfigureAwait(false), null); + } + catch (Exception ex) + { + return new SiteStreamEndpointsResolved(null, ex); + } + }).PipeTo(Self); + } + + private void ApplyResolvedEndpoints(SiteStreamEndpointsResolved msg) + { + if (msg.Error is not null) + { + _log.Warning("Could not re-read gRPC endpoints for site {0} before reconnecting ({1}); using {2} / {3}", + _siteIdentifier, msg.Error.Message, _grpcNodeAAddress, _grpcNodeBAddress); + return; + } + if (msg.Endpoints is not { } endpoints + || (endpoints.NodeA == _grpcNodeAAddress && endpoints.NodeB == _grpcNodeBAddress)) + { + return; + } + + _log.Info("Site {0} gRPC endpoints changed ({1} / {2} -> {3} / {4}); reconnecting the debug stream for {5} to the new address", + _siteIdentifier, _grpcNodeAAddress, _grpcNodeBAddress, endpoints.NodeA, endpoints.NodeB, _instanceUniqueName); + _grpcNodeAAddress = endpoints.NodeA; + _grpcNodeBAddress = endpoints.NodeB; + } + private void OpenGrpcStream() { if (_stopped) return; var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress; + _streamEndpoint = endpoint; _log.Info("Opening gRPC stream for {0} to {1}", _instanceUniqueName, endpoint); _grpcCts?.Cancel(); @@ -637,7 +747,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers _grpcCts?.Cancel(); _grpcCts?.Dispose(); _grpcCts = null; - _grpcFactory.TryGet(_siteIdentifier, _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress) + _grpcFactory.TryGet(_siteIdentifier, _streamEndpoint ?? (_useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress)) ?.Unsubscribe(_correlationId); Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectGrpcStream(), ReconnectDelay); @@ -668,7 +778,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers // alive -- propagates gRPC cancellation to the site so its SiteStreamGrpcServer // stops the StreamRelayActor for this correlation ID, rather than leaving a // zombie relay actor until TCP RST / keepalive eventually detects the loss. - var previousEndpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress; + var previousEndpoint = _streamEndpoint ?? (_useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress); // TryGet, not GetOrCreate: unsubscribing a failed stream must never open a // fresh channel (and, with (site,endpoint) keying, must never touch another // session's healthy channel). Absent client => the channel is already gone @@ -697,7 +807,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers // TryGet, not GetOrCreate: teardown must never open a fresh channel just to // unsubscribe. Absent client => nothing to cancel. - var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress; + var endpoint = _streamEndpoint ?? (_useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress); _grpcFactory.TryGet(_siteIdentifier, endpoint)?.Unsubscribe(_correlationId); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs index ceb49fa3..22ef5b05 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs @@ -53,8 +53,25 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers private readonly Func>> _seedFn; private readonly Action, bool> _publish; private readonly SiteStreamGrpcClientFactory _grpcFactory; - private readonly string _grpcNodeAAddress; - private readonly string _grpcNodeBAddress; + // Mutable: refreshed from _endpointResolver before each reopen (review 2026-10-21 03/R-2). + private string _grpcNodeAAddress; + private string _grpcNodeBAddress; + + /// + /// Re-reads the site's current gRPC endpoints before every reopen, so an address edited while + /// viewers are attached is picked up instead of reopening against the start-time address + /// forever. Null (tests, legacy ctor) keeps the ctor-supplied addresses for the actor's life. + /// + private readonly SiteStreamEndpointResolver? _endpointResolver; + + /// True while a pre-reopen endpoint lookup is running. Actor-thread only. + private bool _resolveInFlight; + + /// The endpoint the current/last stream was opened against (for its unsubscribe). Actor-thread only. + private string? _streamEndpoint; + + /// Upper bound on one pre-reopen endpoint lookup; on expiry the reopen uses the known endpoints. + internal static readonly TimeSpan EndpointResolveTimeout = TimeSpan.FromSeconds(10); private readonly TimeSpan _reconcileInterval; private readonly TimeSpan _publishCoalesce; private readonly double _reconcileJitterFraction; @@ -302,7 +319,47 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers TimeSpan stabilityWindow, double reconcileJitterFraction = 0.0, TimeSpan? openAttemptTimeout = null) + : this(siteIdentifier, correlationId, seedFn, publish, grpcFactory, grpcNodeAAddress, grpcNodeBAddress, + reconcileInterval, publishCoalesce, reconnectDelay, stabilityWindow, reconcileJitterFraction, + openAttemptTimeout, endpointResolver: null) { + } + + /// + /// Initializes the aggregator with an endpoint resolver that is consulted before every stream + /// reopen (review 2026-10-21 03/R-2). The address arguments seed the first open. + /// + /// Site identifier (used for gRPC client keying and logging). + /// Correlation id for the site-wide alarm stream subscription. + /// Snapshot fan-out returning the whole site's current alarm rows. + /// Publishes a fresh immutable cache snapshot plus the stream-liveness flag. + /// Factory caching one gRPC client per (site, endpoint). + /// gRPC address of the site's node A at start. + /// gRPC address of the site's node B at start. + /// Periodic reconcile snapshot cadence. + /// Publish-coalescing window for live deltas. + /// Delay between gRPC reconnection attempts. + /// How long a fresh gRPC stream must stay up before its retry budget recovers. + /// Fraction of the reconcile interval added as per-tick jitter. + /// Deadline on one stream open attempt; null takes the default. + /// Re-reads the site's endpoints before each reopen; null keeps the start-time ones. + public SiteAlarmAggregatorActor( + string siteIdentifier, + string correlationId, + Func>> seedFn, + Action, bool> publish, + SiteStreamGrpcClientFactory grpcFactory, + string grpcNodeAAddress, + string grpcNodeBAddress, + TimeSpan reconcileInterval, + TimeSpan publishCoalesce, + TimeSpan reconnectDelay, + TimeSpan stabilityWindow, + double reconcileJitterFraction, + TimeSpan? openAttemptTimeout, + SiteStreamEndpointResolver? endpointResolver) + { + _endpointResolver = endpointResolver; _siteIdentifier = siteIdentifier; _correlationId = correlationId; _seedFn = seedFn; @@ -414,7 +471,16 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers HandleGrpcCompleted(); }); - Receive(_ => OpenGrpcStream()); + Receive(_ => ReopenGrpcStream()); + + // Pre-reopen endpoint lookup finished (review 2026-10-21 03/R-2). + Receive(msg => + { + _resolveInFlight = false; + if (_stopped) return; + ApplyResolvedEndpoints(msg); + OpenGrpcStream(); + }); // Owning service asks us to stop (last viewer left + linger elapsed). Receive(_ => @@ -513,7 +579,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // Telemetry: a reconcile-driven reopen after the stream was given up is a reconnect. ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect(); _seedOnConnect = true; - OpenGrpcStream(); + ReopenGrpcStream(); reopening = true; } @@ -843,6 +909,66 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // ── gRPC stream lifecycle (mirrors DebugStreamBridgeActor) ────────────────── + /// + /// Reopens the stream, first re-reading the site's endpoints when a resolver is configured + /// (review 2026-10-21 03/R-2): an address edited while the aggregator runs used to be + /// ignored, so every reopen — retries and the reconcile tick's self-heal alike — dialled + /// the dead start-time address for as long as viewers stayed attached. The lookup holds the + /// open-in-flight guard so a reconcile tick cannot start a second reopen meanwhile; a lookup + /// that fails, times out or finds no address reopens against the endpoints already known. + /// + private void ReopenGrpcStream() + { + if (_stopped) return; + if (_endpointResolver is null) + { + OpenGrpcStream(); + return; + } + if (_resolveInFlight) return; // the pending lookup opens when it completes + + _resolveInFlight = true; + _openInFlight = true; + var resolver = _endpointResolver; + var lifetime = _lifetimeCts?.Token ?? CancellationToken.None; + Task.Run(async () => + { + using var cts = CancellationTokenSource.CreateLinkedTokenSource(lifetime); + cts.CancelAfter(EndpointResolveTimeout); + try + { + return new SiteStreamEndpointsResolved(await resolver(cts.Token).ConfigureAwait(false), null); + } + catch (Exception ex) + { + return new SiteStreamEndpointsResolved(null, ex); + } + }).PipeTo(Self); + } + + private void ApplyResolvedEndpoints(SiteStreamEndpointsResolved msg) + { + if (msg.Error is not null) + { + _log.Warning("Could not re-read gRPC endpoints for site {0} before reopening ({1}); using {2} / {3}", + _siteIdentifier, msg.Error.Message, _grpcNodeAAddress, _grpcNodeBAddress); + return; + } + if (msg.Endpoints is not { } endpoints) + { + _log.Warning("Site {0} has no gRPC endpoints configured any more; reopening against {1} / {2}", + _siteIdentifier, _grpcNodeAAddress, _grpcNodeBAddress); + return; + } + if (endpoints.NodeA == _grpcNodeAAddress && endpoints.NodeB == _grpcNodeBAddress) + return; + + _log.Info("Site {0} gRPC endpoints changed ({1} / {2} -> {3} / {4}); reopening the alarm stream against the new address", + _siteIdentifier, _grpcNodeAAddress, _grpcNodeBAddress, endpoints.NodeA, endpoints.NodeB); + _grpcNodeAAddress = endpoints.NodeA; + _grpcNodeBAddress = endpoints.NodeB; + } + private void OpenGrpcStream() { if (_stopped) return; @@ -850,6 +976,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // The stream is not live again until the server actually accepts it — liveness flips // on GrpcAlarmStreamConnected, not on the attempt. var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress; + _streamEndpoint = endpoint; _log.Info("Opening site-alarm gRPC stream for {0} to {1}", _siteIdentifier, endpoint); _grpcCts?.Cancel(); @@ -962,7 +1089,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // Unsubscribe the failed stream on the previous endpoint (TryGet, never // GetOrCreate) so the site releases its relay actor instead of leaving a zombie. - var previousEndpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress; + var previousEndpoint = _streamEndpoint ?? (_useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress); _grpcFactory.TryGet(_siteIdentifier, previousEndpoint)?.Unsubscribe(_correlationId); // Flip to the other node. @@ -1016,7 +1143,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers _grpcCts?.Dispose(); _grpcCts = null; - var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress; + var endpoint = _streamEndpoint ?? (_useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress); _grpcFactory.TryGet(_siteIdentifier, endpoint)?.Unsubscribe(_correlationId); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs index 66893f2c..bade9cea 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs @@ -100,6 +100,7 @@ public class DebugStreamService : IDisposable string siteIdentifier; string grpcNodeAAddress; string grpcNodeBAddress; + int siteId; using (var scope = _serviceProvider.CreateScope()) { @@ -119,18 +120,19 @@ public class DebugStreamService : IDisposable // sole configured endpoint for both slots, so the bridge actor's NodeA↔NodeB // failover flip degenerates to a reconnect against the same node. Requiring BOTH // addresses made Debug View the only surface that refused a one-node site. - var nodeA = site.GrpcNodeAAddress; - var nodeB = site.GrpcNodeBAddress; - if (string.IsNullOrWhiteSpace(nodeA) && string.IsNullOrWhiteSpace(nodeB)) - { - throw new InvalidOperationException( + var endpoints = SiteStreamEndpoints.FromSite(site) + ?? throw new InvalidOperationException( $"Site {siteIdentifier} has no gRPC node addresses configured."); - } - grpcNodeAAddress = !string.IsNullOrWhiteSpace(nodeA) ? nodeA! : nodeB!; - grpcNodeBAddress = !string.IsNullOrWhiteSpace(nodeB) ? nodeB! : nodeA!; + grpcNodeAAddress = endpoints.NodeA; + grpcNodeBAddress = endpoints.NodeB; + siteId = site.Id; } + // Re-read the site's endpoints before each reconnect so an address edited mid-session + // is used (review 2026-10-21 03/R-2). + SiteStreamEndpointResolver endpointResolver = ct => ResolveEndpointsAsync(siteId, ct); + var sessionId = Guid.NewGuid().ToString("N"); // Capture the initial snapshot via a TaskCompletionSource @@ -167,7 +169,8 @@ public class DebugStreamService : IDisposable onTerminatedWrapper, _grpcClientFactory, grpcNodeAAddress, - grpcNodeBAddress); + grpcNodeBAddress, + endpointResolver); var bridgeActor = system.ActorOf(props, $"debug-stream-{sessionId}"); @@ -210,6 +213,20 @@ public class DebugStreamService : IDisposable return new DebugStreamSession(sessionId, snapshot); } + /// + /// Re-reads a site's current streaming endpoints for a running session's reconnect. + /// + /// Site primary key. + /// Cancellation token bounding the lookup. + /// The site's current endpoints, or null when the site is gone or has none. + private async Task ResolveEndpointsAsync(int siteId, CancellationToken ct) + { + using var scope = _serviceProvider.CreateScope(); + var siteRepo = scope.ServiceProvider.GetRequiredService(); + var site = await siteRepo.GetSiteByIdAsync(siteId, ct); + return site is null ? null : SiteStreamEndpoints.FromSite(site); + } + /// /// Stops an active debug stream session. /// diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamEndpoints.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamEndpoints.cs new file mode 100644 index 00000000..115698d7 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamEndpoints.cs @@ -0,0 +1,49 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +/// +/// The pair of gRPC endpoints a central streaming consumer (Alarm Summary aggregator, Debug View +/// bridge) dials for one site. A single-endpoint site reuses its sole address for both slots, so +/// the consumer's NodeA↔NodeB failover flip degenerates to a reconnect against the same node. +/// +/// NodeA gRPC endpoint (or the sole configured one). +/// NodeB gRPC endpoint (or the sole configured one). +public readonly record struct SiteStreamEndpoints(string NodeA, string NodeB) +{ + /// + /// Builds the endpoint pair from a site's GrpcNodeAAddress/GrpcNodeBAddress. + /// + /// The site row. + /// The pair, or null when the site has neither address configured. + public static SiteStreamEndpoints? FromSite(Site site) + { + ArgumentNullException.ThrowIfNull(site); + var nodeA = site.GrpcNodeAAddress; + var nodeB = site.GrpcNodeBAddress; + if (string.IsNullOrWhiteSpace(nodeA) && string.IsNullOrWhiteSpace(nodeB)) + { + return null; + } + + return new SiteStreamEndpoints( + !string.IsNullOrWhiteSpace(nodeA) ? nodeA! : nodeB!, + !string.IsNullOrWhiteSpace(nodeB) ? nodeB! : nodeA!); + } +} + +/// +/// Re-reads a site's current streaming endpoints from configuration (review 2026-10-21 03/R-2). +/// Streaming consumers call it before every REOPEN so an address edited while a session is running +/// is picked up, instead of reconnecting against the address frozen at session start forever. +/// Returns null when the site is gone or has no gRPC address — the consumer then keeps the +/// endpoints it has. +/// +/// Cancellation token bounding the lookup. +/// The site's current endpoint pair, or null. +public delegate Task SiteStreamEndpointResolver(CancellationToken ct); + +/// Internal actor message: the result of a lookup before a reopen. +/// The resolved pair, or null when unavailable. +/// The lookup failure, when it threw or timed out. +internal sealed record SiteStreamEndpointsResolved(SiteStreamEndpoints? Endpoints, Exception? Error); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs index 9c48a587..aab0f8c1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs @@ -217,6 +217,8 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache return; // logged inside ResolveSiteAsync var (siteIdentifier, grpcA, grpcB) = resolved.Value; + var siteId = entry.SiteId; + SiteStreamEndpointResolver resolver = ct => ResolveEndpointsAsync(siteId, ct); var correlationId = $"site-alarm-{entry.SiteId}-{Guid.NewGuid():N}"; Func>> seedFn = @@ -249,7 +251,10 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache // Open-attempt deadline. Passed explicitly rather than defaulted: // Props.Create takes an expression tree, which may not omit an optional // argument (CS0854). - SiteAlarmAggregatorActor.DefaultOpenAttemptTimeout)); + SiteAlarmAggregatorActor.DefaultOpenAttemptTimeout, + // Re-read the site's gRPC endpoints before every reopen, so an address edit + // reaches a running aggregator (review 2026-10-21 03/R-2). + resolver)); entry.Actor = system.ActorOf(props, $"site-alarm-aggregator-{entry.SiteId}-{Guid.NewGuid():N}"); @@ -315,9 +320,11 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache return null; } - var nodeA = site.GrpcNodeAAddress; - var nodeB = site.GrpcNodeBAddress; - if (string.IsNullOrWhiteSpace(nodeA) && string.IsNullOrWhiteSpace(nodeB)) + // Single-endpoint site (review 02 round 2, N9): reuse the sole configured + // endpoint for both slots — the aggregator's NodeA↔NodeB failover flip + // degenerates to a reconnect against the same node, which is exactly the + // right behavior for a one-node site. + if (SiteStreamEndpoints.FromSite(site) is not { } endpoints) { _logger.LogWarning( "Live alarm cache: site {SiteId} ({SiteIdentifier}) has no gRPC node addresses; " + @@ -325,13 +332,23 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache return null; } - // Single-endpoint site (review 02 round 2, N9): reuse the sole configured - // endpoint for both slots — the aggregator's NodeA↔NodeB failover flip - // degenerates to a reconnect against the same node, which is exactly the - // right behavior for a one-node site. - var grpcA = !string.IsNullOrWhiteSpace(nodeA) ? nodeA! : nodeB!; - var grpcB = !string.IsNullOrWhiteSpace(nodeB) ? nodeB! : nodeA!; - return (site.SiteIdentifier, grpcA, grpcB); + return (site.SiteIdentifier, endpoints.NodeA, endpoints.NodeB); + } + + /// + /// The running aggregator's pre-reopen endpoint lookup (review 2026-10-21 03/R-2): re-reads + /// the site row so an edited GrpcNodeAAddress/GrpcNodeBAddress is used on the + /// next reopen. Null when the site is gone or has no address (the aggregator keeps its own). + /// + /// Site primary key. + /// Cancellation token bounding the lookup. + /// The site's current endpoints, or null. + private async Task ResolveEndpointsAsync(int siteId, CancellationToken ct) + { + using var scope = _serviceProvider.CreateScope(); + var siteRepo = scope.ServiceProvider.GetRequiredService(); + var site = await siteRepo.GetSiteByIdAsync(siteId, ct); + return site is null ? null : SiteStreamEndpoints.FromSite(site); } // ── Snapshot fan-out (the seed / reconcile) ───────────────────────────────── diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs index 0fef8e4b..2d641c75 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs @@ -420,6 +420,33 @@ public class DebugStreamBridgeActorTests : TestKit Assert.Equal("corr-1", factory.ClientFor(GrpcNodeB).SubscribeCalls[0].CorrelationId); } + /// Verifies that a reconnect after the site's address was edited dials the new address (review 2026-10-21 03/R-2). + [Fact] + public void Reconnect_AfterSiteAddressEdit_DialsTheNewAddress() + { + const string NewNodeB = "http://localhost:5400"; + var commProbe = CreateTestProbe(); + var factory = new EndpointTrackingGrpcClientFactory(); + SiteStreamEndpointResolver resolver = + _ => Task.FromResult(new SiteStreamEndpoints(GrpcNodeA, NewNodeB)); + + var props = Props.Create(typeof(DebugStreamBridgeActor), + SiteId, InstanceName, "corr-1", commProbe.Ref, + (Action)(_ => { }), (Action)(() => { }), + factory, GrpcNodeA, GrpcNodeB, resolver); + Sys.ActorOf(props); + commProbe.ExpectMsg(); + + AwaitCondition(() => factory.ClientFor(GrpcNodeA).SubscribeCalls.Count == 1, TimeSpan.FromSeconds(3)); + factory.ClientFor(GrpcNodeA).SubscribeCalls[0].OnError(new Exception("NodeA down")); + + // The failover reconnect goes to the EDITED NodeB, never the start-time one. + AwaitCondition(() => factory.ClientFor(NewNodeB).SubscribeCalls.Count == 1, TimeSpan.FromSeconds(5)); + Assert.Empty(factory.ClientFor(GrpcNodeB).SubscribeCalls); + // The failed stream was released on the endpoint it was opened against. + Assert.Contains("corr-1", factory.ClientFor(GrpcNodeA).UnsubscribedCorrelationIds); + } + // ── graceful (status OK) stream completion is a reconnect trigger ── /// Verifies that a graceful stream completion reopens the stream on the same node. diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs index aca1557b..ea81176c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs @@ -1013,4 +1013,61 @@ public class SiteAlarmAggregatorActorTests : TestKit seed.FaultNext(); AwaitCondition(() => seed.CallCount == 3, TimeSpan.FromSeconds(5)); } + + // ── review 2026-10-21 03/R-2: reopen re-resolves the site's endpoints ── + + /// Verifies that a reopen after the site's address was edited dials the NEW address, not the start-time one. + [Fact] + public void Reopen_AfterSiteAddressEdit_DialsTheNewAddress() + { + const string NewNodeA = "http://localhost:5300"; + const string NewNodeB = "http://localhost:5400"; + var edited = 0; + SiteStreamEndpointResolver resolver = _ => Task.FromResult( + Volatile.Read(ref edited) == 1 ? new(NewNodeA, NewNodeB) : new(GrpcNodeA, GrpcNodeB)); + + var seed = new SeedStub(); + var sink = new PublishSink(); + var factory = new MockSiteAlarmStreamClientFactory(); + Sys.ActorOf(Props.Create(() => new SiteAlarmAggregatorActor( + SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB, + TimeSpan.FromMilliseconds(300), TimeSpan.Zero, TestReconnectDelay, TestStabilityWindow, + 0.0, SiteAlarmAggregatorActor.DefaultOpenAttemptTimeout, resolver))); + + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + + // The operator re-points the site; the running stream then ends gracefully. + Volatile.Write(ref edited, 1); + factory.ClientFor(GrpcNodeA).Subs[0].OnCompleted(); + + // The reconcile tick's reopen dials the edited NodeA, never the stale one again. + AwaitCondition(() => factory.ClientFor(NewNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + Assert.Single(factory.ClientFor(GrpcNodeA).Subs); + + // A fault now fails over to the edited NodeB. + factory.ClientFor(NewNodeA).Subs[0].OnError(new Exception("down")); + AwaitCondition(() => factory.ClientFor(NewNodeB).Subs.Count == 1, TimeSpan.FromSeconds(3)); + Assert.Empty(factory.ClientFor(GrpcNodeB).Subs); + } + + /// Verifies that a failing endpoint lookup still reopens against the known address. + [Fact] + public void Reopen_WhenTheEndpointLookupFails_UsesTheKnownAddress() + { + SiteStreamEndpointResolver resolver = _ => throw new InvalidOperationException("db down"); + var seed = new SeedStub(); + var sink = new PublishSink(); + var factory = new MockSiteAlarmStreamClientFactory(); + Sys.ActorOf(Props.Create(() => new SiteAlarmAggregatorActor( + SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB, + TimeSpan.FromMilliseconds(300), TimeSpan.Zero, TestReconnectDelay, TestStabilityWindow, + 0.0, SiteAlarmAggregatorActor.DefaultOpenAttemptTimeout, resolver))); + + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + factory.ClientFor(GrpcNodeA).Subs[0].OnCompleted(); + + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(3)); + } }