feat(engine): orchestrator core (state machine + tick + lock)

StepResult, ProvisioningStep contract, PipelineRegistry, RunRunner (per-run
lock, advance/retry/fail, backoff, timeout, append-only events + Reverb
StepAdvanced), AdvanceRunJob (provisioning queue), minutely Tick, admin.runs
channel. 21 tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
feat/portal-design
nexxo 2026-07-25 09:54:52 +02:00
parent 11cdcdd4da
commit 4f724d57c9
19 changed files with 718 additions and 1 deletions

View File

@ -0,0 +1,25 @@
<?php
namespace App\Console;
use App\Models\ProvisioningRun;
use App\Provisioning\Jobs\AdvanceRunJob;
/**
* Scheduler tick (every minute): dispatch an advance job for every run that is
* running/waiting and due. Immediate dispatch on advance handles the fast path;
* this catches waiting/retrying runs and anything a crashed worker left behind.
*/
class TickProvisioning
{
public function __invoke(): void
{
ProvisioningRun::query()
->whereIn('status', [ProvisioningRun::STATUS_RUNNING, ProvisioningRun::STATUS_WAITING])
->where(function ($q) {
$q->whereNull('next_attempt_at')->orWhere('next_attempt_at', '<=', now());
})
->get()
->each(fn (ProvisioningRun $run) => AdvanceRunJob::dispatch($run->uuid));
}
}

View File

@ -2,6 +2,13 @@
namespace App\Providers;
use App\Provisioning\PipelineRegistry;
use App\Services\Proxmox\HttpProxmoxClient;
use App\Services\Proxmox\ProxmoxClient;
use App\Services\Ssh\PhpseclibRemoteShell;
use App\Services\Ssh\RemoteShell;
use App\Services\Wireguard\LocalWireguardHub;
use App\Services\Wireguard\WireguardHub;
use Illuminate\Support\ServiceProvider;
class AppServiceProvider extends ServiceProvider
@ -11,7 +18,14 @@ class AppServiceProvider extends ServiceProvider
*/
public function register(): void
{
//
$this->app->singleton(PipelineRegistry::class, fn () => new PipelineRegistry(
config('provisioning.pipelines', []),
));
// Real I/O implementations; tests swap in fakes via app()->instance().
$this->app->bind(RemoteShell::class, PhpseclibRemoteShell::class);
$this->app->bind(WireguardHub::class, LocalWireguardHub::class);
$this->app->bind(ProxmoxClient::class, HttpProxmoxClient::class);
}
/**

View File

@ -0,0 +1,24 @@
<?php
namespace App\Provisioning\Contracts;
use App\Models\ProvisioningRun;
use App\Provisioning\StepResult;
/**
* One idempotent unit of a provisioning pipeline. Steps never call each other;
* they depend only on the run's state and the service interfaces they need.
*/
interface ProvisioningStep
{
/** Stable snake_case identifier used in events and idempotency checks. */
public function key(): string;
/** i18n key for the display label, e.g. hosts.step.<key>. */
public function label(): string;
/** Seconds this step may run before the runner treats it as timed out. */
public function maxDuration(): int;
public function execute(ProvisioningRun $run): StepResult;
}

View File

@ -0,0 +1,36 @@
<?php
namespace App\Provisioning\Events;
use Illuminate\Broadcasting\InteractsWithSockets;
use Illuminate\Broadcasting\PrivateChannel;
use Illuminate\Contracts\Broadcasting\ShouldBroadcast;
use Illuminate\Foundation\Events\Dispatchable;
use Illuminate\Queue\SerializesModels;
/**
* Broadcast on every recorded run event so admin/customer views update live.
*/
class StepAdvanced implements ShouldBroadcast
{
use Dispatchable, InteractsWithSockets, SerializesModels;
public function __construct(
public string $runUuid,
public string $step,
public string $outcome,
public int $currentStep,
public string $status,
) {}
/** @return array<int, PrivateChannel> */
public function broadcastOn(): array
{
return [new PrivateChannel('admin.runs')];
}
public function broadcastAs(): string
{
return 'StepAdvanced';
}
}

View File

