diff --git a/app/Console/Commands/DispatchScheduledJobs.php b/app/Console/Commands/DispatchScheduledJobs.php index 2c1cc8df04..d47b782f94 100644 --- a/app/Console/Commands/DispatchScheduledJobs.php +++ b/app/Console/Commands/DispatchScheduledJobs.php @@ -7,13 +7,22 @@ use Illuminate\Console\Command; class DispatchScheduledJobs extends Command { - protected $signature = 'scheduled:dispatch'; + protected $signature = 'scheduled:dispatch + {--type= : Only one schedule type: backups, tasks, volume-backups, or docker-cleanups. Empty runs all types.}'; protected $description = 'Dispatch the due scheduled backups, tasks, volume backups, and Docker cleanups'; public function handle(): int { - ScheduledJobManager::dispatchSync(); + $type = $this->option('type') ?: null; + + if ($type !== null && ! array_key_exists($type, ScheduledJobManager::TYPES)) { + $this->error("Unknown type [{$type}]. Use one of: ".implode(', ', array_keys(ScheduledJobManager::TYPES)).'.'); + + return self::INVALID; + } + + ScheduledJobManager::dispatchSync($type); return self::SUCCESS; } diff --git a/app/Console/Kernel.php b/app/Console/Kernel.php index 7d45ff4b53..4e4d921b08 100644 --- a/app/Console/Kernel.php +++ b/app/Console/Kernel.php @@ -14,6 +14,7 @@ use App\Jobs\PullTemplatesFromCDN; use App\Jobs\ReconcileGithubRunnersJob; use App\Jobs\RegenerateSslCertJob; use App\Jobs\RevalidateUnusableS3StoragesJob; +use App\Jobs\ScheduledJobManager; use App\Jobs\ServerManagerJob; use App\Jobs\UpdateCoolifyJob; use App\Models\InstanceSettings; @@ -122,14 +123,18 @@ class Kernel extends ConsoleKernel /** * Run the manager from the scheduler, not from a queue worker. A busy queue could delay it * past the catch-up window, and then due backups and tasks would be skipped. + * Each schedule type runs in its own process with its own overlap lock, so the types run in + * parallel and a slow type cannot make another type skip a run. */ private function scheduleScheduledJobManager(): void { - $this->scheduleInstance->command('scheduled:dispatch') - ->everyMinute() - ->onOneServer() - ->withoutOverlapping(5) - ->runInBackground(); + foreach (array_keys(ScheduledJobManager::TYPES) as $type) { + $this->scheduleInstance->command("scheduled:dispatch --type={$type}") + ->everyMinute() + ->onOneServer() + ->withoutOverlapping(5) + ->runInBackground(); + } } private function scheduleUpdates(): void diff --git a/app/Jobs/ScheduledJobManager.php b/app/Jobs/ScheduledJobManager.php index 8e4a5b26a8..0ec09c361f 100644 --- a/app/Jobs/ScheduledJobManager.php +++ b/app/Jobs/ScheduledJobManager.php @@ -33,6 +33,17 @@ class ScheduledJobManager implements ShouldQueue private const CHUNK_SIZE = 100; + /** + * The schedule types that run as separate dispatchers, with the delivery job types they own. + * Each type has its own overlap lock, so a slow type cannot make another type skip a run. + */ + public const TYPES = [ + 'backups' => ['database-backup'], + 'tasks' => ['scheduled-task'], + 'volume-backups' => ['volume-backup'], + 'docker-cleanups' => ['docker-cleanup'], + ]; + /** * The time when this job execution started. * Used to ensure all scheduled items are evaluated against the same point in time. @@ -44,10 +55,14 @@ class ScheduledJobManager implements ShouldQueue private int $skippedCount = 0; /** - * Create a new job instance. + * @param string|null $type One key of TYPES, or null for all types. */ - public function __construct() + public function __construct(public ?string $type = null) { + if ($type !== null && ! array_key_exists($type, self::TYPES)) { + throw new \InvalidArgumentException("Unknown scheduled job type [{$type}]."); + } + $this->onQueue(crons_queue()); } @@ -59,10 +74,10 @@ class ScheduledJobManager implements ShouldQueue // Self-healing: clear any stale lock before WithoutOverlapping tries to acquire it. // Stale locks (TTL = -1) can occur during upgrades, Redis restarts, or edge cases. // @see https://github.com/coollabsio/coolify/issues/8327 - self::clearStaleLockIfPresent(); + $this->clearStaleLockIfPresent(); return [ - (new WithoutOverlapping('scheduled-job-manager')) + (new WithoutOverlapping($this->lockName())) ->expireAfter(90) // Lock expires after 90s to handle high-load environments with many tasks ->dontRelease(), // Don't re-queue on lock conflict ]; @@ -74,11 +89,11 @@ class ScheduledJobManager implements ShouldQueue * This provides continuous self-healing since it runs every time the job is dispatched. * Stale locks permanently block all scheduled job executions with no user-visible error. */ - private static function clearStaleLockIfPresent(): void + private function clearStaleLockIfPresent(): void { try { $cachePrefix = config('cache.prefix', ''); - $lockKey = $cachePrefix.'laravel-queue-overlap:'.self::class.':scheduled-job-manager'; + $lockKey = $cachePrefix.'laravel-queue-overlap:'.self::class.':'.$this->lockName(); $ttl = Redis::connection('default')->ttl($lockKey); @@ -96,6 +111,16 @@ class ScheduledJobManager implements ShouldQueue } } + private function lockName(): string + { + return $this->type === null ? 'scheduled-job-manager' : "scheduled-job-manager:{$this->type}"; + } + + private function includes(string $type): bool + { + return $this->type === null || $this->type === $type; + } + public function handle(): void { // Freeze the execution time at the start of the job @@ -104,20 +129,23 @@ class ScheduledJobManager implements ShouldQueue $this->skippedCount = 0; $this->logStart(); + $jobTypes = $this->type === null ? null : self::TYPES[$this->type]; try { - app(ScheduledJobDeliveryService::class)->recoverStaleEnqueued(); + app(ScheduledJobDeliveryService::class)->recoverStaleEnqueued($jobTypes); } catch (\Throwable $e) { Log::channel('scheduled-errors')->error('Failed to recover stale enqueued occurrences', [ 'error' => $e->getMessage(), ]); } - app(ScheduledJobDeliveryService::class)->publishPending(); + app(ScheduledJobDeliveryService::class)->publishPending($jobTypes); // Process scheduled backups and tasks together so neither type starves the other. try { - $this->processScheduledBackupsAndTasks(); + if ($this->includes('backups') || $this->includes('tasks')) { + $this->processScheduledBackupsAndTasks($this->includes('backups'), $this->includes('tasks')); + } } catch (\Throwable $e) { Log::channel('scheduled-errors')->error('Failed to process scheduled backups and tasks', [ 'error' => $e->getMessage(), @@ -126,8 +154,10 @@ class ScheduledJobManager implements ShouldQueue } try { - $this->recoverStoppedVolumeBackupContainers(); - $this->processScheduledVolumeBackups(); + if ($this->includes('volume-backups')) { + $this->recoverStoppedVolumeBackupContainers(); + $this->processScheduledVolumeBackups(); + } } catch (\Throwable $e) { Log::channel('scheduled-errors')->error('Failed to process scheduled volume backups', [ 'error' => $e->getMessage(), @@ -137,7 +167,9 @@ class ScheduledJobManager implements ShouldQueue // Process Docker cleanups - don't let failures stop the job manager try { - $this->processDockerCleanups(); + if ($this->includes('docker-cleanups')) { + $this->processDockerCleanups(); + } } catch (\Throwable $e) { Log::channel('scheduled-errors')->error('Failed to process docker cleanups', [ 'error' => $e->getMessage(), @@ -146,6 +178,7 @@ class ScheduledJobManager implements ShouldQueue } Log::channel('scheduled')->info('ScheduledJobManager completed', [ + 'type' => $this->type ?? 'all', 'execution_time' => $this->executionTime->toIso8601String(), 'duration_ms' => $this->executionTime->diffInMilliseconds(Carbon::now()), 'dispatched' => $this->dispatchedCount, @@ -171,11 +204,13 @@ class ScheduledJobManager implements ShouldQueue */ private function logStart(): void { - $previousStartedAt = rescue(fn () => Cache::get('scheduled-job-manager:last-started-at'), report: false); - rescue(fn () => Cache::put('scheduled-job-manager:last-started-at', $this->executionTime->toIso8601String(), 86400), report: false); + $lastStartedKey = $this->lockName().':last-started-at'; + $previousStartedAt = rescue(fn () => Cache::get($lastStartedKey), report: false); + rescue(fn () => Cache::put($lastStartedKey, $this->executionTime->toIso8601String(), 86400), report: false); $queuedAt = data_get($this->job?->payload(), 'createdAt'); $context = [ + 'type' => $this->type ?? 'all', 'execution_time' => $this->executionTime->toIso8601String(), 'host' => gethostname(), 'pid' => getmypid(), @@ -192,14 +227,14 @@ class ScheduledJobManager implements ShouldQueue } } - private function processScheduledBackupsAndTasks(): void + private function processScheduledBackupsAndTasks(bool $withBackups, bool $withTasks): void { $lastBackupId = null; $lastTaskId = null; do { - $backups = $this->scheduledBackupQuery($lastBackupId)->get(); - $tasks = $this->scheduledTaskQuery($lastTaskId)->get(); + $backups = $withBackups ? $this->scheduledBackupQuery($lastBackupId)->get() : collect(); + $tasks = $withTasks ? $this->scheduledTaskQuery($lastTaskId)->get() : collect(); if ($backups->isNotEmpty()) { $lastBackupId = $backups->last()->id; diff --git a/app/Services/ScheduledJobDeliveryService.php b/app/Services/ScheduledJobDeliveryService.php index 5dce4691dc..84e8d4210f 100644 --- a/app/Services/ScheduledJobDeliveryService.php +++ b/app/Services/ScheduledJobDeliveryService.php @@ -123,10 +123,14 @@ class ScheduledJobDeliveryService }); } - public function publishPending(): void + /** + * @param array|null $jobTypes Only these job types, or all when null. + */ + public function publishPending(?array $jobTypes = null): void { ScheduledJobDelivery::query() ->where('status', 'pending') + ->when($jobTypes !== null, fn ($query) => $query->whereIn('job_type', $jobTypes)) ->orderBy('id') ->chunkById(100, function ($occurrences): void { foreach ($occurrences as $occurrence) { @@ -166,11 +170,14 @@ class ScheduledJobDeliveryService * Publish backups again whose queued job was not started, and log lost tasks and Docker * cleanups as missed. A lost backup is not run when a newer occurrence of the same schedule * exists, so a backup that runs longer than its interval does not build up a backlog. + * + * @param array|null $jobTypes Only these job types, or all when null. */ - public function recoverStaleEnqueued(): void + public function recoverStaleEnqueued(?array $jobTypes = null): void { ScheduledJobDelivery::query() ->where('status', 'enqueued') + ->when($jobTypes !== null, fn ($query) => $query->whereIn('job_type', $jobTypes)) ->where('enqueued_at', '<', now()->subMinutes(self::ENQUEUED_STALE_AFTER_MINUTES)) ->chunkById(100, function ($occurrences): void { foreach ($occurrences as $occurrence) { diff --git a/tests/Feature/ScheduleOnOneServerTest.php b/tests/Feature/ScheduleOnOneServerTest.php index adc7ded3e8..03aab32a84 100644 --- a/tests/Feature/ScheduleOnOneServerTest.php +++ b/tests/Feature/ScheduleOnOneServerTest.php @@ -74,15 +74,21 @@ it('schedules stuck resource cleanup in the background once per day', function ( ->and($event->runInBackground)->toBeTrue(); }); -it('runs the scheduled job dispatcher from the scheduler instead of a queue', function () { +it('runs one scheduled job dispatcher for each schedule type from the scheduler instead of a queue', function () { $events = collect(app(Schedule::class)->events()); - $event = $events->first(fn ($event) => str_contains((string) $event->command, 'scheduled:dispatch')); + $dispatchers = $events->filter(fn ($event) => str_contains((string) $event->command, 'scheduled:dispatch')); - expect($event)->not->toBeNull() - ->and($event->expression)->toBe('* * * * *') - ->and($event->onOneServer)->toBeTrue() - ->and($event->runInBackground)->toBeTrue() + expect($dispatchers->map(fn ($event) => str((string) $event->command)->after('--type=')->value())->values()->all()) + ->toBe(['backups', 'tasks', 'volume-backups', 'docker-cleanups']); + $dispatchers->each(function ($event) { + expect($event->expression)->toBe('* * * * *') + ->and($event->onOneServer)->toBeTrue() + ->and($event->withoutOverlapping)->toBeTrue() + ->and($event->runInBackground)->toBeTrue(); + }); + // Each type has its own overlap lock. + expect($dispatchers->map->mutexName()->unique())->toHaveCount(4) ->and($events->contains(fn ($event) => str_contains((string) $event->description, 'ScheduledJobManager')))->toBeFalse(); }); diff --git a/tests/Feature/ScheduledJobManagerDispatchTest.php b/tests/Feature/ScheduledJobManagerDispatchTest.php index eaff025bfc..95f241b046 100644 --- a/tests/Feature/ScheduledJobManagerDispatchTest.php +++ b/tests/Feature/ScheduledJobManagerDispatchTest.php @@ -2,6 +2,7 @@ use App\Events\BackupCreated; use App\Jobs\DatabaseBackupJob; +use App\Jobs\DockerCleanupJob; use App\Jobs\ScheduledJobManager; use App\Jobs\ScheduledTaskJob; use App\Models\Application; @@ -21,6 +22,7 @@ use App\Models\StandalonePostgresql; use App\Models\Team; use App\Services\ScheduledJobDeliveryService; use Illuminate\Foundation\Testing\RefreshDatabase; +use Illuminate\Queue\Middleware\WithoutOverlapping; use Illuminate\Support\Carbon; use Illuminate\Support\Facades\Bus; use Illuminate\Support\Facades\Cache; @@ -702,6 +704,90 @@ it('leaves recently enqueued occurrences alone', function () { expect($occurrence->fresh()->enqueued_at->toDateTimeString())->toBe('2026-09-17 11:01:00'); }); +it('dispatches only the schedules of its own type', function (string $type, array $expected) { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 1, 0, 0, 'UTC')); + Queue::fakeExcept(ScheduledJobManager::class); + $application = createScheduledTaskApplication(); + createScheduledApplicationTask($application, ['frequency' => '* * * * *']); + createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => '* * * * *']); + + $this->artisan('scheduled:dispatch', ['--type' => $type])->assertSuccessful(); + + Queue::assertPushed(DatabaseBackupJob::class, $expected['backups']); + Queue::assertPushed(ScheduledTaskJob::class, $expected['tasks']); + Queue::assertPushed(DockerCleanupJob::class, $expected['docker-cleanups']); +})->with([ + 'backups' => ['backups', ['backups' => 1, 'tasks' => 0, 'docker-cleanups' => 0]], + 'tasks' => ['tasks', ['backups' => 0, 'tasks' => 1, 'docker-cleanups' => 0]], + // Two servers: the task application and the backup database are on separate test servers. + 'docker cleanups' => ['docker-cleanups', ['backups' => 0, 'tasks' => 0, 'docker-cleanups' => 2]], +]); + +it('dispatches each occurrence once when the type dispatchers run after each other', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 1, 0, 0, 'UTC')); + Queue::fakeExcept(ScheduledJobManager::class); + createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => '* * * * *']); + createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => '* * * * *']); + + foreach (array_keys(ScheduledJobManager::TYPES) as $type) { + $this->artisan('scheduled:dispatch', ['--type' => $type])->assertSuccessful(); + } + (new ScheduledJobManager)->handle(); + + Queue::assertPushed(DatabaseBackupJob::class, 1); + Queue::assertPushed(ScheduledTaskJob::class, 1); + Queue::assertPushed(DockerCleanupJob::class, 2); +}); + +it('publishes and recovers only the deliveries of its own type', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + Queue::fakeExcept(ScheduledJobManager::class); + $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => 'daily']); + $backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => 'daily']); + $pendingTask = 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, + ]); + $lostBackup = createStaleEnqueuedOccurrence("scheduled-backup:{$backup->id}", 'database-backup', $backup->id, Carbon::create(2026, 9, 17, 0, 0, 0, 'UTC')); + + $this->artisan('scheduled:dispatch', ['--type' => 'tasks'])->assertSuccessful(); + + Queue::assertPushed(ScheduledTaskJob::class, 1); + Queue::assertNotPushed(DatabaseBackupJob::class); + expect($pendingTask->fresh()->status)->toBe('enqueued') + ->and($lostBackup->fresh()->status)->toBe('enqueued'); + + $this->artisan('scheduled:dispatch', ['--type' => 'backups'])->assertSuccessful(); + + Queue::assertPushed(DatabaseBackupJob::class, fn (DatabaseBackupJob $job) => $job->occurrenceUuid === $lostBackup->uuid); +}); + +it('uses a separate overlap lock for each type', function () { + $lockKeys = collect([null, ...array_keys(ScheduledJobManager::TYPES)]) + ->map(fn (?string $type) => (new ReflectionProperty(WithoutOverlapping::class, 'key'))->getValue((new ScheduledJobManager($type))->middleware()[0])); + + expect($lockKeys->all())->toBe([ + 'scheduled-job-manager', + 'scheduled-job-manager:backups', + 'scheduled-job-manager:tasks', + 'scheduled-job-manager:volume-backups', + 'scheduled-job-manager:docker-cleanups', + ]); +}); + +it('rejects an unknown schedule type', function () { + Queue::fake(); + + $this->artisan('scheduled:dispatch', ['--type' => 'invalid'])->assertFailed(); + + Queue::assertNothingPushed(); +}); + function createStaleEnqueuedOccurrence(string $scheduleKey, string $jobType, int $resourceId, Carbon $scheduledFor, int $minutesAgo = ScheduledJobDeliveryService::ENQUEUED_STALE_AFTER_MINUTES + 1): ScheduledJobDelivery { return ScheduledJobDelivery::create([