scheduler = $scheduler ?? new cron_scheduler(); $this->name = $this->stringOption($options, 'name', 'CRON_WORKER_NAME', 'cron-worker'); $this->worker_id = $this->stringOption($options, 'worker_id', 'CRON_WORKER_ID', $this->name); $this->source = $this->stringOption($options, 'source', 'CRON_WORKER_SOURCE', 'coolify_worker'); $this->poll_seconds = $this->intOption($options, 'poll_seconds', 'CRON_WORKER_POLL_SECONDS', 15, 1, 300); $this->heartbeat_seconds = $this->intOption($options, 'heartbeat_seconds', 'CRON_WORKER_HEARTBEAT_SECONDS', 30, 5, 300); $this->max_runtime_seconds = $this->intOption($options, 'max_runtime_seconds', 'CRON_WORKER_MAX_RUNTIME_SECONDS', 0, 0, 86400); } public function run(): int { if (!$this->boolOption('CRON_WORKER_ENABLED', true)) { $this->heartbeat('disabled', 0, 0, null, true); return 0; } $this->registerSignalHandlers(); $started = time(); $this->heartbeat('starting', 0, 0, null, true); while (!$this->should_stop) { $pollStarted = microtime(true); $result = $this->tick(); $this->writeStatusLine($result); if ($this->max_runtime_seconds > 0 && time() - $started >= $this->max_runtime_seconds) { $this->should_stop = true; break; } $this->sleepUntilNextPoll($pollStarted + $this->poll_seconds); } $this->heartbeat('stopped', 0, 0, null, true, true); return 0; } public function tick(): array { $this->heartbeat('running'); $loopStartedAt = date('Y-m-d H:i:s'); $staleRuns = 0; $ran = ['count' => 0, 'ran' => []]; $error = null; $status = 'running'; try { $staleRuns = $this->scheduler->markExpiredRunningRuns(); $ran = $this->scheduler->runDue($this->source); } catch (Throwable $throwable) { $status = 'failed'; $error = $throwable->getMessage(); } $this->heartbeat($status, (int)($ran['count'] ?? 0), $staleRuns, $error, true, false, $loopStartedAt); return [ 'worker_id' => $this->worker_id, 'status' => $status, 'ran' => (int)($ran['count'] ?? 0), 'stale_runs' => $staleRuns, 'error' => $error, ]; } public function listWorkers(): array { cron_schema_bootstrap::ensureTables(); $rows = $this->fetchAll('SELECT * FROM cron_worker_state ORDER BY last_heartbeat_at DESC, worker_id'); $workers = []; foreach ($rows as $row) { $workers[] = $this->publicWorker($row); } return [ 'workers' => $workers, 'summary' => [ 'total' => count($workers), 'running' => count(array_filter($workers, static fn(array $worker): bool => ($worker['status'] ?? '') === 'running')), 'stale' => count(array_filter($workers, static fn(array $worker): bool => (bool)($worker['stale'] ?? false))), ], ]; } private function registerSignalHandlers(): void { if (!function_exists('pcntl_signal')) { return; } if (function_exists('pcntl_async_signals')) { pcntl_async_signals(true); } pcntl_signal(SIGTERM, function (): void { $this->should_stop = true; }); pcntl_signal(SIGINT, function (): void { $this->should_stop = true; }); } private function sleepUntilNextPoll(float $nextPollAt): void { 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'); } } } private function heartbeat( string $status, int $runCount = 0, int $staleRunCount = 0, ?string $error = null, bool $force = false, bool $stopped = false, ?string $loopStartedAt = null ): void { if (!$force && time() - $this->last_heartbeat < $this->heartbeat_seconds) { return; } cron_schema_bootstrap::ensureTables(); $this->last_heartbeat = time(); $now = date('Y-m-d H:i:s'); $workerId = $this->sql($this->worker_id); $name = $this->sql($this->name); $hostname = $this->nullableSql(gethostname() ?: null); $pid = getmypid() ?: 0; $source = $this->sql($this->source); $statusSql = $this->sql($status); $releaseChannelId = $this->nullableInt($this->env('CRON_WORKER_RELEASE_CHANNEL_ID')); $releaseTargetId = $this->nullableInt($this->env('CRON_WORKER_RELEASE_TARGET_ID')); $resourceUuid = $this->nullableSql($this->env('COOLIFY_RESOURCE_UUID') ?: $this->env('CRON_WORKER_COOLIFY_RESOURCE_UUID')); $resourceType = $this->nullableSql($this->env('COOLIFY_RESOURCE_TYPE') ?: $this->env('CRON_WORKER_COOLIFY_RESOURCE_TYPE') ?: 'application'); $commitSha = $this->nullableSql($this->commitSha()); $errorSql = $this->nullableSql($error); $loopStarted = $this->nullableSql($loopStartedAt); $stoppedAt = $stopped ? $this->sql($now) : 'NULL'; $nowSql = $this->sql($now); $this->query( "INSERT INTO cron_worker_state ( 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, 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, NULL, " . ($loopStartedAt !== null ? '1' : '0') . ", $stoppedAt ) ON DUPLICATE KEY UPDATE name = VALUES(name), hostname = VALUES(hostname), pid = VALUES(pid), source = VALUES(source), status = VALUES(status), release_channel_id = VALUES(release_channel_id), release_target_id = VALUES(release_target_id), coolify_resource_uuid = VALUES(coolify_resource_uuid), coolify_resource_type = VALUES(coolify_resource_type), commit_sha = VALUES(commit_sha), poll_seconds = VALUES(poll_seconds), last_run_count = VALUES(last_run_count), 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)" ); } private function publicWorker(array $row): array { $heartbeatAt = (string)($row['last_heartbeat_at'] ?? ''); $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'] ?? ''), 'name' => (string)($row['name'] ?? ''), 'hostname' => $row['hostname'] ?? null, 'pid' => isset($row['pid']) ? (int)$row['pid'] : null, 'source' => (string)($row['source'] ?? ''), 'status' => (string)($row['status'] ?? 'unknown'), 'release_channel_id' => isset($row['release_channel_id']) ? (int)$row['release_channel_id'] : null, 'release_target_id' => isset($row['release_target_id']) ? (int)$row['release_target_id'] : null, 'coolify_resource_uuid' => $row['coolify_resource_uuid'] ?? null, 'coolify_resource_type' => $row['coolify_resource_type'] ?? null, 'commit_sha' => $row['commit_sha'] ?? null, 'poll_seconds' => (int)($row['poll_seconds'] ?? 0), 'last_run_count' => (int)($row['last_run_count'] ?? 0), 'last_stale_run_count' => (int)($row['last_stale_run_count'] ?? 0), 'last_error' => $row['last_error'] ?? null, 'started_at' => $row['started_at'] ?? null, 'last_heartbeat_at' => $heartbeatAt !== '' ? $heartbeatAt : null, '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, ]; } private function writeStatusLine(array $result): void { echo '[' . date('Y-m-d H:i:s') . '][CRON_WORKER] ' . json_encode($result, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) . PHP_EOL; } private function stringOption(array $options, string $key, string $env, string $default): string { $value = trim((string)($options[$key] ?? $this->env($env) ?? '')); return $value !== '' ? $value : $default; } private function intOption(array $options, string $key, string $env, int $default, int $min, int $max): int { $value = (int)($options[$key] ?? $this->env($env) ?? $default); return max($min, min($max, $value)); } private function boolOption(string $env, bool $default): bool { $value = $this->env($env); if ($value === null || trim($value) === '') { return $default; } return self::normalizeBoolean($value); } private function commitSha(): string { foreach (['CRON_WORKER_COMMIT_SHA', 'API_COMMIT_SHA', 'RELEASE_COMMIT_SHA', 'COMMIT_SHA', 'GITHUB_SHA'] as $key) { $value = trim((string)($this->env($key) ?? '')); if ($value !== '') { return $value; } } return ''; } private function env(string $key): ?string { $value = getenv($key); if ($value !== false) { return (string)$value; } return isset($_SERVER[$key]) ? (string)$_SERVER[$key] : null; } private function nullableInt(?string $value): string { $value = trim((string)$value); if ($value === '' || filter_var($value, FILTER_VALIDATE_INT) === false) { return 'NULL'; } return (string)max(0, (int)$value); } private function nullableSql(?string $value): string { $value = $value !== null ? trim($value) : ''; return $value === '' ? 'NULL' : $this->sql($value); } private function fetchAll(string $sql): array { $result = $this->query($sql); if ($result === false || $result === true) { return []; } return $result->fetch_all(MYSQLI_ASSOC); } private function query(string $sql): \mysqli_result|bool { global $db; return $db->query($sql); } private function sql(string $value): string { global $db; return "'" . $db->escape_string($value) . "'"; } }