ez-ffmpeg 0.17.0

//! Join burst tests: packet flags, byte budget and GOP selection.

use super::super::join::{
    build_join_burst, gop_wire_size, join_replay_prefix_bytes, select_replay_start,
    CONT_HEADER_MAX, JOIN_REPLAY_BUDGET_BYTES, MSG_HEADER_MAX,
};
use super::*;
use crate::flv::flv_tag_body::is_video_keyframe;
use rml_rtmp::sessions::ServerSessionResult;

// ---- H8: join burst (flags, budget trim, current-GOP inclusion) ----

#[test]
fn join_burst_flags_every_packet_like_the_live_path() {
    let mut channel = MediaChannel::new(10);
    channel.metadata = Some(Rc::new(StreamMetadata {
        video_width: None,
        video_height: None,
        video_codec_id: Some(7), // AVC
        video_frame_rate: None,
        video_bitrate_kbps: None,
        audio_codec_id: Some(10), // AAC
        audio_bitrate_kbps: None,
        audio_sample_rate: None,
        audio_channels: None,
        audio_is_stereo: None,
        encoder: None,
    }));
    channel.video_sequence_header = Some(Bytes::from_static(VIDEO_SEQ));
    channel.audio_sequence_header = Some(Bytes::from_static(AUDIO_SEQ));

    // Frozen GOP: keyframe + audio + delta; the second keyframe freezes
    // it and opens the current GOP.
    channel
        .gops
        .save_frame_data(video_frame(0, Bytes::from_static(KEYFRAME)), true);
    channel
        .gops
        .save_frame_data(audio_frame(10, Bytes::from_static(AUDIO_FRAME)), false);
    channel
        .gops
        .save_frame_data(video_frame(33, Bytes::from_static(DELTA)), false);
    channel
        .gops
        .save_frame_data(video_frame(66, Bytes::from_static(KEYFRAME)), true);

    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(&channel, &mut client, 7, 1, 0, &mut out);

    for result in &out {
        if let ServerResult::OutboundPacket {
            target_connection_id,
            ..
        } = result
        {
            assert_eq!(*target_connection_id, 7);
        }
    }
    assert_eq!(
        burst_flags(&out),
        vec![
            (false, false, false), // metadata
            (false, true, true),   // video sequence header
            (false, true, false),  // audio sequence header
            (true, false, true),   // frozen GOP keyframe
            (false, false, false), // frozen GOP audio
            (false, false, true),  // frozen GOP delta
            (true, false, true),   // current GOP keyframe
        ],
        "every replayed packet must carry the flags the live path computes"
    );
    assert!(
        client.has_received_video_keyframe,
        "replaying a flagged keyframe must open the keyframe gate"
    );
}

#[test]
fn join_burst_trims_whole_oldest_gops_to_the_byte_budget() {
    let mut channel = MediaChannel::new(10);
    channel.video_sequence_header = Some(Bytes::from_static(VIDEO_SEQ));
    channel.audio_sequence_header = Some(Bytes::from_static(AUDIO_SEQ));

    // Three ~400KiB single-keyframe GOPs (marker at data[2]) plus a small
    // current GOP. Newest-first the 960KiB budget admits current + GOP3 +
    // GOP2 (~800KiB) but not GOP1.
    let make_keyframe = |marker: u8| {
        let mut data = vec![0u8; 400 * 1024];
        data[0] = 0x17;
        data[1] = 0x01;
        data[2] = marker;
        Bytes::from(data)
    };
    for (i, marker) in [1u8, 2, 3].into_iter().enumerate() {
        channel
            .gops
            .save_frame_data(video_frame(i as u32 * 100, make_keyframe(marker)), true);
    }
    // The fourth keyframe freezes GOP3 and becomes the (small) current GOP.
    channel
        .gops
        .save_frame_data(video_frame(300, Bytes::from_static(KEYFRAME)), true);

    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(&channel, &mut client, 7, 1, 0, &mut out);

    // vseq + aseq + GOP2 + GOP3 + current — GOP1 trimmed as a whole.
    assert_eq!(out.len(), 5, "the burst must trim the oldest GOP entirely");
    assert_eq!(
        burst_flags(&out)[..2],
        [(false, true, true), (false, true, false)],
        "sequence headers are sent outside the budget trim"
    );
    assert!(
        out.iter().all(|p| !packet_contains(p, &[0x17, 0x01, 1])),
        "no fragment of the trimmed GOP1 may be replayed"
    );
    assert!(
        packet_contains(&out[2], &[0x17, 0x01, 2]),
        "the replay must start at GOP2's keyframe"
    );
    assert!(packet_contains(&out[3], &[0x17, 0x01, 3]));
    assert!(client.has_received_video_keyframe);
}

