[B-01][critical] 客户端 CONNECT 带 Receive Maximum 时 mochi 发送配额路径递归读锁死锁，单个客户端即可冻结全站下发 #8

New Issue

2026-09-30T13:56:51+08:00

nixevol commented

2026-09-30 13:56:51 +08:00

编号：B-01　严重级：critical　工作线：broker（internal/broker）　来源：总审查人实证
依赖：无　被依赖：B-03 (#10)（同在 hooks.go，按顺序合入）

现象与影响

只要有一个客户端在 CONNECT 里带了 MQTT 5 的 Receive Maximum 属性（这是标准属性，很多设备库会带；F20 允许裸 MQTT 设备接入），mochi v2.7.9（当前最新版）就会走发送配额路径。该路径有两个缺陷：

递归读锁死锁：Inflight.NextImmediate 先 RLock，再调 GetAll 又 RLock（mochi inflight.go:95-105、74-90）。Go 的 sync.RWMutex 在有写锁排队时会阻塞新的读锁，明确禁止递归读锁。只要此时另一个 goroutine 向该客户端发布 QoS 1（publishToClient → Inflight.Set 写锁，server.go:1083），两边就永久互等。
配额泄漏导致静默停滞：processPacket 补发被配额扣住的包后，立刻把它从 inflight 删除（server.go:722-726）。客户端回的 PUBACK 在 processPuback 里找不到条目（server.go:1162），发送配额永远少一格。配额耗尽后，publishToClient 把所有 QoS 1 扣在 inflight 里却返回成功（server.go:1089-1093），这个连接从此收不到任何 QoS 1，而上层以为已推送。

触发条件：processPacket 在 Inflight.Len() > 0 && sendQuota > 0 时对每个收到的包都会调用 NextImmediate（server.go:719-728）；sendQuota 来自客户端 CONNECT 的 Receive Maximum（clients.go:233）。测试客户端和目前的验收都不带这个属性，所以一直没暴露。

放到 NixMsg 里，死锁后以下都会永久阻塞：

所有向该客户端发 QoS 1 的 goroutine：全局 messageLoops（cmd/nixmsg/serve.go:322-348，串行对所有在线端推送，导致全站推送、到点分发、清理停止）、其他端的上行 worker（回执、撤回推送）、管理接口（作废后的 revoked、fatal）。
mochi 自己的全局 eventLoop（clearExpiredInflights → GetAll 读锁），以及该编号重连时的 inheritClientSession（ClearInflights），新连接也挂住；srv.Close() 无法返回。

实证

用原版 mochi v2.7.9 复现：客户端 CONNECT 带 ReceiveMaximum=20，订阅主题，边收边回 PUBACK 并每 200µs 发一次 PINGREQ；服务端 8 个 goroutine 持续 server.Publish(QoS1)。只收到 31 条就永久卡死，用时不到 4 秒。卡死时的栈：

该客户端读循环：server.go:720 processPacket → inflight.go:99 NextImmediate → inflight.go:75 GetAll 阻塞在 RWMutex.RLock（手里握着第一次读锁）；
发布方：server.go:1083 publishToClient → inflight.go:34 Inflight.Set 阻塞在 RWMutex.Lock；
mochi eventLoop：clients.go:344 ClearExpiredInflights → GetAll 阻塞；srv.Close() 挂住直到测试超时。

只修掉递归读锁、再修掉"补发后删除"之后，6 轮里仍有 1 轮在 935 条时停滞，说明这条路径不可依赖。

为何判定为真实缺陷

DEVELOPMENT §5 只提到 inflight 满时 mochi 会静默丢弃，没有提到发送配额；DEVIATIONS 无相关条目。这是依赖库缺陷，NixMsg 需要规避。

解决方案（已验证）

在 internal/broker/hooks.go 的 nixHook.OnConnect（第 33 行起，mochi 在 ParseConnect 之后调用它）里执行 cl.State.Inflight.ResetSendQuota(0)。这样 sendQuota 与 maximumSendQuota 都为 0：processPacket 不再进入 NextImmediate 分支，publishToClient 不再扣包，整条有缺陷的路径不会被执行。
- 验证：同一压测在原版 mochi 上连跑 6 轮、每轮 30 秒，每轮稳定收到约 250 万条，无死锁、无停滞。
后果与补充：服务端不再由 mochi 执行客户端的 Receive Maximum，改由应用层窗口约束——单连接未 PUBACK 的 QoS 1 大约不超过推送窗口 32 + 回执窗口 64 + 在途请求的 resp + 少量 revoked/fatal。因此：
- DEVELOPMENT §5 与 §10（非 SDK 设备）写明：客户端不要带 Receive Maximum，或设为不小于 256。
- CONNECT 里 Receive Maximum 小于 256 时仍然接受连接，但每个连接打一条 warn 日志；如需计数，新增指标要同时写进 DEVELOPMENT 的指标清单，不要自造名字。
- K 线核对四套 SDK 的 CONNECT 不带过小的 Receive Maximum（见 SDK 相关 issue）。
可选：向 mochi 上游报告 NextImmediate 递归读锁与 processPacket 补发后删除 inflight 的问题。

不采用的备选：fork 并修补 mochi（replace 指令）。只修递归读锁仍观察到停滞，且长期维护 fork 成本高；置 0 直接绕开整条路径更稳。

改动文件

internal/broker/hooks.go（OnConnect）、internal/broker/broker_test.go 或新测试文件、docs/DEVELOPMENT.md §5/§10、docs/DEVIATIONS.md。

交互 / 冲突说明

与 B-02、B-03 同在 broker，由 broker 线按 B-01 → B-02 → B-03 → B-04 顺序合入，避免 hooks.go 冲突。
不改 port.Downlink 接口，对消息线透明。

验收与测试

新增 broker 回归测试（带看门狗，不能挂住整个测试进程）：MQTT 5 客户端带 ReceiveMaximum=20 订阅自己的 down 主题；8 个 goroutine 持续 PublishDown QoS 1 10 秒，客户端回 PUBACK 并定时 PINGREQ；断言进度停滞从不超过 2 秒。修复前应在数秒内失败。
task check 通过。

复现测试要点（可作为回归测试起点）

// 客户端 CONNECT 带 Receive Maximum
cp := packets.Packet{
	FixedHeader: packets.FixedHeader{Type: packets.Connect}, ProtocolVersion: 5,
	Connect:    packets.ConnectParams{ProtocolName: []byte("MQTT"), Clean: true, ClientIdentifier: "c1", Keepalive: 30},
	Properties: packets.Properties{ReceiveMaximum: 20},
}
// 读 goroutine：每收到 QoS1 PUBLISH 就回 PUBACK（写操作加锁）；另一个 goroutine 每 200µs 发 PINGREQ
// 8 个 goroutine：for { srv.Publish("t", []byte("x"), false, 1) }
// 看门狗：received 计数 3 秒不变 → runtime.Stack(buf, true) 打印含 NextImmediate / (*Inflight).Set 的栈并 t.Fatal
// 修复验证：在 OnConnect 钩子里 cl.State.Inflight.ResetSendQuota(0) 后，30 秒内持续增长、不停滞

_{复审基线：main 4059a15（2026-09-30）。编号说明、各工作线的合并顺序、共享文件归属见总览 #7。}

**编号**：B-01　**严重级**：critical　**工作线**：broker（internal/broker）　**来源**：总审查人实证 **依赖**：无　**被依赖**：B-03 (#10)（同在 `hooks.go`，按顺序合入） ### 现象与影响只要有一个客户端在 CONNECT 里带了 MQTT 5 的 `Receive Maximum` 属性（这是标准属性，很多设备库会带；F20 允许裸 MQTT 设备接入），mochi v2.7.9（当前最新版）就会走发送配额路径。该路径有两个缺陷： 1. **递归读锁死锁**：`Inflight.NextImmediate` 先 `RLock`，再调 `GetAll` 又 `RLock`（mochi `inflight.go:95-105`、`74-90`）。Go 的 `sync.RWMutex` 在有写锁排队时会阻塞新的读锁，明确禁止递归读锁。只要此时另一个 goroutine 向该客户端发布 QoS 1（`publishToClient` → `Inflight.Set` 写锁，`server.go:1083`），两边就永久互等。 2. **配额泄漏导致静默停滞**：`processPacket` 补发被配额扣住的包后，立刻把它从 inflight 删除（`server.go:722-726`）。客户端回的 PUBACK 在 `processPuback` 里找不到条目（`server.go:1162`），发送配额永远少一格。配额耗尽后，`publishToClient` 把所有 QoS 1 扣在 inflight 里却返回成功（`server.go:1089-1093`），这个连接从此收不到任何 QoS 1，而上层以为已推送。触发条件：`processPacket` 在 `Inflight.Len() > 0 && sendQuota > 0` 时对每个收到的包都会调用 `NextImmediate`（`server.go:719-728`）；`sendQuota` 来自客户端 CONNECT 的 Receive Maximum（`clients.go:233`）。测试客户端和目前的验收都不带这个属性，所以一直没暴露。放到 NixMsg 里，死锁后以下都会永久阻塞： - 所有向该客户端发 QoS 1 的 goroutine：全局 `messageLoops`（`cmd/nixmsg/serve.go:322-348`，串行对所有在线端推送，导致全站推送、到点分发、清理停止）、其他端的上行 worker（回执、撤回推送）、管理接口（作废后的 revoked、fatal）。 - mochi 自己的全局 `eventLoop`（`clearExpiredInflights` → `GetAll` 读锁），以及该编号重连时的 `inheritClientSession`（`ClearInflights`），新连接也挂住；`srv.Close()` 无法返回。 ### 实证用原版 mochi v2.7.9 复现：客户端 CONNECT 带 `ReceiveMaximum=20`，订阅主题，边收边回 PUBACK 并每 200µs 发一次 PINGREQ；服务端 8 个 goroutine 持续 `server.Publish(QoS1)`。**只收到 31 条就永久卡死，用时不到 4 秒**。卡死时的栈： - 该客户端读循环：`server.go:720 processPacket` → `inflight.go:99 NextImmediate` → `inflight.go:75 GetAll` 阻塞在 `RWMutex.RLock`（手里握着第一次读锁）； - 发布方：`server.go:1083 publishToClient` → `inflight.go:34 Inflight.Set` 阻塞在 `RWMutex.Lock`； - mochi `eventLoop`：`clients.go:344 ClearExpiredInflights` → `GetAll` 阻塞；`srv.Close()` 挂住直到测试超时。只修掉递归读锁、再修掉"补发后删除"之后，6 轮里仍有 1 轮在 935 条时停滞，说明这条路径不可依赖。 ### 为何判定为真实缺陷 DEVELOPMENT §5 只提到 inflight 满时 mochi 会静默丢弃，没有提到发送配额；DEVIATIONS 无相关条目。这是依赖库缺陷，NixMsg 需要规避。 ### 解决方案（已验证） 1. 在 `internal/broker/hooks.go` 的 `nixHook.OnConnect`（第 33 行起，mochi 在 `ParseConnect` 之后调用它）里执行 `cl.State.Inflight.ResetSendQuota(0)`。这样 `sendQuota` 与 `maximumSendQuota` 都为 0：`processPacket` 不再进入 `NextImmediate` 分支，`publishToClient` 不再扣包，整条有缺陷的路径不会被执行。 - 验证：同一压测在原版 mochi 上连跑 6 轮、每轮 30 秒，每轮稳定收到约 250 万条，无死锁、无停滞。 2. 后果与补充：服务端不再由 mochi 执行客户端的 Receive Maximum，改由应用层窗口约束——单连接未 PUBACK 的 QoS 1 大约不超过推送窗口 32 + 回执窗口 64 + 在途请求的 resp + 少量 revoked/fatal。因此： - DEVELOPMENT §5 与 §10（非 SDK 设备）写明：客户端不要带 Receive Maximum，或设为不小于 256。 - CONNECT 里 Receive Maximum 小于 256 时仍然接受连接，但每个连接打一条 warn 日志；如需计数，新增指标要同时写进 DEVELOPMENT 的指标清单，不要自造名字。 - K 线核对四套 SDK 的 CONNECT 不带过小的 Receive Maximum（见 SDK 相关 issue）。 3. 可选：向 mochi 上游报告 `NextImmediate` 递归读锁与 `processPacket` 补发后删除 inflight 的问题。不采用的备选：fork 并修补 mochi（`replace` 指令）。只修递归读锁仍观察到停滞，且长期维护 fork 成本高；置 0 直接绕开整条路径更稳。 ### 改动文件 `internal/broker/hooks.go`（`OnConnect`）、`internal/broker/broker_test.go` 或新测试文件、`docs/DEVELOPMENT.md` §5/§10、`docs/DEVIATIONS.md`。 ### 交互 / 冲突说明 - 与 B-02、B-03 同在 broker，由 broker 线按 B-01 → B-02 → B-03 → B-04 顺序合入，避免 `hooks.go` 冲突。 - 不改 `port.Downlink` 接口，对消息线透明。 ### 验收与测试 - 新增 broker 回归测试（带看门狗，不能挂住整个测试进程）：MQTT 5 客户端带 `ReceiveMaximum=20` 订阅自己的 down 主题；8 个 goroutine 持续 `PublishDown` QoS 1 10 秒，客户端回 PUBACK 并定时 PINGREQ；断言进度停滞从不超过 2 秒。修复前应在数秒内失败。 - `task check` 通过。 <details><summary>复现测试要点（可作为回归测试起点）</summary> ```go // 客户端 CONNECT 带 Receive Maximum cp := packets.Packet{ FixedHeader: packets.FixedHeader{Type: packets.Connect}, ProtocolVersion: 5, Connect: packets.ConnectParams{ProtocolName: []byte("MQTT"), Clean: true, ClientIdentifier: "c1", Keepalive: 30}, Properties: packets.Properties{ReceiveMaximum: 20}, } // 读 goroutine：每收到 QoS1 PUBLISH 就回 PUBACK（写操作加锁）；另一个 goroutine 每 200µs 发 PINGREQ // 8 个 goroutine：for { srv.Publish("t", []byte("x"), false, 1) } // 看门狗：received 计数 3 秒不变 → runtime.Stack(buf, true) 打印含 NextImmediate / (*Inflight).Set 的栈并 t.Fatal // 修复验证：在 OnConnect 钩子里 cl.State.Inflight.ResetSendQuota(0) 后，30 秒内持续增长、不停滞 ``` </details> --- <sub>复审基线：main `4059a15`（2026-09-30）。编号说明、各工作线的合并顺序、共享文件归属见总览 #7。</sub>