diff --git a/app/Jobs/ScheduledJobManager.php b/app/Jobs/ScheduledJobManager.php index 43c35b2b91..4815a853c9 100644 --- a/app/Jobs/ScheduledJobManager.php +++ b/app/Jobs/ScheduledJobManager.php @@ -54,6 +54,7 @@ class ScheduledJobManager Log::channel('scheduled')->info('ScheduledJobManager started', ['execution_time' => $this->now->toIso8601String()]); + $this->runStep('stale enqueued occurrences', fn () => $this->deliveries->recoverStaleEnqueued()); $this->deliveries->publishPending(); $this->runStep('database backups', fn () => $this->processDatabaseBackups()); $this->runStep('scheduled tasks', fn () => $this->processScheduledTasks()); @@ -282,8 +283,11 @@ class ScheduledJobManager if ($stored === null) { // First run after the upgrade, a timezone change, or a move to another server. $first = next_cron_run_at($frequency, $timezone, $this->now, includeCurrentMinute: true); - if ($first === null || ! $this->advance($schedule, $column, null, $first)) { - return $first === null ? ['invalid_frequency' => $frequency] : []; + if ($first === null) { + return $this->invalidFrequency($schedule, $column, null, $frequency, $delivery['schedule_key']); + } + if (! $this->advance($schedule, $column, null, $first)) { + return []; } if ($first->gt($this->now)) { return []; @@ -294,7 +298,7 @@ class ScheduledJobManager $dueAt = CarbonImmutable::instance($schedule->getAttribute($column))->utc(); $next = next_cron_run_at($frequency, $timezone, $this->now); if ($next === null) { - return ['invalid_frequency' => $frequency]; + return $this->invalidFrequency($schedule, $column, $stored, $frequency, $delivery['schedule_key']); } // When the clocks go back, a local time occurs two times. Run it only once. if ($this->sameLocalTime($next, $dueAt, $timezone)) { @@ -334,10 +338,32 @@ class ScheduledJobManager return [...$outcome, 'server_id' => $server->id, 'next_run_at' => $next->toIso8601String()]; } + /** + * Park a schedule whose frequency does not parse: set next_run_at to null, which the UI, the API, + * and `scheduled:diagnostics` show as an invalid frequency, and log it once a day instead of + * every minute. Saving a valid frequency calculates next_run_at again (HasNextRunAt, ServerSetting). + * + * @return array + */ + private function invalidFrequency(Model $schedule, string $column, mixed $stored, string $frequency, string $scheduleKey): array + { + if ($stored !== null) { + $this->advance($schedule, $column, $stored, null); + } + + $firstReportToday = rescue( + fn () => Cache::add("scheduled-job-manager:invalid-frequency:{$scheduleKey}:".md5($frequency), true, now()->addDay()), + true, + report: false, + ); + + return $firstReportToday ? ['invalid_frequency' => $frequency] : []; + } + /** * Move next_run_at from the stored value to $to. Only one process can win this update. */ - private function advance(Model $schedule, string $column, mixed $from, CarbonImmutable $to): bool + private function advance(Model $schedule, string $column, mixed $from, ?CarbonImmutable $to): bool { $updated = $schedule->newQuery() ->whereKey($schedule->getKey()) diff --git a/app/Services/ScheduledJobDeliveryService.php b/app/Services/ScheduledJobDeliveryService.php index 4aab2231cb..9067391e6d 100644 --- a/app/Services/ScheduledJobDeliveryService.php +++ b/app/Services/ScheduledJobDeliveryService.php @@ -16,6 +16,19 @@ use Illuminate\Support\Facades\Log; class ScheduledJobDeliveryService { + /** + * An occurrence that is enqueued longer than this has most likely lost its queued job, for + * example after a Redis restart. A job that is only waiting in a long queue is safe as well: + * the first job that claims the occurrence runs it, and the other job exits. + */ + public const ENQUEUED_STALE_AFTER_MINUTES = 60; + + /** + * Job types that run once when they are late, like the dispatcher does. A lost task or Docker + * cleanup is logged as missed instead. + */ + private const LATE_RUN_JOB_TYPES = ['database-backup', 'volume-backup']; + /** * Record a pending delivery for one occurrence of a schedule. Returns null when the * occurrence already has a delivery. @@ -53,6 +66,49 @@ class ScheduledJobDeliveryService }); } + /** + * Publish backups again whose queued job was not started, and log lost tasks and Docker + * cleanups as missed. Only one scheduler node can move each occurrence. + */ + public function recoverStaleEnqueued(): void + { + ScheduledJobDelivery::query() + ->where('status', 'enqueued') + ->where('enqueued_at', '<', now()->subMinutes(self::ENQUEUED_STALE_AFTER_MINUTES)) + ->chunkById(100, function ($occurrences): void { + foreach ($occurrences as $occurrence) { + try { + $runLate = in_array($occurrence->job_type, self::LATE_RUN_JOB_TYPES, true); + $moved = ScheduledJobDelivery::query() + ->whereKey($occurrence->id) + ->where('status', 'enqueued') + ->where('enqueued_at', '<', now()->subMinutes(self::ENQUEUED_STALE_AFTER_MINUTES)) + ->update(['status' => $runLate ? 'pending' : 'failed', 'updated_at' => now()]) === 1; + + if (! $moved) { + continue; + } + + if (! $runLate) { + $this->logOccurrence($occurrence->uuid, 'Scheduled occurrence missed: its queued job was not started', 'warning'); + + continue; + } + + $occurrence->status = 'pending'; + $this->logOccurrence($occurrence->uuid, 'Scheduled occurrence publishing again: its queued job was not started', 'warning'); + $this->publish($occurrence); + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Failed to recover enqueued scheduled occurrence', [ + 'occurrence_uuid' => $occurrence->uuid, + 'schedule_key' => $occurrence->schedule_key, + 'error' => $e->getMessage(), + ]); + } + } + }); + } + public function deleteOldOccurrences(): void { ScheduledJobDelivery::query() diff --git a/tests/Feature/ScheduledJobManagerDispatchTest.php b/tests/Feature/ScheduledJobManagerDispatchTest.php index a869d8e6a0..61ecfa66ef 100644 --- a/tests/Feature/ScheduledJobManagerDispatchTest.php +++ b/tests/Feature/ScheduledJobManagerDispatchTest.php @@ -466,6 +466,115 @@ it('logs a backup run that ends early, even when it completes before the enqueue ->and(ScheduledJobDelivery::query()->where('schedule_key', "scheduled-backup:{$backup->id}")->exists())->toBeFalse(); }); +it('logs an invalid frequency once and runs the schedule again after the frequency is fixed', function () { + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + $logPath = captureScheduledJobManagerTestLog('scheduled-errors'); + $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => '* * * * *']); + ScheduledTask::query()->whereKey($task->id)->update(['frequency' => 'not a cron', 'next_run_at' => now()->subMinute()]); + + foreach (range(1, 3) as $minute) { + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, $minute, 0, 'UTC')); + (new ScheduledJobManager)->handle(); + } + + $log = file_get_contents($logPath); + @unlink($logPath); + + Queue::assertNotPushed(ScheduledTaskJob::class); + expect(substr_count($log, 'Task has an invalid frequency'))->toBe(1) + ->and($task->fresh()->next_run_at)->toBeNull() + ->and($task->fresh()->enabled)->toBeTrue(); + + $task->fresh()->update(['frequency' => '* * * * *']); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 4, 0, 'UTC')); + (new ScheduledJobManager)->handle(); + + Queue::assertPushed(ScheduledTaskJob::class, 1); +}); + +it('logs an invalid frequency of a schedule without a next run only once', function () { + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + $logPath = captureScheduledJobManagerTestLog('scheduled-errors'); + $database = createScheduledBackupDatabase(); + $backup = createScheduledDatabaseBackup($database, ['frequency' => 'daily']); + ScheduledDatabaseBackup::query()->whereKey($backup->id)->update(['frequency' => '61 * * * *', 'next_run_at' => null]); + + foreach (range(1, 3) as $minute) { + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, $minute, 0, 'UTC')); + (new ScheduledJobManager)->handle(); + } + + $log = file_get_contents($logPath); + @unlink($logPath); + + expect(substr_count($log, 'Backup has an invalid frequency'))->toBe(1); +}); + +it('republishes a backup occurrence whose queued job was lost, and the old job cannot run it again', function () { + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + $backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => 'daily']); + $occurrence = ScheduledJobDelivery::create([ + 'schedule_key' => "scheduled-backup:{$backup->id}", + 'scheduled_for' => Carbon::create(2026, 9, 17, 0, 0, 0, 'UTC'), + 'job_type' => 'database-backup', + 'resource_id' => $backup->id, + 'status' => 'enqueued', + 'enqueued_at' => now()->subMinutes(ScheduledJobDeliveryService::ENQUEUED_STALE_AFTER_MINUTES + 1), + ]); + + (new ScheduledJobManager)->handle(); + + Queue::assertPushed(DatabaseBackupJob::class, fn (DatabaseBackupJob $job) => $job->occurrenceUuid === $occurrence->uuid); + expect($occurrence->fresh()->status)->toBe('enqueued') + ->and($occurrence->fresh()->enqueued_at->toDateTimeString())->toBe('2026-09-17 12:00:00'); + + $service = app(ScheduledJobDeliveryService::class); + expect($service->claim($occurrence->uuid, 'republished-job'))->toBeTrue() + ->and($service->claim($occurrence->uuid, 'original-job'))->toBeFalse(); +}); + +it('logs a lost scheduled task occurrence as missed instead of running it late', function () { + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + $logPath = captureScheduledJobManagerTestLog('scheduled'); + $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => 'daily']); + $occurrence = ScheduledJobDelivery::create([ + 'schedule_key' => "scheduled-task:{$task->id}", + 'scheduled_for' => Carbon::create(2026, 9, 17, 0, 0, 0, 'UTC'), + 'job_type' => 'scheduled-task', + 'resource_id' => $task->id, + 'status' => 'enqueued', + 'enqueued_at' => now()->subMinutes(ScheduledJobDeliveryService::ENQUEUED_STALE_AFTER_MINUTES + 1), + ]); + + (new ScheduledJobManager)->handle(); + + $log = file_get_contents($logPath); + @unlink($logPath); + + Queue::assertNotPushed(ScheduledTaskJob::class); + expect($occurrence->fresh()->status)->toBe('failed') + ->and(app(ScheduledJobDeliveryService::class)->claim($occurrence->uuid, 'original-job'))->toBeFalse() + ->and($log)->toContain('Scheduled occurrence missed: its queued job was not started'); +}); + +it('leaves recently enqueued occurrences alone', function () { + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + $backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => 'daily']); + $occurrence = ScheduledJobDelivery::create([ + 'schedule_key' => "scheduled-backup:{$backup->id}", + 'scheduled_for' => Carbon::create(2026, 9, 17, 11, 0, 0, 'UTC'), + 'job_type' => 'database-backup', + 'resource_id' => $backup->id, + 'status' => 'enqueued', + 'enqueued_at' => now()->subMinutes(ScheduledJobDeliveryService::ENQUEUED_STALE_AFTER_MINUTES - 1), + ]); + + (new ScheduledJobManager)->handle(); + + Queue::assertNotPushed(DatabaseBackupJob::class); + expect($occurrence->fresh()->enqueued_at->toDateTimeString())->toBe('2026-09-17 11:01:00'); +}); + function captureScheduledJobManagerTestLog(string $channel): string { $path = tempnam(sys_get_temp_dir(), 'coolify-scheduled-log-');