fix: handle ServiceConflictError when reusing Actor across sequential context (#804)

### Description

- handle `ServiceConflictError` when reusing `Actor` across sequential
context

### Issues

- Closes: #678

### Testing

- Added new tests

Author: Mantisus

src/apify/_actor.py

@@ -350,18 +350,27 @@ def event_manager(self) -> EventManager:
 
         It uses `ApifyEventManager` on the Apify platform and `LocalEventManager` otherwise.
         """
-        event_manager = (
-            ApifyEventManager(
-                configuration=self.configuration,
-                persist_state_interval=self.configuration.persist_state_interval,
+        try:
+            event_manager = (
+                ApifyEventManager(
+                    configuration=self.configuration,
+                    persist_state_interval=self.configuration.persist_state_interval,
+                )
+                if self.is_at_home()
+                else LocalEventManager(
+                    system_info_interval=self.configuration.system_info_interval,
+                    persist_state_interval=self.configuration.persist_state_interval,
+                )
             )
-            if self.is_at_home()
-            else LocalEventManager(
-                system_info_interval=self.configuration.system_info_interval,
-                persist_state_interval=self.configuration.persist_state_interval,
+            service_locator.set_event_manager(event_manager)
+        except ServiceConflictError:
+            self.log.debug(
+                'Event manager already exists in service locator (set by previous Actor context or explicitly by '
+                'user). Using the existing event manager.'
             )
-        )
-        service_locator.set_event_manager(event_manager)
+            # Use the event manager from the service locator
+            event_manager = service_locator.get_event_manager()
+
         return event_manager
 
     @cached_property

tests/e2e/test_actor_lifecycle.py

@@ -110,3 +110,34 @@ async def default_handler(context: BasicCrawlingContext) -> None:
     run_result = await run_actor(actor)
 
     assert run_result.status == 'SUCCEEDED'
+
+
+async def test_actor_sequential_contexts(make_actor: MakeActorFunction, run_actor: RunActorFunction) -> None:
+    """Test that Actor and Actor() can be used in sequential async context manager blocks."""
+
+    async def main() -> None:
+        async with Actor as actor:
+            actor._exit_process = False
+            assert actor._is_initialized is True
+
+        # Actor after Actor.
+        async with Actor as actor:
+            actor._exit_process = False
+            assert actor._is_initialized is True
+
+        # Actor() after Actor.
+        async with Actor(exit_process=False) as actor:
+            assert actor._is_initialized is True
+
+        # Actor() after Actor().
+        async with Actor(exit_process=False) as actor:
+            assert actor._is_initialized is True
+
+        # Actor after Actor().
+        async with Actor as actor:
+            assert actor._is_initialized is True
+
+    actor = await make_actor(label='actor-sequential-contexts', main_func=main)
+    run_result = await run_actor(actor)
+
+    assert run_result.status == 'SUCCEEDED'

tests/unit/actor/test_actor_lifecycle.py

@@ -4,6 +4,7 @@
 import contextlib
 import logging
 from typing import TYPE_CHECKING
+from unittest.mock import AsyncMock
 
 import pytest
 
@@ -232,3 +233,28 @@ async def test_actor_fail_prevents_further_execution(caplog: pytest.LogCaptureFi
         status_records = [r for r in caplog.records if r.msg == '[Terminal status message]: cde']
         assert len(status_records) == 1
         assert status_records[0].levelno == logging.INFO
+
+
+@pytest.mark.parametrize(
+    ('first_with_call', 'second_with_call'),
+    [
+        pytest.param(False, False, id='both_without_call'),
+        pytest.param(False, True, id='first_without_call'),
+        pytest.param(True, False, id='second_without_call'),
+        pytest.param(True, True, id='both_with_call'),
+    ],
+)
+async def test_actor_sequential_contexts(*, first_with_call: bool, second_with_call: bool) -> None:
+    """Test that Actor and Actor() can be used in two sequential async context manager blocks."""
+    mock = AsyncMock()
+    async with Actor(exit_process=False) if first_with_call else Actor as actor:
+        await mock()
+        assert actor._is_initialized is True
+
+    # After exiting the context, new Actor instance can be created without conflicts.
+    async with Actor() if second_with_call else Actor as actor:
+        await mock()
+        assert actor._is_initialized is True
+
+    # The mock should have been called twice, once in each context.
+    assert mock.call_count == 2