Repository Public
1import fs from 'node:fs';2import path from 'node:path';3import { afterAll, beforeAll, describe, expect, it } from 'vitest';4import type { RtEvent, RtTaskInfo } from '@cr/shared';5import type { Runtime } from './main.js';6import { parseTimeTrailer } from './core-task-script.js';7import { collectEvents, makeTestEnv, openWs, request, startRuntime, taskSpec, waitUntil, type TestEnv } from './core-testkit.js';89const FINAL = ['SUCCEEDED', 'FAILED', 'CANCELLED', 'TIMED_OUT', 'LOST'];10const saved = { ...process.env };1112beforeAll(() => {13 process.env.CR_TASK_KILL_GRACE_MS = '1000';14 process.env.CR_RUNTIME_TOKEN = 'runtime-token-that-must-not-leak';15});16afterAll(() => {17 process.env = saved;18});1920function api(env: TestEnv) {21 const sock = env.config.socketPath;22 const token = env.config.token;23 const get = async (id: string) => (await request(sock, 'GET', `/tasks/${id}`, { token })).json<RtTaskInfo>();24 return {25 create: (command: string, extra: Record<string, unknown> = {}) => request(sock, 'POST', '/tasks', { token, body: taskSpec(env, command, extra) }),26 get,27 done: (id: string, timeoutMs = 15_000) =>28 waitUntil(async () => {29 const t = await get(id);30 return FINAL.includes(t.state) ? t : null;31 }, timeoutMs, `task ${id} to finish`),32 log: (id: string, query = '') => request(sock, 'GET', `/tasks/${id}/log${query}`, { token }),33 cancel: (id: string) => request(sock, 'DELETE', `/tasks/${id}`, { token }),34 };35}3637function alive(pid: number): boolean {38 try {39 process.kill(pid, 0);40 return true;41 } catch {42 return false;43 }44}4546describe('parseTimeTrailer', () => {47 it('parses BSD time -l output in both decimal locales and never guesses', () => {48 expect(parseTimeTrailer(' 1,50 real 0,25 user 0,10 sys\n 2555904 maximum resident set size\n')).toEqual({ cpuSec: 0.35, maxRssMb: 2.4 });49 expect(parseTimeTrailer(' 0.01 real 0.00 user 0.00 sys\n')).toEqual({ cpuSec: 0, maxRssMb: null });50 expect(parseTimeTrailer('')).toEqual({ cpuSec: null, maxRssMb: null });51 });52});5354describe('durable tasks (real tmux)', () => {55 let env: TestEnv;56 let rt: Runtime;57 let t: ReturnType<typeof api>;58 beforeAll(async () => {59 env = makeTestEnv({ execHosts: ['fakehost', 'bad;host'] });60 rt = await startRuntime(env.config);61 t = api(env);62 });63 afterAll(async () => {64 await rt.stop();65 env.cleanup();66 });6768 it('success: exit 0, stdout+stderr in the log, real timings and rusage, events, cleanup', async () => {69 const events = await collectEvents(env.config.socketPath, env.config.token);70 const before = Date.now();71 const res = await t.create(`echo out-$((1+1)); echo err-line 1>&2; printf '%s|' "a'b" 'c"d'; echo "tok=\${CR_RUNTIME_TOKEN:-none}"`);72 expect(res.status).toBe(200);73 const created = res.json<RtTaskInfo>();74 expect(created.state).toBe('RUNNING');75 const done = await t.done(created.id);76 expect(done).toMatchObject({ state: 'SUCCEEDED', exitCode: 0 });77 const log = await t.log(created.id);78 expect(log.body.toString()).toBe(`out-2\nerr-line\na'b|c"d|tok=none\n`);79 expect(log.headers['x-log-size']).toBe(String(log.body.length));80 expect(done.logBytes).toBe(log.body.length);81 expect(done.startedAtMs).toBeGreaterThanOrEqual(before - 1000);82 expect(done.endedAtMs).toBeGreaterThanOrEqual(done.startedAtMs!);83 expect(done.endedAtMs).toBeLessThanOrEqual(Date.now() + 1000);84 if (process.platform === 'darwin') {85 // Measured by /usr/bin/time -l, not invented: a login shell has a non-zero RSS.86 expect(done.cpuSec).toBeGreaterThanOrEqual(0);87 expect(done.maxRssMb).toBeGreaterThan(0);88 }89 const dir = path.join(env.config.dataDir, 'tasks', created.id);90 expect(fs.readFileSync(path.join(dir, 'exit'), 'utf8')).toBe('0\n');91 expect(JSON.parse(fs.readFileSync(path.join(dir, 'meta.json'), 'utf8'))).toMatchObject({ state: 'SUCCEEDED', logTruncated: false });92 expect(fs.statSync(path.join(dir, 'run.sh')).mode & 0o777).toBe(0o700);93 expect(fs.readFileSync(path.join(dir, 'run.sh'), 'utf8')).not.toContain('out-'); // command is not interpolated94 await waitUntil(async () => !(await rt.core.tmux.hasSession(`crt-${created.id}`)), 5000, 'task tmux session cleanup');95 const mine = (e: RtEvent) => e.type === 'task.updated' && e.task.id === created.id;96 await waitUntil(() => events.events.filter(mine).length >= 2, 5000, 'task.updated events');97 expect(events.events.filter(mine).map((e) => (e.type === 'task.updated' ? e.task.state : ''))).toEqual(['RUNNING', 'SUCCEEDED']);98 events.close();99 expect((await request(env.config.socketPath, 'GET', '/tasks', { token: env.config.token })).json<RtTaskInfo[]>().map((x) => x.id)).toContain(created.id);100 });101102 it('failure keeps the exit code', async () => {103 const id = (await t.create('echo before; exit 3')).json<RtTaskInfo>().id;104 expect(await t.done(id)).toMatchObject({ state: 'FAILED', exitCode: 3 });105 });106107 it('cancel: TERM to the process group ends the command', async () => {108 const pidFile = path.join(env.workspace, 'cancel.pid');109 const id = (await t.create(`sleep 60 & echo $! > ${pidFile}; wait`)).json<RtTaskInfo>().id;110 const pid = Number(await waitUntil(() => (fs.existsSync(pidFile) ? fs.readFileSync(pidFile, 'utf8').trim() : ''), 5000, 'pid file'));111 expect(alive(pid)).toBe(true);112 const res = await t.cancel(id);113 expect(res.status).toBe(200);114 const done = await t.done(id);115 expect(done.state).toBe('CANCELLED');116 expect(done.endedAtMs).not.toBeNull();117 await waitUntil(() => !alive(pid), 3000, 'sleep to die');118 expect((await t.cancel(id)).json<RtTaskInfo>().state).toBe('CANCELLED'); // idempotent119 expect((await t.cancel('00000000-0000-4000-8000-000000000000')).status).toBe(404);120 });121122 it('cancel escalates to KILL when the command ignores TERM', async () => {123 const pidFile = path.join(env.workspace, 'stubborn.pid');124 const id = (await t.create(`trap '' TERM; echo $$ > ${pidFile}; while :; do sleep 1; done`)).json<RtTaskInfo>().id;125 const pid = Number(await waitUntil(() => (fs.existsSync(pidFile) ? fs.readFileSync(pidFile, 'utf8').trim() : ''), 5000, 'pid file'));126 await t.cancel(id);127 await new Promise((r) => setTimeout(r, 500));128 expect((await t.get(id)).state).toBe('RUNNING'); // still inside the grace period129 expect(await t.done(id)).toMatchObject({ state: 'CANCELLED' });130 await waitUntil(() => !alive(pid), 3000, 'stubborn shell to die');131 });132133 it('timeout is enforced by the runtime', async () => {134 const id = (await t.create('echo begin; sleep 60', { timeoutSec: 1 })).json<RtTaskInfo>().id;135 const done = await t.done(id);136 expect(done.state).toBe('TIMED_OUT');137 expect((await t.log(id)).body.toString()).toBe('begin\n');138 });139140 it('log reads honour offset/limit and report the total size', async () => {141 const id = (await t.create(`printf 'abcdefghij'`)).json<RtTaskInfo>().id;142 await t.done(id);143 const part = await t.log(id, '?offset=3&limit=4');144 expect(part.body.toString()).toBe('defg');145 expect(part.headers['x-log-size']).toBe('10');146 expect((await t.log(id, '?offset=10')).body.length).toBe(0);147 expect((await t.log(id, '?offset=999')).body.length).toBe(0);148 expect((await t.log(id, '?limit=99999999')).status).toBe(400); // cap: 1 MiB per call149 expect((await t.log('00000000-0000-4000-8000-000000000000')).status).toBe(404);150 });151152 it('follow streams the log as it grows and closes when the task ends', async () => {153 const id = (await t.create('for i in 1 2 3 4; do echo line-$i; sleep 0.4; done')).json<RtTaskInfo>().id;154 const ws = await openWs(env.config.socketPath, `/tasks/${id}/log/follow?offset=0`, env.config.token);155 const chunks: { atMs: number; data: Buffer }[] = [];156 let closeCode: number | null = null;157 let closedAtMs = 0;158 ws.on('message', (d: Buffer, isBinary: boolean) => {159 expect(isBinary).toBe(true);160 chunks.push({ atMs: Date.now(), data: d });161 });162 ws.on('close', (code) => {163 closeCode = code;164 closedAtMs = Date.now();165 });166 await waitUntil(() => closeCode, 15_000, 'follow close');167 expect(closeCode).toBe(1000);168 expect(Buffer.concat(chunks.map((c) => c.data)).toString()).toBe('line-1\nline-2\nline-3\nline-4\n');169 expect(chunks.length).toBeGreaterThanOrEqual(2); // delivered incrementally...170 expect(closedAtMs - chunks[0]!.atMs).toBeGreaterThan(600); // ...well before the end171 expect((await t.get(id)).state).toBe('SUCCEEDED');172173 // Following a finished task from an offset replays the tail and closes.174 const ws2 = await openWs(env.config.socketPath, `/tasks/${id}/log/follow?offset=7`, env.config.token);175 const tail: Buffer[] = [];176 ws2.on('message', (d: Buffer) => tail.push(d));177 await new Promise<void>((resolve) => ws2.on('close', () => resolve()));178 expect(Buffer.concat(tail).toString()).toBe('line-2\nline-3\nline-4\n');179180 const ghost = await openWs(env.config.socketPath, '/tasks/00000000-0000-4000-8000-000000000000/log/follow', env.config.token);181 expect(await new Promise<number>((resolve) => ghost.on('close', (code) => resolve(code)))).toBe(4404);182 });183184 it('validates host allowlist, cwd and duplicates', async () => {185 expect((await t.create('echo x', { host: 'evil' })).status).toBe(403);186 expect((await t.create('echo x', { host: 'bad;host' })).status).toBe(403); // allow-listed but fails the regex187 expect((await t.create('echo x', { cwd: '/usr' })).status).toBe(403);188 expect((await t.create(' ')).status).toBe(400);189 expect((await t.create('echo x', { timeoutSec: 0 })).status).toBe(400);190 const spec = taskSpec(env, 'echo once');191 expect((await request(env.config.socketPath, 'POST', '/tasks', { token: env.config.token, body: spec })).status).toBe(200);192 expect((await request(env.config.socketPath, 'POST', '/tasks', { token: env.config.token, body: spec })).status).toBe(409);193 });194195 it('remote host: runs ssh with the documented argv (fake ssh on PATH); rusage is not claimed', async () => {196 const bin = path.join(env.dir, 'fakebin');197 fs.mkdirSync(bin);198 fs.writeFileSync(path.join(bin, 'ssh'), '#!/bin/sh\nfor a in "$@"; do printf "[%s]" "$a"; done; echo\nexit 4\n', { mode: 0o755 });199 const command = `echo "remote cmd" && uname -a; echo 'q'`;200 const id = (await t.create(command, { host: 'fakehost', env: { PATH: `${bin}:${process.env.PATH ?? ''}` } })).json<RtTaskInfo>().id;201 const done = await t.done(id);202 expect(done).toMatchObject({ state: 'FAILED', exitCode: 4, cpuSec: null, maxRssMb: null });203 expect((await t.log(id)).body.toString()).toBe(`[-o][BatchMode=yes][-o][ConnectTimeout=10][fakehost][--][${command}]\n`);204 });205});206207describe('durable tasks across a runtime restart', () => {208 let env: TestEnv;209 beforeAll(() => {210 env = makeTestEnv();211 });212 afterAll(() => env.cleanup());213214 it('a running task keeps running while the runtime is down and is picked up again', async () => {215 const a = await startRuntime(env.config);216 const t = api(env);217 const keep = (await t.create('sleep 3; echo survived')).json<RtTaskInfo>().id;218 const timed = (await t.create('sleep 60', { timeoutSec: 4 })).json<RtTaskInfo>().id;219 const finished = (await t.create('echo quick')).json<RtTaskInfo>().id;220 const lost = (await t.create('sleep 60')).json<RtTaskInfo>().id;221 await t.done(finished);222 const lostRow = await waitUntil(() => a.core.tmux.getSession(`crt-${lost}`), 5000, 'lost task session');223 await a.stop();224 expect(fs.existsSync(env.config.socketPath)).toBe(false);225226 // While the runtime is down: one runner is destroyed without leaving an exit file.227 process.kill(-lostRow.panePid!, 'SIGKILL');228 await a.core.tmux.killSession(`crt-${lost}`);229 expect(await a.core.tmux.hasSession(`crt-${keep}`)).toBe(true); // tmux did not die with the runtime230231 const b = await startRuntime(env.config);232 try {233 expect((await t.get(keep)).state).toBe('RUNNING');234 expect((await t.get(finished)).state).toBe('SUCCEEDED');235 const lostInfo = await t.get(lost);236 expect(lostInfo).toMatchObject({ state: 'LOST', endedAtMs: null });237 expect(await t.done(keep)).toMatchObject({ state: 'SUCCEEDED', exitCode: 0 });238 expect((await t.log(keep)).body.toString()).toBe('survived\n');239 expect((await t.done(timed)).state).toBe('TIMED_OUT'); // deadline re-armed from meta.json240 } finally {241 await b.stop();242 }243 });244});245246describe('task log cap', () => {247 it('stops appending at the cap, lets the task finish and records the truncation', async () => {248 process.env.CR_TASK_LOG_CAP_BYTES = '1000';249 const env = makeTestEnv();250 const rt = await startRuntime(env.config);251 delete process.env.CR_TASK_LOG_CAP_BYTES;252 try {253 const t = api(env);254 const marker = path.join(env.workspace, 'after-cap');255 const id = (await t.create(`head -c 5000 /dev/zero | tr '\\0' 'x'; touch ${marker}`)).json<RtTaskInfo>().id;256 const done = await t.done(id);257 expect(done).toMatchObject({ state: 'SUCCEEDED', exitCode: 0, logBytes: 1000 });258 expect(fs.existsSync(marker)).toBe(true); // the command was not killed by the full log259 expect(JSON.parse(fs.readFileSync(path.join(env.config.dataDir, 'tasks', id, 'meta.json'), 'utf8')).logTruncated).toBe(true);260 } finally {261 await rt.stop();262 env.cleanup();263 }264 });265});266