Fixed some ttp stuff

packages/ttp/src/core.test.ts

@@ -37,7 +37,7 @@ function createRoundTripMessage(): TypedMessage<Record<string, unknown>> {
   };
 }
 
-function installFakeWebTransport(frameBytes: Uint8Array) {
+function installFakeWebTransport(streamChunks: Uint8Array[]) {
   const globalScope = globalThis as typeof globalThis & {
     WebTransport?: unknown;
   };
@@ -62,11 +62,17 @@ function installFakeWebTransport(frameBytes: Uint8Array) {
           controller.enqueue(
             new ReadableStream<Uint8Array>({
               start(innerController) {
-                const splitIndex = 3;
+                const emitChunk = (index: number) => {
-                innerController.enqueue(frameBytes.subarray(0, splitIndex));
+                  if (index >= streamChunks.length) {
-                setTimeout(() => {
+                    innerController.close();
-                  innerController.enqueue(frameBytes.subarray(splitIndex));
+                    return;
-                }, 10);
+                  }
+
+                  innerController.enqueue(streamChunks[index]);
+                  setTimeout(() => emitChunk(index + 1), 10);
+                };
+
+                emitChunk(0);
               },
             }),
           );
@@ -99,6 +105,40 @@ function installFakeWebTransport(frameBytes: Uint8Array) {
   };
 }
 
+function concatBytes(chunks: Uint8Array[]) {
+  const totalLength = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0);
+  const buffer = new Uint8Array(totalLength);
+  let offset = 0;
+
+  for (const chunk of chunks) {
+    buffer.set(chunk, offset);
+    offset += chunk.byteLength;
+  }
+
+  return buffer;
+}
+
+function chunkBytes(bytes: Uint8Array, sizes: number[]) {
+  const chunks: Uint8Array[] = [];
+  let offset = 0;
+
+  for (const size of sizes) {
+    if (offset >= bytes.byteLength) {
+      break;
+    }
+
+    const end = Math.min(offset + size, bytes.byteLength);
+    chunks.push(bytes.subarray(offset, end));
+    offset = end;
+  }
+
+  if (offset < bytes.byteLength) {
+    chunks.push(bytes.subarray(offset));
+  }
+
+  return chunks;
+}
+
 function installFakeLocalStorage() {
   const globalScope = globalThis as typeof globalThis & {
     localStorage?: Storage;
@@ -162,6 +202,38 @@ describe("TTP communication codec", () => {
     });
   });
 
+  test("decodes an empty error payload as an empty object", () => {
+    const frame = new Uint8Array(6);
+    const view = new DataView(frame.buffer);
+
+    view.setUint32(0, 2, false);
+    frame[4] = 0;
+    frame[5] = 0;
+
+    const decoded = decodeCommunicationMessage(frame);
+
+    expect(decoded.id).toBe(0);
+    expect(decoded.type).toBe("error");
+    expect(decoded.data).toEqual({});
+  });
+
+  test("keeps malformed error payloads visible as error messages", () => {
+    const frame = new Uint8Array(8);
+    const view = new DataView(frame.buffer);
+
+    view.setUint32(0, 4, false);
+    frame[4] = 0;
+    frame[5] = 0;
+    frame[6] = 0;
+    frame[7] = 1;
+
+    const decoded = decodeCommunicationMessage(frame);
+
+    expect(decoded.id).toBe(0);
+    expect(decoded.type).toBe("error");
+    expect(decoded.data).toEqual({});
+  });
+
   test("throws for unknown communication type", () => {
     expect(() =>
       encodeCommunicationMessage({
@@ -207,7 +279,15 @@ describe("TTP communication codec", () => {
   });
 
   test("resolves an identification response before the stream closes", async () => {
-    const frame = encodeCommunicationMessage({
+    const ignoredFrame = encodeCommunicationMessage({
+      id: 0,
+      type: "error_internal",
+      data: {
+        error_type: "transport_noise",
+      },
+    });
+
+    const identificationFrame = encodeCommunicationMessage({
       id: 7,
       type: "identification",
       data: {
@@ -216,7 +296,69 @@ describe("TTP communication codec", () => {
       },
     });
 
-    const restoreWebTransport = installFakeWebTransport(frame);
+    const streamBytes = concatBytes([ignoredFrame, identificationFrame]);
+    const restoreWebTransport = installFakeWebTransport(
+      chunkBytes(streamBytes, [2, 5, 1, 7]),
+    );
+    const restoreLocalStorage = installFakeLocalStorage();
+
+    try {
+      const client = createTransportClient(socket, { url: "https://example.test" });
+
+      await client.connect("https://example.test");
+
+      const response = client.send(
+        "identification",
+        { user_id: 1 },
+        { id: 7 },
+      );
+
+      const timeout = new Promise<never>((_, reject) => {
+        setTimeout(() => {
+          reject(new Error("identification response timed out"));
+        }, 250);
+      });
+
+      const result = await Promise.race([response, timeout]);
+
+      expect(result).toEqual({
+        id: 7,
+        type: "identification",
+        data: {
+          challenge: "Zm9v",
+          public_key: "Zm9v",
+        },
+      });
+
+      await client.close("test-complete");
+    } finally {
+      restoreLocalStorage();
+      restoreWebTransport();
+    }
+  });
+
+  test("resolves an identification response that arrives after another frame on the same stream", async () => {
+    const ignoredFrame = encodeCommunicationMessage({
+      id: 0,
+      type: "error_internal",
+      data: {
+        error_type: "transport_noise",
+      },
+    });
+
+    const identificationFrame = encodeCommunicationMessage({
+      id: 7,
+      type: "identification",
+      data: {
+        challenge: "Zm9v",
+        public_key: "Zm9v",
+      },
+    });
+
+    const streamBytes = concatBytes([ignoredFrame, identificationFrame]);
+    const restoreWebTransport = installFakeWebTransport(
+      chunkBytes(streamBytes, [1, 4, 3, 9, 2]),
+    );
     const restoreLocalStorage = installFakeLocalStorage();
 
     try {

packages/ttp/src/core.ts

@@ -12,7 +12,7 @@ export const READY_STATE = {
 
 const CLOSE_FRAME_LEN = 0xffff_ffff;
 const APPLICATION_CLOSE_CODE = 0;
-const APPLICATION_CLOSE_REASON = "epsilon-close";
+const APPLICATION_CLOSE_REASON = "ttp-close";
 const MAX_REQUEST_ID = 0xffff_fffe;
 
 const DATA_VALUE_KIND_BOOL_TRUE = 0x01;
@@ -688,36 +688,17 @@ export function createTransportClient<T extends SchemaMap>(
             break;
           }
 
-          let frame: TypedMessage | null;
+          const closedByPeer = await processIncomingStream(
-          try {
+            result.value,
-            frame = await readFrame(result.value);
-          } catch (error) {
-            if (error instanceof RecoverableMessageDecodeError) {
-              handleRecoverableDecodeFailure(error);
-              continue;
-            }
-
-            throw error;
-          }
-
-          if (frame === null) {
-            try {
-              connection.transport.close({
-                closeCode: APPLICATION_CLOSE_CODE,
-                reason: APPLICATION_CLOSE_REASON,
-              });
-            } catch {
-              // Ignore close errors during peer shutdown.
-            }
-
-            handleConnectionFailure(
             connection,
-              new Error("Transport closed by peer"),
+            handleIncomingMessage,
+            handleRecoverableDecodeFailure,
+            handleConnectionFailure,
           );
+
+          if (closedByPeer) {
             return;
           }
-
-          handleIncomingMessage(frame);
         }
 
         if (!connection.intentional) {
@@ -1200,14 +1181,22 @@ async function writeCloseFrame(transport: WebTransportLike) {
 }
 
 /**
- * Reads and validates a full transport frame from a stream.
+ * Reads a byte stream and emits each framed protocol message it contains.
- * @param stream Incoming stream for one framed message.
+ * @param stream Incoming byte stream for a single unidirectional transport stream.
- * @returns Decoded typed message or null for close sentinel frames.
+ * @param connection Active connection instance.
+ * @param handleIncomingFrame Handler for decoded protocol messages.
+ * @param handleDecodeFailure Handler for recoverable frame decode failures.
+ * @returns True when the peer close sentinel was received.
  */
-async function readFrame(stream: ReadableStream<Uint8Array>) {
+async function processIncomingStream(
+  stream: ReadableStream<Uint8Array>,
+  connection: ActiveConnection,
+  handleIncomingFrame: (message: TypedMessage) => void,
+  handleDecodeFailure: (error: RecoverableMessageDecodeError) => void,
+  handleStreamFailure: (connection: ActiveConnection, error?: unknown) => void,
+) {
   const reader = stream.getReader();
-  const chunks: Uint8Array[] = [];
+  let bufferedBytes = new Uint8Array(0) as Uint8Array<ArrayBufferLike>;
-  let totalLength = 0;
 
   try {
     while (true) {
@@ -1216,68 +1205,72 @@ async function readFrame(stream: ReadableStream<Uint8Array>) {
         break;
       }
 
-      chunks.push(value);
+      bufferedBytes = appendBytes(bufferedBytes, value);
-      totalLength += value.byteLength;
 
-      if (totalLength >= 4) {
+      while (bufferedBytes.byteLength >= 4) {
-        const payload = concatChunks(chunks, totalLength);
+        const declaredLength = readU32(bufferedBytes, 0);
-        const declaredLength = readU32(payload, 0);
 
         if (declaredLength === CLOSE_FRAME_LEN) {
-          return null;
+          try {
+            connection.transport.close({
+              closeCode: APPLICATION_CLOSE_CODE,
+              reason: APPLICATION_CLOSE_REASON,
+            });
+          } catch {
+            // Ignore close errors during peer shutdown.
+          }
+
+          handleStreamFailure(connection, new Error("Transport closed by peer"));
+          return true;
         }
 
         const expectedLength = 4 + declaredLength;
-        if (totalLength >= expectedLength) {
+        if (bufferedBytes.byteLength < expectedLength) {
-          if (totalLength !== expectedLength) {
+          break;
-            throw new Error(
-              `Transport frame length mismatch: expected ${declaredLength}, received ${totalLength - 4}`,
-            );
         }
 
-          return decodeCommunicationMessage(payload);
+        const frameBytes = bufferedBytes.subarray(0, expectedLength);
-        }
+        bufferedBytes = bufferedBytes.subarray(expectedLength);
-      }
+
-    }
+        try {
-  } finally {
+          handleIncomingFrame(decodeCommunicationMessage(frameBytes));
-    reader.releaseLock();
+        } catch (error) {
+          if (error instanceof RecoverableMessageDecodeError) {
+            handleDecodeFailure(error);
+            continue;
           }
 
-  if (totalLength < 4) {
+          throw error;
+        }
+      }
+    }
+
+    if (bufferedBytes.byteLength > 0) {
       throw new Error("Received truncated transport frame");
     }
 
-  const payload = concatChunks(chunks, totalLength);
+    return false;
-  const declaredLength = readU32(payload, 0);
+  } finally {
-  if (declaredLength === CLOSE_FRAME_LEN) {
+    reader.releaseLock();
-    return null;
   }
-
-  const actualLength = payload.byteLength - 4;
-  if (actualLength !== declaredLength) {
-    throw new Error(
-      `Transport frame length mismatch: expected ${declaredLength}, received ${actualLength}`,
-    );
-  }
-
-  return decodeCommunicationMessage(payload);
 }
 
 /**
- * Concatenates stream chunks into a single byte array.
+ * Concatenates two byte arrays.
- * @param chunks Stream chunks to concatenate.
+ * @param left Existing buffered bytes.
- * @param totalLength Total byte length of the concatenated chunks.
+ * @param right Newly received bytes.
- * @returns Concatenated stream bytes.
+ * @returns Concatenated bytes.
  */
-function concatChunks(chunks: Uint8Array[], totalLength: number) {
+function appendBytes(left: Uint8Array, right: Uint8Array) {
-  const buffer = new Uint8Array(totalLength);
+  if (left.byteLength === 0) {
-  let offset = 0;
+    return right;
-
-  for (const chunk of chunks) {
-    buffer.set(chunk, offset);
-    offset += chunk.byteLength;
   }
 
+  const buffer = new Uint8Array(
+    left.byteLength + right.byteLength,
+  ) as Uint8Array<ArrayBufferLike>;
+  buffer.set(left, 0);
+  buffer.set(right, left.byteLength);
   return buffer;
 }
 
@@ -1347,14 +1340,26 @@ export function decodeCommunicationMessage(frame: Uint8Array): TypedMessage {
   const messageType = COMMUNICATION_TYPES[typeIndex] ?? "error_protocol";
 
   const dataLength = payloadLength - consumedHeaderBytes;
-  const dataReader = new ByteReader(reader.readBytes(dataLength));
   let decodedData: Record<string, unknown>;
 
+  if (dataLength === 0) {
+    decodedData = {};
+  } else {
+    const dataReader = new ByteReader(reader.readBytes(dataLength));
+
     try {
       decodedData = decodeContainerPayload(dataReader);
     } catch (error) {
+      if (messageType.startsWith("error")) {
+        return {
+          id,
+          type: messageType,
+          data: {},
+        };
+      } else {
         throw new RecoverableMessageDecodeError(id, messageType, error);
       }
+    }
 
     if (!dataReader.isAtEnd()) {
       throw new RecoverableMessageDecodeError(
@@ -1363,6 +1368,7 @@ export function decodeCommunicationMessage(frame: Uint8Array): TypedMessage {
         new Error("Trailing bytes found after communication data payload"),
       );
     }
+  }
 
   if (!reader.isAtEnd()) {
     throw new Error("Trailing bytes found after communication payload");