@ -0,0 +1,30 @@
<?php
namespace App\Provisioning\Jobs;
use App\Models\ProvisioningRun;
use App\Provisioning\RunRunner;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
class AdvanceRunJob implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
public function __construct(public string $runUuid)
{
$this->onQueue('provisioning');
}
public function handle(RunRunner $runner): void
{
$run = ProvisioningRun::query()->where('uuid', $this->runUuid)->first();
if ($run !== null) {
$runner->advance($run);
}
}
}

View File

@ -0,0 +1,41 @@
<?php
namespace App\Provisioning;
use App\Provisioning\Contracts\ProvisioningStep;
use InvalidArgumentException;
/**
* Maps a pipeline name (host | customer) to its ordered list of step classes.
*/
class PipelineRegistry
{
/** @param array<string, array<int, class-string<ProvisioningStep>>> $pipelines */
public function __construct(private array $pipelines) {}
/** @return array<int, class-string<ProvisioningStep>> */
public function stepsFor(string $pipeline): array
{
if (! isset($this->pipelines[$pipeline])) {
throw new InvalidArgumentException("Unknown pipeline [{$pipeline}].");
}
return $this->pipelines[$pipeline];
}
public function count(string $pipeline): int
{
return count($this->stepsFor($pipeline));
}
public function resolve(string $pipeline, int $index): ProvisioningStep
{
$steps = $this->stepsFor($pipeline);
if (! isset($steps[$index])) {
throw new InvalidArgumentException("No step at index {$index} for pipeline [{$pipeline}].");
}
return app($steps[$index]);
}
}

View File

@ -0,0 +1,146 @@
<?php
namespace App\Provisioning;
use App\Models\ProvisioningRun;
use App\Provisioning\Contracts\ProvisioningStep;
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) {}
public function advance(ProvisioningRun $run): void
{
$lock = Cache::lock('run:'.$run->uuid, 120);
if (! $lock->get()) {
return; // another worker already holds this run
}
try {
$this->runLocked($run);
} finally {
$lock->release();
}
}
private function runLocked(ProvisioningRun $run): void
{
$run->refresh();
if (! in_array($run->status, [
ProvisioningRun::STATUS_PENDING,
ProvisioningRun::STATUS_RUNNING,
ProvisioningRun::STATUS_WAITING,
], true)) {
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();
$step = $this->registry->resolve($run->pipeline, $run->current_step);
$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::FAIL => $this->onFail($run, $step, $result),
};
}
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, '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, 'advanced', null);
AdvanceRunJob::dispatch($run->uuid);
}
private function onRetry(ProvisioningRun $run, ProvisioningStep $step, StepResult $result): void
{
$run->attempt += 1;
if ($run->attempt >= $run->max_attempts) {
$run->status = ProvisioningRun::STATUS_FAILED;
$run->error = $result->reason;
$run->save();
$this->record($run, $step, 'failed', $result->reason);
return;
}
$run->status = ProvisioningRun::STATUS_WAITING;
$run->next_attempt_at = now()->addSeconds($result->afterSeconds);
$run->save();
$this->record($run, $step, 'retry', $result->reason);
}
private function onFail(ProvisioningRun $run, ProvisioningStep $step, StepResult $result): void
{
$run->status = ProvisioningRun::STATUS_FAILED;
$run->error = $result->reason;
$run->save();
$this->record($run, $step, 'failed', $result->reason);
}
private function record(ProvisioningRun $run, ProvisioningStep $step, string $outcome, ?string $message): void
{
$run->events()->create([
'step' => $step->key(),
'attempt' => $run->attempt,
'outcome' => $outcome,
'message' => $message,
]);
StepAdvanced::dispatch($run->uuid, $step->key(), $outcome, $run->current_step, $run->status);
}
private function backoff(int $attempt): int
{
return (int) min(300, 15 * (2 ** $attempt));
}
}

View File

@ -0,0 +1,35 @@
<?php
namespace App\Provisioning;
/**
* The outcome a step returns to the runner: advance to the next step, retry
* this step after a delay, or fail the run.
*/
final class StepResult
{
public const ADVANCE = 'advance';
public const RETRY = 'retry';
public const FAIL = 'fail';
private function __construct(
public readonly string $type,
public readonly int $afterSeconds = 0,
public readonly string $reason = '',
) {}
public static function advance(): self
{
return new self(self::ADVANCE);
}
public static function retry(int $afterSeconds, string $reason): self
{
return new self(self::RETRY, $afterSeconds, $reason);
}
public static function fail(string $reason): self
{
return new self(self::FAIL, 0, $reason);
}
}

