fix: Correctly limit pushed dataset items in PPE-aware mode

src/patched_apify_client.ts

@@ -7,6 +7,7 @@ import { ApifyClient, DatasetClient } from 'apify-client';
 import {
     type ChargeResult,
     type ChargingManager,
+    DEFAULT_DATASET_ITEM_EVENT,
     mergeChargeResults,
     pushDataAndCharge,
 } from './charging.js';
@@ -44,18 +45,42 @@ export function createPatchedApifyClient(
             unknown
         >,
     > extends DatasetClient<Data> {
+        private normalizeItems(
+            items: string | Data | string[] | Data[],
+        ): Data[] {
+            if (typeof items === 'string') {
+                const parsed = JSON.parse(items);
+                return Array.isArray(parsed) ? parsed : [parsed];
+            }
+
+            if (Array.isArray(items)) {
+                return items.flatMap((item) =>
+                    typeof item === 'string'
+                        ? (JSON.parse(item) as Data | Data[])
+                        : item,
+                ) as Data[];
+            }
+
+            return [items];
+        }
+
         override async pushItems(
             items: string | Data | string[] | Data[],
         ): Promise<void> {
             const context = pushDataChargingContext.getStore();
 
+            // Normalize string inputs: a single JSON string may encode multiple items
+            // (e.g. '[{"a":1},{"b":2}]'), which would be miscounted by the charging logic.
+            // Parse strings into arrays so each logical item is counted individually.
+            const normalizedItems = this.normalizeItems(items);
+
             const result = await pushDataAndCharge({
                 chargingManager: actor.getChargingManager(),
-                items,
+                items: normalizedItems,
                 eventName: context?.eventName,
                 isDefaultDataset: true,
                 pushFn: async (limitedItems) =>
-                    super.pushItems(limitedItems as typeof items),
+                    super.pushItems(JSON.stringify(limitedItems)), // stringify the items for faster validation in Apify client
             });
 
             if (!context) return;
@@ -91,7 +116,11 @@ export function createPatchedApifyClient(
                 httpClient: this.httpClient,
             };
 
-            if (!isDefaultDataset) {
+            const hasDefaultDatasetItemEvent =
+                DEFAULT_DATASET_ITEM_EVENT in
+                actor.getChargingManager().getPricingInfo().perEventPrices;
+
+            if (!isDefaultDataset || !hasDefaultDatasetItemEvent) {
                 return new DatasetClient(datasetOptions);
             }
 

test/e2e/sdk/actorPushDataPPE/src/main.mjs

@@ -0,0 +1,14 @@
+import { Actor, log } from 'apify';
+
+const actor = new Actor();
+
+await actor.init();
+
+const chargeResult = await actor.pushData(
+    new Array(100).fill({ hello: 'world' }),
+    'result',
+);
+
+log.info(`Charged: ${JSON.stringify(chargeResult)}`);
+
+await actor.exit();

test/e2e/sdk/actorPushDataPPE/test.mjs

@@ -0,0 +1,70 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+
+import { ApifyClient } from 'apify';
+import { sleep } from 'crawlee';
+
+const client = new ApifyClient({
+    token: process.env.APIFY_TOKEN,
+});
+
+const actor = client.actor(process.argv[2]);
+await actor.update({
+    pricingInfos: [
+        {
+            pricingModel: 'PAY_PER_EVENT',
+            pricingPerEvent: {
+                actorChargeEvents: {
+                    'apify-actor-start': {
+                        eventTitle: 'Actor start',
+                        eventPriceUsd: 1,
+                        eventDescription:
+                            'Charged automatically at the start of the run',
+                    },
+                    result: {
+                        eventTitle: 'Result',
+                        eventPriceUsd: 0.01,
+                        eventDescription: 'Single dataset item',
+                    },
+                },
+            },
+        },
+    ],
+});
+
+const runActor = async (input, options) => {
+    const { id: runId } = await actor.call(input, options);
+    await client.run(runId).waitForFinish();
+    await sleep(6000); // wait for updates to propagate to MongoDB
+    return await client.run(runId).get();
+};
+
+test('unlimited push', async () => {
+    const run = await runActor({}, { maxTotalChargeUsd: 10 });
+
+    assert.strictEqual(run.status, 'SUCCEEDED');
+    assert.deepEqual(run.chargedEventCounts, {
+        'apify-actor-start': 1,
+        result: 100,
+    });
+});
+
+test('charge limit works as intended', async () => {
+    const run = await runActor({}, { maxTotalChargeUsd: 1.5 });
+
+    assert.strictEqual(run.status, 'SUCCEEDED');
+    assert.deepEqual(run.chargedEventCounts, {
+        'apify-actor-start': 1,
+        result: 50,
+    });
+
+    const { items } = await client
+        .dataset(run.defaultDatasetId)
+        .listItems({ limit: 100 });
+
+    assert.strictEqual(
+        items.length,
+        50,
+        `default items (${items.length}) != 50`,
+    );
+});

test/e2e/sdk/actorPushDataSynth/src/main.mjs

@@ -0,0 +1,14 @@
+import { Actor, log } from 'apify';
+
+const actor = new Actor();
+
+await actor.init();
+
+const chargeResult = await actor.pushData(
+    new Array(100).fill({ hello: 'world' }),
+    'result',
+);
+
+log.info(`Charged: ${JSON.stringify(chargeResult)}`);
+
+await actor.exit();

test/e2e/sdk/actorPushDataSynth/test.mjs

@@ -0,0 +1,82 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+
+import { ApifyClient } from 'apify';
+import { sleep } from 'crawlee';
+
+const client = new ApifyClient({
+    token: process.env.APIFY_TOKEN,
+});
+
+const actor = client.actor(process.argv[2]);
+await actor.update({
+    pricingInfos: [
+        {
+            pricingModel: 'PAY_PER_EVENT',
+            pricingPerEvent: {
+                actorChargeEvents: {
+                    'apify-actor-start': {
+                        eventTitle: 'Actor start',
+                        eventPriceUsd: 1,
+                        eventDescription:
+                            'Charged automatically at the start of the run',
+                    },
+                    'apify-default-dataset-item': {
+                        eventTitle: 'Dataset item',
+                        eventPriceUsd: 0.005,
+                        eventDescription: 'Default charge per dataset item',
+                    },
+                    result: {
+                        eventTitle: 'Result',
+                        eventPriceUsd: 0.01,
+                        eventDescription: 'Custom charge per result',
+                    },
+                },
+            },
+        },
+    ],
+});
+
+const runActor = async (input, options) => {
+    const { id: runId } = await actor.call(input, options);
+    await client.run(runId).waitForFinish();
+    await sleep(9000); // wait for updates to propagate to MongoDB
+    return await client.run(runId).get();
+};
+
+test('pushData charges both apify-default-dataset-item and custom event', async () => {
+    const run = await runActor({}, { maxTotalChargeUsd: 10 });
+
+    assert.strictEqual(run.status, 'SUCCEEDED');
+    assert.deepEqual(run.chargedEventCounts, {
+        'apify-actor-start': 1,
+        'apify-default-dataset-item': 100,
+        result: 100,
+    });
+});
+
+test('charge limit applies to combined cost of both events', async () => {
+    // Budget: $1.50 total
+    // apify-actor-start: 1 × $1 = $1
+    // Remaining: $0.50
+    // Each pushData item costs $0.005 (default) + $0.01 (result) = $0.015
+    // Items within budget: floor($0.50 / $0.015) = 33
+    const run = await runActor({}, { maxTotalChargeUsd: 1.5 });
+
+    assert.strictEqual(run.status, 'SUCCEEDED');
+    assert.deepEqual(run.chargedEventCounts, {
+        'apify-actor-start': 1,
+        'apify-default-dataset-item': 33,
+        result: 33,
+    });
+
+    const { items } = await client
+        .dataset(run.defaultDatasetId)
+        .listItems({ limit: 100 });
+
+    assert.strictEqual(
+        items.length,
+        33,
+        `default items (${items.length}) != 33`,
+    );
+});