diff --git a/packages/client/README.md b/packages/client/README.md index cc11595bf..61d03bdf2 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -38,3 +38,25 @@ Call `handlers.onData(chunk)` for inbound bytes, `handlers.onClose()` for an ord Treat peers as untrusted. Use a secure transport where required and protect the protocol bearer token. Subscriber exceptions are isolated from protocol state. Set `onListenerError` in `PiClientOptions` to report them to application logging or diagnostics. + +## Unix-domain sockets + +Node.js consumers can use the separately exported Unix-domain socket transport: + +```ts +import { PiClient } from "@earendil-works/pi-client"; +import { createUnixTransportFactory } from "@earendil-works/pi-client/unix"; + +const client = new PiClient({ + token: bearerToken, + transportFactory: createUnixTransportFactory({ + path: "/tmp/pi.sock", + }), +}); + +await client.connect(); +``` + +`maxPendingBytes` bounds queued outbound data. It defaults to four times the protocol frame limit. The transport preserves send order and waits for socket backpressure before resolving each send. + +The `@earendil-works/pi-client` root remains transport- and runtime-neutral. Importing the Node-specific transport requires the explicit `@earendil-works/pi-client/unix` subpath. diff --git a/packages/client/package.json b/packages/client/package.json index 5ca572253..56407a299 100644 --- a/packages/client/package.json +++ b/packages/client/package.json @@ -10,6 +10,10 @@ "types": "./dist/index.d.ts", "import": "./dist/index.js" }, + "./unix": { + "types": "./dist/unix.d.ts", + "import": "./dist/unix.js" + }, "./package.json": "./package.json" }, "sideEffects": false, diff --git a/packages/client/src/unix.ts b/packages/client/src/unix.ts new file mode 100644 index 000000000..f735fe545 --- /dev/null +++ b/packages/client/src/unix.ts @@ -0,0 +1,156 @@ +import { createConnection, type Socket } from "node:net"; +import { DEFAULT_MAX_FRAME_LENGTH } from "@earendil-works/pi-protocol"; +import type { ByteTransport, ByteTransportFactory, ByteTransportHandlers } from "./transport.ts"; + +const MAX_UNIX_SOCKET_PATH_BYTES = process.platform === "linux" ? 107 : 103; + +export interface UnixTransportOptions { + path: string; + maxPendingBytes?: number; +} + +/** Creates fresh Node Unix-domain socket transports for PiClient connection attempts. */ +export function createUnixTransportFactory(options: UnixTransportOptions): ByteTransportFactory { + if (options.path.length === 0) throw new TypeError("Unix transport path must not be empty"); + if (Buffer.byteLength(options.path) > MAX_UNIX_SOCKET_PATH_BYTES) { + throw new TypeError(`Unix transport path is too long; maximum is ${MAX_UNIX_SOCKET_PATH_BYTES} UTF-8 bytes`); + } + const maxPendingBytes = options.maxPendingBytes ?? DEFAULT_MAX_FRAME_LENGTH * 4; + if (!Number.isSafeInteger(maxPendingBytes) || maxPendingBytes <= 0) { + throw new TypeError("Unix transport maxPendingBytes must be a positive safe integer"); + } + if (process.platform === "win32") throw new Error("Unix transport is not supported on Windows"); + return (handlers) => connectUnixSocket(options.path, maxPendingBytes, handlers); +} + +function connectUnixSocket( + path: string, + maxPendingBytes: number, + handlers: ByteTransportHandlers, +): Promise { + return new Promise((resolve, reject) => { + const socket = createConnection(path); + let connected = false; + let terminal = false; + + const close = (): void => { + if (terminal) return; + terminal = true; + socket.destroy(); + if (connected) handlers.onClose(); + else reject(new Error("Unix transport closed before connecting")); + }; + + socket.once("connect", () => { + if (terminal) return; + connected = true; + resolve( + new UnixByteTransport(socket, maxPendingBytes, () => { + terminal = true; + }), + ); + }); + socket.on("data", (chunk) => { + if (!terminal) handlers.onData(new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength)); + }); + socket.once("end", close); + socket.once("close", close); + socket.once("error", (error) => { + if (terminal) return; + terminal = true; + socket.destroy(); + if (connected) handlers.onError(error); + else reject(error); + }); + }); +} + +class UnixByteTransport implements ByteTransport { + readonly #socket: Socket; + readonly #maxPendingBytes: number; + readonly #markLocalClose: () => void; + #closed = false; + #pendingBytes = 0; + #writeTail: Promise = Promise.resolve(); + + constructor(socket: Socket, maxPendingBytes: number, markLocalClose: () => void) { + this.#socket = socket; + this.#maxPendingBytes = maxPendingBytes; + this.#markLocalClose = markLocalClose; + } + + send(chunk: Uint8Array): Promise { + if (!(chunk instanceof Uint8Array)) { + return Promise.reject(new TypeError("Unix transport chunks must be Uint8Array")); + } + if (this.#closed) return Promise.reject(new Error("Unix transport is closed")); + if (this.#pendingBytes + chunk.byteLength > this.#maxPendingBytes) { + return Promise.reject(new Error("Unix transport exceeded its pending byte limit")); + } + this.#pendingBytes += chunk.byteLength; + const bytes = chunk.slice(); + const write = this.#writeTail.then(() => this.#write(bytes)); + const tracked = write.finally(() => { + this.#pendingBytes -= bytes.byteLength; + }); + this.#writeTail = tracked.catch(() => {}); + return tracked; + } + + close(): void { + if (this.#closed) return; + this.#closed = true; + this.#markLocalClose(); + this.#socket.destroy(); + } + + #write(chunk: Uint8Array): Promise { + if (this.#closed || !this.#socket.writable) return Promise.reject(new Error("Unix transport is closed")); + return new Promise((resolve, reject) => { + let callbackComplete = false; + let drainComplete = false; + let requiresDrain: boolean | undefined; + let settled = false; + + const onDrain = (): void => { + drainComplete = true; + finish(); + }; + const cleanup = (): void => { + this.#socket.off("drain", onDrain); + this.#socket.off("close", onClose); + }; + const fail = (error: Error): void => { + if (settled) return; + settled = true; + cleanup(); + reject(error); + }; + const finish = (): void => { + if (settled || !callbackComplete || requiresDrain === undefined) return; + if (requiresDrain && !drainComplete) return; + settled = true; + cleanup(); + resolve(); + }; + const onClose = (): void => fail(new Error("Unix transport closed during write")); + + try { + this.#socket.once("close", onClose); + const accepted = this.#socket.write(chunk, (error) => { + if (error) { + fail(error); + return; + } + callbackComplete = true; + finish(); + }); + requiresDrain = !accepted; + if (requiresDrain) this.#socket.once("drain", onDrain); + finish(); + } catch (error) { + fail(error instanceof Error ? error : new Error(String(error))); + } + }); + } +} diff --git a/packages/client/test/unix.test.ts b/packages/client/test/unix.test.ts new file mode 100644 index 000000000..47f1425ed --- /dev/null +++ b/packages/client/test/unix.test.ts @@ -0,0 +1,225 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { createServer, type Server, type Socket } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + ClientMessageDecoder, + encodeServerMessage, + PROTOCOL_VERSION, + type ServerSnapshot, +} from "@earendil-works/pi-protocol"; +import { describe, expect, test } from "vitest"; +import { PiClient } from "../src/index.ts"; +import { createUnixTransportFactory } from "../src/unix.ts"; + +const serverSnapshot: ServerSnapshot = { + serverId: "unix-server", + protocolVersion: PROTOCOL_VERSION, + revision: 4, + sessions: [], + models: [], +}; + +async function listen(server: Server, path: string): Promise { + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(path, () => { + server.off("error", reject); + resolve(); + }); + }); +} + +async function closeServer(server: Server, sockets: Set): Promise { + for (const socket of sockets) socket.destroy(); + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); +} + +test("rejects invalid Unix transport options", () => { + expect(() => createUnixTransportFactory({ path: "" })).toThrow(/must not be empty/); + expect(() => createUnixTransportFactory({ path: `/tmp/${"x".repeat(512)}` })).toThrow(/too long/); + expect(() => createUnixTransportFactory({ path: "/tmp/pi.sock", maxPendingBytes: 0 })).toThrow(/positive/); +}); + +describe.runIf(process.platform !== "win32")("Unix-domain sockets", () => { + test("PiClient exchanges fragmented framed messages over a real Unix socket", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-client-")); + const socketPath = join(directory, "pi.sock"); + const sockets = new Set(); + const server = createServer((socket) => { + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + const decoder = new ClientMessageDecoder(); + socket.on("data", (chunk) => { + for (const message of decoder.push(chunk)) { + if (message.type === "hello") { + const hello = encodeServerMessage({ + type: "hello", + version: PROTOCOL_VERSION, + connectionId: "unix-connection", + snapshot: serverSnapshot, + }); + for (const byte of hello) socket.write(new Uint8Array([byte])); + } else { + const response = encodeServerMessage({ + type: "response", + id: message.id, + ok: true, + result: { command: "list", sessions: [] }, + }); + const split = Math.floor(response.byteLength / 2); + socket.write(response.subarray(0, split)); + socket.write(response.subarray(split)); + } + } + }); + }); + await listen(server, socketPath); + const client = new PiClient({ + token: "unix-secret", + transportFactory: createUnixTransportFactory({ path: socketPath }), + }); + + try { + await expect(client.connect()).resolves.toEqual(serverSnapshot); + await expect(Promise.all([client.listSessions(), client.listSessions()])).resolves.toEqual([[], []]); + } finally { + client.disconnect(); + await closeServer(server, sockets); + await rm(directory, { recursive: true, force: true }); + } + }); + + test("bounds pending writes, preserves order, and reports remote end once", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-client-")); + const socketPath = join(directory, "pi.sock"); + const sockets = new Set(); + const first = new Uint8Array(2 * 1024 * 1024).fill(1); + const second = new Uint8Array(2 * 1024 * 1024).fill(2); + const expectedLength = first.byteLength + second.byteLength; + let receivedLength = 0; + let invalidOrder = false; + let resolveReceived: (() => void) | undefined; + const received = new Promise((resolve) => { + resolveReceived = resolve; + }); + let resumeServer: (() => void) | undefined; + let resolveServerReady: (() => void) | undefined; + const serverReady = new Promise((resolve) => { + resolveServerReady = resolve; + }); + const server = createServer((socket) => { + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + socket.pause(); + resumeServer = () => socket.resume(); + resolveServerReady?.(); + socket.on("data", (chunk) => { + for (let index = 0; index < chunk.byteLength; index++) { + const expected = receivedLength + index < first.byteLength ? 1 : 2; + if (chunk[index] !== expected) invalidOrder = true; + } + receivedLength += chunk.byteLength; + if (receivedLength === expectedLength) { + socket.end(new Uint8Array([9])); + resolveReceived?.(); + } + }); + }); + await listen(server, socketPath); + const inbound: number[] = []; + const errors: Error[] = []; + let closeCount = 0; + let resolveClosed: (() => void) | undefined; + const closed = new Promise((resolve) => { + resolveClosed = resolve; + }); + const transport = await createUnixTransportFactory({ path: socketPath, maxPendingBytes: expectedLength })({ + onData: (chunk) => inbound.push(...chunk), + onClose: () => { + closeCount += 1; + resolveClosed?.(); + }, + onError: (error) => errors.push(error), + }); + + try { + await serverReady; + const firstWrite = transport.send(first); + const secondWrite = transport.send(second); + await expect(transport.send(Uint8Array.of(3))).rejects.toThrow(/pending byte limit/); + resumeServer?.(); + await Promise.all([firstWrite, secondWrite, received, closed]); + expect(receivedLength).toBe(expectedLength); + expect(invalidOrder).toBe(false); + expect(inbound).toEqual([9]); + expect(errors).toEqual([]); + await new Promise((resolve) => setImmediate(resolve)); + expect(closeCount).toBe(1); + } finally { + transport.close(); + await closeServer(server, sockets); + await rm(directory, { recursive: true, force: true }); + } + }); + + test("PiClient rejects a truncated final frame from a real Unix socket", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-client-")); + const socketPath = join(directory, "pi.sock"); + const sockets = new Set(); + const server = createServer((socket) => { + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + const decoder = new ClientMessageDecoder(); + socket.on("data", (chunk) => { + for (const message of decoder.push(chunk)) { + if (message.type === "hello") { + socket.write( + encodeServerMessage({ + type: "hello", + version: PROTOCOL_VERSION, + connectionId: "unix-truncated", + snapshot: serverSnapshot, + }), + ); + } else { + socket.end(new Uint8Array([0, 0, 0, 2, 1])); + } + } + }); + }); + await listen(server, socketPath); + const client = new PiClient({ + token: "unix-secret", + transportFactory: createUnixTransportFactory({ path: socketPath }), + }); + + try { + await client.connect(); + await expect(client.listSessions()).rejects.toMatchObject({ name: "ProtocolValidationError" }); + expect(client.connectionState).toBe("disconnected"); + } finally { + client.disconnect(); + await closeServer(server, sockets); + await rm(directory, { recursive: true, force: true }); + } + }); + + test("rejects connection errors", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-client-")); + const missingPath = join(directory, "missing.sock"); + try { + await expect( + createUnixTransportFactory({ path: missingPath })({ + onData: () => {}, + onClose: () => {}, + onError: () => {}, + }), + ).rejects.toMatchObject({ code: "ENOENT" }); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); +});