38282766118

Committed 27 Sep 2026 11:28PM UTC coverage: 96.442% (+1.0%) from 95.491%

Build # 38282766118

Build Type

push

github

Committed by

mabar

Commit Message

Fix PHPStan reports

Coverage Stats

1572 of 1630 relevant lines covered (96.44%)

18.54 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.12

/src/Executor/ProcessJobExecutor.php

<?php declare(strict_types = 1);

namespace Orisai\Scheduler\Executor;

use Closure;
use DateTimeImmutable;
use DateTimeZone;
use Generator;
use JsonException;
use Orisai\Clock\Adapter\ClockAdapterFactory;
use Orisai\Clock\Clock;
use Orisai\Clock\SystemClock;
use Orisai\Exceptions\Logic\InvalidState;
use Orisai\Exceptions\Message;
use Orisai\Scheduler\Exception\JobProcessFailure;
use Orisai\Scheduler\Exception\RunFailure;
use Orisai\Scheduler\Job\JobSchedule;
use Orisai\Scheduler\Maintenance\MaintenanceJobSummaryFactory;
use Orisai\Scheduler\Status\JobInfo;
use Orisai\Scheduler\Status\JobResult;
use Orisai\Scheduler\Status\JobResultState;
use Orisai\Scheduler\Status\JobSummary;
use Orisai\Scheduler\Status\RunParameters;
use Orisai\Scheduler\Status\RunSummary;
use Psr\Clock\ClockInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use Symfony\Component\Process\PhpExecutableFinder;
use Symfony\Component\Process\Process;
use function array_merge;
use function assert;
use function is_array;
use function json_decode;
use function json_encode;
use function strlen;
use function strpos;
use function substr;
use function trim;
use const JSON_THROW_ON_ERROR;

/**
 * @infection-ignore-all
 */
final class ProcessJobExecutor implements JobExecutor
{

        private Clock $clock;

        private LoggerInterface $logger;

        private MaintenanceJobSummaryFactory $maintenanceSummaryFactory;

        private string $script = 'bin/console';

        private string $command = 'scheduler:run-job';

        public function __construct(?ClockInterface $clock = null, ?LoggerInterface $logger = null)
        {
                $this->clock = ClockAdapterFactory::create($clock ?? new SystemClock());
                $this->logger = $logger ?? new NullLogger();
                $this->maintenanceSummaryFactory = new MaintenanceJobSummaryFactory($this->clock);
        }

        public function setExecutable(string $script, string $command = 'scheduler:run-job'): void
        {
                $this->script = $script;
                $this->command = $command;
        }

        public function runJobs(
                array $jobSchedulesBySecond,
                DateTimeImmutable $runStart,
                Closure $beforeRunCallback,
                Closure $afterRunCallback,
                ?ShutdownCheck $shutdownCheck = null,
                ?Closure $onJobEvent = null
        ): Generator
        {
                $phpCommand = $this->resolvePhpCommand();

                $beforeRunCallback();

                $state = new ProcessRunState($jobSchedulesBySecond, $runStart);

                while ($state->jobExecutions !== [] || $state->jobSchedulesBySecond !== []) {
                        yield from $this->detectShutdown($state, $shutdownCheck);

                        $killed = yield from $this->forceKillAfterGrace($state, $shutdownCheck, $onJobEvent);
                        if ($killed) {
                                break;
                        }

                        $this->startDueJobs($state, $phpCommand);

                        yield from $this->reapFinishedProcesses($state, $shutdownCheck, $onJobEvent);

                        // Nothing to do, wait
                        $this->clock->sleep(0, 1);
                }

                $summary = new RunSummary($runStart, $this->clock->now(), $state->jobSummaries, $state->maintenanceActive);

                $afterRunCallback($summary);

                if ($state->suppressedExceptions !== []) {
                        throw RunFailure::create($summary, $state->suppressedExceptions);
                }

                return $summary;
        }

        /**
         * @return list<string>
         */
        private function resolvePhpCommand(): array
        {
                $finder = new PhpExecutableFinder();
                $binary = $finder->find(false);
                // @codeCoverageIgnoreStart
                if ($binary === false) {
                        throw InvalidState::create()
                                ->withMessage('PHP executable could not be found, subprocess cannot be executed.');
                }

                // @codeCoverageIgnoreEnd
                return array_merge([$binary], $finder->findArguments());
        }

        /**
         * @return Generator<int, JobSummary, void, void>
         */
        private function detectShutdown(ProcessRunState $state, ?ShutdownCheck $shutdownCheck): Generator
        {
                if ($shutdownCheck === null || $state->shutdownDetectedAt !== null) {
                        return;
                }

                // Refresh registry (throttled to every 30 seconds)
                $now = (float) $this->clock->now()->format('U.u');
                if ($now - $state->lastRefreshAt >= 30.0) {
                        $state->lastRefreshAt = $now;
                        $shutdownCheck->refresh();
                }

                // Check for shutdown (throttled to every 1 second)
                $now = (float) $this->clock->now()->format('U.u');
                if ($now - $state->lastShutdownCheckAt >= 1.0) {
                        $state->lastShutdownCheckAt = $now;

                        if ($shutdownCheck->shouldShutdown()) {
                                $state->shutdownDetectedAt = $now;
                                $state->maintenanceActive = true;

                                // Create maintenance summaries for jobs not yet started
                                foreach ($state->jobSchedulesBySecond as $second => $schedules) {
                                        foreach ($schedules as $id => $jobSchedule) {
                                                yield $state->jobSummaries[] = $this->maintenanceSummaryFactory->create(
                                                        $id,
                                                        $jobSchedule,
                                                        $second,
                                                        $state->runStart,
                                                );
                                        }
                                }

                                $state->jobSchedulesBySecond = [];
                        }
                }
        }

