From 90d7f6dfca71b68e87c8e4d2086e56cee113245c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 17:29:45 -0400 Subject: [PATCH] fix(deploy): hold the Applied ack until the address-space apply outcome is known (#601) DriverHostActor no longer acks Applied the moment the revision advances. It asks OpcUaPublishActor for the outcome (RebuildAddressSpace.ReportOutcome) and acks Applied+Degraded(reason) when the apply was degraded; a missing outcome times out (60 s, inside the coordinator's 2 min deadline) to a degraded ack, never a clean one. Re-dispatching a revision whose apply was degraded or is still unconfirmed re-applies instead of taking the clean idempotent short-circuit - that re-apply is also what repairs the degraded nodes. Tests with a bare publish probe use a clean-outcome auto-pilot probe. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe --- .../Drivers/DriverHostActor.cs | 176 ++++++++++++++- .../DriverHostActorBootFromCacheTests.cs | 4 +- .../Drivers/DriverHostActorDbLessTests.cs | 8 +- .../DriverHostActorDegradedAckTests.cs | 203 ++++++++++++++++++ ...verHostActorFetchAndCacheBootstrapTests.cs | 2 +- ...erHostActorFetchAndCacheUnreadableTests.cs | 6 +- .../DriverHostActorHistoryWriterTests.cs | 2 +- .../DriverHostActorNativeAlarmTests.cs | 2 +- .../DriverHostActorPrimaryGateTests.cs | 2 +- .../Drivers/DriverHostActorTests.cs | 2 +- .../Harness/PublishProbes.cs | 34 +++ .../Harness/V3ArtifactSeeder.cs | 2 +- .../Telemetry/PublishSeamEmitsToHubTests.cs | 2 +- 13 files changed, 421 insertions(+), 24 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/PublishProbes.cs 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 33eafca8..af119dc4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -84,6 +84,19 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// public static readonly TimeSpan DefaultApplyWedgeTimeout = TimeSpan.FromMinutes(5); + /// + /// Gitea #601 — how long an Applied ack waits for OpcUaPublishActor's address-space apply + /// outcome () before it is sent anyway. + /// + /// + /// Deliberately well inside the coordinator's apply deadline + /// (ConfigPublishCoordinator.DefaultApplyDeadline, 2 min, which also has to cover the artifact + /// read/fetch): a hung or very slow apply must still produce an ack before central times the + /// deployment out. An ack sent on this timeout is marked degraded ("outcome unknown"), never + /// clean — the revision did advance, but nobody has confirmed the address space matches it. + /// + public static readonly TimeSpan DefaultApplyOutcomeTimeout = TimeSpan.FromSeconds(60); + /// Minimum interval between two supervision reprovisions of the SAME driver child (archreview /// review F3). Akka's default decider restarts a faulting child immediately and without limit, and the /// reprovision hook makes every restart do ShutdownAsync + InitializeAsync + SubscribeAsync — so an @@ -244,6 +257,21 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private IActorRef? _scriptedAlarmHost; private RevisionHash? _currentRevision; + + /// + /// Gitea #601 — the last address-space apply of reported DEGRADED (or no + /// outcome in time). A re-dispatch of that same revision then RE-APPLIES instead of taking the + /// idempotent "already at this revision" short-circuit: the short-circuit would ack a clean Applied + /// for an address space known to be partial, and the re-apply is what repairs it (a degraded + /// non-rebuild apply leaves the publish actor's composition un-advanced, so it re-diffs the missing + /// nodes). Reset by every clean outcome and by every boot path that restores a revision. + /// + private bool _currentRevisionDegraded; + + /// Gitea #601 — Applied acks held until the address-space apply reports its outcome, keyed by + /// the apply's correlation id (echoed back in ). + private readonly Dictionary _pendingAcks = new(); + private DeploymentId? _applyingDeploymentId; private readonly Dictionary _children = new(StringComparer.Ordinal); @@ -431,6 +459,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// private readonly Func? _sharesHistorizationScopeWithLocal; + /// Gitea #601 — how long an Applied ack waits for the address-space outcome (see + /// ). + private readonly TimeSpan _applyOutcomeTimeout; + /// This host's -wedge watchdog delay (see /// ). private readonly TimeSpan _applyWedgeTimeout; @@ -595,6 +627,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Archreview 03/S15: how long after a driver reconnects the /// recovery audit waits before reporting refs still latched BadNoCommunication. Defaults to /// ; tests shorten it. + /// Gitea #601: how long an Applied ack waits for the address-space + /// apply outcome before it is sent anyway, marked degraded ("outcome unknown"). Defaults to + /// ; tests shorten it. /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory? dbFactory, @@ -626,7 +661,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS TimeSpan? applyWedgeTimeout = null, TimeSpan? reprovisionBackoff = null, TimeSpan? commsRecoveryAuditDelay = null, - Func? sharesAlarmQueueWithLocal = null) => + Func? sharesAlarmQueueWithLocal = null, + TimeSpan? applyOutcomeTimeout = null) => // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // expression tree. Six IActorRef? parameters and several interface-typed ones mean a // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds @@ -639,7 +675,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher, alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView, sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay, - sharesAlarmQueueWithLocal)); + sharesAlarmQueueWithLocal, applyOutcomeTimeout)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -737,8 +773,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS TimeSpan? applyWedgeTimeout = null, TimeSpan? reprovisionBackoff = null, TimeSpan? commsRecoveryAuditDelay = null, - Func? sharesAlarmQueueWithLocal = null) + Func? sharesAlarmQueueWithLocal = null, + TimeSpan? applyOutcomeTimeout = null) { + _applyOutcomeTimeout = applyOutcomeTimeout ?? DefaultApplyOutcomeTimeout; _sharesAlarmQueueWithLocal = sharesAlarmQueueWithLocal; _applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout; _reprovisionBackoff = reprovisionBackoff ?? DefaultReprovisionBackoff; @@ -1288,6 +1326,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private void Steady() { Receive(HandleDispatchFromSteady); + // #601 — the held Applied ack is released by the address-space outcome (or its timeout); both can + // land in any behaviour, so all three register them. + Receive(HandleRebuildCompleted); + Receive(HandleApplyOutcomeTimedOut); Receive(HandleArtifactCached); Receive(HandleArtifactCacheFailed); // A wedge tick already in the mailbox when the apply finished — dropped at Debug, see the handler. @@ -1348,6 +1390,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Direct ConfigDb read (BeginFetchAndCacheApply / BeginDirectApply ran in Steady, then // Become(Applying)); completing the apply transitions back to Steady. Receive(HandleArtifactForApply); + Receive(HandleRebuildCompleted); + Receive(HandleApplyOutcomeTimedOut); // The cache write is piped back too, and it outlives the apply that started it by design — so // its completion can land in EITHER state and both must handle it or it dead-letters. Receive(HandleArtifactCached); @@ -2173,6 +2217,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes). // We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor). Receive(_ => { }); + // #601 — an apply that completed just before the host went Stale still owes its ack; the + // coordinator is not behind the ConfigDb, so release it exactly as in Steady. + Receive(HandleRebuildCompleted); + Receive(HandleApplyOutcomeTimedOut); // An inbound operator write can't be serviced while the config DB is unreachable — fast-fail so the // node-manager's bounded Ask gets an immediate clear status instead of dead-lettering into a timeout. Receive(_ => @@ -2219,7 +2267,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private void HandleDispatchFromSteady(DispatchDeployment msg) { - if (_currentRevision is { } cur && cur == msg.RevisionHash) + // #601: the short-circuit claims a CLEAN Applied, so it is only honest while the current revision's + // address space is known-good — not after a degraded apply, and not while its outcome is still + // pending. Either way the dispatch re-applies instead, which is also what repairs a degraded apply. + var currentRevisionUnconfirmed = _currentRevisionDegraded + || _pendingAcks.Values.Any(p => p.Revision == msg.RevisionHash); + if (_currentRevision is { } cur && cur == msg.RevisionHash && !currentRevisionUnconfirmed) { // Idempotent — already at this rev. Ack and stay Steady. _log.Debug("DriverHost {Node}: dispatch {Id} matches current rev {Rev}; immediate ACK", @@ -2228,6 +2281,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS return; } + if (_currentRevision is { } same && same == msg.RevisionHash) + _log.Info( + "DriverHost {Node}: dispatch {Id} matches current rev {Rev}, but its address-space apply was degraded or is unconfirmed — re-applying rather than acking clean", + _localNode, msg.DeploymentId, msg.RevisionHash); + if (_fetchAndCacheMode) { BeginFetchAndCacheApply(msg.DeploymentId, msg.RevisionHash, msg.CorrelationId); @@ -2387,7 +2445,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS _currentRevision = msg.Revision; UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Applied, failureReason: null); - SendAck(msg.DeploymentId, ApplyAckOutcome.Applied, failureReason: null, msg.Correlation); + // Gitea #601: the Applied ack is NOT sent here any more. It used to go out right now, before the + // address-space apply below had even been dequeued, so a degraded apply (swallowed sink faults, + // _lastApplied not advanced) still sealed the deployment green fleet-wide. The ack is now held + // until OpcUaPublishActor replies RebuildCompleted (HandleRebuildCompleted) or the outcome + // timeout fires (sent degraded, "outcome unknown"). With no publish actor there is no address + // space to wait for, so the ack goes out clean immediately — exactly the old behaviour. + AwaitAddressSpaceOutcomeThenAck(msg.DeploymentId, msg.Revision, msg.Correlation); // Rebuild the address space + push subscriptions FROM the bytes this apply already has, so // nothing downstream re-reads the artifact: under FetchAndCache OpcUaPublishActor has no // central SQL to read at all, and under Direct a bytes-less RebuildAddressSpace would make @@ -2399,8 +2463,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // different actor with its own mailbox, and ParsedDeploymentArtifact is a disposable, // non-thread-safe object owned by this handler (disposed in the finally below, likely before // the other actor even dequeues). It opens its own single parse on the far side. + // ReportOutcome: the reply (to Self — this Tell runs inside a receive, so the implicit sender is + // this actor) releases the held ack. Plain Tell + reply message, no Ask/await, so nothing here + // resumes off the actor thread. _opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace( - msg.Correlation, msg.DeploymentId, appliedBlob)); + msg.Correlation, msg.DeploymentId, appliedBlob, ReportOutcome: true)); // SubscribeBulk pass: hand each driver its desired tag references so live values flow into // the just-rebuilt address space instead of staying BadWaitingForInitialData. PushDesiredSubscriptionsFromParsed(msg.DeploymentId, parsed); @@ -2420,6 +2487,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS } catch (Exception ex) { + // #601: a throw after the Applied ack was put on hold — the Failed ack below is this apply's one + // answer, so the held Applied must never follow it when the address-space outcome lands. + DropPendingAck(msg.Correlation); UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Failed, ex.Message); SendAck(msg.DeploymentId, ApplyAckOutcome.Failed, ex.Message, msg.Correlation); OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair("outcome", "reject")); @@ -2440,6 +2510,94 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS } } + /// Gitea #601 — an Applied ack held until the address-space apply reports its outcome. + private sealed record PendingAck(DeploymentId DeploymentId, RevisionHash Revision, CorrelationId Correlation); + + /// Gitea #601 — self-timer: the address-space outcome for this correlation did not arrive in time. + private sealed record ApplyOutcomeTimedOut(CorrelationId Correlation); + + private static string ApplyOutcomeTimerKey(CorrelationId correlation) => $"apply-outcome-{correlation}"; + + /// + /// Gitea #601 — holds the Applied ack for an apply whose revision has just advanced until the + /// address-space apply outcome is known, arming as the backstop. + /// With no publish actor wired (tests, a node with no OPC UA data plane) there is no outcome to wait + /// for, so the ack is sent clean at once — byte-for-byte the pre-#601 behaviour. + /// + private void AwaitAddressSpaceOutcomeThenAck(DeploymentId deploymentId, RevisionHash revision, CorrelationId correlation) + { + if (_opcUaPublishActor is null) + { + _currentRevisionDegraded = false; + SendAck(deploymentId, ApplyAckOutcome.Applied, failureReason: null, correlation); + return; + } + + if (_pendingAcks.Remove(correlation, out var superseded)) + _log.Warning( + "DriverHost {Node}: a second apply reused correlation {Correlation} while deployment {Old}'s ack was still waiting for its address-space outcome — that ack is superseded", + _localNode, correlation, superseded.DeploymentId); + _pendingAcks[correlation] = new PendingAck(deploymentId, revision, correlation); + Timers.StartSingleTimer(ApplyOutcomeTimerKey(correlation), new ApplyOutcomeTimedOut(correlation), _applyOutcomeTimeout); + } + + /// Gitea #601 — forgets a held ack (its apply failed after all, and acks Failed instead). + private void DropPendingAck(CorrelationId correlation) + { + if (_pendingAcks.Remove(correlation)) + Timers.Cancel(ApplyOutcomeTimerKey(correlation)); + } + + /// + /// Gitea #601 — the address-space apply reported its outcome: release the held ack, clean or + /// degraded. Registered in every behaviour (Steady, Applying, Stale) because the reply can land in + /// any of them — the host returns to Steady before the publish actor has even dequeued the rebuild. + /// + private void HandleRebuildCompleted(OpcUa.OpcUaPublishActor.RebuildCompleted msg) + { + if (!_pendingAcks.Remove(msg.Correlation, out var pending)) + { + _log.Debug("DriverHost {Node}: address-space outcome for {Correlation} has no held ack (already timed out or superseded); dropping", + _localNode, msg.Correlation); + return; + } + + Timers.Cancel(ApplyOutcomeTimerKey(msg.Correlation)); + ReleasePendingAck(pending, msg.Degraded, msg.DegradedReason); + } + + /// + /// Gitea #601 — the address-space outcome did not arrive within . + /// The ack is sent anyway so the coordinator does not time the deployment out on a node that DID + /// advance its revision — but degraded, never clean: nobody has confirmed the address space. + /// + private void HandleApplyOutcomeTimedOut(ApplyOutcomeTimedOut msg) + { + if (!_pendingAcks.Remove(msg.Correlation, out var pending)) return; + + ReleasePendingAck(pending, degraded: true, + $"the address-space apply did not report an outcome within {_applyOutcomeTimeout.TotalSeconds:0}s; it may still be running or may have been lost"); + } + + private void ReleasePendingAck(PendingAck pending, bool degraded, string? reason) + { + // Only the CURRENT revision's flag is ours to set — an older apply's late outcome must not mark a + // newer revision degraded (or clean). + if (_currentRevision is { } cur && cur == pending.Revision) + _currentRevisionDegraded = degraded; + + if (degraded) + { + OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair("outcome", "degraded")); + _log.Warning( + "DriverHost {Node}: deployment {Id} (rev {Rev}) applied DEGRADED — {Reason}. Acking Applied+degraded; re-dispatching the same revision re-applies it", + _localNode, pending.DeploymentId, pending.Revision, reason); + } + + SendAck(pending.DeploymentId, ApplyAckOutcome.Applied, failureReason: null, pending.Correlation, + degraded, degraded ? reason ?? "address-space apply degraded" : null); + } + /// /// Self-timer telling this host it has been in for /// without finishing. @@ -3662,9 +3820,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS } } - private void SendAck(DeploymentId deploymentId, ApplyAckOutcome outcome, string? failureReason, CorrelationId correlation) + private void SendAck( + DeploymentId deploymentId, ApplyAckOutcome outcome, string? failureReason, CorrelationId correlation, + bool degraded = false, string? degradedReason = null) { - var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation); + var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation, degraded, degradedReason); if (_ackRouter is not null) { // Either a direct coordinator handle (tests) or, under MeshTransport:Mode = Grpc, the diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs index 11da96f3..bdc671f2 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs @@ -67,7 +67,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase var blob = EmptyArtifact(); var cached = new CachedDeploymentArtifact( DeploymentId.NewId().ToString(), CachedRev.Value, blob, DateTimeOffset.UtcNow.AddHours(-1)); - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, ackRouter: null, @@ -92,7 +92,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase var cached = new CachedDeploymentArtifact( DeploymentId.NewId().ToString(), CachedRev.Value, System.Text.Encoding.Latin1.GetBytes(text), DateTimeOffset.UtcNow.AddHours(-1)); - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, ackRouter: null, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDbLessTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDbLessTests.cs index dc3504f9..fbf049f3 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDbLessTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDbLessTests.cs @@ -112,7 +112,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase var cache = new RecordingArtifactCache(); var fetcher = new RecordingFetcher(ArtifactBytes()); var deploymentId = DeploymentId.NewId(); - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var coordinator = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( @@ -141,7 +141,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase [Fact] public void DbLessNode_WithLocalDbStore_SpawnsScriptedAlarmHost() { - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( dbFactory: null, TestNode, @@ -162,7 +162,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase [Fact] public void DbLessNode_WithNoStoreAndNoDbFactory_SkipsScriptedAlarmHost() { - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( dbFactory: null, TestNode, @@ -185,7 +185,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase [Fact] public void DbBackedNode_WithNoStore_SkipsScriptedAlarmHost() { - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( dbFactory: NewInMemoryDbFactory(), TestNode, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs new file mode 100644 index 00000000..570d4e48 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs @@ -0,0 +1,203 @@ +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Gitea #601 (archreview 01/S-9) — the deploy ack must be honest about the address-space apply. It used +/// to be sent the moment the revision advanced, before OpcUaPublishActor had even dequeued the +/// rebuild, so a degraded apply (swallowed sink faults; _lastApplied deliberately not advanced) +/// still acked a clean and the deployment sealed green fleet-wide. +/// Now the Applied ack waits for the publish actor's and +/// carries its verdict; a missing verdict times out to degraded, never clean. +/// +[Collection("opcua-publish-meters")] +public sealed class DriverHostActorDegradedAckTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + + [Fact] + public void The_Applied_ack_waits_for_the_address_space_outcome_and_a_clean_one_acks_clean() + { + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + var rebuild = publish.FishForMessage(_ => true, PresenceBudget); + rebuild.ReportOutcome.ShouldBeTrue(); + // The ack is HELD until the outcome lands — the pre-#601 code had already sent it by now. + coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + + publish.Reply(new OpcUaPublishActor.RebuildCompleted(rebuild.Correlation, rebuild.DeploymentId, false, null)); + + var ack = coordinator.ExpectMsg(PresenceBudget); + ack.Outcome.ShouldBe(ApplyAckOutcome.Applied); + ack.Degraded.ShouldBeFalse(); + ack.DegradedReason.ShouldBeNull(); + } + + [Fact] + public void A_degraded_address_space_outcome_acks_Applied_but_degraded_with_the_reason() + { + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + var rebuild = publish.FishForMessage(_ => true, PresenceBudget); + publish.Reply(new OpcUaPublishActor.RebuildCompleted( + rebuild.Correlation, rebuild.DeploymentId, Degraded: true, DegradedReason: "failedNodes=2")); + + var ack = coordinator.ExpectMsg(PresenceBudget); + ack.Outcome.ShouldBe(ApplyAckOutcome.Applied); + ack.Degraded.ShouldBeTrue(); + ack.DegradedReason.ShouldBe("failedNodes=2"); + ack.DeploymentId.ShouldBe(deploymentId); + } + + [Fact] + public void A_real_publish_actor_whose_sink_faults_produces_a_degraded_ack_end_to_end() + { + // No hand-written verdict: the real OpcUaPublishActor + AddressSpaceApplier over a sink that faults on + // every node. This is the production shape of the defect — a degraded apply that sealed green. + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var sink = new FaultingSink(); + var publish = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + sink: sink, applier: new AddressSpaceApplier(sink, NullLogger.Instance))); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + var ack = coordinator.ExpectMsg(PresenceBudget); + ack.Outcome.ShouldBe(ApplyAckOutcome.Applied); + ack.Degraded.ShouldBeTrue(); + ack.DegradedReason.ShouldNotBeNull(); + ack.DegradedReason.ShouldContain("failedNodes="); + } + + [Fact] + public void A_real_publish_actor_with_a_healthy_sink_produces_a_clean_ack_end_to_end() + { + // The control for the test above: same wiring, a sink that does not fault ⇒ clean. + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var sink = new FaultingSink(fault: false); + var publish = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + sink: sink, applier: new AddressSpaceApplier(sink, NullLogger.Instance))); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + var ack = coordinator.ExpectMsg(PresenceBudget); + ack.Outcome.ShouldBe(ApplyAckOutcome.Applied); + ack.Degraded.ShouldBeFalse(); + } + + [Fact] + public void No_outcome_within_the_timeout_acks_degraded_never_clean() + { + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); // never replies — a hung / lost apply + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref, applyOutcomeTimeout: TimeSpan.FromMilliseconds(300))); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + var ack = coordinator.ExpectMsg(PresenceBudget); + ack.Outcome.ShouldBe(ApplyAckOutcome.Applied); + ack.Degraded.ShouldBeTrue(); + ack.DegradedReason.ShouldNotBeNull(); + ack.DegradedReason.ShouldContain("did not report an outcome"); + + // A late outcome after the timeout must not produce a second ack. + var rebuild = publish.FishForMessage(_ => true, PresenceBudget); + publish.Reply(new OpcUaPublishActor.RebuildCompleted(rebuild.Correlation, rebuild.DeploymentId, false, null)); + coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + [Fact] + public void Redispatching_a_degraded_revision_reapplies_instead_of_acking_clean() + { + var (db, deploymentId) = SeedOneTagDeployment(); + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + var first = publish.FishForMessage(_ => true, PresenceBudget); + publish.Reply(new OpcUaPublishActor.RebuildCompleted(first.Correlation, first.DeploymentId, true, "failedNodes=1")); + coordinator.ExpectMsg(PresenceBudget).Degraded.ShouldBeTrue(); + + // Same revision again: the old idempotent short-circuit would ack a CLEAN Applied right here. It must + // re-apply instead (a second rebuild request), and ack whatever THAT apply reports. + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + var second = publish.FishForMessage(_ => true, PresenceBudget); + coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + publish.Reply(new OpcUaPublishActor.RebuildCompleted(second.Correlation, second.DeploymentId, false, null)); + coordinator.ExpectMsg(PresenceBudget).Degraded.ShouldBeFalse(); + + // Now clean: a third dispatch of the same revision takes the short-circuit again (no rebuild). + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + var third = coordinator.ExpectMsg(PresenceBudget); + third.Degraded.ShouldBeFalse(); + publish.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + + private (IDbContextFactory Db, DeploymentId Id) SeedOneTagDeployment() + { + var db = NewInMemoryDbFactory(); + var id = V3ArtifactSeeder.SeedV3Deployment(db, RevA, + rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) }, + unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) }); + return (db, id); + } + + /// A sink that throws from every node-creating call when fault is set. + private sealed class FaultingSink(bool fault = true) : IOpcUaAddressSpaceSink + { + private void MaybeFault() + { + if (fault) throw new InvalidOperationException("simulated sink fault"); + } + + public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { } + public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { } + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { } + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) => MaybeFault(); + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm = AddressSpaceRealm.Uns) => MaybeFault(); + public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null, bool refuseArrayOnScalar = false) => MaybeFault(); + public void RebuildAddressSpace() { } + public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { } + public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { } + public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs index f7e113b6..b24bf7b8 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs @@ -42,7 +42,7 @@ public sealed class DriverHostActorFetchAndCacheBootstrapTests : RuntimeActorTes var cached = new CachedDeploymentArtifact( DeploymentId.NewId().ToString(), CachedRev.Value, blob, DateTimeOffset.UtcNow.AddHours(-1)); var cache = new StubArtifactCache(cached); - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, ackRouter: null, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheUnreadableTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheUnreadableTests.cs index f7d2b8d7..a1963a56 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheUnreadableTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheUnreadableTests.cs @@ -42,7 +42,7 @@ public sealed class DriverHostActorFetchAndCacheUnreadableTests : RuntimeActorTe [RevA.Value] = Artifact(), [RevB.Value] = null, }); - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); var coordinator = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( @@ -72,7 +72,7 @@ public sealed class DriverHostActorFetchAndCacheUnreadableTests : RuntimeActorTe // empty blob handed to it directly: ReconcileDriversFromBlob's #485 guard fails the apply. var db = NewInMemoryDbFactory(); var fetcher = new MapFetcher(new Dictionary { [RevA.Value] = Array.Empty() }); - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); var coordinator = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( @@ -95,7 +95,7 @@ public sealed class DriverHostActorFetchAndCacheUnreadableTests : RuntimeActorTe // A truncated/corrupt cached artifact surfaces as GetCurrentUnkeyedAsync == null (the cache's // integrity check is deliberately indistinguishable from a miss). Boot must land // Steady-no-revision and apply NOTHING — never a partial address space. - var publishProbe = CreateTestProbe(); + var publishProbe = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, ackRouter: null, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistoryWriterTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistoryWriterTests.cs index 76869edd..ab17f6ad 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistoryWriterTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistoryWriterTests.cs @@ -53,7 +53,7 @@ public sealed class DriverHostActorHistoryWriterTests : RuntimeActorTestBase // A publish actor MUST be wired for SpawnVirtualTagHost to spawn the real host (its ctor // requires a non-null sink). No virtualTagHostOverride ⇒ the real VirtualTagHostActor is // spawned with _historyWriter threaded in. - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var writer = new CapturingHistoryWriter(); var actor = Sys.ActorOf(DriverHostActor.Props( diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmTests.cs index 848ee47d..afb61f5b 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmTests.cs @@ -441,7 +441,7 @@ public sealed class DriverHostActorNativeAlarmTests : RuntimeActorTestBase IDbContextFactory db, DeploymentId deploymentId) { var coordinator = CreateTestProbe(); - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var mux = CreateTestProbe(); var vtHost = CreateTestProbe(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs index e16e1b7c..11d0aaa9 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs @@ -251,7 +251,7 @@ public sealed class DriverHostActorPrimaryGateTests : RuntimeActorTestBase IDbContextFactory db, DeploymentId dep, int driverMemberCount) { var coordinator = CreateTestProbe(); - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var mux = CreateTestProbe(); var vtHost = CreateTestProbe(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorTests.cs index a712a339..eda42d74 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorTests.cs @@ -226,7 +226,7 @@ public sealed class DriverHostActorTests : RuntimeActorTestBase seeded.Length.ShouldBeGreaterThan(0); var coordinator = CreateTestProbe(); - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/PublishProbes.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/PublishProbes.cs new file mode 100644 index 00000000..dae22a2e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/PublishProbes.cs @@ -0,0 +1,34 @@ +using Akka.Actor; +using Akka.TestKit; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +/// +/// Gitea #601 — since the deploy ack waits for 's +/// reply, a bare standing in for the +/// publish actor would hold every Applied ack until the outcome timeout (and then ack it degraded). Tests +/// that are not ABOUT the ack's honesty use this probe instead: it still records every message exactly +/// like a plain probe (the auto-pilot runs in addition to queueing), and answers each +/// that asks for an outcome with a CLEAN one — the +/// reply the real publish actor gives a clean apply. +/// +public static class PublishProbes +{ + /// Creates a publish-actor probe that reports a clean address-space outcome for every rebuild + /// carrying . + /// The calling TestKit. + /// The probe. + public static TestProbe CreateCleanOutcomePublishProbe(this TestKitBase kit) + { + var probe = kit.CreateTestProbe(); + probe.SetAutoPilot(new DelegateAutoPilot((sender, message) => + { + if (message is OpcUaPublishActor.RebuildAddressSpace { ReportOutcome: true } rebuild) + sender.Tell(new OpcUaPublishActor.RebuildCompleted( + rebuild.Correlation, rebuild.DeploymentId, Degraded: false, DegradedReason: null)); + return AutoPilot.KeepRunning; + })); + return probe; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs index da329e79..e928b1b2 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs @@ -151,7 +151,7 @@ public static class V3ArtifactSeeder TimeSpan? reprovisionBackoff = null, TimeSpan? commsRecoveryAuditDelay = null) { - var publish = kit.CreateTestProbe(); + var publish = kit.CreateCleanOutcomePublishProbe(); var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref, driverFactory, reprovisionBackoff, commsRecoveryAuditDelay); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs index 16be3cf8..4cfaf6c7 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs @@ -232,7 +232,7 @@ public sealed class PublishSeamEmitsToHubTests : RuntimeActorTestBase IDbContextFactory db, DeploymentId deploymentId, ITelemetryLocalHub hub) { var coordinator = CreateTestProbe(); - var publish = CreateTestProbe(); + var publish = this.CreateCleanOutcomePublishProbe(); var mux = CreateTestProbe(); var vtHost = CreateTestProbe();