fix(opcuaclient): remove a half-built data subscription when its create throws (R06:O-1)
SubscribeAsync left the SDK Subscription on the session (server-side when CreateAsync succeeded, item handlers attached, tracked by nothing) when CreateAsync/CreateItemsAsync threw, so the host's per-tick reconcile retry accumulated one live subscription per retry. Mirror the alarm path: detach the item handlers, bounded RemoveSubscriptionAsync, rethrow. Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
@@ -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<MonitoredItemNotificationHandle>();
|
||||
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<MonitoredItemNotificationHandle> 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;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
+85
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Review 2026-10-21 R06:O-1 — a data <c>SubscribeAsync</c> whose wire create throws used to leave the
|
||||
/// half-built <see cref="Subscription"/> 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.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// A Moq session never attaches the subscription (<c>AddSubscription</c> is a no-op), so the SDK's
|
||||
/// <c>Subscription.CreateAsync</c> throws — the exact "create failed after AddSubscription" shape.
|
||||
/// </remarks>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class OpcUaClientSubscribeCleanupTests
|
||||
{
|
||||
private const string SpeedRef = "Plant/dev/Speed";
|
||||
|
||||
private static readonly IReadOnlyList<RawTagEntry> Tags =
|
||||
[new(SpeedRef, JsonSerializer.Serialize(new { nodeId = "ns=0;s=Speed" }), WriteIdempotent: true)];
|
||||
|
||||
private static (OpcUaClientDriver Driver, Mock<ISession> Session) NewConnectedDriver()
|
||||
{
|
||||
var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-o1");
|
||||
var session = new Mock<ISession>(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);
|
||||
}
|
||||
|
||||
/// <summary>A failed data subscribe removes the subscription it added to the session, then rethrows.</summary>
|
||||
[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<Exception>(() =>
|
||||
drv.SubscribeAsync([SpeedRef], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken));
|
||||
|
||||
session.Verify(s => s.AddSubscription(It.IsAny<Subscription>()), Times.Once);
|
||||
session.Verify(s => s.RemoveSubscriptionAsync(It.IsAny<Subscription>(), It.IsAny<CancellationToken>()), Times.Once,
|
||||
"the subscription added before the failing create must not stay on the session");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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<Exception>(() =>
|
||||
drv.SubscribeAlarmsAsync([SpeedRef], TestContext.Current.CancellationToken));
|
||||
|
||||
session.Verify(s => s.RemoveSubscriptionAsync(It.IsAny<Subscription>(), It.IsAny<CancellationToken>()), Times.Once);
|
||||
}
|
||||
|
||||
/// <summary>Repeated failing subscribes (the host's reconcile retry) each clean up their own subscription.</summary>
|
||||
[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<Exception>(() =>
|
||||
drv.SubscribeAsync([SpeedRef], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken));
|
||||
|
||||
session.Verify(s => s.AddSubscription(It.IsAny<Subscription>()), Times.Exactly(3));
|
||||
session.Verify(s => s.RemoveSubscriptionAsync(It.IsAny<Subscription>(), It.IsAny<CancellationToken>()), Times.Exactly(3));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user