24634959708

Committed 19 Apr 2026 05:31PM UTC coverage: 95.05% (-2.8%) from 97.882%

Build # 24634959708

Build Type

push

github

Committed by

mabar

Commit Message

Move events from subprocesses to main process, fire after job callbacks for locked jobs and maintenance, parse subprocess events via new protocol

Coverage Stats

171 of 223 new or added lines in 4 files covered. (76.68%)

54 existing lines in 3 files now uncovered.

3053 of 3212 relevant lines covered (95.05%)

76.83 hits per line

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

96.56

/src/ManagedScheduler.php

<?php declare(strict_types = 1);

namespace Orisai\Scheduler;

use Closure;
use DateTimeImmutable;
use Generator;
use Orisai\Clock\Adapter\ClockAdapterFactory;
use Orisai\Clock\Clock;
use Orisai\Clock\SystemClock;
use Orisai\Exceptions\Logic\InvalidArgument;
use Orisai\Exceptions\Message;
use Orisai\Scheduler\Exception\JobFailure;
use Orisai\Scheduler\Executor\BasicJobExecutor;
use Orisai\Scheduler\Executor\JobExecutor;
use Orisai\Scheduler\Executor\ShutdownCheck;
use Orisai\Scheduler\Job\JobLock;
use Orisai\Scheduler\Job\JobSchedule;
use Orisai\Scheduler\Maintenance\CreatesMaintenanceJobSummary;
use Orisai\Scheduler\Maintenance\MaintenanceManager;
use Orisai\Scheduler\Manager\JobManager;
use Orisai\Scheduler\RunRegistry\RunRegistry;
use Orisai\Scheduler\Status\ActivityStatus;
use Orisai\Scheduler\Status\JobInfo;
use Orisai\Scheduler\Status\JobResult;
use Orisai\Scheduler\Status\JobResultState;
use Orisai\Scheduler\Status\JobSummary;
use Orisai\Scheduler\Status\PlannedJobInfo;
use Orisai\Scheduler\Status\RunInfo;
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\Lock\LockFactory;
use Symfony\Component\Lock\LockInterface;
use Symfony\Component\Lock\Store\InMemoryStore;
use Throwable;
use function bin2hex;
use function getmypid;
use function iterator_to_array;
use function random_bytes;
use function time;

class ManagedScheduler implements Scheduler
{

        use CreatesMaintenanceJobSummary;

        private const MinuteLockTtl = 30.0;

        private JobManager $jobManager;

        /** @var Closure(Throwable, JobInfo, JobResult): (void)|null */
        private ?Closure $errorHandler;

        private LockFactory $lockFactory;

        private JobExecutor $executor;

        private Clock $clock;

        private LoggerInterface $logger;

        private ?MaintenanceManager $maintenanceManager;

        private ?RunRegistry $runRegistry;

        /** @var list<array{LockInterface, float}> */
        private array $minuteLocks = [];

        /** @var list<Closure(JobInfo, JobResult): void> */
        private array $lockedJobCallbacks = [];

        /** @var list<Closure(JobInfo): void> */
        private array $beforeJobCallbacks = [];

        /** @var list<Closure(JobInfo, JobResult): void> */
        private array $afterJobCallbacks = [];

        /** @var list<Closure(RunInfo): void> */
        private array $beforeRunCallbacks = [];

        /** @var list<Closure(RunSummary): void> */
        private array $afterRunCallbacks = [];

        /**
         * @param Closure(Throwable, JobInfo, JobResult): (void)|null $errorHandler
         * @param-later-invoked-callable $errorHandler
         */
        public function __construct(
                JobManager $jobManager,
                ?Closure $errorHandler = null,
                ?LockFactory $lockFactory = null,
                ?JobExecutor $executor = null,
                ?ClockInterface $clock = null,
                ?LoggerInterface $logger = null,
                ?MaintenanceManager $maintenanceManager = null,
                ?RunRegistry $runRegistry = null
        )
        {
                $this->jobManager = $jobManager;
                $this->errorHandler = $errorHandler;
                $this->lockFactory = $lockFactory ?? new LockFactory(new InMemoryStore());
                $this->clock = ClockAdapterFactory::create($clock ?? new SystemClock());
                $this->logger = $logger ?? new NullLogger();

                $this->executor = $executor ?? new BasicJobExecutor(
                        $this->clock,
                        fn ($id, JobSchedule $jobSchedule, int $runSecond): array => $this->runInternal(
                                $id,
                                $jobSchedule,
                                new RunParameters($runSecond, false),
                                true,
                                false,
                        ),
                );

                $this->maintenanceManager = $maintenanceManager;
                $this->runRegistry = $runRegistry;
        }

        public function getStatus(): ActivityStatus
        {
                $maintenance = $this->maintenanceManager !== null
                        ? $this->maintenanceManager->isMaintenance()
                        : null;

                $activeRuns = $this->runRegistry !== null
                        ? $this->runRegistry->getActiveRuns()
                        : [];

                return new ActivityStatus($maintenance, $activeRuns);
        }

        public function getJobSchedules(): array
        {
                return $this->jobManager->getJobSchedules();
        }

