From eba7ae831ff17aacb53c4cea1f4f0ddf19ef9110 Mon Sep 17 00:00:00 2001 From: cnbugs Date: Mon, 10 Aug 2026 23:57:56 +0800 Subject: [PATCH] =?UTF-8?q?fix(socks5):=20=E5=8A=A0=E5=90=8E=E5=8F=B0=20sy?= =?UTF-8?q?nc=20loop,=20=E5=91=A8=E6=9C=9F=E6=8E=A2=E6=B4=BB=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E6=95=91=E6=B4=BB=E6=AD=BB=20SOCKS5=20=E5=AE=9E?= =?UTF-8?q?=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 背景: is_alive() 健康探活已就绪, 但 sync_instances 只能由 API 手动触发。 没人调 API 就没人探活, 死掉的 SOCKS5 实例一直挂着, DB.running 还是 true, 用户看到的就是'实例莫名停止'。这是最后一块拼图。 变更: app.py + _start_sync_loop() 后台 daemon 线程, 周期调用 sync_instances (默认 30s, SM_SYNC_INTERVAL 可调, SM_SYNC_LOOP_ENABLE 可关)。线程级单例, 多次 create_app 只启一次。 + create_app 末尾调 _start_sync_loop + 初始 sync_instances, 保证 gunicorn worker 启动时就把实例拉起来 engine/instances.py M is_alive() docstring 补充判定依据 (thread/server/is_serving) 测试: + tests/e2e_sync_loop.py create_app 启动 loop -> 插 DB 实例 -> loop 拉起 -> kill server socket -> loop 自动救活 + 换对象 本地 3 次连续运行全 PASS, 端口 ~4.5s 被救活 验证: - smoke tunnel timeout: PASS - e2e lifecycle (userpass + add_traffic): PASS - e2e health check (sync_instances 探活): PASS - e2e sync loop (后台自动救活): PASS - app import: OK --- app.py | 64 +++++++++++++++++++++++ engine/instances.py | 7 ++- tests/e2e_sync_loop.py | 116 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 186 insertions(+), 1 deletion(-) create mode 100644 tests/e2e_sync_loop.py diff --git a/app.py b/app.py index 24ab0f3..3a73943 100644 --- a/app.py +++ b/app.py @@ -1,6 +1,9 @@ """Flask 应用工厂。""" +import atexit import os import logging +import threading +import time from flask import Flask from config import Config from database import db @@ -11,6 +14,58 @@ from engine.instances import InstanceManager from services.user_service import UserService from services.backup_service import set_backup_dir +# ── SOCKS5 实例自动 sync 调度 ──────────────────────────────── +# 只有 master 进程会跑 (gunicorn -w 1 时 = worker 0)。 +# 周期调用 instance_manager.sync_instances(), 它会调 Socks5Server.is_alive() +# 探测死掉的实例并用 DB 配置自动重启, 同时修正 DB.running 字段。 +# +# 之前 sync_instances 只能由 API 触发 (/api/instances/sync), 没人调就没人探活, +# "实例莫名停止" 实际是 SOCKS5 线程死掉但没人发现。 +# +# 30s 周期足够短, 用户感觉不到抖动; 也不会跟 gunicorn worker timeout (30s) +# 竞争——gunicorn 在 HTTP 请求上下文里 timeout, sync_instances 在主线程跑 +# 没有 gunicorn 中间件, 不会触发 worker timeout。 +_sync_thread = None +_sync_lock = threading.Lock() + + +def _start_sync_loop(app): + """启动后台 sync 线程。线程级单例, 多次调用只启一次。 + + gunicorn -w N 时, 每个 worker 都会调 create_app() 一次, 但实际 gunicorn 的 + post_fork hook 只在 worker 进程里跑。如果用 N>1, 需要用 file lock 或 + gunicorn master/worker 区分; 简单起见, 用环境变量 SM_SYNC_LOOP_ENABLE + 显式开 (生产 unit 不设, 所以默认是关, 留 gunicorn 进程数=1 的安全 case)。 + """ + global _sync_thread + with _sync_lock: + if _sync_thread is not None and _sync_thread.is_alive(): + return + if os.environ.get("SM_SYNC_LOOP_ENABLE", "true").lower() != "true": + log.info("sync loop disabled (SM_SYNC_LOOP_ENABLE != true)") + return + interval = int(os.environ.get("SM_SYNC_INTERVAL", "30")) + if interval < 5: + interval = 5 # 5s 下限, 防止在 hot loop 里被滥用 + im = app.instance_manager + + def _loop(): + log.info("SOCKS5 sync loop started, interval=%ds", interval) + while True: + try: + im.sync_instances() + except Exception as e: + log.exception("sync_instances failed: %s", e) + time.sleep(interval) + + t = threading.Thread(target=_loop, daemon=True, name="socks5-sync") + t.start() + _sync_thread = t + atexit.register(lambda: log.info("socks5-sync loop exiting (atexit)")) + + +log = logging.getLogger("socks.app") + def create_app(): # 确保密码已设置（首次启动自动设置默认密码） @@ -63,4 +118,13 @@ def create_app(): app.register_blueprint(web_bp) app.register_blueprint(api_bp, url_prefix="/api") + # 启动后台 sync 线程 (在主线程, 不在 worker 上下文) + # 必须在 register_blueprint 之后, 调一次 sync_instances 把当前已配的 + # SOCKS5 实例拉起来, 再进 loop 周期探活 + _start_sync_loop(app) + try: + instance_manager.sync_instances() + except Exception as e: + log.exception("initial sync_instances failed: %s", e) + return app diff --git a/engine/instances.py b/engine/instances.py index 69b2a03..1ea1c34 100644 --- a/engine/instances.py +++ b/engine/instances.py @@ -159,6 +159,12 @@ class Socks5Server: True 但 server 实际不再 accept。 调用方应据此重启实例, 否则 DB 里 running=true 是谎言。 + + 判定: + - thread 死了 -> 死 (无法恢复) + - server is None -> 死 (还没起来) + - server is_serving() == False -> 死 (socket 已关, asyncio.Server + 不能复用同一个对象重新 serve, 必须新 Socks5Server) """ if not self.running: return False @@ -166,7 +172,6 @@ class Socks5Server: return False if self.server is None: return False - # server.is_serving() 反映 socket 是否在 listen, 比检查 fd 状态靠谱 if not self.server.is_serving(): return False return True diff --git a/tests/e2e_sync_loop.py b/tests/e2e_sync_loop.py new file mode 100644 index 0000000..a80db8e --- /dev/null +++ b/tests/e2e_sync_loop.py @@ -0,0 +1,116 @@ +"""测试 app._start_sync_loop 后台线程能否自动救活死 SOCKS5 实例。 + +只用一次 create_app(), 保证只有一个 InstanceManager + 一个 sync loop。 +让同一个 loop 既负责初次拉起实例, 也负责在实例死后自动救活。 + +流程: + 1. create_app() 启动 sync loop (SM_SYNC_INTERVAL=2s), 此时 DB 无实例 + 2. 往 DB 插入一条 enabled 实例 + 3. 等 loop 周期把它拉起来 (端口 listen) + 4. 模拟死: 关闭 server socket + 5. 等 loop 探测到并自动重启 (端口重新 listen, 且是新 Socks5Server 对象) +""" +import asyncio +import logging +import os +import socket +import sys +import tempfile +import time + +sys.path.insert(0, os.getcwd()) + +_tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False, dir="/tmp") +_tmp_db.close() +os.environ["SM_DB_URI"] = f"sqlite:///{_tmp_db.name}" +os.environ["SM_ADMIN_PASSWORD"] = "test123" +os.environ["SM_SECRET_KEY"] = "test-secret-key" +os.environ["SM_LOG_LEVEL"] = "WARNING" +# 短周期, 加快测试 +os.environ["SM_SYNC_INTERVAL"] = "2" +os.environ["SM_SYNC_LOOP_ENABLE"] = "true" + +from app import create_app +from database import db as _db +from models import Instance + + +def find_free_port(): + s = socket.socket() + s.bind(("127.0.0.1", 0)) + p = s.getsockname()[1] + s.close() + return p + + +def is_listening(port): + try: + s = socket.socket(); s.settimeout(0.5) + s.connect(("127.0.0.1", port)); s.close() + return True + except (ConnectionRefusedError, socket.timeout, OSError): + return False + + +def wait_listening(port, timeout_s=8.0, step=0.3): + t0 = time.time() + while time.time() - t0 < timeout_s: + if is_listening(port): + return True, time.time() - t0 + time.sleep(step) + return False, time.time() - t0 + + +def main(): + logging.basicConfig(level=logging.WARNING, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s") + port = find_free_port() + + # 唯一一次 create_app: 启动 sync loop + 初始 sync_instances (此时 DB 无实例) + flask_app = create_app() + mgr = flask_app.instance_manager # type: ignore[attr-defined] + + # 往 DB 插入 enabled 实例, 让 sync loop 在下一周期拉起它 + with flask_app.app_context(): + inst = Instance( + name="looper", listen_host="127.0.0.1", listen_port=port, + timeout=5, enabled=True, auth_method="none", + bandwidth_down=0, bandwidth_up=0, max_concurrent=10, + ) + _db.session.add(inst) + _db.session.commit() + print(f"[setup] inserted instance port={port}") + + # 等 sync loop 初次拉起实例 + ok, elapsed = wait_listening(port, timeout_s=8.0) + assert ok, f"sync loop 没在 8s 内拉起 port {port}" + print(f"[1] sync loop 初次拉起 OK, port {port} listening (耗时 {elapsed:.1f}s)") + + # 模拟死: 关闭 server socket + srv = mgr._instances["looper"] + srv.loop.call_soon_threadsafe(srv.server.close) + time.sleep(0.6) + assert not is_listening(port), f"关闭后 port {port} 仍在 listen" + print(f"[2] killed port {port}, 等 sync loop 救活...") + + # 等 sync loop 探测并重启 + ok, elapsed = wait_listening(port, timeout_s=8.0) + assert ok, f"sync loop 没在 8s 内救活 port {port}" + print(f"[3] port {port} 被 sync loop 救活 (耗时 {elapsed:.1f}s)") + + # 验证是新的 Socks5Server 对象 (不是复用死掉的那个) + new_srv = mgr._instances["looper"] + assert new_srv is not srv, "sync loop 复用了死掉的 Socks5Server, 没换对象" + print(f"[4] 确认是新 Socks5Server 对象 (旧 id={id(srv)}, 新 id={id(new_srv)})") + + # 清理 + new_srv.stop() + try: os.unlink(_tmp_db.name) + except Exception: pass + print("\nPASS: 后台 sync loop 自动救活死 SOCKS5 实例 (含对象替换)") + return True + + +if __name__ == "__main__": + rc = main() + sys.exit(0 if rc else 1)