Something went wrong. Try again.
Monorepo for Aesthetic.Computer aesthetic.computer
Something went wrong. Try again.
JavaScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465#!/usr/bin/env node// AC fleet worker — a deliberately narrow, tailnet-only compute endpoint.//// v1 accepts one typed workload: ffmpeg. It never accepts a shell command.// Inputs are content-addressed uploads, arguments may reference only named// inputs and one declared output, and every job runs in its own process group.
import { createHash, randomBytes, timingSafeEqual } from "node:crypto";import { createReadStream, createWriteStream, existsSync, mkdirSync, readFileSync, readdirSync, renameSync, rmSync, statfsSync, statSync, writeFileSync } from "node:fs";import { createServer } from "node:http";import { homedir, hostname, loadavg, freemem, totalmem, cpus } from "node:os";import { basename, dirname, join } from "node:path";import { spawn, spawnSync } from "node:child_process";
const HOME = homedir();const PORT = positiveInt(process.env.AC_FLEET_WORKER_PORT, 5263);const BIND = process.env.AC_FLEET_WORKER_BIND || "127.0.0.1";const ROLE = process.env.AC_FLEET_WORKER_ROLE || "balanced";const HOST_NAME = process.env.AC_FLEET_WORKER_NAME || hostname().split(".")[0];const ROOT = process.env.AC_FLEET_WORKER_STATE || join(HOME, ".local", "share", "ac-fleet-worker");const TOKEN_FILE = process.env.AC_FLEET_WORKER_TOKEN_FILE || join(HOME, ".config", "ac-fleet-worker", "token");const GUARD_LATEST = process.env.AC_FLEET_GUARD_LATEST || join(HOME, ".local", "share", "slab", "performance", "latest.txt");const GUARD_PRESSURE = process.env.AC_FLEET_GUARD_PRESSURE || join(HOME, ".local", "share", "slab", "performance", "pressure-active");const MISSION_FILE = process.env.AC_FLEET_MISSION_FILE || join(HOME, ".local", "share", "desktop-badge", "mission.json");const MAX_BODY = positiveInt(process.env.AC_FLEET_WORKER_MAX_BODY, 1_048_576);const MAX_INPUT = positiveInt(process.env.AC_FLEET_WORKER_MAX_INPUT, 50 * 1024 * 1024 * 1024);const MAX_JOBS = positiveInt(process.env.AC_FLEET_WORKER_CONCURRENCY, 1);const FFMPEG = process.env.AC_FLEET_WORKER_FFMPEG || "/opt/homebrew/bin/ffmpeg";const INPUT_DIR = join(ROOT, "inputs");const JOB_DIR = join(ROOT, "jobs");const LOG_DIR = join(ROOT, "logs");const JOB_ID = /^[a-z0-9][a-z0-9-]{5,80}$/;const SHA256 = /^[a-f0-9]{64}$/;const SAFE_NAME = /^[A-Za-z0-9][A-Za-z0-9._-]{0,119}$/;
for (const dir of [ROOT, INPUT_DIR, JOB_DIR, LOG_DIR]) mkdirSync(dir, { recursive: true });if (!existsSync(TOKEN_FILE)) { console.error(`fleet worker token is missing: ${TOKEN_FILE}`); process.exit(78);}const TOKEN = readFileSync(TOKEN_FILE, "utf8").trim();if (TOKEN.length < 32) { console.error("fleet worker token must contain at least 32 characters"); process.exit(78);}if (!existsSync(FFMPEG)) { console.error(`ffmpeg is missing: ${FFMPEG}`); process.exit(69);}
const jobs = new Map();const queue = [];const active = new Map();loadJobs();
function positiveInt(value, fallback) { const n = Number(value); return Number.isFinite(n) && n > 0 ? Math.floor(n) : fallback;}
function now() { return new Date().toISOString(); }function inputPath(sha) { return join(INPUT_DIR, sha); }function jobPath(id) { return join(JOB_DIR, `${id}.json`); }function jobWorkDir(id) { return join(JOB_DIR, id); }
function publicJob(job) { return { id: job.id, type: job.type, host: HOST_NAME, status: job.status, stage: job.stage, progress: job.progress, createdAt: job.createdAt, startedAt: job.startedAt || null, updatedAt: job.updatedAt, finishedAt: job.finishedAt || null, exitCode: job.exitCode ?? null, error: job.error || null, output: job.output, outputBytes: job.outputBytes || null, outputSha256: job.outputSha256 || null, origin: job.origin || null, };}
function persist(job) { job.updatedAt = now(); writeFileSync(jobPath(job.id), JSON.stringify(job, null, 2) + "\n", { mode: 0o600 });}
function loadJobs() { for (const file of readdirSync(JOB_DIR)) { if (!file.endsWith(".json")) continue; try { const job = JSON.parse(readFileSync(join(JOB_DIR, file), "utf8")); if (!JOB_ID.test(job.id)) continue; if (["queued", "running"].includes(job.status)) { job.status = "failed"; job.stage = "interrupted"; job.error = "worker restarted before the job completed"; job.finishedAt = now(); persist(job); } jobs.set(job.id, job); } catch {} }}
function parseGuard() { const values = {}; try { for (const line of readFileSync(GUARD_LATEST, "utf8").split("\n")) { for (const match of line.matchAll(/(?:^|\s)([A-Za-z0-9_]+)=([^\s;]+)/g)) { values[match[1]] = match[2]; } } } catch {} return values;}
function capacity() { const guard = parseGuard(); const cores = cpus().length || 1; const load1 = Number(guard.load1 ?? loadavg()[0]) || 0; const freePct = Number(guard.free_pct ?? macMemoryFreePct() ?? (freemem() / Math.max(1, totalmem()) * 100)) || 0; const cpuPct = Math.max(0, Math.min(100, load1 / cores * 100)); const pressure = existsSync(GUARD_PRESSURE) || guard.pressure === "1"; let diskUsedPct = 0; try { const disk = statfsSync(ROOT); const blocks = Number(disk.blocks) || 0; const available = Number(disk.bavail) || 0; diskUsedPct = blocks > 0 ? Math.max(0, Math.min(100, (blocks - available) / blocks * 100)) : 0; } catch {} const roleThreshold = ROLE === "interactive" ? { free: 32, load: 0.72 } : ROLE === "light" ? { free: 24, load: 0.95 } : { free: 18, load: 1.20 }; const headroom = !pressure && freePct >= roleThreshold.free && load1 <= cores * roleThreshold.load; return { cpuPct, load1, cores, memoryUsedPct: Math.max(0, Math.min(100, 100 - freePct)), memoryFreePct: freePct, memoryTotalGB: totalmem() / 1_073_741_824, diskUsedPct, pressure, pressureReason: guard.reason || (pressure ? "guard" : "none"), headroom, accepting: headroom && active.size < MAX_JOBS, };}
function macMemoryFreePct() { if (process.platform !== "darwin") return null; const result = spawnSync("/usr/bin/memory_pressure", [], { encoding: "utf8", timeout: 1500 }); const match = result.stdout?.match(/System-wide memory free percentage:\s*([0-9.]+)%/); return match ? Number(match[1]) : null;}
function health() { const cap = capacity(); return { ok: true, service: "ac-fleet-worker", version: 1, host: HOST_NAME, role: ROLE, bind: BIND, capabilities: ["ffmpeg-render"], concurrency: MAX_JOBS, active: [...active.keys()], queued: queue.length, ...cap, checkedAt: now(), };}
function authorized(req) { const value = req.headers.authorization || ""; const supplied = value.startsWith("Bearer ") ? value.slice(7) : ""; if (supplied.length !== TOKEN.length) return false; const a = Buffer.from(supplied), b = Buffer.from(TOKEN); return a.length === b.length && timingSafeEqual(a, b);}
function json(res, status, value) { const body = JSON.stringify(value) + "\n"; res.writeHead(status, { "content-type": "application/json", "content-length": Buffer.byteLength(body), "cache-control": "no-store" }); res.end(body);}
async function readJson(req) { const chunks = []; let size = 0; for await (const chunk of req) { size += chunk.length; if (size > MAX_BODY) throw new Error("request body is too large"); chunks.push(chunk); } try { return JSON.parse(Buffer.concat(chunks).toString("utf8")); } catch { throw new Error("request body is not valid JSON"); }}
function cleanName(value, label) { const name = String(value || ""); if (!SAFE_NAME.test(name) || basename(name) !== name) throw new Error(`${label} has an unsafe name`); return name;}
function validateSpec(spec) { if (!spec || spec.type !== "ffmpeg") throw new Error("only type=ffmpeg is supported"); if (!Array.isArray(spec.inputs) || spec.inputs.length < 1 || spec.inputs.length > 16) throw new Error("inputs must contain 1-16 items"); const names = new Set(); const inputs = spec.inputs.map((item) => { const name = cleanName(item?.name, "input"); const sha256 = String(item?.sha256 || "").toLowerCase(); if (names.has(name)) throw new Error(`duplicate input name: ${name}`); if (!SHA256.test(sha256)) throw new Error(`invalid sha256 for ${name}`); if (!existsSync(inputPath(sha256))) throw new Error(`input is not uploaded: ${name}`); names.add(name); return { name, sha256 }; }); const output = cleanName(spec.output, "output"); if (!Array.isArray(spec.args) || spec.args.length < 2 || spec.args.length > 160) throw new Error("args must contain 2-160 tokens"); let sawInput = false, sawOutput = false; const args = spec.args.map((raw) => { const token = String(raw); if (!token || token.length > 4096 || /[\0\r\n]/.test(token)) throw new Error("ffmpeg argument contains invalid characters"); if (token === "@output") { sawOutput = true; return token; } if (token.startsWith("@input:")) { const name = token.slice(7); if (!names.has(name)) throw new Error(`argument references an unknown input: ${name}`); sawInput = true; return token; } if (token.includes("..") || token.startsWith("/") || /(?:^|[=,])(https?|tcp|udp|file|concat):/i.test(token)) { throw new Error(`ffmpeg argument is outside the typed sandbox: ${token.slice(0, 120)}`); } if (["-filter_script", "-filter_complex_script", "-attach", "-dump_attachment"].includes(token)) { throw new Error(`ffmpeg option is not allowed: ${token}`); } return token; }); if (!sawInput) throw new Error("args must reference at least one @input:name"); if (!sawOutput || args.filter((x) => x === "@output").length !== 1) throw new Error("args must reference @output exactly once"); const expectedDurationSeconds = Number(spec.expectedDurationSeconds); return { type: "ffmpeg", inputs, output, args, expectedDurationSeconds: Number.isFinite(expectedDurationSeconds) && expectedDurationSeconds > 0 ? expectedDurationSeconds : null, origin: spec.origin && typeof spec.origin === "object" ? { host: String(spec.origin.host || "").slice(0, 80), handle: String(spec.origin.handle || "").slice(0, 160), } : null, };}
function createJob(spec) { const id = `ffmpeg-${Date.now().toString(36)}-${randomBytes(4).toString("hex")}`; const job = { id, ...spec, status: "queued", stage: "queued", progress: 0, createdAt: now(), updatedAt: now() }; mkdirSync(jobWorkDir(id), { recursive: true }); persist(job); jobs.set(id, job); queue.push(id); pump(); return job;}
function resolvedArgs(job) { const work = jobWorkDir(job.id); const paths = new Map(job.inputs.map((item) => [item.name, inputPath(item.sha256)])); return job.args.map((token) => token === "@output" ? join(work, job.output) : token.startsWith("@input:") ? paths.get(token.slice(7)) : token);}
function writeMission(job, detail) { let current = null; try { current = JSON.parse(readFileSync(MISSION_FILE, "utf8")); } catch {} if (current?.agent && current.agent !== "fleet-worker") return; mkdirSync(dirname(MISSION_FILE), { recursive: true }); const status = job.status === "success" ? "done" : job.status === "failed" ? "pending" : "in_progress"; const percent = Number.isFinite(job.progress) ? ` · ${Math.round(job.progress * 100)}%` : ""; const doc = { mission: `fleet ${job.type} on ${HOST_NAME}`, emoji: "🧱", agent: "fleet-worker", updatedAt: now(), bounded: !!job.expectedDurationSeconds, progress: job.progress, items: [{ text: `${detail}${percent}`, status }], }; writeFileSync(MISSION_FILE, JSON.stringify(doc, null, 2) + "\n", { mode: 0o600 });}
function clearMissionIfOwned() { setTimeout(() => { try { const current = JSON.parse(readFileSync(MISSION_FILE, "utf8")); if (current?.agent === "fleet-worker") rmSync(MISSION_FILE, { force: true }); } catch {} }, 30_000).unref();}
function pump() { while (active.size < MAX_JOBS && queue.length) { if (!capacity().headroom) return; const id = queue.shift(); const job = jobs.get(id); if (job?.status === "queued") runJob(job); }}
function runJob(job) { job.status = "running"; job.stage = "ffmpeg"; job.startedAt = now(); persist(job); writeMission(job, job.output); const logPath = join(LOG_DIR, `${job.id}.log`); const log = createWriteStream(logPath, { flags: "a", mode: 0o600 }); const ffmpegArgs = ["-nostdin", "-y", "-hide_banner", "-progress", "pipe:1", "-nostats", ...resolvedArgs(job)]; const launcher = existsSync("/usr/sbin/taskpolicy") ? "/usr/sbin/taskpolicy" : FFMPEG; const args = launcher === FFMPEG ? ffmpegArgs : ["-c", "utility", "/usr/bin/nice", "-n", "10", FFMPEG, ...ffmpegArgs]; const proc = spawn(launcher, args, { cwd: jobWorkDir(job.id), detached: true, stdio: ["ignore", "pipe", "pipe"], env: { ...process.env, PATH: "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin" }, }); active.set(job.id, proc); let progressBuffer = ""; proc.stdout.on("data", (chunk) => { log.write(chunk); progressBuffer += chunk.toString("utf8"); const lines = progressBuffer.split("\n"); progressBuffer = lines.pop() || ""; for (const line of lines) { const [key, raw] = line.trim().split("=", 2); if ((key === "out_time_us" || key === "out_time_ms") && job.expectedDurationSeconds) { const seconds = Number(raw) / 1_000_000; if (Number.isFinite(seconds)) job.progress = Math.max(job.progress, Math.min(0.99, seconds / job.expectedDurationSeconds)); } if (key === "progress") { if (raw === "end") job.progress = 1; persist(job); writeMission(job, job.output); } } }); proc.stderr.on("data", (chunk) => log.write(chunk)); proc.on("error", (error) => { void finishJob(job, null, error); }); proc.on("close", (code) => { void finishJob(job, code, null); });}
async function finishJob(job, code, spawnError) { if (!active.has(job.id)) return; active.delete(job.id); job.exitCode = Number.isInteger(code) ? code : null; job.finishedAt = now(); const output = join(jobWorkDir(job.id), job.output); if (!spawnError && code === 0 && existsSync(output) && statSync(output).isFile()) { job.status = "success"; job.stage = "done"; job.progress = 1; job.outputBytes = statSync(output).size; job.outputSha256 = await shaFile(output); } else { job.status = job.status === "cancelled" ? "cancelled" : "failed"; job.stage = job.status; job.error = spawnError?.message || `ffmpeg exited with code ${code}`; } persist(job); writeMission(job, job.output); clearMissionIfOwned(); pump();}
function shaFile(path) { return new Promise((resolveHash, rejectHash) => { const hash = createHash("sha256"); const stream = createReadStream(path); stream.on("data", (chunk) => hash.update(chunk)); stream.on("error", rejectHash); stream.on("end", () => resolveHash(hash.digest("hex"))); });}
function cancelJob(job) { if (job.status === "queued") { const index = queue.indexOf(job.id); if (index >= 0) queue.splice(index, 1); job.status = "cancelled"; job.stage = "cancelled"; job.finishedAt = now(); persist(job); return true; } const proc = active.get(job.id); if (!proc) return false; job.status = "cancelled"; try { process.kill(-proc.pid, "SIGTERM"); } catch { try { proc.kill("SIGTERM"); } catch {} } setTimeout(() => { try { process.kill(-proc.pid, "SIGKILL"); } catch {} }, 5000).unref(); return true;}
async function receiveInput(req, res, sha) { if (!SHA256.test(sha)) return json(res, 400, { error: "invalid sha256" }); const length = Number(req.headers["content-length"] || 0); if (!Number.isFinite(length) || length < 1 || length > MAX_INPUT) return json(res, 413, { error: "invalid input size" }); const final = inputPath(sha); if (existsSync(final) && statSync(final).size === length) return json(res, 200, { ok: true, cached: true, sha256: sha, bytes: length }); const temp = `${final}.${process.pid}.part`; const hash = createHash("sha256"); const out = createWriteStream(temp, { mode: 0o600 }); let bytes = 0; try { for await (const chunk of req) { bytes += chunk.length; if (bytes > MAX_INPUT || bytes > length) throw new Error("input exceeded declared size"); hash.update(chunk); if (!out.write(chunk)) await new Promise((resolveDrain) => out.once("drain", resolveDrain)); } await new Promise((resolveClose, rejectClose) => out.end((error) => error ? rejectClose(error) : resolveClose())); const actual = hash.digest("hex"); if (bytes !== length || actual !== sha) throw new Error("input size or sha256 mismatch"); renameSync(temp, final); return json(res, 201, { ok: true, cached: false, sha256: sha, bytes }); } catch (error) { out.destroy(); rmSync(temp, { force: true }); return json(res, 400, { error: error.message }); }}
const server = createServer(async (req, res) => { const url = new URL(req.url || "/", `http://${req.headers.host || "localhost"}`); if (req.method === "GET" && url.pathname === "/health") return json(res, 200, health()); if (!authorized(req)) return json(res, 401, { error: "unauthorized" }); try { const inputMatch = url.pathname.match(/^\/v1\/inputs\/([a-f0-9]{64})$/); if (req.method === "PUT" && inputMatch) return receiveInput(req, res, inputMatch[1]); if (req.method === "POST" && url.pathname === "/v1/jobs") { if (!capacity().headroom) return json(res, 503, { error: "worker is under pressure", health: health() }); const spec = validateSpec(await readJson(req)); return json(res, 202, { job: publicJob(createJob(spec)) }); } if (req.method === "GET" && url.pathname === "/v1/jobs") { return json(res, 200, { jobs: [...jobs.values()].slice(-30).reverse().map(publicJob) }); } const jobMatch = url.pathname.match(/^\/v1\/jobs\/([a-z0-9-]+)$/); if (jobMatch && req.method === "GET") { const job = jobs.get(jobMatch[1]); return job ? json(res, 200, { job: publicJob(job) }) : json(res, 404, { error: "job not found" }); } if (jobMatch && req.method === "DELETE") { const job = jobs.get(jobMatch[1]); return job ? json(res, cancelJob(job) ? 202 : 409, { job: publicJob(job) }) : json(res, 404, { error: "job not found" }); } const artifactMatch = url.pathname.match(/^\/v1\/jobs\/([a-z0-9-]+)\/artifact$/); if (artifactMatch && req.method === "GET") { const job = jobs.get(artifactMatch[1]); if (!job) return json(res, 404, { error: "job not found" }); if (job.status !== "success") return json(res, 409, { error: "artifact is not ready" }); const path = join(jobWorkDir(job.id), job.output); res.writeHead(200, { "content-type": "application/octet-stream", "content-length": statSync(path).size, "x-content-sha256": job.outputSha256, "content-disposition": `attachment; filename="${job.output}"`, }); return createReadStream(path).pipe(res); } return json(res, 404, { error: "not found" }); } catch (error) { return json(res, 400, { error: error.message || String(error) }); }});
server.listen(PORT, BIND, () => { console.log(`ac-fleet-worker listening on http://${BIND}:${PORT} as ${ROLE}`);});
setInterval(pump, 5000).unref();
for (const signal of ["SIGINT", "SIGTERM"]) { process.on(signal, () => { server.close(); for (const proc of active.values()) { try { process.kill(-proc.pid, "SIGTERM"); } catch {} } process.exit(0); });}