Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
99.00%
99 / 100
96.67%
29 / 30
 CRAP
0.00%
0 / 1
WorkerRecord
99.00%
99 / 100
96.67%
29 / 30
 40
0.00%
0 / 1
 __construct
100.00%
5 / 5
100.00%
1 / 1
 1
 create
100.00%
11 / 11
100.00%
1 / 1
 3
 getId
100.00%
1 / 1
100.00%
1 / 1
 1
 setName
100.00%
2 / 2
100.00%
1 / 1
 1
 getName
100.00%
1 / 1
100.00%
1 / 1
 1
 getHost
100.00%
1 / 1
100.00%
1 / 1
 1
 getPid
100.00%
1 / 1
100.00%
1 / 1
 1
 getStartedAt
100.00%
1 / 1
100.00%
1 / 1
 1
 getLastSeenAt
100.00%
1 / 1
100.00%
1 / 1
 1
 setQueues
100.00%
2 / 2
100.00%
1 / 1
 1
 getQueues
100.00%
1 / 1
100.00%
1 / 1
 1
 setMode
100.00%
2 / 2
100.00%
1 / 1
 1
 getMode
100.00%
1 / 1
100.00%
1 / 1
 1
 touch
100.00%
2 / 2
100.00%
1 / 1
 1
 setCurrentJob
100.00%
5 / 5
100.00%
1 / 1
 1
 setCurrentJobStartedAt
100.00%
2 / 2
100.00%
1 / 1
 1
 clearCurrentJob
100.00%
5 / 5
100.00%
1 / 1
 1
 getCurrentJobId
100.00%
1 / 1
100.00%
1 / 1
 1
 getCurrentQueue
100.00%
1 / 1
100.00%
1 / 1
 1
 getCurrentJobStartedAt
100.00%
1 / 1
100.00%
1 / 1
 1
 getCurrentJobTimeout
100.00%
1 / 1
100.00%
1 / 1
 1
 incrementProcessed
100.00%
2 / 2
100.00%
1 / 1
 1
 incrementFailed
100.00%
2 / 2
100.00%
1 / 1
 1
 getJobsProcessed
100.00%
1 / 1
100.00%
1 / 1
 1
 getJobsFailed
100.00%
1 / 1
100.00%
1 / 1
 1
 isStale
100.00%
1 / 1
100.00%
1 / 1
 1
 getCurrentJobDuration
100.00%
1 / 1
100.00%
1 / 1
 2
 isLikelyStuck
87.50%
7 / 8
0.00%
0 / 1
 5.05
 toArray
100.00%
16 / 16
100.00%
1 / 1
 1
 fromArray
