共享LLM客户端的线程安全陷阱：per-request header污染实战

漏洞的形态

这类 bug 上报时从来不会被称为并发问题。它总是以"用户 A 看到了用户 B 的数据"、或"账单报告把费用算错了账户"、或"trace 中某个 span 属于了错误的请求"的形式出现。机制始终如一：写在一个长生命周期对象上的可变状态，在写入和发送之间被另一个请求读取到了。

候选名单少到可以一一列举，这也是它可以被测试而非漫无边际的原因。

每次请求设置默认 header。 调用方提供的 key 的授权、租户 id、trace id、幂等 key。把其中任何一个设置在共享客户端上而不是每次调用时传入，都是典型的出事形态。

可变默认参数或模块级累加器。 对话历史列表、token 计数器、重试次数。在 Python 中，可变默认参数会被该函数的每次调用永久共享，甚至不需要线程参与就能产生这个 bug。

单次请求替换 base URL 或 model。 当有人通过修改客户端来实现"这个路径用便宜 model"的功能时，很常见。

缓存在客户端上的 prompt 或 system message。 高效，但只有当它真的是全局的时候才正确。

大多数厂商 SDK 客户端是线程安全的；不安全的是围绕它们写的包装层。这一点值得明确说出来，因为看到这类 bug 后的本能反应是停止共享客户端，这样做会失去连接池却没有解决任何问题。

强制交错执行的测试工具

起一千个线程然后碰运气，是一个又慢又不稳定的测试。改为强制交错执行：每个线程设置自己的值，然后在一个 barrier 上等待，直到所有线程都设置完毕，之后才开始读取。如果状态是共享的，每个线程都会读到最后一个写人的值，除了一个之外所有断言都会失败，而且每次运行都是如此。

# test_client_thread_safety.py
import threading
from concurrent.futures import ThreadPoolExecutor

def test_tenant_header_does_not_leak_between_threads(client, capture_server):
    n = 16
    barrier = threading.Barrier(n)

    def one(i):
        tenant = "tenant-%d" % i
        with client.scoped(tenant_id=tenant):     # the API under test
            barrier.wait(timeout=5)               # every thread has now set its value
            resp = client.complete(prompt="ping")
        return i, capture_server.header_for(resp.request_id, "x-tenant-id")

    with ThreadPoolExecutor(max_workers=n) as pool:
        results = list(pool.map(one, range(n)))

    for i, seen in results:
        assert seen == "tenant-%d" % i, "thread %d sent %s" % (i, seen)

barrier 把一个概率性的测试变成了确定性的。没有它，十六个线程很可能依次跑完，测试在一个有问题的客户端上也会通过。有了它，状态错误的那段窗口在读取发生时必然是打开的。

capture server 是一个本地的 HTTP server，它根据响应中回传的标识符来记录每个请求的 header。记录 server 端而非在客户端自身视角中断言它打算发送什么：重要的是离开进程的那些字节，而一个客户端报告了正确的 header 却发送了错误的，这正是你在追查的失败模式。

断言什么，以及通过意味着什么

断言每个线程的身份，而非聚合计数。当十六个不同的租户值都出现了但被附在了错误的请求上时，一个断言"看到了十六个不同租户值"的测试就会通过——而这恰恰就是泄漏。断言必须将每个请求与发起它的线程所设置的值配对。

同一个文件里还值得加三个断言：没有请求带着空的或缺失的租户 header 离开、没有请求携带了已结束的线程的值、以及——如果你在累加用量的话——每个租户的 token 总和等于全局总数。最后那个能捕捉到计数器版本的 bug：没有任何东西被误算，但增量因为非原子的读-改-写而丢失了。

在这里对绿色通过意味着什么要诚实。一次并发的通过表明这种交错执行不会泄漏；它并不能证明不存在竞态。这就是为什么最后一部分存在——结构化断言才是重构后依然成立的，而竞态测试是用来让审阅者相信那条结构化规则是承重的。

单线程异步运行时让这个 bug 更容易误以为不存在，而且形式更隐蔽

单线程环境下没有语句间的抢占，所以对计数器的读-改-写是安全的——但每个 await 都是一个让出点，在 await 之前设置的可变状态如果在 await 之后被读取，就可能被另一个 task 在中间覆盖掉。在共享客户端上设置一个 header 然后 await 请求，正是这种模式。

等价的测试工具把线程换成 task，把 barrier 换成 event，写起来还要更容易一些，因为你控制了调度：让 task 并发运行，让假的 transport 在读取 header 之前 yield 一次。在 Python 中修复通常是使用 ContextVar，它是 per-task 而非 per-thread 的，并且能正确地跨 await 传播；在 Node 中则是 async local storage API。这两个都值得用一个测试来断言值跨越了 await 边界依然存活、并且不会泄漏到兄弟 task 中。

真正有效的结构性测试

既然竞态测试在坏代码上也可能通过，所以要加一个不可能通过的测试。设计规则是：每次请求的状态永不驻留在共享对象上，而这条规则是可以直接检查的。

在构造后将客户端冻结，或者让包装层完全不暴露 setter。然后在测试中断言尝试在共享实例上设置请求级字段会抛出异常。编译期等价物——readonly 类型、frozen dataclass——更好，因为在测试运行前就会失败。

断言每次调用的 API 携带了所有作用域内的东西。检查你的 complete 包装层的签名，断言 scoped 字段是参数，这样将来有人把它们移到客户端上时就会因为一个带解释性名称的测试而失败。

断言共享客户端真的是共享的。如果有人通过每个请求构造一个客户端来"修复"这个问题，泄漏会消失，连接复用也会跟着消失。一个断言工厂两次返回相同实例的测试，记录了这曾经是一个决策。

在 CI 中以高 worker 数量跑竞态测试，但不要放在快速单元测试路径上，让它成为有人重新引入可变状态时大声失败的那个。在失败输出中记录哪个线程看到了哪个值；并发测试上一个简单的断言失败几乎无法操作，而你记录的内容决定了它是否可以被诊断。