diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs index 4e2642ca..99cbb907 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -1787,13 +1787,39 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit ?? throw new InvalidOperationException( "OpcUaClient session was lost before the subscription could be created."); session.AddSubscription(subscription); - await subscription.CreateAsync(cancellationToken).ConfigureAwait(false); - // Track each (MonitoredItem, handler) pair so UnsubscribeAsync / ShutdownAsync // can detach the Notification delegate before disposing the session. The lambda // captures `handle`, so we must hold the exact delegate instance returned by // `+=` to be able to remove it. var itemHandlers = new List(); + try + { + await CreateDataItemsAsync(session, itemHandlers).ConfigureAwait(false); + } + catch + { + // R06:O-1 — the host's reconcile retries a failed subscribe every health tick, so a half-built + // subscription left on the session (created server-side when CreateAsync succeeded, its item + // handlers still attached, tracked by nothing) would accumulate one live, publishing subscription + // per retry. Mirror the alarm path: detach, bounded remove, rethrow. + DetachNotificationHandlers(itemHandlers); + try + { + using var removeCts = new CancellationTokenSource(SessionCloseTimeout); + await session.RemoveSubscriptionAsync(subscription, removeCts.Token).ConfigureAwait(false); + } + catch { /* best-effort — the session may be the thing that failed */ } + throw; + } + _subscriptions[id] = new RemoteSubscription(subscription, handle, itemHandlers); + } + finally { _gate.Release(); } + + return handle; + + async Task CreateDataItemsAsync(ISession session, List itemHandlers) + { + await subscription.CreateAsync(cancellationToken).ConfigureAwait(false); foreach (var fullRef in fullReferences) { if (!TryResolveTagNode(session, fullRef, out var nodeId)) continue; @@ -1821,11 +1847,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit } await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false); - _subscriptions[id] = new RemoteSubscription(subscription, handle, itemHandlers); } - finally { _gate.Release(); } - - return handle; } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSubscribeCleanupTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSubscribeCleanupTests.cs new file mode 100644 index 00000000..b3ee7ce6 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientSubscribeCleanupTests.cs @@ -0,0 +1,85 @@ +using System.Text.Json; +using Moq; +using Opc.Ua; +using Opc.Ua.Client; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests; + +/// +/// Review 2026-10-21 R06:O-1 — a data SubscribeAsync whose wire create throws used to leave the +/// half-built on the session (untracked, so neither Unsubscribe nor Shutdown +/// ever deleted it). The host's reconcile retries every health tick, so a server that refused the create +/// accumulated one live subscription per retry. The alarm path already cleaned up; the data path now does too. +/// +/// +/// A Moq session never attaches the subscription (AddSubscription is a no-op), so the SDK's +/// Subscription.CreateAsync throws — the exact "create failed after AddSubscription" shape. +/// +[Trait("Category", "Unit")] +public sealed class OpcUaClientSubscribeCleanupTests +{ + private const string SpeedRef = "Plant/dev/Speed"; + + private static readonly IReadOnlyList Tags = + [new(SpeedRef, JsonSerializer.Serialize(new { nodeId = "ns=0;s=Speed" }), WriteIdempotent: true)]; + + private static (OpcUaClientDriver Driver, Mock Session) NewConnectedDriver() + { + var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-o1"); + var session = new Mock(MockBehavior.Loose); + session.SetupGet(s => s.MessageContext).Returns(new ServiceMessageContext(SdkTelemetry.Context)); + session.SetupGet(s => s.NamespaceUris).Returns(new NamespaceTable()); + drv.SetSessionForTest(session.Object); + drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags)); + return (drv, session); + } + + /// A failed data subscribe removes the subscription it added to the session, then rethrows. + [Fact] + public async Task A_failed_data_subscribe_removes_the_half_built_subscription_from_the_session() + { + var (drv, session) = NewConnectedDriver(); + using var _ = drv; + + await Should.ThrowAsync(() => + drv.SubscribeAsync([SpeedRef], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); + + session.Verify(s => s.AddSubscription(It.IsAny()), Times.Once); + session.Verify(s => s.RemoveSubscriptionAsync(It.IsAny(), It.IsAny()), Times.Once, + "the subscription added before the failing create must not stay on the session"); + } + + /// + /// Positive control for the observation above: the alarm path (which always cleaned up) is seen removing + /// its subscription through the same mock, so a "Times.Once" pass on the data path is not vacuous. + /// + [Fact] + public async Task Positive_control_the_alarm_path_is_observed_removing_its_half_built_subscription() + { + var (drv, session) = NewConnectedDriver(); + using var _ = drv; + + await Should.ThrowAsync(() => + drv.SubscribeAlarmsAsync([SpeedRef], TestContext.Current.CancellationToken)); + + session.Verify(s => s.RemoveSubscriptionAsync(It.IsAny(), It.IsAny()), Times.Once); + } + + /// Repeated failing subscribes (the host's reconcile retry) each clean up their own subscription. + [Fact] + public async Task Each_retry_of_a_failing_subscribe_cleans_up_after_itself() + { + var (drv, session) = NewConnectedDriver(); + using var _ = drv; + + for (var i = 0; i < 3; i++) + await Should.ThrowAsync(() => + drv.SubscribeAsync([SpeedRef], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); + + session.Verify(s => s.AddSubscription(It.IsAny()), Times.Exactly(3)); + session.Verify(s => s.RemoveSubscriptionAsync(It.IsAny(), It.IsAny()), Times.Exactly(3)); + } +}