faucet-core 1.14.0

//! Plan-time feature × write-path compatibility (RFC 0013, #883).
//!
//! Whether a guarantee holds depends on which path a run takes: the native
//! byte path, the columnar (Arrow) path, the row (`Value`) path, the
//! exactly-once watermark write, or the staged overwrite. This module is the
//! one table saying, for each pipeline feature, what happens on each path and
//! against each sink capability:
//!
//! - [`Verdict::Supported`] — the feature holds on that path.
//! - [`Verdict::DisablesPath`] — the path would skip the feature, so the
//!   pipeline falls back to a slower path that honours it.
//! - [`Verdict::Refuse`] — no path honours the combination; it is refused
//!   when the config is loaded (and again when the pipeline starts).
//!
//! [`Pipeline::run`](crate::Pipeline::run) selects its path through
//! [`path_allowed`]; the CLI's load-time gates read [`refusals`]; the
//! reference page `docs/book/src/reference/compatibility.md` is generated by
//! [`render_markdown`]. A combination not listed is supported.

use crate::capabilities::{Capabilities, SchemaReport};
use serde::Serialize;

/// A pipeline feature (or one of its modes) that constrains the write path.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum Feature {
    /// A transform chain on the source.
    Transforms,
    /// A PII masking policy.
    Masking,
    /// A masking rule that changes a column's type by value (value detector,
    /// number/bool coerced to a string).
    MaskingRetypesByValue,
    /// Data-quality checks.
    Quality,
    /// A data contract.
    Contract,
    /// Any `schema:` drift policy.
    SchemaDrift,
    /// `schema.on_drift: fail`.
    SchemaDriftFail,
    /// `schema.on_drift: evolve`.
    SchemaDriftEvolve,
    /// A DLQ with `on_batch_error: propagate`.
    DlqPropagate,
    /// A DLQ with `on_batch_error: dlq_all`.
    DlqAll,
    /// `delivery: exactly_once` through the atomic watermark.
    ExactlyOnce,
    /// `write_mode: upsert|delete` with a `key`.
    KeyedWrites,
    /// `write_mode: overwrite`.
    Overwrite,
    /// A fan-out overwrite whose rows share one staged destination.
    GroupedOverwrite,
    /// Adaptive batch sizing.
    Adaptive,
    /// A `resilience:` retry / circuit-breaker policy.
    Resilience,
    /// Scoped cleanup (`complete_for` with `on_missing: delete`).
    Cleanup,
    /// A data-flow policy rule with `on_runtime: quarantine`.
    PolicyQuarantine,
    /// A data-flow policy runtime backstop (any rule).
    Policy,
    /// A run budget (`budget:` ceilings).
    Budget,
    /// Column profiling (`profiling:`).
    Profiling,
    /// `_faucet_*` metadata columns (`metadata_columns:`).
    MetadataColumns,
}

