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. started_at is NOT * reset here, so a step's own deadline (measured from started_at) actually * accumulates across polls instead of resetting every time. */ private function onPoll(ProvisioningRun $run, ProvisioningStep $step, StepResult $result): void { $run->status = ProvisioningRun::STATUS_WAITING; $run->next_attempt_at = now()->addSeconds($result->afterSeconds); $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)); } }