Follow a background process

export class MyContainer extends DurableObject<Env> {
	// ...

	async follow(id: string, stream: "stdout" | "stderr"): Promise<Response | undefined> {
		const status = await this.status(id);

		if (!status) {
			return undefined;
		}

		const path = `${ROOT}/${id}/${stream}.log`;
		const argv =
			status.state === "running"
				? ["tail", "-n", "+1", "-F", "--pid", `${status.pid}`, path]
				: ["cat", path];
		const process = await this.ctx.container!.exec(argv, { stderr: "ignore" });
		const { readable, writable } = new TransformStream<Uint8Array, Uint8Array>();
		const writer = writable.getWriter();
		const encoder = new TextEncoder();

		const send = (event: string, data: unknown) =>
			writer.write(
				encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`),
			);

		let exited = false;
		const markExited = () => {
			exited = true;
		};
		process.exitCode.then(markExited, markExited);

		// Signaling a process that has exited records an internal error.
		const stop = () => {
			if (!exited) process.kill();
		};

		// Stop following when the client disconnects.
		writer.closed.catch(stop);

		// Detect a disconnected client while the log is quiet.
		const heartbeat = setInterval(() => {
			writer.write(encoder.encode(": keep-alive\n\n")).catch(() => {});
		}, 5_000);

		const pump = async () => {
			try {
				for await (const text of process.stdout!.pipeThrough(
					new TextDecoderStream(),
				)) {
					await send(stream, text);
				}
				const ended = await this.status(id);
				const exitCode = ended?.state === "exited" ? ended.exitCode : null;
				await send("exit", { exitCode });
				await writer.close();
			} catch (error) {
				stop();
				// End the response, so a connected client does not wait forever.
				await writer.abort(error).catch(() => {});
			} finally {
				clearInterval(heartbeat);
			}
		};

		void pump();

		return new Response(readable, {
			headers: {
				"Content-Type": "text/event-stream",
				"Cache-Control": "no-cache",
			},
		});
	}
}

if (url.searchParams.has("follow")) {
	const response = await sandbox.follow(id, stream);
	return response ?? new Response("Not found", { status: 404 });
}

export class MyContainer extends DurableObject<Env> {
	// ...

	async waitForLog(id: string, pattern: string, timeoutMs: number) {
		const container = this.ctx.container;

		if (!container?.running) {
			return undefined;
		}

		const result = await run(container, [
			"timeout",
			"--kill-after=5",
			`${timeoutMs / 1000}`,
			"sh",
			"-c",
			WAIT,
			"sh",
			`${ROOT}/${id}`,
			pattern,
		]);

		if (result.exitCode === 0) {
			return { state: "matched", line: result.stdout.trimEnd() };
		}

		if (result.exitCode === 3) {
			return { state: "exited" };
		}

		if (result.exitCode === 4) {
			return undefined;
		}

		if (result.exitCode === 5) {
			return { state: "lost" };
		}

		// 124 when the time limit passes, or 137 if timeout also sends SIGKILL.
		if (result.exitCode === 124 || result.exitCode === 137) {
			return { state: "timed-out" };
		}

		throw new Error(`The log check failed with exit code ${result.exitCode}`);
	}
}

const ready = await sandbox.waitForLog(id, "^ready", 30_000);