impl Feature {
    /// Every feature, in table order.
    pub const ALL: &'static [Feature] = &[
        Feature::Transforms,
        Feature::Masking,
        Feature::MaskingRetypesByValue,
        Feature::Quality,
        Feature::Contract,
        Feature::SchemaDrift,
        Feature::SchemaDriftFail,
        Feature::SchemaDriftEvolve,
        Feature::DlqPropagate,
        Feature::DlqAll,
        Feature::ExactlyOnce,
        Feature::KeyedWrites,
        Feature::Overwrite,
        Feature::GroupedOverwrite,
        Feature::Adaptive,
        Feature::Resilience,
        Feature::Cleanup,
        Feature::PolicyQuarantine,
        Feature::Policy,
        Feature::Budget,
        Feature::Profiling,
        Feature::MetadataColumns,
    ];

    /// How the feature is spelled in a config.
    pub fn label(&self) -> &'static str {
        match self {
            Feature::Transforms => "`transforms:`",
            Feature::Masking => "`masking:`",
            Feature::MaskingRetypesByValue => "`masking:` rule that retypes a value",
            Feature::Quality => "`quality:`",
            Feature::Contract => "`contract:`",
            Feature::SchemaDrift => "`schema:` (any `on_drift`)",
            Feature::SchemaDriftFail => "`schema: { on_drift: fail }`",
            Feature::SchemaDriftEvolve => "`schema: { on_drift: evolve }`",
            Feature::DlqPropagate => "`dlq:` (`on_batch_error: propagate`)",
            Feature::DlqAll => "`dlq:` (`on_batch_error: dlq_all`)",
            Feature::ExactlyOnce => "`delivery: exactly_once` (atomic watermark)",
            Feature::KeyedWrites => "`write_mode: upsert` / `delete`",
            Feature::Overwrite => "`write_mode: overwrite`",
            Feature::GroupedOverwrite => "`write_mode: overwrite` shared by fan-out rows",
            Feature::Adaptive => "adaptive batch sizing",
            Feature::Resilience => "`resilience:`",
            Feature::Cleanup => "`complete_for: { on_missing: delete }`",
            Feature::PolicyQuarantine => "`policy:` rule with `on_runtime: quarantine`",
            Feature::Policy => "`policy:`",
            Feature::Budget => "`budget:`",
            Feature::Profiling => "`profiling:`",
            Feature::MetadataColumns => "`metadata_columns:`",
        }
    }
}

/// The paths a run can write through.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum WritePath {
    /// Native byte passthrough (#633): source bytes loaded as-is.
    Native,
    /// Arrow `RecordBatch`es end to end (#375).
    Columnar,
    /// Per-record `Value` pages — the path that honours every feature.
    Row,
    /// The exactly-once atomic-watermark write (`write_batch_idempotent`).
    Watermark,
    /// The staged overwrite lifecycle (begin → write → commit swap).
    OverwriteStaging,
}

impl WritePath {
    /// Every path, in table order.
    pub const ALL: &'static [WritePath] = &[
        WritePath::Native,
        WritePath::Columnar,
        WritePath::Row,
        WritePath::Watermark,
        WritePath::OverwriteStaging,
    ];

    /// Column heading.
    pub fn label(&self) -> &'static str {
        match self {
            WritePath::Native => "native",
            WritePath::Columnar => "columnar",
            WritePath::Row => "row",
            WritePath::Watermark => "exactly-once watermark",
            WritePath::OverwriteStaging => "overwrite staging",
        }
    }
}

/// A sink capability (or its absence) a feature depends on.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum CapabilityCondition {
    /// The sink drops fields its destination lacks
    /// ([`SchemaReport::DropsUnknownFields`]).
    DropsUnknownFields,
    /// The sink cannot evolve its destination schema.
    NoSchemaEvolution,
    /// The sink cannot clean up a scope.
    NoCleanup,
}

impl CapabilityCondition {
    /// Every condition, in table order.
    pub const ALL: &'static [CapabilityCondition] = &[
        CapabilityCondition::DropsUnknownFields,
        CapabilityCondition::NoSchemaEvolution,
        CapabilityCondition::NoCleanup,
    ];

    /// Column heading.
    pub fn label(&self) -> &'static str {
        match self {
            CapabilityCondition::DropsUnknownFields => "sink drops unknown fields",
            CapabilityCondition::NoSchemaEvolution => "sink without schema evolution",
            CapabilityCondition::NoCleanup => "sink without cleanup",
        }
    }

    /// Whether a sink with `caps` meets the condition.
    pub fn holds(&self, caps: &Capabilities) -> bool {
        match self {
            CapabilityCondition::DropsUnknownFields => {
                caps.schema_report() == SchemaReport::DropsUnknownFields
            }
            CapabilityCondition::NoSchemaEvolution => !caps.schema_evolution(),
            CapabilityCondition::NoCleanup => !caps.cleanup(),
        }
    }
}

/// A column of the table.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
#[serde(rename_all = "snake_case", tag = "kind", content = "value")]
#[non_exhaustive]
pub enum Target {
    /// A write path.
    Path(WritePath),
    /// A sink capability condition.
    Capability(CapabilityCondition),
}

