From 9a6d442311722727870596c3ede08d828201315a Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 30 Sep 2026 10:10:19 +0200 Subject: [PATCH 1/9] feat: abort named child runs when the parent is gracefully aborted --- src/apify/_actor.py | 22 +++ src/apify/_child_runs.py | 74 +++++++++- tests/unit/actor/test_actor_child_runs.py | 161 +++++++++++++++++++++- 3 files changed, 249 insertions(+), 8 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index a00a98bf..28b2f5a0 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -206,6 +206,7 @@ async def __aenter__(self) -> Self: # Initialize the event manager and register it in the service locator. await self.event_manager.__aenter__() + self.event_manager.on(event=Event.ABORTING, listener=self._abort_child_runs) self.log.debug('Event manager initialized') # Initialize the charging manager. @@ -947,6 +948,7 @@ async def start( force_permission_level: ActorPermissionLevel | None = None, webhooks: list[Webhook] | None = None, name: str | None = None, + abort_with_parent: bool = False, ) -> Run: """Run an Actor on the Apify platform. @@ -979,10 +981,16 @@ async def start( 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`. + abort_with_parent: If true, the child run is gracefully aborted when this Actor run is gracefully + aborted. It requires `name`, and the value is recorded under it, replacing the one from an earlier + call. A hard abort, a timeout or a crash of this Actor run leaves the child running. Returns: Info about the started Actor run """ + if abort_with_parent and name is None: + raise ValueError('`abort_with_parent` requires `name`, since only named child runs are tracked.') + client = self.new_client(token=token) if token else self.apify_client if timeout == 'inherit': @@ -1021,6 +1029,7 @@ async def start( restart_on_error=restart_on_error, memory_mbytes=memory_mbytes, run_timeout=actor_start_timeout, + abort_with_parent=abort_with_parent, ) return run @@ -1079,6 +1088,7 @@ async def call( wait: timedelta | None = None, logger: logging.Logger | Literal['default'] | None = 'default', name: str | None = None, + abort_with_parent: bool = False, ) -> Run: """Start an Actor on the Apify Platform and wait for it to finish before returning. @@ -1114,10 +1124,16 @@ async def call( 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`. + abort_with_parent: If true, the child run is gracefully aborted when this Actor run is gracefully + aborted. It requires `name`, and the value is recorded under it, replacing the one from an earlier + call. A hard abort, a timeout or a crash of this Actor run leaves the child running. Returns: Info about the started Actor run. """ + if abort_with_parent and name is None: + raise ValueError('`abort_with_parent` requires `name`, since only named child runs are tracked.') + client = self.new_client(token=token) if token else self.apify_client if timeout == 'inherit': @@ -1167,6 +1183,7 @@ async def call( restart_on_error=restart_on_error, memory_mbytes=memory_mbytes, run_timeout=actor_call_timeout, + abort_with_parent=abort_with_parent, ) # The earlier attempt of this call already streamed the log of a reattached or resurrected run. run = await self._wait_for_child_run( @@ -1190,6 +1207,7 @@ async def _find_or_start_child_run( restart_on_error: bool | None, memory_mbytes: int | None, run_timeout: timedelta | None, + abort_with_parent: bool, ) -> tuple[Run, bool]: return await self._child_run_registry.find_or_start( name, @@ -1203,8 +1221,12 @@ async def _find_or_start_child_run( memory_mbytes=memory_mbytes, run_timeout=run_timeout, ), + abort_with_parent=abort_with_parent, ) + async def _abort_child_runs(self) -> None: + await self._child_run_registry.abort_runs_with_parent(self.apify_client) + async def _wait_for_child_run( self, run_client: RunClientAsync, diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py index 85befe3f..8b0d5558 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -30,6 +30,8 @@ _RESURRECTABLE_STATUSES = frozenset({'ABORTED', 'TIMED-OUT'}) +_ABORTABLE_STATUSES = frozenset({'READY', 'RUNNING'}) + class ChildRunRecord(BaseModel): """A child run tracked under a name in the child run registry.""" @@ -45,6 +47,9 @@ class ChildRunRecord(BaseModel): previous_run_ids: list[str] = Field(default_factory=list) """IDs of earlier runs under this name that failed or went missing and were replaced by a new run, oldest first.""" + abort_with_parent: bool = False + """Whether the current run is aborted when this Actor run is gracefully aborted.""" + @docs_group('Actor') @dataclass(frozen=True) @@ -63,6 +68,9 @@ class ChildRunInfo: previous_run_ids: list[str] """IDs of earlier runs under this name that failed or went missing and were replaced by a new run, oldest first.""" + abort_with_parent: bool + """Whether the current run is aborted when this Actor run is gracefully aborted.""" + _records_adapter = TypeAdapter(dict[str, ChildRunRecord]) @@ -80,6 +88,8 @@ def __init__(self, open_key_value_store: Callable[[], Awaitable[KeyValueStore]]) self._load_lock = asyncio.Lock() self._write_lock = asyncio.Lock() self._name_locks: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock) + self._clients: dict[str, ApifyClientAsync] = {} + """Client each name was last started or reattached with in this process, used to abort its run.""" async def find_or_start( self, @@ -89,6 +99,7 @@ async def find_or_start( client: ApifyClientAsync, start_run: Callable[[], Awaitable[Run]], resurrect_run: Callable[[RunClientAsync], Awaitable[Run]], + abort_with_parent: bool = False, ) -> tuple[Run, bool]: """Return the run recorded under `name`, or start one when there is none to reuse. @@ -102,6 +113,8 @@ async def find_or_start( 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. + abort_with_parent: Whether to abort the run when this Actor run is gracefully aborted. It replaces + the value recorded under `name`. Returns: The run, and whether it was newly started. @@ -109,9 +122,17 @@ async def find_or_start( async with self._name_locks[name]: records = await self._load() record = records.get(name) + self._clients[name] = client 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, + start_run=start_run, + previous_run_ids=[], + abort_with_parent=abort_with_parent, + ) + return run, True if record.actor_id != actor_id: raise ValueError( @@ -126,10 +147,18 @@ 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] - 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, + start_run=start_run, + previous_run_ids=[*record.previous_run_ids, record.run_id], + abort_with_parent=abort_with_parent, + ) return run, True + if record.abort_with_parent != abort_with_parent: + await self._save(name, record.model_copy(update={'abort_with_parent': abort_with_parent})) + 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 @@ -152,10 +181,40 @@ async def list_runs(self, client: ApifyClientAsync) -> dict[str, ChildRunInfo]: run_id=record.run_id, run=run, previous_run_ids=list(record.previous_run_ids), + abort_with_parent=record.abort_with_parent, ) for (name, record), run in zip(records.items(), runs, strict=True) } + async def abort_runs_with_parent(self, client: ApifyClientAsync) -> None: + """Gracefully abort every recorded run marked `abort_with_parent` that is still `READY` or `RUNNING`. + + A failure to abort one run is logged and does not stop the others. + + Args: + client: Client used for a name not started or reattached in this process, e.g. after a migration. + """ + records = dict(await self._load()) + await asyncio.gather( + *( + self._abort(name, record, self._clients.get(name, client)) + for name, record in records.items() + if record.abort_with_parent + ) + ) + + async def _abort(self, name: str, record: ChildRunRecord, client: ApifyClientAsync) -> None: + run_client = client.run(record.run_id) + try: + run = await run_client.get() + if run is None or run.status not in _ABORTABLE_STATUSES: + return + await run_client.abort(gracefully=True) + except Exception: + logger.exception(f'Failed to abort child run "{name}"', extra={'run_id': record.run_id}) + else: + logger.info(f'Aborted child run "{name}" with the parent', extra={'run_id': record.run_id}) + async def _start( self, name: str, @@ -163,9 +222,16 @@ async def _start( actor_id: str, start_run: Callable[[], Awaitable[Run]], previous_run_ids: list[str], + abort_with_parent: bool, ) -> 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, + run_id=run.id, + previous_run_ids=previous_run_ids, + abort_with_parent=abort_with_parent, + ) + 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 cb4a5fca..3ef00c18 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -3,15 +3,16 @@ import asyncio from datetime import timedelta from typing import TYPE_CHECKING, Any -from unittest.mock import MagicMock, Mock +from unittest.mock import AsyncMock, MagicMock, Mock import pytest from apify_client._models import Run +from crawlee.events import Event, EventAbortingData from apify import Actor from apify._actor import _ActorType -from apify._child_runs import CHILD_RUNS_KEY +from apify._child_runs import CHILD_RUNS_KEY, ChildRunRegistry if TYPE_CHECKING: from ..conftest import ApifyClientAsyncPatcher @@ -54,7 +55,9 @@ 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', 'runId': 'new-run', 'previousRunIds': [], 'abortWithParent': False} + } async def test_unnamed_start_is_not_recorded(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: @@ -166,7 +169,14 @@ 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', + 'runId': 'new-run', + 'previousRunIds': ['old-run'], + 'abortWithParent': False, + } + } async def test_named_start_rejects_name_recorded_for_another_actor( @@ -390,3 +400,146 @@ async def test_child_runs_includes_run_started_in_this_attempt( assert child_runs['scrape-eu'].run_id == 'new-run' assert child_runs['scrape-eu'].run is not None assert child_runs['scrape-eu'].run.status == 'RUNNING' + + +@pytest.mark.parametrize( + 'method', + [ + pytest.param('start', id='start'), + pytest.param('call', id='call'), + ], +) +async def test_abort_with_parent_requires_name( + apify_client_async_patcher: ApifyClientAsyncPatcher, method: str +) -> None: + """`abort_with_parent` without a `name` raises before any run is started.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + apify_client_async_patcher.patch('actor', 'call', return_value=make_run('new-run', 'SUCCEEDED')) + + async with Actor: + with pytest.raises(ValueError, match='requires `name`'): + await getattr(Actor, method)('some-actor', abort_with_parent=True) + + assert apify_client_async_patcher.calls['actor']['start'] == [] + assert apify_client_async_patcher.calls['actor']['call'] == [] + + +async def test_named_start_records_abort_with_parent(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """A named start with `abort_with_parent` records the flag with the run.""" + apify_client_async_patcher.patch('actor', 'start', return_value=make_run('new-run', 'READY')) + + async with Actor: + await Actor.start('some-actor', name='scrape-eu', abort_with_parent=True) + kvs = await Actor.open_key_value_store() + stored = await kvs.get_value(CHILD_RUNS_KEY) + + assert stored['scrape-eu']['abortWithParent'] is True + + +async def test_reattach_replaces_recorded_abort_with_parent( + apify_client_async_patcher: ApifyClientAsyncPatcher, +) -> None: + """Reattaching under a name records the `abort_with_parent` value of the latest call.""" + 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') + await Actor.start('some-actor', name='scrape-eu', abort_with_parent=True) + child_runs = await Actor.child_runs() + + assert child_runs['scrape-eu'].run_id == 'old-run' + assert child_runs['scrape-eu'].abort_with_parent is True + + +async def test_aborting_event_aborts_marked_active_child_runs( + apify_client_async_patcher: ApifyClientAsyncPatcher, +) -> None: + """On `ABORTING`, only child runs marked `abort_with_parent` that are still active are gracefully aborted.""" + runs = { + 'running-run': make_run('running-run', 'RUNNING'), + 'ready-run': make_run('ready-run', 'READY'), + 'finished-run': make_run('finished-run', 'SUCCEEDED'), + 'unmarked-run': make_run('unmarked-run', 'RUNNING'), + } + + async def get_run(run_client: Any, *_args: Any, **_kwargs: Any) -> Run | None: + return runs[run_client.resource_id] + + apify_client_async_patcher.patch('run', 'get', replacement_method=get_run) + apify_client_async_patcher.patch('run', 'abort', return_value=None) + + async with Actor: + kvs = await Actor.open_key_value_store() + await kvs.set_value( + CHILD_RUNS_KEY, + { + name: {'actorId': 'some-actor', 'runId': run_id, 'previousRunIds': [], 'abortWithParent': marked} + for name, run_id, marked in [ + ('running', 'running-run', True), + ('ready', 'ready-run', True), + ('finished', 'finished-run', True), + ('unmarked', 'unmarked-run', False), + ] + }, + ) + Actor.event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await Actor.event_manager.wait_for_all_listeners_to_complete() + + aborts = apify_client_async_patcher.calls['run']['abort'] + assert sorted(args[0].resource_id for args, _ in aborts) == ['ready-run', 'running-run'] + assert all(kwargs == {'gracefully': True} for _, kwargs in aborts) + + +async def test_failed_child_run_abort_does_not_stop_others( + apify_client_async_patcher: ApifyClientAsyncPatcher, caplog: pytest.LogCaptureFixture +) -> None: + """A child run that fails to abort is logged, and the other marked child runs are still aborted.""" + aborted: list[str] = [] + + async def abort_run(run_client: Any, *_args: Any, **_kwargs: Any) -> None: + if run_client.resource_id == 'broken-run': + raise RuntimeError('abort failed') + aborted.append(run_client.resource_id) + + apify_client_async_patcher.patch( + 'run', 'get', replacement_method=lambda run_client: make_run(run_client.resource_id, 'RUNNING') + ) + apify_client_async_patcher.patch('run', 'abort', replacement_method=abort_run) + + async with Actor: + kvs = await Actor.open_key_value_store() + await kvs.set_value( + CHILD_RUNS_KEY, + { + name: {'actorId': 'some-actor', 'runId': f'{name}-run', 'previousRunIds': [], 'abortWithParent': True} + for name in ['broken', 'healthy'] + }, + ) + Actor.event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await Actor.event_manager.wait_for_all_listeners_to_complete() + + assert aborted == ['healthy-run'] + assert 'Failed to abort child run "broken"' in caplog.text + + +async def test_child_run_is_aborted_with_the_client_it_was_started_with() -> None: + """A child run started with its own client is aborted with that client, not the default one.""" + default_client = Mock() + child_client = Mock() + child_client.run.return_value.get = AsyncMock(return_value=make_run('new-run', 'RUNNING')) + child_client.run.return_value.abort = AsyncMock() + + async with Actor: + registry = ChildRunRegistry(Actor.open_key_value_store) + await registry.find_or_start( + 'scrape-eu', + actor_id='some-actor', + client=child_client, + start_run=AsyncMock(return_value=make_run('new-run', 'READY')), + resurrect_run=AsyncMock(), + abort_with_parent=True, + ) + await registry.abort_runs_with_parent(default_client) + + child_client.run.return_value.abort.assert_awaited_once_with(gracefully=True) + default_client.run.assert_not_called() From 29bf9434df5551e1ee78d058c08cc56e6ea792a7 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 30 Sep 2026 10:10:19 +0200 Subject: [PATCH 2/9] test: check a child run is aborted with its parent in E2E --- tests/e2e/test_actor_child_runs.py | 43 ++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/tests/e2e/test_actor_child_runs.py b/tests/e2e/test_actor_child_runs.py index 0eee0ecd..0881ae7b 100644 --- a/tests/e2e/test_actor_child_runs.py +++ b/tests/e2e/test_actor_child_runs.py @@ -1,11 +1,15 @@ from __future__ import annotations import asyncio +from datetime import timedelta from typing import TYPE_CHECKING from apify import Actor +from apify._child_runs import CHILD_RUNS_KEY if TYPE_CHECKING: + from apify_client import ApifyClientAsync + from .conftest import MakeActorFunction, RunActorFunction @@ -91,3 +95,42 @@ async def main() -> None: assert run_result.status == 'SUCCEEDED' # The parent run and the one child run it resurrected. assert (await actor.runs().list()).total == 2 + + +async def test_named_child_run_is_aborted_with_parent( + make_actor: MakeActorFunction, + apify_client_async: ApifyClientAsync, +) -> None: + """A named child run started with `abort_with_parent` is aborted when the parent is gracefully aborted.""" + + 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 '' + await Actor.start(actor_id=actor_id, run_input={'is_child': True}, name='child', abort_with_parent=True) + await asyncio.sleep(300) + + actor = await make_actor(label='child-run-abort-with-parent', main_func=main) + parent_run = await actor.start() + parent_kvs = apify_client_async.key_value_store(parent_run.default_key_value_store_id) + + # Wait for the parent to record the child run. + for _ in range(60): + if record := await parent_kvs.get_record(CHILD_RUNS_KEY): + break + await asyncio.sleep(2) + else: + raise AssertionError('The parent run did not record the child run in time.') + + child_run_id = record['value']['child']['runId'] + parent_run_client = apify_client_async.run(parent_run.id) + await parent_run_client.abort(gracefully=True) + await parent_run_client.wait_for_finish(wait_duration=timedelta(seconds=120)) + + child_run = await apify_client_async.run(child_run_id).wait_for_finish(wait_duration=timedelta(seconds=120)) + assert child_run is not None + assert child_run.status == 'ABORTED' From 19976af0616aa4e20cc9256681fd6a4eb6b9aab1 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 30 Sep 2026 10:10:20 +0200 Subject: [PATCH 3/9] docs: describe aborting child runs with the parent --- .../06_interacting_with_other_actors.mdx | 17 ++++++++++++++++ .../code/06_interacting_abort_with_parent.py | 20 +++++++++++++++++++ 2 files changed, 37 insertions(+) create mode 100644 docs/02_concepts/code/06_interacting_abort_with_parent.py diff --git a/docs/02_concepts/06_interacting_with_other_actors.mdx b/docs/02_concepts/06_interacting_with_other_actors.mdx index c426846c..6ddbc9ed 100644 --- a/docs/02_concepts/06_interacting_with_other_actors.mdx +++ b/docs/02_concepts/06_interacting_with_other_actors.mdx @@ -10,6 +10,7 @@ import InteractingStartExample from '!!raw-loader!roa-loader!./code/06_interacti import InteractingCallExample from '!!raw-loader!roa-loader!./code/06_interacting_call.py'; import InteractingNamedCallExample from '!!raw-loader!roa-loader!./code/06_interacting_named_call.py'; import InteractingChildRunsExample from '!!raw-loader!roa-loader!./code/06_interacting_child_runs.py'; +import InteractingAbortWithParentExample from '!!raw-loader!roa-loader!./code/06_interacting_abort_with_parent.py'; import InteractingCallTaskExample from '!!raw-loader!roa-loader!./code/06_interacting_call_task.py'; import InteractingMetamorphExample from '!!raw-loader!roa-loader!./code/06_interacting_metamorph.py'; import InteractingAbortExample from '!!raw-loader!roa-loader!./code/06_interacting_abort.py'; @@ -67,6 +68,22 @@ The `Actor.child_runs` method ret {InteractingChildRunsExample} +### Aborting child runs with the parent + +When your Actor run is aborted, its child runs keep running, and you pay for them until they finish on their own. To abort a named child run together with your Actor run, pass `abort_with_parent=True`. When your Actor run receives the `ABORTING` event of a graceful abort, the SDK gracefully aborts every child run marked this way that's still `READY` or `RUNNING`. The flag is recorded with the name, so it also covers child runs started before a migration or resurrection. + + + {InteractingAbortWithParentExample} + + +Note that: + +- The option is off by default, since aborting a child run throws away the work it hasn't finished. +- It requires `name`. Without one, `Actor.start` and `Actor.call` raise a `ValueError`. +- Each call under a name records its own value, so the latest call decides whether the run is aborted. +- A child run aborted this way ends as `ABORTED`. If your Actor run is resurrected later, the same named call resurrects the child run too. +- Only a graceful abort gives the SDK time to act. A hard abort, a timeout, or a crash of your Actor run leaves the child runs running. + ## Actor call task The `Actor.call_task` method starts an [Actor task](https://docs.apify.com/platform/actors/tasks) on the Apify platform, and waits for the started Actor run to finish. diff --git a/docs/02_concepts/code/06_interacting_abort_with_parent.py b/docs/02_concepts/code/06_interacting_abort_with_parent.py new file mode 100644 index 00000000..44d39c9e --- /dev/null +++ b/docs/02_concepts/code/06_interacting_abort_with_parent.py @@ -0,0 +1,20 @@ +import asyncio + +from apify import Actor + + +async def main() -> None: + async with Actor: + # Start the child run, and abort it if this Actor run is gracefully aborted. + actor_run = await Actor.start( + actor_id='apify/screenshot-url', + run_input={'urls': [{'url': 'https://www.apify.com/'}]}, + name='screenshot', + abort_with_parent=True, + ) + + Actor.log.info(f'Started child run {actor_run.id}') + + +if __name__ == '__main__': + asyncio.run(main()) From fbd4f3198c9422274962f2f3e9fd42aec8353b2a Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 30 Sep 2026 10:45:19 +0200 Subject: [PATCH 4/9] fix: keep the abort client of a child run when a start under its name is rejected --- src/apify/_child_runs.py | 15 ++++++------ tests/unit/actor/test_actor_child_runs.py | 28 +++++++++++++++++++++++ 2 files changed, 36 insertions(+), 7 deletions(-) diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py index 8b0d5558..472a904b 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -110,7 +110,7 @@ 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`. - client: Client used to look up and resurrect the recorded run. + client: Client used to look up, resurrect and abort the recorded run. start_run: Starts a new run of the Actor. resurrect_run: Resurrects the recorded run, given its run client. abort_with_parent: Whether to abort the run when this Actor run is gracefully aborted. It replaces @@ -122,6 +122,13 @@ async def find_or_start( async with self._name_locks[name]: records = await self._load() record = records.get(name) + + if record is not None and 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}".' + ) + self._clients[name] = client if record is None: @@ -134,12 +141,6 @@ async def find_or_start( ) return run, 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() diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index 3ef00c18..939097e9 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -543,3 +543,31 @@ async def test_child_run_is_aborted_with_the_client_it_was_started_with() -> Non child_client.run.return_value.abort.assert_awaited_once_with(gracefully=True) default_client.run.assert_not_called() + + +async def test_rejected_named_start_keeps_the_client_used_to_abort() -> None: + """A named start rejected for another Actor does not change the client its recorded run is aborted with.""" + default_client = Mock() + default_client.run.return_value.get = AsyncMock(return_value=make_run('old-run', 'RUNNING')) + default_client.run.return_value.abort = AsyncMock() + other_client = Mock() + + async with Actor: + kvs = await Actor.open_key_value_store() + await kvs.set_value( + CHILD_RUNS_KEY, + {'scrape-eu': {'actorId': 'some-actor', 'runId': 'old-run', 'previousRunIds': [], 'abortWithParent': True}}, + ) + registry = ChildRunRegistry(Actor.open_key_value_store) + with pytest.raises(ValueError, match='cannot be reused'): + await registry.find_or_start( + 'scrape-eu', + actor_id='other-actor', + client=other_client, + start_run=AsyncMock(), + resurrect_run=AsyncMock(), + ) + await registry.abort_runs_with_parent(default_client) + + default_client.run.return_value.abort.assert_awaited_once_with(gracefully=True) + other_client.run.assert_not_called() From 37e41b484fc81b6fe4f9bc0e0fb47a711e053fd5 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 30 Sep 2026 10:45:25 +0200 Subject: [PATCH 5/9] fix: abort a named child run whose start is still in flight --- src/apify/_child_runs.py | 35 +++++++++++------------ tests/unit/actor/test_actor_child_runs.py | 35 +++++++++++++++++++++++ 2 files changed, 52 insertions(+), 18 deletions(-) diff --git a/src/apify/_child_runs.py b/src/apify/_child_runs.py index 472a904b..76f53bed 100644 --- a/src/apify/_child_runs.py +++ b/src/apify/_child_runs.py @@ -195,26 +195,25 @@ async def abort_runs_with_parent(self, client: ApifyClientAsync) -> None: Args: client: Client used for a name not started or reattached in this process, e.g. after a migration. """ - records = dict(await self._load()) - await asyncio.gather( - *( - self._abort(name, record, self._clients.get(name, client)) - for name, record in records.items() - if record.abort_with_parent - ) - ) + records = await self._load() + # Names with a start in flight are not recorded yet, so their locks are awaited too. + await asyncio.gather(*(self._abort(name, client) for name in {*records, *self._name_locks})) - async def _abort(self, name: str, record: ChildRunRecord, client: ApifyClientAsync) -> None: - run_client = client.run(record.run_id) - try: - run = await run_client.get() - if run is None or run.status not in _ABORTABLE_STATUSES: + async def _abort(self, name: str, default_client: ApifyClientAsync) -> None: + async with self._name_locks[name]: + record = (await self._load()).get(name) + if record is None or not record.abort_with_parent: return - await run_client.abort(gracefully=True) - except Exception: - logger.exception(f'Failed to abort child run "{name}"', extra={'run_id': record.run_id}) - else: - logger.info(f'Aborted child run "{name}" with the parent', extra={'run_id': record.run_id}) + run_client = self._clients.get(name, default_client).run(record.run_id) + try: + run = await run_client.get() + if run is None or run.status not in _ABORTABLE_STATUSES: + return + await run_client.abort(gracefully=True) + except Exception: + logger.exception(f'Failed to abort child run "{name}"', extra={'run_id': record.run_id}) + else: + logger.info(f'Aborted child run "{name}" with the parent', extra={'run_id': record.run_id}) async def _start( self, diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index 939097e9..6a1c11b3 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -571,3 +571,38 @@ async def test_rejected_named_start_keeps_the_client_used_to_abort() -> None: default_client.run.return_value.abort.assert_awaited_once_with(gracefully=True) other_client.run.assert_not_called() + + +async def test_aborting_waits_for_a_named_start_in_flight() -> None: + """A named start in flight when the parent is aborted has its run aborted once the run is recorded.""" + client = Mock() + client.run.return_value.get = AsyncMock(return_value=make_run('new-run', 'RUNNING')) + client.run.return_value.abort = AsyncMock() + started = asyncio.Event() + release = asyncio.Event() + + async def start_run() -> Run: + started.set() + await release.wait() + return make_run('new-run', 'READY') + + async with Actor: + registry = ChildRunRegistry(Actor.open_key_value_store) + start_task = asyncio.create_task( + registry.find_or_start( + 'scrape-eu', + actor_id='some-actor', + client=client, + start_run=start_run, + resurrect_run=AsyncMock(), + abort_with_parent=True, + ) + ) + await started.wait() + abort_task = asyncio.create_task(registry.abort_runs_with_parent(client)) + await asyncio.sleep(0) + assert not abort_task.done() + release.set() + await asyncio.gather(start_task, abort_task) + + client.run.return_value.abort.assert_awaited_once_with(gracefully=True) From 2f891cd8d7aa15109f0d5e52588c8f4c3202ee10 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 30 Sep 2026 10:45:32 +0200 Subject: [PATCH 6/9] fix: remove the ABORTING listener when the Actor exits --- src/apify/_actor.py | 2 ++ tests/unit/actor/test_actor_child_runs.py | 15 +++++++++++++++ 2 files changed, 17 insertions(+) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index 28b2f5a0..e8050596 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -214,6 +214,7 @@ async def __aenter__(self) -> Self: await self._charging_manager_implementation.__aenter__() except BaseException: # Exit the already-entered event manager so its recurring tasks do not leak. + self.event_manager.off(event=Event.ABORTING, listener=self._abort_child_runs) await self.event_manager.__aexit__(None, None, None) raise self.log.debug('Charging manager initialized') @@ -305,6 +306,7 @@ async def finalize() -> None: except TimeoutError: self.log.exception('Actor cleanup timed out') finally: + self.event_manager.off(event=Event.ABORTING, listener=self._abort_child_runs) self._active = False if reraise_control_flow: diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index 6a1c11b3..6cfc5127 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -606,3 +606,18 @@ async def start_run() -> Run: await asyncio.gather(start_task, abort_task) client.run.return_value.abort.assert_awaited_once_with(gracefully=True) + + +async def test_exit_removes_the_aborting_listener(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: + """After the Actor exits, an `ABORTING` event on a still-active event manager aborts no child run.""" + 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('new-run', 'RUNNING')) + apify_client_async_patcher.patch('run', 'abort', return_value=None) + + async with Actor.event_manager: + async with Actor: + await Actor.start('some-actor', name='scrape-eu', abort_with_parent=True) + Actor.event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await Actor.event_manager.wait_for_all_listeners_to_complete() + + assert apify_client_async_patcher.calls['run']['abort'] == [] From 8ca7ad8546228b6bd05d5a6ca76c767fe2f32415 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 30 Sep 2026 10:45:38 +0200 Subject: [PATCH 7/9] test: check child run aborts through the client patcher --- tests/unit/actor/test_actor_child_runs.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index 6cfc5127..99e8b070 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -494,12 +494,10 @@ async def test_failed_child_run_abort_does_not_stop_others( apify_client_async_patcher: ApifyClientAsyncPatcher, caplog: pytest.LogCaptureFixture ) -> None: """A child run that fails to abort is logged, and the other marked child runs are still aborted.""" - aborted: list[str] = [] async def abort_run(run_client: Any, *_args: Any, **_kwargs: Any) -> None: if run_client.resource_id == 'broken-run': raise RuntimeError('abort failed') - aborted.append(run_client.resource_id) apify_client_async_patcher.patch( 'run', 'get', replacement_method=lambda run_client: make_run(run_client.resource_id, 'RUNNING') @@ -518,8 +516,10 @@ async def abort_run(run_client: Any, *_args: Any, **_kwargs: Any) -> None: Actor.event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) await Actor.event_manager.wait_for_all_listeners_to_complete() - assert aborted == ['healthy-run'] + aborts = apify_client_async_patcher.calls['run']['abort'] + assert sorted(args[0].resource_id for args, _ in aborts) == ['broken-run', 'healthy-run'] assert 'Failed to abort child run "broken"' in caplog.text + assert 'Aborted child run "healthy" with the parent' in caplog.text async def test_child_run_is_aborted_with_the_client_it_was_started_with() -> None: From 8346dad00dbae4d2268d09899a5653aff3792ff5 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 30 Sep 2026 10:57:04 +0200 Subject: [PATCH 8/9] fix: keep aborting child runs when all ABORTING listeners are removed --- src/apify/_actor.py | 12 +++-- src/apify/events/_apify_event_manager.py | 27 +++++++++- tests/unit/actor/test_actor_child_runs.py | 53 +++++++++++++++---- tests/unit/events/test_apify_event_manager.py | 30 ++++++++++- 4 files changed, 105 insertions(+), 17 deletions(-) diff --git a/src/apify/_actor.py b/src/apify/_actor.py index e8050596..cbbf4bd2 100644 --- a/src/apify/_actor.py +++ b/src/apify/_actor.py @@ -206,7 +206,9 @@ async def __aenter__(self) -> Self: # Initialize the event manager and register it in the service locator. await self.event_manager.__aenter__() - self.event_manager.on(event=Event.ABORTING, listener=self._abort_child_runs) + # Only the platform emits `ABORTING`, and it does so through `ApifyEventManager`. + if isinstance(self.event_manager, ApifyEventManager): + self.event_manager._on_internal(event=Event.ABORTING, listener=self._abort_child_runs) # noqa: SLF001 self.log.debug('Event manager initialized') # Initialize the charging manager. @@ -214,7 +216,7 @@ async def __aenter__(self) -> Self: await self._charging_manager_implementation.__aenter__() except BaseException: # Exit the already-entered event manager so its recurring tasks do not leak. - self.event_manager.off(event=Event.ABORTING, listener=self._abort_child_runs) + self._remove_internal_listeners() await self.event_manager.__aexit__(None, None, None) raise self.log.debug('Charging manager initialized') @@ -306,7 +308,7 @@ async def finalize() -> None: except TimeoutError: self.log.exception('Actor cleanup timed out') finally: - self.event_manager.off(event=Event.ABORTING, listener=self._abort_child_runs) + self._remove_internal_listeners() self._active = False if reraise_control_flow: @@ -1226,6 +1228,10 @@ async def _find_or_start_child_run( abort_with_parent=abort_with_parent, ) + def _remove_internal_listeners(self) -> None: + if isinstance(self.event_manager, ApifyEventManager): + self.event_manager._off_internal(event=Event.ABORTING, listener=self._abort_child_runs) # noqa: SLF001 + async def _abort_child_runs(self) -> None: await self._child_run_registry.abort_runs_with_parent(self.apify_client) diff --git a/src/apify/events/_apify_event_manager.py b/src/apify/events/_apify_event_manager.py index 4cb2524a..6710b8c1 100644 --- a/src/apify/events/_apify_event_manager.py +++ b/src/apify/events/_apify_event_manager.py @@ -3,8 +3,9 @@ import asyncio import contextlib import time +from collections import defaultdict from logging import getLogger -from typing import TYPE_CHECKING, Annotated, Self, cast +from typing import TYPE_CHECKING, Annotated, Any, Self, cast import websockets.asyncio.client import websockets.client @@ -24,6 +25,7 @@ from types import TracebackType from crawlee.events._event_manager import EventManagerOptions + from crawlee.events._types import EventData, EventListener, WrappedListener from apify._configuration import Configuration @@ -94,6 +96,11 @@ def __init__(self, configuration: Configuration, **kwargs: Unpack[EventManagerOp connection, so that `__aenter__` can report it. """ + self._internal_listeners: defaultdict[Event, dict[EventListener[Any], WrappedListener]] = defaultdict(dict) + """Listeners of the SDK itself, mapped as `event -> listener -> wrapper`. `off` doesn't remove them, so user + code removing all listeners of an event keeps the SDK's own handling of it. + """ + @override async def __aenter__(self) -> Self: """Initialize the event manager upon entering the async context. @@ -149,6 +156,24 @@ async def __aexit__( # emitting `PersistState` again, as re-entering the context would be a no-op. await super().__aexit__(exc_type, exc_value, exc_traceback) + @override + def emit(self, *, event: Event, event_data: EventData) -> None: + super().emit(event=event, event_data=event_data) + + for listener, listener_wrapper in self._internal_listeners.get(event, {}).items(): + task_name = f'Task-{event.value}-{self._get_listener_name(listener)}' + listener_task = asyncio.create_task(listener_wrapper(event_data), name=task_name) + self._listener_tasks.add(listener_task) + listener_task.add_done_callback(self._listener_tasks.discard) + + def _on_internal(self, *, event: Event, listener: EventListener[Any]) -> None: + """Register a listener of the SDK itself, which `off` doesn't remove.""" + self._internal_listeners[event][listener] = self._wrap_listener(event, listener) + + def _off_internal(self, *, event: Event, listener: EventListener[Any]) -> None: + """Remove a listener registered by `_on_internal`.""" + self._internal_listeners.get(event, {}).pop(listener, None) + async def _teardown_platform_websocket(self) -> None: """Stop consuming the platform messages and close the websocket connection to the platform events.""" try: diff --git a/tests/unit/actor/test_actor_child_runs.py b/tests/unit/actor/test_actor_child_runs.py index 99e8b070..95411649 100644 --- a/tests/unit/actor/test_actor_child_runs.py +++ b/tests/unit/actor/test_actor_child_runs.py @@ -8,11 +8,13 @@ import pytest from apify_client._models import Run +from crawlee import service_locator from crawlee.events import Event, EventAbortingData -from apify import Actor +from apify import Actor, Configuration from apify._actor import _ActorType from apify._child_runs import CHILD_RUNS_KEY, ChildRunRegistry +from apify.events import ApifyEventManager if TYPE_CHECKING: from ..conftest import ApifyClientAsyncPatcher @@ -39,6 +41,14 @@ def make_run(run_id: str, status: str) -> Run: ) +@pytest.fixture +def apify_event_manager() -> ApifyEventManager: + """Make the Actor use `ApifyEventManager`, which delivers `ABORTING` on the platform, without a websocket.""" + event_manager = ApifyEventManager(Configuration.get_global_configuration()) + service_locator.set_event_manager(event_manager) + return event_manager + + 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() @@ -452,7 +462,7 @@ async def test_reattach_replaces_recorded_abort_with_parent( async def test_aborting_event_aborts_marked_active_child_runs( - apify_client_async_patcher: ApifyClientAsyncPatcher, + apify_client_async_patcher: ApifyClientAsyncPatcher, apify_event_manager: ApifyEventManager ) -> None: """On `ABORTING`, only child runs marked `abort_with_parent` that are still active are gracefully aborted.""" runs = { @@ -482,8 +492,8 @@ async def get_run(run_client: Any, *_args: Any, **_kwargs: Any) -> Run | None: ] }, ) - Actor.event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) - await Actor.event_manager.wait_for_all_listeners_to_complete() + apify_event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await apify_event_manager.wait_for_all_listeners_to_complete() aborts = apify_client_async_patcher.calls['run']['abort'] assert sorted(args[0].resource_id for args, _ in aborts) == ['ready-run', 'running-run'] @@ -491,7 +501,9 @@ async def get_run(run_client: Any, *_args: Any, **_kwargs: Any) -> Run | None: async def test_failed_child_run_abort_does_not_stop_others( - apify_client_async_patcher: ApifyClientAsyncPatcher, caplog: pytest.LogCaptureFixture + apify_client_async_patcher: ApifyClientAsyncPatcher, + caplog: pytest.LogCaptureFixture, + apify_event_manager: ApifyEventManager, ) -> None: """A child run that fails to abort is logged, and the other marked child runs are still aborted.""" @@ -513,8 +525,8 @@ async def abort_run(run_client: Any, *_args: Any, **_kwargs: Any) -> None: for name in ['broken', 'healthy'] }, ) - Actor.event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) - await Actor.event_manager.wait_for_all_listeners_to_complete() + apify_event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await apify_event_manager.wait_for_all_listeners_to_complete() aborts = apify_client_async_patcher.calls['run']['abort'] assert sorted(args[0].resource_id for args, _ in aborts) == ['broken-run', 'healthy-run'] @@ -608,16 +620,35 @@ async def start_run() -> Run: client.run.return_value.abort.assert_awaited_once_with(gracefully=True) -async def test_exit_removes_the_aborting_listener(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None: +async def test_exit_removes_the_aborting_listener( + apify_client_async_patcher: ApifyClientAsyncPatcher, apify_event_manager: ApifyEventManager +) -> None: """After the Actor exits, an `ABORTING` event on a still-active event manager aborts no child run.""" 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('new-run', 'RUNNING')) apify_client_async_patcher.patch('run', 'abort', return_value=None) - async with Actor.event_manager: + async with apify_event_manager: async with Actor: await Actor.start('some-actor', name='scrape-eu', abort_with_parent=True) - Actor.event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) - await Actor.event_manager.wait_for_all_listeners_to_complete() + apify_event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await apify_event_manager.wait_for_all_listeners_to_complete() assert apify_client_async_patcher.calls['run']['abort'] == [] + + +async def test_removing_all_aborting_listeners_keeps_aborting_child_runs( + apify_client_async_patcher: ApifyClientAsyncPatcher, apify_event_manager: ApifyEventManager +) -> None: + """Removing all `ABORTING` listeners from the event manager still aborts child runs marked `abort_with_parent`.""" + 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('new-run', 'RUNNING')) + apify_client_async_patcher.patch('run', 'abort', return_value=None) + + async with Actor: + await Actor.start('some-actor', name='scrape-eu', abort_with_parent=True) + apify_event_manager.off(event=Event.ABORTING) + apify_event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await apify_event_manager.wait_for_all_listeners_to_complete() + + assert len(apify_client_async_patcher.calls['run']['abort']) == 1 diff --git a/tests/unit/events/test_apify_event_manager.py b/tests/unit/events/test_apify_event_manager.py index 6d6ec65c..11f13d0f 100644 --- a/tests/unit/events/test_apify_event_manager.py +++ b/tests/unit/events/test_apify_event_manager.py @@ -9,14 +9,14 @@ from collections import defaultdict from datetime import timedelta from typing import TYPE_CHECKING, Any -from unittest.mock import Mock +from unittest.mock import AsyncMock, Mock import pytest import websockets import websockets.asyncio.client import websockets.asyncio.server -from crawlee.events._types import Event +from crawlee.events._types import Event, EventAbortingData from ..._utils import poll_until_condition from apify import Configuration @@ -489,6 +489,32 @@ async def handler(_data: Any) -> None: assert persist_state_counter == 0 +async def test_internal_listener_is_kept_by_off() -> None: + """A listener registered by `_on_internal` still runs after `off` removes all listeners of its event.""" + listener = AsyncMock() + + async with ApifyEventManager(Configuration.get_global_configuration()) as event_manager: + event_manager._on_internal(event=Event.ABORTING, listener=listener) + event_manager.off(event=Event.ABORTING) + event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await event_manager.wait_for_all_listeners_to_complete() + + listener.assert_awaited_once() + + +async def test_off_internal_removes_the_listener() -> None: + """A listener removed by `_off_internal` no longer runs on its event.""" + listener = AsyncMock() + + async with ApifyEventManager(Configuration.get_global_configuration()) as event_manager: + event_manager._on_internal(event=Event.ABORTING, listener=listener) + event_manager._off_internal(event=Event.ABORTING, listener=listener) + event_manager.emit(event=Event.ABORTING, event_data=EventAbortingData()) + await event_manager.wait_for_all_listeners_to_complete() + + listener.assert_not_awaited() + + async def test_deprecated_event_is_skipped(monkeypatch: pytest.MonkeyPatch) -> None: """Test that deprecated events (like CPU_INFO) are silently skipped.""" async with ( From 9554a97a0de428e7e83baab75564e4e7994af21b Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 30 Sep 2026 10:57:05 +0200 Subject: [PATCH 9/9] docs: note which child runs aren't aborted with the parent --- docs/02_concepts/06_interacting_with_other_actors.mdx | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/02_concepts/06_interacting_with_other_actors.mdx b/docs/02_concepts/06_interacting_with_other_actors.mdx index 6ddbc9ed..7d73ec82 100644 --- a/docs/02_concepts/06_interacting_with_other_actors.mdx +++ b/docs/02_concepts/06_interacting_with_other_actors.mdx @@ -83,6 +83,8 @@ Note that: - Each call under a name records its own value, so the latest call decides whether the run is aborted. - A child run aborted this way ends as `ABORTED`. If your Actor run is resurrected later, the same named call resurrects the child run too. - Only a graceful abort gives the SDK time to act. A hard abort, a timeout, or a crash of your Actor run leaves the child runs running. +- Child runs started after your Actor run received `ABORTING` aren't aborted, so don't start new ones while it's shutting down. +- A child run started with its own `token` is aborted with that token. After a migration or resurrection, the SDK uses your Actor's token for it until the same named call runs again. If that token can't access the child run, the abort fails and the error is logged. ## Actor call task