边缘 AI MQTT 加固 + 离线缓冲可执行手册

调研日期：约 2026-10-25（Asia/Shanghai）
定位：站内自托管 Broker 选型、鉴权/ACL、Topic/QoS、断线有界缓冲、断 MQTT 不堵推理硬门禁；与 IEC 60870-5-104 只读边界对齐。
上游衔接：edge-ai-event-schema-acceptance.md（Schema/Topic）、edge-ai-model-signstore-mender-cookbook.md（OTA 轨）、edge-ai-substation-gateway-bom.md（AI VLAN）、edge-ai-scada-104-pointlist-template.md（104 点表）、edge-ai-license-ledger-template.md（许可台账）。
许可声明：下文 SPDX/许可以官网与 GitHub LICENSE 核验为准；嵌入产品/集群等场景须法务复核。不确定处标 unknown。

1. 目标与硬红线

验收否决项（对齐 Schema 手册）：人为断开 Broker ≥5 分钟，解码与推理 FPS 相对断线前跌落 ≤5%；恢复后不得 OOM。

2. Broker 短名单（站内自托管）

选型口诀

1. 单站试点 / A–B 档网关旁：Mosquitto（稳、文档熟、许可清晰）。
2. 算力盒内存紧、要同机旁路：NanoMQ（MIT，边向）。
3. 多站汇聚、规则引擎：评估 EMQX 钉版本（≤5.8 Apache 轨，或 5.9+ 单节点 BSL 条款）；禁止未审就把 EMQX 嵌进对外销售的网关镜像。
4. 全部进许可台账：组件名、版本、SPDX、是否嵌入分发、法务结论。

3. Topic / QoS / ACL（对齐既有 Schema）

3.1 Topic 树（与 Schema 手册一致）

edgeai/{org}/{site}/{device_id}/events/{task}   # 告警/检出，QoS1
edgeai/{org}/{site}/{device_id}/frames/{cam}    # 可选帧摘要，QoS0，强节流
edgeai/{org}/{site}/{device_id}/telemetry       # 功耗/温度/FPS/mqtt_q，QoS0（或1）
edgeai/{org}/{site}/{device_id}/status          # retained online/offline，QoS1
edgeai/{org}/{site}/{device_id}/cmd/#           # 下行运维，QoS1；ACL 严控
edgeai/{org}/{site}/{device_id}/ota/ack         # 可选：模型/Artifact 校验结果上报

另：MODEL_VERIFY_FAIL 等运维事件走 events/ops 或 telemetry 扩展字段，勿塞保护通道。

3.2 QoS 与 Retain

禁止对高频 events/frames 开 Retain（Broker 磁盘与陈旧告警风险）。

3.3 ACL 最小权限（示意）

设备证书/device_id 绑定用户名，例如 dev-box-rk3588-07：

跨站通配 # 禁止给设备账号。cmd 禁止设备互发。

4. 鉴权：口令 vs mTLS

4.1 推荐分层

口令轨：mosquitto_passwd（或等价）每设备独立哈希；禁止共享 admin/admin。
mTLS 轨：站内 CA 签设备证；吊销用 CRL/短有效期轮换（对齐签名仓密钥轮换节奏）。
匿名：生产与联调对外口一律关闭。

4.2 Mosquitto 配置骨架（可复制）

# /etc/mosquitto/conf.d/edgeai.conf  — 示例，按站改路径
listener 8883
protocol mqtt
cafile   /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/broker.crt
keyfile  /etc/mosquitto/certs/broker.key
require_certificate true          # mTLS；若仅口令则 false + password_file
use_identity_as_username true     # CN 作 username，便于 ACL

allow_anonymous false
# password_file /etc/mosquitto/passwd   # 口令轨启用
acl_file /etc/mosquitto/acl

persistence true
persistence_location /var/lib/mosquitto/
max_queued_messages 1000          # Broker 侧上限；边缘侧仍要有本地有界队列

ACL 片段：

user dev-box-rk3588-07
topic write edgeai/grid-east/SS-220-01/box-rk3588-07/events/#
topic write edgeai/grid-east/SS-220-01/box-rk3588-07/frames/#
topic write edgeai/grid-east/SS-220-01/box-rk3588-07/telemetry
topic write edgeai/grid-east/SS-220-01/box-rk3588-07/status
topic read  edgeai/grid-east/SS-220-01/box-rk3588-07/cmd/#

user gw-104-readonly
topic read edgeai/grid-east/SS-220-01/+/events/#
topic read edgeai/grid-east/SS-220-01/+/status

