fix(logging): keep idle system and plugin log streams alive

This commit is contained in:
Freeman
2026-11-03 23:39:13 +08:00
committed by InfinityPacer
parent 26f066df53
commit c301783bc2
5 changed files with 200 additions and 10 deletions

View File

@@ -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(

View File

@@ -103,12 +103,12 @@ class SystemLogPort(Protocol):
disconnected: Callable[[], Awaitable[bool]],
) -> AsyncIterator[str]:
"""
返回历史尾部及后续新增日志行。
返回历史尾部、后续新增日志行及维持空闲连接的心跳。
:param logfile: 受准入规则约束的日志文件标识
:param length: 首次发送的历史尾部行数
:param disconnected: 异步判断客户端是否断开,返回 True 时停止继续跟随
:return: 先包含历史尾部、随后产生新增日志行的异步迭代器
:return: 日志行的异步迭代器;空字符串表示心跳,传输层应编码为不触发消息事件的 SSE 注释
"""
...

View File

@@ -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(

View File

@@ -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

View File

@@ -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()