        public function runJob(
                $id,
                bool $force = true,
                ?RunParameters $parameters = null,
                ?Closure $onJobStarted = null,
                ?Closure $onJobFinished = null
        ): ?JobSummary
        {
                $this->releaseMinuteLocks();

                $jobSchedule = $this->jobManager->getJobSchedule($id);
                // Explicit RunParameters signals subprocess context — parent's runPromise() handles
                // all callback firing, so runInternal must suppress to prevent double-firing.
                $fireCallbacks = $parameters === null;
                $parameters ??= new RunParameters(0, true);

                if ($jobSchedule === null) {
                        $message = Message::create()
                                ->withContext("Running job with ID '$id'")
                                ->withProblem('Job is not registered by scheduler.')
                                ->with(
                                        'Tip',
                                        "Inspect keys in 'Scheduler->getJobSchedules()' or run command 'scheduler:list' to find correct job ID.",
                                );

                        throw InvalidArgument::create()
                                ->withMessage($message);
                }

                $expression = $jobSchedule->getExpression();

                $timeZone = $jobSchedule->getTimeZone();
                $jobDueTime = $timeZone !== null
                        ? $this->clock->now()->setTimezone($timeZone)
                        : $this->clock->now();

                // Intentionally ignores repeat after seconds
                if (!$force && !$expression->isDue($jobDueTime)) {
                        return null;
                }

                if (!$force && $this->maintenanceManager !== null && $this->maintenanceManager->isMaintenance()) {
                        return null;
                }

                try {
                        [$summary, $throwable] = $this->runInternal(
                                $id,
                                $jobSchedule,
                                $parameters,
                                $fireCallbacks,
                                $fireCallbacks,
                                $onJobStarted,
                                $onJobFinished,
                        );
                } finally {
                        $this->releaseMinuteLocks();
                }

                if ($throwable !== null) {
                        throw JobFailure::create($summary, $throwable);
                }

                return $summary;
        }

        /**
         * @param array<int|string, JobSchedule> $jobSchedules
         * @return array<int, array<int|string, JobSchedule>>
         */
        private function groupJobSchedulesBySecond(array $jobSchedules): array
        {
                $scheduledJobsBySecond = [];
                foreach ($jobSchedules as $id => $jobSchedule) {
                        $repeatAfterSeconds = $jobSchedule->getRepeatAfterSeconds();

                        if ($repeatAfterSeconds === 0) {
                                $scheduledJobsBySecond[0][$id] = $jobSchedule;
                        } else {
                                for ($second = 0; $second <= 59; $second += $repeatAfterSeconds) {
                                        $scheduledJobsBySecond[$second][$id] = $jobSchedule;
                                }
                        }
                }

                return $scheduledJobsBySecond;
        }

        public function runPromise(): Generator
        {
                $this->releaseMinuteLocks();

                $runStart = $this->clock->now();
                $runId = time() . '-' . bin2hex(random_bytes(3));

                if ($this->runRegistry !== null) {
                        $pid = getmypid();
                        $this->runRegistry->register($runId, $pid !== false ? $pid : 0);
                }

                try {
                        $jobSchedules = [];
                        foreach ($this->jobManager->getJobSchedules() as $id => $schedule) {
                                $timeZone = $schedule->getTimeZone();
                                $jobDueTime = $timeZone !== null
                                        ? $runStart->setTimezone($timeZone)
                                        : $runStart;

                                if ($schedule->getExpression()->isDue($jobDueTime)) {
                                        $jobSchedules[$id] = $schedule;
                                }
                        }

                        // Check maintenance before starting any jobs
                        if ($this->maintenanceManager !== null && $this->maintenanceManager->isMaintenance()) {
                                $generator = $this->createMaintenanceRunSummary($runStart, $jobSchedules);

                                yield from $this->wrapGeneratorWithJobCallbacks($generator);

                                return $generator->getReturn();
                        }

                        $shutdownCheck = $this->createShutdownCheck($runId);

                        $generator = $this->executor->runJobs(
                                $this->groupJobSchedulesBySecond($jobSchedules),
                                $runStart,
                                $this->getBeforeRunCallback($runStart, $jobSchedules),
                                $this->getAfterRunCallback(),
                                $shutdownCheck,
                                $this->getOnJobEventCallback(),
                        );

                        yield from $this->wrapGeneratorWithJobCallbacks($generator);

                        return $generator->getReturn();
                } finally {
                        if ($this->runRegistry !== null) {
                                $this->runRegistry->deregister($runId);
                        }

                        $this->releaseMinuteLocks();
                }
        }

        private function releaseMinuteLocks(): void
        {
                $now = (float) $this->clock->now()->format('U.u');
                $remaining = [];

                foreach ($this->minuteLocks as [$lock, $createdAt]) {
                        if ($now - $createdAt >= self::MinuteLockTtl) {
                                $lock->release();
                        } else {
                                $remaining[] = [$lock, $createdAt];
                        }
                }

                $this->minuteLocks = $remaining;
        }

        private function createShutdownCheck(string $runId): ?ShutdownCheck
        {
                if ($this->maintenanceManager === null) {
                        return null;
                }

                $manager = $this->maintenanceManager;
                $registry = $this->runRegistry;

                return new ShutdownCheck(
                        static fn (): bool => $manager->isShutdownRequested() || $manager->isMaintenance(),
                        $manager->getGracePeriodSeconds(),
                        static function () use ($registry, $runId): void {
                                if ($registry !== null) {
                                        $registry->refresh($runId);
                                }
                        },
                );
        }

        /**
         * @param array<int|string, JobSchedule> $jobSchedules
         * @return Generator<int, JobSummary, void, RunSummary>
         */
        private function createMaintenanceRunSummary(DateTimeImmutable $runStart, array $jobSchedules): Generator
        {
                $beforeRunCb = $this->getBeforeRunCallback($runStart, $jobSchedules);
                $beforeRunCb();

                $jobSummaries = [];
                foreach ($jobSchedules as $id => $jobSchedule) {
                        yield $jobSummaries[] = $this->createMaintenanceJobSummary($id, $jobSchedule, 0, $runStart);
                }

                $runSummary = new RunSummary($runStart, $this->clock->now(), $jobSummaries, true);

                foreach ($this->afterRunCallbacks as $cb) {
                        $cb($runSummary);
                }

                return $runSummary;
        }

