fix(communication): ReconcileSite threads the gRPC call's cancellation into central reconcile work (review 2026-10-21 03/P-1b)

This commit is contained in:
Joseph Doherty
2026-10-22 08:32:07 -04:00
parent 429ccb8d90
commit dd60e236bd
5 changed files with 94 additions and 5 deletions
@@ -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<ReconcileSiteRequest>(HandleReconcileSiteRequest);
Receive<ReconcileSiteRequest>(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<ReconcileSiteCall>(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.
/// </summary>
private void HandleReconcileSiteRequest(ReconcileSiteRequest msg)
/// <param name="msg">The site node's reconcile request.</param>
/// <param name="callCancelled">
/// 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. <see cref="CancellationToken.None"/> for an in-process caller.
/// </param>
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<ReconcileService>();
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<string, SiteGrpcEndpoints> GrpcContacts,
long Sequence);
/// <summary>
/// In-process envelope (never on the wire) pairing a site's <see cref="ReconcileSiteRequest"/>
/// 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).
/// </summary>
/// <param name="Request">The reconcile request.</param>
/// <param name="CallCancelled">The gRPC call's <c>ServerCallContext.CancellationToken</c>.</param>
internal sealed record ReconcileSiteCall(ReconcileSiteRequest Request, CancellationToken CallCancelled);
/// <summary>A failed (or deadline-cancelled) site-address DB load, piped back to release the overlap guard.</summary>
/// <param name="Sequence">The failed load's sequence stamp.</param>
/// <param name="Cause">The failure.</param>
@@ -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<ReconcileSiteResponse>(
central,
CentralControlDtoMapper.FromDto(request),
new ReconcileSiteCall(CentralControlDtoMapper.FromDto(request), context.CancellationToken),
_options.QueryTimeout,
context).ConfigureAwait(false);