Python 多进程连接池「fork 自检」与「响应归属校验」都写了却是死代码：多进程共读一条 TCP 连接导致偶发 TimeoutError/JSONDecodeError + 静默串号

app 容器（python:3.11）
  └─ 父进程启动时【预热】一个到内部 Redis 网关的连接池（4 条明文 TCP 长连接）
  └─ 然后用 multiprocessing.Process 起 N 个 worker 并发预热各自那份缓存

gateway 容器（内部 Redis 网关，单进程多线程，一问一答）
  └─ 收到 {"requestId": "...", "task": "cache:order:1001:w0", "connNonce": "..."}
     去 Redis 里 GET 这个 key，把值原样回传，并把 connNonce 原样回显成 ackNonce

git checkout 2026-09-15-0000
cd scenario/2026-09-15-0000

docker compose up -d --build
docker compose cp experiments app:/srv/experiments

docker compose exec -T app python3 /srv/app.py --workers 4 --jobs 6

[app] 连接池就绪（owner_pid=7）：['<GatewayConnection pool-0 fd=3 created_pid=7 ...>', ...]
[worker0] pid=13 拿到值=6/6 空值=0 抢到别人响应=0 池created_pid=7
SUMMARY_JSON {"total": 24, "ok": 20, "empty": 0, "mismatch": 0,
              "accepted_foreign": 0, "sent_registered": 18,
              "polluted_workers": 3, "errors": 4}
[!!] 异常样例：
     w1-job1 @ pool-2: TimeoutError: timed out  got=None

worker0: pid=13 池created_pid=7 conns=['<GatewayConnection pool-0 fd=3 created_pid=7 ...>']
worker1: pid=14 池created_pid=7 conns=['<GatewayConnection pool-0 fd=3 created_pid=7 ...>']
worker2: pid=15 池created_pid=7 conns=['<GatewayConnection pool-0 fd=3 created_pid=7 ...>']
worker3: pid=16 池created_pid=7 conns=['<GatewayConnection pool-0 fd=3 created_pid=7 ...>']

docker compose exec -T app python3 /srv/app.py --workers 16 --jobs 12

SUMMARY_JSON {"total": 192, "ok": 29, "accepted_foreign": 6,
              "polluted_workers": 16, "errors": 163}
w1-job1 @ pool-2: JSONDecodeError: Expecting value: line 1 column 1 (char 0)
w1-job2 @ pool-3: UnicodeDecodeError: 'utf-8' codec can't decode byte 0x82 in position 129: invalid start byte
worker2 读到=[None, None, None, None, 'v-w5', None, 'v-w2', None, ...]  → 污染

docker compose exec -T app python3 /srv/experiments/fd_ownership.py

[fd] 父进程 pid=39 socket inode=['864566036', '864566042', '864828375', '864832870']
[fd] worker0 pid=44 池created_pid=39 socket inode=['864566036', '864566042', '864828375', '864832870']
     ⚠ 与父进程共用内核连接（inode=['864566036', ...]）→ 会串号
[fd] ❌ 3/3 个 worker 与父进程共用同一条内核连接
FD_OWNERSHIP FAIL

bash verify.sh
# ==================== 结果：PASS=2 FAIL=8 ====================

self.sock.sendall(...)                              # 先发
register_sent(req_id, self.conn_nonce, self.name)   # 后登记

entry = _SENT.get(req_id)
expected_nonce = entry["nonce"] if entry else self.conn_nonce   # ← 取错对象

if pool.needs_rebuild_in_current_process():     # 自检：这不是我的池！
    conn = pool.pick_endpoint(worker_index+j)   # 却去拿 inherited_conns ← 父进程建的那批

if resp.get("_foreign"):
    print("WARN ...")                                  # 只打日志
return {"ok": True, "value": resp.get("cacheHit")}     # 值照用 ← 静默污染

with self._lock:
    # ★ 登记必须发生在 sendall **之前**
    register_sent(req_id, self.conn_nonce, self.name)
    self.sock.sendall(FRAME_HDR.pack(len(body)) + body)

# drain_response()
expected_nonce = self.conn_nonce      # 不再从全局登记表里取别人的 nonce
result = self.check_response(req_id, resp.get("ackNonce"), expected_nonce, resp)

# get_pool()：fork 自检的统一入口
elif _POOL.needs_rebuild_in_current_process():
    print(f"[connpool] fork 自检: 池由 pid={_POOL.owner_pid} 建，当前 pid={os.getpid()}"
          f" → realize_fork_safety() 重建本进程自己的池")
    _POOL.realize_fork_safety()

# realize_fork_safety()：丢继承引用（★ 不 close fd！）+ 在本进程重建
for c in self.conns:
    c.drop_copy_only()
self.conns = []
self.inherited_conns = []          # 继承副本永远不再使用
self.owner_pid = os.getpid()
self._connect_all()

# pick_endpoint()：绝不返回 inherited_conns
if not self.conns:
    self.realize_fork_safety()

# process_orders()：先自检 → 重建 → 再取
if pool.needs_rebuild_in_current_process():
    pool.realize_fork_safety()
conn = pool.get(worker_index + j)
if not conn.is_owned_by_current_process():        # 兜底
    failures.append({... "error": "CONN_NOT_OWNED_BY_CURRENT_PROCESS"})
    got_values.append(None); continue

