[采集时钟] → [RTP Timestamp] → [网络传输] → [RTCP SR (NTP + RTP)] → [接收端建立映射] → [统一渲染时钟] → [播放对齐]

 0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|                           Seconds                             |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|                          Fraction                             |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

// 伪代码：发送端生成 SR
void SendRtcpSr() {
  uint32_t rtp_ts = last_packet_rtp_timestamp_; // 最近发送包的 RTP TS
  NtpTime ntp_now = NtpTime::Now();             // 当前 NTP 时间
  rtcp_sr->SetSenderInfo(ntp_now, rtp_ts, ...);
}

#include <cstdint>
#include <vector>
#include <optional>
#include <algorithm>

struct NtpTime {
  uint32_t seconds;
  uint32_t fractions;
  
  // 转换为微秒 (int64_t)
  int64_t ToUs() const {
    // 使用 __int128 避免溢出: fractions * 1e6 / 2^32
    __int128 frac_us = (__int128)fractions * 1000000;
    frac_us >>= 32;
    return (__int128)seconds * 1000000 + (int64_t)frac_us;
  }
  
  static NtpTime FromUs(int64_t us) {
    NtpTime ntp;
    ntp.seconds = us / 1000000;
    __int128 frac = (__int128)(us % 1000000) << 32;
    frac /= 1000000;
    ntp.fractions = (uint32_t)frac;
    return ntp;
  }
};

struct RtpNtpPair {
  uint32_t rtp_timestamp; // 90kHz or 48kHz
  int64_t  ntp_time_us;   // 统一微秒时间基
};

class RtpToNtpMapper {
public:
  // 时钟频率 (Hz)，视频 90000，音频 48000 等
  explicit RtpToNtpMapper(int clock_rate_hz) : clock_rate_hz_(clock_rate_hz) {}

  // 更新映射点 (来自 RTCP SR 解析)
  void Update(uint32_t rtp_ts, const NtpTime& ntp) {
    int64_t ntp_us = ntp.ToUs();
    // RTP 时间戳回绕处理：展开为 64 位单调时间线
    uint64_t unwrapped_rtp = UnwrapRtp(rtp_ts);
    
    samples_.push_back({unwrapped_rtp, ntp_us});
    if (samples_.size() > kMaxSamples) samples_.erase(samples_.begin());
    
    Refit();
  }

  // 查询：RTP TS -> NTP 微秒时间
  std::optional<int64_t> Map(uint32_t rtp_ts) const {
    if (!valid_) return std::nullopt;
    uint64_t unwrapped = UnwrapRtp(rtp_ts);
    // y = slope * x + intercept
    __int128 val = (__int128)slope_q32_ * unwrapped + intercept_q32_;
    return (int64_t)(val >> 32); // Q32.32 定点数还原
  }

private:
  static constexpr size_t kMaxSamples = 50;
  const int clock_rate_hz_;
  std::vector<RtpNtpPair> samples_;
  
  // Q32.32 定点数存储斜率与截距，避免浮点误差
  int64_t slope_q32_ = 0;      // (1/clock_rate) * 2^32
  int64_t intercept_q32_ = 0;  // NTP_us * 2^32
  bool valid_ = false;
  uint32_t last_rtp_ts_ = 0;
  uint64_t unwrapped_base_ = 0;

  // RTP 32位回绕展开 (假设调用间隔 < 2^31/clock_rate 秒)
  uint64_t UnwrapRtp(uint32_t rtp_ts) {
    if (rtp_ts < last_rtp_ts_ && (last_rtp_ts_ - rtp_ts) > 0x80000000) {
      unwrapped_base_ += (1ULL << 32);
    }
    last_rtp_ts_ = rtp_ts;
    return unwrapped_base_ + rtp_ts;
  }

  // 核心拟合：最小二乘法 + 定点数运算
  void Refit() {
    if (samples_.size() < 3) return; // 至少 3 点拟合
    
    // 简单线性回归 (生产环境建议加权/卡尔曼)
    __int128 sum_x = 0, sum_y = 0, sum_xy = 0, sum_x2 = 0;
    for (auto& s : samples_) {
      sum_x  += s.rtp_timestamp;
      sum_y  += s.ntp_time_us;
      sum_xy += (__int128)s.rtp_timestamp * s.ntp_time_us;
      sum_x2 += (__int128)s.rtp_timestamp * s.rtp_timestamp;
    }
    size_t n = samples_.size();
    __int128 denom = (__int128)n * sum_x2 - sum_x * sum_x;
    if (denom == 0) return;

    // slope = (n*sum_xy - sum_x*sum_y) / denom
    // 乘以 2^32 存为 Q32.32
    slope_q32_ = (int64_t)(((__int128)n * sum_xy - sum_x * sum_y) * (1ULL << 32) / denom);
    
    // intercept = (sum_y - slope * sum_x) / n
    intercept_q32_ = (int64_t)((sum_y * (1ULL << 32) - slope_q32_ * sum_x) / n);
    
    valid_ = true;
  }
};

