func Route(conferenceID string) (nodeID string, err error) {
    // 1. 计算会议在环上的位置
    pos := hashRing.Locate(hash(conferenceID))
    
    // 2. 顺时针查找首个负载未满的虚拟节点
    for vNode := pos.Next(); vNode != pos; vNode = vNode.Next() {
        physicalNode := vNode.PhysicalNode()
        if atomic.LoadInt64(&physicalNode.currentLoad) < physicalNode.capacity {
            // 3. 原子性尝试占位（CAS）
            if atomic.CompareAndSwapInt64(&physicalNode.currentLoad, 
                physicalNode.currentLoad, physicalNode.currentLoad+1) {
                return physicalNode.ID, nil
            }
            // CAS 失败说明并发竞争，继续找下一个
        }
    }
    return "", ErrAllNodesFull // 触发熔断/排队降级
}

type VNode struct {
    id             uint64
    physicalNode   *PhysicalNode
    next           *VNode
    nextAvailable  *VNode  // 指针跳跃：指向下一个可用虚拟节点
}

┌─────────────────────────────────────────────────────────────┐
│                     接入网关层 (Gateway)                     │
│  - TLS 终结、鉴权、限流、协议转换 (WebRTC/SIP/RTMP)          │
└──────────────────────────┬──────────────────────────────────┘
                           │ gRPC (RouteRequest)
┌──────────────────────────▼──────────────────────────────────┐
│                  路由决策服务 (Router Service)                │
│  - 一致性哈希有界负载算法核心                                  │
│  - 节点健康度感知 (心跳 + 主动探测)                            │
│  - 熔断/降级策略引擎                                           │
└──────────────────────────┬──────────────────────────────────┘
                           │ etcd Watch / Redis PubSub
┌──────────────────────────▼──────────────────────────────────┐
│                   媒体节点集群 (Media Nodes)                  │
│  - 无状态 SFU 进程 (Janus/Mediasoup 自研改造)                 │
│  - 本地关键帧缓存 (LRU, TTL=30s)                              │
│  - 指标上报: load, cpu, mem, bandwidth, keyframe_cache_hit   │
└─────────────────────────────────────────────────────────────┘

// /media/nodes/{nodeID}
{
  "id": "media-node-07",
  "status": "SERVING",           // SERVING | WARMING | DRAINING | OFFLINE
  "capacity": 1200,              // 单节点会议上限
  "current_load": 843,
  "virtual_nodes": 150,          // 虚拟节点数
  "zone": "cn-hangzhou-a",
  "start_time": "2024-11-15T08:30:00Z",
  "warmup_progress": 1.0         // 0.0~1.0
}

// /media/conferences/{confID}/binding
{
  "node_id": "media-node-07",
  "bound_at": "2024-11-15T10:15:22Z",
  "migration_epoch": 3           // 乐观锁，防止并发迁移竞争
}

func (n *MediaNode) prewarmKeyframe(confID, oldNodeID string) error {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    
    // 1. 向旧节点请求关键帧快照
    resp, err := n.grpcClient.GetKeyframeSnapshot(ctx, &pb.KeyframeRequest{
        ConferenceId: confID,
        MaxSizeBytes: 50 * 1024 * 1024,
    })
    if err != nil {
        return fmt.Errorf("prewarm failed: %w", err)
    }
    
    // 2. 写入本地 Ring Buffer（无锁环形缓冲区）
    n.keyframeCache.Store(confID, &KeyframeCacheEntry{
        Data:      resp.FrameData,
        Timestamp: time.Now(),
        RefCount:  0,
    })
    
    // 3. 异步预热下游 CDN/转码节点（可选）
    go n.pushToDownstream(confID, resp.FrameData)
    return nil
}

# 单节点负载利用率（应 ≤ 1.1）
media_node_load_ratio = media_node_current_load / media_node_capacity

# 关键帧请求风暴抑制效果
rate(media_keyframe_requests_total[1m]) / rate(media_conference_migrations_total[1m])

# 迁移成功率
sum(rate(media_migration_success_total[5m])) / sum(rate(media_migration_total[5m]))

