fix(transport): bound logical messages queued behind slow writers

packages/protocol/src/chunked-transport.test.ts

@@ -126,3 +126,23 @@ it("reconnect discards partial state and requires negotiation again", () => {
   fresh.receive(new Uint8Array([0x31, 1, 0, 3, 67]));
   expect(delivered).toEqual(["C"]);
 });
+
+it("bounds logical messages waiting behind a blocked writer", async () => {
+  let drain!: () => void;
+  const transport = new ChunkedTransport({
+    frameBytes: 32 * 1024 * 1024,
+    send: () =>
+      new Promise<void>((resolve) => {
+        drain = resolve;
+      }),
+    deliver: () => {},
+  });
+  const first = transport.send(new Uint8Array(100 * 1024 * 1024));
+  const overflow = transport.send("one more byte");
+  try {
+    await expect(overflow).rejects.toThrow("queue");
+  } finally {
+    drain();
+    await first;
+  }
+});

packages/protocol/src/chunked-transport.ts

@@ -13,6 +13,7 @@ export class ChunkedTransport {
   private kind = 0;
   private closed = false;
   private pending: Promise<void> | undefined;
+  private queuedBytes = 0;
 
   constructor(
     private readonly io: {
@@ -34,12 +35,30 @@ export class ChunkedTransport {
       if (this.closed) throw new Error("Transport closed");
       return chunk ? this.sendChunks(bytes, typeof frame === "string") : this.io.send(frame);
     };
-    const result = this.pending ? this.pending.then(write) : write();
-    if (!result) return Promise.resolve();
-    const pending = Promise.resolve(result).catch((error) => {
+    if (this.queuedBytes + bytes.byteLength > MAX_WEBSOCKET_MESSAGE_BYTES) {
       this.close();
+      return Promise.reject(new Error("Logical output queue exceeded its byte limit"));
+    }
+    this.queuedBytes += bytes.byteLength;
+    let result: void | Promise<void>;
+    try {
+      result = this.pending ? this.pending.then(write) : write();
+    } catch (error) {
+      this.queuedBytes -= bytes.byteLength;
       throw error;
-    });
+    }
+    if (!result) {
+      this.queuedBytes -= bytes.byteLength;
+      return Promise.resolve();
+    }
+    const pending = Promise.resolve(result)
+      .catch((error) => {
+        this.close();
+        throw error;
+      })
+      .finally(() => {
+        this.queuedBytes -= bytes.byteLength;
+      });
     this.pending = pending;
     void pending.then(
       () => {