stateDiagram-v2
    [*] --> IDLE : 初始化
    IDLE --> HANDSHAKE : 交换 FileManifest
    HANDSHAKE --> TRANSFERRING : 握手确认/续传索引对齐
    TRANSFERRING --> PAUSED : 网络中断/用户暂停/背压过高
    PAUSED --> TRANSFERRING : 网络恢复/用户继续/续传触发
    TRANSFERRING --> VERIFYING : 全量分片接收完成
    VERIFYING --> COMPLETED : 校验通过 (Hash一致)
    VERIFYING --> TRANSFERRING : 校验失败 -> 请求缺失分片
    COMPLETED --> [*]
    TRANSFERRING --> ERROR : 致命错误 (对端关闭/协议版本不兼容)

// worker.hasher.js
self.onmessage = async (e) => {
  const { file, chunkSize = 32768 } = e.data;
  const chunks = [];
  const spark = new SparkMD5.ArrayBuffer(); // 或使用 Web Crypto API SHA-256
  let offset = 0, index = 0;

  while (offset < file.size) {
    const slice = file.slice(offset, offset + chunkSize);
    const buffer = await slice.arrayBuffer();
    
    // 计算分片哈希（用于校验）
    const chunkHash = await crypto.subtle.digest('SHA-256', buffer);
    // 累加总哈希
    spark.append(buffer);
    
    chunks.push({ index, size: slice.size, hash: bytesToHex(chunkHash) });
    offset += chunkSize;
    index++;
    
    // 进度反馈
    self.postMessage({ type: 'progress', loaded: offset, total: file.size });
  }
  
  const fileHash = spark.end(true);
  self.postMessage({ type: 'complete', fileHash: bytesToHex(fileHash), chunks });
};

class SenderController {
  private windowSize = 64; // 窗口大小 (分片数)
  private base = 0;        // 窗口基序号
  private nextSeq = 0;     // 下一个待发序号
  private timers = new Map<number, NodeJS.Timeout>(); // 重传定时器
  private rttEstimator = new RTTEstimator(); // RTT估算器 (Jacobson/Karels算法)

  // 由 bufferedAmountLow 事件驱动
  onChannelReady() {
    while (this.nextSeq < this.base + this.windowSize && this.nextSeq < this.totalChunks) {
      this.sendChunk(this.nextSeq);
      this.startTimer(this.nextSeq);
      this.nextSeq++;
    }
  }

  private sendChunk(seq: number) {
    const chunkData = this.fileChunks[seq];
    // 协议封装: [Type:1byte][Seq:4bytes][Payload:Nbytes]
    const packet = encodePacket(PacketType.DATA, seq, chunkData);
    this.dataChannel.send(packet);
  }

  handleAck(ackSeq: number, sackBlocks: SACKBlock[]) {
    // 1. 更新 RTT
    this.rttEstimator.update(Date.now() - this.sendTimeMap.get(ackSeq));
    
    // 2. 清理定时器
    this.clearTimer(ackSeq);
    
    // 3. 滑动窗口 (处理 SACK 块)
    this.processSACK(ackSeq, sackBlocks);
    
    // 4. 继续发送
    this.onChannelReady();
  }
  
  // 指数退避重传
  private startTimer(seq: number) {
    const rto = this.rttEstimator.getRTO();
    this.timers.set(seq, setTimeout(() => this.retransmit(seq), rto));
  }
}

class ReceiverController {
  private receiveBuffer = new Map<number, ArrayBuffer>(); // 乱序缓冲
  private expectedSeq = 0;
  private lastAckTime = 0;

  handleData(packet: DataPacket) {
    const { seq, data } = packet;
    
    // 1. 去重
    if (seq < this.expectedSeq || this.receiveBuffer.has(seq)) {
      this.sendAck(); // 立即回复 ACK 避免对端超时
      return;
    }

    // 2. 缓存
    this.receiveBuffer.set(seq, data);

    // 3. 尝试交付有序数据
    while (this.receiveBuffer.has(this.expectedSeq)) {
      this.writeToDisk(this.receiveBuffer.get(this.expectedSeq)!);
      this.receiveBuffer.delete(this.expectedSeq);
      this.expectedSeq++;
      this.updateProgress();
    }

    // 4. 生成 SACK (Selective Acknowledgment)
    this.sendAck();
  }