#[test]
fn join_burst_with_only_oversized_gops_sends_headers_only() {
    let mut channel = MediaChannel::new(10);
    channel.video_sequence_header = Some(Bytes::from_static(VIDEO_SEQ));
    channel.audio_sequence_header = Some(Bytes::from_static(AUDIO_SEQ));

    // One frozen GOP and a current GOP, each alone above the budget: no
    // segment fits, so the burst degrades to the sequence headers.
    let huge = {
        let mut data = vec![0u8; JOIN_REPLAY_BUDGET_BYTES + 1];
        data[0] = 0x17;
        data[1] = 0x01;
        Bytes::from(data)
    };
    channel
        .gops
        .save_frame_data(video_frame(0, huge.clone()), true);
    channel.gops.save_frame_data(video_frame(100, huge), true);

    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(&channel, &mut client, 7, 1, 0, &mut out);

    assert_eq!(
        burst_flags(&out),
        vec![(false, true, true), (false, true, false)],
        "an oversized cache must degrade to sequence headers only"
    );
    assert!(
        !client.has_received_video_keyframe,
        "no keyframe was replayed, so live deltas must stay gated until a live keyframe"
    );
}

// H8.a regression: the replay used to iterate frozen GOPs only. A joiner
// then received live deltas referencing the open GOP's keyframe it never got —
// a smeared picture until the next keyframe.
#[test]
fn join_burst_includes_the_open_current_gop_as_the_last_segment() {
    const CURRENT_KEYFRAME: &[u8] = &[0x17, 0x01, 0xB2, 0x00, 0x00];
    const CURRENT_DELTA: &[u8] = &[0x27, 0x01, 0xB3, 0x00, 0x00];

    let mut channel = MediaChannel::new(10);
    channel.video_sequence_header = Some(Bytes::from_static(VIDEO_SEQ));

    // Frozen GOP: keyframe + delta. Open GOP: a second keyframe + delta
    // not yet frozen by any later keyframe.
    channel
        .gops
        .save_frame_data(video_frame(0, Bytes::from_static(KEYFRAME)), true);
    channel
        .gops
        .save_frame_data(video_frame(33, Bytes::from_static(DELTA)), false);
    channel
        .gops
        .save_frame_data(video_frame(66, Bytes::from_static(CURRENT_KEYFRAME)), true);
    channel
        .gops
        .save_frame_data(video_frame(99, Bytes::from_static(CURRENT_DELTA)), false);

    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(&channel, &mut client, 7, 1, 0, &mut out);

    // vseq + frozen (keyframe, delta) + current (keyframe, delta).
    assert_eq!(
        out.len(),
        5,
        "the open GOP must be replayed after the frozen ones"
    );
    assert!(
        packet_contains(&out[3], CURRENT_KEYFRAME),
        "the open GOP's keyframe must be replayed — live deltas reference it"
    );
    assert!(packet_contains(&out[4], CURRENT_DELTA));
    assert_eq!(
        burst_flags(&out)[3..],
        [(true, false, true), (false, false, true)]
    );
}

#[test]
fn keyframeless_current_gop_replays_audio_only_and_keeps_the_gate_closed() {
    let mut channel = MediaChannel::new(10);
    channel.video_sequence_header = Some(Bytes::from_static(VIDEO_SEQ));
    channel.audio_sequence_header = Some(Bytes::from_static(AUDIO_SEQ));

    // Publish started mid-GOP: the open GOP holds deltas and audio, no
    // keyframe, and nothing is frozen yet.
    channel
        .gops
        .save_frame_data(video_frame(0, Bytes::from_static(DELTA)), false);
    channel
        .gops
        .save_frame_data(audio_frame(10, Bytes::from_static(AUDIO_FRAME)), false);
    channel
        .gops
        .save_frame_data(video_frame(33, Bytes::from_static(DELTA)), false);
    channel
        .gops
        .save_frame_data(audio_frame(43, Bytes::from_static(AUDIO_FRAME)), false);

    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(&channel, &mut client, 7, 1, 0, &mut out);

    assert_eq!(
        burst_flags(&out),
        vec![
            (false, true, true),   // video sequence header
            (false, true, false),  // audio sequence header
            (false, false, false), // audio
            (false, false, false), // audio
        ],
        "undecodable pre-keyframe deltas must be skipped while audio still flows"
    );
    assert!(
        !client.has_received_video_keyframe,
        "a keyframeless replay must not open the gate"
    );
}

