307 lines
12 KiB
PHP
307 lines
12 KiB
PHP
<?php
|
|
|
|
namespace MintyPHP\Service\Scheduler;
|
|
|
|
use MintyPHP\DB;
|
|
use MintyPHP\Repository\Scheduler\ScheduledJobRepository;
|
|
use MintyPHP\Repository\Scheduler\ScheduledJobRunRepository;
|
|
use MintyPHP\Repository\Scheduler\SchedulerRuntimeRepository;
|
|
use MintyPHP\Repository\Support\RepoQuery;
|
|
|
|
/**
|
|
* Orchestrates the execution of scheduled jobs.
|
|
*
|
|
* Acquires a MySQL advisory lock before any work to ensure only one runner
|
|
* process is active at a time (safe for minute-by-minute cron invocations).
|
|
* Individual jobs are dispatched via ScheduledJobRegistry, and every execution
|
|
* attempt - including skips and failures - is written to the run log.
|
|
*/
|
|
class SchedulerRunService
|
|
{
|
|
private const RUNNER_LOCK_NAME = 'scheduled_jobs_runner';
|
|
|
|
public function __construct(
|
|
private readonly ScheduledJobService $scheduledJobService,
|
|
private readonly ScheduledJobRepository $scheduledJobRepository,
|
|
private readonly ScheduledJobRunRepository $scheduledJobRunRepository,
|
|
private readonly SchedulerRuntimeRepository $schedulerRuntimeRepository,
|
|
private readonly ScheduledJobRegistry $scheduledJobRegistry,
|
|
private readonly ScheduleCalculator $scheduleCalculator
|
|
) {
|
|
}
|
|
|
|
public function runDueJobs(int $limit = 20): array
|
|
{
|
|
$this->scheduledJobService->ensureSystemJobs();
|
|
$startedAt = microtime(true);
|
|
$result = [
|
|
'ok' => true,
|
|
'error' => null,
|
|
'duration_ms' => 0,
|
|
'processed' => 0,
|
|
'success' => 0,
|
|
'failed' => 0,
|
|
'skipped' => 0,
|
|
];
|
|
|
|
if (!$this->acquireRunnerLock()) {
|
|
$result['ok'] = false;
|
|
$result['error'] = 'lock_not_acquired';
|
|
$result['duration_ms'] = $this->durationMs($startedAt);
|
|
$this->updateRuntimeHeartbeat('lock_not_acquired', 'lock_not_acquired');
|
|
return $result;
|
|
}
|
|
|
|
try {
|
|
$nowUtc = gmdate('Y-m-d H:i:s');
|
|
$dueJobs = $this->scheduledJobRepository->listDueJobs($nowUtc, $limit);
|
|
foreach ($dueJobs as $job) {
|
|
$run = $this->runOne($job, 'scheduler', null, false);
|
|
$result['processed']++;
|
|
$status = (string) ($run['status'] ?? 'failed');
|
|
if ($status === 'success') {
|
|
$result['success']++;
|
|
} elseif ($status === 'skipped') {
|
|
$result['skipped']++;
|
|
} else {
|
|
$result['failed']++;
|
|
}
|
|
}
|
|
} catch (\Throwable $exception) {
|
|
$result['ok'] = false;
|
|
$result['error'] = 'unexpected_error';
|
|
} finally {
|
|
$this->releaseRunnerLock();
|
|
$result['duration_ms'] = $this->durationMs($startedAt);
|
|
if ($result['ok']) {
|
|
$this->updateRuntimeHeartbeat('ok', null);
|
|
} else {
|
|
$errorCode = $this->nullableString($result['error']) ?? 'unexpected_error';
|
|
$this->updateRuntimeHeartbeat('unexpected_error', $errorCode);
|
|
}
|
|
}
|
|
|
|
return $result;
|
|
}
|
|
|
|
public function runJobNow(int $jobId, int $actorUserId): array
|
|
{
|
|
$this->scheduledJobService->ensureSystemJobs();
|
|
if ($jobId <= 0 || $actorUserId <= 0) {
|
|
return ['ok' => false, 'error' => 'invalid_request'];
|
|
}
|
|
if (!$this->acquireRunnerLock()) {
|
|
return ['ok' => false, 'error' => 'lock_not_acquired'];
|
|
}
|
|
|
|
try {
|
|
$job = $this->scheduledJobRepository->find($jobId);
|
|
if (!$job) {
|
|
return ['ok' => false, 'error' => 'job_not_found'];
|
|
}
|
|
$run = $this->runOne($job, 'manual', $actorUserId, true);
|
|
return [
|
|
'ok' => in_array((string) ($run['status'] ?? ''), ['success', 'skipped'], true),
|
|
'status' => (string) ($run['status'] ?? 'failed'),
|
|
'error' => (string) ($run['error_code'] ?? ''),
|
|
];
|
|
} finally {
|
|
$this->releaseRunnerLock();
|
|
}
|
|
}
|
|
|
|
private function runOne(array $job, string $triggerType, ?int $actorUserId, bool $force): array
|
|
{
|
|
$jobId = (int) ($job['id'] ?? 0);
|
|
$nowUtc = new \DateTimeImmutable('now', new \DateTimeZone('UTC'));
|
|
$startedAtUtc = $nowUtc->format('Y-m-d H:i:s');
|
|
|
|
$triggerType = in_array($triggerType, ['scheduler', 'manual'], true) ? $triggerType : 'scheduler';
|
|
if ($jobId <= 0) {
|
|
return ['status' => 'failed', 'error_code' => 'job_not_found', 'error_message' => null];
|
|
}
|
|
|
|
if (!$force && ((int) ($job['enabled'] ?? 0) !== 1 || empty($job['next_run_at']))) {
|
|
return ['status' => 'skipped', 'error_code' => 'job_disabled', 'error_message' => null];
|
|
}
|
|
|
|
$markedRunning = $this->scheduledJobRepository->markRunning($jobId, $startedAtUtc);
|
|
if (!$markedRunning) {
|
|
$this->insertRunLog([
|
|
'job_id' => $jobId,
|
|
'job_key' => (string) ($job['job_key'] ?? ''),
|
|
'trigger_type' => $triggerType,
|
|
'status' => 'skipped',
|
|
'actor_user_id' => $actorUserId,
|
|
'started_at' => $startedAtUtc,
|
|
'finished_at' => $startedAtUtc,
|
|
'duration_ms' => 0,
|
|
'error_code' => 'job_already_running',
|
|
'error_message' => null,
|
|
'result_json' => null,
|
|
]);
|
|
return ['status' => 'skipped', 'error_code' => 'job_already_running', 'error_message' => null];
|
|
}
|
|
|
|
$startedMs = microtime(true);
|
|
$status = 'failed';
|
|
$errorCode = null;
|
|
$errorMessage = null;
|
|
$resultPayload = [];
|
|
|
|
try {
|
|
$definition = $this->scheduledJobRegistry->get((string) ($job['job_key'] ?? ''));
|
|
if (!$definition) {
|
|
$status = 'failed';
|
|
$errorCode = 'job_not_registered';
|
|
} else {
|
|
$execution = $this->scheduledJobRegistry->execute((string) $job['job_key'], $actorUserId);
|
|
$status = in_array((string) $execution['status'], ['success', 'failed', 'skipped'], true)
|
|
? (string) $execution['status']
|
|
: 'failed';
|
|
$errorCode = $this->nullableString($execution['error_code']);
|
|
$errorMessage = $this->nullableString($execution['error_message']);
|
|
$resultPayload = $execution['result'];
|
|
}
|
|
} catch (\Throwable $exception) {
|
|
$status = 'failed';
|
|
$errorCode = 'unexpected_error';
|
|
$errorMessage = substr($exception->getMessage(), 0, 255) ?: null;
|
|
}
|
|
|
|
$finishedAt = new \DateTimeImmutable('now', new \DateTimeZone('UTC'));
|
|
$finishedAtUtc = $finishedAt->format('Y-m-d H:i:s');
|
|
$durationMs = max(0, (int) round((microtime(true) - $startedMs) * 1000));
|
|
|
|
$nextRunUtc = null;
|
|
if ((int) ($job['enabled'] ?? 0) === 1) {
|
|
if ($triggerType === 'scheduler' && (int) ($job['catchup_once'] ?? 1) === 0 && !empty($job['next_run_at'])) {
|
|
// catchup_once = 0: advance next_run_at strictly from the originally scheduled
|
|
// time, skipping any windows that already passed. This prevents a backlog storm
|
|
// after downtime - missed slots are dropped, only the next future slot is kept.
|
|
// The guard cap (500) prevents an infinite loop if calculateNextRunUtc has a bug
|
|
// and keeps returning a time that is still in the past.
|
|
$reference = $this->parseUtc((string) $job['next_run_at']) ?? $finishedAt;
|
|
$next = $this->scheduleCalculator->calculateNextRunUtc($job, $reference);
|
|
$guard = 0;
|
|
while ($next && $next <= $finishedAt && $guard < 500) {
|
|
$next = $this->scheduleCalculator->calculateNextRunUtc($job, $next);
|
|
$guard++;
|
|
}
|
|
$nextRunUtc = $next ? $next->format('Y-m-d H:i:s') : null;
|
|
} else {
|
|
// catchup_once = 1 (default): calculate next_run_at from the actual finish time.
|
|
// If a run was delayed or a slot was missed, exactly one catch-up run is scheduled
|
|
// starting from now, then normal scheduling resumes.
|
|
$next = $this->scheduleCalculator->calculateNextRunUtc($job, $finishedAt);
|
|
$nextRunUtc = $next ? $next->format('Y-m-d H:i:s') : null;
|
|
}
|
|
}
|
|
|
|
$this->scheduledJobRepository->finishRun(
|
|
$jobId,
|
|
$status,
|
|
$startedAtUtc,
|
|
$finishedAtUtc,
|
|
$nextRunUtc,
|
|
$errorCode,
|
|
$errorMessage
|
|
);
|
|
|
|
$this->insertRunLog([
|
|
'job_id' => $jobId,
|
|
'job_key' => (string) ($job['job_key'] ?? ''),
|
|
'trigger_type' => $triggerType,
|
|
'status' => $status,
|
|
'actor_user_id' => $actorUserId,
|
|
'started_at' => $startedAtUtc,
|
|
'finished_at' => $finishedAtUtc,
|
|
'duration_ms' => $durationMs,
|
|
'error_code' => $errorCode,
|
|
'error_message' => $errorMessage,
|
|
'result_json' => $this->encodeResult($resultPayload),
|
|
]);
|
|
|
|
return ['status' => $status, 'error_code' => $errorCode, 'error_message' => $errorMessage];
|
|
}
|
|
|
|
private function insertRunLog(array $data): void
|
|
{
|
|
try {
|
|
$this->scheduledJobRunRepository->create([
|
|
'run_uuid' => RepoQuery::uuidV4(),
|
|
'job_id' => (int) ($data['job_id'] ?? 0),
|
|
'job_key' => (string) ($data['job_key'] ?? ''),
|
|
'trigger_type' => (string) ($data['trigger_type'] ?? 'scheduler'),
|
|
'status' => (string) ($data['status'] ?? 'failed'),
|
|
'actor_user_id' => ($data['actor_user_id'] ?? null) !== null ? (int) $data['actor_user_id'] : null,
|
|
'started_at' => (string) ($data['started_at'] ?? ''),
|
|
'finished_at' => $data['finished_at'] ?? null,
|
|
'duration_ms' => ($data['duration_ms'] ?? null) !== null ? (int) $data['duration_ms'] : null,
|
|
'error_code' => $data['error_code'] ?? null,
|
|
'error_message' => $data['error_message'] ?? null,
|
|
'result_json' => $data['result_json'] ?? null,
|
|
]);
|
|
} catch (\Throwable $exception) {
|
|
// fail-open
|
|
}
|
|
}
|
|
|
|
private function encodeResult(array $result): ?string
|
|
{
|
|
if (!$result) {
|
|
return null;
|
|
}
|
|
$encoded = json_encode($result, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
|
|
return is_string($encoded) ? $encoded : null;
|
|
}
|
|
|
|
private function parseUtc(string $value): ?\DateTimeImmutable
|
|
{
|
|
$value = trim($value);
|
|
if ($value === '') {
|
|
return null;
|
|
}
|
|
try {
|
|
return new \DateTimeImmutable($value, new \DateTimeZone('UTC'));
|
|
} catch (\Throwable $exception) {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
private function nullableString(mixed $value): ?string
|
|
{
|
|
$value = trim((string) $value);
|
|
return $value !== '' ? $value : null;
|
|
}
|
|
|
|
private function acquireRunnerLock(): bool
|
|
{
|
|
$got = DB::selectValue('select GET_LOCK(?, 0) as got_lock', self::RUNNER_LOCK_NAME);
|
|
return (int) $got === 1;
|
|
}
|
|
|
|
private function updateRuntimeHeartbeat(string $status, ?string $errorCode): void
|
|
{
|
|
try {
|
|
$this->schedulerRuntimeRepository->touchHeartbeat($status, $errorCode);
|
|
} catch (\Throwable $exception) {
|
|
// fail-open
|
|
}
|
|
}
|
|
|
|
private function releaseRunnerLock(): void
|
|
{
|
|
try {
|
|
DB::selectValue('select RELEASE_LOCK(?) as released_lock', self::RUNNER_LOCK_NAME);
|
|
} catch (\Throwable $exception) {
|
|
// no-op
|
|
}
|
|
}
|
|
|
|
private function durationMs(float $startedAt): int
|
|
{
|
|
return (int) max(0, round((microtime(true) - $startedAt) * 1000));
|
|
}
|
|
}
|