后端系统工程 06：发送确认与可恢复的消息投递

本批导航：00 技术方案怎样接受验证｜01 API 契约与业务不变量｜02 负载模型与容量预算｜03 单库基线与故障实验工具｜04 持久化边界｜05 长连接接入｜06 发送确认｜07 排序与增量同步｜08 群消息扇出｜09 多设备已读｜10 网关故障与路由收敛｜11 撤回与集成验收

一条消息从发送方到接收方，要经过两段互相独立的确认：发送方需要知道“服务端已经接收并保存”，服务端需要知道“接收设备已经保存”。两段确认的承担者、观察位置和失败方式都不同。把它们混为一谈，常见的结果是用“已推送”代替“已送达”，或者把“已送达”的状态记在发送方的消息上，导致多设备场景下状态无法表达。

本篇把这两段确认分开处理。前一段由第 01 篇的幂等契约和第 04 篇的事务保证，本篇用三个崩溃点验证它；后一段引入设备送达游标和重投扫描，处理推送丢失的情况。

1. 需求与不变量

编号 不变量 观察位置
D1 发送方收到 sent{seq} 时，消息所在事务已经提交 events 行
D2 发送方没有收到回复时，用同一 cmid 重发，最终恰好存在一行 events 行数 + 重发回复的 dup
D3 设备送达游标 delivered_seq 只在设备确认后前进，表示“该设备已在本地保存 ≤ 该序号的全部可见事件” device_cursors
D4 送达游标单调，不超过会话 head device_cursors 与 conversations.next_seq
D5 推送丢失时，在有界时间内由服务端发现并触发补发 重投扫描 + 设备游标
D6 数据库进程崩溃后，已确认的消息仍然存在 崩溃恢复后的行数

D3 的定义决定了送达游标能表达什么、不能表达什么。它是“连续保存到哪里”，而不是“收到了哪些”。一台设备收到了序号 5 和 7 但缺 6 时，游标只能停在 5。第 07 篇说明客户端怎样维护这个连续性，这里先接受定义。

2. 规模假设

第 02 篇的模型给出：基线下推送约 9 067 帧/s，如果每条推送都确认一次，游标写入约 9 067 行/s，是消息写入量的 7 倍多，也是数据库写入超出预算的主要原因。确认合并到每 4 条一次，写入利用率就降到预算以内。因此 D3 的定义必须支持合并：确认 seq=n 隐含确认了 n 之前的全部事件。

重投比例假设为 3%。本篇的重投扫描每隔 IM_SWEEP_MS 检查一次所有在线连接，扫描本身对数据库的读取量与在线设备数和会话数成正比，在目标规模下需要单独评估（NOT_RUN）。

3. 接口

发送与确认使用第 01 篇定义的请求：

1
2
3
4
5
{"t":"send","rid":"r1","conv":"c-ab","cmid":"7f3c…","body":"…"}
{"t":"sent","rid":"r1","conv":"c-ab","seq":7,"mid":"…","dup":false,"epoch":1}

{"t":"ack","rid":"r2","conv":"c-ab","seq":7}
{"t":"acked","rid":"r2","conv":"c-ab","delivered":7}

服务端主动下发：

1
2
{"t":"event","ev":{"conv":"c-ab","seq":7,"kind":"msg","mid":"…","sender":"u-ana","cmid":"7f3c…","body":"…","at":…}}
{"t":"hint","conv":"c-ab","head":8,"why":"sweep"}

acked.delivered 返回合并后的游标值，而不是请求中的值。设备发送一个较小的确认时（例如重启后恢复了旧状态），回复中的值仍然是服务端记录的较大值，设备可以据此知道服务端的视图。

4. 数据模型

1
2
3
4
5
6
7
8
CREATE TABLE device_cursors (
    device_id     text NOT NULL REFERENCES devices,
    conv_id       text NOT NULL REFERENCES conversations,
    install_epoch integer NOT NULL,
    delivered_seq bigint NOT NULL CHECK (delivered_seq >= 0),
    updated_at    timestamptz NOT NULL DEFAULT clock_timestamp(),
    PRIMARY KEY (device_id, conv_id)
);