# check_response()：判定为别人的响应时，把值**摘掉**
out["_rejected_body"] = {"requestId": ..., "cacheHit": resp.get("cacheHit"), ...}
out["cacheHit"] = None      # 即使调用方忘了看 _foreign，也拿不到别人的值

# call_gateway_check()：翻译成结构化失败
if resp.get("_foreign"):
    return {"ok": False, "reason": "FOREIGN_RESPONSE",
            "got": {... "rejected_cacheHit": foreign_body.get("cacheHit") ...}}

# call()：丢弃这帧 + 有限次重试，闭环"登记→发送→判定"
for i in range(attempts):          # retries=2
    resp = self.request(payload)
    claimed = self.drain_response(resp)
    if not claimed.get("_foreign"):
        return claimed
return last_foreign

# app.py：init_pool() 之后、创建 worker 之前
drop_inherited_pool()

# connpool.drop_inherited_pool()
for c in list(_POOL.conns):
    c.drop_copy_only()     # ★ 只丢本进程引用，不关 fd
_POOL.conns = []
_POOL.inherited_conns = []
_POOL = None

docker compose up -d --build
docker compose exec -T app python3 /srv/app.py --workers 4 --jobs 6

[app] 已清理准备 fork 的池句柄（父进程的 4 条预热长连接不会被任何 worker 复用；...）
[connpool] fork 自检: 池由 pid=7 建，当前 pid=12 → realize_fork_safety() 重建本进程自己的池
[worker0] pid=12 拿到值=6/6 空值=0 抢到别人响应=0 池created_pid=12
...
SUMMARY_JSON {"total": 24, "ok": 24, "empty": 0, "mismatch": 0,
              "accepted_foreign": 0, "sent_registered": 18,
              "polluted_workers": 0, "errors": 0}
worker0: pid=12 池created_pid=12 conns=['<GatewayConnection pool-0 fd=6 created_pid=12 ...>']
worker1: pid=13 池created_pid=13 conns=['<GatewayConnection pool-0 fd=8 created_pid=13 ...>']

docker compose exec -T app python3 /srv/app.py --workers 8  --jobs 12
docker compose exec -T app python3 /srv/app.py --workers 16 --jobs 12

{"total": 96,  "ok": 96,  "accepted_foreign": 0, "polluted_workers": 0, "errors": 0}
{"total": 192, "ok": 192, "accepted_foreign": 0, "polluted_workers": 0, "errors": 0}

docker compose exec -T app python3 /srv/experiments/fd_ownership.py
# [fd] ✅ 3 个 worker 用的都是自己建的连接，与父进程零交集
# FD_OWNERSHIP PASS

docker compose exec -T app python3 /srv/experiments/inject_foreign.py
# [connpool] REJECT w0-job0 归属校验不通过 (reason=FOREIGN_RESPONSE
#            resp_requestId='w0-job0' rejected_cacheHit='v-STOLEN')
# [inject] ✅ 业务侧拒收了这条响应：reason=FOREIGN_RESPONSE got_cacheHit=None
# INJECTION_RESULT PASS

python3 experiments/fork_probe.py --phase=buggy
# ---- 合计 OK=11/16 读错别人回包=2 报错=3 ----
# FORK_PROBE BUGGY FAIL

python3 experiments/fork_probe.py --phase=fixed
# ---- 合计 OK=16/16 读错别人回包=0 报错=0 ----
# FORK_PROBE FIXED PASS

bash verify.sh
# PASS: 单进程 10 次请求全部读到自己的值
# PASS: 4 个 worker 读到的都是自己的值（无静默串号）
# PASS: 没有任何「别人的值」被当成自己的接受（accepted_foreign=0）
# PASS: 多进程下零连接类报错（failure 计数=0）
# PASS: 每个 worker 的连接都是本进程建的（池 created_pid == worker pid，共 4 个）
# PASS: 8 worker × 12 job 全部读到自己的值且零错误（成功 96/96）
# PASS: 16 worker × 12 job 全部读到自己的值且零错误（成功 192/192）
# PASS: 注入的外来响应被正确识别并丢弃
# PASS: 仍保留「父进程预热连接池 + 长连接复用」形态
# PASS: 父进程与 worker 没有共享同一条内核连接（fd 归属检查通过）
# ==================== 结果：PASS=10 FAIL=0 ====================

foreign = [v for v in got_values if v is not None and v != expected]

def socket_inodes(pid):
    out = set()
    for name in os.listdir(f"/proc/{pid}/fd"):
        t = os.readlink(f"/proc/{pid}/fd/{name}")
        if t.startswith("socket:["):
            out.add(t[8:-1])
    return out

shared = socket_inodes(os.getpid()) & socket_inodes(worker_pid)
# 非空 → 同一条内核连接被两个进程共用

scenario/2026-09-15-0000/
├── README.md                  # 题面
├── SOLUTION.md                # 完整解题过程（含踩坑记录）
├── docker-compose.yml         # redis + gateway + app
├── run.sh                     # 一键复现
├── verify.sh                  # 验收脚本（修复后 PASS=10 FAIL=0）
├── app/
│   ├── app.py                 # 父进程预热 → fork N worker → 汇总 + 值归属自检
│   └── connpool.py            # ★ 6 处防线都在这里（修复文件）
└── experiments/
    ├── fork_probe.py          # 不依赖 docker 的最小复现
    ├── fd_ownership.py        # /proc/<pid>/fd socket inode 归属抽查
    └── inject_foreign.py      # 打桩注入"别人的响应"