diff --git a/package-lock.json b/package-lock.json index ed3d635..98a0d29 100644 --- a/package-lock.json +++ b/package-lock.json @@ -11,7 +11,6 @@ "devDependencies": { "@apify/eslint-config": "^1.2.0", "@apify/tsconfig": "^0.1.2", - "@crawlee/core": "3.16.0", "@types/node": "^25.9.1", "@vitest/coverage-v8": "^4.0.18", "apify": "3.7.2", @@ -32,14 +31,10 @@ }, "peerDependencies": { "@apify/log": "^2", - "@crawlee/core": "^3", "apify": "^3", "apify-client": "^2" }, "peerDependenciesMeta": { - "@crawlee/core": { - "optional": true - }, "apify": { "optional": true }, diff --git a/package.json b/package.json index 259f099..52f99ff 100644 --- a/package.json +++ b/package.json @@ -76,7 +76,6 @@ "devDependencies": { "@apify/eslint-config": "^1.2.0", "@apify/tsconfig": "^0.1.2", - "@crawlee/core": "3.16.0", "@types/node": "^25.9.1", "@vitest/coverage-v8": "^4.0.18", "apify": "3.7.2", @@ -95,14 +94,10 @@ "license": "ISC", "peerDependencies": { "@apify/log": "^2", - "@crawlee/core": "^3", "apify": "^3", "apify-client": "^2" }, "peerDependenciesMeta": { - "@crawlee/core": { - "optional": true - }, "apify": { "optional": true }, diff --git a/src/gteam-internal/README.md b/src/gteam-internal/README.md index 50544c4..51328fa 100644 --- a/src/gteam-internal/README.md +++ b/src/gteam-internal/README.md @@ -19,10 +19,10 @@ const { chargeableWithinLimit } = await safePushData(item, { eventName: 'result- await safePushData(item, { alias: 'competitorAnalysis' }); ``` -It converts a dataset schema-validation failure into a `NonRetryableError` -instead of leaving it to be retried indefinitely. `apify`, `apify-client`, -and `@crawlee/core` are optional peer dependencies, needed only if you use -this subpath. +Items that fail dataset schema-validation are repaired where possible and +dropped (with a logged error) otherwise, instead of failing the whole push — +see `pushDataWithSchemaRepair`. `apify` and `apify-client` are optional peer +dependencies, needed only if you use this subpath. -Its test suite mocks `apify`/`apify-client`/`@crawlee/core` with Vitest's -`vi.mock`, and runs as part of the same `npm test` as everything else. +Its test suite mocks `apify`/`apify-client` with Vitest's `vi.mock`, and runs +as part of the same `npm test` as everything else. diff --git a/src/gteam-internal/push-data.ts b/src/gteam-internal/push-data.ts index b11cfc4..af0dd26 100644 --- a/src/gteam-internal/push-data.ts +++ b/src/gteam-internal/push-data.ts @@ -1,32 +1,39 @@ -import { NonRetryableError } from '@crawlee/core'; import type { ChargeResult } from 'apify'; import { Actor, log } from 'apify'; -import { ApifyApiError } from 'apify-client'; + +import { pushDataWithSchemaRepair } from '../push-data-with-schema-repair/index.js'; + +// Actor.pushData(items, eventName)'s resolved value when every item got +// dropped and the push itself never ran — mirrors what a charge-less push +// would report, so callers checking `chargeableWithinLimit` don't crash on +// a missing result. +const NO_CHARGE_RESULT: ChargeResult = { + eventChargeLimitReached: false, + chargedCount: 0, + chargeableWithinLimit: {}, +}; /** - * Wraps a push call, converting a dataset schema-validation failure into a - * {@link NonRetryableError} instead of a retryable one — a schema mismatch - * will never fix itself on retry, so retrying just burns compute. + * Wraps a push call, repairing or dropping items that fail dataset + * schema-validation instead of letting the whole batch fail. Every dropped + * item is logged, so existing log-based monitoring keeps working. */ -async function wrapPushData(pushDataFn: () => Promise): Promise { - try { - return await pushDataFn(); - } catch (error) { - if (!(error instanceof ApifyApiError) || !Array.isArray(error.data?.invalidItems)) { - throw error; - } - const msg = 'Dataset validation failed'; - for (const { validationErrors } of error.data.invalidItems) { - log.error(msg, { msg, error: `${error}`, validationErrors }); - } - throw new NonRetryableError(msg); +async function wrapPushData( + data: T | T[], + pushFn: (items: T[]) => Promise, +): Promise { + const msg = 'Dataset validation failed'; + const { droppedItems, pushResult } = await pushDataWithSchemaRepair(pushFn, data); + for (const { errors } of droppedItems) { + log.error(msg, { msg, validationErrors: errors }); } + return pushResult; } /** - * Pushes to the default dataset, or a named dataset when `alias` is given, - * converting a schema-validation failure into a {@link NonRetryableError} - * instead of leaving it to be retried. + * Pushes to the default dataset, or a named dataset when `alias` is given. + * Items that fail dataset schema-validation are repaired where possible and + * dropped (with a logged error) otherwise, instead of failing the whole push. * * @param data - A single item or array of items to push. * @param options.alias - Alias of the dataset to push to, opened via {@link Actor.openDataset}. @@ -45,9 +52,9 @@ export async function safePushData( ): Promise; /** * Pushes to the default dataset and atomically charges for `eventName`, via - * {@link Actor.pushData}'s built-in pay-per-event support. Converts a - * schema-validation failure into a {@link NonRetryableError} instead of - * leaving it to be retried. + * {@link Actor.pushData}'s built-in pay-per-event support. Items that fail + * dataset schema-validation are repaired where possible and dropped (with a + * logged error) otherwise, instead of failing the whole push. * * @param data - A single item or array of items to push. * @param options.eventName - Pay-per-event event name to charge for this push. @@ -68,18 +75,17 @@ export async function safePushData( ): Promise { const { alias, eventName } = options ?? {}; - // Default dataset: Actor.pushData(data, eventName) already pushes and + // Default dataset: Actor.pushData(items, eventName) already pushes and // charges atomically, so delegate to it as-is instead of reimplementing charging. if (!alias) { if (eventName) { - return wrapPushData(async () => Actor.pushData(data, eventName)); + const pushResult = await wrapPushData(data, async (items) => Actor.pushData(items, eventName)); + return pushResult ?? NO_CHARGE_RESULT; } - return wrapPushData(async () => Actor.pushData(data)); + return wrapPushData(data, async (items) => Actor.pushData(items)); } // Named dataset: charging is handled by the caller — call Actor.charge() themselves after this push. const dataset = await Actor.openDataset({ alias }); - return wrapPushData(async () => { - await dataset.pushData(data); - }); + return wrapPushData(data, async (items) => dataset.pushData(items)); } diff --git a/test/gteam-internal/push-data.test.ts b/test/gteam-internal/push-data.test.ts index 11862d9..7ef4a98 100644 --- a/test/gteam-internal/push-data.test.ts +++ b/test/gteam-internal/push-data.test.ts @@ -1,6 +1,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; import { safePushData } from '../../src/gteam-internal/index.js'; +import type { ValidationError } from '../../src/push-data-with-schema-repair/index.js'; vi.mock('apify', () => ({ Actor: { @@ -25,20 +26,9 @@ vi.mock('apify-client', () => { return { ApifyApiError }; }); -vi.mock('@crawlee/core', () => { - class NonRetryableError extends Error { - constructor(message: string) { - super(message); - this.name = 'NonRetryableError'; - } - } - return { NonRetryableError }; -}); - // Import after mocks are defined so vi.mocked() works const { Actor } = await import('apify'); const { ApifyApiError } = await import('apify-client'); -const { NonRetryableError } = await import('@crawlee/core'); // The real ApifyApiError takes (response, attempt); the mock above takes // just a message. Cast the constructor to the mock's actual runtime shape @@ -50,6 +40,33 @@ function makeApiError(message: string, data: unknown): Error & { data: unknown } return error; } +// pushDataWithSchemaRepair only recognises an error as repairable when it +// carries this exact top-level shape (see isSchemaValidationError) — mirrors +// what the real Apify API actually reports on a schema-validation 400. +function createMockSchemaValidationError( + invalidItems: { itemPosition: number; validationErrors: ValidationError[] }[], +) { + const error = new MockApifyApiError('Validation failed') as Error & { + type: string; + statusCode: number; + data: { invalidItems: typeof invalidItems }; + }; + error.type = 'schema-validation-error'; + error.statusCode = 400; + error.data = { invalidItems }; + return error; +} + +// A root-level error (empty instancePath) that cleanItemFields can never +// repair — there's no field to strip, so the item is dropped on round one +// with no extra pushFn call. Keeps these tests to a single push attempt. +const UNFIXABLE_ROOT_ERROR: ValidationError = { + instancePath: '', + keyword: 'type', + params: { type: 'object' }, + message: 'must be object', +}; + describe('safePushData', () => { let mockDataset: { pushData: ReturnType }; @@ -108,22 +125,43 @@ describe('safePushData', () => { }); describe('error handling', () => { - it('wraps ApifyApiError with invalidItems into NonRetryableError (default dataset)', async () => { - const apiError = makeApiError('Validation failed', { - invalidItems: [{ validationErrors: ['field required'] }], - }); + it('drops an unfixable item, logs it, and resolves instead of throwing (default dataset)', async () => { + const apiError = createMockSchemaValidationError([ + { itemPosition: 0, validationErrors: [UNFIXABLE_ROOT_ERROR] }, + ]); vi.mocked(Actor.pushData).mockRejectedValueOnce(apiError as never); - await expect(safePushData([{ id: 1 }])).rejects.toBeInstanceOf(NonRetryableError); + const result = await safePushData([{ id: 1 }]); + + expect(result).toBeUndefined(); + expect(Actor.pushData).toHaveBeenCalledTimes(1); + expect(vi.mocked((await import('apify')).log.error)).toHaveBeenCalledWith( + 'Dataset validation failed', + expect.objectContaining({ validationErrors: [UNFIXABLE_ROOT_ERROR] }), + ); }); - it('wraps ApifyApiError with invalidItems into NonRetryableError (named dataset)', async () => { - const apiError = makeApiError('Validation failed', { - invalidItems: [{ validationErrors: ['field required'] }], - }); + it('drops an unfixable item, logs it, and resolves instead of throwing (named dataset)', async () => { + const apiError = createMockSchemaValidationError([ + { itemPosition: 0, validationErrors: [UNFIXABLE_ROOT_ERROR] }, + ]); mockDataset.pushData.mockRejectedValueOnce(apiError); - await expect(safePushData([{ id: 1 }], { alias: 'ds' })).rejects.toBeInstanceOf(NonRetryableError); + const result = await safePushData([{ id: 1 }], { alias: 'ds' }); + + expect(result).toBeUndefined(); + expect(mockDataset.pushData).toHaveBeenCalledTimes(1); + }); + + it('returns the NO_CHARGE_RESULT fallback when every item is dropped (eventName given)', async () => { + const apiError = createMockSchemaValidationError([ + { itemPosition: 0, validationErrors: [UNFIXABLE_ROOT_ERROR] }, + ]); + vi.mocked(Actor.pushData).mockRejectedValueOnce(apiError as never); + + const result = await safePushData([{ id: 1 }], { eventName: 'place-scraped' }); + + expect(result).toEqual({ eventChargeLimitReached: false, chargedCount: 0, chargeableWithinLimit: {} }); }); it('rethrows ApifyApiError without invalidItems unchanged', async () => { @@ -154,13 +192,13 @@ describe('safePushData', () => { await expect(safePushData([{ id: 1 }])).rejects.toBe(err); }); - it('wraps a failing aliased push and never charges', async () => { - const apiError = makeApiError('Validation failed', { - invalidItems: [{ validationErrors: ['field required'] }], - }); + it('never charges on a dropped aliased push', async () => { + const apiError = createMockSchemaValidationError([ + { itemPosition: 0, validationErrors: [UNFIXABLE_ROOT_ERROR] }, + ]); mockDataset.pushData.mockRejectedValueOnce(apiError); - await expect(safePushData([{ id: 1 }], { alias: 'ds' })).rejects.toThrow(); + await safePushData([{ id: 1 }], { alias: 'ds' }); expect(Actor.charge).not.toHaveBeenCalled(); }); });