Skip to content

Optionally keep message payload when moving message to poison queue #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
booooza merged 1 commit into from
Feb 23, 2026
Merged

Optionally keep message payload when moving message to poison queue #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 76 additions & 20 deletions Classes/Queue/AzureQueueStorage.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,11 @@ class AzureQueueStorage implements QueueInterface
*/
protected string $poisonSuffix = '-poison';

/**
* Whether to preserve original payload in poison queue (if false, only metadata will be stored)
*/
protected bool $preservePoisonPayload = false;

/**
* @Flow\Inject
* @var AzureStorageClientFactory
Expand Down Expand Up @@ -150,6 +155,9 @@ public function __construct(string $name, array $options = [])
$this->validateQueueSuffix($options['poisonSuffix']);
$this->poisonSuffix = $options['poisonSuffix'];
}
if (isset($options['preservePoisonPayload'])) {
$this->preservePoisonPayload = (bool)$options['preservePoisonPayload'];
}

$this->normalPriorityQueueName = $name;
$this->priorityQueueName = $name . $this->prioritySuffix;
Expand Down Expand Up @@ -327,6 +335,7 @@ public function waitAndReserve(?int $timeout = null): ?Message
'popReceipt' => $message->getPopReceipt(),
'blobName' => $message->getBlobName(),
'queueName' => $message->getQueueName(),
'payload' => $message->getPayload(),
];

return $message;
Expand Down Expand Up @@ -356,6 +365,13 @@ public function release(string $messageId, array $options = []): void

unset($this->reservedMessages[$messageId]);
} catch (Exception $e) {
if ($e->getCode() === 404) {
$this->systemLogger->warning('Message no longer available during release(), likely taken by another worker', [
'messageId' => $messageId,
]);
unset($this->reservedMessages[$messageId]);
return;
}
throw new JobQueueException('Failed to release message: ' . $e->getMessage(), 1234567897);
}
}
Expand All @@ -373,35 +389,63 @@ public function abort(string $messageId): void

try {
// Push a failed record to the poison queue
if ($this->usePoisonQueue) {
$failedPayload = json_encode([
$isProcessingPoisonQueue = $messageInfo['queueName'] === $this->poisonQueueName;
if ($this->usePoisonQueue && !$isProcessingPoisonQueue) {
$poisonPayload = [
'messageId' => $messageId,
'queueMessageId' => $messageInfo['queueMessageId'],
'originalQueue' => $messageInfo['queueName'],
'blobName' => $messageInfo['blobName'] ?? null,
'timestamp' => time(),
]);
$this->getQueueService()->createMessage($this->poisonQueueName, $failedPayload);
'timestamp' => time(),
];
if ($this->preservePoisonPayload) {
$poisonPayload['payload'] = $messageInfo['payload'];
$serialized = json_encode($poisonPayload);

if (strlen($serialized) > $this->claimCheckThreshold) {
unset($poisonPayload['payload']);
$poisonBlobName = $this->generateBlobName('poison-' . $messageId);
$this->getBlobService()->createBlockBlob(
$this->containerName,
$poisonBlobName,
json_encode($messageInfo['payload'])
);
$poisonPayload['blobName'] = $poisonBlobName;
$poisonPayload['isClaimCheck'] = true;
}
}
$this->getQueueService()->createMessage($this->poisonQueueName, json_encode($poisonPayload));
}

// Delete the message from original queue
$this->getQueueService()->deleteMessage(
$messageInfo['queueName'],
$messageInfo['queueMessageId'],
$messageInfo['popReceipt']
);
if (!$isProcessingPoisonQueue) {
// Delete the message from original queue
$this->getQueueService()->deleteMessage(
$messageInfo['queueName'],
$messageInfo['queueMessageId'],
$messageInfo['popReceipt']
);

// Clean up blob if it exists
if (!empty($messageInfo['blobName'])) {
try {
$this->getBlobService()->deleteBlob($this->containerName, $messageInfo['blobName']);
} catch (Exception $e) {
// Log but don't throw - message is already aborted
// Clean up blob if it exists
if (!empty($messageInfo['blobName'])) {
try {
$this->getBlobService()->deleteBlob($this->containerName, $messageInfo['blobName']);
} catch (Exception $e) {
$this->systemLogger->warning('Failed to delete blob after message abort', [
'blobName' => $messageInfo['blobName'],
'error' => $e->getMessage(),
]);
}
}
}

unset($this->reservedMessages[$messageId]);
} catch (Exception $e) {
if ($e->getCode() === 404) {
$this->systemLogger->warning('Message no longer available during abort(), likely taken by another worker', [
'messageId' => $messageId,
]);
unset($this->reservedMessages[$messageId]);
return;
}
throw new JobQueueException('Failed to abort message: ' . $e->getMessage(), 1234567899);
}
}
Expand Down Expand Up @@ -431,13 +475,23 @@ public function finish(string $messageId): bool
try {
$this->getBlobService()->deleteBlob($this->containerName, $messageInfo['blobName']);
} catch (Exception $e) {
// Log but don't throw - message is already finished
$this->systemLogger->warning('Failed to delete blob after message finish', [
'blobName' => $messageInfo['blobName'],
'error' => $e->getMessage(),
]);
}
}

unset($this->reservedMessages[$messageId]);
return true;
} catch (Exception $e) {
if ($e->getCode() === 404) {
$this->systemLogger->info('Message already deleted by another worker during finish', [
'messageId' => $messageId,
]);
unset($this->reservedMessages[$messageId]);
return true;
}
throw new JobQueueException('Failed to finish message: ' . $e->getMessage(), 1234567900);
}
}
Expand Down Expand Up @@ -569,7 +623,7 @@ public function flush(): void
$this->getQueueService()->clearMessages($this->priorityQueueName);
}

// Clear poison queue only if priority queue feature is enabled
// Clear poison queue only if poison queue feature is enabled
if ($this->usePoisonQueue) {
$this->getQueueService()->clearMessages($this->poisonQueueName);
}
Expand Down Expand Up @@ -781,6 +835,8 @@ protected function extractPayload(string $messageText)
throw new JobQueueException('Invalid JSON in blob: ' . json_last_error_msg(), 1234567912);
}
return $payload;
} catch (JobQueueException $e) {
throw $e;
} catch (Exception $e) {
throw new JobQueueException('Failed to retrieve claim check blob: ' . $e->getMessage(), 1234567904);
}
Expand Down
20 changes: 19 additions & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ Flowpack:
JobQueue:
Common:
queues:
'my-azure-storage-queuee':
'my-azure-storage-queue':
className: 'Oniva\JobQueue\AzureQueueStorage\Queue\AzureQueueStorage'
options:
connectionString: DefaultEndpointsProtocol=https;AccountName=myaccountname;AccountKey=myaccountkey;EndpointSuffix=core.windows.net
Expand All @@ -50,6 +50,7 @@ Flowpack:
blobContainer: jobqueue-blobs # Blob container name for claim check messages
usePriorityQueue: true # Enable priority queueing
usePoisonQueue: true # Enable poison queue for failed jobs
preservePoisonPayload: true # Preserve the original payload for failed jobs in the poison queue
prioritySuffix: '-priority' # Suffix for priority queue
poisonSuffix: '-poison' # Suffix for poison queue
```
Expand All @@ -62,9 +63,26 @@ This allows you to submit high-priority jobs that will be processed before regul
$queue->submit($payload, ['priority' => true]);
```

## Poison Queue

To enable the poison queue, set `usePoisonQueue` to `true`. Failed jobs (those that exceed
`maximumNumberOfReleases`) are automatically moved to a dead-letter queue for inspection.

By default, only metadata is stored in the poison queue. To preserve the original payload for
retry or debugging, also set `preservePoisonPayload: true`.

To retry failed jobs, run a worker directly against the poison queue:
```bash
./flow flowpack.jobqueue.common:job:work my-azure-storage-queue-poison
```

Note: jobs being processed from the poison queue will not be re-poisoned on failure — they
are left visible in the queue for manual inspection instead.

## Caveats
* 7-day message limit - Azure Queue Storage automatically deletes messages after 7 days maximum, even if unprocessed
* 64KB queue message size - While the claim check pattern handles larger payloads, it adds latency and blob storage costs
* No native queue priorities - Priority queues are simulated by polling multiple queues, which increases API calls
* Approximate counts only - Queue metrics like countReady() are estimates, not exact counts, due to Azure's distributed nature
* No message ordering guarantee - Azure Queue Storage doesn't guarantee FIFO ordering, messages may be processed out of sequence
* Poison queue retry is manual - Failed jobs are moved to a dead-letter queue but not automatically retried.
Loading