diff --git a/src/AWS/Orleans.Journaling.S3/README.md b/src/AWS/Orleans.Journaling.S3/README.md index 817a2a130dd..169f4586174 100644 --- a/src/AWS/Orleans.Journaling.S3/README.md +++ b/src/AWS/Orleans.Journaling.S3/README.md @@ -7,3 +7,13 @@ The provider uses S3 Express append writes (`WriteOffsetBytes`) for WAL appends. Buckets should be created ahead of time for AWS S3 Express One Zone. `CreateBucketIfNotExists` is intended for local emulators. Metadata updates rewrite the current WAL using a conditional single-object upload. Publish a checkpoint to compact the WAL before updating metadata when the replacement object would exceed S3's 5 GB (5,000,000,000 byte) single-upload limit. Checkpoint snapshots use the same upload limit. + +## Catalog enumeration + +`IJournalStorageCatalog.ListAsync` returns journal identities incrementally in S3 traversal order, including unordered directory-bucket listings. Set `ListOptions.Prefix` to select an exact journal id and its descendants. + +The provider handles `ListObjectsV2` continuations internally, requests up to 1000 objects per page, and yields canonical WAL identities from that page before fetching more objects. The bucket traversal supports `GetObjectKey` and `TryParseJournalId` mappings; checkpoints, aliases, and unrelated objects consume space in the native page before filtering. + +Client traversal memory is proportional to the current native page. An enumerator advance can cross multiple filtered or empty pages, and the storage service determines scan work, latency, and retries. Enumeration observes the live bucket; concurrent changes follow S3 listing semantics. Use subsequent enumerations to discover later changes and tolerate repeated identities during changes. + +Dispose the enumerator when stopping early and use a cancellation token covering its lifetime. Cancellation and service failures propagate through enumeration. diff --git a/src/AWS/Orleans.Journaling.S3/S3JournalStorageProvider.cs b/src/AWS/Orleans.Journaling.S3/S3JournalStorageProvider.cs index 48fa0321aa9..a5b8d876c91 100644 --- a/src/AWS/Orleans.Journaling.S3/S3JournalStorageProvider.cs +++ b/src/AWS/Orleans.Journaling.S3/S3JournalStorageProvider.cs @@ -46,12 +46,13 @@ public IJournalStorage CreateStorage(JournalId journalId) } public async IAsyncEnumerable ListAsync( - JournalId prefix = default, + ListOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { + cancellationToken.ThrowIfCancellationRequested(); + var prefix = options?.Prefix ?? default; var client = GetClient(); var bucketName = GetBucketName(); - var journalIds = new HashSet(); string? continuationToken = null; do @@ -60,30 +61,18 @@ public async IAsyncEnumerable ListAsync( new ListObjectsV2Request { BucketName = bucketName, + MaxKeys = 1000, ContinuationToken = continuationToken, }, cancellationToken).ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); foreach (var item in response.S3Objects) { cancellationToken.ThrowIfCancellationRequested(); - if (!item.Key.EndsWith("/wal", StringComparison.Ordinal)) - { - continue; - } - - var storageIdValue = item.Key[..^"/wal".Length]; - var journalId = _options.TryParseJournalId(storageIdValue); - if (journalId is not { IsDefault: false } id || !prefix.IsPrefixOf(id)) - { - continue; - } - - var journalObjectKey = _options.GetObjectKeyForJournal(id); - var canonicalWalObjectKey = S3JournalStorageOptions.GetWalObjectKeyForJournal(id, journalObjectKey); - if (string.Equals(item.Key, canonicalWalObjectKey, StringComparison.Ordinal)) + if (TryGetJournalId(item.Key, prefix, out var id)) { - journalIds.Add(id); + yield return id; } } @@ -91,11 +80,26 @@ public async IAsyncEnumerable ListAsync( } while (continuationToken is not null); - foreach (var journalId in journalIds.OrderBy(static journalId => journalId.Value, StringComparer.Ordinal)) + cancellationToken.ThrowIfCancellationRequested(); + } + + private bool TryGetJournalId(string objectKey, JournalId prefix, out JournalId journalId) + { + if (objectKey.EndsWith("/wal", StringComparison.Ordinal) + && _options.TryParseJournalId(objectKey[..^"/wal".Length]) is { IsDefault: false } id + && prefix.IsPrefixOf(id)) { - cancellationToken.ThrowIfCancellationRequested(); - yield return journalId; + var journalObjectKey = _options.GetObjectKeyForJournal(id); + var canonicalWalObjectKey = S3JournalStorageOptions.GetWalObjectKeyForJournal(id, journalObjectKey); + if (string.Equals(objectKey, canonicalWalObjectKey, StringComparison.Ordinal)) + { + journalId = id; + return true; + } } + + journalId = default; + return false; } public void Participate(ISiloLifecycle observer) diff --git a/src/Azure/Orleans.Journaling.AzureStorage/AzureBlobJournalStorageProvider.cs b/src/Azure/Orleans.Journaling.AzureStorage/AzureBlobJournalStorageProvider.cs index 009df4bf7b4..f7492ecf136 100644 --- a/src/Azure/Orleans.Journaling.AzureStorage/AzureBlobJournalStorageProvider.cs +++ b/src/Azure/Orleans.Journaling.AzureStorage/AzureBlobJournalStorageProvider.cs @@ -40,9 +40,10 @@ public AzureBlobJournalStorageProvider( private async Task Initialize(CancellationToken cancellationToken) { var client = await _options.CreateClient!(cancellationToken); - _defaultContainer = client.GetBlobContainerClient(_options.ContainerName); - await _defaultContainer.CreateIfNotExistsAsync(cancellationToken: cancellationToken).ConfigureAwait(false); + var container = client.GetBlobContainerClient(_options.ContainerName); + await container.CreateIfNotExistsAsync(cancellationToken: cancellationToken).ConfigureAwait(false); await _containerFactory.InitializeAsync(client, cancellationToken).ConfigureAwait(false); + _defaultContainer = container; } public IJournalStorage CreateStorage(JournalId journalId) @@ -56,40 +57,37 @@ public IJournalStorage CreateStorage(JournalId journalId) } public async IAsyncEnumerable ListAsync( - JournalId prefix = default, + ListOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { + cancellationToken.ThrowIfCancellationRequested(); + var prefix = options?.Prefix ?? default; var container = GetDefaultContainerClient(); - var blobPrefix = prefix.IsDefault ? null : prefix.Value; - var journalIds = new List(); - await foreach (var item in container.GetBlobsAsync( + + await foreach (var page in container.GetBlobsAsync( traits: BlobTraits.None, states: BlobStates.None, - prefix: blobPrefix, - cancellationToken: cancellationToken)) + prefix: prefix.IsDefault ? null : prefix.Value, + cancellationToken: cancellationToken).AsPages(pageSizeHint: 5000)) { - if (item.Properties.BlobType is { } blobType && blobType != BlobType.Append) - { - continue; - } - - if (!item.Name.EndsWith("/wal", StringComparison.Ordinal)) - { - continue; - } - - var storageIdValue = item.Name[..^"/wal".Length]; - if (TryParseJournalId(storageIdValue, out var journalId) && prefix.IsPrefixOf(journalId)) + cancellationToken.ThrowIfCancellationRequested(); + foreach (var item in page.Values) { - journalIds.Add(journalId); + cancellationToken.ThrowIfCancellationRequested(); + if (item.Properties.BlobType is { } blobType && blobType != BlobType.Append + || !item.Name.EndsWith("/wal", StringComparison.Ordinal)) + { + continue; + } + + if (TryParseJournalId(item.Name[..^"/wal".Length], out var journalId) && prefix.IsPrefixOf(journalId)) + { + yield return journalId; + } } } - foreach (var journalId in journalIds.OrderBy(static journalId => journalId.Value, StringComparer.Ordinal)) - { - cancellationToken.ThrowIfCancellationRequested(); - yield return journalId; - } + cancellationToken.ThrowIfCancellationRequested(); } public void Participate(ISiloLifecycle observer) diff --git a/src/Azure/Orleans.Journaling.AzureStorage/AzureTableJournalStorageProvider.cs b/src/Azure/Orleans.Journaling.AzureStorage/AzureTableJournalStorageProvider.cs index 20cd1e8867e..2263ced4712 100644 --- a/src/Azure/Orleans.Journaling.AzureStorage/AzureTableJournalStorageProvider.cs +++ b/src/Azure/Orleans.Journaling.AzureStorage/AzureTableJournalStorageProvider.cs @@ -57,25 +57,31 @@ public IJournalStorage CreateStorage(JournalId journalId) } public async IAsyncEnumerable ListAsync( - JournalId prefix = default, + ListOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { + cancellationToken.ThrowIfCancellationRequested(); + var prefix = options?.Prefix ?? default; var table = _tableClientProvider.GetTableClient(); var filter = TableClient.CreateQueryFilter($"RowKey eq {AzureTableJournalStorage.HeaderRowKey}"); - var journalIds = new List(); - await foreach (var entity in table.QueryAsync(filter, select: JournalIdSelect, cancellationToken: cancellationToken)) + await foreach (var page in table.QueryAsync( + filter, + maxPerPage: 1000, + select: JournalIdSelect, + cancellationToken: cancellationToken).AsPages(pageSizeHint: 1000)) { - if (TryGetJournalId(entity, out var journalId) && prefix.IsPrefixOf(journalId)) + cancellationToken.ThrowIfCancellationRequested(); + foreach (var entity in page.Values) { - journalIds.Add(journalId); + cancellationToken.ThrowIfCancellationRequested(); + if (TryGetJournalId(entity, out var journalId) && prefix.IsPrefixOf(journalId)) + { + yield return journalId; + } } } - foreach (var journalId in journalIds.OrderBy(static journalId => journalId.Value, StringComparer.Ordinal)) - { - cancellationToken.ThrowIfCancellationRequested(); - yield return journalId; - } + cancellationToken.ThrowIfCancellationRequested(); } private static bool TryGetJournalId(TableEntity entity, out JournalId journalId) diff --git a/src/Azure/Orleans.Journaling.AzureStorage/README.md b/src/Azure/Orleans.Journaling.AzureStorage/README.md index af3ffb7a085..26336d624b9 100644 --- a/src/Azure/Orleans.Journaling.AzureStorage/README.md +++ b/src/Azure/Orleans.Journaling.AzureStorage/README.md @@ -31,6 +31,16 @@ siloBuilder.AddAzureTableJournalStorage(options => `AzureTableJournalStorageOptions.GetPartitionKey` can be used to apply a custom partition layout. The returned key must be unique per journal, satisfy Azure Table partition-key restrictions, and contain at most 1,024 characters. The canonical journal id is stored in the header so catalog listing remains accurate with custom mappings. +## Catalog enumeration + +Both Azure providers implement `IJournalStorageCatalog.ListAsync`, returning identities incrementally in service traversal order. Set `ListOptions.Prefix` to enumerate an exact journal id and its descendants. The provider handles service continuations internally and yields identities from the current page before fetching the next page. + +The Blob catalog scans the configured `ContainerName` and interprets append blobs named `/wal` as journal identities. This traversal applies equally when a custom naming delegate or container factory produces the same entries. Each internal page requests up to 5000 blobs, including checkpoints and other entries before filtering. + +Table enumeration supports custom partition mappings through the canonical journal id stored in each header. It requests up to 1000 header rows per service page; Table Storage determines the internal scan work required by that query. Prefix filtering occurs on the returned headers, so one enumerator advance can cross multiple empty or filtered pages. + +Use `await foreach` or dispose a retained enumerator when stopping early. Pass a cancellation token covering the traversal lifetime; cancellation and service errors propagate to the caller. + ## Getting Started To use this package, install it via NuGet: diff --git a/src/Orleans.DurableJobs/JournaledJobShardManager.cs b/src/Orleans.DurableJobs/JournaledJobShardManager.cs index 6dddb9c3f32..74d7952c43b 100644 --- a/src/Orleans.DurableJobs/JournaledJobShardManager.cs +++ b/src/Orleans.DurableJobs/JournaledJobShardManager.cs @@ -85,7 +85,7 @@ public override async Task> AssignJobShardsAsync(DateTimeOffset var newClaimCount = 0; var membershipSnapshot = _membershipService.CurrentSnapshot; - await foreach (var storageId in _catalog.ListAsync(JobShardId.StoragePrefix, cancellationToken)) + await foreach (var storageId in _catalog.ListAsync(new() { Prefix = JobShardId.StoragePrefix }, cancellationToken)) { var descriptor = await GetDescriptorAsync(storageId, cancellationToken); if (descriptor is null || descriptor.Poisoned || descriptor.StartTime > maxDueTime) diff --git a/src/Orleans.Journaling/CompatibilitySuppressions.xml b/src/Orleans.Journaling/CompatibilitySuppressions.xml new file mode 100644 index 00000000000..cf125a42760 --- /dev/null +++ b/src/Orleans.Journaling/CompatibilitySuppressions.xml @@ -0,0 +1,46 @@ + + + + + CP0002 + M:Orleans.Journaling.IJournalStorageCatalog.ListAsync(Orleans.Journaling.JournalId,System.Threading.CancellationToken) + lib/net10.0/Orleans.Journaling.dll + lib/net10.0/Orleans.Journaling.dll + true + + + CP0002 + M:Orleans.Journaling.VolatileJournalStorageProvider.ListAsync(Orleans.Journaling.JournalId,System.Threading.CancellationToken) + lib/net10.0/Orleans.Journaling.dll + lib/net10.0/Orleans.Journaling.dll + true + + + CP0002 + M:Orleans.Journaling.IJournalStorageCatalog.ListAsync(Orleans.Journaling.JournalId,System.Threading.CancellationToken) + lib/net8.0/Orleans.Journaling.dll + lib/net8.0/Orleans.Journaling.dll + true + + + CP0002 + M:Orleans.Journaling.VolatileJournalStorageProvider.ListAsync(Orleans.Journaling.JournalId,System.Threading.CancellationToken) + lib/net8.0/Orleans.Journaling.dll + lib/net8.0/Orleans.Journaling.dll + true + + + CP0006 + M:Orleans.Journaling.IJournalStorageCatalog.ListAsync(Orleans.Journaling.ListOptions,System.Threading.CancellationToken) + lib/net10.0/Orleans.Journaling.dll + lib/net10.0/Orleans.Journaling.dll + true + + + CP0006 + M:Orleans.Journaling.IJournalStorageCatalog.ListAsync(Orleans.Journaling.ListOptions,System.Threading.CancellationToken) + lib/net8.0/Orleans.Journaling.dll + lib/net8.0/Orleans.Journaling.dll + true + + \ No newline at end of file diff --git a/src/Orleans.Journaling/IJournalStorageCatalog.cs b/src/Orleans.Journaling/IJournalStorageCatalog.cs index 5c875b0069e..7a59456ad7c 100644 --- a/src/Orleans.Journaling/IJournalStorageCatalog.cs +++ b/src/Orleans.Journaling/IJournalStorageCatalog.cs @@ -4,16 +4,28 @@ namespace Orleans.Journaling; /// Provides catalog operations for journal storage instances. /// /// -/// A catalog only discovers storage identities. Storage lifecycle, metadata, and data mutation -/// operations remain on . +/// A catalog discovers storage identities. provides storage lifecycle, +/// metadata, and data mutation operations. /// public interface IJournalStorageCatalog { /// - /// Lists journal ids which match . + /// Enumerates journal ids matching the supplied options. /// - /// The journal id prefix, or the default value to list all ids. + /// The listing options, or to list all ids. /// The cancellation token. - /// Matching ids in lexicographic order. - IAsyncEnumerable ListAsync(JournalId prefix = default, CancellationToken cancellationToken = default); + /// Matching ids in provider traversal order. + /// + /// Options are read when enumeration begins. Providers fetch storage pages internally and yield matching ids + /// as they are discovered. Advancing the enumerator can traverse multiple empty or filtered storage pages. + /// Storage services determine request latency, retries, and internal scan work. + /// Enumeration observes live storage; concurrent changes follow the provider's listing semantics. + /// Callers should tolerate repeated identities during concurrent changes and start a new enumeration to + /// discover later changes. Dispose the enumerator when stopping early. + /// Storage and cancellation errors propagate through enumeration. Start a new enumeration after a listing error. + /// + /// is canceled. + IAsyncEnumerable ListAsync( + ListOptions? options = null, + CancellationToken cancellationToken = default); } diff --git a/src/Orleans.Journaling/ListOptions.cs b/src/Orleans.Journaling/ListOptions.cs new file mode 100644 index 00000000000..a6785ddc995 --- /dev/null +++ b/src/Orleans.Journaling/ListOptions.cs @@ -0,0 +1,15 @@ +namespace Orleans.Journaling; + +/// +/// Options for enumerating journal storage identities. +/// +public sealed class ListOptions +{ + /// + /// Gets or sets the journal id prefix. The default value matches all ids. + /// + /// + /// A prefix matches the exact journal id and its descendant segments. + /// + public JournalId Prefix { get; set; } +} diff --git a/src/Orleans.Journaling/README.md b/src/Orleans.Journaling/README.md index c02bb4328ba..56320decafc 100644 --- a/src/Orleans.Journaling/README.md +++ b/src/Orleans.Journaling/README.md @@ -108,6 +108,26 @@ Inside the operation payload array, element 0 is the command name, followed by c Existing data is read using its stored format metadata, or as legacy OrleansBinary data when metadata is absent, and migrated to the configured write format by the next snapshot write. +## Catalog enumeration + +`IJournalStorageCatalog.ListAsync` returns an `IAsyncEnumerable` in provider traversal order. Pass `ListOptions` with `Prefix` to select an exact journal id and its descendants, or omit the options to enumerate all ids. Options are read when enumeration begins. + +When updating callers of the former prefix overload, pass `new ListOptions { Prefix = prefix }`. Applications which require ordinal ordering can materialize the sequence and sort `JournalId.Value` using `StringComparer.Ordinal`. + +Storage providers fetch pages internally and yield matching identities as they discover them. `await foreach` advances the traversal and disposes the enumerator when the loop ends. Consumers which process identities in batches can retain one enumerator across batches, advance it serially, and dispose it after the last pending `MoveNextAsync` completes. Use a cancellation token whose lifetime covers that enumeration. + +| Provider | Internal traversal | Client memory | +| --- | --- | --- | +| Volatile | Enumerates the existing concurrent storage dictionary, checks journal existence, and applies the prefix | Constant additional traversal state; the storage dictionary holds the journals. | +| Azure Blob | Requests up to 5000 blobs per service page, then applies WAL and hierarchical prefix filtering | Proportional to the current service page. | +| Azure Table | Requests up to 1000 journal headers per service page, then decodes canonical or reversible legacy ids and applies the prefix | Proportional to the current service page. | +| S3 | Requests up to 1000 bucket objects per `ListObjectsV2` page, then applies canonical WAL and prefix filtering | Proportional to the current service page; includes unordered S3 Express listings. | +| Redis | Scans primary-server metadata keys and reads canonical ids in bounded batches, suppressing repeated ids | Read-batch state plus a seen-id set which grows with the catalog. `SCAN` count remains a service work hint. | + +One `MoveNextAsync` can traverse multiple empty or filtered storage pages before yielding an identity. A consumer's identity or metadata-read budget therefore bounds returned candidates, while storage services determine internal scan work, request latency, and retries. Table header queries can inspect additional rows internally, and S3 custom mappings require a bucket traversal. + +Enumeration observes live storage. Concurrent changes follow each provider's listing semantics; callers should tolerate repeated identities during changes and use subsequent enumerations to discover later updates. Journal existence can change between discovery and a storage operation. Cancellation and storage errors propagate through enumeration. Dispose a failed enumerator and begin a new enumeration when retrying a listing operation. + ## Documentation For more comprehensive documentation, please refer to: - [Microsoft Orleans Documentation](https://dotnet.github.io/orleans/docs/) diff --git a/src/Orleans.Journaling/VolatileJournalStorage.cs b/src/Orleans.Journaling/VolatileJournalStorage.cs index db941b7c6a3..ce337459333 100644 --- a/src/Orleans.Journaling/VolatileJournalStorage.cs +++ b/src/Orleans.Journaling/VolatileJournalStorage.cs @@ -47,14 +47,16 @@ public IJournalStorage CreateStorage(JournalId journalId) /// public async IAsyncEnumerable ListAsync( - JournalId prefix = default, + ListOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { - List journalIds = []; + cancellationToken.ThrowIfCancellationRequested(); + var prefix = options?.Prefix ?? default; foreach (var (key, store) in _storage) { cancellationToken.ThrowIfCancellationRequested(); - if (!TryParseJournalId(key, out var journalId) || !prefix.IsPrefixOf(journalId)) + var journalId = new JournalId(key); + if (!prefix.IsPrefixOf(journalId)) { continue; } @@ -67,36 +69,16 @@ public async IAsyncEnumerable ListAsync( } } - journalIds.Add(journalId); - } - - journalIds.Sort(static (left, right) => StringComparer.Ordinal.Compare(left.Value, right.Value)); - - foreach (var journalId in journalIds) - { cancellationToken.ThrowIfCancellationRequested(); yield return journalId; } + cancellationToken.ThrowIfCancellationRequested(); await Task.CompletedTask.ConfigureAwait(false); } private string GetJournalFormatKey() => JournalFormatServices.ValidateJournalFormatKey(_options?.Value.JournalFormatKey ?? JsonJournalExtensions.JournalFormatKey); - - private static bool TryParseJournalId(string value, out JournalId journalId) - { - try - { - journalId = new JournalId(value); - return true; - } - catch (ArgumentException) - { - journalId = default; - return false; - } - } } /// diff --git a/src/Redis/Orleans.Journaling.Redis/README.md b/src/Redis/Orleans.Journaling.Redis/README.md index 6f4b15c9cc7..4596d986c65 100644 --- a/src/Redis/Orleans.Journaling.Redis/README.md +++ b/src/Redis/Orleans.Journaling.Redis/README.md @@ -33,6 +33,25 @@ builder.UseOrleans(siloBuilder => If the Redis connection is already registered in dependency injection, configure it using a keyed service and the `GrainJournaling` provider configuration `ServiceKey`. +## Journal discovery + +Use `IJournalStorageCatalog.ListAsync` with optional `ListOptions` to filter by a journal id prefix: + +```csharp +await foreach (var journalId in catalog.ListAsync( + new ListOptions { Prefix = JournalId.Create("jobs") }, + cancellationToken)) +{ + // Process the discovered journal. +} +``` + +The prefix is read when enumeration begins. Discovery scans primary servers incrementally and reads canonical journal ids in batches of at most 128 metadata keys, yielding matching ids in traversal order without sorting or buffering the whole catalog. Duplicate ids are suppressed across the traversal using O(N) seen-id memory for N distinct matching ids; metadata keys are not retained for the whole traversal, so repeated scan results can cause repeated metadata reads. + +The Redis `SCAN COUNT` value of 250 is a hint, not a strict response-size or server-work bound. One enumerator advancement can traverse many empty or nonmatching scans, and Redis/client-side scan buffering is not bounded by the metadata batch size. Discovery observes live storage rather than a snapshot. + +Storage errors and cancellation propagate without provider-level retries, restarts, or success fallbacks. A disconnected primary or the absence of any primary is an error; failures on later servers can occur after earlier ids have been yielded. Dispose the enumerator when stopping early. Cancellation is checked between scan and metadata operations, but does not abort an in-flight Redis metadata request. + ## Documentation - [Microsoft Orleans Documentation](https://dotnet.github.io/orleans/docs/) diff --git a/src/Redis/Orleans.Journaling.Redis/RedisJournalStorageProvider.cs b/src/Redis/Orleans.Journaling.Redis/RedisJournalStorageProvider.cs index ec873b1995b..ba5eeefe466 100644 --- a/src/Redis/Orleans.Journaling.Redis/RedisJournalStorageProvider.cs +++ b/src/Redis/Orleans.Journaling.Redis/RedisJournalStorageProvider.cs @@ -57,12 +57,15 @@ public IJournalStorage CreateStorage(JournalId journalId) } public async IAsyncEnumerable ListAsync( - JournalId prefix = default, + ListOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { + var prefix = options?.Prefix ?? default; + cancellationToken.ThrowIfCancellationRequested(); var connection = GetConnection(); var database = GetDatabase(); - var metadataKeys = new HashSet(); + var journalIds = new HashSet(); + var batch = new RedisKey[JournalIdReadBatchSize]; var pattern = RedisJournalStorage.GetMetadataKeyPattern(_keyPrefix); var scannedServer = false; foreach (var endpoint in connection.GetEndPoints()) @@ -81,78 +84,92 @@ public async IAsyncEnumerable ListAsync( } scannedServer = true; - await foreach (var metadataKey in server.KeysAsync( + await using var metadataKeys = server.KeysAsync( database.Database, pattern, - pageSize: ScanPageSize).WithCancellation(cancellationToken)) + pageSize: ScanPageSize).GetAsyncEnumerator(cancellationToken); + while (true) { - metadataKeys.Add(metadataKey); - } - } - - if (!scannedServer) - { - throw new InvalidOperationException("No connected primary Redis servers are available for journal discovery."); - } + var count = 0; + while (count < batch.Length) + { + cancellationToken.ThrowIfCancellationRequested(); + if (!await metadataKeys.MoveNextAsync().ConfigureAwait(false)) + { + break; + } - var journalIds = new HashSet(); - foreach (var batch in metadataKeys.Chunk(JournalIdReadBatchSize)) - { - cancellationToken.ThrowIfCancellationRequested(); - var reads = new Task[batch.Length]; - for (var i = 0; i < batch.Length; i++) - { - reads[i] = database.HashGetAsync(batch[i], RedisJournalStorage.JournalIdMetadataKey); - } + batch[count++] = metadataKeys.Current; + } - var values = await Task.WhenAll(reads).ConfigureAwait(false); - for (var i = 0; i < values.Length; i++) - { cancellationToken.ThrowIfCancellationRequested(); - var value = values[i]; - if (value.IsNullOrEmpty) + if (count == 0) { - var result = (RedisResult[]?)await database.ScriptEvaluateAsync( - ReadJournalIdScript, - [batch[i]], - NoValues).ConfigureAwait(false); - if (result is not { Length: > 0 }) - { - throw new InvalidOperationException("The Redis journal discovery script returned an invalid response."); - } + break; + } - var status = (int)result[0]; - if (status == 0) + var reads = new Task[count]; + for (var i = 0; i < count; i++) + { + reads[i] = database.HashGetAsync(batch[i], RedisJournalStorage.JournalIdMetadataKey); + } + + var values = await Task.WhenAll(reads).ConfigureAwait(false); + for (var i = 0; i < values.Length; i++) + { + cancellationToken.ThrowIfCancellationRequested(); + var value = values[i]; + if (value.IsNullOrEmpty) { - continue; + var result = (RedisResult[]?)await database.ScriptEvaluateAsync( + ReadJournalIdScript, + [batch[i]], + NoValues).ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); + if (result is not { Length: > 0 }) + { + throw new InvalidOperationException("The Redis journal discovery script returned an invalid response."); + } + + var status = (int)result[0]; + if (status == 0) + { + continue; + } + + if (status != 1 || result.Length != 2) + { + throw new InvalidOperationException( + $"Redis journal metadata '{batch[i]}' is missing '{RedisJournalStorage.JournalIdMetadataKey}'."); + } + + value = (RedisValue)result[1]; } - if (status != 1 || result.Length != 2) + if (!RedisJournalStorage.TryParseJournalId(value.ToString(), out var journalId)) { throw new InvalidOperationException( - $"Redis journal metadata '{batch[i]}' is missing '{RedisJournalStorage.JournalIdMetadataKey}'."); + $"Redis journal metadata '{batch[i]}' contains an invalid '{RedisJournalStorage.JournalIdMetadataKey}' value."); } - value = (RedisValue)result[1]; - } - - if (!RedisJournalStorage.TryParseJournalId(value.ToString(), out var journalId)) - { - throw new InvalidOperationException( - $"Redis journal metadata '{batch[i]}' contains an invalid '{RedisJournalStorage.JournalIdMetadataKey}' value."); + if (prefix.IsPrefixOf(journalId) && journalIds.Add(journalId)) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return journalId; + } } - if (prefix.IsPrefixOf(journalId)) + if (count < batch.Length) { - journalIds.Add(journalId); + break; } } } - foreach (var journalId in journalIds.OrderBy(static journalId => journalId.Value, StringComparer.Ordinal)) + cancellationToken.ThrowIfCancellationRequested(); + if (!scannedServer) { - cancellationToken.ThrowIfCancellationRequested(); - yield return journalId; + throw new InvalidOperationException("No connected primary Redis servers are available for journal discovery."); } } diff --git a/src/api/Orleans.Journaling/Orleans.Journaling.cs b/src/api/Orleans.Journaling/Orleans.Journaling.cs index e7fcd7aa32d..0b3262c7bff 100644 --- a/src/api/Orleans.Journaling/Orleans.Journaling.cs +++ b/src/api/Orleans.Journaling/Orleans.Journaling.cs @@ -271,7 +271,7 @@ public partial interface IJournalStorage public partial interface IJournalStorageCatalog { - System.Collections.Generic.IAsyncEnumerable ListAsync(JournalId prefix = default, System.Threading.CancellationToken cancellationToken = default); + System.Collections.Generic.IAsyncEnumerable ListAsync(ListOptions? options = null, System.Threading.CancellationToken cancellationToken = default); } public partial interface IJournalStorageConsumer @@ -509,6 +509,11 @@ public readonly partial struct JournalStreamWriter public readonly JournalEntryScope BeginEntry() { throw null; } } + public sealed partial class ListOptions + { + public JournalId Prefix { get { throw null; } set { } } + } + public sealed partial class VolatileJournalStorage : IJournalStorage { public VolatileJournalStorage() { } @@ -540,7 +545,7 @@ public VolatileJournalStorageProvider(Microsoft.Extensions.Options.IOptions ListAsync(JournalId prefix = default, System.Threading.CancellationToken cancellationToken = default) { throw null; } + public System.Collections.Generic.IAsyncEnumerable ListAsync(ListOptions? options = null, System.Threading.CancellationToken cancellationToken = default) { throw null; } } } diff --git a/test/Extensions/Orleans.Redis.Tests/Journaling/RedisJournalStorageCatalogTests.cs b/test/Extensions/Orleans.Redis.Tests/Journaling/RedisJournalStorageCatalogTests.cs new file mode 100644 index 00000000000..4bf0f0fce15 --- /dev/null +++ b/test/Extensions/Orleans.Redis.Tests/Journaling/RedisJournalStorageCatalogTests.cs @@ -0,0 +1,317 @@ +using System.Net; +using System.Runtime.CompilerServices; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using Orleans.Configuration; +using Orleans.Journaling; +using Orleans.Runtime; +using StackExchange.Redis; +using TestExtensions; +using Xunit; + +namespace Tester.Redis.Journaling; + +[TestSuite("BVT")] +[TestProvider("Redis")] +[TestArea("Journaling")] +[TestCategory("BVT")] +public sealed class RedisJournalStorageCatalogTests +{ + private const string KeyPrefix = "catalog-tests"; + + [Fact] + public async Task ListAsync_YieldsFirstBatchBeforeScanCompletesAndSnapshotsPrefix() + { + var firstId = JournalId.Create("redis", "list", "z"); + var secondId = JournalId.Create("redis", "list", "a"); + var nonmatchingId = JournalId.Create("redis", "listing", "other"); + var firstBatch = Enumerable.Range(0, 128) + .Select(index => RedisJournalStorage.GetMetadataKey(KeyPrefix, $"mapped-{index}")) + .ToArray(); + var secondKey = RedisJournalStorage.GetMetadataKey(KeyPrefix, "mapped-second"); + var ids = firstBatch.ToDictionary(key => key, _ => firstId); + ids[firstBatch[^1]] = nonmatchingId; + ids[secondKey] = secondId; + var scannedKeys = 0; + var metadataReads = 0; + var scanCompleted = false; + var scanDisposed = false; + var database = Substitute.For(); + database.HashGetAsync(Arg.Any(), RedisJournalStorage.JournalIdMetadataKey) + .Returns(call => + { + metadataReads++; + return Task.FromResult((RedisValue)ids[call.ArgAt(0)].Value); + }); + var provider = await CreateProviderAsync(database, CreateServer(ScanAsync(TestContext.Current.CancellationToken))); + var options = new ListOptions { Prefix = JournalId.Create("redis", "list") }; + await using var enumerator = provider.ListAsync(options, TestContext.Current.CancellationToken).GetAsyncEnumerator(TestContext.Current.CancellationToken); + + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal(firstId, enumerator.Current); + Assert.Equal(128, scannedKeys); + Assert.Equal(128, metadataReads); + Assert.False(scanCompleted); + Assert.False(scanDisposed); + + options.Prefix = JournalId.Create("redis", "listing"); + + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal(secondId, enumerator.Current); + Assert.Equal(130, scannedKeys); + Assert.Equal(130, metadataReads); + Assert.True(scanCompleted); + Assert.False(await enumerator.MoveNextAsync()); + Assert.True(scanDisposed); + + async IAsyncEnumerable ScanAsync([EnumeratorCancellation] CancellationToken cancellationToken = default) + { + try + { + await Task.CompletedTask; + foreach (var key in firstBatch.Concat([firstBatch[0], secondKey])) + { + cancellationToken.ThrowIfCancellationRequested(); + scannedKeys++; + yield return key; + } + + scanCompleted = true; + } + finally + { + scanDisposed = true; + } + } + } + + [Fact] + public async Task ListAsync_LaterScanFailurePropagatesAfterFirstBatch() + { + var id = JournalId.Create("redis", "failure"); + var key = RedisJournalStorage.GetMetadataKey(KeyPrefix, "mapped"); + var failure = new RedisConnectionException(ConnectionFailureType.UnableToConnect, "Later scan failed."); + var scanDisposed = false; + var scannedKeys = 0; + var database = Substitute.For(); + database.HashGetAsync(key, RedisJournalStorage.JournalIdMetadataKey) + .Returns(Task.FromResult((RedisValue)id.Value)); + var provider = await CreateProviderAsync(database, CreateServer(ScanAsync())); + await using var enumerator = provider.ListAsync(cancellationToken: TestContext.Current.CancellationToken).GetAsyncEnumerator(TestContext.Current.CancellationToken); + + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal(id, enumerator.Current); + Assert.Equal(128, scannedKeys); + Assert.False(scanDisposed); + + var exception = await Assert.ThrowsAsync(() => enumerator.MoveNextAsync().AsTask()); + + Assert.Same(failure, exception); + Assert.Equal(128, scannedKeys); + Assert.True(scanDisposed); + + async IAsyncEnumerable ScanAsync() + { + try + { + await Task.CompletedTask; + for (var i = 0; i < 128; i++) + { + scannedKeys++; + yield return key; + } + + throw failure; + } + finally + { + scanDisposed = true; + } + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ListAsync_CancellationOrEarlyDisposalStopsScan(bool cancel) + { + var id = JournalId.Create("redis", "cancel"); + var key = RedisJournalStorage.GetMetadataKey(KeyPrefix, "mapped"); + var scannedKeys = 0; + var metadataReads = 0; + var scanDisposed = false; + var database = Substitute.For(); + database.HashGetAsync(key, RedisJournalStorage.JournalIdMetadataKey) + .Returns(_ => + { + metadataReads++; + return Task.FromResult((RedisValue)id.Value); + }); + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + var provider = await CreateProviderAsync(database, CreateServer(ScanAsync(cancellation.Token))); + + await using (var enumerator = provider.ListAsync(cancellationToken: cancellation.Token).GetAsyncEnumerator(cancellation.Token)) + { + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal(id, enumerator.Current); + Assert.Equal(128, scannedKeys); + Assert.Equal(128, metadataReads); + Assert.False(scanDisposed); + + if (cancel) + { + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(() => enumerator.MoveNextAsync().AsTask()); + } + } + + Assert.Equal(128, scannedKeys); + Assert.Equal(128, metadataReads); + Assert.True(scanDisposed); + + async IAsyncEnumerable ScanAsync([EnumeratorCancellation] CancellationToken cancellationToken = default) + { + try + { + await Task.CompletedTask; + for (var i = 0; i < 129; i++) + { + cancellationToken.ThrowIfCancellationRequested(); + scannedKeys++; + yield return key; + } + } + finally + { + scanDisposed = true; + } + } + } + + [Fact] + public async Task ListAsync_DisconnectedLaterPrimaryPropagatesAfterEarlierIds() + { + var id = JournalId.Create("redis", "earlier-server"); + var key = RedisJournalStorage.GetMetadataKey(KeyPrefix, "mapped"); + var database = Substitute.For(); + database.HashGetAsync(key, RedisJournalStorage.JournalIdMetadataKey) + .Returns(Task.FromResult((RedisValue)id.Value)); + var disconnected = Substitute.For(); + disconnected.IsConnected.Returns(false); + var provider = await CreateProviderAsync(database, CreateServer(ScanAsync()), disconnected); + await using var enumerator = provider.ListAsync(cancellationToken: TestContext.Current.CancellationToken).GetAsyncEnumerator(TestContext.Current.CancellationToken); + + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal(id, enumerator.Current); + + var exception = await Assert.ThrowsAsync(() => enumerator.MoveNextAsync().AsTask()); + + Assert.Contains("not connected", exception.Message); + + async IAsyncEnumerable ScanAsync() + { + await Task.CompletedTask; + yield return key; + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ListAsync_NoPrimaryThrows(bool hasReplica) + { + var replica = Substitute.For(); + replica.IsReplica.Returns(true); + replica.IsConnected.Returns(true); + var provider = await CreateProviderAsync(Substitute.For(), hasReplica ? [replica] : []); + await using var enumerator = provider.ListAsync(cancellationToken: TestContext.Current.CancellationToken).GetAsyncEnumerator(TestContext.Current.CancellationToken); + + var exception = await Assert.ThrowsAsync(() => enumerator.MoveNextAsync().AsTask()); + + Assert.Equal("No connected primary Redis servers are available for journal discovery.", exception.Message); + } + + [Fact] + public async Task ListAsync_CancellationDuringFinalDeletedMetadataResponsePropagates() + { + var key = RedisJournalStorage.GetMetadataKey(KeyPrefix, "deleted"); + var database = Substitute.For(); + database.HashGetAsync(key, RedisJournalStorage.JournalIdMetadataKey) + .Returns(Task.FromResult(RedisValue.Null)); + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + var scriptCalls = 0; + database.ScriptEvaluateAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(_ => + { + scriptCalls++; + cancellation.Cancel(); + return Task.FromResult(RedisResult.Create(new RedisValue[] { 0 })); + }); + var provider = await CreateProviderAsync(database, CreateServer(ScanAsync())); + await using var enumerator = provider.ListAsync(cancellationToken: cancellation.Token).GetAsyncEnumerator(cancellation.Token); + + var exception = await Assert.ThrowsAnyAsync(() => enumerator.MoveNextAsync().AsTask()); + + Assert.Equal(cancellation.Token, exception.CancellationToken); + Assert.Equal(1, scriptCalls); + + async IAsyncEnumerable ScanAsync() + { + await Task.CompletedTask; + yield return key; + } + } + + [Fact] + public async Task ListAsync_CancellationDuringFinalScanCompletionPropagates() + { + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + var provider = await CreateProviderAsync(Substitute.For(), CreateServer(ScanAsync())); + await using var enumerator = provider.ListAsync(cancellationToken: cancellation.Token).GetAsyncEnumerator(cancellation.Token); + + var exception = await Assert.ThrowsAnyAsync(() => enumerator.MoveNextAsync().AsTask()); + + Assert.Equal(cancellation.Token, exception.CancellationToken); + + async IAsyncEnumerable ScanAsync() + { + await Task.CompletedTask; + cancellation.Cancel(); + yield break; + } + } + + private static IServer CreateServer(IAsyncEnumerable keys) + { + var server = Substitute.For(); + server.IsConnected.Returns(true); + server.KeysAsync(0, RedisJournalStorage.GetMetadataKeyPattern(KeyPrefix), pageSize: 250).Returns(keys); + return server; + } + + private static async Task CreateProviderAsync(IDatabase database, params IServer[] servers) + { + var connection = Substitute.For(); + connection.GetDatabase().Returns(database); + var endpoints = servers.Select((_, index) => (EndPoint)new DnsEndPoint($"primary-{index}", 6379)).ToArray(); + connection.GetEndPoints().Returns(endpoints); + for (var i = 0; i < servers.Length; i++) + { + connection.GetServer(endpoints[i]).Returns(servers[i]); + } + + var provider = new RedisJournalStorageProvider( + Options.Create(new RedisJournalStorageOptions + { + KeyPrefix = KeyPrefix, + CreateMultiplexer = _ => Task.FromResult((connection, true)), + }), + Options.Create(new ClusterOptions { ServiceId = "catalog-tests" }), + Options.Create(new JournaledStateManagerOptions())); + var lifecycle = new SiloLifecycleSubject(NullLogger.Instance); + provider.Participate(lifecycle); + await lifecycle.OnStart(TestContext.Current.CancellationToken); + return provider; + } +} diff --git a/test/Extensions/Orleans.Redis.Tests/Journaling/RedisJournalStorageTests.cs b/test/Extensions/Orleans.Redis.Tests/Journaling/RedisJournalStorageTests.cs index a0587591886..b905c5e0f72 100644 --- a/test/Extensions/Orleans.Redis.Tests/Journaling/RedisJournalStorageTests.cs +++ b/test/Extensions/Orleans.Redis.Tests/Journaling/RedisJournalStorageTests.cs @@ -79,16 +79,24 @@ public async Task ReplaceListAndDelete_UseJournalIds() await provider.CreateStorage(other).CreateIfNotExistsAsync(cancellationToken: TestContext.Current.CancellationToken); var listed = await ToListAsync( - provider.ListAsync(JournalId.Create("redis", "list"), TestContext.Current.CancellationToken), + provider.ListAsync( + new ListOptions { Prefix = JournalId.Create("redis", "list") }, + TestContext.Current.CancellationToken), TestContext.Current.CancellationToken); - Assert.Equal([idA, child, idB], listed); + Assert.Equal(3, listed.Count); + Assert.Equal(listed.Count, listed.Distinct().Count()); + Assert.True(new HashSet { idA, child, idB }.SetEquals(listed)); await provider.CreateStorage(idA).DeleteAsync(TestContext.Current.CancellationToken); listed = await ToListAsync( - provider.ListAsync(JournalId.Create("redis", "list"), TestContext.Current.CancellationToken), + provider.ListAsync( + new ListOptions { Prefix = JournalId.Create("redis", "list") }, + TestContext.Current.CancellationToken), TestContext.Current.CancellationToken); - Assert.Equal([child, idB], listed); + Assert.Equal(2, listed.Count); + Assert.Equal(listed.Count, listed.Distinct().Count()); + Assert.True(new HashSet { child, idB }.SetEquals(listed)); } [Fact] diff --git a/test/Orleans.DurableJobs.Tests/DurableJobs/JournaledJobShardManagerTests.cs b/test/Orleans.DurableJobs.Tests/DurableJobs/JournaledJobShardManagerTests.cs index d17467b3ed3..9693c919b30 100644 --- a/test/Orleans.DurableJobs.Tests/DurableJobs/JournaledJobShardManagerTests.cs +++ b/test/Orleans.DurableJobs.Tests/DurableJobs/JournaledJobShardManagerTests.cs @@ -111,7 +111,7 @@ public async Task EmptyShard_IsDeletedWhenUnregistered() Assert.Null(await storageProvider.CreateStorage(storageId).GetMetadataAsync(cancellationToken)); Assert.Empty(await ToListAsync(storageProvider.ListAsync( - JobShardId.StoragePrefix, + new() { Prefix = JobShardId.StoragePrefix }, cancellationToken), cancellationToken)); } @@ -643,8 +643,8 @@ public void AllowAppends() public IJournalStorage CreateStorage(JournalId journalId) => new CountingJournalStorage(this, _inner.CreateStorage(journalId)); - public IAsyncEnumerable ListAsync(JournalId prefix = default, CancellationToken cancellationToken = default) - => _inner.ListAsync(prefix, cancellationToken); + public IAsyncEnumerable ListAsync(ListOptions? options = null, CancellationToken cancellationToken = default) + => _inner.ListAsync(options, cancellationToken); private async ValueTask OnAppendAsync(CancellationToken cancellationToken) { diff --git a/test/Orleans.Journaling.Tests/AzureTableJournalStorageProviderTests.cs b/test/Orleans.Journaling.Tests/AzureTableJournalStorageProviderTests.cs index fd8f763ee15..0d84b34330c 100644 --- a/test/Orleans.Journaling.Tests/AzureTableJournalStorageProviderTests.cs +++ b/test/Orleans.Journaling.Tests/AzureTableJournalStorageProviderTests.cs @@ -284,7 +284,7 @@ public async Task ListAsync_EmptyTable_ReturnsEmptyAndSelectsOnlyCanonicalJourna } [Fact] - public async Task ListAsync_MultipleHeaders_ReturnsOrdinallySortedJournalIds() + public async Task ListAsync_MultipleHeaders_ReturnsProviderTraversalOrder() { var table = new FakeTableClient(); var lower = new JournalId("catalog/a"); @@ -299,7 +299,7 @@ public async Task ListAsync_MultipleHeaders_ReturnsOrdinallySortedJournalIds() context.Provider.ListAsync(cancellationToken: TestContext.Current.CancellationToken), TestContext.Current.CancellationToken); - Assert.Equal([upper, lower, last], result); + Assert.Equal([last, lower, upper], result); Assert.Equal(3, result.Count); } @@ -318,10 +318,10 @@ public async Task ListAsync_WithPrefix_ReturnsExactIdAndDescendantsOnly() using var context = await CreateStartedProviderAsync(table, TestContext.Current.CancellationToken); var result = await ToListAsync( - context.Provider.ListAsync(prefix, TestContext.Current.CancellationToken), + context.Provider.ListAsync(new() { Prefix = prefix }, TestContext.Current.CancellationToken), TestContext.Current.CancellationToken); - Assert.Equal([exact, child], result); + Assert.Equal([child, exact], result); var query = Assert.Single(table.QueryCalls); Assert.Equal( TableClient.CreateQueryFilter($"RowKey eq {AzureTableJournalStorage.HeaderRowKey}"), @@ -401,7 +401,7 @@ public async Task ListAsync_CustomPartitionMapping_UsesCanonicalJournalIdAndAppl await StartAsync(context.Provider, TestContext.Current.CancellationToken); var result = await ToListAsync( - context.Provider.ListAsync(new JournalId("tenant/orders"), TestContext.Current.CancellationToken), + context.Provider.ListAsync(new() { Prefix = new JournalId("tenant/orders") }, TestContext.Current.CancellationToken), TestContext.Current.CancellationToken); Assert.Equal([included], result); @@ -419,7 +419,7 @@ public async Task ListAsync_LegacyHeaderUsingDefaultPartitionMapping_IsStillRetu using var context = await CreateStartedProviderAsync(table, TestContext.Current.CancellationToken); var result = await ToListAsync( - context.Provider.ListAsync(new JournalId("legacy/orders"), TestContext.Current.CancellationToken), + context.Provider.ListAsync(new() { Prefix = new JournalId("legacy/orders") }, TestContext.Current.CancellationToken), TestContext.Current.CancellationToken); Assert.Equal([journalId], result); @@ -442,7 +442,7 @@ public async Task ListAsync_WhenCanceledBeforeQuery_ThrowsWithoutQuerying() } [Fact] - public async Task ListAsync_WhenCanceledAfterQuery_ThrowsBeforeYieldingBufferedIds() + public async Task ListAsync_WhenCanceledAfterQuery_ThrowsBeforeYieldingIds() { using var cts = new CancellationTokenSource(); var table = new FakeTableClient { AfterQuery = cts.Cancel }; diff --git a/test/Orleans.Journaling.Tests/JournalStorageCatalogTests.cs b/test/Orleans.Journaling.Tests/JournalStorageCatalogTests.cs new file mode 100644 index 00000000000..e77a73f7175 --- /dev/null +++ b/test/Orleans.Journaling.Tests/JournalStorageCatalogTests.cs @@ -0,0 +1,730 @@ +using System.Buffers; +using System.Globalization; +using Amazon.S3; +using Amazon.S3.Model; +using Azure; +using Azure.Core; +using Azure.Data.Tables; +using Azure.Data.Tables.Models; +using Azure.Storage.Blobs; +using Azure.Storage.Blobs.Models; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using Orleans.Runtime; +using Xunit; + +namespace Orleans.Journaling.Tests; + +[TestSuite("BVT")] +[TestProvider("None")] +[TestCategory("BVT")] +public sealed class JournalStorageCatalogTests +{ + [Theory] + [InlineData("Volatile")] + [InlineData("AzureBlob")] + [InlineData("AzureTable")] + [InlineData("S3")] + public async Task ListAsync_OptionsSelectExactAndDescendantIdentities(string kind) + { + string[] ids = + [ + "tenant/z", "tenant2", "tenant", "tenant/a", "tenant/a/child", "tenantish/child", + "tenant%2Fone", "tenant%2Fone/child", " leading space ", "raw\uD800", "raw\uD801", + "unicode/\u00E9", "percent%2f", @"back\slash", "tenant/", + ]; + await using var context = await CreateAsync(kind, ids); + + AssertMembership(ids, await DrainAsync(context.Catalog.ListAsync(cancellationToken: TestContext.Current.CancellationToken))); + AssertMembership(ids, await DrainAsync(context.Catalog.ListAsync(new(), TestContext.Current.CancellationToken))); + AssertMembership( + ["tenant", "tenant/z", "tenant/a", "tenant/a/child", "tenant/"], + await DrainAsync(context.Catalog.ListAsync(new() { Prefix = new("tenant") }, TestContext.Current.CancellationToken))); + AssertMembership( + ["tenant%2Fone", "tenant%2Fone/child"], + await DrainAsync(context.Catalog.ListAsync( + new() { Prefix = JournalId.Create("tenant/one") }, TestContext.Current.CancellationToken))); + } + + [Theory] + [InlineData("Volatile")] + [InlineData("AzureBlob")] + [InlineData("AzureTable")] + [InlineData("S3")] + public async Task ListAsync_OptionsAreReadAtEnumerationStartAndRemainStable(string kind) + { + await using var context = await CreateAsync(kind, ["tenant/z", "tenant/a", "tenant/b", "other/q"]); + var options = new ListOptions { Prefix = new("other") }; + var listing = context.Catalog.ListAsync(options, TestContext.Current.CancellationToken); + options.Prefix = new("tenant"); + await using var enumerator = listing.GetAsyncEnumerator(TestContext.Current.CancellationToken); + Assert.True(await enumerator.MoveNextAsync()); + var result = new List { enumerator.Current.Value }; + + options.Prefix = new("other"); + while (await enumerator.MoveNextAsync()) + { + result.Add(enumerator.Current.Value); + } + + AssertMembership(["tenant/z", "tenant/a", "tenant/b"], result); + AssertMembership(["other/q"], await DrainAsync(listing)); + } + + [Theory] + [InlineData("AzureBlob")] + [InlineData("AzureTable")] + [InlineData("S3")] + public async Task ListAsync_YieldsCurrentPageBeforeFetchingFailingTail(string kind) + { + await using var context = await CreateAsync(kind, ["tenant/z", "tenant/a", "tenant/b"]); + var failure = new InvalidOperationException("later page failed"); + context.Native.Failure = failure; + context.Native.FailureAtRequest = 2; + await using var enumerator = context.Catalog.ListAsync( + new() { Prefix = new("tenant") }, TestContext.Current.CancellationToken).GetAsyncEnumerator(TestContext.Current.CancellationToken); + + Assert.Empty(context.Native.Requests); + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal("tenant/z", enumerator.Current.Value); + Assert.Single(context.Native.Requests); + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal("tenant/a", enumerator.Current.Value); + Assert.Single(context.Native.Requests); + + var actual = await Assert.ThrowsAsync(() => enumerator.MoveNextAsync().AsTask()); + Assert.Same(failure, actual); + Assert.Equal(2, context.Native.Requests.Count); + Assert.False(await enumerator.MoveNextAsync()); + Assert.Equal(2, context.Native.Requests.Count); + if (kind != "S3") + { + Assert.Equal(1, context.Native.DisposedEnumerators); + } + + context.Native.Failure = null; + Assert.Equal( + ["tenant/z", "tenant/a", "tenant/b"], + await DrainAsync(context.Catalog.ListAsync(cancellationToken: TestContext.Current.CancellationToken))); + Assert.Null(context.Native.Requests[2].Cursor); + } + + [Theory] + [InlineData("AzureBlob")] + [InlineData("AzureTable")] + [InlineData("S3")] + public async Task ListAsync_EarlyDisposalStopsFetchingPages(string kind) + { + await using var context = await CreateAsync(kind, ["z", "a", "b"]); + await using (var enumerator = context.Catalog.ListAsync(cancellationToken: TestContext.Current.CancellationToken).GetAsyncEnumerator(TestContext.Current.CancellationToken)) + { + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal("z", enumerator.Current.Value); + Assert.Single(context.Native.Requests); + } + + Assert.Single(context.Native.Requests); + Assert.Equal(kind == "S3" ? 0 : 1, context.Native.DisposedEnumerators); + } + + [Theory] + [InlineData("AzureBlob")] + [InlineData("AzureTable")] + [InlineData("S3")] + public async Task ListAsync_AdvanceCrossesEmptyAndFilteredNativePages(string kind) + { + await using var context = await CreateAsync(kind, ["tenant2", "tenantish/child", "tenant/valid"]); + context.Native.EmptyFirstPage = true; + await using var enumerator = context.Catalog.ListAsync( + new() { Prefix = new("tenant") }, TestContext.Current.CancellationToken).GetAsyncEnumerator(TestContext.Current.CancellationToken); + + Assert.True(await enumerator.MoveNextAsync()); + Assert.Equal("tenant/valid", enumerator.Current.Value); + Assert.Equal(3, context.Native.Requests.Count); + Assert.Equal([0, 2, 1], context.Native.Requests.Select(request => request.ResultCount)); + Assert.False(await enumerator.MoveNextAsync()); + Assert.Equal(3, context.Native.Requests.Count); + } + + [Theory] + [InlineData("Volatile")] + [InlineData("AzureBlob")] + [InlineData("AzureTable")] + [InlineData("S3")] + public async Task ListAsync_CancellationBeforeAndBetweenResultsPropagates(string kind) + { + await using var context = await CreateAsync(kind, ["z", "a", "b"]); + using var canceled = new CancellationTokenSource(); + canceled.Cancel(); + await using (var enumerator = context.Catalog.ListAsync(cancellationToken: canceled.Token).GetAsyncEnumerator(canceled.Token)) + { + var exception = await Assert.ThrowsAnyAsync(() => enumerator.MoveNextAsync().AsTask()); + Assert.Equal(canceled.Token, exception.CancellationToken); + Assert.Empty(context.Native.Requests); + } + + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + await using var active = context.Catalog.ListAsync(cancellationToken: cancellation.Token).GetAsyncEnumerator(cancellation.Token); + Assert.True(await active.MoveNextAsync()); + var requests = context.Native.Requests.Count; + cancellation.Cancel(); + var actual = await Assert.ThrowsAnyAsync(() => active.MoveNextAsync().AsTask()); + Assert.Equal(cancellation.Token, actual.CancellationToken); + Assert.Equal(requests, context.Native.Requests.Count); + } + + [Theory] + [InlineData("AzureBlob")] + [InlineData("AzureTable")] + [InlineData("S3")] + public async Task ListAsync_CancellationAfterNativeResponseIncludesEmptyPages(string kind) + { + foreach (var empty in new[] { false, true }) + { + await using var context = await CreateAsync(kind, empty ? [] : ["z", "a"]); + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + context.Native.BeforeResponse = cancellation.Cancel; + await using var enumerator = context.Catalog.ListAsync(cancellationToken: cancellation.Token).GetAsyncEnumerator(cancellation.Token); + + var actual = await Assert.ThrowsAnyAsync(() => enumerator.MoveNextAsync().AsTask()); + Assert.Equal(cancellation.Token, actual.CancellationToken); + var request = Assert.Single(context.Native.Requests); + Assert.Equal(empty ? 0 : 2, request.ResultCount); + Assert.Equal(cancellation.Token, request.CancellationToken); + } + } + + [Theory] + [InlineData("AzureBlob")] + [InlineData("AzureTable")] + [InlineData("S3")] + public async Task ListAsync_NativeRequestsCarryPrefixContinuationAndCancellation(string kind) + { + await using var context = await CreateAsync(kind, ["tenant/z", "tenant/a", "tenant/b"]); + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + Assert.Equal( + ["tenant/z", "tenant/a", "tenant/b"], + await DrainAsync(context.Catalog.ListAsync(new() { Prefix = new("tenant") }, cancellation.Token))); + + Assert.Equal(2, context.Native.Requests.Count); + Assert.Null(context.Native.Requests[0].Cursor); + Assert.Equal(context.Native.Requests[0].NextCursor, context.Native.Requests[1].Cursor); + Assert.Null(context.Native.Requests[1].NextCursor); + Assert.All(context.Native.Requests, request => + { + Assert.Equal(kind == "AzureBlob" ? 5000 : 1000, request.Maximum); + Assert.Equal(cancellation.Token, request.CancellationToken); + Assert.Equal(kind == "AzureBlob" ? "tenant" : null, request.Prefix); + }); + if (kind == "AzureTable") + { + Assert.Equal( + TableClient.CreateQueryFilter($"RowKey eq {AzureTableJournalStorage.HeaderRowKey}"), + context.Native.Filter); + Assert.Equal([AzureTableJournalStorage.JournalIdPropertyName], Assert.IsType(context.Native.Select)); + } + } + + [Theory] + [InlineData("AzureBlob")] + [InlineData("AzureTable")] + [InlineData("S3")] + public async Task ListAsync_RequiresInitializedProvider(string kind) + { + await using var context = await CreateAsync(kind, ["z"], initialize: false); + var unavailable = await Assert.ThrowsAsync( + () => DrainAsync(context.Catalog.ListAsync(cancellationToken: TestContext.Current.CancellationToken))); + Assert.Contains("has not been initialized", unavailable.Message); + Assert.Empty(context.Native.Requests); + await context.InitializeAsync(); + Assert.Equal(["z"], await DrainAsync(context.Catalog.ListAsync(cancellationToken: TestContext.Current.CancellationToken))); + if (context.Provider is S3JournalStorageProvider s3) + { + await s3.CloseAsync(TestContext.Current.CancellationToken); + await Assert.ThrowsAsync( + () => DrainAsync(context.Catalog.ListAsync(cancellationToken: TestContext.Current.CancellationToken))); + } + } + + [Theory] + [InlineData(1)] + [InlineData(2)] + public async Task AzureBlobListAsync_FailedInitializationKeepsCatalogUnavailable(int failedSetup) + { + await using var context = await CreateAsync("AzureBlob", ["z"], initialize: false); + var failure = new RequestFailedException(503, "setup failed"); + context.Native.FailingSetup = failedSetup; + context.Native.SetupFailure = failure; + + Assert.Same(failure, await Assert.ThrowsAsync(() => context.InitializeAsync())); + Assert.Equal(failedSetup, context.Native.SetupCalls); + await Assert.ThrowsAsync( + () => DrainAsync(context.Catalog.ListAsync(cancellationToken: TestContext.Current.CancellationToken))); + Assert.Empty(context.Native.Requests); + + context.Native.SetupFailure = null; + await context.InitializeAsync(); + Assert.Equal(["z"], await DrainAsync(context.Catalog.ListAsync(cancellationToken: TestContext.Current.CancellationToken))); + } + + [Theory] + [InlineData("EquivalentDelegate")] + [InlineData("WrapperFactory")] + public async Task AzureBlobListAsync_EquivalentConfigurationStreamsSameCatalog(string layout) + { + await using var context = await CreateAsync( + "AzureBlob", ["jobs/shards/z", "jobs/shards2", "jobs/shards/a", "jobs/shards/child"], blobLayout: layout); + var result = await DrainAsync(context.Catalog.ListAsync( + new() { Prefix = new("jobs/shards") }, TestContext.Current.CancellationToken)); + + Assert.Equal(["jobs/shards/z", "jobs/shards/a", "jobs/shards/child"], result); + Assert.Equal(2, context.Native.Requests.Count); + Assert.All(context.Native.Requests, request => Assert.Equal("jobs/shards", request.Prefix)); + } + + [Fact] + public async Task AzureBlobListAsync_FiltersAppendWalEntries() + { + BlobItem[] blobs = + [ + Blob("tenant/z/wal"), Blob("tenant/block/wal", BlobType.Block), + Blob("tenant/page/wal", BlobType.Page), Blob("tenant/chk.1"), + Blob("tenant/child/wal"), Blob("tenant/case/WAL"), Blob("/wal"), + Blob(" \t /wal"), Blob("tenant2/wal"), Blob("tenant/wal"), + ]; + await using var context = await CreateAsync("AzureBlob", [], blobs: blobs); + Assert.Equal( + ["tenant/z", "tenant/child", "tenant"], + await DrainAsync(context.Catalog.ListAsync(new() { Prefix = new("tenant") }, TestContext.Current.CancellationToken))); + Assert.Equal(8, context.Native.Requests.Sum(request => request.ResultCount)); + } + + [Fact] + public async Task AzureTableListAsync_PreservesCanonicalAndReversibleLegacyIds() + { + TableEntity[] headers = + [ + Header("opaque!1", "tenant/z"), Header("other%2Fid", "tenant/a"), + Header("legacy%2F%C3%A9"), Header("legacy%2fchild"), Header("%41"), + Header("bad%ZZ"), Header("fallback%2Fid", ""), Header("%20%09"), + new("orphan", "data") { [AzureTableJournalStorage.JournalIdPropertyName] = "ignored" }, + ]; + await using var context = await CreateAsync("AzureTable", [], headers: headers); + Assert.Equal( + ["tenant/z", "tenant/a", "legacy/\u00E9"], + await DrainAsync(context.Catalog.ListAsync(cancellationToken: TestContext.Current.CancellationToken))); + Assert.Equal(8, context.Native.Requests.Sum(request => request.ResultCount)); + } + + [Fact] + public async Task S3ListAsync_UsesCanonicalMappingAfterPrefixFilter() + { + var mapped = new List(); + string[] keys = + [ + "current/tenant/z/wal", "legacy/tenant/z/wal", "current/tenant/a/chk.1", + "current/tenant/a/wal", "current/other/x/wal", "invalid/wal", "default/wal", + ]; + await using var context = await CreateAsync("S3", [], keys: keys, configureS3: options => + { + options.GetObjectKey = id => + { + mapped.Add(id.Value); + return $"current/{id.Value}"; + }; + options.TryParseJournalId = value => value switch + { + "invalid" => null, + "default" => default(JournalId), + _ => new JournalId(value[(value.IndexOf('/') + 1)..]), + }; + }); + + Assert.Equal( + ["tenant/z", "tenant/a"], + await DrainAsync(context.Catalog.ListAsync(new() { Prefix = new("tenant") }, TestContext.Current.CancellationToken))); + Assert.Equal(["tenant/z", "tenant/z", "tenant/a"], mapped); + } + + [Theory] + [InlineData("Create")] + [InlineData("Append")] + [InlineData("Replace")] + public async Task VolatileListAsync_TracksExistenceAcrossNewEnumerations(string operation) + { + var provider = new VolatileJournalStorageProvider(); + var storage = provider.CreateStorage(new("tenant/id")); + var listing = provider.ListAsync(cancellationToken: TestContext.Current.CancellationToken); + Assert.Empty(await DrainAsync(listing)); + await MaterializeAsync(); + Assert.Equal(["tenant/id"], await DrainAsync(listing)); + await MaterializeAsync(); + Assert.Equal(["tenant/id"], await DrainAsync(listing)); + await storage.DeleteAsync(TestContext.Current.CancellationToken); + Assert.Empty(await DrainAsync(listing)); + await MaterializeAsync(); + Assert.Equal(["tenant/id"], await DrainAsync(listing)); + + async Task MaterializeAsync() + { + switch (operation) + { + case "Create": + await storage.CreateIfNotExistsAsync(cancellationToken: TestContext.Current.CancellationToken); + break; + case "Append": + await storage.AppendAsync(new ReadOnlySequence(new byte[] { 1 }), TestContext.Current.CancellationToken); + break; + case "Replace": + await storage.ReplaceAsync(new ReadOnlySequence(new byte[] { 2 }), TestContext.Current.CancellationToken); + break; + } + } + } + + private static void AssertMembership(string[] expected, IReadOnlyCollection actual) + { + Assert.Equal(expected.Length, actual.Count); + Assert.Equal(expected.Length, actual.Distinct(StringComparer.Ordinal).Count()); + Assert.Equal(expected.Order(StringComparer.Ordinal), actual.Order(StringComparer.Ordinal)); + } + + private static async Task> DrainAsync(IAsyncEnumerable source) + { + var result = new List(); + await foreach (var id in source) + { + result.Add(id.Value); + } + + return result; + } + + private static BlobItem Blob(string name, BlobType type = BlobType.Append) + => BlobsModelFactory.BlobItem( + name: name, deleted: false, + properties: BlobsModelFactory.BlobItemProperties(accessTierInferred: false, blobType: type)); + + private static TableEntity Header(string partition, string? id = null) + { + var result = new TableEntity(partition, AzureTableJournalStorage.HeaderRowKey); + if (id is not null) + { + result[AzureTableJournalStorage.JournalIdPropertyName] = id; + } + + return result; + } + + private static async Task CreateAsync( + string kind, string[] ids, bool initialize = true, string? blobLayout = null, + BlobItem[]? blobs = null, TableEntity[]? headers = null, string[]? keys = null, + Action? configureS3 = null) + { + var context = new ProviderContext(kind, ids, blobLayout, blobs, headers, keys, configureS3); + try + { + if (initialize) + { + await context.InitializeAsync(); + } + + if (kind == "Volatile") + { + foreach (var id in ids) + { + Assert.True(await context.Provider.CreateStorage(new(id)) + .CreateIfNotExistsAsync(cancellationToken: TestContext.Current.CancellationToken)); + } + } + + return context; + } + catch + { + await context.DisposeAsync(); + throw; + } + } + + private sealed class ProviderContext : IAsyncDisposable + { + private readonly ServiceProvider _services; + private readonly RecordingLifecycle _lifecycle = new(); + private readonly IAmazonS3? _client; + + public ProviderContext( + string kind, string[] ids, string? blobLayout, BlobItem[]? blobs, + TableEntity[]? headers, string[]? keys, Action? configureS3) + { + var services = new ServiceCollection(); + services.AddKeyedSingleton("test", new TestFormat()); + _services = services.BuildServiceProvider(); + var manager = Options.Create(new JournaledStateManagerOptions { JournalFormatKey = "test" }); + switch (kind) + { + case "Volatile": + Provider = new VolatileJournalStorageProvider(); + break; + case "AzureBlob": + var container = new FakeContainer(Native, blobs ?? ids.Select(id => Blob($"{id}/wal")).ToArray()); + var blobOptions = new AzureBlobJournalStorageOptions { ContainerName = "journals" }; + blobOptions.ConfigureBlobServiceClient(_ => Task.FromResult(new FakeBlobService(container))); + if (blobLayout == "EquivalentDelegate") + { + blobOptions.GetWalBlobName = id => $"{id.Value}/wal"; + } + else if (blobLayout == "WrapperFactory") + { + blobOptions.BuildContainerFactory = (_, _) => new WrapperFactory(container); + } + + Provider = new AzureBlobJournalStorageProvider( + Options.Create(blobOptions), manager, _services, NullLogger.Instance); + break; + case "AzureTable": + var table = new FakeTable(Native, headers ?? ids.Select((id, index) => Header($"mapped-{index}", id)).ToArray()); + var tableOptions = new AzureTableJournalStorageOptions { TableName = "journals" }; + tableOptions.ConfigureTableServiceClient(_ => Task.FromResult(new FakeTableService(table))); + Provider = new AzureTableJournalStorageProvider( + Options.Create(tableOptions), manager, _services, NullLogger.Instance); + break; + case "S3": + _client = Substitute.For(); + _client.HeadBucketAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromResult(new HeadBucketResponse())); + var objects = (keys ?? ids.Select(id => $"{id}/wal").ToArray()).Select(key => new S3Object { Key = key }).ToArray(); + _client.ListObjectsV2Async(Arg.Any(), Arg.Any()).Returns(call => + { + var request = call.Arg(); + Assert.Equal("journals", request.BucketName); + Assert.Null(request.Prefix); + Assert.Null(request.StartAfter); + var page = Native.Fetch(objects, request.ContinuationToken, request.MaxKeys, request.Prefix, call.Arg()); + return Task.FromResult(new ListObjectsV2Response + { + S3Objects = page.Values.ToList(), + IsTruncated = page.NextCursor is not null, + NextContinuationToken = page.NextCursor, + }); + }); + var s3Options = new S3JournalStorageOptions { BucketName = "journals", S3Client = _client }; + configureS3?.Invoke(s3Options); + Provider = new S3JournalStorageProvider( + Options.Create(s3Options), manager, _services, NullLogger.Instance); + break; + default: + throw new ArgumentOutOfRangeException(nameof(kind)); + } + + Catalog = Assert.IsAssignableFrom(Provider); + if (Provider is ILifecycleParticipant participant) + { + participant.Participate(_lifecycle); + } + } + + public NativeState Native { get; } = new(); + public IJournalStorageProvider Provider { get; } + public IJournalStorageCatalog Catalog { get; } + + public Task InitializeAsync() + => Provider is VolatileJournalStorageProvider + ? Task.CompletedTask : _lifecycle.StartAsync(TestContext.Current.CancellationToken); + + public async ValueTask DisposeAsync() + { + if (Provider is S3JournalStorageProvider s3) + { + await s3.CloseAsync(TestContext.Current.CancellationToken); + _client!.Dispose(); + } + + await _services.DisposeAsync(); + } + } + + private sealed record NativeRequest( + string? Cursor, int? Maximum, string? Prefix, CancellationToken CancellationToken, int ResultCount, string? NextCursor); + private sealed record NativePage(IReadOnlyList Values, string? NextCursor); + + private sealed class NativeState + { + public List Requests { get; } = []; + public bool EmptyFirstPage { get; set; } + public Exception? Failure { get; set; } + public int FailureAtRequest { get; set; } + public Action? BeforeResponse { get; set; } + public int DisposedEnumerators { get; set; } + public int SetupCalls { get; set; } + public int FailingSetup { get; set; } + public Exception? SetupFailure { get; set; } + public string? Filter { get; set; } + public string[]? Select { get; set; } + + public NativePage Fetch(T[] records, string? cursor, int? maximum, string? prefix, CancellationToken cancellationToken) + { + var offset = cursor is null ? 0 : int.Parse(cursor.AsSpan("native:".Length), CultureInfo.InvariantCulture); + var count = cursor is null && EmptyFirstPage ? 0 : Math.Min(2, records.Length - offset); + var next = offset + count < records.Length ? $"native:{offset + count}" : null; + Requests.Add(new(cursor, maximum, prefix, cancellationToken, count, next)); + if (Requests.Count == FailureAtRequest && Failure is { } failure) + { + throw failure; + } + + var callback = BeforeResponse; + BeforeResponse = null; + callback?.Invoke(); + return new(records.AsSpan(offset, count).ToArray(), next); + } + } + + private sealed class FakePageable(NativeState state, T[] records, int? maximum, string? prefix, CancellationToken token) : AsyncPageable + where T : notnull + { + public override async IAsyncEnumerable> AsPages(string? continuationToken = null, int? pageSizeHint = null) + { + try + { + do + { + await Task.CompletedTask; + var page = state.Fetch(records, continuationToken, pageSizeHint ?? maximum, prefix, token); + yield return Page.FromValues(page.Values, page.NextCursor, new FakeResponse()); + continuationToken = page.NextCursor; + } + while (continuationToken is not null); + } + finally + { + state.DisposedEnumerators++; + } + } + } + + private sealed class FakeBlobService(FakeContainer container) : BlobServiceClient + { + public override BlobContainerClient GetBlobContainerClient(string blobContainerName) + { + Assert.Equal("journals", blobContainerName); + return container; + } + } + + private sealed class FakeContainer(NativeState state, BlobItem[] records) : BlobContainerClient + { + public override Task> CreateIfNotExistsAsync( + PublicAccessType publicAccessType = PublicAccessType.None, + IDictionary? metadata = null, + BlobContainerEncryptionScopeOptions? encryptionScopeOptions = null, + CancellationToken cancellationToken = default) + { + if (++state.SetupCalls == state.FailingSetup && state.SetupFailure is { } failure) + { + return Task.FromException>(failure); + } + + return Task.FromResult(Response.FromValue( + BlobsModelFactory.BlobContainerInfo(new ETag("created"), DateTimeOffset.UnixEpoch), new FakeResponse())); + } + + public override AsyncPageable GetBlobsAsync( + BlobTraits traits = BlobTraits.None, BlobStates states = BlobStates.None, + string? prefix = null, CancellationToken cancellationToken = default) + { + Assert.Equal(BlobTraits.None, traits); + Assert.Equal(BlobStates.None, states); + return new FakePageable( + state, records.Where(item => prefix is null || item.Name.StartsWith(prefix, StringComparison.Ordinal)).ToArray(), + null, prefix, cancellationToken); + } + } + + private sealed class WrapperFactory(FakeContainer container) : IBlobContainerFactory + { + public BlobContainerClient GetBlobContainerClient(JournalId journalId) => container; + public async Task InitializeAsync(BlobServiceClient client, CancellationToken cancellationToken) + => await container.CreateIfNotExistsAsync(cancellationToken: cancellationToken); + } + + private sealed class FakeTableService(FakeTable table) : TableServiceClient + { + public override TableClient GetTableClient(string tableName) + { + Assert.Equal("journals", tableName); + return table; + } + } + + private sealed class FakeTable(NativeState state, TableEntity[] records) : TableClient + { + public override Task> CreateIfNotExistsAsync(CancellationToken cancellationToken = default) + => Task.FromResult(Response.FromValue(new TableItem("journals"), new FakeResponse())); + + public override AsyncPageable QueryAsync( + string? filter = null, int? maxPerPage = null, IEnumerable? select = null, + CancellationToken cancellationToken = default) + { + state.Filter = filter; + state.Select = select?.ToArray(); + Assert.Equal(1000, maxPerPage); + Assert.Equal(TableClient.CreateQueryFilter($"RowKey eq {AzureTableJournalStorage.HeaderRowKey}"), filter); + Assert.Equal([AzureTableJournalStorage.JournalIdPropertyName], Assert.IsType(state.Select)); + var values = records.Where(entity => entity.RowKey == AzureTableJournalStorage.HeaderRowKey).Select(entity => + { + var projected = new TableEntity(entity.PartitionKey, entity.RowKey); + if (entity.TryGetValue(AzureTableJournalStorage.JournalIdPropertyName, out var value)) + { + projected[AzureTableJournalStorage.JournalIdPropertyName] = value; + } + + return (T)(ITableEntity)projected; + }).ToArray(); + return new FakePageable(state, values, maxPerPage, null, cancellationToken); + } + } + + private sealed class TestFormat : IJournalFormat + { + public string FormatKey => "test"; + public string? MimeType => "application/test"; + public JournalBufferWriter CreateWriter() => throw new NotSupportedException(); + public void Replay(JournalBufferReader input, JournalReplayContext context) => throw new NotSupportedException(); + } + + private sealed class RecordingLifecycle : ISiloLifecycle + { + private ILifecycleObserver? _observer; + public int HighestCompletedStage => ServiceLifecycleStage.RuntimeInitialize; + public int LowestStoppedStage => ServiceLifecycleStage.RuntimeInitialize; + public IDisposable Subscribe(string observerName, int stage, ILifecycleObserver observer) + { + Assert.Null(_observer); + _observer = observer; + return new NoopDisposable(); + } + + public Task StartAsync(CancellationToken token) => Assert.IsAssignableFrom(_observer).OnStart(token); + } + + private sealed class NoopDisposable : IDisposable + { + public void Dispose() { } + } + + private sealed class FakeResponse : Response + { + public override int Status => 200; + public override string ReasonPhrase => "OK"; + public override Stream? ContentStream { get; set; } + public override string ClientRequestId { get; set; } = string.Empty; + public override void Dispose() { } + protected override bool ContainsHeader(string name) => false; + protected override IEnumerable EnumerateHeaders() => []; + protected override bool TryGetHeader(string name, out string value) { value = string.Empty; return false; } + protected override bool TryGetHeaderValues(string name, out IEnumerable values) { values = []; return false; } + } +} diff --git a/test/Orleans.Journaling.Tests/S3JournalStorageTests.cs b/test/Orleans.Journaling.Tests/S3JournalStorageTests.cs index 72f0d4c5e4a..6b0368882ef 100644 --- a/test/Orleans.Journaling.Tests/S3JournalStorageTests.cs +++ b/test/Orleans.Journaling.Tests/S3JournalStorageTests.cs @@ -217,7 +217,7 @@ public async Task ListAsync_CustomObjectKeyMapping_FiltersParsedJournalIds() await provider.InitializeAsync(CancellationToken.None); var listed = new List(); - await foreach (var journalId in provider.ListAsync(new JournalId("journals"), CancellationToken.None)) + await foreach (var journalId in provider.ListAsync(new() { Prefix = new JournalId("journals") }, CancellationToken.None)) { listed.Add(journalId); } @@ -254,7 +254,7 @@ public async Task ListAsync_WhenAliasMapsToCanonicalJournalId_ReturnsCanonicalId await provider.InitializeAsync(CancellationToken.None); var listed = new List(); - await foreach (var journalId in provider.ListAsync(new JournalId("journals"), CancellationToken.None)) + await foreach (var journalId in provider.ListAsync(new() { Prefix = new JournalId("journals") }, CancellationToken.None)) { listed.Add(journalId); } @@ -287,7 +287,7 @@ public async Task ListAsync_WhenOnlyAliasMapsToJournalId_DoesNotReturnDanglingId await provider.InitializeAsync(CancellationToken.None); var listed = new List(); - await foreach (var journalId in provider.ListAsync(new JournalId("journals"), CancellationToken.None)) + await foreach (var journalId in provider.ListAsync(new() { Prefix = new JournalId("journals") }, CancellationToken.None)) { listed.Add(journalId); } @@ -1593,7 +1593,7 @@ await CreateStorage("journals/format-migration", journalFormatKey: "new-format") } [Fact] - public async Task ListAsync_ReturnsSortedJournalIdsMatchingPrefix() + public async Task ListAsync_ReturnsJournalIdsMatchingPrefix() { EnsureDockerAvailable(); var provider = CreateProvider(); @@ -1602,12 +1602,12 @@ public async Task ListAsync_ReturnsSortedJournalIdsMatchingPrefix() await CreateStorage("other/beta").AppendAsync(new ReadOnlySequence([1]), CancellationToken.None); var listed = new List(); - await foreach (var journalId in provider.ListAsync(new JournalId("journals"), CancellationToken.None)) + await foreach (var journalId in provider.ListAsync(new() { Prefix = new JournalId("journals") }, CancellationToken.None)) { listed.Add(journalId); } - Assert.Equal(["journals/alpha", "journals/zeta"], listed.Select(static id => id.Value)); + Assert.Equal(["journals/alpha", "journals/zeta"], listed.Select(static id => id.Value).Order(StringComparer.Ordinal)); } [Fact] diff --git a/test/Orleans.Journaling.Tests/VolatileJournalStorageProviderTests.cs b/test/Orleans.Journaling.Tests/VolatileJournalStorageProviderTests.cs index d3a69247737..3b78f041902 100644 --- a/test/Orleans.Journaling.Tests/VolatileJournalStorageProviderTests.cs +++ b/test/Orleans.Journaling.Tests/VolatileJournalStorageProviderTests.cs @@ -38,9 +38,9 @@ public async Task CreateIfNotExists_ListAndGetMetadataUseJournalIds() Assert.Equal("one", (await storageA.GetMetadataAsync(TestContext.Current.CancellationToken))!.Properties["owner"]); var listed = await ToListAsync( - provider.ListAsync(JournalId.Create("named", "logs"), TestContext.Current.CancellationToken), + provider.ListAsync(new() { Prefix = JournalId.Create("named", "logs") }, TestContext.Current.CancellationToken), TestContext.Current.CancellationToken); - Assert.Equal([idA, idChild, idB], listed); + Assert.Equal([idA, idChild, idB], listed.OrderBy(id => id.Value, StringComparer.Ordinal)); Assert.NotNull(await provider.CreateStorage(idB).GetMetadataAsync(TestContext.Current.CancellationToken)); Assert.Null(await provider.CreateStorage(JournalId.Create("named", "missing")).GetMetadataAsync(TestContext.Current.CancellationToken)); @@ -116,14 +116,14 @@ public async Task StorageOperationsUpdateMetadataAndDeleteRemovesStorage() Assert.Equal( [storageId], await ToListAsync( - provider.ListAsync(storageId, TestContext.Current.CancellationToken), + provider.ListAsync(new() { Prefix = storageId }, TestContext.Current.CancellationToken), TestContext.Current.CancellationToken)); await storage.DeleteAsync(TestContext.Current.CancellationToken); Assert.Null(await storage.GetMetadataAsync(TestContext.Current.CancellationToken)); Assert.Empty(await ToListAsync( - provider.ListAsync(storageId, TestContext.Current.CancellationToken), + provider.ListAsync(new() { Prefix = storageId }, TestContext.Current.CancellationToken), TestContext.Current.CancellationToken)); }