#[test]
fn select_replay_start_picks_the_longest_fitting_suffix() {
    let cases: &[(&[usize], usize, usize)] = &[
        (&[], 100, 0),    // nothing cached
        (&[10], 100, 0),  // everything fits
        (&[100], 100, 0), // exactly the budget fits (<=)
        (&[101], 100, 1), // a single oversized segment -> none
        (&[100, 200, 300], 600, 0),
        (&[100, 200, 300], 599, 1),
        (&[100, 200, 300], 500, 1),
        (&[100, 200, 300], 499, 2),
        (&[100, 200, 300], 300, 2),
        (&[100, 200, 300], 299, 3),
        (&[0, 0, 0], 0, 0), // zero-size segments always fit
        // Overflowing older segments must fail closed — keep the newest
        // fitting suffix, never wrap around and admit everything.
        (&[usize::MAX, 100], usize::MAX, 1),
        (&[usize::MAX - 50, 100], usize::MAX, 1),
    ];
    for &(sizes, budget, expected) in cases {
        assert_eq!(
            select_replay_start(sizes, budget),
            expected,
            "sizes={sizes:?} budget={budget}"
        );
    }
}

// ---- F1: the replay budget must count real wire bytes, not raw payload ----

/// gop_wire_size adds the RTMP chunk framing on top of the raw payload.
#[test]
fn gop_wire_size_adds_per_frame_and_continuation_framing() {
    // Empty GOP: no payload, no frames -> no framing.
    assert_eq!(gop_wire_size(0, 0), 0);
    // One 100-byte frame -> payload + one type-0 header + one chunk's cont.
    assert_eq!(
        gop_wire_size(100, 1),
        100 + MSG_HEADER_MAX + CONT_HEADER_MAX
    );
    // Many small frames: the per-frame header dominates the payload.
    // 1000 bytes spans a single 4096-byte chunk (one continuation header).
    assert_eq!(
        gop_wire_size(1000, 100),
        1000 + 100 * MSG_HEADER_MAX + CONT_HEADER_MAX
    );
    // A payload spanning several chunks accrues one cont header per chunk.
    let payload = OUTBOUND_CHUNK_SIZE * 3 + 1;
    assert_eq!(
        gop_wire_size(payload, 1),
        payload + MSG_HEADER_MAX + 4 * CONT_HEADER_MAX
    );
}

/// A metadata packet near the 64 KiB headroom must reduce the GOP budget so
/// a GOP that fits the raw-payload budget is trimmed, keeping the real
/// serialized burst under the Warning threshold.
#[test]
fn join_burst_oversized_metadata_trims_gops_below_the_warning_threshold() {
    const GOP_MARKER: &[u8] = &[0x17, 0x01, 0xC1];
    let warn = crate::rtmp::write_queue::QUEUE_WARN_BYTES;

    let mut channel = MediaChannel::new(10);
    // A large-but-legal encoder string (AMF0 UTF8 tops out at 65535 bytes).
    channel.metadata = Some(Rc::new(StreamMetadata {
        video_width: None,
        video_height: None,
        video_codec_id: Some(7),
        video_frame_rate: None,
        video_bitrate_kbps: None,
        audio_codec_id: None,
        audio_bitrate_kbps: None,
        audio_sample_rate: None,
        audio_channels: None,
        audio_is_stereo: None,
        encoder: Some("x".repeat(60_000)),
    }));
    channel.video_sequence_header = Some(Bytes::from_static(VIDEO_SEQ));
    channel.audio_sequence_header = Some(Bytes::from_static(AUDIO_SEQ));

    // A single ~950 KiB one-keyframe GOP: it fits the raw 960 KiB budget, but not
    // once ~60 KiB of metadata is subtracted.
    let mut keyframe = vec![0u8; 950 * 1024];
    keyframe[0] = 0x17;
    keyframe[1] = 0x01;
    keyframe[2] = 0xC1;
    channel
        .gops
        .save_frame_data(video_frame(0, Bytes::from(keyframe)), true);

    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(&channel, &mut client, 7, 1, 0, &mut out);

    assert!(
        !out.iter().any(|p| packet_contains(p, GOP_MARKER)),
        "the oversized-metadata prefix must trim the GOP that the raw budget would have kept"
    );
    assert!(
        serialized_burst_len(&out) <= warn,
        "the real serialized burst must stay within the Warning threshold ({} <= {})",
        serialized_burst_len(&out),
        warn
    );
}

