libwebsockets MQTT QoS2 入站去重与会话恢复：api-test-mqtt-qos2 测试实战解析

• 后端
• 网络
• 通信

libwebsockets 的lws-api-test-mqtt-qos2是一个针对 MQTT 客户端侧入站 QoS2（Quality of Service 2，恰好一次投递）语义的自动化围栏测试（fence test），它通过进程内假 broker 与真实 lws MQTT 客户端在 loopback 上完成七步字节级 lock-step 走查，验证 PUBREC/PUBCOMP 报文格式、重复投递去重、以及跨连接会话恢复等关键行为。阅读本文，你将掌握 QoS2 入站状态机的完整数据流、lws_mqtt_client_qos2_rx_add()恢复 API 的用法与实现细节、历史内存缺陷的成因与回归测试手段，并学会如何在本地构建与运行这一测试。

一、测试定位：为 QoS2 客户端侧行为加一道"围栏"

在 libwebsockets 中，minimal-examples-lowlevel/api-tests/目录下汇集了大量 API 自测程序（API selftest），每个程序针对某一具体功能做"围栏"式验证——即一旦某条路径被破坏，测试会立即以非零退出码报警。api-test-mqtt-qos2正是为 MQTT 客户端侧的入站 QoS2 接收路径（对应 issue #3672）建立的围栏。

该测试覆盖的核心语义包括：

• 客户端收到 QoS2PUBLISH后，必须以固定头标志为 0 的PUBREC应答（规范 MQTT-3.5.1）；
• 收到PUBREL后，必须以固定头标志为 0 的PUBCOMP应答（规范 MQTT-3.7.1）；
• 相同 packet id 的 DUP 重发不得被再次投递给应用；
• 应用通过lws_mqtt_client_qos2_rx_add()恢复的 packet id，必须抑制 broker 对它的重发投递；
• 每个PUBREL必须能命中并释放其对应的接收记录。

测试中客户端侧的所有行为都由真实 lws MQTT 客户端实现，而 broker 侧则是一个完全受控的假 broker，逐字节核对客户端回复，只有上一步完全正确才推进到下一步——这正是"lock-step（锁步）"的含义。测试入口代码位于 main.c，构建配置位于 CMakeLists.txt。

二、整体架构：ONLY_RAW 假 broker + 真实 MQTT 客户端

测试在单个 lws context 中同时运行两个角色：

• 假 broker：挂在LWS_SERVER_OPTION_ONLY_RAWvhost 上监听 TCP 端口，收到原始连接后以LWS_CALLBACK_RAW_ADOPT/LWS_CALLBACK_RAW_RX/LWS_CALLBACK_RAW_WRITEABLE回调手工完成 MQTT broker 的报文解析与应答；
• 真实客户端：通过lws_client_connect_via_info()连接127.0.0.1，以i.method = "MQTT"、i.alpn = "mqtt"、i.protocol = "mqtt"走真实的 lws MQTT 客户端协议栈。

两个协议注册在同一 context 中（见 main.c）：第一个lws-api-test-mqtt-qos2-broker协议作为默认 vhost 监听端口的原始协议，第二个mqtt协议供出站客户端按名字匹配。上下文创建时指定info.options = LWS_SERVER_OPTION_ONLY_RAW（该选项在 lws-context-vhost.h 中定义为(1ll << 22)），确保监听端不启动任何 HTTP 行为。

客户端的连接参数（main.c）如下：

static const lws_mqtt_client_connect_param_t conn_param = { .client_id = "lws-api-test-mqtt-qos2", .keep_alive = 60, .clean_start = 1, .client_id_nofree = 1, .username_nofree = 1, .password_nofree = 1, };

• keep_alive：MQTT 保活间隔，单位为秒，测试设为 60；假 broker 也会回应客户端的PINGREQ（回0xd0 00），保证长连接不被误判为超时；
• clean_start：MQTT 3.1.1 语义下的 clean session 标志，置 1 表示服务端不保留旧会话状态；
• *_nofree：告知 lws 这些字符串由调用方持有，库不负责释放，适用于静态字符串。

客户端建立连接后订阅一个 QoS2 topiclws/api-test-mqtt-qos2（topics[]数组设置qos = QOS2），随后假 broker 才开始走查脚本。测试整体还设置了 10 秒 watchdog（lws_sul_schedule定时watchdog_cb），任何一步若迟迟无法完成，都会触发超时失败并退出事件循环（main.c）。

三、Lock-step 脚本：七步字节级走查

3.1 完整走查表

