From f575b2d1799fe8f5b8fc052790fc07b04ce80303 Mon Sep 17 00:00:00 2001 From: Andras Bacsai <5845193+andrasbacsai@users.noreply.github.com> Date: Fri, 2 Oct 2026 09:19:54 +0200 Subject: [PATCH] feat(scheduler): run one scheduled job dispatcher for each schedule type The scheduler ran one dispatcher for database backups, tasks, volume backups, and Docker cleanups after each other. A slow type, for example many tasks due at the same minute, made the whole run longer, and the shared overlap lock then made all types skip the next minute. Run `scheduled:dispatch --type=` every minute for backups, tasks, volume-backups, and docker-cleanups. Each type runs in its own background process with its own scheduler and job overlap locks, and publishes and recovers only the deliveries of its own job type. Running the command without a type still processes all types. --- .../Commands/DispatchScheduledJobs.php | 13 ++- app/Console/Kernel.php | 15 ++-- app/Jobs/ScheduledJobManager.php | 69 +++++++++++---- app/Services/ScheduledJobDeliveryService.php | 11 ++- tests/Feature/ScheduleOnOneServerTest.php | 18 ++-- .../ScheduledJobManagerDispatchTest.php | 86 +++++++++++++++++++ 6 files changed, 180 insertions(+), 32 deletions(-) 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([