From f37fe8c4cbe4d36109f503c92faf969d61cb3a18 Mon Sep 17 00:00:00 2001 From: yuyr Date: Sun, 2 Aug 2026 13:24:46 +0800 Subject: [PATCH] =?UTF-8?q?20260802=20delta=20wall=E4=BC=98=E5=8C=96?= =?UTF-8?q?=EF=BC=9A=E8=BE=93=E5=87=BA=E6=AE=B5=E5=85=A8=E5=B9=B6=E8=A1=8C?= =?UTF-8?q?+=E5=BD=92=E7=BA=A6=E5=B0=BE=E6=AE=B5=E5=88=86=E7=89=87?= =?UTF-8?q?=E5=B9=B6=E8=A1=8C=E4=B8=8E=E6=89=93=E7=82=B9=EF=BC=88#139?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/cir/accumulator.rs | 111 ++++++++++ src/cli.rs | 179 +++++++++------ src/validation/run_tree_from_tal.rs | 3 + src/validation/tree_parallel.rs | 324 ++++++++++++++++++++++++---- 4 files changed, 506 insertions(+), 111 deletions(-) diff --git a/src/cir/accumulator.rs b/src/cir/accumulator.rs index fbb5fc9..6f6d78c 100644 --- a/src/cir/accumulator.rs +++ b/src/cir/accumulator.rs @@ -140,6 +140,26 @@ impl CirInputAccumulator { rejects.entry(rsync_uri.to_string()).or_insert(reason); } + /// Merge another accumulator into this one. Object digests conflict-check + /// exactly like repeated `insert_object` calls; rejects keep the first + /// reason seen, matching sequential submission order when `other` holds + /// later publication points than `self`. + pub fn merge(&mut self, other: CirInputAccumulator) -> Result<(), CirExportError> { + for (rsync_uri, digest) in other.fresh_objects { + self.insert_object_digest(CirInputSection::Fresh, &rsync_uri, digest)?; + } + for (rsync_uri, digest) in other.cached_objects { + self.insert_object_digest(CirInputSection::Cached, &rsync_uri, digest)?; + } + for (rsync_uri, reason) in other.fresh_rejects { + self.insert_reject(CirInputSection::Fresh, &rsync_uri, reason); + } + for (rsync_uri, reason) in other.cached_rejects { + self.insert_reject(CirInputSection::Cached, &rsync_uri, reason); + } + Ok(()) + } + pub fn finalize(self) -> CirInputSnapshot { CirInputSnapshot { fresh_validated_objects: finalize_objects(self.fresh_objects), @@ -280,4 +300,95 @@ mod tests { assert_eq!(merged.cached_validated_objects.len(), 1); assert_eq!(merged.fresh_rejected_objects.len(), 1); } + + #[test] + fn accumulator_merge_matches_sequential_submission() { + let mut sequential = CirInputAccumulator::default(); + sequential + .submit_audit_entries( + CirInputSection::Fresh, + &[ + entry("rsync://example.net/a.roa", 0x11, AuditObjectResult::Ok), + entry("rsync://example.net/b.roa", 0x22, AuditObjectResult::Error), + ], + ) + .unwrap(); + sequential + .submit_audit_entries( + CirInputSection::Cached, + &[entry( + "rsync://example.net/c.roa", + 0x33, + AuditObjectResult::Ok, + )], + ) + .unwrap(); + + let mut first = CirInputAccumulator::default(); + first + .submit_audit_entries( + CirInputSection::Fresh, + &[entry("rsync://example.net/a.roa", 0x11, AuditObjectResult::Ok)], + ) + .unwrap(); + let mut second = CirInputAccumulator::default(); + second + .submit_audit_entries( + CirInputSection::Fresh, + &[entry( + "rsync://example.net/b.roa", + 0x22, + AuditObjectResult::Error, + )], + ) + .unwrap(); + second + .submit_audit_entries( + CirInputSection::Cached, + &[entry( + "rsync://example.net/c.roa", + 0x33, + AuditObjectResult::Ok, + )], + ) + .unwrap(); + first.merge(second).unwrap(); + assert_eq!(sequential.finalize(), first.finalize()); + } + + #[test] + fn accumulator_merge_detects_conflicting_hashes_across_shards() { + let mut first = CirInputAccumulator::default(); + first + .insert_object(CirInputSection::Fresh, "rsync://example.net/a.roa", &"11".repeat(32)) + .unwrap(); + let mut second = CirInputAccumulator::default(); + second + .insert_object(CirInputSection::Fresh, "rsync://example.net/a.roa", &"22".repeat(32)) + .unwrap(); + let err = first.merge(second).unwrap_err(); + assert!(matches!(err, CirExportError::ConflictingObjectHash { .. })); + } + + #[test] + fn accumulator_merge_keeps_first_reject_reason() { + let mut first = CirInputAccumulator::default(); + first.insert_reject( + CirInputSection::Fresh, + "rsync://example.net/a.roa", + Some("first".to_string()), + ); + let mut second = CirInputAccumulator::default(); + second.insert_reject( + CirInputSection::Fresh, + "rsync://example.net/a.roa", + Some("second".to_string()), + ); + first.merge(second).unwrap(); + let snapshot = first.finalize(); + assert_eq!( + snapshot.fresh_rejected_objects[0].reason, + Some("first".to_string()) + ); + } } diff --git a/src/cli.rs b/src/cli.rs index 01019e9..a902a48 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -2678,48 +2678,129 @@ pub fn run(argv: &[String]) -> Result<(), String> { ReportJsonFormat::Pretty }; let ccr_produced_at = time::OffsetDateTime::now_utc(); - let (report_result, ccr_result) = if args.skip_report_build { - ( - Ok(ReportTaskOutput::skipped()), - run_ccr_task( - store.as_ref(), - &shared, - args.ccr_out_path.as_deref(), - ccr_produced_at, - ), + let compare_view_trust_anchor = args + .compare_view_trust_anchor + .as_deref() + .unwrap_or("unknown"); + let cir_tal_uris = if args.cir_enabled { + Some(effective_cir_tal_uris_for_discoveries( + &args, + &shared, + resolve_cir_export_tal_uris(&args)?, + )?) + } else { + None + }; + let cir_out_path = if args.cir_enabled { + Some( + args.cir_out_path + .as_deref() + .expect("validated by parse_args for cir"), ) } else { + None + }; + // Take the CIR input snapshot before the output stage so the CIR export can + // run inside the same scoped-thread group as report/ccr/compare_view; no + // other output task reads `shared.cir_input`. + let cir_input_owned = args + .cir_enabled + .then(|| std::mem::take(&mut shared.cir_input)); + let (report_result, ccr_result, compare_view_result, cir_result) = std::thread::scope(|scope| { - let report_handle = scope.spawn(|| { - run_report_task( - &policy, - validation_time, - &shared, - args.report_json_path.as_deref(), - report_json_format, - ) - }); + // Reborrow `shared` as a plain reference so the scoped output tasks + // (incl. the `move` CIR task) capture the reference instead of + // moving fields out of the owned value. + let shared = &shared; + let report_handle = if args.skip_report_build { + None + } else { + Some(scope.spawn(|| { + run_report_task( + &policy, + validation_time, + shared, + args.report_json_path.as_deref(), + report_json_format, + ) + })) + }; let ccr_handle = scope.spawn(|| { run_ccr_task( store.as_ref(), - &shared, + shared, args.ccr_out_path.as_deref(), ccr_produced_at, ) }); - let report_result = report_handle - .join() - .map_err(|_| "report task panicked".to_string()) - .and_then(|result| result); + let compare_view_handle = scope.spawn(|| { + run_compare_view_task( + shared, + args.vrps_csv_out_path.as_deref(), + args.vaps_csv_out_path.as_deref(), + compare_view_trust_anchor, + ) + }); + let cir_handle = match (cir_tal_uris.as_ref(), cir_out_path, cir_input_owned) { + (Some(cir_tal_uris), Some(cir_out_path), Some(cir_input)) => { + Some(scope.spawn(move || { + if cir_tal_uris.len() != shared.discoveries.len() { + return Err(format!( + "CIR export TAL URI count ({}) does not match discovery count ({})", + cir_tal_uris.len(), + shared.discoveries.len() + )); + } + let tal_bindings = shared + .discoveries + .iter() + .zip(cir_tal_uris.iter()) + .map(|(discovery, tal_uri)| CirTrustAnchorBinding { + trust_anchor: &discovery.trust_anchor, + tal_uri: tal_uri.as_str(), + }) + .collect::>(); + export_cir_from_input_snapshot_multi( + &tal_bindings, + validation_time, + cir_input, + cir_out_path, + ) + .map_err(|e| e.to_string()) + })) + } + _ => None, + }; + let report_result = match report_handle { + Some(handle) => handle + .join() + .map_err(|_| "report task panicked".to_string()) + .and_then(|result| result), + None => Ok(ReportTaskOutput::skipped()), + }; let ccr_result = ccr_handle .join() .map_err(|_| "ccr task panicked".to_string()) .and_then(|result| result); - (report_result, ccr_result) - }) - }; + let compare_view_result = compare_view_handle + .join() + .map_err(|_| "compare view task panicked".to_string()) + .and_then(|result| result); + let cir_result = cir_handle.map(|handle| { + handle + .join() + .map_err(|_| "cir task panicked".to_string()) + .and_then(|result| result) + }); + (report_result, ccr_result, compare_view_result, cir_result) + }); let report_output = report_result?; let ccr_output = ccr_result?; + let compare_view_output = compare_view_result?; + let cir_summary = match cir_result { + Some(result) => Some(result?), + None => None, + }; record_memory_checkpoint( &mut memory_checkpoints, "after_report_and_ccr", @@ -2740,16 +2821,6 @@ pub fn run(argv: &[String]) -> Result<(), String> { let ccr_build_ms = ccr_output.ccr_build_ms; let ccr_build_breakdown = ccr_output.ccr_build_breakdown; let ccr_write_ms = ccr_output.ccr_write_ms; - let compare_view_trust_anchor = args - .compare_view_trust_anchor - .as_deref() - .unwrap_or("unknown"); - let compare_view_output = run_compare_view_task( - &shared, - args.vrps_csv_out_path.as_deref(), - args.vaps_csv_out_path.as_deref(), - compare_view_trust_anchor, - )?; let compare_view_build_ms = compare_view_output.build_ms; let compare_view_write_ms = compare_view_output.write_ms; record_memory_checkpoint( @@ -2762,45 +2833,13 @@ pub fn run(argv: &[String]) -> Result<(), String> { let mut cir_build_cir_ms = None; let mut cir_write_cir_ms = None; let mut cir_total_ms = None; - if args.cir_enabled { - let cir_tal_uris = effective_cir_tal_uris_for_discoveries( - &args, - &shared, - resolve_cir_export_tal_uris(&args)?, - )?; - if cir_tal_uris.len() != shared.discoveries.len() { - return Err(format!( - "CIR export TAL URI count ({}) does not match discovery count ({})", - cir_tal_uris.len(), - shared.discoveries.len() - )); - } - let cir_out_path = args - .cir_out_path - .as_deref() - .expect("validated by parse_args for cir"); - let tal_bindings = shared - .discoveries - .iter() - .zip(cir_tal_uris.iter()) - .map(|(discovery, tal_uri)| CirTrustAnchorBinding { - trust_anchor: &discovery.trust_anchor, - tal_uri: tal_uri.as_str(), - }) - .collect::>(); - let summary = export_cir_from_input_snapshot_multi( - &tal_bindings, - validation_time, - std::mem::take(&mut shared.cir_input), - cir_out_path, - ) - .map_err(|e| e.to_string())?; + if let Some(summary) = cir_summary { cir_build_cir_ms = Some(summary.timing.build_cir_ms); cir_write_cir_ms = Some(summary.timing.write_cir_ms); cir_total_ms = Some(summary.timing.total_ms); eprintln!( "wrote CIR: {} (objects={}, trust_anchors={}, build_cir_ms={}, write_cir_ms={}, total_ms={})", - cir_out_path.display(), + cir_out_path.expect("cir path present when cir enabled").display(), summary.object_count, summary.trust_anchor_count, summary.timing.build_cir_ms, diff --git a/src/validation/run_tree_from_tal.rs b/src/validation/run_tree_from_tal.rs index 33271c0..4de925f 100644 --- a/src/validation/run_tree_from_tal.rs +++ b/src/validation/run_tree_from_tal.rs @@ -328,15 +328,18 @@ fn persist_transport_request_prefetch_snapshot( } let snapshot = runtime.transport_prefetch_snapshot(); let recorded = snapshot.requests.len() as u64; + let persist_started = std::time::Instant::now(); store .put_transport_prefetch_snapshot(&snapshot) .map_err(|err| TreeRunError::Runner(err.to_string()))?; + let persist_ms = persist_started.elapsed().as_millis() as u64; record_transport_prefetch_count(timing, "transport_prefetch_recorded_requests", recorded); crate::progress_log::emit( "phase1_repo_prefetch_persisted", serde_json::json!({ "recorded_requests": recorded, "schema_version": snapshot.schema_version, + "persist_ms": persist_ms, }), ); Ok(()) diff --git a/src/validation/tree_parallel.rs b/src/validation/tree_parallel.rs index 43d7bef..faadca3 100644 --- a/src/validation/tree_parallel.rs +++ b/src/validation/tree_parallel.rs @@ -2720,19 +2720,45 @@ fn is_complete( && staging_inflight == 0 } -fn build_tree_output(mut finished: Vec) -> TreeRunAuditOutput { - finished.sort_by_key(|item| item.node.id); - let mut instances_processed = 0usize; - let mut instances_failed = 0usize; - let mut warnings = Vec::new(); - let mut vrps = Vec::new(); - let mut aspas = Vec::new(); - let mut router_keys = Vec::new(); - let mut publication_points = Vec::new(); - let mut roa_cache_stats = crate::validation::objects::RoaValidationCacheStats::default(); - let mut cir_input = CirInputAccumulator::default(); +/// Minimum publication points per reduction shard; smaller runs stay +/// single-threaded to avoid thread-spawn overhead dominating the reduction. +const TREE_OUTPUT_MIN_SHARD_LEN: usize = 4096; +/// Hard cap on reduction shards regardless of core count. +const TREE_OUTPUT_MAX_SHARDS: usize = 8; - for item in finished { +fn tree_output_shard_count(len: usize) -> usize { + if len < TREE_OUTPUT_MIN_SHARD_LEN * 2 { + return 1; + } + let parallel = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(1); + parallel + .min(TREE_OUTPUT_MAX_SHARDS) + .min(len / TREE_OUTPUT_MIN_SHARD_LEN) + .max(1) +} + +/// Per-shard reduction result of the phase2 output merge. Shards cover +/// contiguous ranges of the id-sorted finished list, so concatenating shard +/// outputs in shard order reproduces the sequential single-threaded order +/// exactly. +#[derive(Default)] +struct TreeOutputShardReduction { + instances_processed: usize, + instances_failed: usize, + warnings: Vec, + vrps: Vec, + aspas: Vec, + router_keys: Vec, + publication_points: Vec, + roa_cache_stats: crate::validation::objects::RoaValidationCacheStats, + cir_input: CirInputAccumulator, +} + +fn reduce_finished_shard(items: Vec) -> TreeOutputShardReduction { + let mut reduction = TreeOutputShardReduction::default(); + for item in items { match item.result { FinishedPublicationPointResult::Ok { source, @@ -2742,51 +2768,136 @@ fn build_tree_output(mut finished: Vec) -> TreeRunAudi cir_fresh_objects, cir_cached_objects, } => { - instances_processed += 1; - warnings.extend(result_warnings); - warnings.extend(objects.warnings); - roa_cache_stats.add_assign(&objects.roa_cache_stats); - vrps.extend(objects.vrps); - aspas.extend(objects.aspas); - router_keys.extend(objects.router_keys); + reduction.instances_processed += 1; + reduction.warnings.extend(result_warnings); + reduction.warnings.extend(objects.warnings); + reduction + .roa_cache_stats + .add_assign(&objects.roa_cache_stats); + reduction.vrps.extend(objects.vrps); + reduction.aspas.extend(objects.aspas); + reduction.router_keys.extend(objects.router_keys); let mut audit: PublicationPointAudit = audit; audit.node_id = Some(item.node.id); audit.parent_node_id = item.node.parent_id; audit.discovered_from = item.node.discovered_from; crate::validation::tree::submit_publication_point_cir_input( - &mut cir_input, + &mut reduction.cir_input, source, &audit, &cir_fresh_objects, &cir_cached_objects, ) .expect("CIR input collection from validated audit must not fail"); - publication_points.push(audit); + reduction.publication_points.push(audit); } FinishedPublicationPointResult::Err(err) => { - instances_failed += 1; - warnings.push( + reduction.instances_failed += 1; + reduction.warnings.push( Warning::new(format!("publication point failed: {err}")) .with_context(&item.node.manifest_rsync_uri), ); } } } + reduction +} - TreeRunAuditOutput { - tree: TreeRunOutput { - instances_processed, - instances_failed, - warnings, - vrps, - aspas, - router_keys, - }, - publication_points, - roa_cache_stats, - cir_input: cir_input.finalize(), +fn merge_shard_reductions( + reductions: Vec, +) -> TreeOutputShardReduction { + let mut iter = reductions.into_iter(); + let mut merged = iter.next().unwrap_or_default(); + for mut other in iter { + merged.instances_processed += other.instances_processed; + merged.instances_failed += other.instances_failed; + merged.warnings.append(&mut other.warnings); + merged.vrps.append(&mut other.vrps); + merged.aspas.append(&mut other.aspas); + merged.router_keys.append(&mut other.router_keys); + merged + .publication_points + .append(&mut other.publication_points); + merged.roa_cache_stats.add_assign(&other.roa_cache_stats); + merged + .cir_input + .merge(std::mem::take(&mut other.cir_input)) + .expect("CIR input merge from validated audits must not fail"); } + merged +} + +fn build_tree_output(mut finished: Vec) -> TreeRunAuditOutput { + let total_started = Instant::now(); + finished.sort_by_key(|item| item.node.id); + let sort_ms = total_started.elapsed().as_millis() as u64; + let shard_count = tree_output_shard_count(finished.len()); + + let reduce_started = Instant::now(); + let reductions = if shard_count <= 1 { + vec![reduce_finished_shard(finished)] + } else { + let chunk_len = finished.len().div_ceil(shard_count); + let mut shards: Vec> = Vec::new(); + let mut rest = finished; + while !rest.is_empty() { + let split_at = chunk_len.min(rest.len()); + let tail = rest.split_off(split_at); + shards.push(std::mem::replace(&mut rest, tail)); + } + std::thread::scope(|scope| { + let handles: Vec<_> = shards + .into_iter() + .map(|shard| scope.spawn(move || reduce_finished_shard(shard))) + .collect(); + handles + .into_iter() + .map(|handle| { + handle + .join() + .expect("tree output reduction shard panicked") + }) + .collect() + }) + }; + let reduce_ms = reduce_started.elapsed().as_millis() as u64; + + let merge_started = Instant::now(); + let merged = merge_shard_reductions(reductions); + let merge_ms = merge_started.elapsed().as_millis() as u64; + + let finalize_started = Instant::now(); + let output = TreeRunAuditOutput { + tree: TreeRunOutput { + instances_processed: merged.instances_processed, + instances_failed: merged.instances_failed, + warnings: merged.warnings, + vrps: merged.vrps, + aspas: merged.aspas, + router_keys: merged.router_keys, + }, + publication_points: merged.publication_points, + roa_cache_stats: merged.roa_cache_stats, + cir_input: merged.cir_input.finalize(), + }; + let finalize_ms = finalize_started.elapsed().as_millis() as u64; + + crate::progress_log::emit( + "phase2_build_tree_output", + serde_json::json!({ + "sort_ms": sort_ms, + "shard_count": shard_count, + "reduce_ms": reduce_ms, + "merge_ms": merge_ms, + "finalize_ms": finalize_ms, + "total_ms": total_started.elapsed().as_millis() as u64, + "publication_points": output.publication_points.len(), + "instances_processed": output.tree.instances_processed, + "instances_failed": output.tree.instances_failed, + }), + ); + output } pub fn run_tree_parallel_phase2_audit( @@ -2800,12 +2911,14 @@ pub fn run_tree_parallel_phase2_audit( #[cfg(test)] mod tests { use super::{ - CacheHitOutcome, CompleteOutcome, FinishedPublicationPointResult, InflightPublicationPoint, - QueuedCaInstance, ReadyCaInstance, ReadyStageMetrics, ReadyStageSubmitter, StageOutcome, - apply_ready_publication_point_stage, compact_phase2_finished_result, - compact_phase2_finished_result_result, compute_ready_publication_point_stage, - event_poll_timeout, finalize_metrics_from_output, finalize_publication_point_state, - is_complete, submit_ready_batch_to_stage_pool, + CacheHitOutcome, CompleteOutcome, FinishedPublicationPoint, FinishedPublicationPointNode, + FinishedPublicationPointResult, InflightPublicationPoint, QueuedCaInstance, + ReadyCaInstance, ReadyStageMetrics, ReadyStageSubmitter, StageOutcome, + TREE_OUTPUT_MAX_SHARDS, TREE_OUTPUT_MIN_SHARD_LEN, apply_ready_publication_point_stage, + build_tree_output, compact_phase2_finished_result, compact_phase2_finished_result_result, + compute_ready_publication_point_stage, event_poll_timeout, finalize_metrics_from_output, + finalize_publication_point_state, is_complete, merge_shard_reductions, + reduce_finished_shard, submit_ready_batch_to_stage_pool, tree_output_shard_count, }; use crate::audit::{ AuditObjectKind, AuditObjectResult, DiscoveredFrom, ObjectAuditEntry, @@ -2935,6 +3048,135 @@ mod tests { } } + fn finished_ok_item(id: u64) -> FinishedPublicationPoint { + let mut result = sample_result(); + result + .warnings + .push(crate::report::Warning::new(format!("result-warning-{id:05}"))); + result + .objects + .warnings + .push(crate::report::Warning::new(format!("objects-warning-{id:05}"))); + result.objects.vrps.push(crate::validation::objects::Vrp { + asn: 64496 + id as u32, + prefix: crate::data_model::roa::IpPrefix { + afi: crate::data_model::roa::RoaAfi::Ipv4, + addr: [192, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], + prefix_len: 24, + }, + max_length: 24, + }); + result.audit.objects.push(crate::audit::ObjectAuditEntry { + rsync_uri: format!("rsync://example.test/repo/{id:05}.roa"), + sha256_hex: format!("{id:064x}"), + kind: crate::audit::AuditObjectKind::Roa, + result: crate::audit::AuditObjectResult::Ok, + detail: None, + }); + FinishedPublicationPoint { + node: FinishedPublicationPointNode { + id, + parent_id: None, + discovered_from: None, + manifest_rsync_uri: format!("rsync://example.test/repo/{id:05}.mft"), + }, + result: compact_phase2_finished_result(result, false), + } + } + + fn finished_err_item(id: u64) -> FinishedPublicationPoint { + FinishedPublicationPoint { + node: FinishedPublicationPointNode { + id, + parent_id: None, + discovered_from: None, + manifest_rsync_uri: format!("rsync://example.test/repo/{id:05}.mft"), + }, + result: FinishedPublicationPointResult::Err(format!("boom-{id:05}")), + } + } + + fn mixed_finished_items(n: u64) -> Vec { + (0..n) + .map(|id| { + if id % 97 == 0 { + finished_err_item(id) + } else { + finished_ok_item(id) + } + }) + .collect() + } + + #[test] + fn tree_output_shard_count_respects_thresholds() { + assert_eq!(tree_output_shard_count(0), 1); + assert_eq!(tree_output_shard_count(TREE_OUTPUT_MIN_SHARD_LEN * 2 - 1), 1); + assert!(tree_output_shard_count(TREE_OUTPUT_MIN_SHARD_LEN * 2) >= 1); + let parallel = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(1); + assert_eq!( + tree_output_shard_count(1_000_000), + parallel.min(TREE_OUTPUT_MAX_SHARDS) + ); + } + + #[test] + fn tree_output_sharded_merge_matches_single_reduction() { + let single = merge_shard_reductions(vec![reduce_finished_shard(mixed_finished_items(300))]); + + let mut chunks: Vec> = Vec::new(); + let mut rest = mixed_finished_items(300); + while !rest.is_empty() { + let split_at = 100.min(rest.len()); + let tail = rest.split_off(split_at); + chunks.push(std::mem::replace(&mut rest, tail)); + } + let merged = merge_shard_reductions( + chunks.into_iter().map(reduce_finished_shard).collect(), + ); + + assert_eq!(single.instances_processed, merged.instances_processed); + assert_eq!(single.instances_failed, merged.instances_failed); + assert_eq!(format!("{:?}", single.warnings), format!("{:?}", merged.warnings)); + assert_eq!(format!("{:?}", single.vrps), format!("{:?}", merged.vrps)); + assert_eq!( + format!("{:?}", single.publication_points), + format!("{:?}", merged.publication_points) + ); + assert_eq!( + format!("{:?}", single.roa_cache_stats), + format!("{:?}", merged.roa_cache_stats) + ); + assert_eq!(single.cir_input.finalize(), merged.cir_input.finalize()); + } + + #[test] + fn build_tree_output_orders_results_by_node_id() { + let mut items = mixed_finished_items(50); + items.reverse(); + let output = build_tree_output(items); + let got: Vec = output + .publication_points + .iter() + .map(|pp| pp.node_id.expect("node id set")) + .collect(); + let want: Vec = (1..50).filter(|id| id % 97 != 0).collect(); + assert_eq!(got, want); + assert_eq!(output.tree.instances_failed, 1); + let first_warning = format!("{:?}", output.tree.warnings[0]); + assert!( + first_warning.contains("publication point failed: boom-00000"), + "failed publication point warning keeps id order: {first_warning}" + ); + assert_eq!(output.cir_input.fresh_validated_objects.len(), 49); + assert_eq!( + output.cir_input.fresh_validated_objects[0].rsync_uri, + "rsync://example.test/repo/00001.roa" + ); + } + #[test] fn finalize_metrics_from_output_captures_breakdown_and_counts() { let mut result = sample_result();