假 broker 对客户端的回复做逐字节比对，只有上一步完全正确才发送下一步。下表完整复刻自测试 README（README.md），其中"deliveries"是应用实际收到投递的次数，"QOS2_RX_COMPLETE"是应用收到LWS_CALLBACK_MQTT_QOS2_RX_COMPLETE回调的次数：

其中0x0777（QOS2_RESTORED_PKT_ID）是客户端在连接建立时通过lws_mqtt_client_qos2_rx_add()手工恢复的 id，用于模拟"上次连接中断前已收下消息但未等到 PUBREL"的持久化场景。

3.2 报文字节与固定头标志

测试在源码中把每一步的期望行为编码为qsteps[QS_COUNT]表（main.c），每个条目包含 broker 发送报文的固定头字节（cmd）、packet id、payload（NULL表示 PUBREL）、期望客户端回复的固定头字节（reply）以及步进后的计数期望。关键字节含义：

• 0x34：PUBLISH，QoS 级别 2（qos位为 2）、DUP 位为 0；
• 0x3c：PUBLISH，QoS 级别 2、DUP 位为 1（重发报文）；
• 0x62：PUBREL（固定头标志 0x2）；
• 0x50：PUBREC（固定头标志必须为 0）；
• 0x70：PUBCOMP（固定头标志必须为 0）。

PUBREC与PUBCOMP的固定头标志必须为 0（对应 MQTT 规范 MQTT-3.5.1 与 MQTT-3.7.1 的保留位约定），只有PUBREL携带固定头标志 0x2。测试 README 特别强调：合规的 broker 会对固定头标志写错的客户端直接断开连接，因此一旦客户端写错，入站 QoS2 将永远无法在任何真实 broker 上完成。

3.3 broker 侧实现：解码、发送与比对

假 broker 的每个连接会话状态保存在struct broker_pss中（main.c），包含接收缓冲、发送缓冲与当前步进序号（step，初始 -1，订阅完成前不启动走查）。

• VBI 解码：vbi_decode()按 MQTT 变长字节整数（Variable Byte Integer）规则解析剩余长度，最多 4 字节、每字节取低 7 位并累乘 128，不完整时返回 0 等待更多数据（main.c）；
• 报文发送：broker_step_tx()依据当前步骤构造 PUBLISH（topic + packet id + payload）或 PUBREL（仅 4 字节）并排入发送缓冲；
• 回复比对：broker_step_rx()对客户端回复做三重校验——长度必须为 4 字节且与期望的{reply, 0x02, pkt_id_hi, pkt_id_lo}完全一致；应用投递计数必须等于rx_after；QOS2_RX_COMPLETE计数必须等于rx_complete_after。任一不符即fails++（main.c）。

broker 的LWS_CALLBACK_RAW_RX中还处理了 CONNECT（回CONNACK20 02 00 00）、SUBSCRIBE（回SUBACK90 03 <id> 02，其中末字节0x02表示该 topic 授权为 QoS2）、PINGREQ（回d0 00）等报文，并妥善处理 TCP 粘包/半包：先按 VBI 长度切分完整报文，处理完毕后用memmove消费已处理字节。

四、客户端侧 QoS2 入站状态机与去重原理

测试验证的核心数据结构是客户端连接上的QoS2 接收记录链表qos2_rx_list，其元素类型lws_mqtt_qos2_rx_t定义于 private-lib-roles-mqtt.h：

typedef struct lws_mqtt_qos2_rx { struct lws_dll2 list; uint16_t packet_id; } lws_mqtt_qos2_rx_t;

链表所有者struct lws_dll2_owner qos2_rx_list挂在连接的_lws_mqtt_related结构上（private-lib-roles-mqtt.h），上限为LWS_MQTT_MAX_QOS2_RX = 256（private-lib-roles-mqtt.h）。

4.1 入站 PUBLISH：登记或去重

解析到 QoS2PUBLISH的 packet id 时（mqtt.c），实现会先遍历qos2_rx_list查找该 id：

• 已存在：置qos2_duplicate = 1，表示这是重发报文，随后在投递阶段（mqtt.c）跳过对订阅者的 payload 下发——这是"恰好一次"语义的关键：同一个 id 只向应用投递一次；
• 不存在：若链表已满（达到 256 条）则按协议错误断开连接；否则用lws_zalloc分配条目、记下 packet id、挂到链表尾部，并回调可选的qos2_state_ops->rx_add（若应用配置了该 ops），供应用把未确认 id 写入持久存储。

4.2 入站 PUBREL：释放记录并完成交接

