1
use std::net::Ipv4Addr;
2
use std::time::{Duration, Instant};
3
 
4
use str0m::format::Codec;
5
use str0m::media::MediaKind;
6
use str0m::net::Receive;
7
use str0m::rtp::{RtpWrite, Ssrc};
8
use str0m::{Candidate, Event, Input, Output, Rtc, RtcError};
9
 
10
mod common;
11
use common::{Peer, PendingPacket, TestRtc, init_crypto_default, init_log, progress};
12
 
13
fn connect_direct() -> Result<(TestRtc, TestRtc), RtcError> {
14
    let now = Instant::now();
15
    let mut l = TestRtc::new_with_rtc(Peer::Left.span(), Rtc::new(now));
16
    let mut r = TestRtc::new_with_rtc(
17
        Peer::Right.span(),
18
        Rtc::builder().set_reordering_size_video(10).build(now),
19
    );
20
 
21
    let host1 = Candidate::host((Ipv4Addr::new(1, 1, 1, 1), 1000).into(), "udp")?.clone();
22
    let host2 = Candidate::host((Ipv4Addr::new(2, 2, 2, 2), 2000).into(), "udp")?.clone();
23
    l.add_local_candidate(host1.clone());
24
    l.add_remote_candidate(host2.clone());
25
    r.add_local_candidate(host2);
26
    r.add_remote_candidate(host1);
27
 
28
    let finger_l = l.direct_api().local_dtls_fingerprint().clone();
29
    let finger_r = r.direct_api().local_dtls_fingerprint().clone();
30
    l.direct_api().set_remote_fingerprint(finger_r);
31
    r.direct_api().set_remote_fingerprint(finger_l);
32
 
33
    let creds_l = l.direct_api().local_ice_credentials();
34
    let creds_r = r.direct_api().local_ice_credentials();
35
    l.direct_api().set_remote_ice_credentials(creds_r);
36
    r.direct_api().set_remote_ice_credentials(creds_l);
37
 
38
    l.direct_api().set_ice_controlling(true);
39
    r.direct_api().set_ice_controlling(false);
40
 
41
    l.direct_api().start_dtls(true)?;
42
    r.direct_api().start_dtls(false)?;
43
    l.direct_api().start_sctp(true);
44
    r.direct_api().start_sctp(false);
45
 
46
    while !l.is_connected() || !r.is_connected() {
47
        progress(&mut l, &mut r)?;
48
    }
49
 
50
    Ok((l, r))
51
}
52
 
53
fn flush_transmits(rtc: &mut TestRtc, time: Instant) -> Result<Vec<PendingPacket>, RtcError> {
54
    let mut packets = Vec::new();
55
 
56
    rtc.rtc.handle_input(Input::Timeout(time))?;
57
 
58
    loop {
59
        match rtc.rtc.poll_output()? {
60
            Output::Timeout(v) => {
61
                let tick = rtc.last + rtc.forced_time_advance;
62
                rtc.last = if v == rtc.last { tick } else { tick.min(v) };
63
                break;
64
            }
65
            Output::Transmit(v) => {
66
                packets.push(PendingPacket {
67
                    proto: v.proto,
68
                    source: v.source,
69
                    destination: v.destination,
70
                    contents: v.contents.to_vec(),
71
                });
72
            }
73
            Output::Event(v) => rtc.events.push((time, v)),
74
        }
75
    }
76
 
77
    Ok(packets)
78
}
79
 
80
fn wait_for_transmits(
81
    rtc: &mut TestRtc,
82
    earliest: Instant,
83
    max_wait: Duration,
84
) -> Result<(Instant, Vec<PendingPacket>), RtcError> {
85
    let deadline = earliest + max_wait;
86
 
87
    loop {
88
        let time = rtc.last.max(earliest);
89
        let packets = flush_transmits(rtc, time)?;
90
 
91
        if !packets.is_empty() {
92
            return Ok((time, packets));
93
        }
94
 
95
        assert!(
96
            time < deadline,
97
            "expected at least one packet by {:?}, but sender emitted none",
98
            deadline.saturating_duration_since(earliest)
99
        );
100
    }
101
}
102
 