NanoMQ / EMQX：等价关闭匿名、TLS 监听、按 clientid/device_id ACL；具体键名以所选版本文档为准（EMQX 5.9+ 许可见上表）。

5. 离线缓冲：本地队列 + 水位 + 补传

5.1 架构硬门禁

[解码] → [NPU 推理] → [Schema 适配] → [有界 Publish Queue] ──async──► MQTT Client
                              │                    │
                              │ 永不 join/阻塞      ├─ 在线：直发
                              ▼                    └─ 离线/背压：落盘缓冲
                         继续下一帧

规则：

1. 推理线程 只 try_enqueue；队列满 → 按丢弃策略，0 阻塞。
2. MQTT 客户端独占 IO 线程/协程；PUBLISH/reconnect 超时不得回调进推理线程。
3. 本地缓冲与 Broker max_queued_messages 解耦：边缘必须自有持久层。

5.2 存储选型

5.3 优先级与丢弃（水位）

定义优先级（高→低）：

1. P0 status / LWT 相关、MODEL_VERIFY_FAIL、关键 events（severity≥high 或点表已映射的 event.code）
2. P1 普通 events
3. P2 telemetry
4. P3 frames

水位示例（可按盘容量改）：

原则：优先丢 telemetry（与 frames），保事件。

5.4 恢复后补传顺序

1. 先发 status=online（retained）。
2. 按 ts_utc 升序、同刻按 优先级高先 补传；使用原 msg_id 便于云端幂等。
3. 限速（如 50–200 msg/s）避免打爆 Broker/4G；补传期间推理仍只走 try_enqueue。
4. 成功收到 PUBACK（QoS1）或本地确认策略后删行；失败退避重试，禁止忙等。
5. 断源过久：协议网关侧 104 点品质置「源不确定」（见点表模板），不得把最后一条告警当「已确认真实」无限保持。

5.5 缓冲伪代码骨架

# 示意：推理回调中禁止 network I/O
PRI = {"status": 0, "events": 1, "telemetry": 2, "frames": 3}

def on_infer_result(msg: dict, channel: str):
    item = pack(msg, channel)  # topic/qos/payload/msg_id/ts
    if not pub_q.try_enqueue(item, pri=PRI[channel]):
        metrics.inc(f"mqtt_drop_{channel}")
        # 不抛、不等、不重试阻塞
        return
    # 立刻返回，继续下一帧

# 独立线程
def mqtt_worker():
    while True:
        if client.connected:
            drain_online(pub_q, client)      # 先内存再 SQLite
        else:
            spill_to_sqlite(pub_q)            # 达黄/红水位执行丢弃策略
            client.reconnect_backoff()

6. 「断 MQTT 不堵推理」验收测法

否决：推理线程 join 网络调用、无界 std::queue、断线后 FPS 腰斩、恢复后内存涨死。

7. 与保护网 / 104 只读边界

摄像机/传感 ──► AI 边缘网关 ──AI VLAN──► MQTT Broker
                      │                      │
                      │                      ├─► 告警平台 / 对象存储
                      │                      └─► 协议网关（只订 events/status）
                      │                                │
                      │                                ▼
                      │                         IEC 60870-5-104 只读上送
                      │                         （YX/YC 镜像，品质可标 suspect）
                      x 禁止：MQTT cmd → 跳闸/软压板/GOOSE/定值

落地条款：

1. Broker、模型仓、推理盒与 保护装置不同二层域；交换机 ACL 默认 deny。
2. MQTT cmd 白名单：set_threshold / start_cam / stop_cam / reload_model / snap_now 等；无 trip/breaker/interlock 类 action。
3. 104 点表主键 = MQTT event.code；网关 只订不发。
4. 4G 备路仅解析 MQTT/OTA 域名；不承载保护业务。

8. 两周落地清单

9. 许可台账预填（须法务复核）

10. 与系列文档交叉引用

11. 小结

站端 MQTT 以 Mosquitto/NanoMQ 自托管 为主，EMQX/VerneMQ 按汇聚与许可审慎选用；Topic 严格对齐 Schema；TLS + 禁匿名 + 按设备 ACL；本地 SQLite/环形缓冲 且 优先丢 telemetry 保事件；推理路径 异步有界、断线 FPS 跌落 ≤5% 为否决项；MQTT 绝不充当遥控/保护总线，104 仅只读镜像。按两周清单可完成从 Broker 加固到断线验收的最小闭环。