graph LR
    A[上游发送端] -->|RTP + RTCP SR| B(SFU Ingress Mapper)
    B --> C{核心转发逻辑}
    C -->|重写 RTP TS<br/>重写 SSRC| D[SFU Egress Mapper]
    D -->|生成新 RTCP SR| E[下游接收端]
    
    subgraph SFU Ingress Mapper
    B1[解析上游 SR<br/>(NTP_u, RTP_u)]
    B2[建立映射: RTP_u -> NTP_u]
    B3[转换为 SFU 内部统一时间基<br/>(Internal Media Clock)]
    end
    
    subgraph SFU Egress Mapper
    D1[分配下游 RTP TS<br/>(基于 Internal Media Clock)]
    D2[记录映射点: (RTP_d, NTP_sfu)]
    D3[定时发送 RTCP SR<br/>(NTP_sfu, RTP_d)]
    end

// SFU 内部统一媒体时钟 (单位: 微秒，单调递增)
using InternalMediaTimeUs = int64_t;

struct SfuStreamContext {
  // 上游映射器
  RtpToNtpMapper ingress_mapper_;      // 上游 RTP(90kHz) -> NTP(上游时钟)
  // 下游映射器 (每个下游连接独立)
  struct DownstreamContext {
    RtpToNtpMapper egress_mapper_;     // 下游 RTP(90kHz) -> NTP(SFU本地时钟)
    uint32_t last_out_rtp_ts_ = 0;     // 下游输出 RTP TS
    uint64_t last_out_rtp_unwrapped_ = 0;
    // 关键：下游首帧 RTP TS 与 InternalMediaTimeUs 的锚点
    std::optional<std::pair<uint32_t, InternalMediaTimeUs>> anchor_; 
  };
  std::map<uint32_t, DownstreamContext> downstreams_; // key: downstream SSRC
};

// 接收端：视频帧渲染时间戳对齐音频时钟
const transform = new TransformStream({
  async transform(frame, controller) {
    if (frame.type === 'video') {
      // 1. 获取音频渲染时钟基准 (AudioContext.currentTime * 1e6 -> 微秒)
      const audioRenderTimeUs = getAudioRenderReferenceTimeUs(); 
      
      // 2. 计算视频帧目标呈现时间 (基于 RTP TS -> NTP 映射)
      const videoTargetNtpUs = rtpToNtpMapper.map(frame.timestamp); 
      
      // 3. 计算偏移
      const offsetMs = (videoTargetNtpUs - audioRenderTimeUs) / 1000;
      
      // 4. 动态策略
      if (offsetMs > 50) { // 视频超前 > 50ms
        // 策略 A: 丢帧 (直接不 controller.enqueue)
        // 策略 B: 延迟入队 (setTimeout 重新 enqueue) - 需配合解码器缓冲区管理
        return; 
      } else if (offsetMs < -100) { // 视频滞后 > 100ms
        // 请求关键帧 (PLI/FIR) 或加速解码 (需编解码器支持)
        requestKeyFrame();
      }
      
      // 5. 修正帧时间戳供上层渲染器使用 (可选)
      frame.presentationTime = videoTargetNtpUs; 
    }
    controller.enqueue(frame);
  }
});

const transceiver = pc.getTransceivers().find(t => t.receiver.track.kind === 'video');
transceiver.receiver.setStreams([transform.readable]);

// audio-sync-clock-processor.js (AudioWorkletProcessor)
class SyncClockProcessor extends AudioWorkletProcessor {
  constructor() {
    super();
    this.port.onmessage = (e) => {
      if (e.data === 'getTime') {
        // currentFrame 为当前渲染帧起始采样点索引
        // sampleRate 为 AudioContext 采样率 (通常 48000)
        const preciseTimeUs = (this.currentFrame / sampleRate) * 1e6;
        this.port.postMessage({ timeUs: preciseTimeUs });
      }
    };
  }
  process() { return true; } // 保持节点存活
}
registerProcessor('sync-clock', SyncClockProcessor);

