refactor(gteam-internal): delegate safePushData to pushDataWithSchemaRepair

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
         },

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.

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<R>(pushDataFn: () => Promise<R>): Promise<R> {
-    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<T extends object, R>(
+    data: T | T[],
+    pushFn: (items: T[]) => Promise<R>,
+): Promise<R | undefined> {
+    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<T extends object>(
 ): Promise<void>;
 /**
  * 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<T extends object>(
 ): Promise<ChargeResult | void> {
     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<T>({ alias });
-    return wrapPushData(async () => {
-        await dataset.pushData(data);
-    });
+    return wrapPushData(data, async (items) => dataset.pushData(items));
 }

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<typeof vi.fn> };
 
@@ -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();
         });
     });