diff --git a/.ai/lessons.md b/.ai/lessons.md index 040431f38c..af4080da3b 100644 --- a/.ai/lessons.md +++ b/.ai/lessons.md @@ -60,8 +60,8 @@ ## 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. -- Store the next due time on each schedule and select only due rows; claim a run with `UPDATE ... WHERE next_run_at = :old`. Do not evaluate every cron expression each minute: on a remote database the per-row queries alone exceed the one-minute budget. - Run a per-minute dispatcher from the scheduler process, not as a queued job, so queue backlog cannot delay it. +- A Redis instance that holds queues must use `maxmemory-policy noeviction`; an evicting policy deletes queued jobs without an error. Before changing scheduler code for missed runs, check `INFO stats` `evicted_keys`. - Keep pending occurrences recoverable across publisher interruptions, and define an explicit bounded policy for late or offline schedules. - Horizon workers are long-lived: flush every static or `once()` cache (for example `Server::flushIdentityMap()`) in `Queue::before`, or later jobs decide with stale state. diff --git a/app/Actions/Shared/MigrateResourceToDestination.php b/app/Actions/Shared/MigrateResourceToDestination.php index ad69093412..b0979e1677 100644 --- a/app/Actions/Shared/MigrateResourceToDestination.php +++ b/app/Actions/Shared/MigrateResourceToDestination.php @@ -23,7 +23,6 @@ use App\Models\StandalonePostgresql; use App\Models\StandaloneRedis; use App\Models\StandaloneSqlite; use App\Models\SwarmDocker; -use App\Services\ScheduleNextRunRecalculator; use Illuminate\Support\Collection; use Illuminate\Support\Facades\Bus; use Illuminate\Validation\ValidationException; @@ -152,7 +151,6 @@ class MigrateResourceToDestination } $resource->fill($payload)->save(); - app(ScheduleNextRunRecalculator::class)->forResource($resource); if ($resource instanceof Application) { $resource->additional_networks()->detach(); diff --git a/app/Console/Commands/DispatchScheduledJobs.php b/app/Console/Commands/DispatchScheduledJobs.php index 3d4b53c2ae..2c1cc8df04 100644 --- a/app/Console/Commands/DispatchScheduledJobs.php +++ b/app/Console/Commands/DispatchScheduledJobs.php @@ -11,9 +11,9 @@ class DispatchScheduledJobs extends Command protected $description = 'Dispatch the due scheduled backups, tasks, volume backups, and Docker cleanups'; - public function handle(ScheduledJobManager $manager): int + public function handle(): int { - $manager->handle(); + ScheduledJobManager::dispatchSync(); return self::SUCCESS; } diff --git a/app/Console/Commands/ScheduledJobDiagnostics.php b/app/Console/Commands/ScheduledJobDiagnostics.php index f66efd57b2..61f26265a9 100644 --- a/app/Console/Commands/ScheduledJobDiagnostics.php +++ b/app/Console/Commands/ScheduledJobDiagnostics.php @@ -2,67 +2,254 @@ namespace App\Console\Commands; -use App\Jobs\ScheduledJobManager; +use App\Models\DockerCleanupExecution; use App\Models\ScheduledDatabaseBackup; -use App\Models\ScheduledJobDelivery; use App\Models\ScheduledTask; -use App\Models\ScheduledVolumeBackup; -use App\Models\ServerSetting; +use App\Models\Server; +use App\Models\Team; use Illuminate\Console\Command; -use Illuminate\Database\Eloquent\Builder; use Illuminate\Support\Carbon; +use Illuminate\Support\Collection; use Illuminate\Support\Facades\Cache; class ScheduledJobDiagnostics extends Command { - protected $signature = 'scheduled:diagnostics {--all : List every schedule, not only the overdue ones and the ones without a next run}'; + protected $signature = 'scheduled:diagnostics + {--type=all : Type to inspect: docker-cleanup, backups, tasks, server-jobs, all} + {--server= : Filter by server ID}'; - protected $description = 'Show the scheduler heartbeat, overdue schedules, schedules without a next run, and open deliveries'; + protected $description = 'Inspect dedup cache state and scheduling decisions for all scheduled jobs'; public function handle(): int { - $heartbeat = Cache::get('scheduled-job-manager:heartbeat'); - $heartbeat - ? $this->info("Scheduler heartbeat: {$heartbeat} (".Carbon::parse($heartbeat)->diffForHumans().')') - : $this->error('Scheduler heartbeat: missing. The scheduled:dispatch command does not run.'); - $this->newLine(); + $type = $this->option('type'); + $serverFilter = $this->option('server'); - $overdueBefore = now()->subMinutes(ScheduledJobManager::LATE_RUN_WINDOW_MINUTES); - $schedules = [ - 'Database backups' => [ScheduledDatabaseBackup::query()->where('enabled', true), 'next_run_at', 'frequency'], - 'Scheduled tasks' => [ScheduledTask::query()->where('enabled', true), 'next_run_at', 'frequency'], - 'Volume backups' => [ScheduledVolumeBackup::query()->where('enabled', true), 'next_run_at', 'frequency'], - 'Docker cleanups' => [ServerSetting::query(), 'docker_cleanup_next_run_at', 'docker_cleanup_frequency'], - ]; + $this->outputHeartbeat(); - foreach ($schedules as $label => [$query, $column, $frequencyColumn]) { - /** @var Builder $query */ - $problems = (clone $query)->where(fn (Builder $query) => $query->whereNull($column)->orWhere($column, '<', $overdueBefore)); - $this->info(sprintf('=== %s: %d enabled, %d overdue or without a next run ===', $label, (clone $query)->count(), (clone $problems)->count())); - - $rows = ($this->option('all') ? $query : $problems) - ->orderBy($column) - ->limit(100) - ->get() - ->map(fn ($schedule) => [$schedule->getKey(), $schedule->getRawOriginal($frequencyColumn), $schedule->{$column}?->toIso8601String() ?? (next_cron_run_at((string) $schedule->getRawOriginal($frequencyColumn), null, now()) ? 'not calculated' : 'invalid frequency')]); - - if ($rows->isNotEmpty()) { - $this->table(['ID', 'Frequency', 'Next run (UTC)'], $rows); - } - $this->newLine(); + if (in_array($type, ['all', 'docker-cleanup'])) { + $this->inspectDockerCleanups($serverFilter); } - $this->info('=== Open and failed deliveries (failed ones are kept for 30 days) ==='); - $this->table( - ['Job type', 'Status', 'Count', 'Oldest due (UTC)'], - ScheduledJobDelivery::query() - ->whereIn('status', ['pending', 'enqueued', 'claimed', 'failed']) - ->selectRaw('job_type, status, count(*) as total, min(scheduled_for) as oldest') - ->groupBy('job_type', 'status') - ->get() - ->map(fn ($row) => [$row->job_type, $row->status, $row->total, $row->oldest]), - ); + if (in_array($type, ['all', 'backups'])) { + $this->inspectBackups(); + } + + if (in_array($type, ['all', 'tasks'])) { + $this->inspectTasks(); + } + + if (in_array($type, ['all', 'server-jobs'])) { + $this->inspectServerJobs($serverFilter); + } return self::SUCCESS; } + + private function outputHeartbeat(): void + { + $heartbeat = Cache::get('scheduled-job-manager:heartbeat'); + if ($heartbeat) { + $age = Carbon::parse($heartbeat)->diffForHumans(); + $this->info("Scheduler heartbeat: {$heartbeat} ({$age})"); + } else { + $this->error('Scheduler heartbeat: MISSING — ScheduledJobManager may not be running'); + } + $this->newLine(); + } + + private function inspectDockerCleanups(?string $serverFilter): void + { + $this->info('=== Docker Cleanup Jobs ==='); + + $servers = $this->getServers($serverFilter); + + $rows = []; + foreach ($servers as $server) { + $frequency = data_get($server->settings, 'docker_cleanup_frequency', '0 * * * *'); + if (isset(VALID_CRON_STRINGS[$frequency])) { + $frequency = VALID_CRON_STRINGS[$frequency]; + } + + $dedupKey = "docker-cleanup:{$server->id}"; + $cacheValue = Cache::get($dedupKey); + $timezone = data_get($server->settings, 'server_timezone', config('app.timezone')); + + if (validate_timezone($timezone) === false) { + $timezone = config('app.timezone'); + } + + $wouldFire = shouldRunCronNow($frequency, $timezone, $dedupKey); + + $lastExecution = DockerCleanupExecution::where('server_id', $server->id) + ->latest() + ->first(); + + $rows[] = [ + $server->id, + $server->name, + $timezone, + $frequency, + $dedupKey, + $cacheValue ?? '', + $wouldFire ? 'YES' : 'no', + $lastExecution ? $lastExecution->status.' @ '.$lastExecution->created_at : 'never', + ]; + } + + $this->table( + ['ID', 'Server', 'TZ', 'Frequency', 'Dedup Key', 'Cache Value', 'Would Fire', 'Last Execution'], + $rows + ); + $this->newLine(); + } + + private function inspectBackups(): void + { + $this->info('=== Scheduled Backups ==='); + + $backups = ScheduledDatabaseBackup::with(['database']) + ->where('enabled', true) + ->get(); + + $rows = []; + foreach ($backups as $backup) { + $server = $backup->server(); + $frequency = $backup->frequency; + if (isset(VALID_CRON_STRINGS[$frequency])) { + $frequency = VALID_CRON_STRINGS[$frequency]; + } + + $dedupKey = "scheduled-backup:{$backup->id}"; + $cacheValue = Cache::get($dedupKey); + $timezone = $server ? data_get($server->settings, 'server_timezone', config('app.timezone')) : config('app.timezone'); + + if (validate_timezone($timezone) === false) { + $timezone = config('app.timezone'); + } + + $wouldFire = shouldRunCronNow($frequency, $timezone, $dedupKey); + + $rows[] = [ + $backup->id, + $backup->database_type ?? 'unknown', + $server?->name ?? 'N/A', + $frequency, + $cacheValue ?? '', + $wouldFire ? 'YES' : 'no', + ]; + } + + $this->table( + ['Backup ID', 'DB Type', 'Server', 'Frequency', 'Cache Value', 'Would Fire'], + $rows + ); + $this->newLine(); + } + + private function inspectTasks(): void + { + $this->info('=== Scheduled Tasks ==='); + + $tasks = ScheduledTask::with(['service', 'application']) + ->where('enabled', true) + ->get(); + + $rows = []; + foreach ($tasks as $task) { + $server = $task->server(); + $frequency = $task->frequency; + if (isset(VALID_CRON_STRINGS[$frequency])) { + $frequency = VALID_CRON_STRINGS[$frequency]; + } + + $dedupKey = "scheduled-task:{$task->id}"; + $cacheValue = Cache::get($dedupKey); + $timezone = $server ? data_get($server->settings, 'server_timezone', config('app.timezone')) : config('app.timezone'); + + if (validate_timezone($timezone) === false) { + $timezone = config('app.timezone'); + } + + $wouldFire = shouldRunCronNow($frequency, $timezone, $dedupKey); + + $rows[] = [ + $task->id, + $task->name, + $server?->name ?? 'N/A', + $frequency, + $cacheValue ?? '', + $wouldFire ? 'YES' : 'no', + ]; + } + + $this->table( + ['Task ID', 'Name', 'Server', 'Frequency', 'Cache Value', 'Would Fire'], + $rows + ); + $this->newLine(); + } + + private function inspectServerJobs(?string $serverFilter): void + { + $this->info('=== Server Manager Jobs ==='); + + $servers = $this->getServers($serverFilter); + + $rows = []; + foreach ($servers as $server) { + $timezone = data_get($server->settings, 'server_timezone', config('app.timezone')); + if (validate_timezone($timezone) === false) { + $timezone = config('app.timezone'); + } + + $dedupKeys = [ + "server-patch-check:{$server->id}" => '0 0 * * 0', + "server-check:{$server->id}" => isCloud() ? '*/5 * * * *' : '* * * * *', + "server-storage-check:{$server->id}" => data_get($server->settings, 'server_disk_usage_check_frequency', '0 23 * * *'), + ]; + + foreach ($dedupKeys as $dedupKey => $frequency) { + if (isset(VALID_CRON_STRINGS[$frequency])) { + $frequency = VALID_CRON_STRINGS[$frequency]; + } + + $cacheValue = Cache::get($dedupKey); + $wouldFire = shouldRunCronNow($frequency, $timezone, $dedupKey); + + $rows[] = [ + $server->id, + $server->name, + $dedupKey, + $frequency, + $cacheValue ?? '', + $wouldFire ? 'YES' : 'no', + ]; + } + } + + $this->table( + ['Server ID', 'Server', 'Dedup Key', 'Frequency', 'Cache Value', 'Would Fire'], + $rows + ); + $this->newLine(); + } + + private function getServers(?string $serverFilter): Collection + { + $query = Server::with('settings')->where('ip', '!=', '1.2.3.4'); + + if ($serverFilter) { + $query->where('id', $serverFilter); + } + + if (isCloud()) { + $servers = $query->whereRelation('team.subscription', 'stripe_invoice_paid', true)->get(); + $own = Team::find(0)?->servers()->with('settings')->get() ?? collect(); + + return $servers->merge($own); + } + + return $query->get(); + } } diff --git a/app/Console/Kernel.php b/app/Console/Kernel.php index 4bf442d3bf..7d45ff4b53 100644 --- a/app/Console/Kernel.php +++ b/app/Console/Kernel.php @@ -78,7 +78,7 @@ class Kernel extends ConsoleKernel $this->scheduleInstance->job(new ServerManagerJob)->everyMinute()->onOneServer(); // Scheduled Jobs (Backups & Tasks) - $this->scheduleScheduledJobs(); + $this->scheduleScheduledJobManager(); $this->scheduleInstance->command('uploads:clear')->everyTwoMinutes(); @@ -99,7 +99,7 @@ class Kernel extends ConsoleKernel $this->pullImages(); // Scheduled Jobs (Backups & Tasks) - $this->scheduleScheduledJobs(); + $this->scheduleScheduledJobManager(); $this->scheduleInstance->job(new RegenerateSslCertJob)->twiceDaily()->onOneServer(); @@ -120,14 +120,15 @@ class Kernel extends ConsoleKernel } /** - * Run the dispatcher in the scheduler process, so a busy queue cannot delay it. Parallel runs - * are safe because each occurrence is claimed with an atomic update of next_run_at. + * 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. */ - private function scheduleScheduledJobs(): void + private function scheduleScheduledJobManager(): void { $this->scheduleInstance->command('scheduled:dispatch') ->everyMinute() ->onOneServer() + ->withoutOverlapping(5) ->runInBackground(); } diff --git a/app/Jobs/ScheduledJobManager.php b/app/Jobs/ScheduledJobManager.php index 4815a853c9..8e4a5b26a8 100644 --- a/app/Jobs/ScheduledJobManager.php +++ b/app/Jobs/ScheduledJobManager.php @@ -4,199 +4,438 @@ namespace App\Jobs; use App\Actions\Shared\DeleteScheduledVolumeBackup; use App\Models\ScheduledDatabaseBackup; -use App\Models\ScheduledJobDelivery; use App\Models\ScheduledTask; use App\Models\ScheduledVolumeBackup; use App\Models\ScheduledVolumeBackupExecution; use App\Models\Server; +use App\Models\ServerSetting; use App\Models\ServiceDatabase; +use App\Models\Subscription; +use App\Models\Team; use App\Services\ScheduledJobDeliveryService; -use Carbon\CarbonImmutable; +use Cron\CronExpression; +use Illuminate\Bus\Queueable; +use Illuminate\Contracts\Queue\ShouldQueue; use Illuminate\Database\Eloquent\Builder; -use Illuminate\Database\Eloquent\Model; use Illuminate\Database\Eloquent\Relations\MorphTo; +use Illuminate\Foundation\Bus\Dispatchable; +use Illuminate\Queue\InteractsWithQueue; +use Illuminate\Queue\Middleware\WithoutOverlapping; +use Illuminate\Queue\SerializesModels; +use Illuminate\Support\Carbon; use Illuminate\Support\Facades\Cache; -use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Log; +use Illuminate\Support\Facades\Redis; -/** - * Dispatches the due scheduled backups, tasks, volume backups, and Docker cleanups. - * - * Each schedule stores its next due time in next_run_at. A run selects only due rows and claims - * each occurrence with an atomic update of next_run_at, so parallel runs on several nodes never - * dispatch the same occurrence twice. The `scheduled:dispatch` command runs it every minute. - */ -class ScheduledJobManager +class ScheduledJobManager implements ShouldQueue { + use Dispatchable, InteractsWithQueue, Queueable, SerializesModels; + private const CHUNK_SIZE = 100; /** - * A task or Docker cleanup that is due longer than this is logged as missed and not run. - * Database and volume backups always run once, also when they are late. + * The time when this job execution started. + * Used to ensure all scheduled items are evaluated against the same point in time. */ - public const LATE_RUN_WINDOW_MINUTES = 10; + private ?Carbon $executionTime = null; - private CarbonImmutable $now; + private int $dispatchedCount = 0; - private ScheduledJobDeliveryService $deliveries; + private int $skippedCount = 0; - /** @var array{dispatched: int, skipped: int, missed: int} */ - private array $counts = ['dispatched' => 0, 'skipped' => 0, 'missed' => 0]; - - public function handle(): void + /** + * Create a new job instance. + */ + public function __construct() { - $this->now = CarbonImmutable::now(); - $this->deliveries = app(ScheduledJobDeliveryService::class); - $this->counts = ['dispatched' => 0, 'skipped' => 0, 'missed' => 0]; - - // Read the current server state, not servers cached by an earlier run in this process. - Server::flushIdentityMap(); - - 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()); - $this->runStep('volume backups', function () { - $this->recoverStoppedVolumeBackupContainers(); - $this->processVolumeBackups(); - }); - $this->runStep('docker cleanups', fn () => $this->processDockerCleanups()); - - Log::channel('scheduled')->info('ScheduledJobManager completed', [ - 'execution_time' => $this->now->toIso8601String(), - 'duration_ms' => $this->now->diffInMilliseconds(CarbonImmutable::now()), - ...$this->counts, - ]); - - // The UI uses the heartbeat to show when the scheduler has stopped. - rescue(fn () => Cache::put('scheduled-job-manager:heartbeat', now()->toIso8601String(), 300), report: false); + $this->onQueue(crons_queue()); } - private function runStep(string $name, callable $step): void + /** + * Get the middleware the job should pass through. + */ + public function middleware(): array + { + // 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(); + + return [ + (new WithoutOverlapping('scheduled-job-manager')) + ->expireAfter(90) // Lock expires after 90s to handle high-load environments with many tasks + ->dontRelease(), // Don't re-queue on lock conflict + ]; + } + + /** + * Clear a stale WithoutOverlapping lock if it has no TTL (TTL = -1). + * + * 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 { try { - $step(); + $cachePrefix = config('cache.prefix', ''); + $lockKey = $cachePrefix.'laravel-queue-overlap:'.self::class.':scheduled-job-manager'; + + $ttl = Redis::connection('default')->ttl($lockKey); + + if ($ttl === -1) { + Redis::connection('default')->del($lockKey); + Log::channel('scheduled')->warning('Cleared stale ScheduledJobManager lock', [ + 'lock_key' => $lockKey, + ]); + } } catch (\Throwable $e) { - Log::channel('scheduled-errors')->error("Failed to process {$name}", [ + // Never let lock cleanup failure prevent the job from running + Log::channel('scheduled-errors')->error('Failed to check/clear stale lock', [ 'error' => $e->getMessage(), - 'trace' => $e->getTraceAsString(), ]); } } - private function processDatabaseBackups(): void + public function handle(): void { - $this->due(ScheduledDatabaseBackup::query()) - ->with([ - 'team.subscription', - 'database' => fn (MorphTo $morphTo) => $morphTo->morphWith([ - ServiceDatabase::class => ['service.destination.server.settings', 'service.destination.server.team.subscription'], - ...array_fill_keys(STANDALONE_DATABASE_MODELS, ['destination.server.settings', 'destination.server.team.subscription']), - ]), - ]) - ->chunkById(self::CHUNK_SIZE, fn ($backups) => $backups->each(fn (ScheduledDatabaseBackup $backup) => $this->processItem( - 'backup', - ['backup_id' => $backup->id, 'database_id' => $backup->database_id, 'database_type' => $backup->database_type, 'team_id' => $backup->team_id], - fn () => $this->processDatabaseBackup($backup), - ))); - } + // Freeze the execution time at the start of the job + $this->executionTime = Carbon::now(); + $this->dispatchedCount = 0; + $this->skippedCount = 0; - private function processDatabaseBackup(ScheduledDatabaseBackup $backup): array - { - $server = $backup->server(); + $this->logStart(); - if (blank($backup->database) || blank($server)) { - $backup->delete(); - - return ['deleted' => blank($backup->database) ? 'database_deleted' : 'server_deleted']; + try { + app(ScheduledJobDeliveryService::class)->recoverStaleEnqueued(); + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Failed to recover stale enqueued occurrences', [ + 'error' => $e->getMessage(), + ]); } - return $this->runOccurrence( - schedule: $backup, - frequency: $backup->frequency, - server: $server, - runLate: true, - skipReason: fn () => $this->serverSkipReason($server), - delivery: ['schedule_key' => "scheduled-backup:{$backup->id}", 'job_type' => 'database-backup', 'resource_id' => $backup->id], - ); - } + app(ScheduledJobDeliveryService::class)->publishPending(); - private function processScheduledTasks(): void - { - $this->due(ScheduledTask::query()) - ->with([ - 'service.destination.server.settings', - 'service.destination.server.team.subscription', - 'application.destination.server.settings', - 'application.destination.server.team.subscription', - ]) - ->chunkById(self::CHUNK_SIZE, fn ($tasks) => $tasks->each(fn (ScheduledTask $task) => $this->processItem( - 'task', - ['task_id' => $task->id, 'task_name' => $task->name, 'team_id' => $task->team_id], - fn () => $this->processScheduledTask($task), - ))); - } - - private function processScheduledTask(ScheduledTask $task): array - { - $server = $task->server(); - - if (blank($server) || (! $task->service && ! $task->application)) { - $task->delete(); - - return ['deleted' => blank($server) ? 'server_deleted' : 'resource_deleted']; + // Process scheduled backups and tasks together so neither type starves the other. + try { + $this->processScheduledBackupsAndTasks(); + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Failed to process scheduled backups and tasks', [ + 'error' => $e->getMessage(), + 'trace' => $e->getTraceAsString(), + ]); } - return $this->runOccurrence( - schedule: $task, - frequency: $task->frequency, - server: $server, - runLate: false, - skipReason: fn () => $this->serverSkipReason($server) ?? $this->taskResourceSkipReason($task), - delivery: ['schedule_key' => "scheduled-task:{$task->id}", 'job_type' => 'scheduled-task', 'resource_id' => $task->id], - ); + try { + $this->recoverStoppedVolumeBackupContainers(); + $this->processScheduledVolumeBackups(); + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Failed to process scheduled volume backups', [ + 'error' => $e->getMessage(), + 'trace' => $e->getTraceAsString(), + ]); + } + + // Process Docker cleanups - don't let failures stop the job manager + try { + $this->processDockerCleanups(); + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Failed to process docker cleanups', [ + 'error' => $e->getMessage(), + 'trace' => $e->getTraceAsString(), + ]); + } + + Log::channel('scheduled')->info('ScheduledJobManager completed', [ + 'execution_time' => $this->executionTime->toIso8601String(), + 'duration_ms' => $this->executionTime->diffInMilliseconds(Carbon::now()), + 'dispatched' => $this->dispatchedCount, + 'skipped' => $this->skippedCount, + 'host' => gethostname(), + 'pid' => getmypid(), + 'memory_mb' => round(memory_get_usage(true) / 1048576, 1), + 'memory_peak_mb' => round(memory_get_peak_usage(true) / 1048576, 1), + 'open_occurrences_over_15_minutes' => rescue(fn () => app(ScheduledJobDeliveryService::class)->staleOpenOccurrenceCounts(), [], report: false), + ]); + + // Write heartbeat so the UI can detect when the scheduler has stopped + try { + Cache::put('scheduled-job-manager:heartbeat', now()->toIso8601String(), 300); + } catch (\Throwable) { + // Non-critical; don't let heartbeat failure affect the job + } } - private function processVolumeBackups(): void + /** + * Log which worker runs the manager and how late it starts. The manager only catches up + * occurrences from the last CATCH_UP_WINDOW_MINUTES, so a larger delay skips them silently. + */ + private function logStart(): void { - $this->due(ScheduledVolumeBackup::query()) + $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); + $queuedAt = data_get($this->job?->payload(), 'createdAt'); + + $context = [ + 'execution_time' => $this->executionTime->toIso8601String(), + 'host' => gethostname(), + 'pid' => getmypid(), + 'queue_wait_seconds' => $queuedAt ? $this->executionTime->timestamp - (int) $queuedAt : null, + 'seconds_since_previous_start' => $previousStartedAt ? (int) Carbon::parse($previousStartedAt)->diffInSeconds($this->executionTime) : null, + 'memory_mb' => round(memory_get_usage(true) / 1048576, 1), + ]; + + Log::channel('scheduled')->info('ScheduledJobManager started', $context); + + $catchUpWindowSeconds = ScheduledJobDeliveryService::CATCH_UP_WINDOW_MINUTES * 60; + if ($context['queue_wait_seconds'] > $catchUpWindowSeconds || $context['seconds_since_previous_start'] > $catchUpWindowSeconds) { + Log::channel('scheduled-errors')->warning('ScheduledJobManager started late; occurrences due before the catch-up window were not run', $context); + } + } + + private function processScheduledBackupsAndTasks(): void + { + $lastBackupId = null; + $lastTaskId = null; + + do { + $backups = $this->scheduledBackupQuery($lastBackupId)->get(); + $tasks = $this->scheduledTaskQuery($lastTaskId)->get(); + + if ($backups->isNotEmpty()) { + $lastBackupId = $backups->last()->id; + } + + if ($tasks->isNotEmpty()) { + $lastTaskId = $tasks->last()->id; + } + + $this->processInterleavedDueSchedules( + $this->dueScheduledBackups($backups), + $this->dueScheduledTasks($tasks), + ); + } while ($backups->isNotEmpty() || $tasks->isNotEmpty()); + } + + /** + * @param array $dueBackups + * @param array $dueTasks + */ + private function processInterleavedDueSchedules(array $dueBackups, array $dueTasks): void + { + $maxCount = max(count($dueBackups), count($dueTasks)); + + for ($index = 0; $index < $maxCount; $index++) { + if (isset($dueBackups[$index])) { + $this->processScheduledBackup($dueBackups[$index]['backup'], $dueBackups[$index]['server']); + } + + if (isset($dueTasks[$index])) { + $this->processScheduledTask($dueTasks[$index]['task'], $dueTasks[$index]['server']); + } + } + } + + /** + * Load the server chain of each backup with the chunk. Otherwise $backup->server() + * queries the destination, service, server, and settings once for each backup. + */ + private function scheduledBackupQuery(?int $lastBackupId): Builder + { + return ScheduledDatabaseBackup::with([ + 'team.subscription', + 'database' => fn (MorphTo $morphTo) => $morphTo->morphWith([ + ServiceDatabase::class => ['service.destination.server.settings', 'service.destination.server.team.subscription'], + ...array_fill_keys(STANDALONE_DATABASE_MODELS, ['destination.server.settings', 'destination.server.team.subscription']), + ]), + ]) + ->where('enabled', true) + ->when($lastBackupId !== null, fn (Builder $query) => $query->where('id', '>', $lastBackupId)) + ->orderBy('id') + ->limit(self::CHUNK_SIZE); + } + + private function scheduledTaskQuery(?int $lastTaskId): Builder + { + return ScheduledTask::with([ + 'service.destination.server.settings', + 'service.destination.server.team.subscription', + 'application.destination.server.settings', + 'application.destination.server.team.subscription', + ]) + ->where('enabled', true) + ->when($lastTaskId !== null, fn (Builder $query) => $query->where('id', '>', $lastTaskId)) + ->orderBy('id') + ->limit(self::CHUNK_SIZE); + } + + /** + * @param iterable $backups + * @return array + */ + private function dueScheduledBackups(iterable $backups): array + { + $dueBackups = []; + + foreach ($backups as $backup) { + try { + $server = $backup->server(); + + if (blank(data_get($backup, 'database')) || blank($server)) { + $this->processScheduledBackup($backup, $server); + + continue; + } + + if ($this->isDueCandidateBeforeExpensiveChecks($backup->frequency, $server)) { + $dueBackups[] = [ + 'backup' => $backup, + 'server' => $server, + ]; + } + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Error prechecking backup', [ + 'backup_id' => $backup->id, + 'error' => $e->getMessage(), + ]); + } + } + + return $dueBackups; + } + + /** + * @param iterable $tasks + * @return array + */ + private function dueScheduledTasks(iterable $tasks): array + { + $dueTasks = []; + + foreach ($tasks as $task) { + try { + $server = $task->server(); + + if (blank($server) || (! $task->service && ! $task->application)) { + $this->processScheduledTask($task, $server); + + continue; + } + + if ($this->isDueCandidateBeforeExpensiveChecks($task->frequency, $server)) { + $dueTasks[] = [ + 'task' => $task, + 'server' => $server, + ]; + } + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Error prechecking task', [ + 'task_id' => $task->id, + 'error' => $e->getMessage(), + ]); + } + } + + return $dueTasks; + } + + private function processScheduledBackup(ScheduledDatabaseBackup $backup, ?Server $precheckedServer = null): void + { + try { + $server = $precheckedServer ?? $backup->server(); + $skipReason = $this->getBackupSkipReason($backup, $server); + if ($skipReason !== null) { + if ($server === null || $this->recordSkippedOccurrence($backup->frequency, $server, "scheduled-backup:{$backup->id}")) { + $this->skippedCount++; + $this->logBackupSkip($backup, $skipReason, $server); + } + + return; + } + + 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, + 'database_id' => $backup->database_id, + 'database_type' => $backup->database_type, + 'team_id' => $backup->team_id ?? null, + 'server_id' => $server->id, + ]); + } + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Error processing backup', [ + 'backup_id' => $backup->id, + 'error' => $e->getMessage(), + ]); + } + } + + private function processScheduledTask(ScheduledTask $task, ?Server $precheckedServer = null): void + { + try { + $server = $precheckedServer ?? $task->server(); + $criticalSkip = $this->getTaskCriticalSkipReason($task, $server); + if ($criticalSkip !== null) { + if ($server === null || $this->recordSkippedOccurrence($task->frequency, $server, "scheduled-task:{$task->id}")) { + $this->skippedCount++; + $this->logTaskSkip($task, $criticalSkip, $server); + } + + return; + } + + $runtimeSkip = $this->getTaskRuntimeSkipReason($task); + if ($runtimeSkip !== null) { + 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; + } + + $this->dispatchedCount++; + Log::channel('scheduled')->info('Task dispatched', [ + 'task_id' => $task->id, + 'task_name' => $task->name, + 'team_id' => $server->team_id, + 'server_id' => $server->id, + ]); + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Error processing task', [ + 'task_id' => $task->id, + 'error' => $e->getMessage(), + ]); + } + } + + private function processScheduledVolumeBackups(): void + { + ScheduledVolumeBackup::query() ->with(['backupable', 'team.subscription']) - ->chunkById(self::CHUNK_SIZE, fn ($backups) => $backups->each(fn (ScheduledVolumeBackup $backup) => $this->processItem( - 'volume_backup', - ['backup_id' => $backup->id, 'backupable_type' => $backup->backupable_type, 'backupable_id' => $backup->backupable_id, 'team_id' => $backup->team_id], - fn () => $this->processVolumeBackup($backup), - ))); - } - - private function processVolumeBackup(ScheduledVolumeBackup $backup): array - { - $server = $backup->server(); - - // Wait without claiming: the backup runs late when the recovery is done. - if ($backup->executions()->where(fn (Builder $query) => $query->where('stop_recovery_pending', true)->orWhere('s3_cleanup_pending', true))->exists()) { - return ['waiting' => 'container_recovery_pending']; - } - - if (! $backup->backupable) { - DeleteScheduledVolumeBackup::run($backup, $server); - - return ['deleted' => 'resource_deleted']; - } - - if (! $server) { - return ['waiting' => 'server_missing']; - } - - return $this->runOccurrence( - schedule: $backup, - frequency: $backup->frequency, - server: $server, - runLate: true, - skipReason: fn () => $this->serverSkipReason($server), - delivery: ['schedule_key' => "scheduled-volume-backup:{$backup->id}", 'job_type' => 'volume-backup', 'resource_id' => $backup->id], - ); + ->where('enabled', true) + ->chunkById(self::CHUNK_SIZE, function ($backups): void { + foreach ($backups as $backup) { + $this->processScheduledVolumeBackup($backup); + } + }); } private function recoverStoppedVolumeBackupContainers(): void @@ -212,241 +451,367 @@ class ScheduledJobManager }); } - private function processDockerCleanups(): void + private function processScheduledVolumeBackup(ScheduledVolumeBackup $backup): void { - $query = Server::query() - ->with('settings') - ->whereNotNull('ip') - ->where('ip', '!=', '') - ->whereNotIn('ip', Server::PLACEHOLDER_IPS) - ->whereHas('settings', fn (Builder $query) => $query - ->whereNull('docker_cleanup_next_run_at') - ->orWhere('docker_cleanup_next_run_at', '<=', $this->now)); + try { + $server = $backup->server(); - if (isCloud()) { - $query->with('team.subscription')->where(fn (Builder $query) => $query - ->where('team_id', 0) - ->orWhereRelation('team.subscription', 'stripe_invoice_paid', true)); - } + if ($backup->executions() + ->where(fn (Builder $query) => $query + ->where('stop_recovery_pending', true) + ->orWhere('s3_cleanup_pending', true)) + ->exists()) { + $this->skippedCount++; + $this->logSkip('volume_backup', 'container_recovery_pending', [ + 'backup_id' => $backup->id, + 'team_id' => $backup->team_id, + ]); - $query->chunkById(self::CHUNK_SIZE, fn ($servers) => $servers->each(fn (Server $server) => $this->processItem( - 'docker_cleanup', - ['server_id' => $server->id, 'server_name' => $server->name, 'team_id' => $server->team_id], - fn () => $this->runOccurrence( - schedule: $server->settings, - frequency: $server->settings->docker_cleanup_frequency ?? '0 0 * * *', - server: $server, - runLate: false, - skipReason: fn () => $this->serverSkipReason($server), - delivery: [ - 'schedule_key' => "docker-cleanup:{$server->id}", - 'job_type' => 'docker-cleanup', - 'resource_id' => $server->id, - 'payload' => [ - 'delete_unused_volumes' => $server->settings->delete_unused_volumes, - 'delete_unused_networks' => $server->settings->delete_unused_networks, - ], - ], - column: 'docker_cleanup_next_run_at', - ), - ))); - } - - /** - * Enabled schedules that are due, or that have no next run yet. - */ - private function due(Builder $query): Builder - { - return $query - ->where('enabled', true) - ->where(fn (Builder $query) => $query->whereNull('next_run_at')->orWhere('next_run_at', '<=', $this->now)); - } - - /** - * Claim the due occurrence of a schedule and dispatch it, skip it, or log it as missed. - * - * @param array{schedule_key: string, job_type: string, resource_id: int, payload?: array} $delivery - * @return array The outcome, for the log. - */ - private function runOccurrence( - Model $schedule, - string $frequency, - Server $server, - bool $runLate, - callable $skipReason, - array $delivery, - string $column = 'next_run_at', - ): array { - $timezone = data_get($server, 'settings.server_timezone'); - $stored = $schedule->getRawOriginal($column); - - 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) { - return $this->invalidFrequency($schedule, $column, null, $frequency, $delivery['schedule_key']); - } - if (! $this->advance($schedule, $column, null, $first)) { - return []; - } - if ($first->gt($this->now)) { - return []; - } - $stored = $first; - } - - $dueAt = CarbonImmutable::instance($schedule->getAttribute($column))->utc(); - $next = next_cron_run_at($frequency, $timezone, $this->now); - if ($next === null) { - 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)) { - $next = next_cron_run_at($frequency, $timezone, $next) ?? $next; - } - - $minutesLate = (int) $dueAt->diffInMinutes($this->now); - if (! $runLate && $minutesLate > self::LATE_RUN_WINDOW_MINUTES) { - $outcome = ['missed' => $dueAt->toIso8601String(), 'minutes_late' => $minutesLate]; - } elseif (($reason = $skipReason()) !== null) { - $outcome = ['skipped' => $reason]; - } else { - $outcome = ['dispatched' => $dueAt->toIso8601String()]; - } - - // Claim and record the occurrence together, so a crash cannot claim it without a delivery. - $occurrence = DB::transaction(function () use ($schedule, $column, $stored, $next, $outcome, $delivery, $dueAt): ScheduledJobDelivery|bool { - if (! $this->advance($schedule, $column, $stored, $next)) { - return false; + return; } - if (! isset($outcome['dispatched'])) { - return true; + if (! $backup->backupable) { + DeleteScheduledVolumeBackup::run($backup, $server); + $this->skippedCount++; + $this->logSkip('volume_backup', 'resource_deleted', [ + 'backup_id' => $backup->id, + 'team_id' => $backup->team_id, + ]); + + return; } - return $this->deliveries->create($delivery['schedule_key'], $dueAt, $delivery['job_type'], $delivery['resource_id'], $delivery['payload'] ?? []) ?? true; - }); + if (! $server) { + $this->skippedCount++; + $this->logSkip('volume_backup', 'server_missing', [ + 'backup_id' => $backup->id, + 'team_id' => $backup->team_id, + ]); - if ($occurrence === false) { - return []; + return; + } + + if (! $this->isDueCandidateBeforeExpensiveChecks($backup->frequency, $server)) { + return; + } + + if (! $server->isFunctional()) { + 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, + ...$this->serverStateContext($server), + ]); + } + + return; + } + + if (isCloud() && $backup->team_id !== 0 && ! data_get($backup, 'team.subscription.stripe_invoice_paid', false)) { + 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, + ...$this->serverStateContext($server), + ]); + } + + return; + } + + 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, + 'backupable_type' => $backup->backupable_type, + 'backupable_id' => $backup->backupable_id, + 'team_id' => $backup->team_id, + 'server_id' => $server->id, + ]); + } + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Error processing volume backup', [ + 'backup_id' => $backup->id, + 'error' => $e->getMessage(), + ]); } - - if ($occurrence instanceof ScheduledJobDelivery) { - $this->deliveries->publish($occurrence); - } - - 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 + private function getBackupSkipReason(ScheduledDatabaseBackup $backup, ?Server $server): ?string { - if ($stored !== null) { - $this->advance($schedule, $column, $stored, null); + if (blank(data_get($backup, 'database'))) { + $backup->delete(); + + return 'database_deleted'; } - $firstReportToday = rescue( - fn () => Cache::add("scheduled-job-manager:invalid-frequency:{$scheduleKey}:".md5($frequency), true, now()->addDay()), - true, - report: false, - ); + if (blank($server)) { + $backup->delete(); - 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 - { - $updated = $schedule->newQuery() - ->whereKey($schedule->getKey()) - ->when($from === null, fn (Builder $query) => $query->whereNull($column), fn (Builder $query) => $query->where($column, $from)) - ->toBase() - ->update([$column => $to]); - - if ($updated === 1) { - $schedule->setAttribute($column, $to)->syncOriginalAttribute($column); + return 'server_deleted'; } - return $updated === 1; - } - - private function sameLocalTime(CarbonImmutable $first, CarbonImmutable $second, ?string $timezone): bool - { - $timezone = filled($timezone) && validate_timezone($timezone) ? $timezone : config('app.timezone'); - - return $first->setTimezone($timezone)->format('Y-m-d H:i') === $second->setTimezone($timezone)->format('Y-m-d H:i'); - } - - private function serverSkipReason(Server $server): ?string - { - if (! $server->isFunctional()) { + if ($server->isFunctional() === false) { return 'server_not_functional'; } - if (isCloud() && $server->team_id !== 0 && data_get($server->team?->subscription, 'stripe_invoice_paid', false) === false) { + if (isCloud() && data_get($server->team->subscription, 'stripe_invoice_paid', false) === false && $server->team->id !== 0) { return 'subscription_unpaid'; } return null; } - private function taskResourceSkipReason(ScheduledTask $task): ?string + private function getTaskCriticalSkipReason(ScheduledTask $task, ?Server $server): ?string { - if ($task->application && ! str($task->application->status)->contains('running')) { + if (blank($server)) { + $task->delete(); + + return 'server_deleted'; + } + + if ($server->isFunctional() === false) { + return 'server_not_functional'; + } + + if (isCloud() && data_get($server->team->subscription, 'stripe_invoice_paid', false) === false && $server->team->id !== 0) { + return 'subscription_unpaid'; + } + + if (! $task->service && ! $task->application) { + $task->delete(); + + return 'resource_deleted'; + } + + return null; + } + + private function getTaskRuntimeSkipReason(ScheduledTask $task): ?string + { + if ($task->application && str($task->application->status)->contains('running') === false) { return 'application_not_running'; } - if ($task->service && ! str($task->service->status)->contains('running')) { + if ($task->service && str($task->service->status)->contains('running') === false) { return 'service_not_running'; } return null; } - /** - * Process one schedule and log its outcome. An error in one schedule never stops the others. - * - * @param array $context - */ - private function processItem(string $type, array $context, callable $process): void + private function processDockerCleanups(): void + { + $this->getServersForCleanupQuery() + ->chunkById(self::CHUNK_SIZE, function ($servers): void { + foreach ($servers as $server) { + $this->processDockerCleanup($server); + } + }); + } + + private function processDockerCleanup(Server $server): void { try { - $outcome = $process(); - } catch (\Throwable $e) { - Log::channel('scheduled-errors')->error("Error processing {$type}", [...$context, 'error' => $e->getMessage()]); + $frequency = data_get($server->settings, 'docker_cleanup_frequency', '0 * * * *'); + if (! $this->isDueCandidateBeforeExpensiveChecks($frequency, $server)) { + return; + } - return; + $skipReason = $this->getDockerCleanupSkipReason($server); + if ($skipReason !== null) { + 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; + } + + 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, + 'server_name' => $server->name, + 'team_id' => $server->team_id, + ]); + } + } catch (\Throwable $e) { + Log::channel('scheduled-errors')->error('Error processing docker cleanup', [ + 'server_id' => $server->id, + 'server_name' => $server->name, + 'error' => $e->getMessage(), + ]); + } + } + + private function getServersForCleanupQuery(): Builder + { + $query = Server::with('settings') + ->whereNotNull('ip') + ->where('ip', '!=', '') + ->whereNotIn('ip', Server::PLACEHOLDER_IPS); + + if (isCloud()) { + $query + ->with('team.subscription') + ->where(function (Builder $query): void { + $query + ->where('team_id', 0) + ->orWhereRelation('team.subscription', 'stripe_invoice_paid', true); + }); } - $label = ucfirst(str_replace('_', ' ', $type)); - $context = [...$context, ...$outcome, 'execution_time' => $this->now->toIso8601String()]; + return $query; + } - match (true) { - isset($outcome['dispatched']) => $this->log('dispatched', 'info', "{$label} dispatched", $context), - isset($outcome['skipped']) => $this->log('skipped', 'info', "{$label} skipped", ['skip_reason' => $outcome['skipped'], ...$context]), - isset($outcome['deleted']) => $this->log('skipped', 'info', "{$label} skipped", ['skip_reason' => $outcome['deleted'], ...$context]), - isset($outcome['waiting']) => $this->log('skipped', 'info', "{$label} skipped", ['skip_reason' => $outcome['waiting'], ...$context]), - isset($outcome['missed']) => $this->log('missed', 'warning', "{$label} missed", $context), - isset($outcome['invalid_frequency']) => Log::channel('scheduled-errors')->warning("{$label} has an invalid frequency", $context), - default => null, - }; + private function getDockerCleanupSkipReason(Server $server): ?string + { + if (! $server->isFunctional()) { + return 'server_not_functional'; + } + + // In cloud, check subscription status (except team 0) + if (isCloud() && $server->team_id !== 0) { + if (data_get($server->team->subscription, 'stripe_invoice_paid', false) === false) { + return 'subscription_unpaid'; + } + } + + return null; + } + + private function logSkip(string $type, string $reason, array $context = []): void + { + Log::channel('scheduled')->info(ucfirst(str_replace('_', ' ', $type)).' skipped', array_merge([ + 'type' => $type, + 'skip_reason' => $reason, + 'execution_time' => $this->executionTime?->toIso8601String(), + ], $context)); + } + + 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), + $jobType, + $resourceId, + $payload, + $this->executionTime, + ); + } + + 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)); + $previousDue = Carbon::instance($cron->getPreviousRunDate($executionTime, allowCurrentDate: true)); + + return $previousDue->gte( + $executionTime->copy()->subMinutes(ScheduledJobDeliveryService::CATCH_UP_WINDOW_MINUTES) + ); + } + + private function normalizeFrequency(string $frequency): string + { + return VALID_CRON_STRINGS[$frequency] ?? $frequency; + } + + private function serverTimezone(Server $server): string + { + $timezone = data_get($server->settings, 'server_timezone', config('app.timezone')); + + return validate_timezone($timezone) ? $timezone : config('app.timezone'); + } + + private function logBackupSkip(ScheduledDatabaseBackup $backup, string $reason, ?Server $server): void + { + $this->logSkip('backup', $reason, [ + 'backup_id' => $backup->id, + 'database_id' => $backup->database_id, + 'database_type' => $backup->database_type, + 'team_id' => $backup->team_id ?? null, + ...$this->serverStateContext($server), + ]); + } + + private function logTaskSkip(ScheduledTask $task, string $reason, ?Server $server): void + { + $this->logSkip('task', $reason, [ + 'task_id' => $task->id, + 'task_name' => $task->name, + 'team_id' => $server?->team_id, + ...$this->serverStateContext($server), + ]); } /** - * @param 'dispatched'|'skipped'|'missed' $counter - * @param array $context + * Compare the server state that decided the skip with the current database state. + * A difference shows that the worker used stale server data. + * + * @return array */ - private function log(string $counter, string $level, string $message, array $context): void + private function serverStateContext(?Server $server): array { - $this->counts[$counter]++; - Log::channel('scheduled')->log($level, $message, $context); + if ($server === null) { + return []; + } + + $settings = ServerSetting::query() + ->where('server_id', $server->id) + ->first(['is_reachable', 'is_usable', 'force_disabled']); + + return [ + 'server_id' => $server->id, + 'server_ip_is_placeholder' => $server->hasPlaceholderIp(), + 'used_is_reachable' => data_get($server->settings, 'is_reachable'), + 'used_is_usable' => data_get($server->settings, 'is_usable'), + 'used_force_disabled' => data_get($server->settings, 'force_disabled'), + 'used_invoice_paid' => data_get($server->team?->subscription, 'stripe_invoice_paid'), + 'db_is_reachable' => $settings?->is_reachable, + 'db_is_usable' => $settings?->is_usable, + 'db_force_disabled' => $settings?->force_disabled, + 'db_invoice_paid' => Subscription::query()->where('team_id', $server->team_id)->value('stripe_invoice_paid'), + ]; } } diff --git a/app/Livewire/Project/Shared/Destination.php b/app/Livewire/Project/Shared/Destination.php index 24c3b51090..1e5a6fb2d4 100644 --- a/app/Livewire/Project/Shared/Destination.php +++ b/app/Livewire/Project/Shared/Destination.php @@ -8,7 +8,6 @@ use App\Events\ApplicationStatusChanged; use App\Models\Application; use App\Models\Server; use App\Models\StandaloneDocker; -use App\Services\ScheduleNextRunRecalculator; use App\Traits\ListensToTeamChannel; use Illuminate\Foundation\Auth\Access\AuthorizesRequests; use Illuminate\Support\Collection; @@ -139,7 +138,6 @@ class Destination extends Component ->wherePivot('server_id', $server->id) ->detach($network->id); $this->resource->additional_networks()->attach($mainDestination->id, ['server_id' => $mainDestination->server->id]); - app(ScheduleNextRunRecalculator::class)->forResource($this->resource); }); $this->resource->refresh(); $this->refreshServers(); diff --git a/app/Livewire/Project/Shared/ScheduledTask/Show.php b/app/Livewire/Project/Shared/ScheduledTask/Show.php index 2428ff4e69..3bfe0700d9 100644 --- a/app/Livewire/Project/Shared/ScheduledTask/Show.php +++ b/app/Livewire/Project/Shared/ScheduledTask/Show.php @@ -7,7 +7,6 @@ use App\Models\Application; use App\Models\ScheduledTask; use App\Models\Service; use Illuminate\Foundation\Auth\Access\AuthorizesRequests; -use Livewire\Attributes\Computed; use Livewire\Attributes\Locked; use Livewire\Attributes\Validate; use Livewire\Component; @@ -88,15 +87,6 @@ class Show extends Component } } - /** - * Timezone of the server the task runs on, used to display its next run. - */ - #[Computed] - public function serverTimezone(): ?string - { - return $this->task->scheduleTimezone(); - } - private function syncData(bool $toModel = false): void { if ($toModel) { diff --git a/app/Livewire/Server/DockerCleanup.php b/app/Livewire/Server/DockerCleanup.php index 6e6d1480b2..d0a8d8ca9d 100644 --- a/app/Livewire/Server/DockerCleanup.php +++ b/app/Livewire/Server/DockerCleanup.php @@ -5,7 +5,9 @@ namespace App\Livewire\Server; use App\Jobs\DockerCleanupJob; use App\Models\DockerCleanupExecution; use App\Models\Server; +use Cron\CronExpression; use Illuminate\Foundation\Auth\Access\AuthorizesRequests; +use Illuminate\Support\Carbon; use Illuminate\Support\Facades\Cache; use Livewire\Attributes\Computed; use Livewire\Attributes\Validate; @@ -37,21 +39,35 @@ class DockerCleanup extends Component #[Validate('boolean')] public bool $disableApplicationImageRetention = false; - /** - * The scheduled cleanup is stale when its next run is more than 10 minutes overdue, - * meaning the scheduled job dispatcher did not run it. - */ #[Computed] public function isCleanupStale(): bool { - // The dispatcher does not run cleanups on servers without a real IP address. - if (blank($this->server->ip) || $this->server->hasPlaceholderIp()) { + try { + $lastExecution = DockerCleanupExecution::where('server_id', $this->server->id) + ->orderBy('created_at', 'desc') + ->first(); + + if (! $lastExecution) { + return false; + } + + $frequency = $this->server->settings->docker_cleanup_frequency ?? '0 0 * * *'; + if (isset(VALID_CRON_STRINGS[$frequency])) { + $frequency = VALID_CRON_STRINGS[$frequency]; + } + + $cron = new CronExpression($frequency); + $now = Carbon::now(); + $nextRun = Carbon::parse($cron->getNextRunDate($now)); + $afterThat = Carbon::parse($cron->getNextRunDate($nextRun)); + $intervalMinutes = $nextRun->diffInMinutes($afterThat); + + $threshold = max($intervalMinutes * 2, 10); + + return Carbon::parse($lastExecution->created_at)->diffInMinutes($now) > $threshold; + } catch (\Throwable) { return false; } - - $nextRunAt = $this->server->settings->docker_cleanup_next_run_at; - - return $nextRunAt !== null && $nextRunAt->lt(now()->subMinutes(10)); } #[Computed] diff --git a/app/Models/ScheduledDatabaseBackup.php b/app/Models/ScheduledDatabaseBackup.php index b44d1b16f4..53ae084349 100644 --- a/app/Models/ScheduledDatabaseBackup.php +++ b/app/Models/ScheduledDatabaseBackup.php @@ -2,15 +2,12 @@ namespace App\Models; -use App\Traits\HasNextRunAt; use Illuminate\Database\Eloquent\Relations\HasMany; use Illuminate\Database\Eloquent\Relations\HasOne; use Illuminate\Database\Eloquent\Relations\MorphTo; class ScheduledDatabaseBackup extends BaseModel { - use HasNextRunAt; - protected function casts(): array { return [ @@ -19,7 +16,6 @@ class ScheduledDatabaseBackup extends BaseModel 'database_backup_retention_max_storage_s3' => 'float', 'missing_backup_notification_days' => 'integer', 'last_execution_at' => 'datetime', - 'next_run_at' => '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/Models/ScheduledTask.php b/app/Models/ScheduledTask.php index f69ae77b82..0a53395d31 100644 --- a/app/Models/ScheduledTask.php +++ b/app/Models/ScheduledTask.php @@ -2,7 +2,6 @@ namespace App\Models; -use App\Traits\HasNextRunAt; use App\Traits\HasSafeStringAttribute; use Illuminate\Database\Eloquent\Factories\HasFactory; use Illuminate\Database\Eloquent\Relations\HasMany; @@ -21,7 +20,6 @@ use OpenApi\Attributes as OA; 'frequency' => ['type' => 'string', 'description' => 'The frequency of the scheduled task.'], 'container' => ['type' => 'string', 'nullable' => true, 'description' => 'The container where the command should be executed.'], 'timeout' => ['type' => 'integer', 'description' => 'The timeout of the scheduled task in seconds.'], - 'next_run_at' => ['type' => 'string', 'format' => 'date-time', 'nullable' => true, 'description' => 'The next time the scheduled task runs, in UTC. Null when it is disabled or not calculated yet.'], 'created_at' => ['type' => 'string', 'format' => 'date-time', 'description' => 'The date and time when the scheduled task was created.'], 'updated_at' => ['type' => 'string', 'format' => 'date-time', 'description' => 'The date and time when the scheduled task was last updated.'], ], @@ -29,7 +27,6 @@ use OpenApi\Attributes as OA; class ScheduledTask extends BaseModel { use HasFactory; - use HasNextRunAt; use HasSafeStringAttribute; protected $fillable = [ @@ -55,7 +52,6 @@ class ScheduledTask extends BaseModel return [ 'enabled' => 'boolean', 'timeout' => 'integer', - 'next_run_at' => 'datetime', ]; } diff --git a/app/Models/ScheduledVolumeBackup.php b/app/Models/ScheduledVolumeBackup.php index 4732d60adc..7f33fd92b9 100644 --- a/app/Models/ScheduledVolumeBackup.php +++ b/app/Models/ScheduledVolumeBackup.php @@ -2,7 +2,6 @@ namespace App\Models; -use App\Traits\HasNextRunAt; use Illuminate\Database\Eloquent\Builder; use Illuminate\Database\Eloquent\Model; use Illuminate\Database\Eloquent\Relations\BelongsTo; @@ -12,8 +11,6 @@ use Illuminate\Database\Eloquent\Relations\MorphTo; class ScheduledVolumeBackup extends BaseModel { - use HasNextRunAt; - public const int DEFAULT_TIMEOUT = 36000; protected $fillable = [ @@ -50,7 +47,6 @@ class ScheduledVolumeBackup extends BaseModel 'retention_days_s3' => 'integer', 'retention_max_storage_s3' => 'float', 'timeout' => 'integer', - 'next_run_at' => 'datetime', ]; } diff --git a/app/Models/ServerSetting.php b/app/Models/ServerSetting.php index 4f7e76bbd6..1fb3af6d6c 100644 --- a/app/Models/ServerSetting.php +++ b/app/Models/ServerSetting.php @@ -3,7 +3,6 @@ namespace App\Models; use App\Enums\ServerRole; -use App\Services\ScheduleNextRunRecalculator; use Illuminate\Contracts\Encryption\DecryptException; use Illuminate\Database\Eloquent\Casts\Attribute; use Illuminate\Database\Eloquent\Model; @@ -171,7 +170,6 @@ class ServerSetting extends Model 'docker_version_checked_at' => 'datetime', 'compose_version_checked_at' => 'datetime', 'backup_compression_cpu_percentage' => 'integer', - 'docker_cleanup_next_run_at' => 'datetime', 'server_disk_usage_notification_interval_hours' => 'integer', ]; @@ -193,15 +191,6 @@ class ServerSetting extends Model protected static function booted() { - static::saving(function ($setting) { - if (! $setting->exists || $setting->isDirty(['docker_cleanup_frequency', 'server_timezone'])) { - // Without a frequency (a new row that uses the column default), the dispatcher calculates it. - $frequency = $setting->getAttributes()['docker_cleanup_frequency'] ?? null; - $setting->docker_cleanup_next_run_at = $frequency === null - ? null - : next_cron_run_at($frequency, $setting->server_timezone, now()); - } - }); static::creating(function ($setting) { try { if (str($setting->sentinel_token)->isEmpty()) { @@ -215,11 +204,6 @@ class ServerSetting extends Model } }); static::updated(function ($settings) { - if ($settings->wasChanged('server_timezone') && $settings->server) { - // A failure must not fail the settings save: the schedules correct themselves after their next run. - rescue(fn () => app(ScheduleNextRunRecalculator::class)->forServer($settings->server)); - } - if ( $settings->wasChanged('sentinel_token') || $settings->wasChanged('sentinel_custom_url') || diff --git a/app/Services/ScheduleNextRunRecalculator.php b/app/Services/ScheduleNextRunRecalculator.php deleted file mode 100644 index 87127c27c2..0000000000 --- a/app/Services/ScheduleNextRunRecalculator.php +++ /dev/null @@ -1,85 +0,0 @@ -serverTimezone($resource); - - foreach (['scheduled_tasks', 'scheduledBackups'] as $relation) { - if (method_exists($resource, $relation)) { - $this->recalculate($resource->{$relation}(), $timezone); - } - } - - foreach (['persistentStorages', 'fileStorages'] as $relation) { - if (method_exists($resource, $relation)) { - $storages = $resource->{$relation}(); - $this->recalculate(ScheduledVolumeBackup::query() - ->where('backupable_type', $storages->getRelated()->getMorphClass()) - ->whereIn('backupable_id', $storages->select('id')), $timezone); - } - } - - if ($resource instanceof Service) { - $resource->applications()->get()->each(fn (Model $application) => $this->forResource($application, $timezone)); - $resource->databases()->get()->each(fn (Model $database) => $this->forResource($database, $timezone)); - } - } - - /** - * Each resource uses the timezone of its own main server: Server::applications() also returns - * applications for which this server is only an additional server. - */ - public function forServer(Server $server): void - { - $server->applications()->each(fn (Model $application) => $this->forResource($application)); - $server->databases()->each(fn (Model $database) => $this->forResource($database)); - $server->services()->get()->each(fn (Service $service) => $this->forResource($service)); - - // Server::databases() leaves out the instance database, which has the instance backup. - if ($server->id === 0 && ($instanceDatabase = StandalonePostgresql::find(0))) { - $this->forResource($instanceDatabase); - } - } - - /** - * Move enabled schedules to their next due time after now. The current minute is excluded, - * so an occurrence that already ran in this minute does not run again. - */ - private function recalculate(Relation|Builder $schedules, ?string $timezone): void - { - $schedules->where('enabled', true)->get()->each(fn (Model $schedule) => $schedule->forceFill([ - 'next_run_at' => next_cron_run_at((string) $schedule->frequency, $timezone, now()), - ])->saveQuietly()); - } - - private function serverTimezone(Model $resource): ?string - { - $owner = $resource instanceof ServiceApplication || $resource instanceof ServiceDatabase - ? $resource->service()->first() - : $resource; - $server = $owner?->destination()->first()?->server()->first(); - - return $server?->settings()->value('server_timezone'); - } -} diff --git a/app/Services/ScheduledJobDeliveryService.php b/app/Services/ScheduledJobDeliveryService.php index 9067391e6d..5dce4691dc 100644 --- a/app/Services/ScheduledJobDeliveryService.php +++ b/app/Services/ScheduledJobDeliveryService.php @@ -8,41 +8,119 @@ use App\Jobs\ScheduledTaskJob; use App\Jobs\VolumeBackupJob; use App\Models\ScheduledDatabaseBackup; use App\Models\ScheduledJobDelivery; +use App\Models\ScheduledJobState; use App\Models\ScheduledTask; use App\Models\ScheduledVolumeBackup; use App\Models\Server; -use Carbon\CarbonInterface; +use Cron\CronExpression; +use Illuminate\Support\Carbon; +use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Log; class ScheduledJobDeliveryService { + public const CATCH_UP_WINDOW_MINUTES = 10; + /** * 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. + * example when Redis evicted or lost the queue. 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. + * Job types that run once when they are late. A lost task or Docker cleanup is logged as missed. */ 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. - * - * @param array $payload - */ - public function create(string $scheduleKey, CarbonInterface $scheduledFor, string $jobType, int $resourceId, array $payload = []): ?ScheduledJobDelivery - { - $delivery = ScheduledJobDelivery::createOrFirst( - ['schedule_key' => $scheduleKey, 'scheduled_for' => $scheduledFor->utc()], - ['job_type' => $jobType, 'resource_id' => $resourceId, 'payload' => $payload, 'status' => 'pending'], - ); + public function recordAndPublish( + string $scheduleKey, + string $frequency, + string $timezone, + string $jobType, + int $resourceId, + array $payload = [], + ?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)); - return $delivery->wasRecentlyCreated ? $delivery : null; + 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 @@ -66,9 +144,28 @@ class ScheduledJobDeliveryService }); } + /** + * Count occurrences that are still open 15 minutes after they were created. Pending or + * enqueued rows here never started; claimed rows are still running or their worker died. + * + * @return array + */ + public function staleOpenOccurrenceCounts(): array + { + return ScheduledJobDelivery::query() + ->whereIn('status', ['pending', 'enqueued', 'claimed']) + ->where('created_at', '<', now()->subMinutes(15)) + ->selectRaw('status, count(*) as aggregate') + ->groupBy('status') + ->pluck('aggregate', 'status') + ->map(fn ($count) => (int) $count) + ->all(); + } + /** * 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. + * 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. */ public function recoverStaleEnqueued(): void { @@ -78,26 +175,7 @@ class ScheduledJobDeliveryService ->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); + $this->recoverStaleEnqueuedOccurrence($occurrence); } catch (\Throwable $e) { Log::channel('scheduled-errors')->error('Failed to recover enqueued scheduled occurrence', [ 'occurrence_uuid' => $occurrence->uuid, @@ -109,6 +187,38 @@ class ScheduledJobDeliveryService }); } + private function recoverStaleEnqueuedOccurrence(ScheduledJobDelivery $occurrence): void + { + $hasNewerOccurrence = ScheduledJobDelivery::query() + ->where('schedule_key', $occurrence->schedule_key) + ->where('scheduled_for', '>', $occurrence->scheduled_for) + ->exists(); + $runLate = in_array($occurrence->job_type, self::LATE_RUN_JOB_TYPES, true) && ! $hasNewerOccurrence; + + // Only one scheduler node can move each occurrence. + $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) { + return; + } + + if (! $runLate) { + $this->logOccurrence($occurrence->uuid, $hasNewerOccurrence + ? 'Scheduled occurrence skipped: its queued job was not started and a newer occurrence exists' + : 'Scheduled occurrence missed: its queued job was not started', 'warning'); + + return; + } + + $occurrence->status = 'pending'; + $this->logOccurrence($occurrence->uuid, 'Scheduled occurrence publishing again: its queued job was not started', 'warning'); + $this->publish($occurrence); + } + public function deleteOldOccurrences(): void { ScheduledJobDelivery::query() @@ -203,7 +313,7 @@ class ScheduledJobDeliveryService } } - public function publish(ScheduledJobDelivery $occurrence): bool + private function publish(ScheduledJobDelivery $occurrence): bool { if ($occurrence->status !== 'pending') { return false; @@ -242,6 +352,15 @@ class ScheduledJobDeliveryService return false; } + // An open previous occurrence usually means the previous run is still running. Jobs with + // WithoutOverlapping(...)->dontRelease() then drop this one without running it. + $previousOpen = ScheduledJobDelivery::query() + ->where('schedule_key', $occurrence->schedule_key) + ->where('scheduled_for', '<', $occurrence->scheduled_for) + ->whereIn('status', ['pending', 'enqueued', 'claimed']) + ->orderByDesc('scheduled_for') + ->first(['uuid', 'status', 'scheduled_for']); + dispatch($job); ScheduledJobDelivery::query() @@ -254,7 +373,17 @@ class ScheduledJobDeliveryService ]); // Log from the loaded row: a fast worker can complete and delete the occurrence already. - $this->logOccurrence($occurrence, 'Scheduled occurrence enqueued', context: ['status' => 'enqueued', 'enqueued_at' => now()->toIso8601String()]); + $enqueued = ['status' => 'enqueued', 'enqueued_at' => now()->toIso8601String()]; + if ($previousOpen) { + $this->logOccurrence($occurrence, 'Scheduled occurrence enqueued while the previous occurrence is still open', 'warning', [ + ...$enqueued, + 'previous_occurrence_uuid' => $previousOpen->uuid, + 'previous_status' => $previousOpen->status, + 'previous_scheduled_for' => $previousOpen->scheduled_for->toIso8601String(), + ]); + } else { + $this->logOccurrence($occurrence, 'Scheduled occurrence enqueued', context: $enqueued); + } return true; } diff --git a/app/Traits/HasNextRunAt.php b/app/Traits/HasNextRunAt.php deleted file mode 100644 index 856ebe3391..0000000000 --- a/app/Traits/HasNextRunAt.php +++ /dev/null @@ -1,38 +0,0 @@ -exists && ! $schedule->isDirty(['frequency', 'enabled'])) { - return; - } - - $schedule->next_run_at = $schedule->calculateNextRunAt(now()); - }); - } - - public function calculateNextRunAt(DateTimeInterface $after, bool $includeCurrentMinute = false): ?CarbonImmutable - { - if ($this->getAttribute('enabled') !== null && ! $this->enabled) { - return null; - } - - return next_cron_run_at((string) $this->frequency, $this->scheduleTimezone(), $after, $includeCurrentMinute); - } - - public function scheduleTimezone(): ?string - { - return data_get($this->server(), 'settings.server_timezone'); - } -} diff --git a/bootstrap/helpers/shared.php b/bootstrap/helpers/shared.php index 94a35140c0..9b97c551c5 100644 --- a/bootstrap/helpers/shared.php +++ b/bootstrap/helpers/shared.php @@ -1112,28 +1112,6 @@ function validate_timezone(string $timezone): bool return in_array($timezone, timezone_identifiers_list()); } -/** - * The next due time of a cron frequency after $after, in UTC. An invalid frequency returns null. - * With $includeCurrentMinute, a frequency that is due in the minute of $after returns that minute. - */ -function next_cron_run_at(string $frequency, ?string $timezone, DateTimeInterface $after, bool $includeCurrentMinute = false): ?CarbonImmutable -{ - $frequency = trim($frequency); - $timezone = filled($timezone) && validate_timezone($timezone) ? $timezone : config('app.timezone'); - - try { - $cron = new Cron\CronExpression(VALID_CRON_STRINGS[$frequency] ?? $frequency); - $time = CarbonImmutable::instance($after)->setTimezone($timezone); - if ($includeCurrentMinute) { - $time = $time->startOfMinute(); - } - - return CarbonImmutable::instance($cron->getNextRunDate($time, 0, $includeCurrentMinute))->utc(); - } catch (Throwable) { - return null; - } -} - function parseEnvFormatToArray($env_file_contents) { $env_array = []; diff --git a/database/migrations/2026_09_28_194652_add_next_run_at_to_scheduled_jobs.php b/database/migrations/2026_09_28_194652_add_next_run_at_to_scheduled_jobs.php deleted file mode 100644 index dd413fb653..0000000000 --- a/database/migrations/2026_09_28_194652_add_next_run_at_to_scheduled_jobs.php +++ /dev/null @@ -1,94 +0,0 @@ -timestampTz('next_run_at')->nullable(); - $table->index(['enabled', 'next_run_at']); - }); - } - - Schema::table('server_settings', function (Blueprint $table) { - $table->timestampTz('docker_cleanup_next_run_at')->nullable()->index(); - }); - - $this->continueAfterLastScheduledOccurrences(); - - Schema::dropIfExists('scheduled_job_states'); - } - - public function down(): void - { - Schema::create('scheduled_job_states', function (Blueprint $table) { - $table->id(); - $table->string('uuid')->unique(); - $table->string('schedule_key')->unique(); - $table->timestampTz('last_scheduled_for')->nullable(); - $table->timestamps(); - }); - - Schema::table('server_settings', function (Blueprint $table) { - $table->dropIndex(['docker_cleanup_next_run_at']); - $table->dropColumn('docker_cleanup_next_run_at'); - }); - - foreach (['scheduled_database_backups', 'scheduled_tasks', 'scheduled_volume_backups'] as $table) { - Schema::table($table, function (Blueprint $table) { - $table->dropIndex(['enabled', 'next_run_at']); - $table->dropColumn('next_run_at'); - }); - } - } - - /** - * Continue each schedule after the last occurrence that the previous dispatcher recorded, so - * an occurrence that was due during the upgrade still runs late instead of being skipped. - */ - public function continueAfterLastScheduledOccurrences(): void - { - if (! Schema::hasTable('scheduled_job_states')) { - return; - } - - DB::table('scheduled_job_states')->whereNotNull('last_scheduled_for')->orderBy('id')->chunkById(500, function ($states) { - foreach ($states as $state) { - try { - [$type, $id] = explode(':', $state->schedule_key, 2) + [null, null]; - $schedule = match ($type) { - 'scheduled-backup' => ScheduledDatabaseBackup::find($id), - 'scheduled-task' => ScheduledTask::find($id), - 'scheduled-volume-backup' => ScheduledVolumeBackup::find($id), - 'docker-cleanup' => ServerSetting::query()->where('server_id', $id)->first(), - default => null, - }; - - if ($schedule instanceof ServerSetting) { - $schedule->forceFill(['docker_cleanup_next_run_at' => next_cron_run_at( - (string) $schedule->docker_cleanup_frequency, $schedule->server_timezone, Carbon::parse($state->last_scheduled_for), - )])->saveQuietly(); - } elseif ($schedule?->enabled) { - $schedule->forceFill(['next_run_at' => $schedule->calculateNextRunAt(Carbon::parse($state->last_scheduled_for))])->saveQuietly(); - } - } catch (Throwable) { - // The dispatcher calculates the next run of this schedule on its first run. - } - } - }); - } -}; diff --git a/database/migrations/2026_10_02_055938_remove_next_run_at_from_scheduled_jobs.php b/database/migrations/2026_10_02_055938_remove_next_run_at_from_scheduled_jobs.php new file mode 100644 index 0000000000..e3d205a4ce --- /dev/null +++ b/database/migrations/2026_10_02_055938_remove_next_run_at_from_scheduled_jobs.php @@ -0,0 +1,46 @@ +dropIndex(['enabled', 'next_run_at']); + $table->dropColumn('next_run_at'); + }); + } + } + + if (Schema::hasColumn('server_settings', 'docker_cleanup_next_run_at')) { + Schema::table('server_settings', function (Blueprint $table) { + $table->dropIndex(['docker_cleanup_next_run_at']); + $table->dropColumn('docker_cleanup_next_run_at'); + }); + } + + if (! Schema::hasTable('scheduled_job_states')) { + Schema::create('scheduled_job_states', function (Blueprint $table) { + $table->id(); + $table->string('uuid')->unique(); + $table->string('schedule_key')->unique(); + $table->timestampTz('last_scheduled_for')->nullable(); + $table->timestamps(); + }); + } + } + + public function down(): void + { + // The removed columns are not added again. + } +}; diff --git a/openapi.json b/openapi.json index 8d2aefe1ad..9e6bf53c3e 100644 --- a/openapi.json +++ b/openapi.json @@ -23565,12 +23565,6 @@ "type": "integer", "description": "The timeout of the scheduled task in seconds." }, - "next_run_at": { - "type": "string", - "format": "date-time", - "nullable": true, - "description": "The next time the scheduled task runs, in UTC. Null when it is disabled or not calculated yet." - }, "created_at": { "type": "string", "format": "date-time", diff --git a/openapi.yaml b/openapi.yaml index fd86ff91b1..68950d480a 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -15066,11 +15066,6 @@ components: timeout: type: integer description: 'The timeout of the scheduled task in seconds.' - next_run_at: - type: string - format: date-time - nullable: true - description: 'The next time the scheduled task runs, in UTC. Null when it is disabled or not calculated yet.' created_at: type: string format: date-time diff --git a/resources/views/components/next-run.blade.php b/resources/views/components/next-run.blade.php deleted file mode 100644 index f0db83ffda..0000000000 --- a/resources/views/components/next-run.blade.php +++ /dev/null @@ -1,30 +0,0 @@ -@props([ - 'at' => null, - 'timezone' => null, - 'enabled' => true, - 'frequency' => null, -]) - -@php - // Same fallback as next_cron_run_at(): schedules without a valid server timezone run in the instance timezone. - $displayTimezone = filled($timezone) && validate_timezone((string) $timezone) ? $timezone : config('app.timezone'); - $nextRunAt = $at ? \Illuminate\Support\Carbon::parse($at) : null; - // Without a next run, the frequency is either not calculated yet or cannot be calculated at all. - $isFrequencyInvalid = ! $nextRunAt && filled($frequency) && next_cron_run_at((string) $frequency, $displayTimezone, now()) === null; -@endphp - -

