Fix publish queue lost wake-up

Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs

@@ -108,7 +108,7 @@ public Task<ISubscription> PublishAsync(string secureChannelId,
             }
 
             QueuedSubscription subscriptionToPublish;
-            lock (m_subscriptionPublishLock)
+            lock (m_lock)
             {
                 // find the waiting subscription with the highest priority.
                 subscriptionToPublish = GetSubscriptionToPublish();
@@ -117,10 +117,7 @@ public Task<ISubscription> PublishAsync(string secureChannelId,
                 {
                     return Task.FromResult(subscriptionToPublish.Subscription);
                 }
-            }
 
-            lock (m_lock)
-            {
                 // check if queue is full.
                 if (m_queuedRequests.Count >= m_maxRequestCount)
                 {
@@ -150,6 +147,8 @@ public Task<ISubscription> PublishAsync(string secureChannelId,
         /// <returns>The list of subscriptions in the queue.</returns>
         public IList<ISubscription> Close()
         {
+            var subscriptions = new List<ISubscription>();
+
             lock (m_lock)
             {
                 // TraceState("SESSION CLOSED");
@@ -164,19 +163,21 @@ public IList<ISubscription> Close()
                 }
 
                 // tell the subscriptions that the session is closed.
-                var subscriptions = new List<ISubscription>(m_queuedSubscriptions.Count);
-
                 foreach (KeyValuePair<uint, QueuedSubscription> entry in m_queuedSubscriptions)
                 {
                     subscriptions.Add(entry.Value.Subscription);
-                    entry.Value.Subscription.SessionClosed();
                 }
 
                 // clear the queue.
                 m_queuedSubscriptions.Clear();
+            }
 
-                return subscriptions;
+            foreach (ISubscription subscription in subscriptions)
+            {
+                subscription.SessionClosed();
             }
+
+            return subscriptions;
         }
 
         /// <summary>
@@ -375,19 +376,19 @@ public void PublishCompleted(ISubscription subscription, bool moreNotifications)
             if (m_queuedSubscriptions.TryGetValue(subscription.Id,
                 out QueuedSubscription queuedSubscription))
             {
-                queuedSubscription.Publishing = false;
-
-                if (moreNotifications)
+                lock (m_lock)
                 {
-                    lock (m_subscriptionPublishLock)
+                    queuedSubscription.Publishing = false;
+
+                    if (moreNotifications)
                     {
                         AssignSubscriptionToRequest(queuedSubscription);
                     }
-                }
-                else
-                {
-                    queuedSubscription.ReadyToPublish = false;
-                    queuedSubscription.Timestamp = DateTime.UtcNow;
+                    else
+                    {
+                        queuedSubscription.ReadyToPublish = false;
+                        queuedSubscription.Timestamp = DateTime.UtcNow;
+                    }
                 }
             }
         }
@@ -399,8 +400,11 @@ public void Requeue(ISubscription subscription)
         {
             if (m_queuedSubscriptions.TryGetValue(subscription.Id, out QueuedSubscription queuedSubscription))
             {
-                queuedSubscription.Publishing = false;
-                queuedSubscription.ReadyToPublish = true;
+                lock (m_lock)
+                {
+                    queuedSubscription.Publishing = false;
+                    queuedSubscription.ReadyToPublish = true;
+                }
             }
         }
 
@@ -443,9 +447,12 @@ public void PublishTimerExpired()
                 // assign subscription to request if one is available.
                 if (!subscription.Publishing)
                 {
-                    lock (m_subscriptionPublishLock)
+                    lock (m_lock)
                     {
-                        AssignSubscriptionToRequest(subscription);
+                        if (!subscription.Publishing)
+                        {
+                            AssignSubscriptionToRequest(subscription);
+                        }
                     }
                 }
             }
@@ -663,7 +670,6 @@ internal void TraceState(string context, params object[] args)
         }
 
         private readonly Lock m_lock = new();
-        private readonly Lock m_subscriptionPublishLock = new();
         private readonly ILogger m_logger;
         private readonly IServerInternal m_server;
         private readonly ISession m_session;

Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs

