diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index 75663484..1cb58a6c 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -369,7 +369,7 @@ Central discovers site addresses through the **configuration database**, not run ### Site → Central Communication - Site nodes configure a list of **CentralGrpcEndpoints** — the gRPC h2c URLs of the central nodes (e.g., `http://scadabridge-central-a:8083`, on the central's `CentralGrpcPort`, default 8083) — replacing the former `CentralContactPoints` list of Akka addresses. A site node must list at least one; central nodes leave it empty (they host `CentralControlService`, they do not dial it). Note these are the central nodes' **direct** gRPC endpoints, not Traefik (which is HTTP/1 only). -- `GrpcCentralTransport` dials `CentralControlService` over these endpoints as a **sticky central-a→central-b failover channel pair**, and the site sends heartbeats, health reports, notification submit/status, audit/telemetry ingest, and reconcile over it. +- `GrpcCentralTransport` dials `CentralControlService` over these endpoints as a **sticky central-a→central-b failover channel pair**, and the site sends heartbeats, health reports, notification submit/status, audit/telemetry ingest, and reconcile over it. `CentralControlService.ReconcileSite` hands `CentralCommunicationActor` the gRPC call's cancellation token alongside the request (an in-process `ReconcileSiteCall` envelope), linked into `ReconcileService.ReconcileAsync`, so central stops its DB reads and pending-row staging as soon as the site's deadline expires or it disconnects. - If the active central node goes down, the site's transport flips to the standby central endpoint transparently. ## Message Timeouts diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs index ee0136b1..0cbac147 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs @@ -177,7 +177,11 @@ public class CentralCommunicationActor : ReceiveActor // startup. Resolve the scoped ReconcileService, diff the // inventory against central's expected set, and pipe the ReconcileSiteResponse // (gap fetch tokens + orphans) straight back to the site node. - Receive(HandleReconcileSiteRequest); + Receive(msg => HandleReconcileSiteRequest(msg, CancellationToken.None)); + + // The same request as received by CentralControlGrpcService.ReconcileSite, carrying the + // gRPC call's cancellation token so central stops staging work for a site that gave up. + Receive(call => HandleReconcileSiteRequest(call.Request, call.CallCancelled)); } private void HandleNotificationSubmit(NotificationSubmit msg) @@ -230,7 +234,13 @@ public class CentralCommunicationActor : ReceiveActor /// Ask faults and it simply retries reconcile on the next startup — reconcile is /// best-effort, so the fault is allowed to propagate rather than being swallowed. /// - private void HandleReconcileSiteRequest(ReconcileSiteRequest msg) + /// The site node's reconcile request. + /// + /// The originating gRPC call's token (review 2026-10-21 03/P-1b): when the site's deadline + /// expires or it disconnects, the reconcile stops instead of staging pending rows for a caller + /// that is gone. for an in-process caller. + /// + private void HandleReconcileSiteRequest(ReconcileSiteRequest msg, CancellationToken callCancelled) { // Capture Sender before the async/PipeTo — Akka resets Sender between dispatches. var replyTo = Sender; @@ -253,9 +263,10 @@ public class CentralCommunicationActor : ReceiveActor Task.Run(async () => { + using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, callCancelled); using var scope = _serviceProvider.CreateScope(); var service = scope.ServiceProvider.GetRequiredService(); - return await service.ReconcileAsync(msg, ct).ConfigureAwait(false); + return await service.ReconcileAsync(msg, linked.Token).ConfigureAwait(false); }).PipeTo(replyTo); } @@ -609,6 +620,15 @@ public sealed record SiteAddressCacheLoaded( IReadOnlyDictionary GrpcContacts, long Sequence); +/// +/// In-process envelope (never on the wire) pairing a site's +/// with the cancellation token of the gRPC call that carried it, so the central reconcile stops +/// when the site abandons the call (review 2026-10-21 03/P-1b). +/// +/// The reconcile request. +/// The gRPC call's ServerCallContext.CancellationToken. +internal sealed record ReconcileSiteCall(ReconcileSiteRequest Request, CancellationToken CallCancelled); + /// A failed (or deadline-cancelled) site-address DB load, piped back to release the overlap guard. /// The failed load's sequence stamp. /// The failure. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs index 2ab14894..75bd4e96 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs @@ -7,6 +7,7 @@ using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; using GrpcStatus = Grpc.Core.Status; namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; @@ -254,9 +255,11 @@ public sealed class CentralControlGrpcService : CentralControlService.CentralCon ReconcileSiteRequestDto request, ServerCallContext context) { var central = RequireReady(context); + // Carry the call's token into the actor so central stops reconcile work (DB reads, + // pending-row staging) when the site's deadline expires or it disconnects. var response = await AskAsync( central, - CentralControlDtoMapper.FromDto(request), + new ReconcileSiteCall(CentralControlDtoMapper.FromDto(request), context.CancellationToken), _options.QueryTimeout, context).ConfigureAwait(false); diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs index b9be76fe..5d82688b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs @@ -81,4 +81,47 @@ public class CentralCommunicationActorReconcileTests : TestKit Assert.Equal("dep-B", gap.DeploymentId); Assert.False(string.IsNullOrWhiteSpace(gap.FetchToken)); } + + /// Verifies that cancelling the originating gRPC call stops central reconcile work (review 2026-10-21 03/P-1b). + [Fact] + public void CancelledCall_StopsCentralReconcileWork() + { + var deploymentRepo = Substitute.For(); + var siteRepo = Substitute.For(); + siteRepo.GetAllSitesAsync(Arg.Any()).Returns(new List()); + + var lookupStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + CancellationToken observed = default; + siteRepo.GetSiteByIdentifierAsync("site1", Arg.Any()).Returns(async ci => + { + observed = ci.Arg(); + lookupStarted.TrySetResult(); + await Task.Delay(Timeout.Infinite, observed); // a slow central DB + return (Site?)new Site("Site One", "site1") { Id = 7 }; + }); + + var services = new ServiceCollection(); + services.AddScoped(_ => deploymentRepo); + services.AddScoped(_ => siteRepo); + services.AddSingleton(Options.Create(new CommunicationOptions())); + services.AddSingleton>( + NullLogger.Instance); + services.AddScoped(); + var sp = services.BuildServiceProvider(); + + var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor( + sp, Substitute.For()))); + + using var call = new CancellationTokenSource(); + actor.Tell(new ReconcileSiteCall( + new ReconcileSiteRequest("site1", "node-b", new Dictionary()), call.Token)); + AwaitCondition(() => lookupStarted.Task.IsCompleted, TimeSpan.FromSeconds(3)); + + // The site gives up (deadline / disconnect). + call.Cancel(); + + AwaitCondition(() => observed.IsCancellationRequested, TimeSpan.FromSeconds(3)); + ExpectMsg(TimeSpan.FromSeconds(3)); + deploymentRepo.DidNotReceiveWithAnyArgs().GetExpectedDeploymentsForSiteAsync(default, default); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlGrpcServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlGrpcServiceTests.cs index 6bcc2957..75300dd5 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlGrpcServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlGrpcServiceTests.cs @@ -139,6 +139,29 @@ public class CentralControlGrpcServiceTests : TestKit Assert.Empty(ack.AcceptedEventIds); } + /// Verifies that ReconcileSite hands the actor the gRPC call's cancellation token (review 2026-10-21 03/P-1b). + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReconcileSite_CarriesTheCallCancellationToken_ToTheActor() + { + var probe = CreateTestProbe(); + var service = CreateService(); + service.SetReady(probe.Ref); + using var callCts = new CancellationTokenSource(); + + var reply = service.ReconcileSite( + new ReconcileSiteRequestDto { SiteIdentifier = "site1", NodeId = "node-a" }, NewContext(callCts.Token)); + + var call = probe.ExpectMsg(TimeSpan.FromSeconds(3)); + Assert.Equal("site1", call.Request.SiteIdentifier); + Assert.Equal(callCts.Token, call.CallCancelled); + + probe.Reply(new ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment.ReconcileSiteResponse( + new List(), + new List(), "http://central")); + await reply; + } + /// Verifies that the audit ingest timeout is the single shared constant. [Fact] public void TheAuditIngestTimeout_IsTheOneSharedConstant()