Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
92.96%
66 / 71
93.75%
15 / 16
 CRAP
0.00%
0 / 1
WorkerRegistry
92.96%
66 / 71
93.75%
15 / 16
 31.34
0.00%
0 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 getRegistry
100.00%
1 / 1
100.00%
1 / 1
 1
 register
100.00%
4 / 4
100.00%
1 / 1
 1
 isRegistered
100.00%
1 / 1
100.00%
1 / 1
 1
 getRecord
100.00%
1 / 1
100.00%
1 / 1
 1
 heartbeat
100.00%
4 / 4
100.00%
1 / 1
 2
 deregister
100.00%
4 / 4
100.00%
1 / 1
 2
 getWorkers
100.00%
1 / 1
100.00%
1 / 1
 1
 getWorker
100.00%
1 / 1
100.00%
1 / 1
 1
 countWorkers
100.00%
1 / 1
100.00%
1 / 1
 1
 getStaleWorkers
100.00%
3 / 3
100.00%
1 / 1
 1
 getStuckWorkers
100.00%
3 / 3
100.00%
1 / 1
 1
 prune
100.00%
1 / 1
100.00%
1 / 1
 1
 attachTo
100.00%
2 / 2
100.00%
1 / 1
 2
 attachListeners
86.11%
31 / 36
0.00%
0 / 1
 10.27
 resolveEventManager
100.00%
7 / 7
100.00%
1 / 1
 4
