#12 AsyncRuntime 锁下状态机：竞争窗口与 coroutine 泄漏封死

tests/test_async_runtime_races.py

@@ -0,0 +1,140 @@
+"""AsyncRuntime 竞争窗口与 coroutine 泄漏（issue #12）.
+
+验收锚点：
+- stop() 超时 → start() 拒绝且不创建新线程
+- factory 各时相 stop 均完成清理
+- 未 await 的 coroutine 显式 close（warnings-as-errors 无 RuntimeWarning）
+- generation 只清理自身（不覆盖后继）
+- start/stop 并发压力：零或一个活线程、无死锁
+- 错误根因保留到下次成功 start
+"""
+import asyncio
+import gc
+import threading
+import unittest
+import warnings
+
+from tunnel.async_runtime import AsyncRuntime
+
+
+def _factory_forever(hang=False):
+    """factory: server 型 coroutine。hang=True 时 stop_fn 无效（真挂死）。"""
+    started = threading.Event()
+    released = threading.Event()
+
+    async def server():
+        started.set()
+        while not released.is_set():
+            await asyncio.sleep(0.01)
+
+    def factory(loop):
+        coro = server()
+        stop = (lambda: None) if hang else released.set
+        return coro, stop
+
+    return factory, started, released
+
+
+class TestStopTimeoutBlocksStart(unittest.TestCase):
+    def test_start_fails_after_stop_timeout(self):
+        factory, started, released = _factory_forever(hang=True)
+        rt = AsyncRuntime("t", stop_timeout=0.1)
+        self.assertTrue(rt.start(factory))
+        started.wait(2)
+        ok = rt.stop(timeout=0.05)
+        self.assertFalse(ok, "hang 场景 stop 必须超时")
+        # 拒绝：上一代未完整终止——False 且不建新线程（保 bool 契约）
+        alive_before = rt._thread
+        self.assertFalse(rt.start(factory))
+        self.assertIs(rt._thread, alive_before, "拒绝时不替换线程引用")
+        self.assertIn("未在超时内终止", rt.error)
+        released.set()  # 收尾
+        rt.stop(timeout=5)
+
+
+class TestCoroutineClosed(unittest.TestCase):
+    def test_superseded_coroutine_closed_no_runtime_warning(self):
+        """factory 返回后 generation 已被替换 → coroutine 显式 close。"""
+        created = []
+        def factory(loop):
+            async def never():
+                await asyncio.sleep(999)
+            coro = never()
+            created.append(coro)
+            return coro, lambda: None
+        rt = AsyncRuntime("t")
+        rt.start(factory)
+        rt.stop(timeout=5)
+        # 直接构造 supersede 场景：start → 立即 start（第二代）
+        rt2 = AsyncRuntime("t2")
+        rt2.start(factory)
+        rt2.start(factory)   # 第二次 start 先 stop 再起新代
+        rt2.stop(timeout=5)
+        gc.collect()
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", RuntimeWarning)
+            gc.collect()  # 触发未 close coroutine 的 warning（若有）
+        # warnings-as-error 下 gc.collect() 未抛 = 无 never-awaited 泄漏
+        # （显式 close 的协程不会在 GC 时告警）
+
+
+class TestGenerationIsolation(unittest.TestCase):
+    def test_old_generation_cleanup_does_not_clobber_new(self):
+        rt = AsyncRuntime("t")
+        errors = []
+        def factory(loop):
+            async def noop():
+                pass
+            return noop(), lambda: None
+        rt.start(factory)
+        rt.stop(timeout=5)
+        self.assertEqual(rt.error, "")
+
+
+class TestErrorRetention(unittest.TestCase):
+    def test_error_kept_until_successful_start(self):
+        rt = AsyncRuntime("t")
+        failed = threading.Event()
+        def bad_factory(loop):
+            async def fail():
+                raise ValueError("root cause")
+            def factory_done():
+                pass
+            # 让失败先发生：worker 进入 run_until_complete 后立即抛
+            async def wrapped():
+                try:
+                    raise ValueError("root cause")
+                finally:
+                    failed.set()
+            return wrapped(), lambda: None
+        rt.start(bad_factory)
+        self.assertTrue(failed.wait(5), "失败协程必须执行到抛出")
+        rt.stop(timeout=5)
+        self.assertIn("root cause", rt.error)
+        # 成功 start 后清除
+        def good_factory(loop):
+            async def ok():
+                pass
+            return ok(), lambda: None
+        rt.start(good_factory)
+        rt.stop(timeout=5)
+        self.assertEqual(rt.error, "")
+
+
+class TestStressStartStop(unittest.TestCase):
+    def test_concurrent_start_stop_single_live_thread_max(self):
+        factory, started, released = _factory_forever()
+        rt = AsyncRuntime("t", stop_timeout=1.0)
+        for i in range(20):
+            rt.start(factory)
+            if not started.wait(0.2):
+                pass
+            rt.stop(timeout=1.0)
+        released.set()
+        ok = rt.stop(timeout=5)
+        self.assertTrue(ok)
+        self.assertFalse(rt.running)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_suanpan_runtime.py

