//! `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 { 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 = 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 = text[at..end].lines().filter(|l| l.starts_with("| `")).filter_map(|l| l.split('`').nth(1).map(String::from)).collect(); let want: Vec = 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("").expect("the table's start"); let b = text.find("").expect("the table's end"); let listed: Vec = 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 { (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> { 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 = (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> { 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 = (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> = (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 = (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 = c["claims"] .as_array() .expect("claims") .iter() .map(|w| serde_json::from_str::(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 = c["claims"] .as_array() .expect("claims") .iter() .map(|w| serde_json::from_str::(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"); }