diff --git a/packages/web/server/lib/relay/tunnel-host.js b/packages/web/server/lib/relay/tunnel-host.js index a3ec6089..b052b385 100644 --- a/packages/web/server/lib/relay/tunnel-host.js +++ b/packages/web/server/lib/relay/tunnel-host.js @@ -96,10 +96,11 @@ const isWsClosePayload = (parsed) => Boolean(parsed && typeof parsed === 'object * getLocalPort: () => number, * sendFrame: (plaintextFrame: Uint8Array) => void | Promise, * getBufferedAmount: () => number, + * bodyDeliveryTimeoutMs?: number, * }} deps */ -export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBufferedAmount }) => { - /** @type {Map} */ +export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBufferedAmount, bodyDeliveryTimeoutMs = BODY_DELIVERY_TIMEOUT_MS }) => { + /** @type {Map} */ const streams = new Map(); const assembler = createFragmentAssembler(); let closed = false; @@ -329,8 +330,12 @@ export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBuf if (streams.get(streamId) === stream && !completed && !liveStream) { dropStream(streamId); void sendAbort(streamId, 'tunnel request body was not delivered in time'); + // Settle the buffered-body wait below so the stream's buffered chunks + // and this call frame are released (the post-wait guard sees the + // dropped stream and returns without a second abort). + finishBody(new Error('tunnel request body was not delivered in time')); } - }, BODY_DELIVERY_TIMEOUT_MS); + }, bodyDeliveryTimeoutMs); deliveryDeadline.unref?.(); await bodyEnded; diff --git a/packages/web/server/lib/relay/tunnel-host.test.js b/packages/web/server/lib/relay/tunnel-host.test.js index cda4da33..382e2c55 100644 --- a/packages/web/server/lib/relay/tunnel-host.test.js +++ b/packages/web/server/lib/relay/tunnel-host.test.js @@ -28,7 +28,7 @@ const startLoopback = () => })); }); -const createHarness = async () => { +const createHarness = async (hostOverrides = {}) => { const loopback = await startLoopback(); const sentFrames = []; const host = createTunnelHost({ @@ -38,6 +38,7 @@ const createHarness = async () => { sentFrames.push(decodeTunnelFrame(frame)); }, getBufferedAmount: () => 0, + ...hostOverrides, }); return { host, loopback, sentFrames }; }; @@ -98,6 +99,36 @@ describe('tunnel-host HTTP body forwarding', () => { await loopback.stop(); }); + test('aborts a buffered body that never completes within the delivery deadline', async () => { + const { host, loopback, sentFrames } = await createHarness({ bodyDeliveryTimeoutMs: 50 }); + await host.handleFrame(httpHead({ hasBody: true })); + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpBody, 1, new TextEncoder().encode('partial'))); + // No StreamEnd — the tunnel stalled mid-body. + + const aborted = await waitFor(() => sentFrames.some((f) => f.frameType === TunnelFrameType.StreamAbort)); + expect(aborted).toBe(true); + expect(loopback.requests.length).toBe(0); + // A late StreamEnd for the dropped stream must not trigger a second abort + // or forward the stale body. + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.StreamEnd, 1, new Uint8Array(0))); + await new Promise((r) => setTimeout(r, 50)); + expect(sentFrames.filter((f) => f.frameType === TunnelFrameType.StreamAbort).length).toBe(1); + expect(loopback.requests.length).toBe(0); + await loopback.stop(); + }); + + test('forwards an empty body when the client delivered an explicit empty frame', async () => { + const { host, loopback } = await createHarness(); + await host.handleFrame(httpHead({ hasBody: true })); + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpBody, 1, new Uint8Array(0))); + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.StreamEnd, 1, new Uint8Array(0))); + + const received = await waitFor(() => loopback.requests.length === 1); + expect(received).toBe(true); + expect(loopback.requests[0].body).toBe(''); + await loopback.stop(); + }); + test('GET forwards immediately with no body wait', async () => { const { host, loopback } = await createHarness(); await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpRequest, 1, encodeJsonPayload({