收到PUBREL时（mqtt.c），实现遍历qos2_rx_list查找该 id：

• 命中：将该条目从链表移除并释放，回调qos2_state_ops->rx_remove通知应用清除持久化记录；随后向所有订阅了相关 topic 的子流触发LWS_CALLBACK_MQTT_QOS2_RX_COMPLETE，in指向完成的uint16_tpacket id（lws-callbacks.h 中定义为回调值 213，语义为"PUBREL 已收到、PUBCOMP 已发送"）；最后设置send_pubcomp并在可写时回发PUBCOMP；
• 未命中：仍会回PUBCOMP（peer 仍然欠它一个应答），但不会触发QOS2_RX_COMPLETE——正如源码注释所强调的："基于QOS2_RX_COMPLETE提交事务的应用，绝不能由它从未收下过的 id 驱动"（mqtt.c）。

代码中还有一处值得注意的细节：PUBCOMP引用的 packet id 取自当前PUBREL（pubcomp_pkt_id），而不是上次 PUBLISH 的 id，避免交错报文导致"完成错误的 id"。

4.3 连接关闭时的清理

rops_close_role_mqtt()在连接销毁时会取消相关超时定时器、释放订阅条目，并遍历清空qos2_rx_list中的全部条目（ops-mqtt.c），保证没有泄漏。

五、会话恢复 API：lws_mqtt_client_qos2_rx_add

lws_mqtt_client_qos2_rx_add()是本次测试的核心被测 API，其头文件文档（lws-mqtt.h）明确说明：它允许应用在会话恢复时，从持久化存储重新填充未确认的 QoS2 接收记录。

测试中客户端在LWS_CALLBACK_MQTT_CLIENT_ESTABLISHED回调里注入QOS2_RESTORED_PKT_ID (0x0777)：

/* As if restoring from persistent storage an id we took * delivery of, but did not see the PUBREL for, before the * last connection was lost */ if (lws_mqtt_client_qos2_rx_add(wsi, QOS2_RESTORED_PKT_ID)) { lwsl_err("%s: qos2_rx_add failed\n", __func__); fails++; return -1; }

（见 main.c。）

实现位于 mqtt.c，逻辑为：

1. 取得网络连接 wsi，若其mqtt状态不存在则返回 1；
2. 幂等性：先遍历链表，若该 id 已存在则直接返回 0——重复恢复同一个 id 不是错误；
3. 容量上限：链表达到LWS_MQTT_MAX_QOS2_RX（256）时打印 notice 并返回 1，防止对端故意不 PUBREL 而钉住每个 id 导致内存被占满；
4. 用lws_zalloc分配并清零条目、填入 packet id、lws_dll2_add_tail挂入链表。

5.1 配套持久化钩子：qos2_state_ops

对于真正需要跨连接保存 QoS2 状态的应用程序，头文件还提供了lws_mqtt_qos2_state_ops_t（lws-mqtt.h），包含两个回调：

typedef struct lws_mqtt_qos2_state_ops { int (*rx_add)(struct lws *wsi, const char *client_id, uint16_t pkt_id); int (*rx_remove)(struct lws *wsi, const char *client_id, uint16_t pkt_id); } lws_mqtt_qos2_state_ops_t;

• rx_add：客户端收到新的 QoS2 PUBLISH 并登记 id 时被调用，应用可在此时把 id 写入存储；
• rx_remove：收到 PUBREL 释放 id 时被调用，应用同步清除存储中的该 id。

两者通过lws_mqtt_client_connect_param_t.qos2_state_ops挂到连接参数上。仓库中的配套示例 minimal-mqtt-client-qos2 演示了完整闭环：客户端用-f参数模拟 QoS2 交接中途断连，rx_add把未确认 id 存入模拟存储，断线重连后在ESTABLISHED中调用lws_mqtt_client_qos2_rx_add恢复，broker（Mosquitto）随后以 DUP=1 重发同一 PUBLISH，lws 识别为重复、丢弃 payload，无缝续接PUBREC → PUBREL → PUBCOMP流程。这与本测试的"publish restored / pubrel restored"两步本质相同，只是前者用 ops 钩子持久化、后者直接硬编码注入。

六、历史缺陷：未清零链表条目的教训

测试 README 明确记录了一个曾被此测试捕获的历史缺陷：QoS2 rx 列表条目曾经在分配时没有清零（非lws_zalloc），其链表链接字段（struct lws_dll2 list）初始值为堆上残留数据。后果是：

