Complete client subscriptions by identity instead of id

src/Opc.Ua.Client/Subscription/IMessageAckQueue.cs

@@ -51,10 +51,13 @@ ValueTask QueueAsync(SubscriptionAcknowledgement ack,
         /// <summary>
         /// Complete subscription
         /// </summary>
-        /// <param name="subscriptionId"></param>
+        /// <param name="subscription">The completing subscription.</param>
+        /// <param name="subscriptionId">The last server-assigned id of the
+        /// subscription (0 when it was never created).</param>
         /// <param name="ct"></param>
         /// <returns></returns>
-        ValueTask CompleteAsync(uint subscriptionId,
+        ValueTask CompleteAsync(IMessageProcessor subscription,
+            uint subscriptionId,
             CancellationToken ct = default);
 
         /// <summary>

src/Opc.Ua.Client/Subscription/MessageProcessor.cs

@@ -45,7 +45,19 @@ namespace Opc.Ua.Client.Subscriptions
     internal abstract class MessageProcessor : IMessageProcessor, IAsyncDisposable
     {
         /// <inheritdoc/>
-        public uint Id { get; internal set; }
+        public uint Id
+        {
+            get => m_id;
+            internal set
+            {
+                m_id = value;
+                if (value != 0)
+                {
+                    // Save non-zero server id for completion
+                    m_lastServerId = value;
+                }
+            }
+        }
 
         /// <summary>
         /// Number of notification messages detected as missing during
@@ -275,7 +287,7 @@ private async Task ProcessReceivedMessagesAsync(CancellationToken ct)
                 // to be removed
                 throw;
             }
-            await AckQueue.CompleteAsync(Id, default).ConfigureAwait(false);
+            await AckQueue.CompleteAsync(this, m_lastServerId, default).ConfigureAwait(false);
             OnPublishStateChanged(PublishState.Completed);
         }
 
@@ -645,6 +657,8 @@ public static int Compare(IncomingMessage message, IncomingMessage other)
         internal long LastNotificationTimestamp;
         internal uint LastSequenceNumberProcessed;
         internal uint LastDataSequenceNumberProcessed;
+        private uint m_id;
+        private uint m_lastServerId;
         internal long m_missingCount;
         internal long m_republishCount;
         internal IReadOnlyList<uint> AvailableInRetransmissionQueue;

src/Opc.Ua.Client/Subscription/SubscriptionManager.cs

@@ -403,21 +403,24 @@ public int DropPendingForSubscription(uint subscriptionId)
         }
 
         /// <inheritdoc/>