/// A high-frame-count GOP whose payload alone fits must still be trimmed
/// once per-frame chunk framing is counted.
#[test]
fn join_burst_many_small_frames_framing_trims_older_gops() {
    const OLD_KEYFRAME: &[u8] = &[0x17, 0x01, 0xD1];
    const NEW_KEYFRAME: &[u8] = &[0x17, 0x01, 0xD2];
    let warn = crate::rtmp::write_queue::QUEUE_WARN_BYTES;

    let mut channel = MediaChannel::new(10);
    channel.video_sequence_header = Some(Bytes::from_static(VIDEO_SEQ));
    channel.audio_sequence_header = Some(Bytes::from_static(AUDIO_SEQ));

    // Two GOPs of 3000 x 150-byte frames each. Payload sum (~880 KiB) fits
    // the 960 KiB budget, but the per-frame framing (3000 x 18 bytes/GOP)
    // pushes the pair over it, so the older GOP is trimmed as a whole.
    let keyframe = |marker: u8| {
        let mut d = vec![0u8; 150];
        d[0] = 0x17;
        d[1] = 0x01;
        d[2] = marker;
        Bytes::from(d)
    };
    let delta = || {
        let mut d = vec![0u8; 150];
        d[0] = 0x27;
        d[1] = 0x01;
        Bytes::from(d)
    };
    channel
        .gops
        .save_frame_data(video_frame(0, keyframe(0xD1)), true);
    for i in 0..2999u32 {
        channel
            .gops
            .save_frame_data(video_frame(i + 1, delta()), false);
    }
    // The second keyframe freezes the first GOP and opens the second.
    channel
        .gops
        .save_frame_data(video_frame(3000, keyframe(0xD2)), true);
    for i in 0..2999u32 {
        channel
            .gops
            .save_frame_data(video_frame(3001 + i, delta()), false);
    }

    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(&channel, &mut client, 7, 1, 0, &mut out);

    assert!(
        out.iter().any(|p| packet_contains(p, NEW_KEYFRAME)),
        "the newest GOP must be replayed"
    );
    assert!(
        !out.iter().any(|p| packet_contains(p, OLD_KEYFRAME)),
        "framing must trim the older GOP even though its payload alone fits"
    );
    assert!(
        serialized_burst_len(&out) <= warn,
        "the real serialized burst must stay within the Warning threshold"
    );
}

/// An oversized sequence header exhausts the budget by itself: the prefix
/// subtraction saturates to zero and no delta GOP is replayed — no panic,
/// no underflow.
#[test]
fn join_burst_oversized_sequence_header_replays_zero_gops() {
    const GOP_MARKER: &[u8] = &[0x17, 0x01, 0xE1];

    let mut channel = MediaChannel::new(10);
    // A 2 MiB sequence header (adversarial but legal), far over the budget.
    let mut header = vec![0u8; 2 * 1024 * 1024];
    header[0] = 0x17;
    header[1] = 0x00;
    channel.video_sequence_header = Some(Bytes::from(header));

    let mut keyframe = vec![0u8; 4096];
    keyframe[0] = 0x17;
    keyframe[1] = 0x01;
    keyframe[2] = 0xE1;
    channel
        .gops
        .save_frame_data(video_frame(0, Bytes::from(keyframe)), true);

    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    // Must not panic on the budget subtraction (saturating to zero).
    build_join_burst(&channel, &mut client, 7, 1, 0, &mut out);

    assert!(
        !out.iter().any(|p| packet_contains(p, GOP_MARKER)),
        "a prefix that exhausts the budget must replay zero GOPs"
    );
    assert!(
        !client.has_received_video_keyframe,
        "no keyframe was replayed, so the keyframe gate must stay closed"
    );
}

/// F1: the play-accept control packets enqueued ahead of the burst count
/// against the GOP budget. A ~64 KiB accept prefix (an oversized stream key
/// echoed by NetStream.Play.Start) must trim a GOP the raw budget would
/// keep, so the real enqueued bytes (accept prefix + burst) stay under the
/// Warning threshold and no delta frame is dropped by backpressure.
#[test]
fn join_burst_counts_the_accept_prefix_against_the_gop_budget() {
    const GOP_MARKER: &[u8] = &[0x17, 0x01, 0xF1];
    let warn = crate::rtmp::write_queue::QUEUE_WARN_BYTES;

    // A single ~950 KiB one-keyframe GOP: it fits the raw 960 KiB budget on its own.
    let make_channel = || {
        let mut channel = MediaChannel::new(10);
        channel.video_sequence_header = Some(Bytes::from_static(VIDEO_SEQ));
        let mut keyframe = vec![0u8; 950 * 1024];
        keyframe[0] = 0x17;
        keyframe[1] = 0x01;
        keyframe[2] = 0xF1;
        channel
            .gops
            .save_frame_data(video_frame(0, Bytes::from(keyframe)), true);
        channel
    };

    // Sanity: with no accept prefix the GOP fits and is replayed.
    let channel = make_channel();
    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(&channel, &mut client, 7, 1, 0, &mut out);
    assert!(
        out.iter().any(|p| packet_contains(p, GOP_MARKER)),
        "the GOP fits the budget when no accept prefix is charged"
    );

    // With a ~64 KiB accept prefix the same GOP no longer fits and is
    // trimmed as a whole, degrading the burst to the sequence header.
    let channel = make_channel();
    let accept_prefix = 64 * 1024;
    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(&channel, &mut client, 7, 1, accept_prefix, &mut out);
    assert!(
        !out.iter().any(|p| packet_contains(p, GOP_MARKER)),
        "the accept prefix must trim the GOP the raw budget would have kept"
    );
    assert!(
        accept_prefix + serialized_burst_len(&out) <= warn,
        "accept prefix + burst must stay within the Warning threshold ({} + {} <= {})",
        accept_prefix,
        serialized_burst_len(&out),
        warn
    );
}