  private sendAck() {
    const now = Date.now();
    // 限制 ACK 频率 (如 20ms 一次) 或 累计 2 个分片后发送
    if (now - this.lastAckTime < 20 && this.receiveBuffer.size < 2) return;
    
    const blocks = this.generateSACKBlocks(); // 将缓存的连续区间编码
    this.dataChannel.send(encodePacket(PacketType.ACK, this.expectedSeq, blocks));
    this.lastAckTime = now;
  }
}

interface TransferState {
  fileId: string;
  fileName: string;
  totalSize: number;
  chunkSize: number;
  totalChunks: number;
  // 位图 或 区间列表，标记已发送/已接收并校验通过的分片
  // 发送端存：已收到ACK的分片
  // 接收端存：已写入磁盘并校验通过的分片
  completedChunks: IntervalTree | BitSet; 
  remotePeerId: string; // 对端标识，防止状态错配
  updatedAt: number;    // 最后更新时间戳
}

const iceServers = [
  { urls: 'stun:stun.example.com:3478' },
  { 
    urls: 'turn:turn.example.com:3478?transport=udp', 
    username: 'user', credential: 'pass' 
  },
  { 
    urls: 'turns:turn.example.com:443?transport=tcp', // TLS over 443
    username: 'user', credential: 'pass' 
  }
];
const pc = new RTCPeerConnection({ iceServers, iceTransportPolicy: 'all' });

# benchmark_runner.py
import subprocess, json, time, statistics

def run_scenario(scenario_config, file_size_mb=500):
    # 1. 启动网络模拟
    apply_tc_rules(scenario_config)
    
    # 2. 启动发送端 & 接收端 (Headless Chrome / Playwright / Electron)
    sender_proc = launch_client(role='sender', file_id=f'test_{file_size_mb}mb')
    receiver_proc = launch_client(role='receiver')
    
    # 3. 等待完成并收集性能日志
    result = wait_for_completion(sender_proc, receiver_proc, timeout=3600)
    
    # 4. 清理网络规则
    clear_tc_rules()
    
    return parse_metrics(result)

if __name__ == '__main__':
    scenarios = load_scenarios('network_matrix.json')
    report = []
    for sc in scenarios:
        print(f"Running {sc['name']}...")
        metrics = run_scenario(sc)
        report.append({**sc, **metrics})
    
    # 生成 HTML 报告
    generate_html_report(report, 'webrtc_transfer_benchmark.html')

// 发送端：每发送 K 个源分片，立即生成 P 个校验分片插入发送队列
class FecEncoder {
  private rs = new ReedSolomon(16, 4); // K=16, P=4
  private groupBuffer: ArrayBuffer[] = [];

  addSourcePacket(pkt: DataPacket) {
    this.groupBuffer.push(pkt.payload);
    if (this.groupBuffer.length === 16) {
      const parity = this.rs.encode(this.groupBuffer); // 返回 4 个 ArrayBuffer
      parity.forEach((p, i) => this.sendQueue.push(createFecPacket(p, groupId, i)));
      this.groupBuffer = [];
    }
  }
}

// 接收端：收集源包+校验包，缺包时尝试解码
class FecDecoder {
  private groups = new Map<number, { received: Map<number, ArrayBuffer>, fec: Map<number, ArrayBuffer> }>();

  handlePacket(pkt: Packet) {
    const g = this.getOrCreateGroup(pkt.groupId);
    if (pkt.isFec) g.fec.set(pkt.fecIndex, pkt.payload);
    else g.received.set(pkt.seqInGroup, pkt.payload);

    // 触发解码条件：收到任意 16 个包 (源包+校验包任意组合)
    if (g.received.size + g.fec.size >= 16 && g.received.size < 16) {
      this.attemptDecode(g);
    }
  }