送达状态按 (设备, 会话) 存储一个数，而不是按 (设备, 消息) 存储一行。后者在第 02 篇的基线下意味着每秒新增约 9 000 行、保留 30 天，存储量远超消息本身。install_epoch 用于识别设备重装（第 09 篇）。

“某条消息是否已送达某台设备”可以由此推出：seq <= delivered_seq 即已送达。“某条消息已送达多少台设备”需要统计所有成员设备的游标，属于派生查询，本批没有实现对外接口。

5. 正常路径

1
2
3
4
5
6
7
8
9
10
11
12
13
图 6-1  发送确认与送达确认的两段路径
发送设备 A            网关                    PostgreSQL                    网关(接收侧)          接收设备 B
  | send(cmid)          |                          |                              |                      |
  |-------------------->| BEGIN…INSERT…notify      |                              |                      |
  |                     | COMMIT ----------------->| 提交后投递 im_events 通知 ---->|                      |
  | sent{seq,dup=false} |                          |                              | fanout(conv, seq)    |
  |<--------------------|                          |<-----------------------------| 查事件与可见用户     |
  |                     |                          |                              | event{seq} --------->|
  |                     |                          |                              |                      | 写本地存储并持久化
  |                     |                          |                              |<------- ack{seq} ----|
  |                     |                          |<-- UPSERT GREATEST(delivered)| acked{delivered}     |
  |                     |                          |                              |--------------------->|
        第一段：发送确认（D1、D2）                                第二段：送达确认（D3、D4）

两段之间通过 PostgreSQL 的 LISTEN/NOTIFY 衔接。pg_notify 在发送事务内调用，但 PostgreSQL 保证通知只在事务提交后投递，回滚的事务不产生通知；来自同一事务的通知按发送顺序投递，来自不同事务的通知按提交顺序投递。因此网关收到的每一条通知都对应一个已提交的事件，不会推送一条最终被回滚的消息。

通知只携带 conv:seq。网关收到后重新从数据库读取事件内容和此刻有权看到它的用户，再推送。不在通知中携带正文，一是因为通知载荷有长度限制（默认配置下必须小于 8000 字节），二是因为权限需要在读取时判定（第 08 篇）。

接收设备在本地持久化之后才发送确认。顺序反过来（先确认再持久化）的话，设备在两步之间崩溃，服务端认为已送达，设备本地却没有这条消息，而重投机制依据游标判断，不会再补发。

6. 方案比较

6.1 发送确认的时机

方案 回复时机 回复后网关崩溃 回复前网关崩溃
A. 收到即回复 写库之前 消息可能丢失，而发送方认为已发送 —
B. 写入消息队列后回复 队列确认后 取决于队列的持久化配置 发送方重发，需要幂等
C. 数据库提交后回复 COMMIT 返回后 消息已持久 发送方重发，由幂等保证一行

方案 C 是本实现的选择，也是 D1 的直接表达。它的成本是回复延迟包含一次同步提交（WAL 刷盘），这个延迟在本机实验中没有意义，未作测量。方案 B 把持久化边界移到队列，需要回答队列与数据库之间的一致性问题，本批不涉及。

6.2 送达状态的表示

方案 存储 合并确认 能回答“设备缺哪些”
每消息每设备一行 推送数 × 保留期 不能 能，但查询代价高
每消息一个已送达设备集合 正文行上的数组或位图 不能，且热点行争用 能
每设备每会话一个连续游标 设备数 × 会话数 能 能：seq > delivered_seq 的可见事件

6.3 推送丢失怎样发现

推送可能在多处丢失：网关在投递前崩溃、通知在监听连接断开期间到达、待发队列溢出（第 05 篇）、写入套接字后对端崩溃。

方案 发现者 时延 代价
逐条确认 + 网关内重传定时器 网关 一个重传超时 网关需要为每条推送保存状态；网关崩溃后状态丢失
客户端检测缺口 客户端 下一条推送到达时 会话最后一条消息丢失时无法发现
客户端定期同步 客户端 一个同步周期 所有会话定期查询，空转成本高
服务端扫描：游标落后于 head 时发提示 网关 一个扫描周期 扫描查询的成本

