, locked: array}|null */ private ?array $logRedactionSecrets = 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(); } /** * The lock outlives the worker timeout, so a run killed at the timeout still blocks a new run * while its remote command can still be running. */ public function middleware(): array { $expireAfter = ($this->task->timeout ?? 300) + 300; return [ScheduledJobDeliveryService::withoutOverlapping('scheduled-task-'.$this->task->id, $this->occurrenceUuid)->expireAfter($expireAfter)->dontRelease()]; } 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') { // `docker ps -a` also lists stopped containers, for example the old container of a rolling update. getCurrentApplicationContainerStatus($this->server, $this->resource, 0) ->filter(fn ($container) => data_get($container, 'State') === 'running') ->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 || $containerName === $this->task->container || 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 = $this->redact(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; $safeErrorMessage = $this->redact($e->getMessage()); if ($this->task_log) { $this->task_log->update([ 'status' => 'failed', 'message' => $this->task_output ?? $safeErrorMessage, ]); } // 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' => $safeErrorMessage, ]); // Only notify and throw on final failure // Re-throw to trigger Laravel's retry mechanism with backoff throw $this->redactedException($e, $safeErrorMessage); } 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"; } private function redact(string $value): string { try { $this->logRedactionSecrets ??= EnvironmentVariable::logRedactionSecretsFor($this->resource); } catch (\Throwable) { return REDACTED; } return resolve(SensitiveDataRedactor::class)->redactText($value, $this->logRedactionSecrets['known'], $this->logRedactionSecrets['locked']); } /** * Keep the original exception when redaction did not change its message. * Otherwise, drop the original (and its trace) so the secret is not stored in failed_jobs. */ private function redactedException(\Throwable $exception, string $safeMessage): \Throwable { if ($safeMessage === $exception->getMessage()) { return $exception; } return $exception instanceof NonReportableException ? new NonReportableException($safeMessage, (int) $exception->getCode()) : new \RuntimeException($safeMessage, (int) $exception->getCode()); } /** * 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); $this->resource ??= $this->task->application ?? $this->task->service; $safeErrorMessage = $this->redact($exception?->getMessage() ?? 'Unknown error'); $safeTrace = $this->redact($exception?->getTraceAsString() ?? ''); 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' => $safeErrorMessage, 'trace' => $safeTrace, ]); // 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 .= ': '.$safeErrorMessage; } $execution->update([ 'status' => 'failed', 'message' => $errorMessage, 'error_details' => $safeTrace, '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, $safeErrorMessage)); } }