diff --git a/CHANGELOG.md b/CHANGELOG.md index b071deb..96c8df1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ All notable changes to `@openclaw/libterminal` will be documented in this file. ### Fixed +- Split large Worker output acknowledgements into decodable chunks so flow-control accounting stays valid above one MiB. + - Complete fanout delivery and teardown before event callbacks, preserving output order during reentrant publication and closing all subscribers when callbacks throw. - Preserve leading U+FEFF characters in terminal session IDs instead of routing frames under a different identity. diff --git a/src/worker.ts b/src/worker.ts index 880bd14..bd2de1d 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -49,6 +49,7 @@ export type WebSocketBridge = { }; const encoder = new TextEncoder(); +const MAX_OUTPUT_ACKNOWLEDGEMENT_BYTES = 1024 * 1024; export function bridgeWebSockets( left: WebSocketLike, @@ -279,7 +280,7 @@ export function decodeOutputAcknowledgement(value: WebSocketPayload): number | n return parsed.type === "ack" && Number.isInteger(bytes) && Number(bytes) > 0 && - Number(bytes) <= 1024 * 1024 + Number(bytes) <= MAX_OUTPUT_ACKNOWLEDGEMENT_BYTES ? Number(bytes) : null; } catch { @@ -292,8 +293,14 @@ export function terminalMessageByteLength(value: WebSocketPayload): number { } export function sendOutputAcknowledgement(socket: WebSocketLike, bytes: number): void { - if (bytes > 0 && socket.readyState === WEB_SOCKET_OPEN) { - socket.send(JSON.stringify({ type: "ack", bytes })); + if (!Number.isSafeInteger(bytes) || bytes <= 0) { + return; + } + // Each emitted acknowledgement must fit the same limit enforced by the decoder. + while (bytes > 0 && socket.readyState === WEB_SOCKET_OPEN) { + const acknowledgedBytes = Math.min(bytes, MAX_OUTPUT_ACKNOWLEDGEMENT_BYTES); + socket.send(JSON.stringify({ type: "ack", bytes: acknowledgedBytes })); + bytes -= acknowledgedBytes; } } diff --git a/test/worker.test.ts b/test/worker.test.ts index 6644aef..a78a5a8 100644 --- a/test/worker.test.ts +++ b/test/worker.test.ts @@ -5,6 +5,7 @@ import { bridgeWebSockets, decodeOutputAcknowledgement, normalizeWebSocketMessageData, + sendOutputAcknowledgement, } from "../src/worker.js"; function acceptsNativeWebSocket(socket: WebSocket): void { @@ -274,6 +275,40 @@ describe("bridgeWebSockets", () => { expect(bridge.rightOutputAcknowledgementBytes).toBe(0); }); + it.each([false, true])( + "accounts for large output with immediate acknowledgements %s", + async (immediate) => { + const left = new FakeWebSocket(); + const right = new FakeWebSocket(); + const byteLength = 1024 * 1024 + 1; + const bridge = bridgeWebSockets(left, right, { + controlCheckIntervalMs: 0, + acknowledgeRightOutputImmediately: immediate, + forwardRightOutputAcknowledgements: !immediate, + }); + try { + right.receive(new ArrayBuffer(byteLength)); + await vi.waitFor(() => expect(left.sent).toHaveLength(1)); + if (!immediate) { + const acknowledgements = new FakeWebSocket(); + sendOutputAcknowledgement(acknowledgements, byteLength); + for (const acknowledgement of acknowledgements.sent) { + left.receive(acknowledgement); + } + } + await vi.waitFor(() => expect(right.sent).toHaveLength(2)); + expect(right.sent.map((message) => decodeOutputAcknowledgement(message))).toEqual([ + 1024 * 1024, + 1, + ]); + expect(bridge.rightOutputAcknowledgementBytes).toBe(0); + } finally { + bridge.close(); + await bridge.completed; + } + }, + ); + it("sanitizes abnormal peer close metadata", () => { const left = new FakeWebSocket(); const right = new FakeWebSocket(); @@ -323,6 +358,18 @@ describe("bridgeWebSockets", () => { }); describe("Worker message helpers", () => { + it.each([1024 * 1024, 1024 * 1024 + 1, 2 * 1024 * 1024 + 17])( + "emits decodable acknowledgements for %i bytes", + (bytes) => { + const socket = new FakeWebSocket(); + sendOutputAcknowledgement(socket, bytes); + const counts = socket.sent.map((message) => decodeOutputAcknowledgement(message)); + expect(counts.every((count) => count !== null)).toBe(true); + expect(counts.reduce((total, count) => total + count!, 0)).toBe(bytes); + expect(counts).toHaveLength(Math.ceil(bytes / (1024 * 1024))); + }, + ); + it("normalizes views without leaking unrelated backing-buffer bytes", async () => { const backing = Uint8Array.from([1, 2, 3, 4]); const normalized = await normalizeWebSocketMessageData(backing.subarray(1, 3));