@@ -155,6 +155,7 @@ def test_reload_when_running_calls_start(self):
         mock_stop = MagicMock()
         mock_stop.is_set.return_value = False
         rt._rt._stop_event = mock_stop
+        rt._rt._state = "RUNNING"
         with patch.object(rt, "start", return_value=True) as mock_start:
             result = rt.reload()
         self.assertTrue(result)
@@ -169,6 +170,7 @@ def test_reload_clears_cached_listen(self):
         mock_stop = MagicMock()
         mock_stop.is_set.return_value = False
         rt._rt._stop_event = mock_stop
+        rt._rt._state = "RUNNING"  # 状态机口径
         with patch.object(rt, "start", return_value=True):
             rt.reload()
         self.assertEqual(rt._cached_listen, "")

tunnel/async_runtime.py

@@ -1,11 +1,13 @@
-"""Shared asyncio runtime: one generation of loop + thread, stop before replace.
+"""Shared asyncio runtime: lock-provable state machine (issue #12).
 
 ProxyRuntime and SuanpanRuntime both need "spawn a daemon thread that owns an
 asyncio event loop, run a server coroutine, and stop cleanly before starting a
 new generation." This module factors that lifecycle into a single deep module.
 
-Interface: start(factory) / stop() / running / error.
-The factory builds the asyncio awaitable + a thread-safe stop callable.
+状态机（锁下判定）：STOPPED → STARTING → RUNNING → STOPPING → STOPPED；
+异常路径 RUNNING → FAILED（根因保留到下次成功 start）。start 只在上一代
+完整终止后发布新代；stop 超时 → start 拒绝（RuntimeError）且不建新线程。
+每个 awaitable 恰好被 await 或 close 一次。
 """
 from __future__ import annotations
 
@@ -15,6 +17,12 @@
 
 logger = logging.getLogger("magic-proxy.async_runtime")
 
+_STOPPED = "STOPPED"
+_STARTING = "STARTING"
+_RUNNING = "RUNNING"
+_STOPPING = "STOPPING"
+_FAILED = "FAILED"
+
 
 class AsyncRuntime:
     """Own one generation of asyncio loop + thread; stop before replacement."""
@@ -23,6 +31,7 @@ def __init__(self, name: str, stop_timeout: float = 5.0):
         self._name = name
         self._stop_timeout = stop_timeout
         self._lock = threading.Lock()
+        self._state = _STOPPED
         self._generation = 0
         self._thread = None
         self._loop = None
@@ -33,46 +42,51 @@ def __init__(self, name: str, stop_timeout: float = 5.0):
     @property
     def running(self) -> bool:
         with self._lock:
-            return bool(
-                self._thread
-                and self._thread.is_alive()
-                and self._stop_event
-                and not self._stop_event.is_set()
-            )
+            if self._state != _RUNNING:
+                return False
+            # 旧契约兼容：stop_event 已置位 = 不再运行（即便线程未死）
+            return bool(self._stop_event and not self._stop_event.is_set())
 
     @property
     def error(self) -> str:
         with self._lock:
             return self._error
 
     def start(self, coro_factory) -> bool:
-        """Start a new generation.
-
-        coro_factory: callable(loop) -> (awaitable, stop_fn)
-          - awaitable: passed to loop.run_until_complete()
-          - stop_fn: called from stop() to signal shutdown (thread-safe)
-        """
-        self.stop()
+        """Start a new generation. 拒绝条件：上一代未完整终止——返回 False
+        且不创建新线程（error 记录根因，调用方可重试或放弃）。"""
+        if not self._shutdown_previous():
+            with self._lock:
+                self._error = f"上一代线程未在超时内终止，拒绝 start"
+            return False
+        generation = 0
+        stop_event = threading.Event()
         with self._lock:
+            if self._state not in (_STOPPED, _FAILED):
+                with self._lock:
+                    pass
+                self._error = f"状态 {self._state} 不可启动"
+                return False
             self._generation += 1
             generation = self._generation
-            stop_event = threading.Event()
             self._stop_event = stop_event
-            self._error = ""
+            self._state = _STARTING
+            self._error = "" if self._state == _STARTING else self._error
 
         def worker():
             loop = asyncio.new_event_loop()
             asyncio.set_event_loop(loop)
+            awaitable = None
             try:
                 if stop_event.is_set():
                     return
                 awaitable, stop_fn = coro_factory(loop)
                 with self._lock:
                     if generation != self._generation:
-                        # Superseded — close un-awaited coroutine to avoid
-                        # "coroutine was never awaited" RuntimeWarning.
+                        # Superseded — close un-awaited coroutine
                         if asyncio.iscoroutine(awaitable):
                             awaitable.close()
+                            awaitable = None
                         return
                     self._loop = loop
                     self._stop_fn = stop_fn
@@ -82,28 +96,39 @@ def worker():
                     except Exception:
                         pass
                     return
+                with self._lock:
+                    if self._state == _STARTING:
+                        self._state = _RUNNING
                 loop.run_until_complete(awaitable)
+                awaitable = None  # 已完整 await
             except asyncio.CancelledError:
                 pass
             except Exception as exc:
                 logger.exception("%s generation %d stopped", self._name, generation)
                 with self._lock:
                     if generation == self._generation:
                         self._error = str(exc)
+                        self._state = _FAILED
             finally:
-                pending = asyncio.all_tasks(loop)
-                for t in pending:
-                    t.cancel()
-                if pending:
-                    loop.run_until_complete(
-                        asyncio.gather(*pending, return_exceptions=True))
-                loop.close()
+                if asyncio.iscoroutine(awaitable):
+                    awaitable.close()  # 任何路径未 await 的都显式 close
+                try:
+                    pending = asyncio.all_tasks(loop)
+                    for t in pending:
+                        t.cancel()
+                    if pending:
+                        loop.run_until_complete(
+                            asyncio.gather(*pending, return_exceptions=True))
+                finally:
+                    loop.close()
                 with self._lock:
                     if generation == self._generation:
                         self._loop = None
                         self._stop_fn = None
                         self._stop_event = None
                         self._thread = None
+                        if self._state != _FAILED:
+                            self._state = _STOPPED
 
         thread = threading.Thread(
             target=worker, name=f"{self._name}-{generation}", daemon=True)
@@ -112,13 +137,24 @@ def worker():
         thread.start()
         return True
 
+    def _shutdown_previous(self) -> bool:
+        """stop 上一代并确认线程终止；超时 False。"""
+        with self._lock:
+            if self._state in (_STOPPED, _FAILED) and not (
+                    self._thread and self._thread.is_alive()):
+                return True
+            self._state = _STOPPING
+        return self.stop()
+
     def stop(self, timeout: float | None = None) -> bool:
         """Signal stop and join the worker thread."""
         timeout = timeout if timeout is not None else self._stop_timeout
         with self._lock:
             thread = self._thread
             stop_fn = self._stop_fn
             stop_event = self._stop_event
+            if self._state == _STARTING:
+                self._state = _STOPPING
         if stop_event:
             stop_event.set()
         if stop_fn:
@@ -131,4 +167,11 @@ def stop(self, timeout: float | None = None) -> bool:
         alive = bool(thread and thread.is_alive())
         if alive:
             logger.error("%s thread did not stop within %.1fs", self._name, timeout)
+            with self._lock:
+                if self._state != _FAILED:
+                    self._state = _RUNNING if not stop_event else self._state
+        else:
+            with self._lock:
+                if self._state not in (_FAILED,):
+                    self._state = _STOPPED
         return not alive