open-standards/wai/wai-rs/tests/carousel.rs

//! `wai.bearer.carousel` (extensions/carousel.md): the committed corpus `live-conformance/carousel/`
//! regenerates byte for byte from its fixed seeds, holds no other file, every case in it gets the
//! outcome it states, §9 lists exactly the reasons the reference reports, jwp-receipts §3.1 lists
//! the class codes at their steps, and Appendix A is the reference's table of K′. Against an
//! independent implementation of RFC 6330: the table of K′ and the partition agree for every block
//! size, and each RaptorQ object of the corpus that a case holds available at a pass decodes from
//! what a receiver within its loss bound holds by that pass's end. Against an independent
//! implementation of RFC 5053, each Raptor object does the same.
#![cfg(feature = "carousel")]

use std::collections::BTreeMap;
use std::path::PathBuf;

use raptorq::{Decoder, Encoder, EncodingPacket, ObjectTransmissionInformation};
use serde_json::Value;
use wai::carousel::{corpus, k_prime, missed, CODES, K_PRIME};
use wai::claim_receipt::schedule::{partition, K_PRIME_MAX};
use wai::claim_receipt::{Esi, Oti, ScheduleEntry};

fn wai_dir() -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("..")
}

fn committed() -> BTreeMap<String, String> {
    let dir = wai_dir().join("live-conformance/carousel");
    std::fs::read_dir(&dir)
        .expect("the committed corpus")
        .flatten()
        .map(|e| (e.file_name().to_string_lossy().into_owned(), e.path()))
        .filter(|(n, _)| n != "README.md")
        .map(|(n, p)| (n, std::fs::read_to_string(p).expect("a text file")))
        .collect()
}

fn extension() -> String {
    std::fs::read_to_string(wai_dir().join("extensions/carousel.md")).expect("the extension")
}

#[test]
fn the_committed_corpus_regenerates_from_its_seeds() {
    let files = committed();
    let generated: BTreeMap<String, String> = corpus::generate().into_iter().collect();
    let extra: Vec<&String> = files.keys().filter(|k| !generated.contains_key(*k)).collect();
    let missing: Vec<&String> = generated.keys().filter(|k| !files.contains_key(*k)).collect();
    let differ: Vec<&String> = generated.iter().filter(|(k, v)| files.get(*k).is_some_and(|f| f != *v)).map(|(k, _)| k).collect();
    assert!(extra.is_empty() && missing.is_empty() && differ.is_empty(), "extra {extra:?}, missing {missing:?}, differ {differ:?}: run `wai_carousel_vectors gen`");
}

#[test]
fn every_case_gets_its_stated_outcome() {
    let fails = corpus::verify(&committed());
    assert!(fails.is_empty(), "{}", fails.join("\n"));
}

/// §9 of the extension lists exactly the codes `CODES` has, each once, in order.
#[test]
fn the_extension_lists_the_reasons() {
    let text = extension();
    let at = text.find("\n## 9. Reasons").expect("a §9");
    let end = text[at + 1..].find("\n## ").map_or(text.len(), |e| at + 1 + e);
    let listed: Vec<String> =
        text[at..end].lines().filter(|l| l.starts_with("| `")).filter_map(|l| l.split('`').nth(1).map(String::from)).collect();
    let want: Vec<String> = CODES.iter().map(|c| c.to_string()).collect();
    assert_eq!(listed, want, "§9's table and CODES differ");
}

/// jwp-receipts §3.1's table lists the class's codes: its rules at step 4, its checks against the
/// group and its objects at step 6.
#[test]
fn jwp_receipts_lists_the_class_codes() {
    let text = std::fs::read_to_string(wai_dir().join("extensions/jwp-receipts.md")).expect("jwp-receipts");
    for code in CODES.iter().filter(|c| c.starts_with("schedule_") && **c != "schedule_contradicted" && **c != "schedule_on_bidirectional") {
        let step = if ["schedule_not_scheduled", "schedule_objects_malformed", "schedule_oti_invalid", "schedule_symbols_invalid", "schedule_passes_invalid"].contains(code) {
            4
        } else {
            6
        };
        assert!(text.contains(&format!("| {step} | `{code}` |")), "jwp-receipts §3.1 does not list {code} at step {step}");
    }
}

