1
//! Tests that media is correctly delivered under packet loss conditions.
2
//!
3
//! Uses the netem crate to simulate realistic bursty packet loss and verifies
4
//! that RTP retransmissions recover all lost packets.
5
 
6
mod common;
7
 
8
use std::time::Duration;
9
 
10
use common::{connect_l_r, init_crypto_default, progress, vp8_data};
11
use netem::{GilbertElliot, LossModel, NetemConfig};
12
use str0m::format::Codec;
13
use str0m::media::MediaKind;
14
use str0m::rtp::{RtpWrite, Ssrc};
15
use str0m::{Event, RtcError};
16
 
17
use crate::common::init_log;
18
 
19
/// Run a media transmission test with the given loss model using real VP8 data.
20
///
21
/// Returns the number of packets received.
22
fn run_loss_test(loss_model: impl Into<LossModel>, seed: u64) -> Result<usize, RtcError> {
23
    init_log();
24
    init_crypto_default();
25
 
26
    let (mut l, mut r) = connect_l_r();
27
 
28
    // Configure netem on R's incoming queue (L -> R has loss)
29
    let config = NetemConfig::new().loss(loss_model).seed(seed);
30
    r.set_netem(config);
31
 
32
    let mid = "vid".into();
33
    let ssrc_tx: Ssrc = 42.into();
34
    let ssrc_rtx: Ssrc = 44.into();
35
 
36
    l.direct_api().declare_media(mid, MediaKind::Video);
37
    l.direct_api()
38
        .declare_stream_tx(ssrc_tx, Some(ssrc_rtx), mid, None);
39
 
40
    // Increase the RTX ratio cap to 0.2 to allow more retransmissions.
41
    // With heavy loss, the RTX ratio naturally grows, and the default
42
    // cap (0.15) is too low to allow all needed retransmissions.
43
    l.direct_api().stream_tx(&ssrc_tx).unwrap().set_rtx_cache(
44
        1024,
45
        Duration::from_secs(3),
46
        Some(0.2),
47
    );
48
 
49
    r.direct_api().declare_media(mid, MediaKind::Video);
50
    r.direct_api()
51
        .expect_stream_rx(ssrc_tx, Some(ssrc_rtx), mid, None);
52
 
53
    let max = l.last.max(r.last);
54
    l.last = max;
55
    r.last = max;
56
 
57
    let params = l.params_vp8();
58
    assert_eq!(params.spec().codec, Codec::Vp8);
59
    let pt = params.pt();
60
 
61
    let data = vp8_data();
62
    let packet_count = data.len();
63
 
64
    for (relative, header, payload) in data {
65
        // Keep RTC time progressed to be "in sync" with the test data
66
        while (l.last - max) < relative {
67
            progress(&mut l, &mut r)?;
68
        }
69
 
70
        let absolute = max + relative;
71
 
72
        let mut direct = l.direct_api();
73
        let tx = direct.stream_tx(&ssrc_tx).unwrap();
74
        tx.write_rtp(
75
            RtpWrite::new(
76
                pt,
77
                header.sequence_number(None),
78
                header.timestamp,
79
                absolute,
80
                payload,
81
            )
82
            .marker(header.marker)
83
            .nackable(true),
84
        );
85
 
86
        progress(&mut l, &mut r)?;
87
 
88
        if l.duration() > Duration::from_secs(10) {
89
            break;
90
        }
91
    }
92
 
93
    // Let retransmissions complete (also subject to loss)
94
    let settle_time = l.duration() + Duration::from_secs(2);
95
    while l.duration() < settle_time {
96
        progress(&mut l, &mut r)?;
97
    }
98
 
99
    // Count received RTP packets
100
    let mut received_seqs: Vec<u64> = r
101
        .events
102
        .iter()
103
        .filter_map(|(_, e)| {
104
            if let Event::RtpPacket(v) = e {
105
                Some(*v.seq_no)
106
            } else {
107
                None
108
            }
109
        })
110
        .collect();
111
 
112
    received_seqs.sort();
113
 
114
    println!(
115
        "Sent {} packets, received {} packets",
116
        packet_count,
117
        received_seqs.len()
118
    );
119
 
120
    // Find which sequence numbers are missing
121
    if !received_seqs.is_empty() {
122
        let first = received_seqs[0];
123
        let last = *received_seqs.last().unwrap();
124
        let expected: Vec<u64> = (first..=last).collect();
125
        let missing: Vec<u64> = expected
126
            .iter()
127
            .filter(|s| !received_seqs.contains(s))
128
            .copied()
129
            .collect();
130
        if !missing.is_empty() {
131
            println!("Missing seq numbers from Event::RtpPacket: {:?}", missing);
132
        }
133
    }
134
 
135
    Ok(received_seqs.len())
136
}
137
 
138
#[test]
139
fn loss_light() -> Result<(), RtcError> {
140
    // ~1% loss - should recover everything easily
141
    let loss = GilbertElliot::wifi();
142
    let received = run_loss_test(loss, 12345)?;
143
 
144
    // VP8 data has 104 packets, with light loss we should get all
145
    assert_eq!(
146
        received, 104,
147
        "Expected 104 packets with light loss, got {}",
148
        received
149
    );
150
    Ok(())
151
}
152
 
153
#[test]
154
fn loss_medium() -> Result<(), RtcError> {
155
    // ~5% loss
156
    let loss = GilbertElliot::wifi_lossy();
157
    let received = run_loss_test(loss, 12345)?;
158
 
159
    assert_eq!(
160
        received, 104,
161
        "Expected 104 packets with medium loss, got {}",
162
        received
163
    );
164
    Ok(())
165
}
166
 
167
#[test]
168
fn loss_heavy() -> Result<(), RtcError> {
169
    // ~10% loss
170
    let loss = GilbertElliot::congested();
171
    let received = run_loss_test(loss, 12345)?;
172
 
173
    assert_eq!(
174
        received, 104,
175
        "Expected 104 packets with heavy loss, got {}",
176
        received
177
    );
178
    Ok(())
179
}