fix: reserve the limit of a resurrected child run once

Author: vdusek

src/apify/_child_runs.py

@@ -508,8 +508,8 @@ async def _budget(
         async with charging_manager.charge_lock():
             available = charging_manager.calculate_remaining_budget()
             record = (await self._load()).get(name)
-            if replaces_current and record is not None:
-                available += self._current_charge(name, record)
+            current_charge = self._current_charge(name, record) if replaces_current and record is not None else 0
+            available += current_charge
             if available <= 0:
                 raise RuntimeError(
                     f'Child run "{name}" was not started, since the budget of this Actor run is spent or reserved for '
@@ -521,7 +521,8 @@ async def _budget(
                     f'Lowering the charge limit of child run "{name}" to {limit} USD, the budget left for it',
                     extra={'requested_usd': str(max_total_charge_usd)},
                 )
-            self._reserving[name] = limit
+            # A resurrected run's current charge is reserved by its record already.
+            self._reserving[name] = max(limit - current_charge, Decimal(0))
 
         try:
             yield limit, limit

tests/unit/actor/test_actor_child_runs.py

@@ -1303,6 +1303,34 @@ def resurrect(run_client: Any, **_: Any) -> Run:
     assert charge_result.charged_count == 0
 
 
+async def test_resurrection_in_flight_reserves_its_limit_once(
+    parent_budget: dict[str, Run], apify_client_async_patcher: ApifyClientAsyncPatcher
+) -> None:
+    """While a resurrection is in flight, its limit is reserved once, including the charge its run made before."""
+    parent_budget['old-run'] = finish(make_run('old-run', 'RUNNING'), 'ABORTED', 1, finished_ago=timedelta(minutes=5))
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def resurrect(run_client: Any, **_: Any) -> Run:
+        started.set()
+        await release.wait()
+        return parent_budget[run_client._resource_id].model_copy(update={'status': 'RUNNING', 'finished_at': None})
+
+    async with Actor:
+        await seed_budget_record('child', 'old-run', maxTotalChargeUsd='6')
+
+    apify_client_async_patcher.patch('run', 'resurrect', replacement_method=resurrect, is_async=True)
+
+    async with _ActorType() as actor:
+        start_task = asyncio.create_task(actor.start('some-actor', name='child', max_total_charge_usd=Decimal(4)))
+        await started.wait()
+        charge_result = await actor.charge('some-event', count=10)
+        release.set()
+        await start_task
+
+    assert charge_result.charged_count == 6
+
+
 async def test_failed_resurrection_leaves_the_charge_of_its_run_to_settle(
     parent_budget: dict[str, Run], apify_client_async_patcher: ApifyClientAsyncPatcher, monkeypatch: pytest.MonkeyPatch
 ) -> None: