/* ======================================================================== * 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/ * ======================================================================*/ // CA2000: system-under-test disposables are created per test and released at teardown; // there is no cross-test resource leak. Suppressed file-level for the suite. #pragma warning disable CA2000 // Dispose objects before losing scope // 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.Reflection; using System.Runtime.CompilerServices; using System.Text; using System.Threading; using System.Threading.Tasks; using Moq; using NUnit.Framework; using Opc.Ua.Redundancy; using Opc.Ua.Redundancy.Server; using Opc.Ua.Tests; using TestAsyncEnumerable = Opc.Ua.PubSub.Tests.TestAsyncEnumerable; namespace Opc.Ua.Server.Tests.Redundancy { /// /// Unit tests for . /// [TestFixture] [Category("Distributed")] [Category("Subscription")] [Parallelizable(ParallelScope.All)] public class SharedKeyValueSubscriptionStoreTests { private const ushort NamespaceIndex = 2; private static readonly string[] s_filteredRetainConditionIds = ["ns=2;s=Alarm|", "ns=2;s=Alarm|i=7"]; /// /// Verifies that subscription definitions round-trip through shared storage. /// [Test] public async Task StoreAndRestoreRoundTripsDefinitionAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); StoredSubscription expected = NewSubscription(100, 10); bool stored = await store.StoreSubscriptionsAsync([expected]).ConfigureAwait(false); RestoreSubscriptionResult result = await store.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(stored, Is.True); Assert.That(result.Success, Is.True); var actual = (StoredSubscription)result.Subscriptions!.Single(); AssertSubscription(actual, expected); AssertMonitoredItem((StoredMonitoredItem)actual.MonitoredItems.Single(), NewItem(100, 10)); } [Test] public async Task SubscriptionSnapshotCannotReplaySameIdFromOlderGenerationAsync() { using var kv = new InMemorySharedKeyValueStore(); using var protector = new AesCbcHmacRecordProtector(MakeKey(41)); await using var store = new SharedKeyValueSubscriptionStore(kv, CreateContext(), protector); StoredSubscription initial = NewSubscription(101, 11); await store.StoreSubscriptionsAsync([initial]).ConfigureAwait(false); KeyValuePair source = await GetSingleGenerationRecordAsync(kv).ConfigureAwait(false); StoredSubscription replacement = NewSubscription(101, 11); replacement.PublishingInterval = initial.PublishingInterval + 100; await store.StoreSubscriptionsAsync([replacement]).ConfigureAwait(false); var records = new List>(); await foreach (KeyValuePair entry in kv.ScanAsync( SharedKeyValueSubscriptionStore.SnapshotGenerationRootPrefix()).ConfigureAwait(false)) { if (entry.Key != source.Key) { records.Add(entry); } } Assert.That(records, Has.Count.EqualTo(1)); KeyValuePair target = records[0]; await kv.SetAsync(target.Key, source.Value).ConfigureAwait(false); Assert.That( async () => await store.RestoreSubscriptionsAsync().ConfigureAwait(false), Throws.TypeOf() .With.Property(nameof(ServiceResultException.StatusCode)) .EqualTo(StatusCodes.BadSecurityChecksFailed)); await kv.SetAsync(target.Key, target.Value).ConfigureAwait(false); RestoreSubscriptionResult restored = await store.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(restored.Success, Is.True); Assert.That( restored.Subscriptions!.Single().PublishingInterval, Is.EqualTo(replacement.PublishingInterval)); } [Test] public async Task SubscriptionManifestRejectsEmptyContextAsync() { using var kv = new InMemorySharedKeyValueStore(); using var protector = new AesCbcHmacRecordProtector(MakeKey(42)); await using var store = new SharedKeyValueSubscriptionStore(kv, CreateContext(), protector); await store.StoreSubscriptionsAsync([NewSubscription(102, 12)]).ConfigureAwait(false); string key = SharedKeyValueSubscriptionStore.SnapshotManifestKey(); (_, ByteString record) = await kv.TryGetAsync(key).ConfigureAwait(false); Assert.That(protector.TryUnprotect( RecordProtectionContext.Create("subscription-manifest", key), record, out ByteString plaintext), Is.True); await kv.SetAsync(key, protector.Protect(default, plaintext)).ConfigureAwait(false); Assert.That( async () => await store.RestoreSubscriptionsAsync().ConfigureAwait(false), Throws.TypeOf() .With.Property(nameof(ServiceResultException.StatusCode)) .EqualTo(StatusCodes.BadSecurityChecksFailed)); } [Test] public async Task RetransmissionStateAndMessagesRejectDifferentKeysAsync() { using var kv = new InMemorySharedKeyValueStore(); using var protector = new AesCbcHmacRecordProtector(MakeKey(43)); await using var store = new SharedKeyValueSubscriptionStore(kv, CreateContext(), protector); store.StoreRetransmissionState(703, 2, [NewNotification(1)]); await store.FlushAsync().ConfigureAwait(false); (_, ByteString stateRecord) = await kv.TryGetAsync( SharedKeyValueSubscriptionStore.RetransmissionStateKeyFor(703)).ConfigureAwait(false); (_, ByteString messageRecord) = await kv.TryGetAsync( SharedKeyValueSubscriptionStore.RetransmissionMessageKeyFor(703, 1)).ConfigureAwait(false); await kv.SetAsync(SharedKeyValueSubscriptionStore.RetransmissionStateKeyFor(704), stateRecord) .ConfigureAwait(false); await kv.SetAsync(SharedKeyValueSubscriptionStore.RetransmissionMessageKeyFor(703, 2), messageRecord) .ConfigureAwait(false); Assert.That(await store.LoadRetransmissionStateAsync(704).ConfigureAwait(false), Is.Null); SubscriptionRetransmissionState? state = await store.LoadRetransmissionStateAsync(703) .ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(2)); Assert.That(state.SentMessages.Memory.ToArray().Single().SequenceNumber, Is.EqualTo(1)); } /// /// Verifies that shared definitions preserve publishing state and anonymous owner application URI. /// [TestCase(false)] [TestCase(true)] public async Task StoreAndRestoreRoundTripsSubscriptionStateAsync(bool publishingEnabled) { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore active = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); StoredSubscription expected = NewSubscription(107, 17); expected.PublishingEnabled = publishingEnabled; expected.OwnerClientApplicationUri = "urn:test:client"; await active.StoreSubscriptionsAsync([expected]).ConfigureAwait(false); RestoreSubscriptionResult result = await backup.RestoreSubscriptionsAsync().ConfigureAwait(false); var actual = (StoredSubscription)result.Subscriptions!.Single(); Assert.Multiple(() => { Assert.That(result.Success, Is.True); Assert.That(actual.PublishingEnabled, Is.EqualTo(publishingEnabled)); Assert.That(actual.OwnerClientApplicationUri, Is.EqualTo("urn:test:client")); }); } /// /// Verifies that shared definitions preserve the triggering links between monitored /// items (OPC 10000-4 ยง5.13.1.6). /// [Test] public async Task StoreAndRestoreRoundTripsTriggeringLinksAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore active = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); StoredSubscription expected = NewSubscription(108, 18); expected.TriggeringLinks = new Dictionary> { [18] = [19u, 20u] }; await active.StoreSubscriptionsAsync([expected]).ConfigureAwait(false); RestoreSubscriptionResult result = await backup.RestoreSubscriptionsAsync().ConfigureAwait(false); var actual = (StoredSubscription)result.Subscriptions!.Single(); Assert.Multiple(() => { Assert.That(result.Success, Is.True); Assert.That(actual.TriggeringLinks, Is.Not.Null); Assert.That(actual.TriggeringLinks!.Keys, Is.EquivalentTo(new uint[] { 18 })); Assert.That(actual.TriggeringLinks[18], Is.EqualTo(new uint[] { 19, 20 })); }); } /// /// Verifies that stored subscriptions preserve monitored-item lifecycle state. /// [TestCase(false, false)] [TestCase(false, true)] [TestCase(true, false)] [TestCase(true, true)] public async Task StoreAndRestoreRoundTripsMonitoredItemLifecycleStateAsync( bool isDeleted, bool isDetached) { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore active = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); StoredSubscription expected = NewSubscription(101, 11); var expectedItem = (StoredMonitoredItem)expected.MonitoredItems.Single(); expectedItem.IsDeleted = isDeleted; expectedItem.IsDetached = isDetached; await active.StoreSubscriptionsAsync([expected]).ConfigureAwait(false); RestoreSubscriptionResult result = await backup.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(result.Success, Is.True); var actual = (StoredMonitoredItem)result.Subscriptions!.Single().MonitoredItems.Single(); Assert.Multiple(() => { Assert.That(actual.IsDeleted, Is.EqualTo(isDeleted)); Assert.That(actual.IsDetached, Is.EqualTo(isDetached)); }); } /// /// Verifies that stored subscriptions preserve filtered retained-condition identifiers. /// [Test] public async Task StoreAndRestoreRoundTripsFilteredRetainConditionIdsAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore active = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); StoredSubscription expected = NewSubscription(103, 13); var expectedItem = (StoredMonitoredItem)expected.MonitoredItems.Single(); expectedItem.FilteredRetainConditionIds = [.. s_filteredRetainConditionIds]; await active.StoreSubscriptionsAsync([expected]).ConfigureAwait(false); RestoreSubscriptionResult result = await backup.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(result.Success, Is.True); var actual = (StoredMonitoredItem)result.Subscriptions!.Single().MonitoredItems.Single(); Assert.That( actual.FilteredRetainConditionIds.Memory.ToArray(), Is.EqualTo(s_filteredRetainConditionIds)); } /// /// Verifies that absent filtered retained-condition identifiers remain absent after restoration. /// [Test] public async Task StoreAndRestoreRoundTripsAbsentFilteredRetainConditionIdsAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore active = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); await active.StoreSubscriptionsAsync([NewSubscription(104, 14)]).ConfigureAwait(false); RestoreSubscriptionResult result = await backup.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(result.Success, Is.True); var actual = (StoredMonitoredItem)result.Subscriptions!.Single().MonitoredItems.Single(); Assert.That(actual.FilteredRetainConditionIds.IsEmpty, Is.True); } /// /// Verifies that storage preserves raw errors without persisting transient notification metadata. /// [Test] public async Task StoreAndRestorePreservesRawErrorWithoutTransientNotificationMetadataAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore active = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); StoredSubscription expected = NewSubscription(106, 16); var expectedItem = (StoredMonitoredItem)expected.MonitoredItems.Single(); expectedItem.LastValue = new DataValue( Variant.Null, StatusCodes.BadCommunicationError); expectedItem.LastError = new ServiceResult(StatusCodes.BadCommunicationError); await active.StoreSubscriptionsAsync([expected]) .ConfigureAwait(false); RestoreSubscriptionResult result = await backup.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(result.Success, Is.True); var actual = (StoredMonitoredItem)result.Subscriptions! .Single().MonitoredItems.Single(); Assert.That(actual.LastValue, Is.EqualTo(expectedItem.LastValue)); Assert.That( actual.LastError.StatusCode, Is.EqualTo(StatusCodes.BadCommunicationError)); using var decoder = new BinaryDecoder(EncodeDefinition(active, expected).ToArray(), CreateContext()); Assert.That(decoder.ReadInt32(null), Is.EqualTo(5), "Definitions without triggering links must keep the owner and publishing state format."); } /// /// Verifies that only definitions with triggering links use the newer format, so that /// replicas that predate it can still read link-free snapshots. /// [Test] public async Task DefinitionFormatDependsOnTriggeringLinksAsync() { using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); StoredSubscription withoutLinks = NewSubscription(109, 21); StoredSubscription emptyLinks = NewSubscription(110, 22); emptyLinks.TriggeringLinks = new Dictionary>(); StoredSubscription withLinks = NewSubscription(111, 23); withLinks.TriggeringLinks = new Dictionary> { [23] = [24u] }; Assert.Multiple(() => { Assert.That(ReadDefinitionVersion(store, withoutLinks), Is.EqualTo(5)); Assert.That(ReadDefinitionVersion(store, emptyLinks), Is.EqualTo(5)); Assert.That(ReadDefinitionVersion(store, withLinks), Is.EqualTo(6)); }); } private static int ReadDefinitionVersion( SharedKeyValueSubscriptionStore store, StoredSubscription subscription) { using var decoder = new BinaryDecoder( EncodeDefinition(store, subscription).ToArray(), CreateContext()); return decoder.ReadInt32(null); } /// /// Verifies that the retired interim notification-state definition format is rejected. /// [Test] public async Task RestoreRejectsInterimVersionFourDefinitionAsync() { using var keyValueStore = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(keyValueStore); StoredSubscription expected = NewSubscription(106, 16); var item = (StoredMonitoredItem)expected.MonitoredItems.Single(); item.LastValue = new DataValue(Variant.From(42), StatusCodes.Good); item.FilteredRetainConditionIds = [.. s_filteredRetainConditionIds]; ServiceMessageContext context = CreateContext(); using var encoder = new BinaryEncoder(context); encoder.WriteInt32(null, 4); encoder.WriteStringArray(null, context.NamespaceUris.ToArrayOf()); encoder.WriteStringArray(null, context.ServerUris.ToArrayOf()); GetPrivateMethod( "EncodeSubscription", typeof(BinaryEncoder), typeof(StoredSubscription), typeof(int)).Invoke(null, [encoder, expected, 3]); encoder.WriteBoolean(null, true); encoder.WriteDataValue(null, new DataValue(Variant.Null, StatusCodes.BadCommunicationError)); encoder.WriteStatusCode(null, StatusCodes.BadCommunicationError); await keyValueStore.SetAsync( SharedKeyValueSubscriptionStore.KeyFor(expected.Id), ByteString.From(encoder.CloseAndReturnBuffer())).ConfigureAwait(false); ServiceResultException? error = Assert.ThrowsAsync( () => store.RestoreSubscriptionsAsync().AsTask()); Assert.That(error!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); Assert.That(error.Message, Does.Contain("Unsupported subscription record version")); } /// /// Verifies that cloning a monitored item preserves filtered retained-condition identifiers. /// [Test] public void CloneMonitoredItemPreservesFilteredRetainConditionIds() { StoredMonitoredItem item = NewItem(105, 15); item.FilteredRetainConditionIds = [.. s_filteredRetainConditionIds]; MethodInfo clone = GetPrivateMethod( "CloneMonitoredItem", typeof(IStoredMonitoredItem)); var cloned = (StoredMonitoredItem)clone.Invoke(null, [item])!; Assert.That( cloned.FilteredRetainConditionIds.Memory.ToArray(), Is.EqualTo(s_filteredRetainConditionIds)); } /// /// Verifies that cloning a monitored item preserves lifecycle state. /// [Test] public void CloneMonitoredItemPreservesLifecycleState() { StoredMonitoredItem item = NewItem(102, 12); item.IsDeleted = true; item.IsDetached = true; MethodInfo clone = GetPrivateMethod( "CloneMonitoredItem", typeof(IStoredMonitoredItem)); var cloned = (StoredMonitoredItem)clone.Invoke(null, [item])!; Assert.Multiple(() => { Assert.That(cloned, Is.Not.SameAs(item)); Assert.That(cloned.IsDeleted, Is.True); Assert.That(cloned.IsDetached, Is.True); }); } /// /// Verifies that version-one definitions restore lifecycle flags with false defaults. /// [Test] public async Task RestoreVersionOneDefinitionDefaultsLifecycleStateToFalseAsync() { using var kv = new InMemorySharedKeyValueStore(); StoredSubscription legacy = NewSubscription(103, 13); var legacyItem = (StoredMonitoredItem)legacy.MonitoredItems.Single(); legacyItem.IsDeleted = true; legacyItem.IsDetached = true; await kv .SetAsync( SharedKeyValueSubscriptionStore.KeyFor(legacy.Id), EncodeLegacyDefinition(legacy)) .ConfigureAwait(false); SharedKeyValueSubscriptionStore store = CreateStore(kv); RestoreSubscriptionResult result = await store.RestoreSubscriptionsAsync().ConfigureAwait(false); var restored = (StoredMonitoredItem)result.Subscriptions!.Single().MonitoredItems.Single(); Assert.Multiple(() => { Assert.That(result.Success, Is.True); Assert.That(restored.IsDeleted, Is.False); Assert.That(restored.IsDetached, Is.False); }); var restoredSubscription = (StoredSubscription)result.Subscriptions!.Single(); Assert.That(restoredSubscription.PublishingEnabled, Is.True); Assert.That(restoredSubscription.OwnerClientApplicationUri, Is.Null); } /// /// Verifies that a subscription stored by one replica is visible to another. /// [Test] public async Task StoreIsVisibleToSecondReplicaAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore active = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); StoredSubscription expected = NewSubscription(200, 20); await active.StoreSubscriptionsAsync([expected]).ConfigureAwait(false); RestoreSubscriptionResult result = await backup.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(result.Success, Is.True); var actual = (StoredSubscription)result.Subscriptions!.Single(); AssertSubscription(actual, expected); } /// /// Verifies that restoration observes only a committed generation during concurrent snapshots. /// [Test] public async Task RestoreUsesOnlyCommittedGenerationDuringConcurrentSnapshotsAsync() { using var kv = new InMemorySharedKeyValueStore(); var blockedBackend = new BlockingManifestCommitStore(kv); var firstWriter = new SharedKeyValueSubscriptionStore(blockedBackend, CreateContext()); Task firstCommit = firstWriter .StoreSubscriptionsAsync([NewSubscription(225, 22)]) .AsTask(); await blockedBackend.WaitForBlockedManifestAsync().ConfigureAwait(false); var secondBackend = new AsyncSharedKeyValueStore(kv); var secondWriter = new SharedKeyValueSubscriptionStore(secondBackend, CreateContext()); await secondWriter .StoreSubscriptionsAsync([NewSubscription(226, 23)]) .ConfigureAwait(false); var readerBackend = new AsyncSharedKeyValueStore(kv); var reader = new SharedKeyValueSubscriptionStore(readerBackend, CreateContext()); RestoreSubscriptionResult secondResult = await reader.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(secondResult.Subscriptions!.Select(subscription => subscription.Id), Is.EqualTo(new uint[] { 226 })); blockedBackend.ReleaseManifest(); await firstCommit.ConfigureAwait(false); RestoreSubscriptionResult firstResult = await reader.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(firstResult.Subscriptions!.Select(subscription => subscription.Id), Is.EqualTo(new uint[] { 225 })); } /// /// Verifies that subscription definitions can be restored after process restart. /// [Test] public async Task RestoreLoadsPersistedDefinitionsAfterProcessRestartAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore active = CreateStore(kv); StoredSubscription expected = NewSubscription(250, 25); await active.StoreSubscriptionsAsync([expected]).ConfigureAwait(false); var restartedBackend = new AsyncSharedKeyValueStore(kv); var restarted = new SharedKeyValueSubscriptionStore(restartedBackend, CreateContext()); RestoreSubscriptionResult result = await restarted.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(result.Success, Is.True); var actual = (StoredSubscription)result.Subscriptions!.Single(); AssertSubscription(actual, expected); } /// /// Verifies that an empty persisted snapshot clears cached subscription definitions. /// [Test] public async Task EmptyPersistedSnapshotClearsCachedDefinitionsAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore legacyEncoder = CreateStore(kv); StoredSubscription stale = NewSubscription(275, 27); await kv .SetAsync( SharedKeyValueSubscriptionStore.KeyFor(stale.Id), EncodeDefinition(legacyEncoder, stale)) .ConfigureAwait(false); var restartedBackend = new AsyncSharedKeyValueStore(kv); var restarted = new SharedKeyValueSubscriptionStore(restartedBackend, CreateContext()); await restarted.StoreSubscriptionsAsync([]).ConfigureAwait(false); RestoreSubscriptionResult result = await restarted.RestoreSubscriptionsAsync().ConfigureAwait(false); (bool foundLegacy, _) = await kv .TryGetAsync(SharedKeyValueSubscriptionStore.KeyFor(stale.Id)) .ConfigureAwait(false); (bool foundManifest, _) = await kv .TryGetAsync(SharedKeyValueSubscriptionStore.SnapshotManifestKey()) .ConfigureAwait(false); Assert.That(result.Success, Is.True); Assert.That(result.Subscriptions, Is.Empty); Assert.That(foundLegacy, Is.True); Assert.That(foundManifest, Is.True); Assert.That(GetCachedSubscriptionIds(restarted), Is.Empty); } /// /// Verifies that restoration reads legacy definitions when no manifest exists. /// [Test] public async Task RestoreReadsLegacyDefinitionsWhenManifestIsAbsentAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore legacyEncoder = CreateStore(kv); StoredSubscription expected = NewSubscription(276, 28); await kv .SetAsync( SharedKeyValueSubscriptionStore.KeyFor(expected.Id), EncodeDefinition(legacyEncoder, expected)) .ConfigureAwait(false); var restartedBackend = new AsyncSharedKeyValueStore(kv); var restarted = new SharedKeyValueSubscriptionStore(restartedBackend, CreateContext()); RestoreSubscriptionResult result = await restarted.RestoreSubscriptionsAsync().ConfigureAwait(false); var actual = (StoredSubscription)result.Subscriptions!.Single(); AssertSubscription(actual, expected); } /// /// Verifies that tampering with a protected backend definition causes restoration to fail. /// [Test] public async Task ProtectedDefinitionRestoreFailsWhenBackendRecordIsTamperedAsync() { using var kv = new InMemorySharedKeyValueStore(); using var protector = new AesCbcHmacRecordProtector(MakeKey(11)); var store = new SharedKeyValueSubscriptionStore(kv, CreateContext(), protector); StoredSubscription subscription = NewSubscription(300, 30); await store.StoreSubscriptionsAsync([subscription]).ConfigureAwait(false); KeyValuePair record = await GetSingleGenerationRecordAsync(kv).ConfigureAwait(false); await kv .SetAsync(record.Key, ByteString.From(new byte[] { 1, 2, 3, 4, 5 })) .ConfigureAwait(false); ServiceResultException? exception = Assert.ThrowsAsync( () => store.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadSecurityChecksFailed)); } /// /// Verifies that protected subscription definitions are not persisted in clear text. /// [Test] public async Task ProtectedStoreDoesNotPersistDefinitionInClearTextAsync() { using var kv = new InMemorySharedKeyValueStore(); using var protector = new AesCbcHmacRecordProtector(MakeKey(12)); var store = new SharedKeyValueSubscriptionStore(kv, CreateContext(), protector); StoredSubscription subscription = NewSubscription(400, 40); await store.StoreSubscriptionsAsync([subscription]).ConfigureAwait(false); KeyValuePair record = await GetSingleGenerationRecordAsync(kv).ConfigureAwait(false); Assert.That(Contains(record.Value.ToArray(), BitConverter.GetBytes(subscription.Id)), Is.False); Assert.That((await store.RestoreSubscriptionsAsync().ConfigureAwait(false)).Subscriptions!.Single().Id, Is.EqualTo(subscription.Id)); } /// /// Verifies that a corrupt snapshot fails without partially replacing the definition cache. /// [Test] public async Task CorruptSnapshotFailsWithoutPartiallyReplacingCacheAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore reader = CreateStore(kv); await reader.StoreSubscriptionsAsync([NewSubscription(450, 45)]).ConfigureAwait(false); var writerBackend = new AsyncSharedKeyValueStore(kv); var writer = new SharedKeyValueSubscriptionStore(writerBackend, CreateContext()); await writer .StoreSubscriptionsAsync([NewSubscription(451, 46), NewSubscription(452, 47)]) .ConfigureAwait(false); KeyValuePair currentRecord = await FindGenerationRecordAsync(kv, 452) .ConfigureAwait(false); ByteString validRecord = currentRecord.Value; await kv .SetAsync( currentRecord.Key, ByteString.From(new byte[] { 1, 2, 3, 4, 5 })) .ConfigureAwait(false); ServiceResultException? exception = Assert.ThrowsAsync( () => reader.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); Assert.That(GetCachedSubscriptionIds(reader), Is.EqualTo(new uint[] { 450 })); await kv .SetAsync(currentRecord.Key, validRecord) .ConfigureAwait(false); RestoreSubscriptionResult result = await reader.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(result.Subscriptions!.Select(subscription => subscription.Id), Is.EqualTo(new uint[] { 451, 452 })); Assert.That(GetCachedSubscriptionIds(reader), Is.EqualTo(new uint[] { 451, 452 })); } /// /// Verifies that restoration rejects a subscription identifier that differs from its persisted key. /// [Test] public async Task RestoreRejectsSubscriptionIdThatDoesNotMatchPersistedKeyAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore writer = CreateStore(kv); await writer.StoreSubscriptionsAsync([NewSubscription(475, 48)]).ConfigureAwait(false); KeyValuePair generationRecord = await GetSingleGenerationRecordAsync(kv) .ConfigureAwait(false); await kv.DeleteAsync(SharedKeyValueSubscriptionStore.SnapshotManifestKey()).ConfigureAwait(false); await kv .SetAsync(SharedKeyValueSubscriptionStore.KeyFor(476), generationRecord.Value) .ConfigureAwait(false); var restartedBackend = new AsyncSharedKeyValueStore(kv); var restarted = new SharedKeyValueSubscriptionStore(restartedBackend, CreateContext()); ServiceResultException? exception = Assert.ThrowsAsync( () => restarted.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); } /// /// Verifies that storage rejects invalid monitored-item identity. /// [Test] public async Task StoreRejectsInvalidMonitoredItemIdentityAsync() { using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); StoredSubscription subscription = NewSubscription(490, 49); ((StoredMonitoredItem)subscription.MonitoredItems.Single()).SubscriptionId = 491; ArgumentException? exception = Assert.ThrowsAsync( () => store.StoreSubscriptionsAsync([subscription]).AsTask()); (bool foundManifest, _) = await kv .TryGetAsync(SharedKeyValueSubscriptionStore.SnapshotManifestKey()) .ConfigureAwait(false); Assert.That(exception!.ParamName, Is.EqualTo("subscriptions")); Assert.That(foundManifest, Is.False); } /// /// Verifies that storing a new snapshot removes subscriptions omitted from that snapshot. /// [Test] public async Task StoreReplacesRemovedSubscriptionsAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); await store.StoreSubscriptionsAsync([NewSubscription(500, 50), NewSubscription(501, 51)]).ConfigureAwait(false); await store.StoreSubscriptionsAsync([NewSubscription(501, 51)]).ConfigureAwait(false); RestoreSubscriptionResult result = await store.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(result.Subscriptions!.Select(s => s.Id), Is.EqualTo(new uint[] { 501 })); } /// /// Verifies that restore completion cleans up stale subscription definitions. /// [Test] public async Task OnSubscriptionRestoreCompleteCleansStaleDefinitionsAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); await store.StoreSubscriptionsAsync([NewSubscription(600, 60), NewSubscription(601, 61)]).ConfigureAwait(false); await store.OnSubscriptionRestoreCompleteAsync(new Dictionary> { [601] = new ArrayOf(new uint[] { 61 }) }).ConfigureAwait(false); RestoreSubscriptionResult result = await store.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(result.Subscriptions!.Select(s => s.Id), Is.EqualTo(new uint[] { 601 })); } /// /// Verifies that a subscription snapshot rejects duplicate subscription identifiers. /// [Test] public async Task StoreSubscriptionsRejectsDuplicateSubscriptionIdsAsync() { using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); StoredSubscription first = NewSubscription(950, 95); StoredSubscription duplicate = NewSubscription(950, 96); ArgumentException? exception = Assert.ThrowsAsync( () => store.StoreSubscriptionsAsync([first, duplicate]).AsTask()); (bool foundManifest, _) = await kv .TryGetAsync(SharedKeyValueSubscriptionStore.SnapshotManifestKey()) .ConfigureAwait(false); Assert.That(exception!.ParamName, Is.EqualTo("subscriptions")); Assert.That(exception.Message, Does.Contain("duplicate subscription ids")); Assert.That(foundManifest, Is.False); } /// /// Verifies that subscription storage observes cancellation before commit. /// [Test] public void StoreSubscriptionsHonorsCancellationBeforeCommit() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); using var cts = new CancellationTokenSource(); cts.Cancel(); Assert.That( async () => await store .StoreSubscriptionsAsync([NewSubscription(951, 96)], cts.Token) .ConfigureAwait(false), Throws.InstanceOf()); } /// /// Verifies that subscription restoration observes cancellation before looking up the manifest. /// [Test] public void RestoreSubscriptionsHonorsCancellationBeforeManifestLookup() { using var kv = new InMemorySharedKeyValueStore(); var cancelling = new CancellationCheckingStore(kv); var store = new SharedKeyValueSubscriptionStore(cancelling, CreateContext()); using var cts = new CancellationTokenSource(); cts.Cancel(); Assert.That( async () => await store.RestoreSubscriptionsAsync(cts.Token).ConfigureAwait(false), Throws.InstanceOf()); } /// /// Verifies that a mid-write failure releases commit serialization and preserves cached definitions. /// [Test] public async Task StoreSubscriptionsReleasesCommitLockAndPreservesCacheOnMidWriteFailureAsync() { using var kv = new InMemorySharedKeyValueStore(); var failingBackend = new ThrowOnNthGenerationWriteStore(kv, failOnCallNumber: 3); await using var store = new SharedKeyValueSubscriptionStore(failingBackend, CreateContext()); await store.StoreSubscriptionsAsync([NewSubscription(990, 103)]).ConfigureAwait(false); InvalidOperationException? exception = Assert.ThrowsAsync( () => store.StoreSubscriptionsAsync( [NewSubscription(990, 103), NewSubscription(991, 104)]).AsTask()); Assert.That(exception, Is.Not.Null); RestoreSubscriptionResult resultAfterFailure = await store.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(resultAfterFailure.Subscriptions!.Select(s => s.Id), Is.EqualTo(new uint[] { 990 })); Assert.That(GetCachedSubscriptionIds(store), Is.EqualTo(new uint[] { 990 })); // The commit semaphore must have been released on the failed attempt so a later commit still succeeds. bool stored = await store.StoreSubscriptionsAsync([NewSubscription(992, 105)]).ConfigureAwait(false); RestoreSubscriptionResult resultAfterRecovery = await store.RestoreSubscriptionsAsync().ConfigureAwait(false); Assert.That(stored, Is.True); Assert.That(resultAfterRecovery.Subscriptions!.Select(s => s.Id), Is.EqualTo(new uint[] { 992 })); } /// /// Verifies that sequential subscription commits retain records from the prior generation. /// [Test] public async Task SequentialCommitsRetainPriorGenerationRecordsAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); await store.StoreSubscriptionsAsync([NewSubscription(995, 106)]).ConfigureAwait(false); KeyValuePair firstGenerationRecord = await GetSingleGenerationRecordAsync(kv).ConfigureAwait(false); await store.StoreSubscriptionsAsync([NewSubscription(996, 107)]).ConfigureAwait(false); (bool stillPresent, _) = await kv.TryGetAsync(firstGenerationRecord.Key).ConfigureAwait(false); Assert.That(stillPresent, Is.True); } /// /// Verifies that subscription restoration rejects a tampered manifest. /// [Test] public async Task RestoreRejectsTamperedManifestAsync() { using var kv = new InMemorySharedKeyValueStore(); using var protector = new AesCbcHmacRecordProtector(MakeKey(14)); await using var store = new SharedKeyValueSubscriptionStore(kv, CreateContext(), protector); await store.StoreSubscriptionsAsync([NewSubscription(955, 96)]).ConfigureAwait(false); await kv .SetAsync( SharedKeyValueSubscriptionStore.SnapshotManifestKey(), ByteString.From(0xFF, 0xFE, 0xFD)) .ConfigureAwait(false); ServiceResultException? exception = Assert.ThrowsAsync( () => store.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadSecurityChecksFailed)); } /// /// Verifies that subscription restoration rejects a malformed manifest version. /// [Test] public async Task RestoreRejectsMalformedManifestVersionAsync() { using var kv = new InMemorySharedKeyValueStore(); using var encoder = new BinaryEncoder(CreateContext()); encoder.WriteInt32(null, 999); encoder.WriteByteString(null, ByteString.From(new byte[16])); encoder.WriteUInt32(null, 0); await kv .SetAsync( SharedKeyValueSubscriptionStore.SnapshotManifestKey(), ByteString.From(encoder.CloseAndReturnBuffer())) .ConfigureAwait(false); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); ServiceResultException? exception = Assert.ThrowsAsync( () => store.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); Assert.That(exception.Message, Does.Contain("manifest is malformed")); } /// /// Verifies that restoration rejects generation keys without a subscription identifier. /// [Test] public async Task RestoreRejectsGenerationKeyWithoutSubscriptionIdAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore writer = CreateStore(kv); await writer.StoreSubscriptionsAsync([NewSubscription(960, 96)]).ConfigureAwait(false); KeyValuePair record = await GetSingleGenerationRecordAsync(kv).ConfigureAwait(false); await kv .SetAsync(GenerationPrefixOf(record.Key), ByteString.From(new byte[] { 1 })) .ConfigureAwait(false); var restartedBackend = new AsyncSharedKeyValueStore(kv); var restarted = new SharedKeyValueSubscriptionStore(restartedBackend, CreateContext()); ServiceResultException? exception = Assert.ThrowsAsync( () => restarted.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); Assert.That(exception.Message, Does.Contain("key is malformed")); } /// /// Verifies that restoration rejects generation keys with nonnumeric identifier suffixes. /// [Test] public async Task RestoreRejectsGenerationKeyWithNonNumericSuffixAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore writer = CreateStore(kv); await writer.StoreSubscriptionsAsync([NewSubscription(961, 96)]).ConfigureAwait(false); KeyValuePair record = await GetSingleGenerationRecordAsync(kv).ConfigureAwait(false); await kv .SetAsync(GenerationPrefixOf(record.Key) + "12a", ByteString.From(new byte[] { 1 })) .ConfigureAwait(false); var restartedBackend = new AsyncSharedKeyValueStore(kv); var restarted = new SharedKeyValueSubscriptionStore(restartedBackend, CreateContext()); ServiceResultException? exception = Assert.ThrowsAsync( () => restarted.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); Assert.That(exception.Message, Does.Contain("key is malformed")); } /// /// Verifies that restoration rejects generation keys whose identifier suffixes overflow. /// [Test] public async Task RestoreRejectsGenerationKeyWithOverflowingSuffixAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore writer = CreateStore(kv); await writer.StoreSubscriptionsAsync([NewSubscription(962, 96)]).ConfigureAwait(false); KeyValuePair record = await GetSingleGenerationRecordAsync(kv).ConfigureAwait(false); await kv .SetAsync( GenerationPrefixOf(record.Key) + "99999999999999999999", ByteString.From(new byte[] { 1 })) .ConfigureAwait(false); var restartedBackend = new AsyncSharedKeyValueStore(kv); var restarted = new SharedKeyValueSubscriptionStore(restartedBackend, CreateContext()); ServiceResultException? exception = Assert.ThrowsAsync( () => restarted.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); Assert.That(exception.Message, Does.Contain("key is malformed")); } /// /// Verifies that restoration rejects duplicate generation records returned by scanning. /// [Test] public async Task RestoreRejectsDuplicateGenerationRecordsFromScanAsync() { using var kv = new InMemorySharedKeyValueStore(); await using var writer = new SharedKeyValueSubscriptionStore(kv, CreateContext()); await writer.StoreSubscriptionsAsync([NewSubscription(965, 97)]).ConfigureAwait(false); var duplicating = new DuplicatingScanStore(kv); var reader = new SharedKeyValueSubscriptionStore(duplicating, CreateContext()); ServiceResultException? exception = Assert.ThrowsAsync( () => reader.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); Assert.That(exception.Message, Does.Contain("duplicate records")); } /// /// Verifies that restoration rejects an incomplete subscription generation. /// [Test] public async Task RestoreRejectsIncompleteGenerationAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore writer = CreateStore(kv); await writer .StoreSubscriptionsAsync([NewSubscription(970, 98), NewSubscription(971, 99)]) .ConfigureAwait(false); KeyValuePair record = await FindGenerationRecordAsync(kv, 971).ConfigureAwait(false); await kv.DeleteAsync(record.Key).ConfigureAwait(false); var restartedBackend = new AsyncSharedKeyValueStore(kv); var restarted = new SharedKeyValueSubscriptionStore(restartedBackend, CreateContext()); ServiceResultException? exception = Assert.ThrowsAsync( () => restarted.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); Assert.That(exception.Message, Does.Contain("incomplete")); } /// /// Verifies that restoration rejects persisted records containing invalid monitored-item identity. /// [Test] public async Task RestoreRejectsPersistedRecordWithInvalidMonitoredItemIdentityAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore encoder = CreateStore(kv); StoredSubscription malformed = NewSubscription(975, 100); ((StoredMonitoredItem)malformed.MonitoredItems.Single()).SubscriptionId = 976; await kv .SetAsync(SharedKeyValueSubscriptionStore.KeyFor(malformed.Id), EncodeDefinition(encoder, malformed)) .ConfigureAwait(false); var restartedBackend = new AsyncSharedKeyValueStore(kv); var restarted = new SharedKeyValueSubscriptionStore(restartedBackend, CreateContext()); ServiceResultException? exception = Assert.ThrowsAsync( () => restarted.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); Assert.That(exception.Message, Does.Contain("monitored-item")); } /// /// Verifies that restoration rejects subscription records with trailing data. /// [Test] public async Task RestoreRejectsRecordWithTrailingDataAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore encoder = CreateStore(kv); StoredSubscription subscription = NewSubscription(980, 101); ByteString encoded = EncodeDefinition(encoder, subscription); byte[] withTrailingGarbage = [.. encoded.ToArray(), 1, 2, 3]; await kv .SetAsync( SharedKeyValueSubscriptionStore.KeyFor(subscription.Id), ByteString.From(withTrailingGarbage)) .ConfigureAwait(false); var restartedBackend = new AsyncSharedKeyValueStore(kv); var restarted = new SharedKeyValueSubscriptionStore(restartedBackend, CreateContext()); ServiceResultException? exception = Assert.ThrowsAsync( () => restarted.RestoreSubscriptionsAsync().AsTask()); Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadDecodingError)); Assert.That(exception.Message, Does.Contain("trailing data")); } /// /// Verifies that mirrored retransmission state preserves sequence continuity across replicas. /// [Test] public async Task RetransmissionStateIsVisibleToSecondReplicaAndKeepsSequenceContinuityAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore primary = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); primary.StoreRetransmissionState( 700, 4, [NewNotification(1), NewNotification(2), NewNotification(3)]); await primary.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? state = await backup.LoadRetransmissionStateAsync(700).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(4)); Assert.That( state.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { 1, 2, 3 })); NotificationMessage republished = state.SentMessages.Memory.ToArray().Single(m => m.SequenceNumber == 2); Assert.That(republished.NotificationData, Has.Count.EqualTo(1)); } /// /// Retained messages are restored oldest first across a sequence number rollover, /// so the sent/unsent boundary derived from their order stays correct. /// [Test] public async Task RetransmissionMessagesAreRestoredOldestFirstAcrossRolloverAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore primary = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); primary.StoreRetransmissionState( 705, 3, [NewNotification(uint.MaxValue), NewNotification(1), NewNotification(2)]); await primary.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? state = await backup.LoadRetransmissionStateAsync(705).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That( state!.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { uint.MaxValue, 1, 2 })); } /// /// Verifies that acknowledging a retransmission evicts its mirrored notification. /// [Test] public async Task RetransmissionAcknowledgeEvictsMirroredNotificationAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore primary = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); primary.StoreRetransmissionState( 701, 4, [NewNotification(1), NewNotification(2), NewNotification(3)]); await primary.FlushAsync().ConfigureAwait(false); primary.AcknowledgeNotification(701, 2); await primary.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? state = await backup.LoadRetransmissionStateAsync(701).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That( state!.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { 1, 3 })); } /// /// Verifies that a retransmission snapshot removes stale sequence keys. /// [Test] public async Task RetransmissionSnapshotRemovesStaleSequenceKeysAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); store.StoreRetransmissionState( 702, 4, [NewNotification(1), NewNotification(2), NewNotification(3)]); await store.FlushAsync().ConfigureAwait(false); store.StoreRetransmissionState(702, 5, [NewNotification(3), NewNotification(4)]); await store.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? state = await store.LoadRetransmissionStateAsync(702).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(5)); Assert.That( state.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { 3, 4 })); } /// /// Verifies that retransmission deltas add and remove only changed messages. /// [Test] public async Task RetransmissionDeltaAddsAndRemovesOnlyChangedMessagesAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); SharedKeyValueSubscriptionStore backup = CreateStore(kv); store.StoreRetransmissionStateDelta(708, 4, [NewNotification(1), NewNotification(2)], []); await store.FlushAsync().ConfigureAwait(false); store.StoreRetransmissionStateDelta(708, 5, [NewNotification(3)], [1]); await store.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? state = await backup.LoadRetransmissionStateAsync(708).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(5)); Assert.That( state.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { 2, 3 })); } /// /// Verifies that retransmission messages do not redundantly encode namespace tables. /// [Test] public async Task RetransmissionMessagesDoNotRepeatNamespaceTablesAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); store.StoreRetransmissionStateDelta(709, 2, [NewNotification(1)], []); await store.FlushAsync().ConfigureAwait(false); (bool stateFound, ByteString stateRaw) = await kv.TryGetAsync( SharedKeyValueSubscriptionStore.RetransmissionStateKeyFor(709)).ConfigureAwait(false); (bool messageFound, ByteString messageRaw) = await kv.TryGetAsync( SharedKeyValueSubscriptionStore.RetransmissionMessageKeyFor(709, 1)).ConfigureAwait(false); byte[] namespaceBytes = Encoding.UTF8.GetBytes("urn:test:subscriptions"); Assert.That(stateFound, Is.True); Assert.That(messageFound, Is.True); Assert.That(Contains(stateRaw.ToArray(), namespaceBytes), Is.True); Assert.That(Contains(messageRaw.ToArray(), namespaceBytes), Is.False); } /// /// Verifies that tampered retransmission state fails closed. /// [Test] public async Task RetransmissionTamperFailsClosedAsync() { using var kv = new InMemorySharedKeyValueStore(); using var protector = new AesCbcHmacRecordProtector(MakeKey(13)); var store = new SharedKeyValueSubscriptionStore(kv, CreateContext(), protector); store.StoreRetransmissionState(703, 2, [NewNotification(1)]); await store.FlushAsync().ConfigureAwait(false); await kv.SetAsync( SharedKeyValueSubscriptionStore.RetransmissionStateKeyFor(703), ByteString.From(new byte[] { 9, 8, 7 })).ConfigureAwait(false); SubscriptionRetransmissionState? state = await store.LoadRetransmissionStateAsync(703).ConfigureAwait(false); Assert.That(state, Is.Null); } /// /// Verifies that missing retransmission state is reported as null. /// [Test] public async Task MissingRetransmissionStateReturnsNullAsync() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); SubscriptionRetransmissionState? state = await store.LoadRetransmissionStateAsync(704).ConfigureAwait(false); Assert.That(state, Is.Null); } /// /// Verifies that retransmission mirroring uses the asynchronous backend without blocking its caller. /// [Test] public async Task RetransmissionMirrorUsesAsyncBackendWithoutBlockingAsync() { using var inner = new InMemorySharedKeyValueStore(); var kv = new AsyncSharedKeyValueStore(inner); var primary = new SharedKeyValueSubscriptionStore(kv, CreateContext()); var backup = new SharedKeyValueSubscriptionStore(kv, CreateContext()); primary.StoreRetransmissionState(705, 3, [NewNotification(1), NewNotification(2)]); await primary.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? state = await backup.LoadRetransmissionStateAsync(705).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(3)); Assert.That( state.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { 1, 2 })); } /// /// Verifies that retransmission deletion is ordered after in-flight mirror writes. /// [Test] public async Task RetransmissionDeleteIsOrderedAfterInflightMirrorWritesAsync() { const uint subscriptionId = 706; using var inner = new InMemorySharedKeyValueStore(); var kv = new BlockingRetransmissionSetStore(inner, subscriptionId); await using var store = new SharedKeyValueSubscriptionStore(kv, CreateContext()); await store.StoreSubscriptionsAsync([NewSubscription(subscriptionId, 76)]).ConfigureAwait(false); store.StoreRetransmissionState(subscriptionId, 3, [NewNotification(1), NewNotification(2)]); await kv.WaitForBlockedSetAsync().ConfigureAwait(false); await store.StoreSubscriptionsAsync([]).ConfigureAwait(false); kv.ReleaseBlockedSets(); await store.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? state = await store.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(state, Is.Null); Assert.That(await CountKeysAsync(inner, RetransmissionPrefixFor(subscriptionId)).ConfigureAwait(false), Is.Zero); } /// /// Verifies that a reused subscription identifier cannot load deleted retransmission state. /// [Test] public async Task ReusedSubscriptionIdDoesNotLoadDeletedRetransmissionStateAsync() { const uint subscriptionId = 707; using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); await using SharedKeyValueSubscriptionStore backup = CreateStore(kv); await store.StoreSubscriptionsAsync([NewSubscription(subscriptionId, 77)]).ConfigureAwait(false); store.StoreRetransmissionState(subscriptionId, 4, [NewNotification(1), NewNotification(2), NewNotification(3)]); await store.FlushAsync().ConfigureAwait(false); await store.StoreSubscriptionsAsync([]).ConfigureAwait(false); await store.FlushAsync().ConfigureAwait(false); await store.StoreSubscriptionsAsync([NewSubscription(subscriptionId, 78)]).ConfigureAwait(false); SubscriptionRetransmissionState? deletedState = await backup.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); store.StoreRetransmissionState(subscriptionId, 10, [NewNotification(9)]); await store.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? newState = await backup.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(deletedState, Is.Null); Assert.That(newState, Is.Not.Null); Assert.That(newState!.NextSequenceNumber, Is.EqualTo(10)); Assert.That( newState.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { 9 })); } /// /// Verifies that subscription store construction rejects a null backend or message context. /// [Test] public void ConstructorValidatesArguments() { IServiceMessageContext context = CreateContext(); var kvMock = new Mock(); Assert.That( () => new SharedKeyValueSubscriptionStore(null!, context), Throws.ArgumentNullException); Assert.That( () => new SharedKeyValueSubscriptionStore(kvMock.Object, null!), Throws.ArgumentNullException); } /// /// Verifies that data-change and event queue restoration fallbacks return null. /// [Test] public void RestoreQueueFallbacksReturnNull() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); Assert.That(store.RestoreDataChangeMonitoredItemQueue(1), Is.Null); Assert.That(store.RestoreEventMonitoredItemQueue(1), Is.Null); } /// /// Verifies that restore completion rejects a null subscription collection. /// [Test] public void OnSubscriptionRestoreCompleteValidatesArguments() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); Assert.That( async () => await store.OnSubscriptionRestoreCompleteAsync(null!).ConfigureAwait(false), Throws.ArgumentNullException); } /// /// Verifies that asynchronous disposal cancels blocked shared-store writes. /// [Test] public async Task DisposeAsyncCancelsBlockedSharedStoreWritesAsync() { await using var kv = new HangingWriteSharedKeyValueStore(); var store = new SharedKeyValueSubscriptionStore(kv, CreateContext()); store.StoreRetransmissionState(800, 2, [NewNotification(1)]); await kv.WaitForWriteAsync().ConfigureAwait(false); Task disposeTask = store.DisposeAsync().AsTask(); Task completed = await Task.WhenAny(disposeTask, Task.Delay(2000)).ConfigureAwait(false); Assert.That(completed, Is.SameAs(disposeTask)); await disposeTask.ConfigureAwait(false); } /// /// Verifies that definition encoding preserves absent, event, and aggregate monitored-item filters. /// [Test] public void DefinitionCodecRoundTripsAllSupportedFilterShapes() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); StoredSubscription subscription = NewSubscription(900, 90); subscription.MonitoredItems = [ NewItemWithFilter(900, 90, null), NewItemWithFilter(900, 91, new EventFilter()), NewItemWithFilter(900, 92, new AggregateFilter { AggregateType = ObjectIds.AggregateFunction_Average, ProcessingInterval = 1000, StartTime = new DateTimeUtc(638000000000000000) }) ]; MethodInfo encode = GetPrivateMethod("Encode", typeof(StoredSubscription)); MethodInfo decode = GetPrivateMethod("Decode", typeof(ByteString)); var payload = (ByteString)encode.Invoke(store, [subscription])!; var decoded = (StoredSubscription)decode.Invoke(store, [payload])!; var decodedItems = decoded.MonitoredItems.ToList(); Assert.That(decodedItems, Has.Count.EqualTo(3)); Assert.That(decodedItems[0].FilterToUse, Is.Null); Assert.That(decodedItems[1].FilterToUse, Is.TypeOf()); Assert.That(decodedItems[2].FilterToUse, Is.TypeOf()); } /// /// Verifies that definition decoding rejects unsupported format versions. /// [Test] public void DefinitionDecodeRejectsUnsupportedVersion() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); using var encoder = new BinaryEncoder(CreateContext()); encoder.WriteInt32(null, 999); var payload = ByteString.From(encoder.CloseAndReturnBuffer()); MethodInfo decode = GetPrivateMethod("Decode", typeof(ByteString)); TargetInvocationException? ex = Assert.Throws( () => decode.Invoke(store, [payload])); Assert.That(ex!.InnerException, Is.TypeOf()); } /// /// Verifies that continuation loading ignores unsupported envelope versions. /// [Test] public async Task ContinuationPointLoadIgnoresUnsupportedEnvelopeVersionAsync() { using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); var sessionId = new NodeId(Guid.NewGuid(), 1); var continuationPointId = Guid.NewGuid(); using var encoder = new BinaryEncoder(CreateContext()); encoder.WriteInt32(null, 999); await kv.SetAsync( SharedKeyValueSubscriptionStore.ContinuationPointKeyFor( sessionId, ContinuationPointKind.Browse, continuationPointId), ByteString.From(encoder.CloseAndReturnBuffer())).ConfigureAwait(false); ArrayOf envelopes = await store.LoadContinuationPointsAsync(sessionId).ConfigureAwait(false); Assert.That(envelopes, Is.Empty); } /// /// Verifies that continuation loading ignores invalid envelope identifiers. /// [Test] public async Task ContinuationPointLoadIgnoresInvalidEnvelopeIdAsync() { using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); var sessionId = new NodeId(Guid.NewGuid(), 1); var continuationPointId = Guid.NewGuid(); using var encoder = new BinaryEncoder(CreateContext()); encoder.WriteInt32(null, 1); encoder.WriteByteString(null, ByteString.From(new byte[] { 1, 2, 3 })); await kv.SetAsync( SharedKeyValueSubscriptionStore.ContinuationPointKeyFor( sessionId, ContinuationPointKind.Browse, continuationPointId), ByteString.From(encoder.CloseAndReturnBuffer())).ConfigureAwait(false); ArrayOf envelopes = await store.LoadContinuationPointsAsync(sessionId).ConfigureAwait(false); Assert.That(envelopes, Is.Empty); } /// /// Verifies that storing subscriptions rejects a null collection. /// [Test] public void StoreSubscriptionsRejectsNullArgument() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); Assert.That( async () => await store.StoreSubscriptionsAsync(null!).ConfigureAwait(false), Throws.ArgumentNullException); } /// /// Verifies that storing a continuation point rejects a null argument. /// [Test] public void StoreContinuationPointRejectsNullArgument() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); Assert.That(() => store.StoreContinuationPoint(null!), Throws.ArgumentNullException); } /// /// Verifies that continuation removal tolerates a null session. /// [Test] public void RemoveContinuationPointIgnoresNullSession() { using var kv = new InMemorySharedKeyValueStore(); SharedKeyValueSubscriptionStore store = CreateStore(kv); Assert.That( () => store.RemoveContinuationPoint(NodeId.Null, ContinuationPointKind.Browse, Guid.NewGuid()), Throws.Nothing); } /// /// Verifies that loading continuation points for a null session returns an empty collection. /// [Test] public async Task LoadContinuationPointsReturnsEmptyForNullSessionAsync() { using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); ArrayOf envelopes = await store.LoadContinuationPointsAsync(NodeId.Null).ConfigureAwait(false); Assert.That(envelopes, Is.Empty); } /// /// Verifies that loading an unsupported retransmission format returns null. /// [Test] public async Task LoadRetransmissionStateReturnsNullForUnsupportedVersionAsync() { const uint subscriptionId = 720; using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); using var encoder = new BinaryEncoder(CreateContext()); encoder.WriteInt32(null, 999); encoder.WriteUInt32(null, 5); await kv.SetAsync( SharedKeyValueSubscriptionStore.RetransmissionStateKeyFor(subscriptionId), ByteString.From(encoder.CloseAndReturnBuffer())).ConfigureAwait(false); SubscriptionRetransmissionState? state = await store.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(state, Is.Null); } /// /// Verifies that retransmission mirroring requeues its batch after a transient failure. /// [Test] public async Task RetransmissionMirrorRequeuesBatchAfterTransientFailureAsync() { const uint subscriptionId = 721; using var inner = new InMemorySharedKeyValueStore(); var kv = new ThrowOnceOnPrefixStore(inner, "subscription-retransmission/"); await using var store = new SharedKeyValueSubscriptionStore(kv, CreateContext()); store.StoreRetransmissionState(subscriptionId, 3, [NewNotification(1)]); await store.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? state = await store.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(kv.ThrowCount, Is.EqualTo(1)); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(3)); Assert.That( state.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { 1 })); } /// /// Verifies that requeueing a failed batch does not overwrite a send state that was /// stored while the batch was in flight. /// [Test] public async Task RetransmissionRequeueKeepsNewerSendStateAsync() { const uint subscriptionId = 726; using var inner = new InMemorySharedKeyValueStore(); var kv = new BlockingRetransmissionSetStore(inner, subscriptionId, throwAfterRelease: true); await using var store = new SharedKeyValueSubscriptionStore(kv, CreateContext()); store.StoreRetransmissionStateDelta( subscriptionId, 4, [NewNotification(1), NewNotification(2), NewNotification(3)], []); await kv.WaitForBlockedSetAsync().ConfigureAwait(false); store.StoreFirstUnsentSequenceNumber(subscriptionId, 4, 3); kv.ReleaseBlockedSets(); await store.FlushAsync().ConfigureAwait(false); await using SharedKeyValueSubscriptionStore backup = CreateStore(inner); SubscriptionRetransmissionState? state = await backup.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(4)); Assert.That(state.FirstUnsentSequenceNumber, Is.EqualTo(3)); Assert.That( state.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { 1, 2, 3 })); } /// /// Verifies that a failing batch does not drop the other batches taken with it. /// [Test] public async Task RetransmissionFailureRequeuesRemainingBatchesAsync() { const uint blockedSubscriptionId = 727; const uint failingSubscriptionId = 728; const uint otherSubscriptionId = 729; using var inner = new InMemorySharedKeyValueStore(); var blocking = new BlockingRetransmissionSetStore(inner, blockedSubscriptionId); var kv = new ThrowOnceOnPrefixStore( blocking, SharedKeyValueSubscriptionStore.RetransmissionStateKeyFor(failingSubscriptionId)); await using var store = new SharedKeyValueSubscriptionStore(kv, CreateContext()); // Hold the drain so that the next two subscriptions are taken in one pass. store.StoreRetransmissionState(blockedSubscriptionId, 2, [NewNotification(1)]); await blocking.WaitForBlockedSetAsync().ConfigureAwait(false); store.StoreRetransmissionState(failingSubscriptionId, 5, [NewNotification(4)]); store.StoreRetransmissionState(otherSubscriptionId, 8, [NewNotification(7)]); blocking.ReleaseBlockedSets(); await store.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? failing = await store.LoadRetransmissionStateAsync(failingSubscriptionId).ConfigureAwait(false); SubscriptionRetransmissionState? other = await store.LoadRetransmissionStateAsync(otherSubscriptionId).ConfigureAwait(false); Assert.That(kv.ThrowCount, Is.EqualTo(1)); Assert.That(failing, Is.Not.Null); Assert.That(failing!.NextSequenceNumber, Is.EqualTo(5)); Assert.That(other, Is.Not.Null); Assert.That(other!.NextSequenceNumber, Is.EqualTo(8)); Assert.That( other.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { 7 })); } /// /// Verifies that continuation mirroring requeues its batch after a transient failure. /// [Test] public async Task ContinuationPointMirrorRequeuesBatchAfterTransientFailureAsync() { using var inner = new InMemorySharedKeyValueStore(); var kv = new ThrowOnceOnPrefixStore(inner, "continuation-point/"); await using var store = new SharedKeyValueSubscriptionStore(kv, CreateContext()); var sessionId = new NodeId(Guid.NewGuid(), 1); var continuationPointId = Guid.NewGuid(); store.StoreContinuationPoint(new ContinuationPointEnvelope { Id = continuationPointId, OwnerSessionId = sessionId, Kind = ContinuationPointKind.Browse, Index = 4 }); await store.FlushAsync().ConfigureAwait(false); ArrayOf envelopes = await store.LoadContinuationPointsAsync(sessionId).ConfigureAwait(false); Assert.That(kv.ThrowCount, Is.EqualTo(1)); Assert.That(envelopes, Has.Count.EqualTo(1)); Assert.That(envelopes[0].Id, Is.EqualTo(continuationPointId)); Assert.That(envelopes[0].Index, Is.EqualTo(4)); } /// /// Verifies that the fallback decoder reads legacy retransmission messages. /// [Test] public async Task LegacyRetransmissionMessageDecodesViaFallbackAsync() { const uint subscriptionId = 722; using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); using var stateEncoder = new BinaryEncoder(CreateContext()); stateEncoder.WriteInt32(null, 1); stateEncoder.WriteUInt32(null, 7); await kv.SetAsync( SharedKeyValueSubscriptionStore.RetransmissionStateKeyFor(subscriptionId), ByteString.From(stateEncoder.CloseAndReturnBuffer())).ConfigureAwait(false); ServiceMessageContext messageContext = CreateContext(); using var messageEncoder = new BinaryEncoder(messageContext); messageEncoder.WriteStringArray(null, messageContext.NamespaceUris.ToArrayOf()); messageEncoder.WriteStringArray(null, messageContext.ServerUris.ToArrayOf()); messageEncoder.WriteEncodeable(null, NewNotification(3)); await kv.SetAsync( SharedKeyValueSubscriptionStore.RetransmissionMessageKeyFor(subscriptionId, 3), ByteString.From(messageEncoder.CloseAndReturnBuffer())).ConfigureAwait(false); SubscriptionRetransmissionState? state = await store.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(7)); Assert.That( state.SentMessages.Memory.ToArray().Select(m => m.SequenceNumber), Is.EqualTo(new uint[] { 3 })); } /// /// Verifies that the first unsent sequence number is mirrored to a second replica. /// [Test] public async Task RetransmissionFirstUnsentSequenceNumberIsVisibleToSecondReplicaAsync() { const uint subscriptionId = 723; using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore primary = CreateStore(kv); await using SharedKeyValueSubscriptionStore backup = CreateStore(kv); primary.StoreRetransmissionStateDelta( subscriptionId, 4, [NewNotification(1), NewNotification(2), NewNotification(3)], []); primary.StoreFirstUnsentSequenceNumber(subscriptionId, 4, 2); await primary.FlushAsync().ConfigureAwait(false); SubscriptionRetransmissionState? state = await backup.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(4)); Assert.That(state.FirstUnsentSequenceNumber, Is.EqualTo(2)); primary.StoreFirstUnsentSequenceNumber(subscriptionId, 4, 0); await primary.FlushAsync().ConfigureAwait(false); state = await backup.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(state!.FirstUnsentSequenceNumber, Is.Zero); } /// /// Verifies that a replica which restored a subscription and only reports send-state /// changes keeps the stored next sequence number instead of persisting 0. /// [Test] public async Task SendStateOnRestoredReplicaKeepsNextSequenceNumberAsync() { const uint subscriptionId = 725; using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore primary = CreateStore(kv); await using SharedKeyValueSubscriptionStore backup = CreateStore(kv); primary.StoreRetransmissionStateDelta( subscriptionId, 4, [NewNotification(1), NewNotification(2), NewNotification(3)], []); primary.StoreFirstUnsentSequenceNumber(subscriptionId, 4, 2); await primary.FlushAsync().ConfigureAwait(false); // The backup restores and returns the backlog: only the send state changes. SubscriptionRetransmissionState? state = await backup.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(state, Is.Not.Null); backup.StoreFirstUnsentSequenceNumber(subscriptionId, state!.NextSequenceNumber, 3); await backup.FlushAsync().ConfigureAwait(false); state = await primary.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(4)); Assert.That(state.FirstUnsentSequenceNumber, Is.EqualTo(3)); } /// /// Verifies that a state record written before the first unsent sequence number /// existed still loads, with every retained message treated as sent. /// [Test] public async Task RetransmissionStateWithoutFirstUnsentSequenceNumberLoadsAsync() { const uint subscriptionId = 724; using var kv = new InMemorySharedKeyValueStore(); await using SharedKeyValueSubscriptionStore store = CreateStore(kv); ServiceMessageContext context = CreateContext(); using var encoder = new BinaryEncoder(context); encoder.WriteInt32(null, 2); encoder.WriteUInt32(null, 9); encoder.WriteStringArray(null, context.NamespaceUris.ToArrayOf()); encoder.WriteStringArray(null, context.ServerUris.ToArrayOf()); await kv.SetAsync( SharedKeyValueSubscriptionStore.RetransmissionStateKeyFor(subscriptionId), ByteString.From(encoder.CloseAndReturnBuffer())).ConfigureAwait(false); SubscriptionRetransmissionState? state = await store.LoadRetransmissionStateAsync(subscriptionId).ConfigureAwait(false); Assert.That(state, Is.Not.Null); Assert.That(state!.NextSequenceNumber, Is.EqualTo(9)); Assert.That(state.FirstUnsentSequenceNumber, Is.Zero); } private static SharedKeyValueSubscriptionStore CreateStore(InMemorySharedKeyValueStore kv) { return new SharedKeyValueSubscriptionStore(kv, CreateContext()); } private static ServiceMessageContext CreateContext() { ITelemetryContext telemetry = NUnitTelemetryContext.Create(); var context = ServiceMessageContext.CreateEmpty(telemetry); context.NamespaceUris.GetIndexOrAppend("urn:test:subscriptions"); return context; } private static StoredSubscription NewSubscription(uint subscriptionId, uint monitoredItemId) { return new StoredSubscription { Id = subscriptionId, IsDurable = true, LifetimeCounter = 2, MaxLifetimeCount = 120, MaxKeepaliveCount = 12, MaxMessageCount = 8, MaxNotificationsPerPublish = 99, PublishingInterval = 250, Priority = 7, LastSentMessage = 0, SequenceNumber = 0, SentMessages = [], UserIdentityToken = new AnonymousIdentityToken(), MonitoredItems = [NewItem(subscriptionId, monitoredItemId)] }; } private static StoredMonitoredItem NewItem(uint subscriptionId, uint monitoredItemId) { var filter = new DataChangeFilter { Trigger = DataChangeTrigger.StatusValue, DeadbandType = (uint)DeadbandType.Absolute, DeadbandValue = 1.5 }; return new StoredMonitoredItem { IsRestored = false, AlwaysReportUpdates = true, AttributeId = Attributes.Value, ClientHandle = 987, DiagnosticsMasks = DiagnosticsMasks.OperationAll, DiscardOldest = true, Encoding = new QualifiedName(BrowseNames.DefaultXml, 0), Id = monitoredItemId, IndexRange = "1:3", ParsedIndexRange = NumericRange.Parse("1:3"), IsDurable = true, LastError = ServiceResult.Good, LastValue = new DataValue(new Variant(42)), MonitoringMode = MonitoringMode.Reporting, NodeId = new NodeId("Temperature", NamespaceIndex), FilterToUse = filter, OriginalFilter = filter, QueueSize = 3, Range = 100, SamplingInterval = 50, SourceSamplingInterval = 25, SubscriptionId = subscriptionId, TimestampsToReturn = TimestampsToReturn.Both, TypeMask = 1 }; } private static StoredMonitoredItem NewItemWithFilter( uint subscriptionId, uint monitoredItemId, MonitoringFilter? filter) { StoredMonitoredItem item = NewItem(subscriptionId, monitoredItemId); item.FilterToUse = filter!; item.OriginalFilter = filter!; return item; } private static MethodInfo GetPrivateMethod(string name, params Type[] parameterTypes) { MethodInfo? method = typeof(SharedKeyValueSubscriptionStore).GetMethod( name, BindingFlags.Instance | BindingFlags.Static | BindingFlags.NonPublic, null, parameterTypes, null); Assert.That(method, Is.Not.Null); return method!; } private static ByteString EncodeDefinition( SharedKeyValueSubscriptionStore store, StoredSubscription subscription) { MethodInfo encode = GetPrivateMethod("Encode", typeof(StoredSubscription)); return (ByteString)encode.Invoke(store, [subscription])!; } private static ByteString EncodeLegacyDefinition(StoredSubscription subscription) { ServiceMessageContext context = CreateContext(); using var encoder = new BinaryEncoder(context); encoder.WriteInt32(null, 1); encoder.WriteStringArray(null, context.NamespaceUris.ToArrayOf()); encoder.WriteStringArray(null, context.ServerUris.ToArrayOf()); MethodInfo encodeSubscription = GetPrivateMethod( "EncodeSubscription", typeof(BinaryEncoder), typeof(StoredSubscription), typeof(int)); encodeSubscription.Invoke(null, [encoder, subscription, 1]); return ByteString.From(encoder.CloseAndReturnBuffer()); } private static async Task> GetSingleGenerationRecordAsync( InMemorySharedKeyValueStore store) { var records = new List>(); await foreach (KeyValuePair pair in store .ScanAsync(SharedKeyValueSubscriptionStore.SnapshotGenerationRootPrefix())) { records.Add(pair); } Assert.That(records, Has.Count.EqualTo(1)); return records[0]; } private static string GenerationPrefixOf(string generationRecordKey) { int separator = generationRecordKey.LastIndexOf('/'); Assert.That(separator, Is.GreaterThan(-1)); return generationRecordKey[..(separator + 1)]; } private static async Task> FindGenerationRecordAsync( InMemorySharedKeyValueStore store, uint subscriptionId) { string suffix = "/" + subscriptionId.ToString(System.Globalization.CultureInfo.InvariantCulture); var records = new List>(); await foreach (KeyValuePair pair in store .ScanAsync(SharedKeyValueSubscriptionStore.SnapshotGenerationRootPrefix())) { if (pair.Key.EndsWith(suffix, StringComparison.Ordinal)) { records.Add(pair); } } Assert.That(records, Has.Count.EqualTo(1)); return records[0]; } private static uint[] GetCachedSubscriptionIds(SharedKeyValueSubscriptionStore store) { FieldInfo? cacheField = typeof(SharedKeyValueSubscriptionStore).GetField( "m_definitionCache", BindingFlags.Instance | BindingFlags.NonPublic); Assert.That(cacheField, Is.Not.Null); object? cache = cacheField!.GetValue(store); Assert.That(cache, Is.Not.Null); // The cache guards its own dictionary and exposes only operations, so the // field is private; this white-box helper reaches past that deliberately. FieldInfo? subscriptionsField = cache!.GetType().GetField( "m_subscriptions", BindingFlags.Instance | BindingFlags.NonPublic); Assert.That(subscriptionsField, Is.Not.Null); var subscriptions = (Dictionary)subscriptionsField!.GetValue(cache)!; uint[] ids = [.. subscriptions.Keys]; Array.Sort(ids); return ids; } private static NotificationMessage NewNotification(uint sequenceNumber) { var notification = new DataChangeNotification { MonitoredItems = [ new MonitoredItemNotification { ClientHandle = sequenceNumber, Value = new DataValue(new Variant((int)sequenceNumber)) } ] }; return new NotificationMessage { SequenceNumber = sequenceNumber, PublishTime = DateTimeUtc.Now, NotificationData = [new ExtensionObject(notification)] }; } private static string RetransmissionPrefixFor(uint subscriptionId) { string messageKey = SharedKeyValueSubscriptionStore.RetransmissionMessageKeyFor(subscriptionId, 0); return messageKey[..^10]; } private static async Task CountKeysAsync(InMemorySharedKeyValueStore store, string keyPrefix) { int count = 0; await foreach (KeyValuePair pair in store.ScanAsync(keyPrefix)) { _ = pair; count++; } (bool found, _) = await store.TryGetAsync( SharedKeyValueSubscriptionStore.RetransmissionStateKeyFor( uint.Parse(keyPrefix.Split('/')[1], System.Globalization.CultureInfo.InvariantCulture))).ConfigureAwait(false); return found ? count + 1 : count; } private static void AssertSubscription(StoredSubscription actual, StoredSubscription expected) { Assert.That(actual.Id, Is.EqualTo(expected.Id)); Assert.That(actual.IsDurable, Is.EqualTo(expected.IsDurable)); Assert.That(actual.MaxLifetimeCount, Is.EqualTo(expected.MaxLifetimeCount)); Assert.That(actual.MaxKeepaliveCount, Is.EqualTo(expected.MaxKeepaliveCount)); Assert.That(actual.MaxNotificationsPerPublish, Is.EqualTo(expected.MaxNotificationsPerPublish)); Assert.That(actual.PublishingInterval, Is.EqualTo(expected.PublishingInterval)); Assert.That(actual.Priority, Is.EqualTo(expected.Priority)); } private static void AssertMonitoredItem(StoredMonitoredItem actual, StoredMonitoredItem expected) { Assert.That(actual.IsDeleted, Is.EqualTo(expected.IsDeleted)); Assert.That(actual.IsDetached, Is.EqualTo(expected.IsDetached)); Assert.That(actual.SubscriptionId, Is.EqualTo(expected.SubscriptionId)); Assert.That(actual.Id, Is.EqualTo(expected.Id)); Assert.That(actual.NodeId, Is.EqualTo(expected.NodeId)); Assert.That(actual.AttributeId, Is.EqualTo(expected.AttributeId)); Assert.That(actual.MonitoringMode, Is.EqualTo(expected.MonitoringMode)); Assert.That(actual.SamplingInterval, Is.EqualTo(expected.SamplingInterval)); Assert.That(actual.QueueSize, Is.EqualTo(expected.QueueSize)); Assert.That(actual.DiscardOldest, Is.EqualTo(expected.DiscardOldest)); Assert.That(actual.FilterToUse, Is.TypeOf()); Assert.That(((DataChangeFilter)actual.FilterToUse).DeadbandValue, Is.EqualTo(1.5)); } private static bool Contains(byte[] haystack, byte[] needle) { for (int ii = 0; ii + needle.Length <= haystack.Length; ii++) { bool match = true; for (int jj = 0; jj < needle.Length; jj++) { if (haystack[ii + jj] != needle[jj]) { match = false; break; } } if (match) { return true; } } return false; } private static byte[] MakeKey(byte seed) { byte[] key = new byte[32]; for (int ii = 0; ii < key.Length; ii++) { key[ii] = (byte)(seed + ii); } return key; } private sealed class BlockingManifestCommitStore : ISharedKeyValueStore { public BlockingManifestCommitStore(ISharedKeyValueStore inner) { m_inner = inner; } public ValueTask<(bool Found, ByteString Value)> TryGetAsync( string key, CancellationToken ct = default) { return m_inner.TryGetAsync(key, ct); } public async ValueTask SetAsync( string key, ByteString value, CancellationToken ct = default) { if (string.Equals( key, SharedKeyValueSubscriptionStore.SnapshotManifestKey(), StringComparison.Ordinal) && Interlocked.Exchange(ref m_manifestBlocked, 1) == 0) { m_blocked.TrySetResult(true); await m_release.Task.WaitAsync(ct).ConfigureAwait(false); } await m_inner.SetAsync(key, value, ct).ConfigureAwait(false); } public ValueTask CompareAndSwapAsync( string key, ByteString expected, ByteString value, CancellationToken ct = default) { return m_inner.CompareAndSwapAsync(key, expected, value, ct); } public ValueTask DeleteAsync(string key, CancellationToken ct = default) { return m_inner.DeleteAsync(key, ct); } public IAsyncEnumerable> ScanAsync( string keyPrefix, CancellationToken ct = default) { return m_inner.ScanAsync(keyPrefix, ct); } public IAsyncEnumerable WatchAsync( string keyPrefix, CancellationToken ct = default) { return m_inner.WatchAsync(keyPrefix, ct); } public async Task WaitForBlockedManifestAsync() { await m_blocked.Task .WaitAsync(TimeSpan.FromSeconds(30)) .ConfigureAwait(false); } public void ReleaseManifest() { m_release.TrySetResult(true); } private readonly ISharedKeyValueStore m_inner; private readonly TaskCompletionSource m_blocked = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource m_release = new(TaskCreationOptions.RunContinuationsAsynchronously); private int m_manifestBlocked; } private sealed class AsyncSharedKeyValueStore : ISharedKeyValueStore { public AsyncSharedKeyValueStore(ISharedKeyValueStore inner) { m_inner = inner; } public async ValueTask<(bool Found, ByteString Value)> TryGetAsync( string key, CancellationToken ct = default) { await Task.Yield(); return await m_inner.TryGetAsync(key, ct).ConfigureAwait(false); } public async ValueTask SetAsync(string key, ByteString value, CancellationToken ct = default) { await Task.Yield(); await m_inner.SetAsync(key, value, ct).ConfigureAwait(false); } public async ValueTask CompareAndSwapAsync( string key, ByteString expected, ByteString value, CancellationToken ct = default) { await Task.Yield(); return await m_inner.CompareAndSwapAsync(key, expected, value, ct).ConfigureAwait(false); } public async ValueTask DeleteAsync(string key, CancellationToken ct = default) { await Task.Yield(); return await m_inner.DeleteAsync(key, ct).ConfigureAwait(false); } public async IAsyncEnumerable> ScanAsync( string keyPrefix, [EnumeratorCancellation] CancellationToken ct = default) { await Task.Yield(); await foreach (KeyValuePair pair in m_inner.ScanAsync(keyPrefix, ct)) { yield return pair; } } public async IAsyncEnumerable WatchAsync( string keyPrefix, [EnumeratorCancellation] CancellationToken ct = default) { await Task.Yield(); await foreach (KeyValueChange change in m_inner.WatchAsync(keyPrefix, ct)) { yield return change; } } private readonly ISharedKeyValueStore m_inner; } private sealed class HangingWriteSharedKeyValueStore : ISharedKeyValueStore, IAsyncDisposable { public Task WaitForWriteAsync() { return m_writeStarted.Task; } public ValueTask<(bool Found, ByteString Value)> TryGetAsync( string key, CancellationToken ct = default) { return new ValueTask<(bool Found, ByteString Value)>((false, default)); } public ValueTask SetAsync(string key, ByteString value, CancellationToken ct = default) { m_writeStarted.TrySetResult(true); return new ValueTask(Task.Delay(Timeout.Infinite, ct)); } public ValueTask CompareAndSwapAsync( string key, ByteString expected, ByteString value, CancellationToken ct = default) { return new ValueTask(true); } public ValueTask DeleteAsync(string key, CancellationToken ct = default) { m_writeStarted.TrySetResult(true); return WaitForDeleteCancellationAsync(ct); } public IAsyncEnumerable> ScanAsync( string keyPrefix, CancellationToken ct = default) { return TestAsyncEnumerable.Empty>(); } public IAsyncEnumerable WatchAsync( string keyPrefix, CancellationToken ct = default) { return TestAsyncEnumerable.Empty(); } public ValueTask DisposeAsync() { return default; } private static async ValueTask WaitForDeleteCancellationAsync(CancellationToken ct) { await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false); return false; } private readonly TaskCompletionSource m_writeStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); } private sealed class BlockingRetransmissionSetStore : ISharedKeyValueStore { public BlockingRetransmissionSetStore( ISharedKeyValueStore inner, uint subscriptionId, bool throwAfterRelease = false) { m_inner = inner; m_throwAfterRelease = throwAfterRelease; m_keyPrefix = "subscription-retransmission/" + subscriptionId.ToString("D", System.Globalization.CultureInfo.InvariantCulture) + "/"; } public ValueTask<(bool Found, ByteString Value)> TryGetAsync( string key, CancellationToken ct = default) { return m_inner.TryGetAsync(key, ct); } public async ValueTask SetAsync(string key, ByteString value, CancellationToken ct = default) { if (key.StartsWith(m_keyPrefix, StringComparison.Ordinal) && Interlocked.Exchange(ref m_blockedSetCount, 1) == 0) { m_blocked.SetResult(true); await m_release.Task.ConfigureAwait(false); if (m_throwAfterRelease) { throw new InvalidOperationException("Injected transient mirror failure."); } } await m_inner.SetAsync(key, value, ct).ConfigureAwait(false); } public ValueTask CompareAndSwapAsync( string key, ByteString expected, ByteString value, CancellationToken ct = default) { return m_inner.CompareAndSwapAsync(key, expected, value, ct); } public ValueTask DeleteAsync(string key, CancellationToken ct = default) { return m_inner.DeleteAsync(key, ct); } public async IAsyncEnumerable> ScanAsync( string keyPrefix, [EnumeratorCancellation] CancellationToken ct = default) { await foreach (KeyValuePair pair in m_inner.ScanAsync(keyPrefix, ct)) { yield return pair; } } public async IAsyncEnumerable WatchAsync( string keyPrefix, [EnumeratorCancellation] CancellationToken ct = default) { await foreach (KeyValueChange change in m_inner.WatchAsync(keyPrefix, ct)) { yield return change; } } public async Task WaitForBlockedSetAsync() { Task completed = await Task.WhenAny(m_blocked.Task, Task.Delay(TimeSpan.FromSeconds(10))) .ConfigureAwait(false); Assert.That(completed, Is.SameAs(m_blocked.Task)); } public void ReleaseBlockedSets() { m_release.SetResult(true); } private readonly ISharedKeyValueStore m_inner; private readonly string m_keyPrefix; private readonly bool m_throwAfterRelease; private readonly TaskCompletionSource m_blocked = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource m_release = new(TaskCreationOptions.RunContinuationsAsynchronously); private int m_blockedSetCount; } private sealed class ThrowOnceOnPrefixStore : ISharedKeyValueStore { public ThrowOnceOnPrefixStore(ISharedKeyValueStore inner, string keyPrefix) { m_inner = inner; m_keyPrefix = keyPrefix; } public int ThrowCount => m_throwCount; public ValueTask<(bool Found, ByteString Value)> TryGetAsync( string key, CancellationToken ct = default) { return m_inner.TryGetAsync(key, ct); } public ValueTask SetAsync(string key, ByteString value, CancellationToken ct = default) { if (key.StartsWith(m_keyPrefix, StringComparison.Ordinal) && Interlocked.Exchange(ref m_throwGate, 1) == 0) { Interlocked.Increment(ref m_throwCount); throw new InvalidOperationException("Injected transient mirror failure."); } return m_inner.SetAsync(key, value, ct); } public ValueTask CompareAndSwapAsync( string key, ByteString expected, ByteString value, CancellationToken ct = default) { return m_inner.CompareAndSwapAsync(key, expected, value, ct); } public ValueTask DeleteAsync(string key, CancellationToken ct = default) { return m_inner.DeleteAsync(key, ct); } public IAsyncEnumerable> ScanAsync( string keyPrefix, CancellationToken ct = default) { return m_inner.ScanAsync(keyPrefix, ct); } public IAsyncEnumerable WatchAsync( string keyPrefix, CancellationToken ct = default) { return m_inner.WatchAsync(keyPrefix, ct); } private readonly ISharedKeyValueStore m_inner; private readonly string m_keyPrefix; private int m_throwGate; private int m_throwCount; } /// /// Backend whose yields every matching record twice, simulating a /// non-idempotent enumeration from a misbehaving distributed backend. /// private sealed class DuplicatingScanStore : ISharedKeyValueStore { public DuplicatingScanStore(ISharedKeyValueStore inner) { m_inner = inner; } public ValueTask<(bool Found, ByteString Value)> TryGetAsync( string key, CancellationToken ct = default) { return m_inner.TryGetAsync(key, ct); } public ValueTask SetAsync(string key, ByteString value, CancellationToken ct = default) { return m_inner.SetAsync(key, value, ct); } public ValueTask CompareAndSwapAsync( string key, ByteString expected, ByteString value, CancellationToken ct = default) { return m_inner.CompareAndSwapAsync(key, expected, value, ct); } public ValueTask DeleteAsync(string key, CancellationToken ct = default) { return m_inner.DeleteAsync(key, ct); } public async IAsyncEnumerable> ScanAsync( string keyPrefix, [EnumeratorCancellation] CancellationToken ct = default) { var snapshot = new List>(); await foreach (KeyValuePair pair in m_inner.ScanAsync(keyPrefix, ct) .ConfigureAwait(false)) { snapshot.Add(pair); } foreach (KeyValuePair pair in snapshot) { yield return pair; } foreach (KeyValuePair pair in snapshot) { yield return pair; } } public IAsyncEnumerable WatchAsync( string keyPrefix, CancellationToken ct = default) { return m_inner.WatchAsync(keyPrefix, ct); } private readonly ISharedKeyValueStore m_inner; } /// /// Backend that throws up front on every call, matching a /// well-behaved network backend that observes cancellation before doing any work. /// private sealed class CancellationCheckingStore : ISharedKeyValueStore { public CancellationCheckingStore(ISharedKeyValueStore inner) { m_inner = inner; } public ValueTask<(bool Found, ByteString Value)> TryGetAsync( string key, CancellationToken ct = default) { ct.ThrowIfCancellationRequested(); return m_inner.TryGetAsync(key, ct); } public ValueTask SetAsync(string key, ByteString value, CancellationToken ct = default) { ct.ThrowIfCancellationRequested(); return m_inner.SetAsync(key, value, ct); } public ValueTask CompareAndSwapAsync( string key, ByteString expected, ByteString value, CancellationToken ct = default) { ct.ThrowIfCancellationRequested(); return m_inner.CompareAndSwapAsync(key, expected, value, ct); } public ValueTask DeleteAsync(string key, CancellationToken ct = default) { ct.ThrowIfCancellationRequested(); return m_inner.DeleteAsync(key, ct); } public IAsyncEnumerable> ScanAsync( string keyPrefix, CancellationToken ct = default) { ct.ThrowIfCancellationRequested(); return m_inner.ScanAsync(keyPrefix, ct); } public IAsyncEnumerable WatchAsync( string keyPrefix, CancellationToken ct = default) { ct.ThrowIfCancellationRequested(); return m_inner.WatchAsync(keyPrefix, ct); } private readonly ISharedKeyValueStore m_inner; } /// /// Backend that throws once its call counter (scoped to snapshot-generation /// keys) reaches failOnCallNumber, simulating a shared-store write failure partway through /// committing an immutable snapshot generation. /// private sealed class ThrowOnNthGenerationWriteStore : ISharedKeyValueStore { public ThrowOnNthGenerationWriteStore(ISharedKeyValueStore inner, int failOnCallNumber) { m_inner = inner; m_failOnCallNumber = failOnCallNumber; } public ValueTask<(bool Found, ByteString Value)> TryGetAsync( string key, CancellationToken ct = default) { return m_inner.TryGetAsync(key, ct); } public ValueTask SetAsync(string key, ByteString value, CancellationToken ct = default) { if (key.StartsWith( SharedKeyValueSubscriptionStore.SnapshotGenerationRootPrefix(), StringComparison.Ordinal) && Interlocked.Increment(ref m_callCount) == m_failOnCallNumber) { throw new InvalidOperationException("Injected snapshot write failure."); } return m_inner.SetAsync(key, value, ct); } public ValueTask CompareAndSwapAsync( string key, ByteString expected, ByteString value, CancellationToken ct = default) { return m_inner.CompareAndSwapAsync(key, expected, value, ct); } public ValueTask DeleteAsync(string key, CancellationToken ct = default) { return m_inner.DeleteAsync(key, ct); } public IAsyncEnumerable> ScanAsync( string keyPrefix, CancellationToken ct = default) { return m_inner.ScanAsync(keyPrefix, ct); } public IAsyncEnumerable WatchAsync( string keyPrefix, CancellationToken ct = default) { return m_inner.WatchAsync(keyPrefix, ct); } private readonly ISharedKeyValueStore m_inner; private readonly int m_failOnCallNumber; private int m_callCount; } } }