From b173fb11cfbf2f30d4867b993f675292a46d35c8 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Mon, 5 Oct 2026 16:13:22 -0500 Subject: [PATCH 1/9] Round-trip GroupId in Azure Storage queue envelope Depends on https://github.com/FoundatioFx/Foundatio/pull/579 --- .../Queues/AzureStorageQueue.cs | 16 ++++++++++++++-- .../Queues/AzureStorageQueueTests.cs | 12 ++++++++++++ .../Queues/LegacyAzureStorageQueueTests.cs | 12 ++++++++++++ 3 files changed, 38 insertions(+), 2 deletions(-) diff --git a/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs b/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs index a155a36..63ec565 100644 --- a/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs +++ b/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs @@ -87,6 +87,7 @@ await Task.WhenAll( { CorrelationId = options.CorrelationId, Properties = options.Properties, + GroupId = options.GroupId, Data = data }; messageBodyBytes = _serializer.SerializeToBytes(envelope); @@ -103,7 +104,10 @@ await Task.WhenAll( visibilityTimeout: options.DeliveryDelay, cancellationToken: CancellationToken.None).AnyContext(); - var entry = new QueueEntry(response.Value.MessageId, null, data, this, _timeProvider.GetLocalNow().UtcDateTime, 0); + var entry = new QueueEntry(response.Value.MessageId, options.CorrelationId, data, this, _timeProvider.GetLocalNow().UtcDateTime, 0) { GroupId = options.GroupId }; + foreach (var property in options.Properties) + entry.Properties[property.Key] = property.Value; + await OnEnqueuedAsync(entry).AnyContext(); _logger.LogTrace("Enqueued message {MessageId}", response.Value.MessageId); @@ -179,6 +183,7 @@ await Task.WhenAll( T? data; string? correlationId = null; + string? groupId = null; IDictionary? properties = null; Exception? deserializeException = null; @@ -194,6 +199,7 @@ await Task.WhenAll( { data = envelope.Data; correlationId = envelope.CorrelationId; + groupId = envelope.GroupId; properties = envelope.Properties; } else @@ -231,7 +237,7 @@ await Task.WhenAll( return null; } - var entry = new AzureStorageQueueEntry(message, correlationId, properties, data, this); + var entry = new AzureStorageQueueEntry(message, correlationId, properties, data, this) { GroupId = groupId }; if (entry.Attempts > _options.Retries + 1) { @@ -482,6 +488,7 @@ private async Task DeadLetterMessageAsync(AzureStorageQueueEntry entry) { CorrelationId = entry.CorrelationId, Properties = entry.Properties, + GroupId = entry.GroupId, Data = entry.Value }; messageBody = new BinaryData(_serializer.SerializeToBytes(envelope)); @@ -520,6 +527,11 @@ internal record QueueMessageEnvelope where T : class /// public IDictionary? Properties { get; init; } + /// + /// Logical group or tenant key the message was enqueued with + /// + public string? GroupId { get; init; } + /// /// The actual message payload /// diff --git a/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs b/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs index 65c2c1e..37da4c4 100644 --- a/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs +++ b/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs @@ -112,6 +112,12 @@ public override Task DequeueAsync_AfterAbandonWithMutatedValue_ReturnsOriginalVa return base.DequeueAsync_AfterAbandonWithMutatedValue_ReturnsOriginalValueAsync(); } + [Fact] + public override Task AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync() + { + return base.AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync(); + } + [Fact] public override Task DequeueAsync_WithDispose_AutoAbandonsEntryAsync() { @@ -142,6 +148,12 @@ public override Task DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync( return base.DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync(); } + [Fact] + public override Task EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync() + { + return base.EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync(); + } + [Fact] public override Task EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync() { diff --git a/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs b/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs index ff63b54..d136d0a 100644 --- a/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs +++ b/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs @@ -142,6 +142,18 @@ public override Task DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync( return base.DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync(); } + [Fact(Skip = "Legacy mode does not support GroupId")] + public override Task AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync() + { + return base.AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync(); + } + + [Fact(Skip = "Legacy mode does not support GroupId")] + public override Task EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync() + { + return base.EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync(); + } + [Fact] public override Task EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync() { From da26270010d41eb22a951ddbb2f65b73059f25e9 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Mon, 5 Oct 2026 18:40:07 -0500 Subject: [PATCH 2/9] Align legacy enqueue event metadata with stored messages --- .../Queues/AzureStorageQueue.cs | 13 ++++- .../Queues/AzureStorageQueueOptions.cs | 5 +- .../Queues/AzureStorageQueueTests.cs | 58 +++++++++++++++++++ 3 files changed, 71 insertions(+), 5 deletions(-) diff --git a/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs b/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs index 63ec565..3e1b334 100644 --- a/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs +++ b/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs @@ -104,9 +104,16 @@ await Task.WhenAll( visibilityTimeout: options.DeliveryDelay, cancellationToken: CancellationToken.None).AnyContext(); - var entry = new QueueEntry(response.Value.MessageId, options.CorrelationId, data, this, _timeProvider.GetLocalNow().UtcDateTime, 0) { GroupId = options.GroupId }; - foreach (var property in options.Properties) - entry.Properties[property.Key] = property.Value; + bool preservesMetadata = _options.CompatibilityMode == AzureStorageQueueCompatibilityMode.Default; + var entry = new QueueEntry(response.Value.MessageId, preservesMetadata ? options.CorrelationId : null, data, this, _timeProvider.GetLocalNow().UtcDateTime, 0) + { + GroupId = preservesMetadata ? options.GroupId : null + }; + if (preservesMetadata) + { + foreach (var property in options.Properties) + entry.Properties[property.Key] = property.Value; + } await OnEnqueuedAsync(entry).AnyContext(); diff --git a/src/Foundatio.AzureStorage/Queues/AzureStorageQueueOptions.cs b/src/Foundatio.AzureStorage/Queues/AzureStorageQueueOptions.cs index b1c0d39..a6eb07b 100644 --- a/src/Foundatio.AzureStorage/Queues/AzureStorageQueueOptions.cs +++ b/src/Foundatio.AzureStorage/Queues/AzureStorageQueueOptions.cs @@ -10,12 +10,13 @@ public enum AzureStorageQueueCompatibilityMode { /// /// Default mode: Uses message envelope for full metadata support. - /// Supports CorrelationId and Properties. New installations should use this mode. + /// Supports GroupId, CorrelationId and Properties. New installations should use this mode. /// Default = 0, /// /// Legacy mode: Raw message body with Base64 encoding for v11 SDK compatibility. + /// GroupId, CorrelationId and Properties are not persisted or included in the Enqueued entry. /// Use this for backward compatibility with existing queues that have messages /// written with the v11 Microsoft.Azure.Storage.Queue SDK. /// @@ -41,7 +42,7 @@ public class AzureStorageQueueOptions : SharedQueueOptions where T : class /// /// Controls message format compatibility. - /// Default mode uses an envelope wrapper that supports CorrelationId and Properties. + /// Default mode uses an envelope wrapper that supports GroupId, CorrelationId and Properties. /// Legacy mode is provided for backward compatibility with existing queues that have messages /// written with the v11 Microsoft.Azure.Storage.Queue SDK (uses Base64 encoding and raw payload). /// diff --git a/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs b/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs index 37da4c4..4812fe9 100644 --- a/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs +++ b/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs @@ -148,6 +148,64 @@ public override Task DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync( return base.DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync(); } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task EnqueueAsync_WithCompatibilityMode_EventMatchesDequeuedMetadataAsync(bool legacy) + { + // Arrange + string? connectionString = Configuration.GetConnectionString("AzureStorageConnectionString"); + Assert.False(String.IsNullOrEmpty(connectionString)); +#pragma warning disable CS0618 // Exercise the supported legacy wire format. + using var queue = new AzureStorageQueue(new AzureStorageQueueOptions + { + ConnectionString = connectionString, + Name = "foundatio-metadata-" + Guid.NewGuid().ToString("N"), + CompatibilityMode = legacy ? AzureStorageQueueCompatibilityMode.Legacy : AzureStorageQueueCompatibilityMode.Default, + MetricsPollingEnabled = false, + LoggerFactory = Log + }); +#pragma warning restore CS0618 + IQueueEntry? enqueued = null; + queue.Enqueued.AddHandler((_, args) => + { + enqueued = args.Entry; + return Task.CompletedTask; + }); + var options = new QueueEntryOptions { GroupId = "tenant", CorrelationId = "correlation" }; + options.Properties.Add("custom", "value"); + + // Act + await queue.EnqueueAsync(new SimpleWorkItem { Data = "payload" }, options); + var dequeued = await queue.DequeueAsync(TimeSpan.FromSeconds(5)); + + // Assert + Assert.NotNull(enqueued); + Assert.NotNull(dequeued); + Assert.Equal("payload", dequeued.Value.Data); + Assert.Equal(legacy ? null : "tenant", dequeued.GroupId); + Assert.Equal(legacy ? null : "correlation", dequeued.CorrelationId); + Assert.Equal(dequeued.GroupId, enqueued.GroupId); + Assert.Equal(dequeued.CorrelationId, enqueued.CorrelationId); + Assert.Equal(dequeued.Properties.Count, enqueued.Properties.Count); + if (legacy) + Assert.Empty(dequeued.Properties); + else + { + Assert.Equal("value", dequeued.Properties["custom"]); + Assert.Equal("value", enqueued.Properties["custom"]); + } + Assert.Equal("tenant", options.GroupId); + Assert.Equal("correlation", options.CorrelationId); + await dequeued.CompleteAsync(); + } + + [Fact] + public override Task EnqueueAsync_WithEmptyGroupId_EnqueuesWithoutGroupAsync() + { + return base.EnqueueAsync_WithEmptyGroupId_EnqueuesWithoutGroupAsync(); + } + [Fact] public override Task EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync() { From bd80c80dcc4c4726fbf1bc8882545fb99aa6100f Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Mon, 5 Oct 2026 22:22:21 -0500 Subject: [PATCH 3/9] Pin guarded build workflow for feature branch validation --- .github/workflows/build.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index c10208e..4dd0c32 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -3,7 +3,7 @@ on: [push, pull_request] jobs: build: - uses: FoundatioFx/Foundatio/.github/workflows/build-workflow.yml@main + uses: FoundatioFx/Foundatio/.github/workflows/build-workflow.yml@a28d0f736d9124431de09764c56aff77fbdc1d5d secrets: inherit with: new-test-runner: true From 42ab1e95dbd21ba48771077fadac97e8caedf90b Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Tue, 6 Oct 2026 11:26:10 -0500 Subject: [PATCH 4/9] Consume the queue-group-id nightly of Foundatio and Foundatio.TestHarness The GroupId APIs are not in 13.0.4. Use the nightly until a release is available. --- src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj | 2 +- tests/Directory.Build.props | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj b/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj index be5e135..4c764f5 100644 --- a/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj +++ b/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj @@ -7,7 +7,7 @@ - + diff --git a/tests/Directory.Build.props b/tests/Directory.Build.props index 03bc80d..a4d698f 100644 --- a/tests/Directory.Build.props +++ b/tests/Directory.Build.props @@ -12,7 +12,7 @@ - + From 83ee3eae7b6ac7e5818b5a2a380a24e1fc0fffdb Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Tue, 6 Oct 2026 16:43:19 -0500 Subject: [PATCH 5/9] Forward the original body to the dead-letter queue and back off on worker errors Dead-lettering rebuilt the message from the deserialized value, which a handler may have changed. It now forwards the original body, keeping the payload, GroupId, CorrelationId and properties exactly. The worker loop logs SDK timeouts, waits at least one second after a dequeue error, and does not abandon during shutdown. The metadata test no longer asserts on missing config and deletes the queue it creates. Legacy mode now runs the reused-options test. --- .../Queues/AzureStorageQueue.cs | 48 +++++----- .../Queues/AzureStorageQueueTests.cs | 87 ++++++++++++------- .../Queues/LegacyAzureStorageQueueTests.cs | 24 +++++ 3 files changed, 103 insertions(+), 56 deletions(-) diff --git a/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs b/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs index 3e1b334..dfdaecb 100644 --- a/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs +++ b/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs @@ -23,6 +23,7 @@ public class AzureStorageQueue : QueueBase> wh private long _completedCount; private long _abandonedCount; private long _workerErrorCount; + private static readonly TimeSpan MinWorkerErrorDelay = TimeSpan.FromSeconds(1); private bool _queueCreated; public AzureStorageQueue(AzureStorageQueueOptions options) : base(options) @@ -452,9 +453,16 @@ protected override void StartWorkingImpl(Func, CancellationToken, { queueEntry = await DequeueImplAsync(linkedCancellationToken.Token).AnyContext(); } - catch (OperationCanceledException) + catch (OperationCanceledException) when (linkedCancellationToken.IsCancellationRequested) { } + catch (Exception ex) { - // Ignore cancellation + Interlocked.Increment(ref _workerErrorCount); + _logger.LogError(ex, "Error on Dequeue: {Message}", ex.Message); + try + { + await _timeProvider.Delay(_options.DequeueInterval > MinWorkerErrorDelay ? _options.DequeueInterval : MinWorkerErrorDelay, linkedCancellationToken.Token).AnyContext(); + } + catch (OperationCanceledException) { } } if (linkedCancellationToken.IsCancellationRequested || queueEntry == null) @@ -471,8 +479,17 @@ protected override void StartWorkingImpl(Func, CancellationToken, Interlocked.Increment(ref _workerErrorCount); _logger.LogError(ex, "Worker error: {Message}", ex.Message); - if (!queueEntry.IsAbandoned && !queueEntry.IsCompleted) - await queueEntry.AbandonAsync().AnyContext(); + if (!queueEntry.IsAbandoned && !queueEntry.IsCompleted && !linkedCancellationToken.IsCancellationRequested) + { + try + { + await queueEntry.AbandonAsync().AnyContext(); + } + catch (Exception abandonEx) + { + _logger.LogError(abandonEx, "Worker error abandoning queue entry: {Message}", abandonEx.Message); + } + } } } @@ -484,28 +501,7 @@ private async Task DeadLetterMessageAsync(AzureStorageQueueEntry entry) { _logger.LogInformation("Exceeded retry limit ({Attempts}/{Retries}), moving message {QueueEntryId} to dead letter", entry.Attempts, _options.Retries, entry.Id); - BinaryData messageBody; - if (entry.Value is null) - { - messageBody = entry.UnderlyingMessage.Body; - } - else if (_options.CompatibilityMode == AzureStorageQueueCompatibilityMode.Default) - { - var envelope = new QueueMessageEnvelope - { - CorrelationId = entry.CorrelationId, - Properties = entry.Properties, - GroupId = entry.GroupId, - Data = entry.Value - }; - messageBody = new BinaryData(_serializer.SerializeToBytes(envelope)); - } - else - { - messageBody = new BinaryData(_serializer.SerializeToBytes(entry.Value)); - } - - await _deadletterQueueClient.Value.SendMessageAsync(messageBody).AnyContext(); + await _deadletterQueueClient.Value.SendMessageAsync(entry.UnderlyingMessage.Body).AnyContext(); await _queueClient.Value.DeleteMessageAsync(entry.UnderlyingMessage.MessageId, entry.PopReceipt).AnyContext(); } diff --git a/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs b/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs index 4812fe9..3273b81 100644 --- a/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs +++ b/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs @@ -118,6 +118,12 @@ public override Task AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync() return base.AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync(); } + [Fact] + public override Task AbandonAsync_WithGroupIdAndRetryDelay_PreservesGroupIdOnRetryAsync() + { + return base.AbandonAsync_WithGroupIdAndRetryDelay_PreservesGroupIdOnRetryAsync(); + } + [Fact] public override Task DequeueAsync_WithDispose_AutoAbandonsEntryAsync() { @@ -155,7 +161,9 @@ public async Task EnqueueAsync_WithCompatibilityMode_EventMatchesDequeuedMetadat { // Arrange string? connectionString = Configuration.GetConnectionString("AzureStorageConnectionString"); - Assert.False(String.IsNullOrEmpty(connectionString)); + if (String.IsNullOrEmpty(connectionString)) + return; + #pragma warning disable CS0618 // Exercise the supported legacy wire format. using var queue = new AzureStorageQueue(new AzureStorageQueueOptions { @@ -166,38 +174,51 @@ public async Task EnqueueAsync_WithCompatibilityMode_EventMatchesDequeuedMetadat LoggerFactory = Log }); #pragma warning restore CS0618 - IQueueEntry? enqueued = null; - queue.Enqueued.AddHandler((_, args) => + try { - enqueued = args.Entry; - return Task.CompletedTask; - }); - var options = new QueueEntryOptions { GroupId = "tenant", CorrelationId = "correlation" }; - options.Properties.Add("custom", "value"); - - // Act - await queue.EnqueueAsync(new SimpleWorkItem { Data = "payload" }, options); - var dequeued = await queue.DequeueAsync(TimeSpan.FromSeconds(5)); - - // Assert - Assert.NotNull(enqueued); - Assert.NotNull(dequeued); - Assert.Equal("payload", dequeued.Value.Data); - Assert.Equal(legacy ? null : "tenant", dequeued.GroupId); - Assert.Equal(legacy ? null : "correlation", dequeued.CorrelationId); - Assert.Equal(dequeued.GroupId, enqueued.GroupId); - Assert.Equal(dequeued.CorrelationId, enqueued.CorrelationId); - Assert.Equal(dequeued.Properties.Count, enqueued.Properties.Count); - if (legacy) - Assert.Empty(dequeued.Properties); - else + IQueueEntry? enqueued = null; + queue.Enqueued.AddHandler((_, args) => + { + enqueued = args.Entry; + return Task.CompletedTask; + }); + var options = new QueueEntryOptions { GroupId = "tenant", CorrelationId = "correlation" }; + options.Properties.Add("custom", "value"); + + // Act + await queue.EnqueueAsync(new SimpleWorkItem { Data = "payload" }, options); + var dequeued = await queue.DequeueAsync(TimeSpan.FromSeconds(5)); + + // Assert + Assert.NotNull(enqueued); + Assert.NotNull(dequeued); + Assert.Equal("payload", dequeued.Value.Data); + Assert.Equal(legacy ? null : "tenant", dequeued.GroupId); + Assert.Equal(legacy ? null : "correlation", dequeued.CorrelationId); + Assert.Equal(dequeued.GroupId, enqueued.GroupId); + Assert.Equal(dequeued.CorrelationId, enqueued.CorrelationId); + Assert.Equal(dequeued.Properties.Count, enqueued.Properties.Count); + if (legacy) + Assert.Empty(dequeued.Properties); + else + { + Assert.Equal("value", dequeued.Properties["custom"]); + Assert.Equal("value", enqueued.Properties["custom"]); + } + Assert.Equal("tenant", options.GroupId); + Assert.Equal("correlation", options.CorrelationId); + await dequeued.CompleteAsync(); + } + finally { - Assert.Equal("value", dequeued.Properties["custom"]); - Assert.Equal("value", enqueued.Properties["custom"]); + await queue.DeleteQueueAsync(); } - Assert.Equal("tenant", options.GroupId); - Assert.Equal("correlation", options.CorrelationId); - await dequeued.CompleteAsync(); + } + + [Fact] + public override Task EnqueueAsync_WhenEnqueuingHandlerClearsGroupId_EnqueuesWithoutGroupAsync() + { + return base.EnqueueAsync_WhenEnqueuingHandlerClearsGroupId_EnqueuesWithoutGroupAsync(); } [Fact] @@ -218,6 +239,12 @@ public override Task EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmp return base.EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync(); } + [Fact] + public override Task EnqueueAsync_WithReusedOptions_DoesNotChangeCallerOptionsAsync() + { + return base.EnqueueAsync_WithReusedOptions_DoesNotChangeCallerOptionsAsync(); + } + [Fact(Skip = "Azure Storage Queues do not support custom entry IDs")] public override Task EnqueueAsync_WithUniqueId_UsesProvidedIdAsync() { diff --git a/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs b/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs index d136d0a..3a2a56c 100644 --- a/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs +++ b/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs @@ -148,12 +148,36 @@ public override Task AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync() return base.AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync(); } + [Fact(Skip = "Legacy mode does not support GroupId")] + public override Task AbandonAsync_WithGroupIdAndRetryDelay_PreservesGroupIdOnRetryAsync() + { + return base.AbandonAsync_WithGroupIdAndRetryDelay_PreservesGroupIdOnRetryAsync(); + } + + [Fact] + public override Task EnqueueAsync_WhenEnqueuingHandlerClearsGroupId_EnqueuesWithoutGroupAsync() + { + return base.EnqueueAsync_WhenEnqueuingHandlerClearsGroupId_EnqueuesWithoutGroupAsync(); + } + + [Fact] + public override Task EnqueueAsync_WithEmptyGroupId_EnqueuesWithoutGroupAsync() + { + return base.EnqueueAsync_WithEmptyGroupId_EnqueuesWithoutGroupAsync(); + } + [Fact(Skip = "Legacy mode does not support GroupId")] public override Task EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync() { return base.EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync(); } + [Fact] + public override Task EnqueueAsync_WithReusedOptions_DoesNotChangeCallerOptionsAsync() + { + return base.EnqueueAsync_WithReusedOptions_DoesNotChangeCallerOptionsAsync(); + } + [Fact] public override Task EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync() { From 86bab8115f72e6c91e3055c22dd72a1ef37abf8f Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Thu, 8 Oct 2026 11:51:48 -0500 Subject: [PATCH 6/9] Limit queue changes to GroupId; restore main worker loop and dead-letter envelope --- .../Queues/AzureStorageQueue.cs | 59 +++++++++---------- .../Queues/AzureStorageQueueTests.cs | 54 ++++++++++++++--- 2 files changed, 74 insertions(+), 39 deletions(-) diff --git a/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs b/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs index dfdaecb..432a22a 100644 --- a/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs +++ b/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs @@ -23,7 +23,6 @@ public class AzureStorageQueue : QueueBase> wh private long _completedCount; private long _abandonedCount; private long _workerErrorCount; - private static readonly TimeSpan MinWorkerErrorDelay = TimeSpan.FromSeconds(1); private bool _queueCreated; public AzureStorageQueue(AzureStorageQueueOptions options) : base(options) @@ -105,17 +104,10 @@ await Task.WhenAll( visibilityTimeout: options.DeliveryDelay, cancellationToken: CancellationToken.None).AnyContext(); - bool preservesMetadata = _options.CompatibilityMode == AzureStorageQueueCompatibilityMode.Default; - var entry = new QueueEntry(response.Value.MessageId, preservesMetadata ? options.CorrelationId : null, data, this, _timeProvider.GetLocalNow().UtcDateTime, 0) + var entry = new QueueEntry(response.Value.MessageId, null, data, this, _timeProvider.GetLocalNow().UtcDateTime, 0) { - GroupId = preservesMetadata ? options.GroupId : null + GroupId = _options.CompatibilityMode == AzureStorageQueueCompatibilityMode.Default ? options.GroupId : null }; - if (preservesMetadata) - { - foreach (var property in options.Properties) - entry.Properties[property.Key] = property.Value; - } - await OnEnqueuedAsync(entry).AnyContext(); _logger.LogTrace("Enqueued message {MessageId}", response.Value.MessageId); @@ -453,16 +445,9 @@ protected override void StartWorkingImpl(Func, CancellationToken, { queueEntry = await DequeueImplAsync(linkedCancellationToken.Token).AnyContext(); } - catch (OperationCanceledException) when (linkedCancellationToken.IsCancellationRequested) { } - catch (Exception ex) + catch (OperationCanceledException) { - Interlocked.Increment(ref _workerErrorCount); - _logger.LogError(ex, "Error on Dequeue: {Message}", ex.Message); - try - { - await _timeProvider.Delay(_options.DequeueInterval > MinWorkerErrorDelay ? _options.DequeueInterval : MinWorkerErrorDelay, linkedCancellationToken.Token).AnyContext(); - } - catch (OperationCanceledException) { } + // Ignore cancellation } if (linkedCancellationToken.IsCancellationRequested || queueEntry == null) @@ -479,17 +464,8 @@ protected override void StartWorkingImpl(Func, CancellationToken, Interlocked.Increment(ref _workerErrorCount); _logger.LogError(ex, "Worker error: {Message}", ex.Message); - if (!queueEntry.IsAbandoned && !queueEntry.IsCompleted && !linkedCancellationToken.IsCancellationRequested) - { - try - { - await queueEntry.AbandonAsync().AnyContext(); - } - catch (Exception abandonEx) - { - _logger.LogError(abandonEx, "Worker error abandoning queue entry: {Message}", abandonEx.Message); - } - } + if (!queueEntry.IsAbandoned && !queueEntry.IsCompleted) + await queueEntry.AbandonAsync().AnyContext(); } } @@ -501,7 +477,28 @@ private async Task DeadLetterMessageAsync(AzureStorageQueueEntry entry) { _logger.LogInformation("Exceeded retry limit ({Attempts}/{Retries}), moving message {QueueEntryId} to dead letter", entry.Attempts, _options.Retries, entry.Id); - await _deadletterQueueClient.Value.SendMessageAsync(entry.UnderlyingMessage.Body).AnyContext(); + BinaryData messageBody; + if (entry.Value is null) + { + messageBody = entry.UnderlyingMessage.Body; + } + else if (_options.CompatibilityMode == AzureStorageQueueCompatibilityMode.Default) + { + var envelope = new QueueMessageEnvelope + { + CorrelationId = entry.CorrelationId, + Properties = entry.Properties, + GroupId = entry.GroupId, + Data = entry.Value + }; + messageBody = new BinaryData(_serializer.SerializeToBytes(envelope)); + } + else + { + messageBody = new BinaryData(_serializer.SerializeToBytes(entry.Value)); + } + + await _deadletterQueueClient.Value.SendMessageAsync(messageBody).AnyContext(); await _queueClient.Value.DeleteMessageAsync(entry.UnderlyingMessage.MessageId, entry.PopReceipt).AnyContext(); } diff --git a/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs b/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs index 3273b81..d90e207 100644 --- a/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs +++ b/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs @@ -1,5 +1,7 @@ using System; +using System.Linq; using System.Text; +using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Azure.Storage.Queues; @@ -157,7 +159,7 @@ public override Task DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync( [Theory] [InlineData(false)] [InlineData(true)] - public async Task EnqueueAsync_WithCompatibilityMode_EventMatchesDequeuedMetadataAsync(bool legacy) + public async Task EnqueueAsync_WithCompatibilityMode_PersistsMetadataOnlyInDefaultModeAsync(bool legacy) { // Arrange string? connectionString = Configuration.GetConnectionString("AzureStorageConnectionString"); @@ -196,17 +198,10 @@ public async Task EnqueueAsync_WithCompatibilityMode_EventMatchesDequeuedMetadat Assert.Equal(legacy ? null : "tenant", dequeued.GroupId); Assert.Equal(legacy ? null : "correlation", dequeued.CorrelationId); Assert.Equal(dequeued.GroupId, enqueued.GroupId); - Assert.Equal(dequeued.CorrelationId, enqueued.CorrelationId); - Assert.Equal(dequeued.Properties.Count, enqueued.Properties.Count); if (legacy) Assert.Empty(dequeued.Properties); else - { Assert.Equal("value", dequeued.Properties["custom"]); - Assert.Equal("value", enqueued.Properties["custom"]); - } - Assert.Equal("tenant", options.GroupId); - Assert.Equal("correlation", options.CorrelationId); await dequeued.CompleteAsync(); } finally @@ -503,4 +498,47 @@ public async Task DequeueAsync_WithCorruptMessage_MovesToDeadletterAsync() await queue.DeleteQueueAsync(); await new QueueClient(connectionString, poisonQueueName).DeleteIfExistsAsync(TestContext.Current.CancellationToken); } + + [Fact] + public async Task AbandonAsync_WhenRetriesExceeded_PreservesGroupIdInPoisonMessageAsync() + { + // Arrange + string? connectionString = Configuration.GetConnectionString("AzureStorageConnectionString"); + if (String.IsNullOrEmpty(connectionString)) + return; + + var queueName = "foundatio-" + Guid.NewGuid().ToString("N").Substring(10); + var poisonQueueName = $"{queueName}-poison"; + + using var queue = new AzureStorageQueue(o => o + .ConnectionString(connectionString) + .Name(queueName) + .Retries(0) + .WorkItemTimeout(TimeSpan.FromSeconds(30)) + .DequeueInterval(TimeSpan.FromSeconds(1)) + .MetricsPollingInterval(TimeSpan.Zero) + .LoggerFactory(Log)); + + try + { + await queue.EnqueueAsync(new SimpleWorkItem { Data = "poison" }, new QueueEntryOptions { GroupId = "tenant" }); + var entry = await queue.DequeueAsync(TimeSpan.FromSeconds(5)); + Assert.NotNull(entry); + + // Act + await entry.AbandonAsync(); + + // Assert + var poisonMessage = await new QueueClient(connectionString, poisonQueueName).ReceiveMessageAsync(cancellationToken: TestCancellationToken); + Assert.NotNull(poisonMessage.Value); + using var envelope = JsonDocument.Parse(poisonMessage.Value.Body.ToArray()); + var groupId = envelope.RootElement.EnumerateObject().Single(p => String.Equals(p.Name, "GroupId", StringComparison.OrdinalIgnoreCase)); + Assert.Equal("tenant", groupId.Value.GetString()); + } + finally + { + await queue.DeleteQueueAsync(); + await new QueueClient(connectionString, poisonQueueName).DeleteIfExistsAsync(TestContext.Current.CancellationToken); + } + } } From de98dcf81ed096d80b5240a21db77e18ec3efe79 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Thu, 8 Oct 2026 14:03:26 -0500 Subject: [PATCH 7/9] Use matching GroupId core and TestHarness preview --- src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj | 2 +- tests/Directory.Build.props | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj b/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj index 4c764f5..41d5be0 100644 --- a/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj +++ b/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj @@ -7,7 +7,7 @@ - + diff --git a/tests/Directory.Build.props b/tests/Directory.Build.props index a4d698f..0e6af72 100644 --- a/tests/Directory.Build.props +++ b/tests/Directory.Build.props @@ -12,7 +12,7 @@ - + From 7013937878f9450e20182e082f55f329841d8610 Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Thu, 8 Oct 2026 13:35:33 -0500 Subject: [PATCH 8/9] Use the shared QueueBase worker loop Deletes the provider-specific StartWorkingImpl loop in favor of QueueBase.StartWorker (backoff on dequeue errors, cancellation-safe stop, safe abandon/auto-complete) and runs the shared worker fault-injection tests. Requires the Foundatio core change on feature/queue-worker-loop. --- .../Queues/AzureStorageQueue.cs | 55 ++----------------- .../Queues/AzureStorageQueueTests.cs | 18 ++++++ .../Queues/LegacyAzureStorageQueueTests.cs | 18 ++++++ 3 files changed, 42 insertions(+), 49 deletions(-) diff --git a/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs b/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs index 432a22a..6510f5e 100644 --- a/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs +++ b/src/Foundatio.AzureStorage/Queues/AzureStorageQueue.cs @@ -22,7 +22,6 @@ public class AzureStorageQueue : QueueBase> wh private long _dequeuedCount; private long _completedCount; private long _abandonedCount; - private long _workerErrorCount; private bool _queueCreated; public AzureStorageQueue(AzureStorageQueueOptions options) : base(options) @@ -346,7 +345,7 @@ protected override async Task GetQueueStatsImplAsync() Dequeued = _dequeuedCount, Completed = _completedCount, Abandoned = _abandonedCount, - Errors = _workerErrorCount, + Errors = WorkerErrorCount, Timeouts = 0 }; @@ -366,7 +365,7 @@ protected override async Task GetQueueStatsImplAsync() Dequeued = _dequeuedCount, Completed = _completedCount, Abandoned = _abandonedCount, - Errors = _workerErrorCount, + Errors = WorkerErrorCount, Timeouts = 0 // Azure handles visibility timeout natively; client-side tracking not meaningful }; } @@ -383,7 +382,7 @@ protected override QueueStats GetMetricsQueueStats() Dequeued = _dequeuedCount, Completed = _completedCount, Abandoned = _abandonedCount, - Errors = _workerErrorCount, + Errors = WorkerErrorCount, Timeouts = 0 }; @@ -402,7 +401,7 @@ protected override QueueStats GetMetricsQueueStats() Dequeued = _dequeuedCount, Completed = _completedCount, Abandoned = _abandonedCount, - Errors = _workerErrorCount, + Errors = WorkerErrorCount, Timeouts = 0 }; } @@ -420,7 +419,7 @@ await Task.WhenAll( _dequeuedCount = 0; _completedCount = 0; _abandonedCount = 0; - _workerErrorCount = 0; + ResetWorkerErrorCount(); sw.Stop(); _logger.LogTrace("Deleting queue took {Elapsed:g}", sw.Elapsed); @@ -428,49 +427,7 @@ await Task.WhenAll( protected override void StartWorkingImpl(Func, CancellationToken, Task> handler, bool autoComplete, CancellationToken cancellationToken) { - ArgumentNullException.ThrowIfNull(handler); - - var linkedCancellationToken = GetLinkedDisposableCancellationTokenSource(cancellationToken); - - Task.Run(async () => - { - _logger.LogTrace("WorkerLoop Start {QueueName}", _options.Name); - - while (!linkedCancellationToken.IsCancellationRequested) - { - _logger.LogTrace("WorkerLoop Signaled {QueueName}", _options.Name); - - IQueueEntry? queueEntry = null; - try - { - queueEntry = await DequeueImplAsync(linkedCancellationToken.Token).AnyContext(); - } - catch (OperationCanceledException) - { - // Ignore cancellation - } - - if (linkedCancellationToken.IsCancellationRequested || queueEntry == null) - continue; - - try - { - await handler(queueEntry, linkedCancellationToken.Token).AnyContext(); - if (autoComplete && !queueEntry.IsAbandoned && !queueEntry.IsCompleted) - await queueEntry.CompleteAsync().AnyContext(); - } - catch (Exception ex) - { - Interlocked.Increment(ref _workerErrorCount); - _logger.LogError(ex, "Worker error: {Message}", ex.Message); - - if (!queueEntry.IsAbandoned && !queueEntry.IsCompleted) - await queueEntry.AbandonAsync().AnyContext(); - } - } - - _logger.LogTrace("Worker exiting: {QueueName} Cancel Requested: {IsCancellationRequested}", _options.Name, linkedCancellationToken.IsCancellationRequested); - }, linkedCancellationToken.Token).ContinueWith(t => linkedCancellationToken.Dispose()); + _ = StartWorker(handler, autoComplete, cancellationToken); } private async Task DeadLetterMessageAsync(AzureStorageQueueEntry entry) diff --git a/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs b/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs index d90e207..8d574f8 100644 --- a/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs +++ b/tests/Foundatio.AzureStorage.Tests/Queues/AzureStorageQueueTests.cs @@ -282,6 +282,24 @@ public override Task CanHandleErrorInWorkerAsync() return base.CanHandleErrorInWorkerAsync(); } + [Fact] + public override Task StartWorkingAsync_WhenDequeueThrows_KeepsWorkingAsync() + { + return base.StartWorkingAsync_WhenDequeueThrows_KeepsWorkingAsync(); + } + + [Fact] + public override Task StartWorkingAsync_WhenAbandonThrows_KeepsWorkingAsync() + { + return base.StartWorkingAsync_WhenAbandonThrows_KeepsWorkingAsync(); + } + + [Fact] + public override Task StartWorkingAsync_WhenCancelled_StopsWithoutWorkerErrorsAsync() + { + return base.StartWorkingAsync_WhenCancelled_StopsWithoutWorkerErrorsAsync(); + } + [Fact(Skip = "Azure Storage Queue handles visibility timeout natively; no client-side auto-abandon")] public override Task WorkItemsWillTimeoutAsync() { diff --git a/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs b/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs index 3a2a56c..4be76eb 100644 --- a/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs +++ b/tests/Foundatio.AzureStorage.Tests/Queues/LegacyAzureStorageQueueTests.cs @@ -196,6 +196,24 @@ public override Task CanHandleErrorInWorkerAsync() return base.CanHandleErrorInWorkerAsync(); } + [Fact] + public override Task StartWorkingAsync_WhenDequeueThrows_KeepsWorkingAsync() + { + return base.StartWorkingAsync_WhenDequeueThrows_KeepsWorkingAsync(); + } + + [Fact] + public override Task StartWorkingAsync_WhenAbandonThrows_KeepsWorkingAsync() + { + return base.StartWorkingAsync_WhenAbandonThrows_KeepsWorkingAsync(); + } + + [Fact] + public override Task StartWorkingAsync_WhenCancelled_StopsWithoutWorkerErrorsAsync() + { + return base.StartWorkingAsync_WhenCancelled_StopsWithoutWorkerErrorsAsync(); + } + [Fact(Skip = "Azure Storage Queue handles visibility timeout natively; no client-side auto-abandon")] public override Task WorkItemsWillTimeoutAsync() { From 31ad05f1142064ceeb8c256dd2c479ce129e8d7f Mon Sep 17 00:00:00 2001 From: Blake Niemyjski Date: Thu, 8 Oct 2026 13:39:37 -0500 Subject: [PATCH 9/9] Reference the Foundatio queue-worker-loop preview package --- src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj | 2 +- tests/Directory.Build.props | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj b/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj index 41d5be0..711580e 100644 --- a/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj +++ b/src/Foundatio.AzureStorage/Foundatio.AzureStorage.csproj @@ -7,7 +7,7 @@ - + diff --git a/tests/Directory.Build.props b/tests/Directory.Build.props index 0e6af72..d3cc5e5 100644 --- a/tests/Directory.Build.props +++ b/tests/Directory.Build.props @@ -12,7 +12,7 @@ - +