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:') 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 6010c21651..37bf554bc8 100644 --- a/src/apify_dataset_backend.ts +++ b/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 { - 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 > 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(',')}]`); +} 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..00f84f5f0e 100644 --- a/test/apify/apify_storage_backend.test.ts +++ b/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,6 +146,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,11 +193,14 @@ 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 })), }; } + // 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); @@ -162,7 +208,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 +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', () => { 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 () => {}), }; }