Skip to main content

rivet/pipeline/
validate_manifest.rs

1//! **Layer: Observability**
2//!
3//! Manifest-aware verification for `--validate` (ADR-0012 §M5 / §M6,
4//! constrained by the ADR-0013 trust-flag contract that says: no new flags;
5//! manifest-aware checks live under the existing `--validate`).
6//!
7//! Although this module reads from a `Destination` (technically L2 surface),
8//! it makes **no execution decisions**: it does not write data, advance
9//! cursors, mutate state, or change the pipeline path.  Its only output is
10//! a structured `ManifestVerification` verdict the run report renders.
11//! Per ADR-0003, that places it firmly in L4 Observability — the
12//! destination read surface is just the carrier.
13//!
14//! Inputs (read-only):
15//! - the destination's `manifest.json` body
16//! - the destination's `_SUCCESS` body, if present
17//! - the listing of every object under the destination prefix
18//!
19//! Outputs:
20//! - [`ManifestVerification`] — a structured verdict the run report renders
21//!   into the operator-facing "Verdicts" section.
22//!
23//! Out of scope here:
24//! - per-file row-count check (that runs *during* the export, against the
25//!   local temp file before upload — see `pipeline::validate::validate_output`).
26//! - source-side reconciliation (lives in [`pipeline::reconcile_cmd`] and
27//!   is what `--reconcile` adds on top of this).
28//! - re-fingerprinting parts (`--validate --deep`, future).
29//!
30//! Failure modes are explicit: each check produces a `Failure` enum variant
31//! that is rendered verbatim in `summary.json` so an Airflow / CI consumer
32//! can branch on the kind, not parse strings.
33
34use serde::{Deserialize, Serialize};
35
36use crate::destination::Destination;
37use crate::error::Result;
38use crate::manifest::{
39    MANIFEST_FILENAME, RunManifest, SUCCESS_FILENAME, join_key, parse_success_marker,
40    success_marker_body,
41};
42use crate::pipeline::manifest_reconcile::{PartPresence, reconcile_manifest_against_listing};
43
44/// Upper bound on a destination control artifact (`manifest.json`) the read
45/// path will materialise into memory.  A `manifest.json` is metadata — a few
46/// KB to low single-digit MB even for very large datasets — so 64 MiB is far
47/// above any legitimate body while still bounding the blast radius.
48///
49/// Security (V21, CWE-400): the manifest readers `head()` an object then read
50/// its full body into a `Vec<u8>`.  An attacker who can write the destination
51/// prefix (a shared bucket prefix, a world-writable export dir) can plant a
52/// multi-GB `manifest.json`; an unbounded read would OOM the next `--resume`,
53/// `--validate`, or `rivet repair`.  [`read_capped`] consults the size the
54/// `head()` already reports and bails before the read when it exceeds this cap.
55pub(crate) const MANIFEST_MAX_BYTES: u64 = 64 * 1024 * 1024;
56
57/// How deep a `rivet validate` pass goes — a graded verify layer over the
58/// same checks, letting an operator trade thoroughness for latency / cost.
59///
60/// The variants are a strict superset chain: `Light ⊂ Sample ⊂ Full`.  Each
61/// level runs every check the level below it does, plus more.  Defined here
62/// (the pipeline layer) and re-exported for the CLI grammar so the **same**
63/// enum gates the checks in [`verify_at_destination`] and parses on the
64/// `--depth` flag — no CLI→pipeline back-dependency.
65///
66/// - **Light**: manifest read + self-consistency + `_SUCCESS` only.  Skips the
67///   `list_prefix` reconcile (no per-part presence/size/checksum) and the
68///   untracked-surplus scan, leaving `parts_verified = 0`.  One `head` + one
69///   `read` of `manifest.json` and `_SUCCESS` — a fast "is this prefix a
70///   complete, marked run?" poll with no prefix listing.
71/// - **Sample**: everything Light does **plus** the part reconcile and
72///   untracked surplus (one `list_prefix`).  This is the pre-graded behaviour
73///   minus the Form B value re-read — full structural verification with no
74///   part downloads.
75/// - **Full** (default): everything Sample does **plus** the Form B value-
76///   checksum re-read (re-reads parts, re-derives per-column checksums).  The
77///   most thorough and the only level that downloads part bodies.  Equivalent
78///   to the pre-graded behaviour, so existing callers are unchanged.
79#[derive(clap::ValueEnum, Debug, Clone, Copy, PartialEq, Eq, Default)]
80pub enum ValidateDepth {
81    /// Manifest read + self-consistency + `_SUCCESS` only (no prefix listing).
82    Light,
83    /// Light + part reconcile + untracked surplus (one `list_prefix`).
84    Sample,
85    /// Sample + the Form B value-checksum re-read (downloads parts).
86    #[default]
87    Full,
88}
89
90impl ValidateDepth {
91    /// True iff this level runs the `list_prefix` reconcile (part presence,
92    /// size, checksum) and the untracked-surplus scan — i.e. anything above
93    /// `Light`.  The single predicate the section-3/5 depth gating keys off.
94    fn runs_part_reconcile(self) -> bool {
95        !matches!(self, ValidateDepth::Light)
96    }
97
98    /// True iff this level downloads part *bodies* — the CDC `__pos` continuity
99    /// check and the Form-B value-checksum re-read, both `Full`-only (the
100    /// `part_reconcile` level above only reads listing metadata). The single
101    /// predicate the section-3/5 part-download gating keys off, parallel to
102    /// [`Self::runs_part_reconcile`] — so adding a depth level edits the enum,
103    /// not the call sites in `validate_cmd`.
104    pub(crate) fn runs_part_download(self) -> bool {
105        matches!(self, ValidateDepth::Full)
106    }
107
108    /// Stable operator-/wire-facing label for the depth a verdict was produced
109    /// at, surfaced in `summary.json` (`depth_level`) and `rivet validate`
110    /// output.
111    pub fn label(self) -> &'static str {
112        match self {
113            ValidateDepth::Light => "light",
114            ValidateDepth::Sample => "sample",
115            ValidateDepth::Full => "full",
116        }
117    }
118}
119
120/// Read `key` into memory only if its `head()`-reported size is within
121/// `max_bytes`; otherwise bail without reading a single byte.
122///
123/// The single enforcement point for the V21 (CWE-400) manifest-read cap shared
124/// by the three control-artifact readers (`--resume` M8 preamble, `--validate`,
125/// `rivet repair`).  Each previously did `head()` then an uncapped `read()`,
126/// discarding the size `head()` already returned; routing through here closes
127/// that gap in one place.
128///
129/// Behaviour:
130/// - object absent (`head` → `None`): `Err` — callers invoke this only after
131///   establishing the object exists, so an absent object here is a hard error,
132///   not the benign "no manifest / legacy prefix" case (which the callers
133///   detect with their own `head()` first).
134/// - oversized (`size_bytes > max_bytes`): `Err` naming the cap, **before** any
135///   body is materialised.
136/// - otherwise: the full body via [`Destination::read`].
137pub(crate) fn read_capped(dest: &dyn Destination, key: &str, max_bytes: u64) -> Result<Vec<u8>> {
138    match dest.head(key)? {
139        None => anyhow::bail!("'{key}' not found at the destination"),
140        Some(meta) => {
141            if meta.size_bytes > max_bytes {
142                anyhow::bail!(
143                    "'{key}' is {} bytes, exceeding the {max_bytes}-byte control-artifact \
144                     read cap — refusing to load it into memory (possible tampering)",
145                    meta.size_bytes
146                );
147            }
148            dest.read(key)
149        }
150    }
151}
152
153/// Outcome of a single `--validate` pass over a destination prefix.
154///
155/// Stable enough to be embedded in `summary.json` directly (see
156/// `pipeline::report::ValidationOutcome`).  Forward-compat: consumers MUST
157/// ignore unknown fields (no `deny_unknown_fields`).
158#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
159pub struct ManifestVerification {
160    /// True iff a `manifest.json` was found at the destination and parsed.
161    /// `false` triggers ADR-0012 M6 fallback (legacy run); higher-level
162    /// check results below are then "skipped" rather than "passed".
163    pub manifest_found: bool,
164    /// Mirrors ADR-0012 M6's required `legacy_run` operator-facing label.
165    pub legacy_run: bool,
166    /// Manifest parts whose presence and recorded `size_bytes` were
167    /// confirmed at the destination.  0 when no manifest was found.
168    pub parts_verified: usize,
169    /// Subset of `parts_verified` whose **content** was confirmed via an MD5
170    /// the store surfaced in its listing (no download) — the rest are size-only.
171    /// Lets `passed: true` say how much of the dataset was content-checked
172    /// rather than implying all of it was.  `#[serde(default)]` for back-compat.
173    #[serde(default)]
174    pub parts_md5_verified: usize,
175    /// Manifest parts that were declared `committed` but not actually
176    /// present, present at a different size, or otherwise mismatched.
177    pub parts_failed: usize,
178    /// True iff `_SUCCESS` exists at the destination AND its body matches
179    /// the fingerprint of the bytes we read for `manifest.json`.  An
180    /// existing `_SUCCESS` whose body diverges from the manifest is itself
181    /// an integrity failure — surfaced via `failures`.
182    pub success_marker_consistent: bool,
183    /// Self-consistency of the manifest (`row_count`, `part_count`,
184    /// duplicate `part_id`s).  Skipped when `manifest_found = false`.
185    pub manifest_self_consistent: bool,
186    /// Final verdict, **derived** (not hand-maintained) — `manifest_found` and
187    /// no *fatal* failure ([`Failure::is_fatal`]).  Stored so it stays in the
188    /// `summary.json` contract, but computed in one place
189    /// ([`ManifestVerification::recompute_passed`]) so a new failure variant is
190    /// fatal by default rather than relying on every site to flip a bool.
191    pub passed: bool,
192    /// Per-failure detail.  May be non-empty with `passed = true` for advisory
193    /// (non-fatal) failures like [`Failure::UntrackedObject`].  Stable variant
194    /// set; new variants land under a new manifest version per ADR-0012.
195    pub failures: Vec<Failure>,
196    /// The graded depth this verdict was produced at: `"light"`, `"sample"`,
197    /// or `"full"` (see [`ValidateDepth`]).  Lets a consumer of `summary.json`
198    /// tell **how much** was actually checked — a `passed: true` at `"light"`
199    /// asserts far less than at `"full"` (no part presence was verified).
200    /// `#[serde(default)]` (→ `"full"`) for back-compat: pre-graded verdicts
201    /// always ran the full pass.
202    #[serde(default = "default_depth_level")]
203    pub depth_level: String,
204}
205
206/// `serde(default)` for [`ManifestVerification::depth_level`]: a verdict that
207/// predates the graded layer always ran the full pass, so an absent field
208/// deserializes to `"full"`.
209fn default_depth_level() -> String {
210    ValidateDepth::Full.label().to_string()
211}
212
213#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
214#[serde(tag = "kind", rename_all = "snake_case")]
215pub enum Failure {
216    /// Manifest declared a part that does not exist at the destination.
217    PartMissing { part_id: u32, path: String },
218    /// Manifest declared a part whose actual size differs from `size_bytes`.
219    PartSizeMismatch {
220        part_id: u32,
221        path: String,
222        expected: u64,
223        actual: u64,
224    },
225    /// Part present at the recorded size but its content MD5 (from the store's
226    /// listing metadata) differs from the manifest's — transit / at-rest
227    /// corruption, caught with no download.
228    PartChecksumMismatch {
229        part_id: u32,
230        path: String,
231        expected: String,
232        actual: String,
233    },
234    /// `_SUCCESS` exists but its body is malformed (not `xxh3:<16-hex>` after
235    /// trim).  ADR-0012 M2 — orchestrators rely on this format being strict.
236    SuccessMarkerMalformed { body_preview: String },
237    /// `_SUCCESS` body parsed but does not match `xxh3(manifest.json bytes)`.
238    /// Two legitimate sources: (a) someone overwrote `_SUCCESS` after the
239    /// manifest was rewritten — orchestrator bug; (b) the manifest was
240    /// edited in place after the run — operator bug.  Either way the
241    /// manifest is no longer trustworthy.
242    SuccessMarkerStale {
243        marker_fingerprint: String,
244        manifest_fingerprint: String,
245    },
246    /// `RunManifest::validate_self_consistency` rejected the manifest.
247    /// Usually a writer bug (declared row_count != sum of committed parts'
248    /// rows); blocks the rest of the verification because the manifest
249    /// itself is unreliable.
250    ManifestSelfInconsistent { detail: String },
251    /// Reading `manifest.json` returned an I/O error other than "absent".
252    ManifestReadError { detail: String },
253    /// Reading `_SUCCESS` returned an I/O error other than "absent".
254    SuccessMarkerReadError { detail: String },
255    /// Listing the destination prefix returned an I/O error.  Reduces the
256    /// untracked-parts check (M5 surplus) to a no-op for this run.
257    ListPrefixError { detail: String },
258    /// A file is present at the destination prefix but no manifest entry
259    /// references it.  M9-adjacent: `--validate` only flags it; quarantine
260    /// belongs to `--resume`.
261    UntrackedObject { key: String, size_bytes: u64 },
262    /// The export declared `verify: content` but some parts could only be
263    /// size-verified (no comparable content checksum from the store) — the
264    /// declared integrity contract was not met.
265    ContentVerificationUnmet { size_only: usize, total: usize },
266    /// A manifest was *required* at this prefix (the operator pinned a literal
267    /// `--prefix`, asserting a real dataset lives here) but none was found.
268    /// Without this, an absent manifest at an operator-pinned prefix maps to
269    /// the M6 legacy-run label and exits 0 — indistinguishable from a verified
270    /// run, so a CI gate `rivet validate && deploy` sails past a destination
271    /// that was never written.  Fatal: a required-but-missing manifest is a
272    /// refusal reason, not a "cannot certify" advisory.
273    ManifestRequiredButAbsent { prefix: String },
274}
275
276impl Failure {
277    /// Whether this failure invalidates the dataset (flips `passed` to false).
278    ///
279    /// Every variant is fatal **except** [`Failure::UntrackedObject`]: surplus
280    /// objects are an audit signal whose cleanup is `--resume`'s job (ADR-0012
281    /// M9), not a corruption of the manifest-listed parts.  New variants are
282    /// fatal by default — opt out here explicitly, so a forgotten case fails
283    /// closed (safe) rather than silently passing.
284    pub fn is_fatal(&self) -> bool {
285        !matches!(self, Failure::UntrackedObject { .. })
286    }
287
288    /// Stable `RIVET_VERIFY_*` error code for this failure variant.
289    ///
290    /// One code per variant, intended for orchestrators / CI to branch on
291    /// without parsing the human `Display` string or the per-variant JSON
292    /// fields.  The code is part of the wire contract: it is emitted next to
293    /// `kind` in the JSON report and prefixed in brackets on each pretty line.
294    /// Codes are append-only — never renamed once shipped (a renamed code is a
295    /// silent break for any consumer keying off it).
296    pub fn error_code(&self) -> &'static str {
297        match self {
298            Failure::PartMissing { .. } => "RIVET_VERIFY_PART_MISSING",
299            Failure::PartSizeMismatch { .. } => "RIVET_VERIFY_PART_SIZE_MISMATCH",
300            Failure::PartChecksumMismatch { .. } => "RIVET_VERIFY_PART_CHECKSUM_MISMATCH",
301            Failure::SuccessMarkerMalformed { .. } => "RIVET_VERIFY_SUCCESS_MALFORMED",
302            Failure::SuccessMarkerStale { .. } => "RIVET_VERIFY_SUCCESS_STALE",
303            Failure::ManifestSelfInconsistent { .. } => "RIVET_VERIFY_MANIFEST_INCONSISTENT",
304            Failure::ManifestReadError { .. } => "RIVET_VERIFY_MANIFEST_READ_ERROR",
305            Failure::SuccessMarkerReadError { .. } => "RIVET_VERIFY_SUCCESS_READ_ERROR",
306            Failure::ListPrefixError { .. } => "RIVET_VERIFY_LIST_ERROR",
307            Failure::UntrackedObject { .. } => "RIVET_VERIFY_UNTRACKED_OBJECT",
308            Failure::ContentVerificationUnmet { .. } => "RIVET_VERIFY_CONTENT_UNMET",
309            Failure::ManifestRequiredButAbsent { .. } => "RIVET_VERIFY_MANIFEST_REQUIRED",
310        }
311    }
312}
313
314impl std::fmt::Display for Failure {
315    /// One operator-facing line per failure variant.  Used by:
316    /// - `pipeline::report::render_markdown` (summary.md "failure:" lines)
317    /// - `pipeline::validate_cmd::render_pretty` (`rivet validate` stdout)
318    /// - any future consumer that wants a human-readable failure label
319    ///
320    /// The wire format (`failures[].kind` + per-variant fields) lives in
321    /// the `Serialize` derive above and is the contract Airflow / CI
322    /// consumers branch on.  This `Display` impl is for humans only.
323    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
324        match self {
325            Failure::PartMissing { part_id, path } => {
326                write!(f, "part {} missing at {}", part_id, path)
327            }
328            Failure::PartSizeMismatch {
329                part_id,
330                path,
331                expected,
332                actual,
333            } => write!(
334                f,
335                "part {} size mismatch at {}: manifest {}, dest {}",
336                part_id, path, expected, actual
337            ),
338            Failure::PartChecksumMismatch {
339                part_id,
340                path,
341                expected,
342                actual,
343            } => write!(
344                f,
345                "part {} content mismatch at {}: manifest md5 {}, dest {}",
346                part_id, path, expected, actual
347            ),
348            Failure::SuccessMarkerMalformed { body_preview } => {
349                write!(f, "_SUCCESS body malformed: {body_preview:?}")
350            }
351            Failure::SuccessMarkerStale {
352                marker_fingerprint,
353                manifest_fingerprint,
354            } => write!(
355                f,
356                "_SUCCESS body {} != manifest fingerprint {} (stale marker)",
357                marker_fingerprint, manifest_fingerprint
358            ),
359            Failure::ManifestSelfInconsistent { detail } => {
360                write!(f, "manifest self-consistency: {detail}")
361            }
362            Failure::ManifestReadError { detail } => {
363                write!(f, "manifest read error: {detail}")
364            }
365            Failure::SuccessMarkerReadError { detail } => {
366                write!(f, "_SUCCESS read error: {detail}")
367            }
368            Failure::ListPrefixError { detail } => {
369                write!(f, "destination listing error: {detail}")
370            }
371            Failure::UntrackedObject { key, size_bytes } => {
372                write!(f, "untracked object: {} ({} bytes)", key, size_bytes)
373            }
374            Failure::ContentVerificationUnmet { size_only, total } => write!(
375                f,
376                "verify: content not met — {size_only} of {total} part(s) only \
377                 size-verified (no store checksum); lower max_file_size so parts \
378                 upload as a single PUT, or the backend exposes no checksum"
379            ),
380            Failure::ManifestRequiredButAbsent { prefix } => write!(
381                f,
382                "no manifest at {prefix}: a manifest was required here (operator \
383                 pinned --prefix) but none was found — this prefix was never \
384                 written, or the data was relocated. Run the export first, or \
385                 drop --prefix to validate the config-resolved destination."
386            ),
387        }
388    }
389}
390
391impl ManifestVerification {
392    /// Base verdict: nothing checked yet (no manifest, all counts zero, all
393    /// sub-checks false, `passed = false`).  Every constructor builds on this
394    /// and overrides only what differs, so a new field lands in **one** place
395    /// rather than several near-identical literals.
396    fn empty() -> Self {
397        Self {
398            manifest_found: false,
399            legacy_run: false,
400            parts_verified: 0,
401            parts_md5_verified: 0,
402            parts_failed: 0,
403            success_marker_consistent: false,
404            manifest_self_consistent: false,
405            passed: false,
406            failures: Vec::new(),
407            // Base level; `verify_at_destination` overwrites this with the
408            // depth it was actually called at before returning any verdict.
409            depth_level: default_depth_level(),
410        }
411    }
412
413    /// Recompute `passed` from the verdict's facts: a manifest was found and no
414    /// **fatal** failure was recorded (advisory failures like `UntrackedObject`
415    /// don't count).  The single source of truth — callers set failures and
416    /// call this once, rather than flipping `passed` by hand at every site.
417    fn recompute_passed(&mut self) {
418        self.passed = self.manifest_found && !self.failures.iter().any(Failure::is_fatal);
419    }
420
421    /// Apply the export's `verify` policy (ADR-0013 / review D).  When content
422    /// verification is required but some parts were only size-verified, record
423    /// a fatal [`Failure::ContentVerificationUnmet`] and re-derive `passed`.
424    /// Policy lives here (one place); the composers — run finalize and the
425    /// `rivet validate` command — just call it with their export's intent.
426    pub fn enforce_content_policy(&mut self, require_content: bool) {
427        if require_content && self.manifest_found {
428            let size_only = self.parts_verified.saturating_sub(self.parts_md5_verified);
429            if size_only > 0 {
430                self.failures.push(Failure::ContentVerificationUnmet {
431                    size_only,
432                    total: self.parts_verified,
433                });
434                self.recompute_passed();
435            }
436        }
437    }
438
439    /// Apply the "a manifest must exist here" policy (finding #20).  When the
440    /// operator pinned a literal `--prefix`, an absent manifest is no longer the
441    /// benign M6 legacy-run case — it almost always means the prefix was never
442    /// written (a misconfigured CI gate). Convert that exact verdict — no
443    /// manifest, no other failure (i.e. the [`ManifestVerification::legacy`]
444    /// shape) — into a fatal [`Failure::ManifestRequiredButAbsent`] so the exit
445    /// gate refuses it loudly instead of silently passing.
446    ///
447    /// Deliberately a no-op for every other shape: a real manifest (passed or
448    /// failed), or an absent manifest that already carries a `ManifestReadError`
449    /// / head failure, is left untouched — those are already classified.  Only
450    /// the "legacy / cannot certify" case is escalated, and only when required.
451    pub fn require_manifest_present(&mut self, prefix: &str) {
452        if !self.manifest_found && !self.has_failures() {
453            self.legacy_run = false;
454            self.failures.push(Failure::ManifestRequiredButAbsent {
455                prefix: prefix.to_string(),
456            });
457            self.recompute_passed();
458        }
459    }
460
461    /// Construct the M6 (legacy run) verdict for a destination that has no
462    /// manifest at all.  Caller composes this with the existing per-file
463    /// row-count check; together they form the legacy `--validate` result.
464    pub fn legacy() -> Self {
465        // `passed = false` is intentional — not "validation failed" but "this
466        // verifier cannot certify"; the caller layers per-file row counts on
467        // top and composes the final verdict.
468        Self {
469            legacy_run: true,
470            ..Self::empty()
471        }
472    }
473
474    /// True iff this verification surfaced any explicit failure (i.e. a
475    /// reason an orchestrator should refuse the run).  Distinct from
476    /// `!passed`, which can also mean "legacy / not applicable".
477    pub fn has_failures(&self) -> bool {
478        !self.failures.is_empty()
479    }
480}
481
482/// Run the manifest-aware verification at `manifest_dir` (the destination-
483/// relative directory containing `manifest.json` and `_SUCCESS`).
484///
485/// `manifest_dir` is the same key shape `Destination::write` was called with
486/// for the manifest itself — typically empty (`""`) for prefix-rooted runs,
487/// or the per-export sub-directory.  Trailing `/` is optional.
488///
489/// This function does not panic on any expected I/O outcome — every read
490/// failure becomes a `Failure::*ReadError` so the caller can render a
491/// useful message instead of bailing.
492///
493/// `depth` selects the graded verify layer (see [`ValidateDepth`]):
494/// - [`ValidateDepth::Light`] skips section 3 (the `list_prefix` part
495///   reconcile) and section 5 (untracked surplus), leaving `parts_verified`
496///   at 0 — a fast manifest + `_SUCCESS` poll with no prefix listing.
497/// - [`ValidateDepth::Sample`] and [`ValidateDepth::Full`] run all five
498///   sections here.  The Form B value re-read is **not** in this function;
499///   it is the caller's concern (`run_validate_command`), gated on `Full`.
500///
501/// Regardless of depth, `depth_level` on the returned verdict records the
502/// level this pass ran at.
503pub fn verify_at_destination(
504    dest: &dyn Destination,
505    manifest_dir: &str,
506    depth: ValidateDepth,
507) -> Result<ManifestVerification> {
508    let manifest_key = join_key(manifest_dir, MANIFEST_FILENAME);
509    let success_key = join_key(manifest_dir, SUCCESS_FILENAME);
510
511    // Stamp the depth this pass ran at onto every verdict before it leaves the
512    // function — including the early-return error/legacy shapes — so a consumer
513    // always sees *how much* was checked.  Each `return Ok(v)` below routes
514    // through `with_depth` (or sets `out.depth_level` for the main path).
515    let with_depth = |mut v: ManifestVerification| -> ManifestVerification {
516        v.depth_level = depth.label().to_string();
517        v
518    };
519
520    // ── 1. Manifest read ───────────────────────────────────────────────
521    //
522    // Error-consistency contract: every I/O outcome here surfaces as a
523    // structured `Failure` variant rather than as `Err`.  An operator gets
524    // one verdict shape regardless of whether the destination is missing,
525    // permission-denied, or temporarily unreachable.  The bubbled `Err`
526    // path is reserved for *programmer* errors (caller passes a malformed
527    // `manifest_dir`, a future destination breaks an internal invariant).
528    let manifest_bytes = match dest.head(&manifest_key) {
529        Ok(None) => return Ok(with_depth(ManifestVerification::legacy())),
530        Ok(Some(_)) => match read_capped(dest, &manifest_key, MANIFEST_MAX_BYTES) {
531            Ok(b) => b,
532            Err(e) => {
533                let mut v = ManifestVerification::legacy();
534                v.legacy_run = false;
535                v.failures.push(Failure::ManifestReadError {
536                    detail: format!("{e:#}"),
537                });
538                v.passed = false;
539                return Ok(with_depth(v));
540            }
541        },
542        Err(e) => {
543            // `head` failure is symmetric to a `read` failure — same kind
544            // (`ManifestReadError`) so consumers don't have to branch on
545            // which method tripped.  Distinct from "manifest absent"
546            // (Ok(None) above) which legitimately means "legacy prefix".
547            let mut v = ManifestVerification::legacy();
548            v.legacy_run = false;
549            v.failures.push(Failure::ManifestReadError {
550                detail: format!("manifest head failed: {e:#}"),
551            });
552            v.passed = false;
553            return Ok(with_depth(v));
554        }
555    };
556
557    let manifest: RunManifest = match serde_json::from_slice(&manifest_bytes) {
558        Ok(m) => m,
559        Err(e) => {
560            // A malformed manifest is treated as a self-inconsistency —
561            // semantically equivalent for the operator (the manifest can't
562            // be trusted) but kept distinct in `failures` so the kind is
563            // explicit on the wire.
564            return Ok(with_depth(ManifestVerification {
565                manifest_found: true,
566                failures: vec![Failure::ManifestSelfInconsistent {
567                    detail: format!("manifest.json parse failed: {e}"),
568                }],
569                ..ManifestVerification::empty()
570            }));
571        }
572    };
573
574    // Optimistic base: a found, self-consistent manifest that passes until a
575    // check below flips it.  Overrides only what differs from `empty()`.
576    // Stamp the depth here so the two early `return Ok(out)` paths in section
577    // 4 (success-marker head error, non-utf8 body) carry the right level too.
578    let mut out = ManifestVerification {
579        manifest_found: true,
580        manifest_self_consistent: true,
581        passed: true,
582        depth_level: depth.label().to_string(),
583        ..ManifestVerification::empty()
584    };
585
586    // ── 2. Self-consistency ─────────────────────────────────────────────
587    if let Err(e) = manifest.validate_self_consistency() {
588        out.manifest_self_consistent = false;
589        out.failures.push(Failure::ManifestSelfInconsistent {
590            detail: format!("{e}"),
591        });
592        // Don't short-circuit — we still want to surface part-presence
593        // failures because the operator may want to know both classes at
594        // once rather than fix-then-rerun.
595    }
596
597    // ── 3. Reconcile parts + surplus against ONE prefix listing ────────
598    //
599    // Presence and untracked-surplus both fall out of a single
600    // `reconcile_manifest_against_listing` over one `list_prefix` — the same
601    // pure walk chunked resume uses (`build_resume_plan`).  This replaces the
602    // old per-part `HEAD` loop (N round-trips) and its separate untracked
603    // listing.  Per-part failures are emitted here (step 3); untracked is
604    // emitted at step 5 so the failure ordering an operator reads is stable.
605    //
606    // Trade-off: presence now rides the listing, not per-part `HEAD`.  If the
607    // listing cannot be read, an audit cannot certify the parts — so a list
608    // failure flips `passed = false` (a `ListPrefixError`), rather than the
609    // old behaviour where per-part HEAD still "verified" parts a failed
610    // listing couldn't enumerate.  Every Rivet destination backend offers
611    // strong read-after-write list consistency, so the happy path is one call.
612    //
613    // Graded depth: `Light` skips this `list_prefix` entirely — no part
614    // reconcile, no `ListPrefixError`, `parts_verified` stays 0, and section 5
615    // (untracked) is a no-op since `reconciliation` is `None`.  `Sample` and
616    // `Full` run it.  A `Light` pass therefore certifies only that the
617    // manifest reads, is self-consistent, and `_SUCCESS` matches — never that
618    // the parts are physically present.
619    let reconciliation = if depth.runs_part_reconcile() {
620        match dest.list_prefix(manifest_dir) {
621            Ok(listing) => Some(reconcile_manifest_against_listing(
622                &manifest,
623                &listing,
624                manifest_dir,
625            )),
626            Err(e) => {
627                out.failures.push(Failure::ListPrefixError {
628                    detail: format!("{e:#}"),
629                });
630                None
631            }
632        }
633    } else {
634        None
635    };
636    if let Some(rec) = &reconciliation {
637        for check in &rec.per_part {
638            match &check.presence {
639                PartPresence::Present { md5_verified } => {
640                    out.parts_verified += 1;
641                    if *md5_verified {
642                        out.parts_md5_verified += 1;
643                    }
644                }
645                PartPresence::SizeMismatch { expected, actual } => {
646                    out.parts_failed += 1;
647                    out.failures.push(Failure::PartSizeMismatch {
648                        part_id: check.part_id,
649                        path: check.path.clone(),
650                        expected: *expected,
651                        actual: *actual,
652                    });
653                }
654                PartPresence::Missing => {
655                    out.parts_failed += 1;
656                    out.failures.push(Failure::PartMissing {
657                        part_id: check.part_id,
658                        path: check.path.clone(),
659                    });
660                }
661                PartPresence::ChecksumMismatch { expected, actual } => {
662                    out.parts_failed += 1;
663                    out.failures.push(Failure::PartChecksumMismatch {
664                        part_id: check.part_id,
665                        path: check.path.clone(),
666                        expected: expected.clone(),
667                        actual: actual.clone(),
668                    });
669                }
670            }
671        }
672    }
673
674    // ── 4. _SUCCESS marker consistency ─────────────────────────────────
675    //
676    // Same error-consistency contract as step 1: head/read failures become
677    // `Failure::SuccessMarkerReadError`, not bubbled `Err`.  Absent marker
678    // (Ok(None)) stays informational, not a failure (M2: only successful
679    // runs land _SUCCESS, so its absence on a failed manifest is correct).
680    let success_head = match dest.head(&success_key) {
681        Ok(h) => h,
682        Err(e) => {
683            out.failures.push(Failure::SuccessMarkerReadError {
684                detail: format!("_SUCCESS head failed: {e:#}"),
685            });
686            out.recompute_passed();
687            return Ok(out);
688        }
689    };
690    match success_head {
691        None => {
692            // Absent _SUCCESS is informational, not a failure: per ADR-0012
693            // M2, only successful runs land it.  A failed-then-rewritten
694            // manifest legitimately lacks _SUCCESS.  Leave
695            // `success_marker_consistent = false` (this is a "no signal"
696            // bool, not a "broken" bool) and let the caller decide.
697        }
698        Some(_) => match dest.read(&success_key) {
699            Err(e) => {
700                out.failures.push(Failure::SuccessMarkerReadError {
701                    detail: format!("{e:#}"),
702                });
703            }
704            Ok(body) => {
705                let body_str = match std::str::from_utf8(&body) {
706                    Ok(s) => s,
707                    Err(_) => {
708                        out.failures.push(Failure::SuccessMarkerMalformed {
709                            body_preview: format!("(non-utf8, {} bytes)", body.len()),
710                        });
711                        out.recompute_passed();
712                        return Ok(out);
713                    }
714                };
715                match parse_success_marker(body_str) {
716                    None => {
717                        out.failures.push(Failure::SuccessMarkerMalformed {
718                            body_preview: preview(body_str),
719                        });
720                    }
721                    Some(marker_fp) => {
722                        let manifest_fp = success_marker_body(&manifest_bytes);
723                        // success_marker_body returns the trailing `\n`
724                        // form; trim before comparing to the parsed marker
725                        // (which already trims).
726                        let manifest_fp_trimmed = manifest_fp.trim_end_matches('\n');
727                        if marker_fp == manifest_fp_trimmed {
728                            out.success_marker_consistent = true;
729                        } else {
730                            out.failures.push(Failure::SuccessMarkerStale {
731                                marker_fingerprint: marker_fp.to_string(),
732                                manifest_fingerprint: manifest_fp_trimmed.to_string(),
733                            });
734                        }
735                    }
736                }
737            }
738        },
739    }
740
741    // ── 5. Untracked surplus ───────────────────────────────────────────
742    //
743    // Already computed by the step-3 reconciliation (sidecars, quarantine,
744    // and the doctor probe are filtered there).  Emit it last so the failure
745    // ordering stays parts → marker → untracked.  A list failure left
746    // `reconciliation = None` and already flipped `passed` above.
747    if let Some(rec) = reconciliation {
748        for obj in rec.untracked {
749            out.failures.push(Failure::UntrackedObject {
750                key: obj.key,
751                size_bytes: obj.size_bytes,
752            });
753        }
754    }
755
756    out.recompute_passed();
757    Ok(out)
758}
759
760/// Truncate `s` to a small printable preview for error messages.
761fn preview(s: &str) -> String {
762    let trimmed: String = s.chars().take(40).collect();
763    if s.chars().count() > 40 {
764        format!("{trimmed}…")
765    } else {
766        trimmed
767    }
768}
769
770#[cfg(test)]
771mod tests {
772    use super::*;
773    use crate::config::{DestinationConfig, DestinationType};
774    use crate::destination::local::LocalDestination;
775    use crate::manifest::{
776        MANIFEST_VERSION, ManifestDestination, ManifestPart, ManifestSource, ManifestStatus,
777        PartStatus, RunManifest,
778    };
779    use std::path::Path;
780
781    fn local_dest(base: &Path) -> LocalDestination {
782        LocalDestination::new(&DestinationConfig {
783            destination_type: DestinationType::Local,
784            path: Some(base.to_string_lossy().into_owned()),
785            ..Default::default()
786        })
787        .unwrap()
788    }
789
790    fn part(part_id: u32, rows: i64, size: u64, fp: &str) -> ManifestPart {
791        ManifestPart {
792            part_id,
793            path: format!("part-{part_id:06}.parquet"),
794            rows,
795            size_bytes: size,
796            content_fingerprint: fp.into(),
797            content_md5: String::new(),
798            status: PartStatus::Committed,
799        }
800    }
801
802    fn build_manifest(parts: Vec<ManifestPart>, status: ManifestStatus) -> RunManifest {
803        let row_count: i64 = parts
804            .iter()
805            .filter(|p| p.status == PartStatus::Committed)
806            .map(|p| p.rows)
807            .sum();
808        let part_count = parts
809            .iter()
810            .filter(|p| p.status == PartStatus::Committed)
811            .count() as u32;
812        RunManifest {
813            mode: "batch".to_string(),
814            manifest_version: MANIFEST_VERSION,
815            run_id: "r".into(),
816            export_name: "public.orders".into(),
817            started_at: "2026-05-21T12:00:00Z".into(),
818            finished_at: "2026-05-21T12:01:00Z".into(),
819            status,
820            source: ManifestSource {
821                engine: "postgres".into(),
822                schema: Some("public".into()),
823                table: Some("orders".into()),
824                extraction: None,
825            },
826            destination: ManifestDestination {
827                kind: "local".into(),
828                uri: "file:///tmp/out".into(),
829            },
830            format: "parquet".into(),
831            compression: "zstd".into(),
832            schema_fingerprint: "xxh3:0123456789abcdef".into(),
833            row_count,
834            part_count,
835            parts,
836            column_checksums: None,
837            checksum_key_column: None,
838        }
839    }
840
841    /// Lay out a clean dataset with manifest + _SUCCESS at the root.
842    fn write_dataset(dir: &Path, m: &RunManifest, parts_with_bytes: &[(&str, &[u8])]) {
843        for (name, bytes) in parts_with_bytes {
844            std::fs::write(dir.join(name), bytes).unwrap();
845        }
846        let body = serde_json::to_vec_pretty(m).unwrap();
847        std::fs::write(dir.join(MANIFEST_FILENAME), &body).unwrap();
848        if matches!(m.status, ManifestStatus::Success) {
849            std::fs::write(dir.join(SUCCESS_FILENAME), success_marker_body(&body)).unwrap();
850        }
851    }
852
853    // ── happy path ───────────────────────────────────────────────────────
854
855    #[test]
856    fn happy_path_verifies_all_parts_and_success_marker() {
857        let dir = tempfile::tempdir().unwrap();
858        let m = build_manifest(
859            vec![
860                part(1, 10, 4, "xxh3:1111111111111111"),
861                part(2, 20, 5, "xxh3:2222222222222222"),
862            ],
863            ManifestStatus::Success,
864        );
865        write_dataset(
866            dir.path(),
867            &m,
868            &[
869                ("part-000001.parquet", b"AAAA"),
870                ("part-000002.parquet", b"BBBBB"),
871            ],
872        );
873        let dest = local_dest(dir.path());
874
875        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
876        assert!(v.manifest_found);
877        assert!(!v.legacy_run);
878        assert_eq!(v.parts_verified, 2);
879        assert_eq!(v.parts_failed, 0);
880        assert!(v.success_marker_consistent);
881        assert!(v.manifest_self_consistent);
882        assert!(v.passed);
883        assert!(v.failures.is_empty());
884    }
885
886    // ── M6 legacy run ───────────────────────────────────────────────────
887
888    #[test]
889    fn no_manifest_returns_legacy_run_label() {
890        // Empty prefix — no manifest, no parts.
891        let dir = tempfile::tempdir().unwrap();
892        let dest = local_dest(dir.path());
893        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
894        assert!(!v.manifest_found);
895        assert!(v.legacy_run);
896        assert_eq!(v.parts_verified, 0);
897        assert!(!v.passed);
898        assert!(v.failures.is_empty(), "no failures, just a legacy label");
899    }
900
901    // ── M5 part-presence failures ───────────────────────────────────────
902
903    #[test]
904    fn missing_part_is_flagged_with_part_id_and_path() {
905        let dir = tempfile::tempdir().unwrap();
906        let m = build_manifest(
907            vec![
908                part(1, 10, 4, "xxh3:1111111111111111"),
909                part(2, 20, 5, "xxh3:2222222222222222"),
910            ],
911            ManifestStatus::Success,
912        );
913        write_dataset(
914            dir.path(),
915            &m,
916            &[("part-000001.parquet", b"AAAA")], // part 2 missing
917        );
918        let dest = local_dest(dir.path());
919
920        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
921        assert_eq!(v.parts_verified, 1);
922        assert_eq!(v.parts_failed, 1);
923        assert!(!v.passed);
924        assert!(
925            v.failures
926                .iter()
927                .any(|f| matches!(f, Failure::PartMissing { part_id: 2, .. }))
928        );
929    }
930
931    #[test]
932    fn part_size_mismatch_is_flagged_with_expected_and_actual() {
933        let dir = tempfile::tempdir().unwrap();
934        let m = build_manifest(
935            vec![part(1, 10, 4, "xxh3:1111111111111111")],
936            ManifestStatus::Success,
937        );
938        // Manifest claims 4 bytes; we write 6.
939        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"OOPSIE")]);
940        let dest = local_dest(dir.path());
941
942        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
943        assert!(!v.passed);
944        let mismatch = v
945            .failures
946            .iter()
947            .find_map(|f| match f {
948                Failure::PartSizeMismatch {
949                    part_id,
950                    expected,
951                    actual,
952                    ..
953                } => Some((*part_id, *expected, *actual)),
954                _ => None,
955            })
956            .expect("must surface the size mismatch");
957        assert_eq!(mismatch, (1, 4, 6));
958    }
959
960    // ── _SUCCESS marker integrity ───────────────────────────────────────
961
962    #[test]
963    fn stale_success_marker_is_flagged_as_inconsistent() {
964        // Write a manifest, then overwrite _SUCCESS with the marker for a
965        // *different* manifest body — simulating an orchestrator that
966        // mishandled a re-run.
967        let dir = tempfile::tempdir().unwrap();
968        let m = build_manifest(
969            vec![part(1, 10, 4, "xxh3:1111111111111111")],
970            ManifestStatus::Success,
971        );
972        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"AAAA")]);
973        std::fs::write(
974            dir.path().join(SUCCESS_FILENAME),
975            success_marker_body(b"different manifest body"),
976        )
977        .unwrap();
978        let dest = local_dest(dir.path());
979
980        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
981        assert!(!v.success_marker_consistent);
982        assert!(!v.passed);
983        assert!(
984            v.failures
985                .iter()
986                .any(|f| matches!(f, Failure::SuccessMarkerStale { .. }))
987        );
988    }
989
990    #[test]
991    fn malformed_success_marker_body_is_flagged() {
992        let dir = tempfile::tempdir().unwrap();
993        let m = build_manifest(
994            vec![part(1, 10, 4, "xxh3:1111111111111111")],
995            ManifestStatus::Success,
996        );
997        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"AAAA")]);
998        std::fs::write(dir.path().join(SUCCESS_FILENAME), b"not even xxh3 shaped").unwrap();
999        let dest = local_dest(dir.path());
1000
1001        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
1002        assert!(!v.passed);
1003        assert!(
1004            v.failures
1005                .iter()
1006                .any(|f| matches!(f, Failure::SuccessMarkerMalformed { .. }))
1007        );
1008    }
1009
1010    #[test]
1011    fn absent_success_marker_does_not_fail_validation_alone() {
1012        // ADR-0012 M2: only successful runs land _SUCCESS.  A failed-then-
1013        // rewritten manifest legitimately lacks one — verification must
1014        // not flip `passed` just for that.
1015        let dir = tempfile::tempdir().unwrap();
1016        let m = build_manifest(
1017            vec![part(1, 10, 4, "xxh3:1111111111111111")],
1018            ManifestStatus::Failed,
1019        );
1020        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"AAAA")]);
1021        // Note: write_dataset only writes _SUCCESS for status == Success,
1022        // so no marker exists here.
1023        assert!(!dir.path().join(SUCCESS_FILENAME).exists());
1024        let dest = local_dest(dir.path());
1025
1026        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
1027        assert!(v.manifest_found);
1028        assert!(
1029            !v.success_marker_consistent,
1030            "no marker => false (no signal)"
1031        );
1032        // The parts still verified, so passed = true.
1033        assert!(v.passed);
1034        assert!(v.failures.is_empty());
1035    }
1036
1037    // ── self-consistency ────────────────────────────────────────────────
1038
1039    #[test]
1040    fn self_inconsistent_manifest_is_flagged_but_part_check_still_runs() {
1041        let dir = tempfile::tempdir().unwrap();
1042        let mut m = build_manifest(
1043            vec![part(1, 10, 4, "xxh3:1111111111111111")],
1044            ManifestStatus::Success,
1045        );
1046        m.row_count = 9999; // lie
1047
1048        let body = serde_json::to_vec_pretty(&m).unwrap();
1049        std::fs::write(dir.path().join("part-000001.parquet"), b"AAAA").unwrap();
1050        std::fs::write(dir.path().join(MANIFEST_FILENAME), &body).unwrap();
1051        std::fs::write(
1052            dir.path().join(SUCCESS_FILENAME),
1053            success_marker_body(&body),
1054        )
1055        .unwrap();
1056        let dest = local_dest(dir.path());
1057
1058        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
1059        assert!(v.manifest_found);
1060        assert!(!v.manifest_self_consistent);
1061        assert!(!v.passed);
1062        // Parts that are physically present still get their `parts_verified`
1063        // counter bumped — both signals are independently useful.
1064        assert_eq!(v.parts_verified, 1);
1065        assert!(
1066            v.failures
1067                .iter()
1068                .any(|f| matches!(f, Failure::ManifestSelfInconsistent { .. }))
1069        );
1070    }
1071
1072    // ── untracked objects ───────────────────────────────────────────────
1073
1074    #[test]
1075    fn verify_counters_track_present_and_failed_parts() {
1076        // Mutation-tier2 gap: the verdict-shaping fields (manifest_found /
1077        // passed / failures) and the per-part counters (parts_verified /
1078        // parts_failed) had no assertion — `+= -> -=` survived. One present
1079        // part + one missing part pin both counters and the verdict.
1080        let dir = tempfile::tempdir().unwrap();
1081        let m = build_manifest(
1082            vec![
1083                part(1, 10, 4, "xxh3:1111111111111111"),
1084                part(2, 10, 4, "xxh3:2222222222222222"),
1085            ],
1086            ManifestStatus::Success,
1087        );
1088        // Only part 1 exists at the destination.
1089        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"AAAA")]);
1090        let dest = local_dest(dir.path());
1091
1092        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
1093        assert!(v.manifest_found, "manifest is at the prefix");
1094        assert_eq!(v.parts_verified, 1, "part 1 is present at its size");
1095        assert_eq!(v.parts_failed, 1, "part 2 is missing");
1096        assert!(
1097            v.failures
1098                .iter()
1099                .any(|f| matches!(f, Failure::PartMissing { part_id: 2, .. })),
1100            "the missing part is named in failures: {:?}",
1101            v.failures
1102        );
1103        assert!(!v.passed, "a missing part must fail the verdict");
1104    }
1105
1106    #[test]
1107    fn read_capped_boundary_is_exact() {
1108        // `size > max` (not >=): a body exactly AT the cap must load; one
1109        // byte over must be refused before materialising.
1110        let dir = tempfile::tempdir().unwrap();
1111        std::fs::write(dir.path().join("at-cap.bin"), vec![0u8; 8]).unwrap();
1112        std::fs::write(dir.path().join("over-cap.bin"), vec![0u8; 9]).unwrap();
1113        let dest = local_dest(dir.path());
1114        assert_eq!(
1115            read_capped(&dest, "at-cap.bin", 8)
1116                .expect("exactly at cap loads")
1117                .len(),
1118            8
1119        );
1120        let err = read_capped(&dest, "over-cap.bin", 8).expect_err("over cap refuses");
1121        assert!(
1122            err.to_string().contains("read cap"),
1123            "actionable message: {err:#}"
1124        );
1125    }
1126
1127    #[test]
1128    fn preview_truncates_only_past_forty_chars() {
1129        let s39: String = "x".repeat(39);
1130        let s40: String = "x".repeat(40);
1131        let s41: String = "x".repeat(41);
1132        assert_eq!(preview(&s39), s39, "under the limit: unchanged");
1133        assert_eq!(
1134            preview(&s40),
1135            s40,
1136            "exactly at the limit: unchanged (`>` not `>=`)"
1137        );
1138        assert_eq!(
1139            preview(&s41),
1140            format!("{s40}\u{2026}"),
1141            "past the limit: 40 chars + ellipsis"
1142        );
1143    }
1144
1145    #[test]
1146    fn untracked_object_under_prefix_is_flagged() {
1147        let dir = tempfile::tempdir().unwrap();
1148        let m = build_manifest(
1149            vec![part(1, 10, 4, "xxh3:1111111111111111")],
1150            ManifestStatus::Success,
1151        );
1152        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"AAAA")]);
1153        std::fs::write(dir.path().join("rogue.parquet"), b"XX").unwrap();
1154        let dest = local_dest(dir.path());
1155
1156        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
1157        assert!(
1158            v.failures.iter().any(
1159                |f| matches!(f, Failure::UntrackedObject { key, .. } if key == "rogue.parquet")
1160            )
1161        );
1162        // Untracked objects are surfaced but do NOT flip `passed` — that
1163        // is the resume-side decision (M9).  Parts and marker are fine,
1164        // so passed remains true.
1165        assert!(v.passed);
1166    }
1167
1168    #[test]
1169    fn quarantine_prefix_objects_are_silently_ignored() {
1170        let dir = tempfile::tempdir().unwrap();
1171        let m = build_manifest(
1172            vec![part(1, 10, 4, "xxh3:1111111111111111")],
1173            ManifestStatus::Success,
1174        );
1175        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"AAAA")]);
1176        std::fs::create_dir_all(dir.path().join(crate::manifest::QUARANTINE_PREFIX)).unwrap();
1177        std::fs::write(
1178            dir.path()
1179                .join(crate::manifest::QUARANTINE_PREFIX)
1180                .join("old.parquet"),
1181            b"OO",
1182        )
1183        .unwrap();
1184        let dest = local_dest(dir.path());
1185
1186        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
1187        assert!(v.passed);
1188        assert!(
1189            !v.failures
1190                .iter()
1191                .any(|f| matches!(f, Failure::UntrackedObject { .. })),
1192            "quarantine_prefix is the legitimate home for these — must not flag"
1193        );
1194    }
1195
1196    #[test]
1197    fn doctor_probe_is_not_flagged_as_untracked() {
1198        // Regression: `rivet doctor` writes `.rivet_doctor_probe` at the
1199        // destination prefix and never removes it.  A subsequent
1200        // `rivet run --validate` against the same prefix must treat it as a
1201        // Rivet sidecar, not foreign data — otherwise `has_failures()` trips
1202        // and the run's `validated` flag is downgraded to FAIL.
1203        let dir = tempfile::tempdir().unwrap();
1204        let m = build_manifest(
1205            vec![part(1, 10, 4, "xxh3:1111111111111111")],
1206            ManifestStatus::Success,
1207        );
1208        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"AAAA")]);
1209        std::fs::write(
1210            dir.path().join(crate::manifest::DOCTOR_PROBE_FILENAME),
1211            b"ok",
1212        )
1213        .unwrap();
1214        let dest = local_dest(dir.path());
1215
1216        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
1217        assert!(
1218            !v.has_failures(),
1219            "doctor probe must not surface as a failure: {:?}",
1220            v.failures
1221        );
1222        assert!(v.passed);
1223    }
1224
1225    // ── manifest_dir join semantics ─────────────────────────────────────
1226
1227    #[test]
1228    fn verifies_in_subdirectory_when_manifest_dir_is_non_empty() {
1229        let outer = tempfile::tempdir().unwrap();
1230        std::fs::create_dir_all(outer.path().join("sub/run")).unwrap();
1231        let m = build_manifest(
1232            vec![part(1, 10, 4, "xxh3:1111111111111111")],
1233            ManifestStatus::Success,
1234        );
1235        let body = serde_json::to_vec_pretty(&m).unwrap();
1236        std::fs::write(outer.path().join("sub/run/part-000001.parquet"), b"AAAA").unwrap();
1237        std::fs::write(outer.path().join("sub/run").join(MANIFEST_FILENAME), &body).unwrap();
1238        std::fs::write(
1239            outer.path().join("sub/run").join(SUCCESS_FILENAME),
1240            success_marker_body(&body),
1241        )
1242        .unwrap();
1243        let dest = local_dest(outer.path());
1244
1245        let v = verify_at_destination(&dest, "sub/run", ValidateDepth::Full).unwrap();
1246        assert!(v.passed);
1247        assert_eq!(v.parts_verified, 1);
1248
1249        // Trailing slash is normalised — same outcome.
1250        let v2 = verify_at_destination(&dest, "sub/run/", ValidateDepth::Full).unwrap();
1251        assert!(v2.passed);
1252    }
1253
1254    // ── list-failure semantics (presence now rides the listing) ──────────
1255
1256    /// Wraps a real `LocalDestination` but fails every `list_prefix`. Used to
1257    /// pin the post-refactor contract: presence is derived from the listing,
1258    /// so a listing we cannot read means the audit cannot certify the parts.
1259    struct ListFails(LocalDestination);
1260    impl crate::destination::Destination for ListFails {
1261        fn write(&self, p: &Path, k: &str) -> Result<crate::destination::WriteOutcome> {
1262            self.0.write(p, k)
1263        }
1264        fn capabilities(&self) -> crate::destination::DestinationCapabilities {
1265            self.0.capabilities()
1266        }
1267        fn head(&self, k: &str) -> Result<Option<crate::destination::ObjectMeta>> {
1268            self.0.head(k)
1269        }
1270        fn read(&self, k: &str) -> Result<Vec<u8>> {
1271            self.0.read(k)
1272        }
1273        fn list_prefix(&self, _: &str) -> Result<Vec<crate::destination::ObjectMeta>> {
1274            anyhow::bail!("listing unavailable")
1275        }
1276    }
1277
1278    #[test]
1279    fn list_failure_cannot_certify_parts_and_fails_the_audit() {
1280        let dir = tempfile::tempdir().unwrap();
1281        let m = build_manifest(vec![part(0, 3, 3, "xxh3:0")], ManifestStatus::Success);
1282        write_dataset(dir.path(), &m, &[("part-000000.parquet", b"abc")]);
1283        let dest = ListFails(local_dest(dir.path()));
1284
1285        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
1286        // The manifest itself reads + parses fine (HEAD/read still work)…
1287        assert!(v.manifest_found);
1288        assert!(v.manifest_self_consistent);
1289        // …but with no listing we verify zero parts and refuse to pass.
1290        assert!(
1291            !v.passed,
1292            "an audit that cannot list the prefix must not pass"
1293        );
1294        assert_eq!(v.parts_verified, 0);
1295        assert!(
1296            v.failures
1297                .iter()
1298                .any(|f| matches!(f, Failure::ListPrefixError { .. })),
1299            "expected a ListPrefixError, got: {:?}",
1300            v.failures
1301        );
1302    }
1303
1304    // ── manifest read-error semantics (explicit failure, not legacy) ─────
1305
1306    /// Wraps a real `LocalDestination` but fails reading `manifest.json`
1307    /// (head still sees it) — EACCES or a transient store error on a
1308    /// manifest that exists.
1309    struct ManifestReadFails(LocalDestination);
1310    impl crate::destination::Destination for ManifestReadFails {
1311        fn write(&self, p: &Path, k: &str) -> Result<crate::destination::WriteOutcome> {
1312            self.0.write(p, k)
1313        }
1314        fn capabilities(&self) -> crate::destination::DestinationCapabilities {
1315            self.0.capabilities()
1316        }
1317        fn head(&self, k: &str) -> Result<Option<crate::destination::ObjectMeta>> {
1318            self.0.head(k)
1319        }
1320        fn read(&self, k: &str) -> Result<Vec<u8>> {
1321            if k.ends_with(MANIFEST_FILENAME) {
1322                anyhow::bail!("permission denied (simulated)")
1323            }
1324            self.0.read(k)
1325        }
1326        fn list_prefix(&self, p: &str) -> Result<Vec<crate::destination::ObjectMeta>> {
1327            self.0.list_prefix(p)
1328        }
1329    }
1330
1331    #[test]
1332    fn unreadable_manifest_is_explicit_failure_not_legacy() {
1333        // The exit gates (`rivet validate`, run finalize) key off this exact
1334        // shape: `manifest_found: false` but `has_failures()` — distinct
1335        // from M6 legacy (`legacy_run: true`, no failures, exit 0).
1336        let dir = tempfile::tempdir().unwrap();
1337        let m = build_manifest(
1338            vec![part(1, 10, 4, "xxh3:1111111111111111")],
1339            ManifestStatus::Success,
1340        );
1341        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"AAAA")]);
1342        let dest = ManifestReadFails(local_dest(dir.path()));
1343
1344        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
1345        assert!(!v.manifest_found);
1346        assert!(!v.legacy_run, "a read error is not the M6 legacy label");
1347        assert!(!v.passed);
1348        assert!(v.has_failures(), "orchestrators need a reason to refuse");
1349        assert!(
1350            matches!(v.failures.as_slice(), [Failure::ManifestReadError { .. }]),
1351            "expected exactly one ManifestReadError, got: {:?}",
1352            v.failures
1353        );
1354    }
1355
1356    /// Same contract when `head` itself errors (cannot even stat the
1357    /// manifest): symmetric `ManifestReadError`, never the legacy label.
1358    struct ManifestHeadFails(LocalDestination);
1359    impl crate::destination::Destination for ManifestHeadFails {
1360        fn write(&self, p: &Path, k: &str) -> Result<crate::destination::WriteOutcome> {
1361            self.0.write(p, k)
1362        }
1363        fn capabilities(&self) -> crate::destination::DestinationCapabilities {
1364            self.0.capabilities()
1365        }
1366        fn head(&self, k: &str) -> Result<Option<crate::destination::ObjectMeta>> {
1367            if k.ends_with(MANIFEST_FILENAME) {
1368                anyhow::bail!("io timeout (simulated)")
1369            }
1370            self.0.head(k)
1371        }
1372        fn read(&self, k: &str) -> Result<Vec<u8>> {
1373            self.0.read(k)
1374        }
1375        fn list_prefix(&self, p: &str) -> Result<Vec<crate::destination::ObjectMeta>> {
1376            self.0.list_prefix(p)
1377        }
1378    }
1379
1380    #[test]
1381    fn manifest_head_error_is_explicit_failure_not_legacy() {
1382        let dir = tempfile::tempdir().unwrap();
1383        let m = build_manifest(
1384            vec![part(1, 10, 4, "xxh3:1111111111111111")],
1385            ManifestStatus::Success,
1386        );
1387        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"AAAA")]);
1388        let dest = ManifestHeadFails(local_dest(dir.path()));
1389
1390        let v = verify_at_destination(&dest, "", ValidateDepth::Full).unwrap();
1391        assert!(!v.manifest_found);
1392        assert!(!v.legacy_run);
1393        assert!(!v.passed);
1394        assert!(
1395            matches!(
1396                v.failures.as_slice(),
1397                [Failure::ManifestReadError { detail }] if detail.contains("manifest head failed")
1398            ),
1399            "expected one ManifestReadError naming the head step, got: {:?}",
1400            v.failures
1401        );
1402    }
1403
1404    #[test]
1405    fn passed_is_derived_advisory_failures_do_not_fail() {
1406        // An advisory failure (untracked surplus) keeps the verdict passing…
1407        let mut v = ManifestVerification {
1408            manifest_found: true,
1409            ..ManifestVerification::empty()
1410        };
1411        v.failures.push(Failure::UntrackedObject {
1412            key: "stray.parquet".into(),
1413            size_bytes: 9,
1414        });
1415        v.recompute_passed();
1416        assert!(v.passed, "untracked surplus is advisory, not fatal");
1417
1418        // …while any fatal failure flips it.
1419        v.failures.push(Failure::PartMissing {
1420            part_id: 1,
1421            path: "part-000001.parquet".into(),
1422        });
1423        v.recompute_passed();
1424        assert!(!v.passed, "a missing part is fatal");
1425
1426        // No manifest → never passes regardless of failures.
1427        let mut legacy = ManifestVerification::empty();
1428        legacy.recompute_passed();
1429        assert!(!legacy.passed, "no manifest found → cannot certify");
1430    }
1431
1432    #[test]
1433    fn verify_content_policy_fails_only_size_only_parts() {
1434        // 3 parts, 2 content-checked, 1 size-only.
1435        let base = ManifestVerification {
1436            manifest_found: true,
1437            parts_verified: 3,
1438            parts_md5_verified: 2,
1439            ..ManifestVerification::empty()
1440        };
1441        // verify: size → size-only is acceptable, passes.
1442        let mut sz = base.clone();
1443        sz.recompute_passed();
1444        sz.enforce_content_policy(false);
1445        assert!(sz.passed, "size-only OK under verify: size");
1446
1447        // verify: content → the 1 size-only part is a fatal failure.
1448        let mut ct = base.clone();
1449        ct.recompute_passed();
1450        ct.enforce_content_policy(true);
1451        assert!(!ct.passed, "a size-only part fails verify: content");
1452        assert!(
1453            ct.failures.iter().any(|f| matches!(
1454                f,
1455                Failure::ContentVerificationUnmet {
1456                    size_only: 1,
1457                    total: 3
1458                }
1459            )),
1460            "expected ContentVerificationUnmet, got: {:?}",
1461            ct.failures
1462        );
1463
1464        // verify: content with every part md5-checked → satisfied.
1465        let mut all = ManifestVerification {
1466            parts_md5_verified: 3,
1467            ..base
1468        };
1469        all.recompute_passed();
1470        all.enforce_content_policy(true);
1471        assert!(
1472            all.passed && all.failures.is_empty(),
1473            "all md5 meets verify: content"
1474        );
1475    }
1476
1477    // ── require_manifest_present (finding #20: operator-pinned --prefix) ──────
1478
1479    #[test]
1480    fn require_manifest_escalates_legacy_to_fatal_absent() {
1481        // The exact shape `verify_at_destination` returns for an absent manifest
1482        // (`legacy()`): no manifest, no other failure. With a pinned `--prefix`
1483        // this is escalated to a fatal `ManifestRequiredButAbsent` so the exit
1484        // gate refuses it instead of passing as a benign legacy run.
1485        let mut v = ManifestVerification::legacy();
1486        assert!(v.legacy_run && !v.has_failures());
1487
1488        v.require_manifest_present("exports/2026-06-09/orders/");
1489
1490        assert!(!v.legacy_run, "no longer the benign legacy-run label");
1491        assert!(!v.passed, "an absent-but-required manifest cannot pass");
1492        assert!(
1493            matches!(
1494                v.failures.as_slice(),
1495                [Failure::ManifestRequiredButAbsent { prefix }]
1496                    if prefix == "exports/2026-06-09/orders/"
1497            ),
1498            "expected one ManifestRequiredButAbsent naming the prefix, got: {:?}",
1499            v.failures
1500        );
1501    }
1502
1503    #[test]
1504    fn require_manifest_is_noop_on_a_real_passing_manifest() {
1505        // A found, passing verdict is untouched — `--prefix` plus real data is
1506        // the normal "validate this exact prefix" case and must still pass.
1507        let mut v = ManifestVerification {
1508            manifest_found: true,
1509            manifest_self_consistent: true,
1510            parts_verified: 1,
1511            passed: true,
1512            ..ManifestVerification::empty()
1513        };
1514        v.require_manifest_present("exports/orders/");
1515        assert!(
1516            v.passed && v.failures.is_empty(),
1517            "real dataset still passes"
1518        );
1519    }
1520
1521    #[test]
1522    fn require_manifest_does_not_double_flag_a_read_error() {
1523        // An absent manifest that already carries a `ManifestReadError` (head /
1524        // read failed) is already a fatal, classified failure — requiring a
1525        // manifest here must not add a second, redundant failure.
1526        let mut v = ManifestVerification::legacy();
1527        v.legacy_run = false;
1528        v.failures.push(Failure::ManifestReadError {
1529            detail: "permission denied".into(),
1530        });
1531        v.recompute_passed();
1532
1533        v.require_manifest_present("exports/orders/");
1534
1535        assert!(
1536            matches!(v.failures.as_slice(), [Failure::ManifestReadError { .. }]),
1537            "must leave the existing read-error verdict alone, got: {:?}",
1538            v.failures
1539        );
1540    }
1541
1542    // ── graded verify layer (--depth) ───────────────────────────────────
1543
1544    #[test]
1545    fn light_depth_skips_part_reconcile_even_when_a_part_is_missing() {
1546        // A manifest declaring a part that is NOT on disk. At `Full`/`Sample`
1547        // this is a fatal `PartMissing`; at `Light` the `list_prefix` reconcile
1548        // is skipped entirely, so `parts_verified == 0`, no `ListPrefixError`,
1549        // and — with `_SUCCESS` consistent and the manifest self-consistent —
1550        // the verdict still passes. Light certifies the manifest + marker, not
1551        // the parts.
1552        let dir = tempfile::tempdir().unwrap();
1553        let m = build_manifest(
1554            vec![
1555                part(1, 10, 4, "xxh3:1111111111111111"),
1556                part(2, 20, 5, "xxh3:2222222222222222"),
1557            ],
1558            ManifestStatus::Success,
1559        );
1560        // Deliberately write NEITHER part — only manifest.json + _SUCCESS.
1561        write_dataset(dir.path(), &m, &[]);
1562        let dest = local_dest(dir.path());
1563
1564        let v = verify_at_destination(&dest, "", ValidateDepth::Light).unwrap();
1565        assert_eq!(v.depth_level, "light");
1566        assert_eq!(
1567            v.parts_verified, 0,
1568            "light skips the listing — no part is ever verified"
1569        );
1570        assert_eq!(
1571            v.parts_failed, 0,
1572            "no part reconcile means no part failures"
1573        );
1574        assert!(
1575            !v.failures.iter().any(|f| matches!(
1576                f,
1577                Failure::PartMissing { .. } | Failure::ListPrefixError { .. }
1578            )),
1579            "light must not surface part or list failures, got: {:?}",
1580            v.failures
1581        );
1582        assert!(
1583            v.success_marker_consistent,
1584            "_SUCCESS is still checked at light depth"
1585        );
1586        assert!(v.manifest_self_consistent);
1587        assert!(
1588            v.passed,
1589            "manifest + _SUCCESS are consistent, so a light pass certifies it"
1590        );
1591    }
1592
1593    #[test]
1594    fn light_depth_never_lists_so_a_list_failure_cannot_trip() {
1595        // Even with a destination whose `list_prefix` always errors, a light
1596        // pass succeeds: it never calls `list_prefix`, so no `ListPrefixError`.
1597        // This is the direct contrast to `list_failure_cannot_certify_parts…`
1598        // (which runs at Full and *does* fail on the list error).
1599        let dir = tempfile::tempdir().unwrap();
1600        let m = build_manifest(vec![part(0, 3, 3, "xxh3:0")], ManifestStatus::Success);
1601        write_dataset(dir.path(), &m, &[("part-000000.parquet", b"abc")]);
1602        let dest = ListFails(local_dest(dir.path()));
1603
1604        let v = verify_at_destination(&dest, "", ValidateDepth::Light).unwrap();
1605        assert_eq!(v.depth_level, "light");
1606        assert!(
1607            !v.failures
1608                .iter()
1609                .any(|f| matches!(f, Failure::ListPrefixError { .. })),
1610            "light never lists, so a failing list_prefix cannot surface, got: {:?}",
1611            v.failures
1612        );
1613        assert_eq!(v.parts_verified, 0);
1614        assert!(v.passed, "manifest + _SUCCESS consistent → light passes");
1615    }
1616
1617    #[test]
1618    fn sample_depth_runs_part_reconcile_like_full() {
1619        // `Sample` runs every section `verify_at_destination` owns (1-5) — the
1620        // Form B value re-read it skips lives in the *caller*, not here. So a
1621        // missing part is a fatal `PartMissing` at Sample, identical to Full.
1622        let dir = tempfile::tempdir().unwrap();
1623        let m = build_manifest(
1624            vec![
1625                part(1, 10, 4, "xxh3:1111111111111111"),
1626                part(2, 20, 5, "xxh3:2222222222222222"),
1627            ],
1628            ManifestStatus::Success,
1629        );
1630        write_dataset(dir.path(), &m, &[("part-000001.parquet", b"AAAA")]); // part 2 missing
1631        let dest = local_dest(dir.path());
1632
1633        let v = verify_at_destination(&dest, "", ValidateDepth::Sample).unwrap();
1634        assert_eq!(v.depth_level, "sample");
1635        assert_eq!(v.parts_verified, 1);
1636        assert_eq!(v.parts_failed, 1);
1637        assert!(!v.passed);
1638        assert!(
1639            v.failures
1640                .iter()
1641                .any(|f| matches!(f, Failure::PartMissing { part_id: 2, .. })),
1642            "sample reconciles parts just like full, got: {:?}",
1643            v.failures
1644        );
1645    }
1646
1647    #[test]
1648    fn depth_level_is_stamped_on_every_verdict_shape() {
1649        // The depth label rides the verdict even on the early-return shapes
1650        // (legacy / no manifest), so a consumer always sees how deep the pass
1651        // went.
1652        let dir = tempfile::tempdir().unwrap();
1653        let dest = local_dest(dir.path()); // empty prefix → legacy
1654        for depth in [
1655            ValidateDepth::Light,
1656            ValidateDepth::Sample,
1657            ValidateDepth::Full,
1658        ] {
1659            let v = verify_at_destination(&dest, "", depth).unwrap();
1660            assert!(v.legacy_run, "empty prefix is the legacy shape");
1661            assert_eq!(
1662                v.depth_level,
1663                depth.label(),
1664                "legacy verdict must still carry its depth label"
1665            );
1666        }
1667    }
1668
1669    #[test]
1670    fn error_code_is_stable_and_distinct_per_variant() {
1671        // Each variant maps to its documented `RIVET_VERIFY_*` code. A
1672        // regression guard: renaming a code is a silent break for any CI gate
1673        // keying off it.
1674        let cases: &[(Failure, &str)] = &[
1675            (
1676                Failure::PartMissing {
1677                    part_id: 1,
1678                    path: "p".into(),
1679                },
1680                "RIVET_VERIFY_PART_MISSING",
1681            ),
1682            (
1683                Failure::PartSizeMismatch {
1684                    part_id: 1,
1685                    path: "p".into(),
1686                    expected: 1,
1687                    actual: 2,
1688                },
1689                "RIVET_VERIFY_PART_SIZE_MISMATCH",
1690            ),
1691            (
1692                Failure::PartChecksumMismatch {
1693                    part_id: 1,
1694                    path: "p".into(),
1695                    expected: "a".into(),
1696                    actual: "b".into(),
1697                },
1698                "RIVET_VERIFY_PART_CHECKSUM_MISMATCH",
1699            ),
1700            (
1701                Failure::SuccessMarkerMalformed {
1702                    body_preview: "x".into(),
1703                },
1704                "RIVET_VERIFY_SUCCESS_MALFORMED",
1705            ),
1706            (
1707                Failure::SuccessMarkerStale {
1708                    marker_fingerprint: "a".into(),
1709                    manifest_fingerprint: "b".into(),
1710                },
1711                "RIVET_VERIFY_SUCCESS_STALE",
1712            ),
1713            (
1714                Failure::ManifestSelfInconsistent { detail: "d".into() },
1715                "RIVET_VERIFY_MANIFEST_INCONSISTENT",
1716            ),
1717            (
1718                Failure::ManifestReadError { detail: "d".into() },
1719                "RIVET_VERIFY_MANIFEST_READ_ERROR",
1720            ),
1721            (
1722                Failure::SuccessMarkerReadError { detail: "d".into() },
1723                "RIVET_VERIFY_SUCCESS_READ_ERROR",
1724            ),
1725            (
1726                Failure::ListPrefixError { detail: "d".into() },
1727                "RIVET_VERIFY_LIST_ERROR",
1728            ),
1729            (
1730                Failure::UntrackedObject {
1731                    key: "k".into(),
1732                    size_bytes: 1,
1733                },
1734                "RIVET_VERIFY_UNTRACKED_OBJECT",
1735            ),
1736            (
1737                Failure::ContentVerificationUnmet {
1738                    size_only: 1,
1739                    total: 2,
1740                },
1741                "RIVET_VERIFY_CONTENT_UNMET",
1742            ),
1743            (
1744                Failure::ManifestRequiredButAbsent { prefix: "p".into() },
1745                "RIVET_VERIFY_MANIFEST_REQUIRED",
1746            ),
1747        ];
1748        for (failure, code) in cases {
1749            assert_eq!(&failure.error_code(), code, "code for {failure:?}");
1750            assert!(
1751                failure.error_code().starts_with("RIVET_VERIFY_"),
1752                "every code shares the RIVET_VERIFY_ prefix"
1753            );
1754        }
1755    }
1756}