From c301783bc283a4dfc1f613c807f9b50271bc39d9 Mon Sep 17 00:00:00 2001 From: Freeman Date: Tue, 3 Nov 2026 23:39:13 +0800 Subject: [PATCH] fix(logging): keep idle system and plugin log streams alive --- app/api/endpoints/system.py | 10 +- app/application/system.py | 4 +- app/startup/composition/system.py | 14 +- tests/test_agent_realtime_log.py | 4 +- tests/test_system_logging_keepalive.py | 178 +++++++++++++++++++++++++ 5 files changed, 200 insertions(+), 10 deletions(-) create mode 100644 tests/test_system_logging_keepalive.py diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py index 59859a255..aad715778 100644 --- a/app/api/endpoints/system.py +++ b/app/api/endpoints/system.py @@ -836,18 +836,22 @@ async def _get_logging_impl( async def event_generator() -> AsyncIterator[str]: """ - 仅将原始日志行包装为 SSE 的 data 消息。 + 原始日志行编码为 data 消息，空字符串心跳编码为不触发消息事件的 SSE 注释。 """ try: async for line in source: - yield f"data: {line}\n\n" + yield f"data: {line}\n\n" if line else ": keepalive\n\n" except asyncio.CancelledError: return except Exception as error: logger.error("日志读取异常: %s", error) yield f"data: 日志读取异常: {error}\n\n" - return StreamingResponse(event_generator(), media_type="text/event-stream") + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}, + ) @router.get( diff --git a/app/application/system.py b/app/application/system.py index 1885b68f0..52ebff317 100644 --- a/app/application/system.py +++ b/app/application/system.py @@ -103,12 +103,12 @@ class SystemLogPort(Protocol): disconnected: Callable[[], Awaitable[bool]], ) -> AsyncIterator[str]: """ - 返回历史尾部及后续新增日志行。 + 返回历史尾部、后续新增日志行及维持空闲连接的心跳。 :param logfile: 受准入规则约束的日志文件标识 :param length: 首次发送的历史尾部行数 :param disconnected: 异步判断客户端是否断开，返回 True 时停止继续跟随 - :return: 先包含历史尾部、随后产生新增日志行的异步迭代器 + :return: 日志行的异步迭代器；空字符串表示心跳，传输层应编码为不触发消息事件的 SSE 注释 """ ... diff --git a/app/startup/composition/system.py b/app/startup/composition/system.py index baaeefd6a..8f1c9ccae 100644 --- a/app/startup/composition/system.py +++ b/app/startup/composition/system.py @@ -86,7 +86,12 @@ class _FileLogAdapter(SystemLogPort): disconnected: Callable[[], Awaitable[bool]], ) -> AsyncIterator[str]: """ - 先返回历史尾部，再轮询并返回新增日志行。 + 先返回历史尾部，再轮询并返回新增日志行及空字符串心跳。 + + :param logfile: 日志根内受准入规则保护的日志文件标识 + :param length: 首次发送的历史尾部行数 + :param disconnected: 异步判断客户端是否断开，返回 True 时停止跟随 + :return: 日志行与空字符串心跳的异步迭代器；心跳由传输层编码为 SSE 注释 """ if re.fullmatch(r"subscriptions/[1-9][0-9]*\.log", logfile): path = Path(self._settings.get("LOG_PATH")) / logfile @@ -123,7 +128,9 @@ class _FileLogAdapter(SystemLogPort): disconnected: Callable[[], Awaitable[bool]], ) -> AsyncIterator[str]: """ - 有界读取尾部及完整新增行；重开文件以追踪滚动，并等待订阅首次产生日志。 + 有界读取尾部及完整新增行，每轮提供空字符串心跳以维持空闲连接。 + + 重开文件以追踪滚动，并等待订阅首次产生日志；心跳不改变文件游标。 """ cursor: tuple[int, int] | None = None initial = True @@ -133,8 +140,7 @@ class _FileLogAdapter(SystemLogPort): initial = False for line in lines: yield line - if path.parent.name == "subscriptions": - yield "" + yield "" await asyncio.sleep(0.5) def _log_chunk( diff --git a/tests/test_agent_realtime_log.py b/tests/test_agent_realtime_log.py index 0418dd7d0..f5416011e 100644 --- a/tests/test_agent_realtime_log.py +++ b/tests/test_agent_realtime_log.py @@ -31,9 +31,11 @@ async def test_file_log_follow_drains_all_lines_from_one_append(tmp_path: Path) async def collect_lines() -> None: """ - 收集同一次写入产生的三行日志。 + 忽略空字符串心跳，只收集同一次写入产生的三行日志。 """ async for line in stream: + if not line: + continue collected.append(line) if len(collected) == 3: break diff --git a/tests/test_system_logging_keepalive.py b/tests/test_system_logging_keepalive.py new file mode 100644 index 000000000..d6b1c50a4 --- /dev/null +++ b/tests/test_system_logging_keepalive.py @@ -0,0 +1,178 @@ +""" +主程序与插件日志流的空闲心跳、游标和 SSE 编码回归。 +""" + +import asyncio +from collections.abc import AsyncIterator, Awaitable, Callable +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from app.api.endpoints import system +from app.startup.composition import system as composition_system +from app.startup.composition.system import _FileLogAdapter + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "logfile", + ["moviepilot.log", "plugins/embymetarefresh.log", "subscriptions/1.log"], +) +async def test_file_log_follow_keeps_idle_connection_without_replaying_history( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + logfile: str, +) -> None: + """ + 三类日志在空闲轮询时都产生心跳，历史尾部和随后追加的完整行各输出一次。 + """ + path = tmp_path / logfile + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("older\nhistory\n", encoding="utf-8") + adapter = _FileLogAdapter(SimpleNamespace(get=lambda _key: tmp_path)) + monkeypatch.setattr( + composition_system, + "runtime_stop_state", + SimpleNamespace(is_system_stopped=False), + ) + monkeypatch.setattr(composition_system.asyncio, "sleep", AsyncMock()) + polls = 0 + + async def disconnected() -> bool: + """ + 第二轮保持空闲，第三轮追加一行，第四轮断开以有界消费真实文件流。 + """ + nonlocal polls + polls += 1 + if polls == 3: + with path.open("a", encoding="utf-8") as file: + file.write("appended\n") + return polls > 3 + + source = await adapter.follow(logfile, 1, disconnected) + assert [line async for line in source] == [ + "history", + "", + "", + "appended", + "", + ] + assert polls == 4 + + +@pytest.mark.asyncio +async def test_file_log_follow_stops_before_reading_when_disconnected( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """ + 已断开的连接不推送历史或心跳，日志流自然结束。 + """ + (tmp_path / "moviepilot.log").write_text("history\n", encoding="utf-8") + adapter = _FileLogAdapter(SimpleNamespace(get=lambda _key: tmp_path)) + monkeypatch.setattr( + composition_system, + "runtime_stop_state", + SimpleNamespace(is_system_stopped=False), + ) + source = await adapter.follow( + "moviepilot.log", 50, AsyncMock(return_value=True), + ) + assert [line async for line in source] == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "logfile", + ["moviepilot.log", "plugins/embymetarefresh.log"], +) +async def test_system_log_sse_encodes_heartbeat_as_comment(logfile: str) -> None: + """ + 心跳采用 SSE 注释且禁用缓存和代理缓冲，不向浏览器日志列表推送空 data 消息。 + """ + disconnected = AsyncMock(return_value=False) + + async def source() -> AsyncIterator[str]: + """ + 在历史行前后提供心跳，保留日志行原文。 + """ + yield "" + yield "【INFO】历史日志" + yield "" + + async def follow( + path: str, + length: int, + disconnected_callback: Callable[[], Awaitable[bool]], + ) -> AsyncIterator[str]: + """ + 提供遵循系统日志端口的有限迭代器，并检查请求参数沿用。 + """ + assert path == logfile + assert length == 50 + assert disconnected_callback is disconnected + return source() + + response = await system._get_logging_impl( + SimpleNamespace(is_disconnected=disconnected), + logfile=logfile, + runtime=SimpleNamespace(system=SimpleNamespace(follow_log=follow)), + ) + assert [chunk async for chunk in response.body_iterator] == [ + ": keepalive\n\n", + "data: 【INFO】历史日志\n\n", + ": keepalive\n\n", + ] + assert response.headers["Cache-Control"] == "no-store" + assert response.headers["X-Accel-Buffering"] == "no" + assert response.headers["Content-Type"].startswith("text/event-stream") + + +@pytest.mark.asyncio +async def test_system_log_sse_cancellation_ends_without_error_event() -> None: + """ + 心跳后的客户端取消正常结束 SSE，不添加异常日志事件。 + """ + async def source() -> AsyncIterator[str]: + """ + 模拟心跳已发送后连接取消。 + """ + yield "" + raise asyncio.CancelledError + + async def follow(*_args: object) -> AsyncIterator[str]: + """ + 返回会被取消的日志迭代器。 + """ + return source() + + response = await system._get_logging_impl( + SimpleNamespace(is_disconnected=AsyncMock()), + runtime=SimpleNamespace(system=SimpleNamespace(follow_log=follow)), + ) + assert [chunk async for chunk in response.body_iterator] == [ + ": keepalive\n\n", + ] + + +@pytest.mark.asyncio +async def test_system_log_full_text_does_not_include_heartbeat() -> None: + """ + 全文下载保持纯文本路径，不调用实时日志流或添加 SSE 心跳。 + """ + service = SimpleNamespace( + read_log=AsyncMock(return_value="second\nfirst"), + follow_log=AsyncMock(), + ) + response = await system._get_logging_impl( + SimpleNamespace(is_disconnected=AsyncMock()), + length=-1, + logfile="plugins/embymetarefresh.log", + runtime=SimpleNamespace(system=service), + ) + assert response.body == b"second\nfirst" + assert response.headers["Content-Type"].startswith("text/plain") + service.read_log.assert_awaited_once_with("plugins/embymetarefresh.log") + service.follow_log.assert_not_awaited()