        /**
         * @param (Closure(int|string, JobSchedule, int<0, max>, JobInfo): void)|null $onJobEvent
         * @return Generator<int, JobSummary, void, bool>
         */
        private function forceKillAfterGrace(
                ProcessRunState $state,
                ?ShutdownCheck $shutdownCheck,
                ?Closure $onJobEvent
        ): Generator
        {
                if ($state->shutdownDetectedAt === null || $shutdownCheck === null || $state->jobExecutions === []) {
                        return false;
                }

                $elapsed = (float) $this->clock->now()->format('U.u') - $state->shutdownDetectedAt;
                if ($elapsed < $shutdownCheck->getGracePeriodSeconds()) {
                        return false;
                }

                foreach ($state->jobExecutions as $i => $execution) {
                        if ($execution->process->isRunning()) {
                                $execution->process->stop(10);
                        }

                        // Final drain of subprocess output after stop()
                        $this->pollSubprocess($execution, $onJobEvent);

                        unset($state->jobExecutions[$i]);

                        $summary = $this->tryCollectJobSummary($execution);
                        if ($summary === null) {
                                // Subprocess was killed before emitting `finished`. Reuse the
                                // JobInfo from `started` (if received) to preserve executionId
                                // pairing with the already-fired beforeJob callback.
                                $summary = $execution->startedInfo !== null
                                        ? $this->createJobSummaryFromStartedInfo(
                                                $execution->startedInfo,
                                                $execution->schedule,
                                                JobResultState::maintenance(),
                                        )
                                        : $this->maintenanceSummaryFactory->create(
                                                $execution->id,
                                                $execution->schedule,
                                                0,
                                                $state->runStart,
                                        );
                        }

                        $this->logUnexpectedOutputIfAny($execution);

                        yield $state->jobSummaries[] = $summary;
                }

                return true;
        }

        /**
         * @param list<string> $phpCommand
         */
        private function startDueJobs(ProcessRunState $state, array $phpCommand): void
        {
                if ($state->jobSchedulesBySecond === []) {
                        return;
                }

                $shouldRunSecond = $this->clock->now()->getTimestamp() - $state->runStart->getTimestamp();

                while ($state->lastExecutedSecond < $shouldRunSecond) {
                        $currentSecond = $state->lastExecutedSecond + 1;
                        if (isset($state->jobSchedulesBySecond[$currentSecond])) {
                                $this->startJobs(
                                        $phpCommand,
                                        $state->jobSchedulesBySecond[$currentSecond],
                                        $state,
                                        new RunParameters($currentSecond, false),
                                );
                                unset($state->jobSchedulesBySecond[$currentSecond]);
                        }

                        $state->lastExecutedSecond = $currentSecond;
                }
        }

        /**
         * @param (Closure(int|string, JobSchedule, int<0, max>, JobInfo): void)|null $onJobEvent
         * @return Generator<int, JobSummary, void, void>
         */
        private function reapFinishedProcesses(
                ProcessRunState $state,
                ?ShutdownCheck $shutdownCheck,
                ?Closure $onJobEvent
        ): Generator
        {
                foreach ($state->jobExecutions as $i => $execution) {
                        // Poll subprocess for new framework events (dispatches `started`)
                        $this->pollSubprocess($execution, $onJobEvent);

                        if ($execution->process->isRunning()) {
                                continue;
                        }

                        // Subprocess exited — drain any remaining output (finished event may
                        // arrive right before exit and not be visible until after isRunning() flipped).
                        $this->pollSubprocess($execution, $onJobEvent);

                        unset($state->jobExecutions[$i]);

                        $summary = $this->classifyExitedProcess($execution, $state, $shutdownCheck);
                        if ($summary === null) {
                                continue;
                        }

                        $this->logUnexpectedOutputIfAny($execution);

                        yield $state->jobSummaries[] = $summary;
                }
        }

        private function classifyExitedProcess(
                SubprocessExecutionState $execution,
                ProcessRunState $state,
                ?ShutdownCheck $shutdownCheck
        ): ?JobSummary
        {
                $summary = $this->tryCollectJobSummary($execution);

                if ($summary !== null) {
                        if ($execution->failureEvent !== null) {
                                // Job threw in the subprocess and had no errorHandler — surface it as a
                                // suppressed exception so runPromise throws RunFailure (parity with
                                // BasicJobExecutor behavior, parity with pre-events-protocol behavior).
                                $state->suppressedExceptions[] = $this->createUnhandledJobFailure(
                                        $execution->process,
                                        $execution->failureEvent,
                                );
                        }

                        return $summary;
                }

                // Check shutdown directly - SIGINT may have killed the subprocess before
                // the throttled shutdown check had a chance to set $shutdownDetectedAt
                if ($shutdownCheck !== null && $shutdownCheck->shouldShutdown()) {
                        $state->maintenanceActive = true;

                        // Reuse startedInfo if received (preserves executionId pairing).
                        return $execution->startedInfo !== null
                                ? $this->createJobSummaryFromStartedInfo(
                                        $execution->startedInfo,
                                        $execution->schedule,
                                        JobResultState::maintenance(),
                                )
                                : $this->maintenanceSummaryFactory->create(
                                        $execution->id,
                                        $execution->schedule,
                                        0,
                                        $state->runStart,
                                );
                }

                if ($execution->startedInfo !== null) {
                        // Subprocess crashed after `started` but before `finished`. beforeJob
                        // was already fired for this job — yield a synthetic fail summary so
                        // afterJob fires and the pairing invariant holds. Also surface the
                        // subprocess-level failure via RunFailure.
                        $state->suppressedExceptions[] = $this->createSubprocessFail(
                                $execution->process,
                                trim($execution->process->getOutput()),
                                trim($execution->process->getErrorOutput()),
                        );

                        return $this->createJobSummaryFromStartedInfo(
                                $execution->startedInfo,
                                $execution->schedule,
                                JobResultState::fail(),
                        );
                }

                // Subprocess died before emitting any event. beforeJob never fired
                // either, so skipping yield keeps the invariant intact.
                $state->suppressedExceptions[] = $this->createSubprocessFail(
                        $execution->process,
                        trim($execution->process->getOutput()),
                        trim($execution->process->getErrorOutput()),
                );

                return null;
        }

