coolify/tests/Feature/NodeWorkloadReconciliationTest.php

<?php

use App\Actions\Node\CreateDeploymentOperation;
use App\Actions\Node\CreateLifecycleOperation;
use App\Actions\Node\ReconcileNodeWorkloads;
use App\Enums\NodeContainerManagementState;
use App\Enums\NodeOperationStatus;
use App\Enums\NodeWorkloadAction;
use App\Enums\NodeWorkloadDesiredState;
use App\Jobs\DeployNodeWorkloadJob;
use App\Jobs\ManageNodeWorkloadJob;
use App\Jobs\RefreshNodeContainersJob;
use App\Models\InstanceSettings;
use App\Models\Node;
use App\Models\NodeCluster;
use App\Models\NodeContainer;
use App\Models\NodeOperation;
use App\Models\NodeWorkload;
use App\Models\NodeWorkloadRevision;
use App\Models\PrivateKey;
use App\Models\Team;
use Illuminate\Foundation\Testing\RefreshDatabase;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Http;
use Illuminate\Support\Facades\Queue;

uses(RefreshDatabase::class);

beforeEach(function () {
    InstanceSettings::forceCreate(['id' => 0]);
    $this->team = Team::factory()->create();
    $key = PrivateKey::factory()->create(['team_id' => $this->team->id]);
    $cluster = NodeCluster::factory()->create(['team_id' => $this->team->id]);
    $this->node = Node::factory()->create([
        'team_id' => $this->team->id,
        'private_key_id' => $key->id,
        'is_usable' => true,
        'is_reachable' => true,
        'node_cluster_id' => $cluster->id,
        'metadata' => [
            'cpu_usage_percent' => 10,
            'memory_bytes' => 1_000,
            'memory_used_bytes' => 100,
            'disk_total_bytes' => 1_000,
            'disk_available_bytes' => 900,
            'collected_at' => now()->toIso8601String(),
        ],
    ]);
    $this->workload = NodeWorkload::factory()->create([
        'team_id' => $this->team->id,
        'desired_state' => NodeWorkloadDesiredState::RUNNING,
    ]);
    $this->node->workloads()->attach($this->workload);
    $this->revision = NodeWorkloadRevision::factory()->create([
        'node_workload_id' => $this->workload->id,
        'image' => 'docker.io/library/alpine:latest',
    ]);
    Queue::fake();
});

it('redeploys a missing workload that should be running', function () {
    ReconcileNodeWorkloads::run($this->node);

    $operation = $this->node->operations()->firstOrFail();
    expect($operation->command_type)->toBe('workload.deploy.v1')
        ->and($operation->node_workload_revision_id)->toBe($this->revision->id);
    Queue::assertPushed(DeployNodeWorkloadJob::class, fn ($job) => $job->operationId === $operation->id);
});

it('does not deploy on a move target while the move runs', function () {
    $source = Node::factory()->create([
        'team_id' => $this->team->id,
        'private_key_id' => $this->node->private_key_id,
        'node_cluster_id' => $this->node->node_cluster_id,
    ]);
    NodeOperation::factory()->create([
        'node_id' => $source->id,
        'node_workload_id' => $this->workload->id,
        'node_workload_revision_id' => $this->revision->id,
        'command_type' => 'workload.move.v1',
        'status' => NodeOperationStatus::RUNNING,
    ]);

    expect(ReconcileNodeWorkloads::run($this->node))->toBe(0)
        ->and($this->node->operations()->count())->toBe(0);
    Queue::assertNotPushed(DeployNodeWorkloadJob::class);
});

it('deploys on the former move target after the move failed', function () {
    $source = Node::factory()->create([
        'team_id' => $this->team->id,
        'private_key_id' => $this->node->private_key_id,
        'node_cluster_id' => $this->node->node_cluster_id,
    ]);
    NodeOperation::factory()->create([
        'node_id' => $source->id,
        'node_workload_id' => $this->workload->id,
        'node_workload_revision_id' => $this->revision->id,
        'command_type' => 'workload.move.v1',
        'status' => NodeOperationStatus::FAILED,
    ]);

    expect(ReconcileNodeWorkloads::run($this->node))->toBe(1);
});

it('reconciles desired state after the scheduled inventory refresh', function () {
    config()->set('constants.flux.internal_url', 'http://flux:7080');
    config()->set('constants.flux.internal_token', 'secret');
    Cache::put($this->node->cacheKey(), [
        'status' => 'connected',
        'last_heartbeat_at' => now()->toIso8601String(),
    ]);
    Http::fake(['*/v1/commands/container.list' => Http::response([
        'command_id' => 'inventory-1',
        'observed_at_unix_ms' => 1_700_000_000_000,
        'containers' => [],
    ])]);

    (new RefreshNodeContainersJob($this->node->id))->handle();

    Queue::assertPushed(DeployNodeWorkloadJob::class);
});

it('starts a stopped current workload that should be running', function () {
    createManagedContainer($this, $this->revision, 'exited');

    ReconcileNodeWorkloads::run($this->node);

    $operation = $this->node->operations()->firstOrFail();
    expect($operation->command_type)->toBe('workload.lifecycle.v1')
        ->and(data_get($operation->request, 'action'))->toBe(NodeWorkloadAction::START->value);
    Queue::assertPushed(ManageNodeWorkloadJob::class, fn ($job) => $job->operationId === $operation->id);
});

