From 3c4bdd0cd4781c3a3a7b2839dbb9e2d545dfd2e8 Mon Sep 17 00:00:00 2001 From: Marc Schier Date: Sat, 18 Jul 2026 00:09:11 +0200 Subject: [PATCH 1/3] Fix publish queue lost wake-up (#3997) --- .../Subscription/SessionPublishQueue.cs | 50 ++++--- .../SessionPublishQueueRaceTests.cs | 135 ++++++++++++++++++ 2 files changed, 163 insertions(+), 22 deletions(-) create mode 100644 Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs diff --git a/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs b/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs index 3cae9ad1af..10d63acc9a 100644 --- a/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs +++ b/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs @@ -108,7 +108,7 @@ public Task 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 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 PublishAsync(string secureChannelId, /// The list of subscriptions in the queue. public IList Close() { + var subscriptions = new List(); + lock (m_lock) { // TraceState("SESSION CLOSED"); @@ -164,19 +163,21 @@ public IList Close() } // tell the subscriptions that the session is closed. - var subscriptions = new List(m_queuedSubscriptions.Count); - foreach (KeyValuePair 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; } /// @@ -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; diff --git a/Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs b/Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs new file mode 100644 index 0000000000..c47b3ed117 --- /dev/null +++ b/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(); + var server = new Mock(); + server.Setup(s => s.Telemetry).Returns(telemetry); + server.Setup(s => s.SubscriptionManager).Returns(subscriptionManager.Object); + + var session = new Mock(); + session.Setup(s => s.Id).Returns(new NodeId(Guid.NewGuid())); + session.Setup(s => s.IsSecureChannelValid(It.IsAny())).Returns(true); + + using var queue = new SessionPublishQueue(server.Object, session.Object, 10); + var subscription = new Mock(); + 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 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; + } + } +} From 50bf92e96ec6d1172a4cc86224b27961ccfbe835 Mon Sep 17 00:00:00 2001 From: Marc Schier Date: Sat, 18 Jul 2026 10:50:02 +0200 Subject: [PATCH 2/3] Address publish queue review feedback --- .../Subscription/SessionPublishQueue.cs | 4 +- .../SessionPublishQueueRaceTests.cs | 76 +++++++++++++++---- 2 files changed, 66 insertions(+), 14 deletions(-) diff --git a/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs b/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs index 10d63acc9a..34a26f0713 100644 --- a/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs +++ b/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs @@ -147,10 +147,12 @@ public Task PublishAsync(string secureChannelId, /// The list of subscriptions in the queue. public IList Close() { - var subscriptions = new List(); + List subscriptions; lock (m_lock) { + subscriptions = new List(m_queuedSubscriptions.Count); + // TraceState("SESSION CLOSED"); // set any waiting publish requests to Status BadSessionClosed. diff --git a/Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs b/Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs index c47b3ed117..63cac5d45c 100644 --- a/Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs +++ b/Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs @@ -28,6 +28,7 @@ * ======================================================================*/ using System; +using System.Linq; using System.Reflection; using System.Threading; using System.Threading.Tasks; @@ -39,7 +40,7 @@ namespace Opc.Ua.Server.Tests { [TestFixture] [Category("Subscription")] - [Parallelizable] + [NonParallelizable] public sealed class SessionPublishQueueRaceTests { [Test] @@ -61,8 +62,7 @@ public async Task PublishReadyTransitionCompletesQueuedRequestAsync() subscription.Setup(s => s.Id).Returns(1); queue.Add(subscription.Object); - Lock queueLock = GetPrivateLock(queue, "m_lock"); - Lock splitPublishLock = TryGetPrivateLock(queue, "m_subscriptionPublishLock"); + (Lock queueLock, Lock splitPublishLock) = GetPrivateLocks(queue); using var publishStarted = new ManualResetEventSlim(); Task publishTask; @@ -95,10 +95,11 @@ public async Task PublishReadyTransitionCompletesQueuedRequestAsync() if (splitPublishLock != null) { - Thread.Sleep(100); - splitPublishLock.Enter(); - splitPublishLock.Exit(); + WaitForSplitLockTransition(splitPublishLock); } + + // System.Threading.Lock is reentrant. Keeping the queue lock + // held preserves the historical check-before-enqueue window. queue.Requeue(subscription.Object); } finally @@ -119,17 +120,66 @@ await publishTask.ConfigureAwait(false), Is.SameAs(subscription.Object)); } - private static Lock GetPrivateLock(SessionPublishQueue queue, string fieldName) + private static (Lock QueueLock, Lock SplitPublishLock) GetPrivateLocks( + SessionPublishQueue queue) { - return TryGetPrivateLock(queue, fieldName) ?? - throw new InvalidOperationException($"Field '{fieldName}' was not found."); + FieldInfo[] lockFields = [.. typeof(SessionPublishQueue) + .GetFields(BindingFlags.Instance | BindingFlags.NonPublic) + .Where(field => field.FieldType == typeof(Lock))]; + if (lockFields.Length is 0 or > 2) + { + throw new InvalidOperationException( + $"Expected one or two private Lock fields, found: " + + $"{string.Join(", ", lockFields.Select(field => field.Name))}."); + } + + FieldInfo queueLockField = lockFields.FirstOrDefault(field => field.Name == "m_lock") ?? + (lockFields.Length == 1 + ? lockFields[0] + : throw new InvalidOperationException( + $"Could not identify the queue lock from: " + + $"{string.Join(", ", lockFields.Select(field => field.Name))}.")); + var queueLock = (Lock)queueLockField.GetValue(queue); + Lock splitPublishLock = lockFields + .Where(field => !ReferenceEquals(field, queueLockField)) + .Select(field => (Lock)field.GetValue(queue)) + .SingleOrDefault(); + return (queueLock, splitPublishLock); } - private static Lock TryGetPrivateLock(SessionPublishQueue queue, string fieldName) + private static void WaitForSplitLockTransition(Lock splitPublishLock) { - return typeof(SessionPublishQueue) - .GetField(fieldName, BindingFlags.Instance | BindingFlags.NonPublic)? - .GetValue(queue) as Lock; + bool enteredByPublisher = SpinWait.SpinUntil( + () => + { + if (!splitPublishLock.TryEnter()) + { + return true; + } + splitPublishLock.Exit(); + return false; + }, + TimeSpan.FromSeconds(2)); + Assert.That( + enteredByPublisher, + Is.True, + "PublishAsync did not enter the legacy split lock."); + + bool releasedByPublisher = SpinWait.SpinUntil( + () => + { + if (!splitPublishLock.TryEnter()) + { + return false; + } + splitPublishLock.Exit(); + return true; + }, + TimeSpan.FromSeconds(2)); + Assert.That( + releasedByPublisher, + Is.True, + "PublishAsync did not leave the legacy split lock."); } } } From 059a3daef66ab188f49bd2e051c1aaffe1083553 Mon Sep 17 00:00:00 2001 From: Marc Schier Date: Tue, 21 Jul 2026 08:35:23 +0200 Subject: [PATCH 3/3] Address publish timer review feedback --- .../Subscription/SessionPublishQueue.cs | 6 --- .../SessionPublishQueueRaceTests.cs | 54 +++++++++++++++++++ 2 files changed, 54 insertions(+), 6 deletions(-) diff --git a/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs b/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs index 34a26f0713..5416615b50 100644 --- a/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs +++ b/Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs @@ -440,12 +440,6 @@ public void PublishTimerExpired() continue; } - // do nothing if subscription has already been flagged as available. - if (subscription.ReadyToPublish) - { - continue; - } - // assign subscription to request if one is available. if (!subscription.Publishing) { diff --git a/Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs b/Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs index 63cac5d45c..33c9eef43f 100644 --- a/Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs +++ b/Tests/Opc.Ua.Server.Tests/SessionPublishQueueRaceTests.cs @@ -120,6 +120,60 @@ await publishTask.ConfigureAwait(false), Is.SameAs(subscription.Object)); } + [Test] + [CancelAfter(10000)] + public async Task PublishTimerAssignsRequeuedSubscriptionToWaitingRequestAsync() + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + var subscriptionManager = new Mock(); + var server = new Mock(); + server.Setup(s => s.Telemetry).Returns(telemetry); + server.Setup(s => s.SubscriptionManager).Returns(subscriptionManager.Object); + + var session = new Mock(); + session.Setup(s => s.Id).Returns(new NodeId(Guid.NewGuid())); + session.Setup(s => s.IsSecureChannelValid(It.IsAny())).Returns(true); + + using var queue = new SessionPublishQueue(server.Object, session.Object, 10); + var subscription = new Mock(); + subscription.Setup(s => s.Id).Returns(1); + subscription + .Setup(s => s.PublishTimerExpired()) + .Returns(PublishingState.NotificationsAvailable); + queue.Add(subscription.Object); + + queue.Requeue(subscription.Object); + Assert.That( + await queue.PublishAsync( + "channel1", + DateTime.MaxValue, + false, + CancellationToken.None).ConfigureAwait(false), + Is.SameAs(subscription.Object)); + + Task publishTask = queue.PublishAsync( + "channel1", + DateTime.MaxValue, + false, + CancellationToken.None); + Assert.That(publishTask.IsCompleted, Is.False); + + queue.Requeue(subscription.Object); + queue.PublishTimerExpired(); + + Task completed = await Task.WhenAny( + publishTask, + Task.Delay(TimeSpan.FromSeconds(2))).ConfigureAwait(false); + + Assert.That( + completed, + Is.SameAs(publishTask), + "The timer did not assign the already-ready Subscription to the queued Publish request."); + Assert.That( + await publishTask.ConfigureAwait(false), + Is.SameAs(subscription.Object)); + } + private static (Lock QueueLock, Lock SplitPublishLock) GetPrivateLocks( SessionPublishQueue queue) {