fix(mrt): improve Brotli streaming performance

.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.

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);

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<void>((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'}});