diff --git a/AGENTS.md b/AGENTS.md index c645daaf87..df88a7091f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -171,8 +171,19 @@ class My_Module_Checkout_CartController extends Mage_Checkout_CartController { / - **Events**: `Mage::dispatchEvent('event_name', ['data' => $data])` - **Async queue**: `\Maho\Queue\QueueManager::dispatch($messageDto)` queues a flat DTO for a `#[Maho\Config\MessageHandler]` method (message class inferred from the first parameter type); - cron automatically keeps a detached `queue:work` worker alive, with retries/backoff and an - admin grid under System > Message Queue + cron keeps a detached `queue:work` worker alive per pool, with retries/backoff and an admin + grid under System > Message Queue. Worker pools split latency classes: `fast` is resident, + `slow` is the on-demand catch-all. Pass `queue:` to `dispatch()`, then route that queue with + `fast`; anything + unrouted lands in the catch-all, so a long-running handler never blocks short ones. Pool + resourcing (count, memory/time limits, idle timeout) lives under `global/queue/pools`. A crash + parks a claimed message for an operator instead of redelivering it: retry or discard it in the + grid. A handler may run as long as it needs: the worker refreshes its claim on Symfony's + keepalive alarm, so only a worker that actually died is reported as abandoned (the refresh + needs pcntl and cannot land while the handler holds an open transaction on the shared + connection, so such a worker can be misreported after 5 minutes). Worker startup + failures land in `var/log/queue-worker.log`; production installs should prefer supervisord or + systemd over the cron watchdog - **Layout**: XML-based block hierarchy and template assignment - **Sessions**: `Mage::getSingleton('customer/session')`, `'admin/session'`, `'checkout/session'` - **Translations**: `$this->__('Text')`, CSVs in `app/locale/[locale]/` diff --git a/app/code/core/Maho/Queue/Block/Adminhtml/Message/Grid.php b/app/code/core/Maho/Queue/Block/Adminhtml/Message/Grid.php index e91a70f264..e75f67ebf1 100644 --- a/app/code/core/Maho/Queue/Block/Adminhtml/Message/Grid.php +++ b/app/code/core/Maho/Queue/Block/Adminhtml/Message/Grid.php @@ -114,7 +114,7 @@ protected function _prepareMassaction(): static $this->getMassactionBlock()->addItem('retry', [ 'label' => $helper->__('Retry'), 'url' => $this->getUrl('*/*/massRetry'), - 'confirm' => $helper->__('Re-queue the selected failed messages?'), + 'confirm' => $helper->__('Re-queue the selected messages?'), ]); $this->getMassactionBlock()->addItem('discard', [ diff --git a/app/code/core/Maho/Queue/Block/Adminhtml/Message/View.php b/app/code/core/Maho/Queue/Block/Adminhtml/Message/View.php index 930334b9f6..b2f7ddd852 100644 --- a/app/code/core/Maho/Queue/Block/Adminhtml/Message/View.php +++ b/app/code/core/Maho/Queue/Block/Adminhtml/Message/View.php @@ -30,8 +30,20 @@ public function getDiscardUrl(): string return $this->getUrlSecure('*/*/discard', ['id' => $this->getMessage()?->getId()]); } + /** + * Failed, plus a claim old enough to belong to a dead worker: nothing + * re-queues those automatically, so the grid is the only way back. A fresh + * claim is not offered, since re-queueing it would run the handler a second + * time alongside the worker still holding it. + */ public function isRetryable(): bool { - return $this->getMessage()?->getStatus() === Maho_Queue_Model_Message::STATUS_FAILED; + $message = $this->getMessage(); + + return match ($message?->getStatus()) { + Maho_Queue_Model_Message::STATUS_FAILED => true, + Maho_Queue_Model_Message::STATUS_PROCESSING => \Maho\Queue\Transport\DbTransport::isAbandonedClaim($message->getClaimedAt()), + default => false, + }; } } diff --git a/app/code/core/Maho/Queue/Model/Cron.php b/app/code/core/Maho/Queue/Model/Cron.php index 40170f94ec..186b62f4c2 100644 --- a/app/code/core/Maho/Queue/Model/Cron.php +++ b/app/code/core/Maho/Queue/Model/Cron.php @@ -8,36 +8,23 @@ declare(strict_types=1); +use Maho\Queue\Pool; +use Maho\Queue\PoolRegistry; use Maho\Queue\QueueManager; use Maho\Queue\Transport\DbTransport; class Maho_Queue_Model_Cron { /** - * Held by the exclusive worker for its whole life; deliberately a - * machine-local kernel flock even when the db lock backend is configured, - * so every frontend server runs one worker of its own (parallel - * consumption is safe, the transport claim is atomic) and the flock - * disappears the moment the process dies, doubling as the liveness probe. - */ - public const WORKER_LOCK = 'queue.worker'; - - /** Worker recycling cadence: picks up deployed code and frees memory. */ - public const WORKER_TIME_LIMIT = 3600; - public const WORKER_MEMORY_LIMIT = '256M'; - - private const SPAWN_WAIT_ATTEMPTS = 10; - private const SPAWN_WAIT_MICROSECONDS = 500_000; - - /** - * Watchdog: when no worker holds the lock, start a detached - * `queue:work --exclusive` (respawned within a minute of any death, - * recycled hourly via its time limit). + * Watchdog: start a detached `queue:work --exclusive` for every configured + * pool with no live worker, so each latency tier gets a process of its own + * and a slow handler can never sit in front of a fast one. */ #[Maho\Config\CronJob('queue_process', schedule: '* * * * *')] public function process(): void { - if (Mage::getSingleton('core/lock')->isHeld(self::WORKER_LOCK, machineLocal: true)) { + $pending = $this->workersToSpawn(); + if ($pending === []) { return; } @@ -46,7 +33,70 @@ public function process(): void return; } - $this->spawnWorker(); + foreach ($pending as [$pool, $index]) { + $this->spawnWorker($pool, $index); + } + } + + /** + * Split out from process() so the decision is testable without spawning. + * + * @return list + */ + public function workersToSpawn(): array + { + $lock = Mage::getSingleton('core/lock'); + + // A hand-run `queue:work --exclusive` consumes every queue, so it stands + // in for the whole roster and the watchdog keeps out of its way. + if ($lock->isHeld(Pool::LOCK_PREFIX, machineLocal: true)) { + return []; + } + + $spawn = []; + + foreach (PoolRegistry::all() as $pool) { + $due = null; + $idle = null; + $live = 0; + $free = []; + for ($index = 0; $index < $pool->count; $index++) { + if ($lock->isHeld($pool->lockName($index), machineLocal: true)) { + $live++; + } else { + $free[] = $index; + } + } + + foreach ($free as $index) { + // One process per due message. A busy worker cannot take one, and holds + // exactly one claim, so live claims come off the roster. Due and busy + // counts are cluster-global while locks are machine-local, so on + // multi-server installs each server may spawn for the same backlog; + // the excess is bounded by pool->count and drains via idle timeout. + if ($pool->isOnDemand()) { + $due ??= $this->dueWorkCount($pool); + $idle ??= $live === 0 ? 0 : max(0, $live - $this->busyWorkerCount($pool)); + if ($idle >= $due) { + break; + } + $idle++; + } + $spawn[] = [$pool, $index]; + } + } + + return $spawn; + } + + private function dueWorkCount(Pool $pool): int + { + return QueueManager::workerTransport($pool)->countDue($pool->queues); + } + + private function busyWorkerCount(Pool $pool): int + { + return QueueManager::workerTransport($pool)->countClaimed($pool->queues); } #[Maho\Config\CronJob('queue_clean_up', schedule: '0 2 * * *')] @@ -72,25 +122,15 @@ public function cleanup(): void } } - private function spawnWorker(): void + private function spawnWorker(Pool $pool, int $index): void { exec(sprintf( - 'nohup %s %s queue:work --exclusive --time-limit=%d --memory-limit=%s >> %s 2>&1 &', + 'nohup %s %s queue:work --exclusive --pool=%s --index=%d >> %s 2>&1 &', escapeshellarg(PHP_BINARY), escapeshellarg(Mage::getBaseDir() . '/maho'), - self::WORKER_TIME_LIMIT, - escapeshellarg(self::WORKER_MEMORY_LIMIT), + escapeshellarg($pool->name), + $index, escapeshellarg(Mage::getBaseDir('var') . '/log/queue-worker.log'), )); - - $lock = Mage::getSingleton('core/lock'); - for ($attempt = 0; $attempt < self::SPAWN_WAIT_ATTEMPTS; $attempt++) { - usleep(self::SPAWN_WAIT_MICROSECONDS); - if ($lock->isHeld(self::WORKER_LOCK, machineLocal: true)) { - return; - } - } - - Mage::log('Queue worker did not start after spawning; check var/log/queue-worker.log', Mage::LOG_ERROR); } } diff --git a/app/code/core/Maho/Queue/Model/Observer.php b/app/code/core/Maho/Queue/Model/Observer.php new file mode 100644 index 0000000000..008944cf00 --- /dev/null +++ b/app/code/core/Maho/Queue/Model/Observer.php @@ -0,0 +1,72 @@ + + * SPDX-License-Identifier: OSL-3.0 + * @package Maho_Queue + */ + +declare(strict_types=1); + +use Maho\Queue\QueueManager; + +class Maho_Queue_Model_Observer +{ + /** + * Cached count so every admin page load does not pay a queue-table query; + * well under the abandonment threshold, so the notice is never meaningfully + * late. The retry/discard actions drop the key for instant feedback. + */ + public const ABANDONED_COUNT_CACHE_KEY = 'maho_queue_abandoned_count'; + private const ABANDONED_COUNT_CACHE_SECONDS = 60; + + /** + * Nothing re-queues a claim a dead worker left behind, so the grid is the + * only way one comes back and somebody has to be told to look at it. + */ + #[Maho\Config\Observer('controller_action_layout_generate_blocks_before', area: 'adminhtml')] + public function warnAboutAbandonedMessages(): void + { + // An AJAX response renders no message block; the next full page load adds it back. + if (Mage::app()->getRequest()->isXmlHttpRequest()) { + return; + } + + if (!Mage::getSingleton('admin/session')->isAllowed('system/tools/maho_queue/view')) { + return; + } + + $stuck = $this->abandonedMessageCount(); + if ($stuck === 0) { + return; + } + + $helper = Mage::helper('queue'); + Mage::getSingleton('adminhtml/session')->addUniqueMessages([ + Mage::getSingleton('core/message')->notice($helper->__( + '%s queue message(s) were claimed by a worker that never finished. They are not re-queued automatically: retry or discard them.', + $stuck, + Mage::helper('adminhtml')->getUrl('adminhtml/queue'), + )), + ]); + } + + /** Caught rather than probed: isTableExists() would introspect the schema on every request. */ + private function abandonedMessageCount(): int + { + $cached = Mage::app()->loadCache(self::ABANDONED_COUNT_CACHE_KEY); + if (is_string($cached) && $cached !== '') { + return (int) $cached; + } + + try { + $count = QueueManager::dbTransport()->countAbandoned(); + } catch (Exception) { + return 0; + } + + Mage::app()->saveCache((string) $count, self::ABANDONED_COUNT_CACHE_KEY, [], self::ABANDONED_COUNT_CACHE_SECONDS); + + return $count; + } +} diff --git a/app/code/core/Maho/Queue/controllers/Adminhtml/QueueController.php b/app/code/core/Maho/Queue/controllers/Adminhtml/QueueController.php index 41760f98b8..70d2f4c03f 100644 --- a/app/code/core/Maho/Queue/controllers/Adminhtml/QueueController.php +++ b/app/code/core/Maho/Queue/controllers/Adminhtml/QueueController.php @@ -54,12 +54,6 @@ protected function _initAction(): static #[Maho\Config\Route('/admin/queue')] public function indexAction(): void { - if (QueueManager::transportName() === QueueManager::TRANSPORT_REDIS) { - Mage::getSingleton('adminhtml/session')->addNotice( - Mage::helper('queue')->__('The Redis transport is active: pending messages live in Redis and are not listed here, only failures are.'), - ); - } - $this->_title(Mage::helper('queue')->__('Message Queue')); $this->_initAction(); $this->renderLayout(); @@ -104,8 +98,9 @@ public function retryAction(): void if (QueueManager::retryStoredMessage($id)) { Mage::getSingleton('adminhtml/session')->addSuccess(Mage::helper('queue')->__('Message re-queued.')); } else { - Mage::getSingleton('adminhtml/session')->addError(Mage::helper('queue')->__('Only failed messages can be retried.')); + Mage::getSingleton('adminhtml/session')->addError(Mage::helper('queue')->__('The message was not retried: only failed or stuck messages without a newer copy of their dedupe key can be retried.')); } + $this->refreshAbandonedNotice(); $this->_redirect('*/*/'); } @@ -118,6 +113,7 @@ public function discardAction(): void } else { Mage::getSingleton('adminhtml/session')->addError(Mage::helper('queue')->__('Message not found.')); } + $this->refreshAbandonedNotice(); $this->_redirect('*/*/'); } @@ -125,14 +121,22 @@ public function discardAction(): void public function massRetryAction(): void { $retried = 0; + $skipped = 0; foreach ($this->getMessageIds() as $id) { if (QueueManager::retryStoredMessage($id)) { $retried++; + } else { + $skipped++; } } - Mage::getSingleton('adminhtml/session')->addSuccess( - Mage::helper('queue')->__('%s message(s) re-queued.', $retried), - ); + $session = Mage::getSingleton('adminhtml/session'); + if ($retried > 0) { + $session->addSuccess(Mage::helper('queue')->__('%s message(s) re-queued.', $retried)); + } + if ($skipped > 0) { + $session->addNotice(Mage::helper('queue')->__('%s message(s) were skipped: only failed or stuck messages without a newer copy of their dedupe key can be retried.', $skipped)); + } + $this->refreshAbandonedNotice(); $this->_redirect('*/*/'); } @@ -148,9 +152,16 @@ public function massDiscardAction(): void Mage::getSingleton('adminhtml/session')->addSuccess( Mage::helper('queue')->__('%s message(s) discarded.', $discarded), ); + $this->refreshAbandonedNotice(); $this->_redirect('*/*/'); } + /** The abandoned-count notice is cached; an operator action must not leave it stale. */ + private function refreshAbandonedNotice(): void + { + Mage::app()->removeCache(Maho_Queue_Model_Observer::ABANDONED_COUNT_CACHE_KEY); + } + /** * @return list */ diff --git a/app/code/core/Maho/Queue/etc/config.xml b/app/code/core/Maho/Queue/etc/config.xml index 212112b788..167e8885be 100644 --- a/app/code/core/Maho/Queue/etc/config.xml +++ b/app/code/core/Maho/Queue/etc/config.xml @@ -39,6 +39,29 @@ Maho_Queue_Block + + + + + fast + + + + 10 + + + 1 + 60 + 512M + 20 + + + @@ -67,7 +90,6 @@ 60 4 21600 - 3600 0 30 diff --git a/app/code/core/Maho/Queue/etc/system.xml b/app/code/core/Maho/Queue/etc/system.xml index 7687fd709d..d4cf0a4a07 100644 --- a/app/code/core/Maho/Queue/etc/system.xml +++ b/app/code/core/Maho/Queue/etc/system.xml @@ -13,7 +13,7 @@ SPDX-License-Identifier: AFL-3.0 1 0 0 - Background message processing. Maho cron keeps a detached "queue:work" worker running automatically: respawned within a minute if it dies, recycled hourly. + Background message processing. Maho cron keeps one detached "queue:work" worker running per pool: a resident "fast" worker so short jobs never queue behind long ones, plus a "slow" catch-all started only while it has work. Both are respawned within a minute if they die. Pools are declared in config.xml under global/queue/pools. @@ -50,15 +50,6 @@ SPDX-License-Identifier: AFL-3.0 required-entry validate-digits validate-zero-or-greater Upper bound for the backoff. 0 means no bound. - - - 80 - 1 - 0 - 0 - required-entry validate-digits validate-greater-than-zero - Messages claimed by a worker that died are re-queued after this long. Keep above the runtime of your slowest handler. - 90 diff --git a/app/code/core/Maho/Queue/sql/schema.php b/app/code/core/Maho/Queue/sql/schema.php index 5a519e0274..4553dfaa6f 100644 --- a/app/code/core/Maho/Queue/sql/schema.php +++ b/app/code/core/Maho/Queue/sql/schema.php @@ -26,6 +26,7 @@ $message->addColumn('dedupe_key', Types::STRING, ['length' => 64, 'notnull' => false]); $message->addColumn('available_at', Types::DATETIME_MUTABLE, []); $message->addColumn('claimed_at', Types::DATETIME_MUTABLE, ['notnull' => false]); + $message->addColumn('claim_token', Types::STRING, ['length' => 32, 'notnull' => false]); $message->addColumn('processed_at', Types::DATETIME_MUTABLE, ['notnull' => false]); $message->addColumn('created_at', Types::DATETIME_MUTABLE, ['default' => new CurrentTimestamp()]); // Transport keeps updated_at current on every write; the on-update @@ -34,7 +35,11 @@ $message->addPrimaryKeyConstraint( PrimaryKeyConstraint::editor()->setUnquotedColumnNames('message_id')->create(), ); + // (status, available_at, queue) serves unfiltered polls and countDue's date + // bound; (status, queue, available_at) lets a pool worker's queue IN (...) + // poll seek instead of walking every due row of the other pools' backlogs. $message->addIndex(['status', 'available_at', 'queue']); + $message->addIndex(['status', 'queue', 'available_at']); $message->addIndex(['dedupe_key']); $message->addIndex(['created_at']); $message->addIndex(['status', 'processed_at']); diff --git a/app/etc/local.xml.template b/app/etc/local.xml.template index 6880375f73..99beecc97a 100644 --- a/app/etc/local.xml.template +++ b/app/etc/local.xml.template @@ -46,16 +46,6 @@ SPDX-License-Identifier: AFL-3.0 db --> - - {{admin_frontname}} diff --git a/app/locale/en_US/Maho_Queue.csv b/app/locale/en_US/Maho_Queue.csv index 1dd00e20de..298fdc38e9 100644 --- a/app/locale/en_US/Maho_Queue.csv +++ b/app/locale/en_US/Maho_Queue.csv @@ -1,12 +1,14 @@ "%s message(s) discarded.","%s message(s) discarded." "%s message(s) re-queued.","%s message(s) re-queued." +"%s message(s) were skipped: only failed or stuck messages without a newer copy of their dedupe key can be retried.","%s message(s) were skipped: only failed or stuck messages without a newer copy of their dedupe key can be retried." +"%s queue message(s) were claimed by a worker that never finished. They are not re-queued automatically: retry or discard them.","%s queue message(s) were claimed by a worker that never finished. They are not re-queued automatically: retry or discard them." "0 keeps failed messages forever.","0 keeps failed messages forever." "0 removes messages immediately on success; a positive value keeps them visible in the grid for this many days.","0 removes messages immediately on success; a positive value keeps them visible in the grid for this many days." "Action","Action" "Available","Available" "Available (UTC)","Available (UTC)" "Back","Back" -"Background message processing. Maho cron keeps a detached ""queue:work"" worker running automatically: respawned within a minute if it dies, recycled hourly.","Background message processing. Maho cron keeps a detached ""queue:work"" worker running automatically: respawned within a minute if it dies, recycled hourly." +"Background message processing. Maho cron keeps one detached ""queue:work"" worker running per pool: a resident ""fast"" worker so short jobs never queue behind long ones, plus a ""slow"" catch-all started only while it has work. Both are respawned within a minute if they die. Pools are declared in config.xml under global/queue/pools.","Background message processing. Maho cron keeps one detached ""queue:work"" worker running per pool: a resident ""fast"" worker so short jobs never queue behind long ones, plus a ""slow"" catch-all started only while it has work. Both are respawned within a minute if they die. Pools are declared in config.xml under global/queue/pools." "Body (serialized)","Body (serialized)" "Claimed (UTC)","Claimed (UTC)" "Completed","Completed" @@ -30,8 +32,6 @@ "Message not found.","Message not found." "Message Queue","Message Queue" "Message re-queued.","Message re-queued." -"Messages claimed by a worker that died are re-queued after this long. Keep above the runtime of your slowest handler.","Messages claimed by a worker that died are re-queued after this long. Keep above the runtime of your slowest handler." -"Only failed messages can be retried.","Only failed messages can be retried." "Pending","Pending" "Permanently delete the selected messages?","Permanently delete the selected messages?" "Permanently delete this message?","Permanently delete this message?" @@ -40,15 +40,14 @@ "Queue","Queue" "Queued","Queued" "Queued (UTC)","Queued (UTC)" -"Re-queue the selected failed messages?","Re-queue the selected failed messages?" +"Re-queue the selected messages?","Re-queue the selected messages?" "Re-queue this message?","Re-queue this message?" -"Redeliver Stuck Messages After (seconds)","Redeliver Stuck Messages After (seconds)" "Retries","Retries" "Retry","Retry" "Retry Delay Multiplier","Retry Delay Multiplier" "Retry Messages","Retry Messages" "Status","Status" -"The Redis transport is active: pending messages live in Redis and are not listed here, only failures are.","The Redis transport is active: pending messages live in Redis and are not listed here, only failures are." +"The message was not retried: only failed or stuck messages without a newer copy of their dedupe key can be retried.","The message was not retried: only failed or stuck messages without a newer copy of their dedupe key can be retried." "Upper bound for the backoff. 0 means no bound.","Upper bound for the backoff. 0 means no bound." "View","View" "View Messages","View Messages" diff --git a/composer.json b/composer.json index 7ed7e39160..ea14fa5644 100644 --- a/composer.json +++ b/composer.json @@ -117,7 +117,6 @@ "mahocommerce/module-braintree": "Braintree payment gateway integration", "nyholm/psr7": "Required (together with symfony/mcp-bundle) for the MCP protocol at /api/mcp: supplies the PSR-17 factories its HTTP transport discovers. Only a require-dev of mcp-bundle, so nothing pulls it in for you. Any psr/http-factory-implementation works.", "picqer/php-barcode-generator": "Required for gift card barcode generation", - "symfony/redis-messenger": "Enables the Redis transport for the Maho message queue when global/queue/dsn is set in app/etc/local.xml", "symfony/twig-bundle": "Required (together with symfony/asset) to render the API Platform Swagger UI, ReDoc, and GraphiQL explorer pages. Without it, the JSON API and /api/docs.json still work but the human-browsable docs at /api/docs do not.", "symfony/asset": "Required (together with symfony/twig-bundle) for the API Platform Swagger UI: provides the asset() function its template uses.", "symfony/mcp-bundle": "Required (together with nyholm/psr7) for the MCP protocol at /api/mcp: exposes the REST v2 surface to AI agents. Brings mcp/sdk with it. Without both the toggle in System > Config > Services > API stays inert." diff --git a/lib/Maho/Queue/Pool.php b/lib/Maho/Queue/Pool.php new file mode 100644 index 0000000000..c2adb25fb4 --- /dev/null +++ b/lib/Maho/Queue/Pool.php @@ -0,0 +1,75 @@ + + * SPDX-License-Identifier: OSL-3.0 + */ + +declare(strict_types=1); + +namespace Maho\Queue; + +/** + * A worker pool: one or more `queue:work` processes consuming a subset of the + * logical queues, with their own limits. Pools keep + * latency classes apart, so a ten-minute feed build cannot sit in front of an + * order confirmation email. + */ +final readonly class Pool +{ + public const LOCK_PREFIX = 'queue.worker'; + + /** + * @param list $queues Consume only these queues; empty consumes every queue not excluded + * @param list $excludedQueues Never consume these; the catch-all pool excludes every other pool's queues + * @param ?int $idleTimeout Seconds of continuous idleness before exiting; null keeps the worker resident + */ + public function __construct( + public string $name, + public array $queues = [], + public array $excludedQueues = [], + public int $count = 1, + public ?int $idleTimeout = null, + public string $memoryLimit = '256M', + public int $timeLimit = 3600, + ) {} + + /** + * On-demand pools exit once idle, so the watchdog only starts them when + * their queues have work due. + */ + public function isOnDemand(): bool + { + return $this->idleTimeout !== null; + } + + /** Taken machine-local: the flock dies with the process and doubles as the liveness probe. */ + public function lockName(int $index = 0): string + { + return self::LOCK_PREFIX . '.' . $this->name . '.' . $index; + } + + public function consumes(string $queue): bool + { + if (in_array($queue, $this->excludedQueues, true)) { + return false; + } + + return $this->queues === [] || in_array($queue, $this->queues, true); + } + + /** Bytes for a shorthand like "256M", or null when the string is not a memory limit at all. */ + public static function parseMemoryLimit(string $limit): ?int + { + if (!preg_match('/^(\d+)([KMG]?)$/i', trim($limit), $matches)) { + return null; + } + + return (int) $matches[1] * match (strtoupper($matches[2])) { + 'K' => 1024, + 'M' => 1024 ** 2, + 'G' => 1024 ** 3, + default => 1, + }; + } +} diff --git a/lib/Maho/Queue/PoolRegistry.php b/lib/Maho/Queue/PoolRegistry.php new file mode 100644 index 0000000000..0650ac1f05 --- /dev/null +++ b/lib/Maho/Queue/PoolRegistry.php @@ -0,0 +1,200 @@ + + * SPDX-License-Identifier: OSL-3.0 + */ + +declare(strict_types=1); + +namespace Maho\Queue; + +/** + * Reads the worker pools declared under ``, so a module + * can attach its queue to a tier without touching core. + * + * Exactly one pool should be marked `1`: it consumes + * every queue no other pool claims. Core makes the *slow* pool the catch-all + * on purpose, since an unclassified handler is likelier to be a slow newcomer + * than a latency-critical one, and forgetting to declare a tier should cost a + * message waiting behind a feed rather than a checkout email stuck behind one. + */ +final class PoolRegistry +{ + public const FALLBACK_POOL = 'default'; + + /** @var array|null */ + private static ?array $pools = null; + + /** + * @return array + */ + public static function all(): array + { + return self::$pools ??= self::build(); + } + + public static function get(string $name): ?Pool + { + return self::all()[$name] ?? null; + } + + /** Null means the queue is orphaned and no worker will ever pick it up. */ + public static function poolFor(string $queue): ?Pool + { + foreach (self::all() as $pool) { + if ($pool->consumes($queue)) { + return $pool; + } + } + + return null; + } + + public static function reset(): void + { + self::$pools = null; + } + + /** + * @return array + */ + private static function build(): array + { + $node = \Mage::getConfig()->getNode('global/queue'); + $poolsNode = $node !== false && isset($node->pools) ? $node->pools : false; + + $definitions = []; + if ($poolsNode !== false) { + foreach ($poolsNode->children() as $name => $child) { + if (!self::flag($child, 'active', true)) { + continue; + } + $definitions[(string) $name] = $child; + } + } + + if ($definitions === []) { + return [self::FALLBACK_POOL => new Pool(self::FALLBACK_POOL)]; + } + + uasort($definitions, fn($a, $b) => (int) $a->sort_order <=> (int) $b->sort_order); + + $catchAll = null; + foreach ($definitions as $name => $child) { + if (!self::flag($child, 'catch_all', false)) { + continue; + } + if ($catchAll === null) { + $catchAll = $name; + continue; + } + \Mage::log( + sprintf('Queue pool "%s" is marked catch_all but "%s" already is; dropping it', $name, $catchAll), + \Mage::LOG_ERROR, + ); + unset($definitions[$name]); + } + + if ($catchAll === null) { + \Mage::log( + 'No queue pool is marked catch_all: messages dispatched to an unrouted queue will never be consumed', + \Mage::LOG_ERROR, + ); + } + + $queuesByPool = self::routing($node, $definitions, $catchAll); + + // A pool with nothing routed to it would consume everything and become a + // second catch-all, so it is dropped rather than left to compete. + foreach ($queuesByPool as $name => $queues) { + if ($queues === [] && $name !== $catchAll) { + \Mage::log(sprintf('Queue pool "%s" has no queues routed to it; skipping', $name), \Mage::LOG_NOTICE); + unset($definitions[$name], $queuesByPool[$name]); + } + } + + $pools = []; + foreach ($definitions as $name => $child) { + $excluded = []; + if ($name === $catchAll) { + $excluded = array_values(array_unique(array_merge(...array_values( + array_diff_key($queuesByPool, [$name => true]), + )))); + } + + $pools[$name] = new Pool( + name: (string) $name, + queues: $queuesByPool[$name], + excludedQueues: $excluded, + count: max(1, (int) ($child->count ?? 1)), + idleTimeout: isset($child->idle_timeout) ? max(0, (int) $child->idle_timeout) : null, + memoryLimit: self::memoryLimit($child, (string) $name), + timeLimit: max(0, (int) ($child->time_limit ?? 3600)), + ); + } + + // Every declared pool was dropped (none marked catch_all, none routed a + // queue): without a fallback the watchdog would start nothing at all and + // the queue would stall silently. + return $pools === [] ? [self::FALLBACK_POOL => new Pool(self::FALLBACK_POOL)] : $pools; + } + + /** + * The queue to pool map, one node per queue so a module can route its own + * queue and local.xml can retarget a single one without clobbering the rest. + * An empty value unroutes a queue, handing it back to the catch-all. + * + * @param array $definitions + * @return array> + */ + private static function routing(\Mage_Core_Model_Config_Element|false $node, array $definitions, ?string $catchAll): array + { + $queuesByPool = array_fill_keys(array_keys($definitions), []); + if ($node === false || !isset($node->routing)) { + return $queuesByPool; + } + + foreach ($node->routing->children() as $queue => $target) { + $pool = trim((string) $target); + if ($pool === '' || $pool === $catchAll) { + continue; + } + if (!isset($definitions[$pool])) { + \Mage::log( + sprintf('Queue "%s" is routed to unknown pool "%s"; it falls to the catch-all', $queue, $pool), + \Mage::LOG_ERROR, + ); + continue; + } + $queuesByPool[$pool][] = (string) $queue; + } + + return $queuesByPool; + } + + /** Defers to is() so 'true' means the same as everywhere else in config. */ + private static function flag(\Mage_Core_Model_Config_Element $node, string $child, bool $default): bool + { + return isset($node->{$child}) ? $node->is($child) : $default; + } + + /** + * Rejected here rather than in the spawned worker: an invalid limit there + * makes queue:work exit before taking its lock, so the watchdog would + * respawn it every minute while the pool's queues silently stop draining. + */ + private static function memoryLimit(\Mage_Core_Model_Config_Element $child, string $name): string + { + $limit = isset($child->memory_limit) ? trim((string) $child->memory_limit) : '256M'; + if ($limit !== '' && Pool::parseMemoryLimit($limit) === null) { + \Mage::log( + sprintf('Queue pool "%s" has an invalid memory_limit "%s"; using 256M', $name, $limit), + \Mage::LOG_ERROR, + ); + return '256M'; + } + + return $limit; + } +} diff --git a/lib/Maho/Queue/QueueManager.php b/lib/Maho/Queue/QueueManager.php index 662e2ad91e..73dee41982 100644 --- a/lib/Maho/Queue/QueueManager.php +++ b/lib/Maho/Queue/QueueManager.php @@ -22,7 +22,6 @@ use Symfony\Component\Messenger\Stamp\DelayStamp; use Symfony\Component\Messenger\Stamp\StampInterface; use Symfony\Component\Messenger\Transport\Sender\SendersLocator; -use Symfony\Component\Messenger\Transport\TransportInterface; /** * Entry point of the Maho message queue: dispatch a message object and a @@ -32,30 +31,25 @@ * \Maho\Queue\QueueManager::dispatch(new My_Module_Model_SomeMessage(...)); * ``` * - * The default transport stores messages in the maho_queue_message table; a - * `redis://...` node under `` in - * app/etc/local.xml switches to Redis (requires symfony/redis-messenger). - * With the DB transport, dispatching inside a database transaction - * participates in it: the message becomes visible only on commit. + * Messages are stored in the maho_queue_message table, so dispatching inside a + * database transaction participates in it: the message becomes visible only on + * commit. */ final class QueueManager { + /** The name this transport is registered under with Messenger. */ public const TRANSPORT_DB = 'db'; - public const TRANSPORT_REDIS = 'redis'; public const XML_PATH_MAX_RETRIES = 'system/queue/max_retries'; public const XML_PATH_RETRY_DELAY = 'system/queue/retry_delay'; public const XML_PATH_RETRY_MULTIPLIER = 'system/queue/retry_multiplier'; public const XML_PATH_RETRY_MAX_DELAY = 'system/queue/retry_max_delay'; - public const XML_PATH_REDELIVER_AFTER = 'system/queue/redeliver_after'; public const XML_PATH_COMPLETED_RETENTION = 'system/queue/completed_retention'; public const XML_PATH_FAILED_RETENTION = 'system/queue/failed_retention'; private static ?MessageBus $bus = null; - private static ?TransportInterface $transport = null; private static ?DbTransport $dbTransport = null; private static ?Serializer $serializer = null; - private static ?string $transportName = null; /** * Dispatch a message for asynchronous handling. @@ -93,49 +87,39 @@ public static function bus(): MessageBusInterface return self::$bus ??= new MessageBus([ new AddBusNameStampMiddleware('maho'), new SendMessageMiddleware(new SendersLocator( - ['*' => [self::transportName()]], - new ServiceLocator([self::transportName() => self::transport()]), + ['*' => [self::TRANSPORT_DB]], + new ServiceLocator([self::TRANSPORT_DB => self::dbTransport()]), )), new HandleMessageMiddleware(HandlerRegistry::handlersLocator()), ]); } - public static function transport(): TransportInterface + public static function dbTransport(): DbTransport { - if (self::$transport !== null) { - return self::$transport; - } - - if (self::transportName() === self::TRANSPORT_REDIS) { - $factory = new \Symfony\Component\Messenger\Bridge\Redis\Transport\RedisTransportFactory(); // @phpstan-ignore class.notFound - return self::$transport = $factory->createTransport((string) self::redisDsn(), [], self::serializer()); // @phpstan-ignore class.notFound - } - - return self::$transport = self::dbTransport(); + return self::$dbTransport ??= new DbTransport( + self::writeAdapter(), + self::tableName(), + self::serializer(), + (int) \Mage::getStoreConfig(self::XML_PATH_COMPLETED_RETENTION), + ); } - public static function transportName(): string + /** + * The transport a pool's worker consumes from: the shared one unless the + * pool narrows what it sees, in which case it gets its own instance. + */ + public static function workerTransport(?Pool $pool = null): DbTransport { - if (self::$transportName !== null) { - return self::$transportName; - } - - $dsn = self::redisDsn(); - if ($dsn !== null && !class_exists('Symfony\Component\Messenger\Bridge\Redis\Transport\RedisTransportFactory')) { - throw new \RuntimeException('global/queue/dsn is set in app/etc/local.xml but the symfony/redis-messenger package is not installed; run "composer require symfony/redis-messenger" or remove the node'); + if ($pool === null || $pool->excludedQueues === []) { + return self::dbTransport(); } - return self::$transportName = $dsn !== null ? self::TRANSPORT_REDIS : self::TRANSPORT_DB; - } - - public static function dbTransport(): DbTransport - { - return self::$dbTransport ??= new DbTransport( + return new DbTransport( self::writeAdapter(), self::tableName(), self::serializer(), - (int) \Mage::getStoreConfig(self::XML_PATH_REDELIVER_AFTER), (int) \Mage::getStoreConfig(self::XML_PATH_COMPLETED_RETENTION), + $pool->excludedQueues, ); } @@ -145,46 +129,61 @@ public static function serializer(): Serializer } /** - * Re-queue a stored failed message from the admin grid or CLI. DB mode - * flips the row back to pending with a fresh retry budget; Redis failure - * rows are re-dispatched through the bus and the stored row removed. - * Only failed rows are retryable: flipping a claimed row would race the - * worker's ack. + * Re-queue a stored message from the admin grid or CLI, flipping the row + * back to pending with a fresh retry budget. Failed rows and claims old + * enough to belong to a dead worker are both retryable; nothing else, since + * a pending row needs no help and a completed one is done. Nothing re-queues + * an abandoned claim automatically, so this is the only way one comes back. + * + * A fresh claim is refused on purpose: re-queueing a row a live worker is + * still inside runs the handler a second time, the very thing dropping + * timer-based redelivery was meant to make impossible. The age cut-off is + * applied in the UPDATE, so a worker that claims the row between the read + * and the write keeps it. + * + * A deduped row is refused while a newer copy of its key is in flight: + * once an abandoned claim stops suppressing dispatches, the fresh copy + * supersedes it and retrying the old one would run the job twice. Discard + * is the way out for that row. */ public static function retryStoredMessage(int $messageId): bool { $adapter = self::writeAdapter(); $table = self::tableName(); $row = $adapter->fetchRow( - $adapter->select()->from($table)->where('message_id = ?', $messageId), + $adapter->select()->from($table, ['status', 'dedupe_key'])->where('message_id = ?', $messageId), ); - if ($row === false || $row['status'] !== DbTransport::STATUS_FAILED) { + if ($row === false) { return false; } - if (self::transportName() === self::TRANSPORT_DB) { - $now = \Mage_Core_Model_Locale::nowUtc(); - return $adapter->update($table, [ - 'status' => DbTransport::STATUS_PENDING, - 'retries' => 0, - 'available_at' => $now, - 'claimed_at' => null, - 'processed_at' => null, - 'updated_at' => $now, - ], [ - 'message_id = ?' => $messageId, - 'status = ?' => DbTransport::STATUS_FAILED, - ]) === 1; + $where = ['message_id = ?' => $messageId]; + if ($row['status'] === DbTransport::STATUS_PROCESSING) { + $where['status = ?'] = DbTransport::STATUS_PROCESSING; + $where['claimed_at < ?'] = DbTransport::abandonedBefore(); + } elseif ($row['status'] === DbTransport::STATUS_FAILED) { + $where['status = ?'] = DbTransport::STATUS_FAILED; + } else { + return false; } - $envelope = self::serializer()->decode([ - 'body' => (string) $row['body'], - 'headers' => ['type' => (string) $row['message_class']], - ]); - self::bus()->dispatch($envelope->getMessage(), [new QueueNameStamp((string) $row['queue'])]); - $adapter->delete($table, ['message_id = ?' => $messageId]); + if ($row['dedupe_key'] !== null + && self::dbTransport()->inFlightRowExists((string) $row['dedupe_key'], $messageId) + ) { + return false; + } - return true; + $now = \Mage_Core_Model_Locale::nowUtc(); + + return $adapter->update($table, [ + 'status' => DbTransport::STATUS_PENDING, + 'retries' => 0, + 'available_at' => $now, + 'claimed_at' => null, + 'claim_token' => null, + 'processed_at' => null, + 'updated_at' => $now, + ], $where) === 1; } public static function discardStoredMessage(int $messageId): bool @@ -203,22 +202,10 @@ public static function tableName(): string public static function reset(): void { self::$bus = null; - self::$transport = null; self::$dbTransport = null; self::$serializer = null; - self::$transportName = null; HandlerRegistry::reset(); - } - - private static function redisDsn(): ?string - { - $node = \Mage::getConfig()->getNode('global/queue/dsn'); - if ($node === false) { - return null; - } - $dsn = trim((string) $node); - - return $dsn === '' ? null : $dsn; + PoolRegistry::reset(); } private static function writeAdapter(): \Maho\Db\Adapter\AdapterInterface diff --git a/lib/Maho/Queue/Stamp/ClaimTokenStamp.php b/lib/Maho/Queue/Stamp/ClaimTokenStamp.php new file mode 100644 index 0000000000..f5516322be --- /dev/null +++ b/lib/Maho/Queue/Stamp/ClaimTokenStamp.php @@ -0,0 +1,24 @@ + + * SPDX-License-Identifier: OSL-3.0 + */ + +declare(strict_types=1); + +namespace Maho\Queue\Stamp; + +use Symfony\Component\Messenger\Stamp\StampInterface; + +/** + * Proof of claim ownership: written to the row when a worker claims it and + * carried on the envelope, so a stale worker's late ack, reject, re-send or + * keepalive can never touch a row another worker has since claimed. + */ +final readonly class ClaimTokenStamp implements StampInterface +{ + public function __construct( + public string $token, + ) {} +} diff --git a/lib/Maho/Queue/StopWorkerWhenIdleListener.php b/lib/Maho/Queue/StopWorkerWhenIdleListener.php index e877f83255..e4541b372f 100644 --- a/lib/Maho/Queue/StopWorkerWhenIdleListener.php +++ b/lib/Maho/Queue/StopWorkerWhenIdleListener.php @@ -13,15 +13,30 @@ use Symfony\Component\Messenger\Event\WorkerRunningEvent; /** - * Stops the worker as soon as a poll finds no message, for bounded runs - * (cron consumer, `queue:work --stop-when-empty`). Messenger has no built-in - * stop-on-idle listener. + * Stops the worker once polls stop finding messages, for bounded runs and for + * on-demand pools that free their process between bursts. Messenger has no + * built-in stop-on-idle listener. + * + * On-demand pools want a grace period: exiting on the first empty poll makes a + * job queued seconds later wait a whole cron tick for a replacement. */ final class StopWorkerWhenIdleListener implements EventSubscriberInterface { + private ?int $idleSince = null; + + public function __construct( + private readonly int $idleTimeoutSeconds = 0, + ) {} + public function onWorkerRunning(WorkerRunningEvent $event): void { - if ($event->isWorkerIdle()) { + if (!$event->isWorkerIdle()) { + $this->idleSince = null; + return; + } + + $this->idleSince ??= time(); + if (time() - $this->idleSince >= $this->idleTimeoutSeconds) { $event->getWorker()->stop(); } } diff --git a/lib/Maho/Queue/Transport/DbTransport.php b/lib/Maho/Queue/Transport/DbTransport.php index 8f606cdab7..7921adf5fc 100644 --- a/lib/Maho/Queue/Transport/DbTransport.php +++ b/lib/Maho/Queue/Transport/DbTransport.php @@ -10,6 +10,7 @@ namespace Maho\Queue\Transport; use Maho\Db\Adapter\AdapterInterface; +use Maho\Queue\Stamp\ClaimTokenStamp; use Maho\Queue\Stamp\DedupeKeyStamp; use Maho\Queue\Stamp\QueueNameStamp; use Symfony\Component\Messenger\Envelope; @@ -20,6 +21,7 @@ use Symfony\Component\Messenger\Stamp\RedeliveryStamp; use Symfony\Component\Messenger\Stamp\SentToFailureTransportStamp; use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp; +use Symfony\Component\Messenger\Transport\Receiver\KeepaliveReceiverInterface; use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface; use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface; use Symfony\Component\Messenger\Transport\Receiver\QueueReceiverInterface; @@ -38,7 +40,7 @@ * The table is declared in Maho_Queue's sql/schema.php and is never * auto-created here. */ -final class DbTransport implements TransportInterface, QueueReceiverInterface, ListableReceiverInterface, MessageCountAwareInterface +final class DbTransport implements TransportInterface, QueueReceiverInterface, ListableReceiverInterface, MessageCountAwareInterface, KeepaliveReceiverInterface { public const STATUS_PENDING = 'pending'; public const STATUS_PROCESSING = 'processing'; @@ -47,12 +49,25 @@ final class DbTransport implements TransportInterface, QueueReceiverInterface, L public const DEFAULT_QUEUE = 'default'; + /** How often a worker refreshes the claim of the message it is handling. */ + public const KEEPALIVE_INTERVAL_SECONDS = 5; + + /** + * Sixty missed refreshes: a claim this stale belongs to a worker that died. + * Gates the admin notice, the admin retry and the dedupe check; nothing + * redelivers on it. + */ + public const ABANDONED_AFTER_SECONDS = 300; + + /** + * @param list $excludedQueues Queues this instance never consumes, so a pool worker can be "everything but" + */ public function __construct( private readonly AdapterInterface $adapter, private readonly string $table, private readonly Serializer $serializer, - private readonly int $redeliverAfterSeconds, private readonly int $completedRetentionDays, + private readonly array $excludedQueues = [], ) {} #[\Override] @@ -66,17 +81,21 @@ public function send(Envelope $envelope): Envelope $messageIdStamp = $envelope->last(TransportMessageIdStamp::class); $redeliveryStamp = $envelope->last(RedeliveryStamp::class); - // A failure-transport send carries both stamps too, but its id is a Redis stream id: insert, not update. + // A failure-transport send carries both stamps too, but its id belongs to the origin transport: insert, not update. if ($messageIdStamp !== null && $redeliveryStamp !== null && $envelope->last(SentToFailureTransportStamp::class) === null) { + // The ownership guard mirrors ack(): a row an operator already re-queued, + // completed, deleted or that another worker has since claimed must + // survive a stale worker's late re-send. $this->adapter->update($this->table, [ 'status' => self::STATUS_PENDING, 'retries' => $redeliveryStamp->getRetryCount(), 'available_at' => $availableAt, 'error_message' => $envelope->last(ErrorDetailsStamp::class)?->getExceptionMessage(), 'claimed_at' => null, + 'claim_token' => null, 'updated_at' => $now, - ], ['message_id = ?' => (int) $messageIdStamp->getId()]); + ], $this->ownershipWhere($envelope)); return $envelope; } @@ -119,19 +138,26 @@ public function getFromQueues(array $queueNames): iterable return $this->claimNext($queueNames); } + /** + * The ownership guard mirrors reject() and keepalive(): a row an operator + * already flipped back to pending, or that another worker has since + * claimed, must survive the late ack of the worker that used to hold it, + * or the retry (or the other worker's run) is silently swallowed. + */ #[\Override] public function ack(Envelope $envelope): void { - $messageId = $this->messageId($envelope); + $where = $this->ownershipWhere($envelope); if ($this->completedRetentionDays > 0) { $now = \Mage_Core_Model_Locale::nowUtc(); $this->adapter->update($this->table, [ 'status' => self::STATUS_COMPLETED, + 'claim_token' => null, 'processed_at' => $now, 'updated_at' => $now, - ], ['message_id = ?' => $messageId]); + ], $where); } else { - $this->adapter->delete($this->table, ['message_id = ?' => $messageId]); + $this->adapter->delete($this->table, $where); } } @@ -142,13 +168,56 @@ public function reject(Envelope $envelope): void // A row the retry listener already re-queued in place stays pending. $this->adapter->update($this->table, [ 'status' => self::STATUS_FAILED, + 'claim_token' => null, 'error_message' => $envelope->last(ErrorDetailsStamp::class)?->getExceptionMessage(), 'processed_at' => $now, 'updated_at' => $now, - ], [ + ], $this->ownershipWhere($envelope)); + } + + /** + * Makes claimed_at mean "a worker is alive here", not "a worker started here + * a while ago". The ownership guard leaves a row an operator already retried + * (or another worker has since claimed) alone. + */ + #[\Override] + public function keepalive(Envelope $envelope, ?int $seconds = null): void + { + // The alarm can fire while the handler holds an open transaction on this + // shared connection; joining it would lock the row against the admin's + // retry and lose the refresh on rollback. Skip and let the next one land. + if ($this->adapter->getTransactionLevel() > 0) { + return; + } + + $now = \Mage_Core_Model_Locale::nowUtc(); + $this->adapter->update($this->table, [ + 'claimed_at' => $now, + 'updated_at' => $now, + ], $this->ownershipWhere($envelope)); + } + + /** + * WHERE clause proving the caller still owns the row: the claim token + * written when it claimed, when the envelope carries one. Status alone + * cannot tell "my claim" from a newer claim another worker took after an + * operator retried this one. + * + * @return array + */ + private function ownershipWhere(Envelope $envelope): array + { + $where = [ 'message_id = ?' => $this->messageId($envelope), 'status = ?' => self::STATUS_PROCESSING, - ]); + ]; + + $token = $envelope->last(ClaimTokenStamp::class)?->token; + if ($token !== null) { + $where['claim_token = ?'] = $token; + } + + return $where; } #[\Override] @@ -182,11 +251,72 @@ public function find(mixed $id): ?Envelope #[\Override] public function getMessageCount(): int { - return (int) $this->adapter->fetchOne( - $this->adapter->select() - ->from($this->table, new \Maho\Db\Expr('COUNT(*)')) - ->where('status = ?', self::STATUS_PENDING), - ); + return $this->countRows(['status = ?' => self::STATUS_PENDING]); + } + + /** + * Work this instance would pick up right now. Messages scheduled for the + * future are deliberately excluded, or the watchdog would respawn an + * on-demand worker every cron tick until a delayed campaign's send date. + * + * @param list|null $queues + */ + public function countDue(?array $queues = null): int + { + return $this->countRows([ + 'status = ?' => self::STATUS_PENDING, + 'available_at <= ?' => \Mage_Core_Model_Locale::nowUtc(), + ], $queues); + } + + /** Drives the admin notice. A working worker refreshes its claim, so this counts dead ones. */ + public function countAbandoned(int $olderThanSeconds = self::ABANDONED_AFTER_SECONDS): int + { + return $this->countRows([ + 'status = ?' => self::STATUS_PROCESSING, + 'claimed_at < ?' => self::abandonedBefore($olderThanSeconds), + ]); + } + + /** + * Live claims, so the watchdog can tell a pool's busy workers from its idle ones. + * + * @param list|null $queues + */ + public function countClaimed(?array $queues = null): int + { + return $this->countRows([ + 'status = ?' => self::STATUS_PROCESSING, + 'claimed_at >= ?' => self::abandonedBefore(), + ], $queues); + } + + /** + * @param array $conditions + * @param list|null $queues + */ + private function countRows(array $conditions, ?array $queues = null): int + { + $select = $this->adapter->select()->from($this->table, new \Maho\Db\Expr('COUNT(*)')); + foreach ($conditions as $condition => $value) { + $select->where($condition, $value); + } + $this->applyQueueFilter($select, $queues); + + return (int) $this->adapter->fetchOne($select); + } + + /** + * @param list|null $queues + */ + private function applyQueueFilter(\Maho\Db\Select $select, ?array $queues): void + { + if ($queues !== null && $queues !== []) { + $select->where('queue IN (?)', $queues); + } + if ($this->excludedQueues !== []) { + $select->where('queue NOT IN (?)', $this->excludedQueues); + } } /** @@ -195,7 +325,6 @@ public function getMessageCount(): int */ private function claimNext(?array $queues): array { - $this->requeueStaleClaims(); $now = \Mage_Core_Model_Locale::nowUtc(); for ($attempt = 0; $attempt < 5; $attempt++) { @@ -205,18 +334,18 @@ private function claimNext(?array $queues): array ->where('available_at <= ?', $now) ->order(['available_at ASC', 'message_id ASC']) ->limit(1); - if ($queues !== null && $queues !== []) { - $select->where('queue IN (?)', $queues); - } + $this->applyQueueFilter($select, $queues); $row = $this->adapter->fetchRow($select); if ($row === false) { return []; } + $token = bin2hex(random_bytes(16)); $claimed = $this->adapter->update($this->table, [ 'status' => self::STATUS_PROCESSING, 'claimed_at' => $now, + 'claim_token' => $token, 'updated_at' => $now, ], [ 'message_id = ?' => (int) $row['message_id'], @@ -229,7 +358,7 @@ private function claimNext(?array $queues): array $envelope = $this->hydrateOrFail($row); if ($envelope !== null) { - return [$envelope]; + return [$envelope->with(new ClaimTokenStamp($token))]; } } @@ -237,36 +366,39 @@ private function claimNext(?array $queues): array } /** - * Crash recovery: rows claimed longer ago than redeliver_after belong to a - * worker that died without ack/reject; put them back up for grabs. + * A pending or freshly claimed row already carries this dedupe key: gates + * both new dispatches and the admin retry of an older copy. */ - private function requeueStaleClaims(): void + public function inFlightRowExists(string $dedupeKey, ?int $excludeMessageId = null): bool { - if ($this->redeliverAfterSeconds <= 0) { - return; + // A claim a dead worker left behind waits for an operator forever, so it + // must not keep suppressing new dispatches of the same key: that would + // silently drop every later send instead of parking one message. + // Pending rows always have a null claimed_at, so they are never excluded. + $select = $this->adapter->select() + ->from($this->table, 'message_id') + ->where('dedupe_key = ?', $dedupeKey) + ->where('status IN (?)', [self::STATUS_PENDING, self::STATUS_PROCESSING]) + ->where('claimed_at IS NULL OR claimed_at >= ?', self::abandonedBefore()) + ->limit(1); + if ($excludeMessageId !== null) { + $select->where('message_id != ?', $excludeMessageId); } + $existing = $this->adapter->fetchOne($select); - $this->adapter->update($this->table, [ - 'status' => self::STATUS_PENDING, - 'claimed_at' => null, - 'updated_at' => \Mage_Core_Model_Locale::nowUtc(), - ], [ - 'status = ?' => self::STATUS_PROCESSING, - 'claimed_at < ?' => gmdate(\Mage_Core_Model_Locale::DATETIME_FORMAT, time() - $this->redeliverAfterSeconds), - ]); + return $existing !== false && $existing !== null; } - private function inFlightRowExists(string $dedupeKey): bool + /** UTC cut-off before which a claim counts as abandoned. */ + public static function abandonedBefore(int $olderThanSeconds = self::ABANDONED_AFTER_SECONDS): string { - $existing = $this->adapter->fetchOne( - $this->adapter->select() - ->from($this->table, 'message_id') - ->where('dedupe_key = ?', $dedupeKey) - ->where('status IN (?)', [self::STATUS_PENDING, self::STATUS_PROCESSING]) - ->limit(1), - ); + return gmdate(\Mage_Core_Model_Locale::DATETIME_FORMAT, time() - $olderThanSeconds); + } - return $existing !== false && $existing !== null; + /** True when a processing row's claim is old enough to belong to a dead worker. */ + public static function isAbandonedClaim(?string $claimedAt): bool + { + return $claimedAt !== null && $claimedAt < self::abandonedBefore(); } /** diff --git a/lib/Maho/Queue/WorkerFactory.php b/lib/Maho/Queue/WorkerFactory.php index 5f9807dac1..426f382953 100644 --- a/lib/Maho/Queue/WorkerFactory.php +++ b/lib/Maho/Queue/WorkerFactory.php @@ -14,7 +14,6 @@ use Symfony\Component\Messenger\EventListener\AddErrorDetailsStampListener; use Symfony\Component\Messenger\EventListener\DispatchPcntlSignalListener; use Symfony\Component\Messenger\EventListener\SendFailedMessageForRetryListener; -use Symfony\Component\Messenger\EventListener\SendFailedMessageToFailureTransportListener; use Symfony\Component\Messenger\EventListener\StopWorkerOnMemoryLimitListener; use Symfony\Component\Messenger\EventListener\StopWorkerOnMessageLimitListener; use Symfony\Component\Messenger\Retry\MultiplierRetryStrategy; @@ -28,20 +27,19 @@ final class WorkerFactory { /** - * @param array{limit?: ?int, memoryLimit?: ?int, stopWhenIdle?: bool} $options + * @param array{limit?: ?int, memoryLimit?: ?int, idleTimeout?: ?int, pool?: ?Pool} $options */ public static function create(array $options = []): Worker { - $transportName = QueueManager::transportName(); - $transport = QueueManager::transport(); + $transport = QueueManager::workerTransport($options['pool'] ?? null); $dispatcher = new EventDispatcher(); $dispatcher->addSubscriber(new AddErrorDetailsStampListener()); $dispatcher->addSubscriber(new DispatchPcntlSignalListener()); $dispatcher->addSubscriber(new SendFailedMessageForRetryListener( - new ServiceLocator([$transportName => $transport]), - new ServiceLocator([$transportName => new MultiplierRetryStrategy( + new ServiceLocator([QueueManager::TRANSPORT_DB => $transport]), + new ServiceLocator([QueueManager::TRANSPORT_DB => new MultiplierRetryStrategy( (int) \Mage::getStoreConfig(QueueManager::XML_PATH_MAX_RETRIES), (int) \Mage::getStoreConfig(QueueManager::XML_PATH_RETRY_DELAY) * 1000, (float) \Mage::getStoreConfig(QueueManager::XML_PATH_RETRY_MULTIPLIER), @@ -49,13 +47,6 @@ public static function create(array $options = []): Worker )]), )); - if ($transportName === QueueManager::TRANSPORT_REDIS) { - // Final failures on Redis land in the DB table so the admin grid sees them. - $dispatcher->addSubscriber(new SendFailedMessageToFailureTransportListener( - new ServiceLocator([$transportName => QueueManager::dbTransport()]), - )); - } - $dispatcher->addListener(WorkerMessageFailedEvent::class, static function (WorkerMessageFailedEvent $event): void { if (!$event->willRetry()) { \Mage::logException($event->getThrowable()); @@ -70,10 +61,10 @@ public static function create(array $options = []): Worker if (isset($options['memoryLimit']) && $options['memoryLimit'] > 0) { $dispatcher->addSubscriber(new StopWorkerOnMemoryLimitListener($options['memoryLimit'])); } - if ($options['stopWhenIdle'] ?? false) { - $dispatcher->addSubscriber(new StopWorkerWhenIdleListener()); + if (isset($options['idleTimeout'])) { + $dispatcher->addSubscriber(new StopWorkerWhenIdleListener($options['idleTimeout'])); } - return new Worker([$transportName => $transport], QueueManager::bus(), $dispatcher); + return new Worker([QueueManager::TRANSPORT_DB => $transport], QueueManager::bus(), $dispatcher); } } diff --git a/lib/MahoCLI/Commands/EmailConfigShow.php b/lib/MahoCLI/Commands/EmailConfigShow.php index fd8b6df715..4c9de81b4f 100644 --- a/lib/MahoCLI/Commands/EmailConfigShow.php +++ b/lib/MahoCLI/Commands/EmailConfigShow.php @@ -63,15 +63,11 @@ protected function execute(InputInterface $input, OutputInterface $output): int $table->addRow(['Return Path', $returnPathSetting]); // Queue System Information - if (\Maho\Queue\QueueManager::transportName() === \Maho\Queue\QueueManager::TRANSPORT_REDIS) { - $pendingDisplay = 'Unknown (pending messages live in Redis)'; - } else { - $pendingCount = Mage::getModel('queue/message')->getCollection() - ->addFieldToFilter('queue', \Mage_Core_Model_Email_Queue::QUEUE_NAME) - ->addFieldToFilter('status', \Maho_Queue_Model_Message::STATUS_PENDING) - ->getSize(); - $pendingDisplay = $pendingCount . ' emails'; - } + $pendingCount = Mage::getModel('queue/message')->getCollection() + ->addFieldToFilter('queue', \Mage_Core_Model_Email_Queue::QUEUE_NAME) + ->addFieldToFilter('status', \Maho_Queue_Model_Message::STATUS_PENDING) + ->getSize(); + $pendingDisplay = $pendingCount . ' emails'; // Get cron schedules dynamically $cronHelper = Mage::helper('cron'); diff --git a/lib/MahoCLI/Commands/EmailQueueProcess.php b/lib/MahoCLI/Commands/EmailQueueProcess.php index a8e6884084..65a168ed7c 100644 --- a/lib/MahoCLI/Commands/EmailQueueProcess.php +++ b/lib/MahoCLI/Commands/EmailQueueProcess.php @@ -10,10 +10,9 @@ namespace MahoCLI\Commands; use Mage; -use Maho\Queue\QueueManager; -use Maho\Queue\WorkerFactory; use Symfony\Component\Console\Attribute\AsCommand; use Symfony\Component\Console\Command\Command; +use Symfony\Component\Console\Input\ArrayInput; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Output\OutputInterface; @@ -34,32 +33,29 @@ protected function execute(InputInterface $input, OutputInterface $output): int return Command::FAILURE; } - $isDbTransport = QueueManager::transportName() === QueueManager::TRANSPORT_DB; - if ($isDbTransport) { - $pendingCount = $this->getPendingCount(); - if ($pendingCount === 0) { - $output->writeln('No emails in queue to process.'); - return Command::SUCCESS; - } - $output->writeln("Processing email queue ({$pendingCount} emails pending)..."); - } else { - $output->writeln('Redis transport is active: consuming all queues until empty.'); + $pendingCount = $this->getPendingCount(); + if ($pendingCount === 0) { + $output->writeln('No emails in queue to process.'); + return Command::SUCCESS; } + $output->writeln("Processing email queue ({$pendingCount} emails pending)..."); try { - $worker = WorkerFactory::create(['stopWhenIdle' => true]); - $options = []; - if ($isDbTransport) { - $options['queues'] = [\Mage_Core_Model_Email_Queue::QUEUE_NAME]; + // Not the command's own run(): only the full lifecycle registers its signal handlers. + $status = $this->getApplication()?->doRun(new ArrayInput([ + 'command' => 'queue:work', + '--queue' => [\Mage_Core_Model_Email_Queue::QUEUE_NAME], + '--idle-timeout' => '0', + ]), $output) ?? Command::FAILURE; + + if ($status !== Command::SUCCESS) { + return $status; } - $worker->run($options); $output->writeln('Queue processing completed.'); - if ($isDbTransport) { - $failedCount = $this->getFailedCount(); - if ($failedCount > 0) { - $output->writeln("{$failedCount} email(s) are in failed state; inspect them in System > Tools > Message Queue or with ./maho queue:list."); - } + $failedCount = $this->getFailedCount(); + if ($failedCount > 0) { + $output->writeln("{$failedCount} email(s) are in failed state; inspect them in System > Tools > Message Queue or with ./maho queue:list."); } return Command::SUCCESS; diff --git a/lib/MahoCLI/Commands/QueueList.php b/lib/MahoCLI/Commands/QueueList.php index 34671f2f07..30df84a59a 100644 --- a/lib/MahoCLI/Commands/QueueList.php +++ b/lib/MahoCLI/Commands/QueueList.php @@ -11,6 +11,7 @@ use Mage; use Maho\Db\Expr; +use Maho\Queue\PoolRegistry; use Maho\Queue\QueueManager; use Maho\Queue\Transport\DbTransport; use Symfony\Component\Console\Attribute\AsCommand; @@ -21,7 +22,7 @@ #[AsCommand( name: 'queue:list', - description: 'Show per-queue message counts and the active transport', + description: 'Show per-queue message counts and the worker pool each queue is consumed by', )] class QueueList extends BaseMahoCommand { @@ -56,22 +57,20 @@ protected function execute(InputInterface $input, OutputInterface $output): int } ksort($queues); - $transportName = QueueManager::transportName(); - $output->writeln("Active transport: {$transportName}"); - if ($transportName === QueueManager::TRANSPORT_REDIS) { - $output->writeln('Pending messages live in Redis; the counts below only cover messages stored in the database (failures).'); - } - if ($queues === []) { $output->writeln('The queue is empty.'); return Command::SUCCESS; } $table = new Table($output); - $table->setHeaders(['Queue', 'Pending', 'Processing', 'Failed', 'Completed', 'Oldest pending (UTC)']); + $table->setHeaders(['Queue', 'Pool', 'Pending', 'Processing', 'Failed', 'Completed', 'Oldest pending (UTC)']); + $orphaned = false; foreach ($queues as $queue => $counts) { + $pool = PoolRegistry::poolFor((string) $queue); + $orphaned = $orphaned || $pool === null; $table->addRow([ $queue, + $pool === null ? 'none' : $pool->name, $counts[DbTransport::STATUS_PENDING], $counts[DbTransport::STATUS_PROCESSING], $counts[DbTransport::STATUS_FAILED], @@ -81,6 +80,10 @@ protected function execute(InputInterface $input, OutputInterface $output): int } $table->render(); + if ($orphaned) { + $output->writeln('Queues with no pool are never consumed; mark a pool catch_all under global/queue/pools or route them under global/queue/routing.'); + } + return Command::SUCCESS; } } diff --git a/lib/MahoCLI/Commands/QueueWork.php b/lib/MahoCLI/Commands/QueueWork.php index 695cf3b109..bbda38ac81 100644 --- a/lib/MahoCLI/Commands/QueueWork.php +++ b/lib/MahoCLI/Commands/QueueWork.php @@ -9,7 +9,9 @@ namespace MahoCLI\Commands; -use Maho\Queue\QueueManager; +use Maho\Queue\Pool; +use Maho\Queue\PoolRegistry; +use Maho\Queue\Transport\DbTransport; use Maho\Queue\WorkerFactory; use Symfony\Component\Console\Attribute\AsCommand; use Symfony\Component\Console\Command\Command; @@ -32,13 +34,17 @@ class QueueWork extends BaseMahoCommand implements SignalableCommandInterface protected function configure(): void { $this + ->addOption('pool', null, InputOption::VALUE_REQUIRED, 'Consume as this configured worker pool, taking its queues and limits as defaults') + ->addOption('index', null, InputOption::VALUE_REQUIRED, 'Which worker of the pool this process is, when the pool runs more than one', '0') ->addOption('queue', null, InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, 'Only consume these queues (repeatable); default all') + ->addOption('exclude-queue', null, InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY, 'Never consume these queues (repeatable); how a catch-all worker leaves another pool its own') ->addOption('limit', null, InputOption::VALUE_REQUIRED, 'Stop after handling this many messages') ->addOption('time-limit', null, InputOption::VALUE_REQUIRED, 'Stop after this many seconds') ->addOption('memory-limit', null, InputOption::VALUE_REQUIRED, 'Stop once memory usage exceeds this limit (e.g. 256M)') ->addOption('sleep', null, InputOption::VALUE_REQUIRED, 'Seconds to sleep when the queue is empty', '1') - ->addOption('stop-when-empty', null, InputOption::VALUE_NONE, 'Stop as soon as the queue is empty') - ->addOption('exclusive', null, InputOption::VALUE_NONE, 'Hold the queue.worker lock and refuse to run when another exclusive worker is active (used by the cron watchdog)'); + ->addOption('idle-timeout', null, InputOption::VALUE_REQUIRED, 'Stop after this many seconds with nothing to do; 0 stops on the first empty poll') + ->addOption('stop-when-empty', null, InputOption::VALUE_NONE, 'Stop as soon as the queue is empty (same as --idle-timeout=0)') + ->addOption('exclusive', null, InputOption::VALUE_NONE, 'Hold the pool worker lock and refuse to run when another exclusive worker holds it (used by the cron watchdog)'); } #[\Override] @@ -46,19 +52,75 @@ protected function execute(InputInterface $input, OutputInterface $output): int { $this->initMaho(); - if ($input->getOption('exclusive') - && !\Mage::getSingleton('core/lock')->acquire(\Maho_Queue_Model_Cron::WORKER_LOCK, machineLocal: true) - ) { - $output->writeln('Another exclusive queue worker is already running'); + $pool = null; + $poolName = $input->getOption('pool'); + if ($poolName !== null) { + $pool = PoolRegistry::get((string) $poolName); + if ($pool === null) { + $output->writeln("Unknown queue pool: {$poolName}"); + return Command::INVALID; + } + } + + $index = (int) $input->getOption('index'); + if ($pool !== null && ($index < 0 || $index >= $pool->count)) { + // Out of range takes a lock the watchdog never probes, so it would + // spawn a duplicate worker for the index this one is impersonating. + $output->writeln("Pool {$pool->name} runs {$pool->count} worker(s); --index must be 0.." . ($pool->count - 1) . ''); return Command::INVALID; } + if ($input->getOption('exclusive')) { + // The exclusive lock tells the watchdog this process covers the whole + // roster (or a whole pool); a queue filter would make that a lie and + // silently starve every queue the filter leaves out. + if ($input->getOption('queue') !== [] || $input->getOption('exclude-queue') !== []) { + $output->writeln('--exclusive cannot be combined with --queue or --exclude-queue: the lock claims coverage the filter takes away'); + return Command::INVALID; + } + $lockName = $pool?->lockName($index) ?? Pool::LOCK_PREFIX; + if (!\Mage::getSingleton('core/lock')->acquire($lockName, machineLocal: true)) { + $output->writeln("Another exclusive queue worker already holds {$lockName}"); + return Command::INVALID; + } + if ($pool === null && ($running = $this->livePoolWorkers()) !== []) { + // The bare lock only stops the watchdog from spawning more; it + // does not evict workers already running. + $output->writeln(sprintf( + 'Pool worker(s) %s are still running and keep consuming until their own limits stop them', + implode(', ', $running), + )); + } + } + + // Unbounded unless asked: a hand-run worker keeps the limits it had before pools existed. + $base = $pool ?? new Pool(name: 'ad-hoc', memoryLimit: '', timeLimit: 0); + $queues = $input->getOption('queue'); + $effective = new Pool( + name: $base->name, + queues: $queues ?: $base->queues, + // An explicit allow-list already narrows the worker, and keeping the + // pool's "everything but" list on top of it would leave + // `--pool=slow --queue=email` consuming nothing at all. Without an + // allow-list the pool's own exclusions stay: they are the catch-all's + // isolation boundary, and an extra --exclude-queue must not erase it. + excludedQueues: $queues + ? $input->getOption('exclude-queue') + : array_values(array_unique(array_merge($base->excludedQueues, $input->getOption('exclude-queue')))), + idleTimeout: match (true) { + $input->getOption('idle-timeout') !== null => max(0, (int) $input->getOption('idle-timeout')), + (bool) $input->getOption('stop-when-empty') => 0, + default => $base->idleTimeout, + }, + memoryLimit: (string) ($input->getOption('memory-limit') ?? $base->memoryLimit), + timeLimit: (int) ($input->getOption('time-limit') ?? $base->timeLimit), + ); + $memoryLimit = null; - $memoryLimitOption = $input->getOption('memory-limit'); - if ($memoryLimitOption !== null) { - $memoryLimit = $this->parseMemoryLimit((string) $memoryLimitOption); + if ($effective->memoryLimit !== '') { + $memoryLimit = Pool::parseMemoryLimit($effective->memoryLimit); if ($memoryLimit === null) { - $output->writeln("Invalid memory limit: {$memoryLimitOption}"); + $output->writeln("Invalid memory limit: {$effective->memoryLimit}"); return Command::INVALID; } } @@ -66,25 +128,48 @@ protected function execute(InputInterface $input, OutputInterface $output): int $this->worker = WorkerFactory::create([ 'limit' => $input->getOption('limit') !== null ? (int) $input->getOption('limit') : null, 'memoryLimit' => $memoryLimit, - 'stopWhenIdle' => (bool) $input->getOption('stop-when-empty'), + 'idleTimeout' => $effective->idleTimeout, + 'pool' => $effective, ]); - $queues = $input->getOption('queue'); $output->writeln(sprintf( - 'Consuming messages from the %s transport%s (press Ctrl-C to stop gracefully)', - QueueManager::transportName(), - $queues !== [] ? ', queues: ' . implode(', ', $queues) : '', + 'Consuming messages%s%s%s (press Ctrl-C to stop gracefully)', + $pool !== null ? ', pool: ' . $pool->name : '', + $effective->queues !== [] ? ', queues: ' . implode(', ', $effective->queues) : '', + $effective->excludedQueues !== [] ? ', excluding: ' . implode(', ', $effective->excludedQueues) : '', )); $options = ['sleep' => (int) $input->getOption('sleep') * 1_000_000]; - if ($input->getOption('time-limit') !== null) { - $options['time_limit'] = (int) $input->getOption('time-limit'); + if ($effective->timeLimit > 0) { + $options['time_limit'] = $effective->timeLimit; + } + if ($effective->queues !== []) { + $options['queues'] = $effective->queues; } - if ($queues !== []) { - $options['queues'] = $queues; + + // Drives keepalive(): without it a slow handler is indistinguishable from a dead worker. + if (SignalRegistry::isSupported()) { + $this->getApplication()?->setAlarmInterval(DbTransport::KEEPALIVE_INTERVAL_SECONDS); + } else { + $output->writeln(sprintf( + 'pcntl is unavailable, so claims are not refreshed while a handler runs: a handler longer than %d seconds is reported as abandoned even though its worker is alive', + DbTransport::ABANDONED_AFTER_SECONDS, + )); } - $this->worker->run($options); + try { + $this->worker->run($options); + } finally { + // The recurring alarm outlives run(): the command's return pops the + // SIGALRM handler back to SIG_DFL while an alarm is still pending, + // and SIG_DFL terminates the process. Harmless for a detached worker + // about to exit, fatal for a parent command (email:queue:process) + // that invoked queue:work in-process and still has work to do. + if (SignalRegistry::isSupported()) { + $this->getApplication()?->setAlarmInterval(null); + pcntl_alarm(0); + } + } return Command::SUCCESS; } @@ -95,29 +180,44 @@ protected function execute(InputInterface $input, OutputInterface $output): int #[\Override] public function getSubscribedSignals(): array { - return SignalRegistry::isSupported() ? [\SIGTERM, \SIGINT] : []; + return SignalRegistry::isSupported() ? [\SIGTERM, \SIGINT, \SIGALRM] : []; } #[\Override] public function handleSignal(int $signal, int|false $previousExitCode = 0): int|false { + if ($signal === \SIGALRM) { + try { + $this->worker?->keepalive($this->getApplication()?->getAlarmInterval()); + } catch (\Exception $e) { + // The alarm lands mid-handler, so an escaping error would unwind it half-done. + \Mage::log('Queue worker could not refresh its claim: ' . $e->getMessage(), \Mage::LOG_WARNING); + } + + return false; + } + // Finish the in-flight message, then exit cleanly. $this->worker?->stop(); return false; } - private function parseMemoryLimit(string $limit): ?int + /** + * @return list + */ + private function livePoolWorkers(): array { - if (!preg_match('/^(\d+)([KMG]?)$/i', trim($limit), $matches)) { - return null; + $lock = \Mage::getSingleton('core/lock'); + $running = []; + foreach (PoolRegistry::all() as $pool) { + for ($index = 0; $index < $pool->count; $index++) { + if ($lock->isHeld($pool->lockName($index), machineLocal: true)) { + $running[] = "{$pool->name}.{$index}"; + } + } } - return (int) $matches[1] * match (strtoupper($matches[2])) { - 'K' => 1024, - 'M' => 1024 ** 2, - 'G' => 1024 ** 3, - default => 1, - }; + return $running; } } diff --git a/tests/Backend/Integration/Queue/CronConsumerTest.php b/tests/Backend/Integration/Queue/CronConsumerTest.php index 61f44cf76b..1ef9a653c0 100644 --- a/tests/Backend/Integration/Queue/CronConsumerTest.php +++ b/tests/Backend/Integration/Queue/CronConsumerTest.php @@ -7,11 +7,47 @@ declare(strict_types=1); +use Maho\Queue\Pool; +use Maho\Queue\PoolRegistry; use Maho\Queue\QueueManager; use Maho\Queue\Transport\DbTransport; uses(Tests\MahoBackendTestCase::class); +/** + * @return list "pool:index" for each worker the watchdog would start + */ +function pendingWorkers(): array +{ + return array_map( + fn(array $worker) => $worker[0]->name . ':' . $worker[1], + Mage::getModel('queue/cron')->workersToSpawn(), + ); +} + +/** + * @param callable():void $body + */ +function withAllPoolLocks(callable $body): void +{ + $lock = Mage::getSingleton('core/lock'); + $held = []; + foreach (PoolRegistry::all() as $pool) { + for ($index = 0; $index < $pool->count; $index++) { + expect($lock->acquire($pool->lockName($index), machineLocal: true))->toBeTrue(); + $held[] = $pool->lockName($index); + } + } + + try { + $body(); + } finally { + foreach ($held as $name) { + $lock->release($name); + } + } +} + beforeEach(function () { QueueManager::reset(); clearQueueTable(); @@ -22,21 +58,89 @@ QueueManager::reset(); }); -it('does not spawn a second worker while the lock is held', function () { +it('does not spawn a second worker while the pool lock is held', function () { QueueManager::dispatch(makeEmailMessage()); - $lock = Mage::getSingleton('core/lock'); - expect($lock->acquire(Maho_Queue_Model_Cron::WORKER_LOCK, machineLocal: true))->toBeTrue(); - try { + withAllPoolLocks(function () { + expect(pendingWorkers())->toBe([]); + Mage::getModel('queue/cron')->process(); $rows = fetchQueueRows(); expect($rows)->toHaveCount(1); expect($rows[0]['status'])->toBe(DbTransport::STATUS_PENDING); + }); +}); + +it('keeps the resident tier running even with nothing queued', function () { + expect(pendingWorkers())->toBe(['fast:0']); +}); + +it('starts the on-demand tier only once its queues have work due', function () { + QueueManager::dispatch( + makeEmailMessage(), + delaySeconds: 7 * 86400, + queue: 'newsletter', + ); + expect(pendingWorkers())->toBe(['fast:0']); + + QueueManager::dispatch(makeEmailMessage('due now'), queue: 'newsletter'); + expect(pendingWorkers())->toBe(['fast:0', 'slow:0']); +}); + +it('stands aside for a hand-run exclusive worker', function () { + QueueManager::dispatch(makeEmailMessage()); + + $lock = Mage::getSingleton('core/lock'); + expect($lock->acquire(Pool::LOCK_PREFIX, machineLocal: true))->toBeTrue(); + try { + expect(pendingWorkers())->toBe([]); } finally { - $lock->release(Maho_Queue_Model_Cron::WORKER_LOCK); + $lock->release(Pool::LOCK_PREFIX); } }); +it('counts the on-demand workers already alive against the due budget', function () { + withQueueConfig('3', function () { + $lock = Mage::getSingleton('core/lock'); + $slow = PoolRegistry::get('slow'); + expect($slow?->count)->toBe(3); + $held = $slow->lockName(0); + expect($lock->acquire($held, machineLocal: true))->toBeTrue(); + + try { + QueueManager::dispatch(makeEmailMessage('due now'), queue: 'newsletter'); + + // One message, one worker already consuming it: a second process would + // boot only to idle straight back out. + expect(pendingWorkers())->toBe(['fast:0']); + } finally { + $lock->release($held); + } + }); +}); + +it('does not count a worker inside a long handler as free to take the next message', function () { + withQueueConfig('3', function () { + $lock = Mage::getSingleton('core/lock'); + $slow = PoolRegistry::get('slow'); + expect($slow?->count)->toBe(3); + $held = $slow->lockName(0); + expect($lock->acquire($held, machineLocal: true))->toBeTrue(); + + try { + QueueManager::dispatch(makeEmailMessage('long feed build'), queue: 'feed'); + QueueManager::dispatch(makeEmailMessage('due now'), queue: 'newsletter'); + + // The live worker is busy on the feed build, so it cannot take the + // newsletter: a free slot must be started rather than left idle behind it. + expect([...QueueManager::workerTransport($slow)->get()])->toHaveCount(1); + expect(pendingWorkers())->toBe(['fast:0', 'slow:1']); + } finally { + $lock->release($held); + } + }); +}); + it('removes old failed messages during cleanup', function () { $now = Mage_Core_Model_Locale::nowUtc(); queueAdapter()->insert(QueueManager::tableName(), [ @@ -45,7 +149,7 @@ 'message_class' => Mage_Core_Model_Email_SendMessage::class, 'body' => serialize(makeEmailMessage()), 'available_at' => $now, - 'processed_at' => gmdate(Mage_Core_Model_Locale::DATETIME_FORMAT, time() - 40 * 86400), + 'processed_at' => agoUtc(40 * 86400), 'created_at' => $now, 'updated_at' => $now, ]); diff --git a/tests/Backend/Integration/Queue/DbTransportTest.php b/tests/Backend/Integration/Queue/DbTransportTest.php index 45a9f596b2..857bdfd966 100644 --- a/tests/Backend/Integration/Queue/DbTransportTest.php +++ b/tests/Backend/Integration/Queue/DbTransportTest.php @@ -97,11 +97,28 @@ expect(fetchQueueRows())->toHaveCount(2); }); +it('lets a dedupe key be dispatched again once its claim is abandoned', function () { + QueueManager::dispatch(makeEmailMessage(), dedupeKey: 'abc'); + $transport = QueueManager::dbTransport(); + expect([...$transport->getFromQueues(['default'])])->toHaveCount(1); + + QueueManager::dispatch(makeEmailMessage(), dedupeKey: 'abc'); + expect(fetchQueueRows())->toHaveCount(1); + + // A claim nobody will finish must stop suppressing later sends, or every + // future dispatch of this key is silently dropped instead of one being parked. + queueAdapter()->update(QueueManager::tableName(), [ + 'claimed_at' => agoUtc(7200), + ]); + QueueManager::dispatch(makeEmailMessage(), dedupeKey: 'abc'); + expect(fetchQueueRows())->toHaveCount(2); +}); + it('inserts a failure-transport send as a failed row instead of updating by the foreign message id', function () { $envelope = (new Envelope(makeEmailMessage()))->with( new TransportMessageIdStamp('1712345678901-0'), new RedeliveryStamp(0), - new SentToFailureTransportStamp('redis'), + new SentToFailureTransportStamp('origin'), ErrorDetailsStamp::create(new RuntimeException('handler blew up')), ); @@ -113,33 +130,51 @@ expect($rows[0]['error_message'])->toContain('handler blew up'); }); -it('retries only failed stored messages, refusing rows a worker holds', function () { +it('retries a failed row and an abandoned claim, but not a pending or freshly claimed one', function () { QueueManager::dispatch(makeEmailMessage()); $transport = QueueManager::dbTransport(); - $envelopes = [...$transport->get()]; $id = (int) fetchQueueRows()[0]['message_id']; + // Pending needs no help. expect(QueueManager::retryStoredMessage($id))->toBeFalse(); + + // A live worker is still inside this one: re-queueing it runs the handler twice. + expect([...$transport->get()])->toHaveCount(1); expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_PROCESSING); + expect(QueueManager::retryStoredMessage($id))->toBeFalse(); - $transport->reject($envelopes[0]); - expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_FAILED); + // Old enough to belong to a dead worker: nothing else requeues it, so the grid must. + queueAdapter()->update(QueueManager::tableName(), [ + 'claimed_at' => agoUtc(7200), + ]); + expect(QueueManager::retryStoredMessage($id))->toBeTrue(); + expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_PENDING); + // The row's owner (its re-claimer, not the stale envelope) can still fail it. + $reclaimed = [...$transport->get()]; + $transport->reject($reclaimed[0]); + expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_FAILED); expect(QueueManager::retryStoredMessage($id))->toBeTrue(); expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_PENDING); }); -it('re-queues stale processing claims after redeliver_after', function () { +it('never hands out a claim again on its own, however old it is', function () { QueueManager::dispatch(makeEmailMessage()); $transport = QueueManager::dbTransport(); expect([...$transport->get()])->toHaveCount(1); queueAdapter()->update(QueueManager::tableName(), [ - 'claimed_at' => gmdate(Mage_Core_Model_Locale::DATETIME_FORMAT, time() - 7200), + 'claimed_at' => agoUtc(7200), ]); - $envelopes = [...$transport->get()]; - expect($envelopes)->toHaveCount(1); + // A claim is parked for an operator, never redelivered on a timer: running + // a handler a second time is not something a clock gets to decide. + expect([...$transport->get()])->toHaveCount(0); + expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_PROCESSING); + + // It is still reported, so the admin notice can point an operator at it. + expect($transport->countAbandoned(3600))->toBe(1); + expect($transport->countAbandoned(4 * 3600))->toBe(0); }); it('fails a claimed row whose message class has no registered handler', function () { @@ -161,6 +196,164 @@ expect($rows[0]['error_message'])->toContain('Nonexistent_Evil_Class'); }); +it('keeps a live claim out of the abandoned set', function () { + QueueManager::dispatch(makeEmailMessage()); + + $transport = QueueManager::dbTransport(); + $envelopes = [...$transport->get()]; + expect($envelopes)->toHaveCount(1); + + queueAdapter()->update(QueueManager::tableName(), [ + 'claimed_at' => agoUtc(DbTransport::ABANDONED_AFTER_SECONDS + 60), + ], ['message_id = ?' => (int) fetchQueueRows()[0]['message_id']]); + expect($transport->countAbandoned())->toBe(1); + + $transport->keepalive($envelopes[0]); + + expect($transport->countAbandoned())->toBe(0); + expect($transport->countClaimed())->toBe(1); +}); + +it('does not revive a claim that is no longer processing', function () { + QueueManager::dispatch(makeEmailMessage()); + + $transport = QueueManager::dbTransport(); + $envelopes = [...$transport->get()]; + queueAdapter()->update(QueueManager::tableName(), [ + 'status' => DbTransport::STATUS_PENDING, + 'claimed_at' => null, + ], ['message_id = ?' => (int) fetchQueueRows()[0]['message_id']]); + + $transport->keepalive($envelopes[0]); + + $row = fetchQueueRows()[0]; + expect($row['status'])->toBe(DbTransport::STATUS_PENDING); + expect($row['claimed_at'])->toBeNull(); +}); + +it('does not let a late ack swallow a row an operator already retried', function () { + QueueManager::dispatch(makeEmailMessage()); + + $transport = QueueManager::dbTransport(); + $envelopes = [...$transport->get()]; + $id = (int) fetchQueueRows()[0]['message_id']; + + // The claim goes stale (the worker looks dead), so the operator re-queues it. + queueAdapter()->update(QueueManager::tableName(), [ + 'claimed_at' => agoUtc(7200), + ], ['message_id = ?' => $id]); + expect(QueueManager::retryStoredMessage($id))->toBeTrue(); + + // The worker was alive after all and finishes: its ack must not delete or + // complete the pending row, or the operator's retry silently vanishes. + $transport->ack($envelopes[0]); + + $rows = fetchQueueRows(); + expect($rows)->toHaveCount(1); + expect($rows[0]['status'])->toBe(DbTransport::STATUS_PENDING); +}); + +it('does not let a stale worker touch a row another worker has since claimed', function () { + QueueManager::dispatch(makeEmailMessage()); + + $transport = QueueManager::dbTransport(); + $staleEnvelopes = [...$transport->get()]; + $id = (int) fetchQueueRows()[0]['message_id']; + + // The claim goes stale (worker A looks dead), the operator re-queues it, + // and worker B claims the row again. + queueAdapter()->update(QueueManager::tableName(), ['claimed_at' => agoUtc(7200)], ['message_id = ?' => $id]); + expect(QueueManager::retryStoredMessage($id))->toBeTrue(); + $freshEnvelopes = [...$transport->get()]; + expect($freshEnvelopes)->toHaveCount(1); + + // Worker A was alive after all: its claim token no longer matches, so its + // late ack, reject and re-send must all leave B's claim alone. + $transport->ack($staleEnvelopes[0]); + expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_PROCESSING); + $transport->reject($staleEnvelopes[0]); + expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_PROCESSING); + $transport->send($staleEnvelopes[0]->with( + new RedeliveryStamp(1), + ErrorDetailsStamp::create(new RuntimeException('handler blew up')), + )); + expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_PROCESSING); + + // B still owns the row and its ack lands. + $transport->ack($freshEnvelopes[0]); + expect(fetchQueueRows())->toHaveCount(0); +}); + +it('refuses to retry a parked claim while a newer copy of its dedupe key is in flight', function () { + QueueManager::dispatch(makeEmailMessage(), dedupeKey: 'abc'); + $transport = QueueManager::dbTransport(); + expect([...$transport->get()])->toHaveCount(1); + $id = (int) fetchQueueRows()[0]['message_id']; + + // The claim is abandoned, so a later dispatch of the same key inserts a + // fresh copy that supersedes the parked one. + queueAdapter()->update(QueueManager::tableName(), ['claimed_at' => agoUtc(7200)], ['message_id = ?' => $id]); + QueueManager::dispatch(makeEmailMessage(), dedupeKey: 'abc'); + expect(fetchQueueRows())->toHaveCount(2); + + // Retrying the parked claim would run the deduped job twice; discard is the way out. + expect(QueueManager::retryStoredMessage($id))->toBeFalse(); + + // Once the fresh copy is gone, the parked claim is retryable again. + $freshId = (int) fetchQueueRows()[1]['message_id']; + expect(QueueManager::discardStoredMessage($freshId))->toBeTrue(); + expect(QueueManager::retryStoredMessage($id))->toBeTrue(); + expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_PENDING); +}); + +it('does not let a late retry re-send overwrite a row an operator already retried', function () { + QueueManager::dispatch(makeEmailMessage()); + + $transport = QueueManager::dbTransport(); + $envelopes = [...$transport->get()]; + $id = (int) fetchQueueRows()[0]['message_id']; + + // The claim goes stale (the worker looks dead), so the operator re-queues it. + queueAdapter()->update(QueueManager::tableName(), [ + 'claimed_at' => agoUtc(7200), + ], ['message_id = ?' => $id]); + expect(QueueManager::retryStoredMessage($id))->toBeTrue(); + + // The worker was alive after all and its handler fails: the retry listener's + // in-place re-send must leave the operator's fresh pending row alone. + $transport->send($envelopes[0]->with( + new RedeliveryStamp(1), + ErrorDetailsStamp::create(new RuntimeException('handler blew up')), + )); + + $rows = fetchQueueRows(); + expect($rows)->toHaveCount(1); + expect($rows[0]['status'])->toBe(DbTransport::STATUS_PENDING); + expect((int) $rows[0]['retries'])->toBe(0); +}); + +it('skips the keepalive refresh while the shared connection is inside a transaction', function () { + QueueManager::dispatch(makeEmailMessage()); + + $transport = QueueManager::dbTransport(); + $envelopes = [...$transport->get()]; + $id = (int) fetchQueueRows()[0]['message_id']; + + $stale = agoUtc(7200); + queueAdapter()->update(QueueManager::tableName(), ['claimed_at' => $stale], ['message_id = ?' => $id]); + + queueAdapter()->beginTransaction(); + try { + $transport->keepalive($envelopes[0]); + expect(fetchQueueRows()[0]['claimed_at'])->toBe($stale); + } finally { + queueAdapter()->rollBack(); + } + + $transport->keepalive($envelopes[0]); + expect(fetchQueueRows()[0]['claimed_at'])->not->toBe($stale); +}); + it('counts pending messages and finds stored ones', function () { QueueManager::dispatch(makeEmailMessage()); QueueManager::dispatch(makeEmailMessage('second')); diff --git a/tests/Backend/Integration/Queue/PoolTest.php b/tests/Backend/Integration/Queue/PoolTest.php new file mode 100644 index 0000000000..43ca30e185 --- /dev/null +++ b/tests/Backend/Integration/Queue/PoolTest.php @@ -0,0 +1,201 @@ + + * SPDX-License-Identifier: OSL-3.0 + */ + +declare(strict_types=1); + +use Maho\Queue\Pool; +use Maho\Queue\PoolRegistry; +use Maho\Queue\QueueManager; +use Maho\Queue\StopWorkerWhenIdleListener; +use Maho\Queue\Transport\DbTransport; +use Symfony\Component\Messenger\Event\WorkerRunningEvent; +use Symfony\Component\Messenger\Worker; + +uses(Tests\MahoBackendTestCase::class); + +function queuePool(string $name): Pool +{ + return PoolRegistry::get($name) ?? throw new RuntimeException("pool {$name} is missing"); +} + +function poolTransport(string $name): DbTransport +{ + $transport = QueueManager::workerTransport(queuePool($name)); + assert($transport instanceof DbTransport); + + return $transport; +} + +function insertQueueRow(string $queue, string $status, ?string $claimedAt = null): void +{ + $now = Mage_Core_Model_Locale::nowUtc(); + queueAdapter()->insert(QueueManager::tableName(), [ + 'queue' => $queue, + 'status' => $status, + 'message_class' => Mage_Core_Model_Email_SendMessage::class, + 'body' => serialize(makeEmailMessage()), + 'available_at' => $now, + 'claimed_at' => $claimedAt, + 'created_at' => $now, + 'updated_at' => $now, + ]); +} + +/** + * A Worker that only records stop(), so the idle listener can be driven + * directly without a transport behind it. The parent constructor runs with + * empty receivers so inherited typed properties stay initialized. + */ +class RecordingWorker extends Worker +{ + public int $stopped = 0; + + public function __construct() + { + parent::__construct([], QueueManager::bus(), new Symfony\Component\EventDispatcher\EventDispatcher()); + } + + #[\Override] + public function stop(): void + { + $this->stopped++; + } +} + +beforeEach(function () { + QueueManager::reset(); + clearQueueTable(); +}); + +afterEach(function () { + clearQueueTable(); + QueueManager::reset(); +}); + +it('ships a resident fast tier and an on-demand slow catch-all', function () { + expect(array_keys(PoolRegistry::all()))->toBe(['fast', 'slow']); + expect(queuePool('fast')->queues)->toBe([Mage_Core_Model_Email_Queue::QUEUE_NAME]); + expect(queuePool('fast')->isOnDemand())->toBeFalse(); + expect(queuePool('slow')->queues)->toBe([]); + expect(queuePool('slow')->isOnDemand())->toBeTrue(); +}); + +it('keeps the catch-all off the queues another pool claims', function () { + expect(queuePool('slow')->excludedQueues)->toBe([Mage_Core_Model_Email_Queue::QUEUE_NAME]); + expect(queuePool('slow')->consumes(Mage_Core_Model_Email_Queue::QUEUE_NAME))->toBeFalse(); + expect(queuePool('slow')->consumes('newsletter'))->toBeTrue(); +}); + +it('routes an unclassified queue to the slow tier rather than leaving it unconsumed', function () { + expect(PoolRegistry::poolFor('some_third_party_queue')?->name)->toBe('slow'); + expect(PoolRegistry::poolFor(DbTransport::DEFAULT_QUEUE)?->name)->toBe('slow'); +}); + +it('lets another module route its own queue without clobbering the existing map', function () { + withQueueConfig('fast', function () { + expect(queuePool('fast')->queues)->toBe([Mage_Core_Model_Email_Queue::QUEUE_NAME, 'zz_scratch']); + expect(queuePool('slow')->excludedQueues)->toContain('zz_scratch'); + }); +}); + +it('hands a queue routed to an unknown pool back to the catch-all', function () { + withQueueConfig('nowhere', function () { + expect(PoolRegistry::poolFor('zz_scratch')?->name)->toBe('slow'); + }); +}); + +it('skips a pool with nothing routed to it instead of letting it rival the catch-all', function () { + withQueueConfig('5', function () { + expect(array_keys(PoolRegistry::all()))->toBe(['fast', 'slow']); + }); +}); + +it('reads pool flags with core config semantics, so "true" means on', function () { + withQueueConfig('zz_flagtrue5', function () { + expect(PoolRegistry::get('zz_flag'))->not->toBeNull(); + }); +}); + +it('drops a pool declared inactive with "false"', function () { + withQueueConfig('zz_flagfalse5', function () { + expect(PoolRegistry::get('zz_flag'))->toBeNull(); + expect(PoolRegistry::poolFor('zz_q')?->name)->toBe('slow'); + }); +}); + +it('keeps memory_limit 0 as unbounded instead of falling back to the default', function () { + withQueueConfig('zz_pool50', function () { + expect(queuePool('zz_pool')->memoryLimit)->toBe('0'); + }); +}); + +it('does not hand the catch-all worker a message belonging to another pool', function () { + QueueManager::dispatch(makeEmailMessage(), queue: Mage_Core_Model_Email_Queue::QUEUE_NAME); + QueueManager::dispatch(makeEmailMessage('newsletter batch'), queue: 'newsletter'); + + expect(iterator_to_array(poolTransport('slow')->get()))->toHaveCount(1); + + $processing = array_values(array_filter( + fetchQueueRows(), + fn($row) => $row['status'] === DbTransport::STATUS_PROCESSING, + )); + expect($processing)->toHaveCount(1); + expect($processing[0]['queue'])->toBe('newsletter'); +}); + +it('counts only work that is due, not a campaign scheduled for later', function () { + // The regression this guards: a campaign queued as a long-delayed message + // would make a watchdog probing raw pending counts respawn the on-demand + // worker every cron tick until its send date. + QueueManager::dispatch( + makeEmailMessage(), + delaySeconds: 7 * 86400, + queue: 'newsletter', + ); + + expect(poolTransport('slow')->getMessageCount())->toBe(1); + expect(poolTransport('slow')->countDue())->toBe(0); +}); + +it('never counts a claim as due, however old, so no worker is respawned for it', function () { + insertQueueRow('newsletter', DbTransport::STATUS_PROCESSING, agoUtc(4 * 3600)); + + // An abandoned claim waits for an operator, so the watchdog must not treat + // it as work: respawning for it would start a worker with nothing to do. + expect(poolTransport('slow')->countDue())->toBe(0); + expect(iterator_to_array(poolTransport('slow')->get()))->toHaveCount(0); + expect(fetchQueueRows()[0]['status'])->toBe(DbTransport::STATUS_PROCESSING); +}); + +it('stops an idle worker immediately when no grace period is set', function () { + $worker = new RecordingWorker(); + (new StopWorkerWhenIdleListener(0))->onWorkerRunning(new WorkerRunningEvent($worker, true)); + + expect($worker->stopped)->toBe(1); +}); + +it('holds an idle worker open for the grace period', function () { + $worker = new RecordingWorker(); + $listener = new StopWorkerWhenIdleListener(3600); + + $listener->onWorkerRunning(new WorkerRunningEvent($worker, true)); + $listener->onWorkerRunning(new WorkerRunningEvent($worker, true)); + + expect($worker->stopped)->toBe(0); +}); + +it('restarts the grace period when work arrives', function () { + $worker = new RecordingWorker(); + $listener = new StopWorkerWhenIdleListener(1); + + $listener->onWorkerRunning(new WorkerRunningEvent($worker, true)); + sleep(2); + $listener->onWorkerRunning(new WorkerRunningEvent($worker, false)); + $listener->onWorkerRunning(new WorkerRunningEvent($worker, true)); + + expect($worker->stopped)->toBe(0); +}); diff --git a/tests/Pest.php b/tests/Pest.php index a995b42933..1af950f439 100644 --- a/tests/Pest.php +++ b/tests/Pest.php @@ -471,3 +471,33 @@ function fetchQueueRows(): array queueAdapter()->select()->from(\Maho\Queue\QueueManager::tableName())->order('message_id ASC'), ); } + +function agoUtc(int $seconds): string +{ + return gmdate(Mage_Core_Model_Locale::DATETIME_FORMAT, time() - $seconds); +} + +/** + * Merge extra queue config the way another module's config.xml would, run the + * assertions, then restore the original config exactly (a snapshot, so merging + * into an existing node does not delete it on the way out). + * + * @param callable():void $body + */ +function withQueueConfig(string $xml, callable $body): void +{ + $node = Mage::getConfig()->getNode('global/queue'); + $snapshot = new Maho\Simplexml\Element($node->asXML()); + $node->extend(new Maho\Simplexml\Element($xml), true); + \Maho\Queue\QueueManager::reset(); + + try { + $body(); + } finally { + foreach (array_keys((array) $node->children()) as $child) { + unset($node->{$child}); + } + $node->extend($snapshot, true); + \Maho\Queue\QueueManager::reset(); + } +}