app/Jobs/ScheduledTaskJob.php
Rama main · réplica de lectura del repositorio original.
<?php
namespace App\Jobs;
use App\Events\ScheduledTaskDone;
use App\Exceptions\NonReportableException;
use App\Models\Application;
use App\Models\EnvironmentVariable;
use App\Models\ScheduledTask;
use App\Models\ScheduledTaskExecution;
use App\Models\Server;
use App\Models\Service;
use App\Models\Team;
use App\Notifications\ScheduledJobRecovered;
use App\Notifications\ScheduledTask\TaskFailed;
use App\Notifications\ScheduledTask\TaskSuccess;
use App\Services\ScheduledJobDeliveryService;
use App\Services\Security\SensitiveDataRedactor;
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;
/** @var array{known: array<int, string>, locked: array<int, string>}|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');
}
});
}
// A stopped resource is not a task failure: it sends no notification and does not count toward auto-disable.
if (count($this->containers) == 0) {
$this->task_log->update([
'status' => 'skipped',
'message' => 'Skipped: the resource has no running container.',
]);
return;
}
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)) {
$dockerCommand = $this->server->isNonRoot() ? 'sudo docker' : 'docker';
$execCommand = "{$dockerCommand} exec ".escapeshellarg($containerName).' '.$this->task->containerCommand();
$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));
if ($this->task->recordSuccess()) {
$this->team?->notify(new ScheduledJobRecovered(
failureEvent: 'scheduled_task_failure',
title: "Scheduled task ({$this->task->name}) works again.",
webhookEvent: 'task_recovered',
url: $this->task->url(),
webhookData: ['task_name' => $this->task->name, 'task_uuid' => $this->task->uuid],
));
}
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,
]);
}
try {
$disabled = $this->task->recordFailure();
} catch (\Throwable $e) {
Log::channel('scheduled-errors')->warning('Could not record the scheduled task failure', [
'task_id' => $this->task->uuid,
'error' => $e->getMessage(),
]);
$disabled = false;
}
$this->team?->notify(new TaskFailed($this->task, $safeErrorMessage, disabled: $disabled));
}
}