/// What happens to a feature on a path or against a capability.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case", tag = "verdict", content = "reason")]
#[non_exhaustive]
pub enum Verdict {
    /// The feature holds.
    Supported,
    /// The path would skip the feature, so a run with it falls back to a
    /// path that honours it.
    DisablesPath,
    /// No path honours the combination; refused at load time.
    Refuse(&'static str),
}

/// One row of the table.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct Rule {
    /// The feature.
    pub feature: Feature,
    /// The path or capability.
    pub target: Target,
    /// What happens.
    pub verdict: Verdict,
}

const fn path(feature: Feature, p: WritePath, verdict: Verdict) -> Rule {
    Rule {
        feature,
        target: Target::Path(p),
        verdict,
    }
}

const fn cap(feature: Feature, c: CapabilityCondition, verdict: Verdict) -> Rule {
    Rule {
        feature,
        target: Target::Capability(c),
        verdict,
    }
}

use CapabilityCondition as C;
use Feature as F;
use Verdict::{DisablesPath as D, Refuse as R};
use WritePath as P;

/// Refusal: an exactly-once watermark run with a DLQ.
pub const EO_WITH_DLQ: &str =
    "delivery: exactly_once (atomic watermark) is not compatible with a DLQ in this version";
/// Refusal: an exactly-once watermark run that overwrites.
pub const EO_WITH_OVERWRITE: &str =
    "delivery: exactly_once (atomic watermark) is not compatible with write_mode: overwrite";
/// Refusal: an exactly-once watermark run with policy quarantine.
pub const EO_WITH_POLICY_QUARANTINE: &str = "a policy rule with on_runtime: quarantine is not \
     compatible with delivery: exactly_once (atomic watermark): a quarantined row would fail \
     the whole page's atomic commit instead of reaching the DLQ (only a rule whose `when` can \
     match the row's sink kind and attributes counts)";
/// Refusal: `on_drift: fail` against a sink that drops unknown fields.
pub const DRIFT_FAIL_DROPS_FIELDS: &str = "schema.on_drift: fail cannot be enforced against \
     this sink: it reports no destination schema and silently drops fields its schema lacks, so \
     a drift would never be detected and the fields would be lost";
/// Refusal: `on_drift: evolve` against a sink that cannot evolve.
pub const DRIFT_EVOLVE_UNSUPPORTED: &str =
    "schema.on_drift: evolve requires a sink that can evolve its destination schema";
/// Refusal: cleanup against a sink that cannot clean up.
pub const CLEANUP_UNSUPPORTED: &str = "scoped cleanup requires a sink that can delete a scope";
/// Refusal: an overwrite with scoped cleanup.
pub const OVERWRITE_WITH_CLEANUP: &str =
    "write_mode: overwrite replaces the whole destination, so scoped cleanup has nothing to do";
/// Refusal: an overwrite that evolves the schema.
pub const OVERWRITE_WITH_EVOLVE: &str = "write_mode: overwrite stages into a copy of the \
     destination taken before the run, so schema.on_drift: evolve cannot apply to it";

