Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
90 / 90
100.00%
19 / 19
 CRAP
100.00%
1 / 1
Worker
100.00%
90 / 90
100.00%
19 / 19
 44
100.00%
1 / 1
 __construct
100.00%
2 / 2
100.00%
1 / 1
 1
 getPid
100.00%
1 / 1
100.00%
1 / 1
 1
 isRunning
100.00%
2 / 2
100.00%
1 / 1
 1
 getStatus
100.00%
2 / 2
100.00%
1 / 1
 1
 getExitCode
100.00%
2 / 2
100.00%
1 / 1
 1
 getTerminationSignal
100.00%
2 / 2
100.00%
1 / 1
 1
 getOutput
100.00%
2 / 2
100.00%
1 / 1
 1
 getErrorOutput
100.00%
2 / 2
100.00%
1 / 1
 1
 wait
100.00%
5 / 5
100.00%
1 / 1
 3
 kill
100.00%
6 / 6
100.00%
1 / 1
 4
 startForkedWorker
100.00%
7 / 7
100.00%
1 / 1
 3
 executeCallback
100.00%
31 / 31
100.00%
1 / 1
 5
 normalizeExitCode
100.00%
5 / 5
100.00%
1 / 1
 3
 pollState
100.00%
14 / 14
100.00%
1 / 1
 7
 isCurrentWorkerProcess
100.00%
2 / 2
100.00%
1 / 1
 2
 detectCurrentPid
100.00%
2 / 2
100.00%
1 / 1
 3
 isInterruptedWait
100.00%
1 / 1
100.00%
1 / 1
 2
 isNoChildError
100.00%
1 / 1
100.00%
1 / 1
 2
 isNoSuchProcessError
100.00%
1 / 1
100.00%
1 / 1
 2
1<?php
2
3declare(strict_types=1);
4
5/**
6 * This file is part of fast-forward/fork.
7 *
8 * This source file is subject to the license bundled
9 * with this source code in the file LICENSE.
10 *
11 * @copyright Copyright (c) 2026 Felipe Sayão Lobato Abreu <github@mentordosnerds.com>
12 * @license   https://opensource.org/licenses/MIT MIT License
13 *
14 * @see       https://github.com/php-fast-forward/fork
15 * @see       https://github.com/php-fast-forward
16 * @see       https://datatracker.ietf.org/doc/html/rfc2119
17 */
18
19namespace FastForward\Fork\Worker;
20
21use Closure;
22use FastForward\Fork\Exception\LogicException;
23use FastForward\Fork\Exception\RuntimeException;
24use FastForward\Fork\Manager\ForkManagerInterface;
25use FastForward\Fork\Signal\Signal;
26use Psr\Log\LoggerInterface;
27use Throwable;
28use function error_reporting;
29use function getmypid;
30use function max;
31use function min;
32use function ob_end_flush;
33use function ob_get_level;
34use function ob_start;
35use function pcntl_fork;
36use function posix_get_last_error;
37use function pcntl_waitpid;
38use function posix_kill;
39use function restore_error_handler;
40use function set_error_handler;
41
42/**
43 * Represents a single forked worker process.
44 *
45 * This class encapsulates the lifecycle of a worker created via process forking.
46 * It is responsible for:
47 * - Spawning the child process
48 * - Executing the provided callback in the child context
49 * - Capturing output and errors
50 * - Synchronizing state with the master process
51 *
52 * Instances of this class MUST be created exclusively by a manager implementation.
53 * Consumers SHOULD NOT instantiate this class directly.
54 */
55  class Worker implements WorkerInterface
56{
57    /**
58     * POSIX error code for interrupted system calls.
59     */
60    private const int INTERRUPTED_SYSTEM_CALL_ERROR = 4;
61
62    /**
63     * POSIX error code for missing child processes.
64     */
65    private const int NO_CHILD_PROCESS_ERROR = 10;
66
67    /**
68     * POSIX error code for non-existent processes.
69     */
70    private const int NO_SUCH_PROCESS_ERROR = 3;
71
72    /**
73     * Stores the callback executed inside the child process.
74     *
75     * The callback MUST accept the current worker instance as its only argument.
76     *
77     * @var Closure(WorkerInterface):mixed
78     */
79    private Closure $callback;
80
81    /**
82     * Initializes the worker and immediately forks the process.
83     *
84     * The constructor MUST trigger process forking and configure both parent
85     * and child execution contexts accordingly.
86     *
87     * @param ForkManagerInterface $manager manager responsible for this worker
88     * @param WorkerState $state shared mutable state between parent and child
89     * @param callable(WorkerInterface): mixed $callback callback executed in the child process
90     * @param ?LoggerInterface $logger logger used for lifecycle events
91     */
92    public function __construct(
93        private ForkManagerInterface $manager,
94        private WorkerState $state,
95        callable $callback,
96        private ?LoggerInterface $logger = null,
97    ) {
98        $this->callback = Closure::fromCallable($callback);
99        $this->startForkedWorker();
100    }
101
102    /**
103     * {@inheritDoc}
104     */
105    public function getPid(): int
106    {
107        return $this->state->getPid();
108    }
109
110    /**
111     * {@inheritDoc}
112     */
113    public function isRunning(): bool
114    {
115        $this->pollState();
116
117        return $this->state->isRunning();
118    }
119
120    /**
121     * {@inheritDoc}
122     */
123    public function getStatus(): ?int
124    {
125        $this->pollState();
126
127        return $this->state->getStatus();
128    }
129
130    /**
131     * {@inheritDoc}
132     */
133    public function getExitCode(): ?int
134    {
135        $this->pollState();
136
137        return $this->state->getExitCode();
138    }
139
140    /**
141     * {@inheritDoc}
142     */
143    public function getTerminationSignal(): ?Signal
144    {
145        $this->pollState();
146
147        return $this->state->getTerminationSignal();
148    }
149
150    /**
151     * {@inheritDoc}
152     */
153    public function getOutput(): string
154    {
155        $this->pollState();
156
157        return $this->state->getOutput();
158    }
159
160    /**
161     * {@inheritDoc}
162     */
163    public function getErrorOutput(): string
164    {
165        $this->pollState();
166
167        return $this->state->getErrorOutput();
168    }
169
170    /**
171     * {@inheritDoc}
172     *
173     * This method SHALL block until the worker finishes execution.
174     * It MUST NOT be invoked from within the same worker process.
175     */
176    public function wait(): void
177    {
178        if (! $this->state->isRunning()) {
179            return;
180        }
181
182        if ($this->isCurrentWorkerProcess()) {
183            throw LogicException::forWorkerWaitingOnItself($this->state->getPid());
184        }
185
186        $this->manager->wait($this);
187    }
188
189    /**
190     * {@inheritDoc}
191     *
192     * This method attempts to send a POSIX signal to the worker process.
193     * If the process no longer exists, the worker SHALL be marked as detached.
194     */
195    public function kill(Signal $signal = Signal::Terminate): void
196    {
197        if (! $this->state->isRunning()) {
198            return;
199        }
200
201        if (posix_kill($this->state->getPid(), $signal->value)) {
202            return;
203        }
204
205        if ($this->isNoSuchProcessError(posix_get_last_error())) {
206            $this->state->markDetached($this->logger);
207        }
208    }
209
210    /**
211     * Forks the current process and initializes parent/child execution paths.
212     *
213     * The parent process SHALL retain control and track the worker.
214     * The child process MUST execute the provided callback and terminate.
215     *
216     * @throws RuntimeException if the fork operation fails
217     */
218    private function startForkedWorker(): void
219    {
220        $pid = pcntl_fork();
221
222        // @codeCoverageIgnoreStart
223        if (-1 === $pid) {
224            throw RuntimeException::forUnableToForkWorker();
225        }
226
227        // @codeCoverageIgnoreEnd
228
229        if ($pid > 0) {
230            $this->state->activateParent($pid);
231
232            $this->logger?->info('Forked worker process.', [
233                'worker_pid' => $pid,
234            ]);
235
236            return;
237        }
238
239        // @codeCoverageIgnoreStart
240        $pid = $this->detectCurrentPid();
241        $this->state->activateChild($pid);
242
243        pcntl_async_signals(true);
244
245        $this->logger?->info('Starting forked worker execution.', [
246            'worker_pid' => $pid,
247        ]);
248
249        exit($this->executeCallback());
250        // @codeCoverageIgnoreEnd
251    }
252
253    /**
254     * Executes the worker callback and normalizes the exit code.
255     *
256     * Output buffering MUST be used to capture standard output and forward it
257     * to the parent process. Errors SHALL be intercepted and written to the
258     * worker error output channel.
259     *
260     * @return int normalized process exit code (0–255)
261     */
262    private function executeCallback(): int
263    {
264        $bufferLevel = ob_get_level();
265
266        set_error_handler(function (int $severity, string $message, string $file, int $line): bool {
267            if ((error_reporting() & $severity) === 0) {
268                return false;
269            }
270
271            $this->state->writeErrorOutput(\sprintf(
272                "[fast-forward/fork] PHP error %d: %s in %s on line %d\n",
273                $severity,
274                $message,
275                $file,
276                $line,
277            ));
278
279            return true;
280        });
281
282        ob_start(function (string $buffer): string {
283            $this->state->writeOutput($buffer);
284
285            return '';
286        }, 1);
287
288        try {
289            $result = ($this->callback)($this);
290
291            return $this->normalizeExitCode($result);
292        } catch (Throwable $throwable) {
293            $this->state->writeErrorOutput(\sprintf(
294                "[fast-forward/fork] Worker %d failed with %s: %s\n",
295                $this->state->getPid(),
296                $throwable::class,
297                $throwable->getMessage(),
298            ));
299
300            return $throwable->getCode() > 0 ? $throwable->getCode() : 255;
301        } finally {
302            while (ob_get_level() > $bufferLevel) {
303                ob_end_flush();
304            }
305
306            restore_error_handler();
307            $this->state->closeChildSide();
308        }
309    }
310
311    /**
312     * Normalizes a callback result into a valid exit code.
313     *
314     * Integers MUST be clamped to the range 0–255.
315     * A boolean false SHALL be converted to exit code 1.
316     * Any other value SHALL result in exit code 0.
317     *
318     * @param mixed $result callback return value
319     *
320     * @return int normalized exit code
321     */
322    private function normalizeExitCode(mixed $result): int
323    {
324        if (\is_int($result)) {
325            return max(0, min(255, $result));
326        }
327
328        if (false === $result) {
329            return 1;
330        }
331
332        return 0;
333    }
334
335    /**
336     * Performs a non-blocking state synchronization with the worker process.
337     *
338     * This method MUST only operate in the master process context.
339     *
340     * @throws RuntimeException if waiting for the worker fails unexpectedly
341     */
342    private function pollState(): void
343    {
344        if (! $this->state->isRunning() || $this->isCurrentWorkerProcess()) {
345            return;
346        }
347
348        $this->state->drainOutput(logger: $this->logger);
349
350        $status = 0;
351        $pid = pcntl_waitpid($this->state->getPid(), $status, \WNOHANG);
352
353        if ($pid > 0) {
354            $this->state->markTerminated($status, $this->logger);
355
356            return;
357        }
358
359        if (0 === $pid) {
360            return;
361        }
362
363        $error = pcntl_get_last_error();
364
365        // @codeCoverageIgnoreStart
366        if ($this->isInterruptedWait($error)) {
367            return;
368        }
369
370        // @codeCoverageIgnoreEnd
371
372        if ($this->isNoChildError($error)) {
373            $this->state->markDetached($this->logger);
374
375            return;
376        }
377
378        // @codeCoverageIgnoreStart
379        throw RuntimeException::forWorkerWaitFailure($this->state->getPid(), pcntl_strerror($error));
380        // @codeCoverageIgnoreEnd
381    }
382
383    /**
384     * Determines whether the current process corresponds to this worker.
385     *
386     * @return bool true if executing inside this worker process
387     */
388    private function isCurrentWorkerProcess(): bool
389    {
390        return $this->manager->isWorker()
391            && $this->detectCurrentPid() === $this->state->getPid();
392    }
393
394    /**
395     * Detects the current process identifier (PID).
396     *
397     * @return int the current process ID
398     *
399     * @throws RuntimeException if the PID cannot be determined
400     */
401    private function detectCurrentPid(): int
402    {
403        $pid = getmypid();
404
405        // @codeCoverageIgnoreStart
406        if (! \is_int($pid) || $pid < 1) {
407            throw RuntimeException::forUndetectableProcessIdentifier();
408        }
409
410        // @codeCoverageIgnoreEnd
411
412        return $pid;
413    }
414
415    /**
416     * Determines whether a wait failure was caused by an interrupted system call.
417     *
418     * @param int $error error code returned by pcntl
419     *
420     * @return bool true if interrupted
421     */
422    private function isInterruptedWait(int $error): bool
423    {
424        return (\defined('PCNTL_EINTR') ? \PCNTL_EINTR : self::INTERRUPTED_SYSTEM_CALL_ERROR) === $error;
425    }
426
427    /**
428     * Determines whether no child processes remain.
429     *
430     * @param int $error error code returned by pcntl
431     *
432     * @return bool true if no child processes exist
433     */
434    private function isNoChildError(int $error): bool
435    {
436        return (\defined('PCNTL_ECHILD') ? \PCNTL_ECHILD : self::NO_CHILD_PROCESS_ERROR) === $error;
437    }
438
439    /**
440     * Determines whether the process no longer exists.
441     *
442     * @param int $error error code returned by POSIX APIs
443     *
444     * @return bool true if the target process is gone
445     */
446    private function isNoSuchProcessError(int $error): bool
447    {
448        return (\defined('POSIX_ESRCH') ? \constant('POSIX_ESRCH') : self::NO_SUCH_PROCESS_ERROR) === $error;
449    }
450}