HarmonyOS 7 ArkTS：离线消息Schema演进与幂等重放

李游Leo

13人浏览 · 2026-10-07 20:32:02

李游Leo 鸿蒙领航者 · 2026-10-07 20:32:02 发布

ProtoRelay 是一个离线笔记同步 Demo。手机断网时把编辑操作编码成 Protobuf 消息，联网后按顺序重放。它稳定跑了几周，直到服务端先上线 v3：消息多了 workspaceColor 和 conflictPolicy 两个字段，仍在使用 v2 Schema 的旧客户端接收后，再次转发给另一台新设备，这两个字段竟然消失了。

没有崩溃，没有解析错误，ACK 也正常。真正的问题是旧客户端把“自己不认识的字段”悄悄丢掉，再用 v2 对象重新编码。线上最难排的往往就是这种成功链路里的数据损失。最后我没有让旧版本理解 v3 业务，而是保留原始 payload 与 unknown fields，给每条消息增加幂等账本，并把跨版本重放做成固定回归。

一、兼容不只是“旧版能解码新版消息”

Protobuf 的字段编号允许新旧 Schema 共存。v2 客户端遇到 v3 新字段时，通常仍能读取自己认识的字段，所以第一轮测试很容易得出“兼容”的结论。可我们的链路不是只读：旧设备接收、落盘、恢复网络后还会重新编码并转发。

如果 decode 后只保留已知对象，再 encode，未知字段可能消失。对旧客户端来说这似乎无所谓，对下游 v3 设备却是不可逆的数据降级。workspaceColor 丢失会让协作空间恢复默认颜色，conflictPolicy 丢失则可能改变冲突合并策略。

本轮回归批次为 PR-1002-0514，目标消息 msg_20261002_0514_007，发送端 Schema v3，当前客户端 Schema v2。离线队列 37 条，要求重放 37 / 37、重复提交 0、未知字段保留 2，最终状态 ACKED。

二、先把业务消息和传输信封分开

早期实现直接把 NoteMutation 编码结果写文件，文件名就是时间戳。重试时只能重新解析业务对象，既没有稳定 messageId，也没有原始字节的校验值。修复后增加 OfflineEnvelope：信封由当前客户端完全理解，业务 payload 作为 bytes 原样包裹。

下面的 Schema 解决的是“旧客户端必须转发，却不应改写未知业务字段”。信封字段编号长期稳定；payload 内部的 NoteMutationV3 可以继续演进。

syntax = "proto3";
package relay;

message OfflineEnvelope {
  string messageId = 1;
  uint32 schemaVersion = 2;
  int64 createdAtMs = 3;
  bytes payload = 4;
  bytes payloadSha256 = 5;
  uint32 retryCount = 6;
}

message NoteMutationV3 {
  string noteId = 1;
  string operation = 2;
  string content = 3;
  string workspaceColor = 8;
  string conflictPolicy = 9;
}

字段 4～7 即使暂时未用，也不会拿来复用旧语义。Protobuf 兼容的底线不是“字段名没变”，而是字段编号和 wire type 的含义不能偷偷改。删除字段后应该保留编号，避免后来的人把同一个 tag 分配给另一种数据。

信封的哈希针对原始 payload，而不是 decode 后的对象。只要旧客户端没有改业务内容，转发前后的 SHA-256 就应该一致。这条断言比比较 JSON 更稳，因为 JSON 化会把默认值、bytes 和 64 位数转换成另一种表现形式。

三、protobufjs 的 unknown fields 要显式纳入策略

当前 protobufjs 支持通过 Reader 控制是否丢弃未知字段。项目没有把全局默认改掉，而是在需要跨版本转发的解码器中设置 discardUnknown=false。这样普通只读消息仍采用默认策略，离线中继消息则保留未知 wire 数据。

这段解码器解决的是 v2 客户端读取 v3 payload 时保留 tag 8 和 tag 9。即便 UI 不认识这两个字段，重新编码时也不能把它们吞掉。

import { Reader } from 'protobufjs/minimal'
import { NoteMutationV2 } from './generated/note_v2'

export interface UnknownAwareMutation extends NoteMutationV2 {
  $unknowns?: Uint8Array[]
}

export function decodeForRelay(payload: Uint8Array): UnknownAwareMutation {
  const reader = Reader.create(payload)
  reader.discardUnknown = false
  return NoteMutationV2.decode(reader) as UnknownAwareMutation
}

export function unknownFieldCount(message: UnknownAwareMutation): number {
  return message.$unknowns?.length ?? 0
}

这里不能先 toObject() 再持久化。plain object 是业务互操作边界，unknown fields 并不属于 v2 类型定义，转换时很容易消失。队列里始终保存原始 payload；解码对象只用于展示已知字段和做安全校验。

如果未来切换 protobufjs 版本，$unknowns 的具体表现需要重新验证，不能把内部属性当永久业务协议。工程上的保险仍是原始 payload 与哈希：即便库的 unknown-field 表示变化，我们也能原样转发，并通过端到端回归发现差异。

四、离线队列保存的是事实，不是临时对象

队列项包含 messageId、Schema 版本、原始信封、payload 哈希、状态、下一次重试时间和最近错误码。状态只能按 QUEUED → DECODING_V2 → REPLAYING → ACKED 前进；网络异常时从 REPLAYING 回到 QUEUED，但 messageId 不变。