        /**
         * @param array<int|string, JobSchedule> $jobSchedules
         * @return Closure(): void
         */
        private function getBeforeRunCallback(DateTimeImmutable $runStart, array $jobSchedules): Closure
        {
                return function () use ($runStart, $jobSchedules): void {
                        if ($this->beforeRunCallbacks === []) {
                                return;
                        }

                        $jobInfos = [];
                        foreach ($jobSchedules as $id => $jobSchedule) {
                                $job = $jobSchedule->getJob();
                                $timezone = $jobSchedule->getTimeZone();
                                $jobStart = $timezone !== null
                                        ? $runStart->setTimezone($timezone)
                                        : $runStart;
                                $jobInfos[] = new PlannedJobInfo(
                                        $id,
                                        $job->getName(),
                                        $jobSchedule->getExpression()->getExpression(),
                                        $jobSchedule->getRepeatAfterSeconds(),
                                        $jobStart,
                                        $timezone,
                                );
                        }

                        $info = new RunInfo($runStart, $jobInfos);

                        foreach ($this->beforeRunCallbacks as $cb) {
                                $cb($info);
                        }
                };
        }

        /**
         * @return Closure(RunSummary): void
         */
        private function getAfterRunCallback(): Closure
        {
                return function (RunSummary $runSummary): void {
                        foreach ($this->afterRunCallbacks as $cb) {
                                $cb($runSummary);
                        }
                };
        }

        /**
         * Called by ProcessJobExecutor when the subprocess emits a "started" event.
         * Fires parent's beforeJobCallbacks with the JobInfo from the subprocess.
         *
         * @return Closure(int|string, JobSchedule, int<0, max>, JobInfo): void
         */
        private function getOnJobEventCallback(): Closure
        {
                return function ($id, JobSchedule $jobSchedule, int $runSecond, JobInfo $info): void {
                        foreach ($this->beforeJobCallbacks as $cb) {
                                $cb($info);
                        }
                };
        }

        /**
         * @param Generator<int, JobSummary, void, RunSummary> $generator
         * @return Generator<int, JobSummary, void, void>
         */
        private function wrapGeneratorWithJobCallbacks(Generator $generator): Generator
        {
                foreach ($generator as $summary) {
                        if ($summary->getResult()->getState() === JobResultState::lock()) {
                                foreach ($this->lockedJobCallbacks as $cb) {
                                        $cb($summary->getInfo(), $summary->getResult());
                                }
                        }

                        foreach ($this->afterJobCallbacks as $cb) {
                                $cb($summary->getInfo(), $summary->getResult());
                        }

                        yield $summary;
                }
        }

        public function run(): RunSummary
        {
                $generator = $this->runPromise();
                // Forces generator to execute
                iterator_to_array($generator);

                return $generator->getReturn();
        }

        /**
         * @param string|int $id
         * @param (Closure(JobInfo): void)|null $onJobStarted
         * @param (Closure(JobInfo, JobResult): void)|null $onJobFinished
         * @return array{JobSummary, Throwable|null}
         */
        private function runInternal(
                $id,
                JobSchedule $jobSchedule,
                RunParameters $runParameters,
                bool $fireBeforeJobCallbacks = true,
                bool $fireAfterJobCallbacks = true,
                ?Closure $onJobStarted = null,
                ?Closure $onJobFinished = null
        ): array
        {
                $job = $jobSchedule->getJob();
                $expression = $jobSchedule->getExpression();

                $info = new JobInfo(
                        $id,
                        $job->getName(),
                        $expression->getExpression(),
                        $jobSchedule->getRepeatAfterSeconds(),
                        $runParameters->getSecond(),
                        $this->getCurrentTime($jobSchedule),
                        $jobSchedule->getTimeZone(),
                        $runParameters->isForcedRun(),
                );

                $repeatAfterSeconds = $jobSchedule->getRepeatAfterSeconds();
                $runSecond = $runParameters->getSecond();

                // Minute lock prevents re-execution within the same minute on another server.
                // 30-second TTL with autoRelease=false — survives subprocess exit, expires naturally.
                // Skipped for forced (manual) runs — manual execution should always work.
                $minuteLock = null;
                if (!$runParameters->isForcedRun()) {
                        $minuteLockKey = $repeatAfterSeconds > 0
                                ? "Orisai.Scheduler.Job.Minute/$id/$runSecond"
                                : "Orisai.Scheduler.Job.Minute/$id";
                        $minuteLock = $this->lockFactory->createLock($minuteLockKey, self::MinuteLockTtl, false);

                        if (!$minuteLock->acquire()) {
                                $result = new JobResult($expression, $info->getStart(), JobResultState::lock());

                                if ($fireAfterJobCallbacks) {
                                        foreach ($this->lockedJobCallbacks as $cb) {
                                                $cb($info, $result);
                                        }

                                        foreach ($this->afterJobCallbacks as $cb) {
                                                $cb($info, $result);
                                        }
                                }

                                if ($onJobFinished !== null) {
                                        $onJobFinished($info, $result);
                                }

                                return [
                                        new JobSummary($info, $result),
                                        null,
                                ];
                        }
                }

                // Job lock prevents concurrent execution of the same job.
                $lock = $this->lockFactory->createLock("Orisai.Scheduler.Job/$id");

                if (!$lock->acquire()) {
                        if ($minuteLock !== null) {
                                $minuteLock->release();
                        }

                        $result = new JobResult($expression, $info->getStart(), JobResultState::lock());

                        if ($fireAfterJobCallbacks) {
                                foreach ($this->lockedJobCallbacks as $cb) {
                                        $cb($info, $result);
                                }

                                foreach ($this->afterJobCallbacks as $cb) {
                                        $cb($info, $result);
                                }
                        }

                        if ($onJobFinished !== null) {
                                $onJobFinished($info, $result);
                        }

                        return [
                                new JobSummary($info, $result),
                                null,
                        ];
                }

                $throwable = null;
                try {
                        if ($fireBeforeJobCallbacks) {
                                foreach ($this->beforeJobCallbacks as $cb) {
                                        $cb($info);
                                }
                        }

                        if ($onJobStarted !== null) {
                                $onJobStarted($info);
                        }

                        try {
                                $job->run(new JobLock($lock));
                        } catch (Throwable $throwable) {
                                // Handled bellow
                        }

                        $lockExpired = $lock->isExpired();
                        if ($lockExpired) {
                                $this->logger->warning("Lock of job '$id' expired before the job finished.", [
                                        'id' => $id,
                                ]);
                        }

                        $result = new JobResult(
                                $expression,
                                $this->getCurrentTime($jobSchedule),
                                $throwable === null ? JobResultState::done() : JobResultState::fail(),
                                $lockExpired,
                        );

                        if ($fireAfterJobCallbacks) {
                                foreach ($this->afterJobCallbacks as $cb) {
                                        $cb($info, $result);
                                }
                        }

                        if ($onJobFinished !== null) {
                                $onJobFinished($info, $result);
                        }

                        if ($throwable !== null && $this->errorHandler !== null) {
                                ($this->errorHandler)($throwable, $info, $result);
                                $throwable = null;
                        }
                } finally {
                        $lock->release();
                        // Minute lock NOT released — stays in store until 30s TTL expires.
                        // Stored to prevent GC and released at end of run via releaseMinuteLocks().
                        if ($minuteLock !== null) {
                                $this->minuteLocks[] = [$minuteLock, (float) $this->clock->now()->format('U.u')];
                        }
                }

                return [
                        new JobSummary($info, $result),
                        $throwable,
                ];
        }

