From c04ff4b9e415a962d8e5a7113ccbfdecba8d004d Mon Sep 17 00:00:00 2001 From: Austin Turner Date: Sat, 24 Oct 2026 08:42:28 -0700 Subject: [PATCH] feat(load): gzip large bulk batches and pass them through to Salesforce Bulk load batches (up to ~10 MB of CSV) upload from the browser through the API server, and upload bandwidth is often the bottleneck. Web batches over 64 KB are now gzipped, and the server forwards the compressed bytes to the Bulk API, which decodes them itself. The server never decompresses the body, so the inflate: false hardening on every other parser stays in place. Desktop, extension and canvas call Salesforce in-process and are unchanged. --- .../app/controllers/sf-bulk-api.controller.ts | 6 +- .../gzip-body-passthrough.middleware.spec.ts | 91 +++++++++++++++++++ .../utils/gzip-body-passthrough.middleware.ts | 48 ++++++++++ apps/api/src/main.ts | 8 +- .../src/tests/api/bulk.api.spec.ts | 38 ++++++++ libs/salesforce-api/src/lib/api-bulk.ts | 8 +- .../constants/src/lib/shared-constants.ts | 1 + libs/shared/data/src/lib/client-data.ts | 30 +++++- libs/shared/utils/src/lib/compression.ts | 7 +- 9 files changed, 227 insertions(+), 10 deletions(-) create mode 100644 apps/api/src/app/utils/__tests__/gzip-body-passthrough.middleware.spec.ts create mode 100644 apps/api/src/app/utils/gzip-body-passthrough.middleware.ts diff --git a/apps/api/src/app/controllers/sf-bulk-api.controller.ts b/apps/api/src/app/controllers/sf-bulk-api.controller.ts index c11aa74f0d..9b3b7a7ac0 100644 --- a/apps/api/src/app/controllers/sf-bulk-api.controller.ts +++ b/apps/api/src/app/controllers/sf-bulk-api.controller.ts @@ -142,13 +142,15 @@ const closeOrAbortJob = createRoute(routeDefinition.closeOrAbortJob.validators, const addBatchToJob = createRoute( routeDefinition.addBatchToJob.validators, - async ({ body, params, query, jetstreamConn }, _, res, next) => { + async ({ body, params, query, jetstreamConn }, req, res, next) => { try { const jobId = params.jobId; const csv = body; const closeJob = query.closeJob; + // gzipBodyPassthrough left the body compressed - Salesforce decodes it, this server never does + const contentEncoding = req.get(HTTP.HEADERS.CONTENT_ENCODING)?.toLowerCase() === 'gzip' ? 'gzip' : undefined; - const results = await jetstreamConn.bulk.addBatchToJob(csv, jobId, closeJob); + const results = await jetstreamConn.bulk.addBatchToJob(csv, jobId, closeJob, HTTP.CONTENT_TYPE.CSV, contentEncoding); sendJson(res, results); } catch (ex) { diff --git a/apps/api/src/app/utils/__tests__/gzip-body-passthrough.middleware.spec.ts b/apps/api/src/app/utils/__tests__/gzip-body-passthrough.middleware.spec.ts new file mode 100644 index 0000000000..ba31581e3d --- /dev/null +++ b/apps/api/src/app/utils/__tests__/gzip-body-passthrough.middleware.spec.ts @@ -0,0 +1,91 @@ +import { HTTP } from '@jetstream/shared/constants'; +import { raw } from 'body-parser'; +import express from 'express'; +import type { AddressInfo } from 'node:net'; +import { gzipSync } from 'node:zlib'; +import { gzipBodyPassthrough } from '../gzip-body-passthrough.middleware'; + +const CSV = 'Name\nAcme\nGlobex\n'; + +/** Mirrors main.ts: passthrough first, then the global parser that rejects every Content-Encoding */ +function createServer(limitBytes = 1024) { + const app = express(); + app.post('/api/bulk/:jobId', gzipBodyPassthrough({ limitBytes })); + app.use(raw({ limit: '30mb', type: ['text/csv'], inflate: false })); + app.post('/api/bulk/:jobId', (req, res) => { + res.json({ + isBuffer: Buffer.isBuffer(req.body), + bodyBase64: Buffer.isBuffer(req.body) ? req.body.toString('base64') : null, + }); + }); + app.use((err: { status?: number; type?: string }, _req: express.Request, res: express.Response, _next: express.NextFunction) => { + res.status(err.status ?? 500).json({ type: err.type }); + }); + return new Promise<{ url: string; close: () => void }>((resolve) => { + const server = app.listen(0, () => { + const { port } = server.address() as AddressInfo; + resolve({ url: `http://127.0.0.1:${port}`, close: () => server.close() }); + }); + }); +} + +function post(url: string, body: BodyInit, headers: Record) { + return fetch(`${url}/api/bulk/750000000000001`, { method: 'POST', body, headers }); +} + +describe('gzipBodyPassthrough', () => { + let server: Awaited>; + + afterEach(() => { + server?.close(); + }); + + test('passes gzipped csv through as the original compressed bytes', async () => { + server = await createServer(); + const compressed = gzipSync(CSV); + const response = await post(server.url, compressed, { + [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV, + [HTTP.HEADERS.CONTENT_ENCODING]: 'gzip', + }); + expect(response.status).toBe(200); + const result = await response.json(); + expect(result.isBuffer).toBe(true); + expect(Buffer.from(result.bodyBase64, 'base64').equals(compressed)).toBe(true); + }); + + test('leaves uncompressed csv to the regular parser', async () => { + server = await createServer(); + const response = await post(server.url, CSV, { [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV }); + expect(response.status).toBe(200); + const result = await response.json(); + expect(Buffer.from(result.bodyBase64, 'base64').toString('utf8')).toBe(CSV); + }); + + test('rejects a body that is not gzip', async () => { + server = await createServer(); + const response = await post(server.url, CSV, { + [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV, + [HTTP.HEADERS.CONTENT_ENCODING]: 'gzip', + }); + expect(response.status).toBe(400); + expect((await response.json()).type).toBe('encoding.invalid'); + }); + + test('rejects a compressed body over the limit', async () => { + server = await createServer(16); + const response = await post(server.url, gzipSync('x'.repeat(10_000) + CSV), { + [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV, + [HTTP.HEADERS.CONTENT_ENCODING]: 'gzip', + }); + expect(response.status).toBe(413); + }); + + test('leaves other encodings to the global parser, which rejects them', async () => { + server = await createServer(); + const response = await post(server.url, CSV, { + [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV, + [HTTP.HEADERS.CONTENT_ENCODING]: 'br', + }); + expect(response.status).toBe(415); + }); +}); diff --git a/apps/api/src/app/utils/gzip-body-passthrough.middleware.ts b/apps/api/src/app/utils/gzip-body-passthrough.middleware.ts new file mode 100644 index 0000000000..f1a25e5e64 --- /dev/null +++ b/apps/api/src/app/utils/gzip-body-passthrough.middleware.ts @@ -0,0 +1,48 @@ +import { HTTP } from '@jetstream/shared/constants'; +import express from 'express'; + +function createBodyError(status: number, type: string, message: string) { + return Object.assign(new Error(message), { status, type, expose: true }); +} + +/** + * Reads a gzip-encoded CSV body into a Buffer WITHOUT decompressing it, so it can be forwarded to Salesforce + * as-is (Bulk API decodes gzip request bodies itself). + * + * The global body parsers reject every Content-Encoding (inflate: false) to keep untrusted input away from the + * zlib/brotli decoders, so this must be mounted before them - they skip any request whose body was already read. + * An over-limit body is drained before responding (as body-parser does) so the client receives the 413. + */ +export function gzipBodyPassthrough({ limitBytes }: { limitBytes: number }) { + return (req: express.Request, _res: express.Response, next: express.NextFunction) => { + if (req.get(HTTP.HEADERS.CONTENT_ENCODING)?.toLowerCase() !== 'gzip' || !req.is('text/csv')) { + next(); + return; + } + + const chunks: Buffer[] = []; + let receivedBytes = 0; + + req.on('data', (chunk: Buffer) => { + receivedBytes += chunk.length; + if (receivedBytes <= limitBytes) { + chunks.push(chunk); + } + }); + req.on('error', next); + req.on('end', () => { + if (receivedBytes > limitBytes) { + next(createBodyError(413, 'entity.too.large', 'request entity too large')); + return; + } + const body = Buffer.concat(chunks); + // gzip magic bytes - reject anything else here rather than forwarding garbage to Salesforce + if (body[0] !== 0x1f || body[1] !== 0x8b) { + next(createBodyError(400, 'encoding.invalid', 'request body is not valid gzip')); + return; + } + req.body = body; + next(); + }); + }; +} diff --git a/apps/api/src/main.ts b/apps/api/src/main.ts index 573f93f994..16a0417533 100644 --- a/apps/api/src/main.ts +++ b/apps/api/src/main.ts @@ -64,6 +64,7 @@ import { setCacheControlForApiRoutes, setPermissionPolicy, } from './app/routes/route.middleware'; +import { gzipBodyPassthrough } from './app/utils/gzip-body-passthrough.middleware'; import { healthCheck, uncaughtErrorHandler } from './app/utils/response.handlers'; import { buildAppCspDirectives, buildCspDirectives, buildHstsConfig } from './app/utils/security-headers'; import { handleWorkerExit, primaryClusterInitSideEffects, shutdownPrimaryGracefully } from './app/utils/server-process.utils'; @@ -319,8 +320,11 @@ if (ENV.NODE_ENV === 'production' && !ENV.CI && cluster.isPrimary) { // handles the report payload instead of the 20mb global JSON parser, and responds 204. app.use('/api/csp-report', cspReportRoutes); - // inflate: false - no client sends compressed request bodies; rejecting Content-Encoding (415) keeps - // untrusted input away from zlib/brotli decoders (e.g. CVE-2026-111497) + // Gzipped bulk batches are read as raw bytes and forwarded to Salesforce, which decodes them + app.post('/api/bulk/:jobId', gzipBodyPassthrough({ limitBytes: 30 * 1024 * 1024 })); + + // inflate: false - rejecting Content-Encoding (415) keeps untrusted input away from zlib/brotli decoders + // (e.g. CVE-2026-111497); the one compressed body we accept is passed through above, never decoded app.use(raw({ limit: '30mb', type: ['text/csv'], inflate: false })); app.use(raw({ limit: '30mb', type: ['application/zip'], inflate: false })); app.use(json({ limit: '20mb', type: ['json'], inflate: false })); diff --git a/apps/jetstream-e2e/src/tests/api/bulk.api.spec.ts b/apps/jetstream-e2e/src/tests/api/bulk.api.spec.ts index 6572443267..2b9d045a5f 100644 --- a/apps/jetstream-e2e/src/tests/api/bulk.api.spec.ts +++ b/apps/jetstream-e2e/src/tests/api/bulk.api.spec.ts @@ -1,5 +1,6 @@ import { CreateJobRequest } from '@jetstream/api-types'; import { BulkJob, BulkJobBatchInfo, BulkJobWithBatches } from '@jetstream/types'; +import { gzipSync } from 'node:zlib'; import { expect, test } from '../../fixtures/fixtures'; test.describe.configure({ mode: 'parallel' }); @@ -54,6 +55,43 @@ test.describe('API - Bulk', () => { expect(getJobResponse2.state).toEqual('Closed'); }); + test('Bulk Job - Add gzipped Batch', async ({ apiRequestUtils }) => { + const createJobResponse = await apiRequestUtils.makeRequest('POST', `/api/bulk`, { + externalId: null, + serialMode: false, + sObject: 'LEAD', + type: 'INSERT', + } as CreateJobRequest); + + const timestamp = new Date().getTime(); + // The job uses Salesforce's default comma delimiter + const csv = [ + `LastName,FirstName,Company,Email,Status`, + `Gzip-${timestamp},Kathy,TNR Corp. ${timestamp},ksynder@tnr.${timestamp}.net,Working - Contacted`, + ].join('\n'); + + const addBatchToJobResponse = await apiRequestUtils.makeRequest( + 'POST', + `/api/bulk/${createJobResponse.id}?${new URLSearchParams({ closeJob: 'true' })}`, + gzipSync(csv), + { 'Content-Type': 'text/csv; charset=UTF-8', 'Content-Encoding': 'gzip' }, + ); + + expect(addBatchToJobResponse.id).toBeTruthy(); + + // Salesforce decodes the batch itself - a successful insert proves the bytes arrived intact + await expect + .poll( + async () => { + const { batches } = await apiRequestUtils.makeRequest('GET', `/api/bulk/${createJobResponse.id}`); + const [{ state, stateMessage, numberRecordsProcessed, numberRecordsFailed }] = batches; + return { state, stateMessage, numberRecordsProcessed, numberRecordsFailed }; + }, + { timeout: 90_000 }, + ) + .toMatchObject({ state: 'Completed', numberRecordsProcessed: 1, numberRecordsFailed: 0 }); + }); + test('Bulk Job - Errors', async ({ apiRequestUtils }) => { const createJobResponse = await apiRequestUtils.makeRequestRaw('POST', `/api/bulk`, { externalId: null, diff --git a/libs/salesforce-api/src/lib/api-bulk.ts b/libs/salesforce-api/src/lib/api-bulk.ts index 4fde58c6d8..02cc9ae92c 100644 --- a/libs/salesforce-api/src/lib/api-bulk.ts +++ b/libs/salesforce-api/src/lib/api-bulk.ts @@ -68,13 +68,19 @@ export class ApiBulk extends SalesforceApi { jobId: string, closeJob = false, contentType: typeof HTTP.CONTENT_TYPE.CSV | typeof HTTP.CONTENT_TYPE.ZIP_CSV = HTTP.CONTENT_TYPE.CSV, + /** Set when `body` is already gzip-compressed - Salesforce decodes it */ + contentEncoding?: 'gzip', ) { const result = await this.apiRequest<{ batchInfo: BulkJobBatchInfoUntyped }>({ sessionInfo: this.sessionInfo, url: this.getBulkApiUrl(`/job/${jobId}/batch`), method: 'POST', body, - headers: { [HTTP.HEADERS.CONTENT_TYPE]: contentType, Accept: HTTP.CONTENT_TYPE.XML }, + headers: { + [HTTP.HEADERS.CONTENT_TYPE]: contentType, + Accept: HTTP.CONTENT_TYPE.XML, + ...(contentEncoding && { [HTTP.HEADERS.CONTENT_ENCODING]: contentEncoding }), + }, outputType: 'xml', rawBody: true, }).then(({ batchInfo }) => bulkApiEnsureTyped(batchInfo)); diff --git a/libs/shared/constants/src/lib/shared-constants.ts b/libs/shared/constants/src/lib/shared-constants.ts index 1050f9d966..19982e8dc2 100644 --- a/libs/shared/constants/src/lib/shared-constants.ts +++ b/libs/shared/constants/src/lib/shared-constants.ts @@ -102,6 +102,7 @@ export const HTTP = { X_CLIENT_RUNTIME: 'X-Client-Runtime', X_CLIENT_INSTALL: 'X-Client-Install', CONTENT_TYPE: 'Content-Type', + CONTENT_ENCODING: 'Content-Encoding', X_MOCK_KEY: 'X-MOCK-KEY', X_FORWARDED_FOR: 'X-FORWARDED-FOR', CONTENT_DISPOSITION: 'Content-Disposition', diff --git a/libs/shared/data/src/lib/client-data.ts b/libs/shared/data/src/lib/client-data.ts index 2ce491ae59..aa345af7d0 100644 --- a/libs/shared/data/src/lib/client-data.ts +++ b/libs/shared/data/src/lib/client-data.ts @@ -23,6 +23,7 @@ import type { import { logger } from '@jetstream/shared/client-logger'; import { HTTP, MIME_TYPES } from '@jetstream/shared/constants'; import { + gzipEncodeText, hasIncompleteSubqueries, isIncompleteSubqueryResult, isSubqueryResult, @@ -99,7 +100,7 @@ import { parseISO } from 'date-fns/parseISO'; import isFunction from 'lodash/isFunction'; import isNil from 'lodash/isNil'; import orderBy from 'lodash/orderBy'; -import { handleExternalRequest, handleRequest, transformListMetadataResponse } from './client-data-data-helper'; +import { AxiosAdapterConfig, handleExternalRequest, handleRequest, transformListMetadataResponse } from './client-data-data-helper'; import { getStandardValueSetTypes } from './standardValueSet'; export interface PaginationCursorParams { @@ -1327,19 +1328,42 @@ export async function bulkApiAbortJob(org: SalesforceOrgUi, jobId: string): Prom return handleRequest({ method: 'DELETE', url: `/api/bulk/${jobId}/abort` }, { org }).then(unwrapResponseIgnoreCache); } +const BULK_BATCH_GZIP_MIN_LENGTH = 64 * 1024; + +/** + * Gzip large batches when they travel over the network to the Jetstream server, which forwards the compressed + * bytes to Salesforce untouched. Desktop, extension and canvas set a custom adapter and call Salesforce + * in-process, so their batches are left as plain text. + */ +async function getBulkBatchRequestBody(csv: string): Promise<{ data: string | Uint8Array; contentEncoding?: 'gzip' }> { + if (AxiosAdapterConfig.adapter || csv.length < BULK_BATCH_GZIP_MIN_LENGTH || typeof CompressionStream === 'undefined') { + return { data: csv }; + } + try { + return { data: await gzipEncodeText(csv), contentEncoding: 'gzip' }; + } catch (ex) { + logger.warn('[BULK] Failed to gzip batch, sending uncompressed', ex); + return { data: csv }; + } +} + export async function bulkApiAddBatchToJob( org: SalesforceOrgUi, jobId: string, csv: string, closeJob?: boolean, ): Promise { + const { data, contentEncoding } = await getBulkBatchRequestBody(csv); return handleRequest( { method: 'POST', url: `/api/bulk/${jobId}`, - data: csv, + data, params: { closeJob }, - headers: { [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV }, + headers: { + [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV, + ...(contentEncoding && { [HTTP.HEADERS.CONTENT_ENCODING]: contentEncoding }), + }, }, { org }, ) diff --git a/libs/shared/utils/src/lib/compression.ts b/libs/shared/utils/src/lib/compression.ts index b4eb16a7bb..1c97ce0a47 100644 --- a/libs/shared/utils/src/lib/compression.ts +++ b/libs/shared/utils/src/lib/compression.ts @@ -2,8 +2,11 @@ const TEXT_ENCODER = new TextEncoder(); const TEXT_DECODER = new TextDecoder(); export async function gzipEncode(value: unknown): Promise> { - const json = JSON.stringify(value); - const stream = new Blob([TEXT_ENCODER.encode(json) as BlobPart]).stream().pipeThrough(new CompressionStream('gzip')); + return gzipEncodeText(JSON.stringify(value)); +} + +export async function gzipEncodeText(value: string): Promise> { + const stream = new Blob([TEXT_ENCODER.encode(value) as BlobPart]).stream().pipeThrough(new CompressionStream('gzip')); const buffer = await new Response(stream).arrayBuffer(); return new Uint8Array(buffer); }