fix(hosts): bound encrypted overload and initial acquisition

Check encrypted wire capacity before the async channel converts a resource limit into a physical connection error. Bound initial routed acquisition with the existing client timeout, then enable reconnect after admission.

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);
       });

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;
       }

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 () => {

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<unknown>;
+  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<void>((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<void>((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);
+});

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(() => {

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 () => {