a73bc4c · 1629e0b
@@ -33,6 +33,7 @@
export const GATE: readonly string[] = [
  "test/body-limits.test.ts",
  "test/boards-query.test.ts",
  "test/chaos.test.ts",
  "test/cluster-jobs.test.ts",
  "test/fuzz-found.test.ts",
  "test/fuzz-parsers.test.ts",
  "test/attach-look.test.ts",
@@ -0,0 +1,212 @@
/**
 * Run one Kubernetes Job to the end: create it, poll until it succeeds,
 * fails or the deadline passes, read its log, and delete it in every case.
 * The kube API is spoken over `fetch` with the service account's token and
 * CA from the standard paths; the client, the credentials and the clock are
 * injectable. Plain data in (a `JobSpec`), plain data out (a `JobResult`).
 */
import { readFileSync } from "node:fs";
const SA_DIR = "/var/run/secrets/kubernetes.io/serviceaccount";
const NAME = /^[a-z0-9]([-a-z0-9]{0,51}[a-z0-9])?$/;
const MAX_LOG_BYTES = 1_000_000;
export type Toleration = {
  key?: string;
  operator?: "Exists" | "Equal";
  value?: string;
  effect?: "NoSchedule" | "PreferNoSchedule" | "NoExecute";
};
export type JobSpec = {
  /** A DNS label, at most 53 characters. */
  name: string;
  image: string;
  command: string[];
  env?: Record<string, string>;
  nodeSelector?: Record<string, string>;
  tolerations?: Toleration[];
};
export type JobResult = {
  /** The container's exit code; null when it never ran to the end. */
  exitCode: number | null;
  lines: string[];
  timedOut: boolean;
};
export type Credentials = { server: string; token: string; ca?: string; namespace: string };
export type JobOptions = {
  /** Stand-in for fetch; real calls pass the CA through `tls`. */
  fetch?: (url: string, init: RequestInit & { tls?: { ca: string } }) => Promise<Response>;
  credentials?: Credentials;
  timeoutMs?: number;
  pollMs?: number;
  now?: () => number;
  sleep?: (ms: number) => Promise<void>;
  /** The most log bytes read; the rest of the log is left unread. */
  maxLogBytes?: number;
};
/** The service account the pod runs as; throws when there is none. */
export function serviceAccount(dir = SA_DIR, env: Record<string, string | undefined> = process.env): Credentials {
  const host = env.KUBERNETES_SERVICE_HOST;
  if (!host) throw new Error("not running in a cluster");
  const port = env.KUBERNETES_SERVICE_PORT || "443";
  const read = (f: string) => readFileSync(`${dir}/${f}`, "utf8").trim();
  return {
    server: `https://${host.includes(":") ? `[${host}]` : host}:${port}`,
    token: read("token"),
    ca: read("ca.crt"),
    namespace: read("namespace"),
  };
}
/** The Job manifest for a spec, or the reason it is refused. */
export function manifest(spec: JobSpec): object {
  if (!NAME.test(spec.name ?? "")) throw new Error("job name must be a DNS label of at most 53 characters");
  if (typeof spec.image !== "string" || !spec.image || /\s/.test(spec.image)) throw new Error("image is required");
  if (!Array.isArray(spec.command) || !spec.command.length || spec.command.some((c) => typeof c !== "string"))
    throw new Error("command must be a non-empty list of strings");
  const env = Object.entries(spec.env ?? {}).map(([name, value]) => {
    if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name) || typeof value !== "string") throw new Error(`bad env ${name}`);
    return { name, value };
  });
  return {
    apiVersion: "batch/v1",
    kind: "Job",
    metadata: { name: spec.name },
    spec: {
      backoffLimit: 0,
      ttlSecondsAfterFinished: 600,
      template: {
        metadata: { labels: { "job-name": spec.name } },
        spec: {
          restartPolicy: "Never",
          nodeSelector: spec.nodeSelector ?? {},
          tolerations: spec.tolerations ?? [],
          containers: [{ name: "job", image: spec.image, command: spec.command, env }],
        },
      },
    },
  };
}
      
        junipercrew
        29d ago
        outdated
      
      type Json = any hides the shape of every kube response. Type the few fields used (status.succeeded, status.failed, items[].metadata.name, containerStatuses) as a small type.
    
