refactor: split large pushData calls into chunks fitting the platform payload limit

.github/workflows/test-and-release.yaml

@@ -101,6 +101,9 @@ jobs:
             - name: Format check
               run: pnpm format:check
 
+            - name: Type-check tests
+              run: pnpm tsc-check-tests
+
     publish:
         name: Publish
         if: (github.ref == 'refs/heads/master' || github.ref == 'refs/heads/v4') && !contains(github.event.head_commit.message, 'docs:')

src/actor.ts

@@ -1116,7 +1116,7 @@ export class Actor<Data extends Dictionary = Dictionary> {
         const dataset = await this.openDataset();
 
         // Two code paths for charging:
-        // 1. Intercepted client: PatchedDatasetClient intercepts pushItems() calls, handling charging
+        // 1. Intercepted client: PpeAwareDatasetClient intercepts pushItems() calls, handling charging
         //    internally. This is needed because Crawlee's Dataset may call pushItems() directly,
         //    bypassing Actor.pushData(). We propagate eventName via AsyncLocalStorage context.
         // 2. Direct charging: When using a non-patched client (e.g., forceCloud option or custom client),
@@ -2233,7 +2233,7 @@ export class Actor<Data extends Dictionary = Dictionary> {
         item: Data | Data[],
         eventName: string | undefined,
     ): Promise<ChargeResult> {
-        // PatchedDatasetClient will handle charging and item limiting.
+        // PpeAwareDatasetClient will handle charging and item limiting.
         // We only need to propagate `eventName` and (optionally) return aggregated charge info.
         const context: PpeAwarePushDataContext = {
             eventName,

src/apify_dataset_backend.ts

@@ -1,10 +1,19 @@
 import type { DatasetBackend, DatasetBackendListOptions, DatasetInfo, Dictionary, PaginatedList } from '@crawlee/types';
 import type { DatasetClient } from 'apify-client';
+import { MAX_PAYLOAD_SIZE_BYTES } from '@apify/consts';
+
+/** Slight reduction of the API's 9MB payload limit, to stay safely below it. */
+const SAFETY_BUFFER_PERCENT = 0.01 / 100; // 0.01%
+const EFFECTIVE_LIMIT_BYTES = MAX_PAYLOAD_SIZE_BYTES - Math.ceil(MAX_PAYLOAD_SIZE_BYTES * SAFETY_BUFFER_PERCENT);
+
+/** Per-item ceiling — 2 bytes under the chunk limit, so even a lone item fits its `[]` wrapper. */
+const MAX_ITEM_BYTES = EFFECTIVE_LIMIT_BYTES - 2;
 
 /**
  * Implements crawlee v4's {@link DatasetBackend} interface on top of `apify-client`'s
- * dataset API. A thin method-mapping wrapper — the interfaces differ only in naming
- * (`getMetadata`/`get`, `drop`/`delete`, `pushData`/`pushItems`, `getData`/`listItems`).
+ * dataset API. Mostly a thin method-mapping wrapper (`getMetadata`/`get`, `drop`/`delete`,
+ * `getData`/`listItems`), except `pushData`, which also splits large pushes into chunks
+ * fitting the API's payload size limit.
  *
  * @internal
  */
@@ -31,10 +40,52 @@ export class ApifyDatasetBackend implements DatasetBackend {
     }
 
     async pushData(items: Dictionary[]): Promise<void> {
-        await this.client.pushItems(items);
+        // The platform API rejects payloads over 9MB — split the items into chunks
+        // that fit, pushed sequentially to preserve item order.
+        const payloads = items.map((item, index) => serializeToSizeLimit(item, index));
+        for (const chunk of chunkBySize(payloads, EFFECTIVE_LIMIT_BYTES)) {
+            await this.client.pushItems(chunk);
+        }
     }
 
     async getData(options?: DatasetBackendListOptions): Promise<PaginatedList<Dictionary>> {
         return await this.client.listItems(options);
     }
 }
+
+/** Serializes a dataset item, throwing if it alone exceeds the payload size limit. */
+function serializeToSizeLimit(item: Dictionary, index: number): string {
+    const payload = JSON.stringify(item);
+    const bytes = Buffer.byteLength(payload);
+    if (bytes > MAX_ITEM_BYTES) {
+        throw new Error(
+            `Data item at index ${index} is too large (size: ${bytes} bytes, limit: ${MAX_ITEM_BYTES} bytes)`,
+        );
+    }
+    return payload;
+}
+
+/**
+ * Takes an array of JSON-serialized items and groups them into JSON array strings
+ * of at most `limitBytes` each, preserving item order. Assumes (and does not
+ * validate) that no single item exceeds the limit.
+ */
+function chunkBySize(payloads: string[], limitBytes: number): string[] {
+    const chunks: string[][] = [];
+    let chunkBytes = Infinity; // Forces the first item to open a new chunk.
+
+    for (const payload of payloads) {
+        const bytes = Buffer.byteLength(payload);
+        if (chunkBytes + bytes + 1 <= limitBytes) {
+            // Fits into the current chunk — add 1 byte for the ',' separator.
+            chunks[chunks.length - 1].push(payload);
+            chunkBytes += bytes + 1;
+        } else {
+            // Open a new chunk — add 2 bytes for the '[]' wrapper.
+            chunks.push([payload]);
+            chunkBytes = bytes + 2;
+        }
+    }
+
+    return chunks.map((chunk) => `[${chunk.join(',')}]`);
+}

src/apify_storage_backend.ts

@@ -61,7 +61,8 @@ export const USES_PUSH_DATA_INTERCEPTION = Symbol('apify:uses-push-data-intercep
  * Context of a single `Actor.pushData()` call, shared with the intercepted
  * `pushItems()` calls so they can (1) know which event to charge and
  * (2) aggregate the {@link ChargeResult} across the multiple `pushItems()`
- * calls a single `pushData()` may trigger (Crawlee batches large pushes).
+ * calls a single `pushData()` may trigger (the backend splits pushes exceeding
+ * the API's payload size limit).
  */
 export interface PpeAwarePushDataContext {
     eventName: string | undefined;

test/apify/apify_storage_backend.test.ts

@@ -1,6 +1,9 @@
 import type { ApifyClient, DatasetClient, KeyValueStoreClient } from 'apify-client';
+import { DatasetClient as ApifyDatasetClient } from 'apify-client';
 import { describe, expect, test, vi } from 'vitest';
 
+import { MAX_PAYLOAD_SIZE_BYTES } from '@apify/consts';
+
 import { ApifyDatasetBackend } from '../../src/apify_dataset_backend.js';
 import { ApifyKeyValueStoreBackend } from '../../src/apify_key_value_store_backend.js';
 import { ApifyRequestQueueSharedBackend } from '../../src/apify_request_queue_shared_backend.js';
@@ -143,26 +146,69 @@ describe('ApifyStorageBackend', () => {
         expect((defaultDataset as never)[USES_PUSH_DATA_INTERCEPTION]).toBe(true);
         expect((otherDataset as never)[USES_PUSH_DATA_INTERCEPTION]).toBeUndefined();
     });
+
+    test('charges per parsed item when the backend splits a push into chunks', async () => {
+        const client = createMockApifyClient();
+        const charge = vi.fn(async ({ count }: { eventName: string; count: number }) => ({
+            eventChargeLimitReached: false,
+            chargedCount: count,
+            chargeableWithinLimit: {},
+        }));
+        const calculatePushDataLimits = vi.fn(({ items }: { items: Record<string, unknown>[] }) => ({
+            limitedItems: items,
+            eventsToCharge: { [DEFAULT_DATASET_ITEM_EVENT]: items.length },
+        }));
+        const backend = new ApifyStorageBackend(asApifyClient(client), {
+            configuration: new Configuration({ defaultDatasetId: 'default-dataset' }),
+            getChargingManager: () =>
+                ({
+                    getPricingInfo: () => ({ perEventPrices: { [DEFAULT_DATASET_ITEM_EVENT]: {} } }),
+                    calculatePushDataLimits,
+                    charge,
+                }) as never,
+        });
+        const dataset = await backend.createDatasetBackend({ id: 'default-dataset' });
+
+        // Stub out the plain-client push underneath the charging wrapper.
+        const pushSpy = vi.spyOn(ApifyDatasetClient.prototype, 'pushItems').mockResolvedValue(undefined);
+        try {
+            // Four ~3MB items arrive as multiple pre-serialized JSON chunks — the charging
+            // wrapper must still count every logical item exactly once.
+            const items = Array.from({ length: 4 }, (_, i) => ({ i, payload: 'a'.repeat(3 * 1024 * 1024) }));
+            await dataset.pushData(items);
+
+            expect(pushSpy.mock.calls.length).toBeGreaterThan(1);
+            const countedItems = calculatePushDataLimits.mock.calls.map(([{ items: counted }]) => counted);
+            expect(countedItems.flat()).toEqual(items);
+            const chargedCounts = charge.mock.calls.map(([{ count }]) => count);
+            expect(chargedCounts.reduce((sum, count) => sum + count, 0)).toBe(4);
+        } finally {
+            pushSpy.mockRestore();
+        }
+    });
 });
 
 describe('ApifyDatasetBackend', () => {
     function createMockDatasetClient() {
         return {
             get: vi.fn(async () => ({ id: 'dataset-id', itemCount: 0 })),
             delete: vi.fn(async () => {}),
-            pushItems: vi.fn(async () => {}),
+            pushItems: vi.fn(async (_items: unknown) => {}),
             listItems: vi.fn(async () => ({ items: [{ foo: 'bar' }], total: 1, count: 1, offset: 0, limit: 10 })),
         };
     }
 
+    // Mirrors the backend's per-item ceiling: 9MB API limit - 0.01% safety buffer - 2 bytes for '[]'.
+    const maxItemBytes = MAX_PAYLOAD_SIZE_BYTES - Math.ceil(MAX_PAYLOAD_SIZE_BYTES * 0.0001) - 2;
+
     test('maps the backend interface onto the apify-client dataset client', async () => {
         const client = createMockDatasetClient();
         const backend = new ApifyDatasetBackend(client as unknown as DatasetClient);
 
         await expect(backend.getMetadata()).resolves.toEqual({ id: 'dataset-id', itemCount: 0 });
 
         await backend.pushData([{ foo: 'bar' }]);
-        expect(client.pushItems).toHaveBeenCalledWith([{ foo: 'bar' }]);
+        expect(client.pushItems).toHaveBeenCalledWith('[{"foo":"bar"}]');
 
         await expect(backend.getData({ limit: 10 })).resolves.toEqual(
             expect.objectContaining({ items: [{ foo: 'bar' }], total: 1 }),
@@ -181,6 +227,71 @@ describe('ApifyDatasetBackend', () => {
         await expect(backend.getMetadata()).rejects.toThrow(/not found/);
         await expect(backend.purge()).rejects.toThrow(/not supported on the Apify platform/);
     });
+
+    test('splits pushes exceeding the 9MB payload limit into multiple API calls', async () => {
+        const client = createMockDatasetClient();
+        const backend = new ApifyDatasetBackend(client as unknown as DatasetClient);
+
+        // Four ~3MB items (~12MB total) cannot fit into a single 9MB request.
+        const items = Array.from({ length: 4 }, (_, i) => ({ i, payload: 'a'.repeat(3 * 1024 * 1024) }));
+        await backend.pushData(items);
+
+        const chunks = client.pushItems.mock.calls.map(([chunk]) => chunk as string);
+        expect(chunks.length).toBeGreaterThan(1);
+        for (const chunk of chunks) {
+            expect(Buffer.byteLength(chunk)).toBeLessThanOrEqual(MAX_PAYLOAD_SIZE_BYTES);
+        }
+        // All items arrive, in the original order.
+        expect(chunks.flatMap((chunk) => JSON.parse(chunk))).toEqual(items);
+    });
+
+    test('pushing an empty array makes no API call', async () => {
+        const client = createMockDatasetClient();
+        const backend = new ApifyDatasetBackend(client as unknown as DatasetClient);
+
+        await backend.pushData([]);
+        expect(client.pushItems).not.toHaveBeenCalled();
+    });
+
+    test('rejects an item that alone exceeds the payload limit', async () => {
+        const client = createMockDatasetClient();
+        const backend = new ApifyDatasetBackend(client as unknown as DatasetClient);
+
+        await expect(backend.pushData([{ ok: true }, { payload: 'a'.repeat(10 * 1024 * 1024) }])).rejects.toThrow(
+            /Data item at index 1 is too large/,
+        );
+        expect(client.pushItems).not.toHaveBeenCalled();
+    });
+
+    test('accepts an item at exactly the per-item limit and rejects one byte over', async () => {
+        const client = createMockDatasetClient();
+        const backend = new ApifyDatasetBackend(client as unknown as DatasetClient);
+
+        // `{"payload":"a…a"}` serializes to n + 14 bytes.
+        const maxPayload = 'a'.repeat(maxItemBytes - 14);
+
+        await backend.pushData([{ payload: maxPayload }]);
+        expect(client.pushItems).toHaveBeenCalledTimes(1);
+        const chunk = client.pushItems.mock.calls[0][0] as string;
+        expect(Buffer.byteLength(chunk)).toBeLessThanOrEqual(MAX_PAYLOAD_SIZE_BYTES);
+
+        await expect(backend.pushData([{ payload: `${maxPayload}a` }])).rejects.toThrow(
+            /Data item at index 0 is too large/,
+        );
+    });
+
+    test('measures item size in bytes, not string characters', async () => {
+        const client = createMockDatasetClient();
+        const backend = new ApifyDatasetBackend(client as unknown as DatasetClient);
+
+        // 'é' is one character but two UTF-8 bytes — the item is far below the
+        // limit in characters, yet just over it in bytes, so it must be rejected.
+        const item = { payload: 'é'.repeat(Math.ceil(maxItemBytes / 2)) };
+        expect(JSON.stringify(item).length).toBeLessThan(maxItemBytes);
+
+        await expect(backend.pushData([item])).rejects.toThrow(/Data item at index 0 is too large/);
+        expect(client.pushItems).not.toHaveBeenCalled();
+    });
 });
 
 describe('ApifyKeyValueStoreBackend', () => {

test/apify/events.test.ts

@@ -1,5 +1,5 @@
 import { EventType, serviceLocator } from '@crawlee/core';
-import type { Dictionary } from '@crawlee/utils';
+import type { Dictionary } from '@crawlee/types';
 import { sleep } from '@crawlee/utils';
 import { Actor, Configuration, PlatformEventManager } from 'apify';
 import { WebSocketServer } from 'ws';

test/apify/proxy_configuration.test.ts

@@ -1,6 +1,7 @@
+import type { Dictionary } from '@crawlee/types';
 import { Actor, ArgumentValidationError, ProxyConfiguration } from 'apify';
 import { UserClient } from 'apify-client';
-import { type Dictionary, sleep } from 'crawlee';
+import { sleep } from 'crawlee';
 import type { MockInstance } from 'vitest';
 
 import { APIFY_ENV_VARS, LOCAL_APIFY_ENV_VARS } from '@apify/consts';

test/apify/request_queue_backend.test.ts

@@ -61,7 +61,7 @@ function createMockApiClient() {
             wasAlreadyPresent: true,
             wasAlreadyHandled: false,
         })),
-        getRequest: vi.fn(async () => undefined),
+        getRequest: vi.fn(async (_requestId: string) => undefined),
         deleteRequestLock: vi.fn(async () => {}),
     };
 }