// 主线程使用
const audioCtx = new AudioContext();
await audioCtx.audioWorklet.addModule('audio-sync-clock-processor.js');
const syncNode = new AudioWorkletNode(audioCtx, 'sync-clock');

function getAudioRenderReferenceTimeUs() {
  return new Promise(resolve => {
    syncNode.port.onmessage = (e) => resolve(e.data.timeUs);
    syncNode.port.postMessage('getTime');
  });
}

// AudioUnit 输出回调 - 获取最精确的硬件播放时间
static OSStatus AudioOutputCallback(void *inRefCon,
                                    AudioUnitRenderActionFlags *ioActionFlags,
                                    const AudioTimeStamp *inTimeStamp,
                                    UInt32 inBusNumber,
                                    UInt32 inNumberFrames,
                                    AudioBufferList *ioData) {
    // inTimeStamp->mHostTime: 硬件音频设备当前时间
    // inTimeStamp->mSampleTime: 当前帧起始采样点 (基于 AudioUnit 时钟)
    // 关键：使用 mHostTime + mSampleTime 计算当前渲染时间线位置
    uint64_t hostTime = inTimeStamp->mHostTime;
    double sampleTime = inTimeStamp->mSampleTime;
    double sampleRate = GetAudioUnitSampleRate();
    
    // 当前音频渲染时间 (微秒)
    int64_t currentAudioRenderTimeUs = (hostTime + (sampleTime * 1e6 / sampleRate)) * 1e6 / mach_timebase_info().denom;
    
    // 存入无锁环形缓冲供视频线程读取
    gAudioClockRingBuffer.push(currentAudioRenderTimeUs);
    return noErr;
}

// 视频渲染线程
void VideoRenderLoop() {
  while (running) {
    CMSampleBufferRef frame = [decoder getNextFrame];
    int64_t framePTSUs = CMTimeGetSeconds(frame.presentationTimeStamp) * 1e6;
    
    // 从环形缓冲读取最近的音频渲染时间
    int64_t audioRenderTimeUs = gAudioClockRingBuffer.latest();
    
    int64_t diffMs = (framePTSUs - audioRenderTimeUs) / 1000;
    
    if (diffMs > 30) { // 视频超前
      // 方案: CVBufferRetain 重入队列头, 或调用 [videoOutput requestMediaDataWhenReadyOnQueue:]
      usleep(diffMs * 1000); // 简易睡眠, 生产环境用 CADisplayLink 回调驱动
    } else if (diffMs < -50) { // 视频滞后
      // 丢帧: 直接 release frame, 取下一帧
      CFRelease(frame);
      continue;
    }
    
    // 提交渲染
    [videoOutput enqueueSampleBuffer:frame];
  }
}

// 1. 创建 AudioTrack (关键: 设置 TIMESTAMP_MODE)
AudioTrack audioTrack = new AudioTrack.Builder()
    .setAudioAttributes(new AudioAttributes.Builder().setUsage(AudioAttributes.USAGE_VOICE_COMMUNICATION).build())
    .setAudioFormat(new AudioFormat.Builder().setEncoding(AudioFormat.ENCODING_PCM_16BIT).setSampleRate(48000).setChannelMask(AudioFormat.CHANNEL_OUT_STEREO).build())
    .setBufferSizeInBytes(bufferSize)
    .setTransferMode(AudioTrack.MODE_STREAM)
    // 核心: 允许设置每帧数据的呈现时间戳
    .setTimestampMode(AudioTrack.TIMESTAMP_MODE_UNSPECIFIED) // 或 MODE_CLOCK_MONOTONIC
    .build();

// 2. 创建 MediaSync (API 26+)
MediaSync mediaSync = new MediaSync();
mediaSync.setAudioTrack(audioTrack);

// 3. 视频帧渲染前同步
public void renderVideoFrame(VideoFrame frame) {
    long framePtsUs = frame.getTimestampUs(); // 解码器输出 PTS (微秒)
    
    // 获取音频当前播放位置 (微秒)
    AudioTimestamp audioTs = new AudioTimestamp();
    audioTrack.getTimestamp(audioTs, AudioTrack.TIMEBASE_MONOTONIC);
    long audioPresentUs = audioTs.framePosition * 1_000_000L / 48000;
    
    long diffMs = (framePtsUs - audioPresentUs) / 1000;
    
    if (diffMs > 30) {
        // 视频超前: 使用 MediaSync 等待
        // mediaSync.waitForAudioPresentationTimeUs(framePtsUs, timeoutMs);
        // 或手动 Thread.sleep
    } else if (diffMs < -50) {
        // 视频滞后: 丢帧
        frame.release();
        return;
    }
    
    // 提交 Surface
    surfaceTexture.updateTexImage();
    // ...
}