// the join-replay budget subtracts backlog + same-batch
// prefix + accept packets, every add saturating. The same-batch prefix is now
// a pre-accumulated scalar (see the incremental scan test below).
#[test]
fn join_replay_prefix_sums_backlog_prefix_and_accept_bytes() {
    let accept = vec![ServerSessionResult::OutboundResponse(Packet {
        bytes: vec![0u8; 40],
        can_be_dropped: false,
    })];
    // backlog 500 + same-batch prefix 350 + accept 40.
    assert_eq!(join_replay_prefix_bytes(500, 350, &accept), 500 + 350 + 40);
    // No accept packets → backlog + prefix only.
    assert_eq!(join_replay_prefix_bytes(0, 350, &[]), 350);
    // Nothing queued ahead → zero.
    assert_eq!(join_replay_prefix_bytes(0, 0, &[]), 0);
    // Every add saturates: a pathological backlog cannot wrap.
    assert_eq!(
        join_replay_prefix_bytes(usize::MAX, 350, &accept),
        usize::MAX
    );
}

// repeated plays in one batch must fold the same-batch prefix
// INCREMENTALLY (each server_results entry visited once) rather than rescan
// the growing vec per play (quadratic — a reachable reactor stall).
// advance_serving_prefix advances a cursor and a running total.
#[test]
fn serving_prefix_scan_is_incremental_and_targeted() {
    let mut scheduler = RtmpScheduler::new(10);
    let target = 7usize;
    let other = 9usize;
    let outbound = |conn: usize, n: usize| ServerResult::OutboundPacket {
        target_connection_id: conn,
        bytes: Bytes::from(vec![0u8; n]),
        can_be_dropped: false,
        is_keyframe: false,
        is_sequence_header: false,
        is_video: false,
    };
    // First play folds the createStream prefix — only target packets count
    // (100 + 50); the other watcher's 999 is ignored.
    let mut results = vec![
        outbound(target, 100),
        outbound(other, 999),
        outbound(target, 50),
    ];
    assert_eq!(scheduler.advance_serving_prefix(&results, target), 150);
    assert_eq!(
        scheduler.serving_prefix_scan_pos, 3,
        "all three entries consumed exactly once"
    );
    // The play appended its accept+burst to the target; the NEXT play sees it
    // without rescanning the earlier entries.
    results.push(outbound(target, 200));
    assert_eq!(scheduler.advance_serving_prefix(&results, target), 350);
    assert_eq!(
        scheduler.serving_prefix_scan_pos, 4,
        "only the newly-appended entry is consumed"
    );
    // No new entries → unchanged, cursor stable (idempotent).
    assert_eq!(scheduler.advance_serving_prefix(&results, target), 350);
    assert_eq!(scheduler.serving_prefix_scan_pos, 4);
}

// ---- freeze-time eviction of never-replayable frozen GOPs ----
//
// At each freeze (video-keyframe save) the ingest path drops the frozen
// GOPs whose frozen-only wire suffix already exceeds the maximum possible
// join budget (`evict_unreplayable_frozen`). These tests drive the REAL
// ingest path; the direct-save fixtures used by the tests above bypass the
// trigger on purpose, so the join-side trim stays covered independently.