/** The few fields of the kube API's answers this module reads. */
type Job = { status?: { succeeded?: number; failed?: number } };
type Pod = {
  metadata?: { name?: string };
  status?: { containerStatuses?: { state?: { terminated?: { exitCode?: number } } }[] };
};
type PodList = { items?: Pod[] };
export async function runJob(spec: JobSpec, opts: JobOptions = {}): Promise<JobResult> {
  const body = JSON.stringify(manifest(spec));
  const cred = opts.credentials ?? serviceAccount();
  const doFetch = opts.fetch ?? ((u, i) => fetch(u, i as RequestInit));
  const now = opts.now ?? Date.now;
  const sleep = opts.sleep ?? ((ms) => new Promise<void>((r) => setTimeout(r, ms)));
  const timeoutMs = opts.timeoutMs ?? 600_000;
  const pollMs = opts.pollMs ?? 1000;
  const max = opts.maxLogBytes ?? MAX_LOG_BYTES;
  const ns = encodeURIComponent(cred.namespace);
  const name = encodeURIComponent(spec.name);
  const call = (method: string, path: string, payload?: string) =>
    doFetch(`${cred.server}${path}`, {
      method,
      headers: {
        authorization: `Bearer ${cred.token}`,
        accept: "application/json",
        ...(payload ? { "content-type": "application/json" } : {}),
      },
      body: payload,
      ...(cred.ca ? { tls: { ca: cred.ca } } : {}),
    });
  const json = async <T>(method: string, path: string, payload?: string): Promise<T> => {
      
        junipercrew
        29d ago
        outdated
      
      created = true is set before the POST, so if the POST answers 409 AlreadyExists, the finally block deletes the existing Job of that name, which may belong to another run. Set created only after the POST succeeds. On a non-409 failure, issue the delete only when the POST did not answer 4xx. Add a test where POST returns 409 and no DELETE is sent.
    
    const res = await call(method, path, payload);
    if (!res.ok) throw new Error(`kube ${method} ${res.status}`);
    return (await res.json()) as T;
  };
  const jobPath = `/apis/batch/v1/namespaces/${ns}/jobs/${name}`;
  let created = false;
  try {
    // Only a Job this call made is this call's to delete: a 4xx (409
    // AlreadyExists above all) means the API made nothing; anything else
    // (a 5xx, a dropped connection) may have, so it is deleted.
    let made: Response;
    try {
      made = await call("POST", `/apis/batch/v1/namespaces/${ns}/jobs`, body);
    } catch (e) {
      created = true;
      throw e;
    }
    if (!made.ok) {
      created = made.status >= 500;
      throw new Error(`kube POST ${made.status}`);
    }
    created = true;
    const deadline = now() + timeoutMs;
    let done: "succeeded" | "failed" | null = null;
    while (!done) {
      const job = await json<Job>("GET", jobPath);
      if ((job.status?.succeeded ?? 0) > 0) done = "succeeded";
      else if ((job.status?.failed ?? 0) > 0) done = "failed";
      else if (now() >= deadline) return { exitCode: null, lines: await logLines(), timedOut: true };
      else await sleep(pollMs);
    }
    const pod = await findPod();
    const code: unknown = pod?.status?.containerStatuses?.[0]?.state?.terminated?.exitCode;
      
        junipercrew
        29d ago
        
      
      A Job that fails with no pod found returns exitCode null and no lines. Say so in the JobResult doc so the caller words that error for a person.
    
    const exitCode = typeof code === "number" ? code : done === "succeeded" ? 0 : null;
    return { exitCode, lines: await logLines(pod?.metadata?.name), timedOut: false };
      
        junipercrew
        29d ago
        outdated
      
      await res.text() buffers the whole log and then slices it, so the cap is applied after the bytes are read, and slice counts UTF-16 characters, not bytes. A chatty container can exhaust memory. Ask the kube API to bound it with ?tailLines= or ?limitBytes=, or read res.body as a stream and stop at MAX_LOG_BYTES. Add a test that a log larger than the cap comes back truncated. The title promises streaming, and this does not deliver it.
    
  } finally {
    if (created) {
      try {
        await call("DELETE", `${jobPath}?propagationPolicy=Background`);
      } catch {
        // nothing more can be done; ttlSecondsAfterFinished reaps a finished Job
      }
    }
  }
  async function findPod(): Promise<Pod | undefined> {
    const list = await json<PodList>("GET", `/api/v1/namespaces/${ns}/pods?labelSelector=${encodeURIComponent(`job-name=${spec.name}`)}`);
    return list.items?.[0];
  }
  async function logLines(podName?: string): Promise<string[]> {
    try {
      const pod = podName ?? (await findPod())?.metadata?.name;
      if (typeof pod !== "string") return [];
      const res = await call("GET", `/api/v1/namespaces/${ns}/pods/${encodeURIComponent(pod)}/log?limitBytes=${max}`);
      if (!res.ok || !res.body) return [];
      const text = await readCapped(res.body, max);
      return text.split("\n").filter((l, i, a) => l !== "" || i < a.length - 1);
    } catch {
      return [];
    }
  }
}
/** Read a body chunk by chunk, stopping at `max` bytes; the rest is never held. */
async function readCapped(body: ReadableStream<Uint8Array>, max: number): Promise<string> {
  const reader = body.getReader();
  const decoder = new TextDecoder();
  let text = "";
  let bytes = 0;
  try {
    while (bytes < max) {
      const { done, value } = await reader.read();
      if (done) break;
      const chunk = value.length > max - bytes ? value.subarray(0, max - bytes) : value;
      bytes += chunk.length;
      text += decoder.decode(chunk, { stream: true });
    }
    return text + decoder.decode();
  } finally {
    await reader.cancel().catch(() => {});
  }
}
@@ -0,0 +1,88 @@
import { expect, test } from "bun:test";
import { manifest, runJob, serviceAccount, type JobSpec } from "../src/cluster/jobs.ts";
import { mkdtempSync, writeFileSync } from "node:fs";
import { join } from "node:path";
const spec: JobSpec = { name: "build-1", image: "alpine:3", command: ["sh", "-c", "echo hi"], env: { A: "b" }, nodeSelector: { pool: "ci" }, tolerations: [{ key: "ci", operator: "Exists" }] };
const credentials = { server: "https://k", token: "tok", ca: "CA", namespace: "forge" };
/** A stand-in kube API: `phases` is what successive Job GETs report. */
function kube(phases: object[], exit = 0, log = "one\ntwo\n") {
  const calls: { method: string; url: string; auth: string | null; body?: string }[] = [];
  let i = 0;
  const fetch = async (url: string, init: any) => {
    calls.push({ method: init.method, url, auth: new Headers(init.headers).get("authorization"), body: init.body });
    const path = new URL(url).pathname;
    if (init.method === "POST") return Response.json({});
    if (init.method === "DELETE") return Response.json({});
    if (path.endsWith("/log")) return new Response(log);
    if (path.endsWith("/pods")) return Response.json({ items: [{ metadata: { name: "p1" }, status: { containerStatuses: [{ state: { terminated: { exitCode: exit } } }] } }] });
    return Response.json({ status: phases[Math.min(i++, phases.length - 1)] });
  };
  return { fetch, calls };
}
const clock = () => { let t = 0; return { now: () => t, sleep: async (ms: number) => { t += ms; } }; };
const deleted = (c: { method: string }[]) => c.filter((x) => x.method === "DELETE").length;
test("a Job succeeds: exit code and log lines come back, the Job is deleted", async () => {
  const k = kube([{}, { succeeded: 1 }]);
  const r = await runJob(spec, { fetch: k.fetch, credentials, ...clock() });
  expect(r).toEqual({ exitCode: 0, lines: ["one", "two"], timedOut: false });
  expect(deleted(k.calls)).toBe(1);
  expect(k.calls.every((c) => c.auth === "Bearer tok")).toBe(true);
  const sent = JSON.parse(k.calls[0]!.body!);
  expect(sent.spec.template.spec.nodeSelector).toEqual({ pool: "ci" });
  expect(sent.spec.template.spec.containers[0].env).toEqual([{ name: "A", value: "b" }]);
});
test("a Job fails: the container's exit code is reported, the Job is deleted", async () => {
  const k = kube([{ failed: 1 }], 3);
  const r = await runJob(spec, { fetch: k.fetch, credentials, ...clock() });
  expect(r.exitCode).toBe(3);
  expect(r.timedOut).toBe(false);
  expect(deleted(k.calls)).toBe(1);
});
test("a Job that outlives the deadline times out and is deleted", async () => {
  const k = kube([{ active: 1 }]);
  const r = await runJob(spec, { fetch: k.fetch, credentials, timeoutMs: 3000, pollMs: 1000, ...clock() });
  expect(r.timedOut).toBe(true);
  expect(r.exitCode).toBeNull();
  expect(deleted(k.calls)).toBe(1);
});
test("an API error still deletes the Job", async () => {
  const calls: string[] = [];
  const fetch = async (_u: string, init: any) => { calls.push(init.method); return init.method === "DELETE" ? Response.json({}) : new Response("no", { status: 500 }); };
  await expect(runJob(spec, { fetch, credentials, ...clock() })).rejects.toThrow("kube POST 500");
  expect(calls).toEqual(["POST", "DELETE"]);
});
test("a bad spec is refused before any call", async () => {
  const k = kube([{}]);
  await expect(runJob({ ...spec, name: "Bad Name" }, { fetch: k.fetch, credentials })).rejects.toThrow("DNS label");
  expect(() => manifest({ ...spec, env: { "a b": "x" } })).toThrow("bad env");
  expect(() => manifest({ ...spec, command: [] })).toThrow("command");
  expect(k.calls).toHaveLength(0);
});
test("the service account is read from the standard files", () => {
  const dir = mkdtempSync(join(process.env.TMPDIR ?? "/tmp", "sa-"));
  for (const [f, v] of [["token", "t\n"], ["ca.crt", "C"], ["namespace", "ns"]]) writeFileSync(join(dir, f!), v!);
  expect(serviceAccount(dir, { KUBERNETES_SERVICE_HOST: "10.0.0.1", KUBERNETES_SERVICE_PORT: "6443" })).toEqual({ server: "https://10.0.0.1:6443", token: "t", ca: "C", namespace: "ns" });
  expect(() => serviceAccount(dir, {})).toThrow("cluster");
});
test("a create answering 409 deletes nothing: the Job is another run's", async () => {
  const calls: string[] = [];
  const fetch = async (_u: string, init: any) => { calls.push(init.method); return new Response("exists", { status: 409 }); };
  await expect(runJob(spec, { fetch, credentials, ...clock() })).rejects.toThrow("kube POST 409");
  expect(calls).toEqual(["POST"]);
});
test("a log larger than the cap comes back truncated, by bytes, and is asked for bounded", async () => {
  const k = kube([{ succeeded: 1 }], 0, "é".repeat(50) + "\nlast\n");
  const r = await runJob(spec, { fetch: k.fetch, credentials, maxLogBytes: 10, ...clock() });
  expect(r.lines).toEqual(["ééééé"]);
  expect(k.calls.some((c) => c.url.endsWith("/log?limitBytes=10"))).toBe(true);
});
		@@ -33,6 +33,7 @@ export const GATE: readonly string[] = [
33	33	"test/body-limits.test.ts",
34	34	"test/boards-query.test.ts",
35	35	"test/chaos.test.ts",
	36	"test/cluster-jobs.test.ts",
36	37	"test/fuzz-found.test.ts",
37	38	"test/fuzz-parsers.test.ts",
38	39	"test/attach-look.test.ts",
		@@ -0,0 +1,212 @@
	1	/**
	2	* Run one Kubernetes Job to the end: create it, poll until it succeeds,
	3	* fails or the deadline passes, read its log, and delete it in every case.
	4	* The kube API is spoken over `fetch` with the service account's token and
	5	* CA from the standard paths; the client, the credentials and the clock are
	6	* injectable. Plain data in (a `JobSpec`), plain data out (a `JobResult`).
	7	*/
	8	import { readFileSync } from "node:fs";
	9
	10	const SA_DIR = "/var/run/secrets/kubernetes.io/serviceaccount";
	11	const NAME = /^[a-z0-9]([-a-z0-9]{0,51}[a-z0-9])?$/;
	12	const MAX_LOG_BYTES = 1_000_000;
	13
	14	export type Toleration = {
	15	key?: string;
	16	operator?: "Exists" \| "Equal";
	17	value?: string;
	18	effect?: "NoSchedule" \| "PreferNoSchedule" \| "NoExecute";
	19	};
	20
	21	export type JobSpec = {
	22	/** A DNS label, at most 53 characters. */
	23	name: string;
	24	image: string;
	25	command: string[];
	26	env?: Record<string, string>;
	27	nodeSelector?: Record<string, string>;
	28	tolerations?: Toleration[];
	29	};
	30
	31	export type JobResult = {
	32	/** The container's exit code; null when it never ran to the end. */
	33	exitCode: number \| null;
	34	lines: string[];
	35	timedOut: boolean;
	36	};
	37
	38	export type Credentials = { server: string; token: string; ca?: string; namespace: string };
	39
	40	export type JobOptions = {
	41	/** Stand-in for fetch; real calls pass the CA through `tls`. */
	42	fetch?: (url: string, init: RequestInit & { tls?: { ca: string } }) => Promise<Response>;
	43	credentials?: Credentials;
	44	timeoutMs?: number;
	45	pollMs?: number;
	46	now?: () => number;
	47	sleep?: (ms: number) => Promise<void>;
	48	/** The most log bytes read; the rest of the log is left unread. */
	49	maxLogBytes?: number;
	50	};
	51
	52	/** The service account the pod runs as; throws when there is none. */
	53	export function serviceAccount(dir = SA_DIR, env: Record<string, string \| undefined> = process.env): Credentials {
	54	const host = env.KUBERNETES_SERVICE_HOST;
	55	if (!host) throw new Error("not running in a cluster");
	56	const port = env.KUBERNETES_SERVICE_PORT \|\| "443";
	57	const read = (f: string) => readFileSync(`${dir}/${f}`, "utf8").trim();
	58	return {
	59	server: `https://${host.includes(":") ? `[${host}]` : host}:${port}`,
	60	token: read("token"),
	61	ca: read("ca.crt"),
	62	namespace: read("namespace"),
	63	};
	64	}
	65
	66	/** The Job manifest for a spec, or the reason it is refused. */
	67	export function manifest(spec: JobSpec): object {
	68	if (!NAME.test(spec.name ?? "")) throw new Error("job name must be a DNS label of at most 53 characters");
	69	if (typeof spec.image !== "string" \|\| !spec.image \|\| /\s/.test(spec.image)) throw new Error("image is required");
	70	if (!Array.isArray(spec.command) \|\| !spec.command.length \|\| spec.command.some((c) => typeof c !== "string"))
	71	throw new Error("command must be a non-empty list of strings");
	72	const env = Object.entries(spec.env ?? {}).map(([name, value]) => {
	73	if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name) \|\| typeof value !== "string") throw new Error(`bad env ${name}`);
	74	return { name, value };
	75	});
	76	return {
	77	apiVersion: "batch/v1",
	78	kind: "Job",
	79	metadata: { name: spec.name },
	80	spec: {
	81	backoffLimit: 0,
	82	ttlSecondsAfterFinished: 600,
	83	template: {
	84	metadata: { labels: { "job-name": spec.name } },
	85	spec: {
	86	restartPolicy: "Never",
	87	nodeSelector: spec.nodeSelector ?? {},
	88	tolerations: spec.tolerations ?? [],
	89	containers: [{ name: "job", image: spec.image, command: spec.command, env }],
	90	},
	91	},
	92	},
	93	};
	94	}
		junipercrew 29d ago outdated `type Json = any` hides the shape of every kube response. Type the few fields used (status.succeeded, status.failed, items[].metadata.name, containerStatuses) as a small type.
	95
	96	/** The few fields of the kube API's answers this module reads. */
	97	type Job = { status?: { succeeded?: number; failed?: number } };
	98	type Pod = {
	99	metadata?: { name?: string };
	100	status?: { containerStatuses?: { state?: { terminated?: { exitCode?: number } } }[] };
	101	};
	102	type PodList = { items?: Pod[] };
	103
	104	export async function runJob(spec: JobSpec, opts: JobOptions = {}): Promise<JobResult> {
	105	const body = JSON.stringify(manifest(spec));
	106	const cred = opts.credentials ?? serviceAccount();
	107	const doFetch = opts.fetch ?? ((u, i) => fetch(u, i as RequestInit));
	108	const now = opts.now ?? Date.now;
	109	const sleep = opts.sleep ?? ((ms) => new Promise<void>((r) => setTimeout(r, ms)));
	110	const timeoutMs = opts.timeoutMs ?? 600_000;
	111	const pollMs = opts.pollMs ?? 1000;
	112	const max = opts.maxLogBytes ?? MAX_LOG_BYTES;
	113	const ns = encodeURIComponent(cred.namespace);
	114	const name = encodeURIComponent(spec.name);
	115
	116	const call = (method: string, path: string, payload?: string) =>
	117	doFetch(`${cred.server}${path}`, {
	118	method,
	119	headers: {
	120	authorization: `Bearer ${cred.token}`,
	121	accept: "application/json",
	122	...(payload ? { "content-type": "application/json" } : {}),
	123	},
	124	body: payload,
	125	...(cred.ca ? { tls: { ca: cred.ca } } : {}),
	126	});
	127	const json = async <T>(method: string, path: string, payload?: string): Promise<T> => {
		junipercrew 29d ago outdated `created = true` is set before the POST, so if the POST answers 409 AlreadyExists, the finally block deletes the existing Job of that name, which may belong to another run. Set `created` only after the POST succeeds. On a non-409 failure, issue the delete only when the POST did not answer 4xx. Add a test where POST returns 409 and no DELETE is sent.
	128	const res = await call(method, path, payload);
	129	if (!res.ok) throw new Error(`kube ${method} ${res.status}`);
	130	return (await res.json()) as T;
	131	};
	132
	133	const jobPath = `/apis/batch/v1/namespaces/${ns}/jobs/${name}`;
	134	let created = false;
	135	try {
	136	// Only a Job this call made is this call's to delete: a 4xx (409
	137	// AlreadyExists above all) means the API made nothing; anything else
	138	// (a 5xx, a dropped connection) may have, so it is deleted.
	139	let made: Response;
	140	try {
	141	made = await call("POST", `/apis/batch/v1/namespaces/${ns}/jobs`, body);
	142	} catch (e) {
	143	created = true;
	144	throw e;
	145	}
	146	if (!made.ok) {
	147	created = made.status >= 500;
	148	throw new Error(`kube POST ${made.status}`);
	149	}
	150	created = true;
	151
	152	const deadline = now() + timeoutMs;
	153	let done: "succeeded" \| "failed" \| null = null;
	154	while (!done) {
	155	const job = await json<Job>("GET", jobPath);
	156	if ((job.status?.succeeded ?? 0) > 0) done = "succeeded";
	157	else if ((job.status?.failed ?? 0) > 0) done = "failed";
	158	else if (now() >= deadline) return { exitCode: null, lines: await logLines(), timedOut: true };
	159	else await sleep(pollMs);
	160	}
	161	const pod = await findPod();
	162	const code: unknown = pod?.status?.containerStatuses?.[0]?.state?.terminated?.exitCode;
		junipercrew 29d ago A Job that fails with no pod found returns exitCode null and no lines. Say so in the JobResult doc so the caller words that error for a person.
	163	const exitCode = typeof code === "number" ? code : done === "succeeded" ? 0 : null;
	164	return { exitCode, lines: await logLines(pod?.metadata?.name), timedOut: false };
		junipercrew 29d ago outdated `await res.text()` buffers the whole log and then slices it, so the cap is applied after the bytes are read, and `slice` counts UTF-16 characters, not bytes. A chatty container can exhaust memory. Ask the kube API to bound it with `?tailLines=` or `?limitBytes=`, or read `res.body` as a stream and stop at MAX_LOG_BYTES. Add a test that a log larger than the cap comes back truncated. The title promises streaming, and this does not deliver it.
	165	} finally {
	166	if (created) {
	167	try {
	168	await call("DELETE", `${jobPath}?propagationPolicy=Background`);
	169	} catch {
	170	// nothing more can be done; ttlSecondsAfterFinished reaps a finished Job
	171	}
	172	}
	173	}
	174
	175	async function findPod(): Promise<Pod \| undefined> {
	176	const list = await json<PodList>("GET", `/api/v1/namespaces/${ns}/pods?labelSelector=${encodeURIComponent(`job-name=${spec.name}`)}`);
	177	return list.items?.[0];
	178	}
	179
	180	async function logLines(podName?: string): Promise<string[]> {
	181	try {
	182	const pod = podName ?? (await findPod())?.metadata?.name;
	183	if (typeof pod !== "string") return [];
	184	const res = await call("GET", `/api/v1/namespaces/${ns}/pods/${encodeURIComponent(pod)}/log?limitBytes=${max}`);
	185	if (!res.ok \|\| !res.body) return [];
	186	const text = await readCapped(res.body, max);
	187	return text.split("\n").filter((l, i, a) => l !== "" \|\| i < a.length - 1);
	188	} catch {
	189	return [];
	190	}
	191	}
	192	}
	193
	194	/** Read a body chunk by chunk, stopping at `max` bytes; the rest is never held. */
	195	async function readCapped(body: ReadableStream<Uint8Array>, max: number): Promise<string> {
	196	const reader = body.getReader();
	197	const decoder = new TextDecoder();
	198	let text = "";
	199	let bytes = 0;
	200	try {
	201	while (bytes < max) {
	202	const { done, value } = await reader.read();
	203	if (done) break;
	204	const chunk = value.length > max - bytes ? value.subarray(0, max - bytes) : value;
	205	bytes += chunk.length;
	206	text += decoder.decode(chunk, { stream: true });
	207	}
	208	return text + decoder.decode();
	209	} finally {
	210	await reader.cancel().catch(() => {});
	211	}
	212	}
		@@ -0,0 +1,88 @@
	1	import { expect, test } from "bun:test";
	2	import { manifest, runJob, serviceAccount, type JobSpec } from "../src/cluster/jobs.ts";
	3	import { mkdtempSync, writeFileSync } from "node:fs";
	4	import { join } from "node:path";
	5
	6	const spec: JobSpec = { name: "build-1", image: "alpine:3", command: ["sh", "-c", "echo hi"], env: { A: "b" }, nodeSelector: { pool: "ci" }, tolerations: [{ key: "ci", operator: "Exists" }] };
	7	const credentials = { server: "https://k", token: "tok", ca: "CA", namespace: "forge" };
	8
	9	/** A stand-in kube API: `phases` is what successive Job GETs report. */
	10	function kube(phases: object[], exit = 0, log = "one\ntwo\n") {
	11	const calls: { method: string; url: string; auth: string \| null; body?: string }[] = [];
	12	let i = 0;
	13	const fetch = async (url: string, init: any) => {
	14	calls.push({ method: init.method, url, auth: new Headers(init.headers).get("authorization"), body: init.body });
	15	const path = new URL(url).pathname;
	16	if (init.method === "POST") return Response.json({});
	17	if (init.method === "DELETE") return Response.json({});
	18	if (path.endsWith("/log")) return new Response(log);
	19	if (path.endsWith("/pods")) return Response.json({ items: [{ metadata: { name: "p1" }, status: { containerStatuses: [{ state: { terminated: { exitCode: exit } } }] } }] });
	20	return Response.json({ status: phases[Math.min(i++, phases.length - 1)] });
	21	};
	22	return { fetch, calls };
	23	}
	24	const clock = () => { let t = 0; return { now: () => t, sleep: async (ms: number) => { t += ms; } }; };
	25	const deleted = (c: { method: string }[]) => c.filter((x) => x.method === "DELETE").length;
	26
	27	test("a Job succeeds: exit code and log lines come back, the Job is deleted", async () => {
	28	const k = kube([{}, { succeeded: 1 }]);
	29	const r = await runJob(spec, { fetch: k.fetch, credentials, ...clock() });
	30	expect(r).toEqual({ exitCode: 0, lines: ["one", "two"], timedOut: false });
	31	expect(deleted(k.calls)).toBe(1);
	32	expect(k.calls.every((c) => c.auth === "Bearer tok")).toBe(true);
	33	const sent = JSON.parse(k.calls[0]!.body!);
	34	expect(sent.spec.template.spec.nodeSelector).toEqual({ pool: "ci" });
	35	expect(sent.spec.template.spec.containers[0].env).toEqual([{ name: "A", value: "b" }]);
	36	});
	37
	38	test("a Job fails: the container's exit code is reported, the Job is deleted", async () => {
	39	const k = kube([{ failed: 1 }], 3);
	40	const r = await runJob(spec, { fetch: k.fetch, credentials, ...clock() });
	41	expect(r.exitCode).toBe(3);
	42	expect(r.timedOut).toBe(false);
	43	expect(deleted(k.calls)).toBe(1);
	44	});
	45
	46	test("a Job that outlives the deadline times out and is deleted", async () => {
	47	const k = kube([{ active: 1 }]);
	48	const r = await runJob(spec, { fetch: k.fetch, credentials, timeoutMs: 3000, pollMs: 1000, ...clock() });
	49	expect(r.timedOut).toBe(true);
	50	expect(r.exitCode).toBeNull();
	51	expect(deleted(k.calls)).toBe(1);
	52	});
	53
	54	test("an API error still deletes the Job", async () => {
	55	const calls: string[] = [];
	56	const fetch = async (_u: string, init: any) => { calls.push(init.method); return init.method === "DELETE" ? Response.json({}) : new Response("no", { status: 500 }); };
	57	await expect(runJob(spec, { fetch, credentials, ...clock() })).rejects.toThrow("kube POST 500");
	58	expect(calls).toEqual(["POST", "DELETE"]);
	59	});
	60
	61	test("a bad spec is refused before any call", async () => {
	62	const k = kube([{}]);
	63	await expect(runJob({ ...spec, name: "Bad Name" }, { fetch: k.fetch, credentials })).rejects.toThrow("DNS label");
	64	expect(() => manifest({ ...spec, env: { "a b": "x" } })).toThrow("bad env");
	65	expect(() => manifest({ ...spec, command: [] })).toThrow("command");
	66	expect(k.calls).toHaveLength(0);
	67	});
	68
	69	test("the service account is read from the standard files", () => {
	70	const dir = mkdtempSync(join(process.env.TMPDIR ?? "/tmp", "sa-"));
	71	for (const [f, v] of [["token", "t\n"], ["ca.crt", "C"], ["namespace", "ns"]]) writeFileSync(join(dir, f!), v!);
	72	expect(serviceAccount(dir, { KUBERNETES_SERVICE_HOST: "10.0.0.1", KUBERNETES_SERVICE_PORT: "6443" })).toEqual({ server: "https://10.0.0.1:6443", token: "t", ca: "C", namespace: "ns" });
	73	expect(() => serviceAccount(dir, {})).toThrow("cluster");
	74	});
	75
	76	test("a create answering 409 deletes nothing: the Job is another run's", async () => {
	77	const calls: string[] = [];
	78	const fetch = async (_u: string, init: any) => { calls.push(init.method); return new Response("exists", { status: 409 }); };
	79	await expect(runJob(spec, { fetch, credentials, ...clock() })).rejects.toThrow("kube POST 409");
	80	expect(calls).toEqual(["POST"]);
	81	});
	82
	83	test("a log larger than the cap comes back truncated, by bytes, and is asked for bounded", async () => {
	84	const k = kube([{ succeeded: 1 }], 0, "é".repeat(50) + "\nlast\n");
	85	const r = await runJob(spec, { fetch: k.fetch, credentials, maxLogBytes: 10, ...clock() });
	86	expect(r.lines).toEqual(["ééééé"]);
	87	expect(k.calls.some((c) => c.url.endsWith("/log?limitBytes=10"))).toBe(true);
	88	});