Report queues that have no live worker

README.md

@@ -152,11 +152,16 @@ Manager::unregisterWorker(Resource\Worker $worker): void
 Manager::deleteStaleWorkers(): Resource\Worker[]
 Manager::resetStuckJobs(): Job[]
 Manager::rotateOldJobs(): Job[]
+
+// Same liveness check as the admin dashboard alert
+Manager::unworkedQueues(): Queue[]
+Manager::isProcessed(Queue|string $queue): bool
 ```
 
 Notes:
 - Passing `null` or `'default'` as the queue argument uses the default queue alias; you can also pass a fully-qualified queue class name (FQCN) or an instance.
 - FIFO is approximated by selecting the smallest available `id` where `available_at <= NOW()`.
+- `unworkedQueues()` is the list shown by the admin dashboard alert. A worker counts when its heartbeat is newer than `QUEUE_WORKER_HEARTBEAT_STALE` (300 seconds by default) and it was started for that queue class. `isProcessed()` is true when a live worker lists the queue. For a discovered queue, that is the opposite of appearing in `unworkedQueues()`.
 
 
 ## Creating user-land queues and jobs

src/Admin/Dashboard/Alert/NoWorkers.php

@@ -8,8 +8,6 @@
 use Nails\Factory;
 use Nails\Queue\Constants;
 use Nails\Queue\Interface\Queue;
-use Nails\Queue\Model;
-use Nails\Queue\Resource;
 use Nails\Queue\Service;
 
 /**
@@ -86,41 +84,7 @@ private function getUnworkedQueues(): array
 
         /** @var Service\Manager $manager */
         $manager = Factory::service('Manager', Constants::MODULE_SLUG);
-        /** @var Model\Worker $workerModel */
-        $workerModel = Factory::model('Worker', Constants::MODULE_SLUG);
 
-        $queues = $manager->getQueues();
-        if (empty($queues)) {
-            //  No queues, no need to worry if there are workers
-            return static::$unworkedQueues = [];
-        }
-
-        /** @var Resource\Worker[] $workers */
-        $workers = $workerModel->getAll();
-        $workers = array_filter($workers, fn($worker) => !$worker->isStale());
-
-        if (empty($workers)) {
-            //  No active workers at all, all queues are unworked
-            return static::$unworkedQueues = $queues;
-        }
-
-        //  Check each queue has at least one worker
-        $map = [];
-        foreach ($queues as $queue) {
-            $map[$queue::class] = (object) ['queue' => $queue, 'workerCount' => 0];
-            foreach ($workers as $worker) {
-                foreach ($worker->queues as $workerQueue) {
-                    if ($workerQueue === $queue::class) {
-                        $map[$queue::class]->workerCount++;
-                    }
-                }
-            }
-        }
-
-        $map = array_filter($map, fn($map) => $map->workerCount === 0);
-        $map = array_map(fn($map) => $map->queue, $map);
-        $map = array_values($map);
-
-        return static::$unworkedQueues = $map;
+        return static::$unworkedQueues = $manager->unworkedQueues();
     }
 }

src/Service/Manager.php

@@ -118,6 +118,83 @@ public function getQueues(): array
         return $this->queues;
     }
 
+    /**
+     * Discovered queues with no live worker.
+     *
+     * This is the check behind the admin dashboard alert. No discovered queues
+     * means nothing to report. A worker counts when its heartbeat is still
+     * inside QUEUE_WORKER_HEARTBEAT_STALE and its registered queues include
+     * the queue class.
+     *
+     * @return Queue[]
+     * @throws FactoryException
+     * @throws ModelException
+     */
+    public function unworkedQueues(): array
+    {
+        $queues = $this->getQueues();
+        if ($queues === []) {
+            return [];
+        }
+
+        $covered = $this->coveredQueueClasses();
+        if ($covered === []) {
+            return $queues;
+        }
+
+        $unworked = [];
+
+        foreach ($queues as $queue) {
+            if (!isset($covered[$queue::class])) {
+                $unworked[$queue::class] = $queue;
+            }
+        }
+
+        return array_values($unworked);
+    }
+
+    /**
+     * Whether a live worker is registered for this queue.
+     *
+     * A worker counts when its heartbeat is still inside QUEUE_WORKER_HEARTBEAT_STALE
+     * and it was started for this queue class. A discovered queue is processed
+     * exactly when {@see unworkedQueues()} leaves it out.
+     *
+     * @throws FactoryException
+     * @throws InvalidArgumentException
+     * @throws ModelException
+     */
+    public function isProcessed(Queue|string $queue): bool
+    {
+        return isset($this->coveredQueueClasses()[$this->resolveQueue($queue)::class]);
+    }
+
+    /**
+     * Class names listed by a worker whose heartbeat is still fresh.
+     *
+     * @return array<string, true>
+     * @throws FactoryException
+     * @throws ModelException
+     */
+    private function coveredQueueClasses(): array
+    {
+        $covered = [];
+
+        foreach ($this->workerModel->getAll() as $worker) {
+            if (!$worker instanceof Resource\Worker || $worker->isStale()) {
+                continue;
+            }
+
+            foreach ($worker->queues as $workerQueue) {
+                if (is_string($workerQueue) && $workerQueue !== '') {
+                    $covered[$workerQueue] = true;
+                }
+            }
+        }
+
+        return $covered;
+    }
+
     /**
      * Registers a new worker to process the specified queues and returns the persisted worker resource.
      *

tests/Service/ManagerTest.php

@@ -25,6 +25,7 @@
 use PHPUnit\Framework\TestCase;
 use Tests\Queue\Stub\DatabaseStub;
 use Tests\Queue\Stub\ManagerStub;
+use Tests\Queue\Stub\SideQueue;
 
 /**
  * Tests the Queue Manager service behaviours.
@@ -1686,4 +1687,80 @@ public function first_row()
         // Assert
         self::assertSame(17, $value);
     }
+
+    /**
+     * A queue is unworked when no live worker lists its class. A stale heartbeat does not count.
+     *
+     * @covers \Nails\Queue\Service\Manager::unworkedQueues
+     * @covers \Nails\Queue\Service\Manager::isProcessed
+     */
+    public function test_unworked_queues_ignore_stale_workers(): void
+    {
+        /** @var WorkerModel&MockObject $workerModel */
+        $workerModel = $this->makeMock(WorkerModel::class, onlyMethods: ['getAll']);
+        $workerModel->expects($this->atLeastOnce())->method('getAll')->willReturn([
+            $this->makeWorkerResource([
+                'id'        => 1,
+                'queues'    => json_encode([DefaultQueue::class]),
+                'heartbeat' => '2099-01-01 00:00:00',
+            ]),
+            $this->makeWorkerResource([
+                'id'        => 2,
+                'queues'    => json_encode([PriorityQueue::class]),
+                'heartbeat' => '2000-01-01 00:00:00',
+            ]),
+        ]);
+
+        $manager  = $this->makeManager(workerModel: $workerModel);
+        $unworked = array_map(static fn($queue) => $queue::class, $manager->unworkedQueues());
+
+        self::assertNotContains(DefaultQueue::class, $unworked);
+        self::assertContains(PriorityQueue::class, $unworked);
+        self::assertTrue($manager->isProcessed('default'));
+        self::assertTrue($manager->isProcessed(DefaultQueue::class));
+        self::assertFalse($manager->isProcessed('priority'));
+        self::assertFalse($manager->isProcessed(PriorityQueue::class));
+    }
+
+    /**
+     * With no live workers, every discovered queue is unworked.
+     *
+     * @covers \Nails\Queue\Service\Manager::unworkedQueues
+     * @covers \Nails\Queue\Service\Manager::isProcessed
+     */
+    public function test_no_live_workers_marks_every_discovered_queue_unworked(): void
+    {
+        /** @var WorkerModel&MockObject $workerModel */
+        $workerModel = $this->makeMock(WorkerModel::class, onlyMethods: ['getAll']);
+        $workerModel->expects($this->atLeastOnce())->method('getAll')->willReturn([]);
+
+        $manager  = $this->makeManager(workerModel: $workerModel);
+        $unworked = array_map(static fn($queue) => $queue::class, $manager->unworkedQueues());
+
+        self::assertContains(DefaultQueue::class, $unworked);
+        self::assertContains(PriorityQueue::class, $unworked);
+        self::assertFalse($manager->isProcessed('default'));
+    }
+
+    /**
+     * A queue outside discovery is processed only when a live worker lists it.
+     *
+     * @covers \Nails\Queue\Service\Manager::isProcessed
+     */
+    public function test_undiscovered_queue_follows_the_live_worker(): void
+    {
+        /** @var WorkerModel&MockObject $workerModel */
+        $workerModel = $this->makeMock(WorkerModel::class, onlyMethods: ['getAll']);
+        $workerModel->expects($this->atLeastOnce())->method('getAll')->willReturn([
+            $this->makeWorkerResource([
+                'queues'    => json_encode([SideQueue::class]),
+                'heartbeat' => '2099-01-01 00:00:00',
+            ]),
+        ]);
+
+        $manager = $this->makeManager(workerModel: $workerModel);
+
+        self::assertTrue($manager->isProcessed(new SideQueue()));
+        self::assertFalse($manager->isProcessed(DefaultQueue::class));
+    }
 }

tests/Stub/SideQueue.php

@@ -0,0 +1,17 @@
+<?php
+
+namespace Tests\Queue\Stub;
+
+use Nails\Queue\Interface\Queue;
+use Nails\Queue\Resource\Worker;
+
+class SideQueue implements Queue
+{
+    public static function setup(Worker $worker): void
+    {
+    }
+
+    public static function refresh(Worker $worker): void
+    {
+    }
+}