/// Appendix A holds exactly `K_PRIME`, in order.
#[test]
fn appendix_a_is_the_table_of_k_prime() {
    let text = extension();
    let a = text.find("<!-- k-prime:begin -->").expect("the table's start");
    let b = text.find("<!-- k-prime:end -->").expect("the table's end");
    let listed: Vec<u32> = text[a..b].split(|c: char| !c.is_ascii_digit()).filter(|s| !s.is_empty()).map(|s| s.parse().expect("a number")).collect();
    assert_eq!(listed, K_PRIME.to_vec());
}

/// The table and the partition agree with an independent implementation of RFC 6330 for every
/// block size it admits.
#[test]
fn k_prime_and_partition_agree_with_an_independent_rfc_6330() {
    for k in 1..=K_PRIME_MAX as u32 {
        assert_eq!(k_prime(u64::from(k)), Some(u64::from(raptorq::extended_source_block_symbols(k))), "K' of {k}");
    }
    for (i, j) in [(50, 3), (14, 1), (5, 1), (56_403 * 255, 255), (16, 2), (15, 2), (1, 1)] {
        let (il, is, jl, js) = partition(i, j);
        assert_eq!((il as u32, is as u32, jl as u32, js as u32), raptorq::partition(i as u32, j as u32), "Partition[{i}, {j}]");
    }
}

fn hex(s: &str) -> Vec<u8> {
    (0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).expect("hex")).collect()
}

/// What a receiver within `loss_ppm` holds of `e`'s object by the end of pass `pass`, fed to an
/// independent RFC 6330 decoder until it decodes: in each pass, of each block's m symbols, the
/// first ⌊L · m / 10^6⌋ in ESI order are lost (in the first pass, source symbols, so the
/// decoder must decode). Under `repeat` a later pass is taken to lose the same symbols again,
/// so only the first is fed.
fn receive(e: &ScheduleEntry, data: &[u8], loss_ppm: u64, pass: u64) -> Option<Vec<u8>> {
    let Oti::RaptorQ { transfer_length, symbol_size, source_blocks, sub_blocks, alignment } = e.oti else { return None };
    let config = ObjectTransmissionInformation::new(transfer_length, symbol_size as u16, source_blocks as u8, sub_blocks as u16, alignment as u8);
    let encoder = Encoder::new(data, config);
    let mut decoder = Decoder::new(config);
    let passes = if e.esi == Esi::Advance { pass } else { 1 };
    for block in encoder.get_block_encoders() {
        let source = block.source_packets();
        let k = source.len() as u64;
        let m = k + e.repair;
        let x = missed(loss_ppm, m) as usize;
        for j in 1..=passes {
            let start = if e.esi == Esi::Advance { (j - 1) * m } else { 0 };
            let mut sent: Vec<EncodingPacket> = (start..(start + m).min(k)).map(|esi| source[esi as usize].clone()).collect();
            let repair_from = start.max(k);
            if start + m > repair_from {
                sent.extend(block.repair_packets((repair_from - k) as u32, (start + m - repair_from) as u32));
            }
            assert_eq!(sent.len() as u64, m);
            for p in sent.into_iter().skip(x) {
                if let Some(out) = decoder.decode(p) {
                    return Some(out);
                }
            }
        }
    }
    None
}

/// [`receive`] under an independent RFC 5053 decoder, block by block (RFC 5053 §5.3.1.2): `None`
/// where a block does not decode from what the receiver holds.
fn receive_raptor(e: &ScheduleEntry, data: &[u8], loss_ppm: u64, pass: u64) -> Option<Vec<u8>> {
    let Oti::Raptor { transfer_length, symbol_size: t, source_blocks: z, .. } = e.oti else { return None };
    let kt = transfer_length.div_ceil(t);
    let (kl, ks, zl, zs) = partition(kt, z);
    let mut padded = data.to_vec();
    padded.resize((kt * t) as usize, 0);
    let advance = e.esi == Esi::Advance;
    let mut out = Vec::with_capacity(padded.len());
    let mut at = 0usize;
    for k in std::iter::repeat_n(kl, zl as usize).chain(std::iter::repeat_n(ks, zs as usize)) {
        let length = (k * t) as usize;
        let block = &padded[at..at + length];
        at += length;
        let mut encoder = raptor_code::SourceBlockEncoder::new(block, k as usize).expect("a block RFC 5053 admits");
        assert_eq!(u64::from(encoder.nb_source_symbols()), k);
        let mut decoder = raptor_code::SourceBlockDecoder::new(k as usize);
        let m = k + e.repair;
        let x = missed(loss_ppm, m);
        for j in 1..=(if advance { pass } else { 1 }) {
            let start = if advance { (j - 1) * m } else { 0 };
            for esi in start + x..start + m {
                decoder.push_encoding_symbol(&encoder.fountain(esi as u32), esi as u32);
            }
        }
        if !decoder.fully_specified() {
            return None;
        }
        out.extend(decoder.decode(length)?);
    }
    out.truncate(transfer_length as usize);
    Some(out)
}

