Skip to content

Commit 7365e67

Browse files
Copy content to new blob on quarantine (#10)
We had a race condition during quarantine, that could create duplicate messages in quarantine with a reference to one blob. Problem Analysis Failure Scenario: 1. Message with BlobRef quarantined → quarantine queue has copy 2. DeleteMessage fails (service crash/network issue) → message stays in main queue 3. Message becomes visible again → dequeued with DequeueCount++ 4. Same message quarantined again → 2 identical messages in quarantine with same blob ref 5. First quarantine message processed → blob deleted 6. Second quarantine message processed → blob not found → poison message Solution: create new blob on quarantine and copy its content there
2 parents 5a31449 + 09e352e commit 7365e67

1 file changed

Lines changed: 36 additions & 2 deletions

File tree

‎AzureBatchQueue/MessageQueue.cs‎

Lines changed: 36 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -112,8 +112,22 @@ async Task QuarantineMessage(QueueMessage queueMessage, CancellationToken ct = d
112112
{
113113
try
114114
{
115-
await quarantineQueue.SendMessageAsync(queueMessage.Body, cancellationToken: ct);
116-
await queue.DeleteMessageAsync(queueMessage.MessageId, queueMessage.PopReceipt, ct);
115+
BinaryData messageBody;
116+
117+
if (IsBlobRef(queueMessage.Body, out var blobRef) && blobRef != null)
118+
{
119+
var newBlobRef = await CopyBlobForQuarantine(blobRef, queueMessage.MessageId, ct);
120+
messageBody = Payload.OffloadedToBlob(newBlobRef).Data;
121+
}
122+
else
123+
{
124+
messageBody = queueMessage.Body;
125+
}
126+
127+
await quarantineQueue.SendMessageAsync(messageBody, cancellationToken: ct);
128+
129+
var messageId = new MessageId(queueMessage.MessageId, queueMessage.PopReceipt, blobRef?.BlobName);
130+
await DeleteMessage(messageId, ct);
117131

118132
logger.LogInformation("QueueMessage {msgId} with {popReceipt} was quarantined after {dequeueCount} unsuccessful attempts.",
119133
queueMessage.MessageId, queueMessage.PopReceipt, queueMessage.DequeueCount);
@@ -126,6 +140,26 @@ async Task QuarantineMessage(QueueMessage queueMessage, CancellationToken ct = d
126140
}
127141
}
128142

143+
// Why not just move the message from main queue to quarantine queue?
144+
// Because on message quarantine, we can put the message into quarantine queue, and then the service stops and we don't delete it from the main queue.
145+
// In this case, later we will insert a duplicate message to quarantine, with the same ref to blob.
146+
// When one message is processed, we remove the blob. The other message will have a BlobNotFound error.
147+
// So it's better to do some more heavy-lifting and copy blob content on quarantine.
148+
async Task<BlobRef> CopyBlobForQuarantine(BlobRef sourceBlobRef, string messageId, CancellationToken ct)
149+
{
150+
var sourceBlobClient = container.GetBlobClient(sourceBlobRef.BlobName);
151+
var newBlobRef = BlobRef.Create();
152+
var destinationBlobClient = container.GetBlobClient(newBlobRef.BlobName);
153+
154+
var content = await sourceBlobClient.DownloadContentAsync(ct);
155+
await destinationBlobClient.UploadAsync(content.Value.Content, ct);
156+
157+
logger.LogInformation("Copied blob {SourceBlob} to {DestBlob} for quarantine message {MsgId}",
158+
sourceBlobRef.BlobName, newBlobRef.BlobName, messageId);
159+
160+
return newBlobRef;
161+
}
162+
129163
public async Task Dequarantine(CancellationToken ct = default)
130164
{
131165
do

0 commit comments

Comments
 (0)