mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(client): add Unix socket transport
This commit is contained in:
Christian Klotz
parent
5a38a1c127
commit
1c418f6245
4 changed files with 407 additions and 0 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
156
packages/client/src/unix.ts
Normal file
156
packages/client/src/unix.ts
Normal file
|
|
@ -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<ByteTransport> {
|
||||
return new Promise<ByteTransport>((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<void> = Promise.resolve();
|
||||
|
||||
constructor(socket: Socket, maxPendingBytes: number, markLocalClose: () => void) {
|
||||
this.#socket = socket;
|
||||
this.#maxPendingBytes = maxPendingBytes;
|
||||
this.#markLocalClose = markLocalClose;
|
||||
}
|
||||
|
||||
send(chunk: Uint8Array): Promise<void> {
|
||||
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<void> {
|
||||
if (this.#closed || !this.#socket.writable) return Promise.reject(new Error("Unix transport is closed"));
|
||||
return new Promise<void>((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)));
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
225
packages/client/test/unix.test.ts
Normal file
225
packages/client/test/unix.test.ts
Normal file
|
|
@ -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<void> {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(path, () => {
|
||||
server.off("error", reject);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function closeServer(server: Server, sockets: Set<Socket>): Promise<void> {
|
||||
for (const socket of sockets) socket.destroy();
|
||||
await new Promise<void>((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<Socket>();
|
||||
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<Socket>();
|
||||
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<void>((resolve) => {
|
||||
resolveReceived = resolve;
|
||||
});
|
||||
let resumeServer: (() => void) | undefined;
|
||||
let resolveServerReady: (() => void) | undefined;
|
||||
const serverReady = new Promise<void>((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<void>((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<void>((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<Socket>();
|
||||
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 });
|
||||
}
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue