fa/breadcrumb-the-shire
1
0
Fork 1
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

major update

Browse Source
This commit is contained in:
Felix Schneider
2026-03-04 15:56:58 +01:00
parent 16759a2732
commit 8f4dd5840d
478 changed files with 24313 additions and 8201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
lib/Service/Scheduler/SchedulerRunService.php
View File

@@ -2,11 +2,16 @@
namespace MintyPHP\Service\Scheduler;
use MintyPHP\DB;
use MintyPHP\Domain\Taxonomy\ScheduledJobRunStatus;
use MintyPHP\Domain\Taxonomy\ScheduledJobTriggerType;
use MintyPHP\Domain\Taxonomy\SchedulerRuntimeResult;
use MintyPHP\Domain\Taxonomy\SystemAuditOutcome;
use MintyPHP\Repository\Scheduler\ScheduledJobRepository;
use MintyPHP\Repository\Scheduler\ScheduledJobRunRepository;
use MintyPHP\Repository\Scheduler\SchedulerRuntimeRepository;
use MintyPHP\Repository\Support\DatabaseSessionRepository;
use MintyPHP\Repository\Support\RepoQuery;
use MintyPHP\Service\Audit\SystemAuditService;
/**
* Orchestrates the execution of scheduled jobs.
@@ -26,7 +31,9 @@ class SchedulerRunService
private readonly ScheduledJobRunRepository $scheduledJobRunRepository,
private readonly SchedulerRuntimeRepository $schedulerRuntimeRepository,
private readonly ScheduledJobRegistry $scheduledJobRegistry,
private readonly ScheduleCalculator $scheduleCalculator
private readonly ScheduleCalculator $scheduleCalculator,
private readonly DatabaseSessionRepository $databaseSessionRepository,
private readonly SystemAuditService $systemAuditService
) {
}
@@ -46,9 +53,13 @@ class SchedulerRunService
if (!$this->acquireRunnerLock()) {
$result['ok'] = false;
$result['error'] = 'lock_not_acquired';
$result['error'] = SchedulerRuntimeResult::LockNotAcquired->value;
$result['duration_ms'] = $this->durationMs($startedAt);
$this->updateRuntimeHeartbeat('lock_not_acquired', 'lock_not_acquired');
$this->updateRuntimeHeartbeat(
SchedulerRuntimeResult::LockNotAcquired->value,
SchedulerRuntimeResult::LockNotAcquired->value
);
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Scheduler->value, $result);
return $result;
}
@@ -56,12 +67,15 @@ class SchedulerRunService
$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);
$run = $this->runOne($job, ScheduledJobTriggerType::Scheduler->value, null, false);
$result['processed']++;
$status = (string) ($run['status'] ?? 'failed');
if ($status === 'success') {
$status = ScheduledJobRunStatus::normalizeOr(
(string) ($run['status'] ?? ''),
ScheduledJobRunStatus::Failed
)->value;
if ($status === ScheduledJobRunStatus::Success->value) {
$result['success']++;
} elseif ($status === 'skipped') {
} elseif ($status === ScheduledJobRunStatus::Skipped->value) {
$result['skipped']++;
} else {
$result['failed']++;
@@ -69,18 +83,20 @@ class SchedulerRunService
}
} catch (\Throwable $exception) {
$result['ok'] = false;
$result['error'] = 'unexpected_error';
$result['error'] = SchedulerRuntimeResult::UnexpectedError->value;
} finally {
$this->releaseRunnerLock();
$result['duration_ms'] = $this->durationMs($startedAt);
if ($result['ok']) {
$this->updateRuntimeHeartbeat('ok', null);
$this->updateRuntimeHeartbeat(SchedulerRuntimeResult::Ok->value, null);
} else {
$errorCode = $this->nullableString($result['error']) ?? 'unexpected_error';
$this->updateRuntimeHeartbeat('unexpected_error', $errorCode);
$errorCode = $this->nullableString($result['error']) ?? SchedulerRuntimeResult::UnexpectedError->value;
$this->updateRuntimeHeartbeat(SchedulerRuntimeResult::UnexpectedError->value, $errorCode);
}
}
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Scheduler->value, $result);
return $result;
}
@@ -88,23 +104,40 @@ class SchedulerRunService
{
$this->scheduledJobService->ensureSystemJobs();
if ($jobId <= 0 || $actorUserId <= 0) {
return ['ok' => false, 'error' => 'invalid_request'];
$result = ['ok' => false, 'error' => 'invalid_request'];
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId);
return $result;
}
if (!$this->acquireRunnerLock()) {
return ['ok' => false, 'error' => 'lock_not_acquired'];
$result = ['ok' => false, 'error' => SchedulerRuntimeResult::LockNotAcquired->value];
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId);
return $result;
}
try {
$job = $this->scheduledJobRepository->find($jobId);
if (!$job) {
return ['ok' => false, 'error' => 'job_not_found'];
$result = ['ok' => false, 'error' => 'job_not_found'];
$this->recordSchedulerRunEvent(ScheduledJobTriggerType::Manual->value, $result, $actorUserId);
return $result;
}
$run = $this->runOne($job, 'manual', $actorUserId, true);
return [
'ok' => in_array((string) ($run['status'] ?? ''), ['success', 'skipped'], true),
'status' => (string) ($run['status'] ?? 'failed'),
$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();
}
@@ -116,13 +149,15 @@ class SchedulerRunService
$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';
$triggerType = ScheduledJobTriggerType::normalizeOr($triggerType, ScheduledJobTriggerType::Scheduler)->value;
if ($jobId <= 0) {
return ['status' => 'failed', 'error_code' => 'job_not_found', 'error_message' => null];
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']))) {
return ['status' => 'skipped', 'error_code' => 'job_disabled', 'error_message' => null];
$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;
}
$markedRunning = $this->scheduledJobRepository->markRunning($jobId, $startedAtUtc);
@@ -131,7 +166,7 @@ class SchedulerRunService
'job_id' => $jobId,
'job_key' => (string) ($job['job_key'] ?? ''),
'trigger_type' => $triggerType,
'status' => 'skipped',
'status' => ScheduledJobRunStatus::Skipped->value,
'actor_user_id' => $actorUserId,
'started_at' => $startedAtUtc,
'finished_at' => $startedAtUtc,
@@ -140,11 +175,13 @@ class SchedulerRunService
'error_message' => null,
'result_json' => null,
]);
return ['status' => 'skipped', 'error_code' => 'job_already_running', 'error_message' => 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 = 'failed';
$status = ScheduledJobRunStatus::Failed->value;
$errorCode = null;
$errorMessage = null;
$resultPayload = [];
@@ -152,20 +189,19 @@ class SchedulerRunService
try {
$definition = $this->scheduledJobRegistry->get((string) ($job['job_key'] ?? ''));
if (!$definition) {
$status = 'failed';
$status = ScheduledJobRunStatus::Failed->value;
$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';
$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 = 'failed';
$errorCode = 'unexpected_error';
$status = ScheduledJobRunStatus::Failed->value;
$errorCode = SchedulerRuntimeResult::UnexpectedError->value;
$errorMessage = substr($exception->getMessage(), 0, 255) ?: null;
}
@@ -175,7 +211,11 @@ class SchedulerRunService
$nextRunUtc = null;
if ((int) ($job['enabled'] ?? 0) === 1) {
if ($triggerType === 'scheduler' && (int) ($job['catchup_once'] ?? 1) === 0 && !empty($job['next_run_at'])) {
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.
@@ -222,7 +262,10 @@ class SchedulerRunService
'result_json' => $this->encodeResult($resultPayload),
]);
return ['status' => $status, 'error_code' => $errorCode, 'error_message' => $errorMessage];
$run = ['status' => $status, 'error_code' => $errorCode, 'error_message' => $errorMessage];
$this->recordSchedulerJobEvent($jobId, (string) ($job['job_key'] ?? ''), $triggerType, $actorUserId, $run, $durationMs);
return $run;
}
private function insertRunLog(array $data): void
@@ -232,8 +275,8 @@ class SchedulerRunService
'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'),
'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,
@@ -277,8 +320,7 @@ class SchedulerRunService
private function acquireRunnerLock(): bool
{
$got = DB::selectValue('select GET_LOCK(?, 0) as got_lock', self::RUNNER_LOCK_NAME);
return (int) $got === 1;
return $this->databaseSessionRepository->acquireAdvisoryLock(self::RUNNER_LOCK_NAME, 0);
}
private function updateRuntimeHeartbeat(string $status, ?string $errorCode): void
@@ -293,7 +335,7 @@ class SchedulerRunService
private function releaseRunnerLock(): void
{
try {
DB::selectValue('select RELEASE_LOCK(?) as released_lock', self::RUNNER_LOCK_NAME);
$this->databaseSessionRepository->releaseAdvisoryLock(self::RUNNER_LOCK_NAME);
} catch (\Throwable $exception) {
// no-op
}
@@ -303,4 +345,89 @@ class SchedulerRunService
{
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;
}
}