diff --git a/agent/src/bin/dump_stream.rs b/agent/src/bin/dump_stream.rs new file mode 100644 index 0000000..9d61853 --- /dev/null +++ b/agent/src/bin/dump_stream.rs @@ -0,0 +1,272 @@ +//! 临时诊断工具：抓 agent 下发的码流，逐帧打印 NAL 结构并试解码。 +//! +//! 目的：定位「首帧能解、之后连续 decode failed」的根因，不靠猜。 +//! +//! 用法： +//! cargo run --release -p agent --bin dump_stream -- --port 47805 --code t +//! cargo run --release -p agent --bin dump_stream -- --port 47805 --code t --skip-at 2 +//! +//! `--skip-at N`：模拟主控端的「积压跳帧」行为——第 N 帧读出但不喂给解码器， +//! 用来验证「跳帧是否直接导致后续所有 P 帧解不出来」。 + +use anyhow::{bail, Result}; +use common::{read_control, read_video_frame, write_control, ControlMsg}; +use openh264::decoder::Decoder; +use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}; +use rustls::pki_types::{CertificateDer, ServerName, UnixTime}; +use rustls::{DigitallySignedStruct, SignatureScheme}; +use std::net::ToSocketAddrs; +use std::sync::Arc; +use std::time::Duration; + +#[derive(Debug)] +struct SkipVerify; +impl ServerCertVerifier for SkipVerify { + fn verify_server_cert( + &self, + _e: &CertificateDer<'_>, + _i: &[CertificateDer<'_>], + _s: &ServerName<'_>, + _o: &[u8], + _n: UnixTime, + ) -> Result { + Ok(ServerCertVerified::assertion()) + } + fn verify_tls12_signature( + &self, + _m: &[u8], + _c: &CertificateDer<'_>, + _d: &DigitallySignedStruct, + ) -> Result { + Ok(HandshakeSignatureValid::assertion()) + } + fn verify_tls13_signature( + &self, + _m: &[u8], + _c: &CertificateDer<'_>, + _d: &DigitallySignedStruct, + ) -> Result { + Ok(HandshakeSignatureValid::assertion()) + } + fn supported_verify_schemes(&self) -> Vec { + vec![ + SignatureScheme::RSA_PKCS1_SHA256, + SignatureScheme::ECDSA_NISTP256_SHA256, + SignatureScheme::ED25519, + SignatureScheme::RSA_PSS_SHA256, + ] + } +} + +/// 扫 Annex-B 起始码，返回 NAL 类型列表（如 [7, 8, 5]） +fn nal_types(payload: &[u8]) -> Vec { + let mut out = Vec::new(); + let mut i = 0usize; + while i + 3 < payload.len() { + let sc = if payload[i..].starts_with(&[0, 0, 0, 1]) { + 4 + } else if payload[i..].starts_with(&[0, 0, 1]) { + 3 + } else { + i += 1; + continue; + }; + match payload.get(i + sc) { + Some(&b) => { + out.push(b & 0x1F); + i += sc + 1; + } + None => break, + } + } + out +} + +fn describe(types: &[u8]) -> String { + let has = |t: u8| types.contains(&t); + let mut s = String::new(); + if has(7) { + s.push_str("SPS "); + } + if has(8) { + s.push_str("PPS "); + } + if has(5) { + s.push_str("IDR "); + } + if has(1) { + s.push_str("P "); + } + if has(6) { + s.push_str("SEI "); + } + if s.is_empty() { + s.push_str("(未知)"); + } + format!("{}[{}]", s.trim(), types.len()) +} + +fn arg(name: &str) -> Option { + let a: Vec = std::env::args().collect(); + a.iter() + .position(|x| x == name) + .and_then(|i| a.get(i + 1)) + .cloned() +} + +#[tokio::main] +async fn main() -> Result<()> { + rustls::crypto::ring::default_provider() + .install_default() + .ok(); + + let port: u16 = arg("--port") + .map(|s| s.parse()) + .transpose()? + .unwrap_or(47800); + let code = arg("--code").unwrap_or_else(|| "123456".into()); + let skip_at: Option = arg("--skip-at").map(|s| s.parse()).transpose()?; + let total: usize = arg("--frames") + .map(|s| s.parse()) + .transpose()? + .unwrap_or(30); + + let tls = rustls::ClientConfig::builder() + .dangerous() + .with_custom_certificate_verifier(Arc::new(SkipVerify)) + .with_no_client_auth(); + let qc: quinn::crypto::rustls::QuicClientConfig = Arc::new(tls).try_into()?; + let mut transport = quinn::TransportConfig::default(); + transport.receive_window(quinn::VarInt::from_u32(16 * 1024 * 1024)); + transport.stream_receive_window(quinn::VarInt::from_u32(8 * 1024 * 1024)); + let mut client_config = quinn::ClientConfig::new(Arc::new(qc)); + client_config.transport_config(Arc::new(transport)); + let mut endpoint = quinn::Endpoint::client("0.0.0.0:0".parse()?)?; + endpoint.set_default_client_config(client_config); + + let addr = ("127.0.0.1", port) + .to_socket_addrs()? + .next() + .ok_or_else(|| anyhow::anyhow!("解析失败"))?; + let conn = endpoint.connect(addr, "localhost")?.await?; + println!("[dump] connected"); + + let (mut send, mut recv) = conn.open_bi().await?; + write_control( + &mut send, + &ControlMsg::Auth { + code, + proto_version: common::PROTO_VERSION, + }, + ) + .await?; + match read_control(&mut recv).await? { + ControlMsg::AuthOk { + width, + height, + proto_version, + } => println!("[dump] auth ok {width}x{height} proto={proto_version}"), + other => bail!("认证失败: {other:?}"), + } + + let (mut isend, _irecv) = conn.open_bi().await?; + + // 入会先请求一次 IDR，和 viewer 行为一致。 + // 注意：KeyFrameReq 必须走**流 2**（isend），agent 的读循环只监听流 2。 + write_control(&mut isend, &ControlMsg::KeyFrameReq).await?; + + let mut decoder = Decoder::new()?; + let mut sent_kfreq = false; + + println!("\n序号 seq kf 载荷字节 NAL 结构 解码结果"); + println!("{}", "-".repeat(72)); + + for i in 0..total { + let (hdr, payload) = match tokio::time::timeout( + Duration::from_secs(10), + read_video_frame(&mut recv), + ) + .await + { + Ok(r) => r?, + Err(_) => { + println!("[{i}] 读帧超时（桌面可能完全静止）"); + break; + } + }; + + let types = nal_types(&payload); + let is_idr = types.contains(&5); + + // 交叉校验：帧头 keyframe 标志必须与码流实际 NAL 类型一致。 + // 这条不一致意味着时间戳/序号契约或编码器标志有 bug，比画面花屏更难排查。 + if hdr.keyframe != is_idr { + println!( + " ⚠️ 帧头 keyframe={} 但码流实际 IDR={}（seq={}）", + hdr.keyframe, is_idr, hdr.seq + ); + } + + // 模拟主控端积压跳帧：读出但不解码 + if Some(i) == skip_at { + println!( + "[{i:2}] {:3} {:<5} {:8} {:<14} ⏭ 跳过解码（模拟积压跳帧）", + hdr.seq, + hdr.keyframe, + payload.len(), + describe(&types) + ); + continue; + } + + let res = match decoder.decode(&payload) { + Ok(Some(_)) => "✅ 解出画面".to_string(), + Ok(None) => "○ 无输出（缓冲中）".to_string(), + Err(e) => format!("❌ {e}"), + }; + + println!( + "[{i:2}] {:3} {:<5} {:8} {:<14} {}", + hdr.seq, + hdr.keyframe, + payload.len(), + describe(&types), + res + ); + + // 和 viewer 一样：解码失败就请求关键帧 + if res.starts_with('❌') && !sent_kfreq { + sent_kfreq = true; + write_control(&mut isend, &ControlMsg::KeyFrameReq).await?; + println!(" ↑ 已发 KeyFrameReq"); + } + } + + // 诊断：静止桌面下再请求一次关键帧，看是否真能拿到 IDR + println!("\n[dump] 再发一次 KeyFrameReq，观察 10 帧…"); + write_control(&mut isend, &ControlMsg::KeyFrameReq).await?; + for i in 0..10 { + match tokio::time::timeout(Duration::from_secs(5), read_video_frame(&mut recv)).await { + Ok(Ok((hdr, payload))) => { + let types = nal_types(&payload); + println!( + "[k{i:2}] seq={:3} kf={:<5} {:6}B {}", + hdr.seq, + hdr.keyframe, + payload.len(), + describe(&types) + ); + if types.contains(&5) { + println!(" ↑ 收到 IDR"); + } + } + _ => { + println!("[k{i:2}] 超时/流关闭"); + break; + } + } + } + + let _ = conn.close(0u32.into(), b"done"); + Ok(()) +}