diff --git a/docs/02_concepts/13_exceptions.mdx b/docs/02_concepts/13_exceptions.mdx index 8d7bec3fa..2bec78ed7 100644 --- a/docs/02_concepts/13_exceptions.mdx +++ b/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 `Actor.charge` call crosses the limit, only the part that fits within the budget is billed, and `charged_count` on the returned `ChargeResult` reports how many events went through. `Actor.push_data` 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 `Actor.charge` call crosses the limit, the SDK bills the events that fit within the budget plus one more, and `charged_count` on the returned `ChargeResult` 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`, `Actor.push_data` 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). diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 150cf9245..6026d7898 100644 --- a/src/apify/_actor.py +++ b/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. diff --git a/src/apify/_charging.py b/src/apify/_charging.py index 7300ec216..fc404ffbf 100644 --- a/src/apify/_charging.py +++ b/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,6 +143,9 @@ 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. @@ -147,7 +153,7 @@ def compute_push_data_limit( 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: diff --git a/src/apify/storage_clients/_ppe_dataset_mixin.py b/src/apify/storage_clients/_ppe_dataset_mixin.py index 18663dee8..ec80a4871 100644 --- a/src/apify/storage_clients/_ppe_dataset_mixin.py +++ b/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: diff --git a/tests/e2e/test_actor_charge.py b/tests/e2e/test_actor_charge.py index c48dc89f2..375061ca5 100644 --- a/tests/e2e/test_actor_charge.py +++ b/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( diff --git a/tests/e2e/test_scrapy/test_ppe_spider.py b/tests/e2e/test_scrapy/test_ppe_spider.py index ecd9c5848..08fbd308a 100644 --- a/tests/e2e/test_scrapy/test_ppe_spider.py +++ b/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} diff --git a/tests/unit/actor/test_actor_charge.py b/tests/unit/actor/test_actor_charge.py index 87083db25..583a18e0b 100644 --- a/tests/unit/actor/test_actor_charge.py +++ b/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 diff --git a/tests/unit/actor/test_charging_manager.py b/tests/unit/actor/test_charging_manager.py index 86fa4d789..ae50b041d 100644 --- a/tests/unit/actor/test_charging_manager.py +++ b/tests/unit/actor/test_charging_manager.py @@ -117,8 +117,8 @@ async def test_charge_known_event_locally(mock_client: MagicMock) -> None: assert cm.get_charged_event_count('search') == 5 -async def test_charge_partial_count_within_budget(mock_client: MagicMock) -> None: - """Test that charging is capped when budget is insufficient for the full count.""" +async def test_charge_partial_count_over_budget(mock_client: MagicMock) -> None: + """Test that a count exceeding the budget charges what fits plus one overcharge event.""" pricing_info = _make_ppe_pricing_info({'search': Decimal('1.00')}) config = _make_config( test_pay_per_event=True, @@ -129,8 +129,7 @@ async def test_charge_partial_count_within_budget(mock_client: MagicMock) -> Non cm = ChargingManagerImplementation(config, mock_client) async with cm: result = await cm.charge('search', count=5) - assert result.charged_count == 3 - # Budget is fully exhausted ($3.00 spent of $3.00), so limit is reached + assert result.charged_count == 4 assert result.event_charge_limit_reached is True @@ -288,8 +287,8 @@ async def test_get_max_total_charge_usd(mock_client: MagicMock) -> None: assert cm.get_max_total_charge_usd() == Decimal('42.50') -async def test_max_total_charge_usd_zero_blocks_charging(mock_client: MagicMock) -> None: - """Test max_total_charge_usd=0 is preserved and blocks all charging (not silently converted to inf).""" +async def test_max_total_charge_usd_zero_allows_only_the_overcharge_event(mock_client: MagicMock) -> None: + """Test max_total_charge_usd=0 is preserved (not converted to inf) and lets only one overcharge event through.""" pricing_info = _make_ppe_pricing_info({'search': Decimal('0.01')}) config = _make_config( test_pay_per_event=True, @@ -302,9 +301,12 @@ async def test_max_total_charge_usd_zero_blocks_charging(mock_client: MagicMock) assert cm.get_max_total_charge_usd() == Decimal(0) assert cm.is_event_charge_limit_reached('search') is True result = await cm.charge('search', count=5) - assert result.charged_count == 0 + assert result.charged_count == 1 assert result.event_charge_limit_reached is True + result = await cm.charge('search', count=5) + assert result.charged_count == 0 + async def test_max_total_charge_usd_zero_preserved_without_pricing(mock_client: MagicMock) -> None: """Test max_total_charge_usd=0 is preserved through the no-pricing fallback path.""" @@ -327,6 +329,15 @@ async def test_compute_push_data_limit_no_ppe(mock_client: MagicMock) -> None: assert result == 10 +async def test_compute_push_data_limit_ignores_budget_without_ppe(mock_client: MagicMock) -> None: + """Returns items_count for a run that is not pay-per-event, even with a zero budget set locally.""" + config = _make_config(actor_pricing_info=None, charged_event_counts={}, max_total_charge_usd=Decimal(0)) + cm = ChargingManagerImplementation(config, mock_client) + async with cm: + result = cm.compute_push_data_limit(10, 'some-event', is_default_dataset=True) + assert result == 10 + + async def test_compute_push_data_limit_within_budget(mock_client: MagicMock) -> None: """Returns full items_count when combined budget is sufficient for all items.""" pricing_info = _make_ppe_pricing_info({'click': Decimal('0.01'), 'apify-default-dataset-item': Decimal('0.01')}) @@ -375,19 +386,28 @@ async def test_compute_push_data_limit_without_default_dataset(mock_client: Magi assert result == 3 -async def test_compute_push_data_limit_exhausted_budget(mock_client: MagicMock) -> None: - """Returns 0 when the budget is fully exhausted before the push.""" +@pytest.mark.parametrize( + ('charged_count', 'expected_limit'), + [ + pytest.param(3, 1, id='exhausted budget lets one item through'), + pytest.param(4, 0, id='strictly over budget lets nothing through'), + ], +) +async def test_compute_push_data_limit_exhausted_budget( + mock_client: MagicMock, charged_count: int, expected_limit: int +) -> None: + """Returns one item for the overcharge when the budget is exhausted, and none once the run is over budget.""" pricing_info = _make_ppe_pricing_info({'scrape': Decimal('1.00')}) config = _make_config( test_pay_per_event=True, actor_pricing_info=pricing_info, - charged_event_counts={'scrape': 3}, + charged_event_counts={'scrape': charged_count}, max_total_charge_usd=Decimal('3.00'), ) cm = ChargingManagerImplementation(config, mock_client) async with cm: result = cm.compute_push_data_limit(5, 'scrape', is_default_dataset=False) - assert result == 0 + assert result == expected_limit async def test_charge_limit_reached(mock_client: MagicMock) -> None: @@ -406,9 +426,13 @@ async def test_charge_limit_reached(mock_client: MagicMock) -> None: assert result1.event_charge_limit_reached is True result2 = await cm.charge('search', count=1) - assert result2.charged_count == 0 + assert result2.charged_count == 1 assert result2.event_charge_limit_reached is True + result3 = await cm.charge('search', count=1) + assert result3.charged_count == 0 + assert result3.event_charge_limit_reached is True + async def test_is_event_charge_limit_reached_within_budget(mock_client: MagicMock) -> None: """Test that is_event_charge_limit_reached returns False when budget allows at least one more charge.""" @@ -699,8 +723,8 @@ async def test_charge_deduplicates_idempotency_key_off_platform(mock_client: Mag mock_client.run.return_value.charge.assert_not_awaited() -async def test_charge_registers_the_count_capped_by_the_budget(mock_client: MagicMock) -> None: - """Test that a charge capped by the remaining budget registers the capped count, not the requested one.""" +async def test_charge_registers_the_count_limited_by_the_budget(mock_client: MagicMock) -> None: + """Test that a charge limited by the remaining budget registers the charged count, not the requested one.""" pricing_info = _make_ppe_pricing_info({'search': Decimal('1.00')}) config = _make_config( is_at_home=True, @@ -711,9 +735,49 @@ async def test_charge_registers_the_count_capped_by_the_budget(mock_client: Magi ) cm = ChargingManagerImplementation(config, mock_client) async with cm: - assert (await cm.charge('search', count=5, idempotency_key='key-1')).charged_count == 2 - mock_client.run.return_value.charge.assert_awaited_once_with('search', count=2, idempotency_key='key-1') + assert (await cm.charge('search', count=5, idempotency_key='key-1')).charged_count == 3 + mock_client.run.return_value.charge.assert_awaited_once_with('search', count=3, idempotency_key='key-1') - assert (await cm.charge('search', count=5, idempotency_key='key-1')).charged_count == 2 - assert cm.get_charged_event_count('search') == 2 + assert (await cm.charge('search', count=5, idempotency_key='key-1')).charged_count == 3 + assert cm.get_charged_event_count('search') == 3 assert mock_client.run.return_value.charge.await_count == 1 + + +@pytest.mark.parametrize( + 'is_at_home', + [ + pytest.param(True, id='on platform'), + pytest.param(False, id='local'), + ], +) +async def test_charge_overcharges_by_one_event_when_budget_exceeded( + mock_client: MagicMock, + *, + is_at_home: bool, +) -> None: + """A charge over the budget gets the chargeable count plus one, and a run strictly over budget gets nothing.""" + pricing_info = _make_ppe_pricing_info({'search': Decimal('1.00')}) + config = _make_config( + is_at_home=is_at_home, + actor_run_id='test-run-id' if is_at_home else None, + test_pay_per_event=not is_at_home, + actor_pricing_info=pricing_info, + charged_event_counts={'search': 2}, + max_total_charge_usd=Decimal('3.00'), + ) + cm = ChargingManagerImplementation(config, mock_client) + async with cm: + result = await cm.charge('search', count=5) + assert result.charged_count == 2 + assert result.event_charge_limit_reached is True + assert cm.get_charged_event_count('search') == 4 + + result = await cm.charge('search', count=1) + assert result.charged_count == 0 + assert result.event_charge_limit_reached is True + assert cm.get_charged_event_count('search') == 4 + + if is_at_home: + mock_client.run.return_value.charge.assert_awaited_once_with('search', count=2, idempotency_key=None) + else: + mock_client.run.return_value.charge.assert_not_called()