本实现组合了后两类中的两种：客户端检测缺口（第 07 篇），以及网关的重投扫描。扫描的规则只有一条：对每个已订阅的连接，若该设备在某会话上的送达游标小于 head，就发一个 hint。设备收到提示后执行同步，同步完成后确认，游标追上 head，下一轮扫描不再提示。

网关内重传定时器方案在网关崩溃时失效，因为重传状态只存在于网关内存中。扫描方案的依据（游标和 head）都在数据库中，任何网关都可以执行，网关崩溃不会丢失“谁还没收到”的信息。

7. 关键实现

送达确认：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// examples/im/src/im/Store.java（节选）
public long ackDelivered(Principal who, String convId, long seq) {
    return db.tx(c -> {
        if (currentInstallEpoch(c, who.deviceId()) != who.installEpoch())
            throw new ImException("STALE_INSTALL", "device was reinstalled");
        long head = head(c, convId);
        if (periods(c, convId, who.userId()).isEmpty())
            throw new ImException("NOT_MEMBER", "conversation not visible");
        if (seq < 0 || seq > head)
            throw new ImException("CURSOR_AHEAD", "seq=" + seq + " head=" + head);
        // INSERT ... ON CONFLICT (device_id, conv_id) DO UPDATE SET
        //   delivered_seq = GREATEST(device_cursors.delivered_seq, EXCLUDED.delivered_seq)
        // RETURNING delivered_seq
        ...
    });
}

INSERT ... ON CONFLICT DO UPDATE 在 PostgreSQL 中保证原子地得到“插入”或“更新”两种结果之一，即使在高并发下也是如此（PostgreSQL 文档 INSERT 一节）。GREATEST 保证同一设备的两个确认无论以何种顺序到达，结果都是较大者。

客户端发送路径中，持久发件箱在第一次发送之前写入：

1
2
3
4
5
6
7
8
9
10
11
12
// examples/im/src/im/Client.java（节选）
public Map<String, Object> send(String conv, String cmid, String body, long timeoutMs) ... {
    synchronized (outgoing) {
        outgoing.put(cmid, Json.map("conv", conv, "cmid", cmid, "body", body));
        persist();                     // 写临时文件后原子重命名
    }
    Map<String, Object> r = call(Json.map("t", "send", "conv", conv, "cmid", cmid, "body", body), timeoutMs);
    if ("sent".equals(r.get("t")) || isTerminal(r)) {
        synchronized (outgoing) { outgoing.remove(cmid); persist(); }
    }
    return r;
}

超时或连接断开时 call 抛出异常，outgoing 中的记录保留；重连后 resendPending 按原 cmid 逐条重发。测试客户端的本地持久化是“写临时文件、原子重命名”，没有调用 fsync：它能经受进程崩溃（页缓存仍在），不能经受断电。真实客户端应使用带事务的本地存储。

重投扫描：

1
2
3
4
5
6
7
8
9
10
11
// examples/im/src/im/Gateway.java（节选）
void sweep() {
    for (Conn conn : byConnId.values()) {
        if (!conn.subscribed) continue;
        for (Store.ConvState st : store.conversations(conn.who)) {
            if (st.delivered() < st.head()) {
                conn.offerJson(Json.map("t", "hint", "conv", st.conv(), "head", st.head(), "why", "sweep"));
            }
        }
    }
}

8. 验证

运行方式：真实 PostgreSQL 16；网关为独立 JVM 进程，通过 IM_FAULT=<点>:halt 在指定位置以 Runtime.halt(137) 终止；测试客户端在测试进程内，状态保存在临时文件中，模拟“客户端进程重启”时从文件重新加载。结果见 evidence/im-06/result.json。

8.1 三个崩溃点

每个崩溃点的流程相同：启动带故障的网关 → 客户端发送 → 网关在故障点终止 → 记录此时的行数 → 新建客户端实例（从状态文件加载）→ 连接到正常网关 → 重发发件箱。

检查 崩溃位置 网关退出码 首次发送 崩溃后行数 重发后行数 重发回复 dup 结果
06-crash-send.before_insert 插入之前 137 无回复 0 1 false PASS
06-crash-send.before_commit 插入之后、提交之前 137 无回复 0 1 false PASS
06-crash-api.after_send_commit 提交之后、回复之前 137 无回复 1 1 true PASS

