diff --git a/src/Opc.Ua.Redundancy.Server/Subscriptions/SharedKeyValueSubscriptionStore.cs b/src/Opc.Ua.Redundancy.Server/Subscriptions/SharedKeyValueSubscriptionStore.cs index c2cb44a05a..b7cb3d7c35 100644 --- a/src/Opc.Ua.Redundancy.Server/Subscriptions/SharedKeyValueSubscriptionStore.cs +++ b/src/Opc.Ua.Redundancy.Server/Subscriptions/SharedKeyValueSubscriptionStore.cs @@ -109,46 +109,108 @@ public async ValueTask StoreSubscriptionsAsync( var liveIds = new HashSet(); foreach (StoredSubscription subscription in snapshot) { - liveIds.Add(subscription.Id); - string key = KeyFor(subscription.Id); - await m_store - .SetAsync(key, m_protector.Protect(Encode(subscription)), cancellationToken) - .ConfigureAwait(false); + if (!liveIds.Add(subscription.Id)) + { + throw new ArgumentException( + "The subscription snapshot contains duplicate subscription ids.", + nameof(subscriptions)); + } + ValidateSnapshotSubscription(subscription, nameof(subscriptions)); } - uint[] removedIds; - lock (m_definitionCache.Lock) + await m_definitionCache.SnapshotCommitLock.WaitAsync(cancellationToken).ConfigureAwait(false); + try { - removedIds = [.. m_definitionCache.Subscriptions.Keys.Where(id => !liveIds.Contains(id))]; + var generation = Guid.NewGuid(); foreach (StoredSubscription subscription in snapshot) { - m_definitionCache.Subscriptions[subscription.Id] = CloneSubscription(subscription); + await m_store + .SetAsync( + SnapshotGenerationKeyFor(generation, subscription.Id), + m_protector.Protect(Encode(subscription)), + cancellationToken) + .ConfigureAwait(false); } + + await m_store + .SetAsync( + SnapshotManifestKey(), + m_protector.Protect(EncodeSnapshotManifest(generation, (uint)snapshot.Count)), + cancellationToken) + .ConfigureAwait(false); + + // Keep prior generations so readers that captured an older manifest can complete safely. + // Reclamation requires reader pinning or an external retention policy. + uint[] removedIds; + lock (m_definitionCache.Lock) + { + removedIds = [.. m_definitionCache.Subscriptions.Keys.Where(id => !liveIds.Contains(id))]; + m_definitionCache.Subscriptions.Clear(); + foreach (StoredSubscription subscription in snapshot) + { + m_definitionCache.Subscriptions.Add( + subscription.Id, + CloneSubscription(subscription)); + } + } + foreach (uint subscriptionId in removedIds) { - m_definitionCache.Subscriptions.Remove(subscriptionId); + DeleteRetransmissionState(subscriptionId); } } - - foreach (uint subscriptionId in removedIds) + finally { - await m_store.DeleteAsync(KeyFor(subscriptionId), cancellationToken).ConfigureAwait(false); - DeleteRetransmissionState(subscriptionId); + m_definitionCache.SnapshotCommitLock.Release(); } return true; } /// - public ValueTask RestoreSubscriptionsAsync( + public async ValueTask RestoreSubscriptionsAsync( CancellationToken cancellationToken = default) { + (bool foundManifest, ByteString protectedManifest) = await m_store + .TryGetAsync(SnapshotManifestKey(), cancellationToken) + .ConfigureAwait(false); + + Dictionary restored; + if (foundManifest) + { + if (!m_protector.TryUnprotect(protectedManifest, out ByteString manifestPayload) || + manifestPayload.IsNull) + { + throw new ServiceResultException( + StatusCodes.BadSecurityChecksFailed, + "The persisted subscription snapshot manifest could not be authenticated."); + } + SnapshotManifest manifest = DecodeSnapshotManifest(manifestPayload); + restored = await ReadSnapshotAsync( + SnapshotGenerationPrefixFor(manifest.Generation), + manifest.RecordCount, + cancellationToken).ConfigureAwait(false); + } + else + { + restored = await ReadSnapshotAsync(Prefix, null, cancellationToken).ConfigureAwait(false); + } + + List subscriptions = restored.Values + .OrderBy(static subscription => subscription.Id) + .ToList(); lock (m_definitionCache.Lock) { - return new ValueTask(new RestoreSubscriptionResult( - true, - m_definitionCache.Subscriptions.Values.Select(CloneSubscription).ToList())); + m_definitionCache.Subscriptions.Clear(); + foreach (StoredSubscription subscription in subscriptions) + { + m_definitionCache.Subscriptions.Add( + subscription.Id, + CloneSubscription(subscription)); + } } + + return new RestoreSubscriptionResult(true, subscriptions); } /// @@ -199,22 +261,18 @@ public async ValueTask OnSubscriptionRestoreCompleteAsync( throw new ArgumentNullException(nameof(createdSubscriptions)); } - uint[] removedIds; + List restoredSubscriptions; lock (m_definitionCache.Lock) { var liveIds = new HashSet(createdSubscriptions.Keys); - removedIds = [.. m_definitionCache.Subscriptions.Keys.Where(id => !liveIds.Contains(id))]; - foreach (uint subscriptionId in removedIds) - { - m_definitionCache.Subscriptions.Remove(subscriptionId); - } - } - - foreach (uint subscriptionId in removedIds) - { - await m_store.DeleteAsync(KeyFor(subscriptionId), cancellationToken).ConfigureAwait(false); - DeleteRetransmissionState(subscriptionId); + restoredSubscriptions = + [ + .. m_definitionCache.Subscriptions + .Where(pair => liveIds.Contains(pair.Key)) + .Select(static pair => CloneSubscription(pair.Value)) + ]; } + await StoreSubscriptionsAsync(restoredSubscriptions, cancellationToken).ConfigureAwait(false); if (m_queueFactory != null) { @@ -702,6 +760,48 @@ internal static string KeyFor(uint subscriptionId) return Prefix + subscriptionId.ToString("D", System.Globalization.CultureInfo.InvariantCulture); } + /// + /// Gets the key that atomically commits the current subscription snapshot generation. + /// + /// The manifest key. + internal static string SnapshotManifestKey() + { + return DefinitionSnapshotManifestKey; + } + + /// + /// Gets the root prefix containing immutable subscription snapshot generations. + /// + /// The generation root prefix. + internal static string SnapshotGenerationRootPrefix() + { + return DefinitionSnapshotGenerationPrefix; + } + + /// + /// Gets the prefix containing records for one subscription snapshot generation. + /// + /// The snapshot generation. + /// The generation prefix. + internal static string SnapshotGenerationPrefixFor(Guid generation) + { + return DefinitionSnapshotGenerationPrefix + + generation.ToString("N", System.Globalization.CultureInfo.InvariantCulture) + + "/"; + } + + /// + /// Gets the key for one subscription record in an immutable snapshot generation. + /// + /// The snapshot generation. + /// The subscription id. + /// The generation record key. + internal static string SnapshotGenerationKeyFor(Guid generation, uint subscriptionId) + { + return SnapshotGenerationPrefixFor(generation) + + subscriptionId.ToString("D", System.Globalization.CultureInfo.InvariantCulture); + } + /// /// Computes the shared-store key for a subscription retransmission state record. /// @@ -744,6 +844,126 @@ internal static string ContinuationPointKeyFor( id.ToString("N", System.Globalization.CultureInfo.InvariantCulture); } + private async ValueTask> ReadSnapshotAsync( + string keyPrefix, + uint? expectedRecordCount, + CancellationToken cancellationToken) + { + var restored = new Dictionary(); + await foreach (KeyValuePair pair in m_store + .ScanAsync(keyPrefix, cancellationToken) + .ConfigureAwait(false)) + { + if (!TryParseSubscriptionKey(pair.Key, keyPrefix, out uint subscriptionId)) + { + throw new ServiceResultException( + StatusCodes.BadDecodingError, + "The persisted subscription key is malformed."); + } + if (!m_protector.TryUnprotect(pair.Value, out ByteString payload) || payload.IsNull) + { + throw new ServiceResultException( + StatusCodes.BadSecurityChecksFailed, + "The persisted subscription record could not be authenticated."); + } + + StoredSubscription subscription = Decode(payload); + ValidatePersistedSubscription(subscriptionId, subscription); + if (restored.ContainsKey(subscriptionId)) + { + throw new ServiceResultException( + StatusCodes.BadDecodingError, + "The persisted subscription snapshot contains duplicate records."); + } + restored.Add(subscriptionId, subscription); + } + + if (expectedRecordCount.HasValue && (uint)restored.Count != expectedRecordCount.Value) + { + throw new ServiceResultException( + StatusCodes.BadDecodingError, + "The persisted subscription snapshot is incomplete."); + } + + return restored; + } + + private static bool TryParseSubscriptionKey( + string key, + string keyPrefix, + out uint subscriptionId) + { + subscriptionId = 0; + if (!key.StartsWith(keyPrefix, StringComparison.Ordinal) || key.Length == keyPrefix.Length) + { + return false; + } + + for (int ii = keyPrefix.Length; ii < key.Length; ii++) + { + char character = key[ii]; + if (character is < '0' or > '9') + { + return false; + } + + uint digit = (uint)(character - '0'); + if (subscriptionId > (uint.MaxValue - digit) / 10) + { + return false; + } + subscriptionId = (subscriptionId * 10) + digit; + } + + return string.Equals( + key, + keyPrefix + subscriptionId.ToString("D", System.Globalization.CultureInfo.InvariantCulture), + StringComparison.Ordinal); + } + + private static void ValidateSnapshotSubscription(StoredSubscription subscription, string parameterName) + { + if (!HasValidMonitoredItemIdentities(subscription.Id, subscription.MonitoredItems)) + { + throw new ArgumentException( + "The subscription snapshot contains invalid monitored-item identities.", + parameterName); + } + } + + private static void ValidatePersistedSubscription(uint subscriptionId, StoredSubscription subscription) + { + if (subscription.Id != subscriptionId) + { + throw new ServiceResultException( + StatusCodes.BadDecodingError, + "The persisted subscription id does not match its key."); + } + + if (!HasValidMonitoredItemIdentities(subscriptionId, subscription.MonitoredItems)) + { + throw new ServiceResultException( + StatusCodes.BadDecodingError, + "The persisted subscription contains invalid monitored-item identities."); + } + } + + private static bool HasValidMonitoredItemIdentities( + uint subscriptionId, + IEnumerable monitoredItems) + { + var monitoredItemIds = new HashSet(); + foreach (IStoredMonitoredItem monitoredItem in monitoredItems) + { + if (monitoredItem.SubscriptionId != subscriptionId || + !monitoredItemIds.Add(monitoredItem.Id)) + { + return false; + } + } + return true; + } + private static StoredSubscription CloneSubscription(IStoredSubscription subscription) { return new StoredSubscription @@ -817,6 +1037,16 @@ private ByteString Encode(StoredSubscription subscription) return buffer is null ? ByteString.Empty : ByteString.From(buffer); } + private ByteString EncodeSnapshotManifest(Guid generation, uint recordCount) + { + using var encoder = new BinaryEncoder(m_context); + encoder.WriteInt32(null, DefinitionSnapshotManifestFormatVersion); + encoder.WriteByteString(null, ByteString.From(generation.ToByteArray())); + encoder.WriteUInt32(null, recordCount); + byte[]? buffer = encoder.CloseAndReturnBuffer(); + return buffer is null ? ByteString.Empty : ByteString.From(buffer); + } + private ByteString EncodeRetransmissionState(uint nextSequenceNumber) { using var encoder = new BinaryEncoder(m_context); @@ -923,11 +1153,6 @@ private NotificationMessage DecodeNotificationMessage( }; } - // Decodes a stored subscription record. Retained as the symmetric counterpart to - // Encode and exercised directly by the store's unit tests via reflection, so it is - // not statically referenced from production code paths. -#pragma warning disable IDE0051 // Remove unused private members -#pragma warning disable RCS1213 // Remove unused member declaration private StoredSubscription Decode(ByteString payload) { using var decoder = new BinaryDecoder(payload.ToArray(), m_context); @@ -940,10 +1165,33 @@ private StoredSubscription Decode(ByteString payload) ArrayOf namespaceUris = decoder.ReadStringArray(null); ArrayOf serverUris = decoder.ReadStringArray(null); decoder.SetMappingTables(CreateNamespaceTable(namespaceUris), CreateStringTable(serverUris)); - return DecodeSubscription(decoder); + StoredSubscription subscription = DecodeSubscription(decoder); + if (decoder.Position != payload.Length) + { + throw new ServiceResultException( + StatusCodes.BadDecodingError, + "The persisted subscription record contains trailing data."); + } + return subscription; + } + + private SnapshotManifest DecodeSnapshotManifest(ByteString payload) + { + using var decoder = new BinaryDecoder(payload.ToArray(), m_context); + int version = decoder.ReadInt32(null); + ByteString generationBytes = decoder.ReadByteString(null); + uint recordCount = decoder.ReadUInt32(null); + if (version != DefinitionSnapshotManifestFormatVersion || + generationBytes.Length != 16 || + decoder.Position != payload.Length) + { + throw new ServiceResultException( + StatusCodes.BadDecodingError, + "The persisted subscription snapshot manifest is malformed."); + } + + return new SnapshotManifest(new Guid(generationBytes.ToArray()), recordCount); } -#pragma warning restore RCS1213 // Remove unused member declaration -#pragma warning restore IDE0051 // Remove unused private members private static void EncodeSubscription(BinaryEncoder encoder, StoredSubscription subscription) { @@ -1153,6 +1401,7 @@ private static string ContinuationPointPrefixFor(NodeId ownerSessionId) } private const int DefinitionFormatVersion = 1; + private const int DefinitionSnapshotManifestFormatVersion = 1; private const int ContinuationPointFormatVersion = 1; private const int LegacyRetransmissionStateFormatVersion = 1; private const int RetransmissionStateFormatVersion = 2; @@ -1164,6 +1413,8 @@ private static string ContinuationPointPrefixFor(NodeId ownerSessionId) private const int FilterKindExtensionObject = 4; private const int ChannelCapacity = 1024; private const string Prefix = "subscription/"; + private const string DefinitionSnapshotManifestKey = "subscription-snapshot/manifest"; + private const string DefinitionSnapshotGenerationPrefix = "subscription-snapshot/generation/"; private const string RetransmissionPrefix = "subscription-retransmission/"; private const string ContinuationPointPrefix = "continuation-point/"; private readonly ISharedKeyValueStore m_store; @@ -1194,6 +1445,8 @@ private sealed class SharedDefinitionCache { public Lock Lock { get; } = new(); + public SemaphoreSlim SnapshotCommitLock { get; } = new(1, 1); + public Dictionary Subscriptions { get; } = []; } @@ -1233,6 +1486,8 @@ private readonly record struct ContinuationPointBatch( ContinuationPointEnvelope[] Stores, string[] Deletes); + private readonly record struct SnapshotManifest(Guid Generation, uint RecordCount); + /// /// Command enqueued to the mirror worker; carries an optional completion source used to await the mirror /// operation, or acts as a wake-up signal when none is supplied. diff --git a/tests/Opc.Ua.Redundancy.Server.Tests/Subscriptions/SharedKeyValueSubscriptionStoreTests.cs b/tests/Opc.Ua.Redundancy.Server.Tests/Subscriptions/SharedKeyValueSubscriptionStoreTests.cs index 61217a510d..0dac8b6d68 100644 --- a/tests/Opc.Ua.Redundancy.Server.Tests/Subscriptions/SharedKeyValueSubscriptionStoreTests.cs +++ b/tests/Opc.Ua.Redundancy.Server.Tests/Subscriptions/SharedKeyValueSubscriptionStoreTests.cs @@ -98,7 +98,103 @@ public async Task StoreIsVisibleToSecondReplicaAsync() } [Test] - public async Task ProtectedDefinitionCacheSurvivesTamperedBackendRecordAsync() + 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 })); + } + + [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); + } + + [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); + } + + [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); + } + + [Test] + public async Task ProtectedDefinitionRestoreFailsWhenBackendRecordIsTamperedAsync() { using var kv = new InMemorySharedKeyValueStore(); using var protector = new AesCbcHmacRecordProtector(MakeKey(11)); @@ -106,13 +202,15 @@ public async Task ProtectedDefinitionCacheSurvivesTamperedBackendRecordAsync() StoredSubscription subscription = NewSubscription(300, 30); await store.StoreSubscriptionsAsync([subscription]).ConfigureAwait(false); - await kv.SetAsync( - SharedKeyValueSubscriptionStore.KeyFor(subscription.Id), - ByteString.From(new byte[] { 1, 2, 3, 4, 5 })).ConfigureAwait(false); - RestoreSubscriptionResult result = await store.RestoreSubscriptionsAsync().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); - Assert.That(result.Success, Is.True); - Assert.That(result.Subscriptions, Is.Not.Empty); + ServiceResultException? exception = Assert.ThrowsAsync( + () => store.RestoreSubscriptionsAsync().AsTask()); + + Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadSecurityChecksFailed)); } [Test] @@ -124,13 +222,86 @@ public async Task ProtectedStoreDoesNotPersistDefinitionInClearTextAsync() StoredSubscription subscription = NewSubscription(400, 40); await store.StoreSubscriptionsAsync([subscription]).ConfigureAwait(false); - (bool found, ByteString raw) = await kv.TryGetAsync(SharedKeyValueSubscriptionStore.KeyFor(subscription.Id)).ConfigureAwait(false); + KeyValuePair record = await GetSingleGenerationRecordAsync(kv).ConfigureAwait(false); - Assert.That(found, Is.True); - Assert.That(Contains(raw.ToArray(), BitConverter.GetBytes(subscription.Id)), Is.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)); } + [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 })); + } + + [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)); + } + + [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); + } + [Test] public async Task StoreReplacesRemovedSubscriptionsAsync() { @@ -160,6 +331,277 @@ await store.OnSubscriptionRestoreCompleteAsync(new Dictionary s.Id), Is.EqualTo(new uint[] { 601 })); } + [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); + } + + [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()); + } + + [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()); + } + + [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 })); + } + + [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); + } + + [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(new byte[] { 9, 9, 9 })) + .ConfigureAwait(false); + + ServiceResultException? exception = Assert.ThrowsAsync( + () => store.RestoreSubscriptionsAsync().AsTask()); + + Assert.That(exception!.StatusCode, Is.EqualTo(StatusCodes.BadSecurityChecksFailed)); + } + + [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")); + } + + [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")); + } + + [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")); + } + + [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")); + } + + [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")); + } + + [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")); + } + + [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")); + } + + [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")); + } + [Test] public async Task RetransmissionStateIsVisibleToSecondReplicaAndKeepsSequenceContinuityAsync() { @@ -742,6 +1184,73 @@ private static MethodInfo GetPrivateMethod(string name, params Type[] parameterT return method!; } + private static ByteString EncodeDefinition( + SharedKeyValueSubscriptionStore store, + StoredSubscription subscription) + { + MethodInfo encode = GetPrivateMethod("Encode", typeof(StoredSubscription)); + return (ByteString)encode.Invoke(store, [subscription])!; + } + + 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); + PropertyInfo? subscriptionsProperty = cache!.GetType().GetProperty( + "Subscriptions", + BindingFlags.Instance | BindingFlags.Public); + Assert.That(subscriptionsProperty, Is.Not.Null); + var subscriptions = + (Dictionary)subscriptionsProperty!.GetValue(cache)!; + uint[] ids = [.. subscriptions.Keys]; + Array.Sort(ids); + return ids; + } + private static NotificationMessage NewNotification(uint sequenceNumber) { var notification = new DataChangeNotification @@ -840,6 +1349,86 @@ private static byte[] MakeKey(byte seed) 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) @@ -1115,5 +1704,199 @@ public IAsyncEnumerable WatchAsync( 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; + } } }