Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions src/apify/_actor.py
Original file line number Diff line number Diff line change
Expand Up @@ -1021,8 +1021,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 the Actor and input it was first used with,
so reusing it for a different Actor, task or input raises a `ValueError`.
run `FAILED` or no longer exists. Using `run_name` to attach to a run of a different Actor, task or
with a different input raises a `ValueError`.

Returns:
Info about the started Actor run
Expand Down Expand Up @@ -1212,8 +1212,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 the Actor and input it was first used with,
so reusing it for a different Actor, task or input raises a `ValueError`.
run `FAILED` or no longer exists. Using `run_name` to attach to a run of a different Actor, task or
with a different input raises a `ValueError`.

Returns:
Info about the started Actor run.
Expand Down Expand Up @@ -1384,8 +1384,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 the task and input it was first used with,
so reusing it for a different Actor, task or input raises a `ValueError`.
run `FAILED` or no longer exists. Using `run_name` to attach to a run of a different Actor, task or
with a different input raises a `ValueError`.

Returns:
Info about the started Actor run.
Expand Down Expand Up @@ -1474,8 +1474,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 the task and input it was first used with,
so reusing it for a different Actor, task or input raises a `ValueError`.
run `FAILED` or no longer exists. Using `run_name` to attach to a run of a different Actor, task or
with a different input raises a `ValueError`.

Returns:
Info about the started Actor run.
Expand Down
26 changes: 16 additions & 10 deletions src/apify/_child_runs.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ class ChildRunRecord(ChildRunSnapshot):
"""The current run tracked under a name in the child run registry, with the runs it replaced."""

checksum: str
"""Hash of the Actor or task and the input the name was first used with."""
"""Hash of the Actor or task and the input the current run was started 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."""
Expand Down Expand Up @@ -122,40 +122,38 @@ async def find_or_start(

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.
A `FAILED` run, or one the API no longer knows, is replaced by a new run under the same name, for any Actor,
task or input.

Args:
name: Name of the child run, unique within the parent run.
actor_id: The Actor to start.
task_id: The task to start, in place of `actor_id`.
run_input: Input of the run. With the Actor or task, it must match what `name` was first used with.
run_input: Input of the run. The recorded run is reused only if it was started with the same Actor or task
and input.
client: Client used to look up the recorded run.
start_run: Starts a new run of the Actor or task.
resurrect_run: Resurrects the recorded run, given its ID.

Returns:
The run, and whether it was newly started.

Raises:
ValueError: If the recorded run would be reused, but was started for a different Actor, task or input.
"""
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 not None and record.checksum != checksum:
raise ValueError(
f'The run name "{name}" was already used for a different Actor, task or input. '
'Use a unique `run_name` for each child run.'
)

if record is None:
run = await self._start(name, checksum=checksum, start_run=start_run, history=[])
self._clients[name] = client
return run, True

run_client = client.run(record.run_id)
run = await _get_recorded_run(run_client)
self._clients[name] = client

if run is not None and run.status in _SETTLING_STATUSES:
run = await run_client.wait_for_finish()
Expand All @@ -169,15 +167,23 @@ async def find_or_start(
run = await self._start(
name, checksum=checksum, start_run=start_run, history=[*record.history, replaced]
)
self._clients[name] = client
return run, True

if record.checksum != checksum:
raise ValueError(
f'The run name "{name}" was already used for a different Actor, task or input. '
'Use a unique `run_name` for each child run.'
)

if run.status in _RESURRECTABLE_STATUSES:
logger.info(f'Resurrecting child run "{name}"', extra={'run_id': run.id, 'status': run.status})
run = await resurrect_run(run.id)
else:
logger.info(f'Reattaching to child run "{name}"', extra={'run_id': run.id, 'status': run.status})

await self.update(name, run)
self._clients[name] = client
return run, False

def run_clients(self, default_client: ApifyClientAsync) -> dict[str, RunClientAsync]:
Expand Down
51 changes: 50 additions & 1 deletion tests/unit/actor/test_actor_child_runs.py
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,7 @@ async def test_named_start_rejects_name_recorded_for_another_actor(
) -> 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'))
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='other-actor')
Expand All @@ -229,6 +230,7 @@ async def test_named_start_rejects_name_recorded_for_task(
) -> 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'))
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')
Expand Down Expand Up @@ -439,6 +441,7 @@ async def test_named_call_task_rejects_name_recorded_for_actor(
) -> 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'))
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')
Expand Down Expand Up @@ -534,18 +537,63 @@ async def test_named_start_retries_recorded_run_not_found_yet(
assert apify_client_async_patcher.calls['actor']['start'] == []


@pytest.mark.parametrize(
'status',
[
pytest.param('RUNNING', id='running'),
pytest.param('SUCCEEDED', id='succeeded'),
pytest.param('ABORTED', id='aborted'),
],
)
async def test_named_start_rejects_name_recorded_with_other_input(
apify_client_async_patcher: ApifyClientAsyncPatcher,
status: str,
) -> None:
"""Reusing a name for the same Actor with a different input raises instead of attaching to the earlier run."""
"""Reusing a name with a different input raises instead of reattaching to, or resurrecting, the earlier 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('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_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'] == []
assert apify_client_async_patcher.calls['run']['resurrect'] == []


@pytest.mark.parametrize(
('recorded_actor_id', 'recorded_input'),
[
pytest.param('other-actor', None, id='other actor'),
pytest.param('some-actor', {'since': '2025-01-01'}, id='other input'),
],
)
async def test_named_start_replaces_run_recorded_for_other_request(
apify_client_async_patcher: ApifyClientAsyncPatcher,
recorded_actor_id: str,
recorded_input: Any,
) -> None:
"""A failed run recorded for another Actor or input is replaced, and the name takes the new run's checksum."""
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', 'FAILED'))

async with Actor:
await record_child_run('scrape-eu', 'old-run', actor_id=recorded_actor_id, run_input=recorded_input)
run = await Actor.start('some-actor', {'since': '2026-01-01'}, run_name='scrape-eu')
kvs = await Actor.open_key_value_store()
stored = await kvs.get_value(CHILD_RUNS_KEY)

assert run.id == 'new-run'
assert stored == {
'scrape-eu': stored_record(
'new-run',
'READY',
run_input={'since': '2026-01-01'},
history=[{'runId': 'old-run', 'status': 'FAILED', 'startedAt': STARTED_AT}],
)
}


async def test_named_start_ignores_input_key_order(apify_client_async_patcher: ApifyClientAsyncPatcher) -> None:
Expand Down Expand Up @@ -696,6 +744,7 @@ async def test_child_runs_keeps_client_of_name_after_rejected_reuse(
) -> None:
"""A name reuse rejected for a different input leaves `Actor.child_runs` with the original client."""
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'))

async with Actor:
await Actor.start('some-actor', {'since': '2025-01-01'}, run_name='scrape-eu')
Expand Down
Loading