merge(['class' => 'text-xs leading-5 text-neutral-500 dark:text-fg-dim']) }}> - Next run - @if (! $enabled) - Disabled - @elseif ($isFrequencyInvalid) - Invalid frequency. Change it to a valid cron expression, or this schedule never runs. - @elseif (! $nextRunAt) - Calculating… - @else - - · - {{ $nextRunAt->isFuture() ? 'in '.$nextRunAt->diffForHumans(syntax: \Carbon\CarbonInterface::DIFF_ABSOLUTE) : $nextRunAt->diffForHumans() }} - @endif -

diff --git a/resources/views/livewire/project/database/backup-edit/general.blade.php b/resources/views/livewire/project/database/backup-edit/general.blade.php index 1dd6f28eb8..49385b0258 100644 --- a/resources/views/livewire/project/database/backup-edit/general.blade.php +++ b/resources/views/livewire/project/database/backup-edit/general.blade.php @@ -96,7 +96,6 @@ min="0" max="365" suffix="days" canGate="manageBackups" :canResource="$backup->database" helper="Notify through backup failure channels after this many days without an execution. Use 0 to disable." required /> - diff --git a/resources/views/livewire/project/shared/scheduled-task/show.blade.php b/resources/views/livewire/project/shared/scheduled-task/show.blade.php index 7849f35dcf..c5be9974ba 100644 --- a/resources/views/livewire/project/shared/scheduled-task/show.blade.php +++ b/resources/views/livewire/project/shared/scheduled-task/show.blade.php @@ -44,7 +44,6 @@ - diff --git a/resources/views/livewire/project/shared/storages/volume-backups/general.blade.php b/resources/views/livewire/project/shared/storages/volume-backups/general.blade.php index 48764b9c36..5b4ece52f1 100644 --- a/resources/views/livewire/project/shared/storages/volume-backups/general.blade.php +++ b/resources/views/livewire/project/shared/storages/volume-backups/general.blade.php @@ -45,6 +45,5 @@ - diff --git a/resources/views/livewire/server/docker-cleanup.blade.php b/resources/views/livewire/server/docker-cleanup.blade.php index a530f7c746..e12f4cd3da 100644 --- a/resources/views/livewire/server/docker-cleanup.blade.php +++ b/resources/views/livewire/server/docker-cleanup.blade.php @@ -33,8 +33,6 @@ @if (!isCloud() && $this->isCleanupStale) - A scheduled cleanup was due - {{ $server->settings->docker_cleanup_next_run_at?->diffForHumans() }} but has not started. The last cleanup ran {{ $this->lastExecutionTime ?? 'at an unknown time' }}. @if (!$this->isSchedulerHealthy) The scheduled job manager also appears inactive. @@ -54,8 +52,6 @@

Cleanup runs automatically using the schedule and threshold configured below.

- @endif diff --git a/tests/Feature/Jobs/ScheduledJobManagerStaleLockTest.php b/tests/Feature/Jobs/ScheduledJobManagerStaleLockTest.php new file mode 100644 index 0000000000..dba5b6041a --- /dev/null +++ b/tests/Feature/Jobs/ScheduledJobManagerStaleLockTest.php @@ -0,0 +1,50 @@ +shouldReceive('ttl')->once()->with($lockKey)->andReturn(-1); + $redis->shouldReceive('del')->once()->with($lockKey)->andReturn(1); + + Redis::shouldReceive('connection')->with('default')->andReturn($redis); + + $job = new ScheduledJobManager; + $job->middleware(); +}); + +it('preserves valid lock with positive TTL', function () { + $lockKey = scheduledJobManagerLockKey(); + + $redis = Mockery::mock(); + $redis->shouldReceive('ttl')->once()->with($lockKey)->andReturn(60); + $redis->shouldNotReceive('del'); + + Redis::shouldReceive('connection')->with('default')->andReturn($redis); + + $job = new ScheduledJobManager; + $job->middleware(); +}); + +it('does not fail when no lock exists', function () { + $lockKey = scheduledJobManagerLockKey(); + + $redis = Mockery::mock(); + $redis->shouldReceive('ttl')->once()->with($lockKey)->andReturn(-2); + $redis->shouldNotReceive('del'); + + Redis::shouldReceive('connection')->with('default')->andReturn($redis); + + $job = new ScheduledJobManager; + $middleware = $job->middleware(); + + expect($middleware)->toBeArray()->toHaveCount(1); +}); diff --git a/tests/Feature/QueueRoutingTest.php b/tests/Feature/QueueRoutingTest.php index a35d2d8555..1bb837be89 100644 --- a/tests/Feature/QueueRoutingTest.php +++ b/tests/Feature/QueueRoutingTest.php @@ -4,6 +4,7 @@ use App\Actions\Database\StartDatabase; use App\Actions\Database\StartDatabaseProxy; use App\Actions\Service\StartService; use App\Jobs\DatabaseBackupJob; +use App\Jobs\ScheduledJobManager; use App\Models\ScheduledDatabaseBackup; describe('deployment_queue helper', function () { @@ -60,12 +61,14 @@ describe('scheduled job routing', function () { test('scheduled jobs use the crons queue on cloud', function () { config(['constants.coolify.self_hosted' => false]); + expect((new ScheduledJobManager)->queue)->toBe('crons'); expect((new DatabaseBackupJob(new ScheduledDatabaseBackup))->queue)->toBe('crons'); }); test('scheduled jobs use the high queue on self-hosted', function () { config(['constants.coolify.self_hosted' => true]); + expect((new ScheduledJobManager)->queue)->toBe('high'); expect((new DatabaseBackupJob(new ScheduledDatabaseBackup))->queue)->toBe('high'); }); }); diff --git a/tests/Feature/ScheduleNextRunAtTest.php b/tests/Feature/ScheduleNextRunAtTest.php deleted file mode 100644 index e05acc2fb3..0000000000 --- a/tests/Feature/ScheduleNextRunAtTest.php +++ /dev/null @@ -1,179 +0,0 @@ -toDateTimeString())->toBe($expected); -})->with([ - 'alias' => ['daily', 'UTC', '2026-09-17 10:20:30', false, '2026-09-18 00:00:00'], - 'server timezone' => ['0 2 * * *', 'Europe/Budapest', '2026-09-17 10:20:30', false, '2026-09-18 00:00:00'], - 'strictly after' => ['20 10 * * *', 'UTC', '2026-09-17 10:20:00', false, '2026-09-18 10:20:00'], - 'current minute included' => ['20 10 * * *', 'UTC', '2026-09-17 10:20:30', true, '2026-09-17 10:20:00'], - 'a time that does not exist when daylight saving time starts' => ['30 2 * * *', 'Europe/Budapest', '2026-03-28 12:00:00', false, '2026-03-29 01:30:00'], - 'invalid timezone uses the app timezone' => ['0 2 * * *', 'Mars/Olympus', '2026-09-17 10:20:30', false, '2026-09-18 02:00:00'], - 'invalid frequency' => ['not a cron', 'UTC', '2026-09-17 10:20:30', false, null], -]); - -it('calculates the next run of a schedule when it is created or its frequency or enabled flag changes', function () { - $task = nextRunAtTestTask(nextRunAtTestApplication('Europe/Budapest'), '0 2 * * *'); - - expect($task->next_run_at->toDateTimeString())->toBe('2026-09-18 00:00:00'); - - $task->update(['frequency' => 'hourly']); - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-17 11:00:00'); - - $task->update(['enabled' => false]); - expect($task->fresh()->next_run_at)->toBeNull(); - - Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 30, 0, 'UTC')); - $task->update(['enabled' => true]); - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-17 13:00:00'); - - $task->update(['name' => 'renamed']); - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-17 13:00:00'); -}); - -it('calculates the next Docker cleanup when its frequency or the server timezone changes', function () { - $settings = nextRunAtTestApplication()->destination->server->settings; - - $settings->update(['docker_cleanup_frequency' => '0 2 * * *']); - expect($settings->fresh()->docker_cleanup_next_run_at->toDateTimeString())->toBe('2026-09-18 02:00:00'); - - $settings->update(['server_timezone' => 'Europe/Budapest']); - expect($settings->fresh()->docker_cleanup_next_run_at->toDateTimeString())->toBe('2026-09-18 00:00:00'); -}); - -it('recalculates the next run of the server schedules when the server timezone changes', function () { - $application = nextRunAtTestApplication(); - $task = nextRunAtTestTask($application, '0 2 * * *'); - $otherTask = nextRunAtTestTask(nextRunAtTestApplication(), '0 2 * * *'); - - $application->destination->server->settings->update(['server_timezone' => 'Europe/Budapest']); - - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-18 00:00:00') - ->and($otherTask->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-18 02:00:00'); -}); - -it('keeps the timezone of the main server for applications that also run on the changed server', function () { - $application = nextRunAtTestApplication(); - $task = nextRunAtTestTask($application, '0 2 * * *'); - $additionalServer = nextRunAtTestApplication()->destination->server; - $application->additional_networks()->attach( - StandaloneDocker::where('server_id', $additionalServer->id)->value('id'), - ['server_id' => $additionalServer->id], - ); - - $additionalServer->settings->update(['server_timezone' => 'Europe/Budapest']); - - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-18 02:00:00'); -}); - -it('does not run an occurrence again when the timezone changes in the minute it ran', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 17, 11, 59, 0, 'UTC')); - $application = nextRunAtTestApplication(); - $task = nextRunAtTestTask($application, '0 * * * *'); - - Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 30, 'UTC')); - $application->destination->server->settings->update(['server_timezone' => 'Europe/Budapest']); - - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-17 13:00:00'); -}); - -it('recalculates the next run of the resource schedules with the timezone of its current server', function () { - $application = nextRunAtTestApplication(); - $task = nextRunAtTestTask($application, '0 2 * * *'); - $application->destination->server->settings()->update(['server_timezone' => 'Europe/Budapest']); - - app(ScheduleNextRunRecalculator::class)->forResource($application); - - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-18 00:00:00'); -}); - -it('lists overdue schedules and schedules without a next run in the diagnostics', function () { - $overdue = nextRunAtTestTask(nextRunAtTestApplication(), '0 2 * * *'); - ScheduledTask::query()->whereKey($overdue->id)->update(['next_run_at' => now()->subHour()]); - $invalid = nextRunAtTestTask(nextRunAtTestApplication(), '0 2 * * *'); - ScheduledTask::query()->whereKey($invalid->id)->update(['frequency' => '60 * * * *', 'next_run_at' => null]); - nextRunAtTestTask(nextRunAtTestApplication(), '0 3 * * *'); - - $this->artisan('scheduled:diagnostics') - ->expectsOutputToContain('Scheduled tasks: 3 enabled, 2 overdue or without a next run') - ->expectsOutputToContain('invalid frequency') - ->assertSuccessful(); -}); - -it('continues each schedule after the occurrence that the previous dispatcher recorded last', function () { - $task = nextRunAtTestTask(nextRunAtTestApplication(), '0 2 * * *'); - Schema::create('scheduled_job_states', function (Blueprint $table) { - $table->id(); - $table->string('uuid')->unique(); - $table->string('schedule_key')->unique(); - $table->timestampTz('last_scheduled_for')->nullable(); - $table->timestamps(); - }); - DB::table('scheduled_job_states')->insert([ - 'uuid' => 'state-1', - 'schedule_key' => "scheduled-task:{$task->id}", - 'last_scheduled_for' => '2026-09-16 02:00:00', - ]); - - $migration = require database_path('migrations/2026_09_28_194652_add_next_run_at_to_scheduled_jobs.php'); - $migration->continueAfterLastScheduledOccurrences(); - - // The occurrence of 2026-09-17 02:00 was not recorded, so the dispatcher runs it on its next run. - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-17 02:00:00'); -}); - -function nextRunAtTestApplication(string $timezone = 'UTC'): Application -{ - $team = Team::factory()->create(); - $server = Server::factory()->create([ - 'team_id' => $team->id, - 'private_key_id' => PrivateKey::factory()->create(['team_id' => $team->id])->id, - ]); - $server->settings()->update(['server_timezone' => $timezone]); - $destination = StandaloneDocker::where('server_id', $server->id)->first() - ?? StandaloneDocker::factory()->create(['server_id' => $server->id]); - $environment = Environment::factory()->create([ - 'project_id' => Project::factory()->create(['team_id' => $team->id])->id, - ]); - - return Application::factory()->create([ - 'environment_id' => $environment->id, - 'destination_id' => $destination->id, - 'destination_type' => StandaloneDocker::class, - ]); -} - -function nextRunAtTestTask(Application $application, string $frequency): ScheduledTask -{ - return ScheduledTask::factory()->create([ - 'team_id' => $application->environment->project->team_id, - 'application_id' => $application->id, - 'frequency' => $frequency, - 'enabled' => true, - ])->fresh(); -} diff --git a/tests/Feature/ScheduleNextRunDisplayTest.php b/tests/Feature/ScheduleNextRunDisplayTest.php deleted file mode 100644 index d8910b1d6c..0000000000 --- a/tests/Feature/ScheduleNextRunDisplayTest.php +++ /dev/null @@ -1,222 +0,0 @@ - InstanceSettings::firstOrCreate(['id' => 0])); - - $this->team = Team::factory()->create(); - $this->user = User::factory()->create(); - $this->user->teams()->attach($this->team, ['role' => 'owner']); - $this->actingAs($this->user); - session(['currentTeam' => $this->team]); - - $this->server = Server::factory()->create(['team_id' => $this->team->id]); - $this->server->settings->update(['server_timezone' => 'Europe/Budapest']); - $this->destination = StandaloneDocker::where('server_id', $this->server->id)->firstOrFail(); - $project = Project::factory()->create(['team_id' => $this->team->id]); - $this->environment = Environment::factory()->create(['project_id' => $project->id]); - $this->application = Application::factory()->create([ - 'environment_id' => $this->environment->id, - 'destination_id' => $this->destination->id, - 'destination_type' => $this->destination->getMorphClass(), - ]); -}); - -afterEach(function () { - Event::forget(RouteMatched::class); - Carbon::setTestNow(); -}); - -function nextRunDisplayBindTaskRoute(Application $application, ScheduledTask $task): void -{ - $application->loadMissing('environment.project'); - - Event::listen(RouteMatched::class, function (RouteMatched $event) use ($application, $task): void { - $event->route->setParameter('task_uuid', $task->uuid); - $event->route->setParameter('project_uuid', $application->environment->project->uuid); - $event->route->setParameter('environment_uuid', $application->environment->uuid); - $event->route->setParameter('application_uuid', $application->uuid); - }); -} - -function nextRunDisplaySetCleanupNextRun(Server $server, ?Carbon $nextRunAt): void -{ - ServerSetting::query()->where('server_id', $server->id)->update(['docker_cleanup_next_run_at' => $nextRunAt]); -} - -it('shows the next Docker cleanup in the server timezone and refreshes it after saving a new frequency', function () { - $this->server->settings->update(['docker_cleanup_frequency' => '0 2 * * *']); - - Livewire::test(DockerCleanup::class, ['server_uuid' => $this->server->uuid]) - ->assertSee('Next run') - ->assertSee('2026-09-18 02:00 (Europe/Budapest)', false) - ->set('dockerCleanupFrequency', 'hourly') - ->call('submit') - ->assertSee('2026-09-17 13:00 (Europe/Budapest)', false) - ->assertSee('in 39 minutes'); -}); - -it('shows Calculating when the Docker cleanup next run is not calculated yet', function () { - nextRunDisplaySetCleanupNextRun($this->server, null); - - Livewire::test(DockerCleanup::class, ['server_uuid' => $this->server->uuid]) - ->assertSee('Next run') - ->assertSee('Calculating…'); -}); - -it('marks the Docker cleanup as stale only when its next run is more than 10 minutes overdue', function (?int $minutesFromNow, bool $stale) { - nextRunDisplaySetCleanupNextRun($this->server, $minutesFromNow === null ? null : now()->addMinutes($minutesFromNow)); - - $component = Livewire::test(DockerCleanup::class, ['server_uuid' => $this->server->uuid]); - - expect($component->instance()->isCleanupStale())->toBe($stale); - if ($stale) { - $component->assertSee('Docker cleanup may be stalled'); - } else { - $component->assertDontSee('Docker cleanup may be stalled'); - } -})->with([ - '30 minutes overdue' => [-30, true], - '5 minutes overdue' => [-5, false], - 'in the future' => [30, false], - 'not calculated yet' => [null, false], -]); - -it('does not mark the Docker cleanup as stale on a server without a real IP address', function () { - $this->server->update(['ip' => Server::PLACEHOLDER_IP]); - nextRunDisplaySetCleanupNextRun($this->server, now()->subMinutes(30)); - - $component = Livewire::test(DockerCleanup::class, ['server_uuid' => $this->server->uuid]); - - expect($component->instance()->isCleanupStale())->toBeFalse(); -}); - -it('shows the next scheduled task run in the server timezone and refreshes it on changes', function () { - $task = ScheduledTask::factory()->create([ - 'team_id' => $this->team->id, - 'application_id' => $this->application->id, - 'frequency' => '0 2 * * *', - 'enabled' => true, - ]); - nextRunDisplayBindTaskRoute($this->application, $task); - - $component = Livewire::test(Show::class) - ->assertSee('Next run') - ->assertSee('2026-09-18 02:00 (Europe/Budapest)', false) - ->assertSee('in 13 hours'); - - $component->call('toggleEnabled') - ->assertDontSee('2026-09-18 02:00 (Europe/Budapest)', false) - ->assertSeeInOrder(['Next run', 'Disabled']); - - $component->call('toggleEnabled') - ->set('frequency', 'hourly') - ->call('submit') - ->assertSee('2026-09-17 13:00 (Europe/Budapest)', false); - - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-17 11:00:00'); -}); - -it('shows Calculating for an enabled scheduled task without a next run', function () { - $task = ScheduledTask::factory()->create([ - 'team_id' => $this->team->id, - 'application_id' => $this->application->id, - 'frequency' => '0 2 * * *', - 'enabled' => true, - ]); - ScheduledTask::query()->whereKey($task->id)->update(['next_run_at' => null]); - nextRunDisplayBindTaskRoute($this->application, $task); - - Livewire::test(Show::class)->assertSee('Calculating…'); -}); - -it('shows an invalid frequency instead of Calculating when the next run cannot be calculated', function () { - $task = ScheduledTask::factory()->create([ - 'team_id' => $this->team->id, - 'application_id' => $this->application->id, - 'frequency' => '0 2 * * *', - 'enabled' => true, - ]); - // The forms reject such a value, so only old data or a direct database change can have it. - ScheduledTask::query()->whereKey($task->id)->update(['frequency' => '60 * * * *', 'next_run_at' => null]); - nextRunDisplayBindTaskRoute($this->application, $task); - - Livewire::test(Show::class) - ->assertSee('Invalid frequency') - ->assertDontSee('Calculating…'); -}); - -it('shows an invalid Docker cleanup frequency', function () { - ServerSetting::query()->where('server_id', $this->server->id)->update(['docker_cleanup_frequency' => '0 0 31 2 *', 'docker_cleanup_next_run_at' => null]); - - Livewire::test(DockerCleanup::class, ['server_uuid' => $this->server->uuid]) - ->assertSee('Invalid frequency') - ->assertDontSee('Calculating…'); -}); - -it('shows the next database backup run in the server timezone', function () { - $database = StandalonePostgresql::create([ - 'name' => 'pg-next-run', - 'image' => 'postgres:16-alpine', - 'postgres_user' => 'postgres', - 'postgres_password' => 'password', - 'postgres_db' => 'postgres', - 'environment_id' => $this->environment->id, - 'destination_id' => $this->destination->id, - 'destination_type' => $this->destination->getMorphClass(), - ]); - $backup = ScheduledDatabaseBackup::create([ - 'frequency' => '0 2 * * *', - 'enabled' => true, - 'database_type' => $database->getMorphClass(), - 'database_id' => $database->id, - 'team_id' => $this->team->id, - ]); - - Livewire::test(BackupEdit::class, ['backup' => $backup->fresh(), 'availableS3Storages' => collect()]) - ->assertSee('2026-09-18 02:00 (Europe/Budapest)', false) - ->call('toggleEnabled') - ->assertDontSee('2026-09-18 02:00 (Europe/Budapest)', false) - ->assertSeeInOrder(['Next run', 'Disabled']); -}); - -it('shows the next volume backup run in the server timezone', function () { - $volume = LocalPersistentVolume::create([ - 'name' => 'app-data', - 'mount_path' => '/data', - 'resource_id' => $this->application->id, - 'resource_type' => $this->application->getMorphClass(), - ]); - - Livewire::test(VolumeBackups::class, ['storage' => $volume, 'resource' => $this->application]) - ->assertSeeInOrder(['Next run', 'Disabled']) - ->set('frequency', '0 2 * * *') - ->call('toggleEnabled') - ->assertSee('2026-09-18 02:00 (Europe/Budapest)', false); -}); diff --git a/tests/Feature/ScheduledJobManagerDispatchTest.php b/tests/Feature/ScheduledJobManagerDispatchTest.php index 61ecfa66ef..eaff025bfc 100644 --- a/tests/Feature/ScheduledJobManagerDispatchTest.php +++ b/tests/Feature/ScheduledJobManagerDispatchTest.php @@ -2,7 +2,6 @@ use App\Events\BackupCreated; use App\Jobs\DatabaseBackupJob; -use App\Jobs\DockerCleanupJob; use App\Jobs\ScheduledJobManager; use App\Jobs\ScheduledTaskJob; use App\Models\Application; @@ -11,6 +10,7 @@ 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\ServerSetting; @@ -23,6 +23,7 @@ use App\Services\ScheduledJobDeliveryService; use Illuminate\Foundation\Testing\RefreshDatabase; use Illuminate\Support\Carbon; use Illuminate\Support\Facades\Bus; +use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Event; use Illuminate\Support\Facades\Log; @@ -32,219 +33,173 @@ uses(RefreshDatabase::class); beforeEach(function () { Server::flushIdentityMap(); - config(['constants.coolify.self_hosted' => true]); - Queue::fake(); }); -it('dispatches due scheduled tasks across chunks and moves their next run forward', function () { - Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 0, 30, 'UTC')); - $application = createScheduledTaskApplication(); - foreach (range(1, 101) as $ignored) { - createScheduledApplicationTask($application, ['frequency' => '* * * * *']); - } - +it('dispatches scheduled tasks across chunks', function () { + config(['constants.coolify.self_hosted' => true]); Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 1, 0, 'UTC')); + Queue::fake(); + + $team = Team::factory()->create(); + $privateKey = PrivateKey::create([ + 'name' => 'Test Key', + 'private_key' => '-----BEGIN OPENSSH PRIVATE KEY----- +b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAMwAAAAtzc2gtZW +QyNTUxOQAAACBbhpqHhqv6aI67Mj9abM3DVbmcfYhZAhC7ca4d9UCevAAAAJi/QySHv0Mk +hwAAAAtzc2gtZWQyNTUxOQAAACBbhpqHhqv6aI67Mj9abM3DVbmcfYhZAhC7ca4d9UCevA +AAAECBQw4jg1WRT2IGHMncCiZhURCts2s24HoDS0thHnnRKVuGmoeGq/pojrsyP1pszcNV +uZx9iFkCELtxrh31QJ68AAAAEXNhaWxANzZmZjY2ZDJlMmRkAQIDBA== +-----END OPENSSH PRIVATE KEY-----', + 'team_id' => $team->id, + ]); + $server = Server::factory()->create([ + 'team_id' => $team->id, + 'private_key_id' => $privateKey->id, + ]); + $server->settings()->update([ + 'is_reachable' => true, + 'is_usable' => true, + 'force_disabled' => false, + 'docker_cleanup_frequency' => '0 * * * *', + ]); + + $destination = StandaloneDocker::where('server_id', $server->id)->first() + ?? StandaloneDocker::factory()->create(['server_id' => $server->id]); + $project = Project::factory()->create(['team_id' => $team->id]); + $environment = Environment::factory()->create(['project_id' => $project->id]); + $application = Application::factory()->create([ + 'environment_id' => $environment->id, + 'destination_id' => $destination->id, + 'destination_type' => StandaloneDocker::class, + 'status' => 'running', + ]); + + ScheduledTask::factory() + ->count(101) + ->create([ + 'team_id' => $team->id, + 'application_id' => $application->id, + 'frequency' => '* * * * *', + 'enabled' => true, + ]); + (new ScheduledJobManager)->handle(); Queue::assertPushed(ScheduledTaskJob::class, 101); - expect(ScheduledTask::query()->pluck('next_run_at')->map->toDateTimeString()->unique()->all()) - ->toBe(['2026-05-27 00:02:00']); }); -it('runs a new schedule first at its next due time', function () { - Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 1, 20, 'UTC')); - $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => '* * * * *']); +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(); + + $application = createScheduledTaskApplication(); + + $task = ScheduledTask::factory()->create([ + 'team_id' => $application->environment->project->team_id, + 'application_id' => $application->id, + 'frequency' => '0 2 * * *', + 'enabled' => true, + ]); (new ScheduledJobManager)->handle(); Queue::assertNotPushed(ScheduledTaskJob::class); - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-05-27 00:02:00'); + expect(ScheduledJobDelivery::query()->where('schedule_key', "scheduled-task:{$task->id}")->exists())->toBeFalse(); }); -it('does not query schedules that are not due', function () { - Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 0, 0, 'UTC')); +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(); - $database = createScheduledBackupDatabase(); - $countQueries = function (): int { - DB::flushQueryLog(); - DB::enableQueryLog(); - (new ScheduledJobManager)->handle(); - $count = count(DB::getQueryLog()); - DB::disableQueryLog(); - return $count; - }; - // Initialize the Docker cleanup schedules of both servers, so only the schedules below change. - Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 1, 0, 'UTC')); - $countQueries(); - $withoutSchedules = $countQueries(); + ScheduledTask::factory()->create([ + 'team_id' => $application->environment->project->team_id, + 'application_id' => $application->id, + 'frequency' => 'daily', + 'enabled' => true, + ]); - foreach (range(1, 5) as $ignored) { - createScheduledApplicationTask($application, ['frequency' => '0 2 * * *']); - createScheduledDatabaseBackup($database, ['frequency' => '0 2 * * *']); - } + (new ScheduledJobManager)->handle(); - expect($countQueries())->toBe($withoutSchedules); - Queue::assertNothingPushed(); + Queue::assertPushed(ScheduledTaskJob::class, 1); }); -it('dispatches one job when several managers run for the same occurrence', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 16, 12, 0, 0, 'UTC')); - $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => 'daily']); +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, + ]); - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 0, 5, 'UTC')); (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($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-18 00:00:00'); + ->and(ScheduledJobState::query()->where('schedule_key', "scheduled-task:{$task->id}")->count())->toBe(1); }); -it('does not dispatch an occurrence that another manager claims after this one read it', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 16, 12, 0, 0, 'UTC')); - $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => 'daily']); - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 0, 5, 'UTC')); +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(); - // Another node claims the occurrence between this manager's select and its update. - $claimedByOtherNode = false; - ScheduledTask::retrieved(function (ScheduledTask $retrieved) use (&$claimedByOtherNode) { - if (! $claimedByOtherNode) { - $claimedByOtherNode = true; - ScheduledTask::query()->whereKey($retrieved->id)->toBase()->update(['next_run_at' => '2026-09-18 00:00:00']); - } - }); - - (new ScheduledJobManager)->handle(); - - Queue::assertNotPushed(ScheduledTaskJob::class); - expect($claimedByOtherNode)->toBeTrue() - ->and(ScheduledJobDelivery::query()->where('schedule_key', "scheduled-task:{$task->id}")->exists())->toBeFalse(); -}); - -it('calculates the next run of a schedule without one and runs it when it is due now', function (string $frequency, bool $runsNow, string $nextRunAt) { - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 0, 20, 'UTC')); - $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => $frequency]); - ScheduledTask::query()->whereKey($task->id)->update(['next_run_at' => null]); - - (new ScheduledJobManager)->handle(); - - Queue::assertPushed(ScheduledTaskJob::class, $runsNow ? 1 : 0); - expect($task->fresh()->next_run_at->toDateTimeString())->toBe($nextRunAt); -})->with([ - 'due in the current minute' => ['0 0 * * *', true, '2026-09-18 00:00:00'], - 'due later' => ['30 * * * *', false, '2026-09-17 00:30:00'], -]); - -it('moves the next run forward when an occurrence is skipped, and does not retry it', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 16, 12, 0, 0, 'UTC')); - $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => 'daily']); + $application = createScheduledTaskApplication(); + $task = createScheduledApplicationTask($application, ['frequency' => 'daily']); $server = $task->server(); $server->settings()->update(['is_reachable' => false]); - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 0, 5, 'UTC')); (new ScheduledJobManager)->handle(); + $server->settings()->update(['is_reachable' => true]); - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 1, 0, 'UTC')); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 6, 0, 'UTC')); (new ScheduledJobManager)->handle(); Queue::assertNotPushed(ScheduledTaskJob::class); - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-18 00:00:00') - ->and(ScheduledJobDelivery::query()->where('schedule_key', "scheduled-task:{$task->id}")->exists())->toBeFalse(); + 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 dispatch a task when its application is not running', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 16, 12, 0, 0, 'UTC')); +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']); - $logPath = captureScheduledJobManagerTestLog('scheduled'); - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 0, 5, 'UTC')); (new ScheduledJobManager)->handle(); Queue::assertNotPushed(ScheduledTaskJob::class); - expect(file_get_contents($logPath))->toContain('Task skipped')->toContain('"skip_reason":"application_not_running"'); - @unlink($logPath); -}); - -it('runs late backups once, but logs late tasks outside the late-run window as missed', function (int $minutesLate, bool $taskRuns) { - Carbon::setTestNow(Carbon::create(2026, 9, 16, 12, 0, 0, 'UTC')); - $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => 'daily']); - $backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => 'daily']); - $logPath = captureScheduledJobManagerTestLog('scheduled'); - - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 0, 0, 'UTC')->addMinutes($minutesLate)); - (new ScheduledJobManager)->handle(); - - Queue::assertPushed(ScheduledTaskJob::class, $taskRuns ? 1 : 0); - Queue::assertPushed(DatabaseBackupJob::class, 1); - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-18 00:00:00') - ->and($backup->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-18 00:00:00') - ->and(str_contains(file_get_contents($logPath), 'Task missed'))->toBe(! $taskRuns); - @unlink($logPath); -})->with([ - 'five minutes late' => [5, true], - 'three hours late' => [180, false], -]); - -it('runs a daily schedule once on the night the clocks go back', function () { - Carbon::setTestNow(Carbon::create(2026, 10, 24, 12, 0, 0, 'UTC')); - $application = createScheduledTaskApplication(); - $application->destination->server->settings->update(['server_timezone' => 'Europe/Budapest']); - $task = createScheduledApplicationTask($application, ['frequency' => '30 2 * * *']); - - // 02:30 CEST. At 03:00 CEST the clocks go back to 02:00, so 02:30 CET follows one hour later. - Carbon::setTestNow(Carbon::create(2026, 10, 25, 0, 30, 5, 'UTC')); - (new ScheduledJobManager)->handle(); - - Queue::assertPushed(ScheduledTaskJob::class, 1); - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-10-26 01:30:00'); -}); - -it('dispatches due Docker cleanups with their settings', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 30, 0, 'UTC')); - $server = createScheduledTaskApplication()->destination->server; - $server->settings->update(['docker_cleanup_frequency' => '0 * * * *', 'delete_unused_volumes' => true]); - - Carbon::setTestNow(Carbon::create(2026, 9, 17, 1, 0, 10, 'UTC')); - (new ScheduledJobManager)->handle(); - - Queue::assertPushed(DockerCleanupJob::class, fn (DockerCleanupJob $job) => $job->server->is($server) && $job->deleteUnusedVolumes); - expect($server->settings->fresh()->docker_cleanup_next_run_at->toDateTimeString())->toBe('2026-09-17 02:00:00'); -}); - -it('does not dispatch an occurrence that already has a delivery', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 16, 12, 0, 0, 'UTC')); - $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => 'daily']); - 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' => 'claimed', - ]); - - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 0, 5, 'UTC')); - (new ScheduledJobManager)->handle(); - - Queue::assertNotPushed(ScheduledTaskJob::class); - expect($task->fresh()->next_run_at->toDateTimeString())->toBe('2026-09-18 00:00:00'); -}); - -it('dispatches a due backup of a service database', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 4, 30, 'UTC')); - $backup = createScheduledServiceDatabaseBackup(['frequency' => '* * * * *']); - - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); - (new ScheduledJobManager)->handle(); - - Queue::assertPushed(DatabaseBackupJob::class, fn (DatabaseBackupJob $job) => $job->backup->is($backup)); + 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')); - $task = createScheduledApplicationTask(createScheduledTaskApplication(), ['frequency' => 'daily']); + 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'), @@ -258,37 +213,6 @@ it('publishes a pending occurrence after a previous publisher interruption', fun expect($occurrence->fresh()->status)->toBe('enqueued'); }); -it('keeps publishing pending occurrences after one of them fails', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); - $logPath = captureScheduledJobManagerTestLog('scheduled-errors'); - $database = createScheduledBackupDatabase(); - $occurrences = collect([createScheduledDatabaseBackup($database), createScheduledDatabaseBackup($database)]) - ->map(fn (ScheduledDatabaseBackup $backup) => ScheduledJobDelivery::create([ - 'schedule_key' => "scheduled-backup:{$backup->id}", - 'scheduled_for' => now()->startOfMinute(), - 'job_type' => 'database-backup', - 'resource_id' => $backup->id, - 'status' => 'pending', - ])); - - $dispatchAttempts = 0; - Bus::shouldReceive('dispatch')->andReturnUsing(function () use (&$dispatchAttempts) { - if (++$dispatchAttempts === 1) { - throw new RuntimeException('queue connection lost'); - } - }); - - app(ScheduledJobDeliveryService::class)->publishPending(); - - $log = file_get_contents($logPath); - @unlink($logPath); - - expect($occurrences->first()->fresh()->status)->toBe('pending') - ->and($occurrences->last()->fresh()->status)->toBe('enqueued') - ->and($log)->toContain('Failed to publish pending scheduled occurrence') - ->toContain('queue connection lost'); -}); - it('allows only one worker to claim an occurrence', function () { $occurrence = ScheduledJobDelivery::create([ 'schedule_key' => 'scheduled-task:claim-test', @@ -352,8 +276,30 @@ it('marks stale claimed deliveries as failed', function () { expect($delivery->fresh()->status)->toBe('failed'); }); +it('dispatches the instance coolify-db backup even when its id is zero', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 1, 0, 'UTC')); + Queue::fake(); + + $database = createScheduledBackupDatabase(); + $backup = createScheduledDatabaseBackup($database, [ + 'id' => 0, + 'frequency' => '* * * * *', + ]); + + expect($backup->id)->toBe(0); + + (new ScheduledJobManager)->handle(); + + Queue::assertPushed(DatabaseBackupJob::class, 1); + Queue::assertPushed(DatabaseBackupJob::class, fn (DatabaseBackupJob $job) => $job->backup->id === 0); +}); + it('dispatches zero-id schedules and continues with positive ids', function () { - Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 0, 30, 'UTC')); + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 1, 0, 'UTC')); + Queue::fake(); + $application = createScheduledTaskApplication(); $database = StandalonePostgresql::create([ 'name' => 'coolify-db', @@ -366,14 +312,17 @@ it('dispatches zero-id schedules and continues with positive ids', function () { 'destination_id' => $application->destination_id, 'destination_type' => $application->destination_type, ]); + $zeroIdBackup = createScheduledDatabaseBackup($database, ['id' => 0]); $positiveIdBackup = createScheduledDatabaseBackup($database); $zeroIdTask = createScheduledApplicationTask($application, ['id' => 0]); $positiveIdTask = createScheduledApplicationTask($application); - expect($zeroIdBackup->id)->toBe(0)->and($zeroIdTask->id)->toBe(0); + expect($zeroIdBackup->id)->toBe(0) + ->and($positiveIdBackup->id)->toBeGreaterThan(0) + ->and($zeroIdTask->id)->toBe(0) + ->and($positiveIdTask->id)->toBeGreaterThan(0); - Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 1, 0, 'UTC')); (new ScheduledJobManager)->handle(); Queue::assertPushed(DatabaseBackupJob::class, 2); @@ -385,7 +334,14 @@ it('dispatches zero-id schedules and continues with positive ids', function () { }); it('does not query relationships when constructing scheduled task jobs', function () { - $task = createScheduledApplicationTask(createScheduledTaskApplication()); + $application = createScheduledTaskApplication(); + + $task = ScheduledTask::factory()->create([ + 'team_id' => $application->environment->project->team_id, + 'application_id' => $application->id, + 'frequency' => '* * * * *', + 'enabled' => true, + ])->fresh(); DB::flushQueryLog(); DB::enableQueryLog(); @@ -397,32 +353,56 @@ it('does not query relationships when constructing scheduled task jobs', functio ->and($job->timeout)->toBe(300); }); -it('reads current server settings in each manager run of the same process', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 15, 12, 0, 0, 'UTC')); +it('reads current server settings in each queued manager run of the same worker', function () { + config(['constants.coolify.self_hosted' => true]); + Queue::fakeExcept(ScheduledJobManager::class); + $database = createScheduledBackupDatabase(); createScheduledDatabaseBackup($database, ['frequency' => 'daily']); $serverId = $database->destination->server_id; ServerSetting::query()->where('server_id', $serverId)->update(['is_reachable' => false]); - Carbon::setTestNow(Carbon::create(2026, 9, 16, 0, 0, 5, 'UTC')); - $this->artisan('scheduled:dispatch')->assertSuccessful(); + Carbon::setTestNow(Carbon::create(2026, 9, 16, 0, 5, 0, 'UTC')); + dispatch(new ScheduledJobManager); Queue::assertNotPushed(DatabaseBackupJob::class); // Another worker marks the server reachable again before the next daily run. ServerSetting::query()->where('server_id', $serverId)->update(['is_reachable' => true]); - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 0, 5, 'UTC')); - $this->artisan('scheduled:dispatch')->assertSuccessful(); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); + dispatch(new ScheduledJobManager); Queue::assertPushed(DatabaseBackupJob::class, 1); }); -it('logs each step of a scheduled occurrence', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 16, 12, 0, 0, 'UTC')); - $backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => 'daily']); +it('logs the current database server state when a backup is skipped', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); + Queue::fake(); $logPath = captureScheduledJobManagerTestLog('scheduled'); + $database = createScheduledBackupDatabase(); + $backup = createScheduledDatabaseBackup($database, ['frequency' => 'daily']); + ServerSetting::query()->where('server_id', $database->destination->server_id)->update(['is_reachable' => false]); + + (new ScheduledJobManager)->handle(); + + $log = file_get_contents($logPath); + @unlink($logPath); + + expect($log)->toContain('Backup skipped') + ->toContain('"skip_reason":"server_not_functional"') + ->toContain("\"backup_id\":{$backup->id}") + ->toContain('"db_is_reachable":false'); +}); + +it('logs each step of a scheduled occurrence', function () { + config(['constants.coolify.self_hosted' => true]); Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); + Queue::fake(); + $logPath = captureScheduledJobManagerTestLog('scheduled'); + + $backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => 'daily']); (new ScheduledJobManager)->handle(); $occurrenceUuid = ScheduledJobDelivery::query()->where('schedule_key', "scheduled-backup:{$backup->id}")->value('uuid'); $service = app(ScheduledJobDeliveryService::class); @@ -437,6 +417,7 @@ it('logs each step of a scheduled occurrence', function () { expect($log)->toContain('Backup dispatched') ->toContain('Scheduled occurrence enqueued') ->toContain('"occurrence_uuid":"'.$occurrenceUuid.'"') + ->toContain('"host":"'.gethostname().'"') ->toContain('Scheduled occurrence claimed') ->toContain('Scheduled occurrence claim rejected') ->toContain('Scheduled occurrence completed') @@ -444,83 +425,220 @@ it('logs each step of a scheduled occurrence', function () { ->toContain('"seconds_since_due":300'); }); +it('warns when the manager starts after the catch-up window', function (int $minutesSincePreviousStart, bool $expectWarning) { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 30, 0, 'UTC')); + Queue::fake(); + $logPath = captureScheduledJobManagerTestLog('scheduled-errors'); + Cache::put('scheduled-job-manager:last-started-at', now()->subMinutes($minutesSincePreviousStart)->toIso8601String()); + + (new ScheduledJobManager)->handle(); + + $log = file_get_contents($logPath); + @unlink($logPath); + + expect(str_contains($log, 'ScheduledJobManager started late'))->toBe($expectWarning); +})->with([ + 'normal one-minute interval' => [1, false], + 'twenty-minute gap' => [20, true], +]); + +it('warns when an occurrence is enqueued while the previous one is still running', function () { + config(['constants.coolify.self_hosted' => true]); + Queue::fake(); + $logPath = captureScheduledJobManagerTestLog('scheduled'); + $backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => 'daily']); + + Carbon::setTestNow(Carbon::create(2026, 9, 16, 0, 5, 0, 'UTC')); + (new ScheduledJobManager)->handle(); + $previousUuid = ScheduledJobDelivery::query()->where('schedule_key', "scheduled-backup:{$backup->id}")->value('uuid'); + app(ScheduledJobDeliveryService::class)->claim($previousUuid, 'worker-a'); + + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); + (new ScheduledJobManager)->handle(); + + $log = file_get_contents($logPath); + @unlink($logPath); + + expect($log)->toContain('Scheduled occurrence enqueued while the previous occurrence is still open') + ->toContain('"previous_occurrence_uuid":"'.$previousUuid.'"') + ->toContain('"previous_status":"claimed"'); +}); + +it('reports occurrences that stay open after fifteen minutes', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 30, 0, 'UTC')); + Queue::fake(); + $logPath = captureScheduledJobManagerTestLog('scheduled'); + + foreach (['enqueued' => 20, 'claimed' => 20, 'failed' => 20] as $status => $minutesAgo) { + ScheduledJobDelivery::create([ + 'schedule_key' => "scheduled-backup:{$status}", + 'scheduled_for' => now()->subMinutes($minutesAgo), + 'job_type' => 'database-backup', + 'resource_id' => 1, + 'status' => $status, + ])->forceFill(['created_at' => now()->subMinutes($minutesAgo)])->save(); + } + ScheduledJobDelivery::create([ + 'schedule_key' => 'scheduled-backup:recent', + 'scheduled_for' => now()->subMinutes(2), + 'job_type' => 'database-backup', + 'resource_id' => 1, + 'status' => 'enqueued', + ]); + + (new ScheduledJobManager)->handle(); + + $log = file_get_contents($logPath); + @unlink($logPath); + + expect($log)->toContain('"open_occurrences_over_15_minutes":{"claimed":1,"enqueued":1}'); +}); + +it('keeps publishing pending occurrences after one of them fails', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); + $logPath = captureScheduledJobManagerTestLog('scheduled-errors'); + $database = createScheduledBackupDatabase(); + $occurrences = collect([createScheduledDatabaseBackup($database), createScheduledDatabaseBackup($database)]) + ->map(fn (ScheduledDatabaseBackup $backup) => ScheduledJobDelivery::create([ + 'schedule_key' => "scheduled-backup:{$backup->id}", + 'scheduled_for' => now()->startOfMinute(), + 'job_type' => 'database-backup', + 'resource_id' => $backup->id, + 'status' => 'pending', + ])); + + $dispatchAttempts = 0; + Bus::shouldReceive('dispatch')->andReturnUsing(function () use (&$dispatchAttempts) { + if (++$dispatchAttempts === 1) { + throw new RuntimeException('queue connection lost'); + } + }); + + app(ScheduledJobDeliveryService::class)->publishPending(); + + $log = file_get_contents($logPath); + @unlink($logPath); + + expect($occurrences->first()->fresh()->status)->toBe('pending') + ->and($occurrences->last()->fresh()->status)->toBe('enqueued') + ->and($log)->toContain('Failed to publish pending scheduled occurrence') + ->toContain('queue connection lost'); +}); + it('logs a backup run that ends early, even when it completes before the enqueue log', function () { - Carbon::setTestNow(Carbon::create(2026, 9, 16, 12, 0, 0, 'UTC')); + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); Event::fake([BackupCreated::class]); - Queue::fakeExcept([DatabaseBackupJob::class]); + Queue::fakeExcept([ScheduledJobManager::class, DatabaseBackupJob::class]); + $logPath = captureScheduledJobManagerTestLog('scheduled'); + $database = createScheduledBackupDatabase(); $database->update(['status' => 'exited']); $backup = createScheduledDatabaseBackup($database, ['frequency' => 'daily']); - $logPath = captureScheduledJobManagerTestLog('scheduled'); - Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); - $this->artisan('scheduled:dispatch')->assertSuccessful(); + dispatch(new ScheduledJobManager); $log = file_get_contents($logPath); @unlink($logPath); expect($log)->toContain('Database backup job ended without a backup') ->toContain('"skip_reason":"database_not_running"') + ->toContain('"database_status":"exited:unhealthy"') ->toContain('Scheduled occurrence completed') ->toMatch('/Scheduled occurrence enqueued \{[^\n]*"schedule_key":"scheduled-backup:'.$backup->id.'"[^\n]*"status":"enqueued"/') ->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()]); +it('runs the manager in the scheduler process through the scheduled:dispatch command', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 5, 27, 0, 1, 0, 'UTC')); + Queue::fakeExcept(ScheduledJobManager::class); - foreach (range(1, 3) as $minute) { - Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, $minute, 0, 'UTC')); - (new ScheduledJobManager)->handle(); - } + $application = createScheduledTaskApplication(); + createScheduledApplicationTask($application, ['frequency' => '* * * * *']); - $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(); + $this->artisan('scheduled:dispatch')->assertSuccessful(); + Queue::assertNotPushed(ScheduledJobManager::class); 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')); +it('loads the server chain of scheduled database backups without a query for each backup', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); + Queue::fake(); + + $database = createScheduledBackupDatabase(); + createScheduledDatabaseBackup($database, ['frequency' => '0 2 * * *']); + foreach (range(1, 2) as $number) { + $otherDatabase = StandalonePostgresql::create([ + ...$database->only(['image', 'postgres_user', 'postgres_password', 'postgres_db', 'status', 'environment_id', 'destination_id', 'destination_type']), + 'name' => "database-{$number}", + ]); + createScheduledDatabaseBackup($otherDatabase, ['frequency' => '0 2 * * *']); + } + createScheduledServiceDatabaseBackup(['frequency' => '0 2 * * *'], Application::query()->first()); + + DB::flushQueryLog(); + DB::enableQueryLog(); + (new ScheduledJobManager)->handle(); + $queries = collect(DB::getQueryLog())->pluck('query'); + DB::disableQueryLog(); + + $perBackupLookups = $queries->filter(fn (string $query) => str_contains($query, '"standalone_dockers"."id" = ?') + || str_contains($query, '"services"."id" = ?') + || str_contains($query, '"server_settings"."server_id" = ?')); + + expect($perBackupLookups)->toBeEmpty(); +}); + +it('dispatches a due backup of a service database', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 0, 'UTC')); + Queue::fake(); + + $backup = createScheduledServiceDatabaseBackup(['frequency' => '* * * * *']); + + (new ScheduledJobManager)->handle(); + + Queue::assertPushed(DatabaseBackupJob::class, fn (DatabaseBackupJob $job) => $job->backup->is($backup)); +}); + +it('keeps dispatching other schedules when one schedule throws a PHP error', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 0, 5, 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]); + createScheduledDatabaseBackup($database); + createScheduledDatabaseBackup($database); - foreach (range(1, 3) as $minute) { - Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, $minute, 0, 'UTC')); - (new ScheduledJobManager)->handle(); - } + $dispatchAttempts = 0; + Bus::shouldReceive('dispatch')->andReturnUsing(function () use (&$dispatchAttempts) { + if (++$dispatchAttempts === 1) { + throw new Error('broken schedule'); + } + }); + + (new ScheduledJobManager)->handle(); $log = file_get_contents($logPath); @unlink($logPath); - expect(substr_count($log, 'Backup has an invalid frequency'))->toBe(1); + // The first backup stays pending for publishPending(); the second one is still enqueued. + expect(ScheduledJobDelivery::query()->where('job_type', 'database-backup')->orderBy('id')->pluck('status')->all())->toBe(['pending', 'enqueued']) + ->and($log)->toContain('Error processing backup') + ->toContain('broken schedule'); }); it('republishes a backup occurrence whose queued job was lost, and the old job cannot run it again', function () { + config(['constants.coolify.self_hosted' => true]); Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + Queue::fake(); $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), - ]); + $occurrence = createStaleEnqueuedOccurrence("scheduled-backup:{$backup->id}", 'database-backup', $backup->id, Carbon::create(2026, 9, 17, 0, 0, 0, 'UTC')); (new ScheduledJobManager)->handle(); @@ -533,18 +651,32 @@ it('republishes a backup occurrence whose queued job was lost, and the old job c ->and($service->claim($occurrence->uuid, 'original-job'))->toBeFalse(); }); -it('logs a lost scheduled task occurrence as missed instead of running it late', function () { +it('does not run a lost backup occurrence again when a newer occurrence exists', function () { + config(['constants.coolify.self_hosted' => true]); Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + Queue::fake(); + $logPath = captureScheduledJobManagerTestLog('scheduled'); + $backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => 'hourly']); + $lost = createStaleEnqueuedOccurrence("scheduled-backup:{$backup->id}", 'database-backup', $backup->id, Carbon::create(2026, 9, 17, 10, 0, 0, 'UTC')); + createStaleEnqueuedOccurrence("scheduled-backup:{$backup->id}", 'database-backup', $backup->id, Carbon::create(2026, 9, 17, 11, 0, 0, 'UTC'), minutesAgo: 5); + + (new ScheduledJobManager)->handle(); + + $log = file_get_contents($logPath); + @unlink($logPath); + + Queue::assertNotPushed(DatabaseBackupJob::class, fn (DatabaseBackupJob $job) => $job->occurrenceUuid === $lost->uuid); + expect($lost->fresh()->status)->toBe('failed') + ->and($log)->toContain('Scheduled occurrence skipped: its queued job was not started and a newer occurrence exists'); +}); + +it('logs a lost scheduled task occurrence as missed instead of running it late', function () { + config(['constants.coolify.self_hosted' => true]); + Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + Queue::fake(); $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), - ]); + $occurrence = createStaleEnqueuedOccurrence("scheduled-task:{$task->id}", 'scheduled-task', $task->id, Carbon::create(2026, 9, 17, 0, 0, 0, 'UTC')); (new ScheduledJobManager)->handle(); @@ -558,16 +690,11 @@ it('logs a lost scheduled task occurrence as missed instead of running it late', }); it('leaves recently enqueued occurrences alone', function () { + config(['constants.coolify.self_hosted' => true]); Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC')); + Queue::fake(); $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), - ]); + $occurrence = createStaleEnqueuedOccurrence("scheduled-backup:{$backup->id}", 'database-backup', $backup->id, Carbon::create(2026, 9, 17, 11, 0, 0, 'UTC'), minutesAgo: ScheduledJobDeliveryService::ENQUEUED_STALE_AFTER_MINUTES - 1); (new ScheduledJobManager)->handle(); @@ -575,6 +702,46 @@ it('leaves recently enqueued occurrences alone', function () { expect($occurrence->fresh()->enqueued_at->toDateTimeString())->toBe('2026-09-17 11:01:00'); }); +function createStaleEnqueuedOccurrence(string $scheduleKey, string $jobType, int $resourceId, Carbon $scheduledFor, int $minutesAgo = ScheduledJobDeliveryService::ENQUEUED_STALE_AFTER_MINUTES + 1): ScheduledJobDelivery +{ + return ScheduledJobDelivery::create([ + 'schedule_key' => $scheduleKey, + 'scheduled_for' => $scheduledFor, + 'job_type' => $jobType, + 'resource_id' => $resourceId, + 'status' => 'enqueued', + 'enqueued_at' => now()->subMinutes($minutesAgo), + ]); +} + +function createScheduledServiceDatabaseBackup(array $overrides = [], ?Application $application = null): ScheduledDatabaseBackup +{ + $application ??= createScheduledTaskApplication(); + $service = Service::factory()->create([ + 'environment_id' => $application->environment_id, + 'destination_id' => $application->destination_id, + 'destination_type' => $application->destination_type, + ]); + $database = ServiceDatabase::create([ + 'name' => 'postgres', + 'image' => 'postgres:16-alpine', + 'service_id' => $service->id, + ]); + + $backup = new ScheduledDatabaseBackup; + $backup->forceFill(array_merge([ + 'enabled' => true, + 'save_s3' => false, + 'frequency' => '* * * * *', + 'database_id' => $database->id, + 'database_type' => $database->getMorphClass(), + 'team_id' => $application->environment->project->team_id, + ], $overrides)); + $backup->save(); + + return $backup->fresh(); +} + function captureScheduledJobManagerTestLog(string $channel): string { $path = tempnam(sys_get_temp_dir(), 'coolify-scheduled-log-'); @@ -655,34 +822,6 @@ function createScheduledDatabaseBackup(StandalonePostgresql $database, array $ov return $backup->fresh(); } -function createScheduledServiceDatabaseBackup(array $overrides = []): ScheduledDatabaseBackup -{ - $application = createScheduledTaskApplication(); - $service = Service::factory()->create([ - 'environment_id' => $application->environment_id, - 'destination_id' => $application->destination_id, - 'destination_type' => $application->destination_type, - ]); - $database = ServiceDatabase::create([ - 'name' => 'postgres', - 'image' => 'postgres:16-alpine', - 'service_id' => $service->id, - ]); - - $backup = new ScheduledDatabaseBackup; - $backup->forceFill(array_merge([ - 'enabled' => true, - 'save_s3' => false, - 'frequency' => '* * * * *', - 'database_id' => $database->id, - 'database_type' => $database->getMorphClass(), - 'team_id' => $application->environment->project->team_id, - ], $overrides)); - $backup->save(); - - return $backup->fresh(); -} - function createScheduledApplicationTask(Application $application, array $overrides = []): ScheduledTask { $task = new ScheduledTask; diff --git a/tests/Feature/VolumeBackupTest.php b/tests/Feature/VolumeBackupTest.php index a161864408..623e140dfb 100644 --- a/tests/Feature/VolumeBackupTest.php +++ b/tests/Feature/VolumeBackupTest.php @@ -2519,9 +2519,9 @@ it('dispatches due scheduled volume backups', function () { 'frequency' => '* * * * *', 'enabled' => true, ]); + Cache::forget("scheduled-volume-backup:{$backup->id}"); expect($backup->server()?->isFunctional())->toBeTrue(); - $this->travel(1)->minutes(); (new ScheduledJobManager)->handle(); Queue::assertPushed( @@ -2547,8 +2547,8 @@ it('dispatches due scheduled directory backups', function () { 'frequency' => '* * * * *', 'enabled' => true, ]); + Cache::forget("scheduled-volume-backup:{$backup->id}"); - $this->travel(1)->minutes(); (new ScheduledJobManager)->handle(); Queue::assertPushed( @@ -2577,7 +2577,6 @@ it('retains scheduled volume backups and archive metadata when the server is mis 'destination_type' => null, ]); - $this->travel(1)->minutes(); (new ScheduledJobManager)->handle(); expect($backup->fresh())->not->toBeNull() @@ -2607,7 +2606,6 @@ it('dispatches pending recovery without starting another volume backup', functio 'stop_recovery_pending' => true, ]); - $this->travel(1)->minutes(); (new ScheduledJobManager)->handle(); Queue::assertPushed( diff --git a/tests/Unit/ScheduledJobManagerLockTest.php b/tests/Unit/ScheduledJobManagerLockTest.php new file mode 100644 index 0000000000..924f6f43cf --- /dev/null +++ b/tests/Unit/ScheduledJobManagerLockTest.php @@ -0,0 +1,63 @@ +middleware(); + + // Assert middleware exists + expect($middleware)->toBeArray() + ->and($middleware)->toHaveCount(1); + + $overlappingMiddleware = $middleware[0]; + + // Assert it's a WithoutOverlapping instance + expect($overlappingMiddleware)->toBeInstanceOf(WithoutOverlapping::class); + + // Use reflection to check private properties + $reflection = new ReflectionClass($overlappingMiddleware); + + // Check expireAfter is set (should be 60 seconds - matches job frequency) + $expiresAfterProperty = $reflection->getProperty('expiresAfter'); + $expiresAfterProperty->setAccessible(true); + $expiresAfter = $expiresAfterProperty->getValue($overlappingMiddleware); + + expect($expiresAfter)->toBe(90) + ->and($expiresAfter)->toBeGreaterThan(0, 'expireAfter must be set to prevent stale locks'); + + // Check releaseAfter is NOT set (we use dontRelease) + $releaseAfterProperty = $reflection->getProperty('releaseAfter'); + $releaseAfterProperty->setAccessible(true); + $releaseAfter = $releaseAfterProperty->getValue($overlappingMiddleware); + + expect($releaseAfter)->toBeNull('releaseAfter should be null when using dontRelease()'); + + // Check the lock key + $keyProperty = $reflection->getProperty('key'); + $keyProperty->setAccessible(true); + $key = $keyProperty->getValue($overlappingMiddleware); + + expect($key)->toBe('scheduled-job-manager'); +}); + +it('prevents stale locks by ensuring expireAfter is always set', function () { + $job = new ScheduledJobManager; + $middleware = $job->middleware(); + + $overlappingMiddleware = $middleware[0]; + $reflection = new ReflectionClass($overlappingMiddleware); + + $expiresAfterProperty = $reflection->getProperty('expiresAfter'); + $expiresAfterProperty->setAccessible(true); + $expiresAfter = $expiresAfterProperty->getValue($overlappingMiddleware); + + // Critical check: expireAfter MUST be set to prevent GitHub issue #4539 + expect($expiresAfter)->not->toBeNull( + 'expireAfter() is required to prevent stale locks (see GitHub #4539)' + ); +}); diff --git a/tests/Unit/ServerPlaceholderIpTest.php b/tests/Unit/ServerPlaceholderIpTest.php index e7c0f25069..94787b4d47 100644 --- a/tests/Unit/ServerPlaceholderIpTest.php +++ b/tests/Unit/ServerPlaceholderIpTest.php @@ -4,10 +4,8 @@ use App\Jobs\CleanupOrphanedPreviewContainersJob; use App\Jobs\ScheduledJobManager; use App\Models\PrivateKey; use App\Models\Server; -use App\Models\ServerSetting; use App\Models\Team; use Illuminate\Foundation\Testing\RefreshDatabase; -use Illuminate\Support\Facades\Queue; use Tests\TestCase; uses(TestCase::class, RefreshDatabase::class); @@ -101,13 +99,10 @@ it('excludes every placeholder address from scheduled Docker cleanup', function ]); } - ServerSetting::query()->update(['docker_cleanup_next_run_at' => null]); - Queue::fake(); + $method = new ReflectionMethod(ScheduledJobManager::class, 'getServersForCleanupQuery'); + $servers = $method->invoke(new ScheduledJobManager)->get(); - (new ScheduledJobManager)->handle(); - - // The dispatcher calculates the next cleanup only for servers that it selects. - expect(ServerSetting::query()->whereNotNull('docker_cleanup_next_run_at')->pluck('server_id')->all())->toBe([$realServer->id]); + expect($servers->modelKeys())->toBe([$realServer->id]); }); it('excludes every placeholder address from orphaned preview cleanup', function () {