From 43634965fa1be4f8f58975990577072dd3785836 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Martin=20Ad=C3=A1mek?= Date: Fri, 7 Aug 2026 13:13:09 +0200 Subject: [PATCH 1/4] fix: split large pushData calls into chunks fitting the platform payload limit The Apify API rejects dataset push requests over 9MB. SDK v3 split oversized pushes into chunks under the limit (via crawlee v3's Dataset), but crawlee v4 removed that logic expecting storage backends to handle it, so pushing a large batch failed with an API error. Restore the v3 behavior in ApifyDatasetBackend.pushData: items are serialized, split into chunks fitting the limit, and pushed sequentially, preserving order. A single item exceeding the limit throws a descriptive error. Closes #603 --- src/apify_dataset_backend.ts | 54 +++++++++++++++++- src/apify_storage_backend.ts | 3 +- test/apify/apify_storage_backend.test.ts | 72 +++++++++++++++++++++++- 3 files changed, 123 insertions(+), 6 deletions(-) diff --git a/src/apify_dataset_backend.ts b/src/apify_dataset_backend.ts index 6010c21651..84a78818b7 100644 --- a/src/apify_dataset_backend.ts +++ b/src/apify_dataset_backend.ts @@ -1,10 +1,16 @@ 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); /** * 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 +37,52 @@ export class ApifyDatasetBackend implements DatasetBackend { } async pushData(items: Dictionary[]): Promise { - 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> { 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 > EFFECTIVE_LIMIT_BYTES) { + throw new Error( + `Data item at index ${index} is too large (size: ${bytes} bytes, limit: ${EFFECTIVE_LIMIT_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(',')}]`); +} diff --git a/src/apify_storage_backend.ts b/src/apify_storage_backend.ts index cfb48e6e47..c22b80a4ef 100644 --- a/src/apify_storage_backend.ts +++ b/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; diff --git a/test/apify/apify_storage_backend.test.ts b/test/apify/apify_storage_backend.test.ts index a1863078cd..cbc3daac89 100644 --- a/test/apify/apify_storage_backend.test.ts +++ b/test/apify/apify_storage_backend.test.ts @@ -1,4 +1,5 @@ import type { ApifyClient, DatasetClient, KeyValueStoreClient } from 'apify-client'; +import { DatasetClient as ApifyDatasetClient } from 'apify-client'; import { describe, expect, test, vi } from 'vitest'; import { ApifyDatasetBackend } from '../../src/apify_dataset_backend.js'; @@ -143,6 +144,46 @@ 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[] }) => ({ + 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', () => { @@ -150,7 +191,7 @@ describe('ApifyDatasetBackend', () => { 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 })), }; } @@ -162,7 +203,7 @@ describe('ApifyDatasetBackend', () => { 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 +222,33 @@ 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(9437184); + } + // All items arrive, in the original order. + expect(chunks.flatMap((chunk) => JSON.parse(chunk))).toEqual(items); + }); + + 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(); + }); }); describe('ApifyKeyValueStoreBackend', () => { From fe886c6d62303dee9e96b91d24d346e8608fe8fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Martin=20Ad=C3=A1mek?= Date: Fri, 7 Aug 2026 13:20:58 +0200 Subject: [PATCH 2/4] test: fix test type errors against crawlee 4.0.0-beta.105 The crawlee beta.105 bump stopped re-exporting Dictionary from crawlee and @crawlee/utils (import it from @crawlee/types instead), and the getRequest mock arity no longer matched the mocked signature. These errors are only visible to tsc-check-tests, which CI does not run. --- test/apify/events.test.ts | 2 +- test/apify/proxy_configuration.test.ts | 3 ++- test/apify/request_queue_backend.test.ts | 2 +- 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/test/apify/events.test.ts b/test/apify/events.test.ts index 8944fdad2a..5ca1e54d53 100644 --- a/test/apify/events.test.ts +++ b/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'; diff --git a/test/apify/proxy_configuration.test.ts b/test/apify/proxy_configuration.test.ts index be5b0b814b..a7c026c6f0 100644 --- a/test/apify/proxy_configuration.test.ts +++ b/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'; diff --git a/test/apify/request_queue_backend.test.ts b/test/apify/request_queue_backend.test.ts index ac27500792..158b84a9a0 100644 --- a/test/apify/request_queue_backend.test.ts +++ b/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 () => {}), }; } From e4d0d06a3d4f7b9a09e5f32630964dd0f5400536 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Martin=20Ad=C3=A1mek?= Date: Fri, 7 Aug 2026 13:39:38 +0200 Subject: [PATCH 3/4] ci: type-check test files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tsc-check-tests script existed but no CI job ran it, so type errors in test files went unnoticed (e.g. the crawlee beta.105 bump broke three test imports without failing any check). Run it in the lint job — the test tsconfig resolves the apify package from src/, so no build is needed. --- .github/workflows/test-and-release.yaml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/test-and-release.yaml b/.github/workflows/test-and-release.yaml index fb5049876d..d8bd17bae9 100644 --- a/.github/workflows/test-and-release.yaml +++ b/.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:') From 06f8e5edc699e8d50117321d242383b673c5981b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Martin=20Ad=C3=A1mek?= Date: Fri, 7 Aug 2026 14:20:39 +0200 Subject: [PATCH 4/4] chore: apply review refinements Tighten the per-item size limit by the 2-byte array wrapper so a lone maximal item cannot exceed the chunk size contract, and pin the chunking behavior with boundary tests (at-limit, one byte over, multi-byte UTF-8, empty push). Update stale PatchedDatasetClient comment references to PpeAwareDatasetClient. --- src/actor.ts | 4 +-- src/apify_dataset_backend.ts | 7 ++-- test/apify/apify_storage_backend.test.ts | 45 +++++++++++++++++++++++- 3 files changed, 51 insertions(+), 5 deletions(-) diff --git a/src/actor.ts b/src/actor.ts index 1b29d2e72f..320c833696 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -1116,7 +1116,7 @@ export class Actor { 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 { item: Data | Data[], eventName: string | undefined, ): Promise { - // 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, diff --git a/src/apify_dataset_backend.ts b/src/apify_dataset_backend.ts index 84a78818b7..37bf554bc8 100644 --- a/src/apify_dataset_backend.ts +++ b/src/apify_dataset_backend.ts @@ -6,6 +6,9 @@ import { MAX_PAYLOAD_SIZE_BYTES } from '@apify/consts'; 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. Mostly a thin method-mapping wrapper (`getMetadata`/`get`, `drop`/`delete`, @@ -54,9 +57,9 @@ export class ApifyDatasetBackend implements DatasetBackend { function serializeToSizeLimit(item: Dictionary, index: number): string { const payload = JSON.stringify(item); const bytes = Buffer.byteLength(payload); - if (bytes > EFFECTIVE_LIMIT_BYTES) { + if (bytes > MAX_ITEM_BYTES) { throw new Error( - `Data item at index ${index} is too large (size: ${bytes} bytes, limit: ${EFFECTIVE_LIMIT_BYTES} bytes)`, + `Data item at index ${index} is too large (size: ${bytes} bytes, limit: ${MAX_ITEM_BYTES} bytes)`, ); } return payload; diff --git a/test/apify/apify_storage_backend.test.ts b/test/apify/apify_storage_backend.test.ts index cbc3daac89..00f84f5f0e 100644 --- a/test/apify/apify_storage_backend.test.ts +++ b/test/apify/apify_storage_backend.test.ts @@ -2,6 +2,8 @@ import type { ApifyClient, DatasetClient, KeyValueStoreClient } from 'apify-clie 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'; @@ -196,6 +198,9 @@ describe('ApifyDatasetBackend', () => { }; } + // 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); @@ -234,12 +239,20 @@ describe('ApifyDatasetBackend', () => { 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(9437184); + 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); @@ -249,6 +262,36 @@ describe('ApifyDatasetBackend', () => { ); 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', () => {