fix(communication): command-plane failback probes readiness with a real RPC; backoff resets only after a real call succeeds (review 2026-10-21 03/P-2)

2026-10-22 06:26:13 -04:00
parent 017308a397
commit c5add11546
3 changed files with 255 additions and 15 deletions
@@ -362,7 +362,7 @@ Central discovers site addresses through the **configuration database**, not run
 - 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.
+- 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. While failed over, a background **readiness** probe returns calls to NodeA: it is a real, read-only `ExecuteLifecycle(DeploymentStateQuery)` for a sentinel instance (not a TCP/HTTP2 connect), so a node that is Up but still warming — which refuses every command with a not-dispatched `Unavailable` until `SetReady` — reads as not ready. The probe backoff (1 s → doubling → 60 s) is reset only after a **real call** on NodeA succeeds; if NodeA rejects the first calls after a probe-driven failback, the backoff keeps doubling instead of restarting at 1 s.
 - **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.

@@ -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.Messages.Deployment;
 using ZB.MOM.WW.ScadaBridge.Commons.Validation;

 namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc;
@@ -27,8 +28,9 @@ public sealed class SiteChannelUnavailableException(string siteId)
 /// Central holds one <see cref="GrpcChannel"/> per site node (NodeA/NodeB) rather than one channel
 /// with re-dial logic; NodeA is the preferred node (config order). Every call goes to the current
 /// sticky channel; on connect failure / <see cref="StatusCode.Unavailable"/> / readiness rejection
-/// it flips to the other node and stays there. A background probe returns to the preferred node
-/// once it answers again. Reconnect probing backs off 1 s → doubling → 60 s cap.
+/// it flips to the other node and stays there. A background readiness probe (a real read-only RPC)
+/// returns to the preferred node once it answers again; the backoff (1 s → doubling → 60 s cap) is
+/// reset only after a real call on the preferred node succeeds.
 /// </summary>
 /// <remarks>
 /// <para>
@@ -87,8 +89,9 @@ public sealed class SitePairChannelProvider : IDisposable
    /// a keepalive-configured <see cref="SocketsHttpHandler"/>.
    /// </param>
    /// <param name="reachabilityProbe">
-    /// Test seam deciding whether a preferred node is reachable during a failback sweep. Null in
-    /// production, where <see cref="GrpcChannel.ConnectAsync"/> is used.
+    /// Test seam deciding whether a preferred node is READY during a failback sweep. Null in
+    /// production, where <see cref="ReadinessProbeAsync"/> issues a real, read-only
+    /// <c>ExecuteLifecycle</c> call (review 2026-10-21 03/P-2).
    /// </param>
    public SitePairChannelProvider(
        ISitePskProvider pskProvider,
@@ -105,7 +108,7 @@ public sealed class SitePairChannelProvider : IDisposable
        _options = options.Value;
        _logger = logger;
        _handlerFactory = handlerFactory;
-        _reachabilityProbe = reachabilityProbe ?? DefaultReachabilityProbe;
+        _reachabilityProbe = reachabilityProbe ?? ReadinessProbeAsync;
    }

    /// <summary>
