离线时点下的“发送”
用户在会话页输入一条消息，点发送，界面立刻显示“发送中”，但网络正好断了。此刻应用面对的不是“请求失败”，而是“请求还没发生”：消息必须留在本地，等网络回来再补发。类似场景还有草稿提交、收藏、地址修改、订单备注，凡是用户在离线状态下表达的写意图，都不能因为一次网络失败而消失。
写路径比读路径难，因为它要处理三个读路径没有的问题：顺序（先创建的实体不能后于它的更新到达）、重复（超时重试可能让同一次写入执行两次）、冲突（离线期间服务端可能已被同一用户或其他人修改）。这一节把三者收进一个队列模型，并给出可运行的实现与测试。
文中出现的时长、版本号与重试次数都是案例假设值，用来演示判断过程，不是任何真实服务的协议。
队列要守住的三个不变量
第一，先落盘再确认。用户点发送后，队列条目必须在 UI 显示“已排队”之前写入存储。进程随后被杀，重启后仍能读回这条写入。反过来，“先发请求，成功后再补日志”在崩溃窗口里会丢数据，而且无法判断请求是否已经到达服务端。
第二，重放顺序稳定。队列按 seq 严格先进先出，跨实体也不重排。这样“创建笔记”一定先于“更新同一笔记”到达服务端。代价是队头阻塞：一个持续失败的条目会挡住后面的条目。案例选择接受这个代价，因为端上队列通常很短，顺序错误造成的数据问题比多等一会儿更严重。
第三，幂等键在入队时生成。键由客户端标识、实体、实体标识与队列序号组成，一旦写入就不再变化。重试、重启、换网络都复用同一个键，服务端据此去重。每次重试重新生成键，等于告诉服务端“这是一次新的写入”，重复就不可避免。
重放顺序与幂等键
重放是一个循环：取队头、增加尝试次数并落盘、发送、根据结果决定删除或保留。网络错误中止本轮，保留剩余条目；服务端返回 duplicate 与 applied 一样处理，因为它说明上一次请求其实已经成功，只是响应丢了。
严格 FIFO 的边界要说清楚：如果队头条目因为服务端 5xx 持续失败，后面的条目会一直等待。生产实现通常加两个约束：单条最大尝试次数，以及“可并行的实体分组”。前者防止死循环，后者允许不同实体的写入并行，但同一实体的写入仍然串行。案例刻意只实现 FIFO，是为了让顺序与幂等这两个不变量先被测试固定下来。
幂等键不是“请求 ID”。请求 ID 通常每次发送都变，用于链路追踪；幂等键标识的是一次业务意图，必须在整个重试生命周期内保持不变。两者可以同时存在，但不能混用。
冲突策略与 LWW 的边界
服务端返回冲突时，客户端必须选一个策略。案例给出三种，按资源的协作强度选择：
| 策略 | 行为 | 适用 |
|---|---|---|
server-wins | 放弃本地修改，记录冲突 | 权限、配置、他人可编辑的共享资源 |
client-wins-if-newer | 本地写入严格晚于服务器时带新版本重放 | 单用户私有数据，允许最后写入者覆盖 |
merge-disjoint | 字段无重叠时合并，重叠时回落 LWW | 多字段表单，双方通常改不同字段 |
last-write-wins 最容易被误用。它的“最后”必须有一个可信时间源。案例用的是 mutation.queuedAt（本地入队时间）与 server.updatedAt（服务端更新时间）比较，这两个时钟来自不同设备，存在偏移。因此 LWW 只适合“同一用户的私有数据”，并且比较必须是严格大于：本地时间等于服务器时间时判服务器胜。相等意味着无法证明本地更新，宁可不覆盖。
merge-disjoint 的判据是字段级的三方比较：本地 base 里的旧值、本地 payload 里的新值、服务端当前值。如果服务端当前值仍等于本地看到的旧值，说明这个字段没被服务端改过，可以安全写入；如果不等，说明双方都改了同一字段，合并失败，回落到 LWW。合并结果以服务端数据为底稿，只覆盖安全字段，避免把服务端新增的字段删掉。
队列实现
下面的实现把“先落盘、严格 FIFO、稳定幂等键、三种冲突策略”收进一个类。存储层只要求字符串读写，时间从外部注入，测试不需要真实网络或定时器。
// file: src/data/sync-queue.ts
// tier: L1
// verify: node --test test/data/sync-queue.test.ts
export type MutationKind = "create" | "update" | "delete";
export type ConflictPolicy = "server-wins" | "client-wins-if-newer" | "merge-disjoint";
export type ConflictAction = "retry" | "merged" | "drop";
export interface MutationInput {
entity: string;
entityId: string;
kind: MutationKind;
payload: Record<string, unknown>;
base: Record<string, unknown>;
baseVersion: number;
}
export interface Mutation extends MutationInput {
seq: number; idempotencyKey: string; queuedAt: number; attempts: number; force: boolean;
}
export interface ServerRecord {
id: string; version: number; updatedAt: number; data: Record<string, unknown>;
}
export type SendResult =
| { status: "applied"; record: ServerRecord }
| { status: "duplicate"; record: ServerRecord }
| { status: "conflict"; record: ServerRecord };
export interface Transport {
send(mutation: Mutation): Promise<SendResult>;
}
/** 只要求字符串读写，真实实现可以包 localStorage 或 SQLite。 */
export interface QueueStore {
load(): string | undefined;
save(raw: string): void;
}
export interface ConflictEvent {
idempotencyKey: string; strategy: ConflictPolicy; action: ConflictAction;
serverVersion: number; reason: string;
}
export interface ReplayReport {
applied: string[]; conflicts: ConflictEvent[];
remaining: number; stoppedBy: "network" | null;
}
export interface QueueOptions {
clientId: string; now: () => number; store: QueueStore;
}
interface Resolution {
action: ConflictAction; payload?: Record<string, unknown>;
event: ConflictEvent;
}
/**
* 离线写入队列：先落盘再发送，严格按入队顺序重放。
* 幂等键在入队时生成并持久化，重试与重启都不会变化。
*/
export class SyncQueue {
private readonly clientId: string;
private readonly now: () => number;
private readonly store: QueueStore;
private items: Mutation[] = [];
private seq = 0;
constructor(options: QueueOptions) {
this.clientId = options.clientId;
this.now = options.now;
this.store = options.store;
const raw = this.store.load();
if (raw === undefined) return;
try {
const saved = JSON.parse(raw) as { seq: number; items: Mutation[] };
this.seq = saved.seq;
this.items = saved.items;
} catch {
this.items = [];
this.seq = 0;
}
}
get size(): number {
return this.items.length;
}
peek(): Mutation | undefined {
return this.items[0];
}
enqueue(input: MutationInput): Mutation {
const seq = this.seq + 1;
this.seq = seq;
const mutation: Mutation = {
...input,
payload: { ...input.payload },
base: { ...input.base },
seq,
idempotencyKey: [
this.clientId,
input.entity,
input.entityId,
String(seq),
].map(encodeURIComponent).join(":"),
queuedAt: this.now(),
attempts: 0,
force: false,
};
this.items.push(mutation);
this.persist();
return mutation;
}
/**
* 顺序重放。网络错误中止本轮并保留剩余条目；冲突按策略处理。
* 每次发送前先持久化 attempts，进程在发送途中被杀也能看出尝试过。
*/
async replay(transport: Transport, policy: ConflictPolicy): Promise<ReplayReport> {
const applied: string[] = [];
const conflicts: ConflictEvent[] = [];
let stoppedBy: "network" | null = null;
while (this.items.length > 0) {
const head = this.items[0];
head.attempts += 1;
this.persist();
let result: SendResult;
try {
result = await transport.send({ ...head, payload: { ...head.payload }, base: { ...head.base } });
} catch {
stoppedBy = "network";
break;
}
if (result.status === "applied" || result.status === "duplicate") {
applied.push(head.idempotencyKey);
this.items.shift();
this.persist();
continue;
}
const resolution = this.resolve(head, result.record, policy);
conflicts.push(resolution.event);
if (resolution.action === "drop") {
this.items.shift();
this.persist();
continue;
}
const rebased: Mutation = {
...head,
payload: resolution.action === "merged" ? (resolution.payload ?? head.payload) : head.payload,
base: { ...result.record.data },
baseVersion: result.record.version,
force: true,
};
this.items[0] = rebased;
this.persist();
let retried: SendResult;
try {
retried = await transport.send({ ...rebased, payload: { ...rebased.payload }, base: { ...rebased.base } });
} catch {
stoppedBy = "network";
break;
}
if (retried.status === "applied" || retried.status === "duplicate") {
applied.push(rebased.idempotencyKey);
this.items.shift();
this.persist();
continue;
}
conflicts.push({
idempotencyKey: rebased.idempotencyKey,
strategy: policy,
action: "drop",
serverVersion: retried.record.version,
reason: "重试后仍冲突，停止自动合并",
});
this.items.shift();
this.persist();
}
return { applied, conflicts, remaining: this.items.length, stoppedBy };
}
private resolve(
mutation: Mutation,
server: ServerRecord,
policy: ConflictPolicy,
): Resolution {
const event = (action: ConflictAction, reason: string): ConflictEvent => ({
idempotencyKey: mutation.idempotencyKey,
strategy: policy,
action,
serverVersion: server.version,
reason,
});
if (policy === "server-wins") {
return { action: "drop", event: event("drop", "策略 server-wins，放弃本地变更") };
}
if (policy === "client-wins-if-newer") {
if (mutation.queuedAt > server.updatedAt) {
return { action: "retry", event: event("retry", "本地写入严格晚于服务器，带新版本重放") };
}
return {
action: "drop",
event: event("drop", `LWW 边界：本地 ${mutation.queuedAt} 不晚于服务器 ${server.updatedAt}`),
};
}
const merged: Record<string, unknown> = { ...server.data };
const overlapping: string[] = [];
for (const [field, value] of Object.entries(mutation.payload)) {
if (mutation.base[field] === server.data[field]) merged[field] = value;
else overlapping.push(field);
}
if (overlapping.length === 0) {
return { action: "merged", payload: merged, event: event("merged", "字段无重叠，按字段合并") };
}
if (mutation.queuedAt > server.updatedAt) {
return {
action: "retry",
event: event("retry", `字段重叠 ${overlapping.join(",")}，回落 LWW 且本地较新`),
};
}
return {
action: "drop",
event: event("drop", `字段重叠 ${overlapping.join(",")}，回落 LWW 且服务器较新`),
};
}
private persist(): void {
this.store.save(JSON.stringify({ seq: this.seq, items: this.items }));
}
}
几个实现细节值得注意。
冲突重试只做一次。 第一次冲突按策略重放；如果重放后仍冲突，直接丢弃并记录原因，不再无限重试。冲突通常意味着有人持续修改同一资源，客户端继续自动合并只会制造更多冲突。丢弃不是“数据没救”，事件已经写进报告，产品可以选择提示用户重新编辑。
重放前先写 attempts。 发送途中进程被杀，重启后能看到这条已经尝试过一次。它不改变幂等键，但为“最大尝试次数”提供了持久化依据。
合并后的 base 更新为服务端当前数据。 这样下一次冲突比较的是最新基线，而不是入队时的旧快照。force 标记告诉传输层“客户端已经做过冲突处理”，服务端可以据此决定是否跳过版本校验，但最终仍由服务端裁决。
传输层返回 duplicate 时不抛错。 这是幂等设计的核心：重复不是异常，而是“上一次其实成功了”。把它当失败重试，才会产生真正的重复。
测试与真实输出
测试覆盖 FIFO、重启持久化、网络中断续传、重复响应、LWW 等值边界、本地更新重放、字段合并、重叠回退与 server-wins 九个场景。传输层是脚本化的，按顺序返回预设结果，不依赖真实服务：
node --test test/data/sync-queue.test.ts
本机一次运行的真实输出如下，耗时随机器变化：
✔ 按入队顺序重放，全部成功后队列清空 (1.523292ms)
✔ 重启后队列与幂等键保持不变 (0.348125ms)
✔ 网络错误中止本轮，剩余条目与幂等键保留 (0.2315ms)
✔ 服务端返回 duplicate 视为成功，避免重复写入 (0.119833ms)
✔ LWW 边界：本地时间等于服务器时间时服务器胜 (0.254125ms)
✔ 本地严格更新时带服务器版本重放 (0.164416ms)
✔ 字段无重叠时合并双方修改 (0.268083ms)
✔ 字段重叠且服务器更新时放弃本地修改 (0.130125ms)
✔ server-wins 策略不重放，直接丢弃本地变更 (0.164041ms)
ℹ tests 9
ℹ suites 0
ℹ pass 9
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 357.542167
一次重复下单的事故
某次版本里，下单请求超时后自动重试了一次。第一次请求其实已经到达服务端并创建订单，只是响应在弱网里丢了；客户端重试时没有带幂等键，服务端把它当成第二笔订单。用户回到订单列表，看到两笔相同的待支付订单。
复盘发现三个问题叠加：
- 客户端把“超时”当成“失败”，没有区分“不确定”。
- 重试没有幂等键，服务端无法识别这是同一次业务意图。
- 服务端创建订单的接口不是幂等的，重试必然产生新记录。
修复分三层：客户端在入队时生成稳定幂等键并持久化；服务端在去重窗口内按幂等键返回第一次的结果；产品层在订单列表里合并展示疑似重复的待支付订单，给用户一个取消入口。三层缺一不可，只改客户端不能解决服务端已经重复创建的事实，只改服务端也无法覆盖旧版本客户端。
这个事故说明，离线队列不是“把请求存起来慢慢发”，而是“把业务意图变成可重复提交的幂等操作”。没有幂等键的队列，在网络不确定的环境里只是一个延迟爆炸的重复写入器。
与服务端事务的对照
服务端有事务，客户端没有。两者的对应关系可以列成一张表：
| 维度 | 服务端事务 | 客户端同步队列 |
|---|---|---|
| 原子性 | 多步操作要么全成要么全败 | 逐条提交，失败保留剩余 |
| 隔离性 | 锁与隔离级别 | 无，冲突靠版本与策略 |
| 持久性 | 数据库日志 | 本地存储，可能被系统清理 |
| 重试 | 通常由上层重试整个事务 | 逐条重放，依赖幂等键 |
| 回滚 | 撤销未提交变更 | 只能补偿，不能真正回滚 |
客户端能承诺的是“最终一致”：每条写入至少提交一次，重复由幂等键消解，冲突由策略裁决。它不能承诺原子性，也不应该假装能回滚。产品设计要接受这个边界：离线期间的多步操作，最终可能只有一部分成功，UI 必须能表达“部分完成”。
小结
离线写入队列守住三个不变量：先落盘再确认、严格顺序重放、幂等键入队即固定。三者共同保证“至少一次提交，重复可消解”。冲突策略是第二层：server-wins 适合共享资源，client-wins-if-newer 适合单用户私有数据，merge-disjoint 适合多字段表单；LWW 必须使用严格大于，并且只在可信时间源范围内使用。
队列不能替代服务端事务。它把“网络不确定”转换成“可重放的业务意图”，剩下的原子性、隔离性与去重要由服务端配合完成。下一章从数据层转向组件体系，讨论设计系统如何把端上的交互契约固定下来。
阅读导航：上一节：7.2 端上缓存与离线优先 · 下一节：8.1 设计系统与组件契约 。
常见错误与边界
- 每次重试重新生成幂等键。服务端会把它当成新的写入，重复无法避免。
- 队列只放内存。进程被系统回收后，用户离线期间的写入全部丢失。
- 先发送成功再补日志。崩溃窗口里既不知道请求是否到达，也无法恢复。
- 冲突时直接覆盖服务端。可能抹掉其他人的修改，且无法审计。
- 用客户端时钟做全局 LWW。设备时间可被修改，跨用户比较没有意义。
- 合并时用本地快照整体替换。服务端新增的字段会被静默删除。
- 重放非幂等接口且不做服务端去重。超时重试必然产生重复记录。
- 忽略队头阻塞。一个持续失败的条目会让整个队列停摆。
- 把队列当事务。离线多步操作可能部分成功，UI 必须能表达。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。