coolify/app/Jobs/ProvisionGithubRunnerJob.php

<?php

namespace App\Jobs;

use App\Enums\GithubRunnerStatus;
use App\Models\GithubRunnerConfig;
use App\Models\GithubRunnerExecution;
use App\Models\Server;
use App\Services\GithubRunner\GithubRunnerApi;
use App\Services\GithubRunner\GithubRunnerContainer;
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\DB;
use Illuminate\Support\Sleep;
use RuntimeException;

/**
 * Starts one ephemeral runner for a queued execution on the least busy matching server.
 * When no server has free capacity, the execution stays queued and is retried by
 * CleanupGithubRunnerJob or ReconcileGithubRunnersJob. A runner that fails to start is removed and
 * started again, possibly on another server, up to MAX_ATTEMPTS times.
 */
class ProvisionGithubRunnerJob implements ShouldBeEncrypted, ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public $tries = 1;

    public $timeout = 900;

    private const DOCKER_READY_ATTEMPTS = 60;

    public const MAX_ATTEMPTS = 3;

    private const RETRY_DELAY_SECONDS = 30;

    public function __construct(public int $executionId)
    {
        $this->onQueue('high');
    }

    /**
     * Starts provisioning for the oldest queued executions of a GitHub App.
     */
    public static function dispatchQueued(int $githubAppId, int $limit = 10): void
    {
        GithubRunnerExecution::query()
            ->where('github_app_id', $githubAppId)
            ->where('status', GithubRunnerStatus::Queued)
            ->orderBy('id')
            ->limit($limit)
            ->pluck('id')
            ->each(fn (int $id) => self::dispatch($id));
    }

    public function handle(): void
    {
        $execution = $this->reserveServer();
        if (! $execution) {
            return;
        }

        try {
            $config = $execution->config;
            $server = $execution->server;
            $api = new GithubRunnerApi($execution->githubApp);
            $runnerGroupId = $execution->githubApp->runner_group_id ?: $api->ensureRunnerGroup()['id'];
            $container = new GithubRunnerContainer($execution, $config);

            instant_remote_process($container->prepareCommands(), $server, timeout: $this->timeout);
            if ($container->usesDind()) {
                $this->waitForDocker($container, $server);
            }

            if ($this->stopIfFinished($execution)) {
                return;
            }

            $runner = $api->generateJitConfig($container->name(), $config->registeredLabels(), $runnerGroupId);
            $execution->update(['runner_id' => $runner['runner_id']]);
            if ($this->stopIfFinished($execution)) {
                return;
            }
            // stopIfFinished() reloaded the server, so a transfer that just finished stops the start command.
            instant_remote_process($container->startCommands($runner['encoded_jit_config']), $execution->server);

            if ($this->stopIfFinished($execution)) {
                return;
            }
            GithubRunnerExecution::query()
                ->whereKey($execution->id)
                ->whereIn('status', GithubRunnerStatus::occupying())
                ->update(['ready_at' => now()]);
            GithubRunnerExecution::query()
                ->whereKey($execution->id)
                ->where('status', GithubRunnerStatus::Provisioning)
                ->update(['status' => GithubRunnerStatus::Idle]);
        } catch (\Throwable $e) {
            $this->handleFailure($execution, $e);
        }
    }

    /**
     * Removes the failed runner and queues the job again, or marks it failed after the last attempt.
     * GitHub keeps the job queued, so a later runner with matching labels can still take it.
     */
    private function handleFailure(GithubRunnerExecution $execution, \Throwable $e): void
    {
        if ($this->stopIfFinished($execution)) {
            return;
        }

        $attempt = $execution->provision_attempts;
        if ($attempt >= self::MAX_ATTEMPTS) {
            GithubRunnerExecution::query()
                ->whereKey($execution->id)
                ->where('status', GithubRunnerStatus::Provisioning)
                ->update([
                    'status' => GithubRunnerStatus::Failed,
                    'error_message' => remove_iip("Could not start a runner after {$attempt} attempts: {$e->getMessage()}"),
                    'completed_at' => now(),
                ]);
            CleanupGithubRunnerJob::dispatch($execution->id);

            return;
        }

        // The retry keeps the execution uuid, so its Docker resources must be gone before it starts again.
        (new CleanupGithubRunnerJob($execution->id))->handle();
        $retried = GithubRunnerExecution::query()
            ->whereKey($execution->id)
            ->where('status', GithubRunnerStatus::Provisioning)
            ->update([
                'status' => GithubRunnerStatus::Queued,
                'github_runner_config_id' => null,
                'server_id' => null,
                'runner_name' => null,
                'runner_id' => null,
                'ready_at' => null,
                'error_message' => remove_iip("Attempt {$attempt} of ".self::MAX_ATTEMPTS." failed: {$e->getMessage()}"),
            ]);
        if (! $retried) {
            return;
        }
        self::dispatch($execution->id)->delay(now()->addSeconds(self::RETRY_DELAY_SECONDS * $attempt));
    }

    private function stopIfFinished(GithubRunnerExecution $execution): bool
    {
        // The execution is deleted together with its server.
        if (! GithubRunnerExecution::query()->whereKey($execution->id)->exists()) {
            if ($execution->runner_id) {
                DeregisterGithubRunnerJob::dispatch($execution->github_app_id, $execution->runner_id);
            }

            return true;
        }

        // Reloads the server too: a transfer to another instance can finish while this job runs.
        $execution->refresh();
        if ($execution->isActive() && ! $execution->server?->isTransferredAway()) {
            return false;
        }

        if ($execution->isActive()) {
            $execution->finish(GithubRunnerStatus::Cancelled, 'The server was transferred to another Coolify instance.');
        }
        CleanupGithubRunnerJob::dispatch($execution->id);

        return true;
    }

    /**
     * Picks the least busy matching server and moves the execution to "provisioning".
     * Row locks on the execution and on the App's configs stop parallel jobs from overbooking a server.
     */
    private function reserveServer(): ?GithubRunnerExecution
    {
        return DB::transaction(function () {
            $execution = GithubRunnerExecution::query()->whereKey($this->executionId)->lockForUpdate()->first();
            if (! $execution || $execution->status !== GithubRunnerStatus::Queued) {
                return null;
            }

            $config = GithubRunnerConfig::query()
                ->where('github_app_id', $execution->github_app_id)
                ->where('is_enabled', true)
                ->orderBy('id')
                ->lockForUpdate()
                ->with('server.settings')
                ->get()
                ->filter(fn (GithubRunnerConfig $config) => $config->server !== null
                    && $config->matchesLabels($execution->labels ?? [])
                    && ($config->allow_pull_requests || ! $execution->is_pull_request)
                    && $config->server->canBuildApplications()
                    && ! $config->server->isLocalhost()
                    && $config->server->isFunctional())
                ->map(fn (GithubRunnerConfig $config) => ['config' => $config, 'occupied' => $config->occupiedRunnerCount()])
                ->filter(fn (array $candidate) => $candidate['occupied'] < $candidate['config']->max_runners)
                ->sortBy('occupied')
                ->first()['config'] ?? null;

            if (! $config) {
                return null;
            }

            $execution->update([
                'github_runner_config_id' => $config->id,
                'server_id' => $config->server_id,
                'status' => GithubRunnerStatus::Provisioning,
                'runner_name' => GithubRunnerContainer::nameFor($execution->uuid),
                'provision_attempts' => $execution->provision_attempts + 1,
            ]);

            return $execution;
        });
    }

    private function waitForDocker(GithubRunnerContainer $container, Server $server): void
    {
        for ($attempt = 0; $attempt < self::DOCKER_READY_ATTEMPTS; $attempt++) {
            if (filled(instant_remote_process([$container->dockerReadyCommand()], $server, false))) {
                return;
            }
            Sleep::for(1)->second();
        }

        throw new RuntimeException('The Docker sidecar did not start within '.self::DOCKER_READY_ATTEMPTS.' seconds.');
    }
}