forked from fa/breadcrumb-the-shire
436 lines
19 KiB
PHP
436 lines
19 KiB
PHP
<?php
|
|
|
|
namespace MintyPHP\Service\Scheduler;
|
|
|
|
use MintyPHP\Domain\Taxonomy\ScheduledJobRunStatus;
|
|
use MintyPHP\Domain\Taxonomy\ScheduledJobTriggerType;
|
|
use MintyPHP\Domain\Taxonomy\SchedulerRuntimeResult;
|
|
use MintyPHP\Domain\Taxonomy\SystemAuditOutcome;
|
|
use MintyPHP\Repository\Scheduler\ScheduledJobRepositoryInterface;
|
|
use MintyPHP\Repository\Scheduler\ScheduledJobRunRepositoryInterface;
|
|
use MintyPHP\Repository\Scheduler\SchedulerRuntimeRepositoryInterface;
|
|
use MintyPHP\Repository\Support\DatabaseSessionRepository;
|
|
use MintyPHP\Repository\Support\RepoQuery;
|
|
use MintyPHP\Service\Audit\SystemAuditService;
|
|
|
|
/**
|
|
* 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 ScheduledJobRepositoryInterface $scheduledJobRepository,
|
|
private readonly ScheduledJobRunRepositoryInterface $scheduledJobRunRepository,
|
|
private readonly SchedulerRuntimeRepositoryInterface $schedulerRuntimeRepository,
|
|
private readonly ScheduledJobRegistry $scheduledJobRegistry,
|
|
private readonly ScheduleCalculator $scheduleCalculator,
|
|
private readonly DatabaseSessionRepository $databaseSessionRepository,
|
|
private readonly SystemAuditService $systemAuditService
|
|
) {
|
|
}
|
|
|
|
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'] = SchedulerRuntimeResult::LockNotAcquired->value;
|
|
$result['duration_ms'] = $this->durationMs($startedAt);
|
|
$this->updateRuntimeHeartbeat(
|
|
SchedulerRuntimeResult::LockNotAcquired->value,
|
|
SchedulerRuntimeResult::LockNotAcquired->value
|
|
);
|
|
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Scheduler->value, $result);
|
|
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, ScheduledJobTriggerType::Scheduler->value, null, false);
|
|
$result['processed']++;
|
|
$status = ScheduledJobRunStatus::normalizeOr(
|
|
(string) ($run['status'] ?? ''),
|
|
ScheduledJobRunStatus::Failed
|
|
)->value;
|
|
if ($status === ScheduledJobRunStatus::Success->value) {
|
|
$result['success']++;
|
|
} elseif ($status === ScheduledJobRunStatus::Skipped->value) {
|
|
$result['skipped']++;
|
|
} else {
|
|
$result['failed']++;
|
|
}
|
|
}
|
|
} catch (\Throwable $exception) {
|
|
$result['ok'] = false;
|
|
$result['error'] = SchedulerRuntimeResult::UnexpectedError->value;
|
|
} finally {
|
|
$this->releaseRunnerLock();
|
|
$result['duration_ms'] = $this->durationMs($startedAt);
|
|
if ($result['ok']) {
|
|
$this->updateRuntimeHeartbeat(SchedulerRuntimeResult::Ok->value, null);
|
|
} else {
|
|
$errorCode = $this->nullableString($result['error']) ?? SchedulerRuntimeResult::UnexpectedError->value;
|
|
$this->updateRuntimeHeartbeat(SchedulerRuntimeResult::UnexpectedError->value, $errorCode);
|
|
}
|
|
}
|
|
|
|
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Scheduler->value, $result);
|
|
|
|
return $result;
|
|
}
|
|
|
|
public function runJobNow(int $jobId, int $actorUserId): array
|
|
{
|
|
$this->scheduledJobService->ensureSystemJobs();
|
|
if ($jobId <= 0 || $actorUserId <= 0) {
|
|
$result = ['ok' => false, 'error' => 'invalid_request'];
|
|
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId);
|
|
return $result;
|
|
}
|
|
if (!$this->acquireRunnerLock()) {
|
|
$result = ['ok' => false, 'error' => SchedulerRuntimeResult::LockNotAcquired->value];
|
|
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId);
|
|
return $result;
|
|
}
|
|
|
|
try {
|
|
$job = $this->scheduledJobRepository->find($jobId);
|
|
if (!$job) {
|
|
$result = ['ok' => false, 'error' => 'job_not_found'];
|
|
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId);
|
|
return $result;
|
|
}
|
|
$run = $this->runOne($job, ScheduledJobTriggerType::Manual->value, $actorUserId, true);
|
|
$runStatus = ScheduledJobRunStatus::normalizeOr(
|
|
(string) ($run['status'] ?? ''),
|
|
ScheduledJobRunStatus::Failed
|
|
)->value;
|
|
$result = [
|
|
'ok' => in_array($runStatus, [ScheduledJobRunStatus::Success->value, ScheduledJobRunStatus::Skipped->value], true),
|
|
'status' => $runStatus,
|
|
'error' => (string) ($run['error_code'] ?? ''),
|
|
];
|
|
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId, [
|
|
'processed' => 1,
|
|
'success' => (int) ($result['status'] === ScheduledJobRunStatus::Success->value),
|
|
'failed' => (int) ($result['status'] === ScheduledJobRunStatus::Failed->value),
|
|
'skipped' => (int) ($result['status'] === ScheduledJobRunStatus::Skipped->value),
|
|
]);
|
|
return $result;
|
|
} 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 = ScheduledJobTriggerType::normalizeOr($triggerType, ScheduledJobTriggerType::Scheduler)->value;
|
|
if ($jobId <= 0) {
|
|
return ['status' => ScheduledJobRunStatus::Failed->value, 'error_code' => 'job_not_found', 'error_message' => null];
|
|
}
|
|
|
|
if (!$force && ((int) ($job['enabled'] ?? 0) !== 1 || empty($job['next_run_at']))) {
|
|
$run = ['status' => ScheduledJobRunStatus::Skipped->value, 'error_code' => 'job_disabled', 'error_message' => null];
|
|
$this->recordSchedulerJobEvent($jobId, (string) ($job['job_key'] ?? ''), $triggerType, $actorUserId, $run, 0);
|
|
return $run;
|
|
}
|
|
|
|
// markRunning uses a conditional update — failure means another process already claimed this job.
|
|
$markedRunning = $this->scheduledJobRepository->markRunning($jobId, $startedAtUtc);
|
|
if (!$markedRunning) {
|
|
$this->insertRunLog([
|
|
'job_id' => $jobId,
|
|
'job_key' => (string) ($job['job_key'] ?? ''),
|
|
'trigger_type' => $triggerType,
|
|
'status' => ScheduledJobRunStatus::Skipped->value,
|
|
'actor_user_id' => $actorUserId,
|
|
'started_at' => $startedAtUtc,
|
|
'finished_at' => $startedAtUtc,
|
|
'duration_ms' => 0,
|
|
'error_code' => 'job_already_running',
|
|
'error_message' => null,
|
|
'result_json' => null,
|
|
]);
|
|
$run = ['status' => ScheduledJobRunStatus::Skipped->value, 'error_code' => 'job_already_running', 'error_message' => null];
|
|
$this->recordSchedulerJobEvent($jobId, (string) ($job['job_key'] ?? ''), $triggerType, $actorUserId, $run, 0);
|
|
return $run;
|
|
}
|
|
|
|
$startedMs = microtime(true);
|
|
$status = ScheduledJobRunStatus::Failed->value;
|
|
$errorCode = null;
|
|
$errorMessage = null;
|
|
$resultPayload = [];
|
|
|
|
try {
|
|
$definition = $this->scheduledJobRegistry->get((string) ($job['job_key'] ?? ''));
|
|
if (!$definition) {
|
|
$status = ScheduledJobRunStatus::Failed->value;
|
|
$errorCode = 'job_not_registered';
|
|
} else {
|
|
$execution = $this->scheduledJobRegistry->execute((string) $job['job_key'], $actorUserId);
|
|
$status = ScheduledJobRunStatus::tryNormalize((string) $execution['status'])?->value
|
|
?? ScheduledJobRunStatus::Failed->value;
|
|
$errorCode = $this->nullableString($execution['error_code']);
|
|
$errorMessage = $this->nullableString($execution['error_message']);
|
|
$resultPayload = $execution['result'];
|
|
}
|
|
} catch (\Throwable $exception) {
|
|
$status = ScheduledJobRunStatus::Failed->value;
|
|
$errorCode = SchedulerRuntimeResult::UnexpectedError->value;
|
|
$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 === ScheduledJobTriggerType::Scheduler->value
|
|
&& (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),
|
|
]);
|
|
|
|
$run = ['status' => $status, 'error_code' => $errorCode, 'error_message' => $errorMessage];
|
|
$this->recordSchedulerJobEvent($jobId, (string) ($job['job_key'] ?? ''), $triggerType, $actorUserId, $run, $durationMs);
|
|
|
|
return $run;
|
|
}
|
|
|
|
// Fail-open: a logging failure must not kill the job run that already completed.
|
|
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'] ?? ScheduledJobTriggerType::Scheduler->value),
|
|
'status' => (string) ($data['status'] ?? ScheduledJobRunStatus::Failed->value),
|
|
'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
|
|
{
|
|
return $this->databaseSessionRepository->acquireAdvisoryLock(self::RUNNER_LOCK_NAME, 0);
|
|
}
|
|
|
|
private function updateRuntimeHeartbeat(string $status, ?string $errorCode): void
|
|
{
|
|
try {
|
|
$this->schedulerRuntimeRepository->touchHeartbeat($status, $errorCode);
|
|
} catch (\Throwable $exception) {
|
|
// fail-open
|
|
}
|
|
}
|
|
|
|
private function releaseRunnerLock(): void
|
|
{
|
|
try {
|
|
$this->databaseSessionRepository->releaseAdvisoryLock(self::RUNNER_LOCK_NAME);
|
|
} catch (\Throwable $exception) {
|
|
// no-op
|
|
}
|
|
}
|
|
|
|
private function durationMs(float $startedAt): int
|
|
{
|
|
return (int) max(0, round((microtime(true) - $startedAt) * 1000));
|
|
}
|
|
|
|
private function recordSchedulerRunEvent(string $triggerType, array $result, ?int $actorUserId = null, array $counts = []): void
|
|
{
|
|
$status = ScheduledJobRunStatus::tryNormalize((string) ($result['status'] ?? ''))?->value;
|
|
if ($status === null) {
|
|
if (!($result['ok'] ?? false)) {
|
|
$status = ScheduledJobRunStatus::Failed->value;
|
|
} elseif ((int) ($counts['failed'] ?? $result['failed'] ?? 0) > 0) {
|
|
$status = ScheduledJobRunStatus::Failed->value;
|
|
} elseif ((int) ($counts['success'] ?? $result['success'] ?? 0) > 0) {
|
|
$status = ScheduledJobRunStatus::Success->value;
|
|
} else {
|
|
$status = ScheduledJobRunStatus::Skipped->value;
|
|
}
|
|
}
|
|
|
|
$this->recordAudit(
|
|
'scheduler.run',
|
|
$this->auditOutcomeForRunStatus($status),
|
|
[
|
|
'actor_user_id' => ($actorUserId ?? 0) > 0 ? (int) $actorUserId : null,
|
|
'error_code' => $this->nullableString($result['error'] ?? null),
|
|
'metadata' => [
|
|
'trigger_type' => ScheduledJobTriggerType::normalizeOr($triggerType, ScheduledJobTriggerType::Scheduler)->value,
|
|
'run_status' => $status,
|
|
'processed' => (int) ($counts['processed'] ?? $result['processed'] ?? 0),
|
|
'success_count' => (int) ($counts['success'] ?? $result['success'] ?? 0),
|
|
'failed_count' => (int) ($counts['failed'] ?? $result['failed'] ?? 0),
|
|
'skipped_count' => (int) ($counts['skipped'] ?? $result['skipped'] ?? 0),
|
|
'duration_ms' => (int) ($result['duration_ms'] ?? 0),
|
|
],
|
|
]
|
|
);
|
|
}
|
|
|
|
private function recordSchedulerJobEvent(
|
|
int $jobId,
|
|
string $jobKey,
|
|
string $triggerType,
|
|
?int $actorUserId,
|
|
array $runResult,
|
|
int $durationMs
|
|
): void {
|
|
$status = ScheduledJobRunStatus::normalizeOr((string) ($runResult['status'] ?? ''), ScheduledJobRunStatus::Failed)->value;
|
|
|
|
$this->recordAudit(
|
|
'scheduler.job.run',
|
|
$this->auditOutcomeForRunStatus($status),
|
|
[
|
|
'actor_user_id' => ($actorUserId ?? 0) > 0 ? (int) $actorUserId : null,
|
|
'target_type' => 'scheduled_job',
|
|
'target_id' => $jobId > 0 ? $jobId : null,
|
|
'error_code' => $this->nullableString($runResult['error_code'] ?? null),
|
|
'metadata' => [
|
|
'job_id' => $jobId,
|
|
'job_key' => $jobKey,
|
|
'trigger_type' => ScheduledJobTriggerType::normalizeOr($triggerType, ScheduledJobTriggerType::Scheduler)->value,
|
|
'run_status' => $status,
|
|
'duration_ms' => max(0, $durationMs),
|
|
],
|
|
]
|
|
);
|
|
}
|
|
|
|
private function recordAudit(string $eventType, string $outcome, array $context): void
|
|
{
|
|
try {
|
|
$this->systemAuditService->record($eventType, $outcome, $context);
|
|
} catch (\Throwable) {
|
|
// fail-open
|
|
}
|
|
}
|
|
|
|
private function auditOutcomeForRunStatus(string $status): string
|
|
{
|
|
if ($status === ScheduledJobRunStatus::Success->value) {
|
|
return SystemAuditOutcome::Success->value;
|
|
}
|
|
if ($status === ScheduledJobRunStatus::Failed->value) {
|
|
return SystemAuditOutcome::Failed->value;
|
|
}
|
|
// "skipped" is an operational non-error state (e.g. already running),
|
|
// not an authorization denial.
|
|
return SystemAuditOutcome::Success->value;
|
|
}
|
|
}
|