20260605 移除audit rule index和DB trace

src/tools/sequence_triage_ccr_cir.rs

@@ -7,11 +7,11 @@ mod output;
 mod sandwich;
 
 use args::{Args, parse_args};
-use churn::build_intra_rp_churn;
-use loader::load_sequence;
+use churn::build_intra_rp_churn_streaming;
+use loader::load_sequence_meta;
 use model::Side;
-use output::{build_output, write_json, write_markdown};
-use sandwich::build_sandwich_analysis;
+use output::{build_output_from_meta, write_json, write_markdown};
+use sandwich::build_sandwich_analysis_streaming;
 
 pub fn main_entry() -> Result<(), String> {
     real_main()
@@ -25,15 +25,15 @@ fn real_main() -> Result<(), String> {
 fn run(args: Args) -> Result<(), String> {
     std::fs::create_dir_all(&args.out_dir)
         .map_err(|e| format!("create out-dir failed: {}: {e}", args.out_dir.display()))?;
-    let left = load_sequence(&args.left_sequence, Side::Left)?;
-    let right = load_sequence(&args.right_sequence, Side::Right)?;
+    let left = load_sequence_meta(&args.left_sequence, Side::Left)?;
+    let right = load_sequence_meta(&args.right_sequence, Side::Right)?;
     if left.is_empty() || right.is_empty() {
         return Err("left and right sequences must both contain at least one sample".into());
     }
 
-    let sandwich = build_sandwich_analysis(&args, &left, &right);
-    let churn = build_intra_rp_churn(&left, &right);
-    let output = build_output(&args, &left, &right, &sandwich, &churn);
+    let sandwich = build_sandwich_analysis_streaming(&args, &left, &right)?;
+    let churn = build_intra_rp_churn_streaming(&left, &right)?;
+    let output = build_output_from_meta(&args, &left, &right, &sandwich, &churn);
     write_json(&args.out_dir.join("sequence-triage.json"), &output)?;
     write_markdown(&args.out_dir.join("sequence-triage.md"), &output)?;
     println!("{}", args.out_dir.display());

src/tools/sequence_triage_ccr_cir/churn.rs

@@ -1,71 +1,82 @@
 use std::collections::{BTreeMap, BTreeSet};
 
-use super::model::{ChurnRecord, ChurnSummaryRecord, IntraRpChurn, SequenceSample, Side};
+use super::loader::{load_sample_ccr_from_meta, load_sample_cir_from_meta};
+use super::model::{
+    ChurnRecord, ChurnSummaryRecord, IntraRpChurn, SequenceMeta, SequenceSample, Side,
+};
 
-pub(super) fn build_intra_rp_churn(
-    left: &[SequenceSample],
-    right: &[SequenceSample],
-) -> IntraRpChurn {
-    let left_records = build_side_records(Side::Left, left);
-    let right_records = build_side_records(Side::Right, right);
+pub(super) fn build_intra_rp_churn_streaming(
+    left: &[SequenceMeta],
+    right: &[SequenceMeta],
+) -> Result<IntraRpChurn, String> {
+    let left_records = build_side_records_streaming(Side::Left, left)?;
+    let right_records = build_side_records_streaming(Side::Right, right)?;
     let mut summary = summarize_records(&left_records);
     summary.extend(summarize_records(&right_records));
-    IntraRpChurn {
+    Ok(IntraRpChurn {
         left: left_records,
         right: right_records,
         summary,
-    }
+    })
 }
 
-fn build_side_records(side: Side, samples: &[SequenceSample]) -> Vec<ChurnRecord> {
+fn build_side_records_streaming(
+    side: Side,
+    samples: &[SequenceMeta],
+) -> Result<Vec<ChurnRecord>, String> {
     let mut records = Vec::new();
     for pair in samples.windows(2) {
-        let from = &pair[0];
-        let to = &pair[1];
+        let from_cir = load_sample_cir_from_meta(&pair[0])?;
+        let to_cir = load_sample_cir_from_meta(&pair[1])?;
         records.push(record_churn(
             side,
-            from,
-            to,
+            &from_cir,
+            &to_cir,
             "object",
-            &from.object_hashes,
-            &to.object_hashes,
+            &from_cir.object_hashes,
+            &to_cir.object_hashes,
         ));
-        let from_output = output_keys(from);
-        let to_output = output_keys(to);
         records.push(record_churn(
             side,
-            from,
-            to,
+            &from_cir,
+            &to_cir,
+            "reject",
+            &from_cir.rejects,
+            &to_cir.rejects,
+        ));
+        drop(from_cir);
+        drop(to_cir);
+
+        let from_ccr = load_sample_ccr_from_meta(&pair[0])?;
+        let to_ccr = load_sample_ccr_from_meta(&pair[1])?;
+        let from_output = output_keys(&from_ccr);
+        let to_output = output_keys(&to_ccr);
+        records.push(record_churn(
+            side,
+            &from_ccr,
+            &to_ccr,
             "output",
             &from_output,
             &to_output,
         ));
         records.push(record_churn(
             side,
-            from,
-            to,
+            &from_ccr,
+            &to_ccr,
             "vrp_output",
-            &from.vrps,
-            &to.vrps,
+            &from_ccr.vrps,
+            &to_ccr.vrps,
         ));
         records.push(record_churn(
             side,
-            from,
-            to,
+            &from_ccr,
+            &to_ccr,
             "vap_output",
-            &from.vaps,
-            &to.vaps,
-        ));
-        records.push(record_churn(
-            side,
-            from,
-            to,
-            "reject",
-            &from.rejects,
-            &to.rejects,
+            &from_ccr.vaps,
+            &to_ccr.vaps,
         ));
     }
-    records
+    Ok(records)
 }
 
 fn output_keys(sample: &SequenceSample) -> BTreeSet<String> {

src/tools/sequence_triage_ccr_cir/loader.rs

@@ -5,9 +5,9 @@ use crate::ccr::{decode_ccr_compare_views, decode_content_info};
 use crate::cir::decode_cir;
 
 use super::io::{object_hash_key, parse_rfc3339, read_file, resolve_path};
-use super::model::{SequenceItemRaw, SequenceSample, Side};
+use super::model::{SequenceItemRaw, SequenceMeta, SequenceSample, Side};
 
-pub(super) fn load_sequence(path: &Path, side: Side) -> Result<Vec<SequenceSample>, String> {
+pub(super) fn load_sequence_meta(path: &Path, side: Side) -> Result<Vec<SequenceMeta>, String> {
     let base_dir = path.parent().unwrap_or_else(|| Path::new("."));
     let text = std::fs::read_to_string(path)
         .map_err(|e| format!("read sequence failed: {}: {e}", path.display()))?;
@@ -25,101 +25,34 @@ pub(super) fn load_sequence(path: &Path, side: Side) -> Result<Vec<SequenceSampl
                 line_index + 1
             )
         })?;
-        if raw.schema_version.unwrap_or(1) != 1 {
-            return Err(format!(
-                "unsupported sequence item schemaVersion in {}:{}",
-                path.display(),
-                line_index + 1
-            ));
-        }
-        if !seen_seq.insert(raw.seq) {
-            return Err(format!("duplicate seq {} in {}", raw.seq, path.display()));
-        }
-        if let Some(status) = &raw.status
-            && status != "success"
-        {
-            return Err(format!(
-                "sequence sample {} has non-success status: {status}",
-                raw.run_id
-            ));
-        }
+        validate_raw(path, line_index, &raw, side, &mut seen_seq)?;
         let cir_path = resolve_path(base_dir, &raw.cir_path);
         let ccr_path = resolve_path(base_dir, &raw.ccr_path);
         let cir = decode_cir(&read_file(&cir_path)?).map_err(|e| {
             format!(
-                "decode CIR failed for sample {} ({}): {e}",
+                "decode CIR metadata failed for sample {} ({}): {e}",
                 raw.run_id,
                 cir_path.display()
             )
         })?;
-        let ccr = decode_content_info(&read_file(&ccr_path)?).map_err(|e| {
-            format!(
-                "decode CCR failed for sample {} ({}): {e}",
-                raw.run_id,
-                ccr_path.display()
-            )
-        })?;
         let _sequence_validation_time = raw
             .validation_time
             .as_deref()
             .map(parse_rfc3339)
             .transpose()?;
         let validation_time = cir.validation_time;
-        let objects = cir
-            .objects
-            .iter()
-            .map(|item| (item.rsync_uri.clone(), hex::encode(&item.sha256)))
-            .collect::<BTreeMap<_, _>>();
-        let object_uris = objects.keys().cloned().collect::<BTreeSet<_>>();
-        let object_hashes = objects
-            .iter()
-            .map(|(uri, hash)| object_hash_key(uri, hash))
-            .collect::<BTreeSet<_>>();
-        let rejects = cir
-            .rejected_objects
-            .iter()
-            .map(|item| item.object_uri.clone())
-            .collect::<BTreeSet<_>>();
-        let trust_anchors = cir
-            .trust_anchors
-            .iter()
-            .map(|item| {
-                format!(
-                    "{}|{}|{}|{}",
-                    item.ta_rsync_uri,
-                    item.tal_uri,
-                    hex::encode(crate::cir::sha256(&item.tal_bytes)),
-                    hex::encode(&item.ta_certificate_sha256)
-                )
-            })
-            .collect::<BTreeSet<_>>();
-        let (vrps, vaps) = decode_ccr_compare_views(&ccr).map_err(|e| {
-            format!(
-                "decode CCR compare views failed for sample {} ({}): {e}",
-                raw.run_id,
-                ccr_path.display()
-            )
-        })?;
-        let vrps = vrps
-            .into_iter()
-            .map(|row| format!("{}|{}|{}", row.asn, row.ip_prefix, row.max_length))
-            .collect::<BTreeSet<_>>();
-        let vaps = vaps
-            .into_iter()
-            .map(|row| format!("{}|{}", row.customer_asn, row.providers))
-            .collect::<BTreeSet<_>>();
-        samples.push(SequenceSample {
+        samples.push(SequenceMeta {
+            cir_object_count: raw.cir_object_count.or(Some(cir.objects.len() as u64)),
+            cir_reject_count: raw
+                .cir_reject_count
+                .or(Some(cir.rejected_objects.len() as u64)),
+            cir_trust_anchor_count: raw
+                .cir_trust_anchor_count
+                .or(Some(cir.trust_anchors.len() as u64)),
             raw,
             validation_time,
             ccr_path,
             cir_path,
-            objects,
-            object_uris,
-            object_hashes,
-            rejects,
-            trust_anchors,
-            vrps,
-            vaps,
         });
     }
     samples.sort_by(|left, right| {
@@ -128,17 +61,120 @@ pub(super) fn load_sequence(path: &Path, side: Side) -> Result<Vec<SequenceSampl
             .then_with(|| left.raw.seq.cmp(&right.raw.seq))
             .then_with(|| left.raw.run_id.cmp(&right.raw.run_id))
     });
-    if samples.iter().any(|sample| {
-        sample
-            .raw
-            .side
-            .as_deref()
-            .is_some_and(|item| item != side.as_str())
-    }) {
-        return Err(format!(
-            "sequence side field does not match expected side: {}",
-            side.as_str()
-        ));
-    }
     Ok(samples)
 }
+
+pub(super) fn load_sample_cir_from_meta(meta: &SequenceMeta) -> Result<SequenceSample, String> {
+    load_sample_parts(meta, true, false)
+}
+
+pub(super) fn load_sample_ccr_from_meta(meta: &SequenceMeta) -> Result<SequenceSample, String> {
+    load_sample_parts(meta, false, true)
+}
+
+fn load_sample_parts(
+    meta: &SequenceMeta,
+    include_cir: bool,
+    include_ccr: bool,
+) -> Result<SequenceSample, String> {
+    let mut objects = BTreeMap::new();
+    let mut object_hashes = BTreeSet::new();
+    let mut rejects = BTreeSet::new();
+    if include_cir {
+        let cir = decode_cir(&read_file(&meta.cir_path)?).map_err(|e| {
+            format!(
+                "decode CIR failed for sample {} ({}): {e}",
+                meta.raw.run_id,
+                meta.cir_path.display()
+            )
+        })?;
+        objects = cir
+            .objects
+            .iter()
+            .map(|item| (item.rsync_uri.clone(), hex::encode(&item.sha256)))
+            .collect::<BTreeMap<_, _>>();
+        object_hashes = objects
+            .iter()
+            .map(|(uri, hash)| object_hash_key(uri, hash))
+            .collect::<BTreeSet<_>>();
+        rejects = cir
+            .rejected_objects
+            .iter()
+            .map(|item| item.object_uri.clone())
+            .collect::<BTreeSet<_>>();
+    }
+    let mut vrps = BTreeSet::new();
+    let mut vaps = BTreeSet::new();
+    if include_ccr {
+        let ccr = decode_content_info(&read_file(&meta.ccr_path)?).map_err(|e| {
+            format!(
+                "decode CCR failed for sample {} ({}): {e}",
+                meta.raw.run_id,
+                meta.ccr_path.display()
+            )
+        })?;
+        let (decoded_vrps, decoded_vaps) = decode_ccr_compare_views(&ccr).map_err(|e| {
+            format!(
+                "decode CCR compare views failed for sample {} ({}): {e}",
+                meta.raw.run_id,
+                meta.ccr_path.display()
+            )
+        })?;
+        vrps = decoded_vrps
+            .into_iter()
+            .map(|row| format!("{}|{}|{}", row.asn, row.ip_prefix, row.max_length))
+            .collect::<BTreeSet<_>>();
+        vaps = decoded_vaps
+            .into_iter()
+            .map(|row| format!("{}|{}", row.customer_asn, row.providers))
+            .collect::<BTreeSet<_>>();
+    }
+    Ok(SequenceSample {
+        raw: meta.raw.clone(),
+        validation_time: meta.validation_time,
+        objects,
+        object_hashes,
+        rejects,
+        vrps,
+        vaps,
+    })
+}
+
+fn validate_raw(
+    path: &Path,
+    line_index: usize,
+    raw: &SequenceItemRaw,
+    side: Side,
+    seen_seq: &mut BTreeSet<u32>,
+) -> Result<(), String> {
+    if raw.schema_version.unwrap_or(1) != 1 {
+        return Err(format!(
+            "unsupported sequence item schemaVersion in {}:{}",
+            path.display(),
+            line_index + 1
+        ));
+    }
+    if !seen_seq.insert(raw.seq) {
+        return Err(format!("duplicate seq {} in {}", raw.seq, path.display()));
+    }
+    if let Some(status) = &raw.status
+        && status != "success"
+    {
+        return Err(format!(
+            "sequence sample {} has non-success status: {status}",
+            raw.run_id
+        ));
+    }
+    if raw
+        .side
+        .as_deref()
+        .is_some_and(|item| item != side.as_str())
+    {
+        return Err(format!(
+            "sequence side field does not match expected side in {}:{}",
+            path.display(),
+            line_index + 1
+        ));
+    }
+    Ok(())
+}

src/tools/sequence_triage_ccr_cir/model.rs

@@ -25,23 +25,34 @@ pub(super) struct SequenceItemRaw {
     pub(super) max_rss_kb: Option<u64>,
     pub(super) vrps: Option<u64>,
     pub(super) vaps: Option<u64>,
+    pub(super) publication_points: Option<u64>,
+    pub(super) cir_object_count: Option<u64>,
+    pub(super) cir_reject_count: Option<u64>,
+    pub(super) cir_trust_anchor_count: Option<u64>,
 }
 
 #[derive(Clone, Debug)]
 pub(super) struct SequenceSample {
     pub(super) raw: SequenceItemRaw,
     pub(super) validation_time: OffsetDateTime,
-    pub(super) ccr_path: PathBuf,
-    pub(super) cir_path: PathBuf,
     pub(super) objects: BTreeMap<String, String>,
-    pub(super) object_uris: BTreeSet<String>,
     pub(super) object_hashes: BTreeSet<String>,
     pub(super) rejects: BTreeSet<String>,
-    pub(super) trust_anchors: BTreeSet<String>,
     pub(super) vrps: BTreeSet<String>,
     pub(super) vaps: BTreeSet<String>,
 }
 
+#[derive(Clone, Debug)]
+pub(super) struct SequenceMeta {
+    pub(super) raw: SequenceItemRaw,
+    pub(super) validation_time: OffsetDateTime,
+    pub(super) ccr_path: PathBuf,
+    pub(super) cir_path: PathBuf,
+    pub(super) cir_object_count: Option<u64>,
+    pub(super) cir_reject_count: Option<u64>,
+    pub(super) cir_trust_anchor_count: Option<u64>,
+}
+
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
 pub(super) enum Side {
     #[default]

src/tools/sequence_triage_ccr_cir/output.rs

@@ -7,13 +7,13 @@ use super::args::Args;
 use super::io::{format_time, path_string};
 use super::model::{
     ChurnRecord, ChurnSummaryRecord, IntraRpChurn, SandwichAnalysis, SandwichGroupStats,
-    SandwichHeatmapRow, SandwichRecord, SequenceSample,
+    SandwichHeatmapRow, SandwichRecord, SequenceMeta,
 };
 
-pub(super) fn build_output(
+pub(super) fn build_output_from_meta(
     args: &Args,
-    left: &[SequenceSample],
-    right: &[SequenceSample],
+    left: &[SequenceMeta],
+    right: &[SequenceMeta],
     sandwich: &SandwichAnalysis,
     churn: &IntraRpChurn,
 ) -> Value {
@@ -30,6 +30,7 @@ pub(super) fn build_output(
             "warmupSamples": args.warmup_samples,
             "cooldownSamples": args.cooldown_samples,
             "timelineSampleLimit": if args.timeline_sample_limit == 0 { args.sample_limit } else { args.timeline_sample_limit },
+            "executionMode": "streaming-window",
             "sequenceSemantics": {
                 "timeSource": "cir.validation_time",
                 "sourceSampleOrder": "each side is sorted by CIR validation_time, with seq/runId used only as stable tie breakers",
@@ -37,11 +38,11 @@ pub(super) fn build_output(
                 "peerWindow": "all peer samples with source_start.time < peer.time < source_end.time are checked independently",
             },
         },
-        "left": sequence_summary(left),
-        "right": sequence_summary(right),
+        "left": sequence_meta_summary(left),
+        "right": sequence_meta_summary(right),
         "sandwich": {
             "strictTimeWindow": true,
-            "method": "For each side sorted by CIR validation_time, use two adjacent source samples as a stable interval. If source_start.time < peer.time < source_end.time and the source value is identical at both interval endpoints, every peer sample in the interval is expected to contain the same value.",
+            "method": "Streaming mode: first build sample timing metadata, then process each sandwich triple by loading only source_start, peer, and source_end CIR/CCR artifacts.",
             "totals": {
                 "occurrences": sandwich.total_occurrences,
                 "uniqueKeys": sandwich.unique_keys.len(),
@@ -59,7 +60,7 @@ pub(super) fn build_output(
             }
         },
         "intraRpChurn": {
-            "method": "For each side, compare adjacent samples independently. Object keys use object_uri|sha256, output keys use typed VRP/VAP canonical keys, and reject keys use rejected object URI without reason.",
+            "method": "Streaming mode: compare adjacent samples by loading only the two CIR or CCR artifacts needed for the current adjacent pair.",
             "left": churn_records_to_json(&churn.left),
             "right": churn_records_to_json(&churn.right),
             "summary": churn_summary_to_json(&churn.summary),
@@ -77,7 +78,7 @@ pub(super) fn build_output(
     })
 }
 
-fn sequence_summary(samples: &[SequenceSample]) -> Value {
+fn sequence_meta_summary(samples: &[SequenceMeta]) -> Value {
     json!({
         "sampleCount": samples.len(),
         "rpIds": samples.iter().map(|sample| sample.raw.rp_id.clone()).collect::<BTreeSet<_>>(),
@@ -99,12 +100,13 @@ fn sequence_summary(samples: &[SequenceSample]) -> Value {
             "cirSha256": sample.raw.cir_sha256,
             "wallMs": sample.raw.wall_ms,
             "maxRssKb": sample.raw.max_rss_kb,
-            "vrps": sample.raw.vrps.or(Some(sample.vrps.len() as u64)),
-            "vaps": sample.raw.vaps.or(Some(sample.vaps.len() as u64)),
-            "objectCount": sample.object_uris.len(),
-            "objectHashCount": sample.object_hashes.len(),
-            "rejectCount": sample.rejects.len(),
-            "trustAnchorCount": sample.trust_anchors.len(),
+            "vrps": sample.raw.vrps,
+            "vaps": sample.raw.vaps,
+            "publicationPoints": sample.raw.publication_points,
+            "objectCount": sample.cir_object_count,
+            "objectHashCount": sample.cir_object_count,
+            "rejectCount": sample.cir_reject_count,
+            "trustAnchorCount": sample.cir_trust_anchor_count,
         })).collect::<Vec<_>>(),
     })
 }

src/tools/sequence_triage_ccr_cir/sandwich.rs

@@ -2,209 +2,222 @@ use std::collections::BTreeSet;
 
 use super::args::Args;
 use super::io::format_time;
+use super::loader::{load_sample_ccr_from_meta, load_sample_cir_from_meta};
 use super::model::{
-    SandwichAnalysis, SandwichGroupStats, SandwichHeatmapRow, SandwichRecord, SequenceSample, Side,
+    SandwichAnalysis, SandwichGroupStats, SandwichHeatmapRow, SandwichRecord, SequenceMeta,
+    SequenceSample, Side,
 };
 
-pub(super) fn build_sandwich_analysis(
+pub(super) fn build_sandwich_analysis_streaming(
     args: &Args,
-    left: &[SequenceSample],
-    right: &[SequenceSample],
-) -> SandwichAnalysis {
+    left: &[SequenceMeta],
+    right: &[SequenceMeta],
+) -> Result<SandwichAnalysis, String> {
     let mut analysis = SandwichAnalysis::default();
-    analyze_sandwich_objects(&mut analysis, Side::Left, left, right, args);
-    analyze_sandwich_objects(&mut analysis, Side::Right, right, left, args);
-    analyze_sandwich_sets(
-        &mut analysis,
-        "reject_uri",
-        "PEER_MISSING_STABLE_REJECT",
-        Side::Left,
-        left,
-        right,
-        |sample| &sample.rejects,
-        args,
-    );
-    analyze_sandwich_sets(
-        &mut analysis,
-        "reject_uri",
-        "PEER_MISSING_STABLE_REJECT",
-        Side::Right,
-        right,
-        left,
-        |sample| &sample.rejects,
-        args,
-    );
-    analyze_sandwich_sets(
-        &mut analysis,
-        "vrp_output",
-        "PEER_MISSING_STABLE_OUTPUT",
-        Side::Left,
-        left,
-        right,
-        |sample| &sample.vrps,
-        args,
-    );
-    analyze_sandwich_sets(
-        &mut analysis,
-        "vrp_output",
-        "PEER_MISSING_STABLE_OUTPUT",
-        Side::Right,
-        right,
-        left,
-        |sample| &sample.vrps,
-        args,
-    );
-    analyze_sandwich_sets(
-        &mut analysis,
-        "vap_output",
-        "PEER_MISSING_STABLE_OUTPUT",
-        Side::Left,
-        left,
-        right,
-        |sample| &sample.vaps,
-        args,
-    );
-    analyze_sandwich_sets(
-        &mut analysis,
-        "vap_output",
-        "PEER_MISSING_STABLE_OUTPUT",
-        Side::Right,
-        right,
-        left,
-        |sample| &sample.vaps,
-        args,
-    );
-    analysis
+    analyze_sandwich_cir_streaming(&mut analysis, Side::Left, left, right, args)?;
+    analyze_sandwich_cir_streaming(&mut analysis, Side::Right, right, left, args)?;
+    analyze_sandwich_ccr_streaming(&mut analysis, Side::Left, left, right, args)?;
+    analyze_sandwich_ccr_streaming(&mut analysis, Side::Right, right, left, args)?;
+    Ok(analysis)
 }
 
-fn analyze_sandwich_objects(
+fn analyze_sandwich_cir_streaming(
     analysis: &mut SandwichAnalysis,
     source_side: Side,
-    source: &[SequenceSample],
-    peer: &[SequenceSample],
+    source: &[SequenceMeta],
+    peer: &[SequenceMeta],
+    args: &Args,
+) -> Result<(), String> {
+    for pair in source.windows(2) {
+        let source_start_meta = &pair[0];
+        let source_end_meta = &pair[1];
+        if source_start_meta.validation_time >= source_end_meta.validation_time {
+            continue;
+        }
+        let peer_indices = peer_meta_indices_between(peer, source_start_meta, source_end_meta);
+        if peer_indices.is_empty() {
+            continue;
+        }
+        let source_start = load_sample_cir_from_meta(source_start_meta)?;
+        let source_end = load_sample_cir_from_meta(source_end_meta)?;
+        for peer_index in peer_indices {
+            let peer_sample = load_sample_cir_from_meta(&peer[peer_index])?;
+            analyze_loaded_sandwich_objects(
+                analysis,
+                source_side,
+                &source_start,
+                &source_end,
+                &peer_sample,
+                args,
+            );
+            analyze_loaded_sandwich_set(
+                analysis,
+                "reject_uri",
+                "PEER_MISSING_STABLE_REJECT",
+                source_side,
+                &source_start,
+                &source_end,
+                &peer_sample,
+                |sample| &sample.rejects,
+                args,
+            );
+        }
+    }
+    Ok(())
+}
+
+fn analyze_sandwich_ccr_streaming(
+    analysis: &mut SandwichAnalysis,
+    source_side: Side,
+    source: &[SequenceMeta],
+    peer: &[SequenceMeta],
+    args: &Args,
+) -> Result<(), String> {
+    for pair in source.windows(2) {
+        let source_start_meta = &pair[0];
+        let source_end_meta = &pair[1];
+        if source_start_meta.validation_time >= source_end_meta.validation_time {
+            continue;
+        }
+        let peer_indices = peer_meta_indices_between(peer, source_start_meta, source_end_meta);
+        if peer_indices.is_empty() {
+            continue;
+        }
+        let source_start = load_sample_ccr_from_meta(source_start_meta)?;
+        let source_end = load_sample_ccr_from_meta(source_end_meta)?;
+        for peer_index in peer_indices {
+            let peer_sample = load_sample_ccr_from_meta(&peer[peer_index])?;
+            analyze_loaded_sandwich_set(
+                analysis,
+                "vrp_output",
+                "PEER_MISSING_STABLE_OUTPUT",
+                source_side,
+                &source_start,
+                &source_end,
+                &peer_sample,
+                |sample| &sample.vrps,
+                args,
+            );
+            analyze_loaded_sandwich_set(
+                analysis,
+                "vap_output",
+                "PEER_MISSING_STABLE_OUTPUT",
+                source_side,
+                &source_start,
+                &source_end,
+                &peer_sample,
+                |sample| &sample.vaps,
+                args,
+            );
+        }
+    }
+    Ok(())
+}
+
+fn peer_meta_indices_between(
+    peer: &[SequenceMeta],
+    source_start: &SequenceMeta,
+    source_end: &SequenceMeta,
+) -> Vec<usize> {
+    peer.iter()
+        .enumerate()
+        .filter_map(|(index, sample)| {
+            (source_start.validation_time < sample.validation_time
+                && sample.validation_time < source_end.validation_time)
+                .then_some(index)
+        })
+        .collect()
+}
+
+fn analyze_loaded_sandwich_objects(
+    analysis: &mut SandwichAnalysis,
+    source_side: Side,
+    source_start: &SequenceSample,
+    source_end: &SequenceSample,
+    peer_sample: &SequenceSample,
     args: &Args,
 ) {
-    for pair in source.windows(2) {
-        let source_start = &pair[0];
-        let source_end = &pair[1];
-        if source_start.validation_time >= source_end.validation_time {
+    for (uri, source_hash) in &source_start.objects {
+        if source_end.objects.get(uri) != Some(source_hash) {
             continue;
         }
-        let peers = peer_samples_between(peer, source_start, source_end);
-        if peers.is_empty() {
-            continue;
-        }
-        for (uri, source_hash) in &source_start.objects {
-            if source_end.objects.get(uri) != Some(source_hash) {
-                continue;
-            }
-            for peer_sample in &peers {
-                match peer_sample.objects.get(uri) {
-                    Some(peer_hash) if peer_hash == source_hash => {}
-                    Some(peer_hash) => analysis.add(
-                        "PEER_HASH_MISMATCH_STABLE_OBJECT",
-                        sandwich_record(
-                            "PEER_HASH_MISMATCH_STABLE_OBJECT",
-                            "object",
-                            uri.clone(),
-                            source_side,
-                            source_start,
-                            source_end,
-                            peer_sample,
-                            Some(source_hash.clone()),
-                            Some(peer_hash.clone()),
-                            "source interval has stable object hash; peer sample has same URI with another hash",
-                        ),
-                        args.sample_limit,
-                    ),
-                    None => analysis.add(
-                        "PEER_MISSING_STABLE_OBJECT",
-                        sandwich_record(
-                            "PEER_MISSING_STABLE_OBJECT",
-                            "object",
-                            uri.clone(),
-                            source_side,
-                            source_start,
-                            source_end,
-                            peer_sample,
-                            Some(source_hash.clone()),
-                            None,
-                            "source interval has stable object hash; peer sample misses the URI",
-                        ),
-                        args.sample_limit,
-                    ),
-                }
-            }
+        match peer_sample.objects.get(uri) {
+            Some(peer_hash) if peer_hash == source_hash => {}
+            Some(peer_hash) => analysis.add(
+                "PEER_HASH_MISMATCH_STABLE_OBJECT",
+                sandwich_record(
+                    "PEER_HASH_MISMATCH_STABLE_OBJECT",
+                    "object",
+                    uri.clone(),
+                    source_side,
+                    source_start,
+                    source_end,
+                    peer_sample,
+                    Some(source_hash.clone()),
+                    Some(peer_hash.clone()),
+                    "source interval has stable object hash; peer sample has same URI with another hash",
+                ),
+                args.sample_limit,
+            ),
+            None => analysis.add(
+                "PEER_MISSING_STABLE_OBJECT",
+                sandwich_record(
+                    "PEER_MISSING_STABLE_OBJECT",
+                    "object",
+                    uri.clone(),
+                    source_side,
+                    source_start,
+                    source_end,
+                    peer_sample,
+                    Some(source_hash.clone()),
+                    None,
+                    "source interval has stable object hash; peer sample misses the URI",
+                ),
+                args.sample_limit,
+            ),
         }
     }
 }
 
-fn analyze_sandwich_sets<F>(
+fn analyze_loaded_sandwich_set<F>(
     analysis: &mut SandwichAnalysis,
     set_type: &'static str,
     classification: &'static str,
     source_side: Side,
-    source: &[SequenceSample],
-    peer: &[SequenceSample],
+    source_start: &SequenceSample,
+    source_end: &SequenceSample,
+    peer_sample: &SequenceSample,
     extract: F,
     args: &Args,
 ) where
     F: for<'a> Fn(&'a SequenceSample) -> &'a BTreeSet<String>,
 {
-    for pair in source.windows(2) {
-        let source_start = &pair[0];
-        let source_end = &pair[1];
-        if source_start.validation_time >= source_end.validation_time {
+    let start_set = extract(source_start);
+    let end_set = extract(source_end);
+    let peer_set = extract(peer_sample);
+    for key in start_set {
+        if !end_set.contains(key) || peer_set.contains(key) {
             continue;
         }
-        let peers = peer_samples_between(peer, source_start, source_end);
-        if peers.is_empty() {
-            continue;
-        }
-        let start_set = extract(source_start);
-        let end_set = extract(source_end);
-        for key in start_set {
-            if !end_set.contains(key) {
-                continue;
-            }
-            for peer_sample in &peers {
-                if extract(peer_sample).contains(key) {
-                    continue;
-                }
-                analysis.add(
-                    classification,
-                    sandwich_record(
-                        classification,
-                        set_type,
-                        key.clone(),
-                        source_side,
-                        source_start,
-                        source_end,
-                        peer_sample,
-                        Some(key.clone()),
-                        None,
-                        "source interval has a stable key; peer sample misses the key",
-                    ),
-                    args.sample_limit,
-                );
-            }
-        }
+        analysis.add(
+            classification,
+            sandwich_record(
+                classification,
+                set_type,
+                key.clone(),
+                source_side,
+                source_start,
+                source_end,
+                peer_sample,
+                Some(key.clone()),
+                None,
+                "source interval has a stable key; peer sample misses the key",
+            ),
+            args.sample_limit,
+        );
     }
 }
 
-fn peer_samples_between<'a>(
-    peer: &'a [SequenceSample],
-    source_start: &SequenceSample,
-    source_end: &SequenceSample,
-) -> Vec<&'a SequenceSample> {
-    peer.iter()
-        .filter(|sample| {
-            source_start.validation_time < sample.validation_time
-                && sample.validation_time < source_end.validation_time
-        })
-        .collect()
-}
-
 #[allow(clippy::too_many_arguments)]
 fn sandwich_record(
     classification: &'static str,
@@ -239,7 +252,7 @@ fn sandwich_record(
 }
 
 impl SandwichAnalysis {
-    fn add(&mut self, class: &'static str, record: SandwichRecord, sample_limit: usize) {
+    pub(super) fn add(&mut self, class: &'static str, record: SandwichRecord, sample_limit: usize) {
         self.total_occurrences += 1;
         self.unique_keys.insert(sandwich_unique_key(&record));
         *self.by_set_type.entry(record.set_type).or_default() += 1;