From f966efcfad1832b9816fce17f5387aeac574097f Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 14:30:11 -0400 Subject: [PATCH] fix(opcua): replay writes deferred during an off-thread plan; cache-ref and guard fixes (Lane B fix round 1) C1 (critical): B6 computes the plan off the pinned thread while B1 made writes non-creating, so a value/status/alarm write for a node the in-flight plan adds was skipped, and change-driven drivers never republish it (BadWaitingForInitialData indefinitely; on a full rebuild every re-subscribe value landed on discarded nodes). OpcUaPublishActor now applies writes immediately AND remembers the latest per (realm, id) while a plan is in flight or queued (value+status compose; alarm state/quality coalesce to the latest, which the node's delta gate makes inert on an existing condition), replays it after each apply before answering, and drops it once the pipeline is idle. Bounded (250k keys by default; beyond it new keys are not held, logged once). I2: the data-plane-authority parse runs in its own try inside the plan task; a throw is logged and never degrades the address-space apply. I3: the SDK's DeleteMonitoredItems releases an item's component-cache reference on a Good delete; a tombstoned item's reference was already released, so the delete took one from the re-created node's entry. Released items are remembered and the orphan delete takes one back first. Event items never held a reference (EventManager + SubscribeToEvents), so the three event-item releases are gone. I4: direct tests for the HistoryUpdate override's per-request updater check and the HistoryReadEvents override's source guard (both reached by restoring the bit after a reset - the B7 race window); fixed the false remark. Minors: the MaterialiseAlarmCondition kind-swap tombstones the dropped instance; a bytes-less rebuild arriving mid-plan queues behind it (replies in order); a PlanReady/PlanFailed from a previous actor incarnation is ignored and its requester answered degraded; the tie-cluster end bound of an uncapped read fits against the backend ceiling (defence in depth - guarded by a sweep test); the lock-time benchmark stays in the default run, headroom noted. Refs #732 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../OtOpcUaNodeManager.cs | 49 +- .../OpcUa/OpcUaPublishActor.cs | 324 ++++++++++++- .../StaleMonitoredNodeWireTests.cs | 35 ++ .../History/HistoryUpdateTests.cs | 32 ++ .../NodeManagerHistoryReadBoundsTests.cs | 48 ++ .../NodeManagerHistoryReadEventsTests.cs | 43 +- .../NodeManagerLockTimeBenchmarkTests.cs | 5 + .../NodeManagerStaleMonitoredNodeTests.cs | 33 ++ .../OpcUaPublishActorDeferredWriteTests.cs | 438 ++++++++++++++++++ 9 files changed, 977 insertions(+), 30 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorDeferredWriteTests.cs 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(), + UnsLines = Array.Empty(), + Equipment = Array.Empty(), + RawFolders = new[] { new { RawFolderId = "rf-1", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } }, + DriverInstances = new[] { new { DriverInstanceId = "drv-1", DriverType = "Modbus", DriverConfig = "{}", Name = "Modbus", RawFolderId = "rf-1", ClusterId = "c1" } }, + Devices = new[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } }, + TagGroups = Array.Empty(), + Tags = Enumerable.Range(0, tagCount).Select(i => new + { + TagId = $"t-{i}", DeviceId = "dev-1", TagGroupId = (string?)null, Name = $"T{i}", + DataType = "Float", AccessLevel = 0, TagConfig = "{}", + }).ToArray(), + UnsTagReferences = Array.Empty(), + ScriptedAlarms = Array.Empty(), + }); + + private static byte[] AlarmArtifact() => + JsonSerializer.SerializeToUtf8Bytes(new + { + UnsAreas = new[] { new { UnsAreaId = "area-1", ClusterId = "c1", Name = "Area" } }, + UnsLines = new[] { new { UnsLineId = "line-1", UnsAreaId = "area-1", Name = "Line" } }, + Equipment = new[] + { + new { EquipmentId = "eq-1", DriverInstanceId = (string?)null, UnsLineId = "line-1", Name = "Eq", MachineCode = "EQ-1" }, + }, + DriverInstances = Array.Empty(), + Tags = Array.Empty(), + ScriptedAlarms = new[] + { + new + { + ScriptedAlarmId = "al-1", EquipmentId = "eq-1", Name = "High", AlarmType = "AlarmCondition", + Severity = 500, MessageTemplate = "high", PredicateScriptId = "s-1", HistorizeToAveva = false, + Retain = true, Enabled = true, + }, + }, + Scripts = new[] { new { ScriptId = "s-1", SourceCode = "return false;" } }, + }); + + /// Restarts its one child on any failure, and lets the test reach the child. + private sealed class RestartingParent : ReceiveActor + { + public sealed class GetChild { public static readonly GetChild Instance = new(); } + + public RestartingParent(Props childProps) + { + var child = Context.ActorOf(childProps, "publish"); + Receive(_ => Sender.Tell(child)); + } + + protected override SupervisorStrategy SupervisorStrategy() => + new OneForOneStrategy(_ => Directive.Restart); + } + + /// An authority sink that counts installs. + private sealed class RecordingAuthoritySink : IDataPlaneAuthoritySink + { + private int _applies; + public int Applies => Volatile.Read(ref _applies); + + public DataPlaneAuthoritySnapshot Apply( + string? clusterId, string revision, IReadOnlyList keys, + IReadOnlyList grants, + IReadOnlyDictionary> scopePaths) + { + Interlocked.Increment(ref _applies); + return new DataPlaneAuthorityState().Apply(clusterId, revision, keys, grants, scopePaths); + } + } + + /// + /// Models the node manager's non-creating writes: a write to a node that was never ensured (or was removed) + /// is skipped and recorded; a full sink rebuild forgets every node. counts the + /// items written after an apply closed and before the next apply starts — the actor's replay, since the + /// tests send nothing between releasing a plan and its answer. + /// + private sealed class ExistenceSink : IOpcUaAddressSpaceSink, ISurgicalAddressSpaceSink + { + private readonly object _gate = new(); + private readonly HashSet _variables = new(); + private readonly HashSet _conditions = new(); + private readonly Dictionary _values = new(); + private readonly Dictionary _alarmStates = new(); + private readonly Dictionary _alarmQualities = new(); + + public ConcurrentQueue SkippedWrites { get; } = new(); + public ConcurrentQueue SkippedAlarms { get; } = new(); + public int ReplayedItems; + + public bool Exists(string id) { lock (_gate) return _variables.Contains(id); } + public bool ConditionExists(string id) { lock (_gate) return _conditions.Contains(id); } + public (object? Value, uint Status, DateTime At)? ValueOf(string id) { lock (_gate) return _values.TryGetValue(id, out var v) ? v : null; } + public AlarmConditionSnapshot? AlarmStateOf(string id) { lock (_gate) return _alarmStates.GetValueOrDefault(id); } + public uint? AlarmQualityOf(string id) { lock (_gate) return _alarmQualities.TryGetValue(id, out var q) ? q : null; } + + public void WriteValue(string nodeId, object? value, uint statusCode, DateTime ts, AddressSpaceRealm realm) => + WriteValues([new ValueWrite(nodeId, value, statusCode, ts, realm)]); + + public void WriteValues(IReadOnlyList values) + { + lock (_gate) + { + if (_applyClosed) ReplayedItems += values.Count; + foreach (var v in values) + { + if (!_variables.Contains(v.NodeId)) { SkippedWrites.Enqueue(v.NodeId); continue; } + _values[v.NodeId] = (v.Value, v.StatusCode, v.SourceTimestampUtc); + } + } + } + + public void WriteStatuses(IReadOnlyList statuses) + { + lock (_gate) + { + if (_applyClosed) ReplayedItems += statuses.Count; + foreach (var s in statuses) + { + if (!_variables.Contains(s.NodeId)) { SkippedWrites.Enqueue(s.NodeId); continue; } + var current = _values.GetValueOrDefault(s.NodeId); + _values[s.NodeId] = (current.Value, s.StatusCode, s.SourceTimestampUtc); + } + } + } + + public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime ts, AddressSpaceRealm realm) + { + lock (_gate) + { + if (!_conditions.Contains(alarmNodeId)) { SkippedAlarms.Enqueue(alarmNodeId); return; } + _alarmStates[alarmNodeId] = state; + } + } + + public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime ts, AddressSpaceRealm realm) + { + lock (_gate) + { + if (!_conditions.Contains(alarmNodeId)) { SkippedAlarms.Enqueue(alarmNodeId); return; } + _alarmQualities[alarmNodeId] = statusCode; + } + } + + public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) + { + lock (_gate) { _applyClosed = false; _conditions.Add(alarmNodeId); } + } + + public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm) + { + lock (_gate) _applyClosed = false; + } + + 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) + { + lock (_gate) { _applyClosed = false; _variables.Add(variableNodeId); } + } + + public void RebuildAddressSpace() + { + lock (_gate) + { + _applyClosed = false; + _variables.Clear(); + _conditions.Clear(); + _values.Clear(); + _alarmStates.Clear(); + _alarmQualities.Clear(); + } + } + + private bool _applyClosed; + + public void CompleteAddressSpaceApply() + { + lock (_gate) _applyClosed = true; + } + + public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm) { } + public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { } + public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { } + public bool UpdateTagAttributes(string variableNodeId, bool writable, string? historianTagname, string dataType, bool isArray, uint? arrayLength, AddressSpaceRealm realm) => true; + public bool UpdateFolderDisplayName(string folderNodeId, string displayName, AddressSpaceRealm realm) => true; + public bool RemoveVariableNode(string variableNodeId, AddressSpaceRealm realm) { lock (_gate) return _variables.Remove(variableNodeId); } + public bool RemoveAlarmConditionNode(string alarmNodeId, AddressSpaceRealm realm) { lock (_gate) return _conditions.Remove(alarmNodeId); } + public bool RemoveEquipmentSubtree(string equipmentNodeId, AddressSpaceRealm realm) => true; + } +}