-        public ValueTask CompleteAsync(uint subscriptionId, CancellationToken ct)
+        public ValueTask CompleteAsync(IMessageProcessor completing,
+            uint subscriptionId, CancellationToken ct)
         {
             IManagedSubscription? subscription;
             LogicalSubscription? logical = null;
             lock (m_subscriptionLock)
             {
                 // Drop the partition from the dispatch registry.
-                subscription = m_subscriptions
-                    .FirstOrDefault(s => s.Id == subscriptionId);
+                subscription = completing as IManagedSubscription;
                 if (subscription == null ||
                     !m_subscriptions.Remove(subscription))
                 {
                     return default;
                 }
-                m_subscriptionHistory.Enqueue(subscriptionId);
+                if (subscriptionId != 0)
+                {
+                    m_subscriptionHistory.Enqueue(subscriptionId);
+                }
 
                 // If the removed partition was the primary of any
                 // logical wrapper, the wrapper has no usable
@@ -445,7 +448,7 @@ public ValueTask CompleteAsync(uint subscriptionId, CancellationToken ct)
             {
                 m_subscriptionHistory.TryDequeue(out _);
             }
-            m_logger.SubscriptionRemoved(subscription.Id);
+            m_logger.SubscriptionRemoved(subscriptionId);
             m_publishControl.Set();
             return default;
         }

tests/Opc.Ua.Client.Tests/Subscription/Fakes/FakeMessageAckQueue.cs

@@ -67,7 +67,8 @@ public ValueTask QueueAsync(SubscriptionAcknowledgement ack,
             return OnQueueAsync?.Invoke(ack, ct) ?? default;
         }
 
-        public ValueTask CompleteAsync(uint subscriptionId,
+        public ValueTask CompleteAsync(IMessageProcessor subscription,
+            uint subscriptionId,
             CancellationToken ct = default)
         {
             CompletedSubscriptions.Add(subscriptionId);

tests/Opc.Ua.Client.Tests/Subscription/SubscriptionManagerTests.cs

@@ -106,7 +106,7 @@ public async Task AddAndRemoveOfSubscription1Async()
             Assert.That(ex.StatusCode, Is.EqualTo(StatusCodes.BadAlreadyExists));
             await Task.Delay(100).ConfigureAwait(false); // Give time to workers to start
             Assert.That(sut.PublishWorkerCount, Is.Zero);
-            await sut.CompleteAsync(1, default).ConfigureAwait(false);
+            await sut.CompleteAsync(ms1, 1, default).ConfigureAwait(false);
             Assert.That(sut.Count, Is.EqualTo(1));
             Assert.That(sut.Items, Does.Contain(s2));
 
@@ -160,15 +160,15 @@ public async Task AddAndRemoveOfSubscription2Async()
             // but MinPublishWorkerCount defaults to 2, so the desired count is 2.
             await WaitForPublishWorkerCountAsync(sut, 2).ConfigureAwait(false);
 
-            await sut.CompleteAsync(2, default).ConfigureAwait(false); // Remove s2
+            await sut.CompleteAsync(ms2, 2, default).ConfigureAwait(false); // Remove s2
             Assert.That(sut.Count, Is.EqualTo(1));
             Assert.That(sut.Items, Does.Not.Contain(s2));
 
             // Workers stay at 2 because MinPublishWorkerCount=2 (default).
             Assert.That(sut.PublishWorkerCount, Is.EqualTo(2));
             Assert.That(sut.PublishControlCycles, Is.GreaterThan(0));
 
-            await sut.CompleteAsync(1, default).ConfigureAwait(false); // Remove s1
+            await sut.CompleteAsync(ms1, 1, default).ConfigureAwait(false); // Remove s1
             Assert.That(sut.Count, Is.Zero);
             Assert.That(sut.Items, Does.Not.Contain(s1));
 
@@ -179,6 +179,44 @@ public async Task AddAndRemoveOfSubscription2Async()
             Assert.That(sut.PublishWorkerCount, Is.Zero);
         }
 
+        [Test]
+        public async Task CompleteForUncreatedIdDoesNotEvictPendingSubscriptionAsync()
+        {
+            ILoggerFactory loggerFactory = m_telemetry.LoggerFactory;
+            var session = new FakeSubscriptionManagerContext();
+            OptionsMonitor<SubscriptionOptions> so1 = OptionsFactory.Create<SubscriptionOptions>();
+
+            // Subscription awaiting CreateSubscription
+            var ms1 = new FakeManagedSubscription();
+
+            var sut = new SubscriptionManager(session,
+                loggerFactory, DiagnosticsMasks.None);
+
+            session.CreateSubscriptionFactory = (handler, options, queue) =>
+            {
+                Assert.That(queue, Is.SameAs(sut));
+                if (ReferenceEquals(options, so1))
+                {
+                    return ms1;
+                }
+                throw new InvalidOperationException("unexpected options");
+            };
+
+            ISubscription s1 = sut.Add(m_mockNotificationDataHandler.Object, so1);
+            Assert.That(sut.Count, Is.EqualTo(1));
+
+            // A deleted subscription acknowledges completion with an id already reset to 0
+            await sut.CompleteAsync(new FakeManagedSubscription(), 0, default).ConfigureAwait(false);
+
+            // Verify that the pending subscription still exists
+            Assert.That(sut.Count, Is.EqualTo(1));
+            Assert.That(sut.Items, Does.Contain(s1));
+
+            await sut.DisposeAsync().ConfigureAwait(false);
+            Assert.That(sut.Count, Is.Zero);
+            Assert.That(sut.PublishWorkerCount, Is.Zero);
+        }
+
         [Test]
         public async Task ScaleOutAndInOfPublishWorkersAsync()
         {
@@ -411,7 +449,8 @@ public async Task CompleteAsyncCompletesSubscriptionAsync()
 
             m_subscriptionManager.Add(m_mockNotificationDataHandler.Object,
                 Mock.Of<IOptionsMonitor<SubscriptionOptions>>());
-            await m_subscriptionManager.CompleteAsync(1, CancellationToken.None).ConfigureAwait(false);
+            await m_subscriptionManager.CompleteAsync(mockSubscription, 1, CancellationToken.None)
+                .ConfigureAwait(false);
             Assert.That(m_subscriptionManager.Items, Is.Empty);
         }