diff --git a/Libraries/Opc.Ua.Client/Subscription/Subscription.cs b/Libraries/Opc.Ua.Client/Subscription/Subscription.cs index a849d7e24f..65a7e6f335 100644 --- a/Libraries/Opc.Ua.Client/Subscription/Subscription.cs +++ b/Libraries/Opc.Ua.Client/Subscription/Subscription.cs @@ -46,9 +46,13 @@ public class Subscription : ISnapshotRestore, IDisposable, IC { private const int kMinKeepAliveTimerInterval = 1000; private const int kKeepAliveTimerMargin = 1000; - private const int kRepublishMessageTimeout = 2500; private const int kRepublishMessageExpiredTimeout = 10000; + /// + /// Duration to wait before republishing missed notification + /// + public const int RepublishMessageTimeout = 2500; + /// /// Create subscription /// @@ -1619,9 +1623,33 @@ public void SaveMessageInCache( // fill in any gaps in the queue LinkedListNode? node = m_incomingMessages.First; + if (node is not null) + { + //gaps between m_lastSequenceNumberProcessed and starting node + LinkedListNode currentNode = node; + for (uint i = node.Value.SequenceNumber; i > (m_lastSequenceNumberProcessed + 1); i--) + { + var placeholder = new IncomingMessage + { + SequenceNumber = i - 1, + Timestamp = now, + TickCount = tickCount + }; + currentNode = m_incomingMessages.AddBefore(currentNode, placeholder); + + m_logger.LogInformation( + "Session {SessionId}, subscription {SubscriptionName} ({SubscriptionId}): " + + "added placeholder for missing incoming message with sequence number {MissingSequenceNumber}", + Session?.SessionId, + DisplayName, + Id, + placeholder.SequenceNumber); + } + } while (node != null) { + //gaps between neighbouring nodes entry = node.Value; LinkedListNode? next = node.Next; @@ -1969,8 +1997,8 @@ private void ProcessTransferredSequenceNumbers(UInt32Collection availableSequenc // only republish consecutive sequence numbers // triggers the republish mechanism immediately, // if event is in the past - DateTime now = DateTime.UtcNow.AddMilliseconds(-kRepublishMessageTimeout * 2); - int tickCount = HiResClock.TickCount - (kRepublishMessageTimeout * 2); + DateTime now = DateTime.UtcNow.AddMilliseconds(-RepublishMessageTimeout * 2); + int tickCount = HiResClock.TickCount - (RepublishMessageTimeout * 2); uint lastSequenceNumberToRepublish = m_lastSequenceNumberProcessed - 1; int availableNumbers = availableSequenceNumbers.Count; int republishMessages = 0; @@ -2505,7 +2533,7 @@ private async Task OnMessageReceivedAsync(CancellationToken ct) // tolerate if a single request was received out of order if (ii.Next.Next != null && (HiResClock.TickCount - - ii.Value.TickCount) > kRepublishMessageTimeout) + ii.Value.TickCount) > RepublishMessageTimeout) { ii.Value.Republished = true; publishStateChangedMask |= PublishStateChangedMask.Republish; diff --git a/Tests/Opc.Ua.Client.Tests/SubscriptionUnitTests.cs b/Tests/Opc.Ua.Client.Tests/SubscriptionUnitTests.cs new file mode 100644 index 0000000000..ad0ffdc219 --- /dev/null +++ b/Tests/Opc.Ua.Client.Tests/SubscriptionUnitTests.cs @@ -0,0 +1,354 @@ +/* ======================================================================== + * Copyright (c) 2005-2025 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.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Moq; +using NUnit.Framework; +using Opc.Ua.Tests; + +namespace Opc.Ua.Client.Tests +{ + [TestFixture] + [Parallelizable] + public class SubscriptionUnitTests + { + private class SubscriptionContainer : IDisposable + { + private readonly CancellationTokenRegistration m_tokedCancellation; + public Subscription Subscription { get; } + public Task[] ProcessedMessages { get; } + + public SubscriptionContainer(Subscription subscription, TaskCompletionSource[] awaiters, CancellationToken token) + { + Subscription = subscription; + ProcessedMessages = [.. awaiters.Select(x => x.Task)]; + m_tokedCancellation = token.Register(() => + { + foreach (TaskCompletionSource awaiter in awaiters) + { + awaiter.TrySetCanceled(); + } + }); + } + + public void Dispose() + { + m_tokedCancellation.Dispose(); + Subscription.Dispose(); + } + } + + private static Task AwaitForRepublishTimeout(CancellationToken ct) + { + return Task.Delay(Subscription.RepublishMessageTimeout + 100, ct); + } + + private static ISession BuildSessionMock(Func republishHandler) + { + uint subscriptionIdSeed = 0u; + + var session = new Mock(); + session + .Setup(x => x.RepublishAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(( + subscriptionId, + sequenceNumber, + ct) => + { + if (subscriptionId > subscriptionIdSeed) + { + return (true, StatusCodes.BadSubscriptionIdInvalid); + } + if (republishHandler(subscriptionId, sequenceNumber)) + { + return (true, ServiceResult.Good); + } + return (true, StatusCodes.BadMessageNotAvailable); + }); + session + .Setup(x => x + .CreateSubscriptionAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny() + ) + ) + .ReturnsAsync< + RequestHeader, + double, + uint, + uint, + uint, + bool, + byte, + CancellationToken, + ISession, + CreateSubscriptionResponse + >(( + requestHeader, + requestedPublishingInterval, + requestedLifetimeCount, + requestedMaxKeepAliveCount, + maxNotificationsPerPublish, + publishingEnabled, + priority, + ct + ) => new() { SubscriptionId = ++subscriptionIdSeed }); + session + .Setup(x => x.SetPublishingModeAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()) + ) + .ReturnsAsync< + RequestHeader, + bool, + UInt32Collection, + CancellationToken, + ISession, + SetPublishingModeResponse + >((requestHeader, publishingEnabled, subscriptionIds, ct) => new() + { + Results = [.. subscriptionIds.Select(id => id > subscriptionIdSeed ? StatusCodes.BadSubscriptionIdInvalid : StatusCodes.Good)], + DiagnosticInfos = [.. subscriptionIds.Select(_ => new DiagnosticInfo())] + }); + return session.Object; + } + + private static NotificationMessage[] BuildMessages(int count) + { + 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(); + } + + private static async Task BuildSubscriptionAsync( + NotificationMessage[] messagesToProcess, + bool sequentialPublishing, + CancellationToken cancellationToken) + { + TaskCompletionSource[] messageAwaiters = messagesToProcess + .Select(_ => new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)) + .ToArray(); + messageAwaiters[0].SetResult(true); + List availableSequenceNumbers = [.. messagesToProcess.Skip(1).Select(x => x.SequenceNumber)]; + + var subscription = new Subscription( + NUnitTelemetryContext.Create(), + new() + { + PublishingEnabled = true, + SequentialPublishing = sequentialPublishing, + MaxMessageCount = messagesToProcess.Length, + }) + { + FastDataChangeCallback = (_, message, _) => + { + messageAwaiters[message.SequenceNumber].SetResult(true); + }, + }; + subscription.Session = BuildSessionMock((subscriptionId, sequenceNumber) => + { + //simplified republish emulation + if (subscription.Id == subscriptionId && availableSequenceNumbers.Remove(sequenceNumber)) + { + subscription.SaveMessageInCache(null, messagesToProcess[sequenceNumber]); + return true; + } + return false; + }); + await subscription.CreateAsync(cancellationToken).ConfigureAwait(false); + return new(subscription, messageAwaiters, cancellationToken); + } + + /// + /// Set up a Server and a Client instance. + /// + [OneTimeSetUp] + public void OneTimeSetUp() + { + TestContext.AddFormatter((obj) => + { + if (obj is NotificationMessage other) + { + return $"{nameof(NotificationMessage)}: {other.SequenceNumber}"; + } + return null; + }); + } + + [Test] + [Explicit("Test shows possibility for broken order of notifications during sequential publishing")] + [CancelAfter(Subscription.RepublishMessageTimeout * 6)] + public async Task UnorderedMessagesWouldBeLostForSequentialPublishingAsync(CancellationToken ct) + { + NotificationMessage[] messages = BuildMessages(5); + using SubscriptionContainer container = await BuildSubscriptionAsync(messages, sequentialPublishing: true, ct).ConfigureAwait(false); + Subscription subscription = container.Subscription; + + subscription.SaveMessageInCache([], messages[3]); + await AwaitForRepublishTimeout(ct).ConfigureAwait(false); + subscription.SaveMessageInCache([], messages[2]);//this one will be processed, because one message out of order is always awaited + await AwaitForRepublishTimeout(ct).ConfigureAwait(false); + subscription.SaveMessageInCache([], messages[4]); + await AwaitForRepublishTimeout(ct).ConfigureAwait(false); + subscription.SaveMessageInCache([], messages[1]);//this one should be lost due to expiration timout + await AwaitForRepublishTimeout(ct).ConfigureAwait(false); + subscription.SaveMessageInCache([], messages[5]); + + await Task.WhenAll( + container.ProcessedMessages[2], + container.ProcessedMessages[3], + container.ProcessedMessages[4], + container.ProcessedMessages[5] + ).ConfigureAwait(false); + + Assert.That(subscription.Notifications, Is.EqualTo([messages[2], messages[3], messages[4], messages[5]])); + } + + [Test] + [CancelAfter(Subscription.RepublishMessageTimeout * 3)] + public async Task WillRestoreOrderOfTwoMessagesForSequentialPublishingAsync(CancellationToken ct) + { + NotificationMessage[] messages = BuildMessages(3); + using SubscriptionContainer container = await BuildSubscriptionAsync(messages, sequentialPublishing: true, ct).ConfigureAwait(false); + Subscription subscription = container.Subscription; + + subscription.SaveMessageInCache([], messages[2]); + await AwaitForRepublishTimeout(ct).ConfigureAwait(false); + subscription.SaveMessageInCache([], messages[1]); + await AwaitForRepublishTimeout(ct).ConfigureAwait(false); + subscription.SaveMessageInCache([], messages[3]); + + await Task.WhenAll(container.ProcessedMessages).ConfigureAwait(false); + Assert.That(subscription.Notifications, Is.EqualTo(messages.Skip(1))); + } + + [Theory] + [CancelAfter(Subscription.RepublishMessageTimeout * 7)] + public async Task WillAbandonMissedMessagesIfThereAreNoAvailableSequenceNumbersAsync(bool sequentialPublishing, CancellationToken ct) + { + int[] sequenceNumbersToPublish = [/*1..2 gap*/ 3, 4, /*5..7 gap*/ 8, 9, 10]; + NotificationMessage[] messages = BuildMessages(sequenceNumbersToPublish[^1]); + NotificationMessage[] messagesToPublish = [.. sequenceNumbersToPublish.Select(x => messages[x])]; + using SubscriptionContainer container = await BuildSubscriptionAsync(messages, sequentialPublishing, ct).ConfigureAwait(false); + Subscription subscription = container.Subscription; + + foreach (NotificationMessage message in messagesToPublish) + { + subscription.SaveMessageInCache([], message); + await AwaitForRepublishTimeout(ct).ConfigureAwait(false); + } + await Task.WhenAll(sequenceNumbersToPublish.Select(x => container.ProcessedMessages[x])).ConfigureAwait(false); + + if (sequentialPublishing) + { + Assert.That(subscription.Notifications, Is.EqualTo(messagesToPublish)); + } + else + { + Assert.That(subscription.Notifications, Is.EquivalentTo(messagesToPublish)); + } + } + + [Theory] + [CancelAfter(Subscription.RepublishMessageTimeout * 3)] + public async Task WillRepublishIfMissedMessagesOnFirstPublishAsync( + bool sequentialPublishing, + [Values(2, 5, 8, 11)] int gapEnd, + CancellationToken ct) + { + int finalMessage = gapEnd + 1; + NotificationMessage[] messages = BuildMessages(finalMessage); + uint[] sequenceNumbersOfGap = [.. messages.Skip(1).Take(gapEnd).Select(x => x.SequenceNumber)]; + using SubscriptionContainer container = await BuildSubscriptionAsync(messages, sequentialPublishing, ct).ConfigureAwait(false); + Subscription subscription = container.Subscription; + + subscription.SaveMessageInCache(sequenceNumbersOfGap, messages[gapEnd]); + await AwaitForRepublishTimeout(ct).ConfigureAwait(false); + subscription.SaveMessageInCache(sequenceNumbersOfGap, messages[finalMessage]); + await Task.WhenAll(container.ProcessedMessages).ConfigureAwait(false); + + if (sequentialPublishing) + { + Assert.That(subscription.Notifications, Is.EqualTo(messages.Skip(1))); + } + else + { + Assert.That(subscription.Notifications, Is.EquivalentTo(messages.Skip(1))); + } + } + + [Theory] + [CancelAfter(Subscription.RepublishMessageTimeout * 3)] + public async Task WillRepublishIfMissedMessagesInBetweenOfPublishesAsync( + bool sequentialPublishing, + [Values(1, 6, 9)] int gapStart, + [Values(1, 6, 9)] int gapSize, + CancellationToken ct) + { + int gapEnd = gapStart + gapSize + 1; + int finalMessage = gapEnd + 1; + NotificationMessage[] messages = BuildMessages(finalMessage); + uint[] sequenceNumbersOfGap = [.. messages.Skip(1).Skip(gapStart).Take(gapSize).Select(x => x.SequenceNumber)]; + using SubscriptionContainer container = await BuildSubscriptionAsync(messages, sequentialPublishing, ct).ConfigureAwait(false); + Subscription subscription = container.Subscription; + + for (int i = 1; i <= gapStart; i++) + { + subscription.SaveMessageInCache([], messages[i]); + } + subscription.SaveMessageInCache(sequenceNumbersOfGap, messages[gapEnd]); + await AwaitForRepublishTimeout(ct).ConfigureAwait(false); + subscription.SaveMessageInCache(sequenceNumbersOfGap, messages[finalMessage]); + await Task.WhenAll(container.ProcessedMessages).ConfigureAwait(false); + + if (sequentialPublishing) + { + Assert.That(subscription.Notifications, Is.EqualTo(messages.Skip(1))); + } + else + { + Assert.That(subscription.Notifications, Is.EquivalentTo(messages.Skip(1))); + } + } + } +}