feat: overcharge by one event when the PPE budget is exhausted

docs/02_concepts/13_exceptions.mdx

@@ -55,7 +55,7 @@ The SDK raises standard Python exceptions when it's used incorrectly or given in
 
 ## The pay-per-event charge limit
 
-Reaching the pay-per-event charge limit doesn't raise an error. Instead, the SDK caps charging and data pushing, while your Actor keeps running. When a single <ApiLink to="class/Actor#charge">`Actor.charge`</ApiLink> call crosses the limit, only the part that fits within the budget is billed, and `charged_count` on the returned <ApiLink to="class/ChargeResult">`ChargeResult`</ApiLink> reports how many events went through. <ApiLink to="class/Actor#push_data">`Actor.push_data`</ApiLink> behaves the same way when given a `charged_event_name`. It writes only the items that fit within the budget.
+Reaching the pay-per-event charge limit doesn't raise an error. Instead, the SDK limits charging and data pushing. When a single <ApiLink to="class/Actor#charge">`Actor.charge`</ApiLink> call crosses the limit, the SDK bills the events that fit within the budget plus one more, and `charged_count` on the returned <ApiLink to="class/ChargeResult">`ChargeResult`</ApiLink> reports how many events went through. The extra event lets the platform detect the exhausted budget and terminate the run. Once the run is over budget, further calls charge nothing. When given a `charged_event_name`, <ApiLink to="class/Actor#push_data">`Actor.push_data`</ApiLink> writes only the items that fit within the budget, or a single item when the budget can't cover even one.
 
 To detect the limit, check the `event_charge_limit_reached` field on the `ChargeResult`. It's a return value and not an exception, so you can read it in a tight charging loop and stop your work once the budget runs out. For details, see [Pay-per-event monetization](./pay-per-event).
 

src/apify/_actor.py

@@ -852,6 +852,9 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str
 
         This is relevant only for the pay-per-event pricing model.
 
+        When `count` exceeds the remaining budget, the events that fit are charged plus one more, so that the platform
+        detects the exhausted budget and terminates the run. A run already strictly over budget charges nothing.
+
         Args:
             event_name: Name of the event to be charged for.
             count: Number of events to charge for.

src/apify/_charging.py

@@ -93,6 +93,9 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str
 
         This is relevant only for the pay-per-event pricing model.
 
+        When `count` exceeds the remaining budget, the events that fit are charged plus one more, so that the platform
+        detects the exhausted budget and terminates the run. A run already strictly over budget charges nothing.
+
         Args:
             event_name: Name of the event to be charged for.
             count: Number of events to charge for.