it('replaces an outdated running workload with its latest revision', function () {
    $latestRevision = NodeWorkloadRevision::factory()->create([
        'node_workload_id' => $this->workload->id,
        'image' => 'docker.io/library/alpine:3.21',
    ]);
    createManagedContainer($this, $this->revision, 'running');

    ReconcileNodeWorkloads::run($this->node);

    $operation = $this->node->operations()->firstOrFail();
    expect($operation->command_type)->toBe('workload.deploy.v1')
        ->and($operation->node_workload_revision_id)->toBe($latestRevision->id);
    Queue::assertPushed(DeployNodeWorkloadJob::class);
});

it('stops or removes a container to match its desired state', function (NodeWorkloadDesiredState $desiredState, NodeWorkloadAction $action) {
    $this->workload->update(['desired_state' => $desiredState]);
    createManagedContainer($this, $this->revision, 'running');

    ReconcileNodeWorkloads::run($this->node);

    $operation = $this->node->operations()->firstOrFail();
    expect(data_get($operation->request, 'action'))->toBe($action->value);
    Queue::assertPushed(ManageNodeWorkloadJob::class);
})->with([
    'stopped' => [NodeWorkloadDesiredState::STOPPED, NodeWorkloadAction::STOP],
    'removed' => [NodeWorkloadDesiredState::REMOVED, NodeWorkloadAction::REMOVE],
]);

it('does not create work when state matches or another operation is active', function () {
    createManagedContainer($this, $this->revision, 'running');

    ReconcileNodeWorkloads::run($this->node);

    expect($this->node->operations()->count())->toBe(0);
    Queue::assertNothingPushed();

    NodeContainer::query()->delete();
    CreateDeploymentOperation::run($this->node, $this->revision);
    Queue::fake();

    ReconcileNodeWorkloads::run($this->node);

    expect($this->node->operations()->count())->toBe(1);
    Queue::assertNothingPushed();
});

it('persists deployment and lifecycle intent for later reconciliation', function () {
    $this->workload->update(['desired_state' => NodeWorkloadDesiredState::REMOVED]);

    CreateDeploymentOperation::run($this->node, $this->revision);

    expect($this->workload->refresh()->desired_state)->toBe(NodeWorkloadDesiredState::RUNNING);
    $this->node->operations()->delete();

    CreateLifecycleOperation::run($this->node, $this->revision, NodeWorkloadAction::STOP);

    expect($this->workload->refresh()->desired_state)->toBe(NodeWorkloadDesiredState::STOPPED);
});

it('does not replay an uncertain operation or create a duplicate', function () {
    $deployment = CreateDeploymentOperation::run($this->node, $this->revision);
    $deployment['operation']->update(['status' => NodeOperationStatus::UNCERTAIN]);
    Queue::fake();

    $count = ReconcileNodeWorkloads::run($this->node);

    expect($count)->toBe(0)
        ->and($this->node->operations()->count())->toBe(1);
    Queue::assertNothingPushed();
});

it('keeps reconciling later workloads when one workload cannot be reconciled', function () {
    // Memory pressure rejects the deployment of the first workload; the stop of the second must still be sent.
    $this->node->update(['metadata' => [...$this->node->metadata, 'memory_used_bytes' => 990]]);
    $stoppedWorkload = NodeWorkload::factory()->create([
        'team_id' => $this->team->id,
        'desired_state' => NodeWorkloadDesiredState::STOPPED,
    ]);
    $this->node->workloads()->attach($stoppedWorkload);
    $stoppedRevision = NodeWorkloadRevision::factory()->create(['node_workload_id' => $stoppedWorkload->id]);
    NodeContainer::factory()->create([
        'node_id' => $this->node->id,
        'node_workload_id' => $stoppedWorkload->id,
        'node_workload_revision_id' => $stoppedRevision->id,
        'image' => $stoppedRevision->image,
        'state' => 'running',
        'management_state' => NodeContainerManagementState::MANAGED,
        'is_managed' => true,
    ]);

    $count = ReconcileNodeWorkloads::run($this->node);

    $operation = $this->node->operations()->sole();
    expect($count)->toBe(1)
        ->and($operation->node_workload_id)->toBe($stoppedWorkload->id)
        ->and(data_get($operation->request, 'action'))->toBe(NodeWorkloadAction::STOP->value);
    Queue::assertPushed(ManageNodeWorkloadJob::class, fn ($job) => $job->operationId === $operation->id);
    Queue::assertNotPushed(DeployNodeWorkloadJob::class);
});

function createManagedContainer(object $test, NodeWorkloadRevision $revision, string $state): NodeContainer
{
    return NodeContainer::factory()->create([
        'node_id' => $test->node->id,
        'node_workload_id' => $test->workload->id,
        'node_workload_revision_id' => $revision->id,
        'image' => $revision->image,
        'state' => $state,
        'management_state' => NodeContainerManagementState::MANAGED,
        'is_managed' => true,
    ]);
}