starred/coolify
1
0
Fork 0
mirror of https://github.com/coollabsio/coolify.git synced 2026-09-26 01:10:30 -04:00
Code Issues Packages Projects Releases Wiki Activity

feat(scheduling): persist and claim scheduled job occurrences

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