/// The table. A (feature, target) pair not listed is [`Verdict::Supported`].
pub const RULES: &[Rule] = &[
    // Native byte passthrough: nothing may need to see a record.
    path(F::Transforms, P::Native, D),
    path(F::Masking, P::Native, D),
    path(F::MaskingRetypesByValue, P::Native, D),
    path(F::Quality, P::Native, D),
    path(F::Contract, P::Native, D),
    path(F::SchemaDrift, P::Native, D),
    path(F::SchemaDriftFail, P::Native, D),
    path(F::SchemaDriftEvolve, P::Native, D),
    path(F::DlqPropagate, P::Native, D),
    path(F::DlqAll, P::Native, D),
    path(F::ExactlyOnce, P::Native, D),
    path(F::KeyedWrites, P::Native, D),
    path(F::GroupedOverwrite, P::Native, D),
    path(F::Adaptive, P::Native, D),
    path(F::Resilience, P::Native, D),
    path(F::Cleanup, P::Native, D),
    path(F::PolicyQuarantine, P::Native, D),
    path(F::Policy, P::Native, D),
    path(F::Budget, P::Native, D),
    path(F::Profiling, P::Native, D),
    path(F::MetadataColumns, P::Native, D),
    // Columnar: governance runs inside the loop; per-row outcomes do not.
    path(F::MaskingRetypesByValue, P::Columnar, D),
    path(F::DlqAll, P::Columnar, D),
    path(F::ExactlyOnce, P::Columnar, D),
    path(F::Overwrite, P::Columnar, D),
    path(F::GroupedOverwrite, P::Columnar, D),
    path(F::Adaptive, P::Columnar, D),
    path(F::Resilience, P::Columnar, D),
    path(F::Cleanup, P::Columnar, D),
    path(F::PolicyQuarantine, P::Columnar, D),
    path(F::MetadataColumns, P::Columnar, D),
    // The exactly-once watermark commits a whole page atomically.
    path(F::DlqPropagate, P::Watermark, R(EO_WITH_DLQ)),
    path(F::DlqAll, P::Watermark, R(EO_WITH_DLQ)),
    path(F::Overwrite, P::Watermark, R(EO_WITH_OVERWRITE)),
    path(F::GroupedOverwrite, P::Watermark, R(EO_WITH_OVERWRITE)),
    path(
        F::PolicyQuarantine,
        P::Watermark,
        R(EO_WITH_POLICY_QUARANTINE),
    ),
    // The staged overwrite swaps a pre-run copy in whole.
    path(F::Cleanup, P::OverwriteStaging, R(OVERWRITE_WITH_CLEANUP)),
    path(
        F::SchemaDriftEvolve,
        P::OverwriteStaging,
        R(OVERWRITE_WITH_EVOLVE),
    ),
    // Sink capabilities.
    cap(
        F::SchemaDriftFail,
        C::DropsUnknownFields,
        R(DRIFT_FAIL_DROPS_FIELDS),
    ),
    cap(
        F::SchemaDriftEvolve,
        C::NoSchemaEvolution,
        R(DRIFT_EVOLVE_UNSUPPORTED),
    ),
    cap(F::Cleanup, C::NoCleanup, R(CLEANUP_UNSUPPORTED)),
];

/// The verdict for `feature` on `target`.
pub fn verdict(feature: Feature, target: Target) -> Verdict {
    RULES
        .iter()
        .find(|r| r.feature == feature && r.target == target)
        .map_or(Verdict::Supported, |r| r.verdict)
}

/// Whether a run with `features` may take `path`: no feature disables or
/// refuses it.
pub fn path_allowed(path: WritePath, features: &[Feature]) -> bool {
    features
        .iter()
        .all(|f| verdict(*f, Target::Path(path)) == Verdict::Supported)
}

/// The path a feature itself selects, when it does.
pub fn selected_path(feature: Feature) -> Option<WritePath> {
    match feature {
        Feature::ExactlyOnce => Some(WritePath::Watermark),
        Feature::Overwrite | Feature::GroupedOverwrite => Some(WritePath::OverwriteStaging),
        _ => None,
    }
}

/// Every refusal for a run with `features` against a sink with `sink`
/// capabilities: each (feature, target, reason) whose target is a path one of
/// the features selects, or a capability condition the sink meets.
pub fn refusals(
    features: &[Feature],
    sink: Option<&Capabilities>,
) -> Vec<(Feature, Target, &'static str)> {
    let selected: Vec<WritePath> = features.iter().filter_map(|f| selected_path(*f)).collect();
    RULES
        .iter()
        .filter(|r| features.contains(&r.feature))
        .filter(|r| match r.target {
            Target::Path(p) => selected.contains(&p),
            Target::Capability(c) => sink.is_some_and(|s| c.holds(s)),
        })
        .filter_map(|r| match r.verdict {
            Verdict::Refuse(reason) => Some((r.feature, r.target, reason)),
            _ => None,
        })
        .collect()
}

