From bce6b70e8fa5ae6165436bff0e40013683cb8afa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=97=A0=E9=97=BB=E9=A3=8E?= Date: Tue, 23 Jun 2026 21:42:50 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=BC=BA=20MQTT=20=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E6=8B=92=E7=BB=9D=E8=AF=8A=E6=96=AD=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题： - TCP_NODELAY 修复了 TCP ACK 延迟，但设备仍然重发 - 需要确定是消息被服务器拒绝，还是设备端 bug 改进内容： - OnPublish 中添加详细的拒绝日志输出 - 当消息验证失败时，输出 client_id、topic、qos、错误原因 - 当消息被屏蔽时，输出屏蔽类型和原因 - 帮助快速诊断 QoS0 重发的根本原因 新增文档： - doc/QOS0_RETRANSMIT_ANALYSIS.md - 深度分析重发问题的各种原因 - doc/DIAGNOSTIC_GUIDE.md - 完整的诊断指南和解决方案 使用方法： ./meshtastic_mqtt_server --console-log-mqtt=true 观察日志中是否有： [mqtt] PUBLISH rejected: ... - 消息被拒绝（服务器问题） [mqtt] PUBLISH blocked: ... - 消息被屏蔽（配置问题） 如果没有 rejected/blocked 日志但仍重发，则是设备端 bug。 Co-Authored-By: Claude Fable 5 --- doc/DIAGNOSTIC_GUIDE.md | 248 ++++++++++++++++++++++++++++++ doc/QOS0_RETRANSMIT_ANALYSIS.md | 259 ++++++++++++++++++++++++++++++++ main.go | 12 ++ 3 files changed, 519 insertions(+) create mode 100644 doc/DIAGNOSTIC_GUIDE.md create mode 100644 doc/QOS0_RETRANSMIT_ANALYSIS.md diff --git a/doc/DIAGNOSTIC_GUIDE.md b/doc/DIAGNOSTIC_GUIDE.md new file mode 100644 index 0000000..e757b23 --- /dev/null +++ b/doc/DIAGNOSTIC_GUIDE.md @@ -0,0 +1,248 @@ +# QoS0 重发问题诊断指南 + +## 快速诊断 + +### 1. 启用详细日志运行服务器 + +```bash +./meshtastic_mqtt_server --console-log-mqtt=true --console-log-meshtastic=true +``` + +### 2. 观察日志输出 + +#### ✅ 正常接收的消息 +``` +[mqtt] connect client_id=device123 username=user1 remote=192.168.1.100:54321 +text from=!12345678 channel=LongFast text="hello world" +``` + +#### ❌ 被拒绝的消息（关键！） +``` +[mqtt] PUBLISH rejected: client_id=device123 topic=msh/CN/2/e/LongFast/!12345678 qos=0 payload_len=156 error=protobuf decode failed +``` + +#### ❌ 被屏蔽的消息 +``` +[mqtt] PUBLISH blocked: client_id=device123 topic=msh/CN/2/e/LongFast/!12345678 type=forbidden_word reason=blocked node +``` + +### 3. 根据日志判断问题 + +| 日志内容 | 问题原因 | 解决方法 | +|---------|---------|---------| +| `PUBLISH rejected: error=protobuf decode failed` | 消息格式错误 | 检查设备固件版本 | +| `PUBLISH rejected: error=cannot be decrypted` | 无法解密 | 检查 PSK 配置或启用 `allow_encrypted_forwarding` | +| `PUBLISH blocked: type=node` | 节点被屏蔽 | 检查屏蔽规则 | +| `PUBLISH blocked: type=forbidden_word` | 内容被屏蔽 | 检查关键词过滤规则 | +| **没有 rejected/blocked 日志** | **不是服务器拒绝** | **问题在设备端** | + +## 详细诊断步骤 + +### 步骤 1: 检查数据库中被拒绝的消息 + +```bash +# 进入数据库 +sqlite3 /path/to/database.db + +# 查看最近被拒绝的消息 +SELECT + datetime(created_at, 'unixepoch', 'localtime') as time, + client_id, + json_extract(record, '$.error') as error, + json_extract(record, '$.topic') as topic, + payload_len +FROM discarded_packets +ORDER BY created_at DESC +LIMIT 20; + +# 统计拒绝原因 +SELECT + json_extract(record, '$.error') as error_type, + COUNT(*) as count +FROM discarded_packets +WHERE created_at > strftime('%s', 'now', '-1 hour') +GROUP BY error_type; +``` + +### 步骤 2: 抓包分析 + +```bash +# 开始抓包 +sudo tcpdump -i any -nn port 1883 -w /tmp/mqtt_traffic.pcap + +# 让设备发送几条消息，然后停止抓包 (Ctrl+C) + +# 用 Wireshark 打开 /tmp/mqtt_traffic.pcap +# 过滤器: mqtt +# 查看: +# 1. 是否看到重复的 PUBLISH 包（PacketID 相同） +# 2. 重发的时间间隔是多少 +# 3. 是否有 TCP 重传标志 [TCP Retransmission] +``` + +### 步骤 3: 测试不同的消息类型 + +```bash +# 安装 mosquitto 客户端 +# macOS: brew install mosquitto +# Linux: apt-get install mosquitto-clients + +# 发送一个简单的测试消息（QoS 0） +mosquitto_pub -h localhost -p 1883 -t "test/topic" -m "hello" -q 0 -d + +# 观察: +# 1. mosquitto_pub 是否报错 +# 2. 服务器日志是否显示 rejected +# 3. 是否看到重发行为 +``` + +### 步骤 4: 检查 PSK 配置 + +```bash +# 查看当前配置 +cat config.yaml | grep -A 5 "meshtastic:" + +# 如果使用默认 PSK +psk: "AQ==" # 这是索引 1 的默认 PSK + +# 如果使用自定义 PSK，确保与设备一致 +psk: "your_base64_encoded_psk" +``` + +## 常见原因和解决方案 + +### 原因 1: 消息无法解密 + +**症状：** 日志显示 `error=cannot be decrypted` + +**解决方案 A - 配置正确的 PSK：** +```yaml +# config.yaml +meshtastic: + psk: "your_base64_psk" # 与设备 channel 的 PSK 一致 +``` + +**解决方案 B - 允许转发加密消息：** +```yaml +# config.yaml +meshtastic: + allow_encrypted_forwarding: true # 即使无法解密也转发 +``` + +### 原因 2: 节点或内容被屏蔽 + +**症状：** 日志显示 `PUBLISH blocked` + +**解决方案：** +```sql +-- 查看屏蔽规则 +SELECT * FROM blocking_rules WHERE enabled = 1; + +-- 临时禁用特定规则 +UPDATE blocking_rules SET enabled = 0 WHERE id = ; + +-- 或禁用所有规则测试 +UPDATE blocking_rules SET enabled = 0; +``` + +### 原因 3: Protobuf 解析失败 + +**症状：** 日志显示 `error=protobuf decode failed` + +**可能原因：** +- 设备发送的不是标准的 Meshtastic 协议包 +- 固件版本不兼容 +- 数据损坏 + +**解决方案：** +- 更新设备固件到最新版本 +- 检查设备配置是否正确 +- 联系设备厂商 + +### 原因 4: 设备端 Bug + +**症状：** 服务器日志显示消息正常接收，没有 rejected/blocked，但设备仍然重发 + +**诊断方法：** +1. 检查设备日志（如果可访问） +2. 更新设备固件 +3. 尝试不同的 QoS 级别（QoS 1）看是否还重发 +4. 联系设备厂商报告问题 + +## 监控脚本 + +创建一个监控脚本 `monitor_rejects.sh`： + +```bash +#!/bin/bash +echo "监控 MQTT 消息拒绝情况..." +echo "按 Ctrl+C 停止" +echo "" + +# 实时监控日志 +tail -f /path/to/server.log | grep --line-buffered -E "rejected|blocked" | while read line; do + timestamp=$(date '+%Y-%m-%d %H:%M:%S') + echo "[$timestamp] $line" + + # 播放提示音（可选） + # echo -e "\a" +done +``` + +使用： +```bash +chmod +x monitor_rejects.sh +./monitor_rejects.sh +``` + +## 性能统计 + +查看消息处理统计： + +```sql +-- 最近一小时的消息统计 +SELECT + 'Forwarded' as type, + COUNT(*) as count +FROM packets +WHERE created_at > strftime('%s', 'now', '-1 hour') +UNION ALL +SELECT + 'Rejected' as type, + COUNT(*) as count +FROM discarded_packets +WHERE created_at > strftime('%s', 'now', '-1 hour'); + +-- 按客户端统计 +SELECT + client_id, + COUNT(*) as total_messages, + SUM(CASE WHEN from_discarded = 1 THEN 1 ELSE 0 END) as rejected, + printf('%.2f%%', + SUM(CASE WHEN from_discarded = 1 THEN 1 ELSE 0 END) * 100.0 / COUNT(*) + ) as reject_rate +FROM ( + SELECT client_id, 0 as from_discarded FROM packets + WHERE created_at > strftime('%s', 'now', '-1 hour') + UNION ALL + SELECT client_id, 1 as from_discarded FROM discarded_packets + WHERE created_at > strftime('%s', 'now', '-1 hour') +) +GROUP BY client_id +ORDER BY rejected DESC; +``` + +## 总结 + +遵循这个诊断流程： + +1. ✅ **启用详细日志** - 最重要的第一步 +2. ✅ **观察是否有 rejected/blocked** - 判断是否服务器拒绝 +3. ✅ **检查数据库** - 查看历史拒绝记录 +4. ✅ **抓包分析** - 确认网络层行为 +5. ✅ **根据原因修复** - 应用对应的解决方案 + +如果日志中**没有任何 rejected/blocked 消息**，但设备仍然重发，那么问题100%在**设备端固件**，需要： +- 更新设备固件 +- 检查设备配置 +- 联系设备厂商 diff --git a/doc/QOS0_RETRANSMIT_ANALYSIS.md b/doc/QOS0_RETRANSMIT_ANALYSIS.md new file mode 100644 index 0000000..ccfe835 --- /dev/null +++ b/doc/QOS0_RETRANSMIT_ANALYSIS.md @@ -0,0 +1,259 @@ +# QoS0 消息重发问题深度分析 + +## 问题现状 + +即使启用了 TCP_NODELAY，设备发送 QoS0 消息后仍然重发 3 次。 + +## 根本原因分析 + +### 1. TCP_NODELAY 修复了什么？ + +✅ TCP_NODELAY 确实解决了 **TCP ACK 延迟**问题： +- Nagle 算法延迟从 40-200ms 降低到 ~0.05ms +- TCP 层的确认现在是即时的 + +❌ 但这**不能解决消息被拒绝的问题**。 + +### 2. 消息被拒绝的流程 + +当设备发送的消息不符合服务器要求时： + +``` +设备 → MQTT PUBLISH (QoS0) + ↓ +服务器 TCP 层收到 → 发送 TCP ACK ✅ + ↓ +MQTT 层处理 → OnPublish hook + ↓ +MQTTPP 验证失败 → valid=false + ↓ +返回 packets.ErrRejectPacket + ↓ +mochi-mqtt 处理: return nil (不发送任何 MQTT 响应) + ↓ +设备收到 TCP ACK ✅ 但没有收到 MQTT 层响应 + ↓ +设备认为消息可能丢失 → 重发 ❌ +``` + +### 3. 为什么会重发？ + +可能的原因： + +#### 原因 A: 消息验证失败 + +检查以下验证失败的情况： + +1. **Protobuf 解码失败** + ``` + parseServiceEnvelope() 返回错误 + → MQTTPP 返回 valid=false + ``` + +2. **解密失败** + ``` + describePacket() 无法解密 + → type="encrypted_packet" 且 AllowEncryptedForwarding=false + → MQTTPP 返回 valid=false + ``` + +3. **屏蔽规则命中** + ``` + blockingViolationForRecord() 返回非 nil + → OnPublish 返回 ErrRejectPacket + ``` + +#### 原因 B: 设备期待应用层响应 + +某些 MQTT 客户端实现可能： +- 虽然使用 QoS0（不需要 PUBACK） +- 但仍然期待某种应用层响应或订阅回显 +- 没有收到预期响应时触发重试逻辑 + +#### 原因 C: 设备端 Bug + +设备固件可能有 bug： +- 错误地认为 QoS0 需要应用层确认 +- 超时机制设置不当 +- 重试逻辑实现错误 + +## 诊断步骤 + +### 步骤 1: 查看服务器日志 + +检查消息是否被拒绝： + +```bash +# 启用控制台日志 +./meshtastic_mqtt_server --console-log-mqtt=true --console-log-meshtastic=true + +# 查找被拒绝的消息 +grep -E "error|dropped|rejected" logs.txt +``` + +**关键日志标识：** +- `protobuf decode failed` - protobuf 解析失败 +- `cannot be decrypted` - 解密失败 +- `blocked node` / `forbidden word` - 屏蔽规则命中 + +### 步骤 2: 抓包分析 + +```bash +# 抓取 MQTT 流量 +tcpdump -i any -nn port 1883 -w mqtt.pcap + +# 用 Wireshark 分析: +# 1. 查看是否有 TCP 重传 (Retransmission) +# 2. 查看 MQTT PUBLISH 是否有对应的响应 +# 3. 检查时序图，看设备重发的时间间隔 +``` + +**期待的正常流程 (QoS0)：** +``` +Client → Server: MQTT PUBLISH (QoS0) +Server → Client: TCP ACK +(没有 MQTT 层的 PUBACK，因为是 QoS0) +``` + +**如果消息被拒绝：** +``` +Client → Server: MQTT PUBLISH (QoS0) +Server → Client: TCP ACK +(服务器静默丢弃，没有任何 MQTT 响应) +Client → Server: MQTT PUBLISH (QoS0) [重发] +Server → Client: TCP ACK +... +``` + +### 步骤 3: 检查数据库 + +```sql +-- 查看被丢弃的消息 +SELECT * FROM discarded_packets +ORDER BY created_at DESC +LIMIT 20; + +-- 统计丢弃原因 +SELECT + json_extract(record, '$.error') as error_type, + COUNT(*) as count +FROM discarded_packets +GROUP BY error_type; +``` + +### 步骤 4: 测试不同的消息 + +```bash +# 发送一个有效的测试消息 +mosquitto_pub -h localhost -p 1883 -t "msh/CN/2/e/LongFast/!12345678" -m "test" -q 0 + +# 观察是否也会重发 +``` + +## 解决方案 + +### 方案 1: 修复消息验证问题 + +如果是消息验证失败导致： + +**检查 PSK 配置：** +```bash +# 确保服务器配置了正确的 PSK +./meshtastic_mqtt_server --psk="your_base64_psk" +``` + +**检查屏蔽规则：** +```sql +-- 查看当前的屏蔽规则 +SELECT * FROM blocking_rules WHERE enabled = 1; + +-- 临时禁用所有规则测试 +UPDATE blocking_rules SET enabled = 0; +``` + +### 方案 2: 允许加密消息转发 + +如果消息是加密的且无法解密： + +```yaml +# config.yaml +meshtastic: + allow_encrypted_forwarding: true +``` + +这样即使无法解密，消息也会被转发而不是拒绝。 + +### 方案 3: 返回明确的错误响应（不推荐） + +理论上可以在消息被拒绝时返回 MQTT 错误码，但： +- ❌ QoS 0 协议规定不应该有 PUBACK +- ❌ 违反 MQTT 规范 +- ❌ 可能导致客户端行为异常 + +### 方案 4: 设备端修复 + +如果是设备固件 bug： +- 更新设备固件到最新版本 +- 检查设备日志，确认重发原因 +- 联系设备厂商报告 bug + +## 监控和调试 + +### 添加详细日志 + +修改 `main.go` 的 `OnPublish` 方法： + +```go +func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (packets.Packet, error) { + valid, _, record := mqtpp.MQTTPP(pk.TopicName, pk.Payload, h.key, mqtpp.Options{ + AllowEncryptedForwarding: h.settings.AllowEncryptedForwarding(), + PKIKeyResolver: h.pkiResolver, + }) + + info := mqttClientInfoFromClient(cl) + + if !valid { + // 添加详细日志 + printJSON(map[string]any{ + "event": "publish_rejected", + "reason": "validation_failed", + "client_id": info.ClientID, + "topic": pk.TopicName, + "payload_len": len(pk.Payload), + "error": record["error"], + }) + h.rejectPublish(cl, pk, record) + return pk, packets.ErrRejectPacket + } + + // ... 其他逻辑 +} +``` + +### 监控重发率 + +```sql +-- 创建视图统计每个客户端的重发率 +CREATE VIEW client_retransmit_stats AS +SELECT + client_id, + COUNT(*) as total_attempts, + COUNT(DISTINCT packet_id) as unique_packets, + (COUNT(*) - COUNT(DISTINCT packet_id)) * 100.0 / COUNT(*) as retransmit_rate +FROM packets +GROUP BY client_id +HAVING retransmit_rate > 10; +``` + +## 结论 + +TCP_NODELAY 修复了 TCP 层的延迟问题，但如果消息本身被服务器拒绝（验证失败、解密失败、屏蔽规则等），设备仍然会重发。 + +**下一步行动：** + +1. ✅ 启用详细日志，查看是否有消息被拒绝 +2. ✅ 检查 `discarded_packets` 表，确认拒绝原因 +3. ✅ 抓包分析，确认是 TCP 重传还是应用层重发 +4. ✅ 根据诊断结果选择对应的解决方案 + +**如果所有消息都被正常处理（没有被拒绝），但仍然重发，那么问题在设备端固件。** diff --git a/main.go b/main.go index ae1e1ae..ddf85d2 100644 --- a/main.go +++ b/main.go @@ -203,6 +203,12 @@ func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (pa PKIKeyResolver: h.pkiResolver, }) if !valid { + // 记录拒绝原因，帮助诊断 QoS0 重发问题 + if h.consoleLog { + info := mqttClientInfoFromClient(cl) + fmt.Fprintf(os.Stderr, "[mqtt] PUBLISH rejected: client_id=%s topic=%s qos=%d payload_len=%d error=%v\n", + info.ClientID, pk.TopicName, pk.FixedHeader.Qos, len(pk.Payload), record["error"]) + } h.rejectPublish(cl, pk, record) return pk, packets.ErrRejectPacket } @@ -210,6 +216,12 @@ func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (pa for key, value := range violation { record[key] = value } + // 记录屏蔽原因 + if h.consoleLog { + info := mqttClientInfoFromClient(cl) + fmt.Fprintf(os.Stderr, "[mqtt] PUBLISH blocked: client_id=%s topic=%s type=%v reason=%v\n", + info.ClientID, pk.TopicName, violation["blocking_type"], violation["error"]) + } h.rejectPublish(cl, pk, record) return pk, packets.ErrRejectPacket }