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 'success'; } if ($status === ScheduledJobRunStatus::Failed->value) { return 'failed'; } // "skipped" is an operational non-error state (e.g. already running), // not an authorization denial. return 'success'; } }