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((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"], ); }, );