From deaa31479bacd0f254f8411976b5eb3fe8b0c509 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 10:36:34 -0400 Subject: [PATCH] perf(runtime): parse + plan a deploy off the pinned OPC UA thread (B6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review 02 P5. OpcUaPublishActor ran the whole apply — DeploymentArtifact .Open, ParseComposition, AddressSpacePlanner.Compute and the sink materialisation — on the opcua-synchronized-dispatcher, so a large artifact serialised every live value batch (and, because the node manager lock is the SDK's, every client Publish) behind the parse. StartRebuild now runs Open + ParseDataPlaneAuthority + ParseComposition + Compute in Task.Run and pipes a PlanReady (or PlanFailed) back to Self; only ApplyPlan — the sink passes — runs on the actor thread. Rebuilds are applied strictly in arrival order (a plan is a diff against _lastApplied, which only the apply advances): one in flight, later ones queue in _pendingRebuilds while value/status batches keep flowing. The apply is closed (CompleteAddressSpaceApply) and the RebuildCompleted reply sent on every path, PlanFailed included. The bytes-less ConfigDb fallback (LoadArtifact / LoadLatestArtifact) is deleted: it had no production producer and would have blocked the pinned thread on central SQL; the dbFactory Props/ctor parameter is kept for wiring compatibility (F4). Tests: OpcUaPublishActorOffThreadPlanTests — a value batch sent right after a 50k-tag rebuild lands before that rebuild's first materialise call (positive control: all 50k tags still materialise); back-to-back rebuilds apply in arrival order (the second diffs against the first); unreadable bytes answer degraded with nothing torn down. The rebuild / apply-failure suites now hand the actor the artifact bytes every production producer carries (SealArtifact records them; Rebuild(dep)). Refs #732 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../OpcUa/OpcUaPublishActor.cs | 328 +++++++++--------- .../OpcUaPublishActorApplyFailureTests.cs | 14 +- .../OpcUaPublishActorOffThreadPlanTests.cs | 128 +++++++ .../OpcUa/OpcUaPublishActorRebuildTests.cs | 145 +++----- 4 files changed, 360 insertions(+), 255 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorOffThreadPlanTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs index 207585c9..cbe96919 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -86,19 +86,17 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// The namespace realm the condition lives in ( for native). public sealed record AlarmQualityUpdate(string AlarmNodeId, uint StatusCode, DateTime TimestampUtc, AddressSpaceRealm Realm); /// - /// Triggers an address-space rebuild. is the deployment - /// just applied by the host; the rebuild loads THAT artifact so materialisation matches the - /// applied config + the SubscribeBulk pass. It is null only for legacy/dev callers, which - /// fall back to the latest sealed deployment (lags a not-yet-sealed apply by one revision). + /// Triggers an address-space rebuild of the artifact carried in . + /// is the deployment just applied by the host (echoed in the reply and + /// the logs). Review 02 P5: the ConfigDb fallback for a message carrying no bytes is gone — it had no + /// production producer and would have blocked the pinned OPC UA thread on central SQL — so a message + /// without bytes is "no answer" and holds the served address space. /// /// Correlation id for tracing this rebuild. - /// The applied deployment whose artifact to materialise, or null for the latest-sealed fallback. + /// The applied deployment whose artifact this is, or null for a legacy caller. /// - /// The artifact bytes already in hand, used INSTEAD of loading them from the ConfigDb. This is - /// what makes boot-from-cache actually serve its address space: on a central-SQL outage the - /// host has the cached blob but alone would drive a ConfigDb read - /// that cannot succeed, leaving the rebuild a no-op and clients browsing an empty server. Null - /// on the normal path, where loading from the ConfigDb by id is correct. + /// The artifact bytes. Every producer (the deploy path, boot-from-cache, FetchAndCache) has them in + /// hand; null or empty is treated as no answer. /// /// /// when , the actor replies to the sender with a @@ -128,11 +126,29 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers public sealed record ServiceLevelChanged(byte ServiceLevel); + /// + /// Review 02 P5 — a rebuild whose artifact parse, cluster-scoped composition and plan were computed + /// OFF the pinned OPC UA thread (a Task.Run piped back to Self), ready for the sink passes, + /// which are the only part that must run on this actor's thread. Carries the and + /// the original so the reply reaches whoever asked. + /// + private sealed record PlanReady( + RebuildAddressSpace Request, IActorRef Requester, byte[] Artifact, DataPlaneAuthorityProjection? Authority, + AddressSpaceComposition Composition, AddressSpacePlan Plan, IReadOnlyList Inconsistencies); + + /// The off-thread parse / plan step threw (unreadable bytes, a composer fault). + private sealed record PlanFailed(RebuildAddressSpace Request, IActorRef Requester, Exception Error); + + /// True while a rebuild's parse / plan step is running off-thread. Rebuilds are applied strictly in + /// arrival order — the plan is a diff against , which only the apply advances — so a + /// rebuild arriving meanwhile waits in ; value and status batches do not. + private bool _planInFlight; + private readonly Queue<(RebuildAddressSpace Request, IActorRef Requester)> _pendingRebuilds = new(); + private readonly IOpcUaAddressSpaceSink _sink; private readonly IServiceLevelPublisher _serviceLevel; private readonly bool _subscribeRedundancyTopic; private readonly NodeId? _localNode; - private readonly IDbContextFactory? _dbFactory; private readonly AddressSpaceApplier? _applier; /// Receives the API keys + ACL grants carried by each applied artifact, so /// OPC UA session activation can authenticate. Null when nothing consumes it (tests, and any node @@ -184,9 +200,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// Production Props — pins the OPC UA dispatcher + subscribes to the /// redundancy-state DPS topic so cluster transitions drive the local ServiceLevel - /// publish path. When + are supplied, - /// reads the latest deployment artifact + drives the - /// applier through the sink. When is supplied the local + /// publish path. When is supplied, + /// diffs + applies the artifact bytes it carries through the sink ( is + /// accepted for wiring compatibility only — review 02 P5 removed the ConfigDb fallback). When is supplied the local /// ServiceLevel is computed via from real DB-health + /// staleness + role-leader inputs; otherwise the legacy role-only switch is used. /// The OPC UA address space sink. @@ -335,7 +351,9 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers _serviceLevel = serviceLevel; _subscribeRedundancyTopic = subscribeRedundancyTopic; _localNode = localNode; - _dbFactory = dbFactory; + // dbFactory is accepted for wiring compatibility only (review 02 P5): the bytes-less ConfigDb read the + // rebuild used to fall back to is gone — every producer carries the artifact bytes in the message. + _ = dbFactory; _applier = applier; _dbHealthProbe = dbHealthProbe; _dbLess = dbLess; @@ -350,6 +368,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers Receive(HandleAlarmUpdate); Receive(HandleAlarmQualityUpdate); Receive(HandleRebuild); + Receive(HandlePlanReady); + Receive(HandlePlanFailed); Receive(HandleServiceLevelChanged); Receive(HandleRedundancyStateChanged); Receive(HandleDbHealthStatus); @@ -480,114 +500,147 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers } } - private void HandleRebuild(RebuildAddressSpace msg) + private void HandleRebuild(RebuildAddressSpace msg) => StartRebuild(msg, Sender); + + /// + /// The entry half of a rebuild, on the actor thread. The raw-sink fallback (no applier) and the + /// no-artifact case complete synchronously; otherwise the artifact parse, the cluster-scoped composition + /// and the plan — the O(artifact) work that used to serialise every live value batch and (because the + /// node manager's lock is the SDK's) every client Publish behind it — run on the thread pool and come + /// back as a . Only the sink materialisation () runs here. + /// + private void StartRebuild(RebuildAddressSpace msg, IActorRef requester) { - // Captured up front: the pipeline below is synchronous on this actor's thread today, but the reply - // must reach whoever asked even if a future edit introduces an await into it. - var requester = Sender; - (bool Degraded, string? Reason) outcome; - try - { - outcome = HandleRebuildCore(msg); - } - finally - { - // close the apply on EVERY exit, early returns and a throwing pass included: the node - // manager reports here the native conditions a rebuild carried over (after the notifier wiring the - // passes did) and discards any captured state no materialise claimed. Skipping it on a throw would - // leave that state for a later surgical re-add of the same id to resurrect. No-op when nothing is - // pending; never throws out of the handler. - try { _sink.CompleteAddressSpaceApply(); } - catch (Exception ex) - { - _log.Error(ex, "OpcUaPublish: sink.CompleteAddressSpaceApply threw (correlation={Correlation})", msg.Correlation); - } - } - - // the outcome goes out only after the apply (including its completion step) is done. - if (msg.ReportOutcome) - requester.Tell(new RebuildCompleted(msg.Correlation, msg.DeploymentId, outcome.Degraded, outcome.Reason)); - } - - /// The rebuild pipeline. Returns whether the apply was degraded, and why. - private (bool Degraded, string? Reason) HandleRebuildCore(RebuildAddressSpace msg) - { - using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan(); - span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString()); - - // Two modes. With an applier wired, do a REAL diff-and-apply pass — driven either by the - // artifact bytes already in hand (a per-cluster-mesh driver-only / FetchAndCache node, whose - // ConfigDb is admin-only now so _dbFactory is null, hands its fetched/cached bytes in - // msg.Artifact) or, when no bytes are carried, by loading the artifact from the ConfigDb (the - // Direct path, which requires _dbFactory). Only a node with NO applier at all (the F10b/dev - // seam) falls back to the raw-sink rebuild — and that fallback WIPES the served address space - // without re-materialising, so it must NEVER be reached by a real driver node that has an - // applier but simply has no ConfigDb (that node diff-and-applies from its in-hand bytes below). + // Only a node with NO applier at all (the F10b/dev seam) falls back to the raw-sink rebuild — and that + // fallback WIPES the served address space without re-materialising, so it must NEVER be reached by a + // real driver node that has an applier (that node diff-and-applies from its in-hand bytes below). if (_applier is null) { + (bool Degraded, string? Reason) outcome; try { _sink.RebuildAddressSpace(); OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair("kind", "rebuild")); + outcome = (false, null); } catch (Exception ex) { - _log.Error(ex, "OpcUaPublish: sink.RebuildAddressSpace threw (correlation={Correlation})", - msg.Correlation); - return (true, $"address-space rebuild threw: {ex.Message}"); + _log.Error(ex, "OpcUaPublish: sink.RebuildAddressSpace threw (correlation={Correlation})", msg.Correlation); + outcome = (true, $"address-space rebuild threw: {ex.Message}"); } - return (false, null); + + FinishRebuild(msg, requester, outcome); + return; } + // An artifact we could not obtain is NOT an empty configuration. Parsing zero bytes yields an empty + // composition, which the planner diffs against the live one as a PureRemove and the applier then + // faithfully tears down. Nothing legitimate produces a zero-length blob (an operator who really deletes + // everything still deploys a JSON document with empty arrays), so the only honest reading of "no bytes" + // is "no answer" — hold the last-known-good address space and let the next deploy (or the next boot) + // supply a real one. Leaving _lastApplied intact means the retry diffs against what is materialised. + if (msg.Artifact is null or { Length: 0 }) + { + if (_hasAppliedComposition) + _log.Warning( + "OpcUaPublish: no usable artifact for deployment {Id} (correlation={Correlation}) — KEEPING the currently served address space rather than tearing it down", + msg.DeploymentId, msg.Correlation); + else + _log.Debug( + "OpcUaPublish: no artifact to materialise yet (deployment={Id}, correlation={Correlation})", + msg.DeploymentId, msg.Correlation); + FinishRebuild(msg, requester, (true, "no usable artifact reached the address-space apply; the previously served address space was kept")); + return; + } + + if (_planInFlight) + { + _pendingRebuilds.Enqueue((msg, requester)); + return; + } + + _planInFlight = true; + var artifact = msg.Artifact; + var baseComposition = _lastApplied; + var nodeId = _localNode?.Value; + var wantAuthority = _dataPlaneAuthority is not null; + Task.Run(() => + { + // Open() throws on bytes that are not valid JSON, where the static wrappers swallowed to an EMPTY + // result — and an empty composition is not a benign no-op here (see the no-bytes guard above). The + // throw lands in PlanFailed: logged at Error, _lastApplied untouched, last-known-good still served. + using var parsed = DeploymentArtifact.Open(artifact); + var authority = wantAuthority ? parsed.ParseDataPlaneAuthority(nodeId ?? string.Empty) : null; + var inconsistencies = new List(); + var composition = nodeId is not null + ? parsed.ParseComposition(nodeId, inconsistencies.Add) + : parsed.ParseComposition(); + var plan = AddressSpacePlanner.Compute(baseComposition, composition); + return new PlanReady(msg, requester, artifact, authority, composition, plan, inconsistencies); + }) + .PipeTo(Self, failure: ex => new PlanFailed(msg, requester, ex)); + } + + private void HandlePlanReady(PlanReady ready) + { + (bool Degraded, string? Reason) outcome; + try + { + outcome = ApplyPlan(ready); + } + finally + { + _planInFlight = false; + } + + FinishRebuild(ready.Request, ready.Requester, outcome); + StartNextPendingRebuild(); + } + + private void HandlePlanFailed(PlanFailed failed) + { + _planInFlight = false; + _log.Error(failed.Error, "OpcUaPublish: rebuild pipeline threw (correlation={Correlation})", failed.Request.Correlation); + FinishRebuild(failed.Request, failed.Requester, (true, $"address-space apply threw: {failed.Error.Message}")); + StartNextPendingRebuild(); + } + + private void StartNextPendingRebuild() + { + if (_pendingRebuilds.TryDequeue(out var next)) StartRebuild(next.Request, next.Requester); + } + + /// + /// Closes an apply on EVERY exit — early returns and a throwing pass included: the node manager reports + /// here the native conditions a rebuild carried over (after the notifier wiring the passes did) and + /// discards any captured state no materialise claimed. Skipping it on a throw would leave that state for a + /// later surgical re-add of the same id to resurrect. No-op when nothing is pending; never throws. The + /// outcome goes out only after the apply (including its completion step) is done. + /// + private void FinishRebuild(RebuildAddressSpace msg, IActorRef requester, (bool Degraded, string? Reason) outcome) + { + try { _sink.CompleteAddressSpaceApply(); } + catch (Exception ex) + { + _log.Error(ex, "OpcUaPublish: sink.CompleteAddressSpaceApply threw (correlation={Correlation})", msg.Correlation); + } + + if (msg.ReportOutcome) + requester.Tell(new RebuildCompleted(msg.Correlation, msg.DeploymentId, outcome.Degraded, outcome.Reason)); + } + + /// The actor-thread half of a rebuild: install the artifact's authority, apply the plan through the + /// sink passes, announce. Returns whether the apply was degraded, and why. + private (bool Degraded, string? Reason) ApplyPlan(PlanReady ready) + { + var msg = ready.Request; + using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan(); + span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString()); + try { - // An in-hand artifact (boot-from-cache / FetchAndCache) wins: the host already has the - // cached/fetched bytes, and the ConfigDb read the DeploymentId path would do is exactly what - // is unreachable during the outage this exists to survive — and is simply absent on a - // driver-only node with no ConfigDb. Otherwise, when a ConfigDb is present (Direct mode), - // prefer the artifact of the deployment the host just applied — at apply time it is not yet - // Sealed, so LoadLatestArtifact would return the PREVIOUS revision and materialise a stale - // composition (variables that don't match the SubscribeBulk refs); fall back to latest-sealed - // only for legacy callers that carry neither. With NO bytes AND no ConfigDb there is nothing - // to load, so yield Array.Empty() — the guard below reads that as "no answer" and - // abandons the rebuild, holding last-known-good rather than wiping the address space. - var artifact = msg.Artifact - ?? (_dbFactory is not null - ? (msg.DeploymentId is { } depId ? LoadArtifact(depId) : LoadLatestArtifact()) - : Array.Empty()); - - // An artifact we could not obtain is NOT an empty configuration. Parsing zero - // bytes yields an empty composition, which the planner diffs against the live one as a - // PureRemove and the applier then faithfully tears down: a transient ConfigDb blip empties the - // served address space. Nothing legitimate produces a zero-length blob (an operator who really - // deletes everything still deploys a JSON document with empty arrays), so the only honest - // reading of "no bytes" is "no answer" — hold the last-known-good address space and let the - // next deploy (or the next boot) supply a real one. Returning here also leaves _lastApplied - // intact, so the retry diffs against what is actually materialised. - if (artifact is null or { Length: 0 }) - { - if (_hasAppliedComposition) - _log.Warning( - "OpcUaPublish: no usable artifact for deployment {Id} (correlation={Correlation}) — KEEPING the currently served address space rather than tearing it down", - msg.DeploymentId, msg.Correlation); - else - _log.Debug( - "OpcUaPublish: no artifact to materialise yet (deployment={Id}, correlation={Correlation})", - msg.DeploymentId, msg.Correlation); - return (true, "no usable artifact reached the address-space apply; the previously served address space was kept"); - } - - // Parse the artifact ONCE for this rebuild. Every consumer below (data-plane authority, then the - // composition) reads this one JsonDocument instead of rebuilding it per question — the rebuild - // used to pay a full parse-and-throw-away per static call. Disposed when the handler returns. - // - // Open() throws on bytes that are not valid JSON, where the static wrappers swallowed to an - // EMPTY result — and an empty composition is not a benign no-op here: the planner diffs it as a - // PureRemove and the applier tears the served address space down. So the throw is the honest - // outcome, and it lands in this method's catch below: logged at Error, _lastApplied untouched, - // last-known-good still served. Same "no answer ≠ empty configuration" rule as the guard - // immediately above, applied to bytes that arrived unreadable rather than absent. - using var parsed = DeploymentArtifact.Open(artifact); + foreach (var inconsistency in ready.Inconsistencies) + _log.Warning("OpcUaPublish {Node}: cross-cluster binding — {Message}", _localNode, inconsistency); // install this artifact's API keys + ACL grants BEFORE the address-space work // below, and deliberately ahead of the `plan.IsEmpty` early-return: an artifact that revokes a @@ -595,13 +648,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers // feeding this after that return would leave a revoked key authenticating until some unrelated // structural change happened to redeploy. Authorization and topology change on their own // schedules. - FeedDataPlaneAuthority(parsed, artifact); + if (ready.Authority is not null) FeedDataPlaneAuthority(ready.Authority, ready.Artifact); - var composition = _localNode is { } ln - ? parsed.ParseComposition(ln.Value, - inconsistency => _log.Warning("OpcUaPublish {Node}: cross-cluster binding — {Message}", ln, inconsistency)) - : parsed.ParseComposition(); - var plan = AddressSpacePlanner.Compute(_lastApplied, composition); + var composition = ready.Composition; + var plan = ready.Plan; if (plan.IsEmpty) { @@ -610,7 +660,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers // The composition is unchanged but the last apply of it was a degraded // rebuild: re-ensure every node (each Ensure* no-ops on a node that stands) so the // re-deploy actually repairs what that rebuild could not place. - var repairFailed = _applier.MaterialiseHierarchy(composition) + var repairFailed = _applier!.MaterialiseHierarchy(composition) + MaterialiseNodes(plan, composition, fullComposition: true); if (repairFailed > 0) { @@ -632,7 +682,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers return (false, null); } - var outcome = _applier.Apply(plan); + var outcome = _applier!.Apply(plan); // Sum swallowed per-node materialise failures across every pass together with Apply's own // removal-pass tally (archreview 01/S-1). A degraded apply used to vanish into per-node @@ -867,17 +917,15 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// the safe direction for availability but means a revocation carried by an artifact this node /// could not decode has not taken effect. It is logged at Error for exactly that reason. /// - /// The rebuild's single parse of the artifact, read for the keys + grants. + /// The keys + grants decoded (off-thread) from the rebuild's single parse of the artifact. /// The RAW artifact bytes — hashed for the revision id, which must be /// SHA-256 of the exact blob (see below); never re-parsed here. - private void FeedDataPlaneAuthority(ParsedDeploymentArtifact parsed, byte[] artifactBytes) + private void FeedDataPlaneAuthority(DataPlaneAuthorityProjection projection, byte[] artifactBytes) { if (_dataPlaneAuthority is null) return; try { - var nodeId = _localNode?.Value ?? string.Empty; - var projection = parsed.ParseDataPlaneAuthority(nodeId); // The revision hash is SHA-256 of the artifact bytes — the same derivation ConfigComposer uses, // which is what lets a node recognise a re-applied revision without the hash riding the wire. // Hashed off the RAW bytes for exactly that reason: a re-serialisation of the parsed document @@ -922,46 +970,6 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers } } - /// Read a specific deployment's artifact blob from ConfigDb (the one just applied, - /// which may not be Sealed yet). Empty array on any failure — treats that as - /// "no answer" and abandons the rebuild, leaving the served address space untouched. - private byte[] LoadArtifact(DeploymentId deploymentId) - { - try - { - using var db = _dbFactory!.CreateDbContext(); - return db.Deployments.AsNoTracking() - .Where(d => d.DeploymentId == deploymentId.Value) - .Select(d => d.ArtifactBlob) - .FirstOrDefault() ?? Array.Empty(); - } - catch (Exception ex) - { - _log.Warning(ex, "OpcUaPublish: failed to load artifact for deployment {Id}; the rebuild is abandoned", deploymentId); - return Array.Empty(); - } - } - - /// Read the most recent Sealed deployment's artifact blob from ConfigDb. - /// Empty array on any failure — see for how that is handled. - private byte[] LoadLatestArtifact() - { - try - { - using var db = _dbFactory!.CreateDbContext(); - return db.Deployments.AsNoTracking() - .Where(d => d.Status == Configuration.Enums.DeploymentStatus.Sealed) - .OrderByDescending(d => d.SealedAtUtc) - .Select(d => d.ArtifactBlob) - .FirstOrDefault() ?? Array.Empty(); - } - catch (Exception ex) - { - _log.Warning(ex, "OpcUaPublish: failed to load latest deployment artifact; the rebuild is abandoned"); - return Array.Empty(); - } - } - private void HandleServiceLevelChanged(ServiceLevelChanged msg) { // Always publish the FIRST computed level, even if it equals the byte-default 0. Otherwise a diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs index 8d1fd67e..6c07ca0f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs @@ -50,13 +50,13 @@ public sealed class OpcUaPublishActorApplyFailureTests : RuntimeActorTestBase var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); // First deploy: PureAdd — no rebuild, so the throwing sink is not hit yet. - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); Thread.Sleep(200); recorder.Total.ShouldBe(0); // Second deploy: eq-1 RENAMED ⇒ ChangedEquipment ⇒ Rebuild ⇒ the sink's RebuildAddressSpace throws. var dep2 = SeedEquipmentDeployment(db, ("eq-1", "eq-1-renamed")); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); // A PRESENCE budget, so it is an upper bound before giving up rather than a wait: the helper polls // and returns the instant the meter fires, so a generous value costs nothing on the happy path and @@ -78,10 +78,10 @@ public sealed class OpcUaPublishActorApplyFailureTests : RuntimeActorTestBase var db = NewInMemoryDbFactory(); var sink = new NoopSink(); var applier = new AddressSpaceApplier(sink, NullLogger.Instance); - SeedEquipmentDeployment(db, ("eq-1", "eq-1")); + var dep = SeedEquipmentDeployment(db, ("eq-1", "eq-1")); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + actor.Tell(Rebuild(dep)); // Give the actor time to process, then confirm the failed meter never fired. Thread.Sleep(400); @@ -100,6 +100,11 @@ public sealed class OpcUaPublishActorApplyFailureTests : RuntimeActorTestBase if (last is not null) throw last; } + private static readonly System.Collections.Concurrent.ConcurrentDictionary Artifacts = new(); + + private static OpcUaPublishActor.RebuildAddressSpace Rebuild(Guid deploymentId) => + new(CorrelationId.NewId(), new DeploymentId(deploymentId), Artifacts[deploymentId]); + private static Guid SeedEquipmentDeployment(IDbContextFactory dbFactory, params (string Id, string Name)[] equipment) { var artifact = JsonSerializer.SerializeToUtf8Bytes(new @@ -116,6 +121,7 @@ public sealed class OpcUaPublishActorApplyFailureTests : RuntimeActorTestBase }); var id = Guid.NewGuid(); + Artifacts[id] = artifact; using var ctx = dbFactory.CreateDbContext(); ctx.Deployments.Add(new Deployment { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorOffThreadPlanTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorOffThreadPlanTests.cs new file mode 100644 index 00000000..734edbba --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorOffThreadPlanTests.cs @@ -0,0 +1,128 @@ +using System.Collections.Concurrent; +using System.Text.Json; +using Akka.Actor; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa; + +/// +/// Review 02 P5 — a rebuild's artifact parse + plan no longer run on the pinned OPC UA thread. The +/// observable: value batches sent right after a large rebuild land on the sink BEFORE that rebuild's first +/// materialise call, because the actor is free while the thread pool parses and plans; two rebuilds sent +/// back to back still apply in arrival order; a plan that throws still answers the requester degraded. +/// +public sealed class OpcUaPublishActorOffThreadPlanTests : RuntimeActorTestBase +{ + [Fact] + public void Value_batches_keep_flowing_while_a_large_plan_computes() + { + var sink = new OrderedSink(); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier)); + // A value node the batch names; the sink is a recorder, so it need not exist. + var batch = new OpcUaPublishActor.AttributeValueBatchUpdate( + [new ValueWrite("Plant/Modbus/dev1/live", 1, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw)]); + + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), RawTagArtifact(50_000), ReportOutcome: true)); + actor.Tell(batch); + + var completed = ExpectMsg(TimeSpan.FromSeconds(120)); + completed.Degraded.ShouldBeFalse(completed.DegradedReason); + + var calls = sink.Calls.ToList(); + var batchIndex = calls.IndexOf("WV:Plant/Modbus/dev1/live"); + var firstMaterialise = calls.FindIndex(c => c.StartsWith("EV:", StringComparison.Ordinal) || c.StartsWith("EF:", StringComparison.Ordinal)); + batchIndex.ShouldBeGreaterThanOrEqualTo(0, "the batch must reach the sink"); + firstMaterialise.ShouldBeGreaterThanOrEqualTo(0, "the plan must still be applied (positive control)"); + batchIndex.ShouldBeLessThan(firstMaterialise, "the batch sent after the rebuild must land before the rebuild's materialisation — the parse/plan step no longer holds the actor thread"); + calls.Count(c => c.StartsWith("EV:", StringComparison.Ordinal)).ShouldBe(50_000); + } + + [Fact] + public void Back_to_back_rebuilds_apply_in_arrival_order() + { + var sink = new OrderedSink(); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier)); + + // The first (large) plan is still computing when the second (tiny) arrives; the second must wait for + // it — its plan is a diff against what the first applies. + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), RawTagArtifact(20_000), ReportOutcome: true)); + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), RawTagArtifact(2), ReportOutcome: true)); + + var first = ExpectMsg(TimeSpan.FromSeconds(120)); + var second = ExpectMsg(TimeSpan.FromSeconds(120)); + first.Degraded.ShouldBeFalse(first.DegradedReason); + second.Degraded.ShouldBeFalse(second.DegradedReason); + + // The second composition is a strict subset (2 of the same tags), so it diffs as a PureRemove of the + // other 19,998 — which only works if it was planned AFTER the first apply, against its composition. + sink.Calls.Count(c => c.StartsWith("RV:", StringComparison.Ordinal)).ShouldBe(19_998); + } + + [Fact] + public void Unreadable_bytes_answer_degraded_and_keep_the_served_space() + { + var sink = new OrderedSink(); + var applier = new AddressSpaceApplier(sink, NullLogger.Instance); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier)); + + actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), "not json"u8.ToArray(), ReportOutcome: true)); + + var completed = ExpectMsg(PresenceBudget); + completed.Degraded.ShouldBeTrue(); + completed.DegradedReason.ShouldNotBeNull().ShouldContain("threw"); + sink.Calls.ShouldBeEmpty("nothing may be torn down for bytes that could not be read"); + sink.Completed.ShouldBe(1, "the apply is still closed"); + } + + private static byte[] RawTagArtifact(int tagCount) => + JsonSerializer.SerializeToUtf8Bytes(new + { + UnsAreas = Array.Empty(), + UnsLines = Array.Empty(), + Equipment = Array.Empty(), + RawFolders = new[] { new { RawFolderId = "rf-1", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } }, + DriverInstances = new[] { new { DriverInstanceId = "drv-1", DriverType = "Modbus", DriverConfig = "{}", Name = "Modbus", RawFolderId = "rf-1", ClusterId = "c1" } }, + Devices = new[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } }, + TagGroups = Array.Empty(), + Tags = Enumerable.Range(0, tagCount).Select(i => new + { + TagId = $"t-{i}", DeviceId = "dev-1", TagGroupId = (string?)null, Name = $"T{i}", + DataType = "Float", AccessLevel = 0, TagConfig = "{}", + }).ToArray(), + UnsTagReferences = Array.Empty(), + ScriptedAlarms = Array.Empty(), + }); + + private sealed class OrderedSink : IOpcUaAddressSpaceSink, ISurgicalAddressSpaceSink + { + public ConcurrentQueue Calls { get; } = new(); + public int Completed; + + public void WriteValue(string nodeId, object? value, uint statusCode, DateTime ts, AddressSpaceRealm realm) => Calls.Enqueue($"WV:{nodeId}"); + public void WriteValues(IReadOnlyList values) { foreach (var v in values) Calls.Enqueue($"WV:{v.NodeId}"); } + public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime ts, AddressSpaceRealm realm) { } + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime ts, AddressSpaceRealm realm) => Calls.Enqueue($"WA:{alarmNodeId}"); + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) => Calls.Enqueue($"MA:{alarmNodeId}"); + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm) => Calls.Enqueue($"EF:{folderNodeId}"); + 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) => Calls.Enqueue($"EV:{variableNodeId}"); + public void RebuildAddressSpace() => Calls.Enqueue("RB"); + public void CompleteAddressSpaceApply() => Interlocked.Increment(ref Completed); + public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm) => Calls.Enqueue($"NA:{affectedNodeId}"); + 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") { } + public bool UpdateTagAttributes(string variableNodeId, bool writable, string? historianTagname, string dataType, bool isArray, uint? arrayLength, AddressSpaceRealm realm) { Calls.Enqueue($"UT:{variableNodeId}"); return true; } + public bool UpdateFolderDisplayName(string folderNodeId, string displayName, AddressSpaceRealm realm) => true; + public bool RemoveVariableNode(string variableNodeId, AddressSpaceRealm realm) { Calls.Enqueue($"RV:{variableNodeId}"); return true; } + public bool RemoveAlarmConditionNode(string alarmNodeId, AddressSpaceRealm realm) { Calls.Enqueue($"RA:{alarmNodeId}"); return true; } + public bool RemoveEquipmentSubtree(string equipmentNodeId, AddressSpaceRealm realm) { Calls.Enqueue($"RE:{equipmentNodeId}"); return true; } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorRebuildTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorRebuildTests.cs index 2e0d239d..fd05748c 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorRebuildTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorRebuildTests.cs @@ -41,14 +41,14 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var sink = new RecordingSink(); var applier = new AddressSpaceApplier(sink, NullLogger.Instance); - SeedDeployment(db, equipmentIds: new[] { "eq-1", "eq-2" }, driverIds: new[] { "drv-1" }); + var dep = SeedDeployment(db, equipmentIds: new[] { "eq-1", "eq-2" }, driverIds: new[] { "drv-1" }); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + actor.Tell(Rebuild(dep)); AwaitAssert(() => { @@ -72,10 +72,10 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var db = NewInMemoryDbFactory(); var sink = new RecordingSink(); var applier = new AddressSpaceApplier(sink, NullLogger.Instance); - SeedDeployment(db, equipmentIds: new[] { "eq-1", "eq-2" }, driverIds: new[] { "drv-1" }); + var dep = SeedDeployment(db, equipmentIds: new[] { "eq-1", "eq-2" }, driverIds: new[] { "drv-1" }); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + actor.Tell(Rebuild(dep)); AwaitAssert(() => sink.CompletedAfterCallCount.Count.ShouldBe(1), duration: PresenceBudget); sink.CompletedAfterCallCount.Single().ShouldBe(sink.Calls.Count, "nothing may follow the completion"); @@ -126,17 +126,17 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var db = NewInMemoryDbFactory(); var sink = new RecordingSink(); var applier = new AddressSpaceApplier(sink, NullLogger.Instance); - SeedDeployment(db, equipmentIds: new[] { "eq-1" }, driverIds: Array.Empty()); + var dep = SeedDeployment(db, equipmentIds: new[] { "eq-1" }, driverIds: Array.Empty()); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + actor.Tell(Rebuild(dep)); AwaitAssert(() => sink.Calls.ShouldContain("EF:eq-1"), duration: PresenceBudget); sink.RebuildCalls.ShouldBe(0); // PureAdd — no full rebuild var callsAfterFirst = sink.Calls.Count; - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + actor.Tell(Rebuild(dep)); Thread.Sleep(200); // Same composition ⇒ plan IsEmpty ⇒ applier + materialise passes not run again. sink.Calls.Count.ShouldBe(callsAfterFirst); @@ -232,7 +232,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase /// and drives the applier. The guard split must leave the Direct path byte-identical. /// [Fact] - public void Rebuild_direct_mode_with_dbFactory_and_deploymentId_still_loads_and_applies() + public void Rebuild_with_deploymentId_and_in_hand_bytes_applies() { var db = NewInMemoryDbFactory(); var sink = new RecordingSink(); @@ -242,7 +242,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep))); + actor.Tell(Rebuild(dep)); AwaitAssert(() => sink.Calls.ShouldContain("EF:eq-1"), duration: PresenceBudget); sink.RebuildCalls.ShouldBe(0); @@ -267,7 +267,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dbA = NewInMemoryDbFactory(); var sinkA = new RecordingSink(); var applierA = new AddressSpaceApplier(sinkA, NullLogger.Instance); - SeedMultiClusterDeployment(dbA); + var depA = SeedMultiClusterDeployment(dbA); var siteActor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( sink: sinkA, @@ -275,7 +275,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase applier: applierA, localNode: NodeId.Parse("site-a-1:4053"))); - siteActor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + siteActor.Tell(Rebuild(depA)); // PureAdd ⇒ no full rebuild; the materialise passes still run the cluster slice. AwaitAssert(() => sinkA.Calls.ShouldContain("EV:SA/Modbus/dev-sa/S1"), duration: PresenceBudget); @@ -287,7 +287,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dbM = NewInMemoryDbFactory(); var sinkM = new RecordingSink(); var applierM = new AddressSpaceApplier(sinkM, NullLogger.Instance); - SeedMultiClusterDeployment(dbM); + var depM = SeedMultiClusterDeployment(dbM); var mainActor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( sink: sinkM, @@ -295,7 +295,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase applier: applierM, localNode: NodeId.Parse("central-1:4053"))); - mainActor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + mainActor.Tell(Rebuild(depM)); AwaitAssert(() => sinkM.Calls.ShouldContain("EV:Main/Modbus/dev-main/M1"), duration: PresenceBudget); sinkM.RebuildCalls.ShouldBe(0); @@ -308,7 +308,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase /// tag chain plus its UNS area/line/equipment. Used by /// . /// - private static void SeedMultiClusterDeployment(IDbContextFactory dbFactory) + private static Guid SeedMultiClusterDeployment(IDbContextFactory dbFactory) { var artifact = JsonSerializer.SerializeToUtf8Bytes(new { @@ -358,17 +358,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase ScriptedAlarms = Array.Empty(), }); - using var ctx = dbFactory.CreateDbContext(); - ctx.Deployments.Add(new Deployment - { - DeploymentId = Guid.NewGuid(), - RevisionHash = new string('b', 64), - Status = DeploymentStatus.Sealed, - CreatedBy = "test", - SealedAtUtc = DateTime.UtcNow, - ArtifactBlob = artifact, - }); - ctx.SaveChanges(); + return SealArtifact(dbFactory, artifact); } /// R2-07 T4b — a PureAdd deploy (here a single new equipment) does NOT rebuild, and the @@ -381,10 +371,10 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var db = NewInMemoryDbFactory(); var sink = new RecordingSink(); var applier = new AddressSpaceApplier(sink, NullLogger.Instance); - SeedDeployment(db, equipmentIds: new[] { "eq-1" }, driverIds: Array.Empty()); + var dep = SeedDeployment(db, equipmentIds: new[] { "eq-1" }, driverIds: Array.Empty()); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId())); + actor.Tell(Rebuild(dep)); AwaitAssert(() => sink.Calls.ShouldContain("NA:line-1"), duration: PresenceBudget); sink.RebuildCalls.ShouldBe(0); @@ -410,7 +400,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dep1 = SeedNamedEquipmentDeployment(db, ("eq-1", "Pump-1")); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.Count(c => c.StartsWith("NA:")).ShouldBe(1), duration: PresenceBudget); sink.RebuildCalls.ShouldBe(0); var naAfterFirst = sink.Calls.Count(c => c.StartsWith("NA:")); @@ -418,7 +408,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase // Second deploy: eq-1 RENAMED (ChangedEquipment ⇒ Rebuild) AND a new eq-2 (an add). The rebuild // fires, so the announce hook is skipped despite eq-2 being added. var dep2 = SeedNamedEquipmentDeployment(db, ("eq-1", "Pump-1-RENAMED"), ("eq-2", "Pump-2")); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); AwaitAssert(() => sink.RebuildCalls.ShouldBe(1), duration: PresenceBudget); // No new NodeAdded announcement was raised for the rebuild-kind deploy. @@ -442,7 +432,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dep1 = SeedNamedEquipmentDeployment(db, ("eq-1", "Pump-1"), ("eq-2", "Pump-2")); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.ShouldContain("EF:eq-2"), duration: PresenceBudget); var callsAfterFirst = sink.Calls.Count; @@ -458,7 +448,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase // …and the actor still remembers WHAT is applied: once the database is back, re-applying the same // composition diffs to an empty plan. A clobbered _lastApplied would re-materialise everything. db.Fail = false; - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); Thread.Sleep(200); sink.Calls.Count.ShouldBe(callsAfterFirst); } @@ -477,12 +467,12 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dep1 = SeedNamedEquipmentDeployment(db, ("eq-1", "Pump-1"), ("eq-2", "Pump-2")); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.ShouldContain("EF:eq-2"), duration: PresenceBudget); // eq-2 really is gone from the next artifact — that IS a configuration change, so it must apply. var dep2 = SeedNamedEquipmentDeployment(db, ("eq-1", "Pump-1")); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); AwaitAssert(() => sink.Calls.ShouldContain("RE:eq-2"), duration: PresenceBudget); } @@ -534,24 +524,10 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase IDbContextFactory dbFactory, params (string Id, string Name)[] equipment) { - var artifact = BuildNamedEquipmentArtifact(equipment); - - var id = Guid.NewGuid(); - using var ctx = dbFactory.CreateDbContext(); - ctx.Deployments.Add(new Deployment - { - DeploymentId = id, - RevisionHash = new string('d', 64), - Status = DeploymentStatus.Sealed, - CreatedBy = "test", - SealedAtUtc = DateTime.UtcNow, - ArtifactBlob = artifact, - }); - ctx.SaveChanges(); - return id; + return SealArtifact(dbFactory, BuildNamedEquipmentArtifact(equipment)); } - private static void SeedDeployment( + private static Guid SeedDeployment( IDbContextFactory dbFactory, string[] equipmentIds, string[] driverIds) @@ -575,17 +551,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase ScriptedAlarms = Array.Empty(), }); - using var ctx = dbFactory.CreateDbContext(); - ctx.Deployments.Add(new Deployment - { - DeploymentId = Guid.NewGuid(), - RevisionHash = new string('a', 64), - Status = DeploymentStatus.Sealed, - CreatedBy = "test", - SealedAtUtc = DateTime.UtcNow, - ArtifactBlob = artifact, - }); - ctx.SaveChanges(); + return SealArtifact(dbFactory, artifact); } /// OpcUaServer-001 — a deploy whose ONLY change is a UNS Area rename (the area Name differs; @@ -606,14 +572,14 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase // First deploy: the area folder is materialised with the OLD name. R2-07 — this is now a PureAdd // (area + line + equipment all added), so NO full rebuild; the folder is materialised directly. - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.ShouldContain("EF:area-1"), duration: PresenceBudget); sink.RebuildCalls.ShouldBe(0); // Second deploy: ONLY the area Name changed — a rename. The actor must reach the apply path and // drive a surgical in-place folder rename (NOT a rebuild, NOT a no-op). var dep2 = SeedAreaDeployment(db, areaName: "Plant South"); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); AwaitAssert(() => { @@ -639,19 +605,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase ScriptedAlarms = Array.Empty(), }); - var id = Guid.NewGuid(); - using var ctx = dbFactory.CreateDbContext(); - ctx.Deployments.Add(new Deployment - { - DeploymentId = id, - RevisionHash = new string('c', 64), - Status = DeploymentStatus.Sealed, - CreatedBy = "test", - SealedAtUtc = DateTime.UtcNow, - ArtifactBlob = artifact, - }); - ctx.SaveChanges(); - return id; + return SealArtifact(dbFactory, artifact); } // --------------------------------------------------------------------------------------------------- @@ -677,13 +631,13 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dep1 = SeedRawTagDeployment(db, new[] { "A", "B", "C" }); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.ShouldContain("EV:Plant/Modbus/dev1/C"), duration: PresenceBudget); var afterFirst = sink.Calls.Count; // One new tag on an otherwise identical configuration. var dep2 = SeedRawTagDeployment(db, new[] { "A", "B", "C", "D" }); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); AwaitAssert(() => sink.Calls.ShouldContain("EV:Plant/Modbus/dev1/D"), duration: PresenceBudget); Thread.Sleep(200); // let any (unwanted) extra pass finish before counting @@ -707,13 +661,13 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dep1 = SeedUnsRawDeployment(db, equipmentName: "station1", referencedTag: "a"); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.ShouldContain("EV:Plant/Modbus/dev1/b"), duration: PresenceBudget); var afterFirst = sink.Calls.Count; // Renaming the equipment is a ChangedEquipment delta ⇒ the classifier's default-closed Rebuild. var dep2 = SeedUnsRawDeployment(db, equipmentName: "station1-renamed", referencedTag: "a"); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); AwaitAssert(() => sink.RebuildCalls.ShouldBe(1), duration: PresenceBudget); AwaitAssert( () => sink.Calls.Skip(afterFirst).ShouldContain("EV:Plant/Modbus/dev1/b"), @@ -746,13 +700,13 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dep1 = SeedUnsRawDeployment(db, equipmentName: "station1", referencedTag: "a"); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.ShouldContain("EV:filling/line1/station1/a"), duration: PresenceBudget); var afterFirst = sink.Calls.Count; // Re-point the SAME reference row at raw tag "b". var dep2 = SeedUnsRawDeployment(db, equipmentName: "station1", referencedTag: "b"); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); AwaitAssert(() => sink.Calls.ShouldContain("EV:filling/line1/station1/b"), duration: PresenceBudget); var second = sink.Calls.Skip(afterFirst).ToList(); @@ -784,13 +738,13 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dep1 = SeedUnsRawDeployment(db, equipmentName: "station1", referencedTag: "a"); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.ShouldContain("EV:filling/line1/station1/a"), duration: PresenceBudget); var afterFirst = sink.Calls.Count; var dep2 = SeedUnsRawDeployment(db, equipmentName: "station1", referencedTag: "a", historianTagnameForA: "WW.A"); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); // The mirror now carries the backing tag's historian tagname — the repair itself, unchanged by I2. AwaitAssert( @@ -818,12 +772,12 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dep1 = SeedRawTagDeployment(db, new[] { "A", "B" }); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.ShouldContain("EV:Plant/Modbus/dev1/B"), duration: PresenceBudget); var afterFirst = sink.Calls.Count; var dep2 = SeedRawTagDeployment(db, new[] { "A", "B" }, historizedTag: "A"); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); AwaitAssert(() => sink.Calls.ShouldContain("UT:Plant/Modbus/dev1/A"), duration: PresenceBudget); Thread.Sleep(200); // let any (unwanted) extra pass finish before counting @@ -855,7 +809,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dep1 = SeedRawTagDeployment(db, new[] { "A", "B" }); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.ShouldContain("EV:Plant/Modbus/dev1/B"), duration: PresenceBudget); Thread.Sleep(200); sink.RebuildCalls.ShouldBe(0); // genuinely a non-rebuild (PureAdd) apply @@ -867,7 +821,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase // EMPTY plan and A would be gone from the served address space forever. sink.FailEnsureVariableFor = null; var dep2 = SeedRawTagDeployment(db, new[] { "A", "B" }); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); AwaitAssert( () => sink.Calls.Skip(afterFirst).ShouldContain("EV:Plant/Modbus/dev1/A"), @@ -894,14 +848,14 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase var dep1 = SeedUnsRawDeployment(db, equipmentName: "station1", referencedTag: "a"); var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier)); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1))); + actor.Tell(Rebuild(dep1)); AwaitAssert(() => sink.Calls.ShouldContain("EV:Plant/Modbus/dev1/b"), duration: PresenceBudget); // Renaming the equipment is ChangedEquipment ⇒ the classifier's default-closed Rebuild; the sink // faults on one variable of the full re-materialise that follows. sink.FailEnsureVariableFor = "Plant/Modbus/dev1/b"; var dep2 = SeedUnsRawDeployment(db, equipmentName: "station1-renamed", referencedTag: "a"); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2))); + actor.Tell(Rebuild(dep2)); AwaitAssert(() => sink.RebuildCalls.ShouldBe(1), duration: PresenceBudget); Thread.Sleep(200); var afterSecond = sink.Calls.Count; @@ -911,7 +865,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase // rebuild again) — but the degraded rebuild is remembered, so the failed variable is re-ensured. sink.FailEnsureVariableFor = null; var dep3 = SeedUnsRawDeployment(db, equipmentName: "station1-renamed", referencedTag: "a"); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep3))); + actor.Tell(Rebuild(dep3)); AwaitAssert( () => sink.Calls.Skip(afterSecond).ShouldContain("EV:Plant/Modbus/dev1/b"), duration: PresenceBudget); @@ -921,7 +875,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase Thread.Sleep(200); var afterRepair = sink.Calls.Count; var dep4 = SeedUnsRawDeployment(db, equipmentName: "station1-renamed", referencedTag: "a"); - actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep4))); + actor.Tell(Rebuild(dep4)); Thread.Sleep(300); sink.Calls.Count.ShouldBe(afterRepair); sink.RebuildCalls.ShouldBe(1); @@ -1017,10 +971,19 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase return SealArtifact(dbFactory, artifact); } + /// The artifact bytes behind each sealed deployment id, so a test can hand the actor the bytes its + /// production producers always carry (review 02 P5 removed the actor's ConfigDb fallback). + private static readonly System.Collections.Concurrent.ConcurrentDictionary Artifacts = new(); + + /// The rebuild message a producer sends for deployment : its id AND its bytes. + private static OpcUaPublishActor.RebuildAddressSpace Rebuild(Guid deploymentId) => + new(CorrelationId.NewId(), new DeploymentId(deploymentId), Artifacts[deploymentId]); + /// Seal as a new deployment and return its id. private static Guid SealArtifact(IDbContextFactory dbFactory, byte[] artifact) { var id = Guid.NewGuid(); + Artifacts[id] = artifact; using var ctx = dbFactory.CreateDbContext(); ctx.Deployments.Add(new Deployment {