From 0008eb433417d97c2bab25ab892eba8fbaa20961 Mon Sep 17 00:00:00 2001 From: Yan Mahdych Date: Fri, 30 Oct 2026 10:47:14 +0200 Subject: [PATCH 1/3] Skip missing blob references when receiving queue messages Delete main-queue messages whose payload returns 404 BlobNotFound, including during transfer to quarantine, so healthy messages in the same receive batch can continue processing. Preserve handling of other storage errors and quarantine reads. Add five regression cases to MessageQueueTests. Verified all 19 MessageQueueTests against local Azurite. --- AzureBatchQueue.Tests/MessageQueueTests.cs | 101 ++++++++++++++++++++- AzureBatchQueue/MessageQueue.cs | 36 +++++++- 2 files changed, 130 insertions(+), 7 deletions(-) diff --git a/AzureBatchQueue.Tests/MessageQueueTests.cs b/AzureBatchQueue.Tests/MessageQueueTests.cs index 90a2739..e8163ab 100644 --- a/AzureBatchQueue.Tests/MessageQueueTests.cs +++ b/AzureBatchQueue.Tests/MessageQueueTests.cs @@ -1,3 +1,8 @@ +using System.Text.Json; +using Azure; +using Azure.Storage.Blobs; +using Azure.Storage.Queues; +using Azure.Storage.Queues.Models; using AzureBatchQueue.JsonNet; using AzureBatchQueue.Tests.Serializers; using FluentAssertions; @@ -168,6 +173,74 @@ public async Task When_exception_on_deserialize() Assert.DoesNotThrowAsync(async () => await gZipQueue.Queue.Receive()); } + [Test] + public async Task When_receiving_a_missing_blob_returns_healthy_messages() + { + using var queueTest = await Queue(); + await queueTest.Queue.Send(new string('*', 65 * 1024)); + await queueTest.DeleteMessageBlob(); + var largeMessage = new string('h', 65 * 1024); + await queueTest.Queue.Send("healthy-inline"); + await queueTest.Queue.Send(largeMessage); + + var messages = await queueTest.Queue.Receive(3); + + messages.Select(x => x.Item).Should().BeEquivalentTo(new[] { "healthy-inline", largeMessage }); + (await queueTest.MessageCount()).Should().Be(2); + (await queueTest.MessageCount(fromQuarantine: true)).Should().Be(0); + } + + [Test] + public async Task When_quarantining_a_missing_blob_keeps_healthy_messages() + { + using var queueTest = await Queue(maxDequeueCount: 0); + await queueTest.Queue.Send(new string('*', 65 * 1024)); + await queueTest.DeleteMessageBlob(); + var largeMessage = new string('h', 65 * 1024); + await queueTest.Queue.Send("healthy-inline"); + await queueTest.Queue.Send(largeMessage); + + (await queueTest.Queue.Receive(3)).Should().BeEmpty(); + + (await queueTest.MessageCount()).Should().Be(0); + (await queueTest.MessageCount(fromQuarantine: true)).Should().Be(2); + var quarantined = await queueTest.Queue.ReceiveFromQuarantine(2); + quarantined.Select(x => x.Item).Should().BeEquivalentTo(new[] { "healthy-inline", largeMessage }); + } + + [TestCase(0)] + [TestCase(5)] + public async Task When_container_is_missing_keeps_message_for_retry(int maxDequeueCount) + { + using var queueTest = await Queue(maxDequeueCount); + await queueTest.Queue.Send(new string('*', 65 * 1024)); + await queueTest.Blobs.DeleteAsync(); + + if (maxDequeueCount == 0) + (await queueTest.Queue.Receive()).Should().BeEmpty(); + else + { + var error = Assert.ThrowsAsync(async () => await queueTest.Queue.Receive()); + error!.ErrorCode.Should().Be("ContainerNotFound"); + } + + (await queueTest.MessageCount()).Should().Be(1); + } + + [Test] + public async Task When_receiving_a_missing_blob_from_quarantine_keeps_message() + { + using var queueTest = await Queue(); + await queueTest.Queue.QuarantineData(new string('*', 65 * 1024), CancellationToken.None); + await queueTest.DeleteMessageBlob(fromQuarantine: true); + + var error = Assert.ThrowsAsync(async () => await queueTest.Queue.ReceiveFromQuarantine()); + + error!.ErrorCode.Should().Be("BlobNotFound"); + (await queueTest.MessageCount(fromQuarantine: true)).Should().Be(1); + (await queueTest.MessageCount()).Should().Be(0); + } + static async Task> Queue(int maxDequeueCount = 5, IMessageQueueSerializer? serializer = null) { var queue = new QueueTest(maxDequeueCount, serializer); @@ -177,16 +250,40 @@ static async Task> Queue(int maxDequeueCount = 5, IMessageQueueS class QueueTest : IDisposable { + readonly QueueClient rawQueue; + readonly QueueClient quarantineQueue; + public QueueTest(int maxDequeueCount = 5, IMessageQueueSerializer? serializer = null) { - Queue = new MessageQueue("UseDevelopmentStorage=true", "test", maxDequeueCount: maxDequeueCount, serializer: serializer); + var connection = Environment.GetEnvironmentVariable("AZURE_BATCH_QUEUE_TEST_CONNECTION") ?? "UseDevelopmentStorage=true"; + Queue = new MessageQueue(connection, "test", maxDequeueCount: maxDequeueCount, serializer: serializer); + var options = new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 }; + rawQueue = new QueueClient(connection, "test", options); + quarantineQueue = new QueueClient(connection, "test-quarantine", options); + Blobs = new BlobContainerClient(connection, "overflow-test"); } public async Task Init() => await Queue.Init(); public MessageQueue Queue { get; } + public BlobContainerClient Blobs { get; } + + public async Task DeleteMessageBlob(bool fromQuarantine = false) + { + var queue = fromQuarantine ? quarantineQueue : rawQueue; + var message = (await queue.PeekMessagesAsync(1)).Value.Single(); + using var envelope = JsonDocument.Parse(message.Body.ToString()); + await Blobs.DeleteBlobAsync(envelope.RootElement.GetProperty("__MSQ_QUEUE_BLOBNAME__").GetString()!); + } + + public async Task MessageCount(bool fromQuarantine = false) + { + var queue = fromQuarantine ? quarantineQueue : rawQueue; + return (await queue.GetPropertiesAsync()).Value.ApproximateMessagesCount; + } + public void Dispose() { - Queue.Delete().GetAwaiter().GetResult(); + Task.WhenAll(Queue.Delete(), Blobs.DeleteIfExistsAsync()).GetAwaiter().GetResult(); } } diff --git a/AzureBatchQueue/MessageQueue.cs b/AzureBatchQueue/MessageQueue.cs index d7e64b1..82dd5be 100644 --- a/AzureBatchQueue/MessageQueue.cs +++ b/AzureBatchQueue/MessageQueue.cs @@ -1,4 +1,5 @@ using System.Text.Json.Serialization; +using Azure; using Azure.Storage.Blobs; using Azure.Storage.Queues; using Azure.Storage.Queues.Models; @@ -117,7 +118,10 @@ async Task QuarantineMessage(QueueMessage queueMessage, CancellationToken ct = d if (IsBlobRef(queueMessage.Body, out blobRef) && blobRef != null) { - var newBlobRef = await CopyBlobForQuarantine(blobRef, queueMessage.MessageId, ct); + var newBlobRef = await TryCopyBlobForQuarantine(blobRef, queueMessage, ct); + if (newBlobRef == null) + return; + messageBody = Payload.OffloadedToBlob(newBlobRef).Data; } else @@ -145,17 +149,26 @@ async Task QuarantineMessage(QueueMessage queueMessage, CancellationToken ct = d // In this case, later we will insert a duplicate message to quarantine, with the same ref to blob. // When one message is processed, we remove the blob. The other message will have a BlobNotFound error. // So it's better to do some more heavy-lifting and copy blob content on quarantine. - async Task CopyBlobForQuarantine(BlobRef sourceBlobRef, string messageId, CancellationToken ct) + async Task TryCopyBlobForQuarantine(BlobRef sourceBlobRef, QueueMessage message, CancellationToken ct) { var sourceBlobClient = container.GetBlobClient(sourceBlobRef.BlobName); var newBlobRef = BlobRef.Create(); var destinationBlobClient = container.GetBlobClient(newBlobRef.BlobName); - var content = await sourceBlobClient.DownloadContentAsync(ct); - await destinationBlobClient.UploadAsync(content.Value.Content, ct); + BinaryData content; + try + { + content = (await sourceBlobClient.DownloadContentAsync(ct)).Value.Content; + } + catch (RequestFailedException ex) when (ex is { Status: 404, ErrorCode: "BlobNotFound" }) + { + await DeleteMissingBlobMessage(message, sourceBlobRef, ex, ct); + return null; + } + await destinationBlobClient.UploadAsync(content, ct); logger.LogInformation("Copied blob {SourceBlob} to {DestBlob} for quarantine message {MsgId}", - sourceBlobRef.BlobName, newBlobRef.BlobName, messageId); + sourceBlobRef.BlobName, newBlobRef.BlobName, message.MessageId); return newBlobRef; } @@ -193,6 +206,11 @@ public async Task QuarantineData(T item, CancellationToken ct) var blobData = await container.GetBlobClient(blobRef!.BlobName).DownloadContentAsync(ct); payload = blobData.Value.Content; } + catch (RequestFailedException ex) when (!fromQuarantine && ex is { Status: 404, ErrorCode: "BlobNotFound" }) + { + await DeleteMissingBlobMessage(m, blobRef!, ex, ct); + return null; + } catch (Exception ex) { logger.LogError(ex, "Exception when loading blob {BlobName} for {MessageId}", blobRef!.BlobName, m.MessageId); @@ -221,6 +239,14 @@ public async Task QuarantineData(T item, CancellationToken ct) return msg; } + async Task DeleteMissingBlobMessage(QueueMessage message, BlobRef blobRef, RequestFailedException error, CancellationToken ct) + { + await queue.DeleteMessageAsync(message.MessageId, message.PopReceipt, ct); + + logger.LogWarning(error, "Deleted message {MessageId} from queue {QueueName} because blob {BlobName} was not found.", + message.MessageId, queue.Name, blobRef.BlobName); + } + static bool IsBlobRef(BinaryData data, out BlobRef? o) { try From ac18b6f4f3f20836136a07a8422057d7bd4fd2c5 Mon Sep 17 00:00:00 2001 From: Yan Mahdych Date: Fri, 30 Oct 2026 10:57:48 +0200 Subject: [PATCH 2/3] Trace queue payload lifecycles with source-generated logs Record blob and queue-message identities across upload, send, receive, update, quarantine and deletion to investigate missing payloads and shared references. Correlate logical sends with operation and Azure request IDs, and record blob ETags and deletion results. Move existing MessageQueue and TimerBatch events to source-generated ILogger extensions while preserving their templates, levels and fields. Verified all 36 tests against local Azurite and the full solution build. Compared all 13 existing events through the consumer logging formatter; output is unchanged except timestamp and EventId metadata. --- AzureBatchQueue.Tests/BatchQueueTests.cs | 3 +- AzureBatchQueue.Tests/MessageQueueTests.cs | 112 ++++++++++++++++++- AzureBatchQueue/MessageQueue.cs | 118 ++++++++++++++++----- AzureBatchQueue/QueueLog.cs | 97 +++++++++++++++++ AzureBatchQueue/TimerBatch.cs | 20 ++-- 5 files changed, 305 insertions(+), 45 deletions(-) create mode 100644 AzureBatchQueue/QueueLog.cs diff --git a/AzureBatchQueue.Tests/BatchQueueTests.cs b/AzureBatchQueue.Tests/BatchQueueTests.cs index 489fff0..832b01b 100644 --- a/AzureBatchQueue.Tests/BatchQueueTests.cs +++ b/AzureBatchQueue.Tests/BatchQueueTests.cs @@ -295,7 +295,8 @@ public async Task Init(int maxDequeueCount) var logger = loggerFactory.CreateLogger>(); var random = new Random().NextInt64(0, int.MaxValue); - BatchQueue = new BatchQueue("UseDevelopmentStorage=true", $"batch-test-{random}", maxDequeueCount, logger: logger); + var connection = Environment.GetEnvironmentVariable("AZURE_BATCH_QUEUE_TEST_CONNECTION") ?? "UseDevelopmentStorage=true"; + BatchQueue = new BatchQueue(connection, $"batch-test-{random}", maxDequeueCount, logger: logger); await BatchQueue.Init(); } diff --git a/AzureBatchQueue.Tests/MessageQueueTests.cs b/AzureBatchQueue.Tests/MessageQueueTests.cs index e8163ab..e7d8ab4 100644 --- a/AzureBatchQueue.Tests/MessageQueueTests.cs +++ b/AzureBatchQueue.Tests/MessageQueueTests.cs @@ -6,6 +6,7 @@ using AzureBatchQueue.JsonNet; using AzureBatchQueue.Tests.Serializers; using FluentAssertions; +using Microsoft.Extensions.Logging; using NUnit.Framework; namespace AzureBatchQueue.Tests; @@ -150,6 +151,8 @@ public async Task When_quarantine_small_message() public async Task When_quarantine_large_message() { using var queueTest = await Queue(maxDequeueCount: 0); // quarantine after first read + var logger = new RecordingLogger(); + queueTest.Queue.WithLogger(logger); var largeMessage = new string('*', 65 * 1024); await queueTest.Queue.Send(largeMessage); @@ -159,6 +162,9 @@ public async Task When_quarantine_large_message() var msgFromQuarantine = (await queueTest.Queue.ReceiveFromQuarantine()).Single(); msgFromQuarantine.Item.Should().Be(largeMessage); msgFromQuarantine.MessageId.BlobName.Should().NotBeEmpty(); + var copy = logger.Entries.Single(x => x.Name == "BlobCopiedForQuarantine"); + copy.Fields["DestBlob"].Should().Be(msgFromQuarantine.MessageId.BlobName); + copy.Fields["SourceBlob"].Should().Be(logger.Entries.Single(x => x.Name == "BlobDeleted").Fields["BlobName"]); } [Test] @@ -241,6 +247,104 @@ public async Task When_receiving_a_missing_blob_from_quarantine_keeps_message() (await queueTest.MessageCount()).Should().Be(0); } + [Test] + public async Task When_processing_a_blob_logs_its_message_and_lifecycle() + { + using var queueTest = await Queue(); + var logger = new RecordingLogger(); + queueTest.Queue.WithLogger(logger); + + await queueTest.Queue.Send(new string('*', 65 * 1024)); + var message = (await queueTest.Queue.Receive()).Single(); + await queueTest.Queue.DeleteMessage(message.MessageId); + + var events = logger.Entries.Where(x => Equals(x.Fields.GetValueOrDefault("BlobName"), message.MessageId.BlobName)).ToArray(); + events.Select(x => x.Name).Should().ContainInOrder("UploadingBlob", "BlobUploaded", "SendingMessage", "MessageSent", + "MessageReceived", "DeletingMessage", "MessageDeleted", "DeletingBlob", "BlobDeleted"); + events.Single(x => x.Name == "MessageSent").Fields["MessageId"].Should().Be(message.MessageId.Id); + events.Single(x => x.Name == "BlobDeleted").Fields["Deleted"].Should().Be(true); + } + + [TestCase(1024)] + [TestCase(65 * 1024)] + public async Task When_updating_a_blob_logs_the_payload_reference(int size) + { + using var queueTest = await Queue(); + var logger = new RecordingLogger(); + queueTest.Queue.WithLogger(logger); + await queueTest.Queue.Send(new string('*', 65 * 1024)); + var original = (await queueTest.Queue.Receive()).Single(); + + await queueTest.Queue.UpdateMessage(original.MessageId, new string('+', size)); + var updated = (await queueTest.Queue.Receive()).Single(); + + var change = logger.Entries.Single(x => x.Name == "MessageUpdated"); + change.Fields["MessageId"].Should().Be(original.MessageId.Id); + change.Fields["BlobName"].Should().Be(updated.MessageId.BlobName); + logger.Entries.Count(x => x.Name == "BlobDeleted").Should().Be(size == 1024 ? 1 : 0); + } + + [Test] + public async Task When_sending_fails_logs_the_blob_and_operation() + { + using var queueTest = await Queue(); + var logger = new RecordingLogger(); + queueTest.Queue.WithLogger(logger); + await queueTest.RawQueue.DeleteAsync(); + try + { + var error = Assert.ThrowsAsync(async () => await queueTest.Queue.Send(new string('*', 65 * 1024))); + + var failed = logger.Entries.Single(x => x.Name == "MessageSendFailed"); + failed.Fields["OperationId"].Should().Be(logger.Entries.Single(x => x.Name == "SendingMessage").Fields["OperationId"]); + failed.Fields["BlobName"].Should().Be(logger.Entries.Single(x => x.Name == "BlobUploaded").Fields["BlobName"]); + failed.Exception.Should().BeSameAs(error); + } + finally + { + await queueTest.Init(); + } + } + + [Test] + public async Task When_messages_share_a_blob_logs_both_ids_and_who_deletes_it() + { + using var queueTest = await Queue(); + var logger = new RecordingLogger(); + queueTest.Queue.WithLogger(logger); + await queueTest.Queue.Send(new string('*', 65 * 1024)); + var original = (await queueTest.RawQueue.PeekMessagesAsync(1)).Value.Single(); + await queueTest.RawQueue.SendMessageAsync(original.Body); + + var messages = await queueTest.Queue.Receive(2); + await queueTest.Queue.DeleteMessage(messages[0].MessageId); + + var received = logger.Entries.Where(x => x.Name == "MessageReceived").ToArray(); + received.Select(x => x.Fields["MessageId"]).Should().BeEquivalentTo(messages.Select(x => x.MessageId.Id)); + received.Select(x => x.Fields["BlobName"]).Should().OnlyContain(x => Equals(x, messages[0].MessageId.BlobName)); + logger.Entries.Single(x => x.Name == "BlobDeleted").Fields["MessageId"].Should().Be(messages[0].MessageId.Id); + + await queueTest.Queue.UpdateMessage(messages[1].MessageId, null); + (await queueTest.Queue.Receive()).Should().BeEmpty(); + logger.Entries.Single(x => x.Name == "MissingBlobMessageDeleted").Fields["MessageId"].Should().Be(messages[1].MessageId.Id); + } + + sealed record LogEntry(string Name, Dictionary Fields, Exception? Exception); + + sealed class RecordingLogger : ILogger + { + public System.Collections.Concurrent.ConcurrentQueue Entries { get; } = new(); + public bool IsEnabled(LogLevel logLevel) => true; + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + var fields = ((IEnumerable>)state!).ToDictionary(x => x.Key, x => x.Value); + Entries.Enqueue(new LogEntry(eventId.Name ?? "", fields, exception)); + } + } + static async Task> Queue(int maxDequeueCount = 5, IMessageQueueSerializer? serializer = null) { var queue = new QueueTest(maxDequeueCount, serializer); @@ -250,7 +354,6 @@ static async Task> Queue(int maxDequeueCount = 5, IMessageQueueS class QueueTest : IDisposable { - readonly QueueClient rawQueue; readonly QueueClient quarantineQueue; public QueueTest(int maxDequeueCount = 5, IMessageQueueSerializer? serializer = null) @@ -258,18 +361,19 @@ public QueueTest(int maxDequeueCount = 5, IMessageQueueSerializer? serializer var connection = Environment.GetEnvironmentVariable("AZURE_BATCH_QUEUE_TEST_CONNECTION") ?? "UseDevelopmentStorage=true"; Queue = new MessageQueue(connection, "test", maxDequeueCount: maxDequeueCount, serializer: serializer); var options = new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 }; - rawQueue = new QueueClient(connection, "test", options); + RawQueue = new QueueClient(connection, "test", options); quarantineQueue = new QueueClient(connection, "test-quarantine", options); Blobs = new BlobContainerClient(connection, "overflow-test"); } public async Task Init() => await Queue.Init(); public MessageQueue Queue { get; } + public QueueClient RawQueue { get; } public BlobContainerClient Blobs { get; } public async Task DeleteMessageBlob(bool fromQuarantine = false) { - var queue = fromQuarantine ? quarantineQueue : rawQueue; + var queue = fromQuarantine ? quarantineQueue : RawQueue; var message = (await queue.PeekMessagesAsync(1)).Value.Single(); using var envelope = JsonDocument.Parse(message.Body.ToString()); await Blobs.DeleteBlobAsync(envelope.RootElement.GetProperty("__MSQ_QUEUE_BLOBNAME__").GetString()!); @@ -277,7 +381,7 @@ public async Task DeleteMessageBlob(bool fromQuarantine = false) public async Task MessageCount(bool fromQuarantine = false) { - var queue = fromQuarantine ? quarantineQueue : rawQueue; + var queue = fromQuarantine ? quarantineQueue : RawQueue; return (await queue.GetPropertiesAsync()).Value.ApproximateMessagesCount; } diff --git a/AzureBatchQueue/MessageQueue.cs b/AzureBatchQueue/MessageQueue.cs index 82dd5be..17f96e4 100644 --- a/AzureBatchQueue/MessageQueue.cs +++ b/AzureBatchQueue/MessageQueue.cs @@ -38,29 +38,36 @@ public MessageQueue(string connectionString, string queueName, public async Task Send(T item, TimeSpan? visibilityTimeout = null, CancellationToken ct = default) { - var (data, _) = await SerializeAndOffloadIfBig(item, ct: ct); - await queue.SendMessageAsync(data, visibilityTimeout ?? TimeSpan.Zero, null, ct); + var payload = await SerializeAndOffloadIfBig(item, ct: ct); + await SendPayload(queue, payload.Data, payload.BlobName, ct, visibilityTimeout ?? TimeSpan.Zero); } public async Task DeleteMessage(MessageId id, CancellationToken ct = default) { - await queue.DeleteMessageAsync(id.Id, id.PopReceipt, ct); + await DeleteQueueMessage(queue, id, ct); if (id.BlobName is not null) - await container.DeleteBlobIfExistsAsync(id.BlobName, cancellationToken: ct); + await DeleteBlob(id, ct); } public async Task UpdateMessage(MessageId id, T? item, TimeSpan visibilityTimeout = default, CancellationToken ct = default) { + logger.UpdatingMessage(id.Id, queue.Name, id.BlobName); if (item == null) { await queue.UpdateMessageAsync(id.Id, id.PopReceipt, (BinaryData?)null, visibilityTimeout, ct); + logger.MessageUpdated(id.Id, queue.Name, id.BlobName, id.BlobName); return; } - var (data, offloaded) = await SerializeAndOffloadIfBig(item, id.BlobName, ct); - await queue.UpdateMessageAsync(id.Id, id.PopReceipt, data, visibilityTimeout, ct); - if (!offloaded && id.BlobName != null) - await container.DeleteBlobAsync(id.BlobName, cancellationToken: ct); + var payload = await SerializeAndOffloadIfBig(item, id.BlobName, ct); + await queue.UpdateMessageAsync(id.Id, id.PopReceipt, payload.Data, visibilityTimeout, ct); + logger.MessageUpdated(id.Id, queue.Name, payload.BlobName, id.BlobName); + if (!payload.Offloaded && id.BlobName != null) + { + logger.DeletingBlob(id.BlobName, id.Id, queue.Name); + var deleted = await container.DeleteBlobAsync(id.BlobName, cancellationToken: ct); + logger.BlobDeleted(id.BlobName, id.Id, queue.Name, true, deleted.Headers.RequestId); + } } async Task SerializeAndOffloadIfBig(T item, string? blobName = null, CancellationToken ct = default) @@ -74,7 +81,9 @@ async Task SerializeAndOffloadIfBig(T item, string? blobName = null, Ca return Payload.FitsInMessage(stream); var blobRef = blobName != null ? BlobRef.Get(blobName) : BlobRef.Create(); - await container.GetBlobClient(blobRef.BlobName).UploadAsync(stream, overwrite: true, ct); + logger.UploadingBlob(blobRef.BlobName, queue.Name, stream.Length); + var uploaded = await container.GetBlobClient(blobRef.BlobName).UploadAsync(stream, overwrite: true, ct); + logger.BlobUploaded(blobRef.BlobName, queue.Name, uploaded.Value.ETag.ToString(), uploaded.GetRawResponse().Headers.RequestId); return Payload.OffloadedToBlob(blobRef); } @@ -86,6 +95,7 @@ public async Task[]> Receive(int maxMessages = MaxMessagesReceiv if (!r.HasValue) return Array.Empty>(); + LogReceivedMessages(queue, r.Value); var quarantineMessages = r.Value.Where(x => x.DequeueCount > maxDequeueCount).ToList(); var validMessages = r.Value.Except(quarantineMessages); @@ -104,6 +114,7 @@ public async Task[]> ReceiveFromQuarantine(int maxMessages = Max if (!r.HasValue) return Array.Empty>(); + LogReceivedMessages(quarantineQueue, r.Value); var queueMessages = await Task.WhenAll(r.Value.Select(x => ToQueueMessage(x, ct, fromQuarantine: true))); return queueMessages.Where(x => x != null).ToArray()!; @@ -115,6 +126,7 @@ async Task QuarantineMessage(QueueMessage queueMessage, CancellationToken ct = d try { BinaryData messageBody; + string? quarantineBlobName = null; if (IsBlobRef(queueMessage.Body, out blobRef) && blobRef != null) { @@ -123,24 +135,23 @@ async Task QuarantineMessage(QueueMessage queueMessage, CancellationToken ct = d return; messageBody = Payload.OffloadedToBlob(newBlobRef).Data; + quarantineBlobName = newBlobRef.BlobName; } else { messageBody = queueMessage.Body; } - await quarantineQueue.SendMessageAsync(messageBody, cancellationToken: ct); + await SendPayload(quarantineQueue, messageBody, quarantineBlobName, ct); var messageId = new MessageId(queueMessage.MessageId, queueMessage.PopReceipt, blobRef?.BlobName); await DeleteMessage(messageId, ct); - logger.LogInformation("QueueMessage {msgId} with {popReceipt} was quarantined after {dequeueCount} unsuccessful attempts.", - queueMessage.MessageId, queueMessage.PopReceipt, queueMessage.DequeueCount); + logger.MessageQuarantined(queueMessage.MessageId, queueMessage.PopReceipt, queueMessage.DequeueCount); } catch (Exception ex) { - logger.LogError(ex, "Failed to quarantine queue message with {messageId}, {popReceipt}, and {blobName}.", - queueMessage.MessageId, queueMessage.PopReceipt, blobRef?.BlobName); + logger.QuarantineFailed(queueMessage.MessageId, queueMessage.PopReceipt, blobRef?.BlobName, ex); } } @@ -158,17 +169,21 @@ async Task QuarantineMessage(QueueMessage queueMessage, CancellationToken ct = d BinaryData content; try { - content = (await sourceBlobClient.DownloadContentAsync(ct)).Value.Content; + var downloaded = await sourceBlobClient.DownloadContentAsync(ct); + content = downloaded.Value.Content; + logger.BlobDownloaded(sourceBlobRef.BlobName, message.MessageId, queue.Name, + downloaded.Value.Details.ETag.ToString(), downloaded.GetRawResponse().Headers.RequestId); } catch (RequestFailedException ex) when (ex is { Status: 404, ErrorCode: "BlobNotFound" }) { await DeleteMissingBlobMessage(message, sourceBlobRef, ex, ct); return null; } - await destinationBlobClient.UploadAsync(content, ct); + logger.UploadingBlob(newBlobRef.BlobName, queue.Name, content.ToMemory().Length); + var uploaded = await destinationBlobClient.UploadAsync(content, ct); + logger.BlobUploaded(newBlobRef.BlobName, queue.Name, uploaded.Value.ETag.ToString(), uploaded.GetRawResponse().Headers.RequestId); - logger.LogInformation("Copied blob {SourceBlob} to {DestBlob} for quarantine message {MsgId}", - sourceBlobRef.BlobName, newBlobRef.BlobName, message.MessageId); + logger.BlobCopiedForQuarantine(sourceBlobRef.BlobName, newBlobRef.BlobName, message.MessageId); return newBlobRef; } @@ -182,10 +197,12 @@ public async Task Dequarantine(CancellationToken ct = default) if (!response.HasValue || response.Value.Length == 0) return; + LogReceivedMessages(quarantineQueue, response.Value); foreach (var msg in response.Value) { - await queue.SendMessageAsync(msg.Body, TimeSpan.Zero, cancellationToken: ct); - await quarantineQueue.DeleteMessageAsync(msg.MessageId, msg.PopReceipt, ct); + IsBlobRef(msg.Body, out var blobRef); + await SendPayload(queue, msg.Body, blobRef?.BlobName, ct, TimeSpan.Zero); + await DeleteQueueMessage(quarantineQueue, new MessageId(msg.MessageId, msg.PopReceipt, blobRef?.BlobName), ct); } } while (!ct.IsCancellationRequested); } @@ -193,7 +210,7 @@ public async Task Dequarantine(CancellationToken ct = default) public async Task QuarantineData(T item, CancellationToken ct) { var payload = await SerializeAndOffloadIfBig(item, ct: ct); - await quarantineQueue.SendMessageAsync(payload.Data, cancellationToken: ct); + await SendPayload(quarantineQueue, payload.Data, payload.BlobName, ct); } async Task?> ToQueueMessage(QueueMessage m, CancellationToken ct, bool fromQuarantine = false) @@ -205,6 +222,8 @@ public async Task QuarantineData(T item, CancellationToken ct) { var blobData = await container.GetBlobClient(blobRef!.BlobName).DownloadContentAsync(ct); payload = blobData.Value.Content; + logger.BlobDownloaded(blobRef.BlobName, m.MessageId, fromQuarantine ? quarantineQueue.Name : queue.Name, + blobData.Value.Details.ETag.ToString(), blobData.GetRawResponse().Headers.RequestId); } catch (RequestFailedException ex) when (!fromQuarantine && ex is { Status: 404, ErrorCode: "BlobNotFound" }) { @@ -213,7 +232,7 @@ public async Task QuarantineData(T item, CancellationToken ct) } catch (Exception ex) { - logger.LogError(ex, "Exception when loading blob {BlobName} for {MessageId}", blobRef!.BlobName, m.MessageId); + logger.BlobDownloadFailed(blobRef!.BlobName, m.MessageId, ex); throw; } } @@ -225,7 +244,7 @@ public async Task QuarantineData(T item, CancellationToken ct) } catch (Exception ex) { - logger.LogError(ex, "Exception when deserializing a message {MessageId}", m.MessageId); + logger.MessageDeserializationFailed(m.MessageId, ex); if (!fromQuarantine) await QuarantineMessage(m, ct); @@ -241,10 +260,51 @@ public async Task QuarantineData(T item, CancellationToken ct) async Task DeleteMissingBlobMessage(QueueMessage message, BlobRef blobRef, RequestFailedException error, CancellationToken ct) { - await queue.DeleteMessageAsync(message.MessageId, message.PopReceipt, ct); + await DeleteQueueMessage(queue, new MessageId(message.MessageId, message.PopReceipt, blobRef.BlobName), ct); + + logger.MissingBlobMessageDeleted(message.MessageId, queue.Name, blobRef.BlobName, error); + } + + async Task SendPayload(QueueClient target, BinaryData data, string? blobName, CancellationToken ct, TimeSpan? visibilityTimeout = null) + { + var operationId = Guid.NewGuid(); + logger.SendingMessage(target.Name, blobName, operationId); + try + { + var sent = await target.SendMessageAsync(data, visibilityTimeout, cancellationToken: ct); + logger.MessageSent(sent.Value.MessageId, target.Name, blobName, operationId, sent.GetRawResponse().Headers.RequestId); + } + catch (RequestFailedException ex) + { + logger.MessageSendFailed(ex, target.Name, blobName, operationId, ex.Status, ex.ErrorCode, ex.GetRawResponse()?.Headers.RequestId); + throw; + } + } + + async Task DeleteQueueMessage(QueueClient source, MessageId id, CancellationToken ct) + { + logger.DeletingMessage(id.Id, source.Name, id.BlobName); + await source.DeleteMessageAsync(id.Id, id.PopReceipt, ct); + logger.MessageDeleted(id.Id, source.Name, id.BlobName); + } + + async Task DeleteBlob(MessageId id, CancellationToken ct) + { + logger.DeletingBlob(id.BlobName!, id.Id, queue.Name); + var deleted = await container.DeleteBlobIfExistsAsync(id.BlobName!, cancellationToken: ct); + logger.BlobDeleted(id.BlobName!, id.Id, queue.Name, deleted.Value, deleted.GetRawResponse().Headers.RequestId); + } + + void LogReceivedMessages(QueueClient source, QueueMessage[] messages) + { + if (!logger.IsEnabled(LogLevel.Information)) + return; - logger.LogWarning(error, "Deleted message {MessageId} from queue {QueueName} because blob {BlobName} was not found.", - message.MessageId, queue.Name, blobRef.BlobName); + foreach (var message in messages) + { + IsBlobRef(message.Body, out var blobRef); + logger.MessageReceived(message.MessageId, source.Name, blobRef?.BlobName, message.DequeueCount); + } } static bool IsBlobRef(BinaryData data, out BlobRef? o) @@ -265,12 +325,14 @@ static bool IsBlobRef(BinaryData data, out BlobRef? o) public async Task Delete(bool deleteBlobs = false) { + logger.DeletingQueueStorage(queue.Name, deleteBlobs); var tasks = new List { queue.DeleteAsync(), quarantineQueue.DeleteAsync() }; if (deleteBlobs) tasks.Add(container.DeleteAsync()); await Task.WhenAll(tasks); + logger.QueueStorageDeleted(queue.Name, deleteBlobs); } public async Task ClearMessages() => await queue.ClearMessagesAsync(); @@ -288,13 +350,13 @@ record BlobRef([property: JsonPropertyName("__MSQ_QUEUE_BLOBNAME__")] string Blo static string FileName() => $"{DateTime.UtcNow:yyyy-MM-dd}/{DateTime.UtcNow:s}{Guid.NewGuid():N}.json.gzip"; } - record Payload(BinaryData Data, bool Offloaded) + record Payload(BinaryData Data, bool Offloaded, string? BlobName = null) { public static Payload FitsInMessage(MemoryStream stream) => new(BinaryData.FromStream(stream), false); public static Payload OffloadedToBlob(BlobRef blobRef) => - new(BinaryData.FromObjectAsJson(blobRef), true); + new(BinaryData.FromObjectAsJson(blobRef), true, blobRef.BlobName); } } diff --git a/AzureBatchQueue/QueueLog.cs b/AzureBatchQueue/QueueLog.cs new file mode 100644 index 0000000..9a3b836 --- /dev/null +++ b/AzureBatchQueue/QueueLog.cs @@ -0,0 +1,97 @@ +using Microsoft.Extensions.Logging; + +namespace AzureBatchQueue; + +internal static partial class QueueLog +{ + [LoggerMessage(1001, LogLevel.Warning, "Deleted message {MessageId} from queue {QueueName} because blob {BlobName} was not found.")] + public static partial void MissingBlobMessageDeleted(this ILogger logger, string messageId, string queueName, string blobName, Exception exception); + + [LoggerMessage(1002, LogLevel.Information, "QueueMessage {msgId} with {popReceipt} was quarantined after {dequeueCount} unsuccessful attempts.")] + public static partial void MessageQuarantined(this ILogger logger, string msgId, string popReceipt, long dequeueCount); + + [LoggerMessage(1003, LogLevel.Error, "Failed to quarantine queue message with {messageId}, {popReceipt}, and {blobName}.")] + public static partial void QuarantineFailed(this ILogger logger, string messageId, string popReceipt, string? blobName, Exception exception); + + [LoggerMessage(1004, LogLevel.Information, "Copied blob {SourceBlob} to {DestBlob} for quarantine message {MsgId}")] + public static partial void BlobCopiedForQuarantine(this ILogger logger, string sourceBlob, string destBlob, string msgId); + + [LoggerMessage(1005, LogLevel.Error, "Exception when loading blob {BlobName} for {MessageId}")] + public static partial void BlobDownloadFailed(this ILogger logger, string blobName, string messageId, Exception exception); + + [LoggerMessage(1006, LogLevel.Error, "Exception when deserializing a message {MessageId}")] + public static partial void MessageDeserializationFailed(this ILogger logger, string messageId, Exception exception); + + [LoggerMessage(1007, LogLevel.Warning, "Queue {queueName} was not found when flushing {messageId} with {itemsCount} items left.")] + public static partial void QueueNotFoundOnFlush(this ILogger logger, Exception exception, string queueName, MessageId messageId, int itemsCount); + + [LoggerMessage(1008, LogLevel.Error, "Accessing already flushed message with {messageId}.")] + public static partial void FlushedMessageAccessed(this ILogger logger, Exception exception, MessageId messageId); + + [LoggerMessage(1009, LogLevel.Error, "Unhandled exception when flushing message {messageId}.")] + public static partial void FlushFailed(this ILogger logger, Exception exception, MessageId messageId); + + [LoggerMessage(1010, LogLevel.Warning, "Message {MsgId} was not fully processed within a timeout ({FlushPeriod}) sec in queue {QueueName}." + + " {RemainingCount} items were not completed ({NotProcessed} not processed on time and {FailedCount} failed) from {TotalCount} total")] + public static partial void BatchIncomplete(this ILogger logger, MessageId msgId, double flushPeriod, string queueName, + int remainingCount, int notProcessed, int failedCount, int totalCount); + + [LoggerMessage(1011, LogLevel.Debug, "Message {msgId} was fully processed with {totalCount} items")] + public static partial void BatchProcessed(this ILogger logger, MessageId msgId, int totalCount); + + [LoggerMessage(1012, LogLevel.Information, "Message {MsgId} was quarantined after {DequeueCount} unsuccessful attempts in queue {QueueName}." + + " {RemainingCount} items were not completed ({NotProcessed} not processed on time and {FailedCount} failed) from {TotalCount} total")] + public static partial void BatchQuarantined(this ILogger logger, MessageId msgId, long dequeueCount, string queueName, + int remainingCount, int notProcessed, int failedCount, int totalCount); + + [LoggerMessage(1013, LogLevel.Error, "VisibilityTimeout {VisibilityTimeout} cannot be less that 1 sec." + + " UtcNow: {UtcNow}, MessageVisibleTime: {MessageVisibleTime}, InsertedOn: {InsertedOn}")] + public static partial void InvalidVisibilityTimeout(this ILogger logger, TimeSpan visibilityTimeout, string utcNow, + string messageVisibleTime, string insertedOn); + + [LoggerMessage(1101, LogLevel.Information, "Uploading blob {BlobName} for queue {QueueName} with {ContentLength} bytes")] + public static partial void UploadingBlob(this ILogger logger, string blobName, string queueName, long contentLength); + + [LoggerMessage(1102, LogLevel.Information, "Uploaded blob {BlobName} for queue {QueueName}; ETag {ETag}, Azure request {RequestId}")] + public static partial void BlobUploaded(this ILogger logger, string blobName, string queueName, string eTag, string? requestId); + + [LoggerMessage(1103, LogLevel.Information, "Sending message to queue {QueueName} referencing blob {BlobName}; operation {OperationId}")] + public static partial void SendingMessage(this ILogger logger, string queueName, string? blobName, Guid operationId); + + [LoggerMessage(1104, LogLevel.Information, "Sent message {MessageId} to queue {QueueName} referencing blob {BlobName}; operation {OperationId}, Azure request {RequestId}")] + public static partial void MessageSent(this ILogger logger, string messageId, string queueName, string? blobName, Guid operationId, string? requestId); + + [LoggerMessage(1105, LogLevel.Information, "Received message {MessageId} from queue {QueueName} referencing blob {BlobName}; dequeue count {DequeueCount}")] + public static partial void MessageReceived(this ILogger logger, string messageId, string queueName, string? blobName, long dequeueCount); + + [LoggerMessage(1106, LogLevel.Information, "Deleting message {MessageId} from queue {QueueName} referencing blob {BlobName}")] + public static partial void DeletingMessage(this ILogger logger, string messageId, string queueName, string? blobName); + + [LoggerMessage(1107, LogLevel.Information, "Deleted message {MessageId} from queue {QueueName} referencing blob {BlobName}")] + public static partial void MessageDeleted(this ILogger logger, string messageId, string queueName, string? blobName); + + [LoggerMessage(1108, LogLevel.Information, "Deleting blob {BlobName} for message {MessageId} in queue {QueueName}")] + public static partial void DeletingBlob(this ILogger logger, string blobName, string messageId, string queueName); + + [LoggerMessage(1109, LogLevel.Information, "Deleted blob {BlobName} for message {MessageId} in queue {QueueName}; deleted {Deleted}, Azure request {RequestId}")] + public static partial void BlobDeleted(this ILogger logger, string blobName, string messageId, string queueName, bool deleted, string? requestId); + + [LoggerMessage(1110, LogLevel.Information, "Updating message {MessageId} in queue {QueueName} referencing blob {BlobName}")] + public static partial void UpdatingMessage(this ILogger logger, string messageId, string queueName, string? blobName); + + [LoggerMessage(1111, LogLevel.Information, "Updated message {MessageId} in queue {QueueName} referencing blob {BlobName}; previous blob {PreviousBlobName}")] + public static partial void MessageUpdated(this ILogger logger, string messageId, string queueName, string? blobName, string? previousBlobName); + + [LoggerMessage(1112, LogLevel.Error, "Failed sending message to queue {QueueName} referencing blob {BlobName}; operation {OperationId}, status {Status}, code {ErrorCode}, Azure request {RequestId}")] + public static partial void MessageSendFailed(this ILogger logger, Exception exception, string queueName, string? blobName, + Guid operationId, int status, string? errorCode, string? requestId); + + [LoggerMessage(1113, LogLevel.Information, "Deleting storage for queue {QueueName}; delete blob container {DeleteBlobs}")] + public static partial void DeletingQueueStorage(this ILogger logger, string queueName, bool deleteBlobs); + + [LoggerMessage(1114, LogLevel.Information, "Deleted storage for queue {QueueName}; delete blob container {DeleteBlobs}")] + public static partial void QueueStorageDeleted(this ILogger logger, string queueName, bool deleteBlobs); + + [LoggerMessage(1115, LogLevel.Information, "Downloaded blob {BlobName} for message {MessageId} from queue {QueueName}; ETag {ETag}, Azure request {RequestId}")] + public static partial void BlobDownloaded(this ILogger logger, string blobName, string messageId, string queueName, string eTag, string? requestId); +} diff --git a/AzureBatchQueue/TimerBatch.cs b/AzureBatchQueue/TimerBatch.cs index a1edddb..36710a0 100644 --- a/AzureBatchQueue/TimerBatch.cs +++ b/AzureBatchQueue/TimerBatch.cs @@ -58,16 +58,15 @@ async Task Flush() } catch (Azure.RequestFailedException ex) when (ex.ErrorCode == "QueueNotFound") { - logger.LogWarning(ex, "Queue {queueName} was not found when flushing {messageId} with {itemsCount} items left.", - batchQueue.Name, msg.MessageId, items.NotProcessedCount()); + logger.QueueNotFoundOnFlush(ex, batchQueue.Name, msg.MessageId, items.NotProcessedCount()); } catch (Azure.RequestFailedException ex) when (ex.ErrorCode == "MessageNotFound") { - logger.LogError(ex, "Accessing already flushed message with {messageId}.", msg.MessageId); + logger.FlushedMessageAccessed(ex, msg.MessageId); } catch (Exception ex) { - logger.LogError(ex, "Unhandled exception when flushing message {messageId}.", msg.MessageId); + logger.FlushFailed(ex, msg.MessageId); throw; } @@ -95,8 +94,7 @@ async Task Update() var remaining = Remaining(); await batchQueue.UpdateMessage(msg.MessageId, remaining); - logger.LogWarning("Message {MsgId} was not fully processed within a timeout ({FlushPeriod}) sec in queue {QueueName}." + - " {RemainingCount} items were not completed ({NotProcessed} not processed on time and {FailedCount} failed) from {TotalCount} total", + logger.BatchIncomplete( msg.MessageId, FlushPeriod.TotalSeconds, batchQueue.Name, @@ -110,7 +108,7 @@ async Task Delete() { await batchQueue.DeleteMessage(msg.MessageId); - logger.LogDebug("Message {msgId} was fully processed with {totalCount} items", msg.MessageId, items.Items().Length); + logger.BatchProcessed(msg.MessageId, items.Items().Length); } async Task Quarantine() @@ -118,8 +116,7 @@ async Task Quarantine() var remaining = Remaining(); await batchQueue.QuarantineData(msg.MessageId, remaining); - logger.LogInformation("Message {MsgId} was quarantined after {DequeueCount} unsuccessful attempts in queue {QueueName}." + - " {RemainingCount} items were not completed ({NotProcessed} not processed on time and {FailedCount} failed) from {TotalCount} total", + logger.BatchQuarantined( msg.MessageId, msg.Metadata.DequeueCount, batchQueue.Name, @@ -197,9 +194,8 @@ TimeSpan CalculateFlushPeriod(TimeSpan visibilityTimeout) if (visibilityTimeout >= TimeSpan.FromSeconds(1)) return visibilityTimeout.Subtract(TimeSpan.FromMilliseconds(100)); - logger.LogError("VisibilityTimeout {VisibilityTimeout} cannot be less that 1 sec." + - " UtcNow: {UtcNow}, MessageVisibleTime: {MessageVisibleTime}, InsertedOn: {InsertedOn}", - visibilityTimeout, DateTimeOffset.UtcNow.ToString("O"), Metadata.VisibilityTime.ToString("O"), Metadata.InsertedOn.ToString("O")); + logger.InvalidVisibilityTimeout(visibilityTimeout, DateTimeOffset.UtcNow.ToString("O"), + Metadata.VisibilityTime.ToString("O"), Metadata.InsertedOn.ToString("O")); throw new ArgumentOutOfRangeException(nameof(visibilityTimeout)); } From 044ae460fdcc95dd86d4048d9b0d18092e7c9663 Mon Sep 17 00:00:00 2001 From: Yan Mahdych Date: Fri, 30 Oct 2026 21:04:04 +0200 Subject: [PATCH 3/3] Restore standard development storage in queue tests Remove the connection override introduced only for local investigation. Keep both fixtures on the existing Azurite development-storage configuration. --- AzureBatchQueue.Tests/BatchQueueTests.cs | 3 +-- AzureBatchQueue.Tests/MessageQueueTests.cs | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/AzureBatchQueue.Tests/BatchQueueTests.cs b/AzureBatchQueue.Tests/BatchQueueTests.cs index 832b01b..489fff0 100644 --- a/AzureBatchQueue.Tests/BatchQueueTests.cs +++ b/AzureBatchQueue.Tests/BatchQueueTests.cs @@ -295,8 +295,7 @@ public async Task Init(int maxDequeueCount) var logger = loggerFactory.CreateLogger>(); var random = new Random().NextInt64(0, int.MaxValue); - var connection = Environment.GetEnvironmentVariable("AZURE_BATCH_QUEUE_TEST_CONNECTION") ?? "UseDevelopmentStorage=true"; - BatchQueue = new BatchQueue(connection, $"batch-test-{random}", maxDequeueCount, logger: logger); + BatchQueue = new BatchQueue("UseDevelopmentStorage=true", $"batch-test-{random}", maxDequeueCount, logger: logger); await BatchQueue.Init(); } diff --git a/AzureBatchQueue.Tests/MessageQueueTests.cs b/AzureBatchQueue.Tests/MessageQueueTests.cs index e7d8ab4..785fba8 100644 --- a/AzureBatchQueue.Tests/MessageQueueTests.cs +++ b/AzureBatchQueue.Tests/MessageQueueTests.cs @@ -358,7 +358,7 @@ class QueueTest : IDisposable public QueueTest(int maxDequeueCount = 5, IMessageQueueSerializer? serializer = null) { - var connection = Environment.GetEnvironmentVariable("AZURE_BATCH_QUEUE_TEST_CONNECTION") ?? "UseDevelopmentStorage=true"; + const string connection = "UseDevelopmentStorage=true"; Queue = new MessageQueue(connection, "test", maxDequeueCount: maxDequeueCount, serializer: serializer); var options = new QueueClientOptions { MessageEncoding = QueueMessageEncoding.Base64 }; RawQueue = new QueueClient(connection, "test", options);