diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs index 6a7ec7b1..5fe5ea78 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/CentralCommandGrpcService.cs @@ -28,14 +28,16 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Grpc; /// The actor is resolved per call, not in the constructor. Kestrel answers before /// Akka.Hosting has finished spawning the actor graph, so a ctor lookup would throw /// during the boot window and surface as an opaque Unknown. Resolving per call turns -/// that window into — the retryable status — so the -/// node's next attempt lands rather than the deploy recording a permanent failure against a -/// node that applied fine. +/// that window into — the retryable status. The node's +/// sender does not in fact retry this endpoint; it advances to the next central endpoint in +/// its list, where the ack lands on a node whose actor graph is up. Either way the deploy does +/// not record a permanent failure against a node that applied fine. /// /// /// Decoding is graded before the registry. A message that cannot decode is /// even when the actor is also missing: Unavailable -/// invites a retry, and retrying a message that can never decode is a loop. The decode itself +/// is the status that says "try elsewhere", and moving a message that can never decode to +/// another endpoint only spends the second one too. The decode itself /// never guesses (see ) — an unspecified outcome, an /// unparseable id or a blank node name is refused rather than resolved to a plausible value, /// because the deploy path seals a revision on Applied and a guessed ack seals a @@ -105,8 +107,8 @@ public sealed class CentralCommandGrpcService : GeneratedServiceBase _log.LogWarning( "Received an ApplyAck for {DeploymentId} (correlation '{CorrelationId}') from {NodeId} " + "at {Peer} before the central communication actor was registered on this node; " - + "refusing with Unavailable so the caller retries rather than recording a failure " - + "against a node that applied", + + "refusing with Unavailable so the caller moves on to another central endpoint " + + "rather than recording a failure against a node that applied", ack.DeploymentId, ack.CorrelationId, ack.NodeId, context.Peer); throw new RpcException(new Status(StatusCode.Unavailable, "central command actor not ready")); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs index 6aa5b497..b47b2a96 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/GrpcApplyAckSender.cs @@ -19,11 +19,20 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; /// /// One ordered pass, stopping at the first success. /// is an availability list, not a -/// fan-out list: either central node hands the ack to the same -/// ConfigPublishCoordinator singleton proxy, so posting to both would have the -/// coordinator count this node twice — and a doubled count can seal a deployment other nodes -/// never applied. That is exactly why the transport being replaced used -/// ClusterClient.Send and not SendToAll. +/// fan-out list — the semantic of the ClusterClient.Send being replaced (Send, +/// deliberately not SendToAll). +/// +/// +/// Delivery is at-most-once per SUCCESSFUL attempt, but the walk as a whole is +/// at-least-once, and that is fine. An endpoint that processes the ack and then exceeds +/// its deadline before answering is indistinguishable here from one that never received it, so +/// the next endpoint is tried and central may see the ack twice. +/// ConfigPublishCoordinator.HandleAck absorbs that idempotently: acks land in a +/// NodeId-keyed dictionary behind an expected-set membership guard and a +/// stale-deployment guard, so a duplicate overwrites this node's own entry rather than +/// inflating the count, and the row upsert behind it is idempotent too. Duplicates therefore +/// cannot seal a deployment other nodes never applied — the single-endpoint rule is defence in +/// depth and a bound on wasted work, not a correctness dependency. /// /// /// No retry beyond that single pass, and no queue. The receptionist buffered nothing @@ -108,10 +117,14 @@ public sealed class GrpcApplyAckSender : IApplyAckSender, IDisposable /// public async Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default) { - ArgumentNullException.ThrowIfNull(ack); - try { + // INSIDE the try, not before it. This is an async method, so a guard clause above the try + // does not throw to the caller — it faults the returned task, which the caller discards. + // That would break this method's "never faults" promise in the one case a caller could + // never see. + ArgumentNullException.ThrowIfNull(ack); + await PostAsync(ack, cancellationToken); } catch (Exception ex) @@ -121,13 +134,15 @@ public sealed class GrpcApplyAckSender : IApplyAckSender, IDisposable // resurface as an unobserved exception at GC — a drop with no line in the node log, which // is precisely the failure mode this transport replaces. Encoding faults and non-RpcException // transport faults both land here. + // + // The ack is null-conditional because the guard above is one of the things this catches. _log.LogWarning( ex, "Dropping the ApplyAck for {DeploymentId} (correlation {CorrelationId}) from {NodeId}: " + "posting it to central failed unexpectedly. It is not buffered and will not be " + "retried, so the deployment will time out at central naming this node even though it " + "applied successfully.", - ack.DeploymentId, ack.CorrelationId, ack.NodeId); + ack?.DeploymentId, ack?.CorrelationId, ack?.NodeId); } } @@ -145,28 +160,53 @@ public sealed class GrpcApplyAckSender : IApplyAckSender, IDisposable } var message = MeshTransportDtoMapper.ToProto(ack); + + // Hoisted: read-only, and the walk is strictly sequential, so one Metadata is shared safely + // across the attempts. The DEADLINE emphatically is not hoisted — see below. var headers = new Metadata { { "authorization", $"Bearer {_options.GrpcPsk}" } }; if (_clusterId is not null) headers.Add(ClusterHeader, _clusterId); - var deadline = _options.AckTimeoutSeconds > 0 - ? DateTime.UtcNow.AddSeconds(_options.AckTimeoutSeconds) - : (DateTime?)null; - var failures = new List(endpoints.Length); foreach (var endpoint in endpoints) { try { + // Computed FRESH per endpoint, which is what MeshTransport:AckTimeoutSeconds documents + // ("applied per endpoint, not per ack"). A single absolute deadline hoisted out of this + // loop defeats failover against the characteristic WAN failure: a central that HANGS + // (SYN black-hole, frozen host) rather than refusing. The first endpoint burns the whole + // timeout, and the second is then dialled with a deadline already in the past — it fails + // instantly as DeadlineExceeded without a packet leaving the node, so the surviving + // central is never actually tried. + var deadline = _options.AckTimeoutSeconds > 0 + ? DateTime.UtcNow.AddSeconds(_options.AckTimeoutSeconds) + : (DateTime?)null; + var client = _clientFactory(endpoint); - await client.ReportApplyAsync(message, headers, deadline, cancellationToken); + using var call = client.ReportApplyAsync(message, headers, deadline, cancellationToken); + await call; - _log.LogDebug( - "Posted the ApplyAck for {DeploymentId} ({Outcome}) to {Endpoint}.", - ack.DeploymentId, ack.Outcome, endpoint); + if (failures.Count == 0) + { + _log.LogDebug( + "Posted the ApplyAck for {DeploymentId} ({Outcome}) to {Endpoint}.", + ack.DeploymentId, ack.Outcome, endpoint); + } + else + { + // Information, not Debug, because this is a FAILOVER: the ack landed, but only + // after one or more endpoints cost their full deadline. The deploy simply took + // that much longer, and without this line the only evidence is the absence of the + // all-endpoints-failed Warning — which reads identically to a healthy first-try. + _log.LogInformation( + "Posted the ApplyAck for {DeploymentId} ({Outcome}) to {Endpoint} after " + + "{FailedCount} endpoint(s) failed ({Failures}).", + ack.DeploymentId, ack.Outcome, endpoint, failures.Count, string.Join("; ", failures)); + } - // The single-count guarantee, and it is this `return`: the pass stops at the first - // endpoint that accepted the ack and never walks back. + // The single-endpoint rule, and it is this `return`: the walk stops at the first + // endpoint that accepted the ack and never goes back to one that did not. return; } catch (RpcException ex) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs index ca631ea6..092d30ab 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/IApplyAckSender.cs @@ -11,10 +11,14 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Communication; /// /// /// One ack per node per dispatch. Whatever the implementation does about redundancy, it -/// must deliver the ack to at most one central endpoint: both central nodes forward to -/// the same ConfigPublishCoordinator singleton proxy, so a fan-out would have the -/// coordinator count one node twice. That is the semantic of the ClusterClient.Send -/// this replaces — Send, deliberately not SendToAll. +/// posts to one central endpoint at a time rather than fanning out — the semantic of the +/// ClusterClient.Send this replaces (Send, deliberately not SendToAll). +/// This is defence in depth, not a correctness dependency: the coordinator counts acks +/// in a NodeId-keyed dictionary behind an expected-set membership guard +/// (ConfigPublishCoordinator.HandleAck), so a duplicate from one node overwrites its own +/// entry and cannot inflate the count or seal a deployment early. What fan-out would actually +/// cost is duplicated work at central, doubled deploy-ack rows written, and a transport whose +/// delivery count nobody can reason about — none of which is worth having. /// /// /// Fire-and-forget, and it never throws. The caller is an actor that discards the diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs index dfa46a12..f8b7d5fd 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Communication/NodeCommunicationActor.cs @@ -92,8 +92,26 @@ public sealed class NodeCommunicationActor : ReceiveActor // // Checked BEFORE the ClusterClient. The registration arm builds exactly one transport, so // both are never set — but were they, posting over both would deliver two acks for one - // apply and the coordinator would count this node twice. - _ = _ackSender.SendAsync(ack); + // apply and duplicate central's work for no benefit. + try + { + _ = _ackSender.SendAsync(ack); + } + catch (Exception ex) + { + // "Never throws" is a contract on a public seam, which makes it documentation until + // something enforces it. An implementation that threw synchronously — a guard clause + // ahead of the async state machine is the easy way to do it by accident — would fault + // this Receive, restart the actor, and take the node's whole inbound command path down + // with it. One dropped ack is a deployment that times out; a restarting comm actor is + // every command to this node going missing. + _log.Error( + ex, + "The ApplyAck sender threw synchronously for {DeploymentId}; dropping the ack. " + + "IApplyAckSender must never throw — this is a defect in the registered sender.", + ack.DeploymentId); + } + return; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 255f2590..7d14508e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -374,8 +374,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Database context factory for configuration database access. /// The local cluster node identifier. /// Where ApplyAcks are sent. A direct coordinator handle in tests; under - /// MeshTransport:Mode = ClusterClient the node's NodeCommunicationActor, which relays - /// across the boundary. Null publishes on the deployment-acks DPS topic. + /// MeshTransport:Mode = ClusterClient and = Grpc the node's + /// NodeCommunicationActor, which relays across the boundary over whichever transport it was + /// built with. Null publishes on the deployment-acks DPS topic. /// Optional driver factory; defaults to null factory if not provided. /// Optional set of roles assigned to the local node. /// Optional actor reference for dependency multiplexing. @@ -458,8 +459,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Database context factory for configuration database access. /// The local cluster node identifier. /// Where ApplyAcks are sent. A direct coordinator handle in tests; under - /// MeshTransport:Mode = ClusterClient the node's NodeCommunicationActor, which relays - /// across the boundary. Null publishes on the deployment-acks DPS topic. + /// MeshTransport:Mode = ClusterClient and = Grpc the node's + /// NodeCommunicationActor, which relays across the boundary over whichever transport it was + /// built with. Null publishes on the deployment-acks DPS topic. /// Optional driver factory; defaults to null factory if not provided. /// Optional set of roles assigned to the local node. /// Optional actor reference for dependency multiplexing. @@ -2490,11 +2492,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation); if (_ackRouter is not null) { - // Either a direct coordinator handle (tests) or, under - // MeshTransport:Mode = ClusterClient, the NodeCommunicationActor that relays this - // across the boundary. Renamed from _coordinatorOverride in per-cluster mesh Phase 2: - // under ClusterClient mode this ref is emphatically NOT a coordinator, and the old - // name would send the next reader looking for one. + // Either a direct coordinator handle (tests) or, under MeshTransport:Mode = ClusterClient + // and = Grpc alike, the NodeCommunicationActor that relays this across the boundary (it + // owns the choice of transport, not this actor). Renamed from _coordinatorOverride in + // per-cluster mesh Phase 2: under those two modes this ref is emphatically NOT a + // coordinator, and the old name would send the next reader looking for one. _ackRouter.Tell(ack); } else diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs index 58befc3e..7974ae99 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Communication/GrpcApplyAckSenderTests.cs @@ -115,8 +115,32 @@ public sealed class GrpcApplyAckSenderTests options.Headers.GetValue("authorization").ShouldBe($"Bearer {Psk}"); options.Headers.GetValue("x-otopcua-cluster").ShouldBe("SITE-A"); options.Deadline.ShouldNotBeNull(); - options.Deadline!.Value.ShouldBeGreaterThan(before); - options.Deadline!.Value.ShouldBeLessThanOrEqualTo(before.AddSeconds(7).AddSeconds(1)); + // Bounded from BOTH sides against the configured 7s. A lower bound of `before` alone would + // accept any positive timeout at all — the value could be ignored entirely and the fact would + // still pass. + options.Deadline!.Value.ShouldBeGreaterThan(before.AddSeconds(6)); + options.Deadline!.Value.ShouldBeLessThanOrEqualTo(before.AddSeconds(8)); + } + + [Fact] + public async Task Each_endpoint_gets_a_FRESH_deadline() + { + // The failure this pins is silent and is the characteristic WAN one: a central that HANGS + // rather than refusing. With one absolute deadline hoisted over the whole walk, endpoint 1 + // burns all of AckTimeoutSeconds, and endpoint 2 is then dialled with a deadline already in + // the past — it fails instantly as DeadlineExceeded without a packet leaving the node, so the + // surviving central is never really tried and failover is defeated while still LOOKING like it + // ran. Both stamps come from real UtcNow, so a delay on the failing path plus a strict `>` is + // the whole test. + var calls = new List<(ApplyAckMessage Request, CallOptions Options)>(); + var factory = (string endpoint) => endpoint.Contains("central-1", StringComparison.Ordinal) + ? (CentralCommandService.CentralCommandServiceClient)new SlowThrowingClient(calls, TimeSpan.FromMilliseconds(50)) + : new RecordingClient(calls); + + await Sender(["http://central-1:4055", "http://central-2:4055"], factory).SendAsync(Ack()); + + calls.Count.ShouldBe(2); + calls[1].Options.Deadline!.Value.ShouldBeGreaterThan(calls[0].Options.Deadline!.Value); } [Fact] @@ -268,9 +292,28 @@ public sealed class GrpcApplyAckSenderTests } [Fact] - public void Dispose_is_idempotent() + public async Task Dispose_releases_the_real_channel_cache_and_is_idempotent() { - var sender = Sender(["http://central-1:4055"], _ => new RecordingClient([])); + // Built with NO client factory, so the production DefaultClientFactory runs and a real + // GrpcChannel lands in the cache — otherwise Dispose is walking an empty dictionary and this + // test proves nothing about the thing it names. The address is loopback port 1: nothing + // listens, the connection is refused immediately, and no traffic leaves the machine. + var log = new CapturingLogger(); + var sender = new GrpcApplyAckSender( + Options.Create(new MeshTransportOptions + { + Mode = MeshTransportOptions.ModeGrpc, + CentralGrpcEndpoints = ["http://127.0.0.1:1"], + GrpcPsk = Psk, + AckTimeoutSeconds = 5, + }), + Options.Create(new AkkaClusterOptions { Roles = ["driver", "cluster-SITE-A"] }), + log); + + await sender.SendAsync(Ack()); + + // The post must have failed (nothing is listening) — which is what put a channel in the cache. + log.Entries.Any(e => e.Level == LogLevel.Warning).ShouldBeTrue(); sender.Dispose(); Should.NotThrow(sender.Dispose); @@ -303,6 +346,32 @@ public sealed class GrpcApplyAckSenderTests => throw toThrow; } + /// + /// A generated client double modelling a HUNG endpoint: it accepts the call, records it, then + /// takes real wall-clock time before failing — the shape a shared deadline cannot survive. + /// + private sealed class SlowThrowingClient( + List<(ApplyAckMessage Request, CallOptions Options)> calls, TimeSpan delay) + : CentralCommandService.CentralCommandServiceClient + { + public override AsyncUnaryCall ReportApplyAsync(ApplyAckMessage request, CallOptions options) + { + calls.Add((request, options)); + return new AsyncUnaryCall( + HangThenFail(delay), + Task.FromResult(new Metadata()), + () => new Status(StatusCode.DeadlineExceeded, "hung"), + () => [], + () => { }); + } + + private static async Task HangThenFail(TimeSpan delay) + { + await Task.Delay(delay); + throw new RpcException(new Status(StatusCode.DeadlineExceeded, "hung")); + } + } + /// A generated client double throwing something that is NOT an . private sealed class FaultingClient(Exception toThrow) : CentralCommandService.CentralCommandServiceClient { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs index 986bcc76..f649d30d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs @@ -1,13 +1,17 @@ +using Akka.Actor; using Akka.Hosting; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Shouldly; using Xunit; +using ZB.MOM.WW.OtOpcUa.Cluster; using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Communication; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests; @@ -121,6 +125,91 @@ public sealed class ServiceCollectionExtensionsTests } } + /// + /// Under MeshTransport:Mode = Grpc an ApplyAck produced by the driver host reaches the + /// registered , and no ClusterClient actor exists + /// (ClusterClient→gRPC migration Task 1B.2). + /// + /// + /// This is the fact that pins the ackRouter gate in WithOtOpcUaRuntimeActors + /// (useClusterClient || useGrpc ? nodeComm : null). Narrowed back to ClusterClient-only + /// that gate still compiles and every unit test still passes, while every ack on a Grpc node + /// goes out on the deployment-acks DPS topic instead — where, in the split-mesh topology + /// Grpc exists to serve, central is not subscribed. The whole transport would be dead and + /// nothing below this level of the stack could tell. + /// + /// The dispatch names a deployment that does not exist in the in-memory ConfigDb, so the apply + /// fails and the host emits a Failed ack — deterministic, synchronous, and it exercises the + /// real SendAck path rather than poking the router directly. + /// + [Fact] + public async Task Under_Grpc_a_driver_host_ack_reaches_the_ack_sender_and_no_ClusterClient_exists() + { + var sender = new RecordingAckSender(); + using var host = BuildRuntimeActorHost(services => + { + services.AddSingleton(sender); + services.Configure(o => + { + o.Mode = MeshTransportOptions.ModeGrpc; + o.CentralGrpcEndpoints = ["http://central-1:4055"]; + o.GrpcPsk = "rig-key"; + }); + }); + + await host.StartAsync(); + try + { + var system = host.Services.GetRequiredService(); + var driverHost = host.Services.GetRequiredService>(); + + driverHost.ActorRef.Tell(new DispatchDeployment( + new DeploymentId(Guid.NewGuid()), + new RevisionHash("rev-that-does-not-exist"), + new CorrelationId(Guid.NewGuid()))); + + // The ack travelled DriverHostActor -> NodeCommunicationActor -> IApplyAckSender. Under a + // ClusterClient-only gate it would have gone to DPS and this never completes. + var deadline = DateTime.UtcNow.AddSeconds(15); + while (sender.Sent.Count == 0 && DateTime.UtcNow < deadline) + await Task.Delay(50); + + sender.Sent.Count.ShouldBe(1); + + // And the Akka cross-mesh association is genuinely gone, not merely unused. + await Should.ThrowAsync(() => system + .ActorSelection($"/user/{ServiceCollectionExtensions.CentralClusterClientName}") + .ResolveOne(TimeSpan.FromMilliseconds(500))); + } + finally + { + await host.StopAsync(); + } + } + + /// An double recording what the node comm actor handed it. + private sealed class RecordingAckSender : IApplyAckSender + { + private readonly Lock _gate = new(); + private readonly List _sent = []; + + /// Gets a snapshot of the acks posted through this sender. + public IReadOnlyList Sent + { + get { lock (_gate) return [.. _sent]; } + } + + /// Records . + /// The ack handed over. + /// Unused. + /// A completed task. + public Task SendAsync(ApplyAck ack, CancellationToken cancellationToken = default) + { + lock (_gate) _sent.Add(ack); + return Task.CompletedTask; + } + } + /// Builds a driver-role host that runs WithOtOpcUaRuntimeActors, with optional /// extra DI registrations applied before AddAkka. private static IHost BuildRuntimeActorHost(Action? extra)