runa/vscode/integration/server.ts

import { ChildProcess, spawn } from "node:child_process";
import assert from "node:assert/strict";
import { once } from "node:events";
import { resolve } from "node:path";
import { createInterface } from "node:readline";
import { json } from "../test/fixtures";
import WebSocket from "ws";
import { decodeNotification, WEBSOCKET_SUBPROTOCOL } from "../src/protocol";
import { ImageClient } from "../src/client";

// Control tests use a finite handler independent of the bundled self-messaging demo.
export const COUNTER_SOURCE = `pub fn handle(ctx, state, sender, message) {
    if message.op == "send" {
        ctx.send(message.target, #{op: "add", value: message.value});
        state + 1
    } else if message.op == "add" { state + message.value }
    else { panic("unknown counter operation") }
}`;

export async function startImage(): Promise<{
  address: string;
  stop: () => Promise<void>;
  diagnostics: () => string;
}> {
  const executable = resolve(
    __dirname,
    "../../../target/debug",
    process.platform === "win32" ? "runa.exe" : "runa",
  );
  const child = spawn(executable, ["--listen", "127.0.0.1:0"], {
    env: { ...process.env, RUNA_RESTORE: undefined },
    stdio: ["ignore", "ignore", "pipe"],
  });
  let diagnostics = "";
  child.stderr.on("data", (chunk: Buffer) => {
    diagnostics = (diagnostics + chunk.toString()).slice(-8192);
  });
  const stop = () => stopImage(child);
  let address: string;
  try {
    address = await new Promise<string>((resolve, reject) => {
      const lines = createInterface({ input: child.stderr });
      let diagnostics = "";
      const cleanup = () => {
        clearTimeout(timer);
        child.off("error", fail);
        child.off("exit", exited);
        lines.close();
        child.stderr.resume();
      };
      const fail = (error: Error) => {
        cleanup();
        reject(error);
      };
      const exited = () => fail(new Error(`Image exited during startup: ${diagnostics}`));
      const timer = setTimeout(
        () => fail(new Error(`Image startup timed out: ${diagnostics}`)),
        10000,
      );
      child.once("error", fail);
      child.once("exit", exited);
      lines.on("line", (line) => {
        diagnostics = (diagnostics + line + "\n").slice(-8192);
        const address = line.match(/^runa: listening on (.+)$/)?.[1];
        if (address) {
          cleanup();
          resolve(address);
        }
      });
    });
  } catch (error) {
    await stop();
    throw error;
  }
  try {
    const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined));
    try {
      const source = await client.readSource("actors/counter");
      await client.writeSource("actors/counter", source.source_id, COUNTER_SOURCE);
    } finally {
      await client.close();
    }
  } catch (error) {
    await stop();
    throw error;
  }
  return { address, stop, diagnostics: () => diagnostics };
}

async function stopImage(child: ChildProcess): Promise<void> {
  if (child.exitCode !== null || child.signalCode !== null || child.pid === undefined) return;
  const exited = once(child, "exit");
  child.kill();
  await exited;
}

export async function control(address: string, command: unknown): Promise<unknown> {
  const socket = new WebSocket(`ws://${address}/`, WEBSOCKET_SUBPROTOCOL);
  const timer = setTimeout(() => socket.terminate(), 5000);
  try {
    await once(socket, "open");
    const response = new Promise<unknown>((resolve, reject) => {
      socket.on("message", (data, binary) => {
        try {
          if (binary) throw new Error("Unexpected binary control reply");
          const text = data.toString();
          if (decodeNotification(text) !== undefined) return;
          const decoded: unknown = JSON.parse(text);
          resolve(decoded);
        } catch (error) {
          reject(error);
        }
      });
      socket.once("close", () => reject(new Error("Control connection closed")));
      socket.once("error", reject);
    });
    socket.send(json({ version: 5, request_id: "integration-control", command }));
    return await response;
  } finally {
    clearTimeout(timer);
    socket.close();
  }
}

export async function spawnCounter(address: string): Promise<string> {
  const reply = await control(address, {
    operation: "spawn",
    parameters: {
      path: "actors/counter",
      state: { type: "integer", data: 9223372036854775807n },
      mailbox_capacity: 4,
    },
  });
  const ref = json(reply).match(/"actor_ref":"([A-Za-z0-9_-]{21})"/)?.[1];
  if (!ref) throw new Error(`Spawn failed: ${json(reply)}`);
  return ref;
}