diff --git a/pi-worker/src/health/server.ts b/pi-worker/src/health/server.ts index 68616e1..7b3092c 100644 --- a/pi-worker/src/health/server.ts +++ b/pi-worker/src/health/server.ts @@ -1,14 +1,22 @@ import { logger } from "../logger.js"; +/** Shape of the JSON body returned by `GET /health`. */ interface HealthStatus { + /** Overall status of the process. Set to `"degraded"` on graceful shutdown. */ status: "ok" | "degraded" | "error"; + /** Seconds elapsed since the process started (computed at request time). */ uptime: number; + /** ISO-8601 timestamp of when the process started. */ startedAt: string; + /** `true` once the {@link TaskLoop} has been started successfully. */ taskLoop: boolean; + /** `true` once the {@link ImprovementLoop} has been started successfully. */ improvementLoop: boolean; + /** ISO-8601 timestamp of the most recent Asana poll, if any. */ lastPoll?: string; } +/** In-memory snapshot of the current health state. */ let healthStatus: HealthStatus = { status: "ok", uptime: 0, @@ -17,10 +25,99 @@ let healthStatus: HealthStatus = { improvementLoop: false, }; +/** + * Merges `updates` into the in-memory health snapshot. + * Call this whenever a subsystem changes state (e.g. after starting a loop + * or during graceful shutdown). + * + * @param updates - Partial fields to overwrite in the current health status. + */ export function updateHealth(updates: Partial) { healthStatus = { ...healthStatus, ...updates }; } +/** + * Security headers applied to every response from the health server. + */ +const SECURITY_HEADERS: Record = { + "X-Content-Type-Options": "nosniff", + "X-Frame-Options": "DENY", + "Cache-Control": "no-store", +}; + +/** + * Starts a systemd watchdog heartbeat if WatchdogSec is configured. + * + * sd_notify is done by writing "WATCHDOG=1" to the NOTIFY_SOCKET. + * Bun doesn't have native sd_notify, so we use a simple UDP/unix datagram. + * Fallback: we just curl the health endpoint and if it responds, we're alive. + */ +let watchdogTimer: ReturnType | null = null; + +function startWatchdog(healthPort: number) { + const watchdogUsec = process.env.WATCHDOG_USEC; + if (!watchdogUsec) { + logger.debug("health", "No WATCHDOG_USEC set, skipping systemd watchdog"); + return; + } + + const intervalMs = Math.floor(parseInt(watchdogUsec) / 1000 / 2); // ping at half the deadline + logger.info("health", `Systemd watchdog enabled, pinging every ${intervalMs}ms`); + + watchdogTimer = setInterval(async () => { + try { + // Self-check: hit our own health endpoint + const res = await fetch(`http://127.0.0.1:${healthPort}/health`); + if (!res.ok) { + logger.error("health", `Watchdog self-check failed: ${res.status}`); + return; // Don't notify systemd — let it kill us + } + + const data = (await res.json()) as HealthStatus; + if (data.status !== "ok") { + logger.warn("health", `Watchdog: status is ${data.status}, not notifying systemd`); + return; + } + + // Notify systemd via NOTIFY_SOCKET + const notifySocket = process.env.NOTIFY_SOCKET; + if (notifySocket) { + const { createSocket } = await import("dgram"); + const socketPath = notifySocket.startsWith("@") + ? `\0${notifySocket.slice(1)}` + : notifySocket; + const client = createSocket("unix_dgram" as any); + const msg = Buffer.from("WATCHDOG=1"); + client.send(msg, 0, msg.length, socketPath, () => client.close()); + } + } catch (e) { + logger.error("health", `Watchdog ping failed: ${e}`); + // Don't notify — let systemd restart us + } + }, intervalMs); +} + +export function stopWatchdog() { + if (watchdogTimer) { + clearInterval(watchdogTimer); + watchdogTimer = null; + } +} + +/** + * Starts the HTTP health server on the given `port`. + * + * ### Endpoints + * + * | Method | Path | Response | + * |--------|-----------|----------| + * | GET | `/health` | `200 application/json` — {@link HealthStatus} with live `uptime` | + * | GET | `/` | `200 text/plain` — simple alive message | + * | * | * | `404 Not Found` | + * + * @param port - TCP port to listen on. + * @returns The `Bun.Server` instance (rarely needed by callers). + */ export function startHealthServer(port: number) { const startTime = Date.now(); @@ -29,23 +126,38 @@ export function startHealthServer(port: number) { fetch(req) { const url = new URL(req.url); - if (url.pathname === "/health") { - return Response.json({ - ...healthStatus, - uptime: Math.floor((Date.now() - startTime) / 1000), + // Only allow safe read-only methods on this server. + if (req.method !== "GET" && req.method !== "HEAD") { + return new Response("Method Not Allowed", { + status: 405, + headers: { ...SECURITY_HEADERS, Allow: "GET, HEAD" }, }); } + if (url.pathname === "/health") { + return Response.json( + { + ...healthStatus, + uptime: Math.floor((Date.now() - startTime) / 1000), + }, + { headers: SECURITY_HEADERS } + ); + } + if (url.pathname === "/") { return new Response("Pi Worker is running. GET /health for status.", { - headers: { "Content-Type": "text/plain" }, + headers: { ...SECURITY_HEADERS, "Content-Type": "text/plain" }, }); } - return new Response("Not Found", { status: 404 }); + return new Response("Not Found", { status: 404, headers: SECURITY_HEADERS }); }, }); logger.info("health", `Health server listening on :${port}`); + + // Start systemd watchdog after health server is up + startWatchdog(port); + return server; } diff --git a/pi-worker/src/scheduler/lock.ts b/pi-worker/src/scheduler/lock.ts index 5f8ab49..e0329dd 100644 --- a/pi-worker/src/scheduler/lock.ts +++ b/pi-worker/src/scheduler/lock.ts @@ -1,46 +1,112 @@ import { writeFileSync, unlinkSync, readFileSync, existsSync } from "fs"; import { join } from "path"; +/** + * A simple file-based mutex used to prevent concurrent Pi CLI executions. + * + * The lock file stores the acquisition timestamp (milliseconds since epoch) + * as plain text. Any lock older than {@link STALE_LOCK_MS} is automatically + * treated as stale and removed, ensuring a crashed process can never + * permanently block future runs. + * + * Lock files are stored in `dir` (default `/tmp`) as + * `pi-worker-.lock`. + * + * @example + * ```ts + * const lock = new FileLock("task-loop"); + * if (!lock.acquire()) return; // another execution in progress + * try { + * await doWork(); + * } finally { + * lock.release(); + * } + * ``` + */ export class FileLock { private lockPath: string; + private held = false; + /** Locks older than this (ms) are considered stale and are removed on the next acquire(). */ + private static readonly STALE_LOCK_MS = 15 * 60 * 1000; // 15 minutes + + /** + * @param name - Logical name for this lock (e.g. `"task-loop"`). + * Combined with the prefix to form the lock filename. + * @param dir - Directory in which to create the lock file. Defaults to `/tmp`. + */ constructor(name: string, dir: string = "/tmp") { this.lockPath = join(dir, `pi-worker-${name}.lock`); } + /** + * Attempts to acquire the lock. + * + * If a lock file already exists and is **not** stale, returns `false` + * immediately (non-blocking). Stale lock files are removed and the + * acquisition is retried once. + * + * Uses the `wx` (exclusive create) flag so that two processes racing to + * acquire the lock cannot both succeed. + * + * @returns `true` if the lock was acquired, `false` if it is already held. + */ acquire(): boolean { // Check for stale lock first if (existsSync(this.lockPath)) { try { const content = readFileSync(this.lockPath, "utf-8"); const lockTime = parseInt(content); - if (Date.now() - lockTime > 15 * 60 * 1000) { - // Stale lock, remove it - try { unlinkSync(this.lockPath); } catch {} + if (Date.now() - lockTime > FileLock.STALE_LOCK_MS) { + // Stale lock — remove it + try { unlinkSync(this.lockPath); } catch (e) { + process.stderr.write(`[lock] Failed to remove stale lock ${this.lockPath}: ${e}\n`); + } } else { return false; } - } catch { - // Corrupted lock file, remove it + } catch (e) { + // Corrupted lock file — remove it so we don't get permanently stuck + process.stderr.write(`[lock] Corrupted lock file ${this.lockPath}, removing: ${e}\n`); try { unlinkSync(this.lockPath); } catch {} } } - // Atomic create + // Atomic create — fails with EEXIST if another process won the race try { writeFileSync(this.lockPath, Date.now().toString(), { flag: 'wx' }); + this.held = true; return true; } catch { - return false; // Another process got it first + return false; } } + /** + * Releases the lock by deleting the lock file. + * + * Only deletes if this instance actually holds the lock. Safe to call + * even if the lock is not currently held (no-op in that case). + */ release() { + if (!this.held) return; try { unlinkSync(this.lockPath); - } catch {} + } catch (e) { + // Suppress ENOENT — lock was already cleaned up (e.g. by stale detection) + if ((e as NodeJS.ErrnoException).code !== "ENOENT") { + process.stderr.write(`[lock] Failed to release lock ${this.lockPath}: ${e}\n`); + } + } finally { + this.held = false; + } } + /** + * Returns `true` if a lock file exists on disk (regardless of whether it + * is stale). Useful for diagnostics; prefer {@link acquire} for actual + * mutual-exclusion logic. + */ isLocked(): boolean { return existsSync(this.lockPath); }