diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index 5f0b2c53..75663484 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. 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. +- 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. That probation is time-boxed (5 min) and cleared by an address change: a quiet site whose NodeA fails long after the failback starts again 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 7d5a7b72..b03dfd81 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs @@ -58,6 +58,13 @@ public sealed class SitePairChannelProvider : IDisposable /// internal static readonly TimeSpan ChannelDisposalGrace = TimeSpan.FromMinutes(5); + /// + /// How long a probe-driven failback stays on probation without a confirming real call; a + /// NodeA failure after this window is treated as a fresh failover (backoff restarts at the + /// initial value). Settable per instance for tests. + /// + internal TimeSpan FailbackProbationWindow { get; set; } = TimeSpan.FromMinutes(5); + private readonly ISitePskProvider _pskProvider; private readonly CommunicationOptions _options; private readonly ILogger _logger; @@ -169,6 +176,7 @@ public sealed class SitePairChannelProvider : IDisposable // Addresses changed — return to the preferred node and reset failback backoff. pair.CurrentIsA = pair.ChannelA is not null; + pair.FailbackOnProbation = false; pair.ResetFailback(); _logger.LogInformation( "gRPC channel pair for site {SiteId} refreshed (nodeA={HasA}, nodeB={HasB})", @@ -373,6 +381,7 @@ public sealed class SitePairChannelProvider : IDisposable // node that answers the probe but not real traffic is retried ever less often. pair.CurrentIsA = true; pair.FailbackOnProbation = true; + pair.ProbationSinceUtc = DateTime.UtcNow; pair.NextProbeUtc = DateTime.MaxValue; _logger.LogInformation("Site {SiteId} preferred node A ready again; failing back", siteId); return true; @@ -535,7 +544,12 @@ public sealed class SitePairChannelProvider : IDisposable pair.CurrentIsA = toIsA; if (!toIsA) { - if (pair.FailbackOnProbation) + // Probation is time-boxed: a site quiet for longer than FailbackProbationWindow + // after a probe-driven failback (no real call to confirm it) is treated as + // confirmed, so a fresh NodeA failure much later starts at InitialBackoff again + // instead of inheriting a stale, grown backoff. + if (pair.FailbackOnProbation + && DateTime.UtcNow - pair.ProbationSinceUtc < FailbackProbationWindow) { // A failed us again right after a probe-driven failback: it is not really // ready yet — grow the backoff rather than restarting it. @@ -545,6 +559,7 @@ public sealed class SitePairChannelProvider : IDisposable else { // Failed over off the preferred node → schedule the first failback probe. + pair.FailbackOnProbation = false; pair.ScheduleFailback(InitialBackoff); } } @@ -660,6 +675,9 @@ public sealed class SitePairChannelProvider : IDisposable /// public bool FailbackOnProbation; + /// When the current probation began (meaningful only while ). + public DateTime ProbationSinceUtc; + /// The current failback backoff. public TimeSpan Backoff => _backoff; diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderFailbackTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderFailbackTests.cs index 942d583a..824c0504 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderFailbackTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderFailbackTests.cs @@ -86,6 +86,46 @@ public class SitePairChannelProviderFailbackTests Assert.Equal(TimeSpan.FromSeconds(1), provider.GetFailbackBackoff(Site)); } + /// Verifies that probation expires: a quiet site whose NodeA dies long after the failback starts again at the initial backoff. + /// A task that represents the asynchronous operation. + [Fact] + public async Task ProbationOlderThanTheWindow_IsTreatedAsConfirmed_FirstProbeAtInitialBackoff() + { + using var provider = Build((_, _) => Task.FromResult(true)); + provider.FailbackProbationWindow = TimeSpan.FromMilliseconds(100); + provider.UpdateSite(Site, "http://localhost:15321", "http://localhost:15322"); + + await CallRejectedByA(provider); + Assert.True(await provider.TryFailbackAsync(Site, CancellationToken.None)); + await CallRejectedByA(provider); + Assert.Equal(TimeSpan.FromSeconds(2), provider.GetFailbackBackoff(Site)); + + // Fail back on probation, then the site goes quiet (no real call) past the window. + Assert.True(await provider.TryFailbackAsync(Site, CancellationToken.None)); + await Task.Delay(300); + + await CallRejectedByA(provider); + Assert.Equal(TimeSpan.FromSeconds(1), provider.GetFailbackBackoff(Site)); + } + + /// Verifies that an address change clears probation. + /// A task that represents the asynchronous operation. + [Fact] + public async Task AddressChange_ClearsProbation() + { + using var provider = Build((_, _) => Task.FromResult(true)); + provider.UpdateSite(Site, "http://localhost:15331", "http://localhost:15332"); + + await CallRejectedByA(provider); + Assert.True(await provider.TryFailbackAsync(Site, CancellationToken.None)); + await CallRejectedByA(provider); + Assert.True(await provider.TryFailbackAsync(Site, CancellationToken.None)); + + provider.UpdateSite(Site, "http://localhost:15333", "http://localhost:15332"); + 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]