diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 1c5a290f..6ab45191 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -7,7 +7,7 @@ import warnings from dataclasses import asdict from datetime import UTC, datetime, timedelta -from functools import cached_property +from functools import cached_property, partial from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast, overload @@ -35,6 +35,7 @@ ChargingManagerImplementation, charge_lock_if_charging, ) +from apify._child_runs import ChildRunRegistry from apify._configuration import Configuration from apify._consts import EVENT_LISTENERS_TIMEOUT, EXIT_CODE_ERROR_USER_FUNCTION_THREW, ActorEnvVars, ApifyEnvVars from apify._crypto import decrypt_input_secrets, load_private_key @@ -49,13 +50,14 @@ if TYPE_CHECKING: import logging - from collections.abc import Callable, MutableMapping + from collections.abc import Awaitable, Callable, MutableMapping from decimal import Decimal from types import TracebackType from typing import Self from apify_client._literals import ActorPermissionLevel from apify_client._models import Run + from apify_client._resource_clients import RunClientAsync from crawlee._types import JsonSerializable from crawlee.proxy_configuration import _NewUrlFunction @@ -161,6 +163,8 @@ def __init__( # Keep track of all used state stores to persist their values on exit self._use_state_stores: set[str | None] = set() + self._child_run_registry = ChildRunRegistry(self.open_key_value_store) + self._active = False """Whether the Actor instance is currently active (initialized and within context).""" @@ -953,6 +957,7 @@ async def start( timeout: timedelta | Literal['inherit'] | None = None, force_permission_level: ActorPermissionLevel | None = None, webhooks: list[Webhook] | None = None, + run_name: str | None = None, ) -> Run: """Run an Actor on the Apify platform. @@ -982,6 +987,12 @@ async def start( webhooks: Optional ad-hoc webhooks (https://docs.apify.com/webhooks/ad-hoc-webhooks) associated with the Actor run which can be used to receive a notification, e.g. when the Actor finished or failed. If you already have a webhook set up for the Actor or task, you do not have to add it again here. + run_name: Optional name of the child run, unique within this Actor run. A named run is recorded in the + default key-value store, so after a migration or resurrection of this Actor the same call reattaches + to the recorded run. A `SUCCEEDED` run is returned as is, an `ABORTED` or `TIMED-OUT` one is + resurrected, and a new run is started only when nothing is recorded under the name, or the recorded + run `FAILED` or no longer exists. The name is bound to the Actor and input it was first used with, + so reusing it for a different Actor, task or input raises a `ValueError`. Returns: Info about the started Actor run @@ -992,7 +1003,8 @@ async def start( client = self.new_client(token=token) if token else self.apify_client actor_client = client.actor(actor_id) - return await actor_client.start( + start_run = partial( + actor_client.start, run_input=run_input, content_type=content_type, build=build, @@ -1005,6 +1017,25 @@ async def start( webhooks=to_client_representations(webhooks), ) + if run_name is None: + return await start_run() + + run, _ = await self._find_or_start_child_run( + run_name, + actor_id=actor_id, + run_input=run_input, + client=client, + start_run=start_run, + token=token, + build=build, + max_items=max_items, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + timeout=timeout, + ) + return run + @_ensure_context async def abort( self, @@ -1115,6 +1146,7 @@ async def call( webhooks: list[Webhook] | None = None, wait: timedelta | None = None, logger: logging.Logger | Literal['default'] | None = 'default', + run_name: str | None = None, ) -> Run: """Start an Actor on the Apify Platform and wait for it to finish before returning. @@ -1147,6 +1179,12 @@ async def call( logger: Logger used to redirect logs from the Actor run. Using "default" literal means that a predefined default logger will be used. Setting `None` will disable any log propagation. Passing custom logger will redirect logs to the provided logger. + run_name: Optional name of the child run, unique within this Actor run. A named run is recorded in the + default key-value store, so after a migration or resurrection of this Actor the same call reattaches + to the recorded run. A `SUCCEEDED` run is returned as is, an `ABORTED` or `TIMED-OUT` one is + resurrected, and a new run is started only when nothing is recorded under the name, or the recorded + run `FAILED` or no longer exists. The name is bound to the Actor and input it was first used with, + so reusing it for a different Actor, task or input raises a `ValueError`. Returns: Info about the started Actor run. @@ -1157,26 +1195,122 @@ async def call( client = self.new_client(token=token) if token else self.apify_client actor_client = client.actor(actor_id) - run = await actor_client.call( - run_input=run_input, - content_type=content_type, - build=build, - max_items=max_items, - max_total_charge_usd=max_total_charge_usd, - restart_on_error=restart_on_error, - memory_mbytes=memory_mbytes, - run_timeout=self._resolve_run_timeout(timeout), - force_permission_level=force_permission_level, - webhooks=to_client_representations(webhooks), - wait_duration=wait, - logger=logger, - ) + + if run_name is None: + run = await actor_client.call( + run_input=run_input, + content_type=content_type, + build=build, + max_items=max_items, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=self._resolve_run_timeout(timeout), + force_permission_level=force_permission_level, + webhooks=to_client_representations(webhooks), + wait_duration=wait, + logger=logger, + ) + else: + started_run, is_new = await self._find_or_start_child_run( + run_name, + actor_id=actor_id, + run_input=run_input, + client=client, + start_run=partial( + actor_client.start, + run_input=run_input, + content_type=content_type, + build=build, + max_items=max_items, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=self._resolve_run_timeout(timeout), + force_permission_level=force_permission_level, + webhooks=to_client_representations(webhooks), + ), + token=token, + build=build, + max_items=max_items, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + timeout=timeout, + ) + # The earlier attempt of this call already streamed the log of a reattached or resurrected run. + run = await self._wait_for_child_run( + run_name, client.run(started_run.id), started_run, wait=wait, logger=logger, from_start=is_new + ) if run is None: raise RuntimeError(f'Failed to call Actor with ID "{actor_id}".') return run + async def _find_or_start_child_run( + self, + name: str, + *, + actor_id: str | None = None, + task_id: str | None = None, + run_input: Any, + client: ApifyClientAsync, + start_run: Callable[[], Awaitable[Run]], + token: str | None, + build: str | None, + max_items: int | None, + max_total_charge_usd: Decimal | None, + restart_on_error: bool | None, + memory_mbytes: int | None, + timeout: timedelta | Literal['inherit'] | None, + ) -> tuple[Run, bool]: + return await self._child_run_registry.find_or_start( + name, + actor_id=actor_id, + task_id=task_id, + run_input=run_input, + client=client, + start_run=start_run, + resurrect_run=partial( + self.resurrect, + token=token, + build=build, + max_items=max_items, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + timeout=timeout, + ), + ) + + async def _wait_for_child_run( + self, + name: str, + run_client: RunClientAsync, + run: Run, + *, + wait: timedelta | None, + logger: logging.Logger | Literal['default'] | None, + from_start: bool, + ) -> Run | None: + if run.status == 'SUCCEEDED': + return run + + if not logger: + finished_run = await run_client.wait_for_finish(wait_duration=wait) + else: + to_logger = None if logger == 'default' else logger + status_redirector = await run_client.get_status_message_watcher(to_logger=to_logger) + streamed_log = await run_client.get_streamed_log(to_logger=to_logger, from_start=from_start) + + async with status_redirector, streamed_log: + finished_run = await run_client.wait_for_finish(wait_duration=wait) + + if finished_run is not None: + await self._child_run_registry.update(name, finished_run) + return finished_run + @_ensure_context async def start_task( self, @@ -1191,6 +1325,7 @@ async def start_task( timeout: timedelta | Literal['inherit'] | None = None, webhooks: list[Webhook] | None = None, token: str | None = None, + run_name: str | None = None, ) -> Run: """Start an Actor task on the Apify Platform. @@ -1220,6 +1355,12 @@ async def start_task( webhooks: Optional webhooks (https://docs.apify.com/webhooks) associated with the Actor run, which can be used to receive a notification, e.g. when the Actor finished or failed. If you already have a webhook set up for the Actor, you do not have to add it again here. + run_name: Optional name of the child run, unique within this Actor run. A named run is recorded in the + default key-value store, so after a migration or resurrection of this Actor the same call reattaches + to the recorded run. A `SUCCEEDED` run is returned as is, an `ABORTED` or `TIMED-OUT` one is + resurrected, and a new run is started only when nothing is recorded under the name, or the recorded + run `FAILED` or no longer exists. The name is bound to the task and input it was first used with, + so reusing it for a different Actor, task or input raises a `ValueError`. Returns: Info about the started Actor run. @@ -1229,7 +1370,8 @@ async def start_task( client = self.new_client(token=token) if token else self.apify_client task_client = client.task(task_id) - return await task_client.start( + start_run = partial( + task_client.start, task_input=task_input, build=build, max_items=max_items, @@ -1240,6 +1382,25 @@ async def start_task( webhooks=to_client_representations(webhooks), ) + if run_name is None: + return await start_run() + + run, _ = await self._find_or_start_child_run( + run_name, + task_id=task_id, + run_input=task_input, + client=client, + start_run=start_run, + token=token, + build=build, + max_items=max_items, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + timeout=timeout, + ) + return run + @_ensure_context async def call_task( self, @@ -1255,6 +1416,7 @@ async def call_task( webhooks: list[Webhook] | None = None, wait: timedelta | None = None, token: str | None = None, + run_name: str | None = None, ) -> Run: """Start an Actor task on the Apify Platform and wait for it to finish before returning. @@ -1284,6 +1446,12 @@ async def call_task( be used to receive a notification, e.g. when the Actor finished or failed. If you already have a webhook set up for the Actor, you do not have to add it again here. wait: The maximum time the server waits for the run to finish. If not provided, waits indefinitely. + run_name: Optional name of the child run, unique within this Actor run. A named run is recorded in the + default key-value store, so after a migration or resurrection of this Actor the same call reattaches + to the recorded run. A `SUCCEEDED` run is returned as is, an `ABORTED` or `TIMED-OUT` one is + resurrected, and a new run is started only when nothing is recorded under the name, or the recorded + run `FAILED` or no longer exists. The name is bound to the task and input it was first used with, + so reusing it for a different Actor, task or input raises a `ValueError`. Returns: Info about the started Actor run. @@ -1294,17 +1462,47 @@ async def call_task( client = self.new_client(token=token) if token else self.apify_client task_client = client.task(task_id) - run = await task_client.call( - task_input=task_input, - build=build, - max_items=max_items, - max_total_charge_usd=max_total_charge_usd, - restart_on_error=restart_on_error, - memory_mbytes=memory_mbytes, - run_timeout=self._resolve_run_timeout(timeout), - webhooks=to_client_representations(webhooks), - wait_duration=wait, - ) + + if run_name is None: + run = await task_client.call( + task_input=task_input, + build=build, + max_items=max_items, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=self._resolve_run_timeout(timeout), + webhooks=to_client_representations(webhooks), + wait_duration=wait, + ) + else: + started_run, _ = await self._find_or_start_child_run( + run_name, + task_id=task_id, + run_input=task_input, + client=client, + start_run=partial( + task_client.start, + task_input=task_input, + build=build, + max_items=max_items, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=self._resolve_run_timeout(timeout), + webhooks=to_client_representations(webhooks), + ), + token=token, + build=build, + max_items=max_items, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + timeout=timeout, + ) + run = await self._wait_for_child_run( + run_name, client.run(started_run.id), started_run, wait=wait, logger=None, from_start=False + ) if run is None: raise RuntimeError(f'Failed to call Task with ID "{task_id}".') diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py new file mode 100644 index 00000000..680c64a4 --- /dev/null +++ b/src/apify/_child_runs.py @@ -0,0 +1,227 @@ +from __future__ import annotations + +import asyncio +import hashlib +import json +from datetime import datetime +from logging import getLogger +from typing import TYPE_CHECKING, Any +from weakref import WeakValueDictionary + +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError +from pydantic.alias_generators import to_camel + +if TYPE_CHECKING: + from collections.abc import Awaitable, Callable + + from apify_client import ApifyClientAsync + from apify_client._models import Run + from apify_client._resource_clients import RunClientAsync + + from apify.storages import KeyValueStore + +logger = getLogger(__name__) + +CHILD_RUNS_KEY = '__ACTOR_CHILD_RUNS' +"""Key in the default key-value store under which the child run registry is persisted.""" + +_SETTLING_STATUSES = frozenset({'ABORTING', 'TIMING-OUT'}) +"""Statuses that end as `ABORTED` / `TIMED-OUT` shortly, and are resurrectable once they do.""" + +_RESURRECTABLE_STATUSES = frozenset({'ABORTED', 'TIMED-OUT'}) + +_NOT_FOUND_GRACE_SECS = 3 +"""How long a recorded run that the API reports as missing is looked up again before it counts as gone.""" + +_NOT_FOUND_RETRY_INTERVAL_SECS = 0.25 + + +class ChildRunSnapshot(BaseModel): + """A child run as last observed by this Actor run.""" + + model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) + + run_id: str + """ID of the run.""" + + status: str + """Last status of the run observed by this Actor run, or `LOST` once the platform no longer returned it.""" + + started_at: datetime + """When the run started.""" + + +class ChildRunRecord(ChildRunSnapshot): + """The current run tracked under a name in the child run registry, with the runs it replaced.""" + + checksum: str + """Hash of the Actor or task and the input the name was first used with.""" + + history: list[ChildRunSnapshot] = Field(default_factory=list) + """Earlier runs under this name that failed or went missing and were replaced by a new run, oldest first.""" + + +def checksum_request(*, actor_id: str | None, task_id: str | None, run_input: Any) -> str: + """Hash the Actor or task and the input of a named start, in the same JSON shape as the JS SDK.""" + request: dict[str, Any] = ( + {'type': 'actor', 'id': actor_id} if actor_id is not None else {'type': 'task', 'id': task_id} + ) + if run_input is not None: + request['input'] = run_input + serialized = json.dumps(request, sort_keys=True, separators=(',', ':'), ensure_ascii=False, default=str) + return hashlib.sha256(serialized.encode()).hexdigest() + + +async def _get_recorded_run(run_client: RunClientAsync) -> Run | None: + """Fetch a recorded run, retrying a 404 for a few seconds. + + A run started moments ago, e.g. by a concurrent call under the same name, may not be on every API replica yet, + and treating it as gone would start a duplicate. + """ + loop = asyncio.get_running_loop() + deadline = loop.time() + _NOT_FOUND_GRACE_SECS + while True: + run = await run_client.get() + if run is not None or loop.time() >= deadline: + return run + await asyncio.sleep(_NOT_FOUND_RETRY_INTERVAL_SECS) + + +_records_adapter = TypeAdapter(dict[str, ChildRunRecord]) + + +class ChildRunRegistry: + """Persisted name -> run map that lets named child runs survive a migration or resurrection of the parent. + + Every change is written to the key-value store right away. A hard kill of the parent between the platform + starting the child and that write can still orphan the child, since nothing but the platform knows about it. + """ + + def __init__(self, open_key_value_store: Callable[[], Awaitable[KeyValueStore]]) -> None: + self._open_key_value_store = open_key_value_store + self._records: dict[str, ChildRunRecord] | None = None + self._lock = asyncio.Lock() + """Guards loading the records and writing them back to the key-value store.""" + self._name_locks: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary() + """Serializes `find_or_start` per name. A lock is dropped once no call under its name holds it.""" + + async def find_or_start( + self, + name: str, + *, + actor_id: str | None = None, + task_id: str | None = None, + run_input: Any, + client: ApifyClientAsync, + start_run: Callable[[], Awaitable[Run]], + resurrect_run: Callable[[str], Awaitable[Run]], + ) -> tuple[Run, bool]: + """Return the run recorded under `name`, or start one when there is none to reuse. + + A recorded run that is `READY` or `RUNNING` is reattached and one that `SUCCEEDED` is returned as is. + An `ABORTED` or `TIMED-OUT` run is resurrected, since Actors are expected to resume from their state. + A `FAILED` run, or one the API no longer knows, is replaced by a new run under the same name. + + Args: + name: Name of the child run, unique within the parent run. + actor_id: The Actor to start. + task_id: The task to start, in place of `actor_id`. + run_input: Input of the run. With the Actor or task, it must match what `name` was first used with. + client: Client used to look up the recorded run. + start_run: Starts a new run of the Actor or task. + resurrect_run: Resurrects the recorded run, given its ID. + + Returns: + The run, and whether it was newly started. + """ + checksum = checksum_request(actor_id=actor_id, task_id=task_id, run_input=run_input) + + async with self._name_locks.setdefault(name, asyncio.Lock()): + records = await self._load() + record = records.get(name) + + if record is None: + run = await self._start(name, checksum=checksum, start_run=start_run, history=[]) + return run, True + + if record.checksum != checksum: + raise ValueError( + f'The run name "{name}" was already used for a different Actor, task or input. ' + 'Use a unique `run_name` for each child run.' + ) + + run_client = client.run(record.run_id) + run = await _get_recorded_run(run_client) + + if run is not None and run.status in _SETTLING_STATUSES: + run = await run_client.wait_for_finish() + + if run is None or run.status == 'FAILED': + replaced = ChildRunSnapshot( + run_id=record.run_id, + status=run.status if run is not None else 'LOST', + started_at=record.started_at, + ) + run = await self._start( + name, checksum=checksum, start_run=start_run, history=[*record.history, replaced] + ) + return run, True + + if run.status in _RESURRECTABLE_STATUSES: + logger.info(f'Resurrecting child run "{name}"', extra={'run_id': run.id, 'status': run.status}) + run = await resurrect_run(run.id) + else: + logger.info(f'Reattaching to child run "{name}"', extra={'run_id': run.id, 'status': run.status}) + + await self.update(name, run) + return run, False + + async def update(self, name: str, run: Run) -> None: + """Record the latest observed status of the run recorded under `name`. + + Args: + name: Name of the child run. + run: The run as just returned by the API. Nothing is recorded unless it is the current run under `name`. + """ + record = (await self._load()).get(name) + if record is None or record.run_id != run.id or record.status == run.status: + return + await self._save(name, record.model_copy(update={'status': run.status})) + + async def _start( + self, + name: str, + *, + checksum: str, + start_run: Callable[[], Awaitable[Run]], + history: list[ChildRunSnapshot], + ) -> Run: + run = await start_run() + record = ChildRunRecord( + run_id=run.id, status=run.status, started_at=run.started_at, checksum=checksum, history=history + ) + await self._save(name, record) + return run + + async def _load(self) -> dict[str, ChildRunRecord]: + async with self._lock: + if self._records is None: + key_value_store = await self._open_key_value_store() + stored = await key_value_store.get_value(CHILD_RUNS_KEY) + try: + self._records = _records_adapter.validate_python(stored or {}) + except ValidationError as exc: + raise ValueError( + f'The child run registry under the "{CHILD_RUNS_KEY}" key in the default key-value store ' + 'is malformed.' + ) from exc + return self._records + + async def _save(self, name: str, record: ChildRunRecord) -> None: + records = await self._load() + key_value_store = await self._open_key_value_store() + async with self._lock: + records[name] = record + await key_value_store.set_value( + CHILD_RUNS_KEY, _records_adapter.dump_python(records, by_alias=True, mode='json') + ) diff --git a/tests/e2e/test_actor_child_runs.py b/tests/e2e/test_actor_child_runs.py new file mode 100644 index 00000000..1da589fd --- /dev/null +++ b/tests/e2e/test_actor_child_runs.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING + +from apify import Actor + +if TYPE_CHECKING: + from .conftest import MakeActorFunction, RunActorFunction + + +async def test_named_child_run_is_reattached_after_reboot( + make_actor: MakeActorFunction, + run_actor: RunActorFunction, +) -> None: + """A named child run started before a reboot is reattached and awaited by a named call after it.""" + + async def main() -> None: + async with Actor: + actor_input = (await Actor.get_input()) or {} + if actor_input.get('is_child') is True: + await asyncio.sleep(20) + return + + actor_id = Actor.configuration.actor_id or '' + child_run_id = await Actor.get_value('child_run_id') + + if child_run_id is None: + run = await Actor.start(actor_id=actor_id, run_input={'is_child': True}, run_name='child') + await Actor.set_value('child_run_id', run.id) + await Actor.reboot() + return + + run = await Actor.call(actor_id=actor_id, run_input={'is_child': True}, run_name='child') + assert run is not None, 'run is None' + assert run.id == child_run_id, f'run.id={run.id}, child_run_id={child_run_id}' + assert run.status == 'SUCCEEDED', f'run.status={run.status}' + + actor = await make_actor(label='child-run-reattach', main_func=main) + run_result = await run_actor(actor) + + assert run_result.status == 'SUCCEEDED' + # The parent run and the one child run it reattached to. + assert (await actor.runs().list()).total == 2 + + +async def test_named_aborted_child_run_is_resurrected_after_reboot( + make_actor: MakeActorFunction, + run_actor: RunActorFunction, +) -> None: + """A named child run aborted before a reboot is resurrected by a named start after it.""" + + async def main() -> None: + async with Actor: + actor_input = (await Actor.get_input()) or {} + if actor_input.get('is_child') is True: + await asyncio.sleep(300) + return + + actor_id = Actor.configuration.actor_id or '' + child_run_id = await Actor.get_value('child_run_id') + + if child_run_id is None: + run = await Actor.start(actor_id=actor_id, run_input={'is_child': True}, run_name='child') + await Actor.set_value('child_run_id', run.id) + run_client = Actor.apify_client.run(run.id) + await run_client.abort() + aborted_run = await run_client.wait_for_finish() + assert aborted_run is not None, 'aborted_run is None' + assert aborted_run.status == 'ABORTED', f'aborted_run.status={aborted_run.status}' + await Actor.reboot() + return + + run = await Actor.start(actor_id=actor_id, run_input={'is_child': True}, run_name='child') + try: + assert run.id == child_run_id, f'run.id={run.id}, child_run_id={child_run_id}' + assert run.status in {'READY', 'RUNNING'}, f'run.status={run.status}' + finally: + await Actor.apify_client.run(run.id).abort() + + actor = await make_actor(label='child-run-resurrect', main_func=main) + run_result = await run_actor(actor) + + assert run_result.status == 'SUCCEEDED' + # The parent run and the one child run it resurrected. + assert (await actor.runs().list()).total == 2 diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py new file mode 100644 index 00000000..10f17a83 --- /dev/null +++ b/tests/unit/actor/test_actor_child_runs.py @@ -0,0 +1,609 @@ +from __future__ import annotations + +import asyncio +from datetime import timedelta +from typing import TYPE_CHECKING, Any +from unittest.mock import MagicMock, Mock + +import pytest + +from apify_client._models import Run + +from apify import Actor, _child_runs +from apify._actor import _ActorType +from apify._child_runs import CHILD_RUNS_KEY, checksum_request + +if TYPE_CHECKING: + from ..conftest import ApifyClientAsyncPatcher + from apify.storages import KeyValueStore + + +def make_run(run_id: str, status: str) -> Run: + return Run.model_validate( + { + 'id': run_id, + 'actId': 'actor_id', + 'userId': 'user_id', + 'startedAt': STARTED_AT, + 'status': status, + 'meta': {'origin': 'API'}, + 'buildId': 'build_id', + 'defaultDatasetId': 'dataset_id', + 'defaultKeyValueStoreId': 'kvs_id', + 'defaultRequestQueueId': 'rq_id', + 'generalAccess': 'RESTRICTED', + 'stats': {'restartCount': 0, 'resurrectCount': 0, 'computeUnits': 0}, + 'options': {'build': '', 'timeoutSecs': 44, 'memoryMbytes': 4096, 'diskMbytes': 16384}, + } + ) + + +STARTED_AT = '2024-08-08T12:12:44Z' + + +def stored_record( + run_id: str, + status: str, + *, + actor_id: str | None = 'some-actor', + task_id: str | None = None, + run_input: Any = None, + history: list[dict[str, str]] | None = None, +) -> dict[str, Any]: + """Build a registry record as it is stored in the default KVS.""" + return { + 'runId': run_id, + 'status': status, + 'startedAt': STARTED_AT, + 'checksum': checksum_request(actor_id=actor_id, task_id=task_id, run_input=run_input), + 'history': history or [], + } + + +async def record_child_run( + name: str, + run_id: str, + *, + actor_id: str | None = 'some-actor', + task_id: str | None = None, + run_input: Any = None, +) -> None: + """Seed the registry the way an earlier attempt of this Actor run would have left it.""" + kvs = await Actor.open_key_value_store() + record = stored_record(run_id, 'RUNNING', actor_id=actor_id, task_id=task_id, run_input=run_input) + await kvs.set_value(CHILD_RUNS_KEY, {name: record}) + + +async def test_named_start_records_run_in_kvs(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named start persists the name -> run ID entry to the default KVS as soon as the run starts.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + + async with Actor: + run = await Actor.start('some-actor', run_name='scrape-eu') + kvs = await Actor.open_key_value_store() + stored = await kvs.get_value(CHILD_RUNS_KEY) + + assert run.id == 'new-run' + assert stored == {'scrape-eu': stored_record('new-run', 'READY')} + + +async def test_unnamed_start_is_not_recorded(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A start without a name leaves the registry untouched.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + + async with Actor: + await Actor.start('some-actor') + kvs = await Actor.open_key_value_store() + stored = await kvs.get_value(CHILD_RUNS_KEY) + + assert stored is None + + +@pytest.mark.parametrize( + 'status', + [ + pytest.param('READY', id='ready'), + pytest.param('RUNNING', id='running'), + pytest.param('SUCCEEDED', id='succeeded'), + ], +) +async def test_named_start_reuses_recorded_run( + apify_client_async_patcher: ApifyClientAsyncPatcher, status: str +) -> None: + """A recorded run that is active or succeeded is returned without starting a new one.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', status)) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + run = await Actor.start('some-actor', run_name='scrape-eu') + + assert run.id == 'old-run' + assert run.status == status + assert apify_client_async_patcher.calls['actor']['start'] == [] + + +@pytest.mark.parametrize( + 'status', + [ + pytest.param('ABORTED', id='aborted'), + pytest.param('TIMED-OUT', id='timed out'), + ], +) +async def test_named_start_resurrects_recorded_run( + apify_client_async_patcher: ApifyClientAsyncPatcher, status: str +) -> None: + """A recorded run that was aborted or timed out is resurrected, and its options are passed through.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', status)) + apify_client_async_patcher.patch('run', 'resurrect', return_value=make_run('old-run', 'RUNNING')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + run = await Actor.start('some-actor', run_name='scrape-eu', memory_mbytes=2048) + + assert run.id == 'old-run' + assert run.status == 'RUNNING' + assert apify_client_async_patcher.calls['actor']['start'] == [] + [(args, kwargs)] = apify_client_async_patcher.calls['run']['resurrect'] + assert args[0].resource_id == 'old-run' + assert kwargs['memory_mbytes'] == 2048 + + +@pytest.mark.parametrize( + 'status', + [ + pytest.param('ABORTING', id='aborting'), + pytest.param('TIMING-OUT', id='timing out'), + ], +) +async def test_named_start_resurrects_settling_run_after_it_finishes( + apify_client_async_patcher: ApifyClientAsyncPatcher, status: str +) -> None: + """A recorded run still aborting or timing out is waited for, then resurrected.""" + finished_status = {'ABORTING': 'ABORTED', 'TIMING-OUT': 'TIMED-OUT'}[status] + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', status)) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=make_run('old-run', finished_status)) + apify_client_async_patcher.patch('run', 'resurrect', return_value=make_run('old-run', 'RUNNING')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + run = await Actor.start('some-actor', run_name='scrape-eu') + + assert run.status == 'RUNNING' + assert len(apify_client_async_patcher.calls['run']['wait_for_finish']) == 1 + assert len(apify_client_async_patcher.calls['run']['resurrect']) == 1 + + +@pytest.mark.parametrize( + ('recorded_run', 'replaced_status'), + [ + pytest.param(make_run('old-run', 'FAILED'), 'FAILED', id='failed'), + pytest.param(None, 'LOST', id='not found'), + ], +) +async def test_named_start_replaces_failed_or_missing_run( + apify_client_async_patcher: ApifyClientAsyncPatcher, + monkeypatch: pytest.MonkeyPatch, + recorded_run: Run | None, + replaced_status: str, +) -> None: + """A recorded run that failed or no longer exists is replaced by a new run and kept in the history.""" + monkeypatch.setattr(_child_runs, '_NOT_FOUND_GRACE_SECS', 0) + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'get', return_value=recorded_run) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + run = await Actor.start('some-actor', run_name='scrape-eu') + kvs = await Actor.open_key_value_store() + stored = await kvs.get_value(CHILD_RUNS_KEY) + + assert run.id == 'new-run' + assert stored == { + 'scrape-eu': stored_record( + 'new-run', 'READY', history=[{'runId': 'old-run', 'status': replaced_status, 'startedAt': STARTED_AT}] + ) + } + + +async def test_named_start_rejects_name_recorded_for_another_actor( + apify_client_async_patcher: ApifyClientAsyncPatcher, +) -> None: + """Reusing a name for a different Actor raises instead of attaching to the other Actor's run.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run', actor_id='other-actor') + with pytest.raises(ValueError, match='already used for a different Actor, task or input'): + await Actor.start('some-actor', run_name='scrape-eu') + + assert apify_client_async_patcher.calls['actor']['start'] == [] + + +async def test_named_start_rejects_name_recorded_for_task( + apify_client_async_patcher: ApifyClientAsyncPatcher, +) -> None: + """Reusing a task run's name for an Actor raises instead of attaching to the task's run.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run', actor_id=None, task_id='some-task') + with pytest.raises(ValueError, match='already used for a different Actor, task or input'): + await Actor.start('some-actor', run_name='scrape-eu') + + assert apify_client_async_patcher.calls['actor']['start'] == [] + + +async def test_concurrent_named_starts_start_one_run(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """Concurrent starts under the same name start a single run and share it.""" + started = make_run('new-run', 'RUNNING') + + async def slow_start(*_args: Any, **_kwargs: Any) -> Run: + await asyncio.sleep(0.05) + return started + + apify_client_async_patcher.patch('actor', 'start', replacement_method=slow_start) + apify_client_async_patcher.patch('run', 'get', return_value=started) + + async with Actor: + runs = await asyncio.gather(*(Actor.start('some-actor', run_name='scrape-eu') for _ in range(3))) + + assert {run.id for run in runs} == {'new-run'} + assert len(apify_client_async_patcher.calls['actor']['start']) == 1 + + +async def test_concurrent_first_named_starts_share_one_registry( + apify_client_async_patcher: ApifyClientAsyncPatcher, monkeypatch: pytest.MonkeyPatch +) -> None: + """Concurrent first named starts start a single run even when opening the default KVS yields to the event loop.""" + started = make_run('new-run', 'RUNNING') + + async def slow_start(*_args: Any, **_kwargs: Any) -> Run: + await asyncio.sleep(0.05) + return started + + apify_client_async_patcher.patch('actor', 'start', replacement_method=slow_start) + apify_client_async_patcher.patch('run', 'get', return_value=started) + open_key_value_store = _ActorType.open_key_value_store + + async def yielding_open_key_value_store(self: _ActorType, *args: Any, **kwargs: Any) -> KeyValueStore: + # On the platform the default KVS is opened lazily through the API, so opening it suspends. + await asyncio.sleep(0.01) + return await open_key_value_store(self, *args, **kwargs) + + monkeypatch.setattr(_ActorType, 'open_key_value_store', yielding_open_key_value_store) + + # A fresh instance, since the registry binds the opener when the Actor is created. + async with _ActorType() as actor: + runs = await asyncio.gather(*(actor.start('some-actor', run_name='scrape-eu') for _ in range(3))) + + assert {run.id for run in runs} == {'new-run'} + assert len(apify_client_async_patcher.calls['actor']['start']) == 1 + + +async def test_named_call_waits_for_reattached_run(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named call waits for the reattached run and streams only its new log lines.""" + streamed_log = MagicMock() + get_streamed_log = Mock(return_value=streamed_log) + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', 'RUNNING')) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=make_run('old-run', 'SUCCEEDED')) + apify_client_async_patcher.patch('run', 'get_status_message_watcher', return_value=MagicMock()) + apify_client_async_patcher.patch('run', 'get_streamed_log', replacement_method=get_streamed_log) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + run = await Actor.call('some-actor', run_name='scrape-eu') + + assert run.status == 'SUCCEEDED' + assert get_streamed_log.call_args.kwargs['from_start'] is False + streamed_log.__aenter__.assert_awaited_once() + + +async def test_named_call_streams_new_run_log_from_start(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named call that starts a new run streams its log from the start.""" + get_streamed_log = Mock(return_value=MagicMock()) + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=make_run('new-run', 'SUCCEEDED')) + apify_client_async_patcher.patch('run', 'get_status_message_watcher', return_value=MagicMock()) + apify_client_async_patcher.patch('run', 'get_streamed_log', replacement_method=get_streamed_log) + + async with Actor: + run = await Actor.call('some-actor', run_name='scrape-eu') + + assert run.id == 'new-run' + assert run.status == 'SUCCEEDED' + assert get_streamed_log.call_args.kwargs['from_start'] is True + assert apify_client_async_patcher.calls['actor']['call'] == [] + + +async def test_named_call_returns_succeeded_run_without_waiting( + apify_client_async_patcher: ApifyClientAsyncPatcher, +) -> None: + """A named call whose recorded run already succeeded returns it without waiting or streaming logs.""" + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', 'SUCCEEDED')) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=None) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + run = await Actor.call('some-actor', run_name='scrape-eu') + + assert run.id == 'old-run' + assert apify_client_async_patcher.calls['run']['wait_for_finish'] == [] + + +async def test_named_call_waits_for_resurrected_run(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named call resurrects an aborted run with its own timeout and streams only the new log lines.""" + get_streamed_log = Mock(return_value=MagicMock()) + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', 'ABORTED')) + apify_client_async_patcher.patch('run', 'resurrect', return_value=make_run('old-run', 'RUNNING')) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=make_run('old-run', 'SUCCEEDED')) + apify_client_async_patcher.patch('run', 'get_status_message_watcher', return_value=MagicMock()) + apify_client_async_patcher.patch('run', 'get_streamed_log', replacement_method=get_streamed_log) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + run = await Actor.call('some-actor', run_name='scrape-eu', timeout=timedelta(minutes=5)) + + assert run.id == 'old-run' + assert run.status == 'SUCCEEDED' + assert apify_client_async_patcher.calls['actor']['start'] == [] + [(_, kwargs)] = apify_client_async_patcher.calls['run']['resurrect'] + assert kwargs['run_timeout'] == timedelta(minutes=5) + assert get_streamed_log.call_args.kwargs['from_start'] is False + + +async def test_named_call_without_logger_only_waits(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named call with `logger=None` waits for the run without redirecting its log or status messages.""" + get_streamed_log = Mock(return_value=MagicMock()) + get_status_message_watcher = Mock(return_value=MagicMock()) + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=make_run('new-run', 'SUCCEEDED')) + apify_client_async_patcher.patch('run', 'get_status_message_watcher', replacement_method=get_status_message_watcher) + apify_client_async_patcher.patch('run', 'get_streamed_log', replacement_method=get_streamed_log) + + async with Actor: + run = await Actor.call('some-actor', run_name='scrape-eu', logger=None) + + assert run.status == 'SUCCEEDED' + assert len(apify_client_async_patcher.calls['run']['wait_for_finish']) == 1 + get_streamed_log.assert_not_called() + get_status_message_watcher.assert_not_called() + + +async def test_named_start_rejects_malformed_registry(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A malformed registry in the default KVS raises a `ValueError` naming the key, without starting a run.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + + async with Actor: + kvs = await Actor.open_key_value_store() + await kvs.set_value(CHILD_RUNS_KEY, {'scrape-eu': {'runId': 'old-run'}}) + with pytest.raises(ValueError, match=CHILD_RUNS_KEY): + await Actor.start('some-actor', run_name='scrape-eu') + + assert apify_client_async_patcher.calls['actor']['start'] == [] + + +async def test_named_call_task_records_run_in_kvs(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named task call starts the task, records the run under the task ID, and waits for it.""" + apify_client_async_patcher.patch('task', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=make_run('new-run', 'SUCCEEDED')) + + async with Actor: + run = await Actor.call_task('some-task', run_name='scrape-eu') + kvs = await Actor.open_key_value_store() + stored = await kvs.get_value(CHILD_RUNS_KEY) + + assert run.status == 'SUCCEEDED' + assert stored == {'scrape-eu': stored_record('new-run', 'SUCCEEDED', actor_id=None, task_id='some-task')} + assert apify_client_async_patcher.calls['task']['call'] == [] + + +async def test_named_call_task_waits_for_reattached_run(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named task call waits for the recorded running run without starting the task again.""" + apify_client_async_patcher.patch('task', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', 'RUNNING')) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=make_run('old-run', 'SUCCEEDED')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run', actor_id=None, task_id='some-task') + run = await Actor.call_task('some-task', run_name='scrape-eu') + + assert run.id == 'old-run' + assert run.status == 'SUCCEEDED' + assert apify_client_async_patcher.calls['task']['start'] == [] + + +async def test_named_call_task_resurrects_recorded_run(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named task call resurrects an aborted run with its own options.""" + apify_client_async_patcher.patch('task', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', 'ABORTED')) + apify_client_async_patcher.patch('run', 'resurrect', return_value=make_run('old-run', 'RUNNING')) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=make_run('old-run', 'SUCCEEDED')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run', actor_id=None, task_id='some-task') + run = await Actor.call_task('some-task', run_name='scrape-eu', memory_mbytes=2048) + + assert run.id == 'old-run' + assert apify_client_async_patcher.calls['task']['start'] == [] + [(_, kwargs)] = apify_client_async_patcher.calls['run']['resurrect'] + assert kwargs['memory_mbytes'] == 2048 + + +async def test_named_call_task_rejects_name_recorded_for_actor( + apify_client_async_patcher: ApifyClientAsyncPatcher, +) -> None: + """Reusing an Actor run's name for a task raises instead of attaching to the Actor's run.""" + apify_client_async_patcher.patch('task', 'start', return_value=make_run('new-run', 'READY')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + with pytest.raises(ValueError, match='already used for a different Actor, task or input'): + await Actor.call_task('some-task', run_name='scrape-eu') + + assert apify_client_async_patcher.calls['task']['start'] == [] + + +async def test_named_runs_forward_max_items_to_start(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """Named `start`, `call` and `call_task` pass `max_items` to the started run.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('task', 'start', return_value=make_run('new-task-run', 'READY')) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=make_run('new-run', 'SUCCEEDED')) + + async with Actor: + await Actor.start('some-actor', run_name='started', max_items=10) + await Actor.call('some-actor', run_name='called', max_items=20, logger=None) + await Actor.call_task('some-task', run_name='task-called', max_items=30) + + assert [kwargs['max_items'] for _, kwargs in apify_client_async_patcher.calls['actor']['start']] == [10, 20] + [(_, task_kwargs)] = apify_client_async_patcher.calls['task']['start'] + assert task_kwargs['max_items'] == 30 + + +async def test_named_start_forwards_max_items_to_resurrect( + apify_client_async_patcher: ApifyClientAsyncPatcher, +) -> None: + """A named start that resurrects the recorded run passes `max_items` to the resurrection.""" + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', 'ABORTED')) + apify_client_async_patcher.patch('run', 'resurrect', return_value=make_run('old-run', 'RUNNING')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + await Actor.start('some-actor', run_name='scrape-eu', max_items=10) + + [(_, kwargs)] = apify_client_async_patcher.calls['run']['resurrect'] + assert kwargs['max_items'] == 10 + + +async def test_named_start_task_records_run_in_kvs(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named task start records the run under the task ID without waiting for it.""" + apify_client_async_patcher.patch('task', 'start', return_value=make_run('new-run', 'READY')) + + async with Actor: + run = await Actor.start_task('some-task', run_name='scrape-eu') + kvs = await Actor.open_key_value_store() + stored = await kvs.get_value(CHILD_RUNS_KEY) + + assert run.id == 'new-run' + assert stored == {'scrape-eu': stored_record('new-run', 'READY', actor_id=None, task_id='some-task')} + assert apify_client_async_patcher.calls['run']['wait_for_finish'] == [] + + +async def test_named_start_task_reuses_recorded_run(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named task start returns the recorded running run without starting the task again.""" + apify_client_async_patcher.patch('task', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', 'RUNNING')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run', actor_id=None, task_id='some-task') + run = await Actor.start_task('some-task', run_name='scrape-eu') + + assert run.id == 'old-run' + assert apify_client_async_patcher.calls['task']['start'] == [] + + +async def test_name_lock_is_dropped_after_named_start(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """The per-name lock is released from the registry once no named start holds it.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + + async with _ActorType() as actor: + await actor.start('some-actor', run_name='scrape-eu') + assert len(actor._child_run_registry._name_locks) == 0 + + +async def test_named_start_retries_recorded_run_not_found_yet( + apify_client_async_patcher: ApifyClientAsyncPatcher, +) -> None: + """A recorded run the API briefly reports as missing is looked up again and reattached, not replaced.""" + get_run = Mock(side_effect=[None, make_run('old-run', 'RUNNING')]) + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'get', replacement_method=get_run) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + run = await Actor.start('some-actor', run_name='scrape-eu') + + assert run.id == 'old-run' + assert get_run.call_count == 2 + assert apify_client_async_patcher.calls['actor']['start'] == [] + + +async def test_named_start_rejects_name_recorded_with_other_input( + apify_client_async_patcher: ApifyClientAsyncPatcher, +) -> None: + """Reusing a name for the same Actor with a different input raises instead of attaching to the earlier run.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run', run_input={'since': '2025-01-01'}) + with pytest.raises(ValueError, match='already used for a different Actor, task or input'): + await Actor.start('some-actor', {'since': '2026-01-01'}, run_name='scrape-eu') + + assert apify_client_async_patcher.calls['actor']['start'] == [] + + +async def test_named_start_ignores_input_key_order(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """An input equal to the recorded one up to key order reattaches to the recorded run.""" + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', 'RUNNING')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run', run_input={'a': 1, 'nested': {'x': 1, 'y': 2}}) + run = await Actor.start('some-actor', {'nested': {'y': 2, 'x': 1}, 'a': 1}, run_name='scrape-eu') + + assert run.id == 'old-run' + + +@pytest.mark.parametrize( + ('request_kwargs', 'expected'), + [ + pytest.param( + { + 'actor_id': 'some-actor', + 'task_id': None, + 'run_input': { + 'urls': ['https://example.com'], + 'maxPages': 10, + 'nested': {'z': True, 'a': None}, + 'name': 'Žluťoučký', + }, + }, + '79f4451cdbbc9bce11e153be41fe35ab3b0e64d1a0122cf7fafe897b6dfbab63', + id='actor with input', + ), + pytest.param( + {'actor_id': None, 'task_id': 'some-task', 'run_input': None}, + '141ee5857ab4cc98bc2cb9db49accebb94a84a44633abfc94a0ae2e80ad3c2cc', + id='task without input', + ), + ], +) +def test_checksum_matches_js_sdk(request_kwargs: dict[str, Any], expected: str) -> None: + """The request checksum equals the one the JS SDK computes for the same Actor or task and input.""" + assert checksum_request(**request_kwargs) == expected + + +async def test_reattached_run_status_is_recorded(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """Reattaching to a recorded run stores its current status.""" + apify_client_async_patcher.patch('run', 'get', return_value=make_run('old-run', 'SUCCEEDED')) + + async with Actor: + await record_child_run('scrape-eu', 'old-run') + await Actor.start('some-actor', run_name='scrape-eu') + kvs = await Actor.open_key_value_store() + stored = await kvs.get_value(CHILD_RUNS_KEY) + + assert stored == {'scrape-eu': stored_record('old-run', 'SUCCEEDED')} + + +async def test_named_call_records_finished_status(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named call stores the status of the run once it finishes.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('run', 'wait_for_finish', return_value=make_run('new-run', 'FAILED')) + + async with Actor: + await Actor.call('some-actor', run_name='scrape-eu', logger=None) + kvs = await Actor.open_key_value_store() + stored = await kvs.get_value(CHILD_RUNS_KEY) + + assert stored == {'scrape-eu': stored_record('new-run', 'FAILED')}