103
fn deliver_packet(rtc: &mut TestRtc, time: Instant, packet: PendingPacket) -> Result<(), RtcError> {
104
    let input = Input::Receive(
105
        time,
106
        Receive {
107
            proto: packet.proto,
108
            source: packet.source,
109
            destination: packet.destination,
110
            contents: (&packet.contents[..]).try_into()?,
111
        },
112
    );
113
 
114
    rtc.rtc.handle_input(input)?;
115
 
116
    loop {
117
        match rtc.rtc.poll_output()? {
118
            Output::Timeout(v) => {
119
                let tick = rtc.last + rtc.forced_time_advance;
120
                rtc.last = if v == rtc.last { tick } else { tick.min(v) };
121
                break;
122
            }
123
            Output::Transmit(_) => {}
124
            Output::Event(v) => rtc.events.push((time, v)),
125
        }
126
    }
127
 
128
    Ok(())
129
}
130
 
131
fn measure_fresh_age_after_pause(pause_duration: Duration) -> Result<Duration, RtcError> {
132
    let (mut l, mut r) = connect_direct()?;
133
 
134
    let mid = "vid".into();
135
    let ssrc: Ssrc = 42.into();
136
 
137
    l.direct_api().declare_media(mid, MediaKind::Video);
138
    l.direct_api().declare_stream_tx(ssrc, None, mid, None);
139
 
140
    r.direct_api().declare_media(mid, MediaKind::Video);
141
    r.direct_api().expect_stream_rx(ssrc, None, mid, None);
142
 
143
    let max = l.last.max(r.last);
144
    l.last = max;
145
    r.last = max;
146
 
147
    let pt = l.params_vp8().pt();
148
    assert_eq!(l.params_vp8().spec().codec, Codec::Vp8);
149
 
150
    let frame_timestamp = 90_000u32;
151
    let first_write_at = l.last + Duration::from_millis(20);
152
 
153
    {
154
        let mut direct = l.direct_api();
155
        let tx = direct.stream_tx_by_mid(mid, None).unwrap();
156
        tx.write_rtp(
157
            RtpWrite::new(
158
                pt,
159
                10_000u64.into(),
160
                frame_timestamp,
161
                first_write_at,
162
                [0x10, 0x00, 0xAA, 0xBB],
163
            )
164
            .nackable(true),
165
        );
166
    }
167
    let (first_emit_at, mut first_packets) =
168
        wait_for_transmits(&mut l, first_write_at, Duration::from_millis(50))?;
169
    assert_eq!(first_packets.len(), 1, "expected one first-fragment packet");
170
    deliver_packet(&mut r, first_emit_at, first_packets.remove(0))?;
171
 
172
    {
173
        let mut direct = l.direct_api();
174
        let tx = direct.stream_tx_by_mid(mid, None).unwrap();
175
        tx.write_rtp(
176
            RtpWrite::new(
177
                pt,
178
                10_001u64.into(),
179
                frame_timestamp,
180
                first_write_at + Duration::from_millis(5),
181
                [0x00, 0xCC, 0xDD],
182
            )
183
            .marker(true)
184
            .nackable(true),
185
        );
186
    }
187
    let (_, delayed_packets) = wait_for_transmits(
188
        &mut l,
189
        first_write_at + Duration::from_millis(5),
190
        Duration::from_millis(50),
191
    )?;
192
    assert!(
193
        !delayed_packets.is_empty(),
194
        "expected at least one packet when flushing delayed tail"
195
    );
196
 
197
    let pause_end = l.last + pause_duration;
198
    while l.last < pause_end || r.last < pause_end {
199
        progress(&mut l, &mut r)?;
200
    }
201
 
202
    assert!(
203
        r.events
204
            .iter()
205
            .any(|(_, e)| matches!(e, Event::StreamPaused(p) if p.paused)),
206
        "receiver never entered paused state"
207
    );
208
 
209
    let before_tail_event_count = r.events.len();
210
    let delivery_time = pause_end + Duration::from_millis(20);
211
    for packet in delayed_packets {
212
        deliver_packet(&mut r, delivery_time, packet)?;
213
    }
214
 
215
    assert!(
216
        !r.events[before_tail_event_count..]
217
            .iter()
218
            .any(|(_, e)| matches!(e, Event::MediaData(_))),
219
        "did not expect stale MediaData from the delayed pre-pause tail"
220
    );
221
 
222
    let fresh_first_at = delivery_time + Duration::from_millis(20);
223
    let fresh_frame_timestamp = frame_timestamp + 3_000;
224
 
225
    {
226
        let mut direct = l.direct_api();
227
        let tx = direct.stream_tx_by_mid(mid, None).unwrap();
228
        tx.write_rtp(
229
            RtpWrite::new(
230
                pt,
231
                10_002u64.into(),
232
                fresh_frame_timestamp,
233
                fresh_first_at,
234
                [0x10, 0x00, 0x11, 0x22],
235
            )
236
            .nackable(true),
237
        );
238
    }
239
    let (fresh_first_emit_at, mut fresh_first_packets) =
240
        wait_for_transmits(&mut l, fresh_first_at, Duration::from_millis(50))?;
241
    assert_eq!(
242
        fresh_first_packets.len(),
243
        1,
244
        "expected one fresh first-fragment packet"
245
    );
246
    deliver_packet(&mut r, fresh_first_emit_at, fresh_first_packets.remove(0))?;
247
 
248
    {
249
        let mut direct = l.direct_api();
250
        let tx = direct.stream_tx_by_mid(mid, None).unwrap();
251
        tx.write_rtp(
252
            RtpWrite::new(
253
                pt,
254
                10_003u64.into(),
255
                fresh_frame_timestamp,
256
                fresh_first_at + Duration::from_millis(5),
257
                [0x00, 0x33, 0x44],
258
            )
259
            .marker(true)
260
            .nackable(true),
261
        );
262
    }
263
    let (fresh_tail_emit_at, fresh_tail_packets) = wait_for_transmits(
264
        &mut l,
265
        fresh_first_at + Duration::from_millis(5),
266
        Duration::from_millis(50),
267
    )?;
268
    let before_fresh_event_count = r.events.len();
269
    for packet in fresh_tail_packets {
270
        deliver_packet(&mut r, fresh_tail_emit_at, packet)?;
271
    }
272
 
273
    let media = r.events[before_fresh_event_count..]
274
        .iter()
275
        .find_map(|(event_time, e)| match e {
276
            Event::MediaData(data) => Some((*event_time, data)),
277
            _ => None,
278
        })
279
        .expect("expected MediaData from the fresh post-pause frame");
280
 
281
    let age = media.0.saturating_duration_since(media.1.network_time);
282
 
283
    assert_eq!(*media.1.seq_range.start(), 10_002u64.into());
284
    assert_eq!(*media.1.seq_range.end(), 10_003u64.into());
285
 
286
    Ok(age)
287
}
288
 
