diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
index 6f536cb0..6b6f5970 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
@@ -1494,6 +1494,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
// same-key kind-swap ever carries wired notifiers), but this is the one asymmetric teardown
// path; keep it symmetric with RemoveAlarmConditionNode / RebuildAddressSpace.
UnwireAlarmNotifiers(conditionKey);
+ // A removal path like any other (review 01 L2): the dropped instance's items are told and its
+ // fan-out forgotten, or a new item on the swapped-in condition would bind to the dead one.
+ TombstoneMonitoredSubtree(existing, UtcNow);
existing.Parent?.RemoveChild(existing);
ForgetPredefinedSubtree(existing);
// The materialise tail raises only the new instance's unit (no full reconcile), so give the
@@ -4413,6 +4416,18 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|| current.DataChangeMonitoredItems?.Contains(dataItem) != true))
{
OnMonitoredItemDeleted(context, handle, dataItem);
+
+ // Review fix I3 — the SDK's DeleteMonitoredItems releases the item's component-cache reference
+ // whenever this returns Good. A tombstoned item's reference was ALREADY released by the removal, so
+ // that second release would take one from the RE-CREATED node's entry (a new item on the same NodeId)
+ // and evict it while still referenced. Take one back first so the SDK's release nets to zero —
+ // returning a non-Good status instead would fail the client's delete.
+ if (_cacheReferenceReleased.TryGetValue(dataItem, out _))
+ {
+ _cacheReferenceReleased.Remove(dataItem);
+ AddNodeToComponentCache(context, handle, handle.Node);
+ }
+
return ServiceResult.Good;
}
}
@@ -4472,11 +4487,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
if (monitored.EventMonitoredItems is { Count: > 0 } eventItems)
{
- foreach (var item in eventItems.ToArray())
- {
- _eventItems.TryRemove(item, out _);
- ReleaseComponentCacheReference(item);
- }
+ // No component-cache release: an event item never took a reference (the master node manager
+ // creates it through the EventManager + SubscribeToEvents, not AddNodeToComponentCache), so a
+ // release here would take one from a DATA item on the same NodeId.
+ foreach (var item in eventItems.ToArray()) _eventItems.TryRemove(item, out _);
}
}
@@ -4534,7 +4548,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
if (!tracked.Suppressed || (nodeId is not null && tracked.Handle.NodeId != nodeId)) continue;
if (!_eventItems.TryRemove(item, out _)) continue;
Interlocked.Decrement(ref _suppressedItems);
- ReleaseComponentCacheReference(item);
+ // Event items hold no component-cache reference (see TombstoneMonitoredNode).
}
}
@@ -4555,13 +4569,21 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
}
}
- /// Gives back the component-cache reference the SDK took for at create, so
- /// the cache stops handing out the removed node for that NodeId.
+ /// Gives back the component-cache reference the SDK took for the DATA item at
+ /// create, so the cache stops handing out the removed node for that NodeId — and remembers that it did, so the
+ /// item's later client delete does not give it back a second time (see ).
private void ReleaseComponentCacheReference(IMonitoredItem item)
{
- if (item.ManagerHandle is NodeHandle handle) RemoveNodeFromComponentCache(SystemContext, handle);
+ if (item.ManagerHandle is not NodeHandle handle) return;
+ RemoveNodeFromComponentCache(SystemContext, handle);
+ _cacheReferenceReleased.AddOrUpdate(item, ReleasedMarker);
}
+ /// Data items whose component-cache reference a removal already gave back (review fix I3). Weak, so an
+ /// item the SDK drops without a delete (it always deletes, but a leak here would be silent) is not kept alive.
+ private readonly System.Runtime.CompilerServices.ConditionalWeakTable _cacheReferenceReleased = new();
+ private static readonly object ReleasedMarker = new();
+
/// Test observability: how many NodeIds currently carry a fan-out.
internal int MonitoredNodeCount
{
@@ -5352,8 +5374,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
{
_eventItems.TryRemove(item, out _);
Interlocked.Decrement(ref _suppressedItems);
- ReleaseComponentCacheReference(item);
- continue;
+ continue; // an event item holds no component-cache reference (see TombstoneMonitoredNode)
}
SubscribeToEvents(SystemContext, tracked.Handle.Node, item, unsubscribe: !permitted);
@@ -7065,7 +7086,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
else if (endBound is not null)
{
// The data is drained — the end bound ends this page, or is the next one when this one is full.
- clusterCp = await AppendEndBoundAsync(slice, startUtc, readCap).ConfigureAwait(false);
+ // An uncapped read (readCap 0) on a backend with a row ceiling is still a page of at most that
+ // ceiling — the MaxReturnDataValues it advertises — so the bound fits against it, not "unbounded".
+ // (Today the over-fetch bound keeps a cluster slice at most L − 1, so this is defence in depth.)
+ var clusterAllowance = readCap > 0 ? readCap : pageLimit > 0 ? (uint)pageLimit : 0u;
+ clusterCp = await AppendEndBoundAsync(slice, startUtc, clusterAllowance).ConfigureAwait(false);
}
results[handle.Index] = new SdkHistoryReadResult
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 702396a8..833525a0 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
@@ -132,16 +132,56 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// which are the only part that must run on this actor's thread. Carries the and
/// the original so the reply reaches whoever asked.
///
+ /// stamps the actor instance that started the plan: a PipeTo(Self)
+ /// outlives a restart (the ActorRef survives it), and a plan computed against a previous instance's
+ /// must not be applied by the next one. carries a
+ /// data-plane-authority parse failure, which never fails the address-space apply (review fix I2).
+ ///
private sealed record PlanReady(
- RebuildAddressSpace Request, IActorRef Requester, byte[] Artifact, DataPlaneAuthorityProjection? Authority,
+ object Incarnation, RebuildAddressSpace Request, IActorRef Requester, byte[] Artifact,
+ DataPlaneAuthorityProjection? Authority, Exception? AuthorityError,
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);
+ private sealed record PlanFailed(object Incarnation, RebuildAddressSpace Request, IActorRef Requester, Exception Error);
+
+ /// Test seams for the off-thread plan step. Production passes none.
+ public sealed class TestHooks
+ {
+ /// Runs on the thread pool at the start of every plan step (a test blocks it to hold a plan in flight).
+ public Action? BeforePlan { get; init; }
+
+ /// Replaces (a test makes it throw).
+ internal Func? ParseAuthority { get; init; }
+ }
+
+ /// The default bound on : distinct (realm, id) keys held while a plan is in flight.
+ public const int DefaultDeferredWriteCapacity = 250_000;
+
+ /// Identity of THIS actor instance (see ).
+ private readonly object _incarnation = new();
+ private readonly TestHooks? _hooks;
+
+ ///
+ /// Review fix C1 — the latest value / status / alarm write per (realm, id) received while a rebuild's plan is
+ /// in flight (or queued), replayed through the sink after each apply. Writes are non-creating (review 01 L1),
+ /// so a driver's first publish for a node the in-flight plan ADDS used to be skipped — and a change-driven
+ /// driver (or a poll group whose last-value cache already holds it) never sends it again, leaving the node
+ /// BadWaitingForInitialData indefinitely. On a full rebuild the same holds for every node: the
+ /// re-subscribe's first values land on old nodes the apply then discards.
+ /// Writes are still applied IMMEDIATELY (existing nodes keep updating during the plan); the buffer
+ /// only remembers each key's latest so the replay can land it on a node the apply created. Replaying onto a
+ /// node that already holds it is inert: the same value/status/timestamp is filtered by every monitored
+ /// item's data-change filter, and an alarm write is delta-gated against the node's current state. That is
+ /// also why alarms coalesce to the latest state per id rather than replaying every transition: replaying an
+ /// older transition onto a live condition would move it backwards and fire spurious events.
+ ///
+ private readonly DeferredWrites _deferred;
/// 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.
+ /// rebuild arriving meanwhile waits in ; value, status and alarm writes do not — they
+ /// are applied at once AND remembered in for replay after the apply.
private bool _planInFlight;
private readonly Queue<(RebuildAddressSpace Request, IActorRef Requester)> _pendingRebuilds = new();
@@ -309,6 +349,18 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
dataPlaneAuthority,
utcNow));
+ /// Test-only Props exposing the deferred-write bound and the plan-step seams.
+ internal static Props PropsForTests(
+ IOpcUaAddressSpaceSink sink,
+ AddressSpaceApplier? applier,
+ TestHooks? hooks,
+ int deferredWriteCapacity = DefaultDeferredWriteCapacity,
+ NodeId? localNode = null,
+ IDataPlaneAuthoritySink? dataPlaneAuthority = null) =>
+ Akka.Actor.Props.Create(() => new OpcUaPublishActor(
+ sink, NullServiceLevelPublisher.Instance, false, localNode, null, applier, null, null, null, null,
+ false, dataPlaneAuthority, null, deferredWriteCapacity, hooks));
+
/// Initializes a new instance of the class.
/// The OPC UA address space sink.
/// The service level publisher.
@@ -331,6 +383,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only
/// staleness demotes). No is wired on such a node.
/// The optional sink that receives the API keys and ACL grants carried by each applied artifact.
+ /// The bound on distinct nodes whose writes are held for replay while a rebuild plan is in flight.
+ /// Test seams for the off-thread plan step; null in production.
public OpcUaPublishActor(
IOpcUaAddressSpaceSink sink,
IServiceLevelPublisher serviceLevel,
@@ -344,8 +398,12 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
TimeSpan? healthTickInterval = null,
bool dbLess = false,
IDataPlaneAuthoritySink? dataPlaneAuthority = null,
- Func? utcNow = null)
+ Func? utcNow = null,
+ int deferredWriteCapacity = DefaultDeferredWriteCapacity,
+ TestHooks? hooks = null)
{
+ _deferred = new DeferredWrites(Math.Max(1, deferredWriteCapacity));
+ _hooks = hooks;
_sink = sink;
_dataPlaneAuthority = dataPlaneAuthority;
_serviceLevel = serviceLevel;
@@ -400,6 +458,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
{
try
{
+ if (PlanPipelineBusy) DeferValue(new ValueWrite(msg.NodeId, msg.Value, msg.StatusCode, msg.TimestampUtc, msg.Realm));
_sink.WriteValue(msg.NodeId, msg.Value, msg.StatusCode, msg.TimestampUtc, msg.Realm);
Interlocked.Increment(ref _writes);
OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair("kind", "value"));
@@ -421,6 +480,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// a whole batch because one item was bad.
private void HandleAttributeBatchUpdate(AttributeValueBatchUpdate msg)
{
+ if (PlanPipelineBusy)
+ foreach (var item in msg.Items) DeferValue(item);
try
{
_sink.WriteValues(msg.Items);
@@ -451,6 +512,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// value write on the same counter.
private void HandleAttributeStatusBatchUpdate(AttributeStatusBatchUpdate msg)
{
+ if (PlanPipelineBusy)
+ foreach (var item in msg.Items) DeferStatus(item);
try
{
_sink.WriteStatuses(msg.Items);
@@ -474,6 +537,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
private void HandleAlarmUpdate(AlarmStateUpdate msg)
{
+ if (PlanPipelineBusy) DeferAlarm(msg);
try
{
_sink.WriteAlarmCondition(msg.AlarmNodeId, msg.State, msg.TimestampUtc, msg.Realm);
@@ -488,6 +552,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
private void HandleAlarmQualityUpdate(AlarmQualityUpdate msg)
{
+ if (PlanPipelineBusy) DeferAlarmQuality(msg);
try
{
_sink.WriteAlarmQuality(msg.AlarmNodeId, msg.StatusCode, msg.TimestampUtc, msg.Realm);
@@ -511,6 +576,16 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
///
private void StartRebuild(RebuildAddressSpace msg, IActorRef requester)
{
+ // Strict arrival order for EVERY rebuild, including the ones that complete synchronously (no applier, no
+ // bytes): a bytes-less request arriving mid-plan used to answer before the plan ahead of it, so replies went
+ // out of order. HandleRebuild is the only entry that can find the pipeline busy; the drain calls this only
+ // when it is idle.
+ if (_planInFlight)
+ {
+ _pendingRebuilds.Enqueue((msg, requester));
+ return;
+ }
+
// 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).
@@ -559,36 +634,58 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
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;
+ var incarnation = _incarnation;
+ var hooks = _hooks;
Task.Run(() =>
{
+ hooks?.BeforePlan?.Invoke();
// 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;
+ // The authority parse is isolated from the address-space plan (review fix I2): before the plan moved
+ // off-thread a throw here was contained by FeedDataPlaneAuthority's own catch; it must not now
+ // become PlanFailed and degrade an apply whose topology is perfectly readable.
+ DataPlaneAuthorityProjection? authority = null;
+ Exception? authorityError = null;
+ if (wantAuthority)
+ {
+ try
+ {
+ authority = hooks?.ParseAuthority is { } parse
+ ? parse(parsed, nodeId ?? string.Empty)
+ : parsed.ParseDataPlaneAuthority(nodeId ?? string.Empty);
+ }
+ catch (Exception ex)
+ {
+ authorityError = ex;
+ }
+ }
+
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);
+ return new PlanReady(incarnation, msg, requester, artifact, authority, authorityError, composition, plan, inconsistencies);
})
- .PipeTo(Self, failure: ex => new PlanFailed(msg, requester, ex));
+ .PipeTo(Self, failure: ex => new PlanFailed(incarnation, msg, requester, ex));
}
private void HandlePlanReady(PlanReady ready)
{
+ if (IsFromAnotherIncarnation(ready.Incarnation, ready.Request, ready.Requester)) return;
+
+ if (ready.AuthorityError is { } authorityError)
+ _log.Error(authorityError,
+ "OpcUaPublish: data-plane authority parse threw (correlation={Correlation}) — the address space is applied; the previously installed keys and grants stay in force",
+ ready.Request.Correlation);
+
(bool Degraded, string? Reason) outcome;
try
{
@@ -600,20 +697,211 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
}
FinishRebuild(ready.Request, ready.Requester, outcome);
- StartNextPendingRebuild();
+ AfterPlanSettled();
}
private void HandlePlanFailed(PlanFailed failed)
{
+ if (IsFromAnotherIncarnation(failed.Incarnation, failed.Request, failed.Requester)) return;
+
_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();
+ AfterPlanSettled();
}
+ /// A plan piped back to a RESTARTED instance (the ActorRef survives a restart) was computed against the
+ /// previous instance's : never applied. Its requester is still owed an answer.
+ private bool IsFromAnotherIncarnation(object incarnation, RebuildAddressSpace request, IActorRef requester)
+ {
+ if (ReferenceEquals(incarnation, _incarnation)) return false;
+ _log.Warning(
+ "OpcUaPublish: ignoring a rebuild plan started by a previous incarnation of this actor (correlation={Correlation})",
+ request.Correlation);
+ if (request.ReportOutcome)
+ requester.Tell(new RebuildCompleted(request.Correlation, request.DeploymentId, true,
+ "the address-space publisher restarted while this rebuild was being planned; it was not applied"));
+ return true;
+ }
+
+ /// After an apply (or a failed plan) has been answered — its deferred writes already replayed by
+ /// — start what queued behind it, and drop the buffer once nothing is in flight or
+ /// queued (a queued plan may add nodes the buffer names, so it is replayed again after that apply).
+ private void AfterPlanSettled()
+ {
+ StartNextPendingRebuild();
+ if (!PlanPipelineBusy) _deferred.Clear();
+ }
+
+ /// Starts queued rebuilds in arrival order until one goes off-thread (or the queue empties) — the
+ /// synchronous completions (no bytes) answer in their turn.
private void StartNextPendingRebuild()
{
- if (_pendingRebuilds.TryDequeue(out var next)) StartRebuild(next.Request, next.Requester);
+ while (!_planInFlight && _pendingRebuilds.TryDequeue(out var next))
+ StartRebuild(next.Request, next.Requester);
+ }
+
+ /// True while a plan is computing or a rebuild is queued behind one.
+ private bool PlanPipelineBusy => _planInFlight || _pendingRebuilds.Count > 0;
+
+ private void DeferValue(in ValueWrite write)
+ {
+ if (!_deferred.Value(write)) WarnDeferredOverflow();
+ }
+
+ private void DeferStatus(in StatusWrite write)
+ {
+ if (!_deferred.Status(write)) WarnDeferredOverflow();
+ }
+
+ private void DeferAlarm(AlarmStateUpdate msg)
+ {
+ if (!_deferred.Alarm(msg)) WarnDeferredOverflow();
+ }
+
+ private void DeferAlarmQuality(AlarmQualityUpdate msg)
+ {
+ if (!_deferred.AlarmQuality(msg)) WarnDeferredOverflow();
+ }
+
+ private void WarnDeferredOverflow()
+ {
+ if (_deferred.Dropped == 1)
+ _log.Warning(
+ "OpcUaPublish: more than {Capacity} distinct nodes written while a rebuild plan is in flight — later new nodes are not held for replay (their first value lands only if the driver republishes it)",
+ _deferred.Capacity);
+ }
+
+ /// Re-applies the deferred writes through the sink: values, then status-only annotations, then alarm
+ /// states and alarm qualities. Each pass isolated like the live handlers; not re-metered (counted on arrival).
+ private void ReplayDeferredWrites()
+ {
+ if (_deferred.IsEmpty) return;
+ var (values, statuses) = _deferred.VariableWrites();
+ if (values.Count > 0)
+ {
+ try { _sink.WriteValues(values); }
+ catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying {Count} deferred value writes threw", values.Count); }
+ }
+
+ if (statuses.Count > 0)
+ {
+ try { _sink.WriteStatuses(statuses); }
+ catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying {Count} deferred status writes threw", statuses.Count); }
+ }
+
+ foreach (var alarm in _deferred.Alarms)
+ {
+ try { _sink.WriteAlarmCondition(alarm.AlarmNodeId, alarm.State, alarm.TimestampUtc, alarm.Realm); }
+ catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying a deferred alarm write threw for {Node}", alarm.AlarmNodeId); }
+ }
+
+ foreach (var quality in _deferred.AlarmQualities)
+ {
+ try { _sink.WriteAlarmQuality(quality.AlarmNodeId, quality.StatusCode, quality.TimestampUtc, quality.Realm); }
+ catch (Exception ex) { _log.Warning(ex, "OpcUaPublish: replaying a deferred alarm-quality write threw for {Node}", quality.AlarmNodeId); }
+ }
+
+ _log.Debug("OpcUaPublish: replayed {Values} value, {Statuses} status and {Alarms} alarm writes deferred during a rebuild plan",
+ values.Count, statuses.Count, _deferred.AlarmCount);
+ }
+
+ ///
+ /// The latest write per (realm, id) received while a plan is in flight — see . A value
+ /// followed by a status-only write for the same node composes to the value under the new status (exactly what
+ /// the two live writes leave on the node). Bounded by distinct keys across all kinds;
+ /// beyond it a NEW key is not held (an existing key still updates), counted in .
+ ///
+ private sealed class DeferredWrites(int capacity)
+ {
+ private struct VariableEntry
+ {
+ public bool HasValue;
+ public object? Value;
+ public uint Status;
+ public DateTime TimestampUtc;
+ }
+
+ private readonly Dictionary<(AddressSpaceRealm, string), VariableEntry> _variables = new();
+ private readonly Dictionary<(AddressSpaceRealm, string), AlarmStateUpdate> _alarms = new();
+ private readonly Dictionary<(AddressSpaceRealm, string), AlarmQualityUpdate> _qualities = new();
+
+ public int Capacity { get; } = capacity;
+ public long Dropped { get; private set; }
+ public int Count => _variables.Count + _alarms.Count + _qualities.Count;
+ public bool IsEmpty => Count == 0;
+ public int AlarmCount => _alarms.Count + _qualities.Count;
+ public IEnumerable Alarms => _alarms.Values;
+ public IEnumerable AlarmQualities => _qualities.Values;
+
+ public bool Value(in ValueWrite write)
+ {
+ if (string.IsNullOrEmpty(write.NodeId)) return true;
+ var key = (write.Realm, write.NodeId);
+ if (!_variables.ContainsKey(key) && !HasRoom()) return false;
+ _variables[key] = new VariableEntry
+ {
+ HasValue = true, Value = write.Value, Status = write.StatusCode, TimestampUtc = write.SourceTimestampUtc,
+ };
+ return true;
+ }
+
+ public bool Status(in StatusWrite write)
+ {
+ if (string.IsNullOrEmpty(write.NodeId)) return true;
+ var key = (write.Realm, write.NodeId);
+ if (_variables.TryGetValue(key, out var entry))
+ {
+ entry.Status = write.StatusCode;
+ entry.TimestampUtc = write.SourceTimestampUtc;
+ _variables[key] = entry;
+ return true;
+ }
+
+ if (!HasRoom()) return false;
+ _variables[key] = new VariableEntry { Status = write.StatusCode, TimestampUtc = write.SourceTimestampUtc };
+ return true;
+ }
+
+ public bool Alarm(AlarmStateUpdate msg) => Put(_alarms, (msg.Realm, msg.AlarmNodeId), msg);
+
+ public bool AlarmQuality(AlarmQualityUpdate msg) => Put(_qualities, (msg.Realm, msg.AlarmNodeId), msg);
+
+ public (List Values, List Statuses) VariableWrites()
+ {
+ var values = new List();
+ var statuses = new List();
+ foreach (var ((realm, id), e) in _variables)
+ {
+ if (e.HasValue) values.Add(new ValueWrite(id, e.Value, e.Status, e.TimestampUtc, realm));
+ else statuses.Add(new StatusWrite(id, e.Status, e.TimestampUtc, realm));
+ }
+
+ return (values, statuses);
+ }
+
+ public void Clear()
+ {
+ _variables.Clear();
+ _alarms.Clear();
+ _qualities.Clear();
+ Dropped = 0;
+ }
+
+ private bool Put(Dictionary<(AddressSpaceRealm, string), T> map, (AddressSpaceRealm, string) key, T value)
+ {
+ if (string.IsNullOrEmpty(key.Item2)) return true;
+ if (!map.ContainsKey(key) && !HasRoom()) return false;
+ map[key] = value;
+ return true;
+ }
+
+ private bool HasRoom()
+ {
+ if (Count < Capacity) return true;
+ Dropped++;
+ return false;
+ }
}
///
@@ -631,6 +919,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
_log.Error(ex, "OpcUaPublish: sink.CompleteAddressSpaceApply threw (correlation={Correlation})", msg.Correlation);
}
+ // Land what arrived while the plan was in flight on the nodes the apply just created — before the answer,
+ // so "applied" means the deferred first values are on the address space too (review fix C1).
+ ReplayDeferredWrites();
+
if (msg.ReportOutcome)
requester.Tell(new RebuildCompleted(msg.Correlation, msg.DeploymentId, outcome.Degraded, outcome.Reason));
}
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/StaleMonitoredNodeWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/StaleMonitoredNodeWireTests.cs
index 6cacc60e..6893fc2e 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/StaleMonitoredNodeWireTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/StaleMonitoredNodeWireTests.cs
@@ -139,6 +139,10 @@ public sealed class StaleMonitoredNodeWireTests
"the new item must receive a value written after its creation");
lock (gate) newValues.ShouldNotContain(v => v.StatusCode == StatusCodes.BadNodeIdUnknown);
+ // Review fix I3: the cache entry the new item holds is ITS reference alone — the old item's was released
+ // by the removal.
+ CacheRefCount(nm, nodeId).ShouldBe(1, "one live item, one component-cache reference");
+
// The old item was told, and the dead node's fan-out is gone.
await WaitUntilAsync(() => { lock (gate) return oldValues.Any(v => v.StatusCode == StatusCodes.BadNodeIdUnknown); }, TimeSpan.FromSeconds(5),
"the item on the removed node must be told BadNodeIdUnknown");
@@ -150,9 +154,18 @@ public sealed class StaleMonitoredNodeWireTests
(oldItem.Status.Error is null || ServiceResult.IsGood(oldItem.Status.Error)).ShouldBeTrue(
$"deleting the orphaned item must succeed, got {oldItem.Status.Error}");
+ // The SDK releases a deleted item's cache reference on a Good delete; the old item's was released by the
+ // removal already, so a second release would evict the NEW item's live entry (review fix I3).
+ CacheRefCount(nm, nodeId).ShouldBe(1, "deleting the orphaned item must not take the new item's reference");
+
// And the new item is unaffected by that delete.
nm.WriteValue(Tag, 44, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
await WaitUntilAsync(() => { lock (gate) return newValues.Any(v => Equals(v.Value, 44)); }, TimeSpan.FromSeconds(5));
+
+ // Positive control: deleting the live item gives its reference back.
+ subscription.RemoveItem(newItem);
+ await subscription.ApplyChangesAsync(ct);
+ CacheRefCount(nm, nodeId).ShouldBe(0);
}
finally
{
@@ -160,6 +173,28 @@ public sealed class StaleMonitoredNodeWireTests
}
}
+ /// The SDK's component-cache reference count for (private SDK state, read by
+ /// reflection: CustomNodeManager2.m_componentCache → CacheEntry.RefCount); 0 when no entry.
+ private static int CacheRefCount(OtOpcUaNodeManager nm, NodeId nodeId)
+ {
+ const System.Reflection.BindingFlags Any =
+ System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.Public | System.Reflection.BindingFlags.NonPublic;
+ var field = typeof(Opc.Ua.Server.CustomNodeManager2).GetField("m_componentCache", Any)
+ .ShouldNotBeNull("the SDK's component cache field (SDK internals moved?)");
+ lock (nm.Lock)
+ {
+ if (field.GetValue(nm) is not System.Collections.IDictionary cache || !cache.Contains(nodeId)) return 0;
+ var entry = cache[nodeId]!;
+ var member = (object?)entry.GetType().GetField("RefCount", Any) ?? entry.GetType().GetProperty("RefCount", Any);
+ return member switch
+ {
+ System.Reflection.FieldInfo f => (int)f.GetValue(entry)!,
+ System.Reflection.PropertyInfo p => (int)p.GetValue(entry)!,
+ _ => throw new InvalidOperationException("CacheEntry.RefCount not found"),
+ };
+ }
+ }
+
private static void Materialise(OtOpcUaNodeManager nm)
{
nm.EnsureFolder(Device, null, "dev1", AddressSpaceRealm.Raw);
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/History/HistoryUpdateTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/History/HistoryUpdateTests.cs
index da4f1a57..1ddf9a28 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/History/HistoryUpdateTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/History/HistoryUpdateTests.cs
@@ -254,6 +254,38 @@ public sealed class HistoryUpdateTests : IDisposable
finally { await host.DisposeAsync(); }
}
+ ///
+ /// Review fix I4 — the override's OWN per-request updater check, which the bit retraction above now hides.
+ /// It is the B7 race window: a request the SDK base admitted on the HistoryWrite bit just before the updater
+ /// was cleared. Reconstructed by restoring the bit after the clear, so the request reaches the override with
+ /// no updater; it must answer unsupported on both slots (the override sets results itself) and never reach the
+ /// old updater.
+ ///
+ [Fact]
+ public async Task The_override_rechecks_the_updater_per_request_when_the_bit_is_still_set()
+ {
+ var (host, server) = await BootAsync();
+ try
+ {
+ var nm = server.NodeManager!;
+ var updater = new FakeUpdater();
+ nm.HistorianDataUpdater = updater;
+ WireAcl(nm, enabled: false, NodePermissions.HistoryUpdate);
+ var nodeId = Historized(nm, AddressSpaceRealm.Raw, RawPath);
+ nm.HistorianDataUpdater = null;
+ var variable = nm.TryGetVariable(RawPath, AddressSpaceRealm.Raw)!;
+ variable.AccessLevel |= AccessLevels.HistoryWrite;
+ variable.UserAccessLevel |= AccessLevels.HistoryWrite;
+
+ var (results, errors) = Insert(nm, KeyIdentity(OpcUaDataPlaneRoles.HistoryUpdate), nodeId, Value(T1, 1d));
+
+ errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
+ results[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported, "set by the override, not left for the master node manager");
+ updater.UpdateCalls.ShouldBeEmpty();
+ }
+ finally { await host.DisposeAsync(); }
+ }
+
/// With an updater wired, both the raw node and its UNS mirror carry HistoryWrite; a non-historized node does not.
[Fact]
public async Task An_updater_grants_history_write_to_raw_and_uns_historized_nodes_only()
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadBoundsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadBoundsTests.cs
index 74096a4f..4875ac61 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadBoundsTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadBoundsTests.cs
@@ -321,6 +321,54 @@ public sealed class NodeManagerHistoryReadBoundsTests(BoundsServerFixture fx) :
cpWhole.ShouldBeNull();
}
+ ///
+ /// Lane B review fix round 1 (minor): an UNCAPPED read (N = 0) against a backend with a row ceiling L pages at
+ /// L, and MaxReturnDataValues advertises L — so no page, the tie-cluster slices and their end bound included,
+ /// may carry more than L values; and the pages still concatenate to the uncapped single-shot answer. Swept
+ /// over ceilings and tie-cluster sizes up to what one over-fetch read can hold (L − 1).
+ ///
+ [Fact]
+ public async Task An_uncapped_read_never_returns_a_page_larger_than_the_backend_ceiling()
+ {
+ foreach (var ceiling in new[] { 2, 3, 4, 6 })
+ {
+ for (var ties = 1; ties < ceiling; ties++)
+ {
+ var series = new List { Snap("5:00", 0) };
+ for (var i = 0; i < ties; i++) series.Add(Snap("5:02", 10 + i));
+ series.Add(Snap("5:03", 3));
+ for (var i = 0; i < ties; i++) series.Add(Snap("5:05", 50 + i));
+ series.Add(Snap("5:06", 6));
+
+ foreach (var (start, end) in new[] { ("5:01", "5:04"), ("5:00", "5:06"), ("5:01", "5:07"), ("5:06", "5:01") })
+ {
+ var expected = Render(
+ (await ReadAsync(new SeriesSource(series, reverse: true, inclusiveForwardEnd: false, resolution: 1), T(start), T(end), 0, true, null)).Values,
+ T(start), T(end));
+
+ var capped = new SeriesSource(series, reverse: true, inclusiveForwardEnd: false, resolution: 1, ceiling: ceiling);
+ var all = new List();
+ byte[]? cp = null;
+ var pages = 0;
+ do
+ {
+ var (values, error, next) = await ReadAsync(capped, T(start), T(end), 0, true, cp);
+ error.StatusCode.Code.ShouldBe(StatusCodes.Good, $"L={ceiling} ties={ties} {start}->{end}");
+ values.Count.ShouldBeLessThanOrEqualTo(ceiling, $"L={ceiling} ties={ties} {start}->{end} page {pages}");
+ all.AddRange(values);
+ cp = next;
+ (++pages).ShouldBeLessThan(60, "paging must terminate");
+ }
+ while (cp is not null);
+
+ Render(all, T(start), T(end)).ShouldBe(expected, $"L={ceiling} ties={ties} {start}->{end}");
+ }
+ }
+ }
+
+ static DataValueSnapshot Snap(string at, double v) => new(v, StatusCodes.Good, T(at), T(at));
+ }
+
///
/// #691 review: no data AND no bound found ⇒ GoodNoData with no values — the answer a tag with no history (and a
/// disabled historian) gave before bounds (§6.5.3.2: bounds make the result Good only "if one or both exist"). Only
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsTests.cs
index 1984fdb8..3df484bc 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadEventsTests.cs
@@ -249,6 +249,45 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
await host.DisposeAsync();
}
+ ///
+ /// Review fix I4 — the HistoryReadEvents override's OWN source guard. Every other "unsupported" case here is
+ /// answered by the SDK base, which admits only notifiers carrying the EventNotifier.HistoryRead bit —
+ /// and since B7 the bit and the source registration move together. So the guard is reached only by a request
+ /// the base admitted on the bit just before a source reset took the registration back; reconstructed by
+ /// restoring the bit after the reset. The override must answer unsupported on both slots and reach no source.
+ ///
+ [Fact]
+ public async Task The_event_override_refuses_a_notifier_with_the_bit_but_no_registered_source()
+ {
+ var (host, server) = await BootAsync();
+ var nm = server.NodeManager!;
+ var fake = new RecordingHistorianDataSource();
+ nm.HistorianDataSource = fake;
+ nm.EnsureFolder("eq-guard", parentNodeId: null, displayName: "Equipment", realm: AddressSpaceRealm.Uns);
+ nm.MaterialiseAlarmCondition("alarm-g", "eq-guard", "Cond", "OffNormalAlarm", severity: 200, realm: AddressSpaceRealm.Uns);
+ nm.IsEventHistorySource("eq-guard").ShouldBeTrue("non-vacuity: registered while the source was wired");
+
+ nm.HistorianDataSource = null!;
+ var folder = nm.TryGetFolder("eq-guard")!;
+ nm.IsEventHistorySource("eq-guard").ShouldBeFalse();
+ folder.EventNotifier |= EventNotifiers.HistoryRead;
+
+ var details = new ReadEventDetails
+ {
+ StartTime = DateTime.UtcNow.AddHours(-1),
+ EndTime = DateTime.UtcNow,
+ NumValuesPerNode = 10,
+ Filter = SelectFilter("EventId"),
+ };
+ var (results, errors) = InvokeHistoryRead(server, nm, details, folder.NodeId);
+
+ errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
+ results[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported, "set by the override's guard");
+ fake.EventsCalled.ShouldBeFalse();
+
+ await host.DisposeAsync();
+ }
+
/// A variable node targeted by a HistoryReadEvents request ⇒ BadHistoryOperationUnsupported;
/// the source is never invoked.
///
@@ -259,8 +298,8 @@ public sealed class NodeManagerHistoryReadEventsTests : IDisposable
/// SDK base therefore rejects it and does NOT pass it to our HistoryReadEvents override;
/// the Bad result comes from the base's pre-seeding, not from our source-guard. This test pins
/// that base-level rejection of variable nodes for event reads.
- /// The override's own source-guard (miss in _eventNotifierSources) is exercised by the
- /// Events_folder_promoted_without_source_is_unsupported_until_a_source_is_wired test instead.
+ /// The override's own source-guard (a miss in _eventNotifierSources) is exercised by
+ /// The_event_override_refuses_a_notifier_with_the_bit_but_no_registered_source.
///
///
[Fact]
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerLockTimeBenchmarkTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerLockTimeBenchmarkTests.cs
index 848529f3..865bef48 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerLockTimeBenchmarkTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerLockTimeBenchmarkTests.cs
@@ -20,6 +20,11 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
///
/// The wall-clock budgets are generous for CI hardware; the before/after numbers are in the commit
/// message. P3 is pinned by the bookkeeping's own visit counter, which does not depend on the machine.
+/// Kept in the default run on purpose (not a Benchmark category the CI filter would have to learn to exclude):
+/// measured after the fix at 5–6 ms (P1) and 58–64 ms (P2) on a loaded dev machine against the 1 s budget —
+/// ≥ 15× headroom, so it does not flake on slow CI. The flip side: the pre-fix code (75–150 ms / 470–680 ms) also
+/// fits 1 s at these sizes, so the wall-clock legs guard against gross (order-of-magnitude) regressions only; the
+/// machine-independent P3 visit counter is the strict pin.
///
[Trait("Category", "Benchmark")]
public sealed class NodeManagerLockTimeBenchmarkTests(ITestOutputHelper output) : IDisposable
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStaleMonitoredNodeTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStaleMonitoredNodeTests.cs
index fcd966e5..5ddd6b71 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStaleMonitoredNodeTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStaleMonitoredNodeTests.cs
@@ -93,6 +93,39 @@ public sealed class NodeManagerStaleMonitoredNodeTests : IDisposable
finally { await host.DisposeAsync(); }
}
+ ///
+ /// Review fix round 1 (minor): the same-id kind swap (scripted ⇄ native) in MaterialiseAlarmCondition
+ /// drops the old condition instance — one more removal path, so it tombstones like the others. Believed
+ /// unreachable from a composition (native ids are RawPaths in Raw, scripted ids live in Uns), so it is driven
+ /// directly.
+ ///
+ [Fact]
+ public async Task A_condition_kind_swap_tells_and_forgets_items_on_the_dropped_instance()
+ {
+ var (host, server) = await BootAsync();
+ try
+ {
+ var nm = server.NodeManager!;
+ nm.EnsureFolder("eq-1", null, "Equipment", AddressSpaceRealm.Uns);
+ nm.MaterialiseAlarmCondition("al-1", "eq-1", "High", "OffNormalAlarm", 500, AddressSpaceRealm.Uns, isNative: false);
+ var nodeId = new NodeId("al-1", nm.TryGetFolder("eq-1")!.NodeId.NamespaceIndex);
+ var handle = (NodeHandle)nm.GetManagerHandle(nodeId);
+ handle.ShouldNotBeNull("non-vacuity: the condition is materialised");
+ var item = new StubItem(handle, Identity());
+ nm.TrackLiveMonitoredItem(handle, item);
+ nm.Authorizer = Authorizer(permits: false);
+ nm.RevalidateMonitoredItems();
+ item.Statuses.ShouldContain((StatusCode)StatusCodes.BadUserAccessDenied, "non-vacuity: stood down");
+
+ nm.MaterialiseAlarmCondition("al-1", "eq-1", "High", "OffNormalAlarm", 500, AddressSpaceRealm.Uns, isNative: true);
+
+ item.Statuses.ShouldContain((StatusCode)StatusCodes.BadNodeIdUnknown, "the dropped instance's item is told");
+ nm.LiveItemCount.ShouldBe(0);
+ ((NodeHandle)nm.GetManagerHandle(nodeId)).ShouldNotBeNull("positive control: the swapped-in condition exists");
+ }
+ finally { await host.DisposeAsync(); }
+ }
+
/// Registers a stub item on and stands it down through a denying pass.
private static async Task<(StubItem Item, NodeId NodeId)> StandDownAsync(OtOpcUaNodeManager nm, string rawPath)
{
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorDeferredWriteTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorDeferredWriteTests.cs
new file mode 100644
index 00000000..7b795e3d
--- /dev/null
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorDeferredWriteTests.cs
@@ -0,0 +1,438 @@
+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.Core.Authorization;
+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;
+
+///
+/// Lane B review fix round 1. C1: the plan runs off the pinned thread (R02:P5) while writes are non-creating
+/// (R01:L1), so a value, status or alarm write for a node the IN-FLIGHT plan adds used to be skipped — and a
+/// change-driven driver never republishes it. The actor now remembers the latest write per node while a plan
+/// is in flight and replays it after the apply. Every test holds the plan open with a gate, so "during the
+/// plan" is deterministic, and runs against a sink that — like the real node manager — drops writes for nodes
+/// that do not exist.
+/// Also here: I2 (an authority parse throw never fails the apply), the bytes-less request queued behind an
+/// in-flight plan, and a plan piped back to a restarted actor.
+///
+public sealed class OpcUaPublishActorDeferredWriteTests : RuntimeActorTestBase
+{
+ private const string NewTag = "Plant/Modbus/dev1/T1";
+ private static readonly TimeSpan PlanBudget = TimeSpan.FromSeconds(30);
+
+ [Fact]
+ public void A_value_for_a_node_the_in_flight_plan_adds_lands_on_that_node_after_the_apply()
+ {
+ var sink = new ExistenceSink();
+ using var gate = new ManualResetEventSlim(false);
+ var actor = Spawn(sink, gate);
+
+ actor.Tell(Rebuild(RawTagArtifact(3)));
+ var at = DateTime.UtcNow;
+ actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(
+ [new ValueWrite(NewTag, 7, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)]));
+ AwaitAssert(() => sink.SkippedWrites.ShouldContain(NewTag), PlanBudget); // non-vacuity: it really missed
+
+ gate.Set();
+ ExpectMsg(PlanBudget).Degraded.ShouldBeFalse();
+
+ sink.Exists(NewTag).ShouldBeTrue("the plan materialised the node");
+ sink.ValueOf(NewTag).ShouldBe((7, OpcUaStatus.Good, at), "the value sent while the plan was in flight must end up on the node");
+ }
+
+ [Fact]
+ public void A_status_after_a_value_composes_onto_the_new_node_in_order()
+ {
+ var sink = new ExistenceSink();
+ using var gate = new ManualResetEventSlim(false);
+ var actor = Spawn(sink, gate);
+
+ actor.Tell(Rebuild(RawTagArtifact(3)));
+ var t1 = DateTime.UtcNow;
+ var t2 = t1.AddSeconds(1);
+ actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(NewTag, 7, OpcUaStatus.Good, t1, AddressSpaceRealm.Raw)]));
+ actor.Tell(new OpcUaPublishActor.AttributeStatusBatchUpdate([new StatusWrite(NewTag, OpcUaStatus.BadNoCommunication, t2, AddressSpaceRealm.Raw)]));
+ gate.Set();
+ ExpectMsg(PlanBudget);
+
+ sink.ValueOf(NewTag).ShouldBe((7, OpcUaStatus.BadNoCommunication, t2), "last known value under the later status");
+ }
+
+ [Fact]
+ public void Existing_nodes_keep_updating_while_the_plan_computes()
+ {
+ var sink = new ExistenceSink();
+ using var gate = new ManualResetEventSlim(false);
+ var actor = Spawn(sink, gate);
+ actor.Tell(Rebuild(RawTagArtifact(1)));
+ gate.Set();
+ ExpectMsg(PlanBudget);
+ var existing = "Plant/Modbus/dev1/T0";
+ sink.Exists(existing).ShouldBeTrue();
+
+ gate.Reset();
+ actor.Tell(Rebuild(RawTagArtifact(3)));
+ var at = DateTime.UtcNow;
+ actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(existing, 5, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)]));
+
+ // Applied immediately, while the plan is still held open.
+ AwaitAssert(() => sink.ValueOf(existing).ShouldBe((5, OpcUaStatus.Good, at)), PlanBudget);
+ sink.Exists(NewTag).ShouldBeFalse("the plan is still in flight");
+
+ gate.Set();
+ ExpectMsg(PlanBudget);
+ sink.ValueOf(existing).ShouldBe((5, OpcUaStatus.Good, at));
+ }
+
+ [Fact]
+ public void An_alarm_state_sent_during_the_plan_lands_on_the_condition_it_materialises()
+ {
+ var sink = new ExistenceSink();
+ using var gate = new ManualResetEventSlim(false);
+ var actor = Spawn(sink, gate);
+
+ actor.Tell(Rebuild(AlarmArtifact()));
+ var state = new AlarmConditionSnapshot(Active: true, Acknowledged: false, Confirmed: true, Enabled: true, AlarmShelvingKind.Unshelved, 500, "high");
+ actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("al-1", state, DateTime.UtcNow, AddressSpaceRealm.Uns));
+ actor.Tell(new OpcUaPublishActor.AlarmQualityUpdate("al-1", OpcUaStatus.BadNoCommunication, DateTime.UtcNow, AddressSpaceRealm.Uns));
+ AwaitAssert(() => sink.SkippedAlarms.ShouldContain("al-1"), PlanBudget);
+
+ gate.Set();
+ ExpectMsg(PlanBudget).Degraded.ShouldBeFalse();
+
+ sink.ConditionExists("al-1").ShouldBeTrue("the plan materialised the condition");
+ sink.AlarmStateOf("al-1").ShouldBeSameAs(state);
+ sink.AlarmQualityOf("al-1").ShouldBe(OpcUaStatus.BadNoCommunication);
+ }
+
+ [Fact]
+ public void The_deferred_buffer_is_bounded_under_a_flood_and_keeps_the_earliest_keys()
+ {
+ var sink = new ExistenceSink();
+ using var gate = new ManualResetEventSlim(false);
+ var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+ sink, new AddressSpaceApplier(sink, NullLogger.Instance),
+ new OpcUaPublishActor.TestHooks { BeforePlan = () => gate.Wait() }, deferredWriteCapacity: 2));
+
+ actor.Tell(Rebuild(RawTagArtifact(5)));
+ var at = DateTime.UtcNow;
+ actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(Enumerable.Range(0, 5)
+ .Select(i => new ValueWrite($"Plant/Modbus/dev1/T{i}", i, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)).ToList()));
+ // An existing key still updates when the buffer is full.
+ actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite("Plant/Modbus/dev1/T0", 100, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)]));
+ AwaitAssert(() => sink.SkippedWrites.Count.ShouldBe(6), PlanBudget);
+
+ gate.Set();
+ ExpectMsg(PlanBudget);
+
+ sink.ValueOf("Plant/Modbus/dev1/T0")!.Value.Value.ShouldBe(100);
+ sink.ValueOf("Plant/Modbus/dev1/T1")!.Value.Value.ShouldBe(1);
+ for (var i = 2; i < 5; i++)
+ sink.ValueOf($"Plant/Modbus/dev1/T{i}").ShouldBeNull("beyond the bound a new key is not held");
+ sink.ReplayedItems.ShouldBe(2);
+ }
+
+ [Fact]
+ public void Writes_after_the_pipeline_drains_are_not_replayed_again()
+ {
+ var sink = new ExistenceSink();
+ using var gate = new ManualResetEventSlim(false);
+ var actor = Spawn(sink, gate);
+ actor.Tell(Rebuild(RawTagArtifact(3)));
+ actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(NewTag, 7, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw)]));
+ gate.Set();
+ ExpectMsg(PlanBudget);
+ var replayed = sink.ReplayedItems;
+ replayed.ShouldBe(1);
+
+ // A later plan replays only what arrived during IT.
+ actor.Tell(Rebuild(RawTagArtifact(3)));
+ ExpectMsg(PlanBudget);
+ sink.ReplayedItems.ShouldBe(replayed);
+ }
+
+ [Fact]
+ public void A_throwing_authority_parse_does_not_fail_the_apply()
+ {
+ var sink = new ExistenceSink();
+ var authority = new RecordingAuthoritySink();
+ var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+ sink, new AddressSpaceApplier(sink, NullLogger.Instance),
+ new OpcUaPublishActor.TestHooks { ParseAuthority = (_, _) => throw new InvalidOperationException("bad grants") },
+ dataPlaneAuthority: authority));
+
+ actor.Tell(Rebuild(RawTagArtifact(2)));
+
+ var completed = ExpectMsg(PlanBudget);
+ completed.Degraded.ShouldBeFalse(completed.DegradedReason);
+ sink.Exists("Plant/Modbus/dev1/T1").ShouldBeTrue("the address space is applied");
+ authority.Applies.ShouldBe(0, "nothing is installed from an authority that could not be read");
+
+ // Positive control: the same actor installs an authority that CAN be read.
+ var healthy = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+ new ExistenceSink(), new AddressSpaceApplier(new ExistenceSink(), NullLogger.Instance),
+ hooks: null, dataPlaneAuthority: authority));
+ healthy.Tell(Rebuild(RawTagArtifact(2)));
+ ExpectMsg(PlanBudget).Degraded.ShouldBeFalse();
+ authority.Applies.ShouldBe(1);
+ }
+
+ [Fact]
+ public void A_bytes_less_rebuild_arriving_mid_plan_answers_after_the_plan_ahead_of_it()
+ {
+ var sink = new ExistenceSink();
+ using var gate = new ManualResetEventSlim(false);
+ var actor = Spawn(sink, gate);
+
+ var first = Rebuild(RawTagArtifact(2));
+ var second = new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), Artifact: null, ReportOutcome: true);
+ actor.Tell(first);
+ actor.Tell(second);
+ ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // the bytes-less one must not overtake the held plan
+
+ gate.Set();
+ var a = ExpectMsg(PlanBudget);
+ var b = ExpectMsg(PlanBudget);
+ a.Correlation.ShouldBe(first.Correlation);
+ a.Degraded.ShouldBeFalse();
+ b.Correlation.ShouldBe(second.Correlation);
+ b.Degraded.ShouldBeTrue();
+ }
+
+ [Fact]
+ public void A_plan_piped_back_to_a_restarted_actor_is_not_applied()
+ {
+ var sink = new ExistenceSink();
+ using var gate = new ManualResetEventSlim(false);
+ var applier = new AddressSpaceApplier(sink, NullLogger.Instance);
+ var childProps = OpcUaPublishActor.PropsForTests(sink, applier, new OpcUaPublishActor.TestHooks { BeforePlan = () => gate.Wait() });
+ var parent = Sys.ActorOf(Props.Create(() => new RestartingParent(childProps)));
+ parent.Tell(RestartingParent.GetChild.Instance);
+ var child = ExpectMsg();
+
+ var stale = Rebuild(RawTagArtifact(3));
+ child.Tell(stale);
+ child.Tell(Kill.Instance); // the parent's strategy RESTARTS on any failure, ActorKilledException included
+ ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+
+ gate.Set();
+ var answer = ExpectMsg(PlanBudget);
+ answer.Correlation.ShouldBe(stale.Correlation);
+ answer.Degraded.ShouldBeTrue("the requester is still answered");
+ sink.Exists(NewTag).ShouldBeFalse("a plan from the previous incarnation must not be applied");
+
+ // Positive control: the restarted instance applies a fresh rebuild.
+ child.Tell(Rebuild(RawTagArtifact(3)));
+ ExpectMsg(PlanBudget).Degraded.ShouldBeFalse();
+ sink.Exists(NewTag).ShouldBeTrue();
+ }
+
+ // ───────────────────────────── helpers ─────────────────────────────
+
+ private IActorRef Spawn(ExistenceSink sink, ManualResetEventSlim gate) =>
+ Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+ sink, new AddressSpaceApplier(sink, NullLogger.Instance),
+ new OpcUaPublishActor.TestHooks { BeforePlan = () => gate.Wait() }));
+
+ private static OpcUaPublishActor.RebuildAddressSpace Rebuild(byte[] artifact) =>
+ new(CorrelationId.NewId(), DeploymentId.NewId(), artifact, ReportOutcome: true);
+
+ private static byte[] RawTagArtifact(int tagCount) =>
+ JsonSerializer.SerializeToUtf8Bytes(new
+ {
+ UnsAreas = Array.Empty