refactor(relay): release buffered request bodies on the delivery deadline
Settle the buffered-body wait when the delivery deadline fires so the stream's buffered chunks are freed immediately, and make the deadline injectable for tests. Covers the stalled-mid-body path with a test.
This commit is contained in:
@@ -96,10 +96,11 @@ const isWsClosePayload = (parsed) => Boolean(parsed && typeof parsed === 'object
|
|||||||
* getLocalPort: () => number,
|
* getLocalPort: () => number,
|
||||||
* sendFrame: (plaintextFrame: Uint8Array) => void | Promise<void>,
|
* sendFrame: (plaintextFrame: Uint8Array) => void | Promise<void>,
|
||||||
* getBufferedAmount: () => number,
|
* getBufferedAmount: () => number,
|
||||||
|
* bodyDeliveryTimeoutMs?: number,
|
||||||
* }} deps
|
* }} deps
|
||||||
*/
|
*/
|
||||||
export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBufferedAmount }) => {
|
export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBufferedAmount, bodyDeliveryTimeoutMs = BODY_DELIVERY_TIMEOUT_MS }) => {
|
||||||
/** @type {Map<number, { kind: 'http', abort: AbortController, body: ReadableStreamDefaultController | null } | { kind: 'ws', socket: WebSocket, opened: boolean }>} */
|
/** @type {Map<number, { kind: 'http', abort: AbortController, body: { enqueue(payload: Uint8Array): void, close(): void, error(error: Error): void } | null, noBody: boolean } | { kind: 'ws', socket: WebSocket, opened: boolean }>} */
|
||||||
const streams = new Map();
|
const streams = new Map();
|
||||||
const assembler = createFragmentAssembler();
|
const assembler = createFragmentAssembler();
|
||||||
let closed = false;
|
let closed = false;
|
||||||
@@ -329,8 +330,12 @@ export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBuf
|
|||||||
if (streams.get(streamId) === stream && !completed && !liveStream) {
|
if (streams.get(streamId) === stream && !completed && !liveStream) {
|
||||||
dropStream(streamId);
|
dropStream(streamId);
|
||||||
void sendAbort(streamId, 'tunnel request body was not delivered in time');
|
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?.();
|
deliveryDeadline.unref?.();
|
||||||
|
|
||||||
await bodyEnded;
|
await bodyEnded;
|
||||||
|
|||||||
@@ -28,7 +28,7 @@ const startLoopback = () =>
|
|||||||
}));
|
}));
|
||||||
});
|
});
|
||||||
|
|
||||||
const createHarness = async () => {
|
const createHarness = async (hostOverrides = {}) => {
|
||||||
const loopback = await startLoopback();
|
const loopback = await startLoopback();
|
||||||
const sentFrames = [];
|
const sentFrames = [];
|
||||||
const host = createTunnelHost({
|
const host = createTunnelHost({
|
||||||
@@ -38,6 +38,7 @@ const createHarness = async () => {
|
|||||||
sentFrames.push(decodeTunnelFrame(frame));
|
sentFrames.push(decodeTunnelFrame(frame));
|
||||||
},
|
},
|
||||||
getBufferedAmount: () => 0,
|
getBufferedAmount: () => 0,
|
||||||
|
...hostOverrides,
|
||||||
});
|
});
|
||||||
return { host, loopback, sentFrames };
|
return { host, loopback, sentFrames };
|
||||||
};
|
};
|
||||||
@@ -98,6 +99,36 @@ describe('tunnel-host HTTP body forwarding', () => {
|
|||||||
await loopback.stop();
|
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 () => {
|
test('GET forwards immediately with no body wait', async () => {
|
||||||
const { host, loopback } = await createHarness();
|
const { host, loopback } = await createHarness();
|
||||||
await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpRequest, 1, encodeJsonPayload({
|
await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpRequest, 1, encodeJsonPayload({
|
||||||
|
|||||||
Reference in New Issue
Block a user