diff --git a/app/adapters/network/http.py b/app/adapters/network/http.py index b821aef39..5fa000192 100644 --- a/app/adapters/network/http.py +++ b/app/adapters/network/http.py @@ -17,6 +17,8 @@ from requests import Response, Session from urllib3.exceptions import InsecureRequestWarning from urllib.parse import unquote, quote +from app.runtime.correlation import with_correlation_header + urllib3.disable_warnings(InsecureRequestWarning) @@ -418,7 +420,9 @@ class RequestUtils: req_method = requests.request else: req_method = self._session.request - kwargs.setdefault("headers", self._headers) + kwargs["headers"] = with_correlation_header( + kwargs.get("headers", self._headers) + ) kwargs.setdefault("cookies", self._cookies) kwargs.setdefault("proxies", self._proxies) kwargs.setdefault("timeout", self._timeout) @@ -1195,7 +1199,9 @@ class AsyncRequestUtils: """ 执行实际的异步请求 """ - kwargs.setdefault("headers", self._headers) + kwargs["headers"] = with_correlation_header( + kwargs.get("headers", self._headers) + ) # 共享池下 client 自带默认 timeout，这里用每请求 timeout 覆盖以尊重实例配置 kwargs.setdefault("timeout", self._timeout) # Cookie 在 request() 入口已按 path 处理： @@ -1329,7 +1335,9 @@ class AsyncRequestUtils: :return: 上下文管理器，进入后 yield httpx.Response（出错时 yield None） """ cookies_dict: Optional[dict] = self._cookies if isinstance(self._cookies, dict) else None - kwargs.setdefault("headers", self._headers) + kwargs["headers"] = with_correlation_header( + kwargs.get("headers", self._headers) + ) # 与 _make_request 保持一致：复用 keep-alive 时偶遇对端 FIN 的连接， # 流式 GET 是幂等的，单次重试即可 diff --git a/app/adapters/web/correlation.py b/app/adapters/web/correlation.py new file mode 100644 index 000000000..76a74be26 --- /dev/null +++ b/app/adapters/web/correlation.py @@ -0,0 +1,44 @@ +"""HTTP 请求关联 ID 的 ASGI 适配器。""" + +from __future__ import annotations + +from typing import Any + +from app.runtime.correlation import ( + CORRELATION_ID_HEADER, + correlation_scope, + normalize_correlation_id, +) + + +class CorrelationIdMiddleware: + """验证入口 ID、绑定请求上下文并把同一 ID 写回响应。""" + + def __init__(self, app: Any) -> None: + """保存下游 ASGI 应用。""" + self._app = app + + async def __call__(self, scope: dict, receive: Any, send: Any) -> None: + """只治理 HTTP scope，并让绑定覆盖完整流式响应生命周期。""" + if scope.get("type") != "http": + await self._app(scope, receive, send) + return + raw_headers = dict(scope.get("headers") or []) + candidate = raw_headers.get(CORRELATION_ID_HEADER.lower().encode("ascii")) + correlation_id = normalize_correlation_id( + candidate.decode("ascii", errors="ignore") if candidate else None + ) + scope.setdefault("state", {})["request_id"] = correlation_id + + async def send_with_correlation(message: dict) -> None: + """在响应开始帧中覆盖为当前请求的安全关联 ID。""" + if message.get("type") == "http.response.start": + headers = list(message.get("headers") or []) + header_name = CORRELATION_ID_HEADER.lower().encode("ascii") + headers = [item for item in headers if item[0].lower() != header_name] + headers.append((header_name, correlation_id.encode("ascii"))) + message["headers"] = headers + await send(message) + + with correlation_scope(correlation_id): + await self._app(scope, receive, send_with_correlation) diff --git a/app/factory.py b/app/factory.py index 8993c11ae..4ad4f09ae 100644 --- a/app/factory.py +++ b/app/factory.py @@ -8,6 +8,7 @@ from fastapi.responses import JSONResponse from starlette.exceptions import HTTPException from app.api.response import ResponseAPIRoute +from app.adapters.web.correlation import CorrelationIdMiddleware from app.adapters.web.plugin.routes import FastAPIDynamicRouteRegistry from app.adapters.web.health import install_health_routes from app.application.plugin.routes import configure_plugin_routes @@ -19,8 +20,9 @@ from app.adapters.web.security.access import ( from app.application.security.token import create_access_token, decode_access_token from app.runtime.extensions.plugin_manager import PluginManager from app.runtime.config import settings +from app.runtime.correlation import get_correlation_id from app.runtime.localization import LocaleHelper -from app.runtime.log import logger +from app.runtime.log import configure_correlation_id_provider, logger from app.schemas.openai import ( AnthropicErrorDetail, AnthropicErrorResponse, @@ -291,6 +293,7 @@ def create_app() -> FastAPI: """ 创建并配置 FastAPI 应用实例。 """ + configure_correlation_id_provider(get_correlation_id) _app = FastAPI( title=settings.PROJECT_NAME, version=APP_VERSION, @@ -317,6 +320,7 @@ def create_app() -> FastAPI: allow_methods=["*"], allow_headers=["*"], ) + _app.add_middleware(CorrelationIdMiddleware) @_app.middleware("http") async def locale_context_middleware( diff --git a/app/runtime/correlation.py b/app/runtime/correlation.py new file mode 100644 index 000000000..6d74895d6 --- /dev/null +++ b/app/runtime/correlation.py @@ -0,0 +1,74 @@ +"""请求与后台工作共用的关联 ID 上下文。""" + +from __future__ import annotations + +import re +import uuid +from contextlib import contextmanager +from contextvars import ContextVar, Token +from typing import Any, Callable, Iterator, Mapping + +CORRELATION_ID_HEADER = "X-Request-ID" +MAX_CORRELATION_ID_LENGTH = 64 +_VALID_CORRELATION_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,63}$") +_CURRENT_CORRELATION_ID: ContextVar[str | None] = ContextVar( + "moviepilot_correlation_id", + default=None, +) + + +def normalize_correlation_id(candidate: str | None) -> str: + """接受安全的调用方 ID；非法、超长或缺失值均替换为随机 ID。""" + if candidate and _VALID_CORRELATION_ID.fullmatch(candidate): + return candidate + return uuid.uuid4().hex + + +def get_correlation_id(default: str | None = None) -> str | None: + """返回当前执行上下文中的关联 ID。""" + return _CURRENT_CORRELATION_ID.get() or default + + +def set_correlation_id(correlation_id: str) -> Token[str | None]: + """设置当前关联 ID，并返回供调用方精确恢复的 token。""" + return _CURRENT_CORRELATION_ID.set(correlation_id) + + +def reset_correlation_id(token: Token[str | None]) -> None: + """恢复设置关联 ID 之前的上下文。""" + _CURRENT_CORRELATION_ID.reset(token) + + +@contextmanager +def correlation_scope(correlation_id: str | None) -> Iterator[str | None]: + """在当前同步或异步任务作用域内绑定并自动恢复关联 ID。""" + if correlation_id is None: + yield None + return + token = set_correlation_id(correlation_id) + try: + yield correlation_id + finally: + reset_correlation_id(token) + + +def with_correlation_header(headers: Mapping[str, str] | None) -> dict[str, str]: + """复制请求头并在调用方未显式指定时加入当前关联 ID。""" + result = dict(headers or {}) + if any(key.lower() == CORRELATION_ID_HEADER.lower() for key in result): + return result + correlation_id = get_correlation_id() + if correlation_id: + result[CORRELATION_ID_HEADER] = correlation_id + return result + + +def call_with_correlation( + correlation_id: str | None, + func: Callable[..., Any], + args: tuple[Any, ...], + kwargs: dict[str, Any], +) -> Any: + """从可序列化参数恢复关联 ID 后调用函数，供子进程入口使用。""" + with correlation_scope(correlation_id): + return func(*args, **kwargs) diff --git a/app/runtime/event/dispatch.py b/app/runtime/event/dispatch.py index 8c2829966..6251c148b 100644 --- a/app/runtime/event/dispatch.py +++ b/app/runtime/event/dispatch.py @@ -12,6 +12,7 @@ from app.runtime.event.binding import EventBindingResolver from app.runtime.event.registry import EventRegistry from app.runtime.execution import run_in_threadpool from app.runtime.log import logger +from app.runtime.correlation import correlation_scope from app.schemas.types import EventType @@ -114,6 +115,7 @@ class EventDispatcher: event_type=event.event_type, event_data=event_data, priority=event.priority, + correlation_id=event.correlation_id, ) if inspect.iscoroutinefunction(handler): asyncio.run_coroutine_threadsafe( @@ -143,16 +145,17 @@ class EventDispatcher: if not resolved: return method, binding, class_name, method_name = resolved - try: - method(event) - except Exception as err: - self._error_handler( - event=event, - module_name=binding.owner_name, - class_name=class_name, - method_name=method_name, - e=err, - ) + with correlation_scope(event.correlation_id): + try: + method(event) + except Exception as err: + self._error_handler( + event=event, + module_name=binding.owner_name, + class_name=class_name, + method_name=method_name, + e=err, + ) async def invoke_async(self, handler: Callable, event: Any) -> None: """解析实例绑定，并按处理器类型选择协程、线程池或同步调用。""" @@ -160,21 +163,22 @@ class EventDispatcher: if not resolved: return method, binding, class_name, method_name = resolved - try: - if inspect.iscoroutinefunction(method): - await method(event) - elif binding.run_sync_in_threadpool or not class_name: - await run_in_threadpool(method, event) - else: - method(event) - except Exception as err: - self._error_handler( - event=event, - module_name=binding.owner_name, - class_name=class_name, - method_name=method_name, - e=err, - ) + with correlation_scope(event.correlation_id): + try: + if inspect.iscoroutinefunction(method): + await method(event) + elif binding.run_sync_in_threadpool or not class_name: + await run_in_threadpool(method, event) + else: + method(event) + except Exception as err: + self._error_handler( + event=event, + module_name=binding.owner_name, + class_name=class_name, + method_name=method_name, + e=err, + ) @staticmethod def should_dispatch_to_target_plugin( diff --git a/app/runtime/events.py b/app/runtime/events.py index 880871386..9f740e522 100644 --- a/app/runtime/events.py +++ b/app/runtime/events.py @@ -21,6 +21,7 @@ from app.runtime.event.dispatch import EventDispatcher from app.runtime.event.errors import EventErrorNotifier, EventErrorPolicy from app.runtime.event.registry import EventRegistry from app.runtime.event.contracts import validate_event_payload +from app.runtime.correlation import get_correlation_id DEFAULT_EVENT_PRIORITY = 10 # 事件的默认优先级 MIN_EVENT_CONSUMER_THREADS = 1 # 最小事件消费者线程数 @@ -35,11 +36,13 @@ class Event: def __init__(self, event_type: Union[EventType, ChainEventType], event_data: Optional[Union[Dict, ChainEventData]] = None, - priority: Optional[int] = DEFAULT_EVENT_PRIORITY): + priority: Optional[int] = DEFAULT_EVENT_PRIORITY, + correlation_id: Optional[str] = None): """ :param event_type: 事件的类型，支持 EventType 或 ChainEventType :param event_data: 可选，事件携带的数据，默认为空字典 :param priority: 可选，事件的优先级，默认为 10 + :param correlation_id: 生产事件时固化的请求关联 ID """ payload_problems = validate_event_payload(event_type, event_data) if payload_problems: @@ -52,6 +55,7 @@ class Event: self.event_type = event_type # 事件类型 self.event_data = event_data or {} # 事件数据 self.priority = priority # 事件优先级 + self.correlation_id = correlation_id or get_correlation_id() def __repr__(self) -> str: """ diff --git a/app/runtime/execution.py b/app/runtime/execution.py index 7b5992660..4a0ab2f55 100644 --- a/app/runtime/execution.py +++ b/app/runtime/execution.py @@ -1,6 +1,7 @@ import asyncio import inspect import time +from contextvars import copy_context from functools import partial, wraps from typing import Any, Callable @@ -16,7 +17,8 @@ async def run_in_threadpool( """在线程中执行同步函数，保持 FastAPI 旧帮助函数的参数语义。""" if kwargs: func = partial(func, **kwargs) - return await run_sync(func, *args) + context = copy_context() + return await run_sync(context.run, func, *args) def retry(ExceptionToCheck: Any, diff --git a/app/runtime/log.py b/app/runtime/log.py index dc0e7c849..6de2f1371 100644 --- a/app/runtime/log.py +++ b/app/runtime/log.py @@ -12,12 +12,11 @@ from concurrent.futures import ThreadPoolExecutor from datetime import datetime from logging.handlers import RotatingFileHandler from pathlib import Path -from typing import Any, Dict, Optional, Protocol +from typing import Any, Callable, Dict, Optional, Protocol import click from pydantic import BaseModel, ConfigDict - class LogConfigModel(BaseModel): """描述日志级别、格式和文件写入策略。""" @@ -28,8 +27,12 @@ class LogConfigModel(BaseModel): LOG_LEVEL: str = "INFO" LOG_MAX_FILE_SIZE: int = 5 LOG_BACKUP_COUNT: int = 10 - LOG_CONSOLE_FORMAT: str = "%(leveltext)s[%(name)s] %(asctime)s %(message)s" - LOG_FILE_FORMAT: str = "【%(levelname)s】%(asctime)s - %(message)s" + LOG_CONSOLE_FORMAT: str = ( + "%(leveltext)s[%(name)s] %(asctime)s [%(correlation_id)s] %(message)s" + ) + LOG_FILE_FORMAT: str = ( + "【%(levelname)s】%(asctime)s [%(correlation_id)s] - %(message)s" + ) ASYNC_FILE_QUEUE_SIZE: int = 1000 ASYNC_FILE_WORKERS: int = 2 BATCH_WRITE_SIZE: int = 50 @@ -60,6 +63,7 @@ class LogEntry: self.message = message self.file_path = file_path self.timestamp = timestamp or datetime.now() + self.correlation_id = _get_log_correlation_id() class LogWriter(Protocol): @@ -73,6 +77,18 @@ class LogWriter(Protocol): log_settings = LogSettings() +_correlation_id_provider: Callable[[], str | None] = lambda: None + + +def configure_correlation_id_provider(provider: Callable[[], str | None]) -> None: + """由组合根注入日志关联 ID 读取端口，保持日志模块为依赖叶节点。""" + global _correlation_id_provider + _correlation_id_provider = provider + + +def _get_log_correlation_id() -> str: + """读取当前关联 ID；未装配或无请求上下文时返回稳定占位符。""" + return _correlation_id_provider() or "-" class NonBlockingFileHandler: @@ -169,7 +185,7 @@ class NonBlockingFileHandler: @staticmethod def _to_record(entry: LogEntry) -> logging.LogRecord: """把日志条目转换为标准库日志记录。""" - return logging.LogRecord( + record = logging.LogRecord( name="", level=getattr(logging, entry.level.upper(), logging.INFO), pathname="", @@ -179,6 +195,8 @@ class NonBlockingFileHandler: exc_info=None, created=entry.timestamp.timestamp(), ) + record.correlation_id = entry.correlation_id + return record def _batch_writer(self) -> None: """持续收集队列日志，并在停止哨兵后排空已有批次。""" @@ -258,6 +276,7 @@ class CustomFormatter(logging.Formatter): separator = " " * max(8 - len(record.levelname), 0) colorizer = _LEVEL_NAME_COLORS.get(record.levelno, str) record.leveltext = colorizer(record.levelname + ":") + separator + record.correlation_id = _get_log_correlation_id() return super().format(record) diff --git a/app/runtime/thread.py b/app/runtime/thread.py index 1132c7ef6..0209e4058 100644 --- a/app/runtime/thread.py +++ b/app/runtime/thread.py @@ -1,4 +1,5 @@ from concurrent.futures import ThreadPoolExecutor +from contextvars import copy_context from app.runtime.config import settings from app.foundation.singleton import Singleton @@ -20,7 +21,8 @@ class ThreadHelper(metaclass=Singleton): :param kwargs: 参数 :return: future """ - return self.pool.submit(func, *args, **kwargs) + context = copy_context() + return self.pool.submit(context.run, func, *args, **kwargs) def shutdown(self): """ diff --git a/app/scheduler.py b/app/scheduler.py index 0f54ebe24..adcdcf563 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -45,6 +45,7 @@ from app.runtime.gc import get_memory_usage from app.runtime.reload import ConfigReloadMixin from app.foundation.singleton import SingletonClass from app.runtime.scheduling import TimerUtils +from app.runtime.correlation import call_with_correlation, get_correlation_id lock = threading.Lock() SCHEDULER_PROGRESS_PREFIX = "scheduler" @@ -856,7 +857,10 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): deferred_finish = __start_coro(func(*args, **kwargs)) elif run_in_process: # 多进程运行 - p = multiprocessing.Process(target=func, args=args, kwargs=kwargs) + p = multiprocessing.Process( + target=call_with_correlation, + args=(get_correlation_id(), func, args, kwargs), + ) p.start() p.join() else: diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 5e17ae909..47c5f098c 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -731,6 +731,17 @@ app/scheduler.py # APScheduler 兼容 Facade 4. 日志 formatter 增加结构字段，不在消息字符串中到处手拼。 5. 外部请求可传标准 trace headers 或项目 correlation header，但不得泄露用户 token。 +**实施记录（2026-08-21）**： + +- 新增受 64 字符安全字符集约束的 `moviepilot_correlation_id` ContextVar 和纯 ASGI middleware；合法 + `X-Request-ID` 原样使用，非法值重新生成，`request.state`、普通响应和 SSE 握手响应回写同一个 ID。 +- 平台日志 formatter 以独立 `correlation_id` 字段输出；`app.runtime.execution`、共享 `ThreadHelper`、 + Event 生产/消费均显式复制或恢复上下文。Event 在生产时固化 ID，广播线程不能用自己的空上下文覆盖它。 +- Scheduler 多进程入口把关联 ID 作为显式可序列化参数传入，不依赖 fork 继承；`RequestUtils` 和 + `AsyncRequestUtils` 在调用方未指定时传播 `X-Request-ID`，不读取或复制任何鉴权 token。 +- 并发请求、非法头、线程池、事件处理、SSE、同步/异步外呼和显式外呼头覆盖均有专项测试；原 API + 响应、健康探针、日志和搜索流式测试保持通过。 + #### ARCH-261：指标与可选 OpenTelemetry Adapter 先定义内部观测端口和低基数指标： diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index ead685fe0..b1f6b11b0 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6223, - "edge_sha256": "f64e5083ab697022127780800617474d12f10b05a6fea26787e751a3f8a00421", + "edge_count": 6232, + "edge_sha256": "2139dcdcfc12b4f29c732471744d09b0990c51867d4ad227026b00fa947c90d8", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -104,6 +104,8 @@ "app.adapters.network.doh -> app.runtime.config", "app.adapters.network.doh -> app.runtime.log", "app.adapters.network.doh -> app.runtime.reload", + "app.adapters.network.http -> app.runtime", + "app.adapters.network.http -> app.runtime.correlation", "app.adapters.system.display -> app.foundation", "app.adapters.system.display -> app.foundation.singleton", "app.adapters.system.display -> app.runtime", @@ -151,6 +153,8 @@ "app.adapters.system.rust -> app.runtime", "app.adapters.system.rust -> app.runtime.config", "app.adapters.system.rust -> app.runtime.log", + "app.adapters.web.correlation -> app.runtime", + "app.adapters.web.correlation -> app.runtime.correlation", "app.adapters.web.health -> app.runtime", "app.adapters.web.health -> app.runtime.health", "app.adapters.web.security.access -> app.runtime", @@ -3719,6 +3723,7 @@ "app.domain.title -> app.schemas.types", "app.factory -> app.adapters", "app.factory -> app.adapters.web", + "app.factory -> app.adapters.web.correlation", "app.factory -> app.adapters.web.health", "app.factory -> app.adapters.web.plugin", "app.factory -> app.adapters.web.plugin.routes", @@ -3733,6 +3738,7 @@ "app.factory -> app.application.security.token", "app.factory -> app.runtime", "app.factory -> app.runtime.config", + "app.factory -> app.runtime.correlation", "app.factory -> app.runtime.extensions", "app.factory -> app.runtime.extensions.plugin_manager", "app.factory -> app.runtime.localization", @@ -5353,6 +5359,7 @@ "app.runtime.event.contracts -> app.schemas.event", "app.runtime.event.contracts -> app.schemas.types", "app.runtime.event.dispatch -> app.runtime", + "app.runtime.event.dispatch -> app.runtime.correlation", "app.runtime.event.dispatch -> app.runtime.event", "app.runtime.event.dispatch -> app.runtime.event.binding", "app.runtime.event.dispatch -> app.runtime.event.registry", @@ -5372,6 +5379,7 @@ "app.runtime.events -> app.foundation.singleton", "app.runtime.events -> app.runtime", "app.runtime.events -> app.runtime.config", + "app.runtime.events -> app.runtime.correlation", "app.runtime.events -> app.runtime.event", "app.runtime.events -> app.runtime.event.binding", "app.runtime.events -> app.runtime.event.contracts", @@ -5558,6 +5566,7 @@ "app.scheduler -> app.foundation.singleton", "app.scheduler -> app.runtime", "app.scheduler -> app.runtime.config", + "app.scheduler -> app.runtime.correlation", "app.scheduler -> app.runtime.events", "app.scheduler -> app.runtime.extensions", "app.scheduler -> app.runtime.extensions.plugin_manager", @@ -6240,7 +6249,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 776, + "module_count": 778, "modules": [ "app", "app.adapters", @@ -6280,6 +6289,7 @@ "app.adapters.system.rust", "app.adapters.system.stdio", "app.adapters.web", + "app.adapters.web.correlation", "app.adapters.web.health", "app.adapters.web.plugin", "app.adapters.web.plugin.routes", @@ -6870,6 +6880,7 @@ "app.runtime.compat.manifest", "app.runtime.compat.resource_imports", "app.runtime.config", + "app.runtime.correlation", "app.runtime.debounce", "app.runtime.event", "app.runtime.event.binding", diff --git a/tests/test_correlation.py b/tests/test_correlation.py new file mode 100644 index 000000000..e38a9be18 --- /dev/null +++ b/tests/test_correlation.py @@ -0,0 +1,180 @@ +"""请求关联 ID 在 HTTP、线程、事件和外部请求边界的传播测试。""" + +import asyncio +import logging +from types import SimpleNamespace +from unittest.mock import MagicMock + +import httpx +import pytest +from starlette.applications import Starlette +from starlette.responses import JSONResponse, StreamingResponse +from starlette.routing import Route + +from app.adapters.network.http import AsyncRequestUtils, RequestUtils +from app.adapters.web.correlation import CorrelationIdMiddleware +from app.runtime.correlation import ( + CORRELATION_ID_HEADER, + call_with_correlation, + correlation_scope, + get_correlation_id, + normalize_correlation_id, +) +from app.runtime.event.dispatch import EventDispatcher +from app.runtime.events import Event +from app.runtime.execution import run_in_threadpool +from app.runtime.log import CustomFormatter, configure_correlation_id_provider +from app.schemas.types import EventType + + +def _correlation_app() -> Starlette: + """构造同时包含普通响应和 SSE 风格流式响应的最小应用。""" + + async def current_id(_request): + """返回当前协程看到的关联 ID。""" + await asyncio.sleep(0) + return JSONResponse({"request_id": get_correlation_id()}) + + async def stream_id(_request): + """在响应开始后读取关联 ID，验证上下文覆盖完整流生命周期。""" + async def content(): + """生成一条 SSE 数据。""" + await asyncio.sleep(0) + yield f"data: {get_correlation_id()}\n\n" + + return StreamingResponse(content(), media_type="text/event-stream") + + app = Starlette( + routes=[ + Route("/id", current_id), + Route("/stream", stream_id), + ] + ) + app.add_middleware(CorrelationIdMiddleware) + return app + + +@pytest.mark.asyncio +async def test_concurrent_requests_keep_isolated_ids_and_stream_context() -> None: + """并发请求及流式响应必须各自保留入口 ID。""" + transport = httpx.ASGITransport(app=_correlation_app()) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + first, second = await asyncio.gather( + client.get("/id", headers={CORRELATION_ID_HEADER: "request-one"}), + client.get("/id", headers={CORRELATION_ID_HEADER: "request-two"}), + ) + stream = await client.get( + "/stream", headers={CORRELATION_ID_HEADER: "stream-request"} + ) + + assert first.json()["request_id"] == first.headers[CORRELATION_ID_HEADER] + assert second.json()["request_id"] == second.headers[CORRELATION_ID_HEADER] + assert first.headers[CORRELATION_ID_HEADER] != second.headers[CORRELATION_ID_HEADER] + assert stream.headers[CORRELATION_ID_HEADER] == "stream-request" + assert stream.text == "data: stream-request\n\n" + assert get_correlation_id() is None + + +@pytest.mark.asyncio +async def test_invalid_request_id_is_replaced_and_threadpool_copies_context() -> None: + """日志注入型入口值被替换，线程池仍读取替换后的安全 ID。""" + transport = httpx.ASGITransport(app=_correlation_app()) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.get( + "/id", headers={CORRELATION_ID_HEADER: "bad id!"} + ) + + generated = response.headers[CORRELATION_ID_HEADER] + assert generated != "bad id!" + assert normalize_correlation_id(generated) == generated + with correlation_scope("thread-request"): + observed = await run_in_threadpool(get_correlation_id) + assert observed == "thread-request" + + +def test_event_dispatch_restores_producer_correlation_id() -> None: + """后台事件处理器使用事件生产时固化的 ID，而非消费者线程上下文。""" + observed = [] + + def handler(_event): + """记录处理器实际看到的关联 ID。""" + observed.append(get_correlation_id()) + + resolver = MagicMock() + resolver.resolve.return_value = ( + handler, + SimpleNamespace(owner_name="test", run_sync_in_threadpool=False), + "Handler", + "handle", + ) + registry = MagicMock() + dispatcher = EventDispatcher( + registry=registry, + binding_resolver=resolver, + executor=MagicMock(), + event_loop=MagicMock(), + event_factory=Event, + error_handler=MagicMock(), + ) + with correlation_scope("producer-request"): + event = Event(EventType.SystemError, {}) + with correlation_scope("consumer-request"): + dispatcher.invoke_sync(handler, event) + + assert event.correlation_id == "producer-request" + assert observed == ["producer-request"] + assert get_correlation_id() is None + + +def test_sync_external_request_and_formatter_receive_correlation_id() -> None: + """同步外呼头和结构化日志字段使用同一个当前 ID。""" + configure_correlation_id_provider(get_correlation_id) + response = httpx.Response(200) + session = MagicMock() + session.request.return_value = response + formatter = CustomFormatter("%(correlation_id)s %(message)s") + record = logging.LogRecord("test", logging.INFO, "", 0, "message", (), None) + + with correlation_scope("outgoing-request"): + RequestUtils(session=session).request("GET", "https://example.com") + rendered = formatter.format(record) + + headers = session.request.call_args.kwargs["headers"] + assert headers[CORRELATION_ID_HEADER] == "outgoing-request" + assert rendered == "outgoing-request message" + + +def test_process_entry_restores_serialized_correlation_id() -> None: + """多进程入口使用显式 payload 恢复 ID，不依赖 fork 偶然继承上下文。""" + with correlation_scope("parent-request"): + observed = call_with_correlation( + "serialized-request", + get_correlation_id, + (), + {}, + ) + + assert observed == "serialized-request" + + +@pytest.mark.asyncio +async def test_async_external_request_preserves_explicit_header() -> None: + """异步外呼默认传播当前 ID，但不得覆盖调用方显式 trace 边界。""" + observed = [] + + async def respond(request: httpx.Request) -> httpx.Response: + """记录 MockTransport 收到的请求头。""" + observed.append(request.headers[CORRELATION_ID_HEADER]) + return httpx.Response(200) + + async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client: + utils = AsyncRequestUtils(client=client) + with correlation_scope("context-request"): + await utils.request("GET", "https://example.com/default") + await utils.request( + "GET", + "https://example.com/explicit", + headers={CORRELATION_ID_HEADER: "explicit-request"}, + ) + + assert observed == ["context-request", "explicit-request"]