From 5704457bf6ccc5302a6bf24da968f4f59028f5ea Mon Sep 17 00:00:00 2001 From: Jason Shave Date: Fri, 16 Oct 2026 15:23:25 -0600 Subject: [PATCH 1/3] Blobs: createContainerIfNotExists option on UseBlobStorageForLargeMessages All three UseBlobStorageForLargeMessages overloads take an optional trailing `bool createContainerIfNotExists = true`, so an identity that can't create containers can use the short registration against a container created ahead of time instead of falling back to UseLargeMessageStore(new BlobLargeMessageStore(...)). Existing calls are unchanged. Docs and both READMEs updated; one Azurite test checks a missing container is not created and the send fails with 404. Co-Authored-By: Claude Opus 5.5 --- README.md | 4 +-- ...BlobStorageQueueClientBuilderExtensions.cs | 35 +++++++++++++------ src/AzureStorage.QueueService.Blobs/README.md | 4 +-- .../BlobLargeMessageEndToEndTests.cs | 15 ++++++++ 4 files changed, 44 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index f92c96c..ab74732 100644 --- a/README.md +++ b/README.md @@ -433,13 +433,13 @@ queues.AddClient("orders", builder.Configuration.GetSection("Queues:Orders")) .UseBlobStorageForLargeMessages(); // container "queue-large-messages" in the queue's own account ``` -With no arguments it uses the queue's own storage account and credentials: a connection string as is, or `https://myaccount.queue.core.windows.net/...` becomes `https://myaccount.blob.core.windows.net/`. You can also pass a container name, a `BlobContainerClient`, or `sp => ...` to build the container client from your services. The container is created the first time a large message is sent. Pass `thresholdBytes` to send smaller bodies to Blob storage as well. Any other store can plug in by implementing `ILargeMessageStore` and calling `UseLargeMessageStore`. +With no arguments it uses the queue's own storage account and credentials: a connection string as is, or `https://myaccount.queue.core.windows.net/...` becomes `https://myaccount.blob.core.windows.net/`. You can also pass a container name, a `BlobContainerClient`, or `sp => ...` to build the container client from your services. The container is created the first time a large message is sent; pass `createContainerIfNotExists: false` if it already exists. Pass `thresholdBytes` to send smaller bodies to Blob storage as well. Any other store can plug in by implementing `ILargeMessageStore` and calling `UseLargeMessageStore`. Things to know before turning it on: - **Every client that reads the queue needs the same store.** One without it can't read those messages and dead-letters them. Large messages travel in the library's envelope, which only 3.1 and later read. - **Some blobs are left behind.** A message that expires on the queue or is cleared from it is never processed, so its blob stays; a send that fails after the upload deletes its blob only on a best-effort basis. Add a [lifecycle management rule](https://learn.microsoft.com/azure/storage/blobs/lifecycle-management-overview) that deletes blobs in the container older than the queue's message time to live plus a margin (for example 9 days for the default 7), longer if you replay dead-lettered messages later. -- **Permissions:** the identity needs **Storage Blob Data Contributor** on the container or the account. One that can't create containers needs the container created beforehand and `UseLargeMessageStore(new BlobLargeMessageStore(container, createContainerIfNotExists: false))`. +- **Permissions:** the identity needs **Storage Blob Data Contributor** on the container or the account. One that can't create containers needs the container created beforehand and `createContainerIfNotExists: false`, for example `.UseBlobStorageForLargeMessages("order-bodies", createContainerIfNotExists: false)`. - **Messages are untrusted input.** The store only reads and deletes blobs in its own container; a message pointing anywhere else is dead-lettered without a request being made. A SAS token on the container client is never written to the queue. ## Testing your code diff --git a/src/AzureStorage.QueueService.Blobs/BlobStorageQueueClientBuilderExtensions.cs b/src/AzureStorage.QueueService.Blobs/BlobStorageQueueClientBuilderExtensions.cs index 9fc2778..e397a5b 100644 --- a/src/AzureStorage.QueueService.Blobs/BlobStorageQueueClientBuilderExtensions.cs +++ b/src/AzureStorage.QueueService.Blobs/BlobStorageQueueClientBuilderExtensions.cs @@ -28,7 +28,8 @@ namespace AzureStorage.QueueService; /// margin (and long enough to replay dead-lettered messages). /// /// -/// The identity needs the Storage Blob Data Contributor role on the container or the account. +/// The identity needs the Storage Blob Data Contributor role on the container or the account. One that is not +/// allowed to create containers needs the container created ahead of time and createContainerIfNotExists: false. /// /// public static class BlobStorageQueueClientBuilderExtensions @@ -38,7 +39,7 @@ public static class BlobStorageQueueClientBuilderExtensions /// /// Keeps bodies too big for a queue message as blobs in , which is created on the first - /// large send if it does not exist. + /// large send if it does not exist (unless is false). /// /// The client to attach the store to. /// The container to keep bodies in, with the credential to reach it. @@ -46,19 +47,24 @@ public static class BlobStorageQueueClientBuilderExtensions /// Also send bodies of at least this many bytes to Blob storage even when they would fit on the queue. Null (the /// default) sends only what would otherwise be over the 64 KiB limit. /// + /// + /// True (the default) creates the container on the first large send if it does not exist. Pass false when the + /// container is created ahead of time and the identity is not allowed to create containers. + /// /// The same builder, for chaining. - public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClientBuilder builder, BlobContainerClient container, int? thresholdBytes = null) + public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClientBuilder builder, BlobContainerClient container, int? thresholdBytes = null, bool createContainerIfNotExists = true) { ArgumentNullException.ThrowIfNull(builder); ArgumentNullException.ThrowIfNull(container); - return builder.UseBlobStorageForLargeMessages(_ => container, thresholdBytes); + return builder.UseBlobStorageForLargeMessages(_ => container, thresholdBytes, createContainerIfNotExists); } /// /// Keeps bodies too big for a queue message as blobs in the container returns. It /// is called once, from the application's services, the first time the client is used, so the container client can /// come from dependency injection (for example one registered with Microsoft.Extensions.Azure). The container - /// is created on the first large send if it does not exist. + /// is created on the first large send if it does not exist (unless is + /// false). /// /// The client to attach the store to. /// Returns the container to keep bodies in, with the credential to reach it. @@ -66,8 +72,12 @@ public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClie /// Also send bodies of at least this many bytes to Blob storage even when they would fit on the queue. Null (the /// default) sends only what would otherwise be over the 64 KiB limit. /// + /// + /// True (the default) creates the container on the first large send if it does not exist. Pass false when the + /// container is created ahead of time and the identity is not allowed to create containers. + /// /// The same builder, for chaining. - public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClientBuilder builder, Func createContainer, int? thresholdBytes = null) + public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClientBuilder builder, Func createContainer, int? thresholdBytes = null, bool createContainerIfNotExists = true) { ArgumentNullException.ThrowIfNull(builder); ArgumentNullException.ThrowIfNull(createContainer); @@ -79,7 +89,7 @@ public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClie { var container = createContainer(provider) ?? throw new InvalidOperationException($"{Describe(clientName)} keeps large messages in Blob storage, but the delegate passed to UseBlobStorageForLargeMessages returned no container."); - return new BlobLargeMessageStore(container, createContainerIfNotExists: true, provider.GetService()); + return new BlobLargeMessageStore(container, createContainerIfNotExists, provider.GetService()); }, thresholdBytes); } @@ -87,7 +97,7 @@ public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClie /// /// Keeps bodies too big for a queue message as blobs in the container , in the same /// storage account as the queue and reached with the same credentials. The container is created on the first large - /// send if it does not exist. + /// send if it does not exist (unless is false). /// /// /// The Blob endpoint is worked out from the client's settings the first time the client is used: a connection @@ -108,9 +118,13 @@ public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClie /// Also send bodies of at least this many bytes to Blob storage even when they would fit on the queue. Null (the /// default) sends only what would otherwise be over the 64 KiB limit. /// + /// + /// True (the default) creates the container on the first large send if it does not exist. Pass false when the + /// container is created ahead of time and the identity is not allowed to create containers. + /// /// The same builder, for chaining. /// is not a valid container name. - public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClientBuilder builder, string containerName = DefaultContainerName, int? thresholdBytes = null) + public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClientBuilder builder, string containerName = DefaultContainerName, int? thresholdBytes = null, bool createContainerIfNotExists = true) { ArgumentNullException.ThrowIfNull(builder); ArgumentNullException.ThrowIfNull(containerName); @@ -123,7 +137,8 @@ public static IQueueClientBuilder UseBlobStorageForLargeMessages(this IQueueClie var settings = provider.GetRequiredService>().Get(clientName); return BlobEndpoints.ServiceClientFor(settings, Describe(clientName)).GetBlobContainerClient(containerName); }, - thresholdBytes); + thresholdBytes, + createContainerIfNotExists); } private static string Describe(string clientName) => diff --git a/src/AzureStorage.QueueService.Blobs/README.md b/src/AzureStorage.QueueService.Blobs/README.md index 2392de6..25999a9 100644 --- a/src/AzureStorage.QueueService.Blobs/README.md +++ b/src/AzureStorage.QueueService.Blobs/README.md @@ -27,7 +27,7 @@ builder.Services.AddAzureStorageQueueClient(queues => }); ``` -The first form works out the Blob endpoint from the queue client's own settings: a connection string is used as is, `https://myaccount.queue.core.windows.net/...` becomes `https://myaccount.blob.core.windows.net/`, and the storage emulator works too. For a custom domain, use the second or third form. Every form creates the container the first time a large message is sent, and takes an optional `thresholdBytes` to send smaller bodies to Blob storage as well. +The first form works out the Blob endpoint from the queue client's own settings: a connection string is used as is, `https://myaccount.queue.core.windows.net/...` becomes `https://myaccount.blob.core.windows.net/`, and the storage emulator works too. For a custom domain, use the second or third form. Every form creates the container the first time a large message is sent (pass `createContainerIfNotExists: false` if it already exists), and takes an optional `thresholdBytes` to send smaller bodies to Blob storage as well. Large messages travel in the library's envelope, which only 3.1 and later read: upgrade every consumer of the queue before turning this on. @@ -37,7 +37,7 @@ Most blobs are deleted automatically, but some are not: a message that expires o ## Permissions -The identity the client connects with needs **Storage Blob Data Contributor** on the container (or the storage account) to create the container and to write, read and delete blobs. A connection string with the account key needs nothing more. If the container is created ahead of time, create the store yourself with `new BlobLargeMessageStore(container, createContainerIfNotExists: false)` and attach it with `UseLargeMessageStore(store)`. +The identity the client connects with needs **Storage Blob Data Contributor** on the container (or the storage account) to create the container and to write, read and delete blobs. A connection string with the account key needs nothing more. An identity that isn't allowed to create containers needs the container created ahead of time, and `createContainerIfNotExists: false` on the registration, for example `.UseBlobStorageForLargeMessages("order-bodies", createContainerIfNotExists: false)`. ## Security diff --git a/tests/AzureStorage.QueueService.Blobs.Tests/BlobLargeMessageEndToEndTests.cs b/tests/AzureStorage.QueueService.Blobs.Tests/BlobLargeMessageEndToEndTests.cs index 97c879b..6488d5d 100644 --- a/tests/AzureStorage.QueueService.Blobs.Tests/BlobLargeMessageEndToEndTests.cs +++ b/tests/AzureStorage.QueueService.Blobs.Tests/BlobLargeMessageEndToEndTests.cs @@ -1,3 +1,4 @@ +using Azure; using Azure.Storage.Blobs; using Azure.Storage.Blobs.Models; using FluentAssertions; @@ -101,4 +102,18 @@ public async Task Dead_Lettered_Message_Keeps_Its_Blob() deadLettered.Error.Should().BeNull(); deadLettered.Body.Should().Be(sent); } + + [Fact(DisplayName = "Blob storage: with createContainerIfNotExists: false a missing container is not created, and the send fails")] + public async Task Does_Not_Create_Container_When_Told_Not_To() + { + var missing = new BlobContainerClient(AzuriteContainer.ConnectionString, $"azqs-missing-{Guid.NewGuid():N}"); + using var host = BuildHost(new FakeTimeProvider(), queues => + queues.AddClient("audits", settings => { settings.ConnectionString = AzuriteContainer.ConnectionString; settings.QueueName = "audits"; }) + .UseBlobStorageForLargeMessages(missing.Name, createContainerIfNotExists: false)); + + var send = () => host.Services.GetRequiredKeyedService("audits").SendAsync(Large("r-3")); + + (await send.Should().ThrowAsync()).Which.Status.Should().Be(404); + (await missing.ExistsAsync()).Value.Should().BeFalse("an identity without permission to create containers must not try"); + } } From 579859ae32367946e6688b24979c7be41d4d33f2 Mon Sep 17 00:00:00 2001 From: Jason Shave Date: Fri, 16 Oct 2026 15:28:05 -0600 Subject: [PATCH 2/3] 3.1: own SendResult/QueueStatus records, RenewVisibilityAsync, QueueServiceBuilder, camelCase JSON by default The four changes #51 held back for 4.0 go into 3.1 as clean breaks, since 3.0 shipped with negligible uptake: - IQueueSender.SendAsync returns SendResult and IStorageQueue.GetPropertiesAsync returns QueueStatus instead of the Azure SDK's SendReceipt and QueueProperties, so fakes no longer need QueuesModelFactory. Both transports convert at the edge. - QueueMessage/IQueueMessage.RenewLockAsync is renamed RenewVisibilityAsync. - QueueClientSettingsBuilder is renamed QueueServiceBuilder. - The default serializer uses JsonSerializerDefaults.Web: camelCase out, case-insensitive in. Options passed by the application are used as given. README lists them as breaking changes from 3.0 with the migration for each; v3-design.md records why they moved into 3.1. Co-Authored-By: Claude Opus 5.5 --- README.md | 30 +++++++++++-------- docs/v3-design.md | 4 +-- src/AzureStorage.QueueService/IQueueSender.cs | 16 ++++++++-- .../IStorageQueue.cs | 14 +++++++-- .../InMemoryQueueStore.cs | 8 ++--- .../JsonQueueMessageSerializer.cs | 8 ++--- .../QueueClientBuilderExtensions.cs | 4 +-- src/AzureStorage.QueueService/QueueMessage.cs | 6 ++-- .../QueueProcessor.cs | 2 +- ...tingsBuilder.cs => QueueServiceBuilder.cs} | 18 +++++------ .../QueueTransport.cs | 22 +++++++++----- .../ServiceCollectionExtensions.cs | 8 ++--- src/AzureStorage.QueueService/StorageQueue.cs | 6 ++-- .../IntegrationTests.cs | 4 +-- .../BlobLargeMessageEndToEndTests.cs | 2 +- .../ProcessorFeatureTests.cs | 2 +- .../QueueClientBuilderTests.cs | 5 ++-- .../QueueClientFactoryTests.cs | 3 ++ .../QueueFeatureTests.cs | 17 +++++++---- .../QueueProcessorHostTests.cs | 5 ++-- .../QueueProcessorTests.cs | 8 ++--- .../StorageQueueTests.cs | 11 +++---- 22 files changed, 121 insertions(+), 82 deletions(-) rename src/AzureStorage.QueueService/{QueueClientSettingsBuilder.cs => QueueServiceBuilder.cs} (94%) diff --git a/README.md b/README.md index f92c96c..da44e49 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ Still on .NET 8? Use the 2.x package line, which stays on .NET 8 and keeps recei - **Process** with `AddProcessor()`: the library runs the loop — backoff, lease renewal, retries, a poison queue, cancellation and graceful shutdown — and hands each message to a handler resolved from its own dependency-injection scope. Handlers can be classes or lambdas, share middleware, and share one queue between several message types. - **Look after the poison queue** with `IDeadLetterQueue`: see what was dead-lettered and why, and move messages back once the cause is fixed. -Upgrading? See [Upgrading from 3.0 to 3.1](#upgrading-from-30-to-31) for the behaviour changes, or [Upgrading from 2.x](#upgrading-from-2x) for the mapping from the old API. +Upgrading? See [Upgrading from 3.0 to 3.1](#upgrading-from-30-to-31) for the breaking and behaviour changes, or [Upgrading from 2.x](#upgrading-from-2x) for the mapping from the old API. ## Install @@ -45,7 +45,7 @@ builder.Services.AddAzureStorageQueueClient(queues => settings.TokenCredential = new DefaultAzureCredential(); settings.CreateIfNotExists = true; }) - .UseJsonSerializer(new JsonSerializerOptions { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }); + .UseJsonSerializer(new JsonSerializerOptions(JsonSerializerDefaults.Web) { DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull }); // Fail host startup if any client is missing its connection details queues.ValidateOnStart(); @@ -130,7 +130,7 @@ queues.AddClient("legacy", settings => }); ``` -Those senders usually write camelCase JSON. The library's default serializer reads property names case-insensitively, so `{"orderId": 7}` fills `OrderId`. It still writes PascalCase; use `UseJsonSerializer(new JsonSerializerOptions(JsonSerializerDefaults.Web))` to write camelCase too. +Those senders usually write camelCase JSON, which is also what the library's default serializer writes. It reads property names case-insensitively, so `{"orderId": 7}` and `{"OrderId": 7}` both fill `OrderId`. ### Client-side encryption @@ -171,7 +171,7 @@ await notifications.SendAsync(new Notification("tagged"), new SendOptions }); ``` -`SendOptions` takes a `Delay` (also called `VisibilityTimeout`: 0 to 7 days), a `TimeToLive` (`Timeout.InfiniteTimeSpan` for never; the service default is 7 days), `Headers`, and a `MessageType`. A delay the message would expire within, or a message over the service's 64 KiB limit, throws before anything is sent: `ArgumentOutOfRangeException` and `QueueMessageTooLargeException` respectively. The size is checked as the message would go on the queue, so Base64 encoding and the library's envelope count. The return value is the SDK's `SendReceipt`. Failures are logged and thrown. +`SendOptions` takes a `Delay` (also called `VisibilityTimeout`: 0 to 7 days), a `TimeToLive` (`Timeout.InfiniteTimeSpan` for never; the service default is 7 days), `Headers`, and a `MessageType`. A delay the message would expire within, or a message over the service's 64 KiB limit, throws before anything is sent: `ArgumentOutOfRangeException` and `QueueMessageTooLargeException` respectively. The size is checked as the message would go on the queue, so Base64 encoding and the library's envelope count. The return value is a `SendResult`: the message's id, its pop receipt (which can delete or update it until someone receives it), and when it was inserted, expires and can first be received. Failures are logged and thrown. `UseCompression()` on a client gzips bodies of 1 KiB or more (pass another threshold if you like), skipping any message compression would not make smaller. Repetitive JSON often shrinks by 80% or more, which also keeps more messages under the 64 KiB limit. @@ -180,7 +180,7 @@ await notifications.SendAsync(new Notification("tagged"), new SendOptions A storage queue message is only a body: no properties, no headers. Anything that has to travel with it — the trace context, headers, a compressed body, a body kept in Blob storage — goes in a small JSON envelope around the body. A message that needs none of these goes out exactly as the serializer wrote it, as in 3.0. ```json -{"$azqs-envelope":1,"traceparent":"00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01","headers":{"tenant":"contoso"},"body":{"OrderId":42}} +{"$azqs-envelope":1,"traceparent":"00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01","headers":{"tenant":"contoso"},"body":{"orderId":42}} ``` Every client from 3.1 on reads the envelope when it receives, whatever its own settings, so `Body` and `RawBody` are the message you sent and `Headers` holds the headers. The full format is under [Wire format](#wire-format). The catch is older readers: a consumer on 3.0 or earlier, or other code reading the queue (an Azure Function, a script), sees the envelope as the message and fails to deserialize it, or worse, deserializes it into an object with empty fields. So headers, compression, `PropagateTraceContext` and `SendMessageType` are all off until you use them; turn them on only once every reader of the queue is on 3.1 or later. @@ -250,7 +250,7 @@ While a message is being delivered, everything logged — by the library and by ### The message -`QueueMessage` carries `MessageId`, `QueueName`, `PopReceipt`, `DequeueCount`, `InsertedOn`, `ExpiresOn`, `NextVisibleOn`, `Headers`, `IsLastAttempt`, the `RawBody`, the typed `Body`, and `Error` when the body could not be read. `CompleteAsync` deletes it; `AbandonAsync(delay)` makes it visible again now or later; `RenewLockAsync(timeout)` extends the lease; `DeadLetterAsync(reason)` moves it to the poison queue; `Settlement` records which of those happened. Timeouts the service would refuse throw `ArgumentOutOfRangeException` before any call. `IQueueMessage` is the same message without its body's type, for code that serves every message type. +`QueueMessage` carries `MessageId`, `QueueName`, `PopReceipt`, `DequeueCount`, `InsertedOn`, `ExpiresOn`, `NextVisibleOn`, `Headers`, `IsLastAttempt`, the `RawBody`, the typed `Body`, and `Error` when the body could not be read. `CompleteAsync` deletes it; `AbandonAsync(delay)` makes it visible again now or later; `RenewVisibilityAsync(timeout)` extends the lease; `DeadLetterAsync(reason)` moves it to the poison queue; `Settlement` records which of those happened. Timeouts the service would refuse throw `ArgumentOutOfRangeException` before any call. `IQueueMessage` is the same message without its body's type, for code that serves every message type. ### Error policy @@ -355,7 +355,6 @@ builder.Services.AddAzureStorageQueueClient(queues => // Event Grid Base64-encodes what it writes to a storage queue unless the subscription says otherwise settings.ClientOptions = new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 }; }) - .UseJsonSerializer(new JsonSerializerOptions(JsonSerializerDefaults.Web)) // event data is usually camelCase .AddCloudEventProcessor()); ``` @@ -370,7 +369,7 @@ A subscription that delivers several event types goes to a router, which picks t It is the same processor underneath, with the same options and behaviour: return to complete, throw to retry, settle through `message` yourself if you want to. A few things are specific to events: - **Base64.** Event Grid's storage queue destination Base64-encodes each message by default. Set `MessageEncoding = QueueMessageEncoding.Base64` as above, or turn off Base64 encoding on the event subscription and leave the client as it is. -- **The data is read with this client's serializer,** so its naming policy applies. An event with no data reaches the handler with `data` null. +- **The data is read with this client's serializer,** so its naming policy applies. The default reads Event Grid's camelCase data as it is. An event with no data reaches the handler with `data` null. - **An event whose data cannot be read as `TData`** is treated like any other unreadable message: moved to the poison queue without calling the handler, with the event id and type in the reason. With `DeadLetterOnDeserializationFailure` off, the handler is called with `data` null and `message.Error` set. A router dead-letters an event of a type nothing is routed for. - **Binary data** (`data_base64`, or any data that is not JSON): use `BinaryData` as `TData` to get the bytes as they are. - **Event Grid's own schema** (`EventGridEvent`, Event Grid's default schema) is not supported; it needs the `Azure.Messaging.EventGrid` package. Choose the CloudEvents v1.0 schema when you create the subscription. @@ -444,7 +443,7 @@ Things to know before turning it on: ## Testing your code -Depend on `IQueueSender` and `IQueueReceiver` and fake them. A received message can be built in a test: `QueueMessage` has a public constructor that takes a `QueueMessageInfo` (whose optional `QueueName`, `Headers` and `IsLastAttempt` set those properties), the body and an `IQueueMessageOperations` — stub the operations to assert that your code completed or abandoned the message. +Depend on `IQueueSender` and `IQueueReceiver` and fake them. What they return are plain records, so a fake builds them directly: `new SendResult(...)` from `SendAsync`, `new QueueStatus(...)` from `GetPropertiesAsync`. A received message can be built in a test: `QueueMessage` has a public constructor that takes a `QueueMessageInfo` (whose optional `QueueName`, `Headers` and `IsLastAttempt` set those properties), the body and an `IQueueMessageOperations` — stub the operations to assert that your code completed or abandoned the message. To test a handler end to end — send, processor, handler, then complete, retry or dead-letter — swap the real queues for in-memory ones. No storage account, emulator or connection string is needed, even with `ValidateOnStart()`; each client still needs a `QueueName`, or an `EndpointUri` that names the queue, checked by the same rules as in production. The in-memory queues run the same queue code as the real ones, so messages go through each client's serializer, envelope and size check, emit the same spans and metrics, and follow the service's rules: a received message is hidden for its visibility timeout and its dequeue count goes up, an abandoned one stays hidden for its retry delay, and settling with an outdated pop receipt throws a `RequestFailedException` (404). Time comes from the registered `TimeProvider`, so a `FakeTimeProvider` drives polling, retries and expiry without real waits. Clients naming the same queue share it. @@ -558,14 +557,21 @@ Lambda handlers are fine: their parameters are read from the delegate itself. ## Upgrading from 3.0 to 3.1 -3.1 adds to the API without removing anything. Recompile and these are the behaviour changes to know about: +3.1 has five breaking changes from 3.0. They were planned for 4.0, but 3.0 had been out only briefly and had almost no users, so they were made now rather than kept waiting for a major version. Each one is a small change to your code: + +- **`SendAsync` returns a `SendResult` instead of the Azure SDK's `SendReceipt`.** It carries the same information. `MessageId` and `PopReceipt` keep their names; `InsertionTime` is now `InsertedOn`, `ExpirationTime` is `ExpiresOn`, and `TimeNextVisible` is `NextVisibleOn`. A fake `IQueueSender` returns `new SendResult(...)` instead of building a receipt with `QueuesModelFactory`. +- **`GetPropertiesAsync` returns a `QueueStatus` instead of the SDK's `QueueProperties`.** `ApproximateMessagesCount` and `Metadata` keep their names; `Metadata` is now read-only. A fake returns `new QueueStatus(count, metadata)`. +- **`RenewLockAsync` is now `RenewVisibilityAsync`**, on `QueueMessage` and `IQueueMessage`, to match the processor's `RenewVisibility` option. Rename the call. +- **`QueueClientSettingsBuilder` is now `QueueServiceBuilder`:** the type of `queues` in `AddAzureStorageQueueClient(queues => ...)`. Only code that names the type, such as a helper that takes an `Action`, needs the new name. +- **The default serializer writes camelCase JSON:** `{"orderId":42}` where 3.0 wrote `{"OrderId":42}`. It now uses System.Text.Json's web defaults (`JsonSerializerDefaults.Web`), which also read property names case-insensitively, so messages from 3.0 senders still arrive intact, and a camelCase message no longer fills none of a PascalCase type's properties and gets completed as if it were fine. Numbers written as strings (`"42"`) are read too. What changes is what other readers of the queue see: an Azure Function, script or service that expects PascalCase names gets empty fields. To keep writing PascalCase, give the client `UseJsonSerializer(new JsonSerializerOptions { PropertyNameCaseInsensitive = true })`, or pass those options as `serializerOptions` to `AddAzureStorageQueueClient` for every client. Options you pass are used exactly as given. + +Recompile, and these are the behaviour changes to know about: -- **The default serializer reads JSON property names case-insensitively.** A camelCase message used to fill none of a PascalCase type's properties, and was completed as if it were fine; now it fills them. Pass your own `JsonSerializerOptions` to keep the old strictness. Output is unchanged. - **Poison messages no longer expire after 7 days.** They used to take the service's default time to live and could disappear unseen; now they stay until deleted or replayed, or for `PoisonTimeToLive`. - **A poison message records why it was dead-lettered**, so the copy on the poison queue is now an envelope around the original body. A 3.1 reader (and `IDeadLetterQueue`) reads it; a tool that read poison messages as plain JSON sees the envelope. Moving one back by hand still works for 3.1 receivers, but `ReplayAsync` is the clean way: it strips the record. - **One receive loop per processor** instead of one per handler slot. An idle processor makes one request per interval rather than `MaxConcurrentMessages` of them, and a burst after a quiet spell is picked up at the next poll instead of whenever each sleeping worker woke. - **Quieter logs.** Healthy messages log at Debug, not Information. A failed poll is logged once, not twice. A handler failure is a warning while retries remain and an error on the last attempt. -- **Arguments are checked before calling the service:** `ReceiveAsync`, `AbandonAsync`, `RenewLockAsync` and `SendOptions` throw `ArgumentOutOfRangeException` where they used to get a 400 back, and a send over 64 KiB throws `QueueMessageTooLargeException` instead of a 413. +- **Arguments are checked before calling the service:** `ReceiveAsync`, `AbandonAsync`, `RenewVisibilityAsync` and `SendOptions` throw `ArgumentOutOfRangeException` where they used to get a 400 back, and a send over 64 KiB throws `QueueMessageTooLargeException` instead of a 413. - **In-memory queues behave more like the service:** they enforce the size limit and emit spans and metrics. - **`IStorageQueue`, `IQueueSender` and `IQueueReceiver` are always registered,** with or without a default client, so injecting one without a default client fails with a message naming the registered clients. Code that checked `GetService() is null` to detect a default client should check its configuration instead. - **The processing-duration and size histograms have explicit buckets** (seconds and bytes) and units. Dashboards built on the old default buckets will see the bucket boundaries change once. diff --git a/docs/v3-design.md b/docs/v3-design.md index 9160d8d..c801af4 100644 --- a/docs/v3-design.md +++ b/docs/v3-design.md @@ -277,7 +277,7 @@ Previews publish from `v3.0.0-preview.N` tags with the existing `nuget.yml`; nug ## 3.1 (issue #51) -Issue #51 collected a review of 3.0 against Azure Functions queue triggers, the Service Bus processor, MassTransit, Wolverine, Rebus and SQS libraries. 3.1 delivered all of it except the 4.0 candidates (breaking changes) and the items the review chose to skip. The decisions that shape the code: +Issue #51 collected a review of 3.0 against Azure Functions queue triggers, the Service Bus processor, MassTransit, Wolverine, Rebus and SQS libraries. 3.1 delivered all of it, the 4.0 candidates included (see below), except the items the review chose to skip. The decisions that shape the code: - **One envelope format, with forward-compatibility rules.** 3.0's trace-context wrapper became the general envelope (`MessageEnvelope`): trace context, headers, compression, a reference to a body kept elsewhere, and the dead-letter record, each optional. A reader skips properties it does not know; anything that changes how the body is read uses a new `encoding` value, a new body kind or a new version, which an older reader refuses (the message gets `Error` and is dead-lettered intact) instead of reading it as a plain message, which System.Text.Json would turn into an object with empty fields. The marker must still be the first property, so an ordinary payload is never mistaken for an envelope. - **One queue implementation over two transports.** `StorageQueue` now sits on `IQueueTransport`, implemented over the SDK client and over the in-memory queue. Encoding, size checks, dead-lettering, replay and telemetry exist once, and a test on in-memory queues runs the production path. This replaced `InMemoryStorageQueue` and `StorageQueueCommon`. @@ -286,5 +286,5 @@ Issue #51 collected a review of 3.0 against Azure Functions queue triggers, the - **Compression lives in the envelope, not in a serializer wrapper.** The issue suggested a serializer wrapper; putting it in the envelope means every 3.1 receiver decompresses without configuration, the same way it reads trace context. - **Dead-letter categories are a bounded set** (`max_dequeue`, `deserialization`, `handler`, `unroutable`) because they are a metric tag and a replay filter. - **Configuration binding is split.** Processor options use the source-generated binder (trim-safe). Client settings keep reflection binding, because the generator cannot create or fill `QueueClientOptions`, which would silently drop `"ClientOptions": { "MessageEncoding": "Base64" }`; those overloads are marked as not trim-safe. -- **Behaviour change taken:** the default serializer reads property names case-insensitively. Writing camelCase by default stays a 4.0 candidate. +- **The 4.0 candidates moved into 3.1.** The issue held back four breaking changes for 4.0: the library's own `SendResult` and `QueueStatus` records instead of the SDK's `SendReceipt` and `QueueProperties` (so a fake needs no `QueuesModelFactory`), `RenewLockAsync` renamed `RenewVisibilityAsync` to match the option, `QueueClientSettingsBuilder` renamed `QueueServiceBuilder` so it no longer reads like a sibling of `IQueueClientBuilder`, and camelCase output from the default serializer (`JsonSerializerDefaults.Web`, which also reads names case-insensitively). 3.0 had shipped only just before with negligible uptake, so they were made as clean changes in 3.1, with no aliases or `[Obsolete]` forwarders, rather than carried until a major version. The README lists them as breaking changes from 3.0. - **Separate packages** for Blob storage (`AzureStorage.QueueService.Blobs`, the claim-check store) and health checks (`AzureStorage.QueueService.HealthChecks`), so the core takes neither dependency. Both build on public seams in the core: `ILargeMessageStore` and `IQueueProcessorController`. diff --git a/src/AzureStorage.QueueService/IQueueSender.cs b/src/AzureStorage.QueueService/IQueueSender.cs index 26eefbd..22faa22 100644 --- a/src/AzureStorage.QueueService/IQueueSender.cs +++ b/src/AzureStorage.QueueService/IQueueSender.cs @@ -1,5 +1,3 @@ -using Azure.Storage.Queues.Models; - namespace AzureStorage.QueueService; /// @@ -8,12 +6,24 @@ namespace AzureStorage.QueueService; public interface IQueueSender : INamedQueue { /// Serializes and puts it on the queue. + /// What the service said about the message it stored: its id, receipt and times. /// asks for a delay or time to live the service would reject. /// The message would be over the service's 64 KiB limit and the client has no large message store. /// The service rejected the send. - ValueTask SendAsync(T message, SendOptions? options = null, CancellationToken cancellationToken = default); + ValueTask SendAsync(T message, SendOptions? options = null, CancellationToken cancellationToken = default); } +/// +/// The message a send put on the queue. A plain record, so a fake can return one with +/// new SendResult(...). +/// +/// The id the service gave the message. +/// The receipt that can delete or update the message until someone receives it. +/// When the message was put on the queue. +/// When the service will discard the message if nobody deletes it first. +/// When the message can first be received: now, or later when the send asked for a delay. +public sealed record SendResult(string MessageId, string PopReceipt, DateTimeOffset InsertedOn, DateTimeOffset ExpiresOn, DateTimeOffset NextVisibleOn); + /// How one message is sent. /// /// How long the message stays hidden after it is sent (0 to 7 days, and less than ); by diff --git a/src/AzureStorage.QueueService/IStorageQueue.cs b/src/AzureStorage.QueueService/IStorageQueue.cs index bea4c64..484e9fd 100644 --- a/src/AzureStorage.QueueService/IStorageQueue.cs +++ b/src/AzureStorage.QueueService/IStorageQueue.cs @@ -1,5 +1,3 @@ -using Azure.Storage.Queues.Models; - namespace AzureStorage.QueueService; /// @@ -18,5 +16,15 @@ public interface IStorageQueue : IQueueSender, IQueueReceiver /// The queue's metadata and approximate message count. The count includes messages that are hidden (being /// processed, or waiting for a retry), so it is not the number a receive would return. /// - ValueTask GetPropertiesAsync(CancellationToken cancellationToken = default); + ValueTask GetPropertiesAsync(CancellationToken cancellationToken = default); } + +/// +/// A queue's approximate message count and metadata, from . A plain record, +/// so a fake can return one with new QueueStatus(...). +/// +/// +/// Roughly how many messages are on the queue, hidden ones (being processed, or waiting for a retry) included. +/// +/// The name and value pairs stored on the queue itself; empty when it has none. +public sealed record QueueStatus(int ApproximateMessagesCount, IReadOnlyDictionary Metadata); diff --git a/src/AzureStorage.QueueService/InMemoryQueueStore.cs b/src/AzureStorage.QueueService/InMemoryQueueStore.cs index f3da28d..85227c0 100644 --- a/src/AzureStorage.QueueService/InMemoryQueueStore.cs +++ b/src/AzureStorage.QueueService/InMemoryQueueStore.cs @@ -53,7 +53,7 @@ public InMemoryQueue(string name, TimeProvider time) /// Takes the time to live as the service does: -1 second for never. The client converts .NET's -1 ms sentinel first /// (), as the real client does before calling the service. /// - public SendReceipt Send(BinaryData body, TimeSpan? visibilityTimeout, TimeSpan? timeToLive) + public SendResult Send(BinaryData body, TimeSpan? visibilityTimeout, TimeSpan? timeToLive) { var visibility = WholeSeconds(visibilityTimeout ?? TimeSpan.Zero); var ttl = WholeSeconds(timeToLive ?? DefaultTimeToLive); @@ -71,7 +71,7 @@ public SendReceipt Send(BinaryData body, TimeSpan? visibilityTimeout, TimeSpan? NextVisibleOn = now + visibility, }; _messages.Add(entry); - return QueuesModelFactory.SendReceipt(entry.MessageId, entry.InsertedOn, entry.ExpiresOn, entry.PopReceipt, entry.NextVisibleOn); + return new SendResult(entry.MessageId, entry.PopReceipt, entry.InsertedOn, entry.ExpiresOn, entry.NextVisibleOn); } } @@ -145,13 +145,13 @@ public void Clear() lock (_lock) _messages.Clear(); } - public QueueProperties GetProperties() + public QueueStatus GetProperties() { lock (_lock) { RemoveExpired(); // The service's count includes hidden messages; so does this one. - return QueuesModelFactory.QueueProperties(_metadata ?? new Dictionary(), _messages.Count); + return new QueueStatus(_messages.Count, new Dictionary(_metadata ?? new Dictionary())); } } diff --git a/src/AzureStorage.QueueService/JsonQueueMessageSerializer.cs b/src/AzureStorage.QueueService/JsonQueueMessageSerializer.cs index bc8237c..7334dfb 100644 --- a/src/AzureStorage.QueueService/JsonQueueMessageSerializer.cs +++ b/src/AzureStorage.QueueService/JsonQueueMessageSerializer.cs @@ -7,11 +7,11 @@ namespace AzureStorage.QueueService; internal sealed class JsonQueueMessageSerializer : IQueueMessageSerializer { /// - /// The library's defaults: System.Text.Json's, except that property names are read case-insensitively, so a camelCase - /// message from JavaScript, Azure Functions or Event Grid fills a PascalCase .NET type instead of arriving with empty - /// fields. Writing is unchanged (PascalCase). + /// The library's defaults: System.Text.Json's web defaults. Property names are written in camelCase, as JavaScript, + /// Azure Functions and Event Grid write them, and read case-insensitively, so a PascalCase message from an older + /// sender still fills its .NET type instead of arriving with empty fields. /// - public static readonly JsonSerializerOptions DefaultOptions = new() { PropertyNameCaseInsensitive = true }; + public static readonly JsonSerializerOptions DefaultOptions = new(JsonSerializerDefaults.Web); private readonly JsonSerializerOptions _options; diff --git a/src/AzureStorage.QueueService/QueueClientBuilderExtensions.cs b/src/AzureStorage.QueueService/QueueClientBuilderExtensions.cs index 0c60554..4f83d78 100644 --- a/src/AzureStorage.QueueService/QueueClientBuilderExtensions.cs +++ b/src/AzureStorage.QueueService/QueueClientBuilderExtensions.cs @@ -48,8 +48,8 @@ public static IQueueClientBuilder UseJsonSerializer(this IQueueClientBuilder bui /// /// Serializes this client's messages with System.Text.Json using source-generated metadata, for example - /// UseJsonSerializer(MyJsonContext.Default), which is what a trimmed or native AOT application needs. Property - /// names are read case-insensitively, as with the default serializer. + /// UseJsonSerializer(MyJsonContext.Default), which is what a trimmed or native AOT application needs. As with + /// the default serializer, property names are written in camelCase and read case-insensitively. /// public static IQueueClientBuilder UseJsonSerializer(this IQueueClientBuilder builder, System.Text.Json.Serialization.Metadata.IJsonTypeInfoResolver typeInfoResolver) { diff --git a/src/AzureStorage.QueueService/QueueMessage.cs b/src/AzureStorage.QueueService/QueueMessage.cs index 371ecc4..5ea899b 100644 --- a/src/AzureStorage.QueueService/QueueMessage.cs +++ b/src/AzureStorage.QueueService/QueueMessage.cs @@ -79,7 +79,7 @@ public interface IQueueMessage ValueTask AbandonAsync(TimeSpan? visibilityTimeout = null, CancellationToken cancellationToken = default); /// Keeps the message hidden for another (more than zero, up to 7 days), measured from now. - ValueTask RenewLockAsync(TimeSpan visibilityTimeout, CancellationToken cancellationToken = default); + ValueTask RenewVisibilityAsync(TimeSpan visibilityTimeout, CancellationToken cancellationToken = default); /// Moves the message to the poison queue, recording with it. ValueTask DeadLetterAsync(string? reason = null, CancellationToken cancellationToken = default); @@ -132,7 +132,7 @@ private QueueMessage(MessageLease lease, T? body, DeserializationException? erro /// public string QueueName => Lease.QueueName; - /// Changes after and . + /// Changes after and . public string PopReceipt => Lease.PopReceipt; /// How many times the message has been received, including this time. @@ -212,7 +212,7 @@ public ValueTask AbandonAsync(TimeSpan? visibilityTimeout = null, CancellationTo /// Keeps the message hidden for another , measured from now. /// is not more than zero, or is longer than 7 days. - public ValueTask RenewLockAsync(TimeSpan visibilityTimeout, CancellationToken cancellationToken = default) => + public ValueTask RenewVisibilityAsync(TimeSpan visibilityTimeout, CancellationToken cancellationToken = default) => Lease.RenewAsync(visibilityTimeout, cancellationToken); /// diff --git a/src/AzureStorage.QueueService/QueueProcessor.cs b/src/AzureStorage.QueueService/QueueProcessor.cs index 285c4ca..dc121c8 100644 --- a/src/AzureStorage.QueueService/QueueProcessor.cs +++ b/src/AzureStorage.QueueService/QueueProcessor.cs @@ -511,7 +511,7 @@ private async Task DeliverAsync(QueueMessage message, IOwnedReceiver? ? LeaseRenewal.Start( async () => { - await message.RenewLockAsync(_options.VisibilityTimeout, CancellationToken.None); + await message.RenewVisibilityAsync(_options.VisibilityTimeout, CancellationToken.None); _metrics.LeaseRenewals.Add(1, QueueTags()); }, () => message.IsSettled, diff --git a/src/AzureStorage.QueueService/QueueClientSettingsBuilder.cs b/src/AzureStorage.QueueService/QueueServiceBuilder.cs similarity index 94% rename from src/AzureStorage.QueueService/QueueClientSettingsBuilder.cs rename to src/AzureStorage.QueueService/QueueServiceBuilder.cs index 1938e77..bd136d5 100644 --- a/src/AzureStorage.QueueService/QueueClientSettingsBuilder.cs +++ b/src/AzureStorage.QueueService/QueueServiceBuilder.cs @@ -11,13 +11,13 @@ namespace AzureStorage.QueueService; /// Registers clients inside AddAzureStorageQueueClient(queues => ...). Each AddClient / /// AddDefaultClient returns a builder for that one client to attach a serializer or a processor to. /// -public class QueueClientSettingsBuilder +public class QueueServiceBuilder { internal IServiceCollection Services { get; } internal QueueClientSettingsRegistry Registry { get; } - internal QueueClientSettingsBuilder(IServiceCollection services, QueueClientSettingsRegistry registry) + internal QueueServiceBuilder(IServiceCollection services, QueueClientSettingsRegistry registry) { Services = services; Registry = registry; @@ -116,7 +116,7 @@ private static void Bind(IConfiguration configuration, QueueClientSettings setti /// poison queue name fails startup with a clear message instead of failing the first request. Applies to /// clients registered before and after this call. /// - public QueueClientSettingsBuilder ValidateOnStart() + public QueueServiceBuilder ValidateOnStart() { if (Registry.ValidateOnStart) return this; @@ -137,7 +137,7 @@ public QueueClientSettingsBuilder ValidateOnStart() /// own credential wins, and clients that use a connection string are left alone. Calling this again replaces the /// credential. /// - public QueueClientSettingsBuilder UseTokenCredential(TokenCredential credential) + public QueueServiceBuilder UseTokenCredential(TokenCredential credential) { ArgumentNullException.ThrowIfNull(credential); @@ -158,7 +158,7 @@ public QueueClientSettingsBuilder UseTokenCredential(TokenCredential credential) /// endpoint needs a credential, from or each client's own. Applies to default and /// named clients, registered before and after this call; calling it again replaces the account. /// - public QueueClientSettingsBuilder UseStorageAccount(string connectionStringOrServiceUri) + public QueueServiceBuilder UseStorageAccount(string connectionStringOrServiceUri) { ArgumentException.ThrowIfNullOrWhiteSpace(connectionStringOrServiceUri); @@ -176,7 +176,7 @@ public QueueClientSettingsBuilder UseStorageAccount(string connectionStringOrSer /// (https://account.queue.core.windows.net/). Clients connect with 's /// credential or their own. /// - public QueueClientSettingsBuilder UseStorageAccount(Uri serviceUri) + public QueueServiceBuilder UseStorageAccount(Uri serviceUri) { ArgumentNullException.ThrowIfNull(serviceUri); @@ -190,7 +190,7 @@ public QueueClientSettingsBuilder UseStorageAccount(Uri serviceUri) /// Runs around every message every client's processors deliver, resolved from the /// message's scope (registered as scoped unless you registered it yourself). See . /// - public QueueClientSettingsBuilder UseMiddleware<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] TMiddleware>() + public QueueServiceBuilder UseMiddleware<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors)] TMiddleware>() where TMiddleware : class, IQueueMessageMiddleware { Services.TryAddScoped(); @@ -199,7 +199,7 @@ public QueueClientSettingsBuilder UseStorageAccount(Uri serviceUri) } /// Runs around every message every client's processors deliver. See . - public QueueClientSettingsBuilder Use(Func middleware) + public QueueServiceBuilder Use(Func middleware) { ArgumentNullException.ThrowIfNull(middleware); @@ -214,7 +214,7 @@ public QueueClientSettingsBuilder Use(Func /// diff --git a/src/AzureStorage.QueueService/IQueueBatchHandler.cs b/src/AzureStorage.QueueService/IQueueBatchHandler.cs new file mode 100644 index 0000000..841d24b --- /dev/null +++ b/src/AzureStorage.QueueService/IQueueBatchHandler.cs @@ -0,0 +1,35 @@ +namespace AzureStorage.QueueService; + +/// +/// Handles a batch of messages for a processor registered with AddBatchProcessor<TMessage, THandler>(): up +/// to messages from one receive, in one call, so the handler can +/// write them to a database or call an API in bulk. A new instance is resolved from a new dependency-injection scope for +/// every batch. +/// +/// +/// +/// Every message in the batch is still its own message: its own lease renewal, dequeue count, retry delay and +/// settlement. Return normally and the processor completes every message you did not settle yourself. Throw and the +/// error policy (DeadLetterOn<TException>(), then ) is +/// applied to each message you had not settled, which otherwise is abandoned and retried after +/// for its own dequeue count. Settle individual messages with +/// , or +/// (for example, dead-letter one bad item and +/// return) and the processor leaves those alone. +/// +/// +/// Messages whose body cannot be read, and messages delivered more than +/// times, are dead-lettered on their own and left out, so the +/// batch can be smaller than what was received; when none are left the handler is not called. With +/// off, an unreadable message stays in the +/// batch with null and set. Middleware does not +/// run for batches. The token is cancelled when the host shuts down. +/// +/// +public interface IQueueBatchHandler +{ + /// Handles one batch. + /// At least one message, each with its body read as , its identity, and the operations that settle it. + /// Cancelled when the host shuts down and the drain timeout has passed. + ValueTask HandleAsync(IReadOnlyList> messages, CancellationToken cancellationToken); +} diff --git a/src/AzureStorage.QueueService/IQueueMessageHandler.cs b/src/AzureStorage.QueueService/IQueueMessageHandler.cs index 9d060c4..1c55e3c 100644 --- a/src/AzureStorage.QueueService/IQueueMessageHandler.cs +++ b/src/AzureStorage.QueueService/IQueueMessageHandler.cs @@ -9,7 +9,7 @@ namespace AzureStorage.QueueService; /// Return normally and the processor completes the message; throw and it is abandoned and retried later; or settle /// it yourself with , or /// and the processor leaves it alone. With -/// off, +/// off, /// can be null with set. The token is cancelled when the host shuts down. /// public interface IQueueMessageHandler diff --git a/src/AzureStorage.QueueService/LoggerMessageExtensions.cs b/src/AzureStorage.QueueService/LoggerMessageExtensions.cs index 856ea61..45f7e9a 100644 --- a/src/AzureStorage.QueueService/LoggerMessageExtensions.cs +++ b/src/AzureStorage.QueueService/LoggerMessageExtensions.cs @@ -113,6 +113,19 @@ internal static partial class LoggerMessageExtensions public static IDisposable? BeginDeliveryScope(this ILogger logger, string queueName, string messageId, long dequeueCount) => DeliveryScope(logger, queueName, messageId, dequeueCount); + private static readonly Func BatchDeliveryScope = + LoggerMessage.DefineScope("Queue {QueueName}, batch of {MessageCount} message(s): {MessageIds}"); + + /// A scope for everything logged while one batch is delivered, the handler's own logs included. + public static IDisposable? BeginBatchDeliveryScope(this ILogger logger, string queueName, int messageCount, string messageIds) => + BatchDeliveryScope(logger, queueName, messageCount, messageIds); + + [LoggerMessage(46, LogLevel.Warning, "Processor {Processor}: the handler for a batch of {MessageCount} message(s) has been running for {Elapsed}, longer than LongRunningHandlerWarning ({Threshold}). It may be hung; their leases are still being renewed{RenewalNote}.")] + public static partial void LogProcessorBatchHandlerLongRunning(this ILogger logger, string processor, int messageCount, TimeSpan elapsed, TimeSpan threshold, string renewalNote); + + [LoggerMessage(47, LogLevel.Warning, "Queue '{QueueName}': the batch handler settled all {MessageCount} message(s) of its batch and then threw. The settlements stand.")] + public static partial void LogBatchHandlerFailedAfterSettling(this ILogger logger, string? queueName, int messageCount, Exception exception); + [LoggerMessage(45, LogLevel.Warning, "{Client}: could not make a private copy of its QueueClientOptions, so the Base64 decoding handler is added to the caller's options while the SDK client is built, then removed. A QueueClient the application builds from the same options object at that moment may pick it up. Logged once.")] public static partial void LogClientOptionsCopyFailed(this ILogger logger, string client, Exception exception); } diff --git a/src/AzureStorage.QueueService/QueueBatchProcessor.cs b/src/AzureStorage.QueueService/QueueBatchProcessor.cs new file mode 100644 index 0000000..481650f --- /dev/null +++ b/src/AzureStorage.QueueService/QueueBatchProcessor.cs @@ -0,0 +1,210 @@ +using AzureStorage.QueueService.Telemetry; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using System.Diagnostics; + +namespace AzureStorage.QueueService; + +/// +/// The hosted processor behind AddBatchProcessor<TMessage, THandler>(). The receive loop, start, stop, +/// pausing and status are 's, with a batch where the single-message +/// processor has a message: each receive fills one or more free batch slots with up to +/// messages each, and each batch is delivered on its own task. +/// +/// +/// Each message keeps its own lease renewal, poison checks, error policy and settlement; the batch shares one scope, +/// one handler call, one span (linked to every sender's) and one log scope. Middleware is per message, so it does not +/// run here. +/// +internal sealed class QueueBatchProcessor : QueueProcessorBase + where THandler : class, IQueueBatchHandler +{ + private readonly string _optionsName; + private readonly IOptionsFactory _optionsFactory; + private QueueBatchProcessorOptions _typedOptions = new(); + + public QueueBatchProcessor( + string clientName, + string optionsName, + IQueueClientFactory factory, + IServiceScopeFactory scopes, + IOptionsFactory optionsFactory, + ILoggerFactory loggerFactory, + QueueServiceMetrics metrics, + TimeProvider time, + IHostApplicationLifetime? lifetime = null, + QueueProcessorController? controller = null) + : base("QueueBatchProcessor", clientName, typeof(THandler).Name, factory, scopes, loggerFactory, metrics, time, lifetime, controller, readBody: null, afterDeserialize: null) + { + _optionsName = optionsName; + _optionsFactory = optionsFactory; + } + + private protected override QueueProcessorOptionsBase CreateOptions() => _typedOptions = _optionsFactory.Create(_optionsName); + + private protected override int MessagesPerDelivery => _typedOptions.MaxBatchSize; + + private protected override int MaxConcurrentDeliveries => _typedOptions.MaxConcurrentBatches; + + // ------------------------------------------------------------------------------------------------------------ + // One batch: every message's lifecycle, with one handler call and one exception boundary. Never throws. + // ------------------------------------------------------------------------------------------------------------ + + private protected override async Task DeliverAsync(IReadOnlyList> received, IOwnedReceiver? owned, long started) + { + var count = received.Count; + var outcomes = Enumerable.Repeat("abandoned", count).ToArray(); + var renewals = new LeaseRenewal?[count]; + Exception? failure = null; + + using var logScope = _logger.BeginBatchDeliveryScope(QueueName, count, string.Join(", ", received.Select(m => m.MessageId))); + + var tags = QueueServiceDiagnostics.OperationTags(QueueName, QueueServiceDiagnostics.Messaging.Process); + tags.Add(QueueServiceDiagnostics.Messaging.BatchMessageCount, count); + using Activity? activity = QueueServiceDiagnostics.Activities.StartProcessingBatchActivity(tags, received.Select(m => m.CreationContext)); + + _metrics.ActiveHandlers.Add(count, QueueTags()); + + using var longRunning = _options.LongRunningHandlerWarning is { } threshold + ? _time.CreateTimer(_ => _logger.LogProcessorBatchHandlerLongRunning(_name, count, _time.GetElapsedTime(started), threshold, RenewalNote()), null, threshold, Timeout.InfiniteTimeSpan) + : null; + + try + { + // Every lease is armed before anything else, downloading and deserializing included, so nothing that happens + // to one message can let another's lease lapse. + for (var i = 0; i < count; i++) + { + var message = received[i]; + message.Lease.IsLastAttempt |= message.DequeueCount >= _options.MaxDequeueCount; + message.Lease.DeliveryActivity = activity; + RecordTimeInQueue(message); + renewals[i] = StartRenewal(message); + } + + // Read every body; the unreadable and the over-delivered are dead-lettered on their own and left out. + var batch = new List>(count); + var positions = new List(count); + for (var i = 0; i < count; i++) + { + var message = received[i]; + if (!await ReadBodyAsync(message, owned)) + { + // Cancelled while downloading a body at shutdown: give back this message, the ones still to be read + // and the ones already read. + await GiveBackAsync(received, renewals, outcomes, positions.Concat(Enumerable.Range(i, count - i))); + return; + } + + if (await RejectIfUnprocessableAsync(message, renewals[i]) is { } rejected) + { + outcomes[i] = rejected; + continue; + } + + batch.Add(message); + positions.Add(i); + } + + if (batch.Count == 0) return; + activity?.SetTag(QueueServiceDiagnostics.Messaging.BatchMessageCount, batch.Count); + + try + { + // Resolving the handler is inside the boundary: a constructor that throws is a failed batch too. + await using var scope = _scopes.CreateAsyncScope(); + var handler = scope.ServiceProvider.GetRequiredService(); + await handler.HandleAsync(batch, _handlerToken); + } + catch (OperationCanceledException) when (_handlerToken.IsCancellationRequested && batch.Any(m => !m.IsSettled)) + { + await GiveBackAsync(received, renewals, outcomes, positions); + return; + } + catch (Exception ex) when (batch.Any(m => !m.IsSettled)) + { + failure = ex; + activity.RecordFailure(ex); + foreach (var i in positions) + { + var message = received[i]; + await StopRenewalAsync(renewals[i]); + if (message.IsSettled) + { + (outcomes[i], _) = await SettleAfterReturnAsync(message, activity); + continue; + } + + _metrics.MessageFailures.Add(1, QueueTags()); + outcomes[i] = await HandleFailureAsync(message, ex); + } + + return; + } + catch (Exception ex) + { + // The handler settled every message and then threw. The settlements stand; report it as what it was: + // an exception on a span that did not fail, so no error status. + _logger.LogBatchHandlerFailedAfterSettling(QueueName, batch.Count, ex); + activity?.AddException(ex); + } + + foreach (var i in positions) + { + // Wait for any renewal already in flight so completion uses the receipt the service holds. + await StopRenewalAsync(renewals[i]); + (outcomes[i], var settleFailure) = await SettleAfterReturnAsync(received[i], activity); + failure ??= settleFailure; + } + } + catch (Exception ex) + { + // Anything the inner boundaries did not cover: scope disposal, an option callback, a logging sink. + // A delivery must never fault silently and leave its messages in limbo. + failure = ex; + activity.RecordFailure(ex); + for (var i = 0; i < count; i++) + { + var message = received[i]; + if (message.IsSettled) continue; + + _logger.LogProcessorDeliveryFailed(_name, message.MessageId, ex); + _metrics.MessageFailures.Add(1, QueueTags()); + await AbandonSafelyAsync(message, QueueProcessorOptionsBase.MaxRetryDelay); + outcomes[i] = "abandoned"; + } + } + finally + { + foreach (var renewal in renewals) await StopRenewalAsync(renewal); + _metrics.ActiveHandlers.Add(-count, QueueTags()); + + var elapsed = _time.GetElapsedTime(started).TotalSeconds; + foreach (var outcome in outcomes) RecordOutcome(outcome, elapsed); + _metrics.ProcessDuration.Record(elapsed, QueueServiceDiagnostics.MessagingTags(QueueName, QueueServiceDiagnostics.Messaging.Process, QueueServiceDiagnostics.Messaging.Process, failure)); + activity?.SetTag(QueueServiceDiagnostics.Names.Outcome, outcomes.Distinct().Count() == 1 ? outcomes[0] : "mixed"); + foreach (var message in received) message.Lease.DeliveryActivity = null; + } + } + + /// At shutdown: the messages at the handler had not settled go straight back. + private async Task GiveBackAsync(IReadOnlyList> received, LeaseRenewal?[] renewals, string[] outcomes, IEnumerable positions) + { + foreach (var i in positions) + { + var message = received[i]; + await StopRenewalAsync(renewals[i]); + if (message.IsSettled) + { + // The handler settled it before it saw the cancellation. + (outcomes[i], _) = await SettleAfterReturnAsync(message, null); + continue; + } + + _logger.LogProcessorHandlerCancelled(_name, message.MessageId); + await AbandonSafelyAsync(message, TimeSpan.Zero); + } + } +} diff --git a/src/AzureStorage.QueueService/QueueClientBuilderExtensions.cs b/src/AzureStorage.QueueService/QueueClientBuilderExtensions.cs index 4f83d78..9631af8 100644 --- a/src/AzureStorage.QueueService/QueueClientBuilderExtensions.cs +++ b/src/AzureStorage.QueueService/QueueClientBuilderExtensions.cs @@ -158,6 +158,60 @@ public static IQueueClientBuilder Use(this IQueueClientBuilder builder, Func + /// Runs a hosted batch processor for this client: each receive takes up to + /// messages of type and hands + /// them to one call of a new , resolved from the batch's own dependency-injection + /// scope; up to batches run at once. Every message keeps + /// its own lease renewal, poison checks, retries and settlement, exactly as with + /// ; see + /// for what returning and throwing mean. Middleware does not run for + /// batches. Registering the same handler on the same client again adds to its options; it does not start a second processor. + /// + public static IQueueClientBuilder AddBatchProcessor(this IQueueClientBuilder builder, Action? configure = null) + where THandler : class, IQueueBatchHandler + { + ArgumentNullException.ThrowIfNull(builder); + + builder.Services.TryAddScoped(); + var clientName = builder.Name; + // Its own options name, so a class that is both a message handler and a batch handler can be registered both ways. + var optionsName = $"{builder.Name}|batch:{typeof(THandler).FullName}"; + builder.Services.AddOptions(optionsName).Configure(configure ?? (_ => { })); + + if (!ServiceCollectionExtensions.GetOrAddRegistry(builder.Services).AddProcessor(optionsName)) return builder; + + builder.Services.AddSingleton(provider => new QueueBatchProcessor( + clientName, + optionsName, + provider.GetRequiredService(), + provider.GetRequiredService(), + provider.GetRequiredService>(), + provider.GetRequiredService(), + provider.GetRequiredService(), + provider.GetService() ?? TimeProvider.System, + provider.GetService(), + provider.GetService())); + + return builder; + } + + /// + /// As , + /// with the options bound from (for example + /// configuration.GetSection("Queues:Orders:BatchProcessor")) and then . + /// + public static IQueueClientBuilder AddBatchProcessor(this IQueueClientBuilder builder, IConfiguration configuration, Action? configure = null) + where THandler : class, IQueueBatchHandler + { + ArgumentNullException.ThrowIfNull(configuration); + return builder.AddBatchProcessor(options => + { + BindBatchProcessorOptions(configuration, options); + configure?.Invoke(options); + }); + } + /// /// Runs a hosted processor whose handler is a delegate, with its parameters filled per message: a parameter of type /// gets the body; or the @@ -190,7 +244,7 @@ public static IQueueClientBuilder AddProcessor(this IQueueClientBuilde /// resolved from its own dependency-injection scope. Everything else is : /// the same options, settlement, retries, poison queue, lease renewal and shutdown. An event whose data cannot be /// read as counts as an unreadable body (see - /// ). + /// ). /// public static IQueueClientBuilder AddCloudEventProcessor(this IQueueClientBuilder builder, Action? configure = null) where THandler : class, ICloudEventHandler @@ -263,6 +317,8 @@ public static IQueueClientBuilder AddCloudEventRouter(this IQueueClientBuilder b // and RateLimiter), as reflection binding would. private static void BindProcessorOptions(IConfiguration configuration, QueueProcessorOptions options) => configuration.Bind(options); + private static void BindBatchProcessorOptions(IConfiguration configuration, QueueBatchProcessorOptions options) => configuration.Bind(options); + private static string Describe(IQueueClientBuilder builder) => builder.Name.Length == 0 ? "The default queue client" : $"Queue client '{builder.Name}'"; // Options are named per client + handler so the same handler type can serve two queues with different settings. diff --git a/src/AzureStorage.QueueService/QueueClientSettings.cs b/src/AzureStorage.QueueService/QueueClientSettings.cs index 84977e8..8ce4dd7 100644 --- a/src/AzureStorage.QueueService/QueueClientSettings.cs +++ b/src/AzureStorage.QueueService/QueueClientSettings.cs @@ -39,7 +39,7 @@ public class QueueClientSettings /// /// How long a message received with stays hidden from other consumers /// when the call does not say. When not set, the service default of 30 seconds applies. A processor uses its own - /// instead. + /// instead. /// public TimeSpan? VisibilityTimeout { get; set; } diff --git a/src/AzureStorage.QueueService/QueueMessage.cs b/src/AzureStorage.QueueService/QueueMessage.cs index 5ea899b..9e48176 100644 --- a/src/AzureStorage.QueueService/QueueMessage.cs +++ b/src/AzureStorage.QueueService/QueueMessage.cs @@ -61,7 +61,7 @@ public interface IQueueMessage /// /// True when a processor will not retry this message if the handler fails: this is its last delivery before - /// moves it to the poison queue. Always false for a message + /// moves it to the poison queue. Always false for a message /// received with , which has no retry limit of its own. /// bool IsLastAttempt { get; } @@ -230,7 +230,7 @@ internal ValueTask DeadLetterAsync(string reason, string category, CancellationT /// The bounded set of reasons the library records for a dead-lettered message, for metrics and replay filters. public static class DeadLetterCategories { - /// Delivered more than times. + /// Delivered more than times. public const string MaxDequeueCount = "max_dequeue"; /// The body could not be decoded, unwrapped, downloaded or deserialized. diff --git a/src/AzureStorage.QueueService/QueueMessageMiddleware.cs b/src/AzureStorage.QueueService/QueueMessageMiddleware.cs index 9ff56bc..af7548e 100644 --- a/src/AzureStorage.QueueService/QueueMessageMiddleware.cs +++ b/src/AzureStorage.QueueService/QueueMessageMiddleware.cs @@ -10,7 +10,7 @@ namespace AzureStorage.QueueService; /// /// /// Call next to go on to the handler, or return without calling it to stop there: the processor then -/// treats the message as handled (with on, it is completed). Throw and +/// treats the message as handled (with on, it is completed). Throw and /// the message counts as failed, exactly as if the handler had thrown. A middleware can settle the message itself /// through . Register it with UseMiddleware<T>() on one client's /// builder, or on AddAzureStorageQueueClient(queues => queues.UseMiddleware<T>()) for every client; each diff --git a/src/AzureStorage.QueueService/QueueProcessor.cs b/src/AzureStorage.QueueService/QueueProcessor.cs index dc121c8..4996378 100644 --- a/src/AzureStorage.QueueService/QueueProcessor.cs +++ b/src/AzureStorage.QueueService/QueueProcessor.cs @@ -31,13 +31,190 @@ internal sealed record ProcessorHooks( /// /// The hosted processor behind AddProcessor<TMessage, THandler>(): one receive loop that takes as many /// messages as there are free handlers (up to at once, 32 per -/// request) and delivers each on its own task. One instance per registration. +/// request) and delivers each on its own task. One instance per registration. Everything but the delivery of one +/// message is , which the batch processor shares. +/// +internal sealed class QueueProcessor : QueueProcessorBase + where THandler : class, IQueueMessageHandler +{ + private readonly string _optionsName; + private readonly IOptionsFactory _optionsFactory; + private readonly ProcessorHooks _hooks; + private QueueProcessorOptions _typedOptions = new(); + + public QueueProcessor( + string clientName, + string optionsName, + IQueueClientFactory factory, + IServiceScopeFactory scopes, + IOptionsFactory optionsFactory, + ILoggerFactory loggerFactory, + QueueServiceMetrics metrics, + TimeProvider time, + IHostApplicationLifetime? lifetime = null, + ProcessorHooks? hooks = null, + QueueProcessorController? controller = null) + : base( + "QueueProcessor", clientName, + // A wrapper handler (the CloudEvent adapter, a router, a lambda) passes the application's handler name, so logs + // name what the application wrote rather than the library's wrapper. + hooks?.HandlerName ?? typeof(THandler).Name, + factory, scopes, loggerFactory, metrics, time, lifetime, controller, hooks?.ReadBody, hooks?.AfterDeserialize) + { + _hooks = hooks ?? new ProcessorHooks(); + _optionsName = optionsName; + _optionsFactory = optionsFactory; + } + + private protected override QueueProcessorOptionsBase CreateOptions() => _typedOptions = _optionsFactory.Create(_optionsName); + + private protected override int MessagesPerDelivery => 1; + + private protected override int MaxConcurrentDeliveries => _typedOptions.MaxConcurrentMessages; + + private protected override Task DeliverAsync(IReadOnlyList> messages, IOwnedReceiver? owned, long started) => + DeliverOneAsync(messages[0], owned, started); + + // ------------------------------------------------------------------------------------------------------------ + // One delivery: the complete lifecycle of one message, with one exception boundary. Never throws. + // ------------------------------------------------------------------------------------------------------------ + + private async Task DeliverOneAsync(QueueMessage message, IOwnedReceiver? owned, long started) + { + var outcome = "abandoned"; + Exception? failure = null; + var lease = message.Lease; + lease.IsLastAttempt |= message.DequeueCount >= _options.MaxDequeueCount; + + using var logScope = _logger.BeginDeliveryScope(QueueName, message.MessageId, message.DequeueCount); + + var tags = QueueServiceDiagnostics.OperationTags(QueueName, QueueServiceDiagnostics.Messaging.Process); + tags.Add(QueueServiceDiagnostics.Names.MessageId, message.MessageId); + tags.Add(QueueServiceDiagnostics.Names.DequeueCount, message.DequeueCount); + tags.Add(QueueServiceDiagnostics.Messaging.MessageId, message.MessageId); + using Activity? activity = QueueServiceDiagnostics.Activities.StartProcessingMessageActivity(tags, message.CreationContext); + lease.DeliveryActivity = activity; + + _metrics.ActiveHandlers.Add(1, QueueTags()); + RecordTimeInQueue(message); + + // Armed before anything else — including downloading and deserializing — so nothing that happens to this + // message while it is ours can let the lease lapse. + await using var renewal = StartRenewal(message); + + using var longRunning = _options.LongRunningHandlerWarning is { } threshold + ? _time.CreateTimer(_ => _logger.LogProcessorHandlerLongRunning(_name, message.MessageId, _time.GetElapsedTime(started), threshold, RenewalNote()), null, threshold, Timeout.InfiniteTimeSpan) + : null; + + try + { + if (!await ReadBodyAsync(message, owned)) + { + // Cancelled while downloading the body at shutdown: give it straight back. + await StopRenewalAsync(renewal); + _logger.LogProcessorHandlerCancelled(_name, message.MessageId); + if (!message.IsSettled) await AbandonSafelyAsync(message, TimeSpan.Zero); + return; + } + + if (await RejectIfUnprocessableAsync(message, renewal) is { } rejected) + { + outcome = rejected; + return; + } + + try + { + // Resolving the handler and the middleware is inside the boundary: a constructor that throws is a failed delivery too. + await using var scope = _scopes.CreateAsyncScope(); + var context = new QueueMessageContext(message, scope.ServiceProvider, _clientName, typeof(TMessage), _handlerName, _handlerToken); + await Pipeline(message)(context); + } + catch (OperationCanceledException) when (_handlerToken.IsCancellationRequested && !message.IsSettled) + { + await StopRenewalAsync(renewal); + _logger.LogProcessorHandlerCancelled(_name, message.MessageId); + if (!message.IsSettled) await AbandonSafelyAsync(message, TimeSpan.Zero); + return; + } + catch (Exception ex) when (!message.IsSettled) + { + await StopRenewalAsync(renewal); + failure = ex; + activity.RecordFailure(ex); + _metrics.MessageFailures.Add(1, QueueTags()); + outcome = await HandleFailureAsync(message, ex); + return; + } + catch (Exception ex) + { + // The handler settled the message and then threw. The settlement stands; report it as what it was: + // an exception on a span that did not fail, so no error status. + _logger.LogHandlerFailedAfterSettling(QueueName, message.MessageId, message.Settlement, ex); + activity?.AddException(ex); + } + + // Wait for any renewal already in flight so completion uses the receipt the service holds. + await StopRenewalAsync(renewal); + (outcome, failure) = await SettleAfterReturnAsync(message, activity); + } + catch (Exception ex) + { + // Anything the inner boundaries did not cover: scope disposal, an option callback, a logging sink. + // A delivery must never fault silently and leave its message in limbo. + _logger.LogProcessorDeliveryFailed(_name, message.MessageId, ex); + failure = ex; + activity.RecordFailure(ex); + _metrics.MessageFailures.Add(1, QueueTags()); + if (!message.IsSettled) await AbandonSafelyAsync(message, QueueProcessorOptions.MaxRetryDelay); + outcome = "abandoned"; + } + finally + { + await StopRenewalAsync(renewal); + _metrics.ActiveHandlers.Add(-1, QueueTags()); + + var elapsed = _time.GetElapsedTime(started).TotalSeconds; + RecordOutcome(outcome, elapsed); + _metrics.ProcessDuration.Record(elapsed, QueueServiceDiagnostics.MessagingTags(QueueName, QueueServiceDiagnostics.Messaging.Process, QueueServiceDiagnostics.Messaging.Process, failure)); + activity?.SetTag(QueueServiceDiagnostics.Names.Outcome, outcome); + lease.DeliveryActivity = null; + } + } + + private QueueMessageDelegate Pipeline(QueueMessage message) + { + QueueMessageDelegate next = async context => + { + var handler = _hooks.CreateHandler is { } create ? create(context.Services) : context.Services.GetRequiredService(); + await handler.HandleAsync(message, context.CancellationToken); + }; + + if (_hooks.Middleware is not { Count: > 0 } middleware) return next; + + for (var i = middleware.Count - 1; i >= 0; i--) + { + var resolve = middleware[i]; + var inner = next; + next = context => resolve(context.Services).InvokeAsync(context, inner); + } + + return next; + } +} + +/// +/// Everything a hosted processor does except deliver what it received: start and stop, the receive loop, pausing, +/// status, and the per-message steps a delivery is built from (lease renewal, reading the body, the poison checks, the +/// error policy, settlement). delivers one message at a time; +/// a batch. /// /// /// -/// No message waits behind another's handler, and the processor never holds a message no handler is free for: a receive -/// asks only for as many messages as there are free slots, and each message's lease renewal starts as its delivery does. -/// A full batch means more may be waiting, so the loop asks again at once; an empty one backs off. +/// One receive loop per processor. It waits for a free delivery slot, takes every other free slot that fits in one +/// receive, asks for that many slots' worth of messages, and starts each slot's share on its own task. No message waits +/// behind another's handler, and the processor never holds a message no handler is free for. A full receive means more +/// may be waiting, so the loop asks again at once; an empty one backs off. /// /// /// Lifecycle: snapshots and validates the options and resolves the queue (a @@ -45,7 +222,7 @@ internal sealed record ProcessorHooks( /// — the signal that every startup callback succeeded — so a /// slow start does not hold leases and a start that fails part-way never polls at all; without a host lifetime (a /// direct test) it starts in . On shutdown, receiving stops; running handlers get -/// to finish; then their token is cancelled and lease renewal +/// to finish; then their token is cancelled and lease renewal /// stops in the same moment; a handler that ignores its token gets the same time again, after which shutdown stops /// waiting for it. If the host's own shutdown deadline arrives first, the handlers are cancelled at that moment and /// returns rather than throwing, as BackgroundService does. The cancellations happen no @@ -53,21 +230,20 @@ internal sealed record ProcessorHooks( /// unexpectedly is logged loudly and restarted rather than silently lost. /// /// -internal sealed class QueueProcessor : IHostedService, IHostedLifecycleService, IControlledProcessor, IDisposable - where THandler : class, IQueueMessageHandler +internal abstract class QueueProcessorBase : IHostedService, IHostedLifecycleService, IControlledProcessor, IDisposable { - private readonly string _clientName; - private readonly string _optionsName; + private protected readonly string _clientName; + private protected readonly string _handlerName; + private protected readonly IServiceScopeFactory _scopes; + private protected readonly ILogger _logger; + private protected readonly QueueServiceMetrics _metrics; + private protected readonly TimeProvider _time; + private readonly IQueueClientFactory _factory; - private readonly IServiceScopeFactory _scopes; - private readonly IOptionsFactory _optionsFactory; - private readonly ILogger _logger; - private readonly QueueServiceMetrics _metrics; - private readonly TimeProvider _time; private readonly IHostApplicationLifetime? _lifetime; private readonly QueueProcessorController? _controller; - private readonly ProcessorHooks _hooks; - private readonly string _handlerName; + private readonly Action>? _readBody; + private readonly Action>? _afterDeserialize; private readonly CancellationTokenSource _stopReceiving = new(); private readonly CancellationTokenSource _cancelHandlers = new(); @@ -75,21 +251,23 @@ internal sealed class QueueProcessor : IHostedService, IHost // Taken once, up front: reading Token from a disposed source throws, and a delivery can still be running when a // host that gave up waiting disposes the processor. - private readonly CancellationToken _handlerToken; + private protected readonly CancellationToken _handlerToken; private readonly CancellationToken _renewalToken; private readonly object _startLock = new(); private CancellationTokenRegistration _startWhenApplicationStarted; - // Deliveries in flight, by a sequence number, with the timestamp each started at: for draining, and for the - // oldest-handler gauge and status. - private readonly ConcurrentDictionary _deliveries = new(); + // Deliveries in flight, by a sequence number, with the timestamp each started at and how many messages it holds: + // for draining, and for the oldest-handler gauge and status. + private readonly ConcurrentDictionary _deliveries = new(); private long _deliverySequence; - private QueueProcessorOptions _options = new(); + /// The options snapshot this processor runs with, taken in . + private protected QueueProcessorOptionsBase _options = new QueueProcessorOptions(); + + private protected string _name; private SemaphoreSlim? _slots; private IStorageQueue? _queue; private Task? _receiver; - private string _name; private IDisposable? _registration; private IDisposable? _ageGauge; private volatile QueueProcessorState _state = QueueProcessorState.NotStarted; @@ -101,39 +279,52 @@ internal sealed class QueueProcessor : IHostedService, IHost private sealed record ReceiveHealth(DateTimeOffset? SucceededAt, DateTimeOffset? FailedAt, int ConsecutiveFailures, string? LastError); - public QueueProcessor( + private protected QueueProcessorBase( + string kind, string clientName, - string optionsName, + string handlerName, IQueueClientFactory factory, IServiceScopeFactory scopes, - IOptionsFactory optionsFactory, ILoggerFactory loggerFactory, QueueServiceMetrics metrics, TimeProvider time, - IHostApplicationLifetime? lifetime = null, - ProcessorHooks? hooks = null, - QueueProcessorController? controller = null) + IHostApplicationLifetime? lifetime, + QueueProcessorController? controller, + Action>? readBody, + Action>? afterDeserialize) { - _hooks = hooks ?? new ProcessorHooks(); - // A wrapper handler (the CloudEvent adapter, a router, a lambda) passes the application's handler name, so logs - // name what the application wrote rather than the library's wrapper. - _handlerName = _hooks.HandlerName ?? typeof(THandler).Name; + _handlerName = handlerName; _clientName = clientName; - _optionsName = optionsName; _factory = factory; _scopes = scopes; - _optionsFactory = optionsFactory; - _logger = loggerFactory.CreateLogger($"AzureStorage.QueueService.QueueProcessor<{typeof(TMessage).Name}, {_handlerName}>"); + _logger = loggerFactory.CreateLogger($"AzureStorage.QueueService.{kind}<{typeof(TMessage).Name}, {_handlerName}>"); _metrics = metrics; _time = time; _lifetime = lifetime; _controller = controller; + _readBody = readBody; + _afterDeserialize = afterDeserialize; _name = _handlerName; _handlerToken = _cancelHandlers.Token; _renewalToken = _stopRenewals.Token; } - private string QueueName => _queue?.QueueName ?? _clientName; + /// Creates the options snapshot; the base validates it. + private protected abstract QueueProcessorOptionsBase CreateOptions(); + + /// How many messages one delivery takes: 1, or a batch processor's . + private protected abstract int MessagesPerDelivery { get; } + + /// How many deliveries may run at once. + private protected abstract int MaxConcurrentDeliveries { get; } + + /// + /// Delivers what one slot received (one message, or a batch of up to ), from + /// arming lease renewal to settlement. Never throws. + /// + private protected abstract Task DeliverAsync(IReadOnlyList> messages, IOwnedReceiver? owned, long started); + + private protected string QueueName => _queue?.QueueName ?? _clientName; public string ClientName => _clientName; @@ -144,9 +335,9 @@ public QueueProcessor( public Task StartAsync(CancellationToken cancellationToken) { // A snapshot: these are the options the processor runs with for its whole life. - _options = _optionsFactory.Create(_optionsName); + _options = CreateOptions(); _options.Validate($"Processor {_name}"); - _slots = new SemaphoreSlim(_options.MaxConcurrentMessages, _options.MaxConcurrentMessages); + _slots = new SemaphoreSlim(MaxConcurrentDeliveries, MaxConcurrentDeliveries); // Throws for a misconfigured client. That fails host startup, which is the right time to find out. _queue = _clientName == Options.DefaultName ? _factory.GetQueue() : _factory.GetQueue(_clientName); @@ -180,7 +371,7 @@ private void StartReceiving() { if (_receiver is not null || _stopReceiving.IsCancellationRequested) return; - _logger.LogProcessorStarted(_name, _options.MaxConcurrentMessages); + _logger.LogProcessorStarted(_name, MaxConcurrentDeliveries); _state = QueueProcessorState.Running; _receiver = Task.Run(() => ReceiverAsync(_stopReceiving.Token), CancellationToken.None); } @@ -285,7 +476,7 @@ public QueueProcessorStatus GetStatus() TimeSpan? oldest = deliveries.Count == 0 ? null : _time.GetElapsedTime(deliveries.Min(d => d.Started)); var health = _health; return new QueueProcessorStatus( - _clientName, QueueName, _handlerName, typeof(TMessage), state, deliveries.Count, _options.MaxConcurrentMessages, + _clientName, QueueName, _handlerName, typeof(TMessage), state, deliveries.Sum(d => d.Messages), MaxConcurrentDeliveries * MessagesPerDelivery, health.SucceededAt, health.FailedAt, health.ConsecutiveFailures, health.LastError, oldest); } @@ -298,7 +489,7 @@ private IEnumerable> ObserveOldestHandler() } // ------------------------------------------------------------------------------------------------------------ - // The receive loop: take as many as there are free handlers, hand each to its own delivery, repeat + // The receive loop: take as many as there are free slots, hand each slot's share to its own delivery, repeat // ------------------------------------------------------------------------------------------------------------ private async Task ReceiverAsync(CancellationToken stop) @@ -341,6 +532,7 @@ private async Task ReceiveUntilStoppedAsync(CancellationToken stop) var queue = _queue!; var slots = _slots!; var owned = queue as IOwnedReceiver; + var perDelivery = MessagesPerDelivery; var emptyBackoff = _options.EmptyQueueBackoffMin; var errorBackoff = _options.EmptyQueueBackoffMin; @@ -348,19 +540,22 @@ private async Task ReceiveUntilStoppedAsync(CancellationToken stop) { await WaitWhilePausedAsync(stop); - // At least one free handler, then every other one that is free right now. + // At least one free slot, then every other one that is free right now and fits in the same receive. For a + // single-message processor that is up to 32 slots; a full-size batch fills a receive on its own. await slots.WaitAsync(stop); - var take = 1; - while (take < StorageQueue.ServiceMaxMessages && slots.Wait(0)) take++; + var taken = 1; + while ((taken + 1) * perDelivery <= StorageQueue.ServiceMaxMessages && slots.Wait(0)) taken++; + var take = taken * perDelivery; IReadOnlyList> messages; try { if (_options.RateLimiter is { } limiter) { - var permitted = await AcquirePermitsAsync(limiter, take, stop); - slots.Release(take - permitted); - take = permitted; + take = await AcquirePermitsAsync(limiter, take, stop); + var needed = (take + perDelivery - 1) / perDelivery; + Release(slots, taken - needed); + taken = needed; } // The library's own queue hands messages over before reading their bodies, so renewal is armed first. @@ -374,12 +569,12 @@ private async Task ReceiveUntilStoppedAsync(CancellationToken stop) } catch (OperationCanceledException) when (stop.IsCancellationRequested) { - slots.Release(take); + Release(slots, taken); return; } catch (Exception ex) { - slots.Release(take); + Release(slots, taken); _health = _health with { FailedAt = _time.GetUtcNow(), ConsecutiveFailures = _health.ConsecutiveFailures + 1, LastError = ex.Message }; // Transport failures, a missing queue, no permission: log once, back off, keep going. @@ -393,8 +588,13 @@ private async Task ReceiveUntilStoppedAsync(CancellationToken stop) continue; } - foreach (var message in messages) StartDelivery(message, owned); - if (messages.Count < take) slots.Release(take - messages.Count); + var started = 0; + for (var offset = 0; offset < messages.Count; offset += perDelivery, started++) + { + StartDelivery(perDelivery == 1 ? [messages[offset]] : messages.Skip(offset).Take(perDelivery).ToList(), owned); + } + + Release(slots, taken - started); if (messages.Count == 0) { @@ -408,6 +608,12 @@ private async Task ReceiveUntilStoppedAsync(CancellationToken stop) } } + // SemaphoreSlim.Release(0) throws, and "none to give back" is the usual case. + private static void Release(SemaphoreSlim slots, int count) + { + if (count > 0) slots.Release(count); + } + private async Task WaitWhilePausedAsync(CancellationToken stop) { if (_controller is null || !_controller.IsPaused(_clientName)) return; @@ -443,19 +649,19 @@ private async Task AcquirePermitsAsync(RateLimiter limiter, int wanted, Can return permitted; } - private void StartDelivery(QueueMessage message, IOwnedReceiver? owned) + private void StartDelivery(IReadOnlyList> messages, IOwnedReceiver? owned) { var id = Interlocked.Increment(ref _deliverySequence); var started = _time.GetTimestamp(); var delivery = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - _deliveries[id] = (delivery.Task, started); + _deliveries[id] = (delivery.Task, started, messages.Count); // On its own task, so neither renewal nor a handler that blocks before its first await can hold up the loop. _ = Task.Run(async () => { try { - await DeliverAsync(message, owned, started); + await DeliverAsync(messages, owned, started); } finally { @@ -482,32 +688,18 @@ private async Task DelayAsync(TimeSpan delay, CancellationToken stop) private static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b; // ------------------------------------------------------------------------------------------------------------ - // One delivery: the complete lifecycle of one message, with one exception boundary. Never throws. + // The steps a delivery is made of, per message // ------------------------------------------------------------------------------------------------------------ - private async Task DeliverAsync(QueueMessage message, IOwnedReceiver? owned, long started) + private protected void RecordTimeInQueue(QueueMessage message) { - var outcome = "abandoned"; - Exception? failure = null; - var lease = message.Lease; - lease.IsLastAttempt |= message.DequeueCount >= _options.MaxDequeueCount; - - using var logScope = _logger.BeginDeliveryScope(QueueName, message.MessageId, message.DequeueCount); - - var tags = QueueServiceDiagnostics.OperationTags(QueueName, QueueServiceDiagnostics.Messaging.Process); - tags.Add(QueueServiceDiagnostics.Names.MessageId, message.MessageId); - tags.Add(QueueServiceDiagnostics.Names.DequeueCount, message.DequeueCount); - tags.Add(QueueServiceDiagnostics.Messaging.MessageId, message.MessageId); - using Activity? activity = QueueServiceDiagnostics.Activities.StartProcessingMessageActivity(tags, message.CreationContext); - lease.DeliveryActivity = activity; - - _metrics.ActiveHandlers.Add(1, QueueTags()); if (message.InsertedOn is { } inserted) _metrics.TimeInQueue.Record(Math.Max(0, (_time.GetUtcNow() - inserted).TotalSeconds), QueueTags()); + } - // Armed before anything else — including downloading and deserializing — so nothing that happens to this - // message while it is ours can let the lease lapse. - await using var renewal = _options.RenewVisibility + /// Starts renewing 's lease at half the visibility timeout; null with renewal off. + private protected LeaseRenewal? StartRenewal(QueueMessage message) => + _options.RenewVisibility ? LeaseRenewal.Start( async () => { @@ -527,146 +719,15 @@ private async Task DeliverAsync(QueueMessage message, IOwnedReceiver? () => _logger.LogProcessorRenewalLimitReached(_name, message.MessageId, _options.MaxRenewalDuration!.Value)) : null; - using var longRunning = _options.LongRunningHandlerWarning is { } threshold - ? _time.CreateTimer(_ => _logger.LogProcessorHandlerLongRunning(_name, message.MessageId, _time.GetElapsedTime(started), threshold, - _options.MaxRenewalDuration is { } max ? $" until MaxRenewalDuration ({max})" : " for as long as it runs"), null, threshold, Timeout.InfiniteTimeSpan) - : null; - - try - { - if (!await ReadBodyAsync(message, owned)) - { - // Cancelled while downloading the body at shutdown: give it straight back. - await StopRenewalAsync(renewal); - _logger.LogProcessorHandlerCancelled(_name, message.MessageId); - if (!message.IsSettled) await AbandonSafelyAsync(message, TimeSpan.Zero); - return; - } - - if (message.Error is UnroutableMessageException unroutable) - { - await StopRenewalAsync(renewal); - outcome = await DeadLetterOrAbandonAsync(message, unroutable.Message, DeadLetterCategories.Unroutable); - return; - } - - if (message.Error is not null && _options.DeadLetterOnDeserializationFailure) - { - await StopRenewalAsync(renewal); - outcome = await DeadLetterOrAbandonAsync(message, $"the body could not be read: {message.Error.Message}", DeadLetterCategories.Deserialization); - return; - } - - if (message.DequeueCount > _options.MaxDequeueCount) - { - await StopRenewalAsync(renewal); - outcome = await DeadLetterOrAbandonAsync(message, $"delivered {message.DequeueCount} times; the limit is {_options.MaxDequeueCount}", DeadLetterCategories.MaxDequeueCount); - return; - } - - try - { - // Resolving the handler and the middleware is inside the boundary: a constructor that throws is a failed delivery too. - await using var scope = _scopes.CreateAsyncScope(); - var context = new QueueMessageContext(message, scope.ServiceProvider, _clientName, typeof(TMessage), _handlerName, _handlerToken); - await Pipeline(message)(context); - } - catch (OperationCanceledException) when (_handlerToken.IsCancellationRequested && !message.IsSettled) - { - await StopRenewalAsync(renewal); - _logger.LogProcessorHandlerCancelled(_name, message.MessageId); - if (!message.IsSettled) await AbandonSafelyAsync(message, TimeSpan.Zero); - return; - } - catch (Exception ex) when (!message.IsSettled) - { - await StopRenewalAsync(renewal); - failure = ex; - activity.RecordFailure(ex); - _metrics.MessageFailures.Add(1, QueueTags()); - outcome = await HandleFailureAsync(message, ex); - return; - } - catch (Exception ex) - { - // The handler settled the message and then threw. The settlement stands; report it as what it was: - // an exception on a span that did not fail, so no error status. - _logger.LogHandlerFailedAfterSettling(QueueName, message.MessageId, message.Settlement, ex); - activity?.AddException(ex); - } - - // Wait for any renewal already in flight so completion uses the receipt the service holds. - await StopRenewalAsync(renewal); - - switch (message.Settlement) - { - case MessageSettlement.Completed: - outcome = "completed"; - break; - case MessageSettlement.Abandoned: - outcome = "abandoned_by_handler"; - return; - case MessageSettlement.DeadLettered: - outcome = "dead_lettered_by_handler"; - return; - case MessageSettlement.None when _options.AutoComplete: - try - { - // Not tied to the shutdown token: cancelling a delete after the handler succeeded only redelivers finished work. - await message.CompleteAsync(CancellationToken.None); - outcome = "completed"; - } - catch (Exception ex) - { - _logger.LogDeleteFailed(QueueName, message.MessageId, _options.VisibilityTimeout, ex); - failure = ex; - activity.RecordFailure(ex); - _metrics.MessageFailures.Add(1, QueueTags()); - outcome = "complete_failed"; - return; - } - break; - default: - // AutoComplete is off and the handler returned without settling: nothing was removed. - outcome = "left_to_handler"; - return; - } - - _logger.LogMessageProcessed(QueueName, message.MessageId); - _metrics.MessagesProcessed.Add(1, QueueTags()); - } - catch (Exception ex) - { - // Anything the inner boundaries did not cover: scope disposal, an option callback, a logging sink. - // A delivery must never fault silently and leave its message in limbo. - _logger.LogProcessorDeliveryFailed(_name, message.MessageId, ex); - failure = ex; - activity.RecordFailure(ex); - _metrics.MessageFailures.Add(1, QueueTags()); - if (!message.IsSettled) await AbandonSafelyAsync(message, QueueProcessorOptions.MaxRetryDelay); - outcome = "abandoned"; - } - finally - { - await StopRenewalAsync(renewal); - _metrics.ActiveHandlers.Add(-1, QueueTags()); - - var elapsed = _time.GetElapsedTime(started).TotalSeconds; - var finalTags = QueueTags(); - finalTags.Add(QueueServiceDiagnostics.Names.QueueProcessSuccess, outcome is "completed" or "left_to_handler"); - finalTags.Add(QueueServiceDiagnostics.Names.Outcome, outcome); - _metrics.MessageProcessingDuration.Record(elapsed, finalTags); - _metrics.ProcessDuration.Record(elapsed, QueueServiceDiagnostics.MessagingTags(QueueName, QueueServiceDiagnostics.Messaging.Process, QueueServiceDiagnostics.Messaging.Process, failure)); - activity?.SetTag(QueueServiceDiagnostics.Names.Outcome, outcome); - lease.DeliveryActivity = null; - } - } + /// The end of the long-running warning: how long the lease goes on being renewed. + private protected string RenewalNote() => + _options.MaxRenewalDuration is { } max ? $" until MaxRenewalDuration ({max})" : " for as long as it runs"; /// /// Brings in a body kept outside the queue and reads it, setting or /// . False when shutdown cancelled the download. /// - private async Task ReadBodyAsync(QueueMessage message, IOwnedReceiver? owned) + private protected async Task ReadBodyAsync(QueueMessage message, IOwnedReceiver? owned) { if (owned is not null) { @@ -683,38 +744,78 @@ private async Task ReadBodyAsync(QueueMessage message, IOwnedRec if (message.Error is null) { - if (_hooks.ReadBody is { } read) read(message); + if (_readBody is { } read) read(message); else owned?.Deserialize(message); } // A wrapper's own reading of the body (the CloudEvent adapter reads the event's data), which sets Error when it // fails, so that failure takes the dead-letter path with this processor's options snapshot, metric and outcome. - if (message.Error is null) _hooks.AfterDeserialize?.Invoke(message); + if (message.Error is null) _afterDeserialize?.Invoke(message); return true; } - private QueueMessageDelegate Pipeline(QueueMessage message) + /// + /// The checks before a handler sees a message: one nothing is routed for, one whose body could not be read (with + /// ), and one delivered too often are + /// dead-lettered. Returns the outcome when it did that, or null when the message can go to the handler. + /// + private protected async Task RejectIfUnprocessableAsync(QueueMessage message, LeaseRenewal? renewal) { - QueueMessageDelegate next = async context => - { - var handler = _hooks.CreateHandler is { } create ? create(context.Services) : context.Services.GetRequiredService(); - await handler.HandleAsync(message, context.CancellationToken); - }; - - if (_hooks.Middleware is not { Count: > 0 } middleware) return next; + string reason, category; + if (message.Error is UnroutableMessageException unroutable) + (reason, category) = (unroutable.Message, DeadLetterCategories.Unroutable); + else if (message.Error is not null && _options.DeadLetterOnDeserializationFailure) + (reason, category) = ($"the body could not be read: {message.Error.Message}", DeadLetterCategories.Deserialization); + else if (message.DequeueCount > _options.MaxDequeueCount) + (reason, category) = ($"delivered {message.DequeueCount} times; the limit is {_options.MaxDequeueCount}", DeadLetterCategories.MaxDequeueCount); + else + return null; + + await StopRenewalAsync(renewal); + return await DeadLetterOrAbandonAsync(message, reason, category); + } - for (var i = middleware.Count - 1; i >= 0; i--) - { - var resolve = middleware[i]; - var inner = next; - next = context => resolve(context.Services).InvokeAsync(context, inner); + /// + /// After the handler returned: what its own settlement of the message means, or, when it left the message alone, + /// completion with . Logs and counts a completion. The failure + /// is a completion that failed. + /// + private protected async Task<(string Outcome, Exception? Failure)> SettleAfterReturnAsync(QueueMessage message, Activity? activity) + { + switch (message.Settlement) + { + case MessageSettlement.Completed: + break; + case MessageSettlement.Abandoned: + return ("abandoned_by_handler", null); + case MessageSettlement.DeadLettered: + return ("dead_lettered_by_handler", null); + case MessageSettlement.None when _options.AutoComplete: + try + { + // Not tied to the shutdown token: cancelling a delete after the handler succeeded only redelivers finished work. + await message.CompleteAsync(CancellationToken.None); + } + catch (Exception ex) + { + _logger.LogDeleteFailed(QueueName, message.MessageId, _options.VisibilityTimeout, ex); + activity.RecordFailure(ex); + _metrics.MessageFailures.Add(1, QueueTags()); + return ("complete_failed", ex); + } + break; + default: + // AutoComplete is off and the handler returned without settling: nothing was removed. + return ("left_to_handler", null); } - return next; + _logger.LogMessageProcessed(QueueName, message.MessageId); + _metrics.MessagesProcessed.Add(1, QueueTags()); + return ("completed", null); } - /// The error policy: , then , then retry. - private async Task HandleFailureAsync(QueueMessage message, Exception ex) + /// The error policy: DeadLetterOn<TException>(), then , then retry. + private protected async Task HandleFailureAsync(QueueMessage message, Exception ex) { var action = DecideOnFailure(message, ex); switch (action) @@ -765,7 +866,7 @@ private HandlerErrorAction DecideOnFailure(QueueMessage message, Excep } } - private static ValueTask StopRenewalAsync(LeaseRenewal? renewal) => renewal?.StopAsync() ?? ValueTask.CompletedTask; + private protected static ValueTask StopRenewalAsync(LeaseRenewal? renewal) => renewal?.StopAsync() ?? ValueTask.CompletedTask; private TimeSpan RetryDelayFor(long dequeueCount) { @@ -776,12 +877,12 @@ private TimeSpan RetryDelayFor(long dequeueCount) } catch (Exception ex) { - _logger.LogProcessorRetryDelayFailed(_name, dequeueCount, QueueProcessorOptions.MaxRetryDelay, ex); - return QueueProcessorOptions.MaxRetryDelay; + _logger.LogProcessorRetryDelayFailed(_name, dequeueCount, QueueProcessorOptionsBase.MaxRetryDelay, ex); + return QueueProcessorOptionsBase.MaxRetryDelay; } // The service accepts 0 to 7 days; anything else would make the abandon itself fail. - return delay < TimeSpan.Zero ? TimeSpan.Zero : Min(delay, QueueProcessorOptions.MaxVisibility); + return delay < TimeSpan.Zero ? TimeSpan.Zero : Min(delay, QueueProcessorOptionsBase.MaxVisibility); } private async Task DeadLetterOrAbandonAsync(QueueMessage message, string reason, string category) @@ -801,7 +902,7 @@ private async Task DeadLetterOrAbandonAsync(QueueMessage messa } } - private async ValueTask AbandonSafelyAsync(QueueMessage message, TimeSpan delay) + private protected async ValueTask AbandonSafelyAsync(QueueMessage message, TimeSpan delay) { try { @@ -813,7 +914,16 @@ private async ValueTask AbandonSafelyAsync(QueueMessage message, TimeS } } - private TagList QueueTags() => QueueServiceDiagnostics.MetricTags(QueueName); + /// One message's processing time, tagged with how it ended. + private protected void RecordOutcome(string outcome, double elapsedSeconds) + { + var tags = QueueTags(); + tags.Add(QueueServiceDiagnostics.Names.QueueProcessSuccess, outcome is "completed" or "left_to_handler"); + tags.Add(QueueServiceDiagnostics.Names.Outcome, outcome); + _metrics.MessageProcessingDuration.Record(elapsedSeconds, tags); + } + + private protected TagList QueueTags() => QueueServiceDiagnostics.MetricTags(QueueName); } /// A message a routed processor has no handler for; always dead-lettered, as no retry can find it one. diff --git a/src/AzureStorage.QueueService/QueueProcessorController.cs b/src/AzureStorage.QueueService/QueueProcessorController.cs index 8988116..9d799ae 100644 --- a/src/AzureStorage.QueueService/QueueProcessorController.cs +++ b/src/AzureStorage.QueueService/QueueProcessorController.cs @@ -54,8 +54,8 @@ public enum QueueProcessorState /// The handler, as it appears in logs. /// The type it reads bodies as. /// Where it is in its life. -/// Messages being delivered right now. -/// The most it delivers at once. +/// Messages being delivered right now; for a batch processor, the messages in the batches being handled. +/// The most messages it delivers at once; for a batch processor, × . /// When a receive last succeeded, empty or not; null before the first. /// When a receive last failed; null if none has. /// Receives that have failed in a row; 0 after a success. diff --git a/src/AzureStorage.QueueService/QueueProcessorOptions.cs b/src/AzureStorage.QueueService/QueueProcessorOptions.cs index 078a2f9..b790683 100644 --- a/src/AzureStorage.QueueService/QueueProcessorOptions.cs +++ b/src/AzureStorage.QueueService/QueueProcessorOptions.cs @@ -8,11 +8,89 @@ namespace AzureStorage.QueueService; /// scope, with its own lease renewal. /// /// -/// Everything except the callbacks (, ), -/// and can be bound from configuration with the +/// Everything except the callbacks (, +/// ), and +/// can be bound from configuration with the /// AddProcessor<TMessage, THandler>(IConfiguration) overload. /// -public sealed class QueueProcessorOptions +public sealed class QueueProcessorOptions : QueueProcessorOptionsBase +{ + /// + /// How many messages may be handled at once, each in its own scope with its own lease renewal. The processor receives + /// only as many messages as there are free handlers (up to 32 per request), so it never holds a message no handler is + /// free for. + /// + public int MaxConcurrentMessages { get; set; } = 1; + + /// + /// Dead-letters a message straight away, without retrying it, when its handler throws + /// or an exception derived from it: one that retrying cannot fix, such as a validation failure. Checked before + /// . + /// + public QueueProcessorOptions DeadLetterOn() where TException : Exception + { + AddDeadLetterType(typeof(TException)); + return this; + } + + internal override void Validate(string processorName) + { + if (MaxConcurrentMessages < 1) Fail(processorName, "MaxConcurrentMessages must be at least 1."); + base.Validate(processorName); + } +} + +/// +/// How a batch processor registered with AddBatchProcessor<TMessage, THandler>() runs. Each receive takes +/// up to messages and hands them to one handler call in one scope; up to +/// batches are handled at once. Every message keeps its own lease renewal, dequeue +/// count, retry delay and settlement. +/// +/// +/// Everything except the callbacks (, +/// ), and +/// can be bound from configuration with the +/// AddBatchProcessor<TMessage, THandler>(IConfiguration) overload. A rate limiter still takes one permit +/// per message, so it can make a batch smaller. +/// +public sealed class QueueBatchProcessorOptions : QueueProcessorOptionsBase +{ + /// + /// The most messages handed to the handler in one call: 1 to 32 (the most one receive returns), default 32. A batch is + /// whatever one receive returns, so it is smaller when fewer messages are waiting; the processor does not wait to fill it. + /// + public int MaxBatchSize { get; set; } = StorageQueue.ServiceMaxMessages; + + /// + /// How many batches may be handled at once, each in its own scope; default 1. The processor holds at most + /// × messages. + /// + public int MaxConcurrentBatches { get; set; } = 1; + + /// + /// When the handler throws or an exception derived from it, dead-letters every + /// message of the batch it had not settled, without retrying them. Checked before + /// . + /// + public QueueBatchProcessorOptions DeadLetterOn() where TException : Exception + { + AddDeadLetterType(typeof(TException)); + return this; + } + + internal override void Validate(string processorName) + { + if (MaxBatchSize is < 1 or > StorageQueue.ServiceMaxMessages) Fail(processorName, $"MaxBatchSize must be between 1 and {StorageQueue.ServiceMaxMessages} (it is {MaxBatchSize})."); + if (MaxConcurrentBatches < 1) Fail(processorName, "MaxConcurrentBatches must be at least 1."); + base.Validate(processorName); + } +} + +/// +/// What and share: leases, retries, the +/// poison queue, the error policy, polling and shutdown. They differ only in how much is handled at once. +/// +public abstract class QueueProcessorOptionsBase { internal static readonly TimeSpan MaxRetryDelay = TimeSpan.FromMinutes(5); @@ -24,13 +102,10 @@ public sealed class QueueProcessorOptions private readonly List _deadLetterOn = new(); - - /// - /// How many messages may be handled at once, each in its own scope with its own lease renewal. The processor receives - /// only as many messages as there are free handlers (up to 32 per request), so it never holds a message no handler is - /// free for. - /// - public int MaxConcurrentMessages { get; set; } = 1; + // Only the library's own option types derive from this. + private protected QueueProcessorOptionsBase() + { + } /// /// How long a received message stays hidden while a handler runs: 1 second to 7 days, in whole seconds (the @@ -83,22 +158,12 @@ public sealed class QueueProcessorOptions }; /// - /// Decides what happens to a message whose handler threw, when no rule matched: - /// retry it (the default without a callback), dead-letter it, or complete it. If the callback throws, the message is retried. + /// Decides what happens to a message whose handler threw, when no DeadLetterOn<TException>() rule + /// matched: retry it (the default without a callback), dead-letter it, or complete it. If the callback throws, the + /// message is retried. A batch processor asks once for each message the handler had not settled. /// public Func? OnHandlerError { get; set; } - /// - /// Dead-letters a message straight away, without retrying it, when its handler throws - /// or an exception derived from it: one that retrying cannot fix, such as a validation failure. Checked before - /// . - /// - public QueueProcessorOptions DeadLetterOn() where TException : Exception - { - if (!_deadLetterOn.Contains(typeof(TException))) _deadLetterOn.Add(typeof(TException)); - return this; - } - internal IReadOnlyList DeadLetterExceptionTypes => _deadLetterOn; /// @@ -127,9 +192,13 @@ public QueueProcessorOptions DeadLetterOn() where TException : Excep /// public TimeSpan ShutdownDrainTimeout { get; set; } = TimeSpan.FromSeconds(30); - internal void Validate(string processorName) + private protected void AddDeadLetterType(Type exceptionType) + { + if (!_deadLetterOn.Contains(exceptionType)) _deadLetterOn.Add(exceptionType); + } + + internal virtual void Validate(string processorName) { - if (MaxConcurrentMessages < 1) Fail(processorName, "MaxConcurrentMessages must be at least 1."); // The SDK sends whole seconds, so anything under a second becomes 0 and the service rejects every receive. if (VisibilityTimeout < TimeSpan.FromSeconds(1) || VisibilityTimeout > MaxVisibility) Fail(processorName, $"VisibilityTimeout must be between 1 second and 7 days (it is {VisibilityTimeout})."); // Renewal runs at half the timeout; with a fraction the service's lease is shorter than that half assumes. @@ -153,13 +222,13 @@ internal void Validate(string processorName) } } - private static void Fail(string processorName, string problem) => throw new InvalidOperationException($"{processorName}: {problem}"); + private protected static void Fail(string processorName, string problem) => throw new InvalidOperationException($"{processorName}: {problem}"); } /// What a processor does with a message whose handler threw. public enum HandlerErrorAction { - /// Abandon it with , to be delivered again; dead-lettered once it passes . + /// Abandon it with , to be delivered again; dead-lettered once it passes . Retry, /// Move it to the poison queue now. @@ -169,7 +238,10 @@ public enum HandlerErrorAction Complete, } -/// What is told about a failed delivery. +/// +/// What is told about a failed delivery. For a batch handler it +/// is asked once for each message the handler had not settled, with the same exception. +/// public sealed class HandlerErrorContext { internal HandlerErrorContext(Exception exception, IQueueMessage message) diff --git a/src/AzureStorage.QueueService/Telemetry/QueueServiceDiagnostics.cs b/src/AzureStorage.QueueService/Telemetry/QueueServiceDiagnostics.cs index 412d6e3..23b9c5a 100644 --- a/src/AzureStorage.QueueService/Telemetry/QueueServiceDiagnostics.cs +++ b/src/AzureStorage.QueueService/Telemetry/QueueServiceDiagnostics.cs @@ -32,6 +32,17 @@ public struct Activities var parent = Activity.Current is null ? creation : default; return ActivitySource.StartActivityWithTags(Names.ProcessingMessage, ActivityKind.Consumer, tags, parent, [new ActivityLink(creation)]); } + + /// + /// Starts the span for processing a batch. It links to every sender's trace context that was carried, as the + /// messaging conventions recommend for batches; no one sender's trace can be its parent, so it keeps the ambient + /// parent, or none. + /// + public static Activity? StartProcessingBatchActivity(in TagList tags, IEnumerable creationContexts) + { + var links = creationContexts.Where(c => c.HasValue).Select(c => new ActivityLink(c!.Value)).ToList(); + return ActivitySource.StartActivityWithTags(Names.ProcessingMessage, ActivityKind.Consumer, tags, links: links.Count == 0 ? null : links); + } } public struct Names diff --git a/tests/AzureStorage.QueueService.Tests/BatchProcessorTests.cs b/tests/AzureStorage.QueueService.Tests/BatchProcessorTests.cs new file mode 100644 index 0000000..56ecf04 --- /dev/null +++ b/tests/AzureStorage.QueueService.Tests/BatchProcessorTests.cs @@ -0,0 +1,151 @@ +using FluentAssertions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Time.Testing; +using Xunit.Abstractions; + +namespace AzureStorage.QueueService.Tests; + +/// +/// Batch handlers end to end on in-memory queues, the way runs the single-message +/// processor: messages are sent before the host starts, the clock is a fake that never moves, and stopping the host +/// drains the batches, so their settlement is finished by then. +/// +public class BatchProcessorTests(ITestOutputHelper output) +{ + /// The batches the handler saw, by body, and what it does with each. + public sealed class BatchState + { + private readonly Signal _handled = new(); + public List Batches { get; } = new(); + public Func>, ValueTask> Behaviour { get; set; } = _ => ValueTask.CompletedTask; + + public async ValueTask HandleAsync(IReadOnlyList> messages) + { + try + { + await Behaviour(messages); + } + finally + { + lock (Batches) Batches.Add(messages.Select(m => m.Body!.Property1!).ToArray()); + _handled.Set(); + } + } + + public Task WaitForBatchesAsync(int count) => _handled.WaitUntilAsync(() => { lock (Batches) return Batches.Count >= count; }); + } + + private sealed class Handler(BatchState state) : IQueueBatchHandler + { + public ValueTask HandleAsync(IReadOnlyList> messages, CancellationToken cancellationToken) => state.HandleAsync(messages); + } + + private IHost Build(Action? configure = null) + { + var builder = Host.CreateEmptyApplicationBuilder(new HostApplicationBuilderSettings()); + builder.Logging.AddProvider(new XUnitLoggerProvider(output)); + builder.Services.AddSingleton(new FakeTimeProvider()); + builder.Services.AddSingleton(); + builder.Services.AddAzureStorageQueueClient(queues => queues + .AddClient("orders", s => s.QueueName = "orders") + .AddBatchProcessor(configure)); + builder.Services.AddInMemoryQueues(); + return builder.Build(); + } + + private static async Task SendAsync(IHost host, params string[] bodies) + { + var sender = host.Services.GetRequiredKeyedService("orders"); + foreach (var body in bodies) await sender.SendAsync(new TestObject { Property1 = body }); + } + + private static async Task RemainingAsync(IHost host) => + (await host.Services.GetRequiredKeyedService("orders").GetPropertiesAsync()).ApproximateMessagesCount; + + private static async Task> DeadLetteredAsync(IHost host) => + await host.Services.GetRequiredKeyedService("orders").PeekAsync(maxMessages: 32); + + [Fact(DisplayName = "A batch handler gets up to MaxBatchSize messages per call, and returning completes them all")] + public async Task Batches_Up_To_MaxBatchSize() + { + using var host = Build(o => o.MaxBatchSize = 3); + await SendAsync(host, "1", "2", "3", "4", "5"); + var state = host.Services.GetRequiredService(); + + await host.StartAsync(); + await state.WaitForBatchesAsync(2); + await host.StopAsync(); + + state.Batches.Select(b => b.Length).Should().Equal(3, 2); + state.Batches.SelectMany(b => b).Should().BeEquivalentTo("1", "2", "3", "4", "5"); + (await RemainingAsync(host)).Should().Be(0); + host.Services.GetRequiredService().GetStatus().Should().ContainSingle() + .Which.MaxConcurrentMessages.Should().Be(3, "one batch of up to three at a time"); + } + + [Fact(DisplayName = "A batch handler can dead-letter one message and return; the rest are completed")] + public async Task Partial_Settlement_Then_Completion() + { + using var host = Build(); + await SendAsync(host, "ok 1", "bad", "ok 2"); + var state = host.Services.GetRequiredService(); + state.Behaviour = async messages => + { + foreach (var message in messages.Where(m => m.Body!.Property1 == "bad")) await message.DeadLetterAsync("not a real order"); + }; + + await host.StartAsync(); + await state.WaitForBatchesAsync(1); + await host.StopAsync(); + + state.Batches.Should().ContainSingle().Which.Should().HaveCount(3); + var deadLettered = (await DeadLetteredAsync(host)).Should().ContainSingle().Subject; + deadLettered.Category.Should().Be(DeadLetterCategories.Handler); + deadLettered.Reason.Should().Be("not a real order"); + (await RemainingAsync(host)).Should().Be(0, "the two the handler left alone were completed"); + } + + [Fact(DisplayName = "A throwing batch handler: settled messages stand, the error policy decides per message, the rest are abandoned for retry")] + public async Task Failure_Applies_The_Policy_Per_Message() + { + using var host = Build(o => o.OnHandlerError = context => + ((TestObject)context.Message.Body!).Property1 == "poison" ? HandlerErrorAction.DeadLetter : HandlerErrorAction.Retry); + await SendAsync(host, "done", "poison", "retry"); + var state = host.Services.GetRequiredService(); + state.Behaviour = async messages => + { + await messages.Single(m => m.Body!.Property1 == "done").CompleteAsync(); + throw new InvalidOperationException("downstream failed"); + }; + + await host.StartAsync(); + await state.WaitForBatchesAsync(1); + await host.StopAsync(); + + state.Batches.Should().ContainSingle(); + (await DeadLetteredAsync(host)).Should().ContainSingle().Which.Reason.Should().Contain("InvalidOperationException"); + (await RemainingAsync(host)).Should().Be(1, "only the retried message is still on the queue"); + (await host.Services.GetRequiredKeyedService("orders").PeekAsync()) + .Should().BeEmpty("it was abandoned with a retry delay and the clock has not moved"); + } + + [Fact(DisplayName = "An unreadable message is dead-lettered on its own and left out of the batch")] + public async Task Unreadable_Messages_Are_Left_Out() + { + using var host = Build(); + await SendAsync(host, "a"); + await host.Services.GetRequiredKeyedService("orders").SendAsync("not an order"); + await SendAsync(host, "b"); + var state = host.Services.GetRequiredService(); + + await host.StartAsync(); + await state.WaitForBatchesAsync(1); + await host.StopAsync(); + + state.Batches.Should().ContainSingle().Which.Should().Equal("a", "b"); + (await DeadLetteredAsync(host)).Should().ContainSingle().Which.Category.Should().Be(DeadLetterCategories.Deserialization); + (await RemainingAsync(host)).Should().Be(0); + } +} diff --git a/tests/AzureStorage.QueueService.Tests/QueueProcessorTests.cs b/tests/AzureStorage.QueueService.Tests/QueueProcessorTests.cs index ff540aa..8583135 100644 --- a/tests/AzureStorage.QueueService.Tests/QueueProcessorTests.cs +++ b/tests/AzureStorage.QueueService.Tests/QueueProcessorTests.cs @@ -877,6 +877,25 @@ public async Task Rate_Limiter_Caps_The_Receive() h.Queue.Receives[0].Take.Should().Be(2, "five handlers are free but only two permits are"); } + [Fact(DisplayName = "With a rate limiter that has a permit for every free handler, messages are still received and handled")] + public async Task Rate_Limiter_With_Permits_To_Spare() + { + using var limiter = new System.Threading.RateLimiting.TokenBucketRateLimiter(new System.Threading.RateLimiting.TokenBucketRateLimiterOptions + { + TokenLimit = 10, + TokensPerPeriod = 1, + ReplenishmentPeriod = TimeSpan.FromHours(1), + AutoReplenishment = false, + }); + await using var h = new Harness(output, o => o.RateLimiter = limiter); + h.Queue.Then(h.Message("1")); + + await h.StartAsync(); + await h.Ops.WaitForAsync("delete"); + + h.Logs.Entries.Should().NotContain(e => e.Level >= LogLevel.Error, "no slot is left to give back, and that is not a failed receive"); + } + [Fact(DisplayName = "MaxRenewalDuration stops renewing a handler's lease and says so, so a hung handler cannot hide its message for ever")] public async Task Renewal_Stops_At_MaxRenewalDuration() {