perf!: serialize requests once in batchAddRequests (#1051)

### Description

- Mirroring apify/apify-client-python#953.
- `batchAddRequests` stringified every request twice: once in
`sliceArrayByByteLength` to measure the batch, and again in the
`serializeRequest` interceptor when sending. Axios then ran its default
`transformRequest` on top, because `transformRequest: undefined` in the
instance config falls back to the defaults, and that default validates a
JSON string body by parsing it in full. Each batch body was stringified
twice and parsed once more.
- Each request is now serialized once up front. The byte lengths decide
the batch boundaries (commas and brackets counted, which the old
measurement skipped) and the same strings are joined into the batch
body, sent with an explicit `content-type: application/json`. The
interceptor passes a string body with an explicit content type through
untouched, and the axios instance sets `transformRequest` and
`transformResponse` to `[]`. That also removes the validation re-parse
for every JSON body the client sends.

### Issue

- Closes #972

### Breaking changes

- A string body declared as JSON but not valid JSON (for example
`setRecord` with `contentType: 'application/json'` and a non-JSON
string) is sent as it is. Axios used to double-encode it into a JSON
string literal.
- A request too large for the payload limit is rejected before any batch
is sent. It used to be detected only when its batch came up, with
earlier batches already in flight.
- The protected `_batchAddRequests` and `_batchAddRequestsWithRetries`
take the serialized entries, and `_batchAddRequests` no longer
re-validates a batch the public method already validated. The API report
is updated.

*✍️ Drafted by Claude Code*

Author: vdusek

docs/04_upgrading/upgrading_v3.md

@@ -272,6 +272,37 @@ A trailing slash appears only on a field the API returns without a path, so on `
 
 A URL field whose value isn't a valid absolute URL now fails response validation and throws <ApiLink to="class/ResponseValidationError">`ResponseValidationError`</ApiLink>, the same as any other field that doesn't match the specification.
 
+## `batchAddRequests()` rejects an oversized request before sending anything
+
+<ApiLink to="class/RequestQueueClient#batchAddRequests">`batchAddRequests()`</ApiLink> now measures the whole input before it sends the first batch. A request too large for the payload limit still throws the same error, and the message still names the request's index. What changed is that nothing has been sent by the time it throws. In v2 each batch was measured as it came up, so every batch before the oversized request had already gone out.
+
+```js
+// In v2 the four batches before the oversized request had already been sent.
+// In v3 nothing is sent.
+await client.requestQueue('my-queue').batchAddRequests([...hundredRequests, oversizedRequest]);
+```
+
+Code that treated the throw as "some of these landed" and reconciled the queue afterwards can drop the reconciliation.
+
+## A string body with a JSON content type is sent as it is
+
+A string body sent with an explicit `application/json` content type now goes out verbatim. Axios used to parse it to check that it was valid JSON, and wrapped it in a JSON string literal when it wasn't. The client skips that step so that `batchAddRequests()` can send a body it has already serialized.
+
+<ApiLink to="class/KeyValueStoreClient#setRecord">`setRecord()`</ApiLink> is where you'd notice. Storing a non-JSON string under `contentType: 'application/json'` used to save `"my value"`, quotes included, and now saves `my value`, which <ApiLink to="class/KeyValueStoreClient#getRecord">`getRecord()`</ApiLink> can't parse back:
+
+```js
+// v2 stored `"my value"`, v3 stores `my value`.
+await client.keyValueStore('my-store').setRecord({
+    key: 'my-key',
+    value: 'my value',
+    contentType: 'application/json',
+});
+```
+
+Give a record a content type matching what it holds, such as `text/plain`, or pass the value as an object and let the client serialize it.
+
+The same applies to a string input passed together with a `contentType` to <ApiLink to="class/ActorClient#start">`start()`</ApiLink>, <ApiLink to="class/ActorClient#call">`call()`</ApiLink>, or <ApiLink to="class/RunClient#metamorph">`metamorph()`</ApiLink>. v2 handed the API the string wrapped in quotes, v3 hands it over as it is.
+
 ## Timeouts are configured per tier
 
 The `timeoutSecs` option of the <ApiLink to="class/ApifyClient">`ApifyClient`</ApiLink> constructor is gone. Every method takes its timeout from one of three tiers, and the constructor sets the duration of each tier, along with a cap on any single request attempt:

docs/public-api/apify-client.api.md

@@ -3282,9 +3282,9 @@ export interface RequestQueue extends Omit<Schemas['RequestQueue'], keyof Reques
 export class RequestQueueClient extends ResourceClient {
     constructor(options: ApiClientSubResourceOptions, userOptions?: RequestQueueUserOptions);
     addRequest(request: RequestQueueClientRequestToAdd, options?: RequestQueueClientAddRequestOptions): Promise<RequestQueueClientAddRequestResult>;
-    protected addRequestBatch(requests: RequestQueueClientRequestToAdd[], options?: RequestQueueClientAddRequestOptions): Promise<RequestQueueClientBatchRequestsOperationResult>;
+    protected addRequestBatch(requests: SerializedRequestToAdd[], options?: RequestQueueClientAddRequestOptions): Promise<RequestQueueClientBatchRequestsOperationResult>;
     // (undocumented)
-    protected addRequestBatchWithRetries(requests: RequestQueueClientRequestToAdd[], options?: RequestQueueClientBatchAddRequestWithRetriesOptions): Promise<RequestQueueClientBatchRequestsOperationResult>;
+    protected addRequestBatchWithRetries(requests: SerializedRequestToAdd[], options?: RequestQueueClientBatchAddRequestWithRetriesOptions): Promise<RequestQueueClientBatchRequestsOperationResult>;
     batchAddRequests(requests: RequestQueueClientRequestToAdd[], options?: RequestQueueClientBatchAddRequestWithRetriesOptions): Promise<RequestQueueClientBatchRequestsOperationResult>;
     batchDeleteRequests(requests: RequestQueueClientRequestToDelete[], options?: TimeoutOptions): Promise<RequestQueueClientBatchDeleteRequestsResult>;
     delete(options?: TimeoutOptions): Promise<void>;
@@ -3738,6 +3738,17 @@ interface ScheduleRePointed {
 // @public (undocumented)
 type Schemas = components['schemas'];
 
+// Not exported by the entry point; reachable only as a referenced type.
+// @public
+interface SerializedRequestToAdd {
+    // (undocumented)
+    byteLength: number;
+    // (undocumented)
+    json: string;
+    // (undocumented)
+    request: RequestQueueClientRequestToAdd;
+}
+
 // @public
 export class ServerError extends ApifyApiError {
 }

src/http_client.ts

@@ -84,9 +84,10 @@ export class HttpClient {
                 return new URLSearchParams(formattedParams).toString();
             },
             validateStatus: null,
-            // Using interceptors for this functionality.
-            transformRequest: undefined,
-            transformResponse: undefined,
+            // Interceptors serialize requests and parse responses instead. Empty arrays rather than `undefined`,
+            // which axios fills in with its default transforms.
+            transformRequest: [],
+            transformResponse: [],
             responseType: 'arraybuffer',
             // Every request sets its own timeout in `createRequestHandler`, so the default only backs a raw
             // `axios.request()` call.

src/interceptors.ts

@@ -43,6 +43,11 @@ function getHeader(config: ApifyRequestConfig, name: string): string | undefined
 }
 
 function serializeRequest(config: ApifyRequestConfig): ApifyRequestConfig {
+    // A string body with an explicit content type is already serialized and goes out as it is. The axios default
+    // transform would otherwise parse a JSON one in full just to check that it is valid, which for a body assembled
+    // from thousands of pre-serialized requests costs about as much as serializing them did.
+    if (typeof config.data === 'string' && getHeader(config, 'content-type')) return config;
+
     const [defaultTransform] = axios.defaults.transformRequest as AxiosRequestTransformer[];
 
     // The function not only serializes data, but it also adds correct headers.

src/resource_clients/request_queue.ts

@@ -33,7 +33,8 @@ import {
     parseArgument,
     parseResponse,
     RequestQueuePaginationIterator,
-    sliceArrayByByteLength,
+    splitIntoJsonArrayBatches,
+    utf8ByteLength,
 } from '../utils.js';
 
 const DEFAULT_PARALLEL_BATCH_ADD_REQUESTS = 5;
@@ -55,7 +56,6 @@ const newRequestSchema = z.custom<RequestQueueClientRequestToAdd>(
     'Expected a request object without an `id`',
 );
 const forefrontOptionsSchema = z.strictObject({ forefront: z.boolean().optional(), ...timeoutOptionsShape });
-const batchAddRequestsSchema = z.array(newRequestSchema).min(1).max(REQUEST_QUEUE_MAX_REQUESTS_PER_BATCH_OPERATION);
 const batchAddRequestsWithRetriesSchema = z.array(newRequestSchema).min(1);
 const optionalBooleanSchema = z.boolean().optional();
 const optionalNumberSchema = z.number().optional();
@@ -92,6 +92,16 @@ const paginateRequestsOptionsSchema = z.strictObject({
     ...timeoutOptionsShape,
 });
 
+/**
+ * A request to add, serialized once up front: `byteLength` decides which batch it goes into, `json` is what the body of
+ * that batch is assembled from, and `request` is what the result bookkeeping needs.
+ */
+interface SerializedRequestToAdd {
+    request: RequestQueueClientRequestToAdd;
+    json: string;
+    byteLength: number;
+}
+
 export type {
     AllowedHttpMethods,
     RequestQueue,
@@ -360,17 +370,19 @@ export class RequestQueueClient extends ResourceClient {
      * @private
      */
     protected async addRequestBatch(
-        requests: RequestQueueClientRequestToAdd[],
+        requests: SerializedRequestToAdd[],
         options: RequestQueueClientAddRequestOptions = {},
     ): Promise<RequestQueueClientBatchRequestsOperationResult> {
-        parseArgument(requests, batchAddRequestsSchema);
         const parsed = parseArgument(options, forefrontOptionsSchema, 'RequestQueueClientAddRequestOptions');
 
         const response = await this.httpClient.call({
             url: this.buildUrl('requests/batch'),
             method: 'POST',
             timeoutSecs: this.#resolveTimeout(parsed.timeoutSecs, 'medium'),
-            data: requests,
+            // The body is assembled from the requests as `batchAddRequests` serialized them; the explicit content type
+            // makes the request interceptor send the string as it is instead of serializing the requests again.
+            headers: { 'content-type': 'application/json' },
+            data: `[${requests.map(({ json }) => json).join(',')}]`,
             params: this.buildParams({
                 forefront: parsed.forefront,
                 clientKey: this.#clientKey,
@@ -381,7 +393,7 @@ export class RequestQueueClient extends ResourceClient {
     }
 
     protected async addRequestBatchWithRetries(
-        requests: RequestQueueClientRequestToAdd[],
+        requests: SerializedRequestToAdd[],
         options: RequestQueueClientBatchAddRequestWithRetriesOptions = {},
     ): Promise<RequestQueueClientBatchRequestsOperationResult> {
         const {
@@ -416,7 +428,7 @@ export class RequestQueueClient extends ResourceClient {
                 const processedRequestsUniqueKeys = processedRequests.map(({ uniqueKey }) => uniqueKey);
                 // Requests remaining to be processed are the all that remain
                 remainingRequests = requests.filter(
-                    ({ uniqueKey }) => !processedRequestsUniqueKeys.includes(uniqueKey),
+                    ({ request }) => !processedRequestsUniqueKeys.includes(request.uniqueKey),
                 );
 
                 // Stop if all requests have been processed
@@ -432,8 +444,8 @@ export class RequestQueueClient extends ResourceClient {
                 // This ensures that this method does not throw and keeps the signature.
                 const processedRequestsUniqueKeys = processedRequests.map(({ uniqueKey }) => uniqueKey);
                 unprocessedRequests = requests
-                    .filter(({ uniqueKey }) => !processedRequestsUniqueKeys.includes(uniqueKey))
-                    .map(({ method, uniqueKey, url }) => ({ method, uniqueKey, url }));
+                    .filter(({ request }) => !processedRequestsUniqueKeys.includes(request.uniqueKey))
+                    .map(({ request: { method, uniqueKey, url } }) => ({ method, uniqueKey, url }));
 
                 break;
             }
@@ -512,12 +524,28 @@ export class RequestQueueClient extends ResourceClient {
         const payloadSizeLimitBytes =
             MAX_PAYLOAD_SIZE_BYTES - Math.ceil(MAX_PAYLOAD_SIZE_BYTES * SAFETY_BUFFER_PERCENT);
 
+        // Serialize every request once: the byte lengths decide the batch boundaries, and the same strings are joined
+        // into the batch bodies, so nothing is stringified a second time when it is sent.
+        const serializedRequests = requests.map((request, index): SerializedRequestToAdd => {
+            const json = JSON.stringify(request);
+            const byteLength = utf8ByteLength(json);
+            // Two more bytes for the brackets, which even a batch of one request carries.
+            if (byteLength + 2 > payloadSizeLimitBytes) {
+                throw new Error(
+                    `RequestQueueClient.batchAddRequests: The size of the request with index: ${index} ` +
+                        `exceeds the maximum allowed size (${payloadSizeLimitBytes} bytes).`,
+                );
+            }
+            return { request, json, byteLength };
+        });
+        const batches = splitIntoJsonArrayBatches(serializedRequests, {
+            maxCount: REQUEST_QUEUE_MAX_REQUESTS_PER_BATCH_OPERATION,
+            maxByteLength: payloadSizeLimitBytes,
+        });
+
         // Keep a pool of up to `maxParallel` requests running at once
-        let i = 0;
-        while (i < requests.length) {
-            const slicedRequests = requests.slice(i, i + REQUEST_QUEUE_MAX_REQUESTS_PER_BATCH_OPERATION);
-            const requestsInBatch = sliceArrayByByteLength(slicedRequests, payloadSizeLimitBytes, i);
-            const requestPromise = this.addRequestBatchWithRetries(requestsInBatch, options);
+        for (const batch of batches) {
+            const requestPromise = this.addRequestBatchWithRetries(batch, options);
             executingRequests.add(requestPromise);
             // A rejection reaches the caller through the awaits below; this bookkeeping chain only has to avoid
             // turning it into an unhandled one of its own.
@@ -531,7 +559,6 @@ export class RequestQueueClient extends ResourceClient {
             if (executingRequests.size >= maxParallel) {
                 await Promise.race(executingRequests);
             }
-            i += requestsInBatch.length;
         }
         // Get results from remaining operations
         await Promise.all(executingRequests);

src/utils.ts

@@ -257,30 +257,39 @@ export async function maybeCompressValue(value: unknown): Promise<CompressedValu
 }
 
 /**
- * Helper function slice the items from array to fit the max byte length.
+ * Returns the UTF-8 byte length of a string.
  */
-export function sliceArrayByByteLength<T>(array: T[], maxByteLength: number, startIndex: number): T[] {
-    const stringByteLength = (str: string) => (isNode() ? Buffer.byteLength(str) : new Blob([str]).size);
-    const arrayByteLength = stringByteLength(JSON.stringify(array));
-    if (arrayByteLength < maxByteLength) return array;
-
-    const slicedArray: T[] = [];
-    let byteLength = 2; // 2 bytes for the empty array []
-    for (let i = 0; i < array.length; i++) {
-        const item = array[i];
-        const itemByteSize = stringByteLength(JSON.stringify(item));
-        if (itemByteSize > maxByteLength) {
-            throw new Error(
-                `RequestQueueClient.batchAddRequests: The size of the request with index: ${startIndex + i} ` +
-                    `exceeds the maximum allowed size (${maxByteLength} bytes).`,
-            );
+export function utf8ByteLength(value: string): number {
+    return isNode() ? Buffer.byteLength(value) : new Blob([value]).size;
+}
+
+/**
+ * Splits JSON-serialized items into consecutive batches of at most `maxCount` items, each of which fits into a JSON
+ * array body - the items joined by commas between brackets - of at most `maxByteLength` bytes. The `byteLength` of an
+ * item is the UTF-8 byte length of its serialization. An item too large for a body of its own still gets one, so a
+ * caller that cannot send such an item has to reject it beforehand.
+ */
+export function splitIntoJsonArrayBatches<T extends { byteLength: number }>(
+    items: readonly T[],
+    { maxCount, maxByteLength }: { maxCount: number; maxByteLength: number },
+): T[][] {
+    const batches: T[][] = [];
+    let batch: T[] = [];
+    // One byte for the opening bracket; each item then adds its own bytes plus one for the comma or the closing
+    // bracket that follows it.
+    let byteLength = 1;
+    for (const item of items) {
+        if (batch.length > 0 && (batch.length >= maxCount || byteLength + item.byteLength + 1 > maxByteLength)) {
+            batches.push(batch);
+            batch = [];
+            byteLength = 1;
         }
-        if (byteLength + itemByteSize >= maxByteLength) break;
-        byteLength += itemByteSize;
-        slicedArray.push(item);
+        batch.push(item);
+        byteLength += item.byteLength + 1;
     }
+    if (batch.length > 0) batches.push(batch);
 
-    return slicedArray;
+    return batches;
 }
 
 export function isNode(): boolean {

test/http_client.test.ts

@@ -196,6 +196,21 @@ describe('HttpClient', () => {
         expect(request?.headers['content-length']).toBe(String(payload.length));
     });
 
+    test('sends a string body with an explicit content type as it is', async () => {
+        // The axios default transform would re-parse the body to validate it and trim this whitespace away.
+        const body = ' [{"uniqueKey": "key-1", "url": "http://example.com/1"}] ';
+
+        const response = await client.httpClient.call({
+            url: `${baseUrl}/v2/request-queues/some-id/requests/batch`,
+            method: 'POST',
+            headers: { 'content-type': 'application/json' },
+            data: body,
+        });
+
+        expect(response.config.data).toBe(body);
+        expect(mockServer.getLastRequest()?.body).toEqual(JSON.parse(body));
+    });
+
     describe('timeout across retries', () => {
         /**
          * Routes the client's axios instance to an in-process adapter that fails `failures` times with a 500 and