dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(runtime): parse + plan a deploy off the pinned OPC UA thread (B6)

Browse Source 
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
This commit is contained in:
Joseph Doherty
2026-10-22 10:36:34 -04:00
parent 118933d5d5
commit deaa31479b
4 changed files with 360 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
+168 -160
View File
@@ -86,19 +86,17 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// <param name="Realm">The namespace realm the condition lives in (<see cref="AddressSpaceRealm.Raw"/> for native).</param>
public sealed record AlarmQualityUpdate(string AlarmNodeId, uint StatusCode, DateTime TimestampUtc, AddressSpaceRealm Realm);
/// <summary>
/// Triggers an address-space rebuild. <paramref name="DeploymentId"/> 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 <paramref name="Artifact"/>.
/// <paramref name="DeploymentId"/> 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.
/// </summary>
/// <param name="Correlation">Correlation id for tracing this rebuild.</param>
/// <param name="DeploymentId">The applied deployment whose artifact to materialise, or null for the latest-sealed fallback.</param>
/// <param name="DeploymentId">The applied deployment whose artifact this is, or null for a legacy caller.</param>
/// <param name="Artifact">
/// 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 <see cref="DeploymentId"/> 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.
/// </param>
/// <param name="ReportOutcome">
/// when <see langword="true"/>, the actor replies to the sender with a
@@ -128,11 +126,29 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
public sealed record ServiceLevelChanged(byte ServiceLevel);
/// <summary>
/// Review 02 P5 — a rebuild whose artifact parse, cluster-scoped composition and plan were computed
/// OFF the pinned OPC UA thread (a <c>Task.Run</c> piped back to <c>Self</c>), ready for the sink passes,
/// which are the only part that must run on this actor's thread. Carries the <see cref="Request"/> and
/// the original <see cref="Requester"/> so the <see cref="RebuildCompleted"/> reply reaches whoever asked.
/// </summary>
private sealed record PlanReady(
RebuildAddressSpace Request, IActorRef Requester, byte[] Artifact, DataPlaneAuthorityProjection? Authority,
AddressSpaceComposition Composition, AddressSpacePlan Plan, IReadOnlyList<string> Inconsistencies);
/// <summary>The off-thread parse / plan step threw (unreadable bytes, a composer fault).</summary>
private sealed record PlanFailed(RebuildAddressSpace Request, IActorRef Requester, Exception Error);
/// <summary>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 <see cref="_lastApplied"/>, which only the apply advances — so a
/// rebuild arriving meanwhile waits in <see cref="_pendingRebuilds"/>; value and status batches do not.</summary>
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<OtOpcUaConfigDbContext>? _dbFactory;
private readonly AddressSpaceApplier? _applier;
/// <summary>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
/// <summary>Production Props — pins the OPC UA dispatcher + subscribes to the
/// <c>redundancy-state</c> DPS topic so cluster transitions drive the local ServiceLevel
/// publish path. When <paramref name="dbFactory"/> + <paramref name="applier"/> are supplied,
/// <see cref="RebuildAddressSpace"/> reads the latest deployment artifact + drives the
/// applier through the sink. When <paramref name="dbHealthProbe"/> is supplied the local
/// publish path. When <paramref name="applier"/> is supplied, <see cref="RebuildAddressSpace"/>
/// diffs + applies the artifact bytes it carries through the sink (<paramref name="dbFactory"/> is
/// accepted for wiring compatibility only — review 02 P5 removed the ConfigDb fallback). When <paramref name="dbHealthProbe"/> is supplied the local
/// ServiceLevel is computed via <see cref="ServiceLevelCalculator"/> from real DB-health +
/// staleness + role-leader inputs; otherwise the legacy role-only switch is used.</summary>
/// <param name="sink">The OPC UA address space sink.</param>
@@ -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<AlarmStateUpdate>(HandleAlarmUpdate);
Receive<AlarmQualityUpdate>(HandleAlarmQualityUpdate);
Receive<RebuildAddressSpace>(HandleRebuild);
Receive<PlanReady>(HandlePlanReady);
Receive<PlanFailed>(HandlePlanFailed);
Receive<ServiceLevelChanged>(HandleServiceLevelChanged);
Receive<RedundancyStateChanged>(HandleRedundancyStateChanged);
Receive<DbHealthProbeActor.DbHealthStatus>(HandleDbHealthStatus);
@@ -480,114 +500,147 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
}
}
private void HandleRebuild(RebuildAddressSpace msg)
private void HandleRebuild(RebuildAddressSpace msg) => StartRebuild(msg, Sender);
/// <summary>
/// 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 <see cref="PlanReady"/>. Only the sink materialisation (<see cref="ApplyPlan"/>) runs here.
/// </summary>
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));
}
/// <summary>The rebuild pipeline. Returns whether the apply was degraded, and why.</summary>
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<string, object?>("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<string>();
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);
}
/// <summary>
/// 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.
/// </summary>
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));
}
/// <summary>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.</summary>
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<byte>() — 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<byte>());
// 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.
/// </remarks>
/// <param name="parsed">The rebuild's single parse of the artifact, read for the keys + grants.</param>
/// <param name="projection">The keys + grants decoded (off-thread) from the rebuild's single parse of the artifact.</param>
/// <param name="artifactBytes">The RAW artifact bytes — hashed for the revision id, which must be
/// SHA-256 of the exact blob (see below); never re-parsed here.</param>
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
}
}
/// <summary>Read a specific deployment's artifact blob from ConfigDb (the one just applied,
/// which may not be Sealed yet). Empty array on any failure — <see cref="HandleRebuild"/> treats that as
/// "no answer" and abandons the rebuild, leaving the served address space untouched.</summary>
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<byte>();
}
catch (Exception ex)
{
_log.Warning(ex, "OpcUaPublish: failed to load artifact for deployment {Id}; the rebuild is abandoned", deploymentId);
return Array.Empty<byte>();
}
}
/// <summary>Read the most recent <c>Sealed</c> deployment's artifact blob from ConfigDb.
/// Empty array on any failure — see <see cref="LoadArtifact"/> for how that is handled.</summary>
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<byte>();
}
catch (Exception ex)
{
_log.Warning(ex, "OpcUaPublish: failed to load latest deployment artifact; the rebuild is abandoned");
return Array.Empty<byte>();
}
}
private void HandleServiceLevelChanged(ServiceLevelChanged msg)
{
// Always publish the FIRST computed level, even if it equals the byte-default 0. Otherwise a
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs
+10 -4
View File
@@ -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<AddressSpaceApplier>.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<Guid, byte[]> Artifacts = new();
private static OpcUaPublishActor.RebuildAddressSpace Rebuild(Guid deploymentId) =>
new(CorrelationId.NewId(), new DeploymentId(deploymentId), Artifacts[deploymentId]);
private static Guid SeedEquipmentDeployment(IDbContextFactory<OtOpcUaConfigDbContext> 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
{
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorOffThreadPlanTests.cs
+128
View File
@@ -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;
/// <summary>
/// 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.
/// </summary>
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<AddressSpaceApplier>.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<OpcUaPublishActor.RebuildCompleted>(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<AddressSpaceApplier>.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<OpcUaPublishActor.RebuildCompleted>(TimeSpan.FromSeconds(120));
var second = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(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<AddressSpaceApplier>.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<OpcUaPublishActor.RebuildCompleted>(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<object>(),
UnsLines = Array.Empty<object>(),
Equipment = Array.Empty<object>(),
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<object>(),
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<object>(),
ScriptedAlarms = Array.Empty<object>(),
});
private sealed class OrderedSink : IOpcUaAddressSpaceSink, ISurgicalAddressSpaceSink
{
public ConcurrentQueue<string> 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<ValueWrite> 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<string> 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; }
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorRebuildTests.cs
+54 -91
View File
@@ -41,14 +41,14 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase
var sink = new RecordingSink();
var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.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<AddressSpaceApplier>.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<AddressSpaceApplier>.Instance);
SeedDeployment(db, equipmentIds: new[] { "eq-1" }, driverIds: Array.Empty<string>());
var dep = SeedDeployment(db, equipmentIds: new[] { "eq-1" }, driverIds: Array.Empty<string>());
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.
/// </summary>
[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<AddressSpaceApplier>.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<AddressSpaceApplier>.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
/// <see cref="Rebuild_materialises_only_the_nodes_cluster"/>.
/// </summary>
private static void SeedMultiClusterDeployment(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory)
private static Guid SeedMultiClusterDeployment(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory)
{
var artifact = JsonSerializer.SerializeToUtf8Bytes(new
{
@@ -358,17 +358,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase
ScriptedAlarms = Array.Empty<object>(),
});
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);
}
/// <summary>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<AddressSpaceApplier>.Instance);
SeedDeployment(db, equipmentIds: new[] { "eq-1" }, driverIds: Array.Empty<string>());
var dep = SeedDeployment(db, equipmentIds: new[] { "eq-1" }, driverIds: Array.Empty<string>());
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<OtOpcUaConfigDbContext> 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<OtOpcUaConfigDbContext> dbFactory,
string[] equipmentIds,
string[] driverIds)
@@ -575,17 +551,7 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase
ScriptedAlarms = Array.Empty<object>(),
});
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);
}
/// <summary>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<object>(),
});
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);
}
/// <summary>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).</summary>
private static readonly System.Collections.Concurrent.ConcurrentDictionary<Guid, byte[]> Artifacts = new();
/// <summary>The rebuild message a producer sends for deployment <paramref name="deploymentId"/>: its id AND its bytes.</summary>
private static OpcUaPublishActor.RebuildAddressSpace Rebuild(Guid deploymentId) =>
new(CorrelationId.NewId(), new DeploymentId(deploymentId), Artifacts[deploymentId]);
/// <summary>Seal <paramref name="artifact"/> as a new deployment and return its id.</summary>
private static Guid SealArtifact(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, byte[] artifact)
{
var id = Guid.NewGuid();
Artifacts[id] = artifact;
using var ctx = dbFactory.CreateDbContext();
ctx.Deployments.Add(new Deployment
{