CluPilotCloud/app/Provisioning/RunRunner.php

<?php

namespace App\Provisioning;

use App\Models\ProvisioningRun;
use App\Provisioning\Contracts\ProvisioningStep;
use App\Provisioning\Contracts\ProvisioningSubject;
use App\Provisioning\Events\StepAdvanced;
use App\Provisioning\Jobs\AdvanceRunJob;
use Illuminate\Support\Facades\Cache;
use Throwable;

/**
 * Executes exactly one step of a run under a per-run lock, records the event,
 * applies the StepResult, and (on advance) queues the next execution.
 */
class RunRunner
{
    public function __construct(private PipelineRegistry $registry) {}

    private ?string $pendingContinuation = null;

    public function advance(ProvisioningRun $run): void
    {
        $this->pendingContinuation = null;

        // Lock must outlive the longest step so a duplicate job (dispatched by
        // the minutely tick while this one still runs) can't execute concurrently.
        $lock = Cache::lock('run:'.$run->uuid, 2100);

        if (! $lock->get()) {
            return; // another worker already holds this run
        }

        try {
            $this->runLocked($run);
        } finally {
            $lock->release();
        }

        // Dispatch the continuation only AFTER releasing the lock, so a second
        // worker can't consume it and then bail on the still-held lock.
        if ($this->pendingContinuation !== null) {
            AdvanceRunJob::dispatch($this->pendingContinuation);
            $this->pendingContinuation = null;
        }
    }

    private function runLocked(ProvisioningRun $run): void
    {
        $run->refresh();

        if (! in_array($run->status, [
            ProvisioningRun::STATUS_PENDING,
            ProvisioningRun::STATUS_RUNNING,
            ProvisioningRun::STATUS_WAITING,
        ], true)) {
            return;
        }

        // Honour backoff: a stale/duplicate job must not run a waiting run early.
        if ($run->status === ProvisioningRun::STATUS_WAITING
            && $run->next_attempt_at !== null
            && $run->next_attempt_at->isFuture()) {
            return;
        }

        // started_at marks when the current step began (drives the timeout).
        if ($run->started_at === null || $run->status === ProvisioningRun::STATUS_PENDING) {
            $run->started_at ??= now();
        }
        $run->status = ProvisioningRun::STATUS_RUNNING;
        $run->save();

        // A missing/renamed pipeline or step (e.g. after a deploy) is fatal, not
        // a transient error — fail terminally instead of looping forever.
        try {
            $step = $this->registry->resolve($run->pipeline, $run->current_step);
        } catch (Throwable $e) {
            $this->failRun($run, 'pipeline_resolution', 'cannot resolve step: '.$e->getMessage());

            return;
        }

        $timedOut = $run->started_at !== null
            && $run->started_at->copy()->addSeconds($step->maxDuration())->isPast();

        try {
            $result = $timedOut
                ? StepResult::retry(0, 'step timed out after '.$step->maxDuration().'s')
                : $step->execute($run);
        } catch (Throwable $e) {
            $result = StepResult::retry($this->backoff($run->attempt), 'exception: '.$e->getMessage());
        }

        $this->apply($run, $step, $result);
    }

    private function apply(ProvisioningRun $run, ProvisioningStep $step, StepResult $result): void
    {
        match ($result->type) {
            StepResult::ADVANCE => $this->onAdvance($run, $step),
            StepResult::RETRY => $this->onRetry($run, $step, $result),
            StepResult::POLL => $this->onPoll($run, $step, $result),
            StepResult::FAIL => $this->failRun($run, $step->key(), $result->reason),
        };
    }

    private function onAdvance(ProvisioningRun $run, ProvisioningStep $step): void
    {
        $isLast = $run->current_step >= $this->registry->count($run->pipeline) - 1;

        if ($isLast) {
            $run->status = ProvisioningRun::STATUS_COMPLETED;
            $run->finished_at = now();
            $run->attempt = 0;
            $run->save();
            $this->record($run, $step->key(), 'advanced', null);

            return;
        }

        $run->current_step += 1;
        $run->attempt = 0;
        $run->started_at = now();
        $run->status = ProvisioningRun::STATUS_RUNNING;
        $run->save();
        $this->record($run, $step->key(), 'advanced', null);

        $this->pendingContinuation = $run->uuid; // dispatched after the lock releases
    }

    private function onRetry(ProvisioningRun $run, ProvisioningStep $step, StepResult $result): void
    {
        $run->attempt += 1;

        if ($run->attempt >= $run->max_attempts) {
            $this->failRun($run, $step->key(), $result->reason);

            return;
        }

        $run->status = ProvisioningRun::STATUS_WAITING;
        $run->next_attempt_at = now()->addSeconds($result->afterSeconds);
        $run->started_at = now(); // fresh per-attempt timer, else a timed-out step re-times-out
        $run->save();
        $this->record($run, $step->key(), 'retry', $result->reason);
    }

    /** Poll: wait and re-run without consuming the retry budget (step owns its deadline). */
    private function onPoll(ProvisioningRun $run, ProvisioningStep $step, StepResult $result): void
    {
        $run->status = ProvisioningRun::STATUS_WAITING;
        $run->next_attempt_at = now()->addSeconds($result->afterSeconds);
        $run->started_at = now();
        $run->save();
        $this->record($run, $step->key(), 'info', $result->reason);
    }

    private function failRun(ProvisioningRun $run, string $stepKey, string $reason): void
    {
        $run->status = ProvisioningRun::STATUS_FAILED;
        $run->error = $reason;
        $run->save();
        $this->record($run, $stepKey, 'failed', $reason);

        // Let the subject react (e.g. a Host moves to the 'error' status).
        $subject = $run->subject;
        if ($subject instanceof ProvisioningSubject) {
            $subject->onProvisioningFailed();
        }
    }

    private function record(ProvisioningRun $run, string $stepKey, string $outcome, ?string $message): void
    {
        $run->events()->create([
            'step' => $stepKey,
            'attempt' => $run->attempt,
            'outcome' => $outcome,
            'message' => $message,
        ]);

        StepAdvanced::dispatch($run->uuid, $stepKey, $outcome, $run->current_step, $run->status);
    }

    private function backoff(int $attempt): int
    {
        return (int) min(300, 15 * (2 ** $attempt));
    }
}