这段仓储入口解决的是应用被杀、重复启动和同一消息多次入队的问题。插入使用 messageId 做唯一键；相同 ID、相同哈希视为重复到达，直接返回现有记录；相同 ID、不同哈希则标记为冲突，绝不能覆盖。

export async function enqueueEnvelope(raw: Uint8Array): Promise<QueueRecord> {
  const envelope = OfflineEnvelope.decode(raw)
  const actualHash = sha256(envelope.payload)
  if (!bytesEqual(actualHash, envelope.payloadSha256)) {
    throw new Error('PAYLOAD_HASH_MISMATCH')
  }

  const existing = await ledger.findByMessageId(envelope.messageId)
  if (existing) {
    if (!bytesEqual(existing.payloadSha256, actualHash)) {
      throw new Error('MESSAGE_ID_COLLISION')
    }
    return existing
  }

  return ledger.insert({
    messageId: envelope.messageId,
    schemaVersion: envelope.schemaVersion,
    rawEnvelope: raw,
    payloadSha256: actualHash,
    state: 'QUEUED',
    retryCount: 0
  })
}

写入数据库时 rawEnvelope 使用二进制列，不做 Base64 往返。Base64 可以用于日志摘要或调试导出，但会增加体积，也容易在文本层被换行、裁剪。日志只打印 messageId、schemaVersion、字节数和哈希前 8 位，不输出笔记正文。

页面退出不删除队列，Ability 销毁也不把 REPLAYING 直接改成 ACKED。启动恢复时，超过租约时间的 REPLAYING 记录回到 QUEUED，然后继续重放；这比在生命周期回调里猜网络请求是否成功可靠。

五、幂等不是客户端“记得发过”，而是双方认同同一个 ID

重放器一次领取 10 条记录，为每条记录申请 30 秒租约。服务端以 messageId 去重：第一次写入返回 APPLIED，重复到达返回 ALREADY_APPLIED，两者都可以转成 ACKED。客户端不能因为 HTTP 200 就删除消息，还要核对响应中的 messageId 与 payload 哈希。

下面的循环解决的是断网重连时重复回调和迟到 ACK。每次执行都有 generation；用户切换账号或页面销毁后，旧代响应只能释放租约，不能提交当前账本。

export class ReplayCoordinator {
  private generation: number = 0

  async replayPending(): Promise<void> {
    const runGeneration = ++this.generation
    const batch = await ledger.leaseQueued(10, 30_000)
    for (const item of batch) {
      await ledger.move(item.messageId, 'REPLAYING')
      try {
        const ack = await relayApi.push(item.rawEnvelope)
        if (runGeneration !== this.generation) break
        if (ack.messageId !== item.messageId || ack.payloadHash !== hex(item.payloadSha256)) {
          throw new Error('ACK_MISMATCH')
        }
        await ledger.move(item.messageId, 'ACKED')
      } catch (error) {
        await ledger.requeueWithBackoff(item.messageId, String(error))
      }
    }
  }

  cancel(): void {
    this.generation++
  }
}

循环没有 Promise.all()。离线编辑有顺序语义，同一 noteId 的 PATCH 必须按序提交；不同 noteId 可以在分组后并行，但要保证每组单通道。盲目并发会把“传输成功”变成“业务顺序错乱”。

退避使用 2、4、8、16、32 秒并加少量抖动，上限 5 分钟。永久错误如 SCHEMA_UNSUPPORTED 不再自动重试，而是进入 NEEDS_UPGRADE，避免后台无限耗电。

六、跨版本回归必须走一遍 v3→v2→v3

单元测试若只做 v2 自己的 encode/decode，看不见未知字段。回归夹具先用 v3 Schema 生成消息，确认 tag 8 和 tag 9 存在；再交给 v2 解码器读取并持久化；最后将队列中的原始 payload 交给 v3 解码，检查两个字段仍为 #4A6CF7 和 MERGE_BY_CLOCK。

第一次测试里，我们故意走旧路径：v2 decode 后调用 toObject()，再用 fromObject() 与 encode() 生成 payload，结果 unknown fields 从 2 变成 0，报告状态 DATA_LOSS_BLOCKED。改为原始 payload 中继后，哈希前后相同，字段保留数恢复为 2。

回归还模拟了三种时序：请求已到服务端但 ACK 丢失、应用在 REPLAYING 时被杀、同一消息在网络恢复回调中被触发两次。最终 37 条全部 ACK，服务端实际应用 37 次，重复写入 0。

七、最终页面把兼容链路完整摊开

ReplayLedgerPage 不做聊天界面，而是把最难核对的证据放在同一屏：批次 PR-1002-0514、目标消息 msg_20261002_0514_007、发送 Schema v3、当前解码 v2、保留未知字段 2、队列 37、已重放 37 / 37、重复写入 0。

页面底部状态为 ACKED，路径显示 QUEUED → DECODING_V2 → REPLAYING → ACKED。payload 哈希前缀 8f3a21c7 在入队、发送和 ACK 三处一致。状态栏时间固定为 05:14，HiLog 也使用同一批次和 messageId。