The test-infrastructure patches asyncio.sleep with a 10 ms cap to speed up retry waits. The two slow-executor Behave step definitions used asyncio.sleep(10) as the "slow" coroutine, which was silently capped to 10 ms — the same duration as the 0.01 s executor timeout — creating a race condition that caused the "Executor times out via thread pool path" and "Executor times out via run_coroutine_threadsafe path" scenarios to fail intermittently.
Fix: use asyncio._original_sleep (falling back to asyncio.sleep when the patch is absent) with a 0.5 s delay, which is 50× longer than the timeout and guarantees the timeout always fires before the coroutine completes.
Replace the no-op on_next handlers in _setup_node_stream_subscriptions
with factory-created callbacks that look up and invoke the per-node
sync_executor registered by _register_node_executor. The previous code
silently discarded every message delivered to node streams.
Key changes:
- Wire on_next to executor via _make_on_next_handler factory method
- Store executors in _node_executors dict (not setattr on stream_router)
- Shared ThreadPoolExecutor with proper lifecycle (start/stop/restart)
- Prefer run_coroutine_threadsafe when scheduler loop is running;
fall back to thread pool with deadlock prevention
- Thread-safe StateManager: Lock on all mutation methods, deep-copy
emission inside lock for immutable subscriber snapshots
- replace_state() and reset() deep-copy input to prevent external
mutation bypassing the lock (review cycle 5 M1 fix)
- Best-effort cancellation documented on timed-out futures (cycle 5 M2)
- CancelledError catch in thread pool path with "graph stopping" message
- is_running guard on both sync_executor and execute() entry point
- Checkpoint filenames include update_count to prevent collisions
- _save_checkpoint docstring warns against calling while _lock is held
- on_next handler distinguishes TimeoutError (warning) from other
exceptions (exception log) for clearer diagnostics
- TODO comment for deferred downstream propagation to successor nodes
- __del__ safety net for executor pool cleanup
- File I/O outside lock for checkpoint save/load
- Bounded execution_history via collections.deque(maxlen=1000)
See PR !10795 for the full change log across 5 review cycles.
ISSUES CLOSED: #6511
HAL9000
hurui200320
freemo