@@ -306,7 +309,9 @@ public sealed class SitePairChannelProvider : IDisposable

        try
        {
-            return await call(primary, ct).ConfigureAwait(false);
+            var result = await call(primary, ct).ConfigureAwait(false);
+            if (primaryIsA) ConfirmFailback(pair);
+            return result;
        }
        catch (RpcException ex) when (ex.StatusCode == StatusCode.Unavailable)
        {
@@ -362,9 +367,14 @@ public sealed class SitePairChannelProvider : IDisposable
        {
            if (reachable)
            {
+                // Return to A on PROBATION (review 2026-10-21 03/P-2): the backoff is kept, not
+                // reset, until a real call on A succeeds (ConfirmFailback). If A rejects the next
+                // call anyway, FlipTo doubles the backoff instead of restarting it at 1 s, so a
+                // node that answers the probe but not real traffic is retried ever less often.
                pair.CurrentIsA = true;
-                pair.ResetFailback();
-                _logger.LogInformation("Site {SiteId} preferred node A reachable again; failing back", siteId);
+                pair.FailbackOnProbation = true;
+                pair.NextProbeUtc = DateTime.MaxValue;
+                _logger.LogInformation("Site {SiteId} preferred node A ready again; failing back", siteId);
                return true;
            }

@@ -494,6 +504,30 @@ public sealed class SitePairChannelProvider : IDisposable
    /// <summary>Test/diagnostic accessor: how many replaced channels are awaiting deferred disposal.</summary>
    internal int ParkedChannelCount => _graveyard.Count;

+    /// <summary>A real call on the preferred node succeeded: a probationary failback is confirmed and the backoff reset.</summary>
+    /// <param name="pair">The site pair.</param>
+    private static void ConfirmFailback(SitePair pair)
+    {
+        if (!pair.FailbackOnProbation) return;
+        lock (pair.Gate)
+        {
+            if (pair.FailbackOnProbation && pair.CurrentIsA)
+            {
+                pair.FailbackOnProbation = false;
+                pair.ResetFailback();
+            }
+        }
+    }
+
+    /// <summary>Test/diagnostic accessor: the site's current failback backoff (null when the site is unknown).</summary>
+    /// <param name="siteId">Site identifier.</param>
+    /// <returns>The backoff the next failback probe is (or would be) scheduled with.</returns>
+    internal TimeSpan? GetFailbackBackoff(string siteId)
+    {
+        if (!_sites.TryGetValue(siteId, out var pair)) return null;
+        lock (pair.Gate) return pair.Backoff;
+    }
+
    private void FlipTo(SitePair pair, bool toIsA)
    {
        lock (pair.Gate)
@@ -501,11 +535,22 @@ public sealed class SitePairChannelProvider : IDisposable
            pair.CurrentIsA = toIsA;
            if (!toIsA)
            {
-                // Failed over off the preferred node → schedule the first failback probe.
-                pair.ScheduleFailback(InitialBackoff);
+                if (pair.FailbackOnProbation)
+                {
+                    // A failed us again right after a probe-driven failback: it is not really
+                    // ready yet — grow the backoff rather than restarting it.
+                    pair.FailbackOnProbation = false;
+                    pair.BumpFailbackBackoff();
+                }
+                else
+                {
+                    // Failed over off the preferred node → schedule the first failback probe.
+                    pair.ScheduleFailback(InitialBackoff);
+                }
            }
            else
            {
+                pair.FailbackOnProbation = false;
                pair.ResetFailback();
            }
        }
@@ -529,15 +574,37 @@ public sealed class SitePairChannelProvider : IDisposable
        return GrpcChannel.ForAddress(endpoint, channelOptions);
    }

-    private static async Task<bool> DefaultReachabilityProbe(GrpcChannel channel, CancellationToken ct)
+    /// <summary>Sentinel instance name the readiness probe queries; never a real instance.</summary>
+    internal const string ReadinessProbeInstance = "__scadabridge-failback-probe__";
+
+    /// <summary>
+    /// Production failback probe (review 2026-10-21 03/P-2): a real, read-only
+    /// <c>ExecuteLifecycle(DeploymentStateQuery)</c> for a sentinel instance, NOT a transport
+    /// connect. <c>SiteCommandGrpcService</c> refuses every RPC with a not-dispatched
+    /// <see cref="StatusCode.Unavailable"/> until the site has finished bring-up, so a node that is
+    /// Up but still warming (minutes on a large site) now reads as NOT ready — a TCP/HTTP2 connect
+    /// succeeded there and failed every command back to it each sweep. The query reads one
+    /// (absent) row from the site's deployed-config store; it is idempotent and side-effect free.
+    /// Any reply is "ready"; any failure, including the readiness rejection, is "not ready".
+    /// </summary>
+    /// <param name="channel">The preferred node's channel.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>True when the node answered a real command.</returns>
+    internal static async Task<bool> ReadinessProbeAsync(GrpcChannel channel, CancellationToken ct)
    {
        try
        {
-            using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
-            timeout.CancelAfter(TimeSpan.FromSeconds(5));
-            await channel.ConnectAsync(timeout.Token).ConfigureAwait(false);
+            var request = SiteCommandDtoMapper.ToLifecycleRequest(new DeploymentStateQueryRequest(
+                "failback-probe-" + Guid.NewGuid().ToString("N"), ReadinessProbeInstance, DateTimeOffset.UtcNow));
+            var client = new SiteCommandService.SiteCommandServiceClient(channel);
+            await client.ExecuteLifecycleAsync(
+                request, deadline: DateTime.UtcNow.AddSeconds(5), cancellationToken: ct).ResponseAsync.ConfigureAwait(false);
            return true;
        }
+        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        {
+            throw;
+        }
        catch
        {
            return false;
@@ -587,6 +654,15 @@ public sealed class SitePairChannelProvider : IDisposable
        /// <summary>Sticky current node; NodeA is preferred. True = pointing at NodeA.</summary>
        public bool CurrentIsA = true;

+        /// <summary>
+        /// True between a probe-driven failback and the first successful real call on NodeA; while
+        /// set, the backoff is kept so a renewed failure doubles it.
+        /// </summary>
+        public bool FailbackOnProbation;
+
+        /// <summary>The current failback backoff.</summary>
+        public TimeSpan Backoff => _backoff;
+
        /// <summary>Next time a failback probe is due (while failed over). MaxValue = not scheduled.</summary>
        public DateTime NextProbeUtc = DateTime.MaxValue;

@@ -0,0 +1,164 @@
+using System.Net;
+using System.Net.Http.Headers;
+using Google.Protobuf;
+using Grpc.Core;
+using Grpc.Net.Client;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
+
+namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc;
+
+/// <summary>
+/// Command-plane failback (review 2026-10-21 03/P-2): the probe must test READINESS with a real
+/// RPC (a node that is Up but still warming refuses commands with a not-dispatched Unavailable),
+/// and the backoff must only reset after a real call on the preferred node succeeds — not after
+/// the probe — so a node that keeps failing real calls is retried ever less often.
+/// </summary>
+public class SitePairChannelProviderFailbackTests
+{
+    private const string Site = "site-1";
+
+    private static SitePairChannelProvider Build(Func<GrpcChannel, CancellationToken, Task<bool>> probe) => new(
+        new NoKeyProvider(),
+        Options.Create(new CommunicationOptions()),
+        NullLogger<SitePairChannelProvider>.Instance,
+        handlerFactory: null,
+        reachabilityProbe: probe);
+
+    /// <summary>Runs one call that node A refuses as not-ready (then succeeds on B), as a warming node does.</summary>
+    private static async Task CallRejectedByA(SitePairChannelProvider provider)
+    {
+        GrpcChannel? first = null;
+        await provider.ExecuteAsync(Site, (channel, _) =>
+        {
+            if (first is null)
+            {
+                first = channel;
+                throw GrpcRetrySafety.NotDispatched("Site command plane not ready.");
+            }
+            return Task.FromResult(1);
+        }, CancellationToken.None);
+    }
+
+    private static Task CallSucceeds(SitePairChannelProvider provider) =>
+        provider.ExecuteAsync(Site, (_, _) => Task.FromResult(1), CancellationToken.None);
+
+    /// <summary>Verifies that a probe success alone does not reset the backoff; repeated rejections after failback grow it.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ProbeSuccess_ThenRealCallRejected_GrowsTheBackoffInsteadOfResettingIt()
+    {
+        using var provider = Build((_, _) => Task.FromResult(true));
+        provider.UpdateSite(Site, "http://localhost:15301", "http://localhost:15302");
+
+        await CallRejectedByA(provider);
+        Assert.False(provider.IsOnPreferredNode(Site));
+        Assert.Equal(TimeSpan.FromSeconds(1), provider.GetFailbackBackoff(Site));
+
+        Assert.True(await provider.TryFailbackAsync(Site, CancellationToken.None));
+        await CallRejectedByA(provider);
+        Assert.Equal(TimeSpan.FromSeconds(2), provider.GetFailbackBackoff(Site));
+
+        Assert.True(await provider.TryFailbackAsync(Site, CancellationToken.None));
+        await CallRejectedByA(provider);
+        Assert.Equal(TimeSpan.FromSeconds(4), provider.GetFailbackBackoff(Site));
+    }
+
+    /// <summary>Verifies that a successful real call on the preferred node after failback resets the backoff.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RealCallSucceedsAfterFailback_ResetsTheBackoff()
+    {
+        using var provider = Build((_, _) => Task.FromResult(true));
+        provider.UpdateSite(Site, "http://localhost:15311", "http://localhost:15312");
+
+        await CallRejectedByA(provider);
+        Assert.True(await provider.TryFailbackAsync(Site, CancellationToken.None));
+        await CallRejectedByA(provider);
+        Assert.Equal(TimeSpan.FromSeconds(2), provider.GetFailbackBackoff(Site));
+
+        Assert.True(await provider.TryFailbackAsync(Site, CancellationToken.None));
+        await CallSucceeds(provider); // A really is ready now
+
+        // A later, unrelated failure starts again from the initial backoff.
+        await CallRejectedByA(provider);
+        Assert.Equal(TimeSpan.FromSeconds(1), provider.GetFailbackBackoff(Site));
+    }
+
+    /// <summary>Verifies that the production probe treats a not-ready (not-dispatched Unavailable) site as not ready.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ReadinessProbe_SiteRefusingAsNotReady_IsNotReady()
+    {
+        var handler = new GrpcStubHandler(ready: false);
+        using var channel = GrpcChannel.ForAddress("http://node-a", new GrpcChannelOptions { HttpHandler = handler });
+
+        Assert.False(await SitePairChannelProvider.ReadinessProbeAsync(channel, CancellationToken.None));
+        Assert.EndsWith(".SiteCommandService/ExecuteLifecycle", handler.LastPath);
+    }
+
+    /// <summary>Verifies that the production probe treats a site answering a real command as ready.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ReadinessProbe_SiteAnsweringARealCommand_IsReady()
+    {
+        var handler = new GrpcStubHandler(ready: true);
+        using var channel = GrpcChannel.ForAddress("http://node-a", new GrpcChannelOptions { HttpHandler = handler });
+
+        Assert.True(await SitePairChannelProvider.ReadinessProbeAsync(channel, CancellationToken.None));
+        Assert.Contains(SitePairChannelProvider.ReadinessProbeInstance, handler.LastRequestBody);
+    }
+
+    /// <summary>
+    /// Minimal in-memory gRPC server stand-in: a trailers-only Unavailable (+ the not-dispatched
+    /// trailer) while not ready, or a framed <see cref="LifecycleReply"/> with OK status when ready.
+    /// </summary>
+    private sealed class GrpcStubHandler(bool ready) : HttpMessageHandler
+    {
+        public string LastPath { get; private set; } = "";
+        public string LastRequestBody { get; private set; } = "";
+
+        protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken ct)
+        {
+            LastPath = request.RequestUri!.AbsolutePath;
+            var body = await request.Content!.ReadAsByteArrayAsync(ct);
+            var parsed = LifecycleRequest.Parser.ParseFrom(body.AsSpan(5).ToArray());
+            LastRequestBody = parsed.ToString();
+
+            var response = new HttpResponseMessage(HttpStatusCode.OK) { Version = HttpVersion.Version20 };
+            if (!ready)
+            {
+                var notDispatched = GrpcRetrySafety.NotDispatched("Site command plane not ready.");
+                response.Content = new ByteArrayContent(Array.Empty<byte>());
+                response.Content.Headers.ContentType = new MediaTypeHeaderValue("application/grpc");
+                response.Headers.Add("grpc-status", ((int)StatusCode.Unavailable).ToString());
+                foreach (var entry in notDispatched.Trailers)
+                {
+                    response.Headers.Add(entry.Key, entry.Value);
+                }
+                return response;
+            }
+
+            var reply = new LifecycleReply { DeploymentStateQuery = new DeploymentStateQueryResponseDto() }.ToByteArray();
+            var framed = new byte[5 + reply.Length];
+            framed[1] = (byte)(reply.Length >> 24);
+            framed[2] = (byte)(reply.Length >> 16);
+            framed[3] = (byte)(reply.Length >> 8);
+            framed[4] = (byte)reply.Length;
+            reply.CopyTo(framed, 5);
+            response.Content = new ByteArrayContent(framed);
+            response.Content.Headers.ContentType = new MediaTypeHeaderValue("application/grpc");
+            response.TrailingHeaders.Add("grpc-status", "0");
+            return response;
+        }
+    }
+
+    private sealed class NoKeyProvider : ISitePskProvider
+    {
+        /// <inheritdoc />
+        public ValueTask<string> GetAsync(string siteId, CancellationToken ct) => new("k");
+        /// <inheritdoc />
+        public void Invalidate(string siteId) { }
+    }
+}