From dd3e7482bf491f2fdcfef441cb8fecbc0af6355c Mon Sep 17 00:00:00 2001
From: goodboy <goodboy_foss@protonmail.com>
Date: Tue, 25 Aug 2026 20:23:53 -0400
Subject: [PATCH] Wait for active pub/sub cancellation targets

Replace `test_dynamic_pub_sub()`'s fixed startup sleep with an RPC
activity probe in the publisher actor. Track the publisher task and
wait until every launched consumer has installed its first
subscription before raising the user cancellation exception.

This keeps slow spawn backends from passing the regression by
cancelling actors which never reached the streaming workload.

Review: PR #484 (OpenCode)
https://github.com/goodboy/tractor/pull/484#pullrequestreview-5025383596

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
---
 tests/test_advanced_streaming.py | 51 +++++++++++++++++++++++++-------
 1 file changed, 41 insertions(+), 10 deletions(-)

diff --git a/tests/test_advanced_streaming.py b/tests/test_advanced_streaming.py
index 194eb7d1..0f584247 100644
--- a/tests/test_advanced_streaming.py
+++ b/tests/test_advanced_streaming.py
@@ -25,6 +25,7 @@ _registry: dict[str, set[tractor.MsgStream]] = {
     'even': set(),
     'odd': set(),
 }
+_publisher_started: bool = False
 
 
 async def publisher(
@@ -33,21 +34,43 @@ async def publisher(
 
 ) -> None:
 
-    global _registry
+    global _publisher_started, _registry
 
     def is_even(i):
         return i % 2 == 0
 
-    for val in itertools.count(seed):
+    _publisher_started = True
+    try:
+        for val in itertools.count(seed):
 
-        sub = 'even' if is_even(val) else 'odd'
+            sub = 'even' if is_even(val) else 'odd'
 
-        for sub_stream in _registry[sub].copy():
-            await sub_stream.send(val)
+            for sub_stream in _registry[sub].copy():
+                await sub_stream.send(val)
 
-        # throttle send rate to ~1kHz
-        # making it readable to a human user
-        await trio.sleep(1/1000)
+            # throttle send rate to ~1kHz
+            # making it readable to a human user
+            await trio.sleep(1/1000)
+
+    finally:
+        _publisher_started = False
+
+
+async def pubsub_active(
+    expected_subs: int,
+) -> bool:
+    '''
+    Report whether the publisher and all subscriber tasks are active.
+
+    Runs as an RPC task in the publisher actor, where `_registry` is
+    mutated by each `subscribe()` context after its consumer sends the
+    first subscription.
+
+    '''
+    return (
+        _publisher_started
+        and sum(map(len, _registry.values())) >= expected_subs
+    )
 
 
 @tractor.context
@@ -270,8 +293,16 @@ def test_dynamic_pub_sub(
                             )
                         )
 
-                        # block until "cancelled by user"
-                        await trio.sleep(3)
+                        expected_subs: int = max(cpus - 2, 0) + 1
+                        async with tractor.wait_for_actor(
+                            'publisher',
+                        ) as portal:
+                            while not await portal.run(
+                                pubsub_active,
+                                expected_subs=expected_subs,
+                            ):
+                                await trio.sleep(0.01)
+
                         test_log.warning(
                             f'Raising user cancel exc: '
                             f'{expect_cancel_exc!r}'