从发送方的角度看，三种情况完全相同：连接断开，没有回复。区别只存在于数据库中，而重发的结果正确地反映了这个区别。第二行说明，INSERT 已经执行但事务未提交时进程终止，PostgreSQL 在连接断开后回滚事务，行数为 0，序号也不会被占用（第 04 篇的 P1）。第三行是 D2 中“恰好一行”最关键的情形：消息已存在，重发得到 dup=true 和原来的序号。每个检查还断言重发后发件箱为空。

8.2 其他用例

类型 检查 预期 观测
失败 06-lost-reply-retry 网关丢弃一次 send 的回复（gw.reply.send:drop:0:1）；客户端 1500 ms 超时后重发 首次 timeout，重发 dup=true，1 行
正常 06-push-ack-advances-cursor 接收设备收到推送并确认后，delivered_seq 等于消息序号 seq=7
边界 06-ack-monotonic-and-bounded 确认 seq=1 不使游标回退；确认 seq=107（超过 head）被拒绝 回复 delivered=7；CURSOR_AHEAD
恢复 06-lost-push-swept 网关丢弃全部推送（gw.push:drop），扫描间隔 500 ms；3000 ms 内设备同步并确认到新序号 449 ms 修复，delivered_seq=8
恢复 06-db-immediate-stop-keeps-acked pg_ctl -m immediate 停止数据库后重启；已确认的消息行数不变 6 → 6

06-lost-push-swept 的观测中 pushes_applied=1，这个 1 来自扫描之前的阶段，与本次修复无关：本次的消息完全通过“扫描提示 → 同步”到达。449 ms 小于一个扫描间隔加同步往返的上界，具体数值受扫描周期的相位影响，没有统计意义。这个检查在第一次全量运行时曾因断言竞态失败，记录见 evidence/im-06-flaky/README.md 与第 03 篇。

06-db-immediate-stop-keeps-acked 只验证了“数据库进程崩溃、操作系统未崩溃”的情形。由于 fsync=on 且 synchronous_commit=on，提交返回前 WAL 已刷盘，因此断电情形下也应成立，但本批没有模拟断电（NOT_RUN）。如果改为 synchronous_commit=off，PostgreSQL 文档说明最近的已提交事务可能在崩溃中丢失，D1 和 D6 都将不再成立。

9. 故障与恢复

故障 发送方 接收方 恢复路径
网关在提交前崩溃 无回复 — 发送方重连后重发，得到新序号
网关在提交后、回复前崩溃 无回复 推送可能已发出，也可能没有 发送方重发得到 dup=true；接收方由扫描或同步补齐
回复帧丢失 超时 正常 重发得到 dup=true
推送帧丢失 正常 未收到 扫描发现游标落后，发提示，接收方同步
接收设备在保存前崩溃 正常 未确认 重连后从本地游标同步
接收设备在保存后、确认前崩溃 正常 已保存，未确认 重连后同步得到重复事件，按序号去重；再确认
确认帧丢失 正常 已保存 扫描再次提示，同步结果为空或重复，再次确认
数据库进程崩溃 进行中的请求得到 UNAVAILABLE 或无回复 — WAL 恢复后重发

表中有三处“重复”：重发、重复推送、重复同步。所有重复都由同一个机制吸收：发送侧靠请求键，接收侧靠事件序号。客户端本地存储以序号为键，同一事件应用两次结果不变。这就是“至少一次投递 + 幂等应用”的组合，第 07 篇在排序与分页上继续使用它。

10. 限制

  • 重投扫描每轮查询每个在线连接的全部会话状态，查询量与在线设备数和会话数的乘积成正比。在第 02 篇的规模下需要改为只扫描最近有写入的会话，或由数据库维护“落后设备”索引。本批未实现，也未测量。
  • 扫描只对已订阅（发送过 sub）的连接生效。
  • 发送回复延迟中的提交耗时未测量；本机数值不能代表目标环境。
  • 断电、磁盘故障、主备切换（U2）均为 NOT_RUN。

下一篇处理接收侧的顺序问题：推送可能乱序到达，同步可能与新消息并发，客户端怎样维护连续游标，分页怎样避免重复与遗漏。

参考资料