289
fn collect_fresh_ages_after_repeated_pauses(
290
    pause_durations: &[Duration],
291
) -> Result<Vec<Duration>, RtcError> {
292
    let (mut l, mut r) = connect_direct()?;
293
 
294
    let mid = "vid".into();
295
    let ssrc: Ssrc = 42.into();
296
 
297
    l.direct_api().declare_media(mid, MediaKind::Video);
298
    l.direct_api().declare_stream_tx(ssrc, None, mid, None);
299
 
300
    r.direct_api().declare_media(mid, MediaKind::Video);
301
    r.direct_api().expect_stream_rx(ssrc, None, mid, None);
302
 
303
    let max = l.last.max(r.last);
304
    l.last = max;
305
    r.last = max;
306
 
307
    let pt = l.params_vp8().pt();
308
    assert_eq!(l.params_vp8().spec().codec, Codec::Vp8);
309
 
310
    let mut fresh_ages = Vec::new();
311
 
312
    for (i, pause_duration) in pause_durations.iter().copied().enumerate() {
313
        let seq = 20_000u64 + (i as u64) * 2;
314
        let frame_timestamp = 90_000u32 + (i as u32) * 3_000;
315
        let first_write_at = l.last + Duration::from_millis(20);
316
 
317
        {
318
            let mut direct = l.direct_api();
319
            let tx = direct.stream_tx_by_mid(mid, None).unwrap();
320
            tx.write_rtp(
321
                RtpWrite::new(
322
                    pt,
323
                    seq.into(),
324
                    frame_timestamp,
325
                    first_write_at,
326
                    [0x10, 0x00, 0xAA, 0xBB, i as u8],
327
                )
328
                .nackable(true),
329
            );
330
        }
331
        let (first_emit_at, mut first_packets) =
332
            wait_for_transmits(&mut l, first_write_at, Duration::from_millis(50))?;
333
        assert_eq!(first_packets.len(), 1, "expected one first-fragment packet");
334
        deliver_packet(&mut r, first_emit_at, first_packets.remove(0))?;
335
 
336
        {
337
            let mut direct = l.direct_api();
338
            let tx = direct.stream_tx_by_mid(mid, None).unwrap();
339
            tx.write_rtp(
340
                RtpWrite::new(
341
                    pt,
342
                    (seq + 1).into(),
343
                    frame_timestamp,
344
                    first_write_at + Duration::from_millis(5),
345
                    [0x00, 0xCC, 0xDD, i as u8],
346
                )
347
                .marker(true)
348
                .nackable(true),
349
            );
350
        }
351
        let (_, packets) = wait_for_transmits(
352
            &mut l,
353
            first_write_at + Duration::from_millis(5),
354
            Duration::from_millis(50),
355
        )?;
356
        assert!(
357
            !packets.is_empty(),
358
            "expected at least one packet when flushing delayed tail"
359
        );
360
        let pause_end = l.last + pause_duration;
361
        while l.last < pause_end || r.last < pause_end {
362
            progress(&mut l, &mut r)?;
363
        }
364
 
365
        assert!(
366
            r.events
367
                .iter()
368
                .any(|(_, e)| matches!(e, Event::StreamPaused(p) if p.paused)),
369
            "receiver never entered paused state"
370
        );
371
 
372
        let before_tail_event_count = r.events.len();
373
        let delivery_time = pause_end + Duration::from_millis(20);
374
        for packet in packets {
375
            deliver_packet(&mut r, delivery_time, packet)?;
376
        }
377
 
378
        assert_eq!(
379
            r.events[before_tail_event_count..]
380
                .iter()
381
                .filter(|(_, e)| matches!(e, Event::MediaData(_)))
382
                .count(),
383
            0,
384
            "did not expect stale MediaData from delayed pre-pause packets"
385
        );
386
 
387
        let fresh_first_at = delivery_time + Duration::from_millis(20);
388
        let fresh_frame_timestamp = frame_timestamp + 3_000;
389
        let fresh_seq = 30_000u64 + (i as u64) * 2;
390
 
391
        {
392
            let mut direct = l.direct_api();
393
            let tx = direct.stream_tx_by_mid(mid, None).unwrap();
394
            tx.write_rtp(
395
                RtpWrite::new(
396
                    pt,
397
                    fresh_seq.into(),
398
                    fresh_frame_timestamp,
399
                    fresh_first_at,
400
                    [0x10, 0x00, 0x55, 0x66, i as u8],
401
                )
402
                .nackable(true),
403
            );
404
        }
405
        let (fresh_first_emit_at, mut fresh_first_packets) =
406
            wait_for_transmits(&mut l, fresh_first_at, Duration::from_millis(50))?;
407
        assert_eq!(
408
            fresh_first_packets.len(),
409
            1,
410
            "expected one fresh first-fragment packet"
411
        );
412
        deliver_packet(&mut r, fresh_first_emit_at, fresh_first_packets.remove(0))?;
413
 
414
        {
415
            let mut direct = l.direct_api();
416
            let tx = direct.stream_tx_by_mid(mid, None).unwrap();
417
            tx.write_rtp(
418
                RtpWrite::new(
419
                    pt,
420
                    (fresh_seq + 1).into(),
421
                    fresh_frame_timestamp,
422
                    fresh_first_at + Duration::from_millis(5),
423
                    [0x00, 0x77, 0x88, i as u8],
424
                )
425
                .marker(true)
426
                .nackable(true),
427
            );
428
        }
429
        let (fresh_tail_emit_at, fresh_tail_packets) = wait_for_transmits(
430
            &mut l,
431
            fresh_first_at + Duration::from_millis(5),
432
            Duration::from_millis(50),
433
        )?;
434
        let before_fresh_event_count = r.events.len();
435
        for packet in fresh_tail_packets {
436
            deliver_packet(&mut r, fresh_tail_emit_at, packet)?;
437
        }
438
 
439
        let (event_time, media) = r.events[before_fresh_event_count..]
440
            .iter()
441
            .find_map(|(event_time, e)| match e {
442
                Event::MediaData(data) => Some((*event_time, data)),
443
                _ => None,
444
            })
445
            .expect("expected MediaData from the fresh post-pause frame");
446
 
447
        assert_eq!(*media.seq_range.start(), fresh_seq.into());
448
        assert_eq!(*media.seq_range.end(), (fresh_seq + 1).into());
449
        fresh_ages.push(event_time.saturating_duration_since(media.network_time));
450
    }
451
 
452
    Ok(fresh_ages)
453
}
454
 
