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

Fix cron worker minute cadence (#324)

Browse Source 
Replace drifting cron loops with persistent cadence-tracked workers and CI-verifiable Compose wiring.
This commit is contained in:
Jeppe B
2026-07-27 18:16:09 +02:00
committed by GitHub
parent 6e24718c1f
commit 8d8f0eccce
9 changed files with 156 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docker-compose.example.yml
+1 -1
View File
@@ -129,7 +129,7 @@ services:
- redis
- mysql
- edge-broker
command: ["sh", "-c", "while true; do php index.php run cron; sleep 60; done"]
command: ["php", "index.php", "run", "cron-worker"]
env_file:
- .env.example
environment:
docker-compose.prod.standalone.yml
+1 -1
View File
@@ -367,7 +367,7 @@ services:
depends_on:
- redis
- edge-broker
command: ["sh", "-c", "while true; do php index.php run cron; sleep 60; done"]
command: ["php", "index.php", "run", "cron-worker"]
env_file:
- .env
environment:
docker-compose.yml
+1 -1
View File
@@ -425,7 +425,7 @@ services:
depends_on:
- redis
- edge-broker
command: ["sh", "-c", "while true; do php index.php run cron; sleep 60; done"]
command: ["php", "index.php", "run", "cron-worker"]
env_file:
- .env
environment:
scripts/php-ci-test.sh
+3
View File
@@ -146,6 +146,9 @@ tar \
-cf - \
Dockerfile \
Dockerfile.coolify-api \
docker-compose.yml \
docker-compose.example.yml \
docker-compose.prod.standalone.yml \
services/php/Dockerfile \
services/php/php-fpm-pool.conf \
| docker compose $compose_files exec -T php1 tar --no-same-owner -C /var/www/repo-root -xf -
services/nginx/app/classes/cron_scheduler.php
+27 -6
View File
@@ -227,10 +227,15 @@ class cron_scheduler
"UPDATE cron_task_state SET current_run_id = $run_id WHERE task_id = " . $this->sql($definition->id)
);
return $this->executeClaimedRun($definition, $run_id, $started);
return $this->executeClaimedRun($definition, $run_id, $started, $scheduled_for);
}
private function executeClaimedRun(cron_task_definition $definition, int $run_id, float $started): array
private function executeClaimedRun(
cron_task_definition $definition,
int $run_id,
float $started,
?string $scheduled_for = null
): array
{
$status = 'succeeded';
$summary = [];
@@ -272,7 +277,7 @@ class cron_scheduler
$completed_at = date('Y-m-d H:i:s', (int)$completed);
$this->completeRun($run_id, $status, $completed_at, $duration_ms, $summary, $error_message);
$this->releaseLock($definition, $status, $error_message, $completed_at);
$this->releaseLock($definition, $status, $error_message, $completed_at, $scheduled_for);
return $this->publicRun($this->fetchOne("SELECT * FROM cron_task_runs WHERE id = $run_id") ?? []);
}
@@ -441,7 +446,12 @@ class cron_scheduler
"UPDATE cron_task_state SET current_run_id = $run_id WHERE task_id = " . $this->sql($definition->id)
);
return $this->executeClaimedRun($definition, $run_id, $started);
return $this->executeClaimedRun(
$definition,
$run_id,
$started,
isset($queuedRun['scheduled_for']) ? (string)$queuedRun['scheduled_for'] : null
);
}
private function skipQueuedRun(int $run_id, string $message, ?cron_task_definition $definition = null): void
@@ -479,7 +489,13 @@ class cron_scheduler
);
}
private function releaseLock(cron_task_definition $definition, string $status, ?string $error_message, string $completed_at): void
private function releaseLock(
cron_task_definition $definition,
string $status,
?string $error_message,
string $completed_at,
?string $scheduled_for = null
): void
{
$state = $this->fetchOne("SELECT schedule_json FROM cron_task_state WHERE task_id = " . $this->sql($definition->id));
$schedule = $this->decodeJson($state['schedule_json'] ?? null);
@@ -487,7 +503,12 @@ class cron_scheduler
$schedule = $definition->schedule;
}
$nextRunAt = cron_schedule::nextRunAt($schedule, $completed_at, time());
// Automatic runs stay anchored to their intended schedule slot. Anchoring
// to completion time causes every task to drift by its execution time.
$scheduleAnchor = $scheduled_for !== null && strtotime($scheduled_for) !== false
? $scheduled_for
: $completed_at;
$nextRunAt = cron_schedule::nextRunAt($schedule, $scheduleAnchor, time());
if ($status !== 'succeeded') {
$retrySeconds = min(300, max(60, (int)$schedule['seconds']));
$nextRunAt = date('Y-m-d H:i:s', time() + $retrySeconds);
services/nginx/app/classes/cron_schema_bootstrap.php
+4
View File
@@ -84,6 +84,8 @@ class cron_schema_bootstrap
last_heartbeat_at DATETIME NULL,
last_loop_started_at DATETIME NULL,
last_loop_finished_at DATETIME NULL,
last_loop_gap_seconds INT UNSIGNED NULL,
consecutive_minute_loops INT UNSIGNED NOT NULL DEFAULT 0,
stopped_at DATETIME NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
@@ -93,6 +95,8 @@ class cron_schema_bootstrap
KEY idx_cron_worker_state_coolify_resource (coolify_resource_uuid)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"
);
self::ensureColumn('cron_worker_state', 'last_loop_gap_seconds', 'INT UNSIGNED NULL AFTER last_loop_finished_at');
self::ensureColumn('cron_worker_state', 'consecutive_minute_loops', 'INT UNSIGNED NOT NULL DEFAULT 0 AFTER last_loop_gap_seconds');
self::$initialized = true;
}
services/nginx/app/classes/cron_worker.php
+42 -9
View File
@@ -39,6 +39,7 @@ class cron_worker
$this->heartbeat('starting', 0, 0, null, true);
while (!$this->should_stop) {
$pollStarted = microtime(true);
$result = $this->tick();
$this->writeStatusLine($result);
@@ -47,7 +48,7 @@ class cron_worker
break;
}
$this->sleepUntilNextPoll();
$this->sleepUntilNextPoll($pollStarted + $this->poll_seconds);
}
$this->heartbeat('stopped', 0, 0, null, true, true);
@@ -119,13 +120,14 @@ class cron_worker
});
}
private function sleepUntilNextPoll(): void
private function sleepUntilNextPoll(float $nextPollAt): void
{
$remaining = $this->poll_seconds;
while ($remaining > 0 && !$this->should_stop) {
$sleep = min(1, $remaining);
sleep($sleep);
$remaining -= $sleep;
while (!$this->should_stop) {
$remaining = $nextPollAt - microtime(true);
if ($remaining <= 0) {
return;
}
usleep((int)(min(1.0, $remaining) * 1000000));
if (time() - $this->last_heartbeat >= $this->heartbeat_seconds) {
$this->heartbeat('running');
}
@@ -169,12 +171,12 @@ class cron_worker
worker_id, name, hostname, pid, source, status, release_channel_id, release_target_id,
coolify_resource_uuid, coolify_resource_type, commit_sha, poll_seconds, last_run_count,
last_stale_run_count, last_error, started_at, last_heartbeat_at, last_loop_started_at,
last_loop_finished_at, stopped_at
last_loop_finished_at, last_loop_gap_seconds, consecutive_minute_loops, stopped_at
) VALUES (
$workerId, $name, $hostname, $pid, $source, $statusSql, $releaseChannelId, $releaseTargetId,
$resourceUuid, $resourceType, $commitSha, $this->poll_seconds, $runCount,
$staleRunCount, $errorSql, $nowSql, $nowSql, $loopStarted,
$nowSql, $stoppedAt
$nowSql, NULL, " . ($loopStartedAt !== null ? '1' : '0') . ", $stoppedAt
)
ON DUPLICATE KEY UPDATE
name = VALUES(name),
@@ -192,6 +194,17 @@ class cron_worker
last_stale_run_count = VALUES(last_stale_run_count),
last_error = VALUES(last_error),
last_heartbeat_at = VALUES(last_heartbeat_at),
last_loop_gap_seconds = CASE
WHEN VALUES(last_loop_started_at) IS NULL OR last_loop_started_at IS NULL THEN last_loop_gap_seconds
ELSE GREATEST(0, TIMESTAMPDIFF(SECOND, last_loop_started_at, VALUES(last_loop_started_at)))
END,
consecutive_minute_loops = CASE
WHEN VALUES(last_loop_started_at) IS NULL THEN consecutive_minute_loops
WHEN last_loop_started_at IS NULL THEN 1
WHEN TIMESTAMPDIFF(SECOND, last_loop_started_at, VALUES(last_loop_started_at)) BETWEEN 0 AND 60
THEN consecutive_minute_loops + 1
ELSE 1
END,
last_loop_started_at = COALESCE(VALUES(last_loop_started_at), last_loop_started_at),
last_loop_finished_at = VALUES(last_loop_finished_at),
stopped_at = VALUES(stopped_at)"
@@ -204,6 +217,17 @@ class cron_worker
$heartbeatTs = strtotime($heartbeatAt);
$threshold = max(60, ((int)($row['poll_seconds'] ?? 15) * 4) + 30);
$age = $heartbeatTs !== false ? max(0, time() - $heartbeatTs) : null;
$loopStartedAt = (string)($row['last_loop_started_at'] ?? '');
$loopStartedTs = strtotime($loopStartedAt);
$loopAge = $loopStartedTs !== false ? max(0, time() - $loopStartedTs) : null;
$loopGap = isset($row['last_loop_gap_seconds']) ? (int)$row['last_loop_gap_seconds'] : null;
$consecutiveMinuteLoops = (int)($row['consecutive_minute_loops'] ?? 0);
$minuteCadenceVerified = ($row['status'] ?? '') === 'running'
&& $loopAge !== null
&& $loopAge <= 60
&& $loopGap !== null
&& $loopGap <= 60
&& $consecutiveMinuteLoops >= 2;
return [
'worker_id' => (string)($row['worker_id'] ?? ''),
@@ -226,6 +250,15 @@ class cron_worker
'last_heartbeat_age_seconds' => $age,
'last_loop_started_at' => $row['last_loop_started_at'] ?? null,
'last_loop_finished_at' => $row['last_loop_finished_at'] ?? null,
'last_loop_age_seconds' => $loopAge,
'last_loop_gap_seconds' => $loopGap,
'consecutive_minute_loops' => $consecutiveMinuteLoops,
'minute_cadence' => [
'verified' => $minuteCadenceVerified,
'maximum_gap_seconds' => 60,
'last_gap_seconds' => $loopGap,
'consecutive_loops' => $consecutiveMinuteLoops,
],
'stopped_at' => $row['stopped_at'] ?? null,
'stale' => $age === null || $age > $threshold,
'stale_after_seconds' => $threshold,
services/nginx/app/tests/Unit/Cron/CronTaskRegistryTest.php
+18
View File
@@ -54,6 +54,24 @@ it('normalizes and advances interval schedules without tight loops', function ()
expect($next)->toBe('2026-07-09 12:15:00');
});
it('keeps minute schedules anchored to intended slots when execution finishes late', function (): void {
$schedule = ['type' => 'interval', 'seconds' => 60];
$first = cron_schedule::nextRunAt(
$schedule,
'2026-07-27 12:00:00',
strtotime('2026-07-27 12:00:47')
);
$second = cron_schedule::nextRunAt(
$schedule,
$first,
strtotime('2026-07-27 12:01:52')
);
expect($first)->toBe('2026-07-27 12:01:00');
expect($second)->toBe('2026-07-27 12:02:00');
});
it('rejects unsafe cron intervals', function (): void {
expect(fn() => cron_schedule::normalize(['type' => 'interval', 'seconds' => 5]))
->toThrow(InvalidArgumentException::class);
services/nginx/app/tests/Unit/Cron/CronWorkerWiringTest.php
+59 -6
View File
@@ -1,12 +1,22 @@
<?php
$cronAppRoot = dirname(__DIR__, 3);
require_once $cronAppRoot . '/classes/cron_worker.php';
it('wires cron workers through schema, CLI, scheduler, and superuser routes', function (): void {
$schema = file_get_contents(app_path('classes/cron_schema_bootstrap.php'));
$worker = file_get_contents(app_path('classes/cron_worker.php'));
$scheduler = file_get_contents(app_path('classes/cron_scheduler.php'));
$cli = file_get_contents(app_path('cli.php'));
$route = file_get_contents(app_path('routes/cronRoute.php'));
$manager = file_get_contents(app_path('classes/release_manager.php'));
$appRoot = dirname(__DIR__, 3);
$repoRoot = getenv('PLENO_REPO_ROOT_FOR_TESTS') ?: dirname($appRoot, 3);
$schema = file_get_contents($appRoot . '/classes/cron_schema_bootstrap.php');
$worker = file_get_contents($appRoot . '/classes/cron_worker.php');
$scheduler = file_get_contents($appRoot . '/classes/cron_scheduler.php');
$cli = file_get_contents($appRoot . '/cli.php');
$route = file_get_contents($appRoot . '/routes/cronRoute.php');
$manager = file_get_contents($appRoot . '/classes/release_manager.php');
$composeFiles = [
$repoRoot . '/docker-compose.yml',
$repoRoot . '/docker-compose.example.yml',
$repoRoot . '/docker-compose.prod.standalone.yml',
];
expect($schema)->toContain('CREATE TABLE IF NOT EXISTS cron_worker_state');
expect($schema)->toContain('last_heartbeat_at');
@@ -18,6 +28,9 @@ it('wires cron workers through schema, CLI, scheduler, and superuser routes', fu
expect($worker)->toContain('CRON_WORKER_RELEASE_TARGET_ID');
expect($worker)->toContain('markExpiredRunningRuns');
expect($worker)->toContain('runDue($this->source)');
expect($worker)->toContain('$pollStarted + $this->poll_seconds');
expect($worker)->toContain("'minute_cadence' => [");
expect($worker)->toContain("'maximum_gap_seconds' => 60");
expect($scheduler)->toContain('function markExpiredRunningRuns');
expect($scheduler)->toContain('function queueTaskRun');
@@ -27,6 +40,8 @@ it('wires cron workers through schema, CLI, scheduler, and superuser routes', fu
expect($scheduler)->toContain("SET status = 'running'");
expect($scheduler)->toContain("r.status = 'timed_out'");
expect($scheduler)->toContain('s.current_run_id = NULL');
expect($scheduler)->toContain('$this->executeClaimedRun($definition, $run_id, $started, $scheduled_for)');
expect($scheduler)->toContain('$scheduleAnchor = $scheduled_for');
expect($cli)->toContain("case 'cron-worker'");
expect($cli)->toContain('new \\classes\\cron_worker()');
@@ -50,4 +65,42 @@ it('wires cron workers through schema, CLI, scheduler, and superuser routes', fu
expect($manager)->toContain('cron_worker_autoprovision_disabled');
expect($manager)->toContain('cron_worker_deploy_failed');
expect($manager)->toContain('auto_deploy = 0');
foreach ($composeFiles as $composeFile) {
$compose = file_get_contents($composeFile);
expect(str_contains($compose, 'command: ["php", "index.php", "run", "cron-worker"]'))->toBeTrue();
expect(str_contains($compose, 'while true; do php index.php run cron; sleep 60; done'))->toBeFalse();
}
});
it('reports consecutive scheduler loops as once-per-minute execution proof', function (): void {
$reflection = new ReflectionClass(\classes\cron_worker::class);
$worker = $reflection->newInstanceWithoutConstructor();
$publicWorker = $reflection->getMethod('publicWorker');
$now = time();
$row = [
'worker_id' => 'proof-worker',
'name' => 'Proof worker',
'source' => 'test',
'status' => 'running',
'poll_seconds' => 15,
'last_heartbeat_at' => date('Y-m-d H:i:s', $now - 5),
'last_loop_started_at' => date('Y-m-d H:i:s', $now - 10),
'last_loop_finished_at' => date('Y-m-d H:i:s', $now - 9),
'last_loop_gap_seconds' => 15,
'consecutive_minute_loops' => 4,
];
$result = $publicWorker->invoke($worker, $row);
expect($result['minute_cadence'])->toBe([
'verified' => true,
'maximum_gap_seconds' => 60,
'last_gap_seconds' => 15,
'consecutive_loops' => 4,
]);
$row['last_loop_gap_seconds'] = 61;
$result = $publicWorker->invoke($worker, $row);
expect($result['minute_cadence']['verified'])->toBeFalse();
});