《互联网产品的客户端架构·内容、社交与实时互动》4.2 断点续传与失败恢复

// file: src/upload/resume.ts
// tier: L2
// verify: node --test --test-concurrency=1 test/upload/resume.test.ts

export interface FileFingerprint { name: string; size: number; lastModified: number }

export interface UploadSession {
  sessionId: string;
  fingerprint: FileFingerprint;
  chunkSize: number;
  totalBytes: number;
  confirmed: number[];
  updatedAt: number;
  expiresAt: number;
}

export type RestoreAction = "resume" | "restart" | "expired" | "mismatch";

export interface RestoreDecision {
  action: RestoreAction;
  session: UploadSession | null;
  confirmedChunks: number;
  missingChunks: number[];
  reason: string;
}

export type UploadErrorKind =
  | "network" | "timeout" | "server" | "rate-limit" | "auth" | "quota" | "checksum";

export interface RetryPolicy { baseDelayMs: number; maxDelayMs: number; maxAttempts: number }
export interface RetryDecision { retry: boolean; delayMs: number; reason: string }

const RETRYABLE = new Set<UploadErrorKind>(["network", "timeout", "server", "rate-limit", "checksum"]);

export function createSession(input: {
  sessionId: string;
  file: FileFingerprint;
  chunkSize: number;
  now: number;
  ttlMs: number;
}): UploadSession {
  if (!Number.isSafeInteger(input.file.size) || input.file.size < 0) {
    throw new RangeError("文件大小必须是非负安全整数");
  }
  if (!Number.isSafeInteger(input.chunkSize) || input.chunkSize <= 0) {
    throw new RangeError("chunkSize 必须是正安全整数");
  }
  return {
    sessionId: input.sessionId,
    fingerprint: { ...input.file },
    chunkSize: input.chunkSize,
    totalBytes: input.file.size,
    confirmed: [],
    updatedAt: input.now,
    expiresAt: input.now + input.ttlMs,
  };
}

export function totalChunks(session: UploadSession): number {
  return Math.ceil(session.totalBytes / session.chunkSize);
}

export function missingChunks(session: UploadSession): number[] {
  const confirmed = new Set(session.confirmed);
  const missing: number[] = [];
  for (let index = 0; index < totalChunks(session); index += 1) {
    if (!confirmed.has(index)) missing.push(index);
  }
  return missing;
}

/** 确认是幂等的：重复 ack 不改变清单，乱序 ack 只按索引记账。 */
export function applyAck(
  session: UploadSession,
  ack: { index: number; now: number },
): { session: UploadSession; added: boolean } {
  const total = totalChunks(session);
  if (!Number.isSafeInteger(ack.index) || ack.index < 0 || ack.index >= total) {
    throw new RangeError(`分片索引 ${ack.index} 超出 0..${total - 1}`);
  }
  if (session.confirmed.includes(ack.index)) return { session, added: false };
  const confirmed = [...session.confirmed, ack.index].sort((a, b) => a - b);
  return { session: { ...session, confirmed, updatedAt: ack.now }, added: true };
}

export function restoreSession(input: {
  snapshot: UploadSession | null;
  file: FileFingerprint;
  now: number;
}): RestoreDecision {
  const { snapshot } = input;
  if (snapshot === null) {
    return { action: "restart", session: null, confirmedChunks: 0, missingChunks: [], reason: "本地没有可恢复的会话" };
  }
  if (snapshot.expiresAt <= input.now) {
    return {
      action: "expired", session: null, confirmedChunks: snapshot.confirmed.length,
      missingChunks: [], reason: "服务端会话已过期，本地清单不再可信",
    };
  }
  const sameFile = snapshot.fingerprint.name === input.file.name
    && snapshot.fingerprint.size === input.file.size
    && snapshot.fingerprint.lastModified === input.file.lastModified;
  if (!sameFile) {
    return {
      action: "mismatch", session: null, confirmedChunks: snapshot.confirmed.length,
      missingChunks: [], reason: "文件已变化，旧分片不能拼到新文件上",
    };
  }
  const missing = missingChunks(snapshot);
  return {
    action: "resume",
    session: snapshot,
    confirmedChunks: snapshot.confirmed.length,
    missingChunks: missing,
    reason: missing.length === 0 ? "分片已全部确认，直接提交" : `从第 ${missing[0]} 片继续`,
  };
}

