《互联网产品的客户端架构·内容、社交与实时互动》4.1 上传编排：分片、并发与进度语义

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

export interface ChunkPlan { index: number; start: number; end: number; size: number }

export interface ChunkSendInput {
  sessionId: string;
  plan: ChunkPlan;
  onProgress: (sentBytes: number) => void;
}

export interface ChunkTransport { send(input: ChunkSendInput): Promise<void> }
export interface Clock { now(): number }

export type ChunkStatus = "pending" | "inflight" | "confirmed" | "failed";

export interface UploadProgress {
  sessionId: string;
  totalBytes: number;
  totalChunks: number;
  confirmedChunks: number;
  confirmedBytes: number;
  sentBytes: number;
  inFlight: number;
  paused: boolean;
  wireRatio: number;
  confirmedRatio: number;
  done: boolean;
}

export interface OrchestratorOptions {
  sessionId: string;
  totalBytes: number;
  chunkSize: number;
  concurrency: number;
  transport: ChunkTransport;
  clock?: Clock;
  throttleMs?: number;
  onProgress?: (progress: UploadProgress) => void;
}

export interface UploadResult { sessionId: string; totalBytes: number; confirmedChunks: number }

export class ChunkUploadError extends Error {
  readonly chunkIndex: number;
  constructor(chunkIndex: number, cause: unknown) {
    super(`分片 ${chunkIndex} 上传失败`);
    this.name = "ChunkUploadError";
    this.chunkIndex = chunkIndex;
    this.cause = cause;
  }
}

export function planChunks(totalBytes: number, chunkSize: number): ChunkPlan[] {
  if (!Number.isSafeInteger(totalBytes) || totalBytes < 0) throw new RangeError("totalBytes 必须是非负安全整数");
  if (!Number.isSafeInteger(chunkSize) || chunkSize <= 0) throw new RangeError("chunkSize 必须是正安全整数");
  const plans: ChunkPlan[] = [];
  for (let start = 0; start < totalBytes; start += chunkSize) {
    const end = Math.min(start + chunkSize, totalBytes);
    plans.push({ index: plans.length, start, end, size: end - start });
  }
  return plans;
}

/** 把线路进度压成一条只前进的展示曲线；未全部确认时封顶 0.99。 */
export function smoothProgress(previous: number, progress: UploadProgress): number {
  const floor = Number.isFinite(previous) ? Math.min(1, Math.max(0, previous)) : 0;
  const target = Math.min(progress.wireRatio, progress.done ? 1 : 0.99);
  return Math.min(1, Math.max(floor, target));
}

export class UploadOrchestrator {
  readonly #sessionId: string;
  readonly #plans: ChunkPlan[];
  readonly #totalBytes: number;
  readonly #concurrency: number;
  readonly #transport: ChunkTransport;
  readonly #clock: Clock;
  readonly #throttleMs: number;
  readonly #onProgress: ((progress: UploadProgress) => void) | undefined;
  readonly #done: Promise<UploadResult>;
  #statuses: ChunkStatus[];
  #sent: number[];
  #inFlight = new Map<number, Promise<void>>();
  #cursor = 0;
  #paused = false;
  #started = false;
  #settled = false;
  #failed = false;
  #lastEmit = Number.NEGATIVE_INFINITY;
  #resolve!: (result: UploadResult) => void;
  #reject!: (error: unknown) => void;

