Stream passthrough bodies directly without wrapping where possible

This provides a significant body streaming throughput boost. It won't
take effect either by default (recordTraffic: true) or in HTTP Toolkit
(which currently uses 'request' events) but it will for most cases that
disable recording and HTTP Toolkit in future (once we start streaming
body chunks instead of buffering via those new event APIs).

Author: pimterry

src/rules/requests/request-rule.ts

@@ -81,14 +81,16 @@ export class RequestRule implements RequestRule {
         record?: boolean,
         debug: boolean,
         keyLogStream?: Writable,
-        emitEventCallback?: (type: string, event: unknown) => void
+        emitEventCallback?: (type: string, event: unknown) => void,
+        reqBodyObserved?: boolean
     }): Promise<void> {
         let stepsPromise = (async () => {
             for (let step of this.steps) {
                 const result = await step.handle(req, res, {
                     emitEventCallback: options.emitEventCallback,
                     keyLogStream: options.keyLogStream,
-                    debug: options.debug
+                    debug: options.debug,
+                    reqBodyObserved: options.reqBodyObserved
                 });
 
                 if (!result || result.continue === false) break;

src/rules/requests/request-step-impls.ts

@@ -1,5 +1,5 @@
 import { Buffer } from 'buffer';
-import { Writable } from 'stream';
+import { Readable, Writable } from 'stream';
 import * as url from 'url';
 import type * as dns from 'dns';
 import * as net from 'net';
@@ -171,6 +171,7 @@ export interface RequestStepOptions {
     emitEventCallback?: (type: string, event: unknown) => void;
     keyLogStream?: Writable;
     debug: boolean;
+    reqBodyObserved?: boolean; // True if the request body will be buffered for inspection (e.g. recordTraffic)
 }
 
 export class FixedResponseStepImpl extends FixedResponseStep {
@@ -488,9 +489,13 @@ export class PassThroughStepImpl extends PassThroughStep {
             `);
         }
 
-        // We have to capture the request stream immediately, to make sure nothing is lost if it
-        // goes past its max length (truncating the data) before we start sending upstream.
-        const clientReqBody = clientReq.body.asStream();
+        // When transforms or callbacks need the body, we capture it immediately via asStream()
+        // (which buffers data and provides replay). When no transforms are configured, we defer
+        // to pipe the raw request stream directly, avoiding the intermediate PassThrough.
+        const needsBufferedBody = !!(this.transformRequest || this.beforeRequest);
+        const clientReqBody: Readable = needsBufferedBody
+            ? clientReq.body.asStream()
+            : clientReq as any as Readable;
 
         const isH2Downstream = isHttp2(clientReq);
 
@@ -1170,6 +1175,12 @@ export class PassThroughStepImpl extends PassThroughStep {
                 if (reqBodyOverride.length > 0) serverReq.end(reqBodyOverride);
                 else serverReq.end(); // http2-wrapper fails given an empty buffer for methods that aren't allowed a body
             } else {
+                if (!needsBufferedBody && options.reqBodyObserved) {
+                    // Start body buffering now (same tick as pipe) so both the buffer
+                    // and the pipe receive all data via Node's multi-listener support.
+                    // This is needed for recordTraffic and async event emission.
+                    clientReq.body.asBuffer().catch(() => {});
+                }
                 // asStream includes all content, including the body before this call
                 clientReqBody.pipe(serverReq);
                 clientReqBody.on('error', () => serverReq.abort());

src/server/mockttp-server.ts

@@ -753,7 +753,11 @@ export class MockttpServer extends AbstractMockttp implements Mockttp {
                     keyLogStream: this.keyLogStream,
                     emitEventCallback: (requestEmitter.listenerCount('rule-event') !== 0)
                         ? (type, event) => this.announceRuleEventAsync(requestEmitter, request.id, nextRule!.id, type, event)
-                        : undefined
+                        : undefined,
+                    // When something will inspect the request body (traffic recording,
+                    // request event listeners) it must be buffered in memory. Otherwise
+                    // it can be streamed directly to the upstream target.
+                    reqBodyObserved: this.recordTraffic || requestEmitter.listenerCount('request') > 0
                 });
             } else {
                 await this.sendUnmatchedRequestError(request, response);

src/util/buffer-utils.ts

@@ -1,5 +1,4 @@
 import { Buffer } from 'buffer';
-import { EventEmitter } from 'events';
 import * as stream from 'stream';
 
 import { isNode } from './util';
@@ -19,7 +18,7 @@ export const asBuffer = (input: Buffer | Uint8Array | string) =>
 export type BufferInProgress = Promise<Buffer> & {
     currentChunks: Buffer[]; // Stores the body chunks as they arrive
     failedWith?: Error; // Stores the error that killed the stream, if one did
-    events: EventEmitter; // Emits events - notably 'truncate' if data is truncated
+    onTruncate?: (chunks: Buffer[]) => void; // Called if data is truncated (maxSize exceeded)
 };
 
 // Takes a buffer and a stream, returns a simple stream that outputs the buffer then the stream. The stream
@@ -66,11 +65,11 @@ export const bufferThenStream = (buffer: BufferInProgress, inputStream: stream.R
         }
     });
 
-    buffer.events.on('truncate', (chunks) => {
+    buffer.onTruncate = () => {
         // If the stream hasn't started yet, start it now, so it grabs the buffer
         // data before it gets truncated:
         if (!active) outputStream.read(0);
-    });
+    };
 
     return outputStream;
 };
@@ -106,7 +105,7 @@ export const streamToBuffer = (input: stream.Readable, maxSize = MAX_BUFFER_SIZE
                 if (currentSize > maxSize) {
                     // Announce truncation, so that other mechanisms (asStream) can
                     // capture this data if they're interested in it.
-                    bufferPromise.events.emit('truncate', chunks);
+                    bufferPromise.onTruncate?.(chunks);
 
                     // Drop all the data so far & stop reading
                     bufferPromise.currentChunks = chunks = [];
@@ -139,7 +138,6 @@ export const streamToBuffer = (input: stream.Readable, maxSize = MAX_BUFFER_SIZE
         }
     );
     bufferPromise.currentChunks = chunks;
-    bufferPromise.events = new EventEmitter();
     return bufferPromise;
 };
 

test/performance/performance-tests.perf.ts

@@ -11,7 +11,7 @@ nodeOnly(() => {
         this.timeout(120000);
 
         describe("Static HTTP mocking", () => {
-            const server = getLocal();
+            const server = getLocal({ recordTraffic: false });
 
             before(async () => {
                 await server.start();
@@ -90,6 +90,7 @@ nodeOnly(() => {
         describe("HTTPS traffic", () => {
 
             const server = getLocal({
+                recordTraffic: false,
                 https: {
                     keyPath: './test/fixtures/test-ca.key',
                     certPath: './test/fixtures/test-ca.pem'
@@ -125,8 +126,8 @@ nodeOnly(() => {
         });
 
         describe("HTTP proxying", () => {
-            const proxyServer = getLocal();
-            const targetServer = getLocal();
+            const proxyServer = getLocal({ recordTraffic: false });
+            const targetServer = getLocal({ recordTraffic: false });
 
             before(async () => {
                 await targetServer.start();
@@ -150,7 +151,7 @@ nodeOnly(() => {
 
                 printResults("HTTP proxy passthrough", result);
 
-                // Seeing ~2500k req/sec locally for proxy + static server
+                // Seeing ~2500 req/sec locally for proxy + static server
                 assertPerformance(result, {
                     minThroughput: 500,
                     maxP99Latency: 50,
@@ -196,13 +197,15 @@ nodeOnly(() => {
         describe("HTTPS proxying", () => {
 
             const proxyServer = getLocal({
+                recordTraffic: false,
                 https: {
                     keyPath: './test/fixtures/test-ca.key',
                     certPath: './test/fixtures/test-ca.pem'
                 }
             });
 
             const targetServer = getLocal({
+                recordTraffic: false,
                 https: {
                     keyPath: './test/fixtures/test-ca.key',
                     certPath: './test/fixtures/test-ca.pem'
@@ -240,5 +243,39 @@ nodeOnly(() => {
             });
         });
 
+        describe("HTTP proxying with traffic recording", () => {
+            const proxyServer = getLocal({ recordTraffic: true });
+            const targetServer = getLocal({ recordTraffic: false });
+
+            before(async () => {
+                await targetServer.start();
+                await proxyServer.start();
+            });
+
+            after(async () => {
+                await proxyServer.stop();
+                await targetServer.stop();
+            });
+
+            it("for Mockttp proxy with recordTraffic enabled", async () => {
+                await targetServer.forGet("/target").thenReply(200, "Target response");
+                await proxyServer.forGet("/proxy").thenForwardTo(targetServer.url);
+
+                const result = await runPerformanceTest({
+                    url: proxyServer.urlFor("/proxy"),
+                    duration: 10,
+                    connections: 10
+                });
+
+                printResults("HTTP proxy passthrough (recordTraffic)", result);
+
+                assertPerformance(result, {
+                    minThroughput: 500,
+                    maxP99Latency: 50,
+                    maxErrors: 0
+                });
+            });
+        });
+
     });
 });