From 233c44ca44b4f13125ab15a7a159d438d6a45eec Mon Sep 17 00:00:00 2001 From: dougefresh Date: Wed, 3 Jun 2026 19:07:55 +0000 Subject: [PATCH 1/2] fix: drop Bun.serve({fd}) socket activation, use direct bind Bun.serve({fd}) was proposed in oven-sh/bun#2852 but never merged. Connections open but Bun never dispatches to the handler when using fd. Switch to direct UNIX socket binding with Bun.serve({unix}). Systemd manages directory lifecycle via RuntimeDirectory=porter-%i, giving each template instance its own isolated socket directory. Also: use unified fetch handler instead of routes (avoids TS fd+routes type incompatibility), clamp runs limit to 500, and prefer PORTER_SOCKET env var for socket path resolution in the daemon. --- docs/control-plane.md | 37 ++++--- resources/systemd/porter-dev.service | 3 + resources/systemd/porter@.service | 3 +- runtime/src/control-server.ts | 145 +++++++++++---------------- 4 files changed, 87 insertions(+), 101 deletions(-) diff --git a/docs/control-plane.md b/docs/control-plane.md index 32dcd80..8fb70f8 100644 --- a/docs/control-plane.md +++ b/docs/control-plane.md @@ -18,30 +18,37 @@ The control server runs inside the daemon process alongside channels, the inbound/outbound workers, and the scheduler registry. No separate process, no TCP port. -## Socket activation +## Socket lifecycle -In production (systemd template pair): +Systemd manages the socket directory via `RuntimeDirectory=porter-%i` in the +service unit. The daemon binds the socket itself -- systemd socket activation +(`Bun.serve({ fd })`) is not used because the feature never shipped in Bun +(see [oven-sh/bun#2852](https://github.com/oven-sh/bun/pull/2852)). ```ini -# porter@.socket -ListenStream=%t/porter/porter-%i.sock -SocketMode=0600 +# porter@.service +Environment=PORTER_SOCKET=%t/porter-%i/porter.sock +RuntimeDirectory=porter-%i +RuntimeDirectoryMode=0700 ``` -Systemd creates the socket before the service starts and passes it as fd 3. -The daemon detects `LISTEN_FDS=1` and calls `Bun.serve({ fd: 3 })`. -`RemoveOnStop=yes` handles cleanup. +Each template instance gets its own directory: -In development (no socket unit, `bun run`): +``` +porter@projects-me-pi-porter → /run/user/1000/porter-projects-me-pi-porter/porter.sock +porter-dev → /run/user/1000/porter-dev/porter.sock +``` + +In development (`bun run`, no systemd): ```bash -# Daemon binds directly: bun run runtime/src/index.ts --serve -# Socket created at $XDG_RUNTIME_DIR/porter/porter.sock (fallback: ~/.local/state/porter/porter.sock) +# Falls back to $XDG_RUNTIME_DIR/porter/porter.sock +# or ~/.local/state/porter/porter.sock ``` -The CLI locates the socket via `PORTER_SOCKET` env var with a fallback to -`$XDG_RUNTIME_DIR/porter/porter.sock`. +The daemon prefers `PORTER_SOCKET` from the environment; the CLI uses the +same variable with the same fallback. ## API Reference @@ -196,8 +203,8 @@ Worker states: `booting` (initializing Pi session), `ready` (idle), `busy` ## CLI ```bash -# Set socket path for template instances: -export PORTER_SOCKET=$XDG_RUNTIME_DIR/porter/porter-projects-me-pi-porter.sock +# Socket path is set by the systemd service unit; export manually for dev: +export PORTER_SOCKET=$XDG_RUNTIME_DIR/porter-projects-me-pi-porter/porter.sock porter task list porter task get morning-brief diff --git a/resources/systemd/porter-dev.service b/resources/systemd/porter-dev.service index 1878eb1..908408e 100644 --- a/resources/systemd/porter-dev.service +++ b/resources/systemd/porter-dev.service @@ -11,6 +11,9 @@ EnvironmentFile=%h/workspace/.env # Dev unit: run the locally built binary. PATH is for agent subprocesses (bun, tools), not ExecStart. ExecStart=%h/.local/bin/porter --serve Environment=PATH=%h/.local/bin:/usr/local/bin:/usr/bin:/bin +Environment=PORTER_SOCKET=%t/porter-dev/porter.sock +RuntimeDirectory=porter-dev +RuntimeDirectoryMode=0700 Restart=on-failure RestartSec=10s RestartSteps=4 diff --git a/resources/systemd/porter@.service b/resources/systemd/porter@.service index 9cebdf1..844b09f 100644 --- a/resources/systemd/porter@.service +++ b/resources/systemd/porter@.service @@ -11,7 +11,8 @@ EnvironmentFile=%h/%i/.env # Prod unit: packaged install at a fixed path. PATH only affects subprocesses, not ExecStart. ExecStart=/usr/bin/porter --serve Environment=PATH=%h/.local/bin:/usr/local/bin:/usr/bin:/bin -RuntimeDirectory=porter +Environment=PORTER_SOCKET=%t/porter-%i/porter.sock +RuntimeDirectory=porter-%i RuntimeDirectoryMode=0700 Restart=on-failure RestartSec=5 diff --git a/runtime/src/control-server.ts b/runtime/src/control-server.ts index 535c9ba..8e9ef89 100644 --- a/runtime/src/control-server.ts +++ b/runtime/src/control-server.ts @@ -24,10 +24,6 @@ import { parseSessionKey } from './routing/session-key.js'; import type { SchedulerRegistry } from './scheduler/registry.js'; import type { NewScheduledTask } from './scheduler/types.js'; -/** Bun extends Request with route params. Use `any` because TS Record access is `string | undefined`. */ -// eslint-disable-next-line @typescript-eslint/no-explicit-any -type RoutedRequest = Request & { params: any }; - // ---- Types ---- export type ControlServerOptions = { @@ -55,19 +51,6 @@ function errorJson(message: string, status: number): Response { return json({ error: message }, status); } -/** - * Detect systemd socket activation. - * - * systemd sets LISTEN_FDS=1 and LISTEN_PID=. The first - * (and typically only) socket fd is 3 (SD_LISTEN_FDS_START). - */ -function systemdFd(): number | undefined { - if (process.env.LISTEN_PID !== String(process.pid)) return undefined; - const count = Number.parseInt(process.env.LISTEN_FDS ?? '0', 10); - if (count < 1) return undefined; - return 3; -} - // ---- ControlServer ---- export class ControlServer { @@ -86,66 +69,67 @@ export class ControlServer { } async start(): Promise { - const fd = systemdFd(); - - const routes = { - '/api/health': { - GET: () => this.handleHealth(), - }, - '/api/scheduled-tasks': { - GET: () => this.handleListTasks(), - POST: (req: Request) => this.handleCreateTask(req), - }, - '/api/scheduled-tasks/:id': { - GET: (req: Request) => this.handleGetTask(req), - DELETE: (req: Request) => this.handleDeleteTask(req), - }, - '/api/scheduled-tasks/:id/pause': { - POST: (req: Request) => this.handlePauseTask(req), - }, - '/api/scheduled-tasks/:id/resume': { - POST: (req: Request) => this.handleResumeTask(req), - }, - '/api/scheduled-tasks/:id/fire': { - POST: (req: Request) => this.handleFireTask(req), - }, - '/api/scheduled-tasks/:id/runs': { - GET: (req: Request) => this.handleGetTaskRuns(req), - }, - '/api/workers': { - GET: () => this.handleWorkers(), - }, - } as const; + const fetchHandler: (req: Request) => Response | Promise = (req) => { + const url = new URL(req.url); + const id = url.pathname.match(/^\/api\/scheduled-tasks\/([^/]+)/)?.[1]; + + // Health + if (url.pathname === '/api/health' && req.method === 'GET') return this.handleHealth(); + + // Scheduled tasks — collection + if (url.pathname === '/api/scheduled-tasks') { + if (req.method === 'GET') return this.handleListTasks(); + if (req.method === 'POST') return this.handleCreateTask(req); + } + + // Scheduled tasks — single + if (id && url.pathname === `/api/scheduled-tasks/${id}`) { + if (req.method === 'GET') return this.handleGetTask(id); + if (req.method === 'DELETE') return this.handleDeleteTask(id); + } + if (id && url.pathname === `/api/scheduled-tasks/${id}/pause` && req.method === 'POST') + return this.handlePauseTask(id); + if (id && url.pathname === `/api/scheduled-tasks/${id}/resume` && req.method === 'POST') + return this.handleResumeTask(id); + if (id && url.pathname === `/api/scheduled-tasks/${id}/fire` && req.method === 'POST') + return this.handleFireTask(id); + if (id && url.pathname === `/api/scheduled-tasks/${id}/runs` && req.method === 'GET') { + const limit = Number.parseInt(url.searchParams.get('limit') ?? '50', 10); + return this.handleGetTaskRuns(id, limit); + } + + // Workers + if (url.pathname === '/api/workers' && req.method === 'GET') return this.handleWorkers(); + + return errorJson('not found', 404); + }; const onError = (err: Error) => { console.error('[control-server] unhandled error', { error: err }); return errorJson('internal server error', 500); }; - if (fd !== undefined) { - console.log('[control-server] binding on systemd socket activation fd', { fd }); - // Bun types for `fd` + `routes` together are incomplete; cast through any. - // eslint-disable-next-line @typescript-eslint/no-explicit-any - this.server = Bun.serve({ fd, routes, error: onError } as any); - } else { - const runtimeDir = process.env.XDG_RUNTIME_DIR; - const stateDir = process.env.PORTER_STATE_DIR || `${process.env.HOME}/.local/state/porter`; - const base = runtimeDir ? `${runtimeDir}/porter` : stateDir; - const socketPath = `${base}/porter.sock`; - - // Clean up stale socket from a previous run. - try { - await Bun.file(socketPath).delete(); - } catch (err) { - // ENOENT is expected; log anything unexpected. - console.warn('[control-server] stale socket cleanup failed', { path: socketPath, err }); - } + const socketPath = + process.env.PORTER_SOCKET || + (() => { + const runtimeDir = process.env.XDG_RUNTIME_DIR; + const stateDir = process.env.PORTER_STATE_DIR || `${process.env.HOME}/.local/state/porter`; + const base = runtimeDir ? `${runtimeDir}/porter` : stateDir; + return `${base}/porter.sock`; + })(); - this.socketPath = socketPath; - console.log('[control-server] binding on unix socket', { path: socketPath }); - this.server = Bun.serve({ unix: socketPath, routes, error: onError }); + // Clean up stale socket from a previous run. + try { + await Bun.file(socketPath).delete(); + } catch (err) { + // ENOENT is expected; log anything unexpected. + console.warn('[control-server] stale socket cleanup failed', { path: socketPath, err }); } + this.socketPath = socketPath; + console.log('[control-server] binding on unix socket', { path: socketPath }); + this.server = Bun.serve({ unix: socketPath, fetch: fetchHandler, error: onError }); + console.log('[control-server] started'); } @@ -185,8 +169,7 @@ export class ControlServer { return json(tasks); } - private async handleGetTask(req: Request): Promise { - const id = (req as RoutedRequest).params.id as string; + private async handleGetTask(id: string): Promise { const task = await this.taskStore.getById(id); if (!task) return errorJson('task not found', 404); return json(task); @@ -242,9 +225,7 @@ export class ControlServer { } } - private async handleDeleteTask(req: Request): Promise { - const id = (req as RoutedRequest).params.id as string; - + private async handleDeleteTask(id: string): Promise { const task = await this.taskStore.getById(id); if (!task) return errorJson('task not found', 404); @@ -258,8 +239,7 @@ export class ControlServer { return new Response(null, { status: 204 }); } - private async handlePauseTask(req: Request): Promise { - const id = (req as RoutedRequest).params.id as string; + private async handlePauseTask(id: string): Promise { const task = await this.taskStore.getById(id); if (!task) return errorJson('task not found', 404); if (task.status !== 'active') { @@ -273,8 +253,7 @@ export class ControlServer { return json({ id, status: 'paused' }); } - private async handleResumeTask(req: Request): Promise { - const id = (req as RoutedRequest).params.id as string; + private async handleResumeTask(id: string): Promise { const task = await this.taskStore.getById(id); if (!task) return errorJson('task not found', 404); if (task.status !== 'paused') { @@ -287,8 +266,7 @@ export class ControlServer { return json({ id, status: 'active' }); } - private async handleFireTask(req: Request): Promise { - const id = (req as RoutedRequest).params.id as string; + private async handleFireTask(id: string): Promise { const result = await this.scheduler.fireNow(id); if (!result.ok) { return errorJson(result.error ?? 'failed to fire task', result.error === 'task not found' ? 404 : 409); @@ -296,12 +274,9 @@ export class ControlServer { return json({ id, fired: true }); } - private async handleGetTaskRuns(req: Request): Promise { - const id = (req as RoutedRequest).params.id as string; - const url = new URL(req.url); - const raw = Number.parseInt(url.searchParams.get('limit') ?? '50', 10); - const limit = Number.isFinite(raw) && raw > 0 ? Math.min(raw, 500) : 50; - const runs = await this.taskStore.getRuns(id, limit); + private async handleGetTaskRuns(id: string, limit: number): Promise { + const clamped = Number.isFinite(limit) && limit > 0 ? Math.min(limit, 500) : 50; + const runs = await this.taskStore.getRuns(id, clamped); return json(runs); } From b631896166ac4fe1fd4ab1cfe365c1855fd60703 Mon Sep 17 00:00:00 2001 From: dougefresh Date: Wed, 3 Jun 2026 19:15:01 +0000 Subject: [PATCH 2/2] refactor: use Bun.serve routes with typed BunRequest params --- runtime/src/control-server.ts | 126 ++++++++++++++++------------------ 1 file changed, 59 insertions(+), 67 deletions(-) diff --git a/runtime/src/control-server.ts b/runtime/src/control-server.ts index 8e9ef89..172a162 100644 --- a/runtime/src/control-server.ts +++ b/runtime/src/control-server.ts @@ -1,8 +1,8 @@ /** * UNIX socket control plane for the porter daemon. * - * Exposes a REST API over HTTP on a systemd-passed file descriptor - * (socket activation) or a directly-bound UNIX socket in development. + * Exposes a REST API over HTTP on a directly-bound UNIX domain socket. + * Systemd manages socket directory lifecycle via RuntimeDirectory=porter-%i. * * Routes: * GET /api/health @@ -17,6 +17,7 @@ * GET /api/workers */ +import type { BunRequest } from 'bun'; import type { SessionWorkerPool } from './agent/worker-pool.js'; import type { ScheduledTaskStore } from './db/scheduled-task-store.js'; import type { SessionStore } from './db/session-store.js'; @@ -27,14 +28,9 @@ import type { NewScheduledTask } from './scheduler/types.js'; // ---- Types ---- export type ControlServerOptions = { - /** File descriptor from systemd socket activation (fd 3). */ - fd?: number; - /** UNIX socket path for direct binding (development fallback). */ - unix?: string; scheduler: SchedulerRegistry; taskStore: ScheduledTaskStore; sessionStore: SessionStore; - /** Optional: worker pool for /api/workers observability. */ workerPool?: SessionWorkerPool; }; @@ -55,7 +51,7 @@ function errorJson(message: string, status: number): Response { export class ControlServer { private server: ReturnType | null = null; - private socketPath: string | null = null; // for dev-mode cleanup + private socketPath: string | null = null; private scheduler: SchedulerRegistry; private taskStore: ScheduledTaskStore; private sessionStore: SessionStore; @@ -69,46 +65,6 @@ export class ControlServer { } async start(): Promise { - const fetchHandler: (req: Request) => Response | Promise = (req) => { - const url = new URL(req.url); - const id = url.pathname.match(/^\/api\/scheduled-tasks\/([^/]+)/)?.[1]; - - // Health - if (url.pathname === '/api/health' && req.method === 'GET') return this.handleHealth(); - - // Scheduled tasks — collection - if (url.pathname === '/api/scheduled-tasks') { - if (req.method === 'GET') return this.handleListTasks(); - if (req.method === 'POST') return this.handleCreateTask(req); - } - - // Scheduled tasks — single - if (id && url.pathname === `/api/scheduled-tasks/${id}`) { - if (req.method === 'GET') return this.handleGetTask(id); - if (req.method === 'DELETE') return this.handleDeleteTask(id); - } - if (id && url.pathname === `/api/scheduled-tasks/${id}/pause` && req.method === 'POST') - return this.handlePauseTask(id); - if (id && url.pathname === `/api/scheduled-tasks/${id}/resume` && req.method === 'POST') - return this.handleResumeTask(id); - if (id && url.pathname === `/api/scheduled-tasks/${id}/fire` && req.method === 'POST') - return this.handleFireTask(id); - if (id && url.pathname === `/api/scheduled-tasks/${id}/runs` && req.method === 'GET') { - const limit = Number.parseInt(url.searchParams.get('limit') ?? '50', 10); - return this.handleGetTaskRuns(id, limit); - } - - // Workers - if (url.pathname === '/api/workers' && req.method === 'GET') return this.handleWorkers(); - - return errorJson('not found', 404); - }; - - const onError = (err: Error) => { - console.error('[control-server] unhandled error', { error: err }); - return errorJson('internal server error', 500); - }; - const socketPath = process.env.PORTER_SOCKET || (() => { @@ -118,17 +74,50 @@ export class ControlServer { return `${base}/porter.sock`; })(); - // Clean up stale socket from a previous run. try { await Bun.file(socketPath).delete(); } catch (err) { - // ENOENT is expected; log anything unexpected. console.warn('[control-server] stale socket cleanup failed', { path: socketPath, err }); } this.socketPath = socketPath; console.log('[control-server] binding on unix socket', { path: socketPath }); - this.server = Bun.serve({ unix: socketPath, fetch: fetchHandler, error: onError }); + + this.server = Bun.serve({ + unix: socketPath, + routes: { + '/api/health': { + GET: () => this.handleHealth(), + }, + '/api/scheduled-tasks': { + GET: () => this.handleListTasks(), + POST: (req) => this.handleCreateTask(req), + }, + '/api/scheduled-tasks/:id': { + GET: (req) => this.handleGetTask(req), + DELETE: (req) => this.handleDeleteTask(req), + }, + '/api/scheduled-tasks/:id/pause': { + POST: (req) => this.handlePauseTask(req), + }, + '/api/scheduled-tasks/:id/resume': { + POST: (req) => this.handleResumeTask(req), + }, + '/api/scheduled-tasks/:id/fire': { + POST: (req) => this.handleFireTask(req), + }, + '/api/scheduled-tasks/:id/runs': { + GET: (req) => this.handleGetTaskRuns(req), + }, + '/api/workers': { + GET: () => this.handleWorkers(), + }, + }, + error: (err: Error) => { + console.error('[control-server] unhandled error', { error: err }); + return errorJson('internal server error', 500); + }, + }); console.log('[control-server] started'); } @@ -138,8 +127,6 @@ export class ControlServer { await this.server.stop(); this.server = null; } - // In dev mode we own the socket; clean it up. Under systemd socket - // activation RemoveOnStop=yes handles unlink. if (this.socketPath) { const path = this.socketPath; await Bun.file(path) @@ -169,13 +156,13 @@ export class ControlServer { return json(tasks); } - private async handleGetTask(id: string): Promise { - const task = await this.taskStore.getById(id); + private async handleGetTask(req: BunRequest<'/api/scheduled-tasks/:id'>): Promise { + const task = await this.taskStore.getById(req.params.id); if (!task) return errorJson('task not found', 404); return json(task); } - private async handleCreateTask(req: Request): Promise { + private async handleCreateTask(req: BunRequest<'/api/scheduled-tasks'>): Promise { let body: unknown; try { body = await req.json(); @@ -200,7 +187,6 @@ export class ControlServer { scheduleValue: String(input.scheduleValue), }; - // Ensure the session row exists so the FK constraint passes. const parsed = parseSessionKey(newTask.agentSessionKey); if (!parsed) return errorJson('invalid agentSessionKey format', 400); await this.sessionStore.ensureSession(newTask.agentSessionKey, parsed); @@ -225,21 +211,23 @@ export class ControlServer { } } - private async handleDeleteTask(id: string): Promise { + private async handleDeleteTask(req: BunRequest<'/api/scheduled-tasks/:id'>): Promise { + const { id } = req.params; + const task = await this.taskStore.getById(id); if (!task) return errorJson('task not found', 404); - // Pause (disarms the timer) before deleting from DB. await this.taskStore.setStatus(id, 'paused'); await this.scheduler.refresh(id); const deleted = await this.taskStore.delete(id); - if (!deleted) return errorJson('task not found', 404); // race + if (!deleted) return errorJson('task not found', 404); console.log('[control-server] deleted task', { taskId: id }); return new Response(null, { status: 204 }); } - private async handlePauseTask(id: string): Promise { + private async handlePauseTask(req: BunRequest<'/api/scheduled-tasks/:id/pause'>): Promise { + const { id } = req.params; const task = await this.taskStore.getById(id); if (!task) return errorJson('task not found', 404); if (task.status !== 'active') { @@ -247,13 +235,13 @@ export class ControlServer { } await this.taskStore.setStatus(id, 'paused'); - // refresh disarms the paused task's handle since status != 'active' await this.scheduler.refresh(id); console.log('[control-server] paused task', { taskId: id }); return json({ id, status: 'paused' }); } - private async handleResumeTask(id: string): Promise { + private async handleResumeTask(req: BunRequest<'/api/scheduled-tasks/:id/resume'>): Promise { + const { id } = req.params; const task = await this.taskStore.getById(id); if (!task) return errorJson('task not found', 404); if (task.status !== 'paused') { @@ -266,7 +254,8 @@ export class ControlServer { return json({ id, status: 'active' }); } - private async handleFireTask(id: string): Promise { + private async handleFireTask(req: BunRequest<'/api/scheduled-tasks/:id/fire'>): Promise { + const { id } = req.params; const result = await this.scheduler.fireNow(id); if (!result.ok) { return errorJson(result.error ?? 'failed to fire task', result.error === 'task not found' ? 404 : 409); @@ -274,9 +263,12 @@ export class ControlServer { return json({ id, fired: true }); } - private async handleGetTaskRuns(id: string, limit: number): Promise { - const clamped = Number.isFinite(limit) && limit > 0 ? Math.min(limit, 500) : 50; - const runs = await this.taskStore.getRuns(id, clamped); + private async handleGetTaskRuns(req: BunRequest<'/api/scheduled-tasks/:id/runs'>): Promise { + const { id } = req.params; + const url = new URL(req.url); + const raw = Number.parseInt(url.searchParams.get('limit') ?? '50', 10); + const limit = Number.isFinite(raw) && raw > 0 ? Math.min(raw, 500) : 50; + const runs = await this.taskStore.getRuns(id, limit); return json(runs); }