mirror of
https://github.com/coollabsio/coolify.git
synced 2026-11-06 13:10:37 -04:00
fix(jobs): resume S3 revalidation and remove orphaned notification throttles
- The S3 revalidation job stops within its time budget and continues from a cursor on the next run. - An hourly sweep deletes throttle rows whose subject no longer exists.
This commit is contained in:
@@ -2,6 +2,7 @@
|
||||
|
||||
namespace App\Jobs;
|
||||
|
||||
use App\Models\NotificationThrottle;
|
||||
use App\Models\ScheduledDatabaseBackup;
|
||||
use App\Models\TeamInvitation;
|
||||
use App\Models\User;
|
||||
@@ -35,6 +36,7 @@ class CleanupInstanceStuffsJob implements ShouldBeEncrypted, ShouldBeUnique, Sho
|
||||
$this->cleanupInvitationLink();
|
||||
$this->cleanupExpiredEmailChangeRequests();
|
||||
$this->enforceBackupRetention();
|
||||
$this->deleteOrphanedNotificationThrottles();
|
||||
} catch (\Throwable $e) {
|
||||
Log::error('CleanupInstanceStuffsJob failed with error: '.$e->getMessage());
|
||||
}
|
||||
@@ -59,6 +61,20 @@ class CleanupInstanceStuffsJob implements ShouldBeEncrypted, ShouldBeUnique, Sho
|
||||
]);
|
||||
}
|
||||
|
||||
private function deleteOrphanedNotificationThrottles(): void
|
||||
{
|
||||
if (! Cache::add('notification-throttle-orphan-cleanup', true, 3600)) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
NotificationThrottle::deleteOrphans();
|
||||
} catch (\Throwable $e) {
|
||||
Log::error('Failed to delete orphaned notification throttles: '.$e->getMessage());
|
||||
Cache::forget('notification-throttle-orphan-cleanup');
|
||||
}
|
||||
}
|
||||
|
||||
private function enforceBackupRetention(): void
|
||||
{
|
||||
if (! Cache::add('backup-retention-enforcement', true, 1800)) {
|
||||
|
||||
@@ -10,9 +10,13 @@ use Illuminate\Contracts\Queue\ShouldQueue;
|
||||
use Illuminate\Foundation\Bus\Dispatchable;
|
||||
use Illuminate\Queue\InteractsWithQueue;
|
||||
use Illuminate\Queue\SerializesModels;
|
||||
use Illuminate\Support\Facades\Cache;
|
||||
|
||||
/**
|
||||
* Re-tests S3 storages that failed validation, so a storage that becomes reachable again is usable without manual action.
|
||||
*
|
||||
* Each run stops after a time budget and saves the last tested id, so the next run continues from there
|
||||
* instead of re-testing the same slow storages from the start.
|
||||
*/
|
||||
class RevalidateUnusableS3StoragesJob implements ShouldBeEncrypted, ShouldBeUnique, ShouldQueue
|
||||
{
|
||||
@@ -22,18 +26,41 @@ class RevalidateUnusableS3StoragesJob implements ShouldBeEncrypted, ShouldBeUniq
|
||||
|
||||
public $timeout = 1800;
|
||||
|
||||
/**
|
||||
* Stay well below $timeout, because a single unreachable endpoint can take a while to fail.
|
||||
*/
|
||||
public const TIME_BUDGET_SECONDS = 1200;
|
||||
|
||||
public const CURSOR_CACHE_KEY = 'revalidate-unusable-s3-storages:last-id';
|
||||
|
||||
public function handle(): void
|
||||
{
|
||||
S3Storage::query()
|
||||
$deadline = now()->addSeconds(self::TIME_BUDGET_SECONDS);
|
||||
$lastId = Cache::get(self::CURSOR_CACHE_KEY);
|
||||
|
||||
$completed = S3Storage::query()
|
||||
->where('is_usable', false)
|
||||
->chunkById(100, function ($storages): void {
|
||||
->when($lastId !== null, fn ($query) => $query->where('id', '>', $lastId))
|
||||
->chunkById(100, function ($storages) use ($deadline): bool {
|
||||
foreach ($storages as $storage) {
|
||||
if (now()->greaterThanOrEqualTo($deadline)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
$storage->testConnection(shouldSave: true);
|
||||
} catch (\Throwable) {
|
||||
// testConnection() keeps the storage unusable and notifies the team once.
|
||||
}
|
||||
|
||||
Cache::forever(self::CURSOR_CACHE_KEY, $storage->id);
|
||||
}
|
||||
|
||||
return true;
|
||||
});
|
||||
|
||||
if ($completed) {
|
||||
Cache::forget(self::CURSOR_CACHE_KEY);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ use Carbon\CarbonInterface;
|
||||
use Closure;
|
||||
use Illuminate\Database\Eloquent\Model;
|
||||
use Illuminate\Database\Eloquent\Relations\MorphTo;
|
||||
use Illuminate\Database\Eloquent\Relations\Relation;
|
||||
use Throwable;
|
||||
|
||||
/**
|
||||
@@ -133,4 +134,36 @@ class NotificationThrottle extends Model
|
||||
->where('notification', $notification)
|
||||
->exists();
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete rows whose subject no longer exists. Subjects are often removed with bulk or cascading
|
||||
* deletes that skip model events, so this sweep is the reliable cleanup. Soft-deleted subjects are kept.
|
||||
*
|
||||
* Returns the number of deleted rows.
|
||||
*/
|
||||
public static function deleteOrphans(): int
|
||||
{
|
||||
$deleted = 0;
|
||||
|
||||
foreach (static::query()->distinct()->pluck('notifiable_type') as $type) {
|
||||
$class = Relation::getMorphedModel($type) ?? $type;
|
||||
$throttles = static::query()->where('notifiable_type', $type);
|
||||
|
||||
if (! is_a($class, Model::class, true)) {
|
||||
$deleted += $throttles->delete();
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
$subject = new $class;
|
||||
$deleted += $throttles->whereNotExists(
|
||||
$class::query()
|
||||
->withoutGlobalScopes()
|
||||
->selectRaw('1')
|
||||
->whereColumn($subject->getQualifiedKeyName(), (new static)->qualifyColumn('notifiable_id'))
|
||||
)->delete();
|
||||
}
|
||||
|
||||
return $deleted;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
<?php
|
||||
|
||||
use App\Jobs\CleanupInstanceStuffsJob;
|
||||
use App\Models\InstanceSettings;
|
||||
use App\Models\NotificationThrottle;
|
||||
use App\Models\ScheduledDatabaseBackup;
|
||||
use App\Models\Server;
|
||||
use App\Models\Team;
|
||||
use App\Models\User;
|
||||
@@ -137,3 +139,36 @@ it('releases the unreachable claim when sending the unreachable notification thr
|
||||
expect(fn () => $this->server->sendUnreachableNotification())->toThrow(RuntimeException::class)
|
||||
->and(NotificationThrottle::wasSent($this->server, Unreachable::class))->toBeFalse();
|
||||
});
|
||||
|
||||
it('deletes throttle rows whose subject no longer exists', function () {
|
||||
$trashedServer = Server::factory()->create(['team_id' => $this->team->id]);
|
||||
$trashedServer->delete();
|
||||
$rows = [
|
||||
'kept' => ['notifiable_type' => $this->server->getMorphClass(), 'notifiable_id' => $this->server->id],
|
||||
'kept-soft-deleted' => ['notifiable_type' => $trashedServer->getMorphClass(), 'notifiable_id' => $trashedServer->id],
|
||||
'missing-server' => ['notifiable_type' => $this->server->getMorphClass(), 'notifiable_id' => 999999],
|
||||
'missing-backup' => ['notifiable_type' => ScheduledDatabaseBackup::class, 'notifiable_id' => 999999],
|
||||
'missing-class' => ['notifiable_type' => 'App\Models\RemovedModel', 'notifiable_id' => 1],
|
||||
];
|
||||
foreach ($rows as $notification => $row) {
|
||||
NotificationThrottle::query()->insert($row + ['notification' => $notification, 'sent_at' => now()]);
|
||||
}
|
||||
|
||||
expect(NotificationThrottle::deleteOrphans())->toBe(3)
|
||||
->and(NotificationThrottle::query()->orderBy('notification')->pluck('notification')->all())->toBe(['kept', 'kept-soft-deleted']);
|
||||
});
|
||||
|
||||
it('deletes orphaned throttle rows during the instance cleanup', function () {
|
||||
InstanceSettings::query()->firstOrCreate(['id' => 0]);
|
||||
NotificationThrottle::record($this->server, Unreachable::class);
|
||||
NotificationThrottle::query()->insert([
|
||||
'notifiable_type' => ScheduledDatabaseBackup::class,
|
||||
'notifiable_id' => 999999,
|
||||
'notification' => 'App\Notifications\Database\BackupMissing',
|
||||
'sent_at' => now(),
|
||||
]);
|
||||
|
||||
(new CleanupInstanceStuffsJob)->handle();
|
||||
|
||||
expect(NotificationThrottle::query()->pluck('notifiable_type')->all())->toBe([$this->server->getMorphClass()]);
|
||||
});
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
use App\Jobs\RevalidateUnusableS3StoragesJob;
|
||||
use App\Models\S3Storage;
|
||||
use App\Models\Team;
|
||||
use Carbon\Carbon;
|
||||
use Illuminate\Filesystem\FilesystemAdapter;
|
||||
use Illuminate\Foundation\Testing\RefreshDatabase;
|
||||
use Illuminate\Support\Facades\Storage;
|
||||
@@ -61,3 +62,48 @@ it('does not test storages that are already usable', function () {
|
||||
|
||||
(new RevalidateUnusableS3StoragesJob)->handle();
|
||||
});
|
||||
|
||||
it('includes the storage with id 0 on the first run', function () {
|
||||
$storage = revalidationStorage(Team::factory()->create());
|
||||
S3Storage::query()->whereKey($storage->id)->update(['id' => 0]);
|
||||
$disk = Mockery::mock(FilesystemAdapter::class);
|
||||
$disk->shouldReceive('files')->once()->andReturn([]);
|
||||
Storage::shouldReceive('build')->once()->andReturn($disk);
|
||||
|
||||
(new RevalidateUnusableS3StoragesJob)->handle();
|
||||
|
||||
expect(S3Storage::find(0)->is_usable)->toBeTrue();
|
||||
});
|
||||
|
||||
it('stops when the time budget is used and continues with the next storage on the next run', function () {
|
||||
$team = Team::factory()->create();
|
||||
$first = revalidationStorage($team, ['bucket' => 'first-bucket']);
|
||||
$second = revalidationStorage($team, ['bucket' => 'second-bucket']);
|
||||
$tested = [];
|
||||
Storage::shouldReceive('build')->andReturnUsing(function (array $config) use (&$tested) {
|
||||
$tested[] = $config['bucket'];
|
||||
$disk = Mockery::mock(FilesystemAdapter::class);
|
||||
$disk->shouldReceive('files')->andReturnUsing(function () {
|
||||
// A slow endpoint uses the whole time budget of the run.
|
||||
Carbon::setTestNow(now()->addSeconds(RevalidateUnusableS3StoragesJob::TIME_BUDGET_SECONDS + 1));
|
||||
|
||||
throw new RuntimeException('Connection timed out');
|
||||
});
|
||||
|
||||
return $disk;
|
||||
});
|
||||
|
||||
(new RevalidateUnusableS3StoragesJob)->handle();
|
||||
expect($tested)->toBe(['first-bucket']);
|
||||
|
||||
(new RevalidateUnusableS3StoragesJob)->handle();
|
||||
expect($tested)->toBe(['first-bucket', 'second-bucket']);
|
||||
|
||||
// After the last storage, the next run starts from the beginning again.
|
||||
(new RevalidateUnusableS3StoragesJob)->handle();
|
||||
expect($tested)->toBe(['first-bucket', 'second-bucket', 'first-bucket'])
|
||||
->and($first->fresh()->is_usable)->toBeFalse()
|
||||
->and($second->fresh()->is_usable)->toBeFalse();
|
||||
|
||||
Carbon::setTestNow();
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user