Split queue workers into pools, parked crashed messages instead of redelivering them, and dropped the Redis transport

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
+  `<global><queue><routing><yourqueue>fast</yourqueue></routing></queue></global>`; 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]/`

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', [

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,
+        };
     }
 }

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<array{Pool, int}>
+     */
+    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);
     }
 }

app/code/core/Maho/Queue/Model/Observer.php

@@ -0,0 +1,72 @@
+<?php
+
+/**
+ * SPDX-FileCopyrightText: 2026 Maho <https://mahocommerce.com>
+ * 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: <a href="%s">retry or discard them</a>.',
+                $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;
+    }
+}

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,21 +113,30 @@ public function discardAction(): void
         } else {
             Mage::getSingleton('adminhtml/session')->addError(Mage::helper('queue')->__('Message not found.'));
         }
+        $this->refreshAbandonedNotice();
         $this->_redirect('*/*/');
     }
 
     #[Maho\Config\Route('/admin/queue/massRetry')]
     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<int>
      */

app/code/core/Maho/Queue/etc/config.xml

@@ -39,6 +39,29 @@
                 <class>Maho_Queue_Block</class>
             </queue>
         </blocks>
+
+        <!--
+            Worker pools, one detached "queue:work" process each, split by whether a human is
+            waiting: "fast" stays resident, "slow" is the catch-all and runs only while it has
+            work. Route your own queue to a tier by adding a <yourqueue>fast</yourqueue> node
+            under routing from your own config.xml; anything unrouted falls to the catch-all.
+        -->
+        <queue>
+            <routing>
+                <email>fast</email>
+            </routing>
+            <pools>
+                <fast>
+                    <sort_order>10</sort_order>
+                </fast>
+                <slow>
+                    <catch_all>1</catch_all>
+                    <idle_timeout>60</idle_timeout>
+                    <memory_limit>512M</memory_limit>
+                    <sort_order>20</sort_order>
+                </slow>
+            </pools>
+        </queue>
     </global>
 
     <adminhtml>
@@ -67,7 +90,6 @@
                 <retry_delay>60</retry_delay>
                 <retry_multiplier>4</retry_multiplier>
                 <retry_max_delay>21600</retry_max_delay>
-                <redeliver_after>3600</redeliver_after>
                 <completed_retention>0</completed_retention>
                 <failed_retention>30</failed_retention>
             </queue>