Repository Public
1/**2 * Durable tasks HTTP/WS surface (see core-task-manager.ts for the lifecycle).3 * POST /tasks · GET /tasks · GET /tasks/:id · DELETE /tasks/:id4 * GET /tasks/:id/log?offset=&limit= (raw bytes, ≤ 1 MiB per call, `X-Log-Size`)5 * WS /tasks/:id/log/follow?offset= (binary log bytes; closes 1000 once the task has6 * ended and the log is fully drained)7 */8import { watch, type FSWatcher } from 'node:fs';9import fs from 'node:fs/promises';10import type { WebSocket } from 'ws';11import { z } from 'zod';12import { RtTaskSpec, WS_CLOSE } from '@cr/shared';13import type { RuntimePlugin } from './plugins.js';14import { getCore, sleep } from './core-util.js';15import { SATURATION_MS } from './core-backpressure.js';16import { MAX_LOG_READ_BYTES, TaskManager } from './core-task-manager.js';1718const IdParams = z.object({ id: z.string().uuid() });19const LogQuery = z.object({20 offset: z.coerce.number().int().min(0).default(0),21 limit: z.coerce.number().int().min(1).max(MAX_LOG_READ_BYTES).default(MAX_LOG_READ_BYTES),22});23const FollowQuery = z.object({ offset: z.coerce.number().int().min(0).default(0) });2425const FOLLOW_CHUNK_BYTES = 64 * 1024;26const FOLLOW_POLL_MS = 500;2728/**29 * Streams the log from `offset`. Reads only as fast as the socket drains (the file itself is30 * the buffer), so memory stays bounded whatever the log size.31 */32export async function followLog(33 ws: WebSocket,34 mgr: TaskManager,35 id: string,36 offset: number,37 highWaterBytes: number,38 saturationMs = SATURATION_MS,39): Promise<void> {40 let closed = false;41 let wake: (() => void) | null = null;42 const poke = (): void => {43 const w = wake;44 wake = null;45 w?.();46 };47 ws.on('close', () => {48 closed = true;49 poke();50 });51 ws.on('error', () => {52 closed = true;53 poke();54 });55 ws.on('message', () => undefined); // follow is one-way5657 const fh = await fs.open(mgr.logPath(id), 'r');58 let watcher: FSWatcher | null = null;59 try {60 watcher = watch(mgr.logPath(id), { persistent: false }, poke);61 watcher.on('error', () => undefined); // polling below is the fallback62 } catch {63 watcher = null;64 }65 const buf = Buffer.allocUnsafe(FOLLOW_CHUNK_BYTES);66 let pos = offset;67 try {68 while (!closed) {69 // Sample "ended" BEFORE reading: the runner drains the log before the task is finalized,70 // so (ended && nothing left to read) really means the stream is complete.71 const ended = !mgr.isActive(id);72 const { bytesRead } = await fh.read(buf, 0, buf.length, pos);73 if (closed) break;74 if (bytesRead > 0) {75 pos += bytesRead;76 ws.send(Buffer.from(buf.subarray(0, bytesRead)), { binary: true });77 if (ws.bufferedAmount > highWaterBytes) {78 const since = Date.now();79 while (!closed && ws.bufferedAmount > highWaterBytes / 2) {80 if (Date.now() - since >= saturationMs) {81 ws.close(WS_CLOSE.BACKPRESSURE, 'client too slow');82 return;83 }84 await sleep(50);85 }86 }87 continue;88 }89 if (ended) break;90 await new Promise<void>((resolve) => {91 const timer = setTimeout(poke, FOLLOW_POLL_MS);92 wake = () => {93 clearTimeout(timer);94 resolve();95 };96 });97 }98 if (!closed) ws.close(1000, 'task ended');99 } finally {100 watcher?.close();101 await fh.close();102 }103}104105const tasksPlugin: RuntimePlugin = async (app) => {106 const core = getCore(app);107 const mgr = new TaskManager(core, (err, what) => app.log.warn({ err }, what));108 await mgr.init();109 app.addHook('onClose', async () => mgr.close());110111 app.post('/tasks', async (req) => mgr.create(RtTaskSpec.parse(req.body)));112 app.get('/tasks', async () => mgr.list());113 app.get('/tasks/:id', async (req) => mgr.get(IdParams.parse(req.params).id));114 app.delete('/tasks/:id', async (req) => mgr.cancel(IdParams.parse(req.params).id));115116 app.get('/tasks/:id/log', async (req, reply) => {117 const { id } = IdParams.parse(req.params);118 const q = LogQuery.parse(req.query);119 const { data, size } = await mgr.readLog(id, q.offset, q.limit);120 return reply.header('X-Log-Size', String(size)).header('Cache-Control', 'no-store').type('application/octet-stream').send(data);121 });122123 app.get('/tasks/:id/log/follow', { websocket: true }, (ws, req) => {124 const params = IdParams.safeParse(req.params);125 const query = FollowQuery.safeParse(req.query);126 if (!params.success || !query.success || !mgr.has(params.data.id)) {127 ws.close(WS_CLOSE.NOT_FOUND, 'no such task');128 return;129 }130 followLog(ws, mgr, params.data.id, query.data.offset, core.config.termBufferBytes).catch((err: unknown) => {131 app.log.warn({ err }, 'task log follow failed');132 if (ws.readyState === ws.OPEN) ws.close(1011, 'follow failed');133 });134 });135};136137export default tasksPlugin;138