/// Feed `frames` through the real ingest path (freeze-time eviction runs)
/// and return the channel's frozen-GOP count, a fresh joiner's burst and
/// its keyframe gate.
fn ingest_burst(frames: &[(ReceivedDataType, u32, Bytes)]) -> (usize, Vec<ServerResult>, bool) {
    let mut scheduler = RtmpScheduler::new(10);
    assert!(scheduler.new_channel("live".to_string(), 100));
    for (data_type, timestamp, data) in frames {
        feed(&mut scheduler, "live", *data_type, *timestamp, data);
    }
    let channel = scheduler.channels.get("live").unwrap();
    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(channel, &mut client, 7, 1, 0, &mut out);
    (
        channel.gops.frozen_count(),
        out,
        client.has_received_video_keyframe,
    )
}

/// Apply the same `frames` with direct `save_frame_data` calls — the fixture
/// idiom of the tests above, which never runs the freeze-time eviction — and
/// return a fresh joiner's burst plus its keyframe gate. Mirrors the ingest
/// path's header caching; no scenario resends a changed header, so the
/// clear-on-change path has nothing to mirror.
fn direct_burst(frames: &[(ReceivedDataType, u32, Bytes)]) -> (Vec<ServerResult>, bool) {
    let mut channel = MediaChannel::new(10);
    for (data_type, timestamp, data) in frames {
        match data_type {
            ReceivedDataType::Video => {
                if is_video_sequence_header(data) {
                    channel.video_sequence_header = Some(data.clone());
                    channel.video_timestamp = RtmpTimestamp { value: *timestamp };
                }
                channel.gops.save_frame_data(
                    video_frame(*timestamp, data.clone()),
                    is_video_keyframe(data),
                );
            }
            ReceivedDataType::Audio => {
                if is_audio_sequence_header(data) {
                    channel.audio_sequence_header = Some(data.clone());
                    channel.audio_timestamp = RtmpTimestamp { value: *timestamp };
                }
                channel
                    .gops
                    .save_frame_data(audio_frame(*timestamp, data.clone()), false);
            }
        }
    }
    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(&channel, &mut client, 7, 1, 0, &mut out);
    (out, client.has_received_video_keyframe)
}

fn small_keyframe(marker: u8) -> Bytes {
    Bytes::from(vec![0x17u8, 0x01, marker, 0x00, 0x00])
}

fn large_video(first: u8, second: u8, marker: u8, len: usize) -> Bytes {
    let mut data = vec![0u8; len];
    data[0] = first;
    data[1] = second;
    data[2] = marker;
    Bytes::from(data)
}

/// Three ~400 KiB single-keyframe GOPs plus a small fourth keyframe: the
/// 960 KiB budget covers two of them, so the pre-keyframe header GOP and
/// GOP1 fall off the servable suffix at the fourth keyframe's freeze.
fn trim_scenario() -> Vec<(ReceivedDataType, u32, Bytes)> {
    let gop = |marker: u8| large_video(0x17, 0x01, marker, 400 * 1024);
    vec![
        (ReceivedDataType::Video, 0, Bytes::from_static(VIDEO_SEQ)),
        (ReceivedDataType::Audio, 5, Bytes::from_static(AUDIO_SEQ)),
        (ReceivedDataType::Video, 100, gop(1)),
        (ReceivedDataType::Video, 200, gop(2)),
        (ReceivedDataType::Video, 300, gop(3)),
        (ReceivedDataType::Video, 400, small_keyframe(4)),
    ]
}

/// Two keyframes each alone above the budget: the first one's freeze (at
/// the second keyframe) creates a frozen GOP no join could ever replay.
fn oversized_scenario() -> Vec<(ReceivedDataType, u32, Bytes)> {
    let len = JOIN_REPLAY_BUDGET_BYTES + 1;
    vec![
        (ReceivedDataType::Video, 0, Bytes::from_static(VIDEO_SEQ)),
        (ReceivedDataType::Audio, 5, Bytes::from_static(AUDIO_SEQ)),
        (ReceivedDataType::Video, 100, large_video(0x17, 0x01, 1, len)),
        (ReceivedDataType::Video, 200, large_video(0x17, 0x01, 2, len)),
    ]
}

/// Small frozen GOPs well under the budget, then a keyframeless run growing
/// the OPEN GOP past the whole budget: no freeze occurs, so nothing may be
/// evicted.
fn masked_scenario() -> Vec<(ReceivedDataType, u32, Bytes)> {
    vec![
        (ReceivedDataType::Video, 0, Bytes::from_static(VIDEO_SEQ)),
        (ReceivedDataType::Audio, 5, Bytes::from_static(AUDIO_SEQ)),
        (ReceivedDataType::Video, 100, small_keyframe(1)),
        (ReceivedDataType::Video, 200, small_keyframe(2)),
        (
            ReceivedDataType::Video,
            300,
            large_video(0x27, 0x01, 3, JOIN_REPLAY_BUDGET_BYTES + 1),
        ),
    ]
}

