mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-20 08:41:51 +08:00
Marc Schier
49305d9085
# Description V2 subscription engine feature parity round to make the classic engine eligible for deprecation. Splits the test surface so the classic and V2 engines can evolve independently, ports the gaps the classic-engine integration tests had been exercising, and lands the public V2 API additions needed to close them. - Splits `Tests/Opc.Ua.Subscriptions.Tests` into a classic (`Opc.Ua.Subscriptions.Classic.Tests`) and a V2 (`Opc.Ua.Subscriptions.Tests`) project to shrink CI time and let the two engines diverge cleanly. - Adds the V2 public-surface gaps required by the test ports (`Subscription.ServerId` on the class, `SetSubscriptionDurableAsync`, `SendInitialValuesOnTransfer`, unified `OnSubscriptionStateChangedAsync` handler callback, per-item `ConditionRefreshAsync`, fluent `LoadSubscriptionsAsync(Stream, …)` and `Subscription.Snapshot()` / `RestoreAsync()` for save/load + transfer). - Lands handler-centric design (no `PublishingStopped` property handlers maintain derived state); always-sequential per-subscription delivery documented + tested. - **Adds short and long haul stability tests for the V2 `ManagedSession` + subscription engine. Closes #3744** (see "Issue #3744 Haul tests" below). - **Migrates SubscriptionStateSnapshot / MonitoredItemStateSnapshot to `[DataType]` source-generator codecs**. - `SubscriptionDurableV2Tests.cs` 5 V2 ports of the classic durable test patterns. - `V2FollowUpCoverageTests.cs` handler state-change callback, fluent stream-based `LoadSubscriptionsAsync`, `SendInitialValuesOnTransfer`, snapshot edge cases (empty / `DataChangeFilter` round-trip / concurrent mutation). - `SubscriptionFailoverV2Tests.cs` forced channel-break with and without `WithTransferSubscriptionsOnRecreate`. - `MonitoredItemConditionRefreshLiveV2Tests.cs` — live `ConditionRefresh` against the reference server's event source; asserts `RefreshStart` / `RefreshEnd` events flow through the handler. ## Fixes - #3744 Haul tests (V2 ManagedSession + subscription engine) `Tests/Opc.Ua.Sessions.Tests/ManagedSessionStabilityTest.cs` adds the short/long haul coverage requested by [#3744](https://github.com/OPCFoundation/UA-.NETStandard/issues/3744) plus the bonus fault-injection variant: | Test | Duration | Category | Purpose | |---|---|---|---| | `ShortHaulManagedSessionV2Async` | 2 minutes | `ManagedSessionHaul` (`[Explicit]`) | Quick local sanity check of the V2 monotonicity contract. | | `LongHaulManagedSessionV2Async` | 90 minutes (configurable via `TEST_DURATION_MINUTES`) | `ManagedSessionHaul` (`[Explicit]`) | Long-running V2 stability + monotonicity. | | `LongHaulManagedSessionWithFaultInjectionV2Async` | 90 minutes (configurable) + fault every `FAULT_INJECTION_INTERVAL_SECONDS` (default 60s) | `ManagedSessionHaul` (`[Explicit]`) | **Bonus** — force-closes the subscriber transport channel periodically; verifies V2 manager transfers/recreates the subscription and continues monotonic delivery across reconnects. | **Pattern.** A writer `ManagedSession` increments a `uint32` counter and writes it into `Scalar_Static_Mass_UInt32_UInt32_00` every 250ms. A subscriber `ManagedSession` runs a V2 subscription with a single monitored item on the same node. `MonotonicCounterHandler` asserts each received sample is *strictly greater than* the previous one. V2 `ISubscriptionNotificationHandler` per-subscription ordering guarantee. Skipped values (sampling rate < write rate) are tolerated; duplicates and reorderings are violations.
837 lines
36 KiB
C#
837 lines
36 KiB
C#
/* ========================================================================
|
|
* 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/
|
|
* ======================================================================*/
|
|
|
|
// CA2007: tests run without a SynchronizationContext; ConfigureAwait(false)
|
|
// adds noise without a behavioural benefit. Disabled file-level for the suite.
|
|
#pragma warning disable CA2007
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Moq;
|
|
using NUnit.Framework;
|
|
using Opc.Ua.Client.Subscriptions.Fakes;
|
|
using Opc.Ua.Tests;
|
|
|
|
namespace Opc.Ua.Client.Subscriptions
|
|
{
|
|
[TestFixture]
|
|
public sealed class MessageProcessorTests
|
|
{
|
|
[SetUp]
|
|
public void SetUp()
|
|
{
|
|
m_completion = new FakeMessageAckQueue();
|
|
m_telemetry = NUnitTelemetryContext.Create();
|
|
m_mockServices = new Mock<ISubscriptionServiceSetClientMethods>();
|
|
}
|
|
|
|
[Test]
|
|
public async Task DisposeAsyncShouldCompleteMessageWriterAndCancelTokenAsync()
|
|
{
|
|
// Arrange
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 3
|
|
};
|
|
|
|
// Act
|
|
await sut.DisposeAsync().ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.PublishState, Is.EqualTo(PublishState.Completed));
|
|
Assert.That(m_completion.CompletedSubscriptions,
|
|
Is.EqualTo(new uint[] { 3 }));
|
|
}
|
|
|
|
[Test]
|
|
public async Task OnPublishReceivedKeepAliveShouldDispatchKeepAliveAsync()
|
|
{
|
|
// Arrange
|
|
var message = new NotificationMessage
|
|
{
|
|
SequenceNumber = 3
|
|
};
|
|
var availableSequenceNumbers = new List<uint> { 1, 2, 3 };
|
|
var stringTable = new List<string> { "test" };
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 3
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// Act
|
|
await sut.OnPublishReceivedAsync(message, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.KeepAliveNotificationReceived.WaitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.True);
|
|
Assert.That(sut.AvailableInRetransmissionQueue, Is.EqualTo(availableSequenceNumbers));
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(3));
|
|
Assert.That(sut.DataChangeNotificationReceived.IsSet, Is.False);
|
|
|
|
// Arrange
|
|
sut.KeepAliveNotificationReceived.Reset();
|
|
sut.DataChangeNotificationReceived.Reset();
|
|
message = new NotificationMessage
|
|
{
|
|
SequenceNumber = 4,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new DataChangeNotification
|
|
{
|
|
MonitoredItems =
|
|
[
|
|
new MonitoredItemNotification()
|
|
]
|
|
})
|
|
]
|
|
};
|
|
|
|
// Act
|
|
await sut.OnPublishReceivedAsync(message, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.DataChangeNotificationReceived.WaitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.AvailableInRetransmissionQueue, Is.EqualTo(availableSequenceNumbers));
|
|
Assert.That(sut.DataChangeNotificationReceived.IsSet, Is.True);
|
|
Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.False);
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(4));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ProcessMessageAsyncShouldRepublishMissingMessagesAsync()
|
|
{
|
|
// Arrange
|
|
var availableSequenceNumbers = new List<uint> { 1, 2, 3 };
|
|
var stringTable = new List<string> { "test" };
|
|
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 2
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 1
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
|
|
m_mockServices
|
|
.Setup(c => c.RepublishAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<uint>(id => id == sut.Id),
|
|
It.Is<uint>(s => s == 2),
|
|
It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(new RepublishResponse
|
|
{
|
|
NotificationMessage = new NotificationMessage
|
|
{
|
|
SequenceNumber = 2,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new DataChangeNotification
|
|
{
|
|
MonitoredItems =
|
|
[
|
|
new MonitoredItemNotification()
|
|
]
|
|
})
|
|
]
|
|
}
|
|
})
|
|
.Verifiable(Times.Once);
|
|
|
|
// Act
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 3,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new EventNotificationList
|
|
{
|
|
Events =
|
|
[
|
|
new EventFieldList()
|
|
]
|
|
})
|
|
]
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.EventNotificationReceived.WaitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.EventNotificationReceived.IsSet, Is.True);
|
|
Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.True);
|
|
Assert.That(sut.DataChangeNotificationReceived.IsSet, Is.True);
|
|
|
|
m_mockServices.Verify();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ProcessReceivedMessagesAsyncShouldProcessMessagesInOrderAsync()
|
|
{
|
|
var availableSequenceNumbers = new List<uint> { 1, 2, 3 };
|
|
var stringTable = new List<string> { "test" };
|
|
|
|
// Arrange
|
|
NotificationMessage[] messages = [.. Enumerable.Range(2, 99).Select(i => new NotificationMessage
|
|
{
|
|
SequenceNumber = (uint)i
|
|
})];
|
|
|
|
UnsecureRandom.Shared.Shuffle(messages);
|
|
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 3
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
sut.Block.Wait();
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 1u
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
foreach (NotificationMessage message in messages)
|
|
{
|
|
await sut.OnPublishReceivedAsync(message, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
}
|
|
sut.Block.Release();
|
|
|
|
// Act
|
|
await Task.Delay(10).ConfigureAwait(false);
|
|
|
|
Assert.That(sut.ReceivedSequenceNumbers, Is.EqualTo(
|
|
Enumerable.Range(1, 100).Select(i => (uint)i)));
|
|
Assert.That(sut.AvailableInRetransmissionQueue, Is.EqualTo(availableSequenceNumbers));
|
|
Assert.That(sut.DataChangeNotificationReceived.IsSet, Is.False);
|
|
Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.True);
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(100));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task DuplicateSequenceNumberShouldNotRedispatchAsync()
|
|
{
|
|
var availableSequenceNumbers = new List<uint> { 1, 2, 3 };
|
|
var stringTable = new List<string> { "test" };
|
|
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 7
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
var message = new NotificationMessage
|
|
{
|
|
SequenceNumber = 5,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new DataChangeNotification
|
|
{
|
|
MonitoredItems =
|
|
[
|
|
new MonitoredItemNotification()
|
|
]
|
|
})
|
|
]
|
|
};
|
|
|
|
// First arrival
|
|
await sut.OnPublishReceivedAsync(message, availableSequenceNumbers, stringTable)
|
|
.ConfigureAwait(false);
|
|
await sut.DataChangeNotificationReceived.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
int firstCount = sut.ReceivedSequenceNumbers.Count;
|
|
Assert.That(firstCount, Is.GreaterThanOrEqualTo(1));
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(5));
|
|
|
|
// Reset signaling
|
|
sut.DataChangeNotificationReceived.Reset();
|
|
|
|
// Same sequence number arriving again
|
|
await sut.OnPublishReceivedAsync(message, availableSequenceNumbers, stringTable)
|
|
.ConfigureAwait(false);
|
|
// Give the processor a moment in case it tries to dispatch
|
|
await Task.Delay(50).ConfigureAwait(false);
|
|
|
|
// The duplicate should not re-fire the data-change handler.
|
|
Assert.That(
|
|
sut.DataChangeNotificationReceived.IsSet, Is.False,
|
|
"Duplicate sequence number must not re-dispatch the notification");
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(5));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task KeepAliveInterleavedWithNotificationsAsync()
|
|
{
|
|
var availableSequenceNumbers = new List<uint> { 1, 2, 3 };
|
|
var stringTable = new List<string> { "test" };
|
|
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 5
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// 1: notification
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 1,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new DataChangeNotification
|
|
{
|
|
MonitoredItems = [new MonitoredItemNotification()]
|
|
})
|
|
]
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.DataChangeNotificationReceived.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
// 2: keep-alive (no NotificationData)
|
|
sut.KeepAliveNotificationReceived.Reset();
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 2
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.KeepAliveNotificationReceived.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
// 3: another notification
|
|
sut.DataChangeNotificationReceived.Reset();
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 3,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new DataChangeNotification
|
|
{
|
|
MonitoredItems = [new MonitoredItemNotification()]
|
|
})
|
|
]
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.DataChangeNotificationReceived.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
// All three sequence numbers should appear in order.
|
|
Assert.That(sut.ReceivedSequenceNumbers,
|
|
Is.EqualTo(new uint[] { 1, 2, 3 }));
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(3));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task EmptyNotificationDataIsTreatedAsKeepAliveAsync()
|
|
{
|
|
var availableSequenceNumbers = new List<uint> { 1, 2, 3 };
|
|
var stringTable = new List<string> { "test" };
|
|
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 9
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// NotificationData is null/default.
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 7
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.KeepAliveNotificationReceived.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.True);
|
|
Assert.That(sut.DataChangeNotificationReceived.IsSet, Is.False);
|
|
Assert.That(sut.EventNotificationReceived.IsSet, Is.False);
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(7));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task RepublishFailureLogsButContinuesProcessingAsync()
|
|
{
|
|
var availableSequenceNumbers = new List<uint> { 1, 2, 3 };
|
|
var stringTable = new List<string> { "test" };
|
|
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 11
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// First message at sequence 1.
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 1
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
|
|
// Republish for the gap (sequence 2) returns an error.
|
|
m_mockServices
|
|
.Setup(c => c.RepublishAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.Is<uint>(id => id == sut.Id),
|
|
It.Is<uint>(s => s == 2),
|
|
It.IsAny<CancellationToken>()))
|
|
.ThrowsAsync(new ServiceResultException(
|
|
StatusCodes.BadMessageNotAvailable))
|
|
.Verifiable(Times.AtLeastOnce);
|
|
|
|
// Skip to sequence 3 to force the gap.
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 3,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new EventNotificationList
|
|
{
|
|
Events = [new EventFieldList()]
|
|
})
|
|
]
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.EventNotificationReceived.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
// Even though republish failed, the next message must still
|
|
// be dispatched.
|
|
Assert.That(sut.EventNotificationReceived.IsSet, Is.True);
|
|
m_mockServices.Verify();
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task SequenceNumberWraparoundAdvancesLastProcessedAsync()
|
|
{
|
|
// Per OPC UA Part 4 §7.30.5, sequence numbers wrap from
|
|
// uint.MaxValue to 1 (skipping 0). The processor must accept
|
|
// this wrap as forward progress, not silently drop the
|
|
// post-wrap message as "old".
|
|
var availableSequenceNumbers = new List<uint>();
|
|
var stringTable = new List<string> { "test" };
|
|
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 13
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// Republish for the first message's "missing" gap (we start
|
|
// from LastSequenceNumberProcessed=0 so no gap is computed
|
|
// for this first one) — but for the second message, the gap
|
|
// is empty because we wrap directly from uint.MaxValue to 1.
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = uint.MaxValue
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.KeepAliveNotificationReceived.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
Assert.That(sut.LastSequenceNumberProcessed,
|
|
Is.EqualTo(uint.MaxValue));
|
|
|
|
sut.KeepAliveNotificationReceived.Reset();
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 1
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.KeepAliveNotificationReceived.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5))
|
|
.ConfigureAwait(false);
|
|
|
|
Assert.That(sut.KeepAliveNotificationReceived.IsSet, Is.True,
|
|
"Wrapped sequence number must be accepted as forward progress, " +
|
|
"not dropped as duplicate/old.");
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(1));
|
|
}
|
|
}
|
|
|
|
[Test]
|
|
public async Task ReceivingTransferStatusUpdateShouldUpdatePublishStateAsync()
|
|
{
|
|
// Arrange
|
|
var availableSequenceNumbers = new List<uint> { 1, 2, 3 };
|
|
var stringTable = new List<string> { "test" };
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 3
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// Act
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 3,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new StatusChangeNotification
|
|
{
|
|
Status = StatusCodes.GoodSubscriptionTransferred
|
|
})
|
|
]
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.StatusChangeNotificationReceived.WaitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.StatusChangeNotificationReceived.IsSet, Is.True);
|
|
Assert.That(sut.ReceivedSequenceNumbers, Does.Contain(3));
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(3));
|
|
Assert.That(sut.PublishState, Is.EqualTo(PublishState.Transferred));
|
|
|
|
sut.StatusChangeNotificationReceived.Reset();
|
|
|
|
// Act
|
|
await sut.OnPublishReceivedAsync(new NotificationMessage
|
|
{
|
|
SequenceNumber = 4,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new StatusChangeNotification
|
|
{
|
|
Status = StatusCodes.BadTimeout
|
|
})
|
|
]
|
|
}, availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.StatusChangeNotificationReceived.WaitAsync().WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
// Assert
|
|
Assert.That(sut.StatusChangeNotificationReceived.IsSet, Is.True);
|
|
Assert.That(sut.ReceivedSequenceNumbers, Does.Contain(4));
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(4));
|
|
Assert.That(sut.PublishState, Is.EqualTo(PublishState.Timeout));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// V2 equivalent of the classic-engine regression captured by
|
|
/// <c>OldMessagesAbandonedAfterRepublishTimeoutInSequentialModeAsync</c>
|
|
/// (currently <c>[Explicit]</c> in <c>Classic/SubscriptionUnitTests.cs</c>).
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Mirrors the classic test's exact arrival order — <c>[3]</c>,
|
|
/// <c>[2]</c>, <c>[4]</c>, <c>[1]</c>, <c>[5]</c> — but queues all five
|
|
/// while the worker is parked inside a sentinel keep-alive dispatch.
|
|
/// The new <see cref="MessageProcessor"/> uses
|
|
/// <see cref="System.Threading.Channels.Channel.CreateUnboundedPrioritized"/>
|
|
/// keyed by <see cref="NotificationMessage.SequenceNumber"/>, so any
|
|
/// items present in the channel at a single read are sorted by
|
|
/// sequence number. With the consumer held, the five data messages
|
|
/// land together and drain as <c>[1, 2, 3, 4, 5]</c> with no
|
|
/// republishes — the reliability guarantee the classic engine fails
|
|
/// to provide. (Without the sentinel, a fast consumer could read the
|
|
/// first arrival before the rest queue, which is what the priority
|
|
/// channel cannot recover from on its own.)
|
|
/// </remarks>
|
|
[Test]
|
|
public async Task OutOfOrderArrivalsReorderedByPriorityChannelAsync()
|
|
{
|
|
var availableSequenceNumbers = new List<uint> { 1, 2, 3, 4, 5 };
|
|
var stringTable = new List<string> { "test" };
|
|
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 11
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// Arrange: drain the gate, then park the worker inside a
|
|
// sentinel keep-alive dispatch so all five data messages
|
|
// queue up in the priority channel before any are read.
|
|
// The keep-alive does not advance the data dedup gate
|
|
// (LastDataSequenceNumberProcessed) so the burst is still
|
|
// processed as a fresh sequence starting at 1.
|
|
sut.Block.Wait();
|
|
await sut.OnPublishReceivedAsync(
|
|
new NotificationMessage { SequenceNumber = 1u },
|
|
availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.KeepAliveNotificationReceived.WaitAsync()
|
|
.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
|
|
foreach (uint seq in new uint[] { 3, 2, 4, 1, 5 })
|
|
{
|
|
await sut.OnPublishReceivedAsync(BuildDataChangeMessage(seq),
|
|
availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
}
|
|
sut.Block.Release();
|
|
|
|
// Act
|
|
await WaitForLastSeqNumberAsync(sut, 5).ConfigureAwait(false);
|
|
|
|
// Assert: the first entry is the sentinel keep-alive; the
|
|
// remaining entries must be the data burst in strict
|
|
// sequence-number order.
|
|
Assert.That(sut.ReceivedSequenceNumbers,
|
|
Has.Count.EqualTo(6));
|
|
Assert.That(sut.ReceivedSequenceNumbers[0], Is.EqualTo(1u),
|
|
"first entry should be the sentinel keep-alive");
|
|
Assert.That(sut.ReceivedSequenceNumbers.Skip(1),
|
|
Is.EqualTo(new uint[] { 1, 2, 3, 4, 5 }));
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(5));
|
|
Assert.That(sut.MissingMessageCount, Is.Zero);
|
|
Assert.That(sut.RepublishMessageCount, Is.Zero);
|
|
m_mockServices.Verify(
|
|
c => c.RepublishAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.IsAny<uint>(),
|
|
It.IsAny<uint>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Never);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// V2 equivalent of the classic-engine regression captured by
|
|
/// <c>OldMessagesAbandonedAfterRepublishTimeoutInSequentialModeAsync</c>,
|
|
/// stressed in serial timing instead of as a burst.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Same arrival order — <c>[3]</c>, <c>[2]</c>, <c>[4]</c>, <c>[1]</c>,
|
|
/// <c>[5]</c> — but each forward-progressing push waits for delivery
|
|
/// before the next is queued. The V2 engine's monotonic dedup gate
|
|
/// (<c>LastDataSequenceNumberProcessed</c>) silently discards the
|
|
/// stale <c>[2]</c> and <c>[1]</c> instead of re-inserting them
|
|
/// out of order; <c>[3, 4, 5]</c> are delivered in strict ascending
|
|
/// order with no republishes.
|
|
/// </remarks>
|
|
[Test]
|
|
public async Task LateArrivalsDiscardedAfterDedupGateAdvancedAsync()
|
|
{
|
|
var availableSequenceNumbers = new List<uint>();
|
|
var stringTable = new List<string> { "test" };
|
|
|
|
var sut = new TestMessageProcessor(m_mockServices.Object,
|
|
m_completion, m_telemetry)
|
|
{
|
|
Id = 12
|
|
};
|
|
await using (sut.ConfigureAwait(false))
|
|
{
|
|
// [3]: first data message after create. The empty
|
|
// AvailableInRetransmissionQueue suppresses the
|
|
// first-after-create speculative republish path, so [3]
|
|
// goes straight to the handler.
|
|
await sut.OnPublishReceivedAsync(BuildDataChangeMessage(3),
|
|
availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await WaitForLastSeqNumberAsync(sut, 3).ConfigureAwait(false);
|
|
|
|
// [2] is strictly behind LastDataSeq=3. Queue [2] then [4]
|
|
// back-to-back; the priority channel picks [2] first
|
|
// (lowest SequenceNumber) so the worker reads [2], discards
|
|
// it silently, then reads [4] and delivers it. Waiting for
|
|
// LastSeq == 4 confirms both happened in that order.
|
|
await sut.OnPublishReceivedAsync(BuildDataChangeMessage(2),
|
|
availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.OnPublishReceivedAsync(BuildDataChangeMessage(4),
|
|
availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await WaitForLastSeqNumberAsync(sut, 4).ConfigureAwait(false);
|
|
|
|
// Same shape for [1] (even staler) followed by [5].
|
|
await sut.OnPublishReceivedAsync(BuildDataChangeMessage(1),
|
|
availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await sut.OnPublishReceivedAsync(BuildDataChangeMessage(5),
|
|
availableSequenceNumbers, stringTable).ConfigureAwait(false);
|
|
await WaitForLastSeqNumberAsync(sut, 5).ConfigureAwait(false);
|
|
|
|
// Assert: only forward-progressing messages were delivered,
|
|
// in strict order; the late [2] and [1] were discarded.
|
|
Assert.That(sut.ReceivedSequenceNumbers,
|
|
Is.EqualTo(new uint[] { 3, 4, 5 }));
|
|
Assert.That(sut.LastSequenceNumberProcessed, Is.EqualTo(5));
|
|
Assert.That(sut.MissingMessageCount, Is.Zero);
|
|
Assert.That(sut.RepublishMessageCount, Is.Zero);
|
|
m_mockServices.Verify(
|
|
c => c.RepublishAsync(
|
|
It.IsAny<RequestHeader>(),
|
|
It.IsAny<uint>(),
|
|
It.IsAny<uint>(),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Never);
|
|
}
|
|
}
|
|
|
|
private static NotificationMessage BuildDataChangeMessage(uint sequenceNumber)
|
|
{
|
|
return new NotificationMessage
|
|
{
|
|
SequenceNumber = sequenceNumber,
|
|
NotificationData =
|
|
[
|
|
new ExtensionObject(new DataChangeNotification
|
|
{
|
|
MonitoredItems =
|
|
[
|
|
new MonitoredItemNotification()
|
|
]
|
|
})
|
|
]
|
|
};
|
|
}
|
|
|
|
private static async Task WaitForLastSeqNumberAsync(
|
|
TestMessageProcessor sut,
|
|
uint expected,
|
|
int timeoutSeconds = 5)
|
|
{
|
|
DateTimeOffset deadline = DateTimeOffset.UtcNow.AddSeconds(timeoutSeconds);
|
|
while (sut.LastSequenceNumberProcessed < expected)
|
|
{
|
|
if (DateTimeOffset.UtcNow > deadline)
|
|
{
|
|
Assert.Fail(
|
|
$"Timeout: LastSequenceNumberProcessed = {sut.LastSequenceNumberProcessed}, " +
|
|
$"expected = {expected}, received = " +
|
|
$"[{string.Join(",", sut.ReceivedSequenceNumbers)}]");
|
|
}
|
|
await Task.Delay(10).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
private sealed class TestMessageProcessor : MessageProcessor
|
|
{
|
|
public TestMessageProcessor(ISubscriptionServiceSetClientMethods session,
|
|
IMessageAckQueue completion, ITelemetryContext telemetry)
|
|
: base(session, completion, telemetry)
|
|
{
|
|
}
|
|
|
|
public new IReadOnlyList<uint> AvailableInRetransmissionQueue
|
|
{
|
|
get => base.AvailableInRetransmissionQueue;
|
|
set => base.AvailableInRetransmissionQueue = value;
|
|
}
|
|
|
|
public SemaphoreSlim Block { get; } = new(1, 1);
|
|
|
|
public AsyncManualResetEvent DataChangeNotificationReceived { get; } = new();
|
|
|
|
public AsyncManualResetEvent EventNotificationReceived { get; } = new();
|
|
|
|
public AsyncManualResetEvent KeepAliveNotificationReceived { get; } = new();
|
|
|
|
public new uint LastSequenceNumberProcessed
|
|
{
|
|
get => base.LastSequenceNumberProcessed;
|
|
set => base.LastSequenceNumberProcessed = value;
|
|
}
|
|
|
|
public PublishState PublishState { get; set; }
|
|
public List<uint> ReceivedSequenceNumbers { get; } = [];
|
|
public AsyncManualResetEvent StatusChangeNotificationReceived { get; } = new();
|
|
|
|
public async ValueTask WaitAsync()
|
|
{
|
|
await Block.WaitAsync().ConfigureAwait(false);
|
|
Block.Release();
|
|
}
|
|
|
|
protected override ValueTask OnDataChangeNotificationAsync(uint sequenceNumber,
|
|
DateTime publishTime, DataChangeNotification notification,
|
|
PublishState publishStateMask, IReadOnlyList<string> stringTable)
|
|
{
|
|
ReceivedSequenceNumbers.Add(sequenceNumber);
|
|
if (publishStateMask != PublishState.None)
|
|
{
|
|
PublishState = publishStateMask;
|
|
}
|
|
DataChangeNotificationReceived.Set();
|
|
return WaitAsync();
|
|
}
|
|
|
|
protected override ValueTask OnEventDataNotificationAsync(uint sequenceNumber,
|
|
DateTime publishTime, EventNotificationList notification,
|
|
PublishState publishStateMask, IReadOnlyList<string> stringTable)
|
|
{
|
|
ReceivedSequenceNumbers.Add(sequenceNumber);
|
|
if (publishStateMask != PublishState.None)
|
|
{
|
|
PublishState = publishStateMask;
|
|
}
|
|
EventNotificationReceived.Set();
|
|
return WaitAsync();
|
|
}
|
|
|
|
protected override ValueTask OnKeepAliveNotificationAsync(uint sequenceNumber,
|
|
DateTime publishTime, PublishState publishStateMask)
|
|
{
|
|
ReceivedSequenceNumbers.Add(sequenceNumber);
|
|
if (publishStateMask != PublishState.None)
|
|
{
|
|
PublishState = publishStateMask;
|
|
}
|
|
KeepAliveNotificationReceived.Set();
|
|
return WaitAsync();
|
|
}
|
|
|
|
protected override void OnPublishStateChanged(PublishState stateMask)
|
|
{
|
|
PublishState = stateMask;
|
|
base.OnPublishStateChanged(stateMask);
|
|
}
|
|
|
|
protected override async ValueTask OnStatusChangeNotificationAsync(uint sequenceNumber,
|
|
DateTime publishTime, StatusChangeNotification notification,
|
|
PublishState publishStateMask, IReadOnlyList<string> stringTable)
|
|
{
|
|
ReceivedSequenceNumbers.Add(sequenceNumber);
|
|
if (publishStateMask != PublishState.None)
|
|
{
|
|
PublishState = publishStateMask;
|
|
}
|
|
StatusChangeNotificationReceived.Set();
|
|
await WaitAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
private FakeMessageAckQueue m_completion;
|
|
private ITelemetryContext m_telemetry;
|
|
private Mock<ISubscriptionServiceSetClientMethods> m_mockServices;
|
|
}
|
|
}
|