/// The reference page, generated from [`RULES`].
pub fn render_markdown() -> String {
    let mut out = String::new();
    out.push_str("# Feature compatibility\n\n");
    out.push_str(
        "<!-- Generated from `faucet_core::compat::RULES`. Do not edit by hand: run \
         `cargo test -p faucet-cli --test compat_doc -- --ignored regenerate`. -->\n\n",
    );
    out.push_str(
        "Which write path a run takes, and whether a combination is allowed at all, comes \
         from one table in `faucet_core::compat`. `Pipeline::run` picks the fastest path every \
         active feature supports, and `faucet validate` refuses a combination no path \
         honours before anything runs.\n\n\
         - **✓** — supported on that path.\n\
         - **falls back** — that path would skip the feature, so the run takes a slower path \
         that honours it (native → columnar → row).\n\
         - **refused** — the combination is rejected when the config is loaded.\n\n\
         A pair the table does not list is supported.\n\n",
    );
    out.push_str("## Write paths\n\n| Feature |");
    for p in WritePath::ALL {
        out.push_str(&format!(" {} |", p.label()));
    }
    out.push_str("\n|---|");
    for _ in WritePath::ALL {
        out.push_str("---|");
    }
    out.push('\n');
    for f in Feature::ALL {
        out.push_str(&format!("| {} |", f.label()));
        for p in WritePath::ALL {
            out.push_str(&format!(" {} |", cell(verdict(*f, Target::Path(*p)))));
        }
        out.push('\n');
    }
    out.push_str(
        "\nThe exactly-once watermark and overwrite staging columns apply only when the run \
         uses them (`delivery: exactly_once` against an idempotent sink, `write_mode: \
         overwrite`).\n\n## Sink capabilities\n\nA sink declares its capabilities \
         (`Sink::capabilities`, `faucet list --json`). These combinations depend on them:\n\n\
         | Feature | Sink | Verdict |\n|---|---|---|\n",
    );
    for r in RULES {
        if let Target::Capability(c) = r.target {
            out.push_str(&format!(
                "| {} | {} | {} |\n",
                r.feature.label(),
                c.label(),
                cell(r.verdict)
            ));
        }
    }
    out.push_str("\n## Refusal reasons\n\n");
    let mut seen: Vec<&str> = Vec::new();
    for r in RULES {
        if let Verdict::Refuse(reason) = r.verdict
            && !seen.contains(&reason)
        {
            seen.push(reason);
            let target = match r.target {
                Target::Path(p) => p.label(),
                Target::Capability(c) => c.label(),
            };
            out.push_str(&format!(
                "- {} × {}: {}.\n",
                r.feature.label(),
                target,
                reason
            ));
        }
    }
    out
}