  private attemptDecode(group: GroupState) {
    const packets = new Array(16).fill(null);
    group.received.forEach((v, k) => packets[k] = v);
    group.fec.forEach((v, k) => packets[16 + k] = v); // 校验包放后面
    
    try {
      const decoded = this.rs.decode(packets); // 补全缺失的源包
      decoded.forEach((payload, idx) => {
        if (!group.received.has(idx)) this.deliverToUpperLayer(payload);
      });
    } catch (e) { /* 解码失败，等待更多包或回退 ARQ */ }
  }
}

// 仅在 HTTPS 或 localhost 可用
async function getOPFSFileHandle(fileName: string) {
  const root = await navigator.storage.getDirectory();
  return await root.getFileHandle(fileName, { create: true });
}

class StreamWriter {
  private writer: FileSystemSyncWriteHandle; // 仅 Worker 中可用同步句柄
  private hasher: HashStream; // 自定义 TransformStream 实现流式 SHA-256

  async init(fileHandle: FileSystemFileHandle) {
    // 创建同步写入句柄 (必须在 Worker 中)
    const syncHandle = await fileHandle.createSyncAccessHandle();
    this.writer = syncHandle;
    this.hasher = new HashStream('SHA-256');
  }

  // 在 Worker 中调用，零拷贝写入
  writeChunk(chunk: ArrayBuffer, offset: number) {
    // 1. 写入磁盘 (同步阻塞，但 Worker 不阻塞主线程)
    this.writer.write(chunk, { at: offset });
    
    // 2. 流式计算哈希 (无需二次读取)
    this.hasher.push(chunk);
  }

  async finalize(): Promise<string> {
    this.writer.flush();
    this.writer.close();
    return this.hasher.digest(); // 返回最终 Hex Hash
  }
}

// signaling.proto
message SignalFrame {
  string session_id = 1;
  uint32 seq = 2;
  FrameType type = 3;
  bytes payload = 4; // 根据 type 反序列化为 Offer/Answer/Candidate/Control
}

message IceCandidateBatch {
  repeated IceCandidate candidates = 1; // 批量发送，减少 RTT
  bool end_of_candidates = 2;
}

message IceCandidate {
  string foundation = 1;
  uint32 component = 2;
  string protocol = 3; // "udp" / "tcp"
  uint32 priority = 4;
  string address = 5;
  uint32 port = 6;
  string type = 7; // "host", "srflx", "relay"
  string related_address = 8;
  uint32 related_port = 9;
  string tcp_type = 10; // "active", "passive", "so"
}

graph LR
    ClientA -->|WS/WSS| Gateway[Stateless Gateway Cluster]
    ClientB -->|WS/WSS| Gateway
    Gateway -->|Pub/Sub| Redis[(Redis Cluster)]
    Gateway -->|gRPC| TURN_Monitor[TURN Health Monitor]
    Gateway -->|HTTP| Auth_Service[Auth & Quota Service]

// 客户端侧连接状态机超时配置
const ICE_TIMEOUTS = {
  gathering: 3000,      // 收集候选超时
  connecting: 15000,    // 连通性检查总超时
  check_interval: 250,  // 单对候选检查间隔
  nomination: 5000,     // 提名有效对超时
};

// 熔断器：连续 3 次 ICE 失败 -> 触发 TURN-TLS 强制中转模式
class CircuitBreaker {
  private failures = 0;
  recordFailure() { if (++this.failures >= 3) this.forceRelayMode = true; }
  recordSuccess() { this.failures = 0; this.forceRelayMode = false; }
}

// 客户端发起传输时创建根 Span
const tracer = trace.getTracer('webrtc-transfer');
const span = tracer.startSpan('transfer_session', {
  attributes: { 'file.id': fileId, 'file.size': fileSize, 'peer.id': peerId }
});

// 将 TraceContext 注入信令消息
const carrier = {};
propagation.inject(context.active(), carrier);
signalingMessage.trace_context = carrier.traceparent;

// DataChannel 打开时创建子 Span
dataChannel.onopen = () => {
  const childSpan = tracer.startSpan('datachannel_open', undefined, trace.setSpan(context.active(), span));
  // ... 传输逻辑中记录 Events
  childSpan.addEvent('first_chunk_sent');
  childSpan.end();
};