• 条目可能无法正确挂入链表，导致重复投递——DUP 重发被当作新消息再次交给应用；
• 对应的PUBREL在链表中找不到该 id，无法释放条目；
• 条目最终泄漏，并且是静默的，不伴随任何报错。

为了在没有 ASan 的环境里也能复现这类"堆垃圾"缺陷，ctest 在运行本测试时显式设置了 glibc 的环境变量MALLOC_PERTURB_=165，迫使malloc返回的新分配块填充非零字节，从而让"忘记初始化"的问题暴露为可观测的测试失败（见 CMakeLists.txt 的注释与配置）。当前实现已统一改用lws_zalloc（mqtt.c、[mqtt.c](https://link.gitcode.com/i/398a6702f0c1366ce6ec99c860f36d1d)），但这条回归路径仍被永久保留在测试矩阵中。

七、构建、集成与运行

7.1 编译前提

该测试通过 CMake 的LwsCheckRequirements机制声明三个硬性配置需求（CMakeLists.txt），任一不满足则整个目标不参与构建：

require_lws_config(LWS_ROLE_MQTT 1 requirements) require_lws_config(LWS_WITH_CLIENT 1 requirements) require_lws_config(LWS_WITH_SERVER 1 requirements)

即构建时需要同时开启 MQTT 角色、客户端与服务端能力。作为api-tests家族成员，它在正常 CMake 构建流程中随minimal-examples-lowlevel一并产出可执行文件lws-api-test-mqtt-qos2。

7.2 手动运行

README 给出了直接运行方式（端口可用-p指定；-p选项在 main.c 中被解析，未指定时默认端口为 7681）：

$ lws-api-test-mqtt-qos2 -p 17682 [2026/10/03 12:00:00:0000] U: LWS API selftest: MQTT client QoS2 rx ... [2026/10/03 12:00:00:0000] U: Completed: OK

测试还通过lws_cmdline_option_handle_builtin()接受 lws 内置命令行选项（如-d <log level>调节日志级别）。运行过程中可观察每步的 broker 动作日志（broker_step_tx: step <name>）、客户端侧MQTT_CLIENT_RX的 payload 内容以及MQTT_QOS2_RX_COMPLETE的 packet id，便于人工核对表 3.1 中的计数曲线。

7.3 ctest 集成

CMake 同时把该程序注册为 ctest 用例（CMakeLists.txt）：

• 端口由lws_get_free_port(PORT_MQTT_QOS2)分配——该函数从 15000 起按文件计数器递增取可用端口（见 LwsCheckRequirements.cmake），避免与并行 CI 中的其他用例冲突；
• 测试命令为lws-api-test-mqtt-qos2 -p ${PORT_MQTT_QOS2}；
• 测试环境注入MALLOC_PERTURB_=165，超时上限 30 秒。

因此一条ctest -R api-test-mqtt-qos2即可在本地复现全部验证；若任何一步回复错误、计数不符或超时，程序将打印Completed: failed并以非零退出。退出码约定：0 表示成功，非零表示某一步失败或交换超时（main.c）。

八、从测试到生产实践

这个测试程序本身就是一个高度浓缩的 QoS2 客户端实现范例，可以直接指导真实应用的设计：

1. 投递计数只算一次：应用侧LWS_CALLBACK_MQTT_CLIENT_RX中，测试通过pub->payload_pos == 0判断"每个消息的第一个分片"，从而在消息可能被分块到达时只计数一次（main.c）；
2. 提交事务以 QOS2_RX_COMPLETE 为准：只有收到LWS_CALLBACK_MQTT_QOS2_RX_COMPLETE（id 已从链表释放、PUBCOMP 已发出）才表示该消息"恰好一次"完成，适合作为幂等业务提交的触发点；
3. 断线恢复三步曲：连接建立后，把存储中未确认的 id 逐个lws_mqtt_client_qos2_rx_add()回去，配合qos2_state_ops的rx_add/rx_remove钩子维护存储一致性，即可在重连后自动抑制 broker 的 DUP 重发投递，同时正常应答 PUBREC/PUBREL，实现严格的一次性语义；
4. 关注资源上限：LWS_MQTT_MAX_QOS2_RX（256）既是内存保护线，也提示应用应尽快处理 PUBREL，避免积压大量未确认 id。

综上，api-test-mqtt-qos2从报文格式、去重逻辑、状态恢复到内存安全四个维度，为 libwebsockets 的 QoS2 客户端入站路径提供了可重复、可回归的自动化保障；理解它的每一步，就等于理解了 lws MQTT 客户端"恰好一次"投递的完整实现。

• 后端
• 网络
• 通信