fn cell(v: Verdict) -> &'static str {
    match v {
        Verdict::Supported => "✓",
        Verdict::DisablesPath => "falls back",
        Verdict::Refuse(_) => "refused",
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::write_mode::WriteMode;

    #[test]
    fn unlisted_pairs_are_supported() {
        assert_eq!(
            verdict(F::Masking, Target::Path(P::Row)),
            Verdict::Supported
        );
        assert_eq!(
            verdict(F::Masking, Target::Path(P::Columnar)),
            Verdict::Supported
        );
        assert_eq!(verdict(F::Masking, Target::Path(P::Native)), D);
    }

    /// The table's constructors build the rows `verdict` reads.
    #[test]
    fn rule_constructors_build_table_rows() {
        let p = path(F::Overwrite, P::Watermark, R(EO_WITH_OVERWRITE));
        assert_eq!(p.target, Target::Path(P::Watermark));
        assert_eq!(verdict(p.feature, p.target), p.verdict);
        let c = cap(
            F::Cleanup,
            CapabilityCondition::NoCleanup,
            R(CLEANUP_UNSUPPORTED),
        );
        assert_eq!(verdict(c.feature, c.target), c.verdict);
    }

    #[test]
    fn row_path_honours_every_feature() {
        assert!(path_allowed(P::Row, Feature::ALL));
        for f in Feature::ALL {
            assert_eq!(verdict(*f, Target::Path(P::Row)), Verdict::Supported);
        }
    }

    #[test]
    fn path_selection_falls_back() {
        assert!(path_allowed(P::Native, &[F::Overwrite]));
        assert!(!path_allowed(P::Native, &[F::Overwrite, F::KeyedWrites]));
        assert!(path_allowed(P::Columnar, &[F::Masking, F::DlqPropagate]));
        assert!(!path_allowed(P::Columnar, &[F::DlqAll]));
        assert!(!path_allowed(P::Columnar, &[F::PolicyQuarantine]));
        assert!(path_allowed(P::Native, &[]));
    }

    #[test]
    fn refusals_apply_only_to_selected_paths_and_held_conditions() {
        assert!(refusals(&[F::DlqPropagate], None).is_empty());
        let r = refusals(&[F::ExactlyOnce, F::DlqPropagate], None);
        assert_eq!(r.len(), 1);
        assert_eq!(r[0].2, EO_WITH_DLQ);
        let r = refusals(&[F::ExactlyOnce, F::PolicyQuarantine], None);
        assert_eq!(r[0].2, EO_WITH_POLICY_QUARANTINE);
        assert_eq!(
            refusals(&[F::Overwrite, F::Cleanup], None)[0].2,
            OVERWRITE_WITH_CLEANUP
        );

        let drops = Capabilities::append_only_sink()
            .to_builder()
            .schema_report(SchemaReport::DropsUnknownFields)
            .build();
        let r = refusals(&[F::SchemaDrift, F::SchemaDriftFail], Some(&drops));
        assert_eq!(r.len(), 1);
        assert_eq!(r[0].1, Target::Capability(C::DropsUnknownFields));
        assert!(refusals(&[F::SchemaDriftFail], None).is_empty());

        let evolvable = Capabilities::builder()
            .write_modes([WriteMode::Append])
            .schema_evolution(true)
            .cleanup(true)
            .build();
        assert!(refusals(&[F::SchemaDriftEvolve, F::Cleanup], Some(&evolvable)).is_empty());
        assert_eq!(
            refusals(&[F::SchemaDriftEvolve, F::Cleanup], Some(&drops)).len(),
            2
        );
    }

    #[test]
    fn conditions_read_capabilities() {
        let none = Capabilities::append_only_sink();
        assert!(!C::DropsUnknownFields.holds(&none));
        assert!(C::NoSchemaEvolution.holds(&none));
        assert!(C::NoCleanup.holds(&none));
        for c in CapabilityCondition::ALL {
            assert!(!c.label().is_empty());
        }
    }

    #[test]
    fn every_feature_and_path_is_labelled_and_in_all() {
        for f in Feature::ALL {
            assert!(!f.label().is_empty());
        }
        for p in WritePath::ALL {
            assert!(!p.label().is_empty());
        }
        for r in RULES {
            assert!(Feature::ALL.contains(&r.feature));
            assert_ne!(r.verdict, Verdict::Supported, "only list exceptions");
        }
        assert_eq!(selected_path(F::Masking), None);
        assert_eq!(
            selected_path(F::GroupedOverwrite),
            Some(P::OverwriteStaging)
        );
    }

    #[test]
    fn rendered_page_lists_every_feature_and_reason() {
        let md = render_markdown();
        for f in Feature::ALL {
            assert!(md.contains(f.label()), "{f:?}");
        }
        assert!(md.contains(EO_WITH_POLICY_QUARANTINE));
        assert!(md.contains(DRIFT_FAIL_DROPS_FIELDS));
        assert!(
            md.contains("| `schema: { on_drift: fail }` | sink drops unknown fields | refused |")
        );
    }

    #[test]
    fn rules_serialize() {
        let v = serde_json::to_value(RULES[0]).unwrap();
        assert_eq!(v["feature"], "transforms");
        assert_eq!(v["target"]["kind"], "path");
        assert_eq!(v["verdict"]["verdict"], "disables_path");
    }
}