From 19949a6f6f860985a5ffb7d32d20f426bb408946 Mon Sep 17 00:00:00 2001 From: Andras Bacsai <5845193+andrasbacsai@users.noreply.github.com> Date: Fri, 18 Sep 2026 13:56:40 +0200 Subject: [PATCH] feat(scheduling): persist and claim scheduled job occurrences Add database-backed schedule states and deliveries so distributed schedulers publish each occurrence once, queue jobs claim executions atomically, and stale occurrences are cleaned up. --- .ai/lessons.md | 5 + app/Console/Kernel.php | 5 + app/Jobs/DatabaseBackupJob.php | 18 +- app/Jobs/DockerCleanupJob.php | 19 +- app/Jobs/ScheduledJobManager.php | 182 +++++++++----- app/Jobs/ScheduledTaskJob.php | 17 +- app/Jobs/VolumeBackupJob.php | 17 +- app/Models/ScheduledJobDelivery.php | 29 +++ app/Models/ScheduledJobState.php | 18 ++ app/Services/ScheduledJobDeliveryService.php | 232 ++++++++++++++++++ ...duled_job_states_and_deliveries_tables.php | 49 ++++ .../ScheduledJobManagerDispatchTest.php | 176 ++++++++++++- 12 files changed, 692 insertions(+), 75 deletions(-) create mode 100644 app/Models/ScheduledJobDelivery.php create mode 100644 app/Models/ScheduledJobState.php create mode 100644 app/Services/ScheduledJobDeliveryService.php create mode 100644 database/migrations/2026_09_18_111936_create_scheduled_job_states_and_deliveries_tables.php diff --git a/.ai/lessons.md b/.ai/lessons.md index 43de6bf878..c2f648ca9e 100644 --- a/.ai/lessons.md +++ b/.ai/lessons.md @@ -41,3 +41,8 @@ - For container image changes, inspect Compose services and every relevant Dockerfile build stage. - Pin a stable release tag instead of using a floating `latest` tag. - A successful image pull does not prove that the complete application build no longer uses the old image. + +## Make distributed schedules durable +- Use the database as the correctness source for dynamic cron occurrences shared by multiple scheduler and Horizon nodes; Redis locks are load controls, not a durable execution ledger. +- Give each schedule occurrence a unique database identity and make queue consumers claim it atomically before external work. +- Keep pending occurrences recoverable across publisher interruptions, and define an explicit bounded policy for late or offline schedules. diff --git a/app/Console/Kernel.php b/app/Console/Kernel.php index a86b26ce5c..d5b0e5a9c5 100644 --- a/app/Console/Kernel.php +++ b/app/Console/Kernel.php @@ -16,6 +16,7 @@ use App\Jobs\ScheduledJobManager; use App\Jobs\ServerManagerJob; use App\Jobs\UpdateCoolifyJob; use App\Models\InstanceSettings; +use App\Services\ScheduledJobDeliveryService; use Illuminate\Console\Scheduling\Schedule; use Illuminate\Foundation\Console\Kernel as ConsoleKernel; @@ -46,6 +47,10 @@ class Kernel extends ConsoleKernel ->hourly() ->when(fn () => config('constants.ssh.mux_enabled') && ! config('constants.coolify.is_windows_docker_desktop')); $this->scheduleInstance->command('cleanup:redis --clear-locks')->daily(); + $this->scheduleInstance->call(fn () => app(ScheduledJobDeliveryService::class)->deleteOldOccurrences()) + ->name('cleanup:scheduled-job-occurrences') + ->dailyAt('04:00') + ->onOneServer(); $this->scheduleInstance->command('cleanup:stucked-resources') ->dailyAt('03:17') ->onOneServer() diff --git a/app/Jobs/DatabaseBackupJob.php b/app/Jobs/DatabaseBackupJob.php index b1f2c38b96..3bc86df903 100644 --- a/app/Jobs/DatabaseBackupJob.php +++ b/app/Jobs/DatabaseBackupJob.php @@ -18,6 +18,7 @@ use App\Notifications\Database\BackupFailed; use App\Notifications\Database\BackupSuccess; use App\Notifications\Database\BackupSuccessWithS3Warning; use App\Rules\SafeWebhookUrl; +use App\Services\ScheduledJobDeliveryService; use App\Support\BackupCompression; use App\Support\ClickhouseBackupCommand; use Carbon\Carbon; @@ -78,7 +79,7 @@ class DatabaseBackupJob implements ShouldBeEncrypted, ShouldQueue public ?string $backup_log_uuid = null; - public function __construct(public ScheduledDatabaseBackup $backup) + public function __construct(public ScheduledDatabaseBackup $backup, public ?string $occurrenceUuid = null) { $this->onQueue(crons_queue()); $this->timeout = $backup->timeout ?? 3600; @@ -93,6 +94,12 @@ class DatabaseBackupJob implements ShouldBeEncrypted, ShouldQueue public function handle(): void { + if ($this->occurrenceUuid && ! app(ScheduledJobDeliveryService::class)->claim($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid)) { + return; + } + + $failed = false; + try { $databasesToBackup = null; @@ -483,8 +490,13 @@ class DatabaseBackupJob implements ShouldBeEncrypted, ShouldQueue $this->removeExpiredBackups(); } } catch (Throwable $e) { + $failed = true; throw $e; } finally { + if (! $failed && $this->occurrenceUuid) { + app(ScheduledJobDeliveryService::class)->complete($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid); + } + if ($this->backup_log) { $this->backup_log->update([ 'finished_at' => Carbon::now()->toImmutable(), @@ -852,6 +864,10 @@ class DatabaseBackupJob implements ShouldBeEncrypted, ShouldQueue public function failed(?Throwable $exception): void { + if ($this->occurrenceUuid) { + app(ScheduledJobDeliveryService::class)->fail($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid); + } + Log::channel('scheduled-errors')->error('DatabaseBackup permanently failed', [ 'job' => 'DatabaseBackupJob', 'backup_id' => $this->backup->uuid, diff --git a/app/Jobs/DockerCleanupJob.php b/app/Jobs/DockerCleanupJob.php index 5a7627e0a9..4c744a6e64 100644 --- a/app/Jobs/DockerCleanupJob.php +++ b/app/Jobs/DockerCleanupJob.php @@ -8,6 +8,7 @@ use App\Models\DockerCleanupExecution; use App\Models\Server; use App\Notifications\Server\DockerCleanupFailed; use App\Notifications\Server\DockerCleanupSuccess; +use App\Services\ScheduledJobDeliveryService; use Carbon\Carbon; use Illuminate\Bus\Queueable; use Illuminate\Contracts\Queue\ShouldBeEncrypted; @@ -38,13 +39,20 @@ class DockerCleanupJob implements ShouldBeEncrypted, ShouldQueue public Server $server, public bool $manualCleanup = false, public bool $deleteUnusedVolumes = false, - public bool $deleteUnusedNetworks = false + public bool $deleteUnusedNetworks = false, + public ?string $occurrenceUuid = null, ) { $this->onQueue('high'); } public function handle(): void { + if ($this->occurrenceUuid && ! app(ScheduledJobDeliveryService::class)->claim($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid)) { + return; + } + + $failed = false; + try { $this->execution_log = DockerCleanupExecution::create([ 'server_id' => $this->server->id, @@ -138,6 +146,7 @@ class DockerCleanupJob implements ShouldBeEncrypted, ShouldQueue event(new DockerCleanupDone($this->execution_log)); } } catch (\Throwable $e) { + $failed = true; if ($this->execution_log) { $this->execution_log->update([ 'status' => 'failed', @@ -148,6 +157,10 @@ class DockerCleanupJob implements ShouldBeEncrypted, ShouldQueue $this->server->team?->notify(new DockerCleanupFailed($this->server, 'Docker cleanup job failed with the following error: '.$e->getMessage())); throw $e; } finally { + if (! $failed && $this->occurrenceUuid) { + app(ScheduledJobDeliveryService::class)->complete($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid); + } + if ($this->execution_log) { $this->execution_log->update([ 'finished_at' => Carbon::now()->toImmutable(), @@ -158,6 +171,10 @@ class DockerCleanupJob implements ShouldBeEncrypted, ShouldQueue public function failed(?\Throwable $exception): void { + if ($this->occurrenceUuid) { + app(ScheduledJobDeliveryService::class)->fail($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid); + } + $execution = DockerCleanupExecution::query() ->where('server_id', $this->server->id) ->where('status', 'running') diff --git a/app/Jobs/ScheduledJobManager.php b/app/Jobs/ScheduledJobManager.php index 156f08d01b..3bfb30c7c1 100644 --- a/app/Jobs/ScheduledJobManager.php +++ b/app/Jobs/ScheduledJobManager.php @@ -9,6 +9,7 @@ use App\Models\ScheduledVolumeBackup; use App\Models\ScheduledVolumeBackupExecution; use App\Models\Server; use App\Models\Team; +use App\Services\ScheduledJobDeliveryService; use Cron\CronExpression; use Illuminate\Bus\Queueable; use Illuminate\Contracts\Queue\ShouldQueue; @@ -102,6 +103,8 @@ class ScheduledJobManager implements ShouldQueue 'execution_time' => $this->executionTime->toIso8601String(), ]); + app(ScheduledJobDeliveryService::class)->publishPending(); + // Process scheduled backups and tasks together so neither type starves the other. try { $this->processScheduledBackupsAndTasks(); @@ -231,7 +234,7 @@ class ScheduledJobManager implements ShouldQueue continue; } - if ($this->isDueCandidateBeforeExpensiveChecks($backup->frequency, $server, "scheduled-backup:{$backup->id}")) { + if ($this->isDueCandidateBeforeExpensiveChecks($backup->frequency, $server)) { $dueBackups[] = [ 'backup' => $backup, 'server' => $server, @@ -266,7 +269,7 @@ class ScheduledJobManager implements ShouldQueue continue; } - if ($this->isDueCandidateBeforeExpensiveChecks($task->frequency, $server, "scheduled-task:{$task->id}")) { + if ($this->isDueCandidateBeforeExpensiveChecks($task->frequency, $server)) { $dueTasks[] = [ 'task' => $task, 'server' => $server, @@ -289,14 +292,21 @@ class ScheduledJobManager implements ShouldQueue $server = $precheckedServer ?? $backup->server(); $skipReason = $this->getBackupSkipReason($backup, $server); if ($skipReason !== null) { - $this->skippedCount++; - $this->logBackupSkip($backup, $skipReason); + if ($server === null || $this->recordSkippedOccurrence($backup->frequency, $server, "scheduled-backup:{$backup->id}")) { + $this->skippedCount++; + $this->logBackupSkip($backup, $skipReason); + } return; } - if ($this->shouldDispatch($backup->frequency, $server, "scheduled-backup:{$backup->id}")) { - DatabaseBackupJob::dispatch($backup); + if ($this->dispatchOccurrence( + $backup->frequency, + $server, + "scheduled-backup:{$backup->id}", + 'database-backup', + $backup->id, + )) { $this->dispatchedCount++; Log::channel('scheduled')->info('Backup dispatched', [ 'backup_id' => $backup->id, @@ -320,25 +330,35 @@ class ScheduledJobManager implements ShouldQueue $server = $precheckedServer ?? $task->server(); $criticalSkip = $this->getTaskCriticalSkipReason($task, $server); if ($criticalSkip !== null) { - $this->skippedCount++; - $this->logTaskSkip($task, $criticalSkip, $server); + if ($server === null || $this->recordSkippedOccurrence($task->frequency, $server, "scheduled-task:{$task->id}")) { + $this->skippedCount++; + $this->logTaskSkip($task, $criticalSkip, $server); + } return; } - if (! $this->shouldDispatch($task->frequency, $server, "scheduled-task:{$task->id}")) { - return; - } - $runtimeSkip = $this->getTaskRuntimeSkipReason($task); if ($runtimeSkip !== null) { - $this->skippedCount++; - $this->logTaskSkip($task, $runtimeSkip, $server); + if ($this->recordSkippedOccurrence($task->frequency, $server, "scheduled-task:{$task->id}")) { + $this->skippedCount++; + $this->logTaskSkip($task, $runtimeSkip, $server); + } + + return; + } + + if (! $this->dispatchOccurrence( + $task->frequency, + $server, + "scheduled-task:{$task->id}", + 'scheduled-task', + $task->id, + )) { return; } - ScheduledTaskJob::dispatch($task); $this->dispatchedCount++; Log::channel('scheduled')->info('Task dispatched', [ 'task_id' => $task->id, @@ -419,30 +439,43 @@ class ScheduledJobManager implements ShouldQueue return; } + if (! $this->isDueCandidateBeforeExpensiveChecks($backup->frequency, $server)) { + return; + } + if (! $server->isFunctional()) { - $this->skippedCount++; - $this->logSkip('volume_backup', 'server_not_functional', [ - 'backup_id' => $backup->id, - 'team_id' => $backup->team_id, - 'server_id' => $server->id, - ]); + if ($this->recordSkippedOccurrence($backup->frequency, $server, "scheduled-volume-backup:{$backup->id}")) { + $this->skippedCount++; + $this->logSkip('volume_backup', 'server_not_functional', [ + 'backup_id' => $backup->id, + 'team_id' => $backup->team_id, + 'server_id' => $server->id, + ]); + } return; } if (isCloud() && $backup->team_id !== 0 && ! data_get($backup, 'team.subscription.stripe_invoice_paid', false)) { - $this->skippedCount++; - $this->logSkip('volume_backup', 'subscription_unpaid', [ - 'backup_id' => $backup->id, - 'team_id' => $backup->team_id, - 'server_id' => $server->id, - ]); + if ($this->recordSkippedOccurrence($backup->frequency, $server, "scheduled-volume-backup:{$backup->id}")) { + $this->skippedCount++; + $this->logSkip('volume_backup', 'subscription_unpaid', [ + 'backup_id' => $backup->id, + 'team_id' => $backup->team_id, + 'server_id' => $server->id, + ]); + } return; } - if ($this->shouldDispatch($backup->frequency, $server, "scheduled-volume-backup:{$backup->id}")) { - VolumeBackupJob::dispatch($backup); + if ($this->dispatchOccurrence( + $backup->frequency, + $server, + "scheduled-volume-backup:{$backup->id}", + 'volume-backup', + $backup->id, + )) { $this->dispatchedCount++; Log::channel('scheduled')->info('Volume backup dispatched', [ 'backup_id' => $backup->id, @@ -536,27 +569,36 @@ class ScheduledJobManager implements ShouldQueue private function processDockerCleanup(Server $server): void { try { + $frequency = data_get($server->settings, 'docker_cleanup_frequency', '0 * * * *'); + if (! $this->isDueCandidateBeforeExpensiveChecks($frequency, $server)) { + return; + } + $skipReason = $this->getDockerCleanupSkipReason($server); if ($skipReason !== null) { - $this->skippedCount++; - $this->logSkip('docker_cleanup', $skipReason, [ - 'server_id' => $server->id, - 'server_name' => $server->name, - 'team_id' => $server->team_id, - ]); + if ($this->recordSkippedOccurrence($frequency, $server, "docker-cleanup:{$server->id}")) { + $this->skippedCount++; + $this->logSkip('docker_cleanup', $skipReason, [ + 'server_id' => $server->id, + 'server_name' => $server->name, + 'team_id' => $server->team_id, + ]); + } return; } - $frequency = data_get($server->settings, 'docker_cleanup_frequency', '0 * * * *'); - - if ($this->shouldDispatch($frequency, $server, "docker-cleanup:{$server->id}")) { - DockerCleanupJob::dispatch( - $server, - false, - $server->settings->delete_unused_volumes, - $server->settings->delete_unused_networks - ); + if ($this->dispatchOccurrence( + $frequency, + $server, + "docker-cleanup:{$server->id}", + 'docker-cleanup', + $server->id, + [ + 'delete_unused_volumes' => $server->settings->delete_unused_volumes, + 'delete_unused_networks' => $server->settings->delete_unused_networks, + ], + )) { $this->dispatchedCount++; Log::channel('scheduled')->info('Docker cleanup dispatched', [ 'server_id' => $server->id, @@ -618,40 +660,44 @@ class ScheduledJobManager implements ShouldQueue ], $context)); } - private function shouldDispatch(string $frequency, Server $server, string $dedupKey): bool - { - return shouldRunCronNow( - $this->normalizeFrequency($frequency), + private function dispatchOccurrence( + string $frequency, + Server $server, + string $scheduleKey, + string $jobType, + int $resourceId, + array $payload = [], + ): bool { + return app(ScheduledJobDeliveryService::class)->recordAndPublish( + $scheduleKey, + $frequency, $this->serverTimezone($server), - $dedupKey, + $jobType, + $resourceId, + $payload, $this->executionTime, ); } - private function isDueCandidateBeforeExpensiveChecks(string $frequency, Server $server, string $dedupKey): bool + private function recordSkippedOccurrence(string $frequency, Server $server, string $scheduleKey): bool + { + return app(ScheduledJobDeliveryService::class)->recordSkipped( + $scheduleKey, + $frequency, + $this->serverTimezone($server), + $this->executionTime, + ); + } + + private function isDueCandidateBeforeExpensiveChecks(string $frequency, Server $server): bool { $cron = new CronExpression($this->normalizeFrequency($frequency)); $executionTime = ($this->executionTime ?? Carbon::now())->copy()->setTimezone($this->serverTimezone($server)); - $lastDispatched = Cache::get($dedupKey); $previousDue = Carbon::instance($cron->getPreviousRunDate($executionTime, allowCurrentDate: true)); - if ($lastDispatched === null) { - $isDue = $cron->isDue($executionTime); - - if (! $isDue) { - Cache::put($dedupKey, $previousDue->toIso8601String(), 2592000); - } - - return $isDue; - } - - $shouldFire = $previousDue->gt(Carbon::parse($lastDispatched)); - - if (! $shouldFire) { - Cache::put($dedupKey, $previousDue->toIso8601String(), 2592000); - } - - return $shouldFire; + return $previousDue->gte( + $executionTime->copy()->subMinutes(ScheduledJobDeliveryService::CATCH_UP_WINDOW_MINUTES) + ); } private function normalizeFrequency(string $frequency): string diff --git a/app/Jobs/ScheduledTaskJob.php b/app/Jobs/ScheduledTaskJob.php index f7bd5f933d..6435c1a64b 100644 --- a/app/Jobs/ScheduledTaskJob.php +++ b/app/Jobs/ScheduledTaskJob.php @@ -12,6 +12,7 @@ use App\Models\Service; use App\Models\Team; use App\Notifications\ScheduledTask\TaskFailed; use App\Notifications\ScheduledTask\TaskSuccess; +use App\Services\ScheduledJobDeliveryService; use Carbon\Carbon; use Illuminate\Bus\Queueable; use Illuminate\Contracts\Queue\ShouldBeEncrypted; @@ -65,7 +66,7 @@ class ScheduledTaskJob implements ShouldBeEncrypted, ShouldQueue public string $server_timezone = 'UTC'; - public function __construct(ScheduledTask $task) + public function __construct(ScheduledTask $task, public ?string $occurrenceUuid = null) { $this->onQueue(crons_queue()); @@ -106,7 +107,12 @@ class ScheduledTaskJob implements ShouldBeEncrypted, ShouldQueue public function handle(): void { + if ($this->occurrenceUuid && ! app(ScheduledJobDeliveryService::class)->claim($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid)) { + return; + } + $startTime = Carbon::now(); + $failed = false; try { $this->initializeExecutionContext(); @@ -170,6 +176,7 @@ class ScheduledTaskJob implements ShouldBeEncrypted, ShouldQueue // No valid container was found. throw new NonReportableException('ScheduledTaskJob failed: No valid container was found. Is the container name correct?'); } catch (\Throwable $e) { + $failed = true; if ($this->task_log) { $this->task_log->update([ 'status' => 'failed', @@ -192,6 +199,10 @@ class ScheduledTaskJob implements ShouldBeEncrypted, ShouldQueue // Re-throw to trigger Laravel's retry mechanism with backoff throw $e; } finally { + if (! $failed && $this->occurrenceUuid) { + app(ScheduledJobDeliveryService::class)->complete($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid); + } + if ($this->team) { ScheduledTaskDone::dispatch($this->team->id); } @@ -229,6 +240,10 @@ class ScheduledTaskJob implements ShouldBeEncrypted, ShouldQueue */ public function failed(?\Throwable $exception): void { + if ($this->occurrenceUuid) { + app(ScheduledJobDeliveryService::class)->fail($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid); + } + $this->team ??= Team::find($this->task->team_id); Log::channel('scheduled-errors')->error('ScheduledTask permanently failed', [ diff --git a/app/Jobs/VolumeBackupJob.php b/app/Jobs/VolumeBackupJob.php index 0b13c66609..fb7e3376b5 100644 --- a/app/Jobs/VolumeBackupJob.php +++ b/app/Jobs/VolumeBackupJob.php @@ -8,6 +8,7 @@ use App\Models\ScheduledVolumeBackup; use App\Models\ScheduledVolumeBackupExecution; use App\Models\Server; use App\Rules\SafeWebhookUrl; +use App\Services\ScheduledJobDeliveryService; use App\Support\BackupCompression; use Carbon\Carbon; use Illuminate\Bus\Queueable; @@ -32,7 +33,7 @@ class VolumeBackupJob implements ShouldBeEncrypted, ShouldQueue private ?ScheduledVolumeBackupExecution $execution = null; - public function __construct(public ScheduledVolumeBackup $backup) + public function __construct(public ScheduledVolumeBackup $backup, public ?string $occurrenceUuid = null) { $this->onQueue(crons_queue()); $this->timeout = $backup->timeout ?? ScheduledVolumeBackup::DEFAULT_TIMEOUT; @@ -55,6 +56,11 @@ class VolumeBackupJob implements ShouldBeEncrypted, ShouldQueue public function handle(): void { + if ($this->occurrenceUuid && ! app(ScheduledJobDeliveryService::class)->claim($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid)) { + return; + } + + $failed = false; $this->backup->loadMissing(['backupable.resource', 'team', 's3']); $server = $this->backup->server(); $target = $this->backup->backupable; @@ -192,6 +198,7 @@ class VolumeBackupJob implements ShouldBeEncrypted, ShouldQueue ]); } } catch (Throwable $exception) { + $failed = true; $recoveryError = $this->recoverIncompleteBackup($this->execution); $archiveDeleted = $streamToS3; @@ -225,6 +232,10 @@ class VolumeBackupJob implements ShouldBeEncrypted, ShouldQueue throw $exception; } finally { + if (! $failed && $this->occurrenceUuid) { + app(ScheduledJobDeliveryService::class)->complete($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid); + } + $this->execution->update(['finished_at' => now()]); BackupCreated::dispatch($team->id); } @@ -232,6 +243,10 @@ class VolumeBackupJob implements ShouldBeEncrypted, ShouldQueue public function failed(?Throwable $exception): void { + if ($this->occurrenceUuid) { + app(ScheduledJobDeliveryService::class)->fail($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid); + } + $execution = $this->execution ?? $this->backup->executions() ->where('status', 'running') ->latest('id') diff --git a/app/Models/ScheduledJobDelivery.php b/app/Models/ScheduledJobDelivery.php new file mode 100644 index 0000000000..7772a3f707 --- /dev/null +++ b/app/Models/ScheduledJobDelivery.php @@ -0,0 +1,29 @@ + 'immutable_datetime', + 'payload' => 'array', + 'enqueued_at' => 'immutable_datetime', + 'started_at' => 'immutable_datetime', + ]; + } +} diff --git a/app/Models/ScheduledJobState.php b/app/Models/ScheduledJobState.php new file mode 100644 index 0000000000..e267cd6a1d --- /dev/null +++ b/app/Models/ScheduledJobState.php @@ -0,0 +1,18 @@ + 'immutable_datetime', + ]; + } +} diff --git a/app/Services/ScheduledJobDeliveryService.php b/app/Services/ScheduledJobDeliveryService.php new file mode 100644 index 0000000000..0bd4c031c5 --- /dev/null +++ b/app/Services/ScheduledJobDeliveryService.php @@ -0,0 +1,232 @@ +copy()->setTimezone($timezone); + $cron = new CronExpression(VALID_CRON_STRINGS[$frequency] ?? $frequency); + $scheduledFor = Carbon::instance($cron->getPreviousRunDate($executionTime, allowCurrentDate: true)); + + if (! $scheduledFor->gte($executionTime->copy()->subMinutes(self::CATCH_UP_WINDOW_MINUTES))) { + return false; + } + + $delivery = DB::transaction(function () use ($scheduleKey, $scheduledFor, $jobType, $resourceId, $payload): ?ScheduledJobDelivery { + ScheduledJobState::query()->insertOrIgnore([ + 'uuid' => new_public_id(), + 'schedule_key' => $scheduleKey, + 'created_at' => now(), + 'updated_at' => now(), + ]); + + $state = ScheduledJobState::query() + ->where('schedule_key', $scheduleKey) + ->lockForUpdate() + ->firstOrFail(); + + if ($state->last_scheduled_for?->gte($scheduledFor)) { + return null; + } + + $state->update(['last_scheduled_for' => $scheduledFor->utc()]); + + return ScheduledJobDelivery::create([ + 'schedule_key' => $scheduleKey, + 'scheduled_for' => $scheduledFor->utc(), + 'job_type' => $jobType, + 'resource_id' => $resourceId, + 'payload' => $payload, + 'status' => 'pending', + ]); + }); + + if ($delivery === null) { + return false; + } + + return $this->publish($delivery); + } + + public function recordSkipped( + string $scheduleKey, + string $frequency, + string $timezone, + ?Carbon $executionTime = null, + ): bool { + $executionTime = ($executionTime ?? Carbon::now())->copy()->setTimezone($timezone); + $cron = new CronExpression(VALID_CRON_STRINGS[$frequency] ?? $frequency); + $scheduledFor = Carbon::instance($cron->getPreviousRunDate($executionTime, allowCurrentDate: true)); + + if (! $scheduledFor->gte($executionTime->copy()->subMinutes(self::CATCH_UP_WINDOW_MINUTES))) { + return false; + } + + return DB::transaction(function () use ($scheduleKey, $scheduledFor): bool { + ScheduledJobState::query()->insertOrIgnore([ + 'uuid' => new_public_id(), + 'schedule_key' => $scheduleKey, + 'created_at' => now(), + 'updated_at' => now(), + ]); + + $state = ScheduledJobState::query() + ->where('schedule_key', $scheduleKey) + ->lockForUpdate() + ->firstOrFail(); + + if ($state->last_scheduled_for?->gte($scheduledFor)) { + return false; + } + + $state->update(['last_scheduled_for' => $scheduledFor->utc()]); + + return true; + }); + } + + public function publishPending(): void + { + ScheduledJobDelivery::query() + ->where('status', 'pending') + ->orderBy('id') + ->chunkById(100, function ($occurrences): void { + foreach ($occurrences as $occurrence) { + $this->publish($occurrence); + } + }); + } + + public function deleteOldOccurrences(): void + { + ScheduledJobDelivery::query() + ->where('status', 'claimed') + ->where('updated_at', '<', now()->subDays(2)) + ->update([ + 'status' => 'failed', + 'updated_at' => now(), + ]); + + ScheduledJobDelivery::query() + ->whereIn('status', ['failed', 'skipped']) + ->where('created_at', '<', now()->subDays(30)) + ->chunkById(100, function ($occurrences): void { + ScheduledJobDelivery::query()->whereKey($occurrences->modelKeys())->delete(); + }); + } + + public function claim(string $uuid, string $claimToken): bool + { + $claimed = ScheduledJobDelivery::query() + ->where('uuid', $uuid) + ->whereIn('status', ['pending', 'enqueued']) + ->update([ + 'status' => 'claimed', + 'claim_token' => $claimToken, + 'started_at' => now(), + 'updated_at' => now(), + ]); + + if ($claimed === 1) { + return true; + } + + return ScheduledJobDelivery::query() + ->where('uuid', $uuid) + ->where('status', 'claimed') + ->where('claim_token', $claimToken) + ->exists(); + } + + public function complete(string $uuid, string $claimToken): void + { + ScheduledJobDelivery::query() + ->where('uuid', $uuid) + ->where('claim_token', $claimToken) + ->delete(); + } + + public function fail(string $uuid, string $claimToken): void + { + ScheduledJobDelivery::query() + ->where('uuid', $uuid) + ->where('claim_token', $claimToken) + ->update([ + 'status' => 'failed', + 'updated_at' => now(), + ]); + } + + private function publish(ScheduledJobDelivery $occurrence): bool + { + if ($occurrence->status !== 'pending') { + return false; + } + + $job = match ($occurrence->job_type) { + 'scheduled-task' => ($task = ScheduledTask::find($occurrence->resource_id)) + ? new ScheduledTaskJob($task, $occurrence->uuid) + : null, + 'database-backup' => ($backup = ScheduledDatabaseBackup::find($occurrence->resource_id)) + ? new DatabaseBackupJob($backup, $occurrence->uuid) + : null, + 'volume-backup' => ($backup = ScheduledVolumeBackup::find($occurrence->resource_id)) + ? new VolumeBackupJob($backup, $occurrence->uuid) + : null, + 'docker-cleanup' => ($server = Server::find($occurrence->resource_id)) + ? new DockerCleanupJob( + $server, + false, + data_get($occurrence->payload, 'delete_unused_volumes', false), + data_get($occurrence->payload, 'delete_unused_networks', false), + $occurrence->uuid, + ) + : null, + default => null, + }; + + if ($job === null) { + ScheduledJobDelivery::query()->whereKey($occurrence->id)->update(['status' => 'skipped']); + + return false; + } + + dispatch($job); + + ScheduledJobDelivery::query() + ->whereKey($occurrence->id) + ->where('status', 'pending') + ->update([ + 'status' => 'enqueued', + 'enqueued_at' => now(), + 'updated_at' => now(), + ]); + + return true; + } +} diff --git a/database/migrations/2026_09_18_111936_create_scheduled_job_states_and_deliveries_tables.php b/database/migrations/2026_09_18_111936_create_scheduled_job_states_and_deliveries_tables.php new file mode 100644 index 0000000000..6057056351 --- /dev/null +++ b/database/migrations/2026_09_18_111936_create_scheduled_job_states_and_deliveries_tables.php @@ -0,0 +1,49 @@ +id(); + $table->string('uuid')->unique(); + $table->string('schedule_key')->unique(); + $table->timestampTz('last_scheduled_for')->nullable(); + $table->timestamps(); + }); + + Schema::create('scheduled_job_deliveries', function (Blueprint $table) { + $table->id(); + $table->string('uuid')->unique(); + $table->string('schedule_key'); + $table->timestampTz('scheduled_for'); + $table->string('job_type'); + $table->unsignedBigInteger('resource_id'); + $table->json('payload')->nullable(); + $table->string('status')->default('pending'); + $table->string('claim_token')->nullable(); + $table->timestampTz('enqueued_at')->nullable(); + $table->timestampTz('started_at')->nullable(); + $table->timestamps(); + + $table->unique(['schedule_key', 'scheduled_for']); + $table->index(['status', 'created_at']); + }); + } + + /** + * Reverse the migrations. + */ + public function down(): void + { + Schema::dropIfExists('scheduled_job_deliveries'); + Schema::dropIfExists('scheduled_job_states'); + } +}; diff --git a/tests/Feature/ScheduledJobManagerDispatchTest.php b/tests/Feature/ScheduledJobManagerDispatchTest.php index b18cc98e41..7ed396a641 100644 --- a/tests/Feature/ScheduledJobManagerDispatchTest.php +++ b/tests/Feature/ScheduledJobManagerDispatchTest.php @@ -8,14 +8,16 @@ use App\Models\Environment; use App\Models\PrivateKey; use App\Models\Project; use App\Models\ScheduledDatabaseBackup; +use App\Models\ScheduledJobDelivery; +use App\Models\ScheduledJobState; use App\Models\ScheduledTask; use App\Models\Server; use App\Models\StandaloneDocker; use App\Models\StandalonePostgresql; use App\Models\Team; +use App\Services\ScheduledJobDeliveryService; use Illuminate\Foundation\Testing\RefreshDatabase; use Illuminate\Support\Carbon; -use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Queue; @@ -74,7 +76,7 @@ uZx9iFkCELtxrh31QJ68AAAAEXNhaWxANzZmZjY2ZDJlMmRkAQIDBA== Queue::assertPushed(ScheduledTaskJob::class, 101); }); -it('skips expensive dispatch for non-due schedules while seeding dedup cache', function () { +it('skips expensive dispatch for schedules outside the catch-up window', function () { config(['constants.coolify.self_hosted' => true]); Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 1, 0, 'UTC')); Queue::fake(); @@ -91,7 +93,175 @@ it('skips expensive dispatch for non-due schedules while seeding dedup cache', f (new ScheduledJobManager)->handle(); Queue::assertNotPushed(ScheduledTaskJob::class); - expect(Cache::get("scheduled-task:{$task->id}"))->not->toBeNull(); + expect(ScheduledJobDelivery::query()->where('schedule_key', "scheduled-task:{$task->id}")->exists())->toBeFalse(); +}); + +it('dispatches a recently missed daily task when deduplication cache is empty', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 10, 0, 'UTC')); + Queue::fake(); + + $application = createScheduledTaskApplication(); + + ScheduledTask::factory()->create([ + 'team_id' => $application->environment->project->team_id, + 'application_id' => $application->id, + 'frequency' => 'daily', + 'enabled' => true, + ]); + + (new ScheduledJobManager)->handle(); + + Queue::assertPushed(ScheduledTaskJob::class, 1); +}); + +it('dispatches one job when multiple managers evaluate the same occurrence', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); + Queue::fake(); + + $application = createScheduledTaskApplication(); + $task = ScheduledTask::factory()->create([ + 'team_id' => $application->environment->project->team_id, + 'application_id' => $application->id, + 'frequency' => 'daily', + 'enabled' => true, + ]); + + (new ScheduledJobManager)->handle(); + (new ScheduledJobManager)->handle(); + + Queue::assertPushed(ScheduledTaskJob::class, 1); + expect(ScheduledJobDelivery::query()->where('schedule_key', "scheduled-task:{$task->id}")->count())->toBe(1) + ->and(ScheduledJobState::query()->where('schedule_key', "scheduled-task:{$task->id}")->count())->toBe(1); +}); + +it('does not retry an occurrence skipped while its server is not functional', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); + Queue::fake(); + + $application = createScheduledTaskApplication(); + $task = createScheduledApplicationTask($application, ['frequency' => 'daily']); + $server = $task->server(); + $server->settings()->update(['is_reachable' => false]); + + (new ScheduledJobManager)->handle(); + + $server->settings()->update(['is_reachable' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 6, 0, 'UTC')); + (new ScheduledJobManager)->handle(); + + Queue::assertNotPushed(ScheduledTaskJob::class); + expect(ScheduledJobState::query()->where('schedule_key', "scheduled-task:{$task->id}")->value('last_scheduled_for')) + ->not->toBeNull() + ->and(ScheduledJobDelivery::query()->where('schedule_key', "scheduled-task:{$task->id}")->exists()) + ->toBeFalse(); +}); + +it('does not publish a task occurrence when its application is not running', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); + Queue::fake(); + + $application = createScheduledTaskApplication(); + $application->update(['status' => 'stopped']); + $task = createScheduledApplicationTask($application, ['frequency' => 'daily']); + + (new ScheduledJobManager)->handle(); + + Queue::assertNotPushed(ScheduledTaskJob::class); + expect(ScheduledJobState::query()->where('schedule_key', "scheduled-task:{$task->id}")->exists())->toBeTrue() + ->and(ScheduledJobDelivery::query()->where('schedule_key', "scheduled-task:{$task->id}")->exists()) + ->toBeFalse(); +}); + +it('publishes a pending occurrence after a previous publisher interruption', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + Queue::fake(); + + $application = createScheduledTaskApplication(); + $task = ScheduledTask::factory()->create([ + 'team_id' => $application->environment->project->team_id, + 'application_id' => $application->id, + 'frequency' => 'daily', + 'enabled' => true, + ]); + $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, + ]); + + (new ScheduledJobManager)->handle(); + + Queue::assertPushed(ScheduledTaskJob::class, 1); + expect($occurrence->fresh()->status)->toBe('enqueued'); +}); + +it('allows only one worker to claim an occurrence', function () { + $occurrence = ScheduledJobDelivery::create([ + 'schedule_key' => 'scheduled-task:claim-test', + 'scheduled_for' => now(), + 'job_type' => 'scheduled-task', + 'resource_id' => 1, + 'status' => 'enqueued', + ]); + $service = app(ScheduledJobDeliveryService::class); + + expect($service->claim($occurrence->uuid, 'worker-a'))->toBeTrue() + ->and($service->claim($occurrence->uuid, 'worker-b'))->toBeFalse() + ->and($service->claim($occurrence->uuid, 'worker-a'))->toBeTrue() + ->and($occurrence->fresh()->status)->toBe('claimed'); + + $service->complete($occurrence->uuid, 'worker-a'); + + expect($occurrence->fresh())->toBeNull(); +}); + +it('deletes only failed old delivery records', function () { + $oldFailed = ScheduledJobDelivery::create([ + 'schedule_key' => 'scheduled-task:old-failed', + 'scheduled_for' => now()->subDays(31), + 'job_type' => 'scheduled-task', + 'resource_id' => 1, + 'status' => 'failed', + ]); + $oldPending = ScheduledJobDelivery::create([ + 'schedule_key' => 'scheduled-task:old-pending', + 'scheduled_for' => now()->subDays(31), + 'job_type' => 'scheduled-task', + 'resource_id' => 1, + 'status' => 'pending', + ]); + $oldFailed->timestamps = false; + $oldFailed->forceFill(['created_at' => now()->subDays(31)])->save(); + $oldPending->timestamps = false; + $oldPending->forceFill(['created_at' => now()->subDays(31)])->save(); + + app(ScheduledJobDeliveryService::class)->deleteOldOccurrences(); + + expect($oldFailed->fresh())->toBeNull() + ->and($oldPending->fresh())->not->toBeNull(); +}); + +it('marks stale claimed deliveries as failed', function () { + $delivery = ScheduledJobDelivery::create([ + 'schedule_key' => 'scheduled-task:stale-claim', + 'scheduled_for' => now()->subDays(3), + 'job_type' => 'scheduled-task', + 'resource_id' => 1, + 'status' => 'claimed', + 'claim_token' => 'lost-worker', + ]); + $delivery->timestamps = false; + $delivery->forceFill(['updated_at' => now()->subDays(3)])->save(); + + app(ScheduledJobDeliveryService::class)->deleteOldOccurrences(); + + expect($delivery->fresh()->status)->toBe('failed'); }); it('dispatches the instance coolify-db backup even when its id is zero', function () {