harden(observe): 防重复 start 泄漏 client + 关停丢余量注释 + 补 stop/失败存活测试

评审建议(Task4 code-quality):
- start 已启动则不重复建 client(避免泄漏旧连接池)
- 关停只发一批、超出丢弃,补注释说明 best-effort
- 补测: _run_loop 失败后存活 / stop drain+发最后一批+关 client

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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] = []

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 已关闭并置空