import assert from "node:assert/strict"; import { once } from "node:events"; import test from "node:test"; import { ImageClient, parseEndpoint } from "../src/client"; import { MAX_MESSAGE_BYTES, ProtocolError } from "../src/protocol"; import { actor, fakeImage, json, notification, REF, response } from "./fixtures"; test("only numeric loopback endpoints and valid ports are accepted", () => { assert.deepEqual(parseEndpoint("127.0.0.1:3737"), { host: "127.0.0.1", port: 3737 }); assert.deepEqual(parseEndpoint("127.2.3.4:1"), { host: "127.2.3.4", port: 1 }); assert.deepEqual(parseEndpoint("[::1]:65535"), { host: "::1", port: 65535 }); for (const input of [ "localhost:3737", "0.0.0.0:3737", "192.168.1.1:3737", "[::]:3737", "[::ffff:127.0.0.1]:3737", "example.org:3737", "127.0.0.1:0", "127.0.0.1:65536", "127.0.0.1:-1", "127.0.0.1:1/path", ]) { assert.throws(() => parseEndpoint(input)); } }); test("fragmented UTF-8 replies and simultaneous reads preserve correlation and values", async (t) => { const operations: string[] = []; const address = await fakeImage(t, (request, socket) => { operations.push(request.command.operation); assert.equal(request.version, 4); const reply = request.command.operation === "list_actors" ? { kind: "actors", data: [actor()] } : { kind: "actor", data: actor() }; const bytes = Buffer.from(response(request, reply)); const split = bytes.indexOf(Buffer.from("雪")) + 1; socket.send(bytes.subarray(0, split), { binary: false, fin: false }); setImmediate(() => socket.send(bytes.subarray(split), { binary: false, fin: true })); }); const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined)); t.after(() => client.close()); const [actors, inspected] = await Promise.all([client.listActors(), client.inspectActor(REF)]); assert.deepEqual(actors, [actor()]); assert.deepEqual(inspected, actor()); assert.deepEqual(operations, ["list_actors", "inspect_actor"]); }); test("protocol errors preserve the connection and are never retried", async (t) => { let requests = 0; const address = await fakeImage(t, (request, socket) => { requests++; socket.send( requests === 1 ? json({ version: 4, request_id: request.request_id, result: { Err: { code: "not_found", message: "Missing actor" } }, }) : response(request, { kind: "actors", data: [] }), ); }); test("outgoing mutations enforce the exact UTF-8 envelope limit without closing the connection", async (t) => { const sizes: number[] = []; const address = await fakeImage(t, (request, socket) => { sizes.push(Buffer.byteLength(json(request))); socket.send( response( request, request.command.operation === "send" ? { kind: "acknowledged" } : { kind: "actors", data: [] }, ), ); }); const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined)); t.after(() => client.close()); const overhead = Buffer.byteLength( json({ version: 4, request_id: "runa-vscode-1", command: { operation: "send", parameters: { actor_ref: REF, message: { type: "string", data: "" } }, }, }), ); const available = MAX_MESSAGE_BYTES - overhead; const data = "雪".repeat(Math.floor(available / 3)) + "x".repeat(available % 3); await client.send(REF, { type: "string", data }); await assert.rejects(client.send(REF, { type: "string", data: data + "x" }), /Request exceeds/); await client.listActors(); assert.equal(sizes[0], MAX_MESSAGE_BYTES); assert.equal(sizes.length, 2); }); test("a lost mutation response rejects queued mutations without retrying", async (t) => { let requests = 0; const address = await fakeImage(t, (_request, socket) => { requests++; socket.terminate(); }); const client = await ImageClient.connect(address, () => undefined); t.after(() => client.close()); const results = await Promise.allSettled([ client.spawn("actors/counter", { type: "integer", data: 0n }), client.send(REF, { type: "null" }), ]); assert.ok(results.every((result) => result.status === "rejected")); assert.equal(requests, 1); }); const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined)); t.after(() => client.close()); await assert.rejects(client.inspectActor(REF), ProtocolError); assert.deepEqual(await client.listActors(), []); assert.equal(requests, 2); }); test("bad correlation, malformed JSON/UTF-8, oversize, truncation, and wrong versions close the stream", async (t) => { for (const mode of [ "correlation", "json", "utf8", "size", "truncated", "version", "kind", "binary", "event", ]) { await t.test(mode, async (t) => { let requests = 0; const failures: Error[] = []; const address = await fakeImage(t, (request, socket) => { requests++; if (mode === "json") socket.send("{broken}"); else if (mode === "utf8") socket.send(Buffer.from([0xff]), { binary: false }); else if (mode === "size") socket.send(" ".repeat(MAX_MESSAGE_BYTES + 1)); else if (mode === "truncated") { socket.send('{"version":3', { fin: false }); socket.terminate(); } else if (mode === "binary") socket.send(Buffer.from("{}")); else if (mode === "event") socket.send(json({ version: 4, event: { event: "unknown" } })); else socket.send( json({ version: mode === "version" ? 2 : 3, request_id: mode === "correlation" ? "wrong" : request.request_id, result: { Ok: { kind: mode === "kind" ? "acknowledged" : "actors", data: [] } }, }), ); }); const client = await ImageClient.connect(address, (error) => { if (error) failures.push(error); }); t.after(() => client.close()); await assert.rejects(client.listActors()); await assert.rejects(client.listActors(), /Not connected/); assert.equal(requests, 1); assert.equal(failures.length, 1); }); } }); test("request timeout rejects pending and queued reads without retry", async (t) => { let requests = 0; const address = await fakeImage(t, () => { requests++; }); const client = await ImageClient.connect( address, () => {}, () => {}, { connect: 1000, request: 30 }, ); t.after(() => client.close()); await Promise.all([ assert.rejects(client.listActors(), /timed out/), assert.rejects(client.inspectActor(REF), /Not connected/), ]); assert.equal(requests, 1); }); test("the 1 MiB response limit is inclusive and framing survives the next request", async (t) => { const address = await fakeImage(t, (request, socket) => { const line = response(request, { kind: "actors", data: [] }); socket.send(line + " ".repeat(MAX_MESSAGE_BYTES - Buffer.byteLength(line))); }); const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined)); t.after(() => client.close()); assert.deepEqual(await client.listActors(), []); assert.deepEqual(await client.listActors(), []); }); test("unsolicited responses invalidate the connection", async (t) => { const failures: Error[] = []; const address = await fakeImage(t, (request, socket) => { socket.send(response(request, { kind: "actors", data: [] })); socket.send(response(request, { kind: "actors", data: [] })); }); const client = await ImageClient.connect(address, (error) => { if (error) failures.push(error); }); t.after(() => client.close()); assert.deepEqual(await client.listActors(), []); await assert.rejects(client.listActors(), /Not connected/); assert.match(failures[0]?.message ?? "", /unsolicited/); }); test("disconnect rejects pending reads and idle peer close is reported", async (t) => { const address = await fakeImage(t, (_request, socket) => socket.close()); let disconnected = false; const client = await ImageClient.connect(address, (error) => { assert.match(error?.message ?? "", /closed/); disconnected = true; }); t.after(() => client.close()); await assert.rejects(client.listActors(), /closed/); assert.equal(disconnected, true); const idleAddress = await fakeImage(t, (request, socket) => { socket.send(response(request, { kind: "actors", data: [] })); setImmediate(() => socket.close()); }); let onClosed = () => {}; const closed = new Promise((resolve) => { onClosed = resolve; }); const idle = await ImageClient.connect(idleAddress, () => onClosed()); t.after(() => idle.close()); await idle.listActors(); await closed; await assert.rejects(idle.listActors(), /Not connected/); }); test("requires the runa-debug WebSocket subprotocol", async (t) => { const address = await fakeImage(t, () => {}, "wrong-protocol"); await assert.rejects( ImageClient.connect(address, () => {}), /subprotocol/, ); }); test("notifications interleave with replies and arrive when no request is pending", async (t) => { const revisions: bigint[] = []; let onIdle = () => {}; const idle = new Promise((resolve) => { onIdle = resolve; }); const address = await fakeImage(t, (request, socket) => { socket.send(notification(1n)); socket.send(response(request, { kind: "actors", data: [] })); setImmediate(() => socket.send(notification(5n))); }); const client = await ImageClient.connect( address, (error) => assert.equal(error, undefined), (revision) => { revisions.push(revision); if (revision === 5n) onIdle(); }, ); t.after(() => client.close()); assert.deepEqual(await client.listActors(), []); await idle; assert.deepEqual(revisions, [1n, 5n]); }); test("repeated or backwards revisions close the connection", async (t) => { const failures: Error[] = []; const address = await fakeImage(t, (_request, socket) => { socket.send(notification(5n)); socket.send(notification(5n)); }); const client = await ImageClient.connect(address, (error) => { if (error) failures.push(error); }); t.after(() => client.close()); await assert.rejects(client.listActors(), /Non-increasing/); assert.equal(failures.length, 1); }); test("failed connection reports the error", async () => { const { createServer } = await import("node:net"); const server = createServer(); server.listen(0, "127.0.0.1"); await once(server, "listening"); const address = server.address(); if (!address || typeof address === "string") throw new Error("Missing address"); await new Promise((resolve) => server.close(() => resolve())); await assert.rejects( ImageClient.connect(`127.0.0.1:${address.port}`, () => {}), /ECONNREFUSED/, ); }); test("source and system commands use the exact v4 wire contract", async (t) => { const requests: unknown[] = []; const file = { path: "actors/counter", source_id: "source:2", source: "pub fn handle() {}\n" }; const address = await fakeImage(t, (request, socket) => { requests.push(request.command); const replies: Record = { list_sources: { kind: "sources", data: [file] }, read_source: { kind: "source", data: file }, write_source: { kind: "source", data: file }, rollback_source: { kind: "source", data: file }, commit_source: { kind: "source_committed", data: { revision: "r1", source_id: "source:2" } }, source_history: { kind: "source_history", data: [{ revision: "r1", source_id: "source:2" }] }, revision_source: { kind: "behavior_source", data: { source: file.source } }, system: { kind: "acknowledged" }, spawn: { kind: "actor_spawned", data: { actor_ref: REF } }, }; socket.send(response(request, replies[request.command.operation])); }); const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined)); t.after(() => client.close()); assert.deepEqual(await client.listSources(), [file]); assert.deepEqual(await client.readSource(file.path), file); assert.deepEqual(await client.writeSource(file.path, null, file.source), file); assert.deepEqual(await client.writeSource(file.path, "source:1", file.source), file); assert.deepEqual(await client.commitSource(file.path), { revision: "r1", source_id: "source:2" }); assert.equal((await client.sourceHistory(file.path)).length, 1); assert.equal(await client.revisionSource(file.path, "r1"), file.source); assert.deepEqual(await client.rollbackSource(file.path, "r1"), file); await client.system(REF, { kind: "stop" }); await client.system(REF, { kind: "become", path: "actors/other" }); assert.equal(await client.spawn(file.path, { type: "null" }), REF); for (const invalid of [ () => client.readSource("actors/counter.rn"), () => client.readSource("source/actors/counter"), () => client.writeSource(file.path, "", "x"), () => client.revisionSource(file.path, "r.1"), () => client.rollbackSource(file.path, ""), () => client.spawn("../counter", { type: "null" }), () => client.system(REF, { kind: "become", path: "actors/x.rn" }), () => client.system(REF, { kind: "explode" } as never), () => client.system("bad", { kind: "stop" }), ]) await assert.rejects(invalid()); assert.deepEqual(requests, [ { operation: "list_sources" }, { operation: "read_source", parameters: { path: file.path } }, { operation: "write_source", parameters: { path: file.path, expected_source_id: null, source: file.source }, }, { operation: "write_source", parameters: { path: file.path, expected_source_id: "source:1", source: file.source }, }, { operation: "commit_source", parameters: { path: file.path } }, { operation: "source_history", parameters: { path: file.path } }, { operation: "revision_source", parameters: { path: file.path, revision: "r1" } }, { operation: "rollback_source", parameters: { path: file.path, revision: "r1" } }, { operation: "system", parameters: { actor_ref: REF, message: { kind: "stop" } } }, { operation: "system", parameters: { actor_ref: REF, message: { kind: "become", path: "actors/other" } }, }, { operation: "spawn", parameters: { path: file.path, state: { type: "null" }, mailbox_capacity: 64 }, }, ]); });