# 会议级可用性
1 - (sum(rate(media_conference_disconnected_total[5m])) / sum(rate(media_conference_joined_total[5m])))

// TxnRoute 在单个 etcd 事务中完成：选节点 + 校验容量 + 写入绑定 + 版本号递增
func (r *Router) TxnRoute(ctx context.Context, confID string) (string, error) {
    // 1. 计算候选节点列表（本地计算，无副作用）
    candidates := r.hashRing.LocateCandidates(confID, 3) // 取前 3 个候选
    
    // 2. 构建 etcd 事务：遍历候选节点，找到首个满足条件的
    // 由于 etcd Txn 不支持循环，需在应用层组装 If-Then-Else 链
    txnOps := make([]clientv3.Op, 0, len(candidates)*3)
    for i, node := range candidates {
        loadKey := fmt.Sprintf("/media/nodes/%s/load", node.ID)
        bindKey := fmt.Sprintf("/media/conferences/%s/binding", confID)
        
        // If: 节点状态=SERVING AND 当前负载 < 容量上界
        txnOps = append(txnOps, clientv3.OpTxn(
            []clientv3.Cmp{
                clientv3.Compare(clientv3.Value(nodeStatusKey(node.ID)), "=", "SERVING"),
                clientv3.Compare(clientv3.Value(loadKey), "<", strconv.Itoa(node.Capacity)),
            },
            // Then: 原子增负载 + 写入绑定关系(带乐观锁版本号)
            []clientv3.Op{
                clientv3.OpPut(loadKey, "", clientv3.WithIgnoreValue(), clientv3.WithPrevKV()), // 利用 IgnoreValue 做原子增？需配合 Lua 或单独 Key
                // 修正：etcd 无原子增，需用单独 Key 存计数器 或 使用 etcd Lock
                // 此处简化：假设使用分布式锁保护负载计数器，或迁移至 Redis Lua 脚本
            },
            // Else: 尝试下一个候选
            nil,
        ))
    }
    // ... 执行事务并解析响应 ...
}

stateDiagram-v2
    [*] --> STABLE: 会议创建
    STABLE --> PENDING: 发起迁移 (CAS epoch)
    PENDING --> COMMITTED: 新节点预热完成 + 信令切换成功
    PENDING --> ROLLED_BACK: 预热超时/信令失败 (CAS epoch)
    COMMITTED --> STABLE: 旧节点资源释放确认
    ROLLED_BACK --> STABLE: 回滚完成
    
    note right of PENDING
        关键约束：仅允许 epoch=N 的操作修改状态
        任何 epoch < N 的请求直接拒绝 (幂等)
    end note

func (r *Router) getBinding(confID) (nodeID string, epoch int64, err error) {
    // 1. 读 Redis 绑定
    bind, _ := r.redis.HGetAll(ctx, bindKey(confID)).Result()
    if bind["epoch"] != "" {
        // 2. 校验目标节点健康度（本地缓存）
        if r.isNodeHealthy(bind["node_id"]) {
            return bind["node_id"], bind["epoch"], nil
        }
        // 3. 目标节点不健康，触发异步修复，但本次路由仍返回旧节点（保会议不中断）
        go r.triggerRepair(confID, bind["epoch"])
        return bind["node_id"], bind["epoch"], nil 
    }
    // 4. 无绑定，走首次路由逻辑
    return r.firstTimeRoute(confID)
}

// 1. 网关 -> 客户端：迁移预警 (提前 500ms 下发，客户端预建 ICE Candidate)
{
  "jsonrpc": "2.0",
  "method": "conference.migration.prepare",
  "params": {
    "conference_id": "conf_abc123",
    "migration_epoch": 5,
    "target_media_node": "media-node-09",
    "target_ice_servers": [{"urls": "turn:turn.example.com:3478", "username": "...", "credential": "..."}],
    "media_transport": "ice-restart", // 或 "dtls-renegotiate"
    "deadline_ms": 2000
  }
}

// 2. 客户端 -> 网关：预热就绪 (并行收集 Candidate，不中断当前媒体流)
{
  "jsonrpc": "2.0",
  "id": "req_1",
  "method": "conference.migration.ready",
  "params": {
    "conference_id": "conf_abc123",
    "migration_epoch": 5,
    "local_candidates": [...], // 新节点的 Candidate
    "dtls_fingerprint": "sha-256 ..."
  }
}

