1use std::path::Path;
32
33use chrono::NaiveDate;
34
35use crate::config::Config;
36use crate::destination::placeholder::PlaceholderContext;
37use crate::error::Result;
38use crate::pipeline::ManifestVerification;
39use crate::pipeline::validate_manifest::{ValidateDepth, verify_at_destination};
40
41pub enum ValidateOutputFormat {
43 Pretty,
45 Json(Option<String>),
47}
48
49#[derive(Debug, Default, Clone)]
55pub struct ValidateTarget {
56 pub date: Option<NaiveDate>,
58 pub run_id: Option<String>,
60 pub prefix_override: Option<String>,
63 pub depth: ValidateDepth,
69}
70
71impl ValidateTarget {
72 fn placeholder_context(&self, export_name: &str) -> PlaceholderContext {
73 let mut ctx = match self.date {
74 Some(d) => PlaceholderContext::for_date(d, export_name),
75 None => PlaceholderContext::for_today(export_name),
76 };
77 if let Some(rid) = &self.run_id {
78 ctx = ctx.with_run_id(rid.clone());
79 }
80 ctx
81 }
82}
83
84pub fn run_validate_command(
93 config_path: &str,
94 export_name: Option<&str>,
95 format: ValidateOutputFormat,
96 target: ValidateTarget,
97) -> Result<()> {
98 let config = Config::load_with_params(config_path, None)?;
99
100 let exports: Vec<&crate::config::ExportConfig> = match export_name {
101 Some(name) => match config.exports.iter().find(|e| e.name == name) {
102 Some(e) => vec![e],
103 None => anyhow::bail!("export '{}' not found in config", name),
104 },
105 None => config.exports.iter().collect(),
106 };
107
108 if exports.is_empty() {
109 anyhow::bail!("no exports defined in config — nothing to validate");
110 }
111
112 if target.prefix_override.is_some() && exports.len() > 1 {
117 anyhow::bail!(
118 "--prefix requires --export <name>: cannot apply one override to {} exports",
119 exports.len()
120 );
121 }
122
123 let mut all_results: Vec<ExportVerdict> = Vec::with_capacity(exports.len());
124 let mut hard_failures: Vec<String> = Vec::new();
125
126 for export in &exports {
127 let ctx = target.placeholder_context(&export.name);
131 let mut expanded_dest =
132 crate::destination::placeholder::expand_destination(export.destination.clone(), &ctx);
133 if let Some(p) = &target.prefix_override {
134 expanded_dest.path = Some(p.clone());
138 expanded_dest.prefix = Some(p.clone());
139 }
140 let multiplex = if export.mode == crate::config::ExportMode::Cdc {
149 export.tables.as_deref().filter(|t| !t.is_empty())
150 } else {
151 None
152 };
153 let has_snapshot = export.mode == crate::config::ExportMode::Cdc
154 && export.cdc.as_ref().and_then(|c| c.initial)
155 == Some(crate::config::CdcInitialMode::Snapshot);
156 match multiplex {
157 Some(tables) => {
158 for table in tables {
159 verify_cdc_table(
160 crate::pipeline::cdc_job::dest_for_table(&expanded_dest, table),
161 format!("{}/{}", export.name, table),
162 export,
163 &target,
164 has_snapshot,
165 &mut all_results,
166 &mut hard_failures,
167 );
168 }
169 }
170 None if export.mode == crate::config::ExportMode::Cdc => {
171 verify_cdc_table(
173 expanded_dest,
174 export.name.clone(),
175 export,
176 &target,
177 has_snapshot,
178 &mut all_results,
179 &mut hard_failures,
180 );
181 }
182 None => {
183 verify_one_prefix(
186 expanded_dest,
187 export.name.clone(),
188 export,
189 &target,
190 false,
191 false,
192 &mut all_results,
193 &mut hard_failures,
194 );
195 }
196 }
197 }
198
199 match format {
200 ValidateOutputFormat::Pretty => render_pretty(&all_results, &hard_failures),
201 ValidateOutputFormat::Json(out_path) => {
202 render_json(&all_results, &hard_failures, out_path)?
203 }
204 }
205
206 let failed_verdicts = all_results
224 .iter()
225 .filter(|r| verdict_fails_exit(&r.verification))
226 .count();
227 if failed_verdicts > 0 {
228 return Err(crate::error::DataIntegrityError::new(format!(
235 "rivet validate: {} export(s) failed verification",
236 hard_failures.len() + failed_verdicts
237 ))
238 .into());
239 }
240 if !hard_failures.is_empty() {
241 anyhow::bail!(
243 "rivet validate: {} export(s) failed verification",
244 hard_failures.len()
245 );
246 }
247 Ok(())
248}
249
250fn verdict_fails_exit(v: &ManifestVerification) -> bool {
257 !v.passed && v.has_failures()
258}
259
260struct ExportVerdict {
264 name: String,
265 resolved_prefix: String,
266 verification: ManifestVerification,
267}
268
269fn resolved_prefix_for_display(dest: &crate::config::DestinationConfig) -> String {
276 dest.prefix
277 .clone()
278 .or_else(|| dest.path.clone())
279 .unwrap_or_else(|| "<unresolved>".into())
280}
281
282#[allow(clippy::too_many_arguments)]
289fn verify_cdc_table(
290 table_dest: crate::config::DestinationConfig,
291 display_name: String,
292 export: &crate::config::ExportConfig,
293 target: &ValidateTarget,
294 has_snapshot: bool,
295 all_results: &mut Vec<ExportVerdict>,
296 hard_failures: &mut Vec<String>,
297) {
298 if has_snapshot {
301 let snap = crate::pipeline::cdc_job::dest_for_table(&table_dest, "snapshot");
302 verify_one_prefix(
303 snap,
304 format!("{display_name}/snapshot"),
305 export,
306 target,
307 false,
308 false,
309 all_results,
310 hard_failures,
311 );
312 }
313 verify_one_prefix(
314 table_dest,
315 display_name,
316 export,
317 target,
318 true,
319 has_snapshot,
320 all_results,
321 hard_failures,
322 );
323}
324
325#[allow(clippy::too_many_arguments)]
331fn verify_one_prefix(
332 expanded_dest: crate::config::DestinationConfig,
333 display_name: String,
334 export: &crate::config::ExportConfig,
335 target: &ValidateTarget,
336 run_cdc_pos_check: bool,
337 drop_snapshot_untracked: bool,
338 all_results: &mut Vec<ExportVerdict>,
339 hard_failures: &mut Vec<String>,
340) {
341 let resolved_prefix = resolved_prefix_for_display(&expanded_dest);
342 let dest = match crate::destination::create_destination(&expanded_dest) {
343 Ok(d) => d,
344 Err(e) => {
345 hard_failures.push(format!(
346 "export '{}' (prefix: {}): could not open destination: {:#}",
347 display_name, resolved_prefix, e
348 ));
349 return;
350 }
351 };
352 if dest.capabilities().commit_protocol == crate::destination::WriteCommitProtocol::Streaming {
354 log::info!(
355 "export '{}': streaming destination, skipping (nothing to verify)",
356 display_name
357 );
358 return;
359 }
360 match verify_at_destination(&*dest, "", target.depth) {
361 Ok(mut v) => {
362 v.enforce_content_policy(export.verify.requires_content());
365 if drop_snapshot_untracked {
372 v.failures.retain(|f| {
373 !matches!(
374 f,
375 crate::pipeline::validate_manifest::Failure::UntrackedObject { key, .. }
376 if key.starts_with("snapshot/")
377 )
378 });
379 }
380 if target.prefix_override.is_some() {
390 v.require_manifest_present(&resolved_prefix);
391 }
392 let manifest_verified = v.manifest_found && v.passed;
396 all_results.push(ExportVerdict {
397 name: display_name.clone(),
398 resolved_prefix,
399 verification: v,
400 });
401 if target.depth.runs_part_download()
407 && run_cdc_pos_check
408 && export.format == crate::config::FormatType::Parquet
409 {
410 match crate::source::cdc::validate::check_positions(&*dest, "") {
411 Ok(pc) if pc.is_ok() => log::info!(
412 "export '{}': cdc __pos continuity OK — {} changes across {} parts, range {:?}..{:?}",
413 display_name,
414 pc.rows,
415 pc.parts,
416 pc.first,
417 pc.last
418 ),
419 Ok(pc) => {
420 for viol in &pc.violations {
421 hard_failures
422 .push(format!("export '{}': cdc __pos: {}", display_name, viol));
423 }
424 }
425 Err(e) => hard_failures.push(format!(
426 "export '{}': cdc __pos check failed: {:#}",
427 display_name, e
428 )),
429 }
430 }
431 if target.depth.runs_part_download()
442 && manifest_verified
443 && export.format == crate::config::FormatType::Parquet
444 && let Err(e) =
445 crate::source::value_checksum::validate_manifest_checksums(&*dest, "")
446 {
447 hard_failures.push(format!(
448 "export '{}': value checksum: {:#}",
449 display_name, e
450 ));
451 }
452 }
453 Err(e) => {
454 hard_failures.push(format!(
455 "export '{}' (prefix: {}): verify_at_destination failed: {:#}",
456 display_name, resolved_prefix, e
457 ));
458 }
459 }
460}
461
462fn render_pretty(results: &[ExportVerdict], hard_failures: &[String]) {
463 use std::io::Write;
464 let stdout = std::io::stdout();
465 let mut h = stdout.lock();
466
467 for r in results {
468 let _ = writeln!(h, "── {} ──", r.name);
469 let _ = writeln!(h, " prefix: {}", r.resolved_prefix);
470 let v = &r.verification;
471 let _ = writeln!(h, " depth: {}", v.depth_level);
475 if v.legacy_run {
476 let _ = writeln!(
477 h,
478 " status: legacy_run (no manifest at destination — pre-0.7.0 prefix)"
479 );
480 continue;
481 }
482 if !v.manifest_found {
483 let _ = writeln!(h, " status: NO MANIFEST");
484 for failure in &v.failures {
490 let _ = writeln!(h, " failure: [{}] {}", failure.error_code(), failure);
491 }
492 continue;
493 }
494 let _ = writeln!(
495 h,
496 " status: {}",
497 if v.passed { "PASSED" } else { "FAILED" }
498 );
499 let _ = writeln!(
500 h,
501 " parts: {} verified ({} md5, {} size-only), {} failed",
502 v.parts_verified,
503 v.parts_md5_verified,
504 v.parts_verified.saturating_sub(v.parts_md5_verified),
505 v.parts_failed
506 );
507 let _ = writeln!(
508 h,
509 " _SUCCESS: {}",
510 if v.success_marker_consistent {
511 "consistent"
512 } else if v.failures.iter().any(|f| matches!(
513 f,
514 crate::pipeline::ManifestVerificationFailure::SuccessMarkerStale { .. }
515 | crate::pipeline::ManifestVerificationFailure::SuccessMarkerMalformed { .. }
516 | crate::pipeline::ManifestVerificationFailure::SuccessMarkerReadError { .. }
517 )) {
518 "INCONSISTENT (see failures)"
519 } else {
520 "absent (no signal)"
521 }
522 );
523 let _ = writeln!(
524 h,
525 " manifest: {}",
526 if v.manifest_self_consistent {
527 "self-consistent"
528 } else {
529 "INCONSISTENT (see failures)"
530 }
531 );
532 for failure in &v.failures {
533 let label = if failure.is_fatal() {
541 "failure:"
542 } else {
543 "warning:"
544 };
545 let _ = writeln!(h, " {} [{}] {}", label, failure.error_code(), failure);
549 }
550 }
551
552 if !hard_failures.is_empty() {
553 let _ = writeln!(h);
554 let _ = writeln!(h, "── errors ──");
555 for e in hard_failures {
556 let _ = writeln!(h, " {}", e);
557 }
558 }
559 let _ = h.flush();
560}
561
562fn failure_json(f: &crate::pipeline::ManifestVerificationFailure) -> serde_json::Value {
570 let mut value = serde_json::json!(f);
571 if let Some(obj) = value.as_object_mut() {
572 obj.insert(
573 "code".to_string(),
574 serde_json::Value::String(f.error_code().to_string()),
575 );
576 }
577 value
578}
579
580fn verification_json(v: &ManifestVerification) -> serde_json::Value {
585 let mut value = serde_json::json!(v);
586 if let Some(obj) = value.as_object_mut() {
587 let failures: Vec<serde_json::Value> = v.failures.iter().map(failure_json).collect();
588 obj.insert("failures".to_string(), serde_json::Value::Array(failures));
589 }
590 value
591}
592
593fn render_json(
594 results: &[ExportVerdict],
595 hard_failures: &[String],
596 out_path: Option<String>,
597) -> Result<()> {
598 let warnings: Vec<serde_json::Value> = results
606 .iter()
607 .flat_map(|r| {
608 r.verification
609 .failures
610 .iter()
611 .filter(|f| !f.is_fatal())
612 .map(move |f| {
613 serde_json::json!({
614 "export_name": r.name,
615 "warning": failure_json(f),
616 })
617 })
618 })
619 .collect();
620
621 let payload = serde_json::json!({
622 "exports": results
623 .iter()
624 .map(|r| {
625 serde_json::json!({
626 "export_name": r.name,
627 "resolved_prefix": r.resolved_prefix,
628 "verification": verification_json(&r.verification),
629 })
630 })
631 .collect::<Vec<_>>(),
632 "warnings": warnings,
633 "errors": hard_failures,
634 });
635 let serialized = serde_json::to_string_pretty(&payload)?;
636 match out_path {
637 Some(p) => {
638 std::fs::write(Path::new(&p), &serialized)?;
639 log::info!("rivet validate: wrote JSON report to {}", p);
640 }
641 None => {
642 println!("{}", serialized);
643 }
644 }
645 Ok(())
646}
647
648#[cfg(test)]
649mod tests {
650 use super::*;
651
652 #[test]
655 fn target_default_uses_today() {
656 let target = ValidateTarget::default();
657 let ctx = target.placeholder_context("orders");
658 assert_eq!(ctx.date, chrono::Utc::now().date_naive());
659 assert_eq!(ctx.export_name, "orders");
660 assert!(ctx.run_id.is_none());
661 }
662
663 #[test]
664 fn target_with_date_overrides_today() {
665 let target = ValidateTarget {
666 date: Some(NaiveDate::from_ymd_opt(2026, 5, 21).unwrap()),
667 ..Default::default()
668 };
669 let ctx = target.placeholder_context("orders");
670 assert_eq!(ctx.date, NaiveDate::from_ymd_opt(2026, 5, 21).unwrap());
671 assert!(ctx.run_id.is_none());
672 }
673
674 #[test]
675 fn target_composes_date_and_run_id() {
676 let target = ValidateTarget {
680 date: Some(NaiveDate::from_ymd_opt(2026, 5, 21).unwrap()),
681 run_id: Some("r-abc123".into()),
682 prefix_override: None,
683 ..Default::default()
684 };
685 let ctx = target.placeholder_context("orders");
686 assert_eq!(ctx.date, NaiveDate::from_ymd_opt(2026, 5, 21).unwrap());
687 assert_eq!(ctx.run_id.as_deref(), Some("r-abc123"));
688 }
689
690 #[test]
693 fn resolved_prefix_prefers_cloud_prefix_over_path() {
694 let dest = crate::config::DestinationConfig {
695 destination_type: crate::config::DestinationType::S3,
696 prefix: Some("exports/2026-05-21/orders/".into()),
697 path: Some("/scratch".into()),
698 ..Default::default()
699 };
700 assert_eq!(
701 resolved_prefix_for_display(&dest),
702 "exports/2026-05-21/orders/",
703 );
704 }
705
706 #[test]
707 fn resolved_prefix_falls_back_to_path_when_prefix_missing() {
708 let dest = crate::config::DestinationConfig {
709 destination_type: crate::config::DestinationType::Local,
710 prefix: None,
711 path: Some("/data/out".into()),
712 ..Default::default()
713 };
714 assert_eq!(resolved_prefix_for_display(&dest), "/data/out");
715 }
716
717 use crate::pipeline::ManifestVerificationFailure as VFailure;
720
721 fn read_error_verdict() -> ManifestVerification {
725 ManifestVerification {
726 legacy_run: false,
727 failures: vec![VFailure::ManifestReadError {
728 detail: "permission denied".into(),
729 }],
730 ..ManifestVerification::legacy()
731 }
732 }
733
734 #[test]
735 fn exit_gate_counts_manifest_read_error_as_failure() {
736 assert!(verdict_fails_exit(&read_error_verdict()));
737 }
738
739 #[test]
740 fn exit_gate_keeps_legacy_run_at_zero() {
741 assert!(!verdict_fails_exit(&ManifestVerification::legacy()));
744 }
745
746 #[test]
747 fn exit_gate_keeps_advisory_untracked_at_zero() {
748 let v = ManifestVerification {
749 manifest_found: true,
750 legacy_run: false,
751 passed: true,
752 parts_verified: 1,
753 failures: vec![VFailure::UntrackedObject {
754 key: "stray.parquet".into(),
755 size_bytes: 9,
756 }],
757 ..ManifestVerification::legacy()
758 };
759 assert!(!verdict_fails_exit(&v));
760 }
761
762 #[test]
763 fn exit_gate_counts_fatal_failure_on_found_manifest() {
764 let v = ManifestVerification {
765 manifest_found: true,
766 legacy_run: false,
767 failures: vec![VFailure::PartMissing {
768 part_id: 1,
769 path: "part-000001.parquet".into(),
770 }],
771 ..ManifestVerification::legacy()
772 };
773 assert!(verdict_fails_exit(&v));
774 }
775
776 use crate::manifest::{
780 MANIFEST_VERSION, ManifestDestination, ManifestPart, ManifestSource, ManifestStatus,
781 PartStatus, RunManifest,
782 };
783
784 fn success_manifest(parts: Vec<ManifestPart>) -> RunManifest {
785 let row_count: i64 = parts.iter().map(|p| p.rows).sum();
786 let part_count = parts.len() as u32;
787 RunManifest {
788 mode: "batch".to_string(),
789 manifest_version: MANIFEST_VERSION,
790 run_id: "r-validate-cmd".into(),
791 export_name: "orders".into(),
792 started_at: "2026-06-09T12:00:00Z".into(),
793 finished_at: "2026-06-09T12:01:00Z".into(),
794 status: ManifestStatus::Success,
795 source: ManifestSource {
796 engine: "postgres".into(),
797 schema: Some("public".into()),
798 table: Some("orders".into()),
799 extraction: None,
800 },
801 destination: ManifestDestination {
802 kind: "local".into(),
803 uri: "file:///tmp/out".into(),
804 },
805 format: "parquet".into(),
806 compression: "zstd".into(),
807 schema_fingerprint: "xxh3:0123456789abcdef".into(),
808 row_count,
809 part_count,
810 parts,
811 column_checksums: None,
812 checksum_key_column: None,
813 }
814 }
815
816 fn stage_dataset(prefix: &Path, m: &RunManifest) {
819 std::fs::create_dir_all(prefix).unwrap();
820 let dest = crate::destination::create_destination(&crate::config::DestinationConfig {
821 destination_type: crate::config::DestinationType::Local,
822 path: Some(prefix.to_string_lossy().into_owned()),
823 ..Default::default()
824 })
825 .unwrap();
826 crate::pipeline::write_manifest(&*dest, m).unwrap();
827 }
828
829 fn write_cfg(dir: &Path, prefix: &Path) -> std::path::PathBuf {
832 let cfg = dir.join("rivet.yaml");
833 let yaml = format!(
834 "source:\n type: postgres\n url: postgresql://nobody@localhost/nope\nexports:\n - name: orders\n query: \"SELECT 1\"\n mode: full\n format: parquet\n destination:\n type: local\n path: \"{}\"\n",
835 prefix.to_string_lossy()
836 );
837 std::fs::write(&cfg, yaml).unwrap();
838 cfg
839 }
840
841 #[cfg(unix)]
846 #[test]
847 fn unreadable_manifest_fails_the_command() {
848 use std::os::unix::fs::PermissionsExt;
849
850 let dir = tempfile::tempdir().unwrap();
851 let prefix = dir.path().join("out");
852 stage_dataset(&prefix, &success_manifest(Vec::new()));
853 let cfg = write_cfg(dir.path(), &prefix);
854
855 let manifest_path = prefix.join(crate::manifest::MANIFEST_FILENAME);
856 std::fs::set_permissions(&manifest_path, std::fs::Permissions::from_mode(0o000)).unwrap();
857 if std::fs::read(&manifest_path).is_ok() {
858 eprintln!("skipping unreadable_manifest_fails_the_command: running as root");
860 return;
861 }
862
863 let report = dir.path().join("report.json");
864 let err = run_validate_command(
865 cfg.to_str().unwrap(),
866 Some("orders"),
867 ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
868 ValidateTarget::default(),
869 )
870 .expect_err("an unreadable manifest is an explicit failure, not exit 0");
871 assert!(
872 format!("{err:#}").contains("1 export(s) failed verification"),
873 "got: {err:#}"
874 );
875
876 let json: serde_json::Value =
879 serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
880 let verification = &json["exports"][0]["verification"];
881 assert_eq!(verification["manifest_found"], false);
882 assert_eq!(verification["legacy_run"], false);
883 assert_eq!(verification["failures"][0]["kind"], "manifest_read_error");
884 }
885
886 #[test]
887 fn untracked_surplus_alone_keeps_exit_zero() {
888 let dir = tempfile::tempdir().unwrap();
892 let prefix = dir.path().join("out");
893 stage_dataset(&prefix, &success_manifest(Vec::new()));
894 std::fs::write(prefix.join("rogue.parquet"), b"XX").unwrap();
895 let cfg = write_cfg(dir.path(), &prefix);
896
897 let report = dir.path().join("report.json");
898 run_validate_command(
899 cfg.to_str().unwrap(),
900 Some("orders"),
901 ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
902 ValidateTarget::default(),
903 )
904 .expect("advisory untracked surplus must not flip the exit code");
905
906 let json: serde_json::Value =
907 serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
908 let verification = &json["exports"][0]["verification"];
909 assert_eq!(verification["passed"], true);
910 assert_eq!(verification["failures"][0]["kind"], "untracked_object");
913
914 let warnings = json["warnings"].as_array().expect("warnings array present");
918 assert_eq!(warnings.len(), 1, "the untracked surplus is one warning");
919 assert_eq!(warnings[0]["export_name"], "orders");
920 assert_eq!(warnings[0]["warning"]["kind"], "untracked_object");
921 assert_eq!(warnings[0]["warning"]["key"], "rogue.parquet");
922 }
923
924 #[test]
925 fn json_warnings_array_is_empty_when_no_advisory_failures() {
926 let dir = tempfile::tempdir().unwrap();
929 let prefix = dir.path().join("out");
930 stage_dataset(&prefix, &success_manifest(Vec::new()));
931 let cfg = write_cfg(dir.path(), &prefix);
932
933 let report = dir.path().join("report.json");
934 run_validate_command(
935 cfg.to_str().unwrap(),
936 Some("orders"),
937 ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
938 ValidateTarget::default(),
939 )
940 .expect("a clean dataset must pass");
941
942 let json: serde_json::Value =
943 serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
944 assert_eq!(
945 json["warnings"]
946 .as_array()
947 .expect("warnings array present")
948 .len(),
949 0,
950 "no surplus → no warnings"
951 );
952 }
953
954 #[test]
955 fn multiplex_cdc_validates_each_table_sub_prefix() {
956 let dir = tempfile::tempdir().unwrap();
962 let base = dir.path().join("cdc");
963 stage_dataset(&base.join("alpha"), &success_manifest(Vec::new()));
964 stage_dataset(&base.join("beta"), &success_manifest(Vec::new()));
965 let cfg = write_multiplex_cfg(dir.path(), &base);
966
967 let report = dir.path().join("report.json");
968 run_validate_command(
969 cfg.to_str().unwrap(),
970 None,
971 ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
972 ValidateTarget {
976 depth: ValidateDepth::Sample,
977 ..Default::default()
978 },
979 )
980 .expect("both table sub-prefixes are complete — the stream must validate");
981
982 let json: serde_json::Value =
983 serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
984 let names: Vec<&str> = json["exports"]
985 .as_array()
986 .unwrap()
987 .iter()
988 .map(|e| e["export_name"].as_str().unwrap())
989 .collect();
990 assert_eq!(names, vec!["cdc/alpha", "cdc/beta"], "per-table descent");
993 for e in json["exports"].as_array().unwrap() {
994 assert_eq!(e["verification"]["passed"], true, "each table passes");
995 }
996 }
997
998 fn write_multiplex_cfg(dir: &Path, base: &Path) -> std::path::PathBuf {
999 let cfg = dir.join("rivet-cdc.yaml");
1000 let yaml = format!(
1001 "source:\n type: mysql\n url: mysql://nobody@localhost/nope\nexports:\n - name: cdc\n tables: [alpha, beta]\n mode: cdc\n format: parquet\n cdc:\n server_id: 1\n destination:\n type: local\n path: \"{}\"\n",
1002 base.to_string_lossy()
1003 );
1004 std::fs::write(&cfg, yaml).unwrap();
1005 cfg
1006 }
1007
1008 #[test]
1009 fn single_table_cdc_certifies_snapshot_and_hides_its_untracked() {
1010 let dir = tempfile::tempdir().unwrap();
1016 let base = dir.path().join("cdc");
1017 stage_dataset(&base, &success_manifest(Vec::new()));
1018 stage_dataset(&base.join("snapshot"), &success_manifest(Vec::new()));
1019 let cfg = write_single_cdc_cfg(dir.path(), &base);
1020
1021 let report = dir.path().join("report.json");
1022 run_validate_command(
1023 cfg.to_str().unwrap(),
1024 None,
1025 ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
1026 ValidateTarget {
1027 depth: ValidateDepth::Sample,
1028 ..Default::default()
1029 },
1030 )
1031 .expect("snapshot + change datasets are complete");
1032
1033 let json: serde_json::Value =
1034 serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
1035 let exports = json["exports"].as_array().unwrap();
1036 let names: Vec<&str> = exports
1037 .iter()
1038 .map(|e| e["export_name"].as_str().unwrap())
1039 .collect();
1040 assert!(
1042 names.contains(&"cdc/snapshot"),
1043 "snapshot certified: {names:?}"
1044 );
1045 assert!(names.contains(&"cdc"), "change-prefix verdict: {names:?}");
1046 let change = exports.iter().find(|e| e["export_name"] == "cdc").unwrap();
1048 let failures = change["verification"]["failures"].as_array().unwrap();
1049 assert!(
1050 failures.iter().all(|f| f["kind"] != "untracked_object"),
1051 "snapshot files must not read as untracked surplus: {failures:?}"
1052 );
1053 }
1054
1055 fn write_single_cdc_cfg(dir: &Path, base: &Path) -> std::path::PathBuf {
1056 let cfg = dir.join("rivet-single-cdc.yaml");
1057 let yaml = format!(
1058 "source:\n type: mysql\n url: mysql://nobody@localhost/nope\nexports:\n - name: cdc\n table: t\n mode: cdc\n format: parquet\n cdc:\n initial: snapshot\n server_id: 1\n checkpoint: ./ck\n destination:\n type: local\n path: \"{}\"\n",
1059 base.to_string_lossy()
1060 );
1061 std::fs::write(&cfg, yaml).unwrap();
1062 cfg
1063 }
1064
1065 #[test]
1066 fn missing_part_fails_the_command() {
1067 let dir = tempfile::tempdir().unwrap();
1068 let prefix = dir.path().join("out");
1069 let m = success_manifest(vec![ManifestPart {
1070 part_id: 1,
1071 path: "part-000001.parquet".into(),
1072 rows: 10,
1073 size_bytes: 4,
1074 content_fingerprint: "xxh3:1111111111111111".into(),
1075 content_md5: String::new(),
1076 status: PartStatus::Committed,
1077 }]);
1078 stage_dataset(&prefix, &m); let cfg = write_cfg(dir.path(), &prefix);
1080
1081 let err = run_validate_command(
1082 cfg.to_str().unwrap(),
1083 Some("orders"),
1084 ValidateOutputFormat::Json(None),
1085 ValidateTarget::default(),
1086 )
1087 .expect_err("a missing committed part must fail verification");
1088 assert!(
1089 format!("{err:#}").contains("1 export(s) failed verification"),
1090 "got: {err:#}"
1091 );
1092 }
1093
1094 #[test]
1099 fn prefix_override_with_real_manifest_passes() {
1100 let dir = tempfile::tempdir().unwrap();
1101 let prefix = dir.path().join("out");
1102 stage_dataset(&prefix, &success_manifest(Vec::new()));
1103 let cfg = write_cfg(dir.path(), &prefix);
1104
1105 run_validate_command(
1106 cfg.to_str().unwrap(),
1107 Some("orders"),
1108 ValidateOutputFormat::Json(None),
1109 ValidateTarget {
1110 prefix_override: Some(prefix.to_string_lossy().into_owned()),
1111 ..Default::default()
1112 },
1113 )
1114 .expect("a real dataset under a pinned --prefix must pass");
1115 }
1116
1117 #[test]
1122 fn prefix_override_at_absent_manifest_fails() {
1123 let dir = tempfile::tempdir().unwrap();
1124 let cfg_prefix = dir.path().join("cfg_dest");
1127 std::fs::create_dir_all(&cfg_prefix).unwrap();
1128 let cfg = write_cfg(dir.path(), &cfg_prefix);
1129 let empty_prefix = dir.path().join("never_written");
1130 std::fs::create_dir_all(&empty_prefix).unwrap();
1131
1132 let report = dir.path().join("report.json");
1133 let err = run_validate_command(
1134 cfg.to_str().unwrap(),
1135 Some("orders"),
1136 ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
1137 ValidateTarget {
1138 prefix_override: Some(empty_prefix.to_string_lossy().into_owned()),
1139 ..Default::default()
1140 },
1141 )
1142 .expect_err("a never-written prefix pinned via --prefix must fail, not legacy-pass");
1143 assert!(
1144 format!("{err:#}").contains("1 export(s) failed verification"),
1145 "got: {err:#}"
1146 );
1147
1148 let json: serde_json::Value =
1151 serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
1152 let verification = &json["exports"][0]["verification"];
1153 assert_eq!(verification["manifest_found"], false);
1154 assert_eq!(verification["legacy_run"], false);
1155 assert_eq!(
1156 verification["failures"][0]["kind"],
1157 "manifest_required_but_absent"
1158 );
1159 }
1160
1161 #[test]
1165 fn absent_manifest_without_prefix_override_stays_legacy_pass() {
1166 let dir = tempfile::tempdir().unwrap();
1167 let prefix = dir.path().join("out");
1168 std::fs::create_dir_all(&prefix).unwrap(); let cfg = write_cfg(dir.path(), &prefix);
1170
1171 run_validate_command(
1172 cfg.to_str().unwrap(),
1173 Some("orders"),
1174 ValidateOutputFormat::Json(None),
1175 ValidateTarget::default(), )
1177 .expect("an absent manifest with no pinned --prefix is a legacy pass (exit 0)");
1178 }
1179
1180 fn stage_dataset_form_b_would_fail(prefix: &Path) {
1189 std::fs::create_dir_all(prefix).unwrap();
1190 let part_body: &[u8] = b"AAAA";
1193 std::fs::write(prefix.join("part-000001.parquet"), part_body).unwrap();
1194
1195 let mut m = success_manifest(vec![ManifestPart {
1196 part_id: 1,
1197 path: "part-000001.parquet".into(),
1198 rows: 1,
1199 size_bytes: part_body.len() as u64,
1200 content_fingerprint: "xxh3:1111111111111111".into(),
1201 content_md5: String::new(),
1202 status: PartStatus::Committed,
1203 }]);
1204 m.column_checksums = Some(vec![crate::manifest::ColumnChecksum {
1207 name: "id".into(),
1208 checksum: "0".into(),
1209 }]);
1210 stage_dataset(prefix, &m);
1211 }
1212
1213 #[test]
1214 fn sample_depth_does_not_run_form_b() {
1215 let dir = tempfile::tempdir().unwrap();
1219 let prefix = dir.path().join("out");
1220 stage_dataset_form_b_would_fail(&prefix);
1221 let cfg = write_cfg(dir.path(), &prefix);
1222
1223 let report = dir.path().join("report.json");
1224 run_validate_command(
1225 cfg.to_str().unwrap(),
1226 Some("orders"),
1227 ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
1228 ValidateTarget {
1229 depth: ValidateDepth::Sample,
1230 ..Default::default()
1231 },
1232 )
1233 .expect("sample depth skips Form B, so a non-Parquet part still passes");
1234
1235 let json: serde_json::Value =
1236 serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
1237 let verification = &json["exports"][0]["verification"];
1238 assert_eq!(verification["passed"], true);
1239 assert_eq!(verification["parts_verified"], 1, "sample reconciles parts");
1240 assert_eq!(verification["depth_level"], "sample");
1241 }
1242
1243 #[test]
1244 fn full_depth_runs_form_b() {
1245 let dir = tempfile::tempdir().unwrap();
1250 let prefix = dir.path().join("out");
1251 stage_dataset_form_b_would_fail(&prefix);
1252 let cfg = write_cfg(dir.path(), &prefix);
1253
1254 let err = run_validate_command(
1255 cfg.to_str().unwrap(),
1256 Some("orders"),
1257 ValidateOutputFormat::Json(None),
1258 ValidateTarget {
1259 depth: ValidateDepth::Full,
1260 ..Default::default()
1261 },
1262 )
1263 .expect_err("full depth runs Form B, which fails on a non-Parquet part");
1264 assert!(
1265 format!("{err:#}").contains("1 export(s) failed verification"),
1266 "got: {err:#}"
1267 );
1268 }
1269
1270 #[test]
1271 fn json_report_carries_failure_code_and_depth_level() {
1272 let dir = tempfile::tempdir().unwrap();
1276 let prefix = dir.path().join("out");
1277 let m = success_manifest(vec![ManifestPart {
1278 part_id: 1,
1279 path: "part-000001.parquet".into(),
1280 rows: 10,
1281 size_bytes: 4,
1282 content_fingerprint: "xxh3:1111111111111111".into(),
1283 content_md5: String::new(),
1284 status: PartStatus::Committed,
1285 }]);
1286 stage_dataset(&prefix, &m); let cfg = write_cfg(dir.path(), &prefix);
1288
1289 let report = dir.path().join("report.json");
1290 let _ = run_validate_command(
1291 cfg.to_str().unwrap(),
1292 Some("orders"),
1293 ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
1294 ValidateTarget {
1295 depth: ValidateDepth::Sample,
1296 ..Default::default()
1297 },
1298 )
1299 .expect_err("a missing part fails the command");
1300
1301 let json: serde_json::Value =
1302 serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
1303 let verification = &json["exports"][0]["verification"];
1304 assert_eq!(verification["depth_level"], "sample");
1306 let failure = &verification["failures"][0];
1308 assert_eq!(failure["kind"], "part_missing");
1309 assert_eq!(failure["code"], "RIVET_VERIFY_PART_MISSING");
1310 assert_eq!(failure["part_id"], 1);
1313 }
1314
1315 #[test]
1316 fn json_warning_entry_also_carries_its_code() {
1317 let dir = tempfile::tempdir().unwrap();
1320 let prefix = dir.path().join("out");
1321 stage_dataset(&prefix, &success_manifest(Vec::new()));
1322 std::fs::write(prefix.join("rogue.parquet"), b"XX").unwrap();
1323 let cfg = write_cfg(dir.path(), &prefix);
1324
1325 let report = dir.path().join("report.json");
1326 run_validate_command(
1327 cfg.to_str().unwrap(),
1328 Some("orders"),
1329 ValidateOutputFormat::Json(Some(report.to_string_lossy().into_owned())),
1330 ValidateTarget::default(),
1331 )
1332 .expect("advisory untracked surplus must not flip the exit code");
1333
1334 let json: serde_json::Value =
1335 serde_json::from_str(&std::fs::read_to_string(&report).unwrap()).unwrap();
1336 let warning = &json["warnings"][0]["warning"];
1337 assert_eq!(warning["kind"], "untracked_object");
1338 assert_eq!(warning["code"], "RIVET_VERIFY_UNTRACKED_OBJECT");
1339 assert_eq!(json["exports"][0]["verification"]["depth_level"], "full");
1341 }
1342}