runa/vscode/integration/persistence.test.ts

import assert from "node:assert/strict";
import { execFile } from "node:child_process";
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { promisify } from "node:util";
import test from "node:test";
import { once } from "node:events";
import WebSocket from "ws";
import { ImageClient } from "../src/client";
import { parseValue } from "../src/values";
import { startImage } from "./server";
import { WEBSOCKET_SUBPROTOCOL } from "../src/protocol";

test(
  "disk image snapshots survive process replacement and resume only previously runnable actors",
  { timeout: 15000 },
  async (t) => {
    const directory = await mkdtemp(join(tmpdir(), "runa-snapshot-"));
    const file = join(directory, "image.json");
    t.after(() => rm(directory, { recursive: true }));
    const first = await startImage();
    const client = await ImageClient.connect(first.address, (error) =>
      assert.equal(error, undefined),
    );
    t.after(async () => {
      await client.close();
      await first.stop();
    });
    const actor = await client.spawn("actors/counter", parseValue("10"));
    const paused = await client.spawn("actors/counter", parseValue("42"));
    await client.system(paused, { kind: "pause" });
    await client.send(actor, parseValue('{"op":"add","value":2}'));
    const commit = await client.commitSource("actors/counter");
    const run = promisify(execFile);
    const executable = resolve(
      __dirname,
      "../../../target/debug",
      process.platform === "win32" ? "runa-cli.exe" : "runa-cli",
    );
    await run(executable, ["--connect", first.address, "save", file], { timeout: 5000 });
    assert.equal((await client.inspectActor(actor)).status.status, "paused");
    const snapshot = JSON.parse(await readFile(file, "utf8"));
    assert.equal(snapshot.version, 1);
    await client.close();
    await first.stop();
    const second = await startImage();
    const restored = await ImageClient.connect(second.address, (error) =>
      assert.equal(error, undefined),
    );
    t.after(async () => {
      await restored.close();
      await second.stop();
    });
    await run(executable, ["--connect", second.address, "restore", file], { timeout: 5000 });
    assert.equal((await restored.inspectActor(actor)).incarnation, 2n);
    assert.equal((await restored.inspectActor(actor)).status.status, "idle");
    assert.equal((await restored.inspectActor(paused)).status.status, "paused");
    assert.deepEqual(await restored.sourceHistory("actors/counter"), [commit]);
    await restored.tick(1, 1000);
    assert.deepEqual((await restored.inspectActor(actor)).state, parseValue("12"));
    assert.deepEqual((await restored.inspectActor(paused)).state, parseValue("42"));
    await restored.restoreImage(file);
    assert.equal((await restored.inspectActor(actor)).incarnation, 3n);
    assert.deepEqual((await restored.inspectActor(actor)).state, parseValue("10"));
    await assert.rejects(restored.saveImage(file), /exist/i);
    assert.equal((await restored.inspectActor(actor)).status.status, "idle");
  },
);

test("snapshot errors preserve active handlers and the connected image", async (t) => {
  const directory = await mkdtemp(join(tmpdir(), "runa-snapshot-errors-"));
  t.after(() => rm(directory, { recursive: true }));
  const image = await startImage();
  const client = await ImageClient.connect(image.address, (error) =>
    assert.equal(error, undefined),
  );
  t.after(async () => {
    await client.close();
    await image.stop();
  });
  const actor = await client.spawn("actors/counter", parseValue("5"));
  await client.send(actor, parseValue('{"op":"add","value":1}'));
  await client.tick(1, 1);
  const before = await client.inspectActor(actor);
  assert.ok(before.execution);
  const file = join(directory, "busy.json");
  await assert.rejects(client.saveImage(file), /unfinished/);
  assert.deepEqual(await client.inspectActor(actor), before);
  await assert.rejects(readFile(file), /ENOENT/);
  await client.system(actor, { kind: "pause" });
  await assert.rejects(client.saveImage(file), /unfinished/);
  await writeFile(file, '{"version":999}');
  await assert.rejects(client.restoreImage(file));
  assert.equal((await client.inspectActor(actor)).status.status, "paused");
  await client.system(actor, { kind: "resume" });
  await client.tick(1, 1000);
  assert.deepEqual((await client.inspectActor(actor)).state, parseValue("6"));
  await assert.rejects(client.restoreImage(file));
  assert.deepEqual((await client.inspectActor(actor)).state, parseValue("6"));
});

test(
  "restore keeps log subscriptions and advances sequences without replaying history",
  { timeout: 10000 },
  async (t) => {
    const directory = await mkdtemp(join(tmpdir(), "runa-snapshot-logs-"));
    t.after(() => rm(directory, { recursive: true }));
    const image = await startImage();
    const client = await ImageClient.connect(image.address, (error) =>
      assert.equal(error, undefined),
    );
    let subscriptionSocket: WebSocket | undefined;
    t.after(async () => {
      if (subscriptionSocket && subscriptionSocket.readyState !== WebSocket.CLOSED) {
        const closed = once(subscriptionSocket, "close");
        subscriptionSocket.close();
        await closed;
      }
      await client.close();
      await image.stop();
    });
    await client.writeSource(
      "actors/logger",
      null,
      "pub fn handle(ctx, state, sender, message) { ctx.info(message); state }",
    );
    const actor = await client.spawn("actors/logger", parseValue("0"));
    await client.send(actor, parseValue('"saved"'));
    await client.tick(1, 1000);
    const file = join(directory, "logs.json");
    await client.saveImage(file);
    await client.system(actor, { kind: "resume" });
    await client.send(actor, parseValue('"discarded"'));
    await client.tick(1, 1000);
    const socket = new WebSocket(`ws://${image.address}/`, WEBSOCKET_SUBPROTOCOL);
    subscriptionSocket = socket;
    await once(socket, "open");
    const subscribed = once(socket, "message");
    socket.send(
      JSON.stringify({
        version: 5,
        request_id: "logs",
        command: { operation: "logs", parameters: { actors: [actor], follow: true, after: 0 } },
      }),
    );
    await subscribed;
    const events: { sequence: number; message: string }[] = [];
    const received = new Promise<void>((resolve, reject) => {
      const timer = setTimeout(
        () => reject(new Error("Restored runtime did not deliver future log notification")),
        5000,
      );
      socket.on("message", (data) => {
        const notification = JSON.parse(data.toString());
        if (notification.event?.event === "log") {
          events.push(notification.event.data);
          clearTimeout(timer);
          resolve();
        }
      });
    });
    await client.restoreImage(file);
    await client.send(actor, parseValue('"future"'));
    await client.tick(1, 1000);
    await received;
    assert.deepEqual(
      events.map((entry) => [entry.sequence, entry.message]),
      [[3, "future"]],
    );
    assert.deepEqual(
      (await client.logs(actor)).map((entry) => entry.message),
      ["saved", "future"],
    );
  },
);