fix(servers): load server manager servers in batches by id

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andras Bacsai
2026-10-28 23:59:34 +01:00
co-authored by Claude Opus 5.5
parent a12f62ed4f
commit 3607abf7eb
2 changed files with 201 additions and 54 deletions
+66 -54
View File
@@ -4,7 +4,6 @@ namespace App\Jobs;
use App\Models\InstanceSettings;
use App\Models\Server;
use App\Models\Team;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldBeEncrypted;
use Illuminate\Contracts\Queue\ShouldQueue;
@@ -38,6 +37,11 @@ class ServerManagerJob implements ShouldBeEncrypted, ShouldQueue
*/
private const CLOUD_PROVIDER_STATUS_CHECK_CRON = '*/5 * * * *';
/**
* Number of servers loaded and processed at a time.
*/
public const SERVER_BATCH_SIZE = 500;
/**
* Create a new job instance.
*/
@@ -60,44 +64,55 @@ class ServerManagerJob implements ShouldBeEncrypted, ShouldQueue
$this->instanceTimezone = config('app.timezone');
}
// Get all servers to process
$servers = $this->getServers();
// Decide once per run: these cron keys are global, so a second call in the same run returns false.
$runCloudProviderStatusChecks = shouldRunCronNow(self::CLOUD_PROVIDER_STATUS_CHECK_CRON, $this->instanceTimezone, 'server-cloud-provider-status-checks', $this->executionTime);
$runConnectionChecks = shouldRunCronNow($this->checkFrequency, $this->instanceTimezone, 'server-connection-checks', $this->executionTime);
// Provider state checks run independently so slow APIs cannot block SSH checks.
$this->dispatchCloudProviderStatusChecks($servers);
$relations = ['settings' => fn (HasOne $query) => $query->withoutPatchCheckResult(), 'cloudProviderToken'];
$servers = $servers
->reject(fn (Server $server) => $server->hasPlaceholderIp())
->values();
foreach ($this->getServerIds()->chunk(self::SERVER_BATCH_SIZE) as $serverIds) {
$servers = Server::with($relations)->whereKey($serverIds)->orderBy('id')->get();
// Dispatch ServerConnectionCheck for all servers efficiently
$this->dispatchConnectionChecks($servers);
// Provider state checks run independently so slow APIs cannot block SSH checks.
if ($runCloudProviderStatusChecks) {
$this->dispatchCloudProviderStatusChecks($servers);
}
// Process server-specific scheduled tasks
$this->processScheduledTasks($servers);
$servers = $servers
->reject(fn (Server $server) => $server->hasPlaceholderIp())
->values();
if ($runConnectionChecks) {
$this->dispatchConnectionChecks($servers);
}
// Process server-specific scheduled tasks
$this->processScheduledTasks($servers);
}
}
private function getServers(): Collection
/**
* Selects only the IDs, so the paid-subscription filter runs once and the full server rows are loaded
* in batches by primary key. Loading every server in one query kept all of them in worker memory.
* toBase() skips model hydration, which an Eloquent pluck of the cast id column would do for every row.
*
* @return Collection<int, int>
*/
private function getServerIds(): Collection
{
$relations = ['settings' => fn (HasOne $query) => $query->withoutPatchCheckResult(), 'cloudProviderToken'];
$allServers = Server::with($relations);
if (isCloud()) {
$servers = $allServers->whereRelation('team.subscription', 'stripe_invoice_paid', true)->get();
$own = Team::find(0)->servers()->with($relations)->get();
return $servers->merge($own)->unique('id')->values();
} else {
return $allServers->get();
if (! isCloud()) {
return Server::query()->orderBy('id')->toBase()->pluck('id');
}
return Server::query()->whereRelation('team.subscription', 'stripe_invoice_paid', true)->toBase()->pluck('id')
->merge(Server::query()->where('team_id', 0)->toBase()->pluck('id'))
->unique()
->sort()
->values();
}
private function dispatchCloudProviderStatusChecks(Collection $servers): void
{
if (! shouldRunCronNow(self::CLOUD_PROVIDER_STATUS_CHECK_CRON, $this->instanceTimezone, 'server-cloud-provider-status-checks', $this->executionTime)) {
return;
}
$servers->each(function (Server $server) {
$hasCloudResource = $server->hetzner_server_id
|| $server->vultr_instance_id
@@ -112,34 +127,31 @@ class ServerManagerJob implements ShouldBeEncrypted, ShouldQueue
private function dispatchConnectionChecks(Collection $servers): void
{
if (shouldRunCronNow($this->checkFrequency, $this->instanceTimezone, 'server-connection-checks', $this->executionTime)) {
$servers->each(function (Server $server) {
try {
if ($server->hasPlaceholderIp()) {
return;
}
// Skip SSH connection check if Sentinel is healthy — its heartbeat already proves connectivity.
// Never skip while the server is marked unreachable or unusable: the heartbeat does not
// restore these flags, so deployments would fail with "Server is not functional" forever.
if ($server->isSentinelEnabled() && $server->isSentinelLive()
&& $server->settings->is_reachable && $server->settings->is_usable) {
return;
}
if ($this->shouldSkipDueToBackoff($server)) {
return;
}
ServerConnectionCheckJob::dispatch($server);
} catch (\Exception $e) {
Log::channel('scheduled-errors')->error('Failed to dispatch ServerConnectionCheck', [
'server_id' => $server->id,
'server_name' => $server->name,
'error' => get_class($e).': '.$e->getMessage(),
]);
$servers->each(function (Server $server) {
try {
if ($server->hasPlaceholderIp()) {
return;
}
});
}
// Skip SSH connection check if Sentinel is healthy — its heartbeat already proves connectivity.
// Never skip while the server is marked unreachable or unusable: the heartbeat does not
// restore these flags, so deployments would fail with "Server is not functional" forever.
if ($server->isSentinelEnabled() && $server->isSentinelLive()
&& $server->settings->is_reachable && $server->settings->is_usable) {
return;
}
if ($this->shouldSkipDueToBackoff($server)) {
return;
}
ServerConnectionCheckJob::dispatch($server);
} catch (\Exception $e) {
Log::channel('scheduled-errors')->error('Failed to dispatch ServerConnectionCheck', [
'server_id' => $server->id,
'server_name' => $server->name,
'error' => get_class($e).': '.$e->getMessage(),
]);
}
});
}
private function processScheduledTasks(Collection $servers): void
@@ -0,0 +1,135 @@
<?php
use App\Jobs\ServerConnectionCheckJob;
use App\Jobs\ServerManagerJob;
use App\Models\InstanceSettings;
use App\Models\PrivateKey;
use App\Models\Server;
use App\Models\Subscription;
use App\Models\Team;
use Illuminate\Foundation\Testing\RefreshDatabase;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Queue;
uses(RefreshDatabase::class);
beforeEach(function () {
InstanceSettings::forceCreate(['id' => 0, 'instance_timezone' => 'UTC']);
$this->rootTeam = Team::factory()->create(['id' => 0]);
config(['constants.coolify.self_hosted' => false]);
// Minute 0 is due for the Cloud connection check frequency (*/5).
Carbon::setTestNow('2026-07-11 12:00:00');
});
afterEach(function () {
Carbon::setTestNow();
});
function createServerForSelectionTest(Team $team): Server
{
$privateKey = PrivateKey::query()->firstWhere('team_id', $team->id)
?? PrivateKey::factory()->create(['team_id' => $team->id]);
return Server::factory()->create(['team_id' => $team->id, 'private_key_id' => $privateKey->id]);
}
/**
* @return array<int, int> server id => number of connection checks queued for it
*/
function connectionChecksPerServer(): array
{
Queue::fake();
(new ServerManagerJob)->handle();
return collect(Queue::pushed(ServerConnectionCheckJob::class))
->countBy(fn (ServerConnectionCheckJob $job) => $job->server->id)
->sortKeys()
->all();
}
it('processes each server of a paid team once, also with more than one paid subscription', function () {
$team = Team::factory()->create();
Subscription::factory()->count(2)->create(['team_id' => $team->id, 'stripe_invoice_paid' => true]);
$server = createServerForSelectionTest($team);
expect(connectionChecksPerServer())->toBe([$server->id => 1]);
});
it('skips servers of teams without a paid subscription', function () {
$unpaid = Team::factory()->create();
Subscription::factory()->create(['team_id' => $unpaid->id, 'stripe_invoice_paid' => false]);
createServerForSelectionTest($unpaid);
$pastDue = Team::factory()->create();
Subscription::factory()->create(['team_id' => $pastDue->id, 'stripe_invoice_paid' => false, 'stripe_past_due' => true]);
createServerForSelectionTest($pastDue);
createServerForSelectionTest(Team::factory()->create());
expect(connectionChecksPerServer())->toBe([]);
});
it('processes a team with one paid and one unpaid subscription', function () {
$team = Team::factory()->create();
Subscription::factory()->create(['team_id' => $team->id, 'stripe_invoice_paid' => false]);
Subscription::factory()->create(['team_id' => $team->id, 'stripe_invoice_paid' => true]);
$server = createServerForSelectionTest($team);
expect(connectionChecksPerServer())->toBe([$server->id => 1]);
});
it('skips soft-deleted servers of a paid team', function () {
$team = Team::factory()->create();
Subscription::factory()->create(['team_id' => $team->id, 'stripe_invoice_paid' => true]);
$kept = createServerForSelectionTest($team);
createServerForSelectionTest($team)->delete();
expect(connectionChecksPerServer())->toBe([$kept->id => 1]);
});
it('processes root team servers once without a subscription', function () {
$rootServer = createServerForSelectionTest($this->rootTeam);
expect(connectionChecksPerServer())->toBe([$rootServer->id => 1]);
Subscription::factory()->create(['team_id' => 0, 'stripe_invoice_paid' => true]);
Carbon::setTestNow('2026-07-11 12:05:00');
expect(connectionChecksPerServer())->toBe([$rootServer->id => 1]);
});
it('follows subscription payment changes between runs', function () {
$team = Team::factory()->create();
$subscription = Subscription::factory()->create(['team_id' => $team->id, 'stripe_invoice_paid' => false]);
$server = createServerForSelectionTest($team);
expect(connectionChecksPerServer())->toBe([]);
$subscription->update(['stripe_invoice_paid' => true]);
Carbon::setTestNow('2026-07-11 12:05:00');
expect(connectionChecksPerServer())->toBe([$server->id => 1]);
$subscription->update(['stripe_invoice_paid' => false]);
Carbon::setTestNow('2026-07-11 12:10:00');
expect(connectionChecksPerServer())->toBe([]);
});
it('processes every server when they span more than one batch', function () {
$team = Team::factory()->create();
Subscription::factory()->create(['team_id' => $team->id, 'stripe_invoice_paid' => true]);
$privateKey = PrivateKey::factory()->create(['team_id' => $team->id]);
$servers = Server::factory()
->count(ServerManagerJob::SERVER_BATCH_SIZE + 1)
->create(['team_id' => $team->id, 'private_key_id' => $privateKey->id]);
expect(connectionChecksPerServer())
->toBe($servers->pluck('id')->sort()->mapWithKeys(fn (int $id) => [$id => 1])->all());
});
it('processes every server in self-hosted mode', function () {
config(['constants.coolify.self_hosted' => true]);
Carbon::setTestNow('2026-07-11 12:01:00');
$server = createServerForSelectionTest(Team::factory()->create());
expect(connectionChecksPerServer())->toBe([$server->id => 1]);
});