# test_sync_framework.py (基于 pytest + webrtc-python-binding / 自建信令)
import pytest
import asyncio
from network_emulator import NetEmulator # 基于 tc/netem 或 mahimahi
from clock_injector import ClockInjector # 修改 webrtc::Clock 实现, 注入虚拟时间

class SyncTestHarness:
    def __init__(self):
        self.sender_clock = ClockInjector(start_us=0, drift_ppm=0)
        self.receiver_clock = ClockInjector(start_us=100_000, drift_ppm=50) # 接收端快 50ppm
        self.net = NetEmulator(loss=0.05, rtt_ms=100, jitter_ms=30)
        
    async def run_scenario(self, duration_sec=30):
        # 启动发送端 (推流 1080p@30 + 48kHz Opus)
        sender = WebRTCPeer(clock=self.sender_clock, role='sender')
        # 启动接收端 (挂载自动化分析器)
        receiver = WebRTCPeer(clock=self.receiver_clock, role='receiver', 
                              analyzer=SyncAnalyzer())
        
        await signaling_connect(sender, receiver)
        await asyncio.sleep(duration_sec)
        
        return receiver.analyzer.generate_report()

class SyncAnalyzer:
    def __init(self):
        self.audio_render_times = [] # (ntp_us, frame_id)
        self.video_render_times = [] # (ntp_us, frame_id)
        
    def on_audio_render(self, ntp_us, frame_id):
        self.audio_render_times.append((ntp_us, frame_id))
        
    def on_video_render(self, ntp_us, frame_id):
        self.video_render_times.append((ntp_us, frame_id))
        
    def generate_report(self):
        # 对齐音视频帧 (最近邻匹配)
        pairs = self._match_av_frames()
        offsets = [v - a for a, v in pairs] # 视频相对音频偏移 (微秒)
        
        return {
            "convergence_time_ms": self._calc_convergence(offsets),
            "steady_state_offset_ms": np.median(offsets[-100:]) / 1000,
            "steady_state_jitter_ms": np.std(offsets[-100:]) / 1000,
            "max_offset_ms": np.max(np.abs(offsets)) / 1000,
            "freeze_count": self._count_freezes(),
            "drift_compensation_ok": self._verify_drift_correction()
        }

# 生成 HTML 报告片段
import plotly.graph_objects as go

def plot_sync_report(report_data, baseline_data=None):
    fig = go.Figure()
    fig.add_trace(go.Scatter(y=report_data['offsets_ms'], name='Current', line=dict(color='blue')))
    if baseline_data:
        fig.add_trace(go.Scatter(y=baseline_data['offsets_ms'], name='Baseline', line=dict(color='gray', dash='dash')))
    fig.add_hline(y=20, line_dash="dot", line_color="green", annotation_text="Target ±20ms")
    fig.add_hline(y=-20, line_dash="dot", line_color="green")
    fig.update_layout(title="A/V Sync Offset Time Series", yaxis_title="Offset (ms, Video - Audio)")
    return fig.to_html()

// 统一的帧时间戳定义 (webrtc/api/video/video_frame.h)
class VideoFrame {
  // ...
  // 核心时间戳字段
  Timestamp timestamp_;        // 采集/编码时间线 (单调, 微秒/纳秒)
  Timestamp render_time_;      // 目标渲染时间 (可选, 用于同步)
  int64_t ntp_time_ms_;        // 对应的 NTP 时间 (ms), 由 RTCP SR 映射填充
  // ...
};

// FrameTransformer 接口允许在编解码前后插桩
class FrameTransformerInterface {
  virtual void Transform(FrameTransformerInterface::Frame* frame) = 0;
};

// 同步插桩点示例
class SyncTimestampInjector : public FrameTransformerInterface {
  void Transform(Frame* frame) override {
    if (frame->type() == Frame::kVideo) {
      // 1. 从 RtpToNtpMapper 查询当前 RTP TS 对应的 NTP
      // 2. 回填 frame->ntp_time_ms_
      // 3. 计算建议渲染时间 render_time_ = ntp_time_ms_ + render_delay_ms
      frame->set_ntp_time_ms(mapper_->MapToNtpMs(frame->rtp_timestamp()));
      frame->set_render_time(Timestamp::Millis(frame->ntp_time_ms_ + kRenderDelayMs));
    }
  }
};