From 2ef9c6ccb983a8766f881229ca1e6b48c3a3d5fc Mon Sep 17 00:00:00 2001 From: Charles Lavery Date: Fri, 21 Aug 2026 14:33:31 -0400 Subject: [PATCH] fix(mrt): improve Brotli streaming performance --- .changeset/fast-brotli-streaming.md | 5 ++ .../src/streaming/create-lambda-adapter.ts | 37 ++++++++++++-- .../create-lambda-adapter-compression.test.ts | 51 +++++++++++++++++++ 3 files changed, 89 insertions(+), 4 deletions(-) create mode 100644 .changeset/fast-brotli-streaming.md diff --git a/.changeset/fast-brotli-streaming.md b/.changeset/fast-brotli-streaming.md new file mode 100644 index 000000000..fd7f9be55 --- /dev/null +++ b/.changeset/fast-brotli-streaming.md @@ -0,0 +1,5 @@ +--- +'@salesforce/mrt-utilities': patch +--- + +Improve streamed response performance by using a runtime-appropriate Brotli quality and periodically flushing compressed output. diff --git a/packages/mrt-utilities/src/streaming/create-lambda-adapter.ts b/packages/mrt-utilities/src/streaming/create-lambda-adapter.ts index 3c4179dcb..988d0c816 100644 --- a/packages/mrt-utilities/src/streaming/create-lambda-adapter.ts +++ b/packages/mrt-utilities/src/streaming/create-lambda-adapter.ts @@ -28,6 +28,11 @@ import {ServerlessRequest} from '@h4ad/serverless-adapter'; */ const REQUEST_HEADERS_TO_COPY = ['x-correlation-id'] as const; +// Node's Brotli default is quality 11, which is optimized for offline compression +// and can consume seconds of Lambda CPU for large streamed HTML responses. +const DEFAULT_BROTLI_QUALITY = 6; +const BROTLI_FLUSH_THRESHOLD_BYTES = 32 * 1024; + // Check if zstd compression is available (Node.js v22.15.0+) let createZstdCompress: ((options?: ZstdOptions) => ZstdCompress) | undefined; try { @@ -306,7 +311,7 @@ function isCompressible(contentType: string | undefined): boolean { return !!compressible(contentType); } -const isNullOrUndefined = (value: unknown): boolean => value == null; +const isNullOrUndefined = (value: unknown): value is null | undefined => value == null; /** * Determines the best encoding based on Accept-Encoding header using the negotiator package @@ -357,8 +362,22 @@ function getBestEncoding( function createCompressionStream(encoding: string, compressionConfig?: CompressionConfig): CompressionStream { const options = compressionConfig?.options || undefined; switch (encoding) { - case 'br': - return zlib.createBrotliCompress(options as BrotliOptions); + case 'br': { + const brotliOptions = options as BrotliOptions | undefined; + const qualityParameter = zlib.constants.BROTLI_PARAM_QUALITY; + + if (brotliOptions?.params?.[qualityParameter] !== undefined) { + return zlib.createBrotliCompress(brotliOptions); + } + + return zlib.createBrotliCompress({ + ...brotliOptions, + params: { + ...brotliOptions?.params, + [qualityParameter]: DEFAULT_BROTLI_QUALITY, + }, + }); + } case 'zstd': if (!createZstdCompress) { throw new Error('zstd compression is not available in this Node.js version (requires v22.15.0+)'); @@ -414,6 +433,7 @@ export function createExpressResponse( let compressionStream: CompressionStream | null = null; let shouldCompress = false; let compressionInitialized = false; + let uncompressedBytesSinceBrotliFlush = 0; // Helper function to check if stream is still writable const isStreamOpen = (): boolean => { @@ -471,7 +491,16 @@ export function createExpressResponse( try { if (shouldCompress && compressionStream && compressionStream.writable) { // Write to compression stream, which will compress and pipe to httpResponseStream - return compressionStream.write(chunk); + const accepted = compressionStream.write(chunk); + if (selectedEncoding === 'br' && typeof compressionStream.flush === 'function') { + uncompressedBytesSinceBrotliFlush += typeof chunk === 'string' ? Buffer.byteLength(chunk) : chunk.byteLength; + + if (uncompressedBytesSinceBrotliFlush >= BROTLI_FLUSH_THRESHOLD_BYTES) { + compressionStream.flush(zlib.constants.BROTLI_OPERATION_FLUSH); + uncompressedBytesSinceBrotliFlush = 0; + } + } + return accepted; } else if (httpResponseStream && httpResponseStream.writable) { // No compression, write directly to httpResponseStream return httpResponseStream.write(chunk); diff --git a/packages/mrt-utilities/test/streaming/create-lambda-adapter-compression.test.ts b/packages/mrt-utilities/test/streaming/create-lambda-adapter-compression.test.ts index 22bc6b765..79a6c6247 100644 --- a/packages/mrt-utilities/test/streaming/create-lambda-adapter-compression.test.ts +++ b/packages/mrt-utilities/test/streaming/create-lambda-adapter-compression.test.ts @@ -212,6 +212,57 @@ describe('Compression Streaming', () => { }); describe('Brotli compression', () => { + it('should default to a streaming-appropriate compression quality', async () => { + const stream = createCollectingStream(); + const event = createMockEvent({headers: {'Accept-Encoding': 'br'}}); + const context = createMockContext(); + const request = createExpressRequest(event, context); + const createBrotliStub = sinon.stub(zlib, 'createBrotliCompress').callThrough(); + const response = createExpressResponse(stream, event, context, request); + + response.setHeader('Content-Type', 'text/html'); + response.end('test data'); + + await stream.waitForEnd(); + + expect( + createBrotliStub.calledWith({ + params: { + [zlib.constants.BROTLI_PARAM_QUALITY]: 6, + }, + }), + ).to.be.true; + createBrotliStub.restore(); + }); + + it('should emit compressed bytes before the response ends', async () => { + const stream = createCollectingStream(); + const event = createMockEvent({headers: {'Accept-Encoding': 'br'}}); + const context = createMockContext(); + const request = createExpressRequest(event, context); + const response = createExpressResponse(stream, event, context, request); + const firstCompressedChunk = new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error('Brotli output was not flushed before end()')), 500); + stream.once('data', () => { + clearTimeout(timeout); + resolve(); + }); + }); + + response.setHeader('Content-Type', 'text/html'); + response.write('streamed html '.repeat(3_000)); + + await firstCompressedChunk; + expect(stream.getData().length).to.be.greaterThan(0); + + response.end('final chunk'); + await stream.waitForEnd(); + + expect(zlib.brotliDecompressSync(stream.getData()).toString()).to.equal( + `${'streamed html '.repeat(3_000)}final chunk`, + ); + }); + it('should compress content with brotli when br is preferred', async () => { const stream = createCollectingStream(); const event = createMockEvent({headers: {'Accept-Encoding': 'br, gzip, deflate'}});