47
config/provisioning.php Normal file
View File

@ -0,0 +1,47 @@
<?php
use App\Provisioning\Steps\Host;
return [
/*
| Ordered step classes per pipeline. ::class is textual listing steps that
| are built incrementally does not autoload them until the pipeline runs.
*/
'pipelines' => [
'host' => [
Host\ValidateHostInput::class,
Host\EstablishSshTrust::class,
Host\PrepareBaseSystem::class,
Host\ConfigureWireguard::class,
Host\InstallProxmoxVe::class,
Host\RebootIntoPveKernel::class,
Host\ConfigureProxmox::class,
Host\CreateAutomationToken::class,
Host\VerifyProxmoxApi::class,
Host\RegisterCapacity::class,
Host\CompleteHostOnboarding::class,
],
],
// CluPilot VM acts as the WireGuard hub; hosts join it during onboarding.
'wireguard' => [
'subnet' => env('CLUPILOT_WG_SUBNET', '10.66.0.0/24'),
'endpoint' => env('CLUPILOT_WG_ENDPOINT', ''), // host:port reachable by peers
'hub_public_key' => env('CLUPILOT_WG_HUB_PUBKEY', ''),
'config_path' => env('CLUPILOT_WG_CONFIG_PATH', '/etc/wireguard/wg0.conf'),
],
// SSH identity CluPilot deploys to each host after first password login.
'ssh' => [
'public_key' => env('CLUPILOT_SSH_PUBLIC_KEY', ''),
'private_key' => env('CLUPILOT_SSH_PRIVATE_KEY', ''),
],
// Proxmox automation role/user created on each host.
'proxmox' => [
'role_id' => 'CluPilotAutomation',
'role_privs' => 'VM.Allocate,VM.Clone,VM.Config.Disk,VM.Config.CPU,VM.Config.Memory,VM.Config.Network,VM.Config.Options,VM.Config.Cloudinit,VM.PowerMgmt,VM.Monitor,VM.Audit,VM.GuestAgent.Audit,VM.GuestAgent.Unrestricted,Datastore.AllocateSpace,Datastore.Audit,Sys.Audit',
'user' => 'automation@pve',
'token_name' => 'clupilot',
],
];

View File

@ -5,3 +5,6 @@ use Illuminate\Support\Facades\Broadcast;
Broadcast::channel('App.Models.User.{id}', function ($user, $id) {
return (int) $user->id === (int) $id;
});
// Operator console live provisioning feed — admins only.
Broadcast::channel('admin.runs', fn ($user) => (bool) $user->is_admin);

View File

@ -1,8 +1,16 @@
<?php
use App\Console\TickProvisioning;
use Illuminate\Foundation\Inspiring;
use Illuminate\Support\Facades\Artisan;
use Illuminate\Support\Facades\Schedule;
Artisan::command('inspire', function () {
$this->comment(Inspiring::quote());
})->purpose('Display an inspiring quote');
// Advance every due provisioning run once a minute (runs in the scheduler service).
Schedule::call(fn () => app(TickProvisioning::class)())
->everyMinute()
->name('provisioning-tick')
->withoutOverlapping();

View File

@ -0,0 +1,21 @@
<?php
use App\Provisioning\PipelineRegistry;
use Tests\Support\Steps\FakeAdvanceStep;
use Tests\Support\Steps\FakeRetryStep;
it('resolves step instances by index', function () {
$registry = new PipelineRegistry(['host' => [FakeAdvanceStep::class, FakeRetryStep::class]]);
expect($registry->count('host'))->toBe(2)
->and($registry->resolve('host', 0))->toBeInstanceOf(FakeAdvanceStep::class)
->and($registry->resolve('host', 1))->toBeInstanceOf(FakeRetryStep::class);
});
it('throws for an unknown pipeline', function () {
(new PipelineRegistry([]))->stepsFor('nope');
})->throws(InvalidArgumentException::class);
it('throws for an out-of-range step index', function () {
(new PipelineRegistry(['host' => [FakeAdvanceStep::class]]))->resolve('host', 5);
})->throws(InvalidArgumentException::class);

