diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index f22bd0e0..5f0b2c53 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -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. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs index 3700f4d3..7d5a7b72 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.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 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 / / 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. /// /// /// @@ -87,8 +89,9 @@ public sealed class SitePairChannelProvider : IDisposable /// a keepalive-configured . /// /// - /// Test seam deciding whether a preferred node is reachable during a failback sweep. Null in - /// production, where is used. + /// Test seam deciding whether a preferred node is READY during a failback sweep. Null in + /// production, where issues a real, read-only + /// ExecuteLifecycle call (review 2026-10-21 03/P-2). /// 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; } /// @@ -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 /// Test/diagnostic accessor: how many replaced channels are awaiting deferred disposal. internal int ParkedChannelCount => _graveyard.Count; + /// A real call on the preferred node succeeded: a probationary failback is confirmed and the backoff reset. + /// The site pair. + private static void ConfirmFailback(SitePair pair) + { + if (!pair.FailbackOnProbation) return; + lock (pair.Gate) + { + if (pair.FailbackOnProbation && pair.CurrentIsA) + { + pair.FailbackOnProbation = false; + pair.ResetFailback(); + } + } + } + + /// Test/diagnostic accessor: the site's current failback backoff (null when the site is unknown). + /// Site identifier. + /// The backoff the next failback probe is (or would be) scheduled with. + 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 DefaultReachabilityProbe(GrpcChannel channel, CancellationToken ct) + /// Sentinel instance name the readiness probe queries; never a real instance. + internal const string ReadinessProbeInstance = "__scadabridge-failback-probe__"; + + /// + /// Production failback probe (review 2026-10-21 03/P-2): a real, read-only + /// ExecuteLifecycle(DeploymentStateQuery) for a sentinel instance, NOT a transport + /// connect. SiteCommandGrpcService refuses every RPC with a not-dispatched + /// 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". + /// + /// The preferred node's channel. + /// Cancellation token. + /// True when the node answered a real command. + internal static async Task 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 /// Sticky current node; NodeA is preferred. True = pointing at NodeA. public bool CurrentIsA = true; + /// + /// 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. + /// + public bool FailbackOnProbation; + + /// The current failback backoff. + public TimeSpan Backoff => _backoff; + /// Next time a failback probe is due (while failed over). MaxValue = not scheduled. public DateTime NextProbeUtc = DateTime.MaxValue; diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderFailbackTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderFailbackTests.cs new file mode 100644 index 00000000..942d583a --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderFailbackTests.cs @@ -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; + +/// +/// 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. +/// +public class SitePairChannelProviderFailbackTests +{ + private const string Site = "site-1"; + + private static SitePairChannelProvider Build(Func> probe) => new( + new NoKeyProvider(), + Options.Create(new CommunicationOptions()), + NullLogger.Instance, + handlerFactory: null, + reachabilityProbe: probe); + + /// Runs one call that node A refuses as not-ready (then succeeds on B), as a warming node does. + 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); + + /// Verifies that a probe success alone does not reset the backoff; repeated rejections after failback grow it. + /// A task that represents the asynchronous operation. + [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)); + } + + /// Verifies that a successful real call on the preferred node after failback resets the backoff. + /// A task that represents the asynchronous operation. + [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)); + } + + /// Verifies that the production probe treats a not-ready (not-dispatched Unavailable) site as not ready. + /// A task that represents the asynchronous operation. + [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); + } + + /// Verifies that the production probe treats a site answering a real command as ready. + /// A task that represents the asynchronous operation. + [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); + } + + /// + /// Minimal in-memory gRPC server stand-in: a trailers-only Unavailable (+ the not-dispatched + /// trailer) while not ready, or a framed with OK status when ready. + /// + private sealed class GrpcStubHandler(bool ready) : HttpMessageHandler + { + public string LastPath { get; private set; } = ""; + public string LastRequestBody { get; private set; } = ""; + + protected override async Task 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()); + 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 + { + /// + public ValueTask GetAsync(string siteId, CancellationToken ct) => new("k"); + /// + public void Invalidate(string siteId) { } + } +}