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 d9767ea5..cdc806cf 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -54,7 +54,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers; /// AttributeValuePublished survives only as the message /// fabricates straight to the dependency mux; the host itself no longer receives it. /// -public sealed class DriverHostActor : ReceiveActor, IWithTimers +public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedStash { public const string DeploymentsTopic = "deployments"; public const string DeploymentAcksTopic = "deployment-acks"; @@ -148,6 +148,20 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers private readonly IActorRef? _scriptedAlarmHostOverride; private readonly ILoggingAdapter _log = Context.GetLogger(); + /// + /// Holds dispatches that arrive for a DIFFERENT deployment while one is being applied, until + /// the apply transitions back to Steady. Assigned by the ActorCell after construction — + /// it is inside the constructor, so nothing there may touch it. + /// + /// + /// The predecessor was Self.Forward(msg), which put the message back at the tail of this + /// actor's OWN mailbox: with nothing else queued, the host spun dequeue → log → enqueue for the + /// whole apply (a FetchAndCache fetch window is seconds of network I/O), burning a core + /// and emitting a log line per iteration — measured at ~570k lines in 1.5 s. The stash holds the + /// message OUTSIDE the mailbox, so the host simply idles until there is something real to do. + /// + public IStash Stash { get; set; } = null!; + /// 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 @@ -567,6 +581,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _scriptedAlarmHostOverride = scriptedAlarmHostOverride; // Default behavior is Steady — PreStart may flip to Stale or replay an orphan apply. + // The ONE Become(Steady) that is not BecomeSteady(): Stash is assigned by the ActorCell after + // the constructor returns, so it is null here — and nothing can have been stashed yet anyway. Become(Steady); } @@ -703,7 +719,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers if (_dbFactory is null) { _log.Info("DriverHost {Node}: no ConfigDb on the Direct bootstrap path; entering Steady (no revision)", _localNode); - Become(Steady); + BecomeSteady(); return; } @@ -722,7 +738,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers if (latest is null) { _log.Info("DriverHost {Node}: no prior deployments; entering Steady (no revision)", _localNode); - Become(Steady); + BecomeSteady(); return; } @@ -738,7 +754,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers case NodeDeploymentStatus.Applied: _currentRevision = revision; _log.Info("DriverHost {Node}: recovered Applied state at rev {Rev}", _localNode, revision); - Become(Steady); + BecomeSteady(); // The revision is recovered but the in-memory driver children + OPC UA address // space were lost on restart. Re-spawn + re-materialise + re-subscribe from the // applied deployment so a restarted/rebuilt node restores its served state instead @@ -749,13 +765,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _log.Warning("DriverHost {Node}: found orphan Applying row for deployment {Id}; replaying", _localNode, latest.DeploymentId); if (revision is not null) ApplyAndAck(new DeploymentId(latest.DeploymentId), revision.Value, CorrelationId.NewId()); - else Become(Steady); + else BecomeSteady(); break; case NodeDeploymentStatus.Failed: default: _log.Info("DriverHost {Node}: prior deployment {Id} failed; entering Steady at last known rev", _localNode, latest.DeploymentId); - Become(Steady); + BecomeSteady(); break; } } @@ -796,7 +812,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers "DriverHost {Node}: FetchAndCache boot — the LocalDb cache is empty; entering Steady " + "(no revision). The first dispatch will fetch this node's configuration from central.", _localNode); - Become(Steady); + BecomeSteady(); return; } @@ -804,7 +820,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers var revision = RevisionHash.Parse(cached.RevisionHash); _currentRevision = revision; - Become(Steady); + BecomeSteady(); // Restore drivers + address space + subscriptions from the cached bytes — no SQL read, no // re-ack (the deployment is already Applied). Same in-hand-bytes apply as the Direct-mode // cache boot, but this is the steady-state config source, not an outage fallback. @@ -821,7 +837,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers "DriverHost {Node}: FetchAndCache boot — failed to read the LocalDb pointer; entering " + "Steady (no revision). The next dispatch will fetch.", _localNode); - Become(Steady); + BecomeSteady(); } } @@ -876,7 +892,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // Stale would have started does not run here, so recovery rides on the next dispatch — // matching how a normally-booted node behaves. _currentRevision = revision; - Become(Steady); + BecomeSteady(); ApplyCachedArtifact(deploymentId, cached.Artifact); @@ -923,9 +939,36 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers PushDesiredSubscriptionsFromArtifact(deploymentId, blob); } + /// + /// Enters and releases anything the state stashed. + /// + /// + /// + /// A deferred dispatch must re-enter the mailbox exactly once and in order — + /// prepends the stash to the front of the mailbox in stash + /// order, which is both. Forgetting it on any one path silently strands that dispatch + /// forever, and the coordinator only learns about it at its apply deadline. + /// + /// + /// Every Become(Steady) goes through here — including the bootstrap and Stale-recovery + /// paths, where the stash is provably empty (only ever stashes, and + /// those paths run before or outside it). Unstashing an empty stash is a no-op, and routing + /// all of them through one helper is what makes "did we cover every transition?" answerable + /// by grep. The constructor is the one exception: is assigned by the + /// ActorCell only after construction returns. + /// + /// + private void BecomeSteady() + { + Become(Steady); + Stash.UnstashAll(); + } + private void Steady() { Receive(HandleDispatchFromSteady); + Receive(HandleArtifactCached); + Receive(HandleArtifactCacheFailed); Receive(HandleGetDiagnostics); Receive(ForwardBatchToMux); Receive(ForwardNativeAlarm); @@ -953,13 +996,22 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _localNode, msg.DeploymentId); return; } - _log.Info("DriverHost {Node}: dispatch for {Id} received while still applying {Cur}; deferring", + + // 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 + // than once per spin iteration. + _log.Debug("DriverHost {Node}: dispatch for {Id} received while still applying {Cur}; deferring", _localNode, msg.DeploymentId, _applyingDeploymentId); - Self.Forward(msg); // re-deliver after we transition back + Stash.Stash(); }); // The FetchAndCache fetch pipes its result back here (BeginFetchAndCacheApply ran in Steady, // then Become(Applying)); completing the apply transitions back to Steady. Receive(HandleFetchedForApply); + // 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); + Receive(HandleArtifactCacheFailed); Receive(HandleGetDiagnostics); Receive(ForwardBatchToMux); Receive(ForwardNativeAlarm); @@ -1614,7 +1666,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers SendAck(deploymentId, ApplyAckOutcome.Failed, reason, correlation); _log.Error("DriverHost {Node}: cannot apply {Id} — {Reason}", _localNode, deploymentId, reason); _applyingDeploymentId = null; - Become(Steady); + BecomeSteady(); return; } @@ -1706,7 +1758,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers parsed?.Dispose(); OtOpcUaTelemetry.DeploymentApplyDurationSec.Record(sw.Elapsed.TotalSeconds); _applyingDeploymentId = null; - Become(Steady); + BecomeSteady(); } } @@ -1805,7 +1857,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers parsed?.Dispose(); OtOpcUaTelemetry.DeploymentApplyDurationSec.Record(sw.Elapsed.TotalSeconds); _applyingDeploymentId = null; - Become(Steady); + BecomeSteady(); } } @@ -2008,10 +2060,31 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// space during the next outage — strictly worse than having no cache at all. /// /// - /// Runs synchronously on the actor thread. That matches every other DB call on this path - /// (all of which already block), and the alternative — piping the result back as a - /// self-message — would need a handler registered in Steady, Applying AND Stale or it - /// dead-letters after the Become(Steady) in the enclosing finally. + /// The store runs OFF the actor thread. It is hundreds of chunk inserts into SQLite; + /// blocking on it (as this did) parked the host for the duration, so it could not dequeue a + /// 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.) + /// + /// + /// The apply does not depend on the outcome — it never did, and the async shape keeps + /// it that way. By the time this runs the deployment is recorded Applied and an Applied ACK + /// has gone to the coordinator; a failure logs a Warning and changes nothing else. + /// + /// + /// Idempotence is unchanged. The skip-if-already-cached check lives inside + /// LocalDbDeploymentArtifactCache.StoreAsync, so it moves off-thread with the rest of + /// the store rather than being lost. + /// + /// + /// No boot-path race. The readers — and + /// — run in , before this actor has + /// processed a single message and therefore before any apply could have started a store. + /// The only concurrent writer a read can meet is the pair peer's replication, which the + /// cache already handles. /// /// /// The deployment whose artifact is being cached. @@ -2040,6 +2113,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // The node's ClusterId is carried inside the artifact (ClusterNode rows), not in config // or on IClusterRoleInfo. A single-cluster or unscoped artifact has none, so it shares // one sentinel key — the pair still converges on a single pointer row either way. + // + // Resolved HERE, synchronously, and not one step later: `parsed` is owned by the calling + // apply and DISPOSED in its finally, which runs long before a slow store completes. The + // async continuation below must therefore capture only these scalars — never `parsed`. var scope = parsed is not null ? parsed.ResolveClusterScope(_localNode.Value) : DeploymentArtifact.ResolveClusterScope(blob, _localNode.Value); @@ -2047,23 +2124,46 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers ? scope.ClusterId : SingleClusterCacheKey; + _log.Debug("DriverHost {Node}: caching deployment {Id} (rev {Rev}, cluster {ClusterId}, {Bytes} bytes)", + _localNode, deploymentId, revision, clusterId, blob.Length); + _deploymentArtifactCache .StoreAsync(clusterId, deploymentId.ToString(), revision.Value, blob) - .GetAwaiter() - .GetResult(); - - _log.Debug("DriverHost {Node}: cached deployment {Id} (rev {Rev}, cluster {ClusterId}, {Bytes} bytes)", - _localNode, deploymentId, revision, clusterId, blob.Length); + .PipeTo( + Self, + success: () => new ArtifactCached(deploymentId), + failure: ex => new ArtifactCacheFailed(deploymentId, ex.Message)); } catch (Exception ex) { - _log.Error(ex, - "DriverHost {Node}: failed to cache applied deployment {Id}. The apply itself succeeded " + - "and is unaffected; this node simply has no local fallback for that revision.", - _localNode, deploymentId); + // 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)); } } + /// The node-local artifact cache accepted this deployment's bytes. + /// + /// Piped back from . Carries the id only — deliberately no + /// ParsedDeploymentArtifact and no other apply-owned state, since the apply that started + /// the store has long since disposed its parse by the time this lands. + /// + 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); + + 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( + "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); + /// /// Restore the served state for an already-applied deployment after a process restart. /// recovers from NodeDeploymentState, @@ -2600,7 +2700,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers if (_fetchAndCacheMode || _dbFactory is null) { Timers.Cancel("retry-db"); - Become(Steady); + BecomeSteady(); return; } @@ -2623,7 +2723,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers UpsertNodeDeploymentState(new DeploymentId(latestSealed.DeploymentId), NodeDeploymentStatus.Applied, failureReason: null); } - Become(Steady); + BecomeSteady(); } catch (Exception ex) { 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 new file mode 100644 index 00000000..d7db6d67 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorApplyingStashTests.cs @@ -0,0 +1,130 @@ +using System.Text.Json; +using Akka.Actor; +using Akka.Event; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Surgical-deploys Task 7 Part B — a dispatch that arrives while an apply is in flight is +/// STASHED, not re-forwarded to Self. +/// +/// +/// +/// Self.Forward put the message back at the tail of the actor's OWN mailbox, so with +/// nothing else to do the host spun dequeue → log → enqueue for the entire fetch window — +/// thousands of iterations and thousands of log lines for one deferred deployment, burning a +/// core and starving nothing but itself of useful work. The stash holds it outside the mailbox +/// until the apply finishes. +/// +/// +/// Runs at DEBUG because the assertion IS a log line: a spin is only distinguishable +/// from a single deferral by how many times it is announced. A log event below +/// akka.loglevel never reaches the EventStream at all, so at the default WARNING this +/// test would count zero either way and pass vacuously. +/// +/// +public sealed class DriverHostActorApplyingStashTests() + : RuntimeActorTestBase(AkkaTestHoconAtLevel("DEBUG")) +{ + 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)); + + /// Minimal parseable artifact — no drivers. + private static byte[] ArtifactBytes() => + JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty() }); + + [Fact] + public void ADispatchArrivingMidApply_IsDeferredOnce_AndAppliedAfterTheApplyCompletes() + { + 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 + { + // A is in flight: the fetch has started and its task will not complete until released. + 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)); + + 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++; + } + } + + deferrals.ShouldBe(1); + + // Nothing was lost: B re-enters exactly once, after A's apply transitions back to Steady. + fetcher.Release(); + coordinator.ExpectMsg(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + coordinator.ExpectMsg(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + fetcher.Calls.ShouldBe(2); + } + finally + { + // Never leave the fetch parked: a blocked apply would hang ActorSystem shutdown. + fetcher.Release(); + } + } + + /// Gates the FIRST fetch on the test; later fetches complete immediately. + private sealed class GatedFetcher(byte[] bytes) : IDeploymentArtifactFetcher + { + private readonly TaskCompletionSource _gate = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _calls; + + public int Calls => Volatile.Read(ref _calls); + + public void Release() => _gate.TrySetResult(); + + public async Task FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct) + { + if (Interlocked.Increment(ref _calls) == 1) + await _gate.Task.ConfigureAwait(false); + return bytes; + } + } + + /// Accepts every store; this class asserts on dispatch handling, not on caching. + private sealed class NullArtifactCache : IDeploymentArtifactCache + { + public Task StoreAsync(string clusterId, string deploymentId, string revisionHash, + byte[] artifact, CancellationToken ct = default) + => Task.CompletedTask; + + public Task GetCurrentAsync(string clusterId, CancellationToken ct = default) + => Task.FromResult(null); + + public Task GetCurrentUnkeyedAsync(CancellationToken ct = default) + => Task.FromResult(null); + } +} 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 new file mode 100644 index 00000000..64ffd060 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorAsyncCacheWriteTests.cs @@ -0,0 +1,175 @@ +using System.Text.Json; +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +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.Configuration; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// Surgical-deploys Task 7 Part A — the artifact-cache write is I/O (hundreds of chunk inserts +/// into LocalDb) and must not run on the actor thread. +/// +/// +/// +/// The ACK was never the thing at risk: ApplyAndAck sends it BEFORE it caches, so a +/// blocking store still acked promptly. What it blocked was everything after — the host +/// could not dequeue a driver value burst, a diagnostics request or the next dispatch until +/// SQLite was done. So the load-bearing assertion here is that the actor answers an unrelated +/// message while the store is still in flight, not that the ACK arrived. +/// +/// +/// Runs at INFO so the completion self-message's log line is observable — the pipe-back +/// is private, and its visible effect is that line. +/// +/// +public sealed class DriverHostActorAsyncCacheWriteTests() + : RuntimeActorTestBase(AkkaTestHoconAtLevel("INFO")) +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + + [Fact] + public void AStillRunningCacheStore_DoesNotBlockTheHostMailbox() + { + var db = NewInMemoryDbFactory(); + var cache = new GatedArtifactCache(); + var deploymentId = SeedDeployment(db, RevA); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: cache)); + + try + { + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + + // The store was entered and has NOT completed — the gate is still closed. + AwaitAssert(() => cache.StoresEntered.ShouldBe(1), duration: PresenceBudget); + cache.StoresCompleted.ShouldBe(0); + + // THE assertion: with the store still in flight the host is still processing its mailbox. + // A blocking `.GetAwaiter().GetResult()` parks the actor thread here and this times out. + actor.Tell(new GetDiagnostics(CorrelationId.NewId()), coordinator.Ref); + coordinator.ExpectMsg(TimeSpan.FromSeconds(3)); + + // Completion is observed rather than fire-and-forgotten: the store pipes back to Self and + // the handler logs it, so an operator can tell a cached apply from an uncached one. + EventFilter.Info(contains: "cached applied deployment") + .ExpectOne(TimeSpan.FromSeconds(5), () => cache.Release()); + } + finally + { + // Never leave the fake gated: a parked store would hang ActorSystem shutdown. + cache.Release(); + } + } + + [Fact] + public void AFailingCacheStore_WarnsAndLeavesTheApplyApplied() + { + // The cache is a boot fallback, not part of the deployment protocol. A failed store must + // surface as a Warning and nothing else — emphatically NOT as a second, contradictory Failed + // ACK for a deployment the fleet already believes is live. + var db = NewInMemoryDbFactory(); + var deploymentId = SeedDeployment(db, RevA); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + 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); + }); + + coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + /// Seeds a sealed, parseable deployment and returns its id. + private static DeploymentId SeedDeployment(IDbContextFactory db, RevisionHash rev) + { + var artifact = JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty() }); + var id = DeploymentId.NewId(); + using var ctx = db.CreateDbContext(); + ctx.Deployments.Add(new Deployment + { + DeploymentId = id.Value, + RevisionHash = rev.Value, + Status = DeploymentStatus.Sealed, + CreatedBy = "test", + SealedAtUtc = DateTime.UtcNow, + ArtifactBlob = artifact, + }); + ctx.SaveChanges(); + return id; + } + + /// + /// A store that never completes until the test releases it — the fake shape of a slow LocalDb + /// write. It must not complete synchronously, or it would prove nothing about the mailbox. + /// + private sealed class GatedArtifactCache : IDeploymentArtifactCache + { + private readonly TaskCompletionSource _gate = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _entered; + private int _completed; + + public int StoresEntered => Volatile.Read(ref _entered); + public int StoresCompleted => Volatile.Read(ref _completed); + + public void Release() => _gate.TrySetResult(); + + public async Task StoreAsync(string clusterId, string deploymentId, string revisionHash, + byte[] artifact, CancellationToken ct = default) + { + Interlocked.Increment(ref _entered); + await _gate.Task.ConfigureAwait(false); + Interlocked.Increment(ref _completed); + } + + public Task GetCurrentAsync(string clusterId, CancellationToken ct = default) + => Task.FromResult(null); + + public Task GetCurrentUnkeyedAsync(CancellationToken ct = default) + => Task.FromResult(null); + } + + /// + /// Fails the store ASYNCHRONOUSLY (a faulted task), which is how a real LocalDb write fails — + /// as distinct from DriverHostActorArtifactCacheTests's synchronously-throwing fake, + /// which exercises the other half of the guard. + /// + private sealed class AsyncFailingArtifactCache : IDeploymentArtifactCache + { + public async Task StoreAsync(string clusterId, string deploymentId, string revisionHash, + byte[] artifact, CancellationToken ct = default) + { + await Task.Yield(); + throw new InvalidOperationException("disk full"); + } + + public Task GetCurrentAsync(string clusterId, CancellationToken ct = default) + => Task.FromResult(null); + + public Task GetCurrentUnkeyedAsync(CancellationToken ct = default) + => Task.FromResult(null); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/RuntimeActorTestBase.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/RuntimeActorTestBase.cs index c547321f..d6f8fafe 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/RuntimeActorTestBase.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/RuntimeActorTestBase.cs @@ -55,8 +55,30 @@ akka { } }"; + /// + /// The same single-node HOCON at a caller-chosen loglevel. + /// + /// + /// A log event below akka.loglevel is never published to the EventStream at all, so + /// neither EventFilter nor an EventStream-subscribed probe can see it. A test whose + /// subject IS a log line (e.g. "this Debug line must be emitted exactly once, not in a spin") + /// therefore has to raise the level for its own ActorSystem; the default stays + /// WARNING so the other 40-odd classes keep their quiet output. + /// + /// An Akka log level — DEBUG, INFO, WARNING, ERROR. + /// The base test HOCON with its loglevel replaced. + protected static string AkkaTestHoconAtLevel(string level) => + AkkaTestHocon.Replace(@"loglevel = ""WARNING""", $@"loglevel = ""{level}""", StringComparison.Ordinal); + /// Initializes a new instance of the RuntimeActorTestBase class with Akka test setup. - protected RuntimeActorTestBase() : base(AkkaTestHocon) + protected RuntimeActorTestBase() : this(AkkaTestHocon) + { + } + + /// Initializes a new instance of the RuntimeActorTestBase class with caller-supplied HOCON. + /// The Akka configuration for this class's ActorSystem — usually + /// . + protected RuntimeActorTestBase(string hocon) : base(hocon) { var cluster = Akka.Cluster.Cluster.Get(Sys); cluster.Join(cluster.SelfAddress);