/// Of `trials` sets of `k` + `margin` distinct encoding symbols of a block of `k`, chosen uniformly
/// at random from a fixed seed, how many an independent RFC 5053 decoder cannot solve.
fn raptor_failures(k: usize, margin: usize, trials: usize, seed: u64) -> usize {
    let block: Vec<u8> = (0..k * 4).map(|i| (i * 131 + 7) as u8).collect();
    let mut encoder = raptor_code::SourceBlockEncoder::new(&block, k).expect("a block RFC 5053 admits");
    let n = k + 3 * margin + 40;
    let symbols: Vec<Vec<u8>> = (0..n as u32).map(|esi| encoder.fountain(esi)).collect();
    let mut state = seed;
    let mut next = move || {
        state ^= state << 13;
        state ^= state >> 7;
        state ^= state << 17;
        state as usize
    };
    (0..trials)
        .filter(|_| {
            let mut esis: Vec<usize> = (0..n).collect();
            for i in 0..k + margin {
                let j = i + next() % (n - i);
                esis.swap(i, j);
            }
            let mut decoder = raptor_code::SourceBlockDecoder::new(k);
            for &esi in &esis[..k + margin] {
                decoder.push_encoding_symbol(&symbols[esi], esi as u32);
            }
            !decoder.fully_specified()
        })
        .count()
}

/// Carousel §5.3 holds a Raptor block at K and a margin of this text's, RFC 5053 stating no bound.
/// An independent decoder, on a fixed sample, fails on many blocks at K + 2 and on few at each
/// margin. This is a measurement of that sample, as §5.3 says, and no bound.
#[test]
fn raptor_margins_keep_decoding_failures_rare() {
    use wai::carousel::RAPTOR_MARGIN;
    for k in [16, 64] {
        let seed = 0x9E37_79B9_7F4A_7C15 ^ k as u64;
        let at_two = raptor_failures(k, 2, 400, seed);
        assert!(at_two >= 40, "K = {k}: {at_two} of 400 at K + 2");
        let first = raptor_failures(k, RAPTOR_MARGIN[0] as usize, 2_000, seed);
        assert!(first <= 20, "K = {k}: {first} of 2,000 at K + {}", RAPTOR_MARGIN[0]);
        let second = raptor_failures(k, RAPTOR_MARGIN[1] as usize, 10_000, seed);
        assert!(second <= 1, "K = {k}: {second} of 10,000 at K + {}", RAPTOR_MARGIN[1]);
        let third = raptor_failures(k, RAPTOR_MARGIN[2] as usize, 10_000, seed);
        assert_eq!(third, 0, "K = {k}: of 10,000 at K + {}", RAPTOR_MARGIN[2]);
    }
}

/// The measurement carousel §5.3 quotes, by block size and margin, from fixed seeds. It takes some
/// minutes in a release build: `cargo test --release --test carousel -- --ignored --nocapture`.
#[test]
#[ignore = "a measurement of some minutes, run by hand in a release build; carousel §5.3 quotes it"]
fn raptor_margin_measurement() {
    let sample: &[(usize, &[(usize, usize)])] = &[
        (2, &[(4, 6_000), (16, 6_000), (64, 6_000), (1_024, 600)]),
        (8, &[(4, 1_200_000), (8, 1_200_000), (16, 1_200_000), (64, 1_200_000), (256, 240_000), (1_024, 42_000), (2_048, 18_000), (8_192, 1_800)]),
        (16, &[(4, 1_800_000), (8, 1_800_000), (16, 1_800_000), (64, 1_800_000), (256, 360_000), (1_024, 60_000), (2_048, 18_000), (8_192, 1_800)]),
        (28, &[(4, 6_000_000), (8, 6_000_000), (16, 6_000_000), (32, 6_000_000), (64, 6_000_000), (128, 3_000_000), (256, 1_200_000), (1_024, 120_000), (2_048, 24_000), (8_192, 1_800)]),
    ];
    for (margin, sizes) in sample {
        for (k, trials) in *sizes {
            let (k, margin, trials) = (*k, *margin, *trials);
            let failures: usize = std::thread::scope(|s| {
                let runs: Vec<_> = (0..6u64)
                    .map(|t| s.spawn(move || raptor_failures(k, margin, trials / 6, 0x9E37_79B9_7F4A_7C15 ^ (((k as u64) << 24) | ((margin as u64) << 8) | (t + 1)))))
                    .collect();
                runs.into_iter().map(|r| r.join().expect("a run")).sum()
            });
            println!("K {k}, K + {margin}: {failures} of {trials}");
        }
    }
}