View File

@ -0,0 +1,132 @@
<?php
use App\Models\ProvisioningRun;
use App\Provisioning\Events\StepAdvanced;
use App\Provisioning\Jobs\AdvanceRunJob;
use App\Provisioning\PipelineRegistry;
use App\Provisioning\RunRunner;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Event;
use Illuminate\Support\Facades\Queue;
use Tests\Support\Steps\FakeAdvanceStep;
use Tests\Support\Steps\FakeFailStep;
use Tests\Support\Steps\FakeRetryStep;
use Tests\Support\Steps\FakeThrowStep;
function bindPipeline(array $pipelines): void
{
app()->instance(PipelineRegistry::class, new PipelineRegistry($pipelines));
}
it('advances to the next step and queues the follow-up job', function () {
Queue::fake();
bindPipeline(['test' => [FakeAdvanceStep::class, FakeAdvanceStep::class]]);
$run = ProvisioningRun::factory()->create(['pipeline' => 'test', 'current_step' => 0, 'status' => 'pending']);
app(RunRunner::class)->advance($run);
$run->refresh();
expect($run->current_step)->toBe(1)
->and($run->status)->toBe('running')
->and($run->events()->where('outcome', 'advanced')->count())->toBe(1);
Queue::assertPushed(AdvanceRunJob::class);
});
it('completes the run on the last step', function () {
Queue::fake();
bindPipeline(['test' => [FakeAdvanceStep::class]]);
$run = ProvisioningRun::factory()->create(['pipeline' => 'test', 'status' => 'pending']);
app(RunRunner::class)->advance($run);
$run->refresh();
expect($run->status)->toBe('completed')->and($run->finished_at)->not->toBeNull();
Queue::assertNotPushed(AdvanceRunJob::class);
});
it('waits and increments attempt on retry', function () {
Queue::fake();
bindPipeline(['test' => [FakeRetryStep::class]]);
$run = ProvisioningRun::factory()->create(['pipeline' => 'test', 'max_attempts' => 5]);
app(RunRunner::class)->advance($run);
$run->refresh();
expect($run->status)->toBe('waiting')
->and($run->attempt)->toBe(1)
->and($run->next_attempt_at)->not->toBeNull();
Queue::assertNotPushed(AdvanceRunJob::class);
});
it('fails once retries are exhausted', function () {
bindPipeline(['test' => [FakeRetryStep::class]]);
$run = ProvisioningRun::factory()->create(['pipeline' => 'test', 'max_attempts' => 1]);
app(RunRunner::class)->advance($run);
expect($run->refresh()->status)->toBe('failed');
});
it('fails immediately on a fail result', function () {
bindPipeline(['test' => [FakeFailStep::class]]);
$run = ProvisioningRun::factory()->create(['pipeline' => 'test']);
app(RunRunner::class)->advance($run);
$run->refresh();
expect($run->status)->toBe('failed')->and($run->error)->toBe('boom');
});
it('treats a thrown exception as a retry', function () {
bindPipeline(['test' => [FakeThrowStep::class]]);
$run = ProvisioningRun::factory()->create(['pipeline' => 'test', 'max_attempts' => 5]);
app(RunRunner::class)->advance($run);
$run->refresh();
expect($run->status)->toBe('waiting')
->and($run->attempt)->toBe(1)
->and($run->events()->where('outcome', 'retry')->exists())->toBeTrue();
});
it('retries when a step exceeds its max duration', function () {
bindPipeline(['test' => [FakeAdvanceStep::class]]);
$run = ProvisioningRun::factory()->create([
'pipeline' => 'test',
'status' => 'running',
'started_at' => now()->subMinutes(10),
'max_attempts' => 5,
]);
app(RunRunner::class)->advance($run);
$run->refresh();
expect($run->status)->toBe('waiting')
->and($run->events()->where('message', 'like', '%timed out%')->exists())->toBeTrue();
});
it('is a no-op while the run lock is held', function () {
bindPipeline(['test' => [FakeAdvanceStep::class, FakeAdvanceStep::class]]);
$run = ProvisioningRun::factory()->create(['pipeline' => 'test', 'current_step' => 0]);
$lock = Cache::lock('run:'.$run->uuid, 120);
expect($lock->get())->toBeTrue();
app(RunRunner::class)->advance($run);
expect($run->refresh()->current_step)->toBe(0);
$lock->release();
});
it('broadcasts a StepAdvanced event', function () {
Event::fake([StepAdvanced::class]);
bindPipeline(['test' => [FakeAdvanceStep::class]]);
$run = ProvisioningRun::factory()->create(['pipeline' => 'test']);
app(RunRunner::class)->advance($run);
Event::assertDispatched(
StepAdvanced::class,
fn ($e) => $e->runUuid === $run->uuid && $e->status === 'completed',
);
});

