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
[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