@@ -0,0 +1,135 @@
+/* ========================================================================
+ * Copyright (c) 2005-2026 The OPC Foundation, Inc. All rights reserved.
+ *
+ * OPC Foundation MIT License 1.00
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following
+ * conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
+ * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+ * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+ * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+ * OTHER DEALINGS IN THE SOFTWARE.
+ *
+ * The complete license agreement can be found here:
+ * http://opcfoundation.org/License/MIT/1.00/
+ * ======================================================================*/
+
+using System;
+using System.Reflection;
+using System.Threading;
+using System.Threading.Tasks;
+using Moq;
+using NUnit.Framework;
+using Opc.Ua.Tests;
+
+namespace Opc.Ua.Server.Tests
+{
+    [TestFixture]
+    [Category("Subscription")]
+    [Parallelizable]
+    public sealed class SessionPublishQueueRaceTests
+    {
+        [Test]
+        [CancelAfter(10000)]
+        public async Task PublishReadyTransitionCompletesQueuedRequestAsync()
+        {
+            ITelemetryContext telemetry = NUnitTelemetryContext.Create();
+            var subscriptionManager = new Mock<ISubscriptionManager>();
+            var server = new Mock<IServerInternal>();
+            server.Setup(s => s.Telemetry).Returns(telemetry);
+            server.Setup(s => s.SubscriptionManager).Returns(subscriptionManager.Object);
+
+            var session = new Mock<ISession>();
+            session.Setup(s => s.Id).Returns(new NodeId(Guid.NewGuid()));
+            session.Setup(s => s.IsSecureChannelValid(It.IsAny<string>())).Returns(true);
+
+            using var queue = new SessionPublishQueue(server.Object, session.Object, 10);
+            var subscription = new Mock<ISubscription>();
+            subscription.Setup(s => s.Id).Returns(1);
+            queue.Add(subscription.Object);
+
+            Lock queueLock = GetPrivateLock(queue, "m_lock");
+            Lock splitPublishLock = TryGetPrivateLock(queue, "m_subscriptionPublishLock");
+            using var publishStarted = new ManualResetEventSlim();
+            Task<ISubscription> publishTask;
+
+            // The legacy split lock lets PublishAsync finish its ready check
+            // before it blocks on the request-queue lock.
+            queueLock.Enter();
+            try
+            {
+                splitPublishLock?.Enter();
+                try
+                {
+                    publishTask = Task.Run(() =>
+                    {
+                        publishStarted.Set();
+                        return queue.PublishAsync(
+                            "channel1",
+                            DateTime.MaxValue,
+                            false,
+                            CancellationToken.None);
+                    });
+                    Assert.That(
+                        publishStarted.Wait(TimeSpan.FromSeconds(2)),
+                        Is.True,
+                        "The Publish request did not start.");
+                }
+                finally
+                {
+                    splitPublishLock?.Exit();
+                }
+
+                if (splitPublishLock != null)
+                {
+                    Thread.Sleep(100);
+                    splitPublishLock.Enter();
+                    splitPublishLock.Exit();
+                }
+                queue.Requeue(subscription.Object);
+            }
+            finally
+            {
+                queueLock.Exit();
+            }
+
+            Task completed = await Task.WhenAny(
+                publishTask,
+                Task.Delay(TimeSpan.FromSeconds(2))).ConfigureAwait(false);
+
+            Assert.That(
+                completed,
+                Is.SameAs(publishTask),
+                "The ready Subscription and queued Publish request lost their wake-up.");
+            Assert.That(
+                await publishTask.ConfigureAwait(false),
+                Is.SameAs(subscription.Object));
+        }
+
+        private static Lock GetPrivateLock(SessionPublishQueue queue, string fieldName)
+        {
+            return TryGetPrivateLock(queue, fieldName) ??
+                throw new InvalidOperationException($"Field '{fieldName}' was not found.");
+        }
+
+        private static Lock TryGetPrivateLock(SessionPublishQueue queue, string fieldName)
+        {
+            return typeof(SessionPublishQueue)
+                .GetField(fieldName, BindingFlags.Instance | BindingFlags.NonPublic)?
+                .GetValue(queue) as Lock;
+        }
+    }
+}