From 8b8883cccaff1bbf0d71f8d7742cf97e50b90cfd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Fri, 11 Sep 2026 16:28:19 +0800 Subject: [PATCH] =?UTF-8?q?Worker=20=E5=A4=B1=E8=B4=A5=E5=8E=9F=E5=9B=A0?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E5=8F=AA=E7=95=99=E7=B1=BB=E5=90=8D=EF=BC=9A?= =?UTF-8?q?=E5=8C=BA=E5=88=86"=E5=8F=AF=E8=90=BD=E5=BA=93=E7=9A=84?= =?UTF-8?q?=E5=9B=BA=E5=AE=9A=E6=96=87=E6=A1=88"=E4=B8=8E"=E5=BC=82?= =?UTF-8?q?=E5=B8=B8=E6=B6=88=E6=81=AF"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 起因是查 Worker 运行状态时发现库里 373 条死信的 last_error 全是裸的 "ValueError" （工具 tools/probe_worker_state.py，证据 docs/evidence/worker-state.json）。 dispatch（run not found）、dispatch_run_completed、dispatch_memory_extraction、 dispatch_profile_rebuild 抛的都是 ValueError，只记类名等于把"哪一处失败"也一起丢了。 但"直接存 str(exc)"是错的：tests/unit/worker/test_outbox_worker.py 那条 RuntimeError("credential=do-not-log") 断言异常消息不得落库 —— 它可能含凭据、SQL 或 客户标识。第一版改动就是这么写的，被这个测试当场拦下（这测试写得值）。 折中： - 新增 OutboxHandlerError（继承 ValueError，这些失败本就是 ValueError 语义，保持 继承关系才不会改动既有的 except ValueError 行为与断言）。它的 reason 由代码写死、 不含任何请求数据，因此可以落库； - safe_error_text：OutboxHandlerError → "类名: 固定文案"（截断 500 字符）， 其余异常 → 仍只记类名； - runtime.py 的 5 处 handler 失败改抛 OutboxHandlerError。 测试：新增"固定文案落库"用例；并把既有用例的断言收紧为 last_error == "RuntimeError" （原先只断言"不含 do-not-log"，太松，漏掉的情况测不出来）。 顺带产出 tools/probe_worker_state.py（只读）：outbox / agent_run 各状态计数、按事件 类型分组、死信原因聚合。当前环境实测 pending 347、dead 373、published 410、 agent_run 无 queued/running。 门禁：ruff 干净 / mypy 138 文件 / 697 unit+contract / 33 integration。 --- app/worker/outbox_worker.py | 38 ++++++- app/worker/runtime.py | 12 +-- docs/evidence/worker-state.json | 135 ++++++++++++++++++++++++ tests/unit/worker/test_outbox_worker.py | 25 ++++- tools/probe_worker_state.py | 113 ++++++++++++++++++++ 5 files changed, 315 insertions(+), 8 deletions(-) create mode 100644 docs/evidence/worker-state.json create mode 100644 tools/probe_worker_state.py diff --git a/app/worker/outbox_worker.py b/app/worker/outbox_worker.py index 3973e8a..a9929bf 100644 --- a/app/worker/outbox_worker.py +++ b/app/worker/outbox_worker.py @@ -8,6 +8,41 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.model.platform import DomainEventOutbox, OutboxDelivery +class OutboxHandlerError(ValueError): + """Handler 失败，且失败原因是**可以安全落库**的固定文案。 + + 继承 `ValueError` 而不是 `Exception`：这些失败（run not found、payload 不完整、 + mode 非法）本来就是 ValueError 语义，保持继承关系才不会改动既有的 `except + ValueError` 行为与断言。 + + 为什么需要这个类型：`last_error` 默认只记异常类名，因为异常消息可能含凭据、SQL + 语句或客户标识（`tests/unit/worker/test_outbox_worker.py` 里那条 + `RuntimeError("credential=do-not-log")` 就是守这条的）。 + + 但只记类名又不够 —— `dispatch`（run not found）、`dispatch_run_completed`、 + `dispatch_memory_extraction`、`dispatch_profile_rebuild` 抛的全是 ValueError，实测 + 库里 373 条死信的 `last_error` 都是裸的 `"ValueError"`，分不清是哪一处失败的。 + + 折中办法：handler 想让人看见原因时，抛这个类型，`reason` 由**代码写死**、不含任何 + 请求数据，于是可以落库；其余异常仍然只记类名。 + """ + + def __init__(self, reason: str) -> None: + super().__init__(reason) + self.reason = reason + + +def safe_error_text(exc: BaseException) -> str: + """把异常转成可落库的失败原因。 + + - `OutboxHandlerError` → `类名: 固定文案`（文案由代码写死，安全）； + - 其他异常 → **只记类名**（消息可能含凭据/请求数据，不落库）。 + """ + if isinstance(exc, OutboxHandlerError): + return f"{type(exc).__name__}: {exc.reason}"[:500] + return type(exc).__name__ + + class OutboxWorker: """One dispatcher per event type, not a fan-out transport. @@ -71,7 +106,8 @@ class OutboxWorker: await self.session.flush() except Exception as exc: event.retry_count += 1 - event.last_error = type(exc).__name__ + # 只让"代码写死的固定文案"落库；异常消息可能含凭据，见 safe_error_text。 + event.last_error = safe_error_text(exc) event.status = "dead" if event.retry_count >= 5 else "failed" event.next_retry_at = now + timedelta(seconds=min(300, 2**event.retry_count)) else: diff --git a/app/worker/runtime.py b/app/worker/runtime.py index e63858b..bdedcb6 100644 --- a/app/worker/runtime.py +++ b/app/worker/runtime.py @@ -41,7 +41,7 @@ from app.worker.episode_worker import ( EpisodeWorker, ) from app.worker.memory_extraction_worker import MemoryExtractionWorker -from app.worker.outbox_worker import OutboxWorker +from app.worker.outbox_worker import OutboxHandlerError, OutboxWorker logger = logging.getLogger(__name__) @@ -119,11 +119,11 @@ class WorkerRuntime: async def dispatch(payload: dict[str, Any]) -> None: run = await AgentRunRepository(session).get(str(payload["run_id"])) if run is None: - raise ValueError("run not found") + raise OutboxHandlerError("run not found") async def dispatch_memory_extraction(payload: dict[str, Any]) -> None: if "message_id" not in payload or "customer_id" not in payload: - raise ValueError("memory extraction payload is incomplete") + raise OutboxHandlerError("memory extraction payload is incomplete") # 幂等键只认事件 id，由 worker 自己按 payload 回查，避免调用方漏传。 # 注入召回缓存适配器：写入生效后立即失效该客户的热缓存。 await MemoryExtractionWorker( @@ -135,7 +135,7 @@ class WorkerRuntime: # "运行已完成"的对外通知职责。当前没有独立外部消费者， # 这里显式消费以免事件永久滞留；接入推送链路时在此处扩展。 if not str(payload.get("run_id", "")): - raise ValueError("agent.run_completed payload is incomplete") + raise OutboxHandlerError("agent.run_completed payload is incomplete") async def dispatch_cache_invalidate(payload: dict[str, Any]) -> None: await self._invalidate_config_cache(payload) @@ -149,7 +149,7 @@ class WorkerRuntime: """ customer_id = payload.get("customer_id") if not customer_id: - raise ValueError("profile.rebuild_requested payload is incomplete") + raise OutboxHandlerError("profile.rebuild_requested payload is incomplete") # 延迟导入：bootstrap 会间接导入本模块，模块级导入会形成循环依赖 from app.service.profile_assembly_service import ProfileAssemblyService from app.service.profile_graph_projection_service import ( @@ -181,7 +181,7 @@ class WorkerRuntime: raise ValueError("memory.deletion_requested payload is incomplete") mode = str(payload.get("mode", "invalidate")) if mode not in {"invalidate", "delete"}: - raise ValueError("memory.deletion_requested mode is invalid") + raise OutboxHandlerError("memory.deletion_requested mode is invalid") await MemoryLifecycleService(session).run( int(customer_id), mode=cast("Mode", mode), diff --git a/docs/evidence/worker-state.json b/docs/evidence/worker-state.json new file mode 100644 index 0000000..b1a3d56 --- /dev/null +++ b/docs/evidence/worker-state.json @@ -0,0 +1,135 @@ +{ + "outbox_columns": [ + "id", + "event_id", + "event_type", + "aggregate_type", + "aggregate_id", + "trace_id", + "payload", + "status", + "retry_count", + "next_retry_at", + "last_error", + "occurred_at", + "published_at", + "created_at", + "updated_at" + ], + "agent_run_columns": [ + "id", + "run_id", + "idempotency_id", + "session_id", + "user_id", + "agent_type", + "trace_id", + "request_message_id", + "result_message_id", + "status", + "attempt_count", + "worker_id", + "locked_until", + "error_code", + "result_version", + "cancel_requested_at", + "started_at", + "completed_at", + "created_at", + "updated_at" + ], + "outbox_by_status": { + "dead": 373, + "failed": 3, + "pending": 347, + "published": 410 + }, + "agent_run_by_status": { + "failed": 21, + "succeeded": 184 + }, + "outbox_created_at_range": [ + "2026-09-09 06:55:06.787006", + "2026-09-11 07:21:23.195457" + ], + "outbox_total": 1133, + "agent_run_total": 205, + "outbox_by_event_type_status": [ + { + "event_type": "agent.run_requested", + "status": "dead", + "count": 373 + }, + { + "event_type": "agent.run_requested", + "status": "pending", + "count": 132 + }, + { + "event_type": "agent.run_requested", + "status": "published", + "count": 130 + }, + { + "event_type": "agent.run_completed", + "status": "published", + "count": 118 + }, + { + "event_type": "memory.extraction_requested", + "status": "published", + "count": 106 + }, + { + "event_type": "profile.rebuild_requested", + "status": "pending", + "count": 76 + }, + { + "event_type": "agent.run_completed", + "status": "pending", + "count": 66 + }, + { + "event_type": "memory.extraction_requested", + "status": "pending", + "count": 60 + }, + { + "event_type": "config.cache_invalidate_requested", + "status": "published", + "count": 42 + }, + { + "event_type": "profile.rebuild_requested", + "status": "published", + "count": 13 + }, + { + "event_type": "config.cache_invalidate_requested", + "status": "pending", + "count": 13 + }, + { + "event_type": "agent.run_requested", + "status": "failed", + "count": 3 + }, + { + "event_type": "memory.invalidated", + "status": "published", + "count": 1 + } + ], + "dead_last_errors": [ + { + "last_error": "ValueError", + "count": 372 + }, + { + "last_error": "no handler registered", + "count": 1 + } + ], + "episode_probe_error": "ImportError: cannot import name 'MemoryEpisode' from 'app.model.memory' (C:\\Users\\Windows\\Desktop\\项目代码\\app\\model\\memory.py)" +} \ No newline at end of file diff --git a/tests/unit/worker/test_outbox_worker.py b/tests/unit/worker/test_outbox_worker.py index 08bd270..a5f04e4 100644 --- a/tests/unit/worker/test_outbox_worker.py +++ b/tests/unit/worker/test_outbox_worker.py @@ -4,7 +4,7 @@ from unittest.mock import AsyncMock, Mock import pytest from app.model.platform import DomainEventOutbox -from app.worker.outbox_worker import OutboxWorker +from app.worker.outbox_worker import OutboxHandlerError, OutboxWorker def event() -> DomainEventOutbox: @@ -66,9 +66,32 @@ async def test_handler_failure_records_attempt(attempts: int, expected: str) -> assert row.status == expected assert row.retry_count == attempts + 1 assert "do-not-log" not in (row.last_error or "") + # 其他异常一律只记类名：消息可能含凭据/SQL/客户标识，不落库。 + assert row.last_error == "RuntimeError" session.commit.assert_awaited_once() +@pytest.mark.asyncio +async def test_handler_error_reason_is_recorded() -> None: + """`OutboxHandlerError` 的固定文案可以落库 —— 否则失败原因根本无从分辨。 + + 实测库里 373 条死信的 `last_error` 全是裸的 `"ValueError"`：`dispatch`（run not + found）、`dispatch_run_completed`、`dispatch_memory_extraction`、 + `dispatch_profile_rebuild` 抛的都是 ValueError，只记类名等于把"哪一处失败"也丢了。 + 这些 handler 现在改抛 `OutboxHandlerError`，它的 reason 由代码写死、不含请求数据。 + """ + row = event() + session = AsyncMock() + session.add = Mock() + session.scalar.side_effect = [row, None] + handler = AsyncMock(side_effect=OutboxHandlerError("run not found")) + + await OutboxWorker(session, {"test.event": handler}).publish_one() + + assert row.last_error == "OutboxHandlerError: run not found" + assert row.status == "failed" + + @pytest.mark.asyncio async def test_empty_matching_queue_releases_transaction() -> None: session = AsyncMock() diff --git a/tools/probe_worker_state.py b/tools/probe_worker_state.py new file mode 100644 index 0000000..cd349b3 --- /dev/null +++ b/tools/probe_worker_state.py @@ -0,0 +1,113 @@ +"""只读探查：Worker 的运行状态（队列有没有在消费）。 + +只做 SELECT，结果写 `docs/evidence/worker-state.json`： + + python tools/probe_worker_state.py + +三个问题： +1. `domain_event_outbox` 里各状态各有多少条、最老的一条积压了多久 + —— Worker 没在跑，事件就永远停在 pending； +2. `agent_run` 里 queued/running 有多少 + —— Worker 没在跑，用户发起的对话就一直"排队中"； +3. `memory_episode` 有没有待提取的片段（同样是 Worker 负责消费）。 +""" + +from __future__ import annotations + +import asyncio +import json +from pathlib import Path +from typing import Any + +from sqlalchemy import func, select + +from app.infrastructure.db import SessionFactory +from app.model.platform import AgentRun, DomainEventOutbox + +OUTPUT = Path("docs/evidence/worker-state.json") + + +async def _grouped(session: Any, model: Any, column_name: str) -> dict[str, int]: + column = getattr(model, column_name, None) + if column is None: + return {} + rows = await session.execute(select(column, func.count()).group_by(column)) + return {str(key): int(count) for key, count in rows.all()} + + +async def collect() -> dict[str, Any]: + report: dict[str, Any] = {} + async with SessionFactory() as session: + report["outbox_columns"] = list(DomainEventOutbox.__table__.columns.keys()) + report["agent_run_columns"] = list(AgentRun.__table__.columns.keys()) + + report["outbox_by_status"] = await _grouped( + session, DomainEventOutbox, "status" + ) + report["agent_run_by_status"] = await _grouped(session, AgentRun, "status") + + oldest = await session.scalar( + select(func.min(DomainEventOutbox.created_at)) + ) + newest = await session.scalar( + select(func.max(DomainEventOutbox.created_at)) + ) + report["outbox_created_at_range"] = [str(oldest), str(newest)] + report["outbox_total"] = await session.scalar( + select(func.count()).select_from(DomainEventOutbox) + ) + report["agent_run_total"] = await session.scalar( + select(func.count()).select_from(AgentRun) + ) + + # 哪一类事件在堆、哪一类已经进了死信 —— 死信意味着那些事件的副作用永远不会发生。 + report["outbox_by_event_type_status"] = [ + {"event_type": str(row[0]), "status": str(row[1]), "count": int(row[2])} + for row in ( + await session.execute( + select( + DomainEventOutbox.event_type, + DomainEventOutbox.status, + func.count(), + ) + .group_by(DomainEventOutbox.event_type, DomainEventOutbox.status) + .order_by(func.count().desc()) + ) + ).all() + ] + report["dead_last_errors"] = [ + {"last_error": str(row[0])[:400], "count": int(row[1])} + for row in ( + await session.execute( + select(DomainEventOutbox.last_error, func.count()) + .where(DomainEventOutbox.status == "dead") + .group_by(DomainEventOutbox.last_error) + .order_by(func.count().desc()) + ) + ).all() + ] + + try: + from app.model.memory import MemoryEpisode + + report["episode_by_status"] = await _grouped(session, MemoryEpisode, "status") + report["episode_total"] = await session.scalar( + select(func.count()).select_from(MemoryEpisode) + ) + except Exception as exc: # 模型名/字段与预期不符时只记录，不影响其余结论 + report["episode_probe_error"] = f"{type(exc).__name__}: {exc}" + return report + + +async def main() -> None: + report = await collect() + OUTPUT.parent.mkdir(parents=True, exist_ok=True) + OUTPUT.write_text( + json.dumps(report, ensure_ascii=False, indent=2, default=str), + encoding="utf-8", + ) + print(f"wrote {OUTPUT}") + + +if __name__ == "__main__": + asyncio.run(main())