@@ -140,14 +143,17 @@ def compute_push_data_limit(
         Accounts for both the explicit event and the synthetic `DEFAULT_DATASET_ITEM_EVENT` event,
         so that the combined cost per item does not exceed the remaining budget.
 
+        When the budget cannot cover a single item, one item is still allowed through, so that the resulting overcharge
+        makes the platform terminate the run. A run that is already strictly over budget gets no items through.
+
         Args:
             items_count: The number of items to be pushed.
             event_name: The explicit event name to charge for each item.
             is_default_dataset: Whether the data is pushed to the default dataset.
                 If True, the synthetic event cost is included in the combined price.
 
         Returns:
-            Max number of items that can be pushed within the budget.
+            Max number of items to push.
         """
 
     def is_event_charge_limit_reached(self, event_name: str) -> bool:
@@ -341,9 +347,7 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str
                     chargeable_within_limit=self.compute_chargeable(),
                 )
 
-            # Determine the maximum amount of events that can be charged within the budget
-            max_chargeable = self.calculate_max_event_charge_count_within_limit(event_name)
-            charged_count = min(count, max_chargeable if max_chargeable is not None else count)
+            charged_count = self._compute_charged_count(event_name, count)
 
             if charged_count == 0:
                 return ChargeResult(
@@ -472,6 +476,9 @@ def compute_push_data_limit(
         *,
         is_default_dataset: bool,
     ) -> int:
+        if self._pricing_model != 'PAY_PER_EVENT' or items_count == 0:
+            return items_count
+
         explicit_price = self._get_event_price(event_name)
         synthetic_price = self._get_event_price(DEFAULT_DATASET_ITEM_EVENT) if is_default_dataset else Decimal(0)
         combined_price = explicit_price + synthetic_price
@@ -481,7 +488,14 @@ def compute_push_data_limit(
 
         result = (self._max_total_charge_usd - self.calculate_total_charged_amount()) / combined_price
         max_count = max(0, math.floor(result)) if result.is_finite() else items_count
-        return min(items_count, max_count)
+
+        if max_count >= items_count:
+            return items_count
+
+        if max_count > 0:
+            return max_count
+
+        return 1 if self.calculate_total_charged_amount() <= self._max_total_charge_usd else 0
 
     @_ensure_context
     def is_event_charge_limit_reached(self, event_name: str) -> bool:
@@ -537,6 +551,19 @@ async def _fetch_pricing_info(self) -> _FetchedPricingInfoDict:
             ),
         )
 
+    def _compute_charged_count(self, event_name: str, count: int) -> int:
+        """Compute how many of `count` events to charge, given the remaining budget."""
+        max_chargeable = self.calculate_max_event_charge_count_within_limit(event_name)
+        if max_chargeable is None or count <= max_chargeable:
+            return count
+
+        # Overcharge by one event, so that the platform detects the exhausted budget and terminates the run. A run
+        # already strictly over budget gets no further charges.
+        if self.calculate_total_charged_amount() <= self._max_total_charge_usd:
+            return max_chargeable + 1
+
+        return 0
+
     def _get_event_price(self, event_name: str) -> Decimal:
         pricing_info = self._pricing_info.get(event_name)
         if pricing_info is not None:

src/apify/storage_clients/_ppe_dataset_mixin.py

@@ -11,10 +11,12 @@ def __init__(self) -> None:
 
     def _compute_limit_for_push(self, items_count: int) -> int:
         if self.is_default_dataset and (charging_manager := charging_manager_ctx.get()):
-            max_charged_count = charging_manager.calculate_max_event_charge_count_within_limit(
-                event_name=DEFAULT_DATASET_ITEM_EVENT
+            # The synthetic event is the only one charged per item, so it is priced as the explicit event.
+            return charging_manager.compute_push_data_limit(
+                items_count,
+                DEFAULT_DATASET_ITEM_EVENT,
+                is_default_dataset=False,
             )
-            return min(max_charged_count, items_count) if max_charged_count is not None else items_count
         return items_count
 
     async def _charge_for_items(self, count_items: int) -> None:

tests/e2e/test_actor_charge.py

@@ -155,18 +155,19 @@ async def test_actor_charge_limit(
 ) -> None:
     run = await run_actor(ppe_actor, max_total_charge_usd=Decimal('0.2'))
 
-    # Reaching `max_total_charge_usd` makes the platform auto-abort the run. The Actor blocks after hitting the
-    # limit (see `ppe_actor_build`) so the abort always wins the race against its clean exit, hence ABORTED.
-    # Refetch until the charged event counts propagate on the platform.
+    # The Actor charges 4 events and the budget covers 2, so the SDK charges 3 - one over the budget, which makes
+    # the platform auto-abort the run. The Actor blocks after hitting the limit (see `ppe_actor_build`) so the abort
+    # always wins the race against its clean exit, hence ABORTED. Refetch until the charged event counts propagate
+    # on the platform.
     run = await poll_until_condition(
         partial(_get_run, apify_client_async, run.id),
-        lambda r: r.status == 'ABORTED' and r.charged_event_counts == {'foobar': 2},
+        lambda r: r.status == 'ABORTED' and r.charged_event_counts == {'foobar': 3},
         timeout=30,
         poll_interval=1,
     )
 
     assert run.status == 'ABORTED'
-    assert run.charged_event_counts == {'foobar': 2}
+    assert run.charged_event_counts == {'foobar': 3}
 
 
 async def test_actor_push_data_charges_both_events(

tests/e2e/test_scrapy/test_ppe_spider.py

@@ -12,12 +12,12 @@
     from ..conftest import MakeActorFunction, RunActorFunction
 
 
-async def test_ppe_spider_charges_pushed_items_within_budget(
+async def test_ppe_spider_stops_pushing_once_over_budget(
     make_actor: MakeActorFunction,
     run_actor: RunActorFunction,
     apify_client_async: ApifyClientAsync,
 ) -> None:
-    """A pay-per-event Scrapy Actor charges the synthetic dataset-item event and stops pushing at the budget cap."""
+    """A pay-per-event Scrapy Actor charges the synthetic dataset-item event and stops pushing once over budget."""
     actor = await make_actor(
         label='scrapy-ppe',
         source_files=get_scrapy_source_files('spider_basic.py', 'BasicSpider'),
@@ -44,12 +44,12 @@ async def test_ppe_spider_charges_pushed_items_within_budget(
         ]
     )
 
-    # The spider scrapes three products. At $0.05 an item, a $0.125 budget covers only two of them, so the
-    # third push has to be dropped and never charged. The budget deliberately does not divide evenly into the
-    # item price - hitting the cap exactly would let the platform auto-abort the run and race the clean exit.
-    run = await run_actor(actor, max_total_charge_usd=Decimal('0.125'))
+    # The spider scrapes three products. At $0.05 an item, a $0.075 budget covers one of them. The second push
+    # overcharges by one item, so the run is over budget and the third push is dropped and never charged.
+    run = await run_actor(actor, max_total_charge_usd=Decimal('0.075'))
 
-    assert run.status == 'SUCCEEDED'
+    # The overcharge makes the platform auto-abort the run, which races the Actor's clean exit.
+    assert run.status in ('SUCCEEDED', 'ABORTED')
 
     expected_charges = {'apify-default-dataset-item': 2}
 

tests/unit/actor/test_actor_charge.py

@@ -71,11 +71,8 @@ async def setup_mocked_charging(
             yield setup
 
 
-async def test_actor_charge_push_data_with_no_remaining_budget() -> None:
-    """Test that the API client is NOT called when budget is exhausted during push_data.
-
-    When push_data can't afford to charge for any items, it correctly avoids calling the API.
-    """
+async def test_actor_push_data_overcharges_one_item_when_budget_cannot_cover_one() -> None:
+    """Test that push_data pushes and charges exactly one item when the budget cannot cover a single one."""
     async with setup_mocked_charging(
         Configuration(max_total_charge_usd=Decimal('1.5'), test_pay_per_event=True),
         {'some-event': Decimal('1.0'), 'another-event': Decimal('1.0')},
@@ -89,22 +86,15 @@ async def test_actor_charge_push_data_with_no_remaining_budget() -> None:
 
         assert result1.charged_count == 1
 
-        # Now try to push data - we can't afford even 1 more event
-        # This will call charge(event_name, count=0) because max_charged_count=0
         result = await Actor.push_data([{'hello': 'world'} for _ in range(10)], charged_event_name='another-event')
 
-        # The API should NOT be called when count=0
-        setup.mock_charge.assert_not_called()
-
-        # Correctly returns result with charged_count=0
-        assert result is not None
-        assert result.charged_count == 0
+        setup.mock_charge.assert_called_once_with('another-event', count=1, idempotency_key=None)
+        assert result.charged_count == 1
         assert result.event_charge_limit_reached is True
 
-        # Verify no items were pushed
         dataset = await Actor.open_dataset()
         items = await dataset.get_data()
-        assert len(items.items) == 0
+        assert len(items.items) == 1
 
 
 async def test_actor_charge_api_call_verification() -> None:
@@ -245,37 +235,36 @@ async def test_charge_lock_concurrent_with_limited_budget() -> None:
     ) as setup:
         dataset = await Actor.open_dataset()
 
-        # Both try to push 5 items, but budget only allows 5 total
+        # Both try to push 5 items. The budget covers 5 in total, plus one overcharge item.
         await asyncio.gather(
             dataset.push_data([{'source': 'a', 'id': i} for i in range(5)]),
             dataset.push_data([{'source': 'b', 'id': i} for i in range(5)]),
         )
 
-        # Verify total items pushed does not exceed budget limit
         items = await dataset.get_data()
-        assert len(items.items) == 5  # Budget allows max 5 items at $0.10 each
+        assert len(items.items) == 6
 
         # Verify total charged events matches items pushed
-        assert setup.charging_mgr.get_charged_event_count('apify-default-dataset-item') == 5
+        assert setup.charging_mgr.get_charged_event_count('apify-default-dataset-item') == 6
 
 
-async def test_concurrent_actor_push_data_stays_within_budget() -> None:
-    """Concurrent `Actor.push_data` calls do not overdraw the budget - the reservation and the charge stay atomic."""
+async def test_concurrent_actor_push_data_overdraws_budget_by_at_most_one_item() -> None:
+    """Concurrent `Actor.push_data` calls overdraw the budget by at most one item, since the charge stays atomic."""
     async with setup_mocked_charging(
         Configuration(max_total_charge_usd=Decimal('0.50'), test_pay_per_event=True),
         {'scrape': Decimal('0.10')},
     ) as setup:
-        # Both try to push 5 items, but the budget only allows 5 in total.
+        # Both try to push 5 items. The budget covers 5 in total, plus one overcharge item.
         await asyncio.gather(
             Actor.push_data([{'source': 'a', 'id': i} for i in range(5)], charged_event_name='scrape'),
             Actor.push_data([{'source': 'b', 'id': i} for i in range(5)], charged_event_name='scrape'),
         )
 
-        assert setup.charging_mgr.get_charged_event_count('scrape') == 5
+        assert setup.charging_mgr.get_charged_event_count('scrape') == 6
 
         dataset = await Actor.open_dataset()
         items = await dataset.get_data()
-        assert len(items.items) == 5
+        assert len(items.items) == 6
 
 
 async def test_push_data_does_not_take_charge_lock_without_pay_per_event(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -320,10 +309,37 @@ async def test_charge_with_overdrawn_budget() -> None:
     )
 
     async with setup_mocked_charging(configuration, {}) as setup:
+        # The budget doesn't cover another event, so the charge overcharges by one.
         charge_result = await Actor.charge('event', count=1)
-        assert charge_result.charged_count == 0  # The budget doesn't allow another event
+        assert charge_result.charged_count == 1
+        assert charge_result.event_charge_limit_reached is True
+        setup.mock_charge.assert_called_once_with('event', count=1, idempotency_key=None)
+        setup.mock_charge.reset_mock()
 
+        # The run is now strictly over budget, so nothing else is charged.
         push_result = await Actor.push_data([{'hello': 'world'}], charged_event_name='event')
-        assert push_result.charged_count == 0  # Nor does the budget allow this
+        assert push_result.charged_count == 0
+        assert push_result.event_charge_limit_reached is True
 
         setup.mock_charge.assert_not_called()
+
+
+async def test_actor_push_data_overcharges_one_item_locally() -> None:
+    """Test that a local pay-per-event run lets one item through when the budget cannot cover a single one."""
+    # Locally, every event without pricing info costs $1, so one item costs $2 (explicit + synthetic event).
+    async with Actor(Configuration(max_total_charge_usd=Decimal('1.00'), test_pay_per_event=True)):
+        result = await Actor.push_data([{'id': i} for i in range(3)], charged_event_name='scrape')
+
+        charging_manager = Actor.get_charging_manager()
+        assert charging_manager.get_charged_event_count('apify-default-dataset-item') == 1
+        assert result.event_charge_limit_reached is True
+
+        dataset = await Actor.open_dataset()
+        items = await dataset.get_data()
+        assert len(items.items) == 1
+
+        result = await Actor.push_data([{'id': 3}], charged_event_name='scrape')
+        assert result.charged_count == 0
+
+        items = await dataset.get_data()
+        assert len(items.items) == 1