From a5f63cb53c6447bf5641685d7e9bbf2071f1a034 Mon Sep 17 00:00:00 2001 From: guke Date: Mon, 6 Jul 2026 12:11:16 +0800 Subject: [PATCH] =?UTF-8?q?harden(observe):=20=E9=98=B2=E9=87=8D=E5=A4=8D?= =?UTF-8?q?=20start=20=E6=B3=84=E6=BC=8F=20client=20+=20=E5=85=B3=E5=81=9C?= =?UTF-8?q?=E4=B8=A2=E4=BD=99=E9=87=8F=E6=B3=A8=E9=87=8A=20+=20=E8=A1=A5?= =?UTF-8?q?=20stop/=E5=A4=B1=E8=B4=A5=E5=AD=98=E6=B4=BB=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 评审建议(Task4 code-quality): - start 已启动则不重复建 client(避免泄漏旧连接池) - 关停只发一批、超出丢弃,补注释说明 best-effort - 补测: _run_loop 失败后存活 / stop drain+发最后一批+关 client Co-Authored-By: Claude Opus 4.8 (1M context) --- app/core/observe_worker.py | 9 ++++-- tests/test_observe.py | 65 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/app/core/observe_worker.py b/app/core/observe_worker.py index c037467..ba44a6e 100644 --- a/app/core/observe_worker.py +++ b/app/core/observe_worker.py @@ -74,10 +74,14 @@ async def _run_loop(client: httpx.AsyncClient) -> None: def start_observe_worker() -> asyncio.Task | None: - """启动上报 worker。未配置观测 → 返回 None(no-op)。""" + """启动上报 worker。未配置观测 → 返回 None(no-op)。约定每进程只调一次(lifespan)。""" global _client if not settings.observe_configured: return None + if _client is not None: + # 约定 start 每进程只调一次;已启动则不重复建 client(避免泄漏旧连接池)。 + logger.warning("observe worker already started; ignoring duplicate start") + return None _client = httpx.AsyncClient( base_url=settings.OBSERVE_ENDPOINT, auth=(settings.OBSERVE_USER, settings.OBSERVE_PASSWORD), @@ -101,7 +105,8 @@ async def stop_observe_worker(task: asyncio.Task | None) -> None: with contextlib.suppress(asyncio.CancelledError): await task if _client is not None: - # worker 已停,安全 drain 剩余并 best-effort 发最后一批(短超时,不卡关停)。 + # worker 已停,安全 drain 剩余并 best-effort 发最后一批(短超时,不卡关停); + # 超过一批(BATCH_MAX)的剩余直接丢,不做多轮 flush(best-effort,关停从速)。 try: queue = get_queue() final: list[dict] = [] diff --git a/tests/test_observe.py b/tests/test_observe.py index ed18a3f..0816c7a 100644 --- a/tests/test_observe.py +++ b/tests/test_observe.py @@ -160,3 +160,68 @@ async def test_post_batch_hits_json_ingest_url(monkeypatch): def test_start_observe_worker_noop_when_not_configured(monkeypatch): monkeypatch.setattr(settings, "OBSERVE_ENABLED", False) assert observe_worker.start_observe_worker() is None + + +async def test_run_loop_survives_post_failure(monkeypatch): + """_post_batch 抛异常时,loop 不崩溃、继续处理后续批次(best-effort 契约)。""" + q = asyncio.Queue(maxsize=100) + monkeypatch.setattr(observe, "_queue", q) + monkeypatch.setattr(settings, "OBSERVE_FLUSH_INTERVAL_SEC", 0.02) + monkeypatch.setattr(settings, "OBSERVE_BATCH_MAX", 200) + seen: list[list[int]] = [] + + async def boom(client, batch): + seen.append([e["n"] for e in batch]) + raise RuntimeError("boom") + + monkeypatch.setattr(observe_worker, "_post_batch", boom) + + q.put_nowait({"n": 1}) + task = asyncio.create_task(observe_worker._run_loop(None)) + try: + for _ in range(50): # 轮询直到第 1 批被处理(失败),最多等 0.5s + await asyncio.sleep(0.01) + if seen: + break + q.put_nowait({"n": 2}) + for _ in range(50): # 第 2 批被处理 → 证明失败后 loop 仍存活 + await asyncio.sleep(0.01) + if len(seen) >= 2: + break + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + assert seen == [[1], [2]] + + +async def test_stop_flushes_remaining_and_closes_client(monkeypatch): + """stop:cancel 后把剩余事件 best-effort 发出最后一批,并关闭 + 置空 client。""" + q = asyncio.Queue(maxsize=100) + monkeypatch.setattr(observe, "_queue", q) + monkeypatch.setattr(settings, "OBSERVE_ORG", "default") + monkeypatch.setattr(settings, "OBSERVE_STREAM", "app_requests") + monkeypatch.setattr(settings, "OBSERVE_BATCH_MAX", 200) + q.put_nowait({"n": 1}) + q.put_nowait({"n": 2}) + posted: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + posted["body"] = request.content + return httpx.Response(200, json={"code": 200}) + + client = httpx.AsyncClient( + base_url="http://oo", transport=httpx.MockTransport(handler) + ) + monkeypatch.setattr(observe_worker, "_client", client) + + async def _noop() -> None: + return None + + task = asyncio.create_task(_noop()) + await observe_worker.stop_observe_worker(task) + + assert b'"n"' in posted["body"] # 关停时把剩余事件发了出去 + assert observe_worker._client is None # client 已关闭并置空