diff --git a/src/muse/启动.py b/src/muse/启动.py index 9031def..9391e1c 100644 --- a/src/muse/启动.py +++ b/src/muse/启动.py @@ -239,11 +239,19 @@ class 应用装配: def 构建(配置: 应用配置, *, 时钟: 时钟 | None = None, 流程: 流程登记 | None = None) -> 应用装配: """注入按用途绑定的数据库工厂；装配与自检不建立连接。""" + 最大连接 = 配置.连接池.最大连接 + if 最大连接 is None: + 最大连接 = 2 if 配置.运行用途.value == "evaluation" else 4 + 等待秒 = 配置.连接池.等待秒 + if 等待秒 is None: + 等待秒 = 2.0 + 数据库 = 数据库工厂( 配置.数据库, 配置.运行用途, 常驻=True, - 最大连接=2 if 配置.运行用途.value == "evaluation" else 4, + 最大连接=最大连接, + 等待秒=等待秒, ) 目录 = 参与者目录() 登记作品参与者(目录, 探索服务(Path(配置.探索草稿)) if 配置.探索草稿 else None, 解析本书规划引用) diff --git a/src/muse/接入/http/事件订阅.py b/src/muse/接入/http/事件订阅.py index 2aa8840..cd69dd6 100644 --- a/src/muse/接入/http/事件订阅.py +++ b/src/muse/接入/http/事件订阅.py @@ -79,6 +79,7 @@ async def 订阅事件( async def 数据流(): 游标 = max(cursor, last_event_id or 0) + 空轮询次数 = 0 while not await request.is_disconnected(): 结果 = await run_in_threadpool(服务.续接事件, task_id, 游标) 页 = 呈现事件(结果) @@ -100,7 +101,11 @@ async def 订阅事件( return if not 页.items: yield ": keep-alive\n\n" - await asyncio.sleep(1) + 空轮询次数 += 1 + else: + 空轮询次数 = 0 + 等待秒 = 2.0 if (protocol == "bounded-v2" and 空轮询次数 >= 2) else 1.0 + await asyncio.sleep(等待秒) return StreamingResponse( 数据流(), diff --git a/src/muse/配置.py b/src/muse/配置.py index 23e3590..329345b 100644 --- a/src/muse/配置.py +++ b/src/muse/配置.py @@ -9,6 +9,7 @@ from __future__ import annotations +import math import tomllib from dataclasses import dataclass, replace from pathlib import Path @@ -21,6 +22,25 @@ from muse.共享.错误 import 配置错误 _允许的取值方式 = frozenset({"环境变量", "受控存储"}) +@dataclass(frozen=True, slots=True) +class 连接池配置: + """运行连接池参数；独立于凭据引用，不参与数据库目标身份相等比较。""" + + 最大连接: int | None = None + 等待秒: float | None = None + + def __post_init__(self) -> None: + if self.最大连接 is not None: + if type(self.最大连接) is not int or self.最大连接 <= 0 or self.最大连接 > 100: + raise 配置错误("连接池.最大连接 必须是 1 到 100 之间的整数") + if self.等待秒 is not None: + if not isinstance(self.等待秒, (int, float)) or type(self.等待秒) is bool: + raise 配置错误("连接池.等待秒 必须是有效数值") + val = float(self.等待秒) + if math.isnan(val) or math.isinf(val) or not (0.1 <= val <= 60.0): + raise 配置错误("连接池.等待秒 必须在 0.1 到 60.0 之间的有限数值") + + @dataclass(frozen=True, slots=True) class 数据库引用: """数据库连接引用：只记录取值方式与位置，不含连接串明文。""" @@ -83,6 +103,7 @@ class 应用配置: 探索草稿: str | None = None 备份目录: str | None = None 运行装配: 运行装配配置 = 运行装配配置() + 连接池: 连接池配置 = 连接池配置() def __post_init__(self) -> None: if not self.资源发布身份.strip(): @@ -130,6 +151,18 @@ def 解析配置(数据: dict[str, Any]) -> 应用配置: 运行装配 = 运行装配配置(**装配节) except TypeError as exc: raise 配置错误("运行装配配置字段不合法") from exc + 池节 = 数据.get("数据库", {}).get("连接池") + if 池节 is None: + 池节 = 数据.get("连接池") + if 池节 is not None: + if not isinstance(池节, dict): + raise 配置错误("数据库.连接池 必须是映射结构") + try: + 连接池 = 连接池配置(**池节) + except TypeError as exc: + raise 配置错误("数据库.连接池 字段不合法") from exc + else: + 连接池 = 连接池配置() return 应用配置( 数据库=_校验数据库节(数据["数据库"]), 资源发布身份=发布身份, @@ -139,6 +172,7 @@ def 解析配置(数据: dict[str, Any]) -> 应用配置: 探索草稿=数据.get("文件", {}).get("探索草稿"), 备份目录=数据.get("文件", {}).get("备份目录"), 运行装配=运行装配, + 连接池=连接池, ) @@ -187,4 +221,12 @@ def 读取配置(路径: str | Path) -> 应用配置: return 配置 -__all__ = ["数据库引用", "应用配置", "服务配置", "运行装配配置", "解析配置", "读取配置"] +__all__ = [ + "数据库引用", + "连接池配置", + "应用配置", + "服务配置", + "运行装配配置", + "解析配置", + "读取配置", +] diff --git a/tests/单元/test_连接池配置与退避.py b/tests/单元/test_连接池配置与退避.py new file mode 100644 index 0000000..ed76a8c --- /dev/null +++ b/tests/单元/test_连接池配置与退避.py @@ -0,0 +1,217 @@ +"""连接池配置解耦、校验及 SSE 空轮询退避机制验证。""" + +import asyncio + +import pytest + +from muse.共享.错误 import 配置错误 +from muse.配置 import 数据库引用, 解析配置, 连接池配置 + + +@pytest.mark.case_id("TC-POOL-CONFIG-VALIDATION") +def test_连接池配置字段强校验(): + # 正常值 + cfg = 连接池配置(最大连接=8, 等待秒=5.0) + assert cfg.最大连接 == 8 + assert cfg.等待秒 == 5.0 + + # 最大连接不接受非 int、bool、越界 + with pytest.raises(配置错误, match="连接池.最大连接"): + 连接池配置(最大连接=True) # type: ignore + with pytest.raises(配置错误, match="连接池.最大连接"): + 连接池配置(最大连接=0) + with pytest.raises(配置错误, match="连接池.最大连接"): + 连接池配置(最大连接=101) + with pytest.raises(配置错误, match="连接池.最大连接"): + 连接池配置(最大连接="8") # type: ignore + + # 等待秒不接受 bool、NaN、Inf、越界 + with pytest.raises(配置错误, match="连接池.等待秒"): + 连接池配置(等待秒=False) # type: ignore + with pytest.raises(配置错误, match="连接池.等待秒"): + 连接池配置(等待秒=float("nan")) + with pytest.raises(配置错误, match="连接池.等待秒"): + 连接池配置(等待秒=float("inf")) + with pytest.raises(配置错误, match="连接池.等待秒"): + 连接池配置(等待秒=0.05) + with pytest.raises(配置错误, match="连接池.等待秒"): + 连接池配置(等待秒=61.0) + + +@pytest.mark.case_id("TC-POOL-CONFIG-DECOUPLING") +def test_连接池配置不改变数据库引用相等性(): + ref1 = 数据库引用("环境变量", "DB_URL") + ref2 = 数据库引用("环境变量", "DB_URL") + assert ref1 == ref2 + + # 配置解析产出独立连接池配置 + cfg = 解析配置( + { + "数据库": { + "取值方式": "环境变量", + "位置": "DB_URL", + "连接池": {"最大连接": 10, "等待秒": 3.0}, + }, + "资源": {"发布身份": "identity-1"}, + } + ) + assert cfg.数据库 == ref1 + assert cfg.连接池.最大连接 == 10 + assert cfg.连接池.等待秒 == 3.0 + + +@pytest.mark.case_id("TC-POOL-CONFIG-STARTUP-INTEGRATION") +def test_启动构建消费连接池配置(monkeypatch): + from muse.启动 import 构建 + + # 1. 显式配置生效 + cfg = 解析配置( + { + "数据库": { + "取值方式": "环境变量", + "位置": "DB_URL", + "连接池": {"最大连接": 12, "等待秒": 4.5}, + }, + "资源": {"发布身份": "identity-1"}, + "运行": {"用途": "production"}, + } + ) + 装配 = 构建(cfg) + assert 装配.数据库.最大连接 == 12 + assert 装配.数据库.等待秒 == 4.5 + + # 2. 未配置时回退默认（生产 4 / 2.0，评测 2 / 2.0） + cfg_prod = 解析配置( + { + "数据库": {"取值方式": "环境变量", "位置": "DB_URL"}, + "资源": {"发布身份": "identity-1"}, + "运行": {"用途": "production"}, + } + ) + 装配_prod = 构建(cfg_prod) + assert 装配_prod.数据库.最大连接 == 4 + assert 装配_prod.数据库.等待秒 == 2.0 + + cfg_eval = 解析配置( + { + "数据库": {"取值方式": "环境变量", "位置": "DB_URL"}, + "资源": {"发布身份": "identity-1"}, + "运行": {"用途": "evaluation"}, + } + ) + 装配_eval = 构建(cfg_eval) + assert 装配_eval.数据库.最大连接 == 2 + assert 装配_eval.数据库.等待秒 == 2.0 + + +@pytest.mark.case_id("TC-SSE-BACKOFF-ADAPTIVE") +def test_SSE空轮询退避机制(monkeypatch): + from types import SimpleNamespace + + from muse.任务运行.模型 import 事件类型, 事件续接, 运行事件 + from muse.接入.http import 事件订阅 + + def 构造事件(seq): + from datetime import UTC, datetime + + return 运行事件( + f"event-{seq}", + "task", + None, + None, + seq, + 事件类型.步骤完成, + datetime(2026, 9, 17, tzinfo=UTC), + 1, + {"step": f"step-{seq}"}, + ) + + class 模拟事件服务: + def __init__(self, 页集): + self.页集 = 页集 + + def 续接事件(self, task_id, cursor, 数量=100): + return self.页集.get(cursor, 事件续接((), False, cursor, cursor, None)) + + def 运行订阅(protocol, 页集, 循环次数=4): + 服务 = 模拟事件服务(页集) + monkeypatch.setattr(事件订阅, "取得任务服务", lambda _: 服务) + monkeypatch.setattr(事件订阅, "读取作者任务", lambda req, tid, wid=None: None) + 等待记录 = [] + + async def 记录等待(秒): + 等待记录.append(秒) + + monkeypatch.setattr(事件订阅.asyncio, "sleep", 记录等待) + + 检查次数 = [0] + + async def 已断开(): + 检查次数[0] += 1 + return 检查次数[0] > 循环次数 + + async def 执行(): + 响应 = await 事件订阅.订阅事件( + "task-1", + SimpleNamespace(is_disconnected=已断开), + "author", + cursor=0, + last_event_id=None, + protocol=protocol, + ) + return [块 async for 块 in 响应.body_iterator] + + asyncio.run(执行()) + return 等待记录 + + # 1. bounded-v2: 连续空轮询从 1.0s 退避到 2.0s + 等待_v2 = 运行订阅("bounded-v2", {}, 循环次数=4) + assert 等待_v2 == [1.0, 2.0, 2.0, 2.0] + + # 2. legacy: 始终保持 1.0s + 等待_legacy = 运行订阅("legacy", {}, 循环次数=4) + assert 等待_legacy == [1.0, 1.0, 1.0, 1.0] + + # 3. bounded-v2: 退避后有新事件到达，立即重置回 1.0s + class 动态服务: + def __init__(self): + self.count = 0 + + def 续接事件(self, task_id, cursor, 数量=100): + self.count += 1 + if self.count == 3: + return 事件续接((构造事件(1),), False, 1, 1, None) + return 事件续接((), False, cursor, cursor, None) + + 服务动态 = 动态服务() + monkeypatch.setattr(事件订阅, "取得任务服务", lambda _: 服务动态) + monkeypatch.setattr(事件订阅, "读取作者任务", lambda req, tid, wid=None: None) + 动态等待 = [] + + async def 记录动态等待(秒): + 动态等待.append(秒) + + monkeypatch.setattr(事件订阅.asyncio, "sleep", 记录动态等待) + 检查 = [0] + + async def 已断开_动态(): + 检查[0] += 1 + return 检查[0] > 4 + + async def 执行动态(): + 响应 = await 事件订阅.订阅事件( + "task-1", + SimpleNamespace(is_disconnected=已断开_动态), + "author", + cursor=0, + last_event_id=None, + protocol="bounded-v2", + ) + return [块 async for 块 in 响应.body_iterator] + + asyncio.run(执行动态()) + # 轮询1: 空 (等待1.0) + # 轮询2: 空 (等待2.0) + # 轮询3: 有事件 (重置为0，等待1.0) + # 轮询4: 空 (等待1.0) + assert 动态等待 == [1.0, 2.0, 1.0, 1.0] diff --git a/tests/用例清单.json b/tests/用例清单.json index c9e5732..1cc1b4d 100644 --- a/tests/用例清单.json +++ b/tests/用例清单.json @@ -35588,6 +35588,72 @@ ], "markers": [] }, + { + "case_id": "TC-POOL-CONFIG-DECOUPLING", + "file": "tests/单元/test_连接池配置与退避.py", + "symbol": "test_连接池配置不改变数据库引用相等性", + "parameter_ids": [], + "node_ids": [ + "tests/单元/test_连接池配置与退避.py::test_连接池配置不改变数据库引用相等性" + ], + "fixtures": [ + "request", + "测试资源接缝", + "源码资源", + "离线防护" + ], + "markers": [] + }, + { + "case_id": "TC-POOL-CONFIG-STARTUP-INTEGRATION", + "file": "tests/单元/test_连接池配置与退避.py", + "symbol": "test_启动构建消费连接池配置", + "parameter_ids": [], + "node_ids": [ + "tests/单元/test_连接池配置与退避.py::test_启动构建消费连接池配置" + ], + "fixtures": [ + "monkeypatch", + "request", + "测试资源接缝", + "源码资源", + "离线防护" + ], + "markers": [] + }, + { + "case_id": "TC-POOL-CONFIG-VALIDATION", + "file": "tests/单元/test_连接池配置与退避.py", + "symbol": "test_连接池配置字段强校验", + "parameter_ids": [], + "node_ids": [ + "tests/单元/test_连接池配置与退避.py::test_连接池配置字段强校验" + ], + "fixtures": [ + "request", + "测试资源接缝", + "源码资源", + "离线防护" + ], + "markers": [] + }, + { + "case_id": "TC-SSE-BACKOFF-ADAPTIVE", + "file": "tests/单元/test_连接池配置与退避.py", + "symbol": "test_SSE空轮询退避机制", + "parameter_ids": [], + "node_ids": [ + "tests/单元/test_连接池配置与退避.py::test_SSE空轮询退避机制" + ], + "fixtures": [ + "monkeypatch", + "request", + "测试资源接缝", + "源码资源", + "离线防护" + ], + "markers": [] + }, { "case_id": "TC-a089d73880d8", "environment": "隔离PG、实际S02与合成HTTP；不认证外部模型文学收益",