From 1e9087a5e5945a419cd4ee79da5cb27429f8d10c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pablo=20de=20la=20Pen=CC=83a?= Date: Wed, 21 Oct 2026 19:26:06 +0100 Subject: [PATCH] feat: Report queues that have no live worker The dashboard alert can ask the manager which queues are unworked, using the same heartbeat rule as before. --- README.md | 5 ++ src/Admin/Dashboard/Alert/NoWorkers.php | 38 +----------- src/Service/Manager.php | 77 +++++++++++++++++++++++++ tests/Service/ManagerTest.php | 77 +++++++++++++++++++++++++ tests/Stub/SideQueue.php | 17 ++++++ 5 files changed, 177 insertions(+), 37 deletions(-) create mode 100644 tests/Stub/SideQueue.php diff --git a/README.md b/README.md index 15cafe1..5dcf7d1 100644 --- a/README.md +++ b/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 diff --git a/src/Admin/Dashboard/Alert/NoWorkers.php b/src/Admin/Dashboard/Alert/NoWorkers.php index 2f71553..c81884a 100644 --- a/src/Admin/Dashboard/Alert/NoWorkers.php +++ b/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(); } } diff --git a/src/Service/Manager.php b/src/Service/Manager.php index adb356d..632b5f3 100644 --- a/src/Service/Manager.php +++ b/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 + * @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. * diff --git a/tests/Service/ManagerTest.php b/tests/Service/ManagerTest.php index 44b3025..30586eb 100644 --- a/tests/Service/ManagerTest.php +++ b/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)); + } } diff --git a/tests/Stub/SideQueue.php b/tests/Stub/SideQueue.php new file mode 100644 index 0000000..f2782dd --- /dev/null +++ b/tests/Stub/SideQueue.php @@ -0,0 +1,17 @@ +