Fixed ttp
This commit is contained in:
parent
dcd31cb0c5
commit
24e969a19c
3 changed files with 281 additions and 36 deletions
|
|
@ -445,9 +445,8 @@ export default function Provider(props: { children: React.ReactNode }) {
|
|||
}
|
||||
|
||||
const isFatal = isFatalIdentificationError(identificationError);
|
||||
const protocolErrorDetails = getProtocolErrorDetails(
|
||||
identificationError,
|
||||
);
|
||||
const protocolErrorDetails =
|
||||
getProtocolErrorDetails(identificationError);
|
||||
|
||||
log(
|
||||
isFatal ? 0 : 1,
|
||||
|
|
|
|||
|
|
@ -50,9 +50,11 @@ function installFakeWebTransport(streamChunks: Uint8Array[]) {
|
|||
|
||||
private resolveClosed!: () => void;
|
||||
|
||||
readonly incomingUnidirectionalStreams: ReadableStream<ReadableStream<Uint8Array>>;
|
||||
readonly incomingUnidirectionalStreams: ReadableStream<
|
||||
ReadableStream<Uint8Array>
|
||||
>;
|
||||
|
||||
constructor(_url: string) {
|
||||
constructor() {
|
||||
this.closed = new Promise<void>((resolve) => {
|
||||
this.resolveClosed = resolve;
|
||||
});
|
||||
|
|
@ -139,6 +141,16 @@ function chunkBytes(bytes: Uint8Array, sizes: number[]) {
|
|||
return chunks;
|
||||
}
|
||||
|
||||
function frameBytes(payload: Uint8Array) {
|
||||
const frame = new Uint8Array(payload.byteLength + 4);
|
||||
const view = new DataView(frame.buffer);
|
||||
|
||||
view.setUint32(0, payload.byteLength, false);
|
||||
frame.set(payload, 4);
|
||||
|
||||
return frame;
|
||||
}
|
||||
|
||||
function installFakeLocalStorage() {
|
||||
const globalScope = globalThis as typeof globalThis & {
|
||||
localStorage?: Storage;
|
||||
|
|
@ -172,12 +184,31 @@ function installFakeLocalStorage() {
|
|||
};
|
||||
}
|
||||
|
||||
function installConsoleLogSpy() {
|
||||
const globalConsole = console as typeof console & {
|
||||
log: (...args: unknown[]) => void;
|
||||
};
|
||||
const originalLog = globalConsole.log;
|
||||
const calls: unknown[][] = [];
|
||||
|
||||
globalConsole.log = (...args: unknown[]) => {
|
||||
calls.push(args);
|
||||
};
|
||||
|
||||
return {
|
||||
calls,
|
||||
restore() {
|
||||
globalConsole.log = originalLog;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("TTP communication codec", () => {
|
||||
test("encodes and decodes a mixed payload message", () => {
|
||||
const input = createRoundTripMessage();
|
||||
|
||||
const encoded = encodeCommunicationMessage(input);
|
||||
const decoded = decodeCommunicationMessage(encoded);
|
||||
const decoded = decodeCommunicationMessage(frameBytes(encoded));
|
||||
|
||||
expect(decoded.id).toBe(41);
|
||||
expect(decoded.type).toBe("message");
|
||||
|
|
@ -203,14 +234,14 @@ 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);
|
||||
const innerFrame = new Uint8Array(6);
|
||||
const view = new DataView(innerFrame.buffer);
|
||||
|
||||
view.setUint32(0, 2, false);
|
||||
frame[4] = 0;
|
||||
frame[5] = 0;
|
||||
innerFrame[4] = 0;
|
||||
innerFrame[5] = 0;
|
||||
|
||||
const decoded = decodeCommunicationMessage(frame);
|
||||
const decoded = decodeCommunicationMessage(frameBytes(innerFrame));
|
||||
|
||||
expect(decoded.id).toBe(0);
|
||||
expect(decoded.type).toBe("error");
|
||||
|
|
@ -218,16 +249,16 @@ describe("TTP communication codec", () => {
|
|||
});
|
||||
|
||||
test("keeps malformed error payloads visible as error messages", () => {
|
||||
const frame = new Uint8Array(8);
|
||||
const view = new DataView(frame.buffer);
|
||||
const innerFrame = new Uint8Array(8);
|
||||
const view = new DataView(innerFrame.buffer);
|
||||
|
||||
view.setUint32(0, 4, false);
|
||||
frame[4] = 0;
|
||||
frame[5] = 0;
|
||||
frame[6] = 0;
|
||||
frame[7] = 1;
|
||||
innerFrame[4] = 0;
|
||||
innerFrame[5] = 0;
|
||||
innerFrame[6] = 0;
|
||||
innerFrame[7] = 1;
|
||||
|
||||
const decoded = decodeCommunicationMessage(frame);
|
||||
const decoded = decodeCommunicationMessage(frameBytes(innerFrame));
|
||||
|
||||
expect(decoded.id).toBe(0);
|
||||
expect(decoded.type).toBe("error");
|
||||
|
|
@ -266,7 +297,7 @@ describe("TTP communication codec", () => {
|
|||
const corrupted = encoded.slice();
|
||||
corrupted[corrupted.length - 1] = 215;
|
||||
|
||||
const decoded = decodeCommunicationMessage(corrupted);
|
||||
const decoded = decodeCommunicationMessage(frameBytes(corrupted));
|
||||
|
||||
expect(decoded.id).toBe(9);
|
||||
expect(decoded.type).toBe("error");
|
||||
|
|
@ -278,6 +309,172 @@ describe("TTP communication codec", () => {
|
|||
expect((socket as Record<string, unknown>).live_message).toEqual(undefined);
|
||||
});
|
||||
|
||||
test("does not log raw binary frames unless enabled", 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([
|
||||
frameBytes(ignoredFrame),
|
||||
frameBytes(identificationFrame),
|
||||
]);
|
||||
const restoreWebTransport = installFakeWebTransport(
|
||||
chunkBytes(streamBytes, [2, 5, 1, 7]),
|
||||
);
|
||||
const restoreLocalStorage = installFakeLocalStorage();
|
||||
const consoleSpy = installConsoleLogSpy();
|
||||
|
||||
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",
|
||||
},
|
||||
});
|
||||
expect(consoleSpy.calls);
|
||||
|
||||
await client.close("test-complete");
|
||||
} finally {
|
||||
consoleSpy.restore();
|
||||
restoreLocalStorage();
|
||||
restoreWebTransport();
|
||||
}
|
||||
});
|
||||
|
||||
test("logs raw binary frames when enabled", async () => {
|
||||
const requestMessage = encodeCommunicationMessage({
|
||||
id: 7,
|
||||
type: "identification",
|
||||
data: {
|
||||
user_id: 1,
|
||||
},
|
||||
});
|
||||
const requestFrame = frameBytes(requestMessage);
|
||||
|
||||
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 ignoredTransportFrame = frameBytes(ignoredFrame);
|
||||
const identificationTransportFrame = frameBytes(identificationFrame);
|
||||
|
||||
const streamBytes = concatBytes([
|
||||
ignoredTransportFrame,
|
||||
identificationTransportFrame,
|
||||
]);
|
||||
const restoreWebTransport = installFakeWebTransport(
|
||||
chunkBytes(streamBytes, [2, 5, 1, 7]),
|
||||
);
|
||||
const restoreLocalStorage = installFakeLocalStorage();
|
||||
const consoleSpy = installConsoleLogSpy();
|
||||
|
||||
try {
|
||||
localStorage.setItem("ttp_logBinary", "true");
|
||||
|
||||
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",
|
||||
},
|
||||
});
|
||||
|
||||
const logMessages = consoleSpy.calls
|
||||
.filter((entry) => typeof entry[0] === "string")
|
||||
.map((entry) => entry[0] as string);
|
||||
|
||||
expect(logMessages).toContain(
|
||||
`[Socket] Outgoing binary message (${requestFrame.byteLength} bytes)`,
|
||||
);
|
||||
expect(logMessages).toContain(
|
||||
`[Socket] Incoming binary message (${ignoredTransportFrame.byteLength} bytes)`,
|
||||
);
|
||||
expect(logMessages).toContain(
|
||||
`[Socket] Incoming binary message (${identificationTransportFrame.byteLength} bytes)`,
|
||||
);
|
||||
|
||||
const outgoingCall = consoleSpy.calls.find(
|
||||
(entry) =>
|
||||
entry[0] ===
|
||||
`[Socket] Outgoing binary message (${requestFrame.byteLength} bytes)`,
|
||||
);
|
||||
expect(outgoingCall?.[1]).toEqual(requestFrame);
|
||||
|
||||
const incomingCall = consoleSpy.calls.find(
|
||||
(entry) =>
|
||||
entry[0] ===
|
||||
`[Socket] Incoming binary message (${identificationTransportFrame.byteLength} bytes)`,
|
||||
);
|
||||
expect(incomingCall?.[1]).toEqual(identificationTransportFrame);
|
||||
|
||||
await client.close("test-complete");
|
||||
} finally {
|
||||
consoleSpy.restore();
|
||||
restoreLocalStorage();
|
||||
restoreWebTransport();
|
||||
}
|
||||
});
|
||||
|
||||
test("resolves an identification response before the stream closes", async () => {
|
||||
const ignoredFrame = encodeCommunicationMessage({
|
||||
id: 0,
|
||||
|
|
@ -296,22 +493,23 @@ describe("TTP communication codec", () => {
|
|||
},
|
||||
});
|
||||
|
||||
const streamBytes = concatBytes([ignoredFrame, identificationFrame]);
|
||||
const streamBytes = concatBytes([
|
||||
frameBytes(ignoredFrame),
|
||||
frameBytes(identificationFrame),
|
||||
]);
|
||||
const restoreWebTransport = installFakeWebTransport(
|
||||
chunkBytes(streamBytes, [2, 5, 1, 7]),
|
||||
);
|
||||
const restoreLocalStorage = installFakeLocalStorage();
|
||||
|
||||
try {
|
||||
const client = createTransportClient(socket, { url: "https://example.test" });
|
||||
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 response = client.send("identification", { user_id: 1 }, { id: 7 });
|
||||
|
||||
const timeout = new Promise<never>((_, reject) => {
|
||||
setTimeout(() => {
|
||||
|
|
@ -355,22 +553,23 @@ describe("TTP communication codec", () => {
|
|||
},
|
||||
});
|
||||
|
||||
const streamBytes = concatBytes([ignoredFrame, identificationFrame]);
|
||||
const streamBytes = concatBytes([
|
||||
frameBytes(ignoredFrame),
|
||||
frameBytes(identificationFrame),
|
||||
]);
|
||||
const restoreWebTransport = installFakeWebTransport(
|
||||
chunkBytes(streamBytes, [1, 4, 3, 9, 2]),
|
||||
);
|
||||
const restoreLocalStorage = installFakeLocalStorage();
|
||||
|
||||
try {
|
||||
const client = createTransportClient(socket, { url: "https://example.test" });
|
||||
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 response = client.send("identification", { user_id: 1 }, { id: 7 });
|
||||
|
||||
const timeout = new Promise<never>((_, reject) => {
|
||||
setTimeout(() => {
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ const CLOSE_FRAME_LEN = 0xffff_ffff;
|
|||
const APPLICATION_CLOSE_CODE = 0;
|
||||
const APPLICATION_CLOSE_REASON = "ttp-close";
|
||||
const MAX_REQUEST_ID = 0xffff_fffe;
|
||||
const BINARY_LOG_STORAGE_KEY = "ttp_logBinary";
|
||||
|
||||
const DATA_VALUE_KIND_BOOL_TRUE = 0x01;
|
||||
const DATA_VALUE_KIND_BOOL_FALSE = 0x02;
|
||||
|
|
@ -1022,6 +1023,38 @@ function formatUnknownError(error: unknown) {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns whether raw transport binary logging is enabled in local storage.
|
||||
* @returns True when binary transport logs should be emitted.
|
||||
*/
|
||||
function isBinaryMessageLoggingEnabled() {
|
||||
try {
|
||||
return localStorage.getItem(BINARY_LOG_STORAGE_KEY) === "true";
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Logs raw transport bytes when binary logging is enabled.
|
||||
* @param direction Message direction label.
|
||||
* @param payload Raw binary payload to log.
|
||||
* @returns Void.
|
||||
*/
|
||||
function logBinaryMessage(
|
||||
direction: "Incoming" | "Outgoing",
|
||||
payload: Uint8Array,
|
||||
) {
|
||||
if (!isBinaryMessageLoggingEnabled()) {
|
||||
return;
|
||||
}
|
||||
|
||||
console.log(
|
||||
`[Socket] ${direction} binary message (${payload.byteLength} bytes)`,
|
||||
payload,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Detects whether an error chain includes STOP_SENDING.
|
||||
* @param error Unknown transport error.
|
||||
|
|
@ -1154,6 +1187,8 @@ async function writeMessage(transport: WebTransportLike, payload: Uint8Array) {
|
|||
writeU32(frame, 0, payload.byteLength);
|
||||
frame.set(payload, 4);
|
||||
|
||||
logBinaryMessage("Outgoing", frame);
|
||||
|
||||
await writer.write(frame);
|
||||
await writer.close();
|
||||
} finally {
|
||||
|
|
@ -1220,7 +1255,10 @@ async function processIncomingStream(
|
|||
// Ignore close errors during peer shutdown.
|
||||
}
|
||||
|
||||
handleStreamFailure(connection, new Error("Transport closed by peer"));
|
||||
handleStreamFailure(
|
||||
connection,
|
||||
new Error("Transport closed by peer"),
|
||||
);
|
||||
return true;
|
||||
}
|
||||
|
||||
|
|
@ -1232,6 +1270,8 @@ async function processIncomingStream(
|
|||
const frameBytes = bufferedBytes.subarray(0, expectedLength);
|
||||
bufferedBytes = bufferedBytes.subarray(expectedLength);
|
||||
|
||||
logBinaryMessage("Incoming", frameBytes);
|
||||
|
||||
try {
|
||||
handleIncomingFrame(decodeCommunicationMessage(frameBytes));
|
||||
} catch (error) {
|
||||
|
|
@ -1310,10 +1350,17 @@ export function encodeCommunicationMessage(
|
|||
*/
|
||||
export function decodeCommunicationMessage(frame: Uint8Array): TypedMessage {
|
||||
const reader = new ByteReader(frame);
|
||||
const payloadLength = reader.readU32();
|
||||
if (payloadLength !== frame.byteLength - 4) {
|
||||
const frameLength = reader.readU32();
|
||||
if (frameLength !== frame.byteLength - 4) {
|
||||
throw new Error(
|
||||
`Communication payload length mismatch: expected ${payloadLength}, received ${frame.byteLength - 4}`,
|
||||
`Communication frame length mismatch: expected ${frameLength}, received ${frame.byteLength - 4}`,
|
||||
);
|
||||
}
|
||||
|
||||
const payloadLength = reader.readU32();
|
||||
if (payloadLength !== frameLength - 4) {
|
||||
throw new Error(
|
||||
`Communication payload length mismatch: expected ${payloadLength}, received ${frameLength - 4}`,
|
||||
);
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue