fix(docker): index interrupted cleanup lookup and process it in batches

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/Jobs/DockerCleanupJob.php

@@ -13,6 +13,7 @@ use Carbon\Carbon;
 use Illuminate\Bus\Queueable;
 use Illuminate\Contracts\Queue\ShouldBeEncrypted;
 use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Database\Eloquent\Collection;
 use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Queue\InteractsWithQueue;
 use Illuminate\Queue\SerializesModels;
@@ -43,6 +44,8 @@ class DockerCleanupJob implements ShouldBeEncrypted, ShouldQueue
 
     public const INTERRUPTED_MESSAGE = 'Docker cleanup was interrupted: its worker stopped before the cleanup finished.';
 
+    public const INTERRUPTED_EXECUTIONS_CHUNK_SIZE = 100;
+
     /**
      * Shares the per-server lock with queued CleanupDocker runs, so two cleanups never run on
      * one server at once. A scheduled run that finds the lock held marks its occurrence
@@ -254,31 +257,32 @@ class DockerCleanupJob implements ShouldBeEncrypted, ShouldQueue
      */
     public static function failInterruptedExecutions(): void
     {
-        $executions = DockerCleanupExecution::query()
+        DockerCleanupExecution::query()
+            ->select(['id', 'server_id'])
             ->with('server.team')
             ->where('status', 'running')
             ->whereNull('finished_at')
             ->where('created_at', '<', now()->subSeconds(CleanupDocker::overlapLockExpiresAfter()))
-            ->get();
+            ->chunkById(self::INTERRUPTED_EXECUTIONS_CHUNK_SIZE, function (Collection $executions): void {
+                foreach ($executions as $execution) {
+                    $updated = DockerCleanupExecution::query()
+                        ->whereKey($execution->id)
+                        ->where('status', 'running')
+                        ->whereNull('finished_at')
+                        ->update([
+                            'status' => 'failed',
+                            'message' => self::INTERRUPTED_MESSAGE,
+                            'finished_at' => Carbon::now()->toImmutable(),
+                        ]);
 
-        foreach ($executions as $execution) {
-            $updated = DockerCleanupExecution::query()
-                ->whereKey($execution->id)
-                ->where('status', 'running')
-                ->whereNull('finished_at')
-                ->update([
-                    'status' => 'failed',
-                    'message' => self::INTERRUPTED_MESSAGE,
-                    'finished_at' => Carbon::now()->toImmutable(),
-                ]);
+                    if ($updated === 0) {
+                        continue;
+                    }
 
-            if ($updated === 0) {
-                continue;
-            }
-
-            event(new DockerCleanupDone($execution->refresh()));
-            $execution->server?->team?->notify(new DockerCleanupFailed($execution->server, 'Docker cleanup job failed with the following error: '.self::INTERRUPTED_MESSAGE));
-        }
+                    event(new DockerCleanupDone($execution->refresh()));
+                    $execution->server?->team?->notify(new DockerCleanupFailed($execution->server, 'Docker cleanup job failed with the following error: '.self::INTERRUPTED_MESSAGE));
+                }
+            });
     }
 
     public function failed(?\Throwable $exception): void

database/migrations/2026_10_28_214953_add_index_for_interrupted_docker_cleanup_executions.php

@@ -0,0 +1,45 @@
+<?php
+
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+
+return new class extends Migration
+{
+    public $withinTransaction = false;
+
+    /**
+     * Supports DockerCleanupJob::failInterruptedExecutions(), which runs on every scheduler tick.
+     * Only unfinished executions are indexed, so the index stays small on large histories.
+     */
+    public function up(): void
+    {
+        if (DB::getDriverName() === 'pgsql') {
+            DB::statement('CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_docker_cleanup_executions_unfinished_status_created ON docker_cleanup_executions (status, created_at) WHERE finished_at IS NULL');
+
+            return;
+        }
+
+        if (! Schema::hasIndex('docker_cleanup_executions', 'idx_docker_cleanup_executions_unfinished_status_created')) {
+            Schema::table('docker_cleanup_executions', function (Blueprint $table) {
+                $table->index(['status', 'created_at'], 'idx_docker_cleanup_executions_unfinished_status_created');
+            });
+        }
+    }
+
+    public function down(): void
+    {
+        if (DB::getDriverName() === 'pgsql') {
+            DB::statement('DROP INDEX CONCURRENTLY IF EXISTS idx_docker_cleanup_executions_unfinished_status_created');
+
+            return;
+        }
+
+        if (Schema::hasIndex('docker_cleanup_executions', 'idx_docker_cleanup_executions_unfinished_status_created')) {
+            Schema::table('docker_cleanup_executions', function (Blueprint $table) {
+                $table->dropIndex('idx_docker_cleanup_executions_unfinished_status_created');
+            });
+        }
+    }
+};

tests/Feature/DockerCleanupJobTest.php

@@ -883,3 +883,93 @@ it('fails an interrupted cleanup of a deleted server without breaking its broadc
     Queue::assertPushed(BroadcastEvent::class, fn (BroadcastEvent $job) => $job->event instanceof DockerCleanupDone
         && $job->event->broadcastOn() === []);
 });