/// Three ~400 KiB single-keyframe GOPs, then a ~300 KiB keyframe whose freeze
/// fires the eviction while the open GOP it just started is itself
/// size-significant, then two deltas growing that open GOP without another
/// freeze (0x27 is not a keyframe). At the fourth keyframe's freeze the
/// frozen-only suffix GOP3+GOP2 (~801 KiB wire) fits the 960 KiB budget and
/// adding GOP1 overflows it, so exactly the header GOP and GOP1 are evicted;
/// a predicate that also counted the ~300 KiB open GOP would already overflow
/// at GOP2 and cut one GOP deeper.
fn open_gop_evict_scenario() -> Vec<(ReceivedDataType, u32, Bytes)> {
    let gop = |marker: u8| large_video(0x17, 0x01, marker, 400 * 1024);
    let delta = |marker: u8| large_video(0x27, 0x01, marker, 10 * 1024);
    vec![
        (ReceivedDataType::Video, 0, Bytes::from_static(VIDEO_SEQ)),
        (ReceivedDataType::Audio, 5, Bytes::from_static(AUDIO_SEQ)),
        (ReceivedDataType::Video, 100, gop(1)),
        (ReceivedDataType::Video, 200, gop(2)),
        (ReceivedDataType::Video, 300, gop(3)),
        (
            ReceivedDataType::Video,
            400,
            large_video(0x17, 0x01, 4, 300 * 1024),
        ),
        (ReceivedDataType::Video, 433, delta(5)),
        (ReceivedDataType::Video, 466, delta(6)),
    ]
}

/// Ingest twin of `join_burst_trims_whole_oldest_gops_to_the_byte_budget`:
/// the eviction retains exactly the frozen suffix the join can serve, and
/// the joiner's burst is unchanged.
#[test]
fn ingest_evicts_frozen_gops_beyond_the_join_budget_at_freeze_time() {
    let (frozen_count, out, got_keyframe) = ingest_burst(&trim_scenario());

    // The header GOP (vseq+aseq, frozen by the first keyframe) and GOP1 sit
    // below the servable suffix once GOP3 freezes; GOP2 + GOP3 survive.
    assert_eq!(
        frozen_count, 2,
        "only the join-servable frozen suffix may be retained"
    );
    // vseq + aseq + GOP2 + GOP3 + current: exactly what the untrimmed cache
    // replays.
    assert_eq!(out.len(), 5, "the burst must trim the oldest GOP entirely");
    assert_eq!(
        burst_flags(&out)[..2],
        [(false, true, true), (false, true, false)],
        "sequence headers still open the burst"
    );
    assert!(
        out.iter().all(|p| !packet_contains(p, &[0x17, 0x01, 1])),
        "no fragment of the evicted GOP1 may be replayed"
    );
    assert!(
        packet_contains(&out[2], &[0x17, 0x01, 2]),
        "the replay must start at GOP2's keyframe"
    );
    assert!(packet_contains(&out[3], &[0x17, 0x01, 3]));
    assert!(packet_contains(&out[4], &[0x17, 0x01, 4]));
    assert!(got_keyframe);
}

/// The eviction predicate is computed over the FROZEN GOPs alone, pinned
/// where it matters: an eviction that FIRES while the open GOP is itself
/// size-significant. At the fourth keyframe's freeze the frozen-only suffix
/// keeps GOP2+GOP3; a predicate that also counted the ~300 KiB open GOP
/// would push GOP2 past the budget and evict it too (frozen_count 1, not
/// 2), while a missing eviction would leave all 4. The open GOP must
/// survive untouched and replay verbatim to a joiner.
#[test]
fn eviction_fires_while_the_open_gop_carries_frames() {
    let mut scheduler = RtmpScheduler::new(10);
    assert!(scheduler.new_channel("live".to_string(), 100));
    for (data_type, timestamp, data) in &open_gop_evict_scenario() {
        feed(&mut scheduler, "live", *data_type, *timestamp, data);
    }

    let channel = scheduler.channels.get("live").unwrap();
    // Exactly the unreplayable prefix (header GOP + GOP1) was evicted.
    assert_eq!(
        channel.gops.frozen_count(),
        2,
        "the eviction cut must be decided by the frozen-only suffix"
    );
    // The open GOP is untouched: the firing keyframe plus both deltas.
    assert_eq!(
        channel.gops.current_frames().len(),
        3,
        "eviction must never touch the open GOP"
    );

    let mut client = make_watching_client(7, "live", 1);
    let mut out = Vec::new();
    build_join_burst(channel, &mut client, 7, 1, 0, &mut out);

    // vseq + aseq + GOP3 + the whole open GOP. GOP2 stays cached but is
    // masked from THIS join: the join-time size list appends the open GOP,
    // and ~320 KiB open + GOP3 pushes GOP2 past the budget — the documented
    // conservative gap between the eviction and the join-time trim.
    assert_eq!(out.len(), 6, "headers + GOP3 + open GOP (kf4, d5, d6)");
    assert_eq!(
        burst_flags(&out)[..2],
        [(false, true, true), (false, true, false)],
        "sequence headers still open the burst"
    );
    assert!(
        out.iter().all(|p| !packet_contains(p, &[0x17, 0x01, 1])),
        "no fragment of the evicted GOP1 may be replayed"
    );
    assert!(
        out.iter().all(|p| !packet_contains(p, &[0x17, 0x01, 2])),
        "the retained GOP2 is masked from this join, not replayed"
    );
    assert!(
        packet_contains(&out[2], &[0x17, 0x01, 3]),
        "the replay must start at GOP3's keyframe"
    );
    assert!(
        packet_contains(&out[3], &[0x17, 0x01, 4]),
        "the open GOP's keyframe must be replayed"
    );
    assert!(packet_contains(&out[4], &[0x27, 0x01, 5]));
    assert!(packet_contains(&out[5], &[0x27, 0x01, 6]));
    assert!(client.has_received_video_keyframe);
}

