diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index a49edfa9..774fc3f6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -290,6 +290,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 private volatile TimeProvider _clock = TimeProvider.System; + /// The manager's "now" (UTC) — read off so a test clock drives every timestamp this + /// manager stamps (condition / shelve / revert / revalidation times), not only the HistoryUpdate bound. + private DateTime UtcNow => _clock.GetUtcNow().UtcDateTime; + private volatile IOpcUaNodeWriteGateway _nodeWriteGateway = NullOpcUaNodeWriteGateway.Instance; /// @@ -3491,6 +3495,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { lock (Lock) { + // Review 01 L2 — every monitored item on a node this rebuild discards is told and forgotten BEFORE + // the nodes go, so a later create on the same NodeId binds to the re-created node (see + // TombstoneMonitoredNode). The shared root survives the rebuild and keeps its items. + TombstoneAllMonitoredNodes(); + foreach (var v in _variables.Values) { v.Parent?.RemoveChild(v); @@ -3602,6 +3611,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // caller (AddressSpaceApplier) falls back to a full rebuild (resync). Mirrors the per-node // teardown inside RebuildAddressSpace, scoped to this one id. if (!_variables.TryRemove(key, out var variable)) return false; + TombstoneMonitoredNode(variable.NodeId, UtcNow); // 2026-08-15 review I3: sever the INVERSE half of every cross-tree Organizes edge this node holds // BEFORE detaching it — the same hazard (and the same fix shape) as UnwireAlarmNotifiers on a // condition removal. AddReference wires the edge bidirectionally: forward on the UNS reference @@ -3668,6 +3678,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // lingering notifier folder is harmless (RemoveEquipmentSubtree demotes it when the whole // equipment goes). if (!_alarmConditions.TryRemove(key, out var condition)) return false; + TombstoneMonitoredSubtree(condition, UtcNow); // unwire this condition's extra equipment-folder notifiers bidirectionally BEFORE // the condition drops — the folders (equipment, UNS realm) SURVIVE this scoped remove, so without // this they would keep a dangling inverse-notifier entry to the removed condition. @@ -3704,11 +3715,13 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 bool InScope(RealmKey key) => key.Realm == realm && (key.Id == equipmentNodeId || key.Id.StartsWith(prefix, StringComparison.Ordinal)); + var now = UtcNow; // Value variables (+ their historized-tagname registrations). foreach (var id in _variables.Keys.Where(InScope).ToList()) { if (_variables.TryRemove(id, out var v)) { + TombstoneMonitoredNode(v.NodeId, now); // Sever the inverse half of each cross-tree Organizes edge // first, exactly as RemoveVariableNode does — a UNS reference variable in scope points at a // RAW node that survives this UNS-scoped removal and would otherwise keep a dangling edge. @@ -3725,6 +3738,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { if (_alarmConditions.TryRemove(id, out var c)) { + TombstoneMonitoredSubtree(c, now); c.Parent?.RemoveChild(c); ForgetPredefinedSubtree(c); } @@ -3765,6 +3779,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { if (_folders.TryRemove(id, out var f)) { + TombstoneMonitoredNode(f.NodeId, now); f.Parent?.RemoveChild(f); PredefinedNodes?.Remove(f.NodeId); } @@ -4080,8 +4095,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 if (unsubscribe) { - // An item RevalidateMonitoredItems stood down is ALREADY off its notifier's fan-out. - // Handing its delete to the SDK would make it remove an item that is not there — at best a + // An item RevalidateMonitoredItems stood down is ALREADY off its notifier's fan-out, and so is + // one whose notifier a rebuild / surgical remove tombstoned (review 01 L2, TombstoneMonitoredNode). + // Handing either delete to the SDK would make it remove an item that is not there — at best a // spurious BadNodeIdUnknown on the client's delete, at worst a NullReferenceException inside // MonitoredNode2.Remove when the node's event list has gone null. Checked and dropped under the // same Lock the revalidation pass re-attaches under, so the two cannot interleave into @@ -4089,7 +4105,17 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 lock (Lock) { if (_eventItems.TryRemove(monitoredItem, out var stoodDown) && stoodDown.Suppressed) + { + Interlocked.Decrement(ref _suppressedItems); return ServiceResult.Good; + } + + if (IsHandleInNamespace(sourceId) is { } h + && (!MonitoredNodes.TryGetValue(h.NodeId, out var current) + || current.EventMonitoredItems?.Contains(monitoredItem) != true)) + { + return ServiceResult.Good; + } } return base.SubscribeToEvents(context, sourceId, subscriptionId, monitoredItem, unsubscribe); @@ -4175,10 +4201,186 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 protected override void OnMonitoredItemDeleted( ServerSystemContext context, NodeHandle handle, ISampledDataChangeMonitoredItem monitoredItem) { - _liveItems.TryRemove(monitoredItem, out _); + if (_liveItems.TryRemove(monitoredItem, out var tracked) && tracked.Suppressed) Interlocked.Decrement(ref _suppressedItems); base.OnMonitoredItemDeleted(context, handle, monitoredItem); } + // Review 01 L2 — a client deleting an item whose node this manager removed from under it (a rebuild or + // a surgical remove tombstoned the node's MonitoredNode2, see TombstoneMonitoredNode) must succeed. The + // SDK's own delete looks the node up by NodeId: with the entry gone it answers BadMonitoredItemIdInvalid, + // and with a FRESH entry (the same NodeId re-created and re-subscribed) it would try to remove the item + // from a fan-out list it is not on — a NullReferenceException when that list has gone null. An item that + // is not on the current fan-out for its NodeId has nothing left to detach, so its delete is completed here. + /// + protected override ServiceResult DeleteMonitoredItem( + ServerSystemContext context, IMonitoredItem monitoredItem, NodeHandle handle) + { + lock (Lock) + { + if (monitoredItem is ISampledDataChangeMonitoredItem dataItem + && (!MonitoredNodes.TryGetValue(handle.NodeId, out var current) + || current.DataChangeMonitoredItems?.Contains(dataItem) != true)) + { + OnMonitoredItemDeleted(context, handle, dataItem); + return ServiceResult.Good; + } + } + + return base.DeleteMonitoredItem(context, monitoredItem, handle); + } + + /// How many items the revalidation pass currently holds detached (stood down). Lets the + /// tombstone sweep for stood-down items — which are not on any fan-out list — run only when there are any. + private int _suppressedItems; + + /// + /// Review 01 L2 — forget the of a node this manager is removing, telling + /// every data-change item on it BadNodeIdUnknown first. MUST be called under Lock, BEFORE + /// the node is detached. + /// + /// + /// + /// The SDK keys its fan-out lists by NodeId and never clears them on the node manager's behalf + /// (DeleteMonitoredItem drops an entry only when its item count reaches zero), and its + /// component cache (AddNodeToComponentCache) hands back the CACHED + /// for a NodeId it already holds. So after a rebuild or a surgical remove+re-add, a NEW item on the + /// same NodeId — from any session — was bound to the DEAD node: created Good, never fed. Only + /// once every session had deleted its items did a create bind to the live node. + /// + /// + /// The fix removes the entry and releases each item's component-cache reference (the refcount the + /// SDK took at create), so the next create binds to the re-created node. The items themselves stay in + /// their client's subscription — a node manager cannot remove them — told BadNodeIdUnknown + /// (data items; an event item has no value slot) and dropped from the revalidation maps; their later + /// client delete is completed by / the SubscribeToEvents + /// unsubscribe arm. Queueing under Lock follows the publish path's own order (Lock, then + /// the item lock) — the inversion the revalidation pass avoids is the other way round. + /// + /// + /// An item the revalidation pass has STOOD DOWN is on no fan-out list, so it is swept from the + /// tracking map by NodeId instead — it still holds a component-cache reference that would otherwise + /// keep the dead node cached. + /// + /// + /// The NodeId being removed. + /// The timestamp the terminal BadNodeIdUnknown carries. + private void TombstoneMonitoredNode(NodeId nodeId, DateTime now) + { + if (MonitoredNodes.TryGetValue(nodeId, out var monitored)) + { + MonitoredNodes.Remove(nodeId); + if (monitored.DataChangeMonitoredItems is { Count: > 0 } dataItems) + { + foreach (var item in dataItems.ToArray()) + { + TellNodeRemoved(item, now); + if (item is ISampledDataChangeMonitoredItem sampled) _liveItems.TryRemove(sampled, out _); + ReleaseComponentCacheReference(item); + } + } + + if (monitored.EventMonitoredItems is { Count: > 0 } eventItems) + { + foreach (var item in eventItems.ToArray()) + { + _eventItems.TryRemove(item, out _); + ReleaseComponentCacheReference(item); + } + } + } + + if (Volatile.Read(ref _suppressedItems) > 0) SweepStoodDownItems(nodeId, now); + } + + /// Tombstones and every descendant (a condition's instance-scoped + /// children are monitorable in their own right). MUST be called under Lock. + private void TombstoneMonitoredSubtree(NodeState node, DateTime now) + { + TombstoneMonitoredNode(node.NodeId, now); + var children = new List(); + node.GetChildren(SystemContext, children); + foreach (var child in children) TombstoneMonitoredSubtree(child, now); + } + + /// The rebuild form: every monitored node this manager serves except the shared root, which the + /// rebuild keeps. MUST be called under Lock. + private void TombstoneAllMonitoredNodes() + { + var now = UtcNow; + var rootId = _root?.NodeId; + foreach (var nodeId in MonitoredNodes.Keys.ToList()) + { + if (rootId is not null && nodeId == rootId) continue; + TombstoneMonitoredNode(nodeId, now); + } + + if (Volatile.Read(ref _suppressedItems) > 0) SweepStoodDownItems(nodeId: null, now); + } + + /// Drops the stood-down items of (every stood-down item when null — + /// the rebuild case) from the revalidation maps, telling the data items BadNodeIdUnknown. + private void SweepStoodDownItems(NodeId? nodeId, DateTime now) + { + foreach (var (item, tracked) in _liveItems) + { + if (!tracked.Suppressed || (nodeId is not null && tracked.Handle.NodeId != nodeId)) continue; + if (!_liveItems.TryRemove(item, out _)) continue; + Interlocked.Decrement(ref _suppressedItems); + TellNodeRemoved(item, now); + ReleaseComponentCacheReference(item); + } + + foreach (var (item, tracked) in _eventItems) + { + if (!tracked.Suppressed || (nodeId is not null && tracked.Handle.NodeId != nodeId)) continue; + if (!_eventItems.TryRemove(item, out _)) continue; + Interlocked.Decrement(ref _suppressedItems); + ReleaseComponentCacheReference(item); + } + } + + /// Queues the terminal BadNodeIdUnknown on a data-change item whose node is going away. + private void TellNodeRemoved(IDataChangeMonitoredItem2 item, DateTime now) + { + try + { + item.QueueValue( + new DataValue(StatusCodes.BadNodeIdUnknown) { SourceTimestamp = now, ServerTimestamp = now }, + new ServiceResult(StatusCodes.BadNodeIdUnknown), + ignoreFilters: true); + } + catch (Exception ex) + { + // A disabled item refuses the value; nothing else here is worth failing a removal for. + _logger.LogDebug(ex, "OtOpcUaNodeManager: could not queue the removal status to a monitored item"); + } + } + + /// Gives back the component-cache reference the SDK took for at create, so + /// the cache stops handing out the removed node for that NodeId. + private void ReleaseComponentCacheReference(IMonitoredItem item) + { + if (item.ManagerHandle is NodeHandle handle) RemoveNodeFromComponentCache(SystemContext, handle); + } + + /// Test observability: how many NodeIds currently carry a fan-out. + internal int MonitoredNodeCount + { + get { lock (Lock) return MonitoredNodes.Count; } + } + + /// Test observability: whether currently carries a fan-out. + internal bool IsNodeMonitored(NodeId nodeId) + { + lock (Lock) return MonitoredNodes.ContainsKey(nodeId); + } + + /// Test observability: how many data-change items the revalidation map tracks. + internal int LiveItemCount => _liveItems.Count; + + /// Test observability: how many direct event items the revalidation map tracks. + internal int LiveEventItemCount => _eventItems.Count; + /// /// Live data-change monitored items on nodes this manager owns, so a grant revoked by a deployment /// can reach the subscriptions it was authorizing. @@ -4817,7 +5019,26 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { if (!tracked.Suppressed) continue; - lock (Lock) tracked.Handle.MonitoredNode?.Add(item); + // Re-resolve the node by NodeId rather than trusting the handle (review 01 L2): the fan-out + // the item came off belongs to the NodeState that existed when it was stood down, and a + // rebuild or surgical remove in between has tombstoned that node — then the item was already + // dropped from this map, so a mismatch here is a race with one; tell the client and let go. + bool reattached; + lock (Lock) + { + reattached = ResolveNodeState(MapKey(nodeId)) is { } liveNode + && ReferenceEquals(liveNode, tracked.Handle.Node); + if (reattached) tracked.Handle.MonitoredNode?.Add(item); + else _liveItems.TryRemove(item, out _); + } + + Interlocked.Decrement(ref _suppressedItems); + if (!reattached) + { + TellNodeRemoved(item, now); + continue; + } + tracked.Suppressed = false; restored++; continue; @@ -4832,8 +5053,13 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 new ServiceResult(StatusCodes.BadUserAccessDenied), ignoreFilters: true); - lock (Lock) tracked.Handle.MonitoredNode?.Remove(item); + lock (Lock) + { + // The fan-out list may have gone null if every other item on the node was deleted meanwhile. + if (tracked.Handle.MonitoredNode is { DataChangeMonitoredItems: not null } fanOut) fanOut.Remove(item); + } tracked.Suppressed = true; + Interlocked.Increment(ref _suppressedItems); stoodDown++; } @@ -4905,8 +5131,22 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { if (!_eventItems.TryGetValue(item, out var current) || !ReferenceEquals(current, tracked)) continue; + // Re-resolve the notifier by NodeId (review 01 L2): re-attaching to the handle's NodeState + // after a rebuild would bind the item to a dead node. A notifier that is gone, or has been + // re-created, cannot be the one this item was stood down from — drop the item instead. + if (permitted + && (ResolveNodeState(MapKey(tracked.Handle.NodeId)) is not { } live || !ReferenceEquals(live, tracked.Handle.Node))) + { + _eventItems.TryRemove(item, out _); + Interlocked.Decrement(ref _suppressedItems); + ReleaseComponentCacheReference(item); + continue; + } + SubscribeToEvents(SystemContext, tracked.Handle.Node, item, unsubscribe: !permitted); tracked.Suppressed = !permitted; + if (permitted) Interlocked.Decrement(ref _suppressedItems); + else Interlocked.Increment(ref _suppressedItems); } if (permitted) restored++; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/StaleMonitoredNodeWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/StaleMonitoredNodeWireTests.cs new file mode 100644 index 00000000..6cacc60e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/StaleMonitoredNodeWireTests.cs @@ -0,0 +1,252 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Opc.Ua; +using Opc.Ua.Client; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Review 01 L2, over the wire against the real SDK: a rebuild (or a surgical remove + re-add) that +/// recreates a NodeId must not leave the SDK's per-NodeId MonitoredNode2 — and its component-cache +/// entry — pointing at the DEAD NodeState. Before the fix, ANY new item on that NodeId, from any +/// session, bound to the dead node while an older item on it still existed: created Good, fed its +/// stale initial value, then never again. +/// +/// Each test keeps the OLD item alive while creating the NEW one — that is exactly the trap — and +/// proves the positive path (the new item receives a value written AFTER its creation) before asserting +/// the old item's BadNodeIdUnknown. Deleting the old item afterwards must succeed too. +/// +/// +public sealed class StaleMonitoredNodeWireTests +{ + private const string ServerUri = "urn:OtOpcUa.StaleMonitoredNode"; + private const string Device = "Plant/Modbus/dev1"; + private const string Tag = "Plant/Modbus/dev1/Speed"; + + /// Rebuild + re-create the same NodeId: the old item is told, a new item on the same NodeId receives values. + [Fact] + public async Task A_new_item_after_a_rebuild_binds_to_the_recreated_node() + { + await RunAsync("rebuild", nm => + { + nm.RebuildAddressSpace(); + nm.CompleteAddressSpaceApply(); + Materialise(nm); + }); + } + + /// Surgical remove + re-add of the same NodeId: same contract as the rebuild. + [Fact] + public async Task A_new_item_after_a_surgical_remove_and_readd_binds_to_the_recreated_node() + { + await RunAsync("surgical", nm => + { + nm.RemoveVariableNode(Tag, AddressSpaceRealm.Raw).ShouldBeTrue(); + nm.EnsureVariable(Tag, Device, "Speed", "Int32", writable: false, realm: AddressSpaceRealm.Raw); + }); + } + + /// + /// The same trap from a SECOND session: a UaExpert operator connecting after a deploy while ScadaBridge's + /// original item is still alive. + /// + [Fact] + public async Task A_second_session_binds_to_the_recreated_node_while_the_first_sessions_item_lives() + { + var pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-stale-2s-{Guid.NewGuid():N}"); + var ct = TestContext.Current.CancellationToken; + try + { + var (server, host, port) = await BootServerAsync(pkiRoot, ServerUri + ".TwoSessions", ct); + await using var _ = host; + var nm = server.NodeManager!; + Materialise(nm); + nm.WriteValue(Tag, 1, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw); + + using var sessionA = await OpenSessionAsync($"opc.tcp://127.0.0.1:{port}/OtOpcUa", ct); + var nodeId = new NodeId(Tag, (ushort)sessionA.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri)); + var (subA, oldItem, oldValues, gate) = await SubscribeAsync(sessionA, nodeId, ct); + await WaitUntilAsync(() => { lock (gate) return oldValues.Count >= 1; }, TimeSpan.FromSeconds(5)); + + nm.RebuildAddressSpace(); + nm.CompleteAddressSpaceApply(); + Materialise(nm); + nm.WriteValue(Tag, 2, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw); + + // Session B arrives AFTER the rebuild, while A's item is still in A's subscription. + using var sessionB = await OpenSessionAsync($"opc.tcp://127.0.0.1:{port}/OtOpcUa", ct); + var (subB, newItem, newValues, gateB) = await SubscribeAsync(sessionB, nodeId, ct); + await WaitUntilAsync(() => { lock (gateB) return newValues.Count >= 1; }, TimeSpan.FromSeconds(5)); + + nm.WriteValue(Tag, 3, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw); + await WaitUntilAsync(() => { lock (gateB) return newValues.Any(v => Equals(v.Value, 3)); }, TimeSpan.FromSeconds(5), + "the second session's item must receive a value written after its creation"); + + lock (gate) oldValues.ShouldContain(v => v.StatusCode == StatusCodes.BadNodeIdUnknown, "the first session's item was told its node went away"); + lock (gateB) newValues.ShouldNotContain(v => v.StatusCode == StatusCodes.BadNodeIdUnknown); + + subA.RemoveItem(oldItem); + await subA.ApplyChangesAsync(ct); + subB.RemoveItem(newItem); + await subB.ApplyChangesAsync(ct); + } + finally + { + SafeDelete(pkiRoot); + } + } + + private static async Task RunAsync(string name, Action recreate) + { + var pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-stale-{name}-{Guid.NewGuid():N}"); + var ct = TestContext.Current.CancellationToken; + try + { + var (server, host, port) = await BootServerAsync(pkiRoot, $"{ServerUri}.{name}", ct); + await using var _ = host; + var nm = server.NodeManager!; + Materialise(nm); + nm.WriteValue(Tag, 41, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw); + + using var session = await OpenSessionAsync($"opc.tcp://127.0.0.1:{port}/OtOpcUa", ct); + var nodeId = new NodeId(Tag, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri)); + var (subscription, oldItem, oldValues, gate) = await SubscribeAsync(session, nodeId, ct); + await WaitUntilAsync(() => { lock (gate) return oldValues.Any(v => Equals(v.Value, 41)); }, TimeSpan.FromSeconds(5)); + nm.IsNodeMonitored(nodeId).ShouldBeTrue("non-vacuity: the SDK really holds a fan-out for the node"); + + // The node goes away and comes back under the same NodeId; a fresh value lands on the NEW node. + recreate(nm); + nm.WriteValue(Tag, 42, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw); + + // A NEW item on the same NodeId, in the same subscription, with the old one still present. + var newValues = new List(); + var newItem = new MonitoredItem(subscription.DefaultItem) + { + StartNodeId = nodeId, AttributeId = Attributes.Value, SamplingInterval = 50, QueueSize = 10, + }; + newItem.Notification += (item, _) => { foreach (var v in item.DequeueValues()) lock (gate) newValues.Add(v); }; + subscription.AddItem(newItem); + await subscription.ApplyChangesAsync(ct); + newItem.Status.Error.ShouldBeNull(); + + // Positive control: the initial value is the NEW node's, and a value written afterwards arrives. + await WaitUntilAsync(() => { lock (gate) return newValues.Any(v => Equals(v.Value, 42)); }, TimeSpan.FromSeconds(5), + "the new item's initial read must come from the re-created node"); + nm.WriteValue(Tag, 43, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw); + await WaitUntilAsync(() => { lock (gate) return newValues.Any(v => Equals(v.Value, 43)); }, TimeSpan.FromSeconds(5), + "the new item must receive a value written after its creation"); + lock (gate) newValues.ShouldNotContain(v => v.StatusCode == StatusCodes.BadNodeIdUnknown); + + // 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"); + nm.IsNodeMonitored(nodeId).ShouldBeTrue("the NEW item's fan-out exists for the re-created node"); + + // The old item's delete completes — its node is gone, not its right to be deleted. + subscription.RemoveItem(oldItem); + await subscription.ApplyChangesAsync(ct); + (oldItem.Status.Error is null || ServiceResult.IsGood(oldItem.Status.Error)).ShouldBeTrue( + $"deleting the orphaned item must succeed, got {oldItem.Status.Error}"); + + // 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)); + } + finally + { + SafeDelete(pkiRoot); + } + } + + private static void Materialise(OtOpcUaNodeManager nm) + { + nm.EnsureFolder(Device, null, "dev1", AddressSpaceRealm.Raw); + nm.EnsureVariable(Tag, Device, "Speed", "Int32", writable: false, realm: AddressSpaceRealm.Raw); + } + + private static async Task<(Subscription Subscription, MonitoredItem Item, List Values, object Gate)> SubscribeAsync( + ISession session, NodeId nodeId, CancellationToken ct) + { + var values = new List(); + var gate = new object(); + var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 }; + session.AddSubscription(subscription); + await subscription.CreateAsync(ct); + var item = new MonitoredItem(subscription.DefaultItem) + { + StartNodeId = nodeId, AttributeId = Attributes.Value, SamplingInterval = 50, QueueSize = 10, + }; + item.Notification += (i, _) => { foreach (var v in i.DequeueValues()) lock (gate) values.Add(v); }; + subscription.AddItem(item); + await subscription.ApplyChangesAsync(ct); + return (subscription, item, values, gate); + } + + private static async Task<(OtOpcUaSdkServer Server, OpcUaApplicationHost Host, int Port)> BootServerAsync( + string pkiRoot, string appUri, CancellationToken ct) + { + var boundPort = 0; + var started = await OpcUaHostStart.StartAsync( + port => + { + boundPort = port; + return new OpcUaApplicationHost( + new OpcUaApplicationHostOptions + { + ApplicationName = appUri, + ApplicationUri = appUri, + OpcUaPort = port, + PublicHostname = "127.0.0.1", + PkiStoreRoot = pkiRoot, + EnabledSecurityProfiles = new List { OpcUaSecurityProfile.None }, + AutoAcceptUntrustedClientCertificates = true, + }, + NullLogger.Instance); + }, + () => new OtOpcUaSdkServer(), ct); + return (started.Server, started.Host, boundPort); + } + + private static async Task OpenSessionAsync(string endpointUrl, CancellationToken ct) + { + var appConfig = new ApplicationConfiguration + { + ApplicationName = "OtOpcUa.StaleMonitoredNodeClient", + ApplicationUri = $"urn:OtOpcUa.StaleMonitoredNodeClient.{Guid.NewGuid():N}", + ApplicationType = ApplicationType.Client, + SecurityConfiguration = TestClientSecurity.Build(TestClientSecurity.AllocatePkiRoot()), + ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60_000 }, + }; + await appConfig.ValidateAsync(ApplicationType.Client, ct); + appConfig.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true; + + var endpoint = await CoreClientUtils.SelectEndpointAsync( + appConfig, endpointUrl, false, DefaultTelemetry.Create(_ => { }), ct); + var configuredEndpoint = new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(appConfig)); + + return await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync( + appConfig, configuredEndpoint, updateBeforeConnect: false, checkDomain: false, + sessionName: "StaleMonitoredNodeWireTests", sessionTimeout: 60_000, + identity: new UserIdentity(new AnonymousIdentityToken()), preferredLocales: null, ct: ct); + } + + private static async Task WaitUntilAsync(Func condition, TimeSpan timeout, string? because = null) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (condition()) return; + await Task.Delay(50); + } + condition().ShouldBeTrue(because ?? "condition not met within timeout"); + } + + private static void SafeDelete(string dir) + { + if (!Directory.Exists(dir)) return; + try { Directory.Delete(dir, recursive: true); } + catch { /* best-effort */ } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStaleMonitoredNodeTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStaleMonitoredNodeTests.cs new file mode 100644 index 00000000..fcd966e5 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStaleMonitoredNodeTests.cs @@ -0,0 +1,197 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Opc.Ua; +using Opc.Ua.Server; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Authorization; +using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests; + +/// +/// Review 01 L2, the half the wire tests (StaleMonitoredNodeWireTests) cannot reach: an item the +/// ACL revalidation pass has STOOD DOWN is on no SDK fan-out list, so the tombstone a removal runs must find +/// it through the revalidation map. Otherwise it keeps a component-cache reference to the dead node — and +/// the cache hands that node to the next item created on the NodeId. +/// +public sealed class NodeManagerStaleMonitoredNodeTests : IDisposable +{ + private const string Cluster = "SITE-A"; + private const string Principal = "svc"; + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-stale-{Guid.NewGuid():N}"); + + /// A surgical remove tells a stood-down item and drops it from the revalidation map. + [Fact] + public async Task A_surgical_remove_tells_and_forgets_a_stood_down_item() + { + var (host, server) = await BootAsync(); + try + { + var nm = server.NodeManager!; + var (item, nodeId) = await StandDownAsync(nm, "Plant/Modbus/dev1/Speed"); + + nm.RemoveVariableNode("Plant/Modbus/dev1/Speed", AddressSpaceRealm.Raw).ShouldBeTrue(); + + item.Statuses.ShouldContain((StatusCode)StatusCodes.BadNodeIdUnknown); + nm.LiveItemCount.ShouldBe(0, "the item's node is gone; the map must not keep re-authorizing it"); + nm.IsNodeMonitored(nodeId).ShouldBeFalse(); + } + finally { await host.DisposeAsync(); } + } + + /// A rebuild sweeps every stood-down item. + [Fact] + public async Task A_rebuild_tells_and_forgets_every_stood_down_item() + { + var (host, server) = await BootAsync(); + try + { + var nm = server.NodeManager!; + var (item, _) = await StandDownAsync(nm, "Plant/Modbus/dev1/Speed"); + + nm.RebuildAddressSpace(); + nm.CompleteAddressSpaceApply(); + + item.Statuses.ShouldContain((StatusCode)StatusCodes.BadNodeIdUnknown); + nm.LiveItemCount.ShouldBe(0); + } + finally { await host.DisposeAsync(); } + } + + /// + /// Positive control for the sweep: a stood-down item on a SURVIVING node is left alone by a surgical + /// remove of a sibling, and the next permitted pass restores it. + /// + [Fact] + public async Task A_stood_down_item_on_a_surviving_node_is_untouched_and_still_restorable() + { + var (host, server) = await BootAsync(); + try + { + var nm = server.NodeManager!; + nm.EnsureVariable("Plant/Modbus/dev1/Other", null, "Other", "Float", writable: false, realm: AddressSpaceRealm.Raw); + var (item, _) = await StandDownAsync(nm, "Plant/Modbus/dev1/Speed"); + + nm.RemoveVariableNode("Plant/Modbus/dev1/Other", AddressSpaceRealm.Raw).ShouldBeTrue(); + + item.Statuses.ShouldNotContain((StatusCode)StatusCodes.BadNodeIdUnknown); + nm.LiveItemCount.ShouldBe(1); + + // Grant restored ⇒ the pass re-attaches it (a second stand-down proves the restore happened). + nm.Authorizer = Authorizer(permits: true); + nm.RevalidateMonitoredItems(); + nm.Authorizer = Authorizer(permits: false); + nm.RevalidateMonitoredItems(); + item.Statuses.Count(s => s == StatusCodes.BadUserAccessDenied).ShouldBe(2); + } + 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) + { + nm.EnsureVariable(rawPath, null, rawPath, "Float", writable: false, realm: AddressSpaceRealm.Raw); + var nodeId = nm.TryGetVariable(rawPath, AddressSpaceRealm.Raw)!.NodeId; + var handle = (NodeHandle)nm.GetManagerHandle(nodeId); + 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: the item really was stood down"); + await Task.CompletedTask; + return (item, nodeId); + } + + private static DataPlaneAuthorizer Authorizer(bool permits) + { + var state = new DataPlaneAuthorityState(); + var grants = permits + ? new[] + { + new NodeAcl + { + NodeAclId = "acl-cluster", ClusterId = Cluster, Principal = Principal, + ScopeKind = NodeAclScopeKind.Cluster, PermissionFlags = NodePermissions.Subscribe, + }, + } + : []; + state.Apply(Cluster, "rev-1", [], grants, new Dictionary>()); + return new DataPlaneAuthorizer(state, new TriePermissionEvaluator(state.Tries), enabled: true, NullLogger.Instance); + } + + private static RoleCarryingUserIdentity Identity() => + new(new UserNameIdentityToken { UserName = "apikey" }, [], Principal, apiKeyId: null); + + private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync() => + await OpcUaHostStart.StartAsync( + port => new OpcUaApplicationHost( + new OpcUaApplicationHostOptions + { + ApplicationName = "OtOpcUa.StaleTest", + ApplicationUri = $"urn:OtOpcUa.StaleTest:{Guid.NewGuid():N}", + OpcUaPort = port, + PublicHostname = "localhost", + PkiStoreRoot = _pkiRoot, + }, + NullLogger.Instance), + () => new OtOpcUaSdkServer(), Ct); + + public void Dispose() + { + if (Directory.Exists(_pkiRoot)) Directory.Delete(_pkiRoot, recursive: true); + } + + /// The minimum of a sampled item the removal path touches: its handle, identity and value queue. + private sealed class StubItem(NodeHandle handle, IUserIdentity identity) : ISampledDataChangeMonitoredItem + { + public List Statuses { get; } = []; + public IUserIdentity EffectiveIdentity => identity; + public object ManagerHandle => handle; + public NodeId NodeId => handle.NodeId; + public void QueueValue(DataValue value, ServiceResult error, bool ignoreFilters) => Statuses.Add(error?.StatusCode ?? value.StatusCode); + public void QueueValue(DataValue value, ServiceResult error) => QueueValue(value, error, false); + public void Dispose() { } + + public INodeManager NodeManager => throw new NotSupportedException(); + public ISession Session => throw new NotSupportedException(); + public uint Id => throw new NotSupportedException(); + public uint SubscriptionId => throw new NotSupportedException(); + public bool IsDurable => throw new NotSupportedException(); + public uint ClientHandle => throw new NotSupportedException(); + public ISubscription SubscriptionCallback { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + public int MonitoredItemType => throw new NotSupportedException(); + public bool IsReadyToPublish => throw new NotSupportedException(); + public bool IsReadyToTrigger { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + public bool IsResendData => throw new NotSupportedException(); + public MonitoringMode MonitoringMode => throw new NotSupportedException(); + public double SamplingInterval => throw new NotSupportedException(); + public DataChangeFilter DataChangeFilter => throw new NotSupportedException(); + public uint AttributeId => throw new NotSupportedException(); + public NumericRange IndexRange => throw new NotSupportedException(); + public QualifiedName DataEncoding => throw new NotSupportedException(); + public bool AlwaysReportUpdates { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + public DiagnosticsMasks DiagnosticsMasks => throw new NotSupportedException(); + public uint QueueSize => throw new NotSupportedException(); + public double MinimumSamplingInterval => throw new NotSupportedException(); + public int TimeToNextSample => throw new NotSupportedException(); + public void SetupResendDataTrigger() => throw new NotSupportedException(); + public ServiceResult GetCreateResult(out MonitoredItemCreateResult result) => throw new NotSupportedException(); + public ServiceResult GetModifyResult(out MonitoredItemModifyResult result) => throw new NotSupportedException(); + public IStoredMonitoredItem ToStorableMonitoredItem() => throw new NotSupportedException(); + public MonitoringMode SetMonitoringMode(MonitoringMode monitoringMode) => throw new NotSupportedException(); + public bool Publish(OperationContext context, Queue notifications, Queue diagnostics, uint maxNotificationsPerPublish, ILogger logger) => throw new NotSupportedException(); + public void SetSemanticsChanged() => throw new NotSupportedException(); + public void SetStructureChanged() => throw new NotSupportedException(); + public bool SamplingIntervalExpired() => throw new NotSupportedException(); + public ReadValueId GetReadValueId() => throw new NotSupportedException(); + public ServiceResult ModifyAttributes(DiagnosticsMasks diagnosticsMasks, TimestampsToReturn timestampsToReturn, uint clientHandle, MonitoringFilter originalFilter, MonitoringFilter filterToUse, Opc.Ua.Range range, double samplingInterval, uint queueSize, bool discardOldest) => throw new NotSupportedException(); + public void SetSamplingInterval(double samplingInterval) => throw new NotSupportedException(); + } +}