From e6a3638505cf5075254fff5bf60006ce5765238b Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 29 Sep 2026 13:42:26 +0200 Subject: [PATCH 01/14] feat: reattach named child runs after migration or resurrection --- src/apify/_actor.py | 146 +++++++++++-- src/apify/_child_runs.py | 144 +++++++++++++ tests/unit/actor/test_actor_child_runs.py | 248 ++++++++++++++++++++++ 3 files changed, 522 insertions(+), 16 deletions(-) create mode 100644 src/apify/_child_runs.py create mode 100644 tests/unit/actor/test_actor_child_runs.py diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 3c5845ee..9179e579 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 @@ -151,6 +153,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 | None = None + self._active = False """Whether the Actor instance is currently active (initialized and within context).""" @@ -942,6 +946,7 @@ async def start( timeout: timedelta | Literal['inherit'] | None = None, force_permission_level: ActorPermissionLevel | None = None, webhooks: list[Webhook] | None = None, + name: str | None = None, ) -> Run: """Run an Actor on the Apify platform. @@ -968,6 +973,11 @@ 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. + 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`. Returns: Info about the started Actor run @@ -984,7 +994,8 @@ async def start( raise ValueError(f'Invalid timeout {timeout!r}: expected `None`, `"inherit"`, or a `timedelta`.') 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, @@ -996,6 +1007,22 @@ async def start( webhooks=to_client_representations(webhooks), ) + if name is None: + return await start_run() + + run, _ = await self._find_or_start_child_run( + name, + actor_id=actor_id, + client=client, + start_run=start_run, + build=build, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=actor_start_timeout, + ) + return run + @_ensure_context async def abort( self, @@ -1050,6 +1077,7 @@ async def call( webhooks: list[Webhook] | None = None, wait: timedelta | None = None, logger: logging.Logger | Literal['default'] | None = 'default', + name: str | None = None, ) -> Run: """Start an Actor on the Apify Platform and wait for it to finish before returning. @@ -1079,6 +1107,11 @@ 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. + 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`. Returns: Info about the started Actor run. @@ -1095,25 +1128,106 @@ async def call( raise ValueError(f'Invalid timeout {timeout!r}: expected `None`, `"inherit"`, or a `timedelta`.') actor_client = client.actor(actor_id) - run = await actor_client.call( - run_input=run_input, - content_type=content_type, - build=build, - max_total_charge_usd=max_total_charge_usd, - restart_on_error=restart_on_error, - memory_mbytes=memory_mbytes, - run_timeout=actor_call_timeout, - force_permission_level=force_permission_level, - webhooks=to_client_representations(webhooks), - wait_duration=wait, - logger=logger, - ) + + if name is None: + run = await actor_client.call( + run_input=run_input, + content_type=content_type, + build=build, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=actor_call_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( + name, + actor_id=actor_id, + client=client, + start_run=partial( + actor_client.start, + run_input=run_input, + content_type=content_type, + build=build, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=actor_call_timeout, + force_permission_level=force_permission_level, + webhooks=to_client_representations(webhooks), + ), + build=build, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=actor_call_timeout, + ) + # The earlier attempt of this call already streamed the log of a reattached or resurrected run. + run = await self._wait_for_child_run( + 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, + client: ApifyClientAsync, + start_run: Callable[[], Awaitable[Run]], + build: str | None, + max_total_charge_usd: Decimal | None, + restart_on_error: bool | None, + memory_mbytes: int | None, + run_timeout: timedelta | None, + ) -> tuple[Run, bool]: + if self._child_run_registry is None: + self._child_run_registry = ChildRunRegistry(await self.open_key_value_store()) + + return await self._child_run_registry.find_or_start( + name, + actor_id=actor_id, + client=client, + start_run=start_run, + resurrect_run=lambda run_client: run_client.resurrect( + build=build, + max_total_charge_usd=max_total_charge_usd, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=run_timeout, + ), + ) + + async def _wait_for_child_run( + self, + 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: + return await run_client.wait_for_finish(wait_duration=wait) + + 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: + return await run_client.wait_for_finish(wait_duration=wait) + @_ensure_context async def call_task( self, diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py new file mode 100644 index 00000000..63fa9024 --- /dev/null +++ b/src/apify/_child_runs.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +import asyncio +from collections import defaultdict +from logging import getLogger +from typing import TYPE_CHECKING + +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter +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 = 'APIFY_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'}) + + +class ChildRunRecord(BaseModel): + """A child run tracked under a name in the child run registry.""" + + model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) + + actor_id: str + """The Actor ID or name the child was started with, as the caller passed it.""" + + run_id: str + """ID of the current run under this name.""" + + previous_run_ids: list[str] = Field(default_factory=list) + """IDs of earlier runs under this name that failed and were replaced by a new run, oldest first.""" + + +_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, key_value_store: KeyValueStore) -> None: + self._key_value_store = key_value_store + self._records: dict[str, ChildRunRecord] | None = None + self._load_lock = asyncio.Lock() + self._write_lock = asyncio.Lock() + self._name_locks: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock) + + async def find_or_start( + self, + name: str, + *, + actor_id: str, + client: ApifyClientAsync, + start_run: Callable[[], Awaitable[Run]], + resurrect_run: Callable[[RunClientAsync], 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. It must match the Actor already recorded under `name`. + client: Client used to look up and resurrect the recorded run. + start_run: Starts a new run of the Actor. + resurrect_run: Resurrects the recorded run, given its run client. + + Returns: + The run, and whether it was newly started. + """ + async with self._name_locks[name]: + records = await self._load() + record = records.get(name) + + if record is None: + return await self._start(name, actor_id=actor_id, start_run=start_run, previous_run_ids=[]), True + + if record.actor_id != actor_id: + raise ValueError( + f'Child run "{name}" is already recorded for Actor "{record.actor_id}", ' + f'it cannot be reused for Actor "{actor_id}".' + ) + + run_client = client.run(record.run_id) + run = await run_client.get() + + 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': + previous_run_ids = [*record.previous_run_ids, record.run_id] + run = await self._start(name, actor_id=actor_id, start_run=start_run, previous_run_ids=previous_run_ids) + return run, True + + if run.status in _RESURRECTABLE_STATUSES: + logger.info(f'Resurrecting child run "{name}"', extra={'run_id': run.id, 'status': run.status}) + return await resurrect_run(run_client), False + + logger.info(f'Reattaching to child run "{name}"', extra={'run_id': run.id, 'status': run.status}) + return run, False + + async def _start( + self, + name: str, + *, + actor_id: str, + start_run: Callable[[], Awaitable[Run]], + previous_run_ids: list[str], + ) -> Run: + run = await start_run() + await self._save(name, ChildRunRecord(actor_id=actor_id, run_id=run.id, previous_run_ids=previous_run_ids)) + return run + + async def _load(self) -> dict[str, ChildRunRecord]: + async with self._load_lock: + if self._records is None: + stored = await self._key_value_store.get_value(CHILD_RUNS_KEY) + self._records = _records_adapter.validate_python(stored or {}) + return self._records + + async def _save(self, name: str, record: ChildRunRecord) -> None: + records = await self._load() + async with self._write_lock: + records[name] = record + await self._key_value_store.set_value( + CHILD_RUNS_KEY, _records_adapter.dump_python(records, by_alias=True, mode='json') + ) 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..26c16225 --- /dev/null +++ b/tests/unit/actor/test_actor_child_runs.py @@ -0,0 +1,248 @@ +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING, Any +from unittest.mock import MagicMock, Mock + +import pytest + +from apify_client._models import Run + +from apify import Actor +from apify._child_runs import CHILD_RUNS_KEY + +if TYPE_CHECKING: + from ..conftest import ApifyClientAsyncPatcher + + +def make_run(run_id: str, status: str) -> Run: + return Run.model_validate( + { + 'id': run_id, + 'actId': 'actor_id', + 'userId': 'user_id', + 'startedAt': '2024-08-08T12:12:44Z', + '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}, + } + ) + + +async def record_child_run(name: str, run_id: str, *, actor_id: str = 'some-actor') -> None: + """Seed the registry the way an earlier attempt of this Actor run would have left it.""" + kvs = await Actor.open_key_value_store() + await kvs.set_value(CHILD_RUNS_KEY, {name: {'actorId': actor_id, 'runId': run_id, 'previousRunIds': []}}) + + +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', 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': {'actorId': 'some-actor', 'runId': 'new-run', 'previousRunIds': []}} + + +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', 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', 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', 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', + [ + pytest.param(make_run('old-run', 'FAILED'), id='failed'), + pytest.param(None, id='not found'), + ], +) +async def test_named_start_replaces_failed_or_missing_run( + apify_client_async_patcher: ApifyClientAsyncPatcher, recorded_run: Run | None +) -> None: + """A recorded run that failed or no longer exists is replaced by a new run and kept in the history.""" + 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', 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': {'actorId': 'some-actor', 'runId': 'new-run', 'previousRunIds': ['old-run']}} + + +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 recorded for Actor "other-actor"'): + await Actor.start('some-actor', 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', 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', 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', 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', name='scrape-eu') + + assert run.id == 'old-run' + assert apify_client_async_patcher.calls['run']['wait_for_finish'] == [] From ee91c8d4009179e0fd702786c2fbd905705d1a74 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 29 Sep 2026 14:00:03 +0200 Subject: [PATCH 02/14] docs: mention missing recorded runs in the child run name docstring --- src/apify/_actor.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 9179e579..36469d53 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -976,8 +976,8 @@ async def start( 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`. + resurrected, and a new run is started only when nothing is recorded under the name, or the recorded + run `FAILED` or no longer exists. Returns: Info about the started Actor run @@ -1110,8 +1110,8 @@ async def call( 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`. + resurrected, and a new run is started only when nothing is recorded under the name, or the recorded + run `FAILED` or no longer exists. Returns: Info about the started Actor run. From 0925146da86b87aa3a97f4964d4f2b41ee424521 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 29 Sep 2026 14:00:04 +0200 Subject: [PATCH 03/14] fix: share one child run registry across concurrent named starts --- src/apify/_actor.py | 5 +--- src/apify/_child_runs.py | 10 +++++--- tests/unit/actor/test_actor_child_runs.py | 31 +++++++++++++++++++++++ 3 files changed, 38 insertions(+), 8 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 36469d53..a67bcf65 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -153,7 +153,7 @@ 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 | None = None + self._child_run_registry = ChildRunRegistry(self.open_key_value_store) self._active = False """Whether the Actor instance is currently active (initialized and within context).""" @@ -1189,9 +1189,6 @@ async def _find_or_start_child_run( memory_mbytes: int | None, run_timeout: timedelta | None, ) -> tuple[Run, bool]: - if self._child_run_registry is None: - self._child_run_registry = ChildRunRegistry(await self.open_key_value_store()) - return await self._child_run_registry.find_or_start( name, actor_id=actor_id, diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py index 63fa9024..5c9f092c 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -53,8 +53,8 @@ class ChildRunRegistry: starting the child and that write can still orphan the child, since nothing but the platform knows about it. """ - def __init__(self, key_value_store: KeyValueStore) -> None: - self._key_value_store = key_value_store + 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._load_lock = asyncio.Lock() self._write_lock = asyncio.Lock() @@ -131,14 +131,16 @@ async def _start( async def _load(self) -> dict[str, ChildRunRecord]: async with self._load_lock: if self._records is None: - stored = await self._key_value_store.get_value(CHILD_RUNS_KEY) + key_value_store = await self._open_key_value_store() + stored = await key_value_store.get_value(CHILD_RUNS_KEY) self._records = _records_adapter.validate_python(stored or {}) 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._write_lock: records[name] = record - await self._key_value_store.set_value( + await key_value_store.set_value( CHILD_RUNS_KEY, _records_adapter.dump_python(records, by_alias=True, mode='json') ) diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index 26c16225..fd6bfeb9 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -9,10 +9,12 @@ from apify_client._models import Run from apify import Actor +from apify._actor import _ActorType from apify._child_runs import CHILD_RUNS_KEY if TYPE_CHECKING: from ..conftest import ApifyClientAsyncPatcher + from apify.storages import KeyValueStore def make_run(run_id: str, status: str) -> Run: @@ -198,6 +200,35 @@ async def slow_start(*_args: Any, **_kwargs: Any) -> 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', 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() From a841e9dc288d1763cfbbbb037ec0b20474cc324e Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 29 Sep 2026 15:01:06 +0200 Subject: [PATCH 04/14] test: add e2e tests for named child runs --- tests/e2e/test_actor_child_runs.py | 86 ++++++++++++++++++++++++++++++ 1 file changed, 86 insertions(+) create mode 100644 tests/e2e/test_actor_child_runs.py diff --git a/tests/e2e/test_actor_child_runs.py b/tests/e2e/test_actor_child_runs.py new file mode 100644 index 00000000..c9d0d9cc --- /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}, 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}, 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}, 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}, 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 From 8e76335d312fb48a26810be6e91060b1d20e4053 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 30 Sep 2026 09:21:49 +0200 Subject: [PATCH 05/14] feat: reject a malformed child run registry and cover the remaining call paths --- src/apify/_actor.py | 6 ++- src/apify/_child_runs.py | 10 ++++- tests/unit/actor/test_actor_child_runs.py | 53 +++++++++++++++++++++++ 3 files changed, 65 insertions(+), 4 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index a67bcf65..8aa532df 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -977,7 +977,8 @@ async def start( 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. + run `FAILED` or no longer exists. The name is bound to `actor_id` exactly as passed, so reusing it with + any other value raises a `ValueError`. Returns: Info about the started Actor run @@ -1111,7 +1112,8 @@ async def call( 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. + run `FAILED` or no longer exists. The name is bound to `actor_id` exactly as passed, so reusing it with + any other value raises a `ValueError`. Returns: Info about the started Actor run. diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py index 5c9f092c..05427df0 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -5,7 +5,7 @@ from logging import getLogger from typing import TYPE_CHECKING -from pydantic import BaseModel, ConfigDict, Field, TypeAdapter +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError from pydantic.alias_generators import to_camel if TYPE_CHECKING: @@ -133,7 +133,13 @@ async def _load(self) -> dict[str, ChildRunRecord]: if self._records is None: key_value_store = await self._open_key_value_store() stored = await key_value_store.get_value(CHILD_RUNS_KEY) - self._records = _records_adapter.validate_python(stored or {}) + 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: diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index fd6bfeb9..b5d29717 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +from datetime import timedelta from typing import TYPE_CHECKING, Any from unittest.mock import MagicMock, Mock @@ -277,3 +278,55 @@ async def test_named_call_returns_succeeded_run_without_waiting( 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', 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', 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', name='scrape-eu') + + assert apify_client_async_patcher.calls['actor']['start'] == [] From d2ff105d2d5922fd8f41b6358358a187561afb4a Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 7 Oct 2026 08:24:41 +0200 Subject: [PATCH 06/14] refactor: rename the child run `name` parameter to `run_name` --- src/apify/_actor.py | 16 ++++++------- tests/e2e/test_actor_child_runs.py | 8 +++---- tests/unit/actor/test_actor_child_runs.py | 28 +++++++++++------------ 3 files changed, 26 insertions(+), 26 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 8aa532df..bfb34344 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -946,7 +946,7 @@ async def start( timeout: timedelta | Literal['inherit'] | None = None, force_permission_level: ActorPermissionLevel | None = None, webhooks: list[Webhook] | None = None, - name: str | None = None, + run_name: str | None = None, ) -> Run: """Run an Actor on the Apify platform. @@ -973,7 +973,7 @@ 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. - name: Optional name of the child run, unique within this Actor run. A named run is recorded in the + 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 @@ -1008,11 +1008,11 @@ async def start( webhooks=to_client_representations(webhooks), ) - if name is None: + if run_name is None: return await start_run() run, _ = await self._find_or_start_child_run( - name, + run_name, actor_id=actor_id, client=client, start_run=start_run, @@ -1078,7 +1078,7 @@ async def call( webhooks: list[Webhook] | None = None, wait: timedelta | None = None, logger: logging.Logger | Literal['default'] | None = 'default', - name: str | None = None, + run_name: str | None = None, ) -> Run: """Start an Actor on the Apify Platform and wait for it to finish before returning. @@ -1108,7 +1108,7 @@ 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. - name: Optional name of the child run, unique within this Actor run. A named run is recorded in the + 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 @@ -1131,7 +1131,7 @@ async def call( actor_client = client.actor(actor_id) - if name is None: + if run_name is None: run = await actor_client.call( run_input=run_input, content_type=content_type, @@ -1147,7 +1147,7 @@ async def call( ) else: started_run, is_new = await self._find_or_start_child_run( - name, + run_name, actor_id=actor_id, client=client, start_run=partial( diff --git a/tests/e2e/test_actor_child_runs.py b/tests/e2e/test_actor_child_runs.py index c9d0d9cc..1da589fd 100644 --- a/tests/e2e/test_actor_child_runs.py +++ b/tests/e2e/test_actor_child_runs.py @@ -26,12 +26,12 @@ async def main() -> None: 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}, name='child') + 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}, name='child') + 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}' @@ -61,7 +61,7 @@ async def main() -> None: 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}, name='child') + 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() @@ -71,7 +71,7 @@ async def main() -> None: await Actor.reboot() return - run = await Actor.start(actor_id=actor_id, run_input={'is_child': True}, name='child') + 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}' diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index b5d29717..89cdc4a9 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -49,7 +49,7 @@ async def test_named_start_records_run_in_kvs(apify_client_async_patcher: ApifyC apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) async with Actor: - run = await Actor.start('some-actor', name='scrape-eu') + 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) @@ -86,7 +86,7 @@ async def test_named_start_reuses_recorded_run( async with Actor: await record_child_run('scrape-eu', 'old-run') - run = await Actor.start('some-actor', name='scrape-eu') + run = await Actor.start('some-actor', run_name='scrape-eu') assert run.id == 'old-run' assert run.status == status @@ -110,7 +110,7 @@ async def test_named_start_resurrects_recorded_run( async with Actor: await record_child_run('scrape-eu', 'old-run') - run = await Actor.start('some-actor', name='scrape-eu', memory_mbytes=2048) + run = await Actor.start('some-actor', run_name='scrape-eu', memory_mbytes=2048) assert run.id == 'old-run' assert run.status == 'RUNNING' @@ -138,7 +138,7 @@ async def test_named_start_resurrects_settling_run_after_it_finishes( async with Actor: await record_child_run('scrape-eu', 'old-run') - run = await Actor.start('some-actor', name='scrape-eu') + 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 @@ -161,7 +161,7 @@ async def test_named_start_replaces_failed_or_missing_run( async with Actor: await record_child_run('scrape-eu', 'old-run') - run = await Actor.start('some-actor', name='scrape-eu') + 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) @@ -178,7 +178,7 @@ async def test_named_start_rejects_name_recorded_for_another_actor( async with Actor: await record_child_run('scrape-eu', 'old-run', actor_id='other-actor') with pytest.raises(ValueError, match='already recorded for Actor "other-actor"'): - await Actor.start('some-actor', name='scrape-eu') + await Actor.start('some-actor', run_name='scrape-eu') assert apify_client_async_patcher.calls['actor']['start'] == [] @@ -195,7 +195,7 @@ async def slow_start(*_args: Any, **_kwargs: Any) -> Run: apify_client_async_patcher.patch('run', 'get', return_value=started) async with Actor: - runs = await asyncio.gather(*(Actor.start('some-actor', name='scrape-eu') for _ in range(3))) + 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 @@ -224,7 +224,7 @@ async def yielding_open_key_value_store(self: _ActorType, *args: Any, **kwargs: # 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', name='scrape-eu') for _ in range(3))) + 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 @@ -241,7 +241,7 @@ async def test_named_call_waits_for_reattached_run(apify_client_async_patcher: A async with Actor: await record_child_run('scrape-eu', 'old-run') - run = await Actor.call('some-actor', name='scrape-eu') + 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 @@ -257,7 +257,7 @@ async def test_named_call_streams_new_run_log_from_start(apify_client_async_patc apify_client_async_patcher.patch('run', 'get_streamed_log', replacement_method=get_streamed_log) async with Actor: - run = await Actor.call('some-actor', name='scrape-eu') + run = await Actor.call('some-actor', run_name='scrape-eu') assert run.id == 'new-run' assert run.status == 'SUCCEEDED' @@ -274,7 +274,7 @@ async def test_named_call_returns_succeeded_run_without_waiting( async with Actor: await record_child_run('scrape-eu', 'old-run') - run = await Actor.call('some-actor', name='scrape-eu') + 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'] == [] @@ -291,7 +291,7 @@ async def test_named_call_waits_for_resurrected_run(apify_client_async_patcher: async with Actor: await record_child_run('scrape-eu', 'old-run') - run = await Actor.call('some-actor', name='scrape-eu', timeout=timedelta(minutes=5)) + run = await Actor.call('some-actor', run_name='scrape-eu', timeout=timedelta(minutes=5)) assert run.id == 'old-run' assert run.status == 'SUCCEEDED' @@ -311,7 +311,7 @@ async def test_named_call_without_logger_only_waits(apify_client_async_patcher: apify_client_async_patcher.patch('run', 'get_streamed_log', replacement_method=get_streamed_log) async with Actor: - run = await Actor.call('some-actor', name='scrape-eu', logger=None) + 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 @@ -327,6 +327,6 @@ async def test_named_start_rejects_malformed_registry(apify_client_async_patcher 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', name='scrape-eu') + await Actor.start('some-actor', run_name='scrape-eu') assert apify_client_async_patcher.calls['actor']['start'] == [] From ed8160ca79c26e1d2376376883c96719470186bd Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 7 Oct 2026 08:27:03 +0200 Subject: [PATCH 07/14] feat: support `run_name` in `Actor.call_task` --- src/apify/_actor.py | 58 +++++++++--- src/apify/_child_runs.py | 49 +++++++--- tests/unit/actor/test_actor_child_runs.py | 103 +++++++++++++++++++++- 3 files changed, 181 insertions(+), 29 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index bfb34344..2eca6151 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -978,7 +978,7 @@ async def start( 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 `actor_id` exactly as passed, so reusing it with - any other value raises a `ValueError`. + any other value, or for a task, raises a `ValueError`. Returns: Info about the started Actor run @@ -1113,7 +1113,7 @@ async def call( 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 `actor_id` exactly as passed, so reusing it with - any other value raises a `ValueError`. + any other value, or for a task, raises a `ValueError`. Returns: Info about the started Actor run. @@ -1182,7 +1182,8 @@ async def _find_or_start_child_run( self, name: str, *, - actor_id: str, + actor_id: str | None = None, + task_id: str | None = None, client: ApifyClientAsync, start_run: Callable[[], Awaitable[Run]], build: str | None, @@ -1194,6 +1195,7 @@ async def _find_or_start_child_run( return await self._child_run_registry.find_or_start( name, actor_id=actor_id, + task_id=task_id, client=client, start_run=start_run, resurrect_run=lambda run_client: run_client.resurrect( @@ -1240,6 +1242,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. @@ -1265,6 +1268,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 `task_id` exactly as passed, so reusing it with + any other value, or for an Actor, raises a `ValueError`. Returns: Info about the started Actor run. @@ -1281,15 +1290,40 @@ async def call_task( raise ValueError(f'Invalid timeout {timeout!r}: expected `None`, `"inherit"`, or a `timedelta`.') task_client = client.task(task_id) - run = await task_client.call( - task_input=task_input, - build=build, - restart_on_error=restart_on_error, - memory_mbytes=memory_mbytes, - run_timeout=task_call_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, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=task_call_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, + client=client, + start_run=partial( + task_client.start, + task_input=task_input, + build=build, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=task_call_timeout, + webhooks=to_client_representations(webhooks), + ), + build=build, + max_total_charge_usd=None, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + run_timeout=task_call_timeout, + ) + run = await self._wait_for_child_run( + 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 index 05427df0..8f1ed9b9 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -3,9 +3,9 @@ import asyncio from collections import defaultdict from logging import getLogger -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Self -from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError, model_validator from pydantic.alias_generators import to_camel if TYPE_CHECKING: @@ -33,8 +33,11 @@ class ChildRunRecord(BaseModel): model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) - actor_id: str - """The Actor ID or name the child was started with, as the caller passed it.""" + actor_id: str | None = None + """The Actor ID or name the child was started with, as the caller passed it, or `None` for a task run.""" + + task_id: str | None = None + """The task ID or name the child was started with, as the caller passed it, or `None` for an Actor run.""" run_id: str """ID of the current run under this name.""" @@ -42,6 +45,16 @@ class ChildRunRecord(BaseModel): previous_run_ids: list[str] = Field(default_factory=list) """IDs of earlier runs under this name that failed and were replaced by a new run, oldest first.""" + @model_validator(mode='after') + def _check_started_from(self) -> Self: + if (self.actor_id is None) == (self.task_id is None): + raise ValueError('Exactly one of `actor_id` and `task_id` must be set.') + return self + + +def _describe_started_from(actor_id: str | None, task_id: str | None) -> str: + return f'Actor "{actor_id}"' if actor_id is not None else f'task "{task_id}"' + _records_adapter = TypeAdapter(dict[str, ChildRunRecord]) @@ -64,7 +77,8 @@ async def find_or_start( self, name: str, *, - actor_id: str, + actor_id: str | None = None, + task_id: str | None = None, client: ApifyClientAsync, start_run: Callable[[], Awaitable[Run]], resurrect_run: Callable[[RunClientAsync], Awaitable[Run]], @@ -78,8 +92,9 @@ async def find_or_start( Args: name: Name of the child run, unique within the parent run. actor_id: The Actor to start. It must match the Actor already recorded under `name`. + task_id: The task to start, in place of `actor_id`. It must match the task already recorded under `name`. client: Client used to look up and resurrect the recorded run. - start_run: Starts a new run of the Actor. + start_run: Starts a new run of the Actor or task. resurrect_run: Resurrects the recorded run, given its run client. Returns: @@ -90,12 +105,16 @@ async def find_or_start( record = records.get(name) if record is None: - return await self._start(name, actor_id=actor_id, start_run=start_run, previous_run_ids=[]), True + run = await self._start( + name, actor_id=actor_id, task_id=task_id, start_run=start_run, previous_run_ids=[] + ) + return run, True - if record.actor_id != actor_id: + if (record.actor_id, record.task_id) != (actor_id, task_id): raise ValueError( - f'Child run "{name}" is already recorded for Actor "{record.actor_id}", ' - f'it cannot be reused for Actor "{actor_id}".' + f'Child run "{name}" is already recorded for ' + f'{_describe_started_from(record.actor_id, record.task_id)}, ' + f'it cannot be reused for {_describe_started_from(actor_id, task_id)}.' ) run_client = client.run(record.run_id) @@ -106,7 +125,9 @@ async def find_or_start( if run is None or run.status == 'FAILED': previous_run_ids = [*record.previous_run_ids, record.run_id] - run = await self._start(name, actor_id=actor_id, start_run=start_run, previous_run_ids=previous_run_ids) + run = await self._start( + name, actor_id=actor_id, task_id=task_id, start_run=start_run, previous_run_ids=previous_run_ids + ) return run, True if run.status in _RESURRECTABLE_STATUSES: @@ -120,12 +141,14 @@ async def _start( self, name: str, *, - actor_id: str, + actor_id: str | None, + task_id: str | None, start_run: Callable[[], Awaitable[Run]], previous_run_ids: list[str], ) -> Run: run = await start_run() - await self._save(name, ChildRunRecord(actor_id=actor_id, run_id=run.id, previous_run_ids=previous_run_ids)) + record = ChildRunRecord(actor_id=actor_id, task_id=task_id, run_id=run.id, previous_run_ids=previous_run_ids) + await self._save(name, record) return run async def _load(self) -> dict[str, ChildRunRecord]: diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index 89cdc4a9..b59029c2 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -38,10 +38,14 @@ def make_run(run_id: str, status: str) -> Run: ) -async def record_child_run(name: str, run_id: str, *, actor_id: str = 'some-actor') -> None: +async def record_child_run( + name: str, run_id: str, *, actor_id: str | None = 'some-actor', task_id: str | None = 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() - await kvs.set_value(CHILD_RUNS_KEY, {name: {'actorId': actor_id, 'runId': run_id, 'previousRunIds': []}}) + await kvs.set_value( + CHILD_RUNS_KEY, {name: {'actorId': actor_id, 'taskId': task_id, 'runId': run_id, 'previousRunIds': []}} + ) async def test_named_start_records_run_in_kvs(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: @@ -54,7 +58,7 @@ async def test_named_start_records_run_in_kvs(apify_client_async_patcher: ApifyC stored = await kvs.get_value(CHILD_RUNS_KEY) assert run.id == 'new-run' - assert stored == {'scrape-eu': {'actorId': 'some-actor', 'runId': 'new-run', 'previousRunIds': []}} + assert stored == {'scrape-eu': {'actorId': 'some-actor', 'taskId': None, 'runId': 'new-run', 'previousRunIds': []}} async def test_unnamed_start_is_not_recorded(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: @@ -166,7 +170,9 @@ async def test_named_start_replaces_failed_or_missing_run( stored = await kvs.get_value(CHILD_RUNS_KEY) assert run.id == 'new-run' - assert stored == {'scrape-eu': {'actorId': 'some-actor', 'runId': 'new-run', 'previousRunIds': ['old-run']}} + assert stored == { + 'scrape-eu': {'actorId': 'some-actor', 'taskId': None, 'runId': 'new-run', 'previousRunIds': ['old-run']} + } async def test_named_start_rejects_name_recorded_for_another_actor( @@ -183,6 +189,20 @@ async def test_named_start_rejects_name_recorded_for_another_actor( 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 recorded for task "some-task"'): + 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') @@ -330,3 +350,78 @@ async def test_named_start_rejects_malformed_registry(apify_client_async_patcher 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': {'actorId': None, 'taskId': 'some-task', 'runId': 'new-run', 'previousRunIds': []}} + 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 recorded for Actor "some-actor", it cannot be reused for task'): + await Actor.call_task('some-task', run_name='scrape-eu') + + assert apify_client_async_patcher.calls['task']['start'] == [] + + +async def test_registry_rejects_record_without_actor_or_task( + apify_client_async_patcher: ApifyClientAsyncPatcher, +) -> None: + """A recorded run with neither an Actor nor a task ID is treated as a malformed registry.""" + 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) + 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'] == [] From 4e9d7c4efb8acaae37d45100dab5ff6a5d0a1e1e Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 8 Oct 2026 13:43:20 +0200 Subject: [PATCH 08/14] refactor: resurrect named child runs through Actor.resurrect --- src/apify/_actor.py | 18 ++++++++++++------ src/apify/_child_runs.py | 9 ++++----- 2 files changed, 16 insertions(+), 11 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index a58c020e..6f7048d2 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -1011,11 +1011,12 @@ async def start( actor_id=actor_id, client=client, start_run=start_run, + token=token, build=build, 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), + timeout=timeout, ) return run @@ -1201,11 +1202,12 @@ async def call( force_permission_level=force_permission_level, webhooks=to_client_representations(webhooks), ), + token=token, build=build, 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), + 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( @@ -1225,11 +1227,12 @@ async def _find_or_start_child_run( task_id: str | None = None, client: ApifyClientAsync, start_run: Callable[[], Awaitable[Run]], + token: str | None, build: str | None, max_total_charge_usd: Decimal | None, restart_on_error: bool | None, memory_mbytes: int | None, - run_timeout: timedelta | None, + timeout: timedelta | Literal['inherit'] | None, ) -> tuple[Run, bool]: return await self._child_run_registry.find_or_start( name, @@ -1237,12 +1240,14 @@ async def _find_or_start_child_run( task_id=task_id, client=client, start_run=start_run, - resurrect_run=lambda run_client: run_client.resurrect( + resurrect_run=partial( + self.resurrect, + token=token, build=build, max_total_charge_usd=max_total_charge_usd, restart_on_error=restart_on_error, memory_mbytes=memory_mbytes, - run_timeout=run_timeout, + timeout=timeout, ), ) @@ -1405,11 +1410,12 @@ async def call_task( run_timeout=self._resolve_run_timeout(timeout), webhooks=to_client_representations(webhooks), ), + token=token, build=build, max_total_charge_usd=None, restart_on_error=restart_on_error, memory_mbytes=memory_mbytes, - run_timeout=self._resolve_run_timeout(timeout), + timeout=timeout, ) run = await self._wait_for_child_run( client.run(started_run.id), started_run, wait=wait, logger=None, from_start=False diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py index 8f1ed9b9..e2b07bcd 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -13,7 +13,6 @@ from apify_client import ApifyClientAsync from apify_client._models import Run - from apify_client._resource_clients import RunClientAsync from apify.storages import KeyValueStore @@ -81,7 +80,7 @@ async def find_or_start( task_id: str | None = None, client: ApifyClientAsync, start_run: Callable[[], Awaitable[Run]], - resurrect_run: Callable[[RunClientAsync], 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. @@ -93,9 +92,9 @@ async def find_or_start( name: Name of the child run, unique within the parent run. actor_id: The Actor to start. It must match the Actor already recorded under `name`. task_id: The task to start, in place of `actor_id`. It must match the task already recorded under `name`. - client: Client used to look up and resurrect the recorded run. + 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 run client. + resurrect_run: Resurrects the recorded run, given its ID. Returns: The run, and whether it was newly started. @@ -132,7 +131,7 @@ async def find_or_start( if run.status in _RESURRECTABLE_STATUSES: logger.info(f'Resurrecting child run "{name}"', extra={'run_id': run.id, 'status': run.status}) - return await resurrect_run(run_client), False + return await resurrect_run(run.id), False logger.info(f'Reattaching to child run "{name}"', extra={'run_id': run.id, 'status': run.status}) return run, False From f7ac5c4c27a0b20ae7b9a545b6c2e287c300ae9c Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 8 Oct 2026 13:43:44 +0200 Subject: [PATCH 09/14] fix: forward max_items to named child runs --- src/apify/_actor.py | 7 +++++ tests/unit/actor/test_actor_child_runs.py | 31 +++++++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 6f7048d2..5b1aa638 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -1013,6 +1013,7 @@ async def start( 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, @@ -1195,6 +1196,7 @@ async def 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, @@ -1204,6 +1206,7 @@ async def call( ), 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, @@ -1229,6 +1232,7 @@ async def _find_or_start_child_run( 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, @@ -1244,6 +1248,7 @@ async def _find_or_start_child_run( 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, @@ -1405,6 +1410,7 @@ async def call_task( task_client.start, task_input=task_input, build=build, + max_items=max_items, restart_on_error=restart_on_error, memory_mbytes=memory_mbytes, run_timeout=self._resolve_run_timeout(timeout), @@ -1412,6 +1418,7 @@ async def call_task( ), token=token, build=build, + max_items=max_items, max_total_charge_usd=None, restart_on_error=restart_on_error, memory_mbytes=memory_mbytes, diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index b59029c2..812c07b9 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -425,3 +425,34 @@ async def test_registry_rejects_record_without_actor_or_task( await Actor.start('some-actor', run_name='scrape-eu') assert apify_client_async_patcher.calls['actor']['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 From 905d15c9f895bdf2475ec7f9916e54a3c860fc9f Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 8 Oct 2026 13:44:06 +0200 Subject: [PATCH 10/14] feat: support run_name in Actor.start_task --- src/apify/_actor.py | 28 ++++++++++++++++++++++- tests/unit/actor/test_actor_child_runs.py | 27 ++++++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 5b1aa638..474bd5d3 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -1291,6 +1291,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. @@ -1318,13 +1319,20 @@ 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 `task_id` exactly as passed, so reusing it with + any other value, or for an Actor, raises a `ValueError`. Returns: Info about the started Actor run. """ 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, @@ -1334,6 +1342,24 @@ 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, + client=client, + start_run=start_run, + token=token, + build=build, + max_items=max_items, + max_total_charge_usd=None, + restart_on_error=restart_on_error, + memory_mbytes=memory_mbytes, + timeout=timeout, + ) + return run + @_ensure_context async def call_task( self, diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index 812c07b9..0f75824d 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -456,3 +456,30 @@ async def test_named_start_forwards_max_items_to_resurrect( [(_, 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': {'actorId': None, 'taskId': 'some-task', 'runId': 'new-run', 'previousRunIds': []}} + 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'] == [] From d7218408c77d5fe23b3e3b46904ab29f9974f88a Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 8 Oct 2026 13:44:28 +0200 Subject: [PATCH 11/14] refactor: simplify child run registry locking --- src/apify/_child_runs.py | 15 ++++++++------- tests/unit/actor/test_actor_child_runs.py | 9 +++++++++ 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py index e2b07bcd..c3fde78f 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -1,9 +1,9 @@ from __future__ import annotations import asyncio -from collections import defaultdict from logging import getLogger from typing import TYPE_CHECKING, Self +from weakref import WeakValueDictionary from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError, model_validator from pydantic.alias_generators import to_camel @@ -68,9 +68,10 @@ class ChildRunRegistry: 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._load_lock = asyncio.Lock() - self._write_lock = asyncio.Lock() - self._name_locks: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock) + 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, @@ -99,7 +100,7 @@ async def find_or_start( Returns: The run, and whether it was newly started. """ - async with self._name_locks[name]: + async with self._name_locks.setdefault(name, asyncio.Lock()): records = await self._load() record = records.get(name) @@ -151,7 +152,7 @@ async def _start( return run async def _load(self) -> dict[str, ChildRunRecord]: - async with self._load_lock: + 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) @@ -167,7 +168,7 @@ async def _load(self) -> dict[str, ChildRunRecord]: 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._write_lock: + 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/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index 0f75824d..8769bb1b 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -483,3 +483,12 @@ async def test_named_start_task_reuses_recorded_run(apify_client_async_patcher: 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 From e42f93b0223697ec3bdd6f57a9d545891e96fb7f Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 8 Oct 2026 13:45:15 +0200 Subject: [PATCH 12/14] fix: retry a briefly missing recorded child run before replacing it --- src/apify/_child_runs.py | 23 ++++++++++++++++++++++- tests/unit/actor/test_actor_child_runs.py | 22 ++++++++++++++++++++-- 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py index c3fde78f..14952c56 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -13,6 +13,7 @@ from apify_client import ApifyClientAsync from apify_client._models import Run + from apify_client._resource_clients import RunClientAsync from apify.storages import KeyValueStore @@ -26,6 +27,11 @@ _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 ChildRunRecord(BaseModel): """A child run tracked under a name in the child run registry.""" @@ -55,6 +61,21 @@ def _describe_started_from(actor_id: str | None, task_id: str | None) -> str: return f'Actor "{actor_id}"' if actor_id is not None else f'task "{task_id}"' +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]) @@ -118,7 +139,7 @@ async def find_or_start( ) run_client = client.run(record.run_id) - run = await run_client.get() + 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() diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index 8769bb1b..d5150ed9 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -9,7 +9,7 @@ from apify_client._models import Run -from apify import Actor +from apify import Actor, _child_runs from apify._actor import _ActorType from apify._child_runs import CHILD_RUNS_KEY @@ -157,9 +157,10 @@ async def test_named_start_resurrects_settling_run_after_it_finishes( ], ) async def test_named_start_replaces_failed_or_missing_run( - apify_client_async_patcher: ApifyClientAsyncPatcher, recorded_run: Run | None + apify_client_async_patcher: ApifyClientAsyncPatcher, monkeypatch: pytest.MonkeyPatch, recorded_run: Run | None ) -> 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) @@ -492,3 +493,20 @@ async def test_name_lock_is_dropped_after_named_start(apify_client_async_patcher 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'] == [] From 0397b29204da6ac80dc400dd60abe796853cb4a2 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 8 Oct 2026 13:45:31 +0200 Subject: [PATCH 13/14] refactor: store named child runs under the __ACTOR_CHILD_RUNS key --- src/apify/_child_runs.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py index 14952c56..931da09e 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -19,7 +19,7 @@ logger = getLogger(__name__) -CHILD_RUNS_KEY = 'APIFY_CHILD_RUNS' +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'}) From c9856b0d8010a116acfc2f0c9974334c19b741b7 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 8 Oct 2026 13:47:44 +0200 Subject: [PATCH 14/14] feat: store named child runs in the JS SDK record format and bind the name to its input --- src/apify/_actor.py | 43 +++--- src/apify/_child_runs.py | 103 +++++++++----- tests/unit/actor/test_actor_child_runs.py | 159 +++++++++++++++++----- 3 files changed, 222 insertions(+), 83 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 474bd5d3..7e6d6088 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -980,8 +980,8 @@ async def start( 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 `actor_id` exactly as passed, so reusing it with - any other value, or for a task, raises a `ValueError`. + 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 @@ -1009,6 +1009,7 @@ async def start( 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, @@ -1161,8 +1162,8 @@ async def call( 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 `actor_id` exactly as passed, so reusing it with - any other value, or for a task, raises a `ValueError`. + 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. @@ -1190,6 +1191,7 @@ async def call( 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, @@ -1214,7 +1216,7 @@ async def call( ) # The earlier attempt of this call already streamed the log of a reattached or resurrected run. run = await self._wait_for_child_run( - client.run(started_run.id), started_run, wait=wait, logger=logger, from_start=is_new + run_name, client.run(started_run.id), started_run, wait=wait, logger=logger, from_start=is_new ) if run is None: @@ -1228,6 +1230,7 @@ async def _find_or_start_child_run( *, actor_id: str | None = None, task_id: str | None = None, + run_input: Any, client: ApifyClientAsync, start_run: Callable[[], Awaitable[Run]], token: str | None, @@ -1242,6 +1245,7 @@ async def _find_or_start_child_run( name, actor_id=actor_id, task_id=task_id, + run_input=run_input, client=client, start_run=start_run, resurrect_run=partial( @@ -1258,6 +1262,7 @@ async def _find_or_start_child_run( async def _wait_for_child_run( self, + name: str, run_client: RunClientAsync, run: Run, *, @@ -1269,14 +1274,18 @@ async def _wait_for_child_run( return run if not logger: - return await run_client.wait_for_finish(wait_duration=wait) + 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) - 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) - async with status_redirector, streamed_log: - return 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( @@ -1323,8 +1332,8 @@ async def start_task( 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 `task_id` exactly as passed, so reusing it with - any other value, or for an Actor, raises a `ValueError`. + 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. @@ -1348,6 +1357,7 @@ async def start_task( 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, @@ -1406,8 +1416,8 @@ async def call_task( 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 `task_id` exactly as passed, so reusing it with - any other value, or for an Actor, raises a `ValueError`. + 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. @@ -1431,6 +1441,7 @@ async def call_task( 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, @@ -1451,7 +1462,7 @@ async def call_task( timeout=timeout, ) run = await self._wait_for_child_run( - client.run(started_run.id), started_run, wait=wait, logger=None, from_start=False + run_name, client.run(started_run.id), started_run, wait=wait, logger=None, from_start=False ) if run is None: diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py index 931da09e..680c64a4 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -1,11 +1,14 @@ from __future__ import annotations import asyncio +import hashlib +import json +from datetime import datetime from logging import getLogger -from typing import TYPE_CHECKING, Self +from typing import TYPE_CHECKING, Any from weakref import WeakValueDictionary -from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError, model_validator +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError from pydantic.alias_generators import to_camel if TYPE_CHECKING: @@ -33,32 +36,40 @@ _NOT_FOUND_RETRY_INTERVAL_SECS = 0.25 -class ChildRunRecord(BaseModel): - """A child run tracked under a name in the child run registry.""" +class ChildRunSnapshot(BaseModel): + """A child run as last observed by this Actor run.""" model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel) - actor_id: str | None = None - """The Actor ID or name the child was started with, as the caller passed it, or `None` for a task run.""" + run_id: str + """ID of the run.""" - task_id: str | None = None - """The task ID or name the child was started with, as the caller passed it, or `None` for an Actor 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.""" - run_id: str - """ID of the current run under this name.""" - previous_run_ids: list[str] = Field(default_factory=list) - """IDs of earlier runs under this name that failed and were replaced by a new run, oldest first.""" +class ChildRunRecord(ChildRunSnapshot): + """The current run tracked under a name in the child run registry, with the runs it replaced.""" - @model_validator(mode='after') - def _check_started_from(self) -> Self: - if (self.actor_id is None) == (self.task_id is None): - raise ValueError('Exactly one of `actor_id` and `task_id` must be set.') - return self + 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 _describe_started_from(actor_id: str | None, task_id: str | None) -> str: - return f'Actor "{actor_id}"' if actor_id is not None else f'task "{task_id}"' + +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: @@ -100,6 +111,7 @@ async def find_or_start( *, 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]], @@ -112,8 +124,9 @@ async def find_or_start( Args: name: Name of the child run, unique within the parent run. - actor_id: The Actor to start. It must match the Actor already recorded under `name`. - task_id: The task to start, in place of `actor_id`. It must match the task already recorded under `name`. + 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. @@ -121,21 +134,20 @@ async def find_or_start( 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, actor_id=actor_id, task_id=task_id, start_run=start_run, previous_run_ids=[] - ) + run = await self._start(name, checksum=checksum, start_run=start_run, history=[]) return run, True - if (record.actor_id, record.task_id) != (actor_id, task_id): + if record.checksum != checksum: raise ValueError( - f'Child run "{name}" is already recorded for ' - f'{_describe_started_from(record.actor_id, record.task_id)}, ' - f'it cannot be reused for {_describe_started_from(actor_id, task_id)}.' + 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) @@ -145,30 +157,49 @@ async def find_or_start( run = await run_client.wait_for_finish() if run is None or run.status == 'FAILED': - previous_run_ids = [*record.previous_run_ids, record.run_id] + 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, actor_id=actor_id, task_id=task_id, start_run=start_run, previous_run_ids=previous_run_ids + 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}) - return await resurrect_run(run.id), False + run = await resurrect_run(run.id) + else: + logger.info(f'Reattaching to child run "{name}"', extra={'run_id': run.id, 'status': run.status}) - 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, *, - actor_id: str | None, - task_id: str | None, + checksum: str, start_run: Callable[[], Awaitable[Run]], - previous_run_ids: list[str], + history: list[ChildRunSnapshot], ) -> Run: run = await start_run() - record = ChildRunRecord(actor_id=actor_id, task_id=task_id, run_id=run.id, previous_run_ids=previous_run_ids) + 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 diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index d5150ed9..10f17a83 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -11,7 +11,7 @@ from apify import Actor, _child_runs from apify._actor import _ActorType -from apify._child_runs import CHILD_RUNS_KEY +from apify._child_runs import CHILD_RUNS_KEY, checksum_request if TYPE_CHECKING: from ..conftest import ApifyClientAsyncPatcher @@ -24,7 +24,7 @@ def make_run(run_id: str, status: str) -> Run: 'id': run_id, 'actId': 'actor_id', 'userId': 'user_id', - 'startedAt': '2024-08-08T12:12:44Z', + 'startedAt': STARTED_AT, 'status': status, 'meta': {'origin': 'API'}, 'buildId': 'build_id', @@ -38,14 +38,40 @@ def make_run(run_id: str, status: str) -> Run: ) +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 + 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() - await kvs.set_value( - CHILD_RUNS_KEY, {name: {'actorId': actor_id, 'taskId': task_id, 'runId': run_id, 'previousRunIds': []}} - ) + 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: @@ -58,7 +84,7 @@ async def test_named_start_records_run_in_kvs(apify_client_async_patcher: ApifyC stored = await kvs.get_value(CHILD_RUNS_KEY) assert run.id == 'new-run' - assert stored == {'scrape-eu': {'actorId': 'some-actor', 'taskId': None, 'runId': 'new-run', 'previousRunIds': []}} + assert stored == {'scrape-eu': stored_record('new-run', 'READY')} async def test_unnamed_start_is_not_recorded(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: @@ -150,14 +176,17 @@ async def test_named_start_resurrects_settling_run_after_it_finishes( @pytest.mark.parametrize( - 'recorded_run', + ('recorded_run', 'replaced_status'), [ - pytest.param(make_run('old-run', 'FAILED'), id='failed'), - pytest.param(None, id='not found'), + 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 + 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) @@ -172,7 +201,9 @@ async def test_named_start_replaces_failed_or_missing_run( assert run.id == 'new-run' assert stored == { - 'scrape-eu': {'actorId': 'some-actor', 'taskId': None, 'runId': 'new-run', 'previousRunIds': ['old-run']} + 'scrape-eu': stored_record( + 'new-run', 'READY', history=[{'runId': 'old-run', 'status': replaced_status, 'startedAt': STARTED_AT}] + ) } @@ -184,7 +215,7 @@ async def test_named_start_rejects_name_recorded_for_another_actor( async with Actor: await record_child_run('scrape-eu', 'old-run', actor_id='other-actor') - with pytest.raises(ValueError, match='already recorded for Actor "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'] == [] @@ -198,7 +229,7 @@ async def test_named_start_rejects_name_recorded_for_task( async with Actor: await record_child_run('scrape-eu', 'old-run', actor_id=None, task_id='some-task') - with pytest.raises(ValueError, match='already recorded for task "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'] == [] @@ -364,7 +395,7 @@ async def test_named_call_task_records_run_in_kvs(apify_client_async_patcher: Ap stored = await kvs.get_value(CHILD_RUNS_KEY) assert run.status == 'SUCCEEDED' - assert stored == {'scrape-eu': {'actorId': None, 'taskId': 'some-task', 'runId': 'new-run', 'previousRunIds': []}} + assert stored == {'scrape-eu': stored_record('new-run', 'SUCCEEDED', actor_id=None, task_id='some-task')} assert apify_client_async_patcher.calls['task']['call'] == [] @@ -408,26 +439,12 @@ async def test_named_call_task_rejects_name_recorded_for_actor( async with Actor: await record_child_run('scrape-eu', 'old-run') - with pytest.raises(ValueError, match='already recorded for Actor "some-actor", it cannot be reused for task'): + 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_registry_rejects_record_without_actor_or_task( - apify_client_async_patcher: ApifyClientAsyncPatcher, -) -> None: - """A recorded run with neither an Actor nor a task ID is treated as a malformed registry.""" - 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) - 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_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')) @@ -469,7 +486,7 @@ async def test_named_start_task_records_run_in_kvs(apify_client_async_patcher: A stored = await kvs.get_value(CHILD_RUNS_KEY) assert run.id == 'new-run' - assert stored == {'scrape-eu': {'actorId': None, 'taskId': 'some-task', 'runId': 'new-run', 'previousRunIds': []}} + 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'] == [] @@ -510,3 +527,83 @@ async def test_named_start_retries_recorded_run_not_found_yet( 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')}