        /**
         * @param list<string> $phpCommand
         * @param array<int|string, JobSchedule> $jobSchedules
         */
        private function startJobs(
                array $phpCommand,
                array $jobSchedules,
                ProcessRunState $state,
                RunParameters $parameters
        ): void
        {
                foreach ($jobSchedules as $id => $jobSchedule) {
                        $execution = new Process(
                                array_merge($phpCommand, [
                                        $this->script,
                                        $this->command,
                                        $id,
                                        '--events',
                                        '--parameters',
                                        json_encode($parameters->toArray(), JSON_THROW_ON_ERROR),
                                ]),
                        );
                        $execution->start();

                        $state->jobExecutions[] = new SubprocessExecutionState($execution, $jobSchedule, $id);
                }
        }

        /**
         * Reads incremental subprocess stdout, parses marker-prefixed JSON lines as
         * framework events, and dispatches `started` events to $onJobEvent. Non-marker
         * lines accumulate in the state's unexpectedStdout buffer.
         *
         * @param (Closure(int|string, JobSchedule, int<0, max>, JobInfo): void)|null $onJobEvent
         */
        private function pollSubprocess(SubprocessExecutionState $state, ?Closure $onJobEvent): void
        {
                $state->unparsedBuffer .= $state->process->getIncrementalOutput();

                $marker = SubprocessEventProtocol::EventMarker;
                $markerLen = strlen($marker);

                while (($nlPos = strpos($state->unparsedBuffer, "\n")) !== false) {
                        $line = substr($state->unparsedBuffer, 0, $nlPos);
                        $state->unparsedBuffer = substr($state->unparsedBuffer, $nlPos + 1);

                        if ($line === '') {
                                continue;
                        }

                        if (strpos($line, $marker) !== 0) {
                                $state->unexpectedStdout .= $line . "\n";

                                continue;
                        }

                        $json = substr($line, $markerLen);
                        try {
                                $event = json_decode($json, true, 512, JSON_THROW_ON_ERROR);
                        } catch (JsonException $e) {
                                // Malformed framework event — treat the whole line as unexpected output
                                $state->unexpectedStdout .= $line . "\n";

                                continue;
                        }

                        assert(is_array($event));

                        $type = $event['type'] ?? null;
                        if ($type === SubprocessEventProtocol::TypeStarted) {
                                if (!$state->startedDispatched) {
                                        $info = $this->buildJobInfo($event['info'], $state->schedule);
                                        $state->startedInfo = $info;
                                        if ($onJobEvent !== null) {
                                                $onJobEvent($state->id, $state->schedule, $info->getRunSecond(), $info);
                                        }

                                        $state->startedDispatched = true;
                                }
                        } elseif ($type === SubprocessEventProtocol::TypeFinished) {
                                $state->finishedEvent = $event;
                        } elseif ($type === SubprocessEventProtocol::TypeFailure) {
                                $state->failureEvent = $event;
                        }
                }
        }

        private function tryCollectJobSummary(SubprocessExecutionState $state): ?JobSummary
        {
                $stderr = trim($state->process->getErrorOutput());
                if ($stderr !== '') {
                        $this->logUnexpectedStderr($state->process, $state->id, $stderr);
                }

                if ($state->finishedEvent === null) {
                        return null;
                }

                return $this->createSummary($state->finishedEvent, $state->schedule);
        }

        private function logUnexpectedOutputIfAny(SubprocessExecutionState $state): void
        {
                $captured = '';
                if ($state->finishedEvent !== null && isset($state->finishedEvent['stdout'])) {
                        $captured = (string) $state->finishedEvent['stdout'];
                }

                $combined = $state->unexpectedStdout . $captured;
                if ($combined !== '') {
                        $this->logUnexpectedStdout($state->process, $state->id, $combined);
                }
        }

        /**
         * Builds a synthetic JobSummary reusing the JobInfo from the subprocess's
         * `started` event. Used when the subprocess exited without emitting `finished`
         * (maintenance-killed or crashed) — reusing the JobInfo preserves the
         * `executionId` pairing between beforeJob and afterJob user callbacks.
         */
        private function createJobSummaryFromStartedInfo(
                JobInfo $startedInfo,
                JobSchedule $jobSchedule,
                JobResultState $state
        ): JobSummary
        {
                $timezone = $jobSchedule->getTimeZone();
                $now = $timezone !== null
                        ? $this->clock->now()->setTimezone($timezone)
                        : $this->clock->now();

                $result = new JobResult($jobSchedule->getExpression(), $now, $state);

                return new JobSummary($startedInfo, $result);
        }