        private function getCurrentTime(JobSchedule $schedule): DateTimeImmutable
        {
                $now = $this->clock->now();
                $timezone = $schedule->getTimeZone();

                return $timezone !== null
                        ? $now->setTimezone($timezone)
                        : $now;
        }

        /**
         * @param Closure(JobInfo, JobResult): void $callback
         * @param-later-invoked-callable $callback
         */
        public function addLockedJobCallback(Closure $callback): void
        {
                $this->lockedJobCallbacks[] = $callback;
        }

        /**
         * @param Closure(JobInfo): void $callback
         * @param-later-invoked-callable $callback
         */
        public function addBeforeJobCallback(Closure $callback): void
        {
                $this->beforeJobCallbacks[] = $callback;
        }

        /**
         * @param Closure(JobInfo, JobResult): void $callback
         * @param-later-invoked-callable $callback
         */
        public function addAfterJobCallback(Closure $callback): void
        {
                $this->afterJobCallbacks[] = $callback;
        }

        /**
         * @param Closure(RunInfo): void $callback
         * @param-later-invoked-callable $callback
         */
        public function addBeforeRunCallback(Closure $callback): void
        {
                $this->beforeRunCallbacks[] = $callback;
        }

        /**
         * @param Closure(RunSummary): void $callback
         * @param-later-invoked-callable $callback
         */
        public function addAfterRunCallback(Closure $callback): void
        {
                $this->afterRunCallbacks[] = $callback;
        }

}