// 3. 网关 -> 旧媒体节点：冻结媒体转发 (停止转发 RTP，保留 RTCP/状态)
{
  "cmd": "freeze_conference",
  "conf_id": "conf_abc123",
  "epoch": 5,
  "graceful_timeout_ms": 1000
}

// 4. 网关 -> 新媒体节点：恢复上下文 (注入预热的关键帧、参会者列表、转码配置)
{
  "cmd": "restore_conference",
  "conf_id": "conf_abc123",
  "epoch": 5,
  "context": {
    "participants": [...],
    "keyframe_cache": "shm://keyframe_conf_abc123",
    "recording_status": "RUNNING"
  }
}

// 5. 网关 -> 客户端：执行切换 (原子切换，单 RTT 内完成)
{
  "jsonrpc": "2.0",
  "method": "conference.migration.commit",
  "params": {
    "conference_id": "conf_abc123",
    "migration_epoch": 5,
    "new_ice_ufrag": "new_ufrag",
    "new_ice_pwd": "new_pwd",
    "dtls_role": "client" // 强制角色避免冲突
  }
}

# 节点标签
node_labels:
  - "high_cpu_freq"      # 高主频 CPU，适合大型会议混音
  - "gpu_h264_encoder"   # 硬编资源
  - "low_latency_zone"   # 金融/交易场景专用区

# 会议调度策略
conference_profile:
  type: "LARGE_MEETING"
  required_labels: ["high_cpu_freq"]
  forbidden_labels: ["spot_instance"] # 禁止抢占式实例
  dedicated_capacity_ratio: 0.3       # 预留 30% 容量给此类会议

# .chaos/schedule/weekly-media-migration.yaml
apiVersion: chaos.example.com/v1alpha1
kind: ChaosExperiment
metadata:
  name: weekly-media-node-drain
spec:
  schedule: "0 2 * * 1" # 每周一凌晨 2 点
  target:
    selector: "role=media-node,zone!=canary"
    mode: "one" # 每次随机选 1 个节点
  action: "drain" # 模拟滚动升级驱逐
  steadyStateHypothesis:
    - name: "conference_availability"
      provider:
        type: "promql"
        query: 'media_conference_availability_ratio > 0.9999'
  rollbackConditions:
    - 'media_keyframe_storm_ratio > 0.05'
    - 'media_migration_failure_rate > 0.01'
  reporting:
    webhook: "https://alert.example.com/chaos-report"

// 中间件自动注入
func ConferenceTraceMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        confID := r.Header.Get("X-Conference-ID")
        epoch := r.Header.Get("X-Migration-Epoch")
        
        // 1. 设置 OpenTelemetry Baggage (跨进程传递)
        ctx := baggage.ContextWithValues(r.Context(),
            attribute.String("conference.id", confID),
            attribute.String("migration.epoch", epoch),
        )
        
        // 2. 启动 Span (采样率 100% 覆盖迁移会议)
        sampler := tracesdk.ParentBased(tracesdk.TraceIDRatioBased(0.01))
        if epoch != "" { sampler = tracesdk.AlwaysSample() } // 迁移会议全采样
        
        ctx, span := tracer.Start(ctx, "media_route", trace.WithSampler(sampler))
        defer span.End()
        
        // 3. 注入 Logger 字段
        logger := log.FromContext(ctx).With("conf_id", confID, "epoch", epoch)
        next.ServeHTTP(w, r.WithContext(logger.WithContext(ctx)))
    })
}

graph LR
    A[Spot 实例启动] --> B{注册为 WARMING}
    B --> C[预热关键帧缓存]
    C --> D{预热通过?}
    D -- Yes --> E[标记 SERVING<br/>权重 0.8]
    D -- No --> F[标记 OFFLINE]
    E --> G[接收生产流量]
    G --> H{收到回收通知?}
    H -- Yes --> I[标记 DRAINING<br/>权重置 0]
    I --> J[驱逐会议/等待自然结束]
    J --> K[资源释放]