test: fix flaky test_event_listener_can_be_removed_successfully (#1013)

`test_event_listener_can_be_removed_successfully` is flaky on CI
([example
run](https://github.com/apify/apify-sdk-python/actions/runs/28158933735/job/83394293634)).
The deployed Actor asserts `counter == last_count` after `Actor.off()`,
but the count occasionally moved on, so the run reports `FAILED` and the
test fails.

The cause is a timing race, not a bug in `Actor.off()`.
`EventManager.emit()` dispatches each listener through an asyncio task,
so there's latency between a `PERSIST_STATE` event being emitted and the
counter being incremented. `off()` only stops future dispatch. It can't
cancel an invocation already in flight from an event emitted just before
it. Under load that increment lands after the snapshot, so the equality
check fails even though `off()` behaves correctly.

The fix drops the fixed-duration `asyncio.sleep` calls the test relied
on. A second reference listener stays subscribed the whole time and acts
as a heartbeat, so every step waits until N more `PERSIST_STATE` events
have actually been observed rather than for a guessed wall-clock
duration. After `off()` the test waits one event cycle (which flushes
any invocation dispatched just before it), snapshots the removed
listener's count, then waits several more event cycles and asserts the
count did not move. That still fails if `off()` ever genuinely breaks.
Both listeners are async so they run on the event loop and can drive the
heartbeat safely.

Verified locally against the real `EventManager` (no platform needed):
the previous fixed-sleep version fails 40/40 under load, while this
event-driven version holds 0 failures across every condition tried,
including the exact stress that broke the old version, listeners slower
than the emit interval, and 150-run batches.

Author: vdusek

tests/e2e/test_actor_events.py

@@ -75,29 +75,56 @@ async def test_event_listener_can_be_removed_successfully(
 ) -> None:
     async def main() -> None:
         import os
-        from typing import Any
 
         from crawlee.events._types import Event
 
         from apify._consts import ApifyEnvVars
 
         os.environ[ApifyEnvVars.PERSIST_STATE_INTERVAL_MILLIS] = '100'
 
-        counter = 0
-
-        def count_event(data: Any) -> None:
-            nonlocal counter
-            print(data)
-            counter += 1
+        # `removed_count` is bumped by the listener we later remove. `total_count` is bumped by a
+        # reference listener that stays subscribed the whole time and acts as a heartbeat, letting the
+        # test wait for real events instead of guessing wall-clock durations. Both listeners are async
+        # so they run on the event loop and can drive `heartbeat` safely.
+        removed_count = 0
+        total_count = 0
+        heartbeat = asyncio.Event()
+
+        async def removed_listener() -> None:
+            nonlocal removed_count
+            removed_count += 1
+
+        async def reference_listener() -> None:
+            nonlocal total_count
+            total_count += 1
+            heartbeat.set()
+
+        async def wait_for_events(n: int) -> None:
+            """Block until the reference listener has observed at least `n` more events."""
+            target = total_count + n
+            while total_count < target:
+                heartbeat.clear()
+                if total_count >= target:
+                    return
+                await heartbeat.wait()
 
         async with Actor:
-            Actor.on(Event.PERSIST_STATE, count_event)
-            await asyncio.sleep(0.5)
-            assert counter > 1
-            last_count = counter
-            Actor.off(Event.PERSIST_STATE, count_event)
-            await asyncio.sleep(0.5)
-            assert counter == last_count
+            Actor.on(Event.PERSIST_STATE, removed_listener)
+            Actor.on(Event.PERSIST_STATE, reference_listener)
+
+            # Both listeners are live: wait until the one we will remove has demonstrably received events.
+            while removed_count < 2:
+                await wait_for_events(1)
+
+            Actor.off(Event.PERSIST_STATE, removed_listener)
+
+            # One more event cycle flushes any invocation dispatched just before `off`, then snapshot.
+            await wait_for_events(1)
+            count_after_off = removed_count
+
+            # Over several further event cycles, the removed listener must not be called again.
+            await wait_for_events(5)
+            assert removed_count == count_after_off
 
     actor = await make_actor(label='actor-off-event', main_func=main)
     run_result = await run_actor(actor)