copenhagentruckwash/api
1
0
Fork 0
Code Issues Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

Queue cron runs through workers

Browse Source
This commit is contained in:
Jeppe Bundgaard
2026-07-13 20:03:39 +02:00
parent 403f93e62c
commit 026492c3bd
7 changed files with 316 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/nginx/app/classes/cron_scheduler.php
+200 -8
View File
@@ -67,15 +67,79 @@ class cron_scheduler
);
}
public function queueTaskRun(string $task_id_or_legacy_name, ?int $actor_user_id = null, bool $force = false): array
{
$this->ensureReady();
$this->syncDefinitions();
$definition = $this->registry->get($task_id_or_legacy_name);
if ($definition === null) {
throw new RuntimeException('Cron task not found.');
}
$state = $this->stateRows()[$definition->id] ?? [];
$enabled = (bool)($state['enabled'] ?? $definition->enabled);
if (!$enabled && !$force) {
throw new RuntimeException('Cron task is disabled.');
}
if ($this->taskIsLocked($state)) {
throw new RuntimeException('Cron task is already running.');
}
$existing = $this->fetchOne(
"SELECT * FROM cron_task_runs
WHERE task_id = " . $this->sql($definition->id) . " AND status = 'queued'
ORDER BY id DESC LIMIT 1"
);
if ($existing !== null) {
$this->markTaskQueued($definition);
return $this->publicRun($existing);
}
$scheduled_for = date('Y-m-d H:i:s');
$this->query(
"INSERT INTO cron_task_runs
(task_id, module, source, status, actor_user_id, scheduled_for, force_run)
VALUES ("
. $this->sql($definition->id) . ', '
. $this->sql($definition->module) . ", 'manual', 'queued', "
. ($actor_user_id === null ? 'NULL' : (string)(int)$actor_user_id) . ', '
. $this->sql($scheduled_for) . ', '
. ($force ? '1' : '0')
. ")"
);
$run_id = (int)$this->insertId();
$this->markTaskQueued($definition);
return $this->publicRun($this->fetchOne("SELECT * FROM cron_task_runs WHERE id = $run_id") ?? []);
}
public function runDue(string $source = 'automatic'): array
{
$this->ensureReady();
$this->syncDefinitions();
$ran = [];
foreach ($this->queuedRuns() as $queuedRun) {
try {
$run = $this->runQueuedRun($queuedRun);
if ($run !== null) {
$ran[] = $run;
}
} catch (Throwable $throwable) {
$ran[] = [
'task_id' => (string)($queuedRun['task_id'] ?? ''),
'module' => (string)($queuedRun['module'] ?? ''),
'source' => (string)($queuedRun['source'] ?? 'manual'),
'status' => 'skipped',
'error_message' => $throwable->getMessage(),
];
}
}
$now = time();
$states = $this->stateRows();
foreach ($this->registry->definitions() as $definition) {
$state = $states[$definition->id] ?? [];
if (!(bool)($state['enabled'] ?? $definition->enabled)) {
@@ -158,11 +222,16 @@ class cron_scheduler
$started = microtime(true);
$started_at = date('Y-m-d H:i:s', (int)$started);
$run_id = $this->createRun($definition, $source, $actor_user_id, $scheduled_for, $started_at);
$run_id = $this->createRun($definition, $source, $actor_user_id, $scheduled_for, $started_at, $force);
$this->query(
"UPDATE cron_task_state SET current_run_id = $run_id WHERE task_id = " . $this->sql($definition->id)
);
return $this->executeClaimedRun($definition, $run_id, $started);
}
private function executeClaimedRun(cron_task_definition $definition, int $run_id, float $started): array
{
$status = 'succeeded';
$summary = [];
$error_message = null;
@@ -205,9 +274,7 @@ class cron_scheduler
$this->completeRun($run_id, $status, $completed_at, $duration_ms, $summary, $error_message);
$this->releaseLock($definition, $status, $error_message, $completed_at);
$run = $this->fetchOne("SELECT * FROM cron_task_runs WHERE id = $run_id") ?? [];
$run['summary'] = $this->decodeJson($run['summary_json'] ?? null);
return $run;
return $this->publicRun($this->fetchOne("SELECT * FROM cron_task_runs WHERE id = $run_id") ?? []);
}
public function updateTaskConfig(string $task_id, array $config): array
@@ -300,6 +367,118 @@ class cron_scheduler
return $this->affectedRows() === 1;
}
private function taskIsLocked(array $state): bool
{
$lockedUntil = (string)($state['locked_until'] ?? '');
return $lockedUntil !== ''
&& strtotime($lockedUntil) !== false
&& strtotime($lockedUntil) >= time();
}
private function markTaskQueued(cron_task_definition $definition): void
{
$now = date('Y-m-d H:i:s');
$this->query(
"UPDATE cron_task_state
SET last_status = 'queued',
last_error = NULL,
next_run_at = CASE
WHEN next_run_at IS NULL OR next_run_at > " . $this->sql($now) . " THEN " . $this->sql($now) . "
ELSE next_run_at
END
WHERE task_id = " . $this->sql($definition->id)
);
}
/**
* @return array<int, array<string, mixed>>
*/
private function queuedRuns(): array
{
return $this->fetchAll("SELECT * FROM cron_task_runs WHERE status = 'queued' ORDER BY id ASC LIMIT 50");
}
private function runQueuedRun(array $queuedRun): ?array
{
$run_id = (int)($queuedRun['id'] ?? 0);
$definition = $this->registry->get((string)($queuedRun['task_id'] ?? ''));
if ($run_id < 1 || $definition === null) {
if ($run_id > 0) {
$this->skipQueuedRun($run_id, 'Cron task not found.');
return $this->publicRun($this->fetchOne("SELECT * FROM cron_task_runs WHERE id = $run_id") ?? []);
}
return null;
}
$force = (bool)($queuedRun['force_run'] ?? false);
$state = $this->stateRows()[$definition->id] ?? [];
$enabled = (bool)($state['enabled'] ?? $definition->enabled);
if (!$enabled && !$force) {
$this->skipQueuedRun($run_id, 'Cron task is disabled.', $definition);
return $this->publicRun($this->fetchOne("SELECT * FROM cron_task_runs WHERE id = $run_id") ?? []);
}
if (!$this->claimLock($definition)) {
return null;
}
$started = microtime(true);
$started_at = date('Y-m-d H:i:s', (int)$started);
$this->query(
"UPDATE cron_task_runs
SET status = 'running',
started_at = " . $this->sql($started_at) . ",
lock_owner = " . $this->sql($this->lock_owner) . "
WHERE id = $run_id AND status = 'queued'"
);
if ($this->affectedRows() !== 1) {
$this->clearClaimedLock($definition);
return null;
}
$this->query(
"UPDATE cron_task_state SET current_run_id = $run_id WHERE task_id = " . $this->sql($definition->id)
);
return $this->executeClaimedRun($definition, $run_id, $started);
}
private function skipQueuedRun(int $run_id, string $message, ?cron_task_definition $definition = null): void
{
$completed_at = date('Y-m-d H:i:s');
$this->query(
"UPDATE cron_task_runs
SET status = 'skipped',
completed_at = " . $this->sql($completed_at) . ",
duration_ms = 0,
error_message = " . $this->sql($message) . "
WHERE id = $run_id AND status = 'queued'"
);
$updated = $this->affectedRows() === 1;
if (!$updated || $definition === null) {
return;
}
$this->query(
"UPDATE cron_task_state
SET last_status = 'skipped',
last_error = " . $this->sql($message) . "
WHERE task_id = " . $this->sql($definition->id)
);
}
private function clearClaimedLock(cron_task_definition $definition): void
{
$this->query(
"UPDATE cron_task_state
SET locked_until = NULL,
lock_owner = NULL
WHERE task_id = " . $this->sql($definition->id) . "
AND lock_owner = " . $this->sql($this->lock_owner)
);
}
private function releaseLock(cron_task_definition $definition, string $status, ?string $error_message, string $completed_at): void
{
$state = $this->fetchOne("SELECT schedule_json FROM cron_task_state WHERE task_id = " . $this->sql($definition->id));
@@ -340,11 +519,12 @@ class cron_scheduler
string $source,
?int $actor_user_id,
?string $scheduled_for,
string $started_at
string $started_at,
bool $force = false
): int {
$this->query(
"INSERT INTO cron_task_runs
(task_id, module, source, status, actor_user_id, scheduled_for, started_at, lock_owner)
(task_id, module, source, status, actor_user_id, scheduled_for, started_at, lock_owner, force_run)
VALUES ("
. $this->sql($definition->id) . ', '
. $this->sql($definition->module) . ', '
@@ -352,7 +532,8 @@ class cron_scheduler
. ($actor_user_id === null ? 'NULL' : (string)(int)$actor_user_id) . ', '
. $this->nullableSql($scheduled_for) . ', '
. $this->sql($started_at) . ', '
. $this->sql($this->lock_owner)
. $this->sql($this->lock_owner) . ', '
. ($force ? '1' : '0')
. ")"
);
@@ -443,6 +624,17 @@ class cron_scheduler
return is_array($decoded) ? $decoded : [];
}
private function publicRun(array $run): array
{
if ($run === []) {
return [];
}
$run['force_run'] = (bool)($run['force_run'] ?? false);
$run['summary'] = $this->decodeJson($run['summary_json'] ?? null);
return $run;
}
private function fetchOne(string $sql): ?array
{
$rows = $this->fetchAll($sql);
services/nginx/app/classes/cron_schema_bootstrap.php
+20
View File
@@ -53,6 +53,7 @@ class cron_schema_bootstrap
summary_json LONGTEXT NULL,
error_message TEXT NULL,
lock_owner VARCHAR(191) NULL,
force_run TINYINT(1) NOT NULL DEFAULT 0,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
KEY idx_cron_task_runs_task_created (task_id, created_at),
@@ -60,6 +61,7 @@ class cron_schema_bootstrap
KEY idx_cron_task_runs_module_created (module, created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"
);
self::ensureColumn('cron_task_runs', 'force_run', 'TINYINT(1) NOT NULL DEFAULT 0 AFTER lock_owner');
$db->query(
"CREATE TABLE IF NOT EXISTS cron_worker_state (
@@ -94,4 +96,22 @@ class cron_schema_bootstrap
self::$initialized = true;
}
private static function ensureColumn(string $table, string $column, string $definition): void
{
global $db;
$table = preg_replace('/[^a-zA-Z0-9_]/', '', $table);
$column = preg_replace('/[^a-zA-Z0-9_]/', '', $column);
if ($table === '' || $column === '') {
return;
}
$result = $db->query("SHOW COLUMNS FROM `{$table}` LIKE '{$column}'");
if ($result && $result->num_rows > 0) {
return;
}
$db->query("ALTER TABLE `{$table}` ADD COLUMN `{$column}` {$definition}");
}
}
services/nginx/app/classes/release_manager.php
+50 -1
View File
@@ -6270,24 +6270,73 @@ class release_manager
private function deployCronWorkerAfterApiDeployment(array $apiTarget, ?string $commitSha, ?int $actorUserId, int $deploymentId): array
{
if (!$this->cronWorkerAutoprovisionEnabled($apiTarget)) {
return [
'ok' => true,
'skipped' => true,
'required' => false,
'reason' => 'cron_worker_autoprovision_disabled',
];
}
try {
return $this->deployCronWorkerForApiTarget($apiTarget, $commitSha, $actorUserId, false, $deploymentId);
$result = $this->deployCronWorkerForApiTarget($apiTarget, $commitSha, $actorUserId, false, $deploymentId);
$result['required'] = $this->cronWorkerAutoprovisionRequired($apiTarget);
return $result;
} catch (Throwable $throwable) {
$channelId = (int)($apiTarget['channel_id'] ?? 0) ?: null;
$this->audit($channelId, $deploymentId, 'cron_worker_deploy_failed', $actorUserId, 'warning', [
'api_target_id' => (int)($apiTarget['id'] ?? 0),
'commit_sha' => $commitSha,
'error' => $throwable->getMessage(),
'required' => $this->cronWorkerAutoprovisionRequired($apiTarget),
]);
if ($this->cronWorkerAutoprovisionRequired($apiTarget)) {
throw new RuntimeException(
'Cron worker deployment is required for API deployments but did not complete: ' . $throwable->getMessage(),
0,
$throwable
);
}
return [
'ok' => false,
'required' => false,
'warning' => 'API deployment completed, but the cron worker deployment did not complete.',
'error' => $throwable->getMessage(),
];
}
}
private function cronWorkerAutoprovisionEnabled(array $apiTarget): bool
{
$context = self::jsonDecode($apiTarget['deploy_context_json'] ?? null);
foreach (['cron_worker_autoprovision', 'cron_worker_enabled'] as $key) {
if (array_key_exists($key, $context)) {
return $this->toBool($context[$key]);
}
}
return true;
}
private function cronWorkerAutoprovisionRequired(array $apiTarget): bool
{
if (!$this->cronWorkerAutoprovisionEnabled($apiTarget)) {
return false;
}
$context = self::jsonDecode($apiTarget['deploy_context_json'] ?? null);
foreach (['cron_worker_autoprovision_required', 'cron_worker_required'] as $key) {
if (array_key_exists($key, $context)) {
return $this->toBool($context[$key]);
}
}
return true;
}
public function promoteDeployment(int $deploymentId, ?int $actorUserId = null): array
{
$this->ensureSchema();
services/nginx/app/routes/cronRoute.php
+3 -4
View File
@@ -88,14 +88,13 @@ class cronRoute
}
try {
$run = (new cron_scheduler())->runTask(
$run = (new cron_scheduler())->queueTaskRun(
$task_id,
'manual',
$this->actorUserId(),
$this->toBool($parameters['force'] ?? false, false)
);
(new logs_o())->add('cron', 'global', 1, $this->actorUserId() ?? 0, 'CRON_JOB_RUN', 'Ran cron task: ' . $task_id);
$response->success($run);
(new logs_o())->add('cron', 'global', 1, $this->actorUserId() ?? 0, 'CRON_JOB_QUEUED', 'Queued cron task: ' . $task_id);
$response->success($run, 202);
} catch (Throwable $throwable) {
$response->error(['message' => $throwable->getMessage()], 409);
}
services/nginx/app/tests/Unit/Cron/CronRouteWiringTest.php
+3
View File
@@ -9,6 +9,9 @@ it('exposes superuser cron status, history, manual run, and config routes', func
expect($content)->toContain('/superuser/cron/run');
expect($content)->toContain('/superuser/cron/config');
expect($content)->toContain('new cron_scheduler()');
expect($content)->toContain('queueTaskRun(');
expect($content)->toContain('$response->success($run, 202)');
expect($content)->toContain('CRON_JOB_QUEUED');
expect($content)->toContain("requireClassicSuperuserPermission('superuser_cron_view')");
expect($content)->toContain("requireClassicSuperuserPermission('SUPERUSER_RUN_CRON')");
expect($content)->toContain("requireClassicSuperuserPermission('superuser_cron_manage')");
services/nginx/app/tests/Unit/Cron/CronWorkerWiringTest.php
+10
View File
@@ -11,6 +11,7 @@ it('wires cron workers through schema, CLI, scheduler, and superuser routes', fu
expect($schema)->toContain('CREATE TABLE IF NOT EXISTS cron_worker_state');
expect($schema)->toContain('last_heartbeat_at');
expect($schema)->toContain('last_stale_run_count');
expect($schema)->toContain('force_run TINYINT(1) NOT NULL DEFAULT 0');
expect($worker)->toContain('CRON_WORKER_POLL_SECONDS');
expect($worker)->toContain('CRON_WORKER_HEARTBEAT_SECONDS');
@@ -19,6 +20,11 @@ it('wires cron workers through schema, CLI, scheduler, and superuser routes', fu
expect($worker)->toContain('runDue($this->source)');
expect($scheduler)->toContain('function markExpiredRunningRuns');
expect($scheduler)->toContain('function queueTaskRun');
expect($scheduler)->toContain('function queuedRuns');
expect($scheduler)->toContain('function runQueuedRun');
expect($scheduler)->toContain("WHERE status = 'queued'");
expect($scheduler)->toContain("SET status = 'running'");
expect($scheduler)->toContain("r.status = 'timed_out'");
expect($scheduler)->toContain('s.current_run_id = NULL');
@@ -27,6 +33,8 @@ it('wires cron workers through schema, CLI, scheduler, and superuser routes', fu
expect($route)->toContain('/superuser/cron/workers');
expect($route)->toContain('/superuser/cron/workers/deploy');
expect($route)->toContain('queueTaskRun(');
expect($route)->toContain('$response->success($run, 202)');
expect($route)->toContain('superuser_cron_view');
expect($route)->toContain('superuser_cron_manage');
expect($route)->toContain('superuser_coolify_manage');
@@ -34,6 +42,8 @@ it('wires cron workers through schema, CLI, scheduler, and superuser routes', fu
expect($manager)->toContain("private const CRON_WORKER_APP = 'cron'");
expect($manager)->toContain("private const CRON_WORKER_START_COMMAND = 'php index.php run cron-worker'");
expect($manager)->toContain('deployCronWorkerAfterApiDeployment');
expect($manager)->toContain('cronWorkerAutoprovisionRequired');
expect($manager)->toContain('cron_worker_autoprovision_disabled');
expect($manager)->toContain('cron_worker_deploy_failed');
expect($manager)->toContain('auto_deploy = 0');
});
services/nginx/app/tests/Unit/ReleaseManager/ReleaseManagerTest.php
+30
View File
@@ -442,6 +442,36 @@ it('derives cron worker deployment context from the API target without public ro
expect($context)->not->toHaveKey('manual_endpoint_host');
});
it('requires Coolify cron worker autoprovisioning for API deployments by default', function (): void {
$manager = new release_manager();
$enabledMethod = new ReflectionMethod(release_manager::class, 'cronWorkerAutoprovisionEnabled');
$requiredMethod = new ReflectionMethod(release_manager::class, 'cronWorkerAutoprovisionRequired');
$enabledMethod->setAccessible(true);
$requiredMethod->setAccessible(true);
expect($enabledMethod->invoke($manager, ['deploy_context_json' => null]))->toBeTrue();
expect($requiredMethod->invoke($manager, ['deploy_context_json' => null]))->toBeTrue();
$optionalTarget = [
'deploy_context_json' => json_encode([
'cron_worker_autoprovision_required' => false,
]),
];
expect($enabledMethod->invoke($manager, $optionalTarget))->toBeTrue();
expect($requiredMethod->invoke($manager, $optionalTarget))->toBeFalse();
$disabledTarget = [
'deploy_context_json' => json_encode([
'cron_worker_autoprovision' => false,
]),
];
expect($enabledMethod->invoke($manager, $disabledTarget))->toBeFalse();
expect($requiredMethod->invoke($manager, $disabledTarget))->toBeFalse();
$managerSource = file_get_contents(app_path('classes/release_manager.php'));
expect($managerSource)->toContain('Cron worker deployment is required for API deployments');
});
it('builds explicit Coolify application route labels for release API targets', function (): void {
$manager = new release_manager();
$payloadMethod = new ReflectionMethod(release_manager::class, 'releaseCoolifyApplicationRoutePayload');