diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index 4aafbfdc..953f8380 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -360,6 +360,7 @@ Central discovers site addresses through the **configuration database**, not run - 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 address cache is **refreshed every 60 seconds** and **on-demand** when site records are added, edited, or deleted via the Central UI or CLI. Channels are rebuilt when a site's gRPC endpoints change. +- 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. - **Heartbeats** from sites serve **health monitoring only** — they do not serve as a registration or address discovery mechanism. - If no gRPC endpoints are configured for a site, commands to that site fail and the caller's call times out. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs index 2b9ddfa4..7e120287 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs @@ -392,8 +392,19 @@ public class CentralCommunicationActor : ReceiveActor private void HandleSiteAddressCacheLoaded(SiteAddressCacheLoaded msg) { // Per-site transport resource reconciliation (build/drop gRPC channel - // pairs). Runs on the actor thread each refresh tick. - _transport.ReconcileSites(msg); + // pairs). Runs on the actor thread each refresh tick. Guarded (review 2026-10-21 03/R-1): + // a throw here would restart this actor, whose PreStart re-reads the Sites table at + // TimeSpan.Zero — a restart loop for as long as a bad row persists — and would skip the + // aggregator prune below. + try + { + _transport.ReconcileSites(msg); + } + catch (Exception ex) + { + _log.Error(ex, "Reconciling site gRPC channels from the refreshed address cache failed; " + + "existing channels are kept until the next refresh"); + } // Self-healing eviction: a site deleted from configuration would otherwise // linger in the aggregator as a permanently-offline tile (and live KPI diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs index 44e6a85f..a83a9a59 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcSiteTransport.cs @@ -245,18 +245,38 @@ public sealed class GrpcSiteTransport : ISiteCommandTransport var desired = cache.GrpcContacts; + // Each site is reconciled independently (review 2026-10-21 03/R-1): one bad row must not + // abort the rest of the tick, nor throw into CentralCommunicationActor's Receive (whose + // restart re-reads the Sites table immediately — a restart loop on a persistent bad row). // Drop sites that lost their gRPC endpoints (or were deleted). foreach (var removed in _knownSites.Where(s => !desired.ContainsKey(s)).ToList()) { - _channels.RemoveSite(removed); - _knownSites.Remove(removed); + try + { + _channels.RemoveSite(removed); + _knownSites.Remove(removed); + } + catch (Exception ex) when (ex is not ObjectDisposedException) + { + _logger.LogError(ex, "Failed to remove the gRPC channel pair for site {SiteId}", removed); + } } // Create/refresh the rest. foreach (var (siteId, endpoints) in desired) { - _channels.UpdateSite(siteId, endpoints.NodeA, endpoints.NodeB); - _knownSites.Add(siteId); + try + { + _channels.UpdateSite(siteId, endpoints.NodeA, endpoints.NodeB); + _knownSites.Add(siteId); + } + catch (Exception ex) when (ex is not ObjectDisposedException) + { + _logger.LogError(ex, + "Failed to reconcile the gRPC channel pair for site {SiteId} (nodeA={NodeA}, nodeB={NodeB}); " + + "its previous channels, if any, are kept", + siteId, endpoints.NodeA ?? "none", endpoints.NodeB ?? "none"); + } } } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs index 9939a673..3700f4d3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs @@ -3,6 +3,7 @@ using Grpc.Core; using Grpc.Net.Client; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Commons.Validation; namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; @@ -129,6 +130,22 @@ public sealed class SitePairChannelProvider : IDisposable { var changedA = !string.Equals(pair.NodeAEndpoint, nodeAEndpoint, StringComparison.Ordinal); var changedB = !string.Equals(pair.NodeBEndpoint, nodeBEndpoint, StringComparison.Ordinal); + + // Build every replacement BEFORE parking anything (review 2026-10-21 03/R-1): a + // malformed address used to throw out of BuildChannel after the old channel was + // already parked, leaving pair.ChannelA pointing at a channel the graveyard would + // dispose — every later call then failed with ObjectDisposedException (not + // Unavailable, so no failover) — and the throw escaped into the refresh actor. + // A side whose new endpoint cannot be built keeps its old channel AND old endpoint. + GrpcChannel? newA = null, newB = null; + if (changedA && !TryBuildReplacement(pair, siteId, nodeAEndpoint, isA: true, out newA)) + { + changedA = false; + } + if (changedB && !TryBuildReplacement(pair, siteId, nodeBEndpoint, isA: false, out newB)) + { + changedB = false; + } if (!changedA && !changedB) { return; @@ -137,13 +154,13 @@ public sealed class SitePairChannelProvider : IDisposable if (changedA) { Park(pair.ChannelA); - pair.ChannelA = string.IsNullOrWhiteSpace(nodeAEndpoint) ? null : BuildChannel(nodeAEndpoint, siteId); + pair.ChannelA = newA; pair.NodeAEndpoint = nodeAEndpoint; } if (changedB) { Park(pair.ChannelB); - pair.ChannelB = string.IsNullOrWhiteSpace(nodeBEndpoint) ? null : BuildChannel(nodeBEndpoint, siteId); + pair.ChannelB = newB; pair.NodeBEndpoint = nodeBEndpoint; } @@ -158,6 +175,62 @@ public sealed class SitePairChannelProvider : IDisposable } } + /// + /// Builds the replacement channel for one side of a pair whose endpoint changed. A null/blank + /// endpoint is a valid "unconfigured" (null channel). A malformed endpoint — validated with + /// , and any channel-construction throw besides — is rejected: + /// logged at Error naming the site and column (once per distinct rejected value, so the 60 s + /// refresh does not repeat it every tick) and the caller keeps the side's existing channel. + /// + /// The site pair (caller holds pair.Gate). + /// Site identifier, for the log. + /// The new endpoint for this side. + /// True for NodeA (GrpcNodeAAddress), false for NodeB. + /// The built channel, or null for an unconfigured side. + /// True when the replacement may be applied; false to keep the existing side. + private bool TryBuildReplacement(SitePair pair, string siteId, string? endpoint, bool isA, out GrpcChannel? channel) + { + channel = null; + if (string.IsNullOrWhiteSpace(endpoint)) + { + SetRejected(pair, isA, null); + return true; + } + + var field = isA ? "GrpcNodeAAddress" : "GrpcNodeBAddress"; + Exception? buildError = null; + if (GrpcEndpointRules.TryValidate(endpoint, allowEmpty: false, out var error)) + { + try + { + channel = BuildChannel(endpoint, siteId); + SetRejected(pair, isA, null); + return true; + } + catch (Exception ex) + { + buildError = ex; + error = ex.Message; + } + } + + if (!string.Equals(isA ? pair.RejectedAEndpoint : pair.RejectedBEndpoint, endpoint, StringComparison.Ordinal)) + { + _logger.LogError(buildError, + "Site {SiteId} {Field} '{Endpoint}' is not a usable gRPC endpoint ({Error}); keeping the " + + "previous channel ({Previous}) for that node until the address is corrected", + siteId, field, endpoint, error, (isA ? pair.NodeAEndpoint : pair.NodeBEndpoint) ?? "none"); + SetRejected(pair, isA, endpoint); + } + return false; + } + + private static void SetRejected(SitePair pair, bool isA, string? endpoint) + { + if (isA) pair.RejectedAEndpoint = endpoint; + else pair.RejectedBEndpoint = endpoint; + } + /// /// Parks a removed site's channels for deferred disposal (see ) /// and invalidates its cached preshared key. @@ -504,6 +577,10 @@ public sealed class SitePairChannelProvider : IDisposable public readonly object Gate = new(); public string? NodeAEndpoint; public string? NodeBEndpoint; + + /// Last malformed NodeA/NodeB endpoint rejected by UpdateSite (log de-duplication only). + public string? RejectedAEndpoint; + public string? RejectedBEndpoint; public GrpcChannel? ChannelA; public GrpcChannel? ChannelB; diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderBadAddressTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderBadAddressTests.cs new file mode 100644 index 00000000..04207de3 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderBadAddressTests.cs @@ -0,0 +1,144 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Grpc.Net.Client; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc; + +/// +/// A malformed GrpcNodeAAddress/GrpcNodeBAddress (review 2026-10-21 03/R-1) must not +/// park-then-dispose the site's still-referenced channel, must not stop the other sites in the same +/// refresh tick from being reconciled, and must not restart . +/// +public class SitePairChannelProviderBadAddressTests : TestKit +{ + /// Initializes the test kit with a quieter Akka log level. + public SitePairChannelProviderBadAddressTests() : base(@"akka.loglevel = WARNING") { } + + private static SitePairChannelProvider Build(Func? handlerFactory = null) => new( + new NoKeyProvider(), + Options.Create(new CommunicationOptions()), + NullLogger.Instance, + handlerFactory); + + private static Task CurrentChannel(SitePairChannelProvider provider, string siteId) => + provider.ExecuteAsync(siteId, (channel, _) => Task.FromResult(channel), CancellationToken.None); + + /// Verifies that an endpoint with no scheme keeps the existing channel and endpoint and parks nothing. + /// A malformed gRPC endpoint. + /// A task that represents the asynchronous operation. + [Theory] + [InlineData("scadabridge-site-a-node-a")] + [InlineData("site-a-node-a:8083")] + [InlineData("ftp://site-a-node-a:8083")] + public async Task UpdateSite_MalformedEndpoint_KeepsTheOldChannelAlive(string bad) + { + using var provider = Build(); + provider.UpdateSite("site-1", "http://localhost:15201", "http://localhost:15202"); + var before = await CurrentChannel(provider, "site-1"); + + provider.UpdateSite("site-1", bad, "http://localhost:15202"); + + Assert.Equal(0, provider.ParkedChannelCount); + provider.SweepGraveyard(force: true); + var after = await CurrentChannel(provider, "site-1"); + Assert.Same(before, after); + _ = after.CreateCallInvoker(); // not disposed + + // The next tick re-presents the same bad value: still nothing parked (no double-park). + provider.UpdateSite("site-1", bad, "http://localhost:15202"); + Assert.Equal(0, provider.ParkedChannelCount); + } + + /// Verifies that a channel build that throws keeps the old channel and still applies the other node's valid change. + /// A task that represents the asynchronous operation. + [Fact] + public async Task UpdateSite_BuildThrows_KeepsOldSideAndAppliesTheOtherSide() + { + using var provider = Build(endpoint => endpoint.Contains("15299", StringComparison.Ordinal) + ? throw new InvalidOperationException("No address resolver configured") + : new SocketsHttpHandler()); + provider.UpdateSite("site-1", "http://localhost:15211", "http://localhost:15212"); + var beforeA = await CurrentChannel(provider, "site-1"); + + provider.UpdateSite("site-1", "http://localhost:15299", "http://localhost:15213"); + + // Only the B side really changed, so only the old B channel is parked. + Assert.Equal(1, provider.ParkedChannelCount); + Assert.Same(beforeA, await CurrentChannel(provider, "site-1")); + } + + /// Verifies that a brand-new site whose only endpoint is malformed simply has no channel. + [Fact] + public async Task UpdateSite_NewSiteWithOnlyMalformedEndpoint_HasNoChannel() + { + using var provider = Build(); + provider.UpdateSite("site-1", "not a url", null); + await Assert.ThrowsAsync(() => CurrentChannel(provider, "site-1")); + } + + /// Verifies that one site that faults the reconcile does not stop later sites being reconciled. + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReconcileSites_OneFaultingSite_OtherSitesStillReconciled() + { + using var provider = Build(); + var transport = new GrpcSiteTransport( + provider, new CommunicationOptions(), NullLogger.Instance); + + // A whitespace site id makes UpdateSite throw; it is inserted FIRST so a foreach abort + // would skip the good site. + var contacts = new Dictionary + { + [" "] = new("http://localhost:15221", null), + ["site-good"] = new("http://localhost:15222", null), + }; + + transport.ReconcileSites(new SiteAddressCacheLoaded(new[] { " ", "site-good" }, contacts)); + + _ = await CurrentChannel(provider, "site-good"); + } + + /// Verifies that a transport reconcile fault does not restart the actor (which would re-read the Sites table in a loop). + [Fact] + public void TransportReconcileFault_DoesNotRestartTheActor() + { + var repo = Substitute.For(); + repo.GetAllSitesAsync(Arg.Any()).Returns(new List + { + new("Bad", "site-bad") { GrpcNodeAAddress = "scadabridge-site-a-node-a" } + }); + var services = new ServiceCollection(); + services.AddScoped(_ => repo); + var sp = services.BuildServiceProvider(); + + var transport = Substitute.For(); + transport.When(t => t.ReconcileSites(Arg.Any())) + .Do(_ => throw new UriFormatException("Invalid URI")); + + Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport))); + + AwaitAssert(() => transport.Received(1).ReconcileSites(Arg.Any()), + TimeSpan.FromSeconds(3)); + + // A restart re-runs PreStart, whose refresh fires at TimeSpan.Zero — so a restart loop would + // show up as repeated repository reads within a second. + Thread.Sleep(TimeSpan.FromSeconds(1)); + repo.Received(1).GetAllSitesAsync(Arg.Any()); + } + + private sealed class NoKeyProvider : ISitePskProvider + { + /// + public ValueTask GetAsync(string siteId, CancellationToken ct) => new("k"); + /// + public void Invalidate(string siteId) { } + } +}