diff --git a/packages/client/src/daemon-client-relay-e2ee-transport.ts b/packages/client/src/daemon-client-relay-e2ee-transport.ts index e25779a7bb..237699e600 100644 --- a/packages/client/src/daemon-client-relay-e2ee-transport.ts +++ b/packages/client/src/daemon-client-relay-e2ee-transport.ts @@ -24,10 +24,16 @@ export function createRelayE2eeTransportFactory(args: { baseFactory: DaemonTransportFactory; daemonPublicKeyB64: string; logger: TransportLogger; + assertOutboundCapacity?: (wireBytes: number) => void; }): DaemonTransportFactory { return ({ url }) => { const base = args.baseFactory({ url }); - return createEncryptedTransport(base, args.daemonPublicKeyB64, args.logger); + return createEncryptedTransport( + base, + args.daemonPublicKeyB64, + args.logger, + args.assertOutboundCapacity, + ); }; } @@ -42,6 +48,7 @@ export function createEncryptedTransport( base: DaemonTransport, daemonPublicKeyB64: string, logger: TransportLogger, + assertOutboundCapacity: (wireBytes: number) => void = () => {}, ): DaemonTransport { let channel: EncryptedChannel | null = null; let opened = false; @@ -148,6 +155,9 @@ export function createEncryptedTransport( if (wireBytes > MAX_RELAY_WEBSOCKET_MESSAGE_BYTES) { throw messageTooLargeError(wireBytes, MAX_RELAY_WEBSOCKET_MESSAGE_BYTES); } + // Encryption and its transport handoff are synchronous up to send's first await. + // Reject capacity failures here so a routed sender loses only its own channel. + assertOutboundCapacity(wireBytes); void channel.send(payload).catch((error) => { emitError(error); }); diff --git a/packages/client/src/hosts/index.ts b/packages/client/src/hosts/index.ts index 45b057fb07..94a7b9b8af 100644 --- a/packages/client/src/hosts/index.ts +++ b/packages/client/src/hosts/index.ts @@ -173,6 +173,8 @@ export function createHostActions( url: "ws://routed-host/ws", clientId: scopeId, transportFactory: createHostChannelTransport(origin, serverId, scopeId, routeId, state), + // Initial acquisition has a deadline; established scopes restore demand on loss. + reconnect: { enabled: false }, routedClient: true, routedServerId: serverId, routedFailure: () => state.failure, @@ -189,10 +191,16 @@ export function createHostActions( })); try { await client.connect(); + client.setReconnectEnabled(true); assertActive(); return { ...api, dispose: () => scope.release() }; } catch (error) { - const failure = state.failure ?? error; + const failure = + state.failure ?? + Object.assign(error instanceof Error ? error : new Error("Host handshake failed"), { + code: error instanceof Error && "code" in error ? error.code : "host_offline", + serverId, + }); await scope.release(); throw failure; } diff --git a/packages/client/src/node/buffering.ts b/packages/client/src/node/buffering.ts index 06bed49bf0..1dbf5be05f 100644 --- a/packages/client/src/node/buffering.ts +++ b/packages/client/src/node/buffering.ts @@ -5,11 +5,14 @@ export class HostTransportBudget { get bufferedBytes(): number { return this.used; } - reserve(bytes: number): () => void { + assertCapacity(bytes: number): void { if (bytes > this.limit - this.used) throw Object.assign(new Error("Connected host buffer limit reached"), { code: "resource_limit", }); + } + reserve(bytes: number): () => void { + this.assertCapacity(bytes); this.used += bytes; let released = false; return () => { diff --git a/packages/client/src/node/connector.test.ts b/packages/client/src/node/connector.test.ts index 3de65c2dcb..6b31107f24 100644 --- a/packages/client/src/node/connector.test.ts +++ b/packages/client/src/node/connector.test.ts @@ -72,3 +72,82 @@ test("direct connector preserves 25 MiB binary frames in both directions", async socket.send(bytes); expect(Buffer.from(await received).equals(bytes)).toBe(true); }); + +import { + createDaemonChannel, + generateKeyPair, + exportPublicKey, + type Transport, +} from "@getpaseo/relay/e2ee"; +test("encrypted connector rejects overload synchronously and preserves the physical link for sibling traffic", async () => { + const server = await endpoint(); + const keys = generateKeyPair(); + const received: (string | ArrayBuffer)[] = []; + let peer!: WebSocket; + let ready!: Promise; + server.wss.on("connection", (socket) => { + peer = socket; + const wire: Transport = { + send: (data) => socket.send(data), + close: () => socket.close(), + onmessage: null, + onclose: null, + onerror: null, + }; + socket.on("message", (data, isBinary) => { + const bytes = Buffer.from(data as Buffer); + wire.onmessage?.({ + data: isBinary + ? bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) + : bytes.toString(), + isBinary, + }); + }); + ready = createDaemonChannel(wire, keys, { + onmessage: (data) => { + received.push(data); + }, + }); + }); + const port = new URL(server.url).port; + const budget = new HostTransportBudget(4 * 1024 * 1024); + const transport = createHostTransportFactory( + `relay://127.0.0.1:${port}/srv_slow?key=${encodeURIComponent(exportPublicKey(keys.publicKey))}`, + budget, + )({ url: server.url }); + cleanup.push(async () => transport.close()); + const errors: unknown[] = []; + let closed = false; + transport.onError((error) => errors.push(error)); + transport.onClose(() => { + closed = true; + }); + await new Promise((resolve) => transport.onOpen(resolve)); + await ready; + peer.pause(); + const frame = new Uint8Array(1024 * 1024).fill(173); + let rejected: unknown; + for (let index = 0; index < 8; index++) { + try { + transport.send(frame); + } catch (error) { + rejected = error; + break; + } + } + expect(rejected).toMatchObject({ code: "resource_limit" }); + expect(budget.bufferedBytes).toBeLessThanOrEqual(budget.limit); + await new Promise((resolve) => setImmediate(resolve)); + expect(errors).toEqual([]); + expect(closed).toBe(false); + transport.send("sibling control"); + peer.resume(); + await expect.poll(() => received.includes("sibling control")).toBe(true); + expect(received.filter((data) => data instanceof ArrayBuffer).length).toBeGreaterThan(0); + expect( + received + .filter((data) => data instanceof ArrayBuffer) + .every((data) => Buffer.from(data).equals(frame)), + ).toBe(true); + expect(server.wss.clients.size).toBe(1); +}); diff --git a/packages/client/src/node/index.ts b/packages/client/src/node/index.ts index e015f53cd4..857c2ba432 100644 --- a/packages/client/src/node/index.ts +++ b/packages/client/src/node/index.ts @@ -34,6 +34,7 @@ export function createHostTransportFactory( let transport: DaemonTransport | undefined; let teardown = () => {}; let disposeSocket = () => {}; + let assertOutboundCapacity = (bytes: number) => budget.assertCapacity(bytes); let sshTunnel: SshTunnel | null = null; const close = (code?: number, reason?: string) => { if (closed) return; @@ -73,6 +74,14 @@ export function createHostTransportFactory( }, ); disposeSocket = () => socket.terminate(); + assertOutboundCapacity = (bytes) => { + if (socket.bufferedAmount + bytes > 128 * 1024 * 1024) { + throw Object.assign(new Error("Host socket buffer limit reached"), { + code: "resource_limit", + }); + } + budget.assertCapacity(bytes); + }; const pending = new Set<() => void>(); const send = socket.send.bind(socket); socket.send = (data, ...args: unknown[]) => { @@ -80,11 +89,7 @@ export function createHostTransportFactory( typeof data === "string" ? Buffer.byteLength(data) : Buffer.byteLength(data as Uint8Array); - if (socket.bufferedAmount + bytes > 128 * 1024 * 1024) { - throw Object.assign(new Error("Host socket buffer limit reached"), { - code: "resource_limit", - }); - } + assertOutboundCapacity(bytes); const release = budget.reserve(bytes); pending.add(release); const callback = args.find( @@ -116,6 +121,7 @@ export function createHostTransportFactory( baseFactory: factory, daemonPublicKeyB64: socketTarget.offer.daemonPublicKeyB64, logger: { warn() {} }, + assertOutboundCapacity: (bytes) => assertOutboundCapacity(bytes), }); transport = factory({ ...attemptOptions, url: socketTarget.url }); transport.onOpen(() => { diff --git a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts index 3fa0064781..1cf36a3bfd 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -753,6 +753,7 @@ test("disposing an origin during routed handshake settles acquisition and closes connections++; socket.on("message", (data) => { const frame = JSON.parse(data.toString()); + if (frame.type === "ping") socket.send(JSON.stringify({ type: "pong" })); if (frame.type === "hello") socket.send( JSON.stringify({ @@ -808,6 +809,17 @@ test("disposing an origin during routed handshake settles acquisition and closes await expect.poll(() => routeClosed).toBe(true); expect(ws.clients.size).toBe(1); expect(connections).toBe(1); + routeClosed = false; + const next = await api(a); + await expect(next.paseo.hosts.client("srv_pending")).rejects.toMatchObject({ + code: "host_offline", + serverId: "srv_pending", + }); + await expect.poll(() => routeClosed).toBe(true); + expect(ws.clients.size).toBe(1); + expect(connections).toBe(1); + expect((await next.paseo.workspaces.list()).entries).toEqual([]); + expect((await next.paseo.hosts.list()).hosts[0]?.status).toBe("online"); }); test("source-key registration remains idempotent at directory capacity", async () => {