coolify/app/Jobs/ScheduledTaskJob.php

<?php

namespace App\Jobs;

use App\Events\ScheduledTaskDone;
use App\Exceptions\NonReportableException;
use App\Models\Application;
use App\Models\ScheduledTask;
use App\Models\ScheduledTaskExecution;
use App\Models\Server;
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;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Log;

class ScheduledTaskJob implements ShouldBeEncrypted, ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public const MAX_OUTPUT_SIZE_BYTES = 5 * 1024 * 1024;

    /**
     * Lets a task that runs too long fail inside handle() before the worker times out.
     */
    public const WORKER_TIMEOUT_MARGIN_SECONDS = 120;

    /**
     * The number of times the job may be attempted.
     */
    public $tries = 3;

    /**
     * The maximum number of unhandled exceptions to allow before failing.
     */
    public $maxExceptions = 1;

    /**
     * The number of seconds the job can run before timing out.
     */
    public $timeout = 300 + self::WORKER_TIMEOUT_MARGIN_SECONDS;

    public ?Team $team = null;

    public ?Server $server = null;

    public ScheduledTask $task;

    public Application|Service|null $resource = null;

    public ?ScheduledTaskExecution $task_log = null;

    /**
     * Queued with the payload, so failed() finds this run's execution after a worker timeout.
     */
    public ?string $executionUuid = null;

    public string $task_status = 'failed';

    public ?string $task_output = null;

    public array $containers = [];

    public string $server_timezone = 'UTC';

    public function __construct(ScheduledTask $task, public ?string $occurrenceUuid = null)
    {
        $this->onQueue(crons_queue());

        $this->task = $task;
        $this->timeout = ($this->task->timeout ?? 300) + self::WORKER_TIMEOUT_MARGIN_SECONDS;
        $this->executionUuid = new_public_id();
    }

    private function initializeExecutionContext(): void
    {
        $this->task->loadMissing([
            'service.destination.server.settings',
            'application.destination.server.settings',
        ]);

        if ($this->task->service) {
            $this->resource = $this->task->service;
        } elseif ($this->task->application) {
            $this->resource = $this->task->application;
        } else {
            throw new \RuntimeException('ScheduledTaskJob failed: No resource found.');
        }

        $this->team = Team::findOrFail($this->task->team_id);
        $this->server_timezone = $this->getServerTimezone();
        $this->server = $this->resource->destination->server;
    }

    private function getServerTimezone(): string
    {
        if ($this->resource instanceof Application) {
            return $this->resource->destination->server->settings->server_timezone;
        } elseif ($this->resource instanceof Service) {
            return $this->resource->server->settings->server_timezone;
        }

        return 'UTC';
    }

    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;
        }

        $startTime = Carbon::now();
        $failed = false;

        try {
            $this->initializeExecutionContext();

            // A retried job reuses the queued payload and its execution uuid.
            if ($this->executionUuid === null || ScheduledTaskExecution::query()->where('uuid', $this->executionUuid)->exists()) {
                $this->executionUuid = new_public_id();
            }
            $this->task_log = ScheduledTaskExecution::forceCreate([
                'uuid' => $this->executionUuid,
                'scheduled_task_id' => $this->task->id,
                'started_at' => $startTime,
                'retry_count' => $this->attempts() - 1,
            ]);

            if ($this->resource->type() === 'application') {
                $containers = getCurrentApplicationContainerStatus($this->server, $this->resource, 0);
                if ($containers->count() > 0) {
                    $containers->each(function ($container) {
                        $this->containers[] = str_replace('/', '', $container['Names']);
                    });
                }
            } elseif ($this->resource->type() === 'service') {
                $this->resource->applications()->get()->each(function ($application) {
                    if (str(data_get($application, 'status'))->contains('running')) {
                        $this->containers[] = data_get($application, 'name').'-'.data_get($this->resource, 'uuid');
                    }
                });
                $this->resource->databases()->get()->each(function ($database) {
                    if (str(data_get($database, 'status'))->contains('running')) {
                        $this->containers[] = data_get($database, 'name').'-'.data_get($this->resource, 'uuid');
                    }
                });
            }
            if (count($this->containers) == 0) {
                throw new \Exception('ScheduledTaskJob failed: No containers running.');
            }

            if (count($this->containers) > 1 && empty($this->task->container)) {
                throw new \Exception('ScheduledTaskJob failed: More than one container exists but no container name was provided.');
            }

            foreach ($this->containers as $containerName) {
                if (count($this->containers) == 1 || str_starts_with($containerName, $this->task->container.'-'.$this->resource->uuid)) {
                    $cmd = "sh -c '".str_replace("'", "'\''", $this->task->command)."'";
                    $dockerCommand = $this->server->isNonRoot() ? 'sudo docker' : 'docker';
                    $execCommand = "{$dockerCommand} exec ".escapeshellarg($containerName)." {$cmd}";
                    $exec = $this->boundedTaskCommand($execCommand);
                    // Disable SSH multiplexing to prevent race conditions when multiple tasks run concurrently
                    // See: https://github.com/coollabsio/coolify/issues/6736
                    $this->task_output = instant_remote_process([$exec], $this->server, throwError: true, no_sudo: true, timeout: $this->task->timeout ?? 300, disableMultiplexing: true);
                    $this->task_log->update([
                        'status' => 'success',
                        'message' => $this->task_output,
                    ]);

                    $this->team?->notify(new TaskSuccess($this->task, $this->task_output));

                    return;
                }
            }

            // 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',
                    'message' => $this->task_output ?? $e->getMessage(),
                ]);
            }

            // Log the error to the scheduled-errors channel
            Log::channel('scheduled-errors')->error('ScheduledTask execution failed', [
                'job' => 'ScheduledTaskJob',
                'task_id' => $this->task->uuid,
                'task_name' => $this->task->name,
                'server' => $this->server?->name ?? 'unknown',
                'attempt' => $this->attempts(),
                'error' => $e->getMessage(),
            ]);

            // Only notify and throw on final failure

            // 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);
            }

            if ($this->task_log) {
                $finishedAt = Carbon::now();
                $duration = round($startTime->floatDiffInSeconds($finishedAt), 2);

                $this->task_log->update([
                    'finished_at' => $finishedAt->toImmutable(),
                    'duration' => $duration,
                ]);
            }
        }
    }

    private function boundedTaskCommand(string $command): string
    {
        $maxOutputBytes = self::MAX_OUTPUT_SIZE_BYTES;
        $readLimit = $maxOutputBytes + 1;

        return "output_file=\$(mktemp); trap 'rm -f \"\$output_file\"' EXIT; set +e; set -o pipefail; {$command} 2>&1 | { head -c {$readLimit} > \"\$output_file\"; cat > /dev/null; }; exit_code=\${PIPESTATUS[0]}; if [ \"\$(wc -c < \"\$output_file\")\" -gt {$maxOutputBytes} ]; then truncate -s {$maxOutputBytes} \"\$output_file\"; printf '\n\n[... Output truncated at 5MB limit ...]' >> \"\$output_file\"; fi; if [ \"\$exit_code\" -eq 0 ]; then cat \"\$output_file\"; else cat \"\$output_file\" >&2; fi; exit \$exit_code";
    }

    /**
     * Calculate the number of seconds to wait before retrying the job.
     */
    public function backoff(): array
    {
        return [30, 60, 120]; // 30s, 60s, 120s between retries
    }

    /**
     * Handle a job failure.
     */
    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', [
            'job' => 'ScheduledTaskJob',
            'task_id' => $this->task->uuid,
            'task_name' => $this->task->name,
            'server' => $this->server?->name ?? 'unknown',
            'total_attempts' => $this->attempts(),
            'error' => $exception?->getMessage(),
            'trace' => $exception?->getTraceAsString(),
        ]);

        // After a worker timeout only the queued uuid identifies this run's execution.
        $execution = $this->executionUuid === null ? null : ScheduledTaskExecution::query()
            ->where('uuid', $this->executionUuid)
            ->whereNull('finished_at')
            ->first();

        if ($execution) {
            $errorMessage = 'Job permanently failed after '.$this->attempts().' attempts';
            if ($exception) {
                $errorMessage .= ': '.$exception->getMessage();
            }

            $execution->update([
                'status' => 'failed',
                'message' => $errorMessage,
                'error_details' => $exception?->getTraceAsString(),
                'finished_at' => Carbon::now()->toImmutable(),
            ]);
        } else {
            Log::channel('scheduled-errors')->warning('Could not find execution log to update', [
                'execution_uuid' => $this->executionUuid,
                'task_id' => $this->task->uuid,
            ]);
        }

        // Notify team about permanent failure
        $this->team?->notify(new TaskFailed($this->task, $exception?->getMessage() ?? 'Unknown error'));
    }
}