1	<?php declare(strict_types = 1);
2
3	namespace Orisai\Scheduler;
4
5	use Closure;
6	use DateTimeImmutable;
7	use Generator;
8	use Orisai\Clock\Adapter\ClockAdapterFactory;
9	use Orisai\Clock\Clock;
10	use Orisai\Clock\SystemClock;
11	use Orisai\Exceptions\Logic\InvalidArgument;
12	use Orisai\Exceptions\Message;
13	use Orisai\Scheduler\Exception\JobFailure;
14	use Orisai\Scheduler\Executor\BasicJobExecutor;
15	use Orisai\Scheduler\Executor\JobExecutor;
16	use Orisai\Scheduler\Executor\ShutdownCheck;
17	use Orisai\Scheduler\Job\JobLock;
18	use Orisai\Scheduler\Job\JobSchedule;
19	use Orisai\Scheduler\Maintenance\CreatesMaintenanceJobSummary;
20	use Orisai\Scheduler\Maintenance\MaintenanceManager;
21	use Orisai\Scheduler\Manager\JobManager;
22	use Orisai\Scheduler\RunRegistry\RunRegistry;
23	use Orisai\Scheduler\Status\ActivityStatus;
24	use Orisai\Scheduler\Status\JobInfo;
25	use Orisai\Scheduler\Status\JobResult;
26	use Orisai\Scheduler\Status\JobResultState;
27	use Orisai\Scheduler\Status\JobSummary;
28	use Orisai\Scheduler\Status\PlannedJobInfo;
29	use Orisai\Scheduler\Status\RunInfo;
30	use Orisai\Scheduler\Status\RunParameters;
31	use Orisai\Scheduler\Status\RunSummary;
32	use Psr\Clock\ClockInterface;
33	use Psr\Log\LoggerInterface;
34	use Psr\Log\NullLogger;
35	use Symfony\Component\Lock\LockFactory;
36	use Symfony\Component\Lock\LockInterface;
37	use Symfony\Component\Lock\Store\InMemoryStore;
38	use Throwable;
39	use function bin2hex;
40	use function getmypid;
41	use function iterator_to_array;
42	use function random_bytes;
43	use function time;
44
45	class ManagedScheduler implements Scheduler
46	{
47
48	use CreatesMaintenanceJobSummary;
49
50	private const MinuteLockTtl = 30.0;
51
52	private JobManager $jobManager;
53
54	/** @var Closure(Throwable, JobInfo, JobResult): (void)\|null */
55	private ?Closure $errorHandler;
56
57	private LockFactory $lockFactory;
58
59	private JobExecutor $executor;
60
61	private Clock $clock;
62
63	private LoggerInterface $logger;
64
65	private ?MaintenanceManager $maintenanceManager;
66
67	private ?RunRegistry $runRegistry;
68
69	/** @var list<array{LockInterface, float}> */
70	private array $minuteLocks = [];
71
72	/** @var list<Closure(JobInfo, JobResult): void> */
73	private array $lockedJobCallbacks = [];
74
75	/** @var list<Closure(JobInfo): void> */
76	private array $beforeJobCallbacks = [];
77
78	/** @var list<Closure(JobInfo, JobResult): void> */
79	private array $afterJobCallbacks = [];
80
81	/** @var list<Closure(RunInfo): void> */
82	private array $beforeRunCallbacks = [];
83
84	/** @var list<Closure(RunSummary): void> */
85	private array $afterRunCallbacks = [];
86
87	/**
88	* @param Closure(Throwable, JobInfo, JobResult): (void)\|null $errorHandler
89	* @param-later-invoked-callable $errorHandler
90	*/
91	public function __construct(
92	JobManager $jobManager,
93	?Closure $errorHandler = null,
94	?LockFactory $lockFactory = null,
95	?JobExecutor $executor = null,
96	?ClockInterface $clock = null,
97	?LoggerInterface $logger = null,
98	?MaintenanceManager $maintenanceManager = null,
99	?RunRegistry $runRegistry = null
100	)
101	{
102	$this->jobManager = $jobManager;	1,032✔
103	$this->errorHandler = $errorHandler;	1,032✔
104	$this->lockFactory = $lockFactory ?? new LockFactory(new InMemoryStore());	1,032✔
105	$this->clock = ClockAdapterFactory::create($clock ?? new SystemClock());	1,032✔
106	$this->logger = $logger ?? new NullLogger();	1,032✔
107
108	$this->executor = $executor ?? new BasicJobExecutor(	1,032✔
109	$this->clock,	1,032✔
110	fn ($id, JobSchedule $jobSchedule, int $runSecond): array => $this->runInternal(	1,032✔
111	$id,	1,032✔
112	$jobSchedule,	1,032✔
113	new RunParameters($runSecond, false),	1,032✔
114	true,	1,032✔
115	false,	1,032✔
116	),	1,032✔
117	);	1,032✔
118
119	$this->maintenanceManager = $maintenanceManager;	1,032✔
120	$this->runRegistry = $runRegistry;	1,032✔
121	}
122
123	public function getStatus(): ActivityStatus
124	{
125	$maintenance = $this->maintenanceManager !== null	24✔
126	? $this->maintenanceManager->isMaintenance()	8✔
127	: null;	18✔
128
129	$activeRuns = $this->runRegistry !== null	24✔
130	? $this->runRegistry->getActiveRuns()	16✔
131	: [];	12✔
132
133	return new ActivityStatus($maintenance, $activeRuns);	24✔
134	}
135
136	public function getJobSchedules(): array
137	{
138	return $this->jobManager->getJobSchedules();	112✔
139	}
140
141	public function runJob(
142	$id,
143	bool $force = true,
144	?RunParameters $parameters = null,
145	?Closure $onJobStarted = null,
146	?Closure $onJobFinished = null
147	): ?JobSummary
148	{
149	$this->releaseMinuteLocks();	200✔
150
151	$jobSchedule = $this->jobManager->getJobSchedule($id);	200✔
152	// Explicit RunParameters signals subprocess context — parent's runPromise() handles
153	// all callback firing, so runInternal must suppress to prevent double-firing.
154	$fireCallbacks = $parameters === null;	200✔
155	$parameters ??= new RunParameters(0, true);	200✔
156
157	if ($jobSchedule === null) {	200✔
158	$message = Message::create()	16✔
159	->withContext("Running job with ID '$id'")	16✔
160	->withProblem('Job is not registered by scheduler.')	16✔
161	->with(	16✔
162	'Tip',	16✔
163	"Inspect keys in 'Scheduler->getJobSchedules()' or run command 'scheduler:list' to find correct job ID.",	16✔
164	);	16✔
165
166	throw InvalidArgument::create()	16✔
167	->withMessage($message);	16✔
168	}
169
170	$expression = $jobSchedule->getExpression();	184✔
171
172	$timeZone = $jobSchedule->getTimeZone();	184✔
173	$jobDueTime = $timeZone !== null	184✔
174	? $this->clock->now()->setTimezone($timeZone)	8✔
175	: $this->clock->now();	178✔
176
177	// Intentionally ignores repeat after seconds
178	if (!$force && !$expression->isDue($jobDueTime)) {	184✔
179	return null;	32✔
180	}
181
182	if (!$force && $this->maintenanceManager !== null && $this->maintenanceManager->isMaintenance()) {	184✔
183	return null;	8✔
184	}
185
186	try {
187	[$summary, $throwable] = $this->runInternal(	184✔
188	$id,	184✔
189	$jobSchedule,	184✔
190	$parameters,	184✔
191	$fireCallbacks,	184✔
192	$fireCallbacks,	184✔
193	$onJobStarted,	184✔
194	$onJobFinished,	184✔
195	);	184✔
196	} finally {
197	$this->releaseMinuteLocks();	184✔
198	}
199
200	if ($throwable !== null) {	184✔
201	throw JobFailure::create($summary, $throwable);	8✔
202	}
203
204	return $summary;	176✔
205	}
206
207	/**
208	* @param array<int\|string, JobSchedule> $jobSchedules
209	* @return array<int, array<int\|string, JobSchedule>>
210	*/
211	private function groupJobSchedulesBySecond(array $jobSchedules): array
212	{
213	$scheduledJobsBySecond = [];	520✔
214	foreach ($jobSchedules as $id => $jobSchedule) {	520✔
215	$repeatAfterSeconds = $jobSchedule->getRepeatAfterSeconds();	448✔
216
217	if ($repeatAfterSeconds === 0) {	448✔
218	$scheduledJobsBySecond[0][$id] = $jobSchedule;	424✔
219	} else {
220	for ($second = 0; $second <= 59; $second += $repeatAfterSeconds) {	80✔
221	$scheduledJobsBySecond[$second][$id] = $jobSchedule;	80✔
222	}
223	}
224	}
225
226	return $scheduledJobsBySecond;	520✔
227	}
228
229	public function runPromise(): Generator
230	{
231	$this->releaseMinuteLocks();	616✔
232
233	$runStart = $this->clock->now();	616✔
234	$runId = time() . '-' . bin2hex(random_bytes(3));	616✔
235
236	if ($this->runRegistry !== null) {	616✔
237	$pid = getmypid();	200✔
238	$this->runRegistry->register($runId, $pid !== false ? $pid : 0);	200✔
239	}
240
241	try {
242	$jobSchedules = [];	616✔
243	foreach ($this->jobManager->getJobSchedules() as $id => $schedule) {	616✔
244	$timeZone = $schedule->getTimeZone();	536✔
245	$jobDueTime = $timeZone !== null	536✔
246	? $runStart->setTimezone($timeZone)	48✔
247	: $runStart;	512✔
248
249	if ($schedule->getExpression()->isDue($jobDueTime)) {	536✔
250	$jobSchedules[$id] = $schedule;	536✔
251	}
252	}
253
254	// Check maintenance before starting any jobs
255	if ($this->maintenanceManager !== null && $this->maintenanceManager->isMaintenance()) {	616✔
256	$generator = $this->createMaintenanceRunSummary($runStart, $jobSchedules);	96✔
257
258	yield from $this->wrapGeneratorWithJobCallbacks($generator);	96✔
259
260	return $generator->getReturn();	96✔
261	}
262
263	$shutdownCheck = $this->createShutdownCheck($runId);	520✔
264
265	$generator = $this->executor->runJobs(	520✔
266	$this->groupJobSchedulesBySecond($jobSchedules),	520✔
267	$runStart,	520✔
268	$this->getBeforeRunCallback($runStart, $jobSchedules),	520✔
269	$this->getAfterRunCallback(),	520✔
270	$shutdownCheck,	520✔
271	$this->getOnJobEventCallback(),	520✔
272	);	520✔
273
274	yield from $this->wrapGeneratorWithJobCallbacks($generator);	520✔
275
276	return $generator->getReturn();	464✔
277	} finally {
278	if ($this->runRegistry !== null) {	616✔
279	$this->runRegistry->deregister($runId);	200✔
280	}
281
282	$this->releaseMinuteLocks();	616✔
283	}
284	}
285
286	private function releaseMinuteLocks(): void
287	{
288	$now = (float) $this->clock->now()->format('U.u');	736✔
289	$remaining = [];	736✔
290
291	foreach ($this->minuteLocks as [$lock, $createdAt]) {	736✔
292	if ($now - $createdAt >= self::MinuteLockTtl) {	280✔
293	$lock->release();	136✔
294	} else {
295	$remaining[] = [$lock, $createdAt];	280✔
296	}
297	}
298
299	$this->minuteLocks = $remaining;	736✔
300	}
301
302	private function createShutdownCheck(string $runId): ?ShutdownCheck
303	{
304	if ($this->maintenanceManager === null) {	520✔
305	return null;	416✔
306	}
307
308	$manager = $this->maintenanceManager;	104✔
309	$registry = $this->runRegistry;	104✔
310
311	return new ShutdownCheck(	104✔
312	static fn (): bool => $manager->isShutdownRequested() \|\| $manager->isMaintenance(),	104✔
313	$manager->getGracePeriodSeconds(),	104✔
314	static function () use ($registry, $runId): void {	104✔
315	if ($registry !== null) {	24✔
316	$registry->refresh($runId);	24✔
317	}
318	},	104✔
319	);	104✔
320	}
321
322	/**
323	* @param array<int\|string, JobSchedule> $jobSchedules
324	* @return Generator<int, JobSummary, void, RunSummary>
325	*/
326	private function createMaintenanceRunSummary(DateTimeImmutable $runStart, array $jobSchedules): Generator
327	{
328	$beforeRunCb = $this->getBeforeRunCallback($runStart, $jobSchedules);	96✔
329	$beforeRunCb();	96✔
330
331	$jobSummaries = [];	96✔
332	foreach ($jobSchedules as $id => $jobSchedule) {	96✔
333	yield $jobSummaries[] = $this->createMaintenanceJobSummary($id, $jobSchedule, 0, $runStart);	88✔
334	}
335
336	$runSummary = new RunSummary($runStart, $this->clock->now(), $jobSummaries, true);	96✔
337
338	foreach ($this->afterRunCallbacks as $cb) {	96✔
339	$cb($runSummary);	16✔
340	}
341
342	return $runSummary;	96✔
343	}
344
345	/**
346	* @param array<int\|string, JobSchedule> $jobSchedules
347	* @return Closure(): void
348	*/
349	private function getBeforeRunCallback(DateTimeImmutable $runStart, array $jobSchedules): Closure
350	{
351	return function () use ($runStart, $jobSchedules): void {	616✔
352	if ($this->beforeRunCallbacks === []) {	616✔
353	return;	576✔
354	}
355
356	$jobInfos = [];	40✔
357	foreach ($jobSchedules as $id => $jobSchedule) {	40✔
358	$job = $jobSchedule->getJob();	32✔
359	$timezone = $jobSchedule->getTimeZone();	32✔
360	$jobStart = $timezone !== null	32✔
361	? $runStart->setTimezone($timezone)	8✔
362	: $runStart;	32✔
363	$jobInfos[] = new PlannedJobInfo(	32✔
364	$id,	32✔
365	$job->getName(),	32✔
366	$jobSchedule->getExpression()->getExpression(),	32✔
367	$jobSchedule->getRepeatAfterSeconds(),	32✔
368	$jobStart,	32✔
369	$timezone,	32✔
370	);	32✔
371	}
372
373	$info = new RunInfo($runStart, $jobInfos);	40✔
374
375	foreach ($this->beforeRunCallbacks as $cb) {	40✔
376	$cb($info);	40✔
377	}
378	};	616✔
379	}
380
381	/**
382	* @return Closure(RunSummary): void
383	*/
384	private function getAfterRunCallback(): Closure
385	{
386	return function (RunSummary $runSummary): void {	520✔
387	foreach ($this->afterRunCallbacks as $cb) {	520✔
388	$cb($runSummary);	32✔
389	}
390	};	520✔
391	}
392
393	/**
394	* Called by ProcessJobExecutor when the subprocess emits a "started" event.
395	* Fires parent's beforeJobCallbacks with the JobInfo from the subprocess.
396	*
397	* @return Closure(int\|string, JobSchedule, int<0, max>, JobInfo): void
398	*/
399	private function getOnJobEventCallback(): Closure
400	{
401	return function ($id, JobSchedule $jobSchedule, int $runSecond, JobInfo $info): void {	520✔
402	foreach ($this->beforeJobCallbacks as $cb) {	112✔
403	$cb($info);	32✔
404	}
405	};	520✔
406	}
407
408	/**
409	* @param Generator<int, JobSummary, void, RunSummary> $generator
410	* @return Generator<int, JobSummary, void, void>
411	*/
412	private function wrapGeneratorWithJobCallbacks(Generator $generator): Generator
413	{
414	foreach ($generator as $summary) {	616✔
415	if ($summary->getResult()->getState() === JobResultState::lock()) {	520✔
416	foreach ($this->lockedJobCallbacks as $cb) {	64✔
417	$cb($summary->getInfo(), $summary->getResult());	8✔
418	}
419	}
420
421	foreach ($this->afterJobCallbacks as $cb) {	520✔
422	$cb($summary->getInfo(), $summary->getResult());	112✔
423	}
424
425	yield $summary;	520✔
426	}
427	}
428
429	public function run(): RunSummary
430	{
431	$generator = $this->runPromise();	528✔
432	// Forces generator to execute
433	iterator_to_array($generator);	528✔
434
435	return $generator->getReturn();	472✔
436	}
437
438	/**
439	* @param string\|int $id
440	* @param (Closure(JobInfo): void)\|null $onJobStarted
441	* @param (Closure(JobInfo, JobResult): void)\|null $onJobFinished
442	* @return array{JobSummary, Throwable\|null}
443	*/
444	private function runInternal(
445	$id,
446	JobSchedule $jobSchedule,
447	RunParameters $runParameters,
448	bool $fireBeforeJobCallbacks = true,
449	bool $fireAfterJobCallbacks = true,
450	?Closure $onJobStarted = null,
451	?Closure $onJobFinished = null
452	): array
453	{
454	$job = $jobSchedule->getJob();	384✔
455	$expression = $jobSchedule->getExpression();	384✔
456
457	$info = new JobInfo(	384✔
458	$id,	384✔
459	$job->getName(),	384✔
460	$expression->getExpression(),	384✔
461	$jobSchedule->getRepeatAfterSeconds(),	384✔
462	$runParameters->getSecond(),	384✔
463	$this->getCurrentTime($jobSchedule),	384✔
464	$jobSchedule->getTimeZone(),	384✔
465	$runParameters->isForcedRun(),	384✔
466	);	384✔
467
468	$repeatAfterSeconds = $jobSchedule->getRepeatAfterSeconds();	384✔
469	$runSecond = $runParameters->getSecond();	384✔
470
471	// Minute lock prevents re-execution within the same minute on another server.
472	// 30-second TTL with autoRelease=false — survives subprocess exit, expires naturally.
473	// Skipped for forced (manual) runs — manual execution should always work.
474	$minuteLock = null;	384✔
475	if (!$runParameters->isForcedRun()) {	384✔
476	$minuteLockKey = $repeatAfterSeconds > 0	304✔
477	? "Orisai.Scheduler.Job.Minute/$id/$runSecond"	32✔
478	: "Orisai.Scheduler.Job.Minute/$id";	286✔
479	$minuteLock = $this->lockFactory->createLock($minuteLockKey, self::MinuteLockTtl, false);	304✔
480
481	if (!$minuteLock->acquire()) {	304✔
482	$result = new JobResult($expression, $info->getStart(), JobResultState::lock());	32✔
483
484	if ($fireAfterJobCallbacks) {	32✔
NEW 485	foreach ($this->lockedJobCallbacks as $cb) {	×
NEW 486	$cb($info, $result);	×
487	}
488
NEW 489	foreach ($this->afterJobCallbacks as $cb) {	×
NEW 490	$cb($info, $result);	×
491	}
492	}
493
494	if ($onJobFinished !== null) {	32✔
NEW 495	$onJobFinished($info, $result);	×
496	}
497
498	return [	32✔
499	new JobSummary($info, $result),	32✔
500	null,	32✔
501	];	32✔
502	}
503	}
504
505	// Job lock prevents concurrent execution of the same job.
506	$lock = $this->lockFactory->createLock("Orisai.Scheduler.Job/$id");	384✔
507
508	if (!$lock->acquire()) {	384✔
509	if ($minuteLock !== null) {	56✔
510	$minuteLock->release();	40✔
511	}
512
513	$result = new JobResult($expression, $info->getStart(), JobResultState::lock());	56✔
514
515	if ($fireAfterJobCallbacks) {	56✔
516	foreach ($this->lockedJobCallbacks as $cb) {	24✔
517	$cb($info, $result);	8✔
518	}
519
520	foreach ($this->afterJobCallbacks as $cb) {	24✔
NEW 521	$cb($info, $result);	×
522	}
523	}
524
525	if ($onJobFinished !== null) {	56✔
NEW 526	$onJobFinished($info, $result);	×
527	}
528
529	return [	56✔
530	new JobSummary($info, $result),	56✔
531	null,	56✔
532	];	56✔
533	}
534
535	$throwable = null;	352✔
536	try {
537	if ($fireBeforeJobCallbacks) {	352✔
538	foreach ($this->beforeJobCallbacks as $cb) {	312✔
539	$cb($info);	56✔
540	}
541	}
542
543	if ($onJobStarted !== null) {	352✔
NEW 544	$onJobStarted($info);	×
545	}
546
547	try {
548	$job->run(new JobLock($lock));	352✔
549	} catch (Throwable $throwable) {	56✔
550	// Handled bellow
551	}
552
553	$lockExpired = $lock->isExpired();	352✔
554	if ($lockExpired) {	352✔
555	$this->logger->warning("Lock of job '$id' expired before the job finished.", [	8✔
556	'id' => $id,	8✔
557	]);	8✔
558	}
559
560	$result = new JobResult(	352✔
561	$expression,	352✔
562	$this->getCurrentTime($jobSchedule),	352✔
563	$throwable === null ? JobResultState::done() : JobResultState::fail(),	352✔
564	$lockExpired,	352✔
565	);	352✔
566
567	if ($fireAfterJobCallbacks) {	352✔
568	foreach ($this->afterJobCallbacks as $cb) {	136✔
569	$cb($info, $result);	16✔
570	}
571	}
572
573	if ($onJobFinished !== null) {	352✔
NEW 574	$onJobFinished($info, $result);	×
575	}
576
577	if ($throwable !== null && $this->errorHandler !== null) {	352✔
578	($this->errorHandler)($throwable, $info, $result);	40✔
579	$throwable = null;	40✔
580	}
581	} finally {
582	$lock->release();	352✔
583	// Minute lock NOT released — stays in store until 30s TTL expires.
584	// Stored to prevent GC and released at end of run via releaseMinuteLocks().
585	if ($minuteLock !== null) {	352✔
586	$this->minuteLocks[] = [$minuteLock, (float) $this->clock->now()->format('U.u')];	280✔
587	}
588	}
589
590	return [	352✔
591	new JobSummary($info, $result),	352✔
592	$throwable,	352✔
593	];	352✔
594	}
595
596	private function getCurrentTime(JobSchedule $schedule): DateTimeImmutable
597	{
598	$now = $this->clock->now();	384✔
599	$timezone = $schedule->getTimeZone();	384✔
600
601	return $timezone !== null	384✔
602	? $now->setTimezone($timezone)	32✔
603	: $now;	384✔
604	}
605
606	/**
607	* @param Closure(JobInfo, JobResult): void $callback
608	* @param-later-invoked-callable $callback
609	*/
610	public function addLockedJobCallback(Closure $callback): void
611	{
612	$this->lockedJobCallbacks[] = $callback;	16✔
613	}
614
615	/**
616	* @param Closure(JobInfo): void $callback
617	* @param-later-invoked-callable $callback
618	*/
619	public function addBeforeJobCallback(Closure $callback): void
620	{
621	$this->beforeJobCallbacks[] = $callback;	104✔
622	}
623
624	/**
625	* @param Closure(JobInfo, JobResult): void $callback
626	* @param-later-invoked-callable $callback
627	*/
628	public function addAfterJobCallback(Closure $callback): void
629	{
630	$this->afterJobCallbacks[] = $callback;	120✔
631	}
632
633	/**
634	* @param Closure(RunInfo): void $callback
635	* @param-later-invoked-callable $callback
636	*/
637	public function addBeforeRunCallback(Closure $callback): void
638	{
639	$this->beforeRunCallbacks[] = $callback;	40✔
640	}
641
642	/**
643	* @param Closure(RunSummary): void $callback
644	* @param-later-invoked-callable $callback
645	*/
646	public function addAfterRunCallback(Closure $callback): void
647	{
648	$this->afterRunCallbacks[] = $callback;	48✔
649	}
650
651	}

orisai / scheduler / 24634959708

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