100.00%
20 / 20
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
17/**
18 * Worker record class
19 *
20 * A point-in-time snapshot of one worker process: who it is, what it is
21 * servicing, what it is working on right now, and when it was last heard
22 * from. Deliberately plain scalars and arrays only, so it serializes as
23 * JSON and the registry carries no unserialize() object-injection surface.
24 *
25 * @category   Pop
26 * @package    Pop\Queue
27 * @author     Nick Sagona, III <nick@popphp.org>
28 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
29 * @license    https://www.popphp.org/license     New BSD License
30 * @version    3.0.0
31 */
32class WorkerRecord
33{
34
35    /**
36     * Worker modes
37     */
38    const MODE_DAEMON      = 'daemon';
39    const MODE_SINGLE_PASS = 'single-pass';
40
41    /**
42     * Unique worker instance ID
43     * @var string
44     */
45    protected string $id;
46
47    /**
48     * Optional operator-facing label
49     * @var ?string
50     */
51    protected ?string $name = null;
52
53    /**
54     * Hostname
55     * @var string
56     */
57    protected string $host;
58
59    /**
60     * Process ID
61     * @var int
62     */
63    protected int $pid;
64
65    /**
66     * When this worker registered
67     * @var int
68     */
69    protected int $startedAt;
70
71    /**
72     * Last heartbeat
73     * @var int
74     */
75    protected int $lastSeenAt;
76
77    /**
78     * Names of the queues being serviced
79     * @var array
80     */
81    protected array $queues = [];
82
83    /**
84     * Worker mode - MODE_DAEMON or MODE_SINGLE_PASS
85     * @var string
86     */
87    protected string $mode = self::MODE_SINGLE_PASS;
88
89    /**
90     * ID of the job currently being executed, if any
91     * @var ?string
92     */
93    protected ?string $currentJobId = null;
94
95    /**
96     * Name of the queue the current job came from
97     * @var ?string
98     */
99    protected ?string $currentQueue = null;
100
101    /**
102     * When the current job started
103     * @var ?int
104     */
105    protected ?int $currentJobStartedAt = null;
106
107    /**
108     * The current job's own timeout, stored so isLikelyStuck() can be
109     * answered without loading the job payload
110     * @var ?int
111     */
112    protected ?int $currentJobTimeout = null;
113
114    /**
115     * Jobs completed since registration
116     * @var int
117     */
118    protected int $jobsProcessed = 0;
119
120    /**
121     * Jobs failed since registration
122     * @var int
123     */
124    protected int $jobsFailed = 0;
125
126    /**
127     * Constructor
128     *
129     * Timestamps are explicit parameters rather than defaulted to time()
130     * so time-dependent behavior is testable without sleeping.
131     *
132     * @param string $id
133     * @param string $host
134     * @param int    $pid
135     * @param int    $startedAt
136     * @param int    $lastSeenAt
137     */
138    public function __construct(string $id, string $host, int $pid, int $startedAt, int $lastSeenAt)
139    {
140        $this->id         = $id;
141        $this->host       = $host;
142        $this->pid        = $pid;
143        $this->startedAt  = $startedAt;
144        $this->lastSeenAt = $lastSeenAt;
145    }
146
147    /**
148     * Create a record for the current process
149     *
150     * The ID carries a random suffix so a recycled PID can never collide
151     * with an older record from the same host.
152     *
153     * @param  ?string $name
154     * @param  array   $queues
155     * @param  string  $mode
156     * @return WorkerRecord
157     */
158    public static function create(?string $name = null, array $queues = [], string $mode = self::MODE_SINGLE_PASS): WorkerRecord
159    {
160        $host = gethostname();
161        $host = ($host !== false) ? $host : 'unknown';
162        $pid  = getmypid();
163        $pid  = ($pid !== false) ? $pid : 0;
164        $now  = time();
165        $id   = $host . ':' . $pid . ':' . bin2hex(random_bytes(6));
166
167        $record = new self($id, $host, $pid, $now, $now);
168        $record->setName($name);
169        $record->setQueues($queues);
170        $record->setMode($mode);
171
172        return $record;
173    }
174
175    public function getId(): string
176    {
177        return $this->id;
178    }
179
180    public function setName(?string $name): WorkerRecord
181    {
182        $this->name = $name;
183        return $this;
184    }
185
186    public function getName(): ?string
187    {
188        return $this->name;
189    }
190
191    public function getHost(): string
192    {
193        return $this->host;
194    }
195
196    public function getPid(): int
197    {
198        return $this->pid;
199    }
200
201    public function getStartedAt(): int
202    {
203        return $this->startedAt;
204    }
205
206    public function getLastSeenAt(): int
207    {
208        return $this->lastSeenAt;
209    }
210
211    public function setQueues(array $queues): WorkerRecord
212    {
213        $this->queues = $queues;
214        return $this;
215    }
216
217    public function getQueues(): array
218    {
219        return $this->queues;
220    }
221
222    public function setMode(string $mode): WorkerRecord
223    {
224        $this->mode = $mode;
225        return $this;
226    }
227
228    public function getMode(): string
229    {
230        return $this->mode;
231    }
232
233    /**
234     * Refresh the heartbeat
235     *
236     * @return WorkerRecord
237     */
238    public function touch(): WorkerRecord
239    {
240        $this->lastSeenAt = time();
241        return $this;
242    }
243
244    /**
245     * Record the job about to be executed
246     *
247     * @param  string  $jobId
248     * @param  ?string $queue
249     * @param  ?int    $timeout
250     * @return WorkerRecord
251     */
252    public function setCurrentJob(string $jobId, ?string $queue = null, ?int $timeout = null): WorkerRecord
253    {
254        $this->currentJobId        = $jobId;
255        $this->currentQueue        = $queue;
256        $this->currentJobTimeout   = $timeout;
257        $this->currentJobStartedAt = time();
258        return $this;
259    }
260
261    /**
262     * Set the current job's start time directly (used when restoring a
263     * record, and by tests that need a deterministic duration)
264     *
265     * @param  ?int $startedAt
266     * @return WorkerRecord
267     */
268    public function setCurrentJobStartedAt(?int $startedAt): WorkerRecord
269    {
270        $this->currentJobStartedAt = $startedAt;
271        return $this;
272    }
273
274    public function clearCurrentJob(): WorkerRecord
275    {
276        $this->currentJobId        = null;
277        $this->currentQueue        = null;
278        $this->currentJobStartedAt = null;
279        $this->currentJobTimeout   = null;
280        return $this;
281    }
282
283    public function getCurrentJobId(): ?string
284    {
285        return $this->currentJobId;
286    }
287
288    public function getCurrentQueue(): ?string
289    {
290        return $this->currentQueue;
291    }
292
293    public function getCurrentJobStartedAt(): ?int
294    {
295        return $this->currentJobStartedAt;
296    }
297
298    public function getCurrentJobTimeout(): ?int
299    {
300        return $this->currentJobTimeout;
301    }
302
303    public function incrementProcessed(): WorkerRecord
304    {
305        $this->jobsProcessed++;
306        return $this;
307    }
308
309    public function incrementFailed(): WorkerRecord
310    {
311        $this->jobsFailed++;
312        return $this;
313    }
314
315    public function getJobsProcessed(): int
316    {
317        return $this->jobsProcessed;
318    }
319
320    public function getJobsFailed(): int
321    {
322        return $this->jobsFailed;
323    }
324
325    /**
326     * Whether this worker's heartbeat is older than the given threshold
327     *
328     * @param  int $seconds
329     * @return bool
330     */
331    public function isStale(int $seconds = 90): bool
332    {
333        return ((time() - $this->lastSeenAt) > $seconds);
334    }
335
336    /**
337     * How long the current job has been running, or null when idle
338     *
339     * @return ?int
340     */
341    public function getCurrentJobDuration(): ?int
342    {
343        return ($this->currentJobStartedAt !== null) ? (time() - $this->currentJobStartedAt) : null;
344    }
345
346    /**
347     * Whether this worker looks stuck: its heartbeat is stale AND it has a
348     * job in flight that has outlived its own timeout (or, when the job set
349     * no timeout, the staleness threshold itself as the fallback yardstick).
350     *
351     * A stale worker with NO current job is deliberately not "stuck" - it is
352     * wedged idle, which getStaleWorkers() surfaces instead.
353     *
354     * @param  int $staleSeconds
355     * @return bool
356     */
357    public function isLikelyStuck(int $staleSeconds = 90): bool
358    {
359        if (!$this->isStale($staleSeconds) || ($this->currentJobId === null)) {
360            return false;
361        }
362
363        $duration = $this->getCurrentJobDuration();
364        if ($duration === null) {
365            return false;
366        }
367
368        return ($this->currentJobTimeout !== null)
369            ? ($duration > $this->currentJobTimeout)
370            : ($duration > $staleSeconds);
371    }
372
373    /**
374     * Flatten to a plain array for storage
375     *
376     * @return array
377     */
378    public function toArray(): array
379    {
380        return [
381            'id'                  => $this->id,
382            'name'                => $this->name,
383            'host'                => $this->host,
384            'pid'                 => $this->pid,
385            'startedAt'           => $this->startedAt,
386            'lastSeenAt'          => $this->lastSeenAt,
387            'queues'              => $this->queues,
388            'mode'                => $this->mode,
389            'currentJobId'        => $this->currentJobId,
390            'currentQueue'        => $this->currentQueue,
391            'currentJobStartedAt' => $this->currentJobStartedAt,
392            'currentJobTimeout'   => $this->currentJobTimeout,
393            'jobsProcessed'       => $this->jobsProcessed,
394            'jobsFailed'          => $this->jobsFailed,
395        ];
396    }
397
398    /**
399     * Rebuild from a stored array
400     *
401     * @param  array $data
402     * @return WorkerRecord
403     */
404    public static function fromArray(array $data): WorkerRecord
405    {
406        $record = new self(
407            (string)($data['id'] ?? ''),
408            (string)($data['host'] ?? ''),
409            (int)($data['pid'] ?? 0),
410            (int)($data['startedAt'] ?? 0),
411            (int)($data['lastSeenAt'] ?? 0)
412        );
413
414        $record->setName($data['name'] ?? null);
415        $record->setQueues($data['queues'] ?? []);
416        $record->setMode((string)($data['mode'] ?? self::MODE_SINGLE_PASS));
417
418        if (!empty($data['currentJobId'])) {
419            $record->currentJobId        = (string)$data['currentJobId'];
420            $record->currentQueue        = $data['currentQueue'] ?? null;
421            $record->currentJobTimeout   = isset($data['currentJobTimeout'])
422                ? (int)$data['currentJobTimeout'] : null;
423            $record->currentJobStartedAt = isset($data['currentJobStartedAt'])
424                ? (int)$data['currentJobStartedAt'] : null;
425        }
426
427        $record->jobsProcessed = (int)($data['jobsProcessed'] ?? 0);
428        $record->jobsFailed    = (int)($data['jobsFailed'] ?? 0);
429
430        return $record;
431    }
432
433}