        /**
         * @param array<mixed> $rawInfo
         */
        private function buildJobInfo(array $rawInfo, JobSchedule $jobSchedule): JobInfo
        {
                return new JobInfo(
                        $rawInfo['id'],
                        $rawInfo['name'],
                        $rawInfo['expression'],
                        $rawInfo['repeatAfterSeconds'],
                        $rawInfo['runSecond'],
                        DateTimeImmutable::createFromFormat('U.u', $rawInfo['start'][0])
                                ->setTimezone(new DateTimeZone($rawInfo['start'][1])),
                        $jobSchedule->getTimeZone(),
                        $rawInfo['manualRun'],
                );
        }

        /**
         * @param array<mixed> $raw
         */
        private function createSummary(array $raw, JobSchedule $jobSchedule): JobSummary
        {
                return new JobSummary(
                        $this->buildJobInfo($raw['info'], $jobSchedule),
                        new JobResult(
                                $jobSchedule->getExpression(),
                                DateTimeImmutable::createFromFormat('U.u', $raw['result']['end'][0])
                                        ->setTimezone(new DateTimeZone($raw['result']['end'][1])),
                                JobResultState::from($raw['result']['state']),
                                $raw['result']['lockExpired'] ?? false,
                        ),
                );
        }

        private function createSubprocessFail(Process $execution, string $output, string $errorOutput): JobProcessFailure
        {
                $message = Message::create()
                        ->withContext("Running job via command {$execution->getCommandLine()}")
                        ->withProblem('Job subprocess did not correctly write job result to stdout.')
                        ->with('Tip', 'Check the documentation for troubleshooting guide.')
                        ->with('Exit code', (string) $execution->getExitCode())
                        ->with('stdout', $output)
                        ->with('stderr', $errorOutput);

                return JobProcessFailure::create()
                        ->withMessage($message);
        }

        /**
         * @param array<mixed> $failureEvent
         */
        private function createUnhandledJobFailure(Process $execution, array $failureEvent): JobProcessFailure
        {
                $class = isset($failureEvent['class']) ? (string) $failureEvent['class'] : 'Throwable';
                $text = isset($failureEvent['message']) ? (string) $failureEvent['message'] : '';

                $message = Message::create()
                        ->withContext("Running job via command {$execution->getCommandLine()}")
                        ->withProblem("Job threw an unhandled $class: $text")
                        ->with('Tip', 'Register an error handler on the scheduler to handle job failures gracefully.');

                return JobProcessFailure::create()
                        ->withMessage($message);
        }

        /**
         * @param int|string $jobId
         */
        private function logUnexpectedStderr(Process $execution, $jobId, string $stderr): void
        {
                $this->logger->warning("Subprocess running job '$jobId' produced unexpected stderr output.", [
                        'id' => $jobId,
                        'command' => $execution->getCommandLine(),
                        'exitCode' => $execution->getExitCode(),
                        'stderr' => $stderr,
                ]);
        }

        /**
         * @param int|string $jobId
         */
        private function logUnexpectedStdout(Process $execution, $jobId, string $stdout): void
        {
                $this->logger->warning("Subprocess running job '$jobId' produced unexpected stdout output.", [
                        'id' => $jobId,
                        'command' => $execution->getCommandLine(),
                        'exitCode' => $execution->getExitCode(),
                        'stdout' => $stdout,
                ]);
        }

}

