fix(opcua): tombstone MonitoredNode2 on every node removal (B2a)

Review 01 L2. The SDK keys its fan-out lists by NodeId and its component
cache hands back the CACHED NodeState for a NodeId it already holds, so a
rebuild or a surgical remove+re-add left a NEW item on the same NodeId —
from any session — bound to the DEAD node: created Good, fed its stale
initial value, never again, until every session had deleted its items.

Every removal path (RebuildAddressSpace, RemoveVariableNode,
RemoveAlarmConditionNode incl. the condition's instance-scoped children,
RemoveEquipmentSubtree) now queues BadNodeIdUnknown to the node's data
items, drops the MonitoredNodes entry, releases each item's component
cache reference and forgets the _liveItems/_eventItems entries. Items
the ACL pass stood down (on no fan-out list) are swept by NodeId. A later
client delete of an orphaned item is completed by the DeleteMonitoredItem
override / the SubscribeToEvents unsubscribe arm instead of answering
BadMonitoredItemIdInvalid or NRE-ing in MonitoredNode2.Remove. The
revalidation restore re-resolves the node by NodeId and no longer
re-attaches to Handle.Node blindly. Timestamps now come from the injected
Clock (UtcNow).

Tests: StaleMonitoredNodeWireTests (real SDK, three legs: rebuild,
surgical remove+re-add, second session; each keeps the old item alive,
proves the new item receives a value written after its creation, the old
one gets BadNodeIdUnknown, and its delete succeeds — all three fail with
the tombstone neutered), NodeManagerStaleMonitoredNodeTests (stood-down
item swept on surgical remove and rebuild; sibling removal leaves it
restorable).

Refs #732

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 06:41:09 -04:00
parent 00300d0f60
commit eaac65286d
3 changed files with 694 additions and 5 deletions
@@ -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;
/// <summary>
/// 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 <c>MonitoredNode2</c> — and its component-cache
/// entry — pointing at the DEAD <c>NodeState</c>. 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 <c>Good</c>, fed its
/// stale initial value, then never again.
/// <para>
/// 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 <c>BadNodeIdUnknown</c>. Deleting the old item afterwards must succeed too.
/// </para>
/// </summary>
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";
/// <summary>Rebuild + re-create the same NodeId: the old item is told, a new item on the same NodeId receives values.</summary>
[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);
});
}
/// <summary>Surgical remove + re-add of the same NodeId: same contract as the rebuild.</summary>
[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);
});
}
/// <summary>
/// The same trap from a SECOND session: a UaExpert operator connecting after a deploy while ScadaBridge's
/// original item is still alive.
/// </summary>
[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<OtOpcUaNodeManager> 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<DataValue>();
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<DataValue> Values, object Gate)> SubscribeAsync(
ISession session, NodeId nodeId, CancellationToken ct)
{
var values = new List<DataValue>();
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> { OpcUaSecurityProfile.None },
AutoAcceptUntrustedClientCertificates = true,
},
NullLogger<OpcUaApplicationHost>.Instance);
},
() => new OtOpcUaSdkServer(), ct);
return (started.Server, started.Host, boundPort);
}
private static async Task<ISession> 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<bool> 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 */ }
}
}
@@ -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;
/// <summary>
/// Review 01 L2, the half the wire tests (<c>StaleMonitoredNodeWireTests</c>) 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.
/// </summary>
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}");
/// <summary>A surgical remove tells a stood-down item and drops it from the revalidation map.</summary>
[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(); }
}
/// <summary>A rebuild sweeps every stood-down item.</summary>
[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(); }
}
/// <summary>
/// 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.
/// </summary>
[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(); }
}
/// <summary>Registers a stub item on <paramref name="rawPath"/> and stands it down through a denying pass.</summary>
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<AclScopeRef, IReadOnlyList<NodeAclPath>>());
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<OpcUaApplicationHost>.Instance),
() => new OtOpcUaSdkServer(), Ct);
public void Dispose()
{
if (Directory.Exists(_pkiRoot)) Directory.Delete(_pkiRoot, recursive: true);
}
/// <summary>The minimum of a sampled item the removal path touches: its handle, identity and value queue.</summary>
private sealed class StubItem(NodeHandle handle, IUserIdentity identity) : ISampledDataChangeMonitoredItem
{
public List<StatusCode> 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<MonitoredItemNotification> notifications, Queue<DiagnosticInfo> 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();
}
}