From 1dd75e9369a83c582d6b3295c4ac7a9e241eff74 Mon Sep 17 00:00:00 2001 From: Alois Date: Sat, 21 Mar 2026 13:22:38 +0100 Subject: [PATCH] Fixed some ttp stuff --- packages/ttp/src/core.test.ts | 158 +++++++++++++++++++++++++++-- packages/ttp/src/core.ts | 182 ++++++++++++++++++---------------- 2 files changed, 244 insertions(+), 96 deletions(-) diff --git a/packages/ttp/src/core.test.ts b/packages/ttp/src/core.test.ts index 3c1ad3c..dd5816c 100644 --- a/packages/ttp/src/core.test.ts +++ b/packages/ttp/src/core.test.ts @@ -37,7 +37,7 @@ function createRoundTripMessage(): TypedMessage> { }; } -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({ start(innerController) { - const splitIndex = 3; - innerController.enqueue(frameBytes.subarray(0, splitIndex)); - setTimeout(() => { - innerController.enqueue(frameBytes.subarray(splitIndex)); - }, 10); + const emitChunk = (index: number) => { + if (index >= streamChunks.length) { + innerController.close(); + return; + } + + 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((_, 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 { diff --git a/packages/ttp/src/core.ts b/packages/ttp/src/core.ts index 6fbf787..5c1c0a5 100644 --- a/packages/ttp/src/core.ts +++ b/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( break; } - let frame: TypedMessage | null; - try { - frame = await readFrame(result.value); - } catch (error) { - if (error instanceof RecoverableMessageDecodeError) { - handleRecoverableDecodeFailure(error); - continue; - } + const closedByPeer = await processIncomingStream( + result.value, + connection, + handleIncomingMessage, + handleRecoverableDecodeFailure, + handleConnectionFailure, + ); - 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"), - ); + 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. - * @param stream Incoming stream for one framed message. - * @returns Decoded typed message or null for close sentinel frames. + * Reads a byte stream and emits each framed protocol message it contains. + * @param stream Incoming byte stream for a single unidirectional transport stream. + * @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) { +async function processIncomingStream( + stream: ReadableStream, + connection: ActiveConnection, + handleIncomingFrame: (message: TypedMessage) => void, + handleDecodeFailure: (error: RecoverableMessageDecodeError) => void, + handleStreamFailure: (connection: ActiveConnection, error?: unknown) => void, +) { const reader = stream.getReader(); - const chunks: Uint8Array[] = []; - let totalLength = 0; + let bufferedBytes = new Uint8Array(0) as Uint8Array; try { while (true) { @@ -1216,68 +1205,72 @@ async function readFrame(stream: ReadableStream) { break; } - chunks.push(value); - totalLength += value.byteLength; + bufferedBytes = appendBytes(bufferedBytes, value); - if (totalLength >= 4) { - const payload = concatChunks(chunks, totalLength); - const declaredLength = readU32(payload, 0); + while (bufferedBytes.byteLength >= 4) { + const declaredLength = readU32(bufferedBytes, 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 (totalLength !== expectedLength) { - throw new Error( - `Transport frame length mismatch: expected ${declaredLength}, received ${totalLength - 4}`, - ); + if (bufferedBytes.byteLength < expectedLength) { + break; + } + + const frameBytes = bufferedBytes.subarray(0, expectedLength); + bufferedBytes = bufferedBytes.subarray(expectedLength); + + try { + handleIncomingFrame(decodeCommunicationMessage(frameBytes)); + } catch (error) { + if (error instanceof RecoverableMessageDecodeError) { + handleDecodeFailure(error); + continue; } - return decodeCommunicationMessage(payload); + throw error; } } } + + if (bufferedBytes.byteLength > 0) { + throw new Error("Received truncated transport frame"); + } + + return false; } finally { reader.releaseLock(); } - - if (totalLength < 4) { - throw new Error("Received truncated transport frame"); - } - - const payload = concatChunks(chunks, totalLength); - const declaredLength = readU32(payload, 0); - if (declaredLength === CLOSE_FRAME_LEN) { - 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. - * @param chunks Stream chunks to concatenate. - * @param totalLength Total byte length of the concatenated chunks. - * @returns Concatenated stream bytes. + * Concatenates two byte arrays. + * @param left Existing buffered bytes. + * @param right Newly received bytes. + * @returns Concatenated bytes. */ -function concatChunks(chunks: Uint8Array[], totalLength: number) { - const buffer = new Uint8Array(totalLength); - let offset = 0; - - for (const chunk of chunks) { - buffer.set(chunk, offset); - offset += chunk.byteLength; +function appendBytes(left: Uint8Array, right: Uint8Array) { + if (left.byteLength === 0) { + return right; } + const buffer = new Uint8Array( + left.byteLength + right.byteLength, + ) as Uint8Array; + buffer.set(left, 0); + buffer.set(right, left.byteLength); return buffer; } @@ -1347,21 +1340,34 @@ 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; - try { - decodedData = decodeContainerPayload(dataReader); - } catch (error) { - throw new RecoverableMessageDecodeError(id, messageType, error); - } + if (dataLength === 0) { + decodedData = {}; + } else { + const dataReader = new ByteReader(reader.readBytes(dataLength)); - if (!dataReader.isAtEnd()) { - throw new RecoverableMessageDecodeError( - id, - messageType, - new Error("Trailing bytes found after communication data payload"), - ); + 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( + id, + messageType, + new Error("Trailing bytes found after communication data payload"), + ); + } } if (!reader.isAtEnd()) {