[Server] Improve Request Queue, fix Event handling, add Tests

Libraries/Opc.Ua.Server/Configuration/ConfigurationNodeManager.cs

@@ -185,7 +185,7 @@ protected override async ValueTask<NodeState> AddBehaviourToPredefinedNodeAsync(
                             }
                             else
                             {
-                                NodeState serverNode = Server.NodeManager.FindNodeInAddressSpaceAsync(ObjectIds.Server).AsTask().GetAwaiter().GetResult();
+                                NodeState serverNode = await Server.NodeManager.FindNodeInAddressSpaceAsync(ObjectIds.Server).ConfigureAwait(false);
                                 serverNode?.ReplaceChild(context, activeNode);
                             }
                             // remove the reference to server node because it is set as parent

Libraries/Opc.Ua.Server/Diagnostics/DiagnosticsNodeManager.cs

@@ -453,9 +453,10 @@ protected override async ValueTask<NodeState> AddBehaviourToPredefinedNodeAsync(
                 {
                     if (passiveNode is ServerObjectState)
                     {
+                        // add the server object as the root notifier.
+                        await AddRootNotifierAsync(passiveNode, cancellationToken).ConfigureAwait(false);
                         break;
                     }
-
                     var activeNode = new ServerObjectState(passiveNode.Parent);
                     activeNode.Create(context, passiveNode);
 

Tests/Opc.Ua.Client.Tests/ClientTest.cs

@@ -981,7 +981,6 @@ public async Task ReconnectSession_ReuseUsertokenPolicyAsync(
                 null,
                 (EndpointDescription)endpoint.Description.MemberwiseClone(),
                 endpoint.Configuration);
-            endpoint.Description.SecurityMode = MessageSecurityMode.Sign;
             ConfiguredEndpoint tokenPolicyEndpoint = await ClientFixture
                 .GetEndpointAsync(ServerUrl, userTokenPolicy, Endpoints)
                 .ConfigureAwait(false);
@@ -997,6 +996,12 @@ public async Task ReconnectSession_ReuseUsertokenPolicyAsync(
                     $"No UserTokenPolicy found for {userIdentity.TokenType}" +
                     $" / {userIdentity.IssuedTokenType}");
             }
+            if (identityPolicy.SecurityPolicyUri != userTokenPolicy)
+            {
+                NUnit.Framework.Assert.Fail(
+                    $"UserTokenPolicy SecurityPolicyUri {identityPolicy.SecurityPolicyUri} does not match test expected SecurityPolicyUri {userTokenPolicy}" +
+                    $"Please fix Test parameters or Test server configuration");
+            }
             userIdentity.PolicyId = identityPolicy.PolicyId;
 
             // the active channel
@@ -1006,6 +1011,8 @@ public async Task ReconnectSession_ReuseUsertokenPolicyAsync(
             try
             {
                 await session1.ReconnectAsync(null, null).ConfigureAwait(false);
+                Assert.That(session1.Identity.PolicyId, Is.EqualTo(identityPolicy.PolicyId),
+                    "User Token PolicyId needs to be preserved after reconnect.");
             }
             finally
             {

Tests/Opc.Ua.Client.Tests/ClientTestFramework.cs

@@ -239,7 +239,12 @@ public virtual async Task CreateReferenceServerFixtureAsync(
                 {
                     IssuedTokenType = Profiles.JwtUserToken
                 });
-
+            ServerFixture.Config.ServerConfiguration.UserTokenPolicies.Add(
+                new UserTokenPolicy(UserTokenType.UserName)
+                {
+                    SecurityPolicyUri
+                        = SecurityPolicies.Basic128Rsa15
+                });
             ServerFixture.Config.ServerConfiguration.UserTokenPolicies.Add(
                 new UserTokenPolicy(UserTokenType.UserName)
                 {

Tests/Opc.Ua.Client.Tests/LoadTest.cs

@@ -35,6 +35,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using NUnit.Framework;
+using Opc.Ua.Server;
 using Assert = NUnit.Framework.Legacy.ClassicAssert;
 
 namespace Opc.Ua.Client.Tests
@@ -504,5 +505,186 @@ public async Task ServerReadLoadTestAsync()
                 await Task.WhenAll(closeTasks).ConfigureAwait(false);
             }
         }
+
+        /// <summary>
+        /// Load test a server with multiple sessions subscribing to events.
+        /// Verifies that event handling remains responsive under load with many concurrent
+        /// event subscriptions, addressing the performance issue where server becomes
+        /// unresponsive as the number of monitored items increases.
+        /// </summary>
+        [Test]
+        [Explicit]
+        [Order(120)]
+        public async Task ServerEventSubscribeLoadTestAsync()
+        {
+            const int sessionCount = 10;
+            const int subscriptionsPerSession = 5;
+            const int publishingInterval = 100;
+            const int testDurationSeconds = 30;
+
+            var sessions = new ConcurrentBag<ISession>();
+            long eventsReceived = 0;
+            long totalDelayTicks = 0;
+
+            try
+            {
+                TestContext.Out.WriteLine($"Creating {sessionCount} sessions with {subscriptionsPerSession} event subscriptions each.");
+
+                // Create sessions with event subscriptions in parallel.
+                var createSessionTasks = new List<Task>();
+                for (int i = 0; i < sessionCount; i++)
+                {
+                    createSessionTasks.Add(Task.Run(async () =>
+                    {
+                        ISession session = await ClientFixture.ConnectAsync(
+                            ServerUrl,
+                            SecurityPolicies.Basic256Sha256).ConfigureAwait(false);
+                        sessions.Add(session);
+
+                        for (int j = 0; j < subscriptionsPerSession; j++)
+                        {
+                            var subscription = new Subscription(session.DefaultSubscription)
+                            {
+                                PublishingInterval = publishingInterval
+                            };
+
+                            // Build an event filter for the base event type.
+                            var eventFilter = new EventFilter();
+                            eventFilter.AddSelectClause(
+                                ObjectTypes.BaseEventType,
+                                QualifiedName.From(BrowseNames.EventId));
+                            eventFilter.AddSelectClause(
+                                ObjectTypes.BaseEventType,
+                                QualifiedName.From(BrowseNames.EventType));
+                            eventFilter.AddSelectClause(
+                                ObjectTypes.BaseEventType,
+                                QualifiedName.From(BrowseNames.SourceNode));
+                            eventFilter.AddSelectClause(
+                                ObjectTypes.BaseEventType,
+                                QualifiedName.From(BrowseNames.SourceName));
+                            eventFilter.AddSelectClause(
+                                ObjectTypes.BaseEventType,
+                                QualifiedName.From(BrowseNames.Time));
+                            eventFilter.AddSelectClause(
+                                ObjectTypes.BaseEventType,
+                                QualifiedName.From(BrowseNames.Message));
+                            eventFilter.AddSelectClause(
+                                ObjectTypes.BaseEventType,
+                                QualifiedName.From(BrowseNames.Severity));
+
+                            // Monitor the Server node for events.
+                            var monitoredItem = new MonitoredItem(subscription.DefaultItem)
+                            {
+                                StartNodeId = ObjectIds.Server,
+                                AttributeId = Attributes.EventNotifier,
+                                MonitoringMode = MonitoringMode.Reporting,
+                                Filter = eventFilter,
+                                QueueSize = 10000
+                            };
+
+                            subscription.FastEventCallback = (sub, notification, _) =>
+                            {
+                                Interlocked.Add(ref eventsReceived, notification.Events.Count);
+                                foreach (EventFieldList fieldList in notification.Events)
+                                {
+                                    if (fieldList.EventFields.Count > 4 && fieldList.EventFields[4].Value is DateTime eventTime)
+                                    {
+                                        TimeSpan delay = DateTime.UtcNow - eventTime.ToUniversalTime();
+                                        Interlocked.Add(ref totalDelayTicks, delay.Ticks);
+                                    }
+                                }
+                            };
+
+                            subscription.AddItem(monitoredItem);
+                            session.AddSubscription(subscription);
+                            await subscription.CreateAsync().ConfigureAwait(false);
+                        }
+                    }));
+                }
+
+                await Task.WhenAll(createSessionTasks).ConfigureAwait(false);
+
+                const int totalSubscriptions = sessionCount * subscriptionsPerSession;
+                TestContext.Out.WriteLine(
+                    $"Created {totalSubscriptions} event subscriptions across {sessionCount} sessions.");
+                TestContext.Out.WriteLine($"Generating events on the server for {testDurationSeconds} seconds...");
+
+                // Generate events directly on the server to stress-test event delivery.
+                IServerInternal serverInternal = ReferenceServer.CurrentInstance;
+                ISystemContext serverContext = serverInternal.DefaultSystemContext;
+                int eventCount = 0;
+
+                using var testCts = new CancellationTokenSource(TimeSpan.FromSeconds(testDurationSeconds));
+
+                var sw = System.Diagnostics.Stopwatch.StartNew();
+                while (!testCts.IsCancellationRequested)
+                {
+                    var e = new BaseEventState(null);
+                    e.Initialize(
+                        serverContext,
+                        serverInternal.ServerObject,
+                        EventSeverity.Medium,
+                        new LocalizedText($"LoadTest event {eventCount}"));
+                    serverInternal.ReportEvent(serverContext, e);
+                    eventCount++;
+                }
+
+                sw.Stop();
+                TestContext.Out.WriteLine($"Generated {eventCount} events in {sw.ElapsedMilliseconds} ms " +
+                    $"({eventCount / sw.Elapsed.TotalSeconds:F0} events/sec).");
+
+                // Wait for subscriptions to deliver the events.
+                // Allow enough publishing intervals for all notifications to be sent and acknowledged.
+                const int publishingIntervalsToWait = 20;
+                await Task.Delay(publishingInterval * publishingIntervalsToWait).ConfigureAwait(false);
+
+                long expectedTotal = (long)eventCount * totalSubscriptions;
+                long received = Interlocked.Read(ref eventsReceived);
+
+                TestContext.Out.WriteLine($"Expected event notifications : {expectedTotal}");
+                TestContext.Out.WriteLine($"Received event notifications : {received}");
+
+                double receiveRatio = expectedTotal > 0 ? (double)received / expectedTotal : 0;
+                TestContext.Out.WriteLine($"Receive ratio: {receiveRatio:P2}");
+
+                if (received > 0)
+                {
+                    long averageDelayTicks = Interlocked.Read(ref totalDelayTicks) / received;
+                    var averageDelay = TimeSpan.FromTicks(averageDelayTicks);
+                    TestContext.Out.WriteLine($"Average event delivery delay: {averageDelay.TotalMilliseconds:F2} ms");
+                }
+
+                NUnit.Framework.Assert.That(
+                    received,
+                    Is.GreaterThan(0),
+                    "No event notifications were received.");
+
+                NUnit.Framework.Assert.That(
+                    receiveRatio,
+                    Is.GreaterThan(0.99),
+                    "The event notification receive ratio is too low.");
+            }
+            finally
+            {
+                // Cleanup all sessions.
+                var closeTasks = sessions.Select(session => Task.Run(async () =>
+                {
+                    try
+                    {
+                        if (session.Connected)
+                        {
+                            await session.CloseAsync().ConfigureAwait(false);
+                        }
+
+                        session.Dispose();
+                    }
+                    catch (Exception ex)
+                    {
+                        TestContext.Out.WriteLine($"Failed to close session: {ex.Message}");
+                    }
+                })).ToList();
+                await Task.WhenAll(closeTasks).ConfigureAwait(false);
+            }
+        }
     }
 }

Tests/Opc.Ua.Client.Tests/SubscriptionUnitTests.cs

@@ -42,7 +42,7 @@ namespace Opc.Ua.Client.Tests
     [Parallelizable]
     public class SubscriptionUnitTests
     {
-        private class SubscriptionContainer : IDisposable
+        private sealed class SubscriptionContainer : IDisposable
         {
             private readonly CancellationTokenRegistration m_tokedCancellation;
             public Subscription Subscription { get; }
@@ -149,25 +149,23 @@ private static ISession BuildSessionMock(Func<uint, uint, bool> republishHandler
 
         private static NotificationMessage[] BuildMessages(int count)
         {
-            return Enumerable
+            return [.. Enumerable
                 .Range(1, count)
                 .Select(sequenceNumber => new NotificationMessage
                 {
                     SequenceNumber = (uint)sequenceNumber,
                     NotificationData = [new(new DataChangeNotification { SequenceNumber = (uint)sequenceNumber })]
                 })
-                .Prepend(new())//stub to compensate sequenceNumbers start from 1. Should be ignored
-                .ToArray();
+                .Prepend(new())];
         }
 
         private static async Task<SubscriptionContainer> BuildSubscriptionAsync(
             NotificationMessage[] messagesToProcess,
             bool sequentialPublishing,
             CancellationToken cancellationToken)
         {
-            TaskCompletionSource<bool>[] messageAwaiters = messagesToProcess
-                .Select(_ => new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously))
-                .ToArray();
+            TaskCompletionSource<bool>[] messageAwaiters =
+                [.. messagesToProcess.Select(_ => new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously))];
             messageAwaiters[0].SetResult(true);
             List<uint> availableSequenceNumbers = [.. messagesToProcess.Skip(1).Select(x => x.SequenceNumber)];
 
@@ -177,13 +175,10 @@ private static async Task<SubscriptionContainer> BuildSubscriptionAsync(
                 {
                     PublishingEnabled = true,
                     SequentialPublishing = sequentialPublishing,
-                    MaxMessageCount = messagesToProcess.Length,
+                    MaxMessageCount = messagesToProcess.Length
                 })
             {
-                FastDataChangeCallback = (_, message, _) =>
-                {
-                    messageAwaiters[message.SequenceNumber].SetResult(true);
-                },
+                FastDataChangeCallback = (_, message, _) => messageAwaiters[message.SequenceNumber].SetResult(true)
             };
             subscription.Session = BuildSessionMock((subscriptionId, sequenceNumber) =>
             {