app/Support/ResourceStartActivity.php
Rama main · réplica de lectura del repositorio original.
<?php
namespace App\Support;
use App\Enums\ProcessStatus;
use App\Models\Service;
use App\Support\DatabaseImport\DatabaseImportCleanup;
use Illuminate\Support\Collection;
use Spatie\Activitylog\Models\Activity;
/**
* State of the activity log rows that track database and service start/restart runs
* and database imports.
*
* A killed worker, a Coolify restart or a lost queue can leave such an activity queued or
* in progress forever. Activities that have not progressed for longer than their job can
* possibly run are treated as stale, so they no longer block Start/Restart/Import.
* Command output is saved to the activity as it arrives, so `updated_at` is the heartbeat.
*/
class ResourceStartActivity
{
public const DATABASE_START_OPERATION = 'database-start';
public const DATABASE_IMPORT_OPERATION = 'database_import';
/**
* A start that is still waiting in the queue after this long is treated as lost.
*/
public const QUEUED_STALE_AFTER_SECONDS = 600;
/**
* DatabaseStartJob and CoolifyTask time out after 600 seconds; allow a margin on top.
*/
public const IN_PROGRESS_STALE_AFTER_SECONDS = 900;
/**
* A restore can be silent for a long time (for example pg_restore without verbose output),
* so the output heartbeat is not reliable for imports. The remote restore process is killed
* after the SSH command timeout, so an import without progress for longer than that plus
* this margin cannot still be running.
*/
public const IMPORT_STALE_MARGIN_SECONDS = 1800;
/**
* Lower bound for the import limit, also used when the SSH command timeout is disabled (0).
*/
public const IMPORT_MIN_STALE_AFTER_SECONDS = 7200;
/**
* The start lock must not outlive a killed job for long: DatabaseStartJob times out after 600 seconds.
*/
public const DATABASE_START_LOCK_SECONDS = self::IN_PROGRESS_STALE_AFTER_SECONDS;
public const INTERRUPTED_MESSAGE = 'Interrupted by a Coolify restart.';
public const STALE_MESSAGE = 'Marked as failed: no progress was recorded for too long.';
/**
* Added to the reason when Coolify queued the stop of the remote restore and the cleanup.
*/
public const IMPORT_STOPPED_MESSAGE = 'The import was stopped. The database may be partly restored.';
/**
* Added to the reason for an import without stored cleanup data (started before Coolify stored it).
*/
public const IMPORT_PARTLY_RESTORED_MESSAGE = 'The database may be partly restored.';
public const SUPERSEDED_MESSAGE = 'Skipped: a newer start of this database was requested.';
public const ALREADY_STARTING_MESSAGE = 'Skipped: another start of this database is still running.';
public const DATABASE_OPERATION_IN_PROGRESS_MESSAGE = 'Another start, restart or import of this database is already in progress.';
private const ACTIVE_STATUSES = [
ProcessStatus::QUEUED->value,
ProcessStatus::IN_PROGRESS->value,
];
/**
* The latest activity for a resource, if it is queued or in progress and not stale.
*/
public static function latestRunning(string $typeUuid): ?Activity
{
$activity = Activity::query()
->where('properties->type_uuid', $typeUuid)
->latest()
->first();
return self::isRunning($activity) ? $activity : null;
}
public static function isRunning(?Activity $activity): bool
{
if (! $activity || ! in_array(data_get($activity, 'properties.status'), self::ACTIVE_STATUSES, true)) {
return false;
}
return ! self::isStale($activity);
}
public static function isStale(Activity $activity): bool
{
$lastProgressAt = $activity->updated_at ?? $activity->created_at;
if (! $lastProgressAt) {
return true;
}
$staleAfterSeconds = match (true) {
data_get($activity, 'properties.operation') === self::DATABASE_IMPORT_OPERATION => self::importStaleAfterSeconds(),
data_get($activity, 'properties.status') === ProcessStatus::QUEUED->value => self::QUEUED_STALE_AFTER_SECONDS,
default => self::IN_PROGRESS_STALE_AFTER_SECONDS,
};
return $lastProgressAt->lte(now()->subSeconds($staleAfterSeconds));
}
public static function importStaleAfterSeconds(): int
{
$commandTimeout = (int) config('constants.ssh.command_timeout');
return max(self::IMPORT_MIN_STALE_AFTER_SECONDS, $commandTimeout + self::IMPORT_STALE_MARGIN_SECONDS);
}
public static function databaseStartLockKey(string $databaseUuid): string
{
return "database-start:{$databaseUuid}";
}
/**
* Queued or in-progress activities of one operation for a resource, oldest first.
*
* @return Collection<int, Activity>
*/
public static function active(string $typeUuid, string $operation, ?int $teamId = null): Collection
{
return Activity::query()
->where('properties->type_uuid', $typeUuid)
->where('properties->operation', $operation)
->whereIn('properties->status', self::ACTIVE_STATUSES)
->when($teamId !== null, fn ($query) => $query->where('properties->team_id', $teamId))
->orderBy('id')
->get();
}
/**
* Fail the stale activities in the list, so the UI and API stop reporting them as running.
*
* @param Collection<int, Activity> $activities
* @return Collection<int, Activity> The activities that are not stale.
*/
public static function failStale(Collection $activities): Collection
{
[$stale, $live] = $activities->partition(fn (Activity $activity): bool => self::isStale($activity));
$stale->each(fn (Activity $activity) => self::fail($activity, self::STALE_MESSAGE));
return $live->values();
}
/**
* Mark an activity as failed, append the redacted reason to its log and stop log polling.
*
* @param array<string, mixed> $extraProperties
*/
public static function markFailed(Activity $activity, string $message, array $extraProperties = []): void
{
// Callers can pass exception or command output text, which can contain secrets.
$message = remove_iip($message);
$properties = [
...$extraProperties,
'status' => ProcessStatus::ERROR->value,
'error' => $message,
'failed_at' => now()->toIso8601String(),
];
if ($activity->properties->get('exitCode') === null) {
$properties['exitCode'] = 1;
}
$activity->properties = $activity->properties->merge($properties);
$activity->description = self::appendOutput($activity->description, $message);
$activity->save();
}
/**
* Fail database starts, database imports and service starts left queued or in progress
* by a restart. Runs during app:init, before any queue worker is started.
*/
public static function failInterrupted(): int
{
$activities = Activity::query()
->whereIn('properties->status', self::ACTIVE_STATUSES)
->get();
$serviceUuids = self::existingServiceUuids($activities);
$interrupted = $activities->filter(
fn (Activity $activity): bool => in_array(
data_get($activity, 'properties.operation'),
[self::DATABASE_START_OPERATION, self::DATABASE_IMPORT_OPERATION],
true,
)
|| $serviceUuids->contains(data_get($activity, 'properties.type_uuid'))
);
$interrupted->each(fn (Activity $activity) => self::fail($activity, self::INTERRUPTED_MESSAGE));
return $interrupted->count();
}
/**
* Fail an interrupted or stale activity. For a database import, also queue the stop of the
* restore that can still run in the database container, and the removal of its helper
* container and temporary files. The stop is queued: this also runs during boot.
*/
private static function fail(Activity $activity, string $reason): void
{
if (data_get($activity, 'properties.operation') !== self::DATABASE_IMPORT_OPERATION) {
self::markFailed($activity, $reason);
return;
}
$cleanup = DatabaseImportCleanup::stored($activity);
if ($cleanup === null) {
self::markFailed($activity, $reason.' '.self::IMPORT_PARTLY_RESTORED_MESSAGE);
return;
}
// Save the stop flag before the stop is queued, so a retried CoolifyTask sees it.
self::markFailed($activity, $reason.' '.self::IMPORT_STOPPED_MESSAGE, [
DatabaseImportCleanup::STOP_REQUESTED_PROPERTY => now()->toIso8601String(),
]);
DatabaseImportCleanup::queueStop($cleanup);
}
/**
* @param Collection<int, Activity> $activities
* @return Collection<int, string>
*/
private static function existingServiceUuids(Collection $activities): Collection
{
$typeUuids = $activities
->map(fn (Activity $activity) => data_get($activity, 'properties.type_uuid'))
->filter(fn ($uuid): bool => is_string($uuid) && $uuid !== '')
->unique()
->values();
if ($typeUuids->isEmpty()) {
return collect();
}
return $typeUuids
->chunk(500)
->flatMap(fn (Collection $chunk) => Service::query()->whereIn('uuid', $chunk->all())->pluck('uuid'))
->values();
}
private static function appendOutput(?string $description, string $message): string
{
try {
$entries = json_decode($description ?: '[]', true, flags: JSON_THROW_ON_ERROR);
} catch (\JsonException) {
$entries = [];
}
if (! is_array($entries)) {
$entries = [];
}
$lastOrder = collect($entries)->max('order') ?? 0;
$entries[] = [
'type' => 'stderr',
'output' => "\n{$message}\n",
'timestamp' => hrtime(true),
'batch' => 1,
'order' => $lastOrder + 1,
];
return json_encode($entries, flags: JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE);
}
}