1<?php
2declare(strict_types=1);
3/**
4 * Pop PHP Framework (https://www.popphp.org/)
5 *
6 * @link       https://github.com/popphp/popphp-framework
7 * @author     Nick Sagona, III <nick@popphp.org>
8 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
9 * @license    https://www.popphp.org/license     New BSD License
10 */
11
12/**
13 * @namespace
14 */
15namespace Pop\Queue\Registry;
16
17use Pop\Event\Manager as EventManager;
18use Pop\Queue\Queue;
19use Pop\Queue\Worker;
20
21/**
22 * Worker registry class
23 *
24 * The read-side facade over a registry backend: who is running, which of
25 * them have gone quiet, and which look genuinely stuck.
26 *
27 * @category   Pop
28 * @package    Pop\Queue
29 * @author     Nick Sagona, III <nick@popphp.org>
30 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
31 * @license    https://www.popphp.org/license     New BSD License
32 * @version    3.0.0
33 */
34class WorkerRegistry
35{
36
37    /**
38     * Registry backend
39     * @var RegistryInterface
40     */
41    protected RegistryInterface $registry;
42
43    /**
44     * This process's own record, once registered. A WorkerRegistry instance
45     * represents this process's view of the registry, which is what lets the
46     * event listeners attached by attachTo() mutate the record by closing
47     * over $this.
48     * @var ?WorkerRecord
49     */
50    protected ?WorkerRecord $record = null;
51
52    /**
53     * Event managers this registry has already attached listeners to, keyed
54     * by spl_object_id. resolveEventManager() returns the SAME manager for
55     * every queue whenever the Application-level fallback is in play, and
56     * Manager::on() is additive - without this guard, N queues would install
57     * N copies of each listener on one manager and a single job would count
58     * N times. It also makes repeated setRegistry()/attachTo() calls safe.
59     * @var array
60     */
61    protected array $attached = [];
62
63    /**
64     * Constructor
65     *
66     * @param RegistryInterface $registry
67     */
68    public function __construct(RegistryInterface $registry)
69    {
70        $this->registry = $registry;
71    }
72
73    /**
74     * Get the underlying backend
75     *
76     * @return RegistryInterface
77     */
78    public function getRegistry(): RegistryInterface
79    {
80        return $this->registry;
81    }
82
83    /**
84     * Register this process, writing its record to the backend
85     *
86     * Note: when a Worker is given this registry, it calls this for you at
87     * the right point in its lifecycle. Calling it directly opts you out of
88     * that - the Worker will see an existing registration, take no
89     * ownership, and never deregister it, so the record outlives the run and
90     * is only reaped by prune().
91     *
92     * @param  ?string $name       optional operator-facing label
93     * @param  array   $queueNames names of the queues being serviced
94     * @param  string  $mode       WorkerRecord::MODE_DAEMON or MODE_SINGLE_PASS
95     * @return WorkerRecord
96     */
97    public function register(?string $name = null, array $queueNames = [], string $mode = WorkerRecord::MODE_SINGLE_PASS): WorkerRecord
98    {
99        $record = WorkerRecord::create($name, $queueNames, $mode);
100
101        // Write BEFORE adopting the record. If the backend throws, this
102        // instance must remain genuinely unregistered - otherwise
103        // isRegistered() would report true for a record that was never
104        // persisted, and Worker::ensureRegistered()'s early-return would
105        // permanently prevent any retry for the life of the process.
106        $this->registry->write($record);
107        $this->record = $record;
108
109        return $this->record;
110    }
111
112    /**
113     * Whether this process has registered
114     *
115     * @return bool
116     */
117    public function isRegistered(): bool
118    {
119        return ($this->record !== null);
120    }
121
122    /**
123     * This process's own record, or null if it hasn't registered
124     *
125     * @return ?WorkerRecord
126     */
127    public function getRecord(): ?WorkerRecord
128    {
129        return $this->record;
130    }
131
132    /**
133     * Refresh this process's heartbeat and flush its record. A no-op when
134     * not registered, so callers never need to guard.
135     *
136     * @return void
137     */
138    public function heartbeat(): void
139    {
140        if ($this->record === null) {
141            return;
142        }
143
144        $this->record->touch();
145        $this->registry->write($this->record);
146    }
147
148    /**
149     * Remove this process's record. A no-op when not registered.
150     *
151     * @return void
152     */
153    public function deregister(): void
154    {
155        if ($this->record === null) {
156            return;
157        }
158
159        $this->registry->delete($this->record->getId());
160        $this->record = null;
161    }
162
163    /**
164     * Every registered worker, keyed by worker ID
165     *
166     * @return array
167     */
168    public function getWorkers(): array
169    {
170        return $this->registry->all();
171    }
172
173    /**
174     * A single worker by ID, or null
175     *
176     * @param  string $id
177     * @return ?WorkerRecord
178     */
179    public function getWorker(string $id): ?WorkerRecord
180    {
181        return $this->registry->read($id);
182    }
183
184    /**
185     * How many workers are registered
186     *
187     * @return int
188     */
189    public function countWorkers(): int
190    {
191        return count($this->registry->all());
192    }
193
194    /**
195     * Workers whose heartbeat has gone quiet. Note this includes workers
196     * that are merely busy inside a long job - a worker executing a job
197     * cannot heartbeat, so use getStuckWorkers() to narrow to the ones
198     * that look genuinely wedged.
199     *
200     * @param  int $seconds
201     * @return array
202     */
203    public function getStaleWorkers(int $seconds = 90): array
204    {
205        return array_filter($this->registry->all(), function($record) use ($seconds) {
206            return $record->isStale($seconds);
207        });
208    }
209
210    /**
211     * Workers that are stale AND holding a job that has outlived its own
212     * timeout - the alerting signal
213     *
214     * @param  int $seconds
215     * @return array
216     */
217    public function getStuckWorkers(int $seconds = 90): array
218    {
219        return array_filter($this->registry->all(), function($record) use ($seconds) {
220            return $record->isLikelyStuck($seconds);
221        });
222    }
223
224    /**
225     * Remove records left behind by processes that are long gone
226     *
227     * @param  int $olderThanSeconds
228     * @return int
229     */
230    public function prune(int $olderThanSeconds = 3600): int
231    {
232        return $this->registry->prune($olderThanSeconds);
233    }
234
235    /**
236     * Wire this registry's current-job and counter tracking onto a worker's
237     * queues, via the queue lifecycle events.
238     *
239     * Tracking rides on the existing events rather than new plumbing because
240     * Queue::work() reserves AND runs a job internally - the Worker only
241     * receives it after it ran, so it structurally cannot record the current
242     * job before execution. queue.job.pre fires before execution, which is
243     * exactly the hook needed.
244     *
245     * @param  Worker $worker
246     * @return void
247     */
248    public function attachTo(Worker $worker): void
249    {
250        foreach ($worker->getQueues() as $queue) {
251            $this->attachListeners($this->resolveEventManager($queue, $worker));
252        }
253    }
254
255    /**
256     * Attach this registry's listeners to one event manager, at most once
257     * per manager for the lifetime of this registry instance.
258     *
259     * @param  EventManager $events
260     * @return void
261     */
262    protected function attachListeners(EventManager $events): void
263    {
264        $id = spl_object_id($events);
265        if (isset($this->attached[$id])) {
266            return;
267        }
268        $this->attached[$id] = true;
269
270        // Listener params are positional, not an array - Manager::trigger()
271        // strips the keys before calling.
272        $events->on('queue.job.pre', function($job, $queue) {
273            if ($this->record !== null) {
274                $this->record->setCurrentJob($job->getJobId(), $queue->getName(), $job->getTimeout());
275                // Persisted BEFORE the job runs: a worker that wedges
276                // mid-job can't write anything afterwards, so this is the
277                // only chance to record what it died on.
278                try {
279                    $this->registry->write($this->record);
280                } catch (\Throwable $e) {
281                    // Best-effort - a failed write costs stuck-detection
282                    // fidelity for this job, never the job itself.
283                }
284            }
285        });
286
287        $events->on('queue.job.post', function($job, $queue) {
288            if ($this->record !== null) {
289                $this->record->clearCurrentJob();
290                $this->record->incrementProcessed();
291                // Deliberately no write - the cleared state and counters
292                // flush on the next heartbeat, keeping steady-state write
293                // volume flat regardless of job throughput.
294            }
295        });
296
297        $events->on('queue.job.failed', function($job, $queue, $exception) {
298            if ($this->record !== null) {
299                $this->record->clearCurrentJob();
300                $this->record->incrementFailed();
301            }
302        });
303
304        $events->on('queue.task.pre', function($task, $queue) {
305            if ($this->record !== null) {
306                $this->record->setCurrentJob($task->getJobId(), $queue->getName(), $task->getTimeout());
307                try {
308                    $this->registry->write($this->record);
309                } catch (\Throwable $e) {
310                    // Best-effort, as with jobs.
311                }
312            }
313        });
314
315        $events->on('queue.task.post', function($task, $queue) {
316            if ($this->record !== null) {
317                $this->record->clearCurrentJob();
318                $this->record->incrementProcessed();
319            }
320        });
321
322        $events->on('queue.task.failed', function($task, $queue, $exception) {
323            if ($this->record !== null) {
324                $this->record->clearCurrentJob();
325                $this->record->incrementFailed();
326            }
327        });
328    }
329
330    /**
331     * Find the event manager a queue's events actually reach, without
332     * disturbing existing wiring.
333     *
334     * Queue::triggerEvent() uses the queue's own manager if set, else the
335     * Application's - and setting a queue-level manager SUPPRESSES the
336     * Application fallback. So attaching to the wrong one, or installing a
337     * new one where a fallback was in play, would silently orphan a user's
338     * app-level listeners.
339     *
340     * @param  Queue  $queue
341     * @param  Worker $worker
342     * @return EventManager
343     */
344    protected function resolveEventManager(Queue $queue, Worker $worker): EventManager
345    {
346        if ($queue->hasEvents()) {
347            return $queue->events();
348        }
349
350        if ($worker->hasApplication() && ($worker->getApplication()->events() !== null)) {
351            return $worker->getApplication()->events();
352        }
353
354        $events = new EventManager();
355        $queue->setEvents($events);
356
357        return $events;
358    }
359
360}