fix(communication): malformed site gRPC address keeps the old channel, never aborts the refresh tick or restarts CentralCommunicationActor (review 2026-10-21 03/R-1)

2026-10-22 06:14:34 -04:00
parent 0fd9bcbce6
commit b0a97a4b67
5 changed files with 261 additions and 8 deletions
@@ -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.
@@ -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
@@ -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");
+            }
        }
    }
 }
@@ -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
        }
    }

+    /// <summary>
+    /// 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
+    /// <see cref="GrpcEndpointRules"/>, 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.
+    /// </summary>
+    /// <param name="pair">The site pair (caller holds <c>pair.Gate</c>).</param>
+    /// <param name="siteId">Site identifier, for the log.</param>
+    /// <param name="endpoint">The new endpoint for this side.</param>
+    /// <param name="isA">True for NodeA (<c>GrpcNodeAAddress</c>), false for NodeB.</param>
+    /// <param name="channel">The built channel, or null for an unconfigured side.</param>
+    /// <returns>True when the replacement may be applied; false to keep the existing side.</returns>
+    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;
+    }
+
    /// <summary>
    /// Parks a removed site's channels for deferred disposal (see <see cref="ChannelDisposalGrace"/>)
    /// 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;
+
+        /// <summary>Last malformed NodeA/NodeB endpoint rejected by <c>UpdateSite</c> (log de-duplication only).</summary>
+        public string? RejectedAEndpoint;
+        public string? RejectedBEndpoint;
        public GrpcChannel? ChannelA;
        public GrpcChannel? ChannelB;

@@ -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;
+
+/// <summary>
+/// A malformed <c>GrpcNodeAAddress</c>/<c>GrpcNodeBAddress</c> (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 <see cref="CentralCommunicationActor"/>.
+/// </summary>
+public class SitePairChannelProviderBadAddressTests : TestKit
+{
+    /// <summary>Initializes the test kit with a quieter Akka log level.</summary>
+    public SitePairChannelProviderBadAddressTests() : base(@"akka.loglevel = WARNING") { }
+
+    private static SitePairChannelProvider Build(Func<string, HttpMessageHandler>? handlerFactory = null) => new(
+        new NoKeyProvider(),
+        Options.Create(new CommunicationOptions()),
+        NullLogger<SitePairChannelProvider>.Instance,
+        handlerFactory);
+
+    private static Task<GrpcChannel> CurrentChannel(SitePairChannelProvider provider, string siteId) =>
+        provider.ExecuteAsync(siteId, (channel, _) => Task.FromResult(channel), CancellationToken.None);
+
+    /// <summary>Verifies that an endpoint with no scheme keeps the existing channel and endpoint and parks nothing.</summary>
+    /// <param name="bad">A malformed gRPC endpoint.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
+    /// <summary>Verifies that a channel build that throws keeps the old channel and still applies the other node's valid change.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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"));
+    }
+
+    /// <summary>Verifies that a brand-new site whose only endpoint is malformed simply has no channel.</summary>
+    [Fact]
+    public async Task UpdateSite_NewSiteWithOnlyMalformedEndpoint_HasNoChannel()
+    {
+        using var provider = Build();
+        provider.UpdateSite("site-1", "not a url", null);
+        await Assert.ThrowsAsync<SiteChannelUnavailableException>(() => CurrentChannel(provider, "site-1"));
+    }
+
+    /// <summary>Verifies that one site that faults the reconcile does not stop later sites being reconciled.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ReconcileSites_OneFaultingSite_OtherSitesStillReconciled()
+    {
+        using var provider = Build();
+        var transport = new GrpcSiteTransport(
+            provider, new CommunicationOptions(), NullLogger<GrpcSiteTransport>.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<string, SiteGrpcEndpoints>
+        {
+            [" "] = 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");
+    }
+
+    /// <summary>Verifies that a transport reconcile fault does not restart the actor (which would re-read the Sites table in a loop).</summary>
+    [Fact]
+    public void TransportReconcileFault_DoesNotRestartTheActor()
+    {
+        var repo = Substitute.For<ISiteRepository>();
+        repo.GetAllSitesAsync(Arg.Any<CancellationToken>()).Returns(new List<Site>
+        {
+            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<ISiteCommandTransport>();
+        transport.When(t => t.ReconcileSites(Arg.Any<SiteAddressCacheLoaded>()))
+            .Do(_ => throw new UriFormatException("Invalid URI"));
+
+        Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport)));
+
+        AwaitAssert(() => transport.Received(1).ReconcileSites(Arg.Any<SiteAddressCacheLoaded>()),
+            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<CancellationToken>());
+    }
+
+    private sealed class NoKeyProvider : ISitePskProvider
+    {
+        /// <inheritdoc />
+        public ValueTask<string> GetAsync(string siteId, CancellationToken ct) => new("k");
+        /// <inheritdoc />
+        public void Invalidate(string siteId) { }
+    }
+}