455
#[test]
456
fn delayed_vp8_tail_packet_after_pause_is_dropped() -> Result<(), RtcError> {
457
    init_log();
458
    init_crypto_default();
459
 
460
    let age = measure_fresh_age_after_pause(Duration::from_millis(2200))?;
461
 
462
    assert!(
463
        age < Duration::from_millis(200),
464
        "expected fresh MediaData after pause recovery, got age {:?}",
465
        age
466
    );
467
 
468
    Ok(())
469
}
470
 
471
#[test]
472
fn longer_pauses_still_recover_with_fresh_media() -> Result<(), RtcError> {
473
    init_log();
474
    init_crypto_default();
475
 
476
    let pause_durations = [
477
        Duration::from_millis(2200),
478
        Duration::from_millis(4200),
479
        Duration::from_millis(8200),
480
    ];
481
 
482
    let observed_ages: Vec<_> = pause_durations
483
        .into_iter()
484
        .map(measure_fresh_age_after_pause)
485
        .collect::<Result<_, _>>()?;
486
 
487
    assert!(
488
        observed_ages
489
            .iter()
490
            .all(|age| *age < Duration::from_millis(200)),
491
        "expected fresh recovery after long pauses, got {:?}",
492
        observed_ages
493
    );
494
 
495
    Ok(())
496
}
497
 
498
#[test]
499
fn repeated_pause_cycles_recover_without_stale_media() -> Result<(), RtcError> {
500
    init_log();
501
    init_crypto_default();
502
 
503
    let ages = collect_fresh_ages_after_repeated_pauses(&[
504
        Duration::from_millis(2200),
505
        Duration::from_millis(4200),
506
        Duration::from_millis(6200),
507
    ])?;
508
 
509
    assert!(
510
        ages.iter().all(|age| *age < Duration::from_millis(200)),
511
        "expected fresh media after each pause cycle, got {:?}",
512
        ages
513
    );
514
 
515
    Ok(())
516
}