/** 恢复时以服务端清单为准：本地记了但服务端没有的分片必须重传。 */
export function reconcileWithServer(
  session: UploadSession,
  serverConfirmed: number[],
  now: number,
): { session: UploadSession; dropped: number[]; added: number[] } {
  const total = totalChunks(session);
  const server = new Set(
    serverConfirmed.filter((index) => Number.isSafeInteger(index) && index >= 0 && index < total),
  );
  const local = new Set(session.confirmed);
  const valid = [...server].sort((a, b) => a - b);
  return {
    session: { ...session, confirmed: valid, updatedAt: now },
    dropped: session.confirmed.filter((index) => !server.has(index)),
    added: valid.filter((index) => !local.has(index)),
  };
}

export function decideRetry(input: {
  kind: UploadErrorKind;
  attempt: number;
  policy: RetryPolicy;
  random?: () => number;
  retryAfterMs?: number;
}): RetryDecision {
  const { kind, attempt, policy } = input;
  if (!RETRYABLE.has(kind)) {
    return { retry: false, delayMs: 0, reason: `${kind} 不是暂时性错误，重试不会改变结果` };
  }
  if (attempt >= policy.maxAttempts) {
    return { retry: false, delayMs: 0, reason: `已连续失败 ${attempt} 次，达到上限` };
  }
  if (kind === "rate-limit" && (input.retryAfterMs ?? 0) > 0) {
    return { retry: true, delayMs: input.retryAfterMs as number, reason: "遵守服务端 Retry-After" };
  }
  const exponential = Math.min(policy.maxDelayMs, policy.baseDelayMs * 2 ** (attempt - 1));
  const random = input.random ?? Math.random;
  const jitter = Math.min(1, Math.max(0, random()));
  const delayMs = Math.min(policy.maxDelayMs, Math.round(exponential * (0.5 + 0.5 * jitter)));
  return { retry: true, delayMs, reason: `第 ${attempt} 次失败，${delayMs}ms 后重试` };
}

export function canComplete(session: UploadSession): boolean {
  return missingChunks(session).length === 0;
}

export function snapshotOf(session: UploadSession): string {
  return JSON.stringify(session);
}

export function readSnapshot(raw: string | null): UploadSession | null {
  if (raw === null || raw.trim() === "") return null;
  let parsed: unknown;
  try {
    parsed = JSON.parse(raw);
  } catch {
    return null;
  }
  if (typeof parsed !== "object" || parsed === null) return null;
  const candidate = parsed as Partial<UploadSession>;
  const file = candidate.fingerprint as Partial<FileFingerprint> | undefined;
  if (typeof candidate.sessionId !== "string" || candidate.sessionId === "") return null;
  if (file === undefined || typeof file.name !== "string") return null;
  if (!Number.isSafeInteger(file.size) || !Number.isSafeInteger(file.lastModified)) return null;
  if (!Number.isSafeInteger(candidate.chunkSize) || (candidate.chunkSize ?? 0) <= 0) return null;
  if (!Number.isSafeInteger(candidate.totalBytes) || (candidate.totalBytes ?? -1) < 0) return null;
  if (candidate.totalBytes !== file.size) return null;
  if (!Array.isArray(candidate.confirmed)) return null;
  if (!candidate.confirmed.every((index) => Number.isSafeInteger(index))) return null;
  if (!Number.isSafeInteger(candidate.updatedAt) || !Number.isSafeInteger(candidate.expiresAt)) return null;
  const total = Math.ceil((candidate.totalBytes as number) / (candidate.chunkSize as number));
  const confirmed = [...new Set(candidate.confirmed as number[])]
    .filter((index) => index >= 0 && index < total)
    .sort((a, b) => a - b);
  return {
    sessionId: candidate.sessionId,
    fingerprint: { name: file.name, size: file.size as number, lastModified: file.lastModified as number },
    chunkSize: candidate.chunkSize as number,
    totalBytes: candidate.totalBytes as number,
    confirmed,
    updatedAt: candidate.updatedAt as number,
    expiresAt: candidate.expiresAt as number,
  };
}

$ node --test --test-concurrency=1 test/upload/resume.test.ts
✔ 会话快照可 JSON 往返，损坏或字段不一致的输入返回 null (1.670042ms)
✔ 名称、大小、修改时间任一变化都判定 mismatch (0.1645ms)
✔ 过期会话不再恢复，边界时刻按过期处理 (0.077292ms)
✔ 恢复沿用会话自己的 chunkSize，不被当前策略重算 (0.13ms)
✔ 确认幂等：重复 ack 不改状态，乱序 ack 只按索引记账 (0.291458ms)
✔ 恢复时以服务端清单为准，丢弃本地虚记并补记服务端确认 (0.152583ms)
✔ 可重试错误按指数退避，429 遵守 Retry-After (0.15975ms)
✔ 鉴权与配额错误不重试，超过次数上限停止，缺片不能提交 (0.116584ms)
ℹ tests 8
ℹ suites 0
ℹ pass 8
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 156.06425