Repository Public
1/**2 * Durable tasks. Everything that matters lives on disk (`<dataDir>/tasks/<id>/`) and in a3 * detached tmux session `crt-<id>`, so a task keeps running across runtime restarts and its4 * state is rebuilt at startup. A task whose runner vanished without an exit file is LOST.5 */6import fs from 'node:fs/promises';7import path from 'node:path';8import { z } from 'zod';9import { RtTaskInfo, type RtTaskSpec } from '@cr/shared';10import { RtError } from './plugins.js';11import { assertInsideWorkspace, atomicWrite, cleanUserEnv, envInt, type CoreContext } from './core-util.js';12import { parseTimeTrailer, TASK_LOG_SINK_CJS, TASK_RUNNER_SH } from './core-task-script.js';13import { paneExitCode, type TmuxSessionRow } from './tmux.js';1415export const TASK_PREFIX = 'crt-';16export const HOST_RE = /^[A-Za-z0-9._-]+$/;17export const MAX_LOG_READ_BYTES = 1024 * 1024;18const MAX_COMMAND_BYTES = 256 * 1024;19const MAX_TIMER_MS = 2 ** 31 - 1;2021const TaskMeta = z.object({22 id: z.string().uuid(),23 host: z.string(),24 cwd: z.string(),25 timeoutSec: z.number().nullable(),26 createdAtMs: z.number(),27 state: RtTaskInfo.shape.state,28 exitCode: z.number().nullable(),29 startedAtMs: z.number().nullable(),30 endedAtMs: z.number().nullable(),31 deadlineMs: z.number().nullable(),32 cancelRequestedAtMs: z.number().nullable(),33 timedOutAtMs: z.number().nullable(),34 logTruncated: z.boolean(),35 logBytes: z.number(),36 cpuSec: z.number().nullable(),37 maxRssMb: z.number().nullable(),38});39export type TaskMeta = z.infer<typeof TaskMeta>;4041type Ending = { kind: 'exit'; code: number | null; mtimeMs: number } | { kind: 'vanished'; paneCode: number | null };4243export class TaskManager {44 readonly tasksDir: string;45 readonly logCapBytes = envInt('CR_TASK_LOG_CAP_BYTES', 256 * 1024 * 1024);46 private readonly killGraceMs = envInt('CR_TASK_KILL_GRACE_MS', 10_000);47 private readonly pollMs = envInt('CR_TASK_POLL_MS', 500);48 private readonly sinkPath: string;49 private readonly metas = new Map<string, TaskMeta>();50 private readonly timeoutTimers = new Map<string, NodeJS.Timeout>();51 private readonly killTimers = new Map<string, NodeJS.Timeout>();52 private ticker: NodeJS.Timeout | null = null;53 private ticking = false;54 private tickCount = 0;55 private closed = false;5657 constructor(58 private readonly core: CoreContext,59 private readonly onError: (err: unknown, what: string) => void,60 ) {61 this.tasksDir = path.join(core.config.dataDir, 'tasks');62 this.sinkPath = path.join(core.config.dataDir, 'run', 'cr-logsink.cjs');63 }6465 dir(id: string): string {66 return path.join(this.tasksDir, id);67 }68 private file(id: string, name: string): string {69 return path.join(this.tasksDir, id, name);70 }7172 /** Startup: load every meta.json and reconcile RUNNING tasks with the exit files and tmux. */73 async init(): Promise<void> {74 await fs.mkdir(this.tasksDir, { recursive: true, mode: 0o700 });75 await fs.mkdir(path.dirname(this.sinkPath), { recursive: true, mode: 0o700 });76 await atomicWrite(this.sinkPath, TASK_LOG_SINK_CJS, 0o600);77 for (const ent of await fs.readdir(this.tasksDir, { withFileTypes: true })) {78 if (!ent.isDirectory()) continue;79 try {80 const meta = TaskMeta.parse(JSON.parse(await fs.readFile(this.file(ent.name, 'meta.json'), 'utf8')));81 if (meta.id === ent.name) this.metas.set(meta.id, meta);82 } catch (err) {83 this.onError(err, `unreadable task meta ${ent.name}`);84 }85 }86 const rows = await this.taskRows();87 for (const meta of this.active()) {88 if (await this.settle(meta, rows)) continue;89 // Still running under tmux: pick it up again.90 this.armTimeout(meta);91 if (meta.cancelRequestedAtMs !== null || meta.timedOutAtMs !== null) await this.terminate(meta);92 }93 this.ticker = setInterval(() => void this.tick(), this.pollMs);94 this.ticker.unref();95 }9697 /** Runtime shutdown: stop watching. Tasks keep running in tmux. */98 close(): void {99 this.closed = true;100 if (this.ticker) clearInterval(this.ticker);101 for (const t of this.timeoutTimers.values()) clearTimeout(t);102 for (const t of this.killTimers.values()) clearTimeout(t);103 this.timeoutTimers.clear();104 this.killTimers.clear();105 }106107 private active(): TaskMeta[] {108 return [...this.metas.values()].filter((m) => m.state === 'RUNNING');109 }110 isActive(id: string): boolean {111 return this.metas.get(id)?.state === 'RUNNING';112 }113 has(id: string): boolean {114 return this.metas.has(id);115 }116117 private async taskRows(): Promise<Map<string, TmuxSessionRow>> {118 const map = new Map<string, TmuxSessionRow>();119 for (const r of await this.core.tmux.listSessions()) if (r.name.startsWith(TASK_PREFIX)) map.set(r.name.slice(TASK_PREFIX.length), r);120 return map;121 }122123 private async save(meta: TaskMeta): Promise<void> {124 await atomicWrite(this.file(meta.id, 'meta.json'), JSON.stringify(meta, null, 2) + '\n', 0o600);125 }126127 private async logSize(id: string): Promise<number> {128 try {129 return (await fs.stat(this.file(id, 'log'))).size;130 } catch {131 return 0;132 }133 }134135 async info(meta: TaskMeta): Promise<RtTaskInfo> {136 return {137 id: meta.id,138 state: meta.state,139 exitCode: meta.exitCode,140 startedAtMs: meta.startedAtMs,141 endedAtMs: meta.endedAtMs,142 logBytes: meta.state === 'RUNNING' ? await this.logSize(meta.id) : meta.logBytes,143 cpuSec: meta.cpuSec,144 maxRssMb: meta.maxRssMb,145 };146 }147148 async get(id: string): Promise<RtTaskInfo> {149 const meta = this.metas.get(id);150 if (!meta) throw new RtError(404, 'not_found', 'no such task');151 return this.info(meta);152 }153154 async list(): Promise<RtTaskInfo[]> {155 const metas = [...this.metas.values()].sort((a, b) => a.createdAtMs - b.createdAtMs);156 return Promise.all(metas.map((m) => this.info(m)));157 }158159 async create(spec: RtTaskSpec): Promise<RtTaskInfo> {160 const { config, tmux } = this.core;161 if (this.metas.has(spec.id)) throw new RtError(409, 'exists', 'task already exists');162 if (spec.command.trim() === '' || spec.command.includes('\0') || Buffer.byteLength(spec.command) > MAX_COMMAND_BYTES) {163 throw new RtError(400, 'bad_request', 'invalid command');164 }165 if (spec.timeoutSec !== null && !(spec.timeoutSec > 0)) throw new RtError(400, 'bad_request', 'timeoutSec must be > 0 or null');166 const remote = spec.host !== 'local';167 if (remote && (!HOST_RE.test(spec.host) || !config.execHosts.includes(spec.host))) {168 throw new RtError(403, 'host_not_allowed', 'host is not in the exec allowlist');169 }170 const env = cleanUserEnv(spec.env);171 const dir = this.dir(spec.id);172 const cwd = remote ? dir : await assertInsideWorkspace(config, spec.cwd);173174 const now = Date.now();175 const meta: TaskMeta = {176 id: spec.id,177 host: spec.host,178 cwd,179 timeoutSec: spec.timeoutSec,180 createdAtMs: now,181 state: 'RUNNING',182 exitCode: null,183 startedAtMs: now,184 endedAtMs: null,185 deadlineMs: spec.timeoutSec === null ? null : now + Math.round(spec.timeoutSec * 1000),186 cancelRequestedAtMs: null,187 timedOutAtMs: null,188 logTruncated: false,189 logBytes: 0,190 cpuSec: null,191 maxRssMb: null,192 };193 this.metas.set(meta.id, meta); // reserve the id before the first await that could interleave194 try {195 await fs.mkdir(dir, { mode: 0o700 });196 await fs.writeFile(this.file(meta.id, 'command'), spec.command, { mode: 0o600 });197 await fs.writeFile(this.file(meta.id, 'log'), '', { mode: 0o600 });198 await fs.writeFile(this.file(meta.id, 'run.sh'), TASK_RUNNER_SH, { mode: 0o700 });199 await this.save(meta);200 await tmux.newSession({201 name: TASK_PREFIX + meta.id,202 cwd,203 cols: 200,204 rows: 50,205 env: {206 ...env,207 CR_TASK_ID: meta.id,208 CR_TASK_NODE: process.execPath,209 CR_TASK_SINK: this.sinkPath,210 CR_TASK_LOG_CAP: String(this.logCapBytes),211 CR_TASK_SHELL: config.shell,212 CR_TASK_MODE: remote ? 'ssh' : 'local',213 CR_TASK_HOST: remote ? spec.host : '',214 },215 argv: ['/bin/sh', this.file(meta.id, 'run.sh')],216 });217 } catch (err) {218 this.metas.delete(meta.id);219 await fs.rm(dir, { recursive: true, force: true });220 this.onError(err, 'task spawn failed');221 throw new RtError(500, 'spawn_failed', 'could not start the task');222 }223 this.armTimeout(meta);224 const info = await this.info(meta);225 this.core.bus.emit({ type: 'task.updated', task: info });226 return info;227 }228229 /** Cancel: TERM the runner's process group, KILL after the grace period. Idempotent. */230 async cancel(id: string): Promise<RtTaskInfo> {231 const meta = this.metas.get(id);232 if (!meta) throw new RtError(404, 'not_found', 'no such task');233 if (meta.state === 'RUNNING') {234 if (meta.cancelRequestedAtMs === null) {235 meta.cancelRequestedAtMs = Date.now();236 await this.save(meta);237 }238 await this.terminate(meta);239 }240 return this.info(meta);241 }242243 private armTimeout(meta: TaskMeta): void {244 if (meta.deadlineMs === null || meta.state !== 'RUNNING' || this.closed) return;245 const old = this.timeoutTimers.get(meta.id);246 if (old) clearTimeout(old);247 const delay = Math.max(0, meta.deadlineMs - Date.now());248 const timer = setTimeout(249 () => {250 this.timeoutTimers.delete(meta.id);251 if (meta.state !== 'RUNNING' || meta.deadlineMs === null) return;252 if (Date.now() < meta.deadlineMs) return this.armTimeout(meta); // long deadlines are re-armed in chunks253 void (async () => {254 if (meta.timedOutAtMs === null) {255 meta.timedOutAtMs = Date.now();256 await this.save(meta);257 }258 await this.terminate(meta);259 })().catch((err: unknown) => this.onError(err, 'task timeout handling failed'));260 },261 Math.min(delay, MAX_TIMER_MS),262 );263 timer.unref();264 this.timeoutTimers.set(meta.id, timer);265 }266267 private signalGroup(pid: number | null, signal: NodeJS.Signals): void {268 if (pid === null || pid <= 1) return;269 try {270 process.kill(-pid, signal); // the runner is its pane's session/group leader271 } catch {272 /* group already gone */273 }274 }275276 private async terminate(meta: TaskMeta): Promise<void> {277 if (this.killTimers.has(meta.id) || meta.state !== 'RUNNING') return;278 const name = TASK_PREFIX + meta.id;279 const row = await this.core.tmux.getSession(name);280 if (row && !row.paneDead) this.signalGroup(row.panePid, 'SIGTERM');281 const timer = setTimeout(() => {282 this.killTimers.delete(meta.id);283 if (meta.state !== 'RUNNING') return;284 void (async () => {285 const r = await this.core.tmux.getSession(name);286 if (r && !r.paneDead) this.signalGroup(r.panePid, 'SIGKILL');287 })().catch((err: unknown) => this.onError(err, 'task kill failed'));288 }, this.killGraceMs);289 timer.unref();290 this.killTimers.set(meta.id, timer);291 }292293 private async readExit(id: string): Promise<Ending | null> {294 try {295 const file = this.file(id, 'exit');296 const [text, st] = await Promise.all([fs.readFile(file, 'utf8'), fs.stat(file)]);297 const code = Number(text.trim());298 return { kind: 'exit', code: text.trim() !== '' && Number.isInteger(code) ? code : null, mtimeMs: Math.round(st.mtimeMs) };299 } catch {300 return null;301 }302 }303304 /** Finalizes `meta` if its runner is done or gone. Returns true when it reached a final state. */305 private async settle(meta: TaskMeta, rows: Map<string, TmuxSessionRow> | null): Promise<boolean> {306 let ending: Ending | null = await this.readExit(meta.id);307 if (!ending && rows) {308 const row = rows.get(meta.id);309 if (!row || row.paneDead) {310 // The runner writes `exit` before it dies, so re-reading closes the race with the listing.311 ending = (await this.readExit(meta.id)) ?? { kind: 'vanished', paneCode: row ? paneExitCode(row) : null };312 }313 }314 if (!ending || meta.state !== 'RUNNING') return meta.state !== 'RUNNING';315316 // The final state is published last: a reader that sees it must also see the final log317 // size and usage (they are read from disk below).318 let state: TaskMeta['state'];319 if (ending.kind === 'exit') {320 meta.exitCode = ending.code;321 meta.endedAtMs = ending.mtimeMs;322 if (meta.cancelRequestedAtMs !== null && ending.mtimeMs >= meta.cancelRequestedAtMs) state = 'CANCELLED';323 else if (meta.timedOutAtMs !== null && ending.mtimeMs >= meta.timedOutAtMs) state = 'TIMED_OUT';324 else state = ending.code === 0 ? 'SUCCEEDED' : 'FAILED';325 } else {326 meta.exitCode = ending.paneCode;327 if (meta.cancelRequestedAtMs !== null) state = 'CANCELLED';328 else if (meta.timedOutAtMs !== null) state = 'TIMED_OUT';329 else state = 'LOST';330 // For LOST the real end time is unknown: leave it null rather than invent one.331 meta.endedAtMs = state === 'LOST' ? null : Date.now();332 }333 try {334 meta.startedAtMs = Math.round((await fs.stat(this.file(meta.id, 'started'))).mtimeMs);335 } catch {336 /* keep the creation-time estimate */337 }338 meta.logBytes = await this.logSize(meta.id);339 meta.logTruncated = meta.logBytes >= this.logCapBytes;340 if (meta.host === 'local') {341 try {342 const usage = parseTimeTrailer(await fs.readFile(this.file(meta.id, 'time'), 'utf8'));343 meta.cpuSec = usage.cpuSec;344 meta.maxRssMb = usage.maxRssMb;345 } catch {346 /* not measured ⇒ stays null */347 }348 }349 if (meta.state !== 'RUNNING') return true; // settled concurrently350 meta.state = state;351 for (const timers of [this.timeoutTimers, this.killTimers]) {352 const t = timers.get(meta.id);353 if (t) clearTimeout(t);354 timers.delete(meta.id);355 }356 await this.save(meta);357 await this.core.tmux.killSession(TASK_PREFIX + meta.id).catch(() => false);358 this.core.bus.emit({ type: 'task.updated', task: await this.info(meta) });359 return true;360 }361362 /** Cheap every tick (one stat per running task); one tmux listing every 4th tick. */363 private async tick(): Promise<void> {364 if (this.ticking || this.closed) return;365 const active = this.active();366 if (active.length === 0) return;367 this.ticking = true;368 try {369 const rows = this.tickCount++ % 4 === 3 ? await this.taskRows() : null;370 for (const meta of active) await this.settle(meta, rows);371 } catch (err) {372 this.onError(err, 'task monitor tick failed');373 } finally {374 this.ticking = false;375 }376 }377378 async readLog(id: string, offset: number, limit: number): Promise<{ data: Buffer; size: number }> {379 if (!this.metas.has(id)) throw new RtError(404, 'not_found', 'no such task');380 const fh = await fs.open(this.file(id, 'log'), 'r');381 try {382 const size = (await fh.stat()).size;383 const want = Math.max(0, Math.min(limit, MAX_LOG_READ_BYTES, size - offset));384 const buf = Buffer.allocUnsafe(want);385 const { bytesRead } = want > 0 ? await fh.read(buf, 0, want, offset) : { bytesRead: 0 };386 return { data: buf.subarray(0, bytesRead), size };387 } finally {388 await fh.close();389 }390 }391392 logPath(id: string): string {393 return this.file(id, 'log');394 }395}396