View File

@ -0,0 +1,20 @@
<?php
use App\Console\TickProvisioning;
use App\Models\ProvisioningRun;
use App\Provisioning\Jobs\AdvanceRunJob;
use Illuminate\Support\Facades\Queue;
it('dispatches advance jobs only for due running or waiting runs', function () {
Queue::fake();
ProvisioningRun::factory()->create(['status' => 'waiting', 'next_attempt_at' => now()->subMinute()]);
ProvisioningRun::factory()->create(['status' => 'running', 'next_attempt_at' => null]);
ProvisioningRun::factory()->create(['status' => 'waiting', 'next_attempt_at' => now()->addHour()]);
ProvisioningRun::factory()->create(['status' => 'completed']);
ProvisioningRun::factory()->create(['status' => 'paused']);
app(TickProvisioning::class)();
Queue::assertPushed(AdvanceRunJob::class, 2);
});

View File

@ -0,0 +1,30 @@
<?php
namespace Tests\Support\Steps;
use App\Models\ProvisioningRun;
use App\Provisioning\Contracts\ProvisioningStep;
use App\Provisioning\StepResult;
class FakeAdvanceStep implements ProvisioningStep
{
public function key(): string
{
return 'fake_advance';
}
public function label(): string
{
return 'Fake advance';
}
public function maxDuration(): int
{
return 120;
}
public function execute(ProvisioningRun $run): StepResult
{
return StepResult::advance();
}
}

View File

@ -0,0 +1,30 @@
<?php
namespace Tests\Support\Steps;
use App\Models\ProvisioningRun;
use App\Provisioning\Contracts\ProvisioningStep;
use App\Provisioning\StepResult;
class FakeFailStep implements ProvisioningStep
{
public function key(): string
{
return 'fake_fail';
}
public function label(): string
{
return 'Fake fail';
}
public function maxDuration(): int
{
return 120;
}
public function execute(ProvisioningRun $run): StepResult
{
return StepResult::fail('boom');
}
}

View File

@ -0,0 +1,30 @@
<?php
namespace Tests\Support\Steps;
use App\Models\ProvisioningRun;
use App\Provisioning\Contracts\ProvisioningStep;
use App\Provisioning\StepResult;
class FakeRetryStep implements ProvisioningStep
{
public function key(): string
{
return 'fake_retry';
}
public function label(): string
{
return 'Fake retry';
}
public function maxDuration(): int
{
return 120;
}
public function execute(ProvisioningRun $run): StepResult
{
return StepResult::retry(30, 'need more time');
}
}

View File

@ -0,0 +1,31 @@
<?php
namespace Tests\Support\Steps;
use App\Models\ProvisioningRun;
use App\Provisioning\Contracts\ProvisioningStep;
use App\Provisioning\StepResult;
use RuntimeException;
class FakeThrowStep implements ProvisioningStep
{
public function key(): string
{
return 'fake_throw';
}
public function label(): string
{
return 'Fake throw';
}
public function maxDuration(): int
{
return 120;
}
public function execute(ProvisioningRun $run): StepResult
{
throw new RuntimeException('kaboom');
}
}

View File

@ -0,0 +1,14 @@
<?php
use App\Provisioning\StepResult;
it('builds advance, retry and fail results', function () {
expect(StepResult::advance()->type)->toBe(StepResult::ADVANCE);
$retry = StepResult::retry(30, 'wait');
expect($retry->type)->toBe(StepResult::RETRY)
->and($retry->afterSeconds)->toBe(30)
->and($retry->reason)->toBe('wait');
expect(StepResult::fail('boom')->reason)->toBe('boom');
});