1	<?php declare(strict_types = 1);
2
3	namespace Orisai\Scheduler\Executor;
4
5	use Closure;
6	use DateTimeImmutable;
7	use DateTimeZone;
8	use Generator;
9	use JsonException;
10	use Orisai\Clock\Adapter\ClockAdapterFactory;
11	use Orisai\Clock\Clock;
12	use Orisai\Clock\SystemClock;
13	use Orisai\Exceptions\Logic\InvalidState;
14	use Orisai\Exceptions\Message;
15	use Orisai\Scheduler\Exception\JobProcessFailure;
16	use Orisai\Scheduler\Exception\RunFailure;
17	use Orisai\Scheduler\Job\JobSchedule;
18	use Orisai\Scheduler\Maintenance\MaintenanceJobSummaryFactory;
19	use Orisai\Scheduler\Status\JobInfo;
20	use Orisai\Scheduler\Status\JobResult;
21	use Orisai\Scheduler\Status\JobResultState;
22	use Orisai\Scheduler\Status\JobSummary;
23	use Orisai\Scheduler\Status\RunParameters;
24	use Orisai\Scheduler\Status\RunSummary;
25	use Psr\Clock\ClockInterface;
26	use Psr\Log\LoggerInterface;
27	use Psr\Log\NullLogger;
28	use Symfony\Component\Process\PhpExecutableFinder;
29	use Symfony\Component\Process\Process;
30	use function array_merge;
31	use function assert;
32	use function is_array;
33	use function json_decode;
34	use function json_encode;
35	use function strlen;
36	use function strpos;
37	use function substr;
38	use function trim;
39	use const JSON_THROW_ON_ERROR;
40
41	/**
42	* @infection-ignore-all
43	*/
44	final class ProcessJobExecutor implements JobExecutor
45	{
46
47	private Clock $clock;
48
49	private LoggerInterface $logger;
50
51	private MaintenanceJobSummaryFactory $maintenanceSummaryFactory;
52
53	private string $script = 'bin/console';
54
55	private string $command = 'scheduler:run-job';
56
57	public function __construct(?ClockInterface $clock = null, ?LoggerInterface $logger = null)
58	{
59	$this->clock = ClockAdapterFactory::create($clock ?? new SystemClock());	35 ✔
60	$this->logger = $logger ?? new NullLogger();	35 ✔
61	$this->maintenanceSummaryFactory = new MaintenanceJobSummaryFactory($this->clock);	35 ✔
62	}
63
64	public function setExecutable(string $script, string $command = 'scheduler:run-job'): void
65	{
66	$this->script = $script;	20 ✔
67	$this->command = $command;	20 ✔
68	}
69
70	public function runJobs(
71	array $jobSchedulesBySecond,
72	DateTimeImmutable $runStart,
73	Closure $beforeRunCallback,
74	Closure $afterRunCallback,
75	?ShutdownCheck $shutdownCheck = null,
76	?Closure $onJobEvent = null
77	): Generator
78	{
79	$phpCommand = $this->resolvePhpCommand();	25 ✔
80
81	$beforeRunCallback();	25 ✔
82
83	$state = new ProcessRunState($jobSchedulesBySecond, $runStart);	25 ✔
84
85	while ($state->jobExecutions !== [] \|\| $state->jobSchedulesBySecond !== []) {	25 ✔
86	yield from $this->detectShutdown($state, $shutdownCheck);	21 ✔
87
88	$killed = yield from $this->forceKillAfterGrace($state, $shutdownCheck, $onJobEvent);	21 ✔
89	if ($killed) {	21 ✔
90	break;	5 ✔
91	}
92
93	$this->startDueJobs($state, $phpCommand);	21 ✔
94
95	yield from $this->reapFinishedProcesses($state, $shutdownCheck, $onJobEvent);	21 ✔
96
97	// Nothing to do, wait
98	$this->clock->sleep(0, 1);	21 ✔
99	}
100
101	$summary = new RunSummary($runStart, $this->clock->now(), $state->jobSummaries, $state->maintenanceActive);	25 ✔
102
103	$afterRunCallback($summary);	25 ✔
104
105	if ($state->suppressedExceptions !== []) {	25 ✔
106	throw RunFailure::create($summary, $state->suppressedExceptions);	5 ✔
107	}
108
109	return $summary;	20 ✔
110	}
111
112	/**
113	* @return list<string>
114	*/
115	private function resolvePhpCommand(): array
116	{
117	$finder = new PhpExecutableFinder();	25 ✔
118	$binary = $finder->find(false);	25 ✔
119	// @codeCoverageIgnoreStart
120	if ($binary === false) {
121	throw InvalidState::create()
122	->withMessage('PHP executable could not be found, subprocess cannot be executed.');
123	}
124
125	// @codeCoverageIgnoreEnd
126	return array_merge([$binary], $finder->findArguments());	25 ✔
127	}
128
129	/**
130	* @return Generator<int, JobSummary, void, void>
131	*/
132	private function detectShutdown(ProcessRunState $state, ?ShutdownCheck $shutdownCheck): Generator
133	{
134	if ($shutdownCheck === null \|\| $state->shutdownDetectedAt !== null) {	21 ✔
135	return;	15 ✔
136	}
137
138	// Refresh registry (throttled to every 30 seconds)
139	$now = (float) $this->clock->now()->format('U.u');	9 ✔
140	if ($now - $state->lastRefreshAt >= 30.0) {	9 ✔
141	$state->lastRefreshAt = $now;	3 ✔
142	$shutdownCheck->refresh();	3 ✔
143	}
144
145	// Check for shutdown (throttled to every 1 second)
146	$now = (float) $this->clock->now()->format('U.u');	9 ✔
147	if ($now - $state->lastShutdownCheckAt >= 1.0) {	9 ✔
148	$state->lastShutdownCheckAt = $now;	9 ✔
149
150	if ($shutdownCheck->shouldShutdown()) {	9 ✔
151	$state->shutdownDetectedAt = $now;	9 ✔
152	$state->maintenanceActive = true;	9 ✔
153
154	// Create maintenance summaries for jobs not yet started
155	foreach ($state->jobSchedulesBySecond as $second => $schedules) {	9 ✔
156	foreach ($schedules as $id => $jobSchedule) {	2 ✔
157	yield $state->jobSummaries[] = $this->maintenanceSummaryFactory->create(	2 ✔
158	$id,	2 ✔
159	$jobSchedule,	2 ✔
160	$second,	2 ✔
161	$state->runStart,	2 ✔
162	);	2 ✔
163	}
164	}
165
166	$state->jobSchedulesBySecond = [];	9 ✔
167	}
168	}
169	}
170
171	/**
172	* @param (Closure(int\|string, JobSchedule, int<0, max>, JobInfo): void)\|null $onJobEvent
173	* @return Generator<int, JobSummary, void, bool>
174	*/
175	private function forceKillAfterGrace(
176	ProcessRunState $state,
177	?ShutdownCheck $shutdownCheck,
178	?Closure $onJobEvent
179	): Generator
180	{
181	if ($state->shutdownDetectedAt === null \|\| $shutdownCheck === null \|\| $state->jobExecutions === []) {	21 ✔
182	return false;	21 ✔
183	}
184
185	$elapsed = (float) $this->clock->now()->format('U.u') - $state->shutdownDetectedAt;	7 ✔
186	if ($elapsed < $shutdownCheck->getGracePeriodSeconds()) {	7 ✔
187	return false;	3 ✔
188	}
189
190	foreach ($state->jobExecutions as $i => $execution) {	5 ✔
191	if ($execution->process->isRunning()) {	5 ✔
192	$execution->process->stop(10);	5 ✔
193	}
194
195	// Final drain of subprocess output after stop()
196	$this->pollSubprocess($execution, $onJobEvent);	5 ✔
197
198	unset($state->jobExecutions[$i]);	5 ✔
199
200	$summary = $this->tryCollectJobSummary($execution);	5 ✔
201	if ($summary === null) {	5 ✔
202	// Subprocess was killed before emitting `finished`. Reuse the
203	// JobInfo from `started` (if received) to preserve executionId
204	// pairing with the already-fired beforeJob callback.
205	$summary = $execution->startedInfo !== null	3 ✔
206	? $this->createJobSummaryFromStartedInfo(	1 ✔
207	$execution->startedInfo,	1 ✔
208	$execution->schedule,	1 ✔
209	JobResultState::maintenance(),	1 ✔
210	)	1 ✔
211	: $this->maintenanceSummaryFactory->create(	2 ✔
212	$execution->id,	2 ✔
213	$execution->schedule,	2 ✔
214	0,	2 ✔
215	$state->runStart,	2 ✔
216	);	2 ✔
217	}
218
219	$this->logUnexpectedOutputIfAny($execution);	5 ✔
220
221	yield $state->jobSummaries[] = $summary;	5 ✔
222	}
223
224	return true;	5 ✔
225	}
226
227	/**
228	* @param list<string> $phpCommand
229	*/
230	private function startDueJobs(ProcessRunState $state, array $phpCommand): void
231	{
232	if ($state->jobSchedulesBySecond === []) {	21 ✔
233	return;	21 ✔
234	}
235
236	$shouldRunSecond = $this->clock->now()->getTimestamp() - $state->runStart->getTimestamp();	19 ✔
237
238	while ($state->lastExecutedSecond < $shouldRunSecond) {	19 ✔
239	$currentSecond = $state->lastExecutedSecond + 1;	19 ✔
240	if (isset($state->jobSchedulesBySecond[$currentSecond])) {	19 ✔
241	$this->startJobs(	19 ✔
242	$phpCommand,	19 ✔
243	$state->jobSchedulesBySecond[$currentSecond],	19 ✔
244	$state,	19 ✔
245	new RunParameters($currentSecond, false),	19 ✔
246	);	19 ✔
247	unset($state->jobSchedulesBySecond[$currentSecond]);	19 ✔
248	}
249
250	$state->lastExecutedSecond = $currentSecond;	19 ✔
251	}
252	}
253
254	/**
255	* @param (Closure(int\|string, JobSchedule, int<0, max>, JobInfo): void)\|null $onJobEvent
256	* @return Generator<int, JobSummary, void, void>
257	*/
258	private function reapFinishedProcesses(
259	ProcessRunState $state,
260	?ShutdownCheck $shutdownCheck,
261	?Closure $onJobEvent
262	): Generator
263	{
264	foreach ($state->jobExecutions as $i => $execution) {	21 ✔
265	// Poll subprocess for new framework events (dispatches `started`)
266	$this->pollSubprocess($execution, $onJobEvent);	19 ✔
267
268	if ($execution->process->isRunning()) {	19 ✔
269	continue;	19 ✔
270	}
271
272	// Subprocess exited — drain any remaining output (finished event may
273	// arrive right before exit and not be visible until after isRunning() flipped).
274	$this->pollSubprocess($execution, $onJobEvent);	14 ✔
275
276	unset($state->jobExecutions[$i]);	14 ✔
277
278	$summary = $this->classifyExitedProcess($execution, $state, $shutdownCheck);	14 ✔
279	if ($summary === null) {	14 ✔
280	continue;	2 ✔
281	}
282
283	$this->logUnexpectedOutputIfAny($execution);	12 ✔
284
285	yield $state->jobSummaries[] = $summary;	12 ✔
286	}
287	}
288
289	private function classifyExitedProcess(
290	SubprocessExecutionState $execution,
291	ProcessRunState $state,
292	?ShutdownCheck $shutdownCheck
293	): ?JobSummary
294	{
295	$summary = $this->tryCollectJobSummary($execution);	14 ✔
296
297	if ($summary !== null) {	14 ✔
298	if ($execution->failureEvent !== null) {	10 ✔
299	// Job threw in the subprocess and had no errorHandler — surface it as a
300	// suppressed exception so runPromise throws RunFailure (parity with
301	// BasicJobExecutor behavior, parity with pre-events-protocol behavior).
302	$state->suppressedExceptions[] = $this->createUnhandledJobFailure(	2 ✔
303	$execution->process,	2 ✔
304	$execution->failureEvent,	2 ✔
305	);	2 ✔
306	}
307
308	return $summary;	10 ✔
309	}
310
311	// Check shutdown directly - SIGINT may have killed the subprocess before
312	// the throttled shutdown check had a chance to set $shutdownDetectedAt
313	if ($shutdownCheck !== null && $shutdownCheck->shouldShutdown()) {	4 ✔
314	$state->maintenanceActive = true;	1 ✔
315
316	// Reuse startedInfo if received (preserves executionId pairing).
317	return $execution->startedInfo !== null	1 ✔
318	? $this->createJobSummaryFromStartedInfo(	×
319	$execution->startedInfo,	×
320	$execution->schedule,	×
321	JobResultState::maintenance(),	×
322	)	×
323	: $this->maintenanceSummaryFactory->create(	1 ✔
324	$execution->id,	1 ✔
325	$execution->schedule,	1 ✔
326	0,	1 ✔
327	$state->runStart,	1 ✔
328	);	1 ✔
329	}
330
331	if ($execution->startedInfo !== null) {	3 ✔
332	// Subprocess crashed after `started` but before `finished`. beforeJob
333	// was already fired for this job — yield a synthetic fail summary so
334	// afterJob fires and the pairing invariant holds. Also surface the
335	// subprocess-level failure via RunFailure.
336	$state->suppressedExceptions[] = $this->createSubprocessFail(	1 ✔
337	$execution->process,	1 ✔
338	trim($execution->process->getOutput()),	1 ✔
339	trim($execution->process->getErrorOutput()),	1 ✔
340	);	1 ✔
341
342	return $this->createJobSummaryFromStartedInfo(	1 ✔
343	$execution->startedInfo,	1 ✔
344	$execution->schedule,	1 ✔
345	JobResultState::fail(),	1 ✔
346	);	1 ✔
347	}
348
349	// Subprocess died before emitting any event. beforeJob never fired
350	// either, so skipping yield keeps the invariant intact.
351	$state->suppressedExceptions[] = $this->createSubprocessFail(	2 ✔
352	$execution->process,	2 ✔
353	trim($execution->process->getOutput()),	2 ✔
354	trim($execution->process->getErrorOutput()),	2 ✔
355	);	2 ✔
356
357	return null;	2 ✔
358	}
359
360	/**
361	* @param list<string> $phpCommand
362	* @param array<int\|string, JobSchedule> $jobSchedules
363	*/
364	private function startJobs(
365	array $phpCommand,
366	array $jobSchedules,
367	ProcessRunState $state,
368	RunParameters $parameters
369	): void
370	{
371	foreach ($jobSchedules as $id => $jobSchedule) {	19 ✔
372	$execution = new Process(	19 ✔
373	array_merge($phpCommand, [	19 ✔
374	$this->script,	19 ✔
375	$this->command,	19 ✔
376	$id,	19 ✔
377	'--events',	19 ✔
378	'--parameters',	19 ✔
379	json_encode($parameters->toArray(), JSON_THROW_ON_ERROR),	19 ✔
380	]),	19 ✔
381	);	19 ✔
382	$execution->start();	19 ✔
383
384	$state->jobExecutions[] = new SubprocessExecutionState($execution, $jobSchedule, $id);	19 ✔
385	}
386	}
387
388	/**
389	* Reads incremental subprocess stdout, parses marker-prefixed JSON lines as
390	* framework events, and dispatches `started` events to $onJobEvent. Non-marker
391	* lines accumulate in the state's unexpectedStdout buffer.
392	*
393	* @param (Closure(int\|string, JobSchedule, int<0, max>, JobInfo): void)\|null $onJobEvent
394	*/
395	private function pollSubprocess(SubprocessExecutionState $state, ?Closure $onJobEvent): void
396	{
397	$state->unparsedBuffer .= $state->process->getIncrementalOutput();	19 ✔
398
399	$marker = SubprocessEventProtocol::EventMarker;	19 ✔
400	$markerLen = strlen($marker);	19 ✔
401
402	while (($nlPos = strpos($state->unparsedBuffer, "\n")) !== false) {	19 ✔
403	$line = substr($state->unparsedBuffer, 0, $nlPos);	14 ✔
404	$state->unparsedBuffer = substr($state->unparsedBuffer, $nlPos + 1);	14 ✔
405
406	if ($line === '') {	14 ✔
407	continue;	×
408	}
409
410	if (strpos($line, $marker) !== 0) {	14 ✔
411	$state->unexpectedStdout .= $line . "\n";	×
412
413	continue;	×
414	}
415
416	$json = substr($line, $markerLen);	14 ✔
417	try {
418	$event = json_decode($json, true, 512, JSON_THROW_ON_ERROR);	14 ✔
419	} catch (JsonException $e) {	×
420	// Malformed framework event — treat the whole line as unexpected output
421	$state->unexpectedStdout .= $line . "\n";	×
422
423	continue;	×
424	}
425
426	assert(is_array($event));	14 ✔
427
428	$type = $event['type'] ?? null;	14 ✔
429	if ($type === SubprocessEventProtocol::TypeStarted) {	14 ✔
430	if (!$state->startedDispatched) {	14 ✔
431	$info = $this->buildJobInfo($event['info'], $state->schedule);	14 ✔
432	$state->startedInfo = $info;	14 ✔
433	if ($onJobEvent !== null) {	14 ✔
434	$onJobEvent($state->id, $state->schedule, $info->getRunSecond(), $info);	14 ✔
435	}
436
437	$state->startedDispatched = true;	14 ✔
438	}
439	} elseif ($type === SubprocessEventProtocol::TypeFinished) {	12 ✔
440	$state->finishedEvent = $event;	12 ✔
441	} elseif ($type === SubprocessEventProtocol::TypeFailure) {	2 ✔
442	$state->failureEvent = $event;	2 ✔
443	}
444	}
445	}
446
447	private function tryCollectJobSummary(SubprocessExecutionState $state): ?JobSummary
448	{
449	$stderr = trim($state->process->getErrorOutput());	19 ✔
450	if ($stderr !== '') {	19 ✔
451	$this->logUnexpectedStderr($state->process, $state->id, $stderr);	4 ✔
452	}
453
454	if ($state->finishedEvent === null) {	19 ✔
455	return null;	7 ✔
456	}
457
458	return $this->createSummary($state->finishedEvent, $state->schedule);	12 ✔
459	}
460
461	private function logUnexpectedOutputIfAny(SubprocessExecutionState $state): void
462	{
463	$captured = '';	17 ✔
464	if ($state->finishedEvent !== null && isset($state->finishedEvent['stdout'])) {	17 ✔
465	$captured = (string) $state->finishedEvent['stdout'];	12 ✔
466	}
467
468	$combined = $state->unexpectedStdout . $captured;	17 ✔
469	if ($combined !== '') {	17 ✔
470	$this->logUnexpectedStdout($state->process, $state->id, $combined);	2 ✔
471	}
472	}
473
474	/**
475	* Builds a synthetic JobSummary reusing the JobInfo from the subprocess's
476	* `started` event. Used when the subprocess exited without emitting `finished`
477	* (maintenance-killed or crashed) — reusing the JobInfo preserves the
478	* `executionId` pairing between beforeJob and afterJob user callbacks.
479	*/
480	private function createJobSummaryFromStartedInfo(
481	JobInfo $startedInfo,
482	JobSchedule $jobSchedule,
483	JobResultState $state
484	): JobSummary
485	{
486	$timezone = $jobSchedule->getTimeZone();	2 ✔
487	$now = $timezone !== null	2 ✔
488	? $this->clock->now()->setTimezone($timezone)	×
489	: $this->clock->now();	2 ✔
490
491	$result = new JobResult($jobSchedule->getExpression(), $now, $state);	2 ✔
492
493	return new JobSummary($startedInfo, $result);	2 ✔
494	}
495
496	/**
497	* @param array<mixed> $rawInfo
498	*/
499	private function buildJobInfo(array $rawInfo, JobSchedule $jobSchedule): JobInfo
500	{
501	return new JobInfo(	14 ✔
502	$rawInfo['id'],	14 ✔
503	$rawInfo['name'],	14 ✔
504	$rawInfo['expression'],	14 ✔
505	$rawInfo['repeatAfterSeconds'],	14 ✔
506	$rawInfo['runSecond'],	14 ✔
507	DateTimeImmutable::createFromFormat('U.u', $rawInfo['start'][0])	14 ✔
508	->setTimezone(new DateTimeZone($rawInfo['start'][1])),	14 ✔
509	$jobSchedule->getTimeZone(),	14 ✔
510	$rawInfo['manualRun'],	14 ✔
511	);	14 ✔
512	}
513
514	/**
515	* @param array<mixed> $raw
516	*/
517	private function createSummary(array $raw, JobSchedule $jobSchedule): JobSummary
518	{
519	return new JobSummary(	12 ✔
520	$this->buildJobInfo($raw['info'], $jobSchedule),	12 ✔
521	new JobResult(	12 ✔
522	$jobSchedule->getExpression(),	12 ✔
523	DateTimeImmutable::createFromFormat('U.u', $raw['result']['end'][0])	12 ✔
524	->setTimezone(new DateTimeZone($raw['result']['end'][1])),	12 ✔
525	JobResultState::from($raw['result']['state']),	12 ✔
526	$raw['result']['lockExpired'] ?? false,	12 ✔
527	),	12 ✔
528	);	12 ✔
529	}
530
531	private function createSubprocessFail(Process $execution, string $output, string $errorOutput): JobProcessFailure
532	{
533	$message = Message::create()	3 ✔
534	->withContext("Running job via command {$execution->getCommandLine()}")	3 ✔
535	->withProblem('Job subprocess did not correctly write job result to stdout.')	3 ✔
536	->with('Tip', 'Check the documentation for troubleshooting guide.')	3 ✔
537	->with('Exit code', (string) $execution->getExitCode())	3 ✔
538	->with('stdout', $output)	3 ✔
539	->with('stderr', $errorOutput);	3 ✔
540
541	return JobProcessFailure::create()	3 ✔
542	->withMessage($message);	3 ✔
543	}
544
545	/**
546	* @param array<mixed> $failureEvent
547	*/
548	private function createUnhandledJobFailure(Process $execution, array $failureEvent): JobProcessFailure
549	{
550	$class = isset($failureEvent['class']) ? (string) $failureEvent['class'] : 'Throwable';	2 ✔
551	$text = isset($failureEvent['message']) ? (string) $failureEvent['message'] : '';	2 ✔
552
553	$message = Message::create()	2 ✔
554	->withContext("Running job via command {$execution->getCommandLine()}")	2 ✔
555	->withProblem("Job threw an unhandled $class: $text")	2 ✔
556	->with('Tip', 'Register an error handler on the scheduler to handle job failures gracefully.');	2 ✔
557
558	return JobProcessFailure::create()	2 ✔
559	->withMessage($message);	2 ✔
560	}
561
562	/**
563	* @param int\|string $jobId
564	*/
565	private function logUnexpectedStderr(Process $execution, $jobId, string $stderr): void
566	{
567	$this->logger->warning("Subprocess running job '$jobId' produced unexpected stderr output.", [	4 ✔
568	'id' => $jobId,	4 ✔
569	'command' => $execution->getCommandLine(),	4 ✔
570	'exitCode' => $execution->getExitCode(),	4 ✔
571	'stderr' => $stderr,	4 ✔
572	]);	4 ✔
573	}
574
575	/**
576	* @param int\|string $jobId
577	*/
578	private function logUnexpectedStdout(Process $execution, $jobId, string $stdout): void
579	{
580	$this->logger->warning("Subprocess running job '$jobId' produced unexpected stdout output.", [	2 ✔
581	'id' => $jobId,	2 ✔
582	'command' => $execution->getCommandLine(),	2 ✔
583	'exitCode' => $execution->getExitCode(),	2 ✔
584	'stdout' => $stdout,	2 ✔
585	]);	2 ✔
586	}
587
588	}

orisai / scheduler / 38282766118

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous