mirror of
https://github.com/coollabsio/coolify.git
synced 2026-11-06 13:10:37 -04:00
fix(scheduler): bound occurrence recovery and fail interrupted claims
- Recover at most 500 open occurrences per scheduler run. - Mark claims older than the worker timeout as failed. - Treat long-pending occurrences like lost queued jobs, and do not run backups late when they are older than 24 hours. - Run fixed-time schedules once when daylight saving time repeats an hour. - Do not run a scheduled task again when Redis hands it out after a killed worker, and do not overwrite a finished execution in failed().
This commit is contained in:
@@ -65,6 +65,7 @@
|
||||
- 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.
|
||||
- A job of a killed worker comes back after `retry_after` (one day) as attempt 2. With `tries > 1` it runs again a day late; queued jobs that must not run late need `tries = 1` or an attempt guard, and `failed()` must not guess its execution row from "the latest" record.
|
||||
|
||||
## Fail closed at public webhook boundaries
|
||||
- Reject missing or blank secrets before signature verification, and return generic errors without logging secrets, signatures, or payloads.
|
||||
|
||||
@@ -136,16 +136,21 @@ class ScheduledJobManager implements ShouldQueue
|
||||
$this->logStart();
|
||||
$jobTypes = $this->type === null ? null : self::TYPES[$this->type];
|
||||
|
||||
try {
|
||||
app(ScheduledJobDeliveryService::class)->recoverStaleEnqueued($jobTypes);
|
||||
} catch (\Throwable $e) {
|
||||
Log::channel('scheduled-errors')->error('Failed to recover stale enqueued occurrences', [
|
||||
'error' => $e->getMessage(),
|
||||
]);
|
||||
$deliveries = app(ScheduledJobDeliveryService::class);
|
||||
foreach ([
|
||||
'Failed to recover stale enqueued occurrences' => fn () => $deliveries->recoverStaleEnqueued($jobTypes),
|
||||
'Failed to recover interrupted occurrences' => fn () => $deliveries->failInterruptedClaims($jobTypes),
|
||||
'Failed to publish pending occurrences' => fn () => $deliveries->publishPending($jobTypes),
|
||||
] as $errorMessage => $recover) {
|
||||
try {
|
||||
$recover();
|
||||
} catch (\Throwable $e) {
|
||||
Log::channel('scheduled-errors')->error($errorMessage, [
|
||||
'error' => $e->getMessage(),
|
||||
]);
|
||||
}
|
||||
}
|
||||
|
||||
app(ScheduledJobDeliveryService::class)->publishPending($jobTypes);
|
||||
|
||||
// Process scheduled backups and tasks together so neither type starves the other.
|
||||
try {
|
||||
if ($this->includes('backups') || $this->includes('tasks')) {
|
||||
|
||||
@@ -107,6 +107,15 @@ class ScheduledTaskJob implements ShouldBeEncrypted, ShouldQueue
|
||||
|
||||
public function handle(): void
|
||||
{
|
||||
// maxExceptions = 1 fails the job on its first error, so a second attempt only happens
|
||||
// when Redis hands the job out again after its worker was killed (retry_after, about a
|
||||
// day later). The command may have run already, so it is not run again late.
|
||||
if ($this->attempts() > 1) {
|
||||
$this->fail(new \RuntimeException('The worker stopped while this task was running. The task is not run again late.'));
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
if ($this->occurrenceUuid && ! app(ScheduledJobDeliveryService::class)->claim($this->occurrenceUuid, $this->job?->uuid() ?? $this->occurrenceUuid)) {
|
||||
return;
|
||||
}
|
||||
@@ -266,10 +275,12 @@ class ScheduledTaskJob implements ShouldBeEncrypted, ShouldQueue
|
||||
$execution = ScheduledTaskExecution::find($this->executionId);
|
||||
}
|
||||
|
||||
// If no stored ID or not found, query for the most recent execution log for this task
|
||||
// If no stored ID or not found, query for the most recent unfinished execution log for this
|
||||
// task. A finished execution belongs to another run and must keep its result.
|
||||
if (! $execution) {
|
||||
$execution = ScheduledTaskExecution::query()
|
||||
->where('scheduled_task_id', $this->task->id)
|
||||
->whereNull('finished_at')
|
||||
->orderBy('created_at', 'desc')
|
||||
->first();
|
||||
}
|
||||
|
||||
@@ -39,6 +39,18 @@ class ScheduledJobDeliveryService
|
||||
*/
|
||||
public const ENQUEUED_STALE_AFTER_MINUTES = 60;
|
||||
|
||||
/**
|
||||
* At most this many open occurrences are published again or recovered per scheduler run, so a
|
||||
* large backlog after an outage is worked off in steps instead of in one long scheduler tick.
|
||||
*/
|
||||
public const RECOVERY_BATCH_SIZE = 500;
|
||||
|
||||
/**
|
||||
* A claimed occurrence that started longer ago than the worker timeout plus this margin has no
|
||||
* live worker: Horizon kills every job at the worker timeout.
|
||||
*/
|
||||
public const INTERRUPTED_CLAIM_MARGIN_MINUTES = 15;
|
||||
|
||||
public const MISSED_OCCURRENCE_NOTIFICATION = 'scheduled-occurrence-missed';
|
||||
|
||||
public const MISSED_OCCURRENCE_NOTIFICATION_INTERVAL_MINUTES = 60;
|
||||
@@ -49,6 +61,12 @@ class ScheduledJobDeliveryService
|
||||
*/
|
||||
private const LATE_RUN_JOB_TYPES = ['database-backup', 'volume-backup'];
|
||||
|
||||
/**
|
||||
* A lost backup occurrence runs late only while it is younger than this. Older work is not
|
||||
* replayed, for example after a long Redis or worker outage.
|
||||
*/
|
||||
public const LATE_RUN_MAX_AGE_HOURS = 24;
|
||||
|
||||
public function recordAndPublish(
|
||||
string $scheduleKey,
|
||||
string $frequency,
|
||||
@@ -66,7 +84,7 @@ class ScheduledJobDeliveryService
|
||||
return false;
|
||||
}
|
||||
|
||||
$delivery = DB::transaction(function () use ($scheduleKey, $scheduledFor, $jobType, $resourceId, $payload): ?ScheduledJobDelivery {
|
||||
$delivery = DB::transaction(function () use ($scheduleKey, $scheduledFor, $cron, $timezone, $jobType, $resourceId, $payload): ?ScheduledJobDelivery {
|
||||
ScheduledJobState::query()->insertOrIgnore([
|
||||
'uuid' => new_public_id(),
|
||||
'schedule_key' => $scheduleKey,
|
||||
@@ -79,7 +97,7 @@ class ScheduledJobDeliveryService
|
||||
->lockForUpdate()
|
||||
->firstOrFail();
|
||||
|
||||
if ($state->last_scheduled_for?->gte($scheduledFor)) {
|
||||
if ($this->isDuplicateOccurrence($state, $scheduledFor, $cron, $timezone)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -116,7 +134,7 @@ class ScheduledJobDeliveryService
|
||||
return false;
|
||||
}
|
||||
|
||||
return DB::transaction(function () use ($scheduleKey, $scheduledFor): bool {
|
||||
return DB::transaction(function () use ($scheduleKey, $scheduledFor, $cron, $timezone): bool {
|
||||
ScheduledJobState::query()->insertOrIgnore([
|
||||
'uuid' => new_public_id(),
|
||||
'schedule_key' => $scheduleKey,
|
||||
@@ -129,7 +147,7 @@ class ScheduledJobDeliveryService
|
||||
->lockForUpdate()
|
||||
->firstOrFail();
|
||||
|
||||
if ($state->last_scheduled_for?->gte($scheduledFor)) {
|
||||
if ($this->isDuplicateOccurrence($state, $scheduledFor, $cron, $timezone)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -140,28 +158,64 @@ class ScheduledJobDeliveryService
|
||||
}
|
||||
|
||||
/**
|
||||
* The same schedule can have the same local time twice when daylight saving time ends. A
|
||||
* schedule with fixed hours (for example `30 2 * * *`) then runs once, like cron does. Interval
|
||||
* schedules (`*` or `/` in the hour field) still run in both hours.
|
||||
*/
|
||||
private function isDuplicateOccurrence(ScheduledJobState $state, Carbon $scheduledFor, CronExpression $cron, string $timezone): bool
|
||||
{
|
||||
$last = $state->last_scheduled_for;
|
||||
if ($last === null) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if ($last->gte($scheduledFor)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (preg_match('/^[\d,\-]+$/', (string) $cron->getExpression(CronExpression::HOUR)) !== 1) {
|
||||
return false;
|
||||
}
|
||||
|
||||
$format = 'Y-m-d H:i';
|
||||
|
||||
return $last->copy()->setTimezone($timezone)->format($format) === $scheduledFor->copy()->setTimezone($timezone)->format($format);
|
||||
}
|
||||
|
||||
/**
|
||||
* Publish occurrences again whose publication did not finish, for example because Redis
|
||||
* rejected the job. An occurrence that stays pending longer than the stale limit follows the
|
||||
* same late-run policy as a lost queued job (see recoverStaleEnqueued()).
|
||||
*
|
||||
* @param array<int, string>|null $jobTypes Only these job types, or all when null.
|
||||
*/
|
||||
public function publishPending(?array $jobTypes = null): void
|
||||
{
|
||||
ScheduledJobDelivery::query()
|
||||
$occurrences = ScheduledJobDelivery::query()
|
||||
->where('status', 'pending')
|
||||
->when($jobTypes !== null, fn ($query) => $query->whereIn('job_type', $jobTypes))
|
||||
->orderBy('id')
|
||||
->chunkById(100, function ($occurrences): void {
|
||||
foreach ($occurrences as $occurrence) {
|
||||
try {
|
||||
$this->logOccurrence($occurrence, 'Scheduled occurrence publishing again', 'warning');
|
||||
$this->publish($occurrence);
|
||||
} catch (\Throwable $e) {
|
||||
Log::channel('scheduled-errors')->error('Failed to publish pending scheduled occurrence', [
|
||||
'occurrence_uuid' => $occurrence->uuid,
|
||||
'schedule_key' => $occurrence->schedule_key,
|
||||
'error' => $e->getMessage(),
|
||||
]);
|
||||
}
|
||||
->limit(self::RECOVERY_BATCH_SIZE)
|
||||
->get();
|
||||
|
||||
foreach ($occurrences as $occurrence) {
|
||||
try {
|
||||
if ($occurrence->created_at->lt(now()->subMinutes(self::ENQUEUED_STALE_AFTER_MINUTES))) {
|
||||
$this->recoverUnstartedOccurrence($occurrence);
|
||||
|
||||
continue;
|
||||
}
|
||||
});
|
||||
|
||||
$this->logOccurrence($occurrence, 'Scheduled occurrence publishing again', 'warning');
|
||||
$this->publish($occurrence);
|
||||
} catch (\Throwable $e) {
|
||||
Log::channel('scheduled-errors')->error('Failed to publish pending scheduled occurrence', [
|
||||
'occurrence_uuid' => $occurrence->uuid,
|
||||
'schedule_key' => $occurrence->schedule_key,
|
||||
'error' => $e->getMessage(),
|
||||
]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -191,27 +245,70 @@ class ScheduledJobDeliveryService
|
||||
*/
|
||||
public function recoverStaleEnqueued(?array $jobTypes = null): void
|
||||
{
|
||||
ScheduledJobDelivery::query()
|
||||
$occurrences = ScheduledJobDelivery::query()
|
||||
->where('status', 'enqueued')
|
||||
->when($jobTypes !== null, fn ($query) => $query->whereIn('job_type', $jobTypes))
|
||||
->where('enqueued_at', '<', now()->subMinutes(self::ENQUEUED_STALE_AFTER_MINUTES))
|
||||
->chunkById(100, function ($occurrences): void {
|
||||
foreach ($occurrences as $occurrence) {
|
||||
try {
|
||||
$this->recoverStaleEnqueuedOccurrence($occurrence);
|
||||
} catch (\Throwable $e) {
|
||||
Log::channel('scheduled-errors')->error('Failed to recover enqueued scheduled occurrence', [
|
||||
'occurrence_uuid' => $occurrence->uuid,
|
||||
'schedule_key' => $occurrence->schedule_key,
|
||||
'error' => $e->getMessage(),
|
||||
]);
|
||||
}
|
||||
}
|
||||
});
|
||||
->orderBy('id')
|
||||
->limit(self::RECOVERY_BATCH_SIZE)
|
||||
->get();
|
||||
|
||||
foreach ($occurrences as $occurrence) {
|
||||
try {
|
||||
$this->recoverUnstartedOccurrence($occurrence);
|
||||
} catch (\Throwable $e) {
|
||||
Log::channel('scheduled-errors')->error('Failed to recover enqueued scheduled occurrence', [
|
||||
'occurrence_uuid' => $occurrence->uuid,
|
||||
'schedule_key' => $occurrence->schedule_key,
|
||||
'error' => $e->getMessage(),
|
||||
]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private function recoverStaleEnqueuedOccurrence(ScheduledJobDelivery $occurrence): void
|
||||
/**
|
||||
* Mark claimed occurrences as failed when no worker can still run them. Horizon kills a job at
|
||||
* the worker timeout, so a claim older than that lost its worker (for example a killed
|
||||
* container). The job is not run again: it can have done part of its remote work already.
|
||||
*
|
||||
* @param array<int, string>|null $jobTypes Only these job types, or all when null.
|
||||
*/
|
||||
public function failInterruptedClaims(?array $jobTypes = null): void
|
||||
{
|
||||
$startedBefore = now()
|
||||
->subSeconds((int) config('horizon.worker_timeout'))
|
||||
->subMinutes(self::INTERRUPTED_CLAIM_MARGIN_MINUTES);
|
||||
|
||||
$occurrences = ScheduledJobDelivery::query()
|
||||
->where('status', 'claimed')
|
||||
->when($jobTypes !== null, fn ($query) => $query->whereIn('job_type', $jobTypes))
|
||||
->where('started_at', '<', $startedBefore)
|
||||
->orderBy('id')
|
||||
->limit(self::RECOVERY_BATCH_SIZE)
|
||||
->get(['id', 'uuid', 'claim_token']);
|
||||
|
||||
foreach ($occurrences as $occurrence) {
|
||||
$failed = ScheduledJobDelivery::query()
|
||||
->whereKey($occurrence->id)
|
||||
->where('status', 'claimed')
|
||||
->where('claim_token', $occurrence->claim_token)
|
||||
->update(['status' => 'failed', 'updated_at' => now()]) === 1;
|
||||
|
||||
if ($failed) {
|
||||
$this->logOccurrence($occurrence->uuid, 'Scheduled occurrence interrupted: its worker stopped after it claimed the occurrence', 'warning');
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Decide what happens to an occurrence that never started: a pending row whose publication kept
|
||||
* failing, or an enqueued row whose queued job was lost. Backups run once late unless a newer
|
||||
* occurrence exists; tasks and Docker cleanups are reported as missed.
|
||||
*/
|
||||
private function recoverUnstartedOccurrence(ScheduledJobDelivery $occurrence): void
|
||||
{
|
||||
$staleBefore = now()->subMinutes(self::ENQUEUED_STALE_AFTER_MINUTES);
|
||||
|
||||
// complete() deletes the row of a finished occurrence, so the schedule state also counts.
|
||||
$hasNewerOccurrence = ScheduledJobState::query()
|
||||
->where('schedule_key', $occurrence->schedule_key)
|
||||
@@ -221,13 +318,18 @@ class ScheduledJobDeliveryService
|
||||
->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;
|
||||
$tooOld = $occurrence->scheduled_for->lt(now()->subHours(self::LATE_RUN_MAX_AGE_HOURS));
|
||||
$runLate = in_array($occurrence->job_type, self::LATE_RUN_JOB_TYPES, true) && ! $hasNewerOccurrence && ! $tooOld;
|
||||
|
||||
// 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))
|
||||
->where('status', $occurrence->status)
|
||||
->when(
|
||||
$occurrence->status === 'enqueued',
|
||||
fn ($query) => $query->where('enqueued_at', '<', $staleBefore),
|
||||
fn ($query) => $query->where('created_at', '<', $staleBefore),
|
||||
)
|
||||
->update(['status' => $runLate ? 'pending' : 'failed', 'updated_at' => now()]) === 1;
|
||||
|
||||
if (! $moved) {
|
||||
@@ -235,9 +337,11 @@ class ScheduledJobDeliveryService
|
||||
}
|
||||
|
||||
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');
|
||||
$this->logOccurrence($occurrence->uuid, match (true) {
|
||||
$hasNewerOccurrence => 'Scheduled occurrence skipped: its queued job was not started and a newer occurrence exists',
|
||||
$tooOld && in_array($occurrence->job_type, self::LATE_RUN_JOB_TYPES, true) => 'Scheduled occurrence missed: its queued job was not started and it is too old to run late',
|
||||
default => 'Scheduled occurrence missed: its queued job was not started',
|
||||
}, 'warning', ['previous_status' => $occurrence->status]);
|
||||
$this->reportMissedOccurrence($occurrence);
|
||||
|
||||
return;
|
||||
|
||||
@@ -456,14 +456,14 @@ it('warns when an occurrence is enqueued while the previous one is still running
|
||||
config(['constants.coolify.self_hosted' => true]);
|
||||
Queue::fake();
|
||||
$logPath = captureScheduledJobManagerTestLog('scheduled');
|
||||
$backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => 'daily']);
|
||||
$backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => 'hourly']);
|
||||
|
||||
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'));
|
||||
Carbon::setTestNow(Carbon::create(2026, 9, 16, 1, 5, 0, 'UTC'));
|
||||
(new ScheduledJobManager)->handle();
|
||||
|
||||
$log = file_get_contents($logPath);
|
||||
@@ -887,6 +887,172 @@ it('rejects an unknown schedule type', function () {
|
||||
Queue::assertNothingPushed();
|
||||
});
|
||||
|
||||
it('reports a task occurrence as missed when its publication kept failing for an hour', function () {
|
||||
config(['constants.coolify.self_hosted' => true]);
|
||||
Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC'));
|
||||
Queue::fake();
|
||||
Notification::fake();
|
||||
$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' => 'pending',
|
||||
]);
|
||||
$occurrence->forceFill(['created_at' => now()->subMinutes(ScheduledJobDeliveryService::ENQUEUED_STALE_AFTER_MINUTES + 1)])->save();
|
||||
|
||||
app(ScheduledJobDeliveryService::class)->publishPending();
|
||||
|
||||
Queue::assertNotPushed(ScheduledTaskJob::class);
|
||||
expect($occurrence->fresh()->status)->toBe('failed')
|
||||
->and(ScheduledTaskExecution::query()->where('scheduled_task_id', $task->id)->value('status'))->toBe('failed');
|
||||
});
|
||||
|
||||
it('publishes a backup occurrence once late when its publication kept failing and no newer occurrence exists', 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' => 'pending',
|
||||
]);
|
||||
$occurrence->forceFill(['created_at' => now()->subHours(2)])->save();
|
||||
|
||||
app(ScheduledJobDeliveryService::class)->publishPending();
|
||||
|
||||
Queue::assertPushed(DatabaseBackupJob::class, fn (DatabaseBackupJob $job) => $job->occurrenceUuid === $occurrence->uuid);
|
||||
expect($occurrence->fresh()->status)->toBe('enqueued');
|
||||
});
|
||||
|
||||
it('does not run a lost backup occurrence late when it is older than the late-run limit', 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' => 'weekly']);
|
||||
$occurrence = createStaleEnqueuedOccurrence(
|
||||
"scheduled-backup:{$backup->id}",
|
||||
'database-backup',
|
||||
$backup->id,
|
||||
now()->subHours(ScheduledJobDeliveryService::LATE_RUN_MAX_AGE_HOURS + 1),
|
||||
minutesAgo: (ScheduledJobDeliveryService::LATE_RUN_MAX_AGE_HOURS + 1) * 60,
|
||||
);
|
||||
|
||||
app(ScheduledJobDeliveryService::class)->recoverStaleEnqueued();
|
||||
|
||||
$log = file_get_contents($logPath);
|
||||
@unlink($logPath);
|
||||
|
||||
Queue::assertNotPushed(DatabaseBackupJob::class);
|
||||
expect($occurrence->fresh()->status)->toBe('failed')
|
||||
->and($log)->toContain('too old to run late');
|
||||
});
|
||||
|
||||
it('fails claimed occurrences whose worker can no longer be running, and keeps recent claims', function () {
|
||||
Carbon::setTestNow(Carbon::create(2026, 9, 17, 12, 0, 0, 'UTC'));
|
||||
$deadAfterSeconds = (int) config('horizon.worker_timeout') + ScheduledJobDeliveryService::INTERRUPTED_CLAIM_MARGIN_MINUTES * 60;
|
||||
$create = fn (string $key, int $startedSecondsAgo) => ScheduledJobDelivery::create([
|
||||
'schedule_key' => $key,
|
||||
'scheduled_for' => now()->subSeconds($startedSecondsAgo),
|
||||
'job_type' => 'scheduled-task',
|
||||
'resource_id' => 1,
|
||||
'status' => 'claimed',
|
||||
'claim_token' => "token-{$key}",
|
||||
'started_at' => now()->subSeconds($startedSecondsAgo),
|
||||
]);
|
||||
$interrupted = $create('scheduled-task:interrupted', $deadAfterSeconds + 60);
|
||||
$running = $create('scheduled-task:running', $deadAfterSeconds - 60);
|
||||
|
||||
$service = app(ScheduledJobDeliveryService::class);
|
||||
$service->failInterruptedClaims();
|
||||
|
||||
expect($interrupted->fresh()->status)->toBe('failed')
|
||||
->and($running->fresh()->status)->toBe('claimed')
|
||||
// The job copy that Redis hands out again later cannot claim the failed occurrence.
|
||||
->and($service->claim($interrupted->uuid, 'token-scheduled-task:interrupted'))->toBeFalse();
|
||||
});
|
||||
|
||||
it('does not run a scheduled task again when Redis hands it out after its worker was killed', function () {
|
||||
$task = createScheduledApplicationTask(createScheduledTaskApplication());
|
||||
$occurrence = ScheduledJobDelivery::create([
|
||||
'schedule_key' => "scheduled-task:{$task->id}",
|
||||
'scheduled_for' => now()->subDay(),
|
||||
'job_type' => 'scheduled-task',
|
||||
'resource_id' => $task->id,
|
||||
'status' => 'claimed',
|
||||
'claim_token' => 'killed-worker',
|
||||
'started_at' => now()->subDay(),
|
||||
]);
|
||||
|
||||
$job = (new ScheduledTaskJob($task, $occurrence->uuid))->withFakeQueueInteractions();
|
||||
$job->job->attempts = 2;
|
||||
$job->handle();
|
||||
|
||||
$job->assertFailed();
|
||||
expect(ScheduledTaskExecution::query()->where('scheduled_task_id', $task->id)->exists())->toBeFalse();
|
||||
});
|
||||
|
||||
it('marks only the unfinished execution as failed when a scheduled task job fails without its execution id', function () {
|
||||
InstanceSettings::forceCreate(['id' => 0]);
|
||||
Notification::fake();
|
||||
$task = createScheduledApplicationTask(createScheduledTaskApplication());
|
||||
$interrupted = ScheduledTaskExecution::create(['scheduled_task_id' => $task->id, 'status' => 'running', 'started_at' => now()->subHours(2)]);
|
||||
$interrupted->forceFill(['created_at' => now()->subHours(2)])->save();
|
||||
$later = ScheduledTaskExecution::create(['scheduled_task_id' => $task->id, 'status' => 'success', 'started_at' => now()->subHour(), 'finished_at' => now()->subMinutes(59)]);
|
||||
|
||||
(new ScheduledTaskJob($task))->failed(new RuntimeException('worker killed'));
|
||||
|
||||
expect($interrupted->fresh()->status)->toBe('failed')
|
||||
->and($later->fresh()->status)->toBe('success');
|
||||
});
|
||||
|
||||
it('runs a fixed-time schedule once when daylight saving time repeats its local time', function () {
|
||||
$service = app(ScheduledJobDeliveryService::class);
|
||||
$record = fn (string $frequency, string $utc) => $service->recordSkipped("dst:{$frequency}", $frequency, 'Europe/Berlin', Carbon::parse($utc, 'UTC'));
|
||||
|
||||
// 2026-10-25: Berlin goes from 03:00 CEST back to 02:00 CET, so 02:30 happens at 00:30Z and 01:30Z.
|
||||
expect($record('30 2 * * *', '2026-10-25 00:31:00'))->toBeTrue()
|
||||
->and($record('30 2 * * *', '2026-10-25 01:31:00'))->toBeFalse()
|
||||
->and($record('30 * * * *', '2026-10-25 00:31:00'))->toBeTrue()
|
||||
->and($record('30 * * * *', '2026-10-25 01:31:00'))->toBeTrue();
|
||||
});
|
||||
|
||||
it('publishes one backup when daylight saving time repeats the local backup time', function () {
|
||||
config(['constants.coolify.self_hosted' => true]);
|
||||
Queue::fake();
|
||||
$backup = createScheduledDatabaseBackup(createScheduledBackupDatabase(), ['frequency' => '30 2 * * *']);
|
||||
$service = app(ScheduledJobDeliveryService::class);
|
||||
|
||||
foreach (['2026-10-25 00:31:00', '2026-10-25 01:31:00'] as $utc) {
|
||||
$service->recordAndPublish("scheduled-backup:{$backup->id}", '30 2 * * *', 'Europe/Berlin', 'database-backup', $backup->id, executionTime: Carbon::parse($utc, 'UTC'));
|
||||
}
|
||||
|
||||
Queue::assertPushed(DatabaseBackupJob::class, 1);
|
||||
});
|
||||
|
||||
it('does not run a fixed time that does not exist when daylight saving time starts', function () {
|
||||
$service = app(ScheduledJobDeliveryService::class);
|
||||
|
||||
// 2026-03-29: Berlin skips from 02:00 CET to 03:00 CEST, so 02:30 does not exist that day.
|
||||
expect($service->recordSkipped('dst:spring', '30 2 * * *', 'Europe/Berlin', Carbon::parse('2026-03-29 01:01:00', 'UTC')))->toBeFalse()
|
||||
->and($service->recordSkipped('dst:spring', '30 2 * * *', 'Europe/Berlin', Carbon::parse('2026-03-30 00:31:00', 'UTC')))->toBeTrue();
|
||||
});
|
||||
|
||||
it('records a midnight schedule in the server timezone, not in UTC', function () {
|
||||
$service = app(ScheduledJobDeliveryService::class);
|
||||
|
||||
// Midnight in New York (EDT) is 04:00Z; 00:05Z is 20:05 the day before in New York.
|
||||
expect($service->recordSkipped('midnight', '0 0 * * *', 'America/New_York', Carbon::parse('2026-10-03 00:05:00', 'UTC')))->toBeFalse()
|
||||
->and($service->recordSkipped('midnight', '0 0 * * *', 'America/New_York', Carbon::parse('2026-10-03 04:05:00', 'UTC')))->toBeTrue()
|
||||
->and($service->recordSkipped('midnight', '0 0 * * *', 'America/New_York', Carbon::parse('2026-10-03 04:09:00', 'UTC')))->toBeFalse()
|
||||
->and(ScheduledJobState::query()->where('schedule_key', 'midnight')->value('last_scheduled_for')->toDateTimeString())->toBe('2026-10-03 04:00:00');
|
||||
});
|
||||
|
||||
function createStaleEnqueuedOccurrence(string $scheduleKey, string $jobType, int $resourceId, Carbon $scheduledFor, int $minutesAgo = ScheduledJobDeliveryService::ENQUEUED_STALE_AFTER_MINUTES + 1): ScheduledJobDelivery
|
||||
{
|
||||
return ScheduledJobDelivery::create([
|
||||
|
||||
Reference in New Issue
Block a user