+
+/**
+ * Creates an execution at an exact age, so tests can place it on either side of the stale cutoff.
+ *
+ * @param  array<string, mixed>  $attributes
+ */
+function dockerCleanupExecutionCreatedSecondsAgo(Server $server, int $secondsAgo, array $attributes = []): DockerCleanupExecution
+{
+    $execution = DockerCleanupExecution::create(['server_id' => $server->id, ...$attributes]);
+    $execution->forceFill(['created_at' => now()->subSeconds($secondsAgo)])->saveQuietly();
+
+    return $execution;
+}
+
+it('fails only running, unfinished cleanups that are older than the server lock and sends one event and notification each', function () {
+    Carbon::setTestNow(Carbon::create(2026, 10, 28, 12, 0, 0, 'UTC'));
+    Event::fake([DockerCleanupDone::class]);
+    Notification::fake();
+    $server = dockerCleanupReachableServer();
+    $server->team->emailNotificationSettings->update(['smtp_enabled' => true, 'docker_cleanup_failure_email_notifications' => true]);
+    $cutoff = CleanupDocker::overlapLockExpiresAfter();
+
+    $stale = dockerCleanupExecutionCreatedSecondsAgo($server, $cutoff + 1, ['message' => 'still running', 'cleanup_log' => '["kept"]']);
+    $atCutoff = dockerCleanupExecutionCreatedSecondsAgo($server, $cutoff);
+    $recent = dockerCleanupExecutionCreatedSecondsAgo($server, 60);
+    $finished = dockerCleanupExecutionCreatedSecondsAgo($server, $cutoff + 3600, ['finished_at' => now()->subHour()]);
+    $succeeded = dockerCleanupExecutionCreatedSecondsAgo($server, $cutoff + 3600, ['status' => 'success', 'message' => 'Saved 3%']);
+    $failed = dockerCleanupExecutionCreatedSecondsAgo($server, $cutoff + 3600, ['status' => 'failed', 'message' => 'original error']);
+
+    DockerCleanupJob::failInterruptedExecutions();
+
+    $stale->refresh();
+    expect($stale->status)->toBe('failed')
+        ->and($stale->message)->toBe(DockerCleanupJob::INTERRUPTED_MESSAGE)
+        ->and($stale->finished_at)->not->toBeNull()
+        ->and($stale->cleanup_log)->toBe('["kept"]');
+    foreach ([$atCutoff, $recent] as $running) {
+        expect($running->fresh()->only(['status', 'finished_at']))->toBe(['status' => 'running', 'finished_at' => null]);
+    }
+    expect($finished->fresh()->status)->toBe('running')
+        ->and($succeeded->fresh()->only(['status', 'message', 'finished_at']))->toBe(['status' => 'success', 'message' => 'Saved 3%', 'finished_at' => null])
+        ->and($failed->fresh()->only(['status', 'message', 'finished_at']))->toBe(['status' => 'failed', 'message' => 'original error', 'finished_at' => null]);
+
+    Event::assertDispatchedTimes(DockerCleanupDone::class, 1);
+    Event::assertDispatched(DockerCleanupDone::class, fn (DockerCleanupDone $event) => $event->execution->is($stale)
+        && $event->execution->status === 'failed'
+        && $event->teamId === $server->team_id);
+    Notification::assertSentToTimes($server->team, DockerCleanupFailed::class, 1);
+    Notification::assertSentTo($server->team, DockerCleanupFailed::class, fn (DockerCleanupFailed $notification) => $notification->server->is($server)
+        && str_contains($notification->message, DockerCleanupJob::INTERRUPTED_MESSAGE));
+});
+
+it('fails interrupted cleanups across more than one batch', function () {
+    Carbon::setTestNow(Carbon::create(2026, 10, 28, 12, 0, 0, 'UTC'));
+    Event::fake([DockerCleanupDone::class]);
+    $server = dockerCleanupReachableServer();
+    $count = DockerCleanupJob::INTERRUPTED_EXECUTIONS_CHUNK_SIZE * 2 + 1;
+    $staleAge = CleanupDocker::overlapLockExpiresAfter() + 1;
+    foreach (range(1, $count) as $ignored) {
+        dockerCleanupExecutionCreatedSecondsAgo($server, $staleAge);
+    }
+
+    DockerCleanupJob::failInterruptedExecutions();
+
+    expect(DockerCleanupExecution::query()->where('status', 'failed')->whereNotNull('finished_at')->count())->toBe($count)
+        ->and(DockerCleanupExecution::query()->where('status', 'running')->count())->toBe(0);
+    Event::assertDispatchedTimes(DockerCleanupDone::class, $count);
+});
+
+it('does not fail a cleanup again when it finished after the stale lookup read it', function () {
+    Carbon::setTestNow(Carbon::create(2026, 10, 28, 12, 0, 0, 'UTC'));
+    Event::fake([DockerCleanupDone::class]);
+    $server = dockerCleanupReachableServer();
+    $staleAge = CleanupDocker::overlapLockExpiresAfter() + 1;
+    $first = dockerCleanupExecutionCreatedSecondsAgo($server, $staleAge);
+    $finishesMeanwhile = dockerCleanupExecutionCreatedSecondsAgo($server, $staleAge);
+    DB::listen(function ($query) use ($first, $finishesMeanwhile) {
+        static $done = false;
+        if (! $done && str_starts_with(strtolower($query->sql), 'update') && in_array($first->id, $query->bindings, true)) {
+            $done = true;
+            DockerCleanupExecution::query()->whereKey($finishesMeanwhile->id)->update(['status' => 'success', 'message' => 'done', 'finished_at' => now()]);
+        }
+    });
+
+    DockerCleanupJob::failInterruptedExecutions();
+
+    expect($first->fresh()->status)->toBe('failed')
+        ->and($finishesMeanwhile->fresh()->only(['status', 'message']))->toBe(['status' => 'success', 'message' => 'done']);
+    Event::assertDispatchedTimes(DockerCleanupDone::class, 1);
+});