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 cdc806cf..cab1515b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -64,6 +64,24 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Publishing interval handed to each driver's SubscribeBulk pass after an apply. private static readonly TimeSpan SubscriptionPublishingInterval = TimeSpan.FromSeconds(1); + /// Timer key for the -state wedge watchdog. Started by + /// , cancelled by . + private const string ApplyWedgeTimerKey = "apply-wedged"; + + /// + /// How long the host may sit in before it says so. + /// + /// + /// Five minutes ≈ 2× the worst configured artifact-fetch deadline, so an ordinary slow fetch + /// never trips it and a genuinely wedged one always does. It is not a timeout: nothing is + /// cancelled or failed, because the apply may still land and the coordinator's own apply + /// deadline is the authority on when a deployment is dead. The watchdog exists purely so a host + /// stuck in is VISIBLE — the pre-stash forward-spin used to make that + /// unmissable (a flood of log lines), and taking the spin away accidentally took the only alarm + /// with it. + /// + public static readonly TimeSpan DefaultApplyWedgeTimeout = TimeSpan.FromMinutes(5); + /// /// Cache key used when an artifact carries no cluster scoping (single-cluster or unscoped /// deployments). A literal rather than an empty string so the row is visibly deliberate when @@ -162,6 +180,37 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// public IStash Stash { get; set; } = null!; + /// + /// The set of deployment ids currently sitting in , so the + /// state stashes each distinct deployment AT MOST ONCE. Cleared in + /// , which is the same moment empties + /// the stash itself — a dispatch that is re-stashed by the NEXT apply re-adds itself naturally. + /// + /// + /// + /// It bounds the stash by DISTINCT deployments. A coordinator that re-dispatches on a + /// timer (or several coordinators racing) can emit the same deployment many times inside one + /// apply window; without this, every copy is stashed and every copy is re-delivered on the + /// way back to — an unbounded queue of messages that will all + /// short-circuit as duplicates anyway. + /// + /// + /// It also moots an Akka stash-equality question — which was raised in review as a + /// restart hazard and, MEASURED, is not one on the pinned version. The string + /// "Can't stash the same message {0} more than once" (the + /// IllegalActorStateException that AbstractStash.Stash() is documented to + /// throw when an envelope equals the last one stashed) IS present verbatim in the shipped + /// Akka.dll 1.5.62. But a throwaway probe actor on this exact package — stashing one + /// message instance twice in a row, then two value-equal instances, both with and without an + /// explicit sender — accepted all four stashes and never threw. So the guard is not on this + /// path in 1.5.62, and the dedup set is NOT fixing a live crash. What it does is make the + /// question moot: since is a sealed record (value + /// equality), a future Akka that does enforce the check would restart this host in the + /// middle of a deployment, and nothing here would have to change to stay safe. + /// + /// + private readonly HashSet _deferredDispatches = new(); + /// The single VirtualTag-host child that spawns/reconciles Equipment-namespace /// VirtualTagActors and bridges their results onto the OPC UA publish actor. Spawned in /// when an OPC UA publish actor is wired; receives @@ -320,6 +369,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// private readonly Func? _sharesHistorizationScopeWithLocal; + /// This host's -wedge watchdog delay (see + /// ). + private readonly TimeSpan _applyWedgeTimeout; + /// /// Host of this node's LocalDb replication partner — the only node that holds a copy of this /// node's alarm queue, and therefore the only node it may ever stand down in favour of. @@ -431,6 +484,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// Per-cluster mesh Phase 5: optional node-local live-telemetry hub each /// Primary-gated native-alarm transition is also emitted into, and which is threaded into the spawned /// VirtualTag + ScriptedAlarm hosts. Defaults to null (admin-only nodes / tests skip the emit). + /// How long the host may sit in before the + /// wedge watchdog logs a Warning. Defaults to . /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory? dbFactory, @@ -458,7 +513,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS ITelemetryLocalHub? telemetryHub = null, IDriverPrimaryRoleSource? primaryRoleSource = null, IHistorizationRoleView? historizationRoleView = null, - Func? sharesHistorizationScopeWithLocal = null) => + Func? sharesHistorizationScopeWithLocal = null, + TimeSpan? applyWedgeTimeout = 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 @@ -470,7 +526,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher, alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView, - sharesHistorizationScopeWithLocal)); + sharesHistorizationScopeWithLocal, applyWedgeTimeout)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -526,6 +582,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// snapshot serves this node's address space and therefore records the same historized refs. Null /// (the default) means the caller could not establish it, and this node never stands down — see /// . + /// How long the host may sit in before the + /// wedge watchdog logs a Warning. Defaults to ; a test shortens + /// it so the watchdog is observable in a test's lifetime. public DriverHostActor( IDbContextFactory? dbFactory, CommonsNodeId localNode, @@ -552,8 +611,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS ITelemetryLocalHub? telemetryHub = null, IDriverPrimaryRoleSource? primaryRoleSource = null, IHistorizationRoleView? historizationRoleView = null, - Func? sharesHistorizationScopeWithLocal = null) + Func? sharesHistorizationScopeWithLocal = null, + TimeSpan? applyWedgeTimeout = null) { + _applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout; _historizationRoleView = historizationRoleView; _sharesHistorizationScopeWithLocal = sharesHistorizationScopeWithLocal; _telemetryHub = telemetryHub; @@ -957,9 +1018,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// by grep. The constructor is the one exception: is assigned by the /// ActorCell only after construction returns. /// + /// + /// It is also the single place the state's two side-tables are + /// released: (which must be cleared in lockstep with the + /// unstash, or the re-delivered dispatch would be treated as an already-deferred duplicate + /// and dropped) and the watchdog. + /// /// private void BecomeSteady() { + // Cancelled BEFORE the unstash so a re-stash by the very next apply starts a fresh window. + Timers.Cancel(ApplyWedgeTimerKey); + _deferredDispatches.Clear(); Become(Steady); Stash.UnstashAll(); } @@ -969,6 +1039,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS Receive(HandleDispatchFromSteady); Receive(HandleArtifactCached); Receive(HandleArtifactCacheFailed); + // A wedge tick already in the mailbox when the apply finished — dropped at Debug, see the handler. + Receive(HandleApplyWedged); Receive(HandleGetDiagnostics); Receive(ForwardBatchToMux); Receive(ForwardNativeAlarm); @@ -997,6 +1069,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS return; } + // At most ONE stashed copy per distinct deployment (see _deferredDispatches): the redundant + // copies would all short-circuit as duplicates on the way out anyway, so a re-dispatching + // coordinator would otherwise grow the stash with retries rather than with work. + if (!_deferredDispatches.Add(msg.DeploymentId)) + { + _log.Debug("DriverHost {Node}: dispatch for {Id} is already deferred behind {Cur}; dropping the duplicate", + _localNode, msg.DeploymentId, _applyingDeploymentId); + return; + } + // Stashed, NOT re-forwarded to Self: a forward re-enqueues on this actor's own mailbox and // is dequeued again immediately, spinning for the whole apply. Every transition back to // Steady unstashes (see BecomeSteady), so this is logged once per deferred dispatch rather @@ -1012,6 +1094,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // its completion can land in EITHER state and both must handle it or it dead-letters. Receive(HandleArtifactCached); Receive(HandleArtifactCacheFailed); + Receive(HandleApplyWedged); Receive(HandleGetDiagnostics); Receive(ForwardBatchToMux); Receive(ForwardNativeAlarm); @@ -1573,6 +1656,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS { _log.Warning("DriverHost {Node}: ignoring DispatchDeployment while Stale (DB unreachable)", _localNode); }); + // A cache store started before this node fell Stale still pipes back here (see + // CacheAppliedArtifact: Bootstrap's orphan-replay ApplyAndAck and RestoreApplied both start + // stores, and either can be followed by the ConfigDb failure that lands us in Stale). Without + // these the pipe-back dead-letters and a failed cache write goes unreported in exactly the + // situation — central unreachable — where the local cache is the thing keeping the node alive. + Receive(HandleArtifactCached); + Receive(HandleArtifactCacheFailed); Receive(HandleGetDiagnostics); Receive(_ => TryRecoverFromStale()); Receive(HandleRestartDriver); @@ -1649,10 +1739,19 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// inside a receive. The apply itself completes synchronously on the actor thread in /// , mirroring . /// + /// + /// This is the ONLY path that arms the watchdog, deliberately. + /// enters and leaves inside a single mailbox + /// turn, so it cannot be "stuck in Applying" in any sense a watchdog could report: if it hung, + /// it would be hanging the actor thread, and a watchdog delivered as a mailbox message could not + /// be dequeued to say so. The fetch is the only unbounded wait during which the host is still + /// alive and dequeuing — which is exactly what makes a silent wedge possible there. + /// private void BeginFetchAndCacheApply(DeploymentId deploymentId, RevisionHash revision, CorrelationId correlation) { _applyingDeploymentId = deploymentId; Become(Applying); + Timers.StartSingleTimer(ApplyWedgeTimerKey, new ApplyWedged(deploymentId), _applyWedgeTimeout); // Persist Applying row (idempotent on PK) — same crash-boundary marker as ApplyAndAck. UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Applying, failureReason: null); @@ -1762,6 +1861,39 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS } } + /// + /// Self-timer telling this host it has been in for + /// without finishing. + /// + private sealed record ApplyWedged(DeploymentId DeploymentId); + + /// + /// Reports a host that is still in long after it should have finished. + /// + /// + /// Purely a signal — nothing is cancelled and no ACK is sent, because the apply may yet land and + /// the coordinator's apply deadline already owns "this deployment is dead". Registered in + /// AND : cancels the timer, + /// but a fire already in the mailbox when the apply completes still has to land somewhere, and + /// dead-lettering it would be a second, more confusing signal. Such a late fire finds + /// null (or naming a different deployment) and drops at Debug. + /// + private void HandleApplyWedged(ApplyWedged msg) + { + if (_applyingDeploymentId is null || msg.DeploymentId != _applyingDeploymentId.Value) + { + _log.Debug("DriverHost {Node}: late apply-wedge tick for {Id} (in-flight={Cur}); dropping", + _localNode, msg.DeploymentId, _applyingDeploymentId); + return; + } + + _log.Warning( + "DriverHost {Node}: deployment {Id} appears WEDGED — still applying after {Timeout} " + + "(the artifact fetch has neither answered nor failed); {Deferred} further dispatch(es) are " + + "deferred behind it. The node keeps serving revision {Rev}.", + _localNode, msg.DeploymentId, _applyWedgeTimeout, _deferredDispatches.Count, _currentRevision); + } + /// Pipe-back carrier for a FetchAndCache fetch result (null bytes = no answer). private sealed record FetchedForApply( DeploymentId DeploymentId, RevisionHash Revision, CorrelationId Correlation, byte[]? Bytes); @@ -2065,9 +2197,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// driver value burst, a diagnostics request or the next dispatch. The task is piped back as /// / so completion is observed /// rather than fire-and-forgotten — the earlier objection to this shape was that the - /// pipe-back needs a handler in every state it can land in, which is exactly why both - /// and register one. (Stale does not: it - /// is only ever entered from , before any apply has started a store.) + /// pipe-back needs a handler in every state it can land in, which is why + /// , AND all register one. + /// + /// + /// Stale is included on purpose. This used to claim Stale could not be reached after a + /// store had started, because it is only ever entered from . That is + /// not a property of the Stale entry — starts stores of its own, via + /// the orphan-replay and via , and both + /// run inside the very try block whose catch Becomes Stale. Today no statement between those + /// calls and the catch happens to throw, so the old claim held by accident of ordering rather + /// than by design; a restart can orphan an in-flight pipe-back the same way. Registering the + /// handlers costs two lines and removes the need to re-derive that argument. /// /// /// The apply does not depend on the outcome — it never did, and the async shape keeps @@ -2132,14 +2273,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS .PipeTo( Self, success: () => new ArtifactCached(deploymentId), - failure: ex => new ArtifactCacheFailed(deploymentId, ex.Message)); + failure: ex => new ArtifactCacheFailed(deploymentId, ex)); } catch (Exception ex) { // A cache that throws SYNCHRONOUSLY (before returning a task) never produces a pipe-back, // so route it to the same handler by hand rather than letting it unwind into the apply's // catch — which would send a second, contradictory Failed ACK for a live deployment. - Self.Tell(new ArtifactCacheFailed(deploymentId, ex.Message)); + Self.Tell(new ArtifactCacheFailed(deploymentId, ex)); } } @@ -2152,17 +2293,25 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private sealed record ArtifactCached(DeploymentId DeploymentId); /// The node-local artifact cache could not store this deployment's bytes. - private sealed record ArtifactCacheFailed(DeploymentId DeploymentId, string Reason); + /// + /// Carries the exception, not a pre-flattened message string. A LocalDb/SQLite failure is + /// routinely a wrapped inner exception, and ex.Message alone ("An error occurred while + /// …") names neither the SQLite result code nor the frame that produced it — the detail an + /// operator needs to tell a full disk from a locked file from a schema mismatch. Logging it as + /// the log event's Cause also lets a test assert on the exception TYPE rather than on + /// prose. + /// + private sealed record ArtifactCacheFailed(DeploymentId DeploymentId, Exception Cause); private void HandleArtifactCached(ArtifactCached msg) => _log.Info("DriverHost {Node}: cached applied deployment {Id} as this node's local fallback", _localNode, msg.DeploymentId); private void HandleArtifactCacheFailed(ArtifactCacheFailed msg) => - _log.Warning( + _log.Warning(msg.Cause, "DriverHost {Node}: deployment {Id} could not be cached locally ({Reason}). The apply itself " + "succeeded and is unaffected; this node simply has no local fallback for that revision.", - _localNode, msg.DeploymentId, msg.Reason); + _localNode, msg.DeploymentId, msg.Cause.Message); /// /// Restore the served state for an already-applied deployment after a process restart. diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorApplyingStashTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorApplyingStashTests.cs index d7db6d67..4ca52a95 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorApplyingStashTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorApplyingStashTests.cs @@ -3,7 +3,9 @@ using Akka.Actor; using Akka.Event; using Shouldly; using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet; using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; @@ -36,6 +38,10 @@ public sealed class DriverHostActorApplyingStashTests() private static readonly NodeId TestNode = NodeId.Parse("driver-test"); private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64)); + private static readonly RevisionHash RevC = RevisionHash.Parse(new string('c', 64)); + + /// How long a forward-spin gets to announce itself thousands of times if it is back. + private static readonly TimeSpan SpinWindow = TimeSpan.FromSeconds(1.5); /// Minimal parseable artifact — no drivers. private static byte[] ArtifactBytes() => @@ -63,24 +69,20 @@ public sealed class DriverHostActorApplyingStashTests() actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId())); AwaitAssert(() => fetcher.Calls.ShouldBe(1), duration: PresenceBudget); - var logs = CreateTestProbe(); - Sys.EventStream.Subscribe(logs.Ref, typeof(LogEvent)); + var logs = SubscribeToLogs("deferring"); actor.Tell(new DispatchDeployment(deploymentB, RevB, CorrelationId.NewId())); - // Collect for long enough that a forward-spin would announce itself thousands of times. - var deferrals = 0; - var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(1.5); - while (DateTime.UtcNow < deadline) - { - if (logs.ReceiveOne(TimeSpan.FromMilliseconds(150)) is LogEvent e && - e.Message?.ToString()?.Contains("deferring", StringComparison.Ordinal) == true) - { - deferrals++; - } - } + // PRESENCE first, on the shared budget: the deferral must actually be ANNOUNCED. Counting + // only over a fixed window would let "the dispatch was silently dropped" score zero and + // then pass the ≤1 half below, which is the wrong bug to be blind to. + logs.DrainUntilAllSeen(PresenceBudget); + logs["deferring"].ShouldBeGreaterThanOrEqualTo(1); - deferrals.ShouldBe(1); + // ABSENCE second: over a window in which a forward-spin would announce itself thousands of + // times, no further deferral appears. + logs.Drain(SpinWindow); + logs["deferring"].ShouldBe(1); // Nothing was lost: B re-enters exactly once, after A's apply transitions back to Steady. fetcher.Release(); @@ -95,21 +97,274 @@ public sealed class DriverHostActorApplyingStashTests() } } - /// Gates the FIRST fetch on the test; later fetches complete immediately. - private sealed class GatedFetcher(byte[] bytes) : IDeploymentArtifactFetcher + /// + /// The SAME dispatch arriving twice mid-apply is stashed ONCE — the stash is bounded by + /// distinct deployments, not by how often the coordinator re-dispatches. + /// + /// + /// + /// Verified RED against the pinned Akka (1.5.62): with the dedup disabled, both copies are + /// stashed and the deferral is announced twice. That is the failure this pins — a + /// coordinator that re-dispatches on a timer would otherwise grow the stash with retries + /// rather than with work, and every extra copy is re-delivered on the way back to Steady + /// only to short-circuit as a duplicate. + /// + /// + /// It is NOT a crash today, and the review's suggestion that it might be was checked rather + /// than assumed. AbstractStash.Stash() is documented to throw + /// IllegalActorStateException when an envelope equals the last stashed one, and that + /// message string is in the shipped assembly — but a probe actor on this package stashed the + /// same instance twice, and two value-equal instances, with and without an explicit sender, + /// and never threw. The RED above is a count, not an exception, for exactly that reason. + /// + /// + [Fact] + public void TheSameDispatchArrivingTwiceMidApply_IsStashedOnce_AndAppliedOnce() { - private readonly TaskCompletionSource _gate = - new(TaskCreationOptions.RunContinuationsAsynchronously); + var db = NewInMemoryDbFactory(); + var fetcher = new GatedFetcher(ArtifactBytes()); + var deploymentA = DeploymentId.NewId(); + var deploymentB = DeploymentId.NewId(); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: new NullArtifactCache(), + fetchAndCacheMode: true, + artifactFetcher: fetcher)); + + try + { + actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId())); + AwaitAssert(() => fetcher.Calls.ShouldBe(1), duration: PresenceBudget); + + var logs = SubscribeToLogs("deferring", "already deferred"); + + // The IDENTICAL message object, twice — a record, so the two envelopes are value-equal. + var dispatchB = new DispatchDeployment(deploymentB, RevB, CorrelationId.NewId()); + actor.Tell(dispatchB); + actor.Tell(dispatchB); + + logs.DrainUntilAllSeen(PresenceBudget); + logs["deferring"].ShouldBe(1); + logs["already deferred"].ShouldBe(1); + + // No second stash, therefore no second re-delivery: exactly one Applied ack for B. + fetcher.Release(); + coordinator.ExpectMsg(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + coordinator.ExpectMsg(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + fetcher.Calls.ShouldBe(2); + } + finally + { + fetcher.Release(); + } + } + + /// + /// Two successive deferrals apply in arrival order, and the second one proves the dedup set is + /// CLEARED on the way back to Steady rather than accumulating across applies. + /// + /// + /// C is stashed by B's apply, i.e. after A's apply already unstashed B and cleared the set. A + /// dedup set that was never cleared would still hold nothing for C, so what this really pins is + /// the re-stash CYCLE working at all — plus the ordering the coordinator depends on: the node + /// must end on C's revision, not on B's. + /// + [Fact] + public void ASecondDeferralAppliesAfterTheFirst_AndTheNodeEndsOnTheLastRevision() + { + var db = NewInMemoryDbFactory(); + // Gate the first TWO fetches so C can arrive while B is applying. + var fetcher = new GatedFetcher(ArtifactBytes(), gatedCalls: 2); + var deploymentA = DeploymentId.NewId(); + var deploymentB = DeploymentId.NewId(); + var deploymentC = DeploymentId.NewId(); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: new NullArtifactCache(), + fetchAndCacheMode: true, + artifactFetcher: fetcher)); + + try + { + actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId())); + AwaitAssert(() => fetcher.Calls.ShouldBe(1), duration: PresenceBudget); + + actor.Tell(new DispatchDeployment(deploymentB, RevB, CorrelationId.NewId())); + + // A completes; B unstashes and starts its own (still gated) fetch. + fetcher.Release(1); + coordinator.ExpectMsg(TimeSpan.FromSeconds(10)).DeploymentId.ShouldBe(deploymentA); + AwaitAssert(() => fetcher.Calls.ShouldBe(2), duration: PresenceBudget); + + // C arrives mid-B — a fresh stash cycle, on a set that was cleared when B was released. + actor.Tell(new DispatchDeployment(deploymentC, RevC, CorrelationId.NewId())); + + fetcher.Release(2); + coordinator.ExpectMsg(TimeSpan.FromSeconds(10)).DeploymentId.ShouldBe(deploymentB); + coordinator.ExpectMsg(TimeSpan.FromSeconds(10)).DeploymentId.ShouldBe(deploymentC); + + actor.Tell(new GetDiagnostics(CorrelationId.NewId()), coordinator.Ref); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)) + .CurrentRevision.ShouldBe(RevC); + } + finally + { + fetcher.Release(); + } + } + + /// + /// A host wedged in Applying says so. The pre-stash forward-spin was accidentally the + /// only alarm for this — a flood of log lines nobody could miss — and removing it removed the + /// signal with it. The watchdog is the deliberate replacement. + /// + /// + /// The wedge delay is a constructor parameter (defaulting to + /// ), shortened here so the fire is + /// observable in a test's lifetime — the same seam every other interval on this actor uses. + /// Nothing is cancelled or failed by the fire, so the assertion is purely the Warning, and it + /// must NAME the wedged deployment or it is useless during an incident. + /// + [Fact] + public void AnApplyThatNeverCompletes_IsReportedAsWedged() + { + var db = NewInMemoryDbFactory(); + var fetcher = new GatedFetcher(ArtifactBytes()); + var deploymentA = DeploymentId.NewId(); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: new NullArtifactCache(), + fetchAndCacheMode: true, + artifactFetcher: fetcher, + applyWedgeTimeout: TimeSpan.FromMilliseconds(300))); + + try + { + EventFilter.Warning(contains: deploymentA.ToString()).ExpectOne(TimeSpan.FromSeconds(10), () => + actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId()))); + } + finally + { + fetcher.Release(); + } + } + + /// + /// A completed apply does NOT report a wedge, however short the delay — the timer is cancelled + /// on the way back to Steady, and a fire already in the mailbox drops at Debug. + /// + /// + /// The falsification control for the test above: a watchdog that fires unconditionally would + /// pass it and would then cry wedge after every ordinary deployment. + /// + [Fact] + public void ACompletedApply_NeverReportsAWedge() + { + var db = NewInMemoryDbFactory(); + var deploymentA = DeploymentId.NewId(); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: new NullArtifactCache(), + fetchAndCacheMode: true, + // Ungated: the fetch answers immediately. + artifactFetcher: new GatedFetcher(ArtifactBytes(), gatedCalls: 0), + applyWedgeTimeout: TimeSpan.FromMilliseconds(200))); + + var logs = SubscribeToLogs("appears WEDGED"); + actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + + // Well past the (200 ms) delay — a live timer would have fired several times over. + logs.Drain(SpinWindow); + logs["appears WEDGED"].ShouldBe(0); + } + + /// Subscribes a fresh counter to the EventStream's log events. + /// + /// Every fragment of interest must be declared up front, because draining is destructive: a + /// probe read for one fragment discards every other log event it passes over, so counting two + /// fragments with two independent drains would always score zero on the second. + /// + private LogCounter SubscribeToLogs(params string[] fragments) + { + var probe = CreateTestProbe(); + Sys.EventStream.Subscribe(probe.Ref, typeof(LogEvent)); + return new LogCounter(probe, fragments); + } + + /// Counts occurrences of a fixed set of log-message fragments off one EventStream probe. + private sealed class LogCounter(Akka.TestKit.TestProbe logs, string[] fragments) + { + private readonly Dictionary _counts = + fragments.ToDictionary(f => f, _ => 0, StringComparer.Ordinal); + + public int this[string fragment] => _counts[fragment]; + + /// Drains the probe for a fixed window, tallying every tracked fragment. + public void Drain(TimeSpan window) + { + var deadline = DateTime.UtcNow + window; + while (DateTime.UtcNow < deadline) + { + if (logs.ReceiveOne(TimeSpan.FromMilliseconds(100)) is LogEvent e && + e.Message?.ToString() is { } text) + { + foreach (var fragment in fragments) + if (text.Contains(fragment, StringComparison.Ordinal)) + _counts[fragment]++; + } + } + } + + /// Drains until every tracked fragment has been seen at least once, or the budget expires. + public void DrainUntilAllSeen(TimeSpan budget) + { + var deadline = DateTime.UtcNow + budget; + while (DateTime.UtcNow < deadline && _counts.Values.Any(n => n == 0)) + Drain(TimeSpan.FromMilliseconds(200)); + } + } + + /// + /// Gates the first gatedCalls fetches on the test; later fetches complete immediately. + /// + private sealed class GatedFetcher(byte[] bytes, int gatedCalls = 1) : IDeploymentArtifactFetcher + { + private readonly TaskCompletionSource[] _gates = Enumerable.Range(0, gatedCalls) + .Select(_ => new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)) + .ToArray(); + private int _calls; public int Calls => Volatile.Read(ref _calls); - public void Release() => _gate.TrySetResult(); + /// Releases every gate — also the test-teardown safety net. + public void Release() + { + foreach (var gate in _gates) gate.TrySetResult(); + } + + /// Releases the gate on the -th fetch (1-based). + public void Release(int call) => _gates[call - 1].TrySetResult(); public async Task FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct) { - if (Interlocked.Increment(ref _calls) == 1) - await _gate.Task.ConfigureAwait(false); + var call = Interlocked.Increment(ref _calls); + if (call <= _gates.Length) + await _gates[call - 1].Task.ConfigureAwait(false); return bytes; } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorAsyncCacheWriteTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorAsyncCacheWriteTests.cs index 64ffd060..a0857409 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorAsyncCacheWriteTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorAsyncCacheWriteTests.cs @@ -1,5 +1,6 @@ using System.Text.Json; using Akka.Actor; +using Akka.Event; using Microsoft.EntityFrameworkCore; using Shouldly; using Xunit; @@ -93,11 +94,20 @@ public sealed class DriverHostActorAsyncCacheWriteTests() localRoles: new HashSet { "driver" }, deploymentArtifactCache: new AsyncFailingArtifactCache())); - EventFilter.Warning(contains: "could not be cached").ExpectOne(TimeSpan.FromSeconds(5), () => - { - actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); - coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); - }); + // Matched on the EXCEPTION, not just the prose: the pipe-back carries the Exception itself and + // the handler logs it as the event's Cause. A SQLite failure is routinely a wrapped inner + // exception whose outer Message names neither the result code nor the failing frame, so a + // Reason-string-only carrier would have thrown that away before the log line was written. + // (Custom rather than EventFilter.Exception, which only ever matches ERROR-level events — + // a failed cache write is deliberately a Warning, since the apply itself succeeded.) + EventFilter.Custom(e => + e.Cause is InvalidOperationException && + e.Message?.ToString()?.Contains("could not be cached", StringComparison.Ordinal) == true) + .ExpectOne(TimeSpan.FromSeconds(5), () => + { + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + }); coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); }