/// The core equivalence claim: for the same frame sequence, a joiner of the
/// evicting ingest path and a joiner of the untrimmed direct-save cache
/// receive byte-identical bursts — the eviction is wire-invisible.
#[test]
fn freeze_time_eviction_keeps_join_bursts_byte_identical() {
    for (name, frames) in [
        ("budget trim", trim_scenario()),
        ("single oversized GOP", oversized_scenario()),
        ("masked by the open GOP", masked_scenario()),
        (
            "evicting freeze with a populated open GOP",
            open_gop_evict_scenario(),
        ),
    ] {
        let (_, ingest_out, ingest_gate) = ingest_burst(&frames);
        let (direct_out, direct_gate) = direct_burst(&frames);
        assert_eq!(
            ingest_out.len(),
            direct_out.len(),
            "{name}: burst packet counts diverge"
        );
        assert_eq!(
            burst_flags(&ingest_out),
            burst_flags(&direct_out),
            "{name}: burst packet flags diverge"
        );
        for (i, (ingest_packet, direct_packet)) in
            ingest_out.iter().zip(direct_out.iter()).enumerate()
        {
            if let (
                ServerResult::OutboundPacket { bytes: a, .. },
                ServerResult::OutboundPacket { bytes: b, .. },
            ) = (ingest_packet, direct_packet)
            {
                assert_same_bytes(a, b, &format!("{name}: burst packet {i}"));
            }
        }
        assert_eq!(ingest_gate, direct_gate, "{name}: keyframe gate diverges");
    }
}

/// Pins the open-GOP exclusion: the open GOP's size is non-monotone (the
/// current-GOP caps clear it on overflow), so an oversized open GOP masks
/// the frozen history from THIS join but must never evict it — once the
/// open GOP resets, the frozen GOPs become replayable again.
#[test]
fn an_over_budget_open_gop_masks_but_never_evicts_frozen_history() {
    let (frozen_count, out, _) = ingest_burst(&masked_scenario());

    assert_eq!(
        frozen_count, 2,
        "an oversized open GOP must not evict under-budget frozen history"
    );
    assert_eq!(
        burst_flags(&out),
        vec![(false, true, true), (false, true, false)],
        "this join is masked down to the sequence headers"
    );
    assert!(
        out.iter().all(|p| !packet_contains(p, &[0x17, 0x01, 1])),
        "the retained GOPs are masked, not replayed"
    );
}

/// A frozen GOP whose wire size alone exceeds the budget is evicted at its
/// own freeze; the joiner burst degrades to sequence headers exactly as
/// `join_burst_with_only_oversized_gops_sends_headers_only` shows for the
/// untrimmed cache.
#[test]
fn a_frozen_gop_over_the_budget_alone_is_evicted_at_its_own_freeze() {
    let (frozen_count, out, got_keyframe) = ingest_burst(&oversized_scenario());

    assert_eq!(
        frozen_count, 0,
        "a never-replayable frozen GOP must not outlive its own freeze"
    );
    assert_eq!(
        burst_flags(&out),
        vec![(false, true, true), (false, true, false)],
        "an oversized cache still degrades to sequence headers only"
    );
    assert!(
        !got_keyframe,
        "no keyframe was replayed, so live deltas stay gated"
    );
}