#[test]
fn each_raptor_object_decodes_from_what_a_receiver_within_its_bound_holds() {
    let files = committed();
    let doc: Value = serde_json::from_str(&files["vectors.json"]).expect("the corpus");
    let mut decoded = 0;
    for c in doc["verdicts"].as_array().expect("verdict cases") {
        let Some(lines) = c["outcome"]["lines"].as_array() else { continue };
        let loss = c["loss_ppm"].as_u64().expect("a loss bound");
        let schedules: Vec<Value> = c["claims"]
            .as_array()
            .expect("claims")
            .iter()
            .map(|w| serde_json::from_str::<Value>(w.as_str().expect("a claim")).expect("json"))
            .filter(|v| v["class"] == "schedule")
            .collect();
        for line in lines {
            let Some(pass) = line["pass"].as_u64() else { continue };
            let leaf = line["leaf_index"].as_u64().expect("a leaf") as usize;
            let Some(data) = c["objects"][leaf]["envelope"].as_str().map(hex) else { continue };
            let entry = schedules
                .iter()
                .filter(|s| s["body"]["bearer_id"] == line["bearer_id"])
                .flat_map(|s| s["body"]["objects"].as_array().expect("entries").iter())
                .rfind(|o| o["leaf_index"] == line["leaf_index"])
                .map(|o| ScheduleEntry::from_value(o).expect("an entry that stands"))
                .expect("the entry of a line");
            if !matches!(entry.oti, Oti::Raptor { .. }) {
                continue;
            }
            let got = receive_raptor(&entry, &data, loss, pass);
            assert_eq!(got.as_deref(), Some(&data[..]), "{} object {leaf} on {} by pass {pass}", c["name"], line["bearer_id"]);
            decoded += 1;
        }
    }
    assert!(decoded >= 10, "only {decoded} objects decoded");
}

#[test]
fn each_raptorq_object_decodes_from_what_a_receiver_within_its_bound_holds() {
    let files = committed();
    let doc: Value = serde_json::from_str(&files["vectors.json"]).expect("the corpus");
    let mut decoded = 0;
    for c in doc["verdicts"].as_array().expect("verdict cases") {
        let Some(lines) = c["outcome"]["lines"].as_array() else { continue };
        let loss = c["loss_ppm"].as_u64().expect("a loss bound");
        let schedules: Vec<Value> = c["claims"]
            .as_array()
            .expect("claims")
            .iter()
            .map(|w| serde_json::from_str::<Value>(w.as_str().expect("a claim")).expect("json"))
            .filter(|v| v["class"] == "schedule")
            .collect();
        for line in lines {
            let Some(pass) = line["pass"].as_u64() else { continue };
            let leaf = line["leaf_index"].as_u64().expect("a leaf") as usize;
            let Some(data) = c["objects"][leaf]["envelope"].as_str().map(hex) else { continue };
            let entry = schedules
                .iter()
                .filter(|s| s["body"]["bearer_id"] == line["bearer_id"])
                .flat_map(|s| s["body"]["objects"].as_array().expect("entries").iter())
                // The schedule that stands is the last of its instance (§6.2 step 4).
                .rfind(|o| o["leaf_index"] == line["leaf_index"])
                .map(|o| ScheduleEntry::from_value(o).expect("an entry that stands"))
                .expect("the entry of a line");
            if !matches!(entry.oti, Oti::RaptorQ { .. }) {
                continue;
            }
            let got = receive(&entry, &data, loss, pass);
            assert_eq!(got.as_deref(), Some(&data[..]), "{} object {leaf} on {} by pass {pass}", c["name"], line["bearer_id"]);
            decoded += 1;
        }
    }
    assert!(decoded >= 10, "only {decoded} objects decoded");
}