  constructor(options: OrchestratorOptions) {
    if (!Number.isSafeInteger(options.concurrency) || options.concurrency <= 0) {
      throw new RangeError("concurrency 必须是正安全整数");
    }
    this.#sessionId = options.sessionId;
    this.#plans = planChunks(options.totalBytes, options.chunkSize);
    this.#totalBytes = this.#plans.at(-1)?.end ?? 0;
    this.#concurrency = options.concurrency;
    this.#transport = options.transport;
    this.#clock = options.clock ?? { now: () => Date.now() };
    this.#throttleMs = options.throttleMs ?? 100;
    this.#onProgress = options.onProgress;
    this.#statuses = this.#plans.map(() => "pending");
    this.#sent = this.#plans.map(() => 0);
    this.#done = new Promise<UploadResult>((resolve, reject) => {
      this.#resolve = resolve;
      this.#reject = reject;
    });
  }

  get plans(): ChunkPlan[] { return this.#plans.map((plan) => ({ ...plan })); }

  status(index: number): ChunkStatus {
    const status = this.#statuses[index];
    if (status === undefined) throw new RangeError(`分片索引越界: ${index}`);
    return status;
  }

  start(): Promise<UploadResult> {
    if (this.#started) throw new Error("同一个编排器只能 start 一次");
    this.#started = true;
    this.#pump();
    return this.#done;
  }

  pause(): void {
    if (this.#paused || !this.#started || this.#settled) return;
    this.#paused = true;
    this.#emit(true);
  }

  resume(): void {
    if (!this.#paused || this.#settled) return;
    this.#paused = false;
    this.#emit(true);
    this.#pump();
  }

  progress(): UploadProgress {
    let confirmedChunks = 0, confirmedBytes = 0, sentBytes = 0, inFlight = 0;
    for (const plan of this.#plans) {
      const status = this.#statuses[plan.index];
      if (status === "confirmed") { confirmedChunks += 1; confirmedBytes += plan.size; }
      if (status === "inflight") inFlight += 1;
      sentBytes += this.#sent[plan.index] ?? 0;
    }
    const ratio = (bytes: number): number =>
      this.#totalBytes > 0 ? bytes / this.#totalBytes : this.#started ? 1 : 0;
    return {
      sessionId: this.#sessionId,
      totalBytes: this.#totalBytes,
      totalChunks: this.#plans.length,
      confirmedChunks,
      confirmedBytes,
      sentBytes,
      inFlight,
      paused: this.#paused,
      wireRatio: ratio(sentBytes),
      confirmedRatio: ratio(confirmedBytes),
      done: this.#started && !this.#failed && confirmedChunks === this.#plans.length,
    };
  }

  #launch(plan: ChunkPlan): void {
    this.#statuses[plan.index] = "inflight";
    this.#sent[plan.index] = 0;
    const sending = this.#transport
      .send({ sessionId: this.#sessionId, plan, onProgress: (bytes) => this.#onChunkProgress(plan, bytes) })
      .then(() => this.#onChunkConfirmed(plan), (error: unknown) => this.#onChunkFailed(plan, error));
    this.#inFlight.set(plan.index, sending);
    this.#emit();
  }

  #onChunkProgress(plan: ChunkPlan, bytes: number): void {
    if (this.#settled || this.#statuses[plan.index] !== "inflight") return;
    const clamped = Math.min(plan.size, Math.max(0, Math.floor(bytes)));
    if (clamped === this.#sent[plan.index]) return;
    this.#sent[plan.index] = clamped;
    this.#emit();
  }

  #onChunkConfirmed(plan: ChunkPlan): void {
    this.#inFlight.delete(plan.index);
    if (this.#settled) return;
    this.#statuses[plan.index] = "confirmed";
    this.#sent[plan.index] = plan.size;
    this.#emit();
    this.#pump();
  }

  #onChunkFailed(plan: ChunkPlan, error: unknown): void {
    this.#inFlight.delete(plan.index);
    if (this.#settled) return;
    this.#statuses[plan.index] = "failed";
    this.#sent[plan.index] = 0;
    this.#failed = true;
    this.#settled = true;
    this.#emit(true);
    this.#reject(new ChunkUploadError(plan.index, error));
  }

  #pump(): void {
    if (this.#paused || this.#settled) return;
    while (this.#inFlight.size < this.#concurrency && this.#cursor < this.#plans.length) {
      const plan = this.#plans[this.#cursor];
      this.#cursor += 1;
      this.#launch(plan);
    }
    if (this.#statuses.every((status) => status === "confirmed")) {
      this.#settled = true;
      this.#emit(true);
      this.#resolve({ sessionId: this.#sessionId, totalBytes: this.#totalBytes, confirmedChunks: this.#plans.length });
    }
  }

  #emit(force = false): void {
    const listener = this.#onProgress;
    if (!listener) return;
    const now = this.#clock.now();
    if (!force && now - this.#lastEmit < this.#throttleMs) return;
    this.#lastEmit = now;
    listener(this.progress());
  }
}

$ node --test --test-concurrency=1 test/upload/orchestrator.test.ts
✔ 分片计划覆盖完整文件，末片截断，0 字节没有分片 (0.994875ms)
✔ 并发窗口只启动 concurrency 个分片 (0.321208ms)
✔ 乱序完成不会提前结束，全部分片确认后才 resolve (0.203209ms)
✔ 进度同时暴露在途字节与已确认字节 (0.181125ms)
✔ 分片失败丢弃在途字节：wireRatio 回退，confirmedRatio 不动 (0.25925ms)
✔ smoothProgress 只前进，未完成时封顶 0.99 (0.091334ms)
✔ 暂停只停新分片，在途分片照常完成，恢复后继续 (0.15125ms)
✔ 进度回调按最小间隔节流，完成时强制发出最后一次 (0.163708ms)
✔ 0 字节文件 start 后立即完成，重复 start 抛错 (0.128792ms)
ℹ tests 9
ℹ suites 0
ℹ pass 9
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 147.426333