krabka_metadata/

feature.rs

//! KIP-584 feature framework. A [`Feature`] owns the *versioning facts* of one
//! cluster feature: the supported range, the per-release default, the downgrade
//! floor, the KIP-1022 dependencies, and an optional level name. The static
//! [`feature_registry`] is the single source of truth for `ApiVersions`,
//! `UpdateFeatures`, the `krabka format` bootstrap, and the Raft range guards.
//! Behavioral *gating*, that is, the rejection of RPCs below a level, lives in
//! the broker handlers that read the finalized level from the image, not here.

use std::collections::BTreeMap;

use crate::{MetadataImage, group_version, krabka_version, metadata_version, transaction_version};

/// One versioned cluster feature (KIP-584).
pub trait Feature: Sync {
    /// KIP-584 feature name, e.g. `"metadata.version"`.
    fn name(&self) -> &'static str;

    /// Inclusive `[min, max]` supported level range. Advertised in
    /// `ApiVersions.supported_features` and accepted by `UpdateFeatures`.
    fn supported_range(&self) -> (i16, i16);

    /// The level finalized at `krabka format` for the bootstrap
    /// `metadata.version` level, which is the resolved `--release-version`.
    /// Kafka derives the default of every feature from the release in this way.
    fn default_level(&self, bootstrap_mv: i16) -> i16;

    /// Lowest level that the live image permits for a finalize or a downgrade:
    /// the "unsafe downgrade" floor. It defaults to the supported min, where no
    /// live-state constraint applies.
    fn min_required_floor(&self, _image: &MetadataImage) -> i16 {
        self.supported_range().0
    }

    /// KIP-1022 dependencies: `(other_feature, min_level)` pairs that must be
    /// finalized at `>= min_level` before THIS feature may be finalized at
    /// `level`. Empty by default.
    fn dependencies(&self, _level: i16) -> &'static [(&'static str, i16)] {
        &[]
    }

    /// Optional Kafka level name (e.g. metadata.version's `"3.7-IV4"`).
    /// `None` for plain integer features.
    fn level_name(&self, _level: i16) -> Option<&'static str> {
        None
    }
}

/// `metadata.version` (KIP-584 / KIP-778). The range, the string table, and the
/// SCRAM and delegation-token floor come from [`crate::metadata_version`].
pub struct MetadataVersionFeature;

impl Feature for MetadataVersionFeature {
    fn name(&self) -> &'static str {
        metadata_version::METADATA_VERSION_FEATURE
    }
    fn supported_range(&self) -> (i16, i16) {
        (
            metadata_version::METADATA_VERSION_MIN,
            metadata_version::METADATA_VERSION_MAX,
        )
    }
    fn default_level(&self, bootstrap_mv: i16) -> i16 {
        // For metadata.version the release string IS the metadata.version, so
        // the default is the bootstrap level itself, clamped into range.
        bootstrap_mv.clamp(
            metadata_version::METADATA_VERSION_MIN,
            metadata_version::METADATA_VERSION_MAX,
        )
    }
    fn min_required_floor(&self, image: &MetadataImage) -> i16 {
        image.min_required_metadata_version()
    }
    fn level_name(&self, level: i16) -> Option<&'static str> {
        metadata_version::from_feature_level(level).map(metadata_version::MetadataVersion::ivn)
    }
}

/// `group.version` (KIP-848). A plain integer feature. The default rises to 1
/// once the bootstrap metadata.version reaches the KIP-848 GA level.
pub struct GroupVersionFeature;

impl Feature for GroupVersionFeature {
    fn name(&self) -> &'static str {
        group_version::GROUP_VERSION_FEATURE
    }
    fn supported_range(&self) -> (i16, i16) {
        (
            group_version::GROUP_VERSION_MIN,
            group_version::GROUP_VERSION_MAX,
        )
    }
    fn default_level(&self, bootstrap_mv: i16) -> i16 {
        if bootstrap_mv >= group_version::GROUP_VERSION_GA_METADATA_LEVEL {
            group_version::GROUP_VERSION_MAX
        } else {
            group_version::GROUP_VERSION_MIN
        }
    }
    // min_required_floor: inherits the default (supported min). Next-gen group
    // state lives in the coordinator / __consumer_offsets, NOT the
    // MetadataImage, so a live-state-aware downgrade floor can't be computed
    // here. dependencies: inherits the empty default — Kafka 4.0
    // declares no hard `UpdateFeatures` dependency for group.version.
}

/// `transaction.version` (KIP-890). The default jumps to 2 once the bootstrap
/// metadata.version reaches 4.0-IV2, and 2 is the highest level Kafka defines.
/// The downgrade floor is the supported min, because in-flight txn state lives
/// in the `__transaction_state` log and not in the [`MetadataImage`], so this
/// module computes no image-derived floor.
pub struct TransactionVersionFeature;

impl Feature for TransactionVersionFeature {
    fn name(&self) -> &'static str {
        transaction_version::TRANSACTION_VERSION_FEATURE
    }
    fn supported_range(&self) -> (i16, i16) {
        (
            transaction_version::TRANSACTION_VERSION_MIN,
            transaction_version::TRANSACTION_VERSION_MAX,
        )
    }
    // The TV_1 tier is retained as an explicit (currently coincident) threshold
    // so a future Kafka release that splits TV_1's bootstrap level below TV_2's
    // is a one-line constant change.
    fn default_level(&self, bootstrap_mv: i16) -> i16 {
        // Both TV_1 and TV_2 bootstrap at level 24 (4.0-IV2) → default jumps
        // 0 -> 2 at >= 24. Empirically pinned.
        use crate::transaction_version::{TV1_METADATA_LEVEL, TV2_METADATA_LEVEL};
        match bootstrap_mv {
            level if level >= TV2_METADATA_LEVEL => 2,
            level if level >= TV1_METADATA_LEVEL => 1,
            _ => 0,
        }
    }
    // dependencies + min_required_floor: inherit the empty/supported-min defaults.
}

/// `share.version` (KIP-932). A plain integer feature that gates share-group
/// membership. The default stays at the supported min, 0, which is disabled,
/// until the bootstrap metadata.version reaches `4.2-IV0`, where Kafka's
/// `ShareVersion.SV_1` bootstraps. Trunk's `SV_2` (KIP-1191 dead-letter
/// queues) bootstraps at `4.4-IV0`. The registry lists it; a node supports
/// level 2 only when it supports unstable feature versions, because Kafka
/// 4.3.1's latest production level is 1.
pub struct ShareVersionFeature;

impl Feature for ShareVersionFeature {
    fn name(&self) -> &'static str {
        metadata_version::SHARE_VERSION_FEATURE
    }
    fn supported_range(&self) -> (i16, i16) {
        (
            metadata_version::SHARE_VERSION_MIN,
            metadata_version::SHARE_VERSION_MAX,
        )
    }
    fn default_level(&self, bootstrap_mv: i16) -> i16 {
        use crate::metadata_version::{
            SHARE_VERSION_2_DEFAULT_METADATA_LEVEL, SHARE_VERSION_DEFAULT_METADATA_LEVEL,
        };
        match bootstrap_mv {
            level if level >= SHARE_VERSION_2_DEFAULT_METADATA_LEVEL => 2,
            level if level >= SHARE_VERSION_DEFAULT_METADATA_LEVEL => 1,
            _ => metadata_version::SHARE_VERSION_MIN,
        }
    }
    // dependencies + min_required_floor: inherit the empty/supported-min defaults.
}

/// `streams.version` (KIP-1071). A plain integer feature that gates the
/// broker-side Streams rebalance protocol. The default stays at the supported
/// min, 0, which is disabled, until the bootstrap metadata.version reaches
/// `4.2-IV1`, where Kafka's `StreamsVersion.SV_1` bootstraps. Below that an
/// operator opts in through `UpdateFeatures`.
pub struct StreamsVersionFeature;

impl Feature for StreamsVersionFeature {
    fn name(&self) -> &'static str {
        metadata_version::STREAMS_VERSION_FEATURE
    }
    fn supported_range(&self) -> (i16, i16) {
        (
            metadata_version::STREAMS_VERSION_MIN,
            metadata_version::STREAMS_VERSION_MAX,
        )
    }
    fn default_level(&self, bootstrap_mv: i16) -> i16 {
        if bootstrap_mv >= metadata_version::STREAMS_VERSION_DEFAULT_METADATA_LEVEL {
            metadata_version::STREAMS_VERSION_MAX
        } else {
            metadata_version::STREAMS_VERSION_MIN
        }
    }
    // dependencies + min_required_floor: inherit the empty/supported-min
    // defaults. Streams group state lives in __consumer_offsets, not the
    // MetadataImage, so a live-state-aware downgrade floor can't be computed
    // here, mirroring group.version / share.version.
}

/// `eligible.leader.replicas.version` (KIP-966). A plain integer feature that
/// gates the controller's ELR bookkeeping: at level 0 no partition gains an
/// eligible-leader or last-known-leader set, at level 1 the controller
/// maintains both.
///
/// `EligibleLeaderReplicasVersion` in apache/kafka bootstraps `ELRV_1` at
/// `4.1-IV0` and declares one KIP-1022 dependency, `metadata.version` at
/// `4.0-IV1`, the level whose `PartitionRecord` carries the ELR fields. The
/// release default is therefore 1 from `4.1-IV0` and 0 below it, exactly as
/// Kafka's `Feature.defaultLevel` computes it. Below `4.1-IV0` an operator
/// turns the feature on with `kafka-features upgrade`.
pub struct ElrVersionFeature;

impl Feature for ElrVersionFeature {
    fn name(&self) -> &'static str {
        metadata_version::ELR_VERSION_FEATURE
    }
    fn supported_range(&self) -> (i16, i16) {
        (
            metadata_version::ELR_VERSION_MIN,
            metadata_version::ELR_VERSION_MAX,
        )
    }
    fn default_level(&self, bootstrap_mv: i16) -> i16 {
        if bootstrap_mv >= metadata_version::ELR_DEFAULT_METADATA_LEVEL {
            metadata_version::ELR_VERSION_MAX
        } else {
            metadata_version::ELR_VERSION_MIN
        }
    }
    fn dependencies(&self, level: i16) -> &'static [(&'static str, i16)] {
        if level >= 1 {
            const ELRV_1_DEPENDENCIES: &[(&str, i16)] = &[(
                metadata_version::METADATA_VERSION_FEATURE,
                metadata_version::ELR_MIN_LEVEL,
            )];
            ELRV_1_DEPENDENCIES
        } else {
            &[]
        }
    }
    // min_required_floor: inherits the supported min. ELR membership lives in
    // the MetadataImage, but Kafka clears it with PartitionChangeRecords on a
    // downgrade to 0 rather than refusing the downgrade, so no floor applies.
}

/// `kraft.version` is finalized by a `KRaft` control record, never by a
/// KIP-584 `FeatureLevelRecord`.
pub struct KRaftVersionFeature;

impl Feature for KRaftVersionFeature {
    fn name(&self) -> &'static str {
        metadata_version::KRAFT_VERSION_FEATURE
    }
    fn supported_range(&self) -> (i16, i16) {
        (0, 1)
    }
    fn default_level(&self, _bootstrap_mv: i16) -> i16 {
        0
    }
}

/// `krabka.version`, the krabka-owned feature that gates krabka-only on-disk
/// and inter-node format changes (see [`crate::krabka_version`]). A fresh
/// cluster bootstraps at level 0 whatever the bootstrap metadata.version is,
/// so the bootstrap writes no record for it. An Apache Kafka node in the same
/// cluster supports only level 0 of a feature it does not know.
pub struct KrabkaVersionFeature;

impl Feature for KrabkaVersionFeature {
    fn name(&self) -> &'static str {
        krabka_version::KRABKA_VERSION_FEATURE
    }
    fn supported_range(&self) -> (i16, i16) {
        (
            krabka_version::KRABKA_VERSION_MIN,
            krabka_version::KRABKA_VERSION_MAX,
        )
    }
    fn default_level(&self, _bootstrap_mv: i16) -> i16 {
        krabka_version::KRABKA_VERSION_MIN
    }
    // dependencies: inherits the empty default. No krabka.version level needs a
    // metadata.version yet. min_required_floor: inherits the supported min.
    // Levels 0 and 1 are the same formats, and every build that supports level
    // 2 or 3 reads that level's formats at every level, so lowering it strands
    // nothing.
}

/// All features this broker supports finalizing. Single source of truth.
#[must_use]
pub fn feature_registry() -> &'static [&'static dyn Feature] {
    const REGISTRY: &[&dyn Feature] = &[
        &MetadataVersionFeature,
        &GroupVersionFeature,
        &TransactionVersionFeature,
        &ShareVersionFeature,
        &StreamsVersionFeature,
        &ElrVersionFeature,
        &KRaftVersionFeature,
        &KrabkaVersionFeature,
    ];
    REGISTRY
}

/// Supported feature ranges advertised by a Krabka broker or controller when
/// it registers with the `KRaft` controller.
#[must_use]
pub fn supported_feature_ranges() -> BTreeMap<String, (i16, i16)> {
    let mut ranges: BTreeMap<_, _> = feature_registry()
        .iter()
        .map(|feature| (feature.name().to_owned(), feature.supported_range()))
        .collect();
    // Registration-only KIP-1155 capability signal. Keep it out of the
    // finalizable feature registry: upstream has not assigned the KIP's
    // promised metadata.version level, and inventing one would make
    // ApiVersions claim support for an undefined metadata record format.
    ranges.insert(
        metadata_version::METADATA_DOWNGRADE_CAPABILITY_FEATURE.into(),
        (
            metadata_version::METADATA_DOWNGRADE_CAPABILITY_LEVEL,
            metadata_version::METADATA_DOWNGRADE_CAPABILITY_LEVEL,
        ),
    );
    ranges
}

/// Look up a registered feature by name.
#[must_use]
pub fn feature(name: &str) -> Option<&'static dyn Feature> {
    feature_registry()
        .iter()
        .copied()
        .find(|f| f.name() == name)
}

/// KIP-584/1022 bootstrap: one `V1FeatureLevel` record per registered feature
/// at its per-release default, derived from `bootstrap_mv`, the bootstrap
/// metadata.version level.
///
/// Both `krabka format` and the broker's standalone self-bootstrap use this, so
/// a fresh cluster finalizes the release default of every feature.
#[must_use]
pub fn bootstrap_feature_records(bootstrap_mv: i16) -> Vec<crate::MetadataRecord> {
    bootstrap_feature_records_with_overrides(bootstrap_mv, &BTreeMap::new())
}

/// KIP-1022 `krabka format` seeding: one `V1FeatureLevel` record per registered
/// feature.
///
/// Each feature is at its explicit `--feature NAME=LEVEL` override if present
/// in `overrides`, else at its per-release default for `bootstrap_mv`.
///
/// Mirrors `kafka-storage format`: this function **omits** a feature whose
/// resolved level is `0`. Level 0 means absent and disabled, which is the
/// default state. The seeded record set therefore matches what Kafka writes to
/// its `bootstrap.checkpoint`.
#[must_use]
pub fn bootstrap_feature_records_with_overrides(
    bootstrap_mv: i16,
    overrides: &BTreeMap<String, i16>,
) -> Vec<crate::MetadataRecord> {
    feature_registry()
        .iter()
        .filter(|f| f.name() != metadata_version::KRAFT_VERSION_FEATURE)
        .filter_map(|f| {
            let level = overrides
                .get(f.name())
                .copied()
                .unwrap_or_else(|| f.default_level(bootstrap_mv));
            (level > 0).then(|| {
                crate::MetadataRecord::V1FeatureLevel(crate::FeatureLevelRecord {
                    name: f.name().to_owned(),
                    level,
                })
            })
        })
        .collect()
}

/// KIP-1022 dependency validation for a fully-resolved feature→level map, as
/// seeded by `krabka format`.
///
/// For every finalized feature, each of its `dependencies(level)` must be
/// present in `resolved` at `>=` the required level. Returns `Err` with the
/// name of the first unmet dependency, enforcing the rule at format time in the
/// same way as the `UpdateFeatures` handler.
/// # Errors
/// Returns an error naming the first finalized feature whose required
/// dependency is absent or finalized below the minimum level.
pub fn validate_feature_dependencies(resolved: &BTreeMap<String, i16>) -> Result<(), String> {
    check_deps(resolved, |name, level| {
        feature(name).map_or(&[][..], |f| f.dependencies(level))
    })
}

/// Core of [`validate_feature_dependencies`], parameterized over the dependency
/// source so the rejection logic is unit-testable without a dependency-bearing
/// feature in the real registry.
fn check_deps(
    resolved: &BTreeMap<String, i16>,
    deps_of: impl Fn(&str, i16) -> &'static [(&'static str, i16)],
) -> Result<(), String> {
    for (name, &level) in resolved {
        for &(dep, min) in deps_of(name, level) {
            let have = resolved.get(dep).copied().unwrap_or(0);
            if have < min {
                return Err(format!(
                    "feature {name}={level} requires {dep}>={min}, but {dep} is finalized at {have}"
                ));
            }
        }
    }
    Ok(())
}

/// True if `level` is within the registered feature's supported range.
/// `false` for an unknown feature (nothing supports that level).
#[must_use]
pub fn is_supported_level(name: &str, level: i16) -> bool {
    feature(name).is_some_and(|f| {
        let (min, max) = f.supported_range();
        (min..=max).contains(&level)
    })
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use assert2::check;

    use super::*;

    #[test]
    fn registry_contains_metadata_version() {
        let f = feature("metadata.version").expect("registered");
        assert2::assert!(f.supported_range() == (7, 34));
        assert2::assert!(feature("not.a.feature").is_none());
    }

    #[test]
    fn metadata_version_default_is_the_bootstrap_level_clamped() {
        let f = feature("metadata.version").unwrap();
        for (_case, bootstrap, want) in [
            ("maximum bootstrap", 34, 34),
            ("minimum bootstrap", 7, 7),
            ("above maximum", 99, 34),
            ("below minimum", 1, 7),
        ] {
            assert2::assert!(f.default_level(bootstrap) == want);
        }
    }

    #[test]
    fn is_supported_level_checks_range() {
        for (name, level, want) in [
            ("metadata.version", 7, true),
            ("metadata.version", 34, true),
            ("metadata.version", 6, false),
            ("metadata.version", 35, false),
            // Kafka's TransactionVersion has TV_0 to TV_2 only (broker #784).
            ("transaction.version", 2, true),
            ("transaction.version", 3, false),
            ("krabka.version", 0, true),
            ("krabka.version", 1, true),
            ("krabka.version", 2, true),
            ("krabka.version", 3, true),
            ("krabka.version", 4, false),
            ("not.a.feature", 1, false),
        ] {
            assert2::assert!(is_supported_level(name, level) == want);
        }
    }

    /// The whole map a node advertises at registration and in `ApiVersions`.
    #[test]
    fn supported_feature_ranges_are_pinned() {
        let want: BTreeMap<String, (i16, i16)> = [
            ("eligible.leader.replicas.version", (0, 1)),
            ("group.version", (0, 1)),
            ("krabka.metadata.downgrade", (1, 1)),
            ("krabka.version", (0, 3)),
            ("kraft.version", (0, 1)),
            ("metadata.version", (7, 34)),
            ("share.version", (0, 2)),
            ("streams.version", (0, 1)),
            ("transaction.version", (0, 2)),
        ]
        .into_iter()
        .map(|(name, range)| (name.to_owned(), range))
        .collect();
        check!(supported_feature_ranges() == want);
    }

    /// `krabka.version` bootstraps at 0, so it earns no record, whatever the
    /// bootstrap metadata.version. An explicit override still finalizes it.
    #[test]
    fn krabka_version_bootstraps_at_zero() {
        use crate::metadata_version::{METADATA_VERSION_MAX, METADATA_VERSION_MIN};
        for (case, bootstrap, overrides, want) in [
            ("minimum release", METADATA_VERSION_MIN, vec![], None),
            ("latest release", METADATA_VERSION_MAX, vec![], None),
            (
                "override to 1 earns a record",
                METADATA_VERSION_MAX,
                vec![("krabka.version".to_owned(), 1)],
                Some(1),
            ),
            (
                "override to 0 omits the record",
                METADATA_VERSION_MAX,
                vec![("krabka.version".to_owned(), 0)],
                None,
            ),
        ] {
            let overrides: BTreeMap<String, i16> = overrides.into_iter().collect();
            let levels = levels_of(&bootstrap_feature_records_with_overrides(
                bootstrap, &overrides,
            ));
            check!(levels.get("krabka.version").copied() == want, "{case}");
        }
    }

    #[test]
    fn krabka_version_declares_no_dependencies_and_no_level_name() {
        let f = feature("krabka.version").expect("registered");
        check!((f.dependencies(0), f.dependencies(1), f.level_name(1)) == (&[][..], &[][..], None));
        let resolved = BTreeMap::from([("krabka.version".to_owned(), 1)]);
        check!(validate_feature_dependencies(&resolved) == Ok(()));
    }

    #[test]
    fn downgrade_capability_is_registration_only() {
        let ranges = supported_feature_ranges();
        check!(
            ranges.get(metadata_version::METADATA_DOWNGRADE_CAPABILITY_FEATURE)
                == Some(&(
                    metadata_version::METADATA_DOWNGRADE_CAPABILITY_LEVEL,
                    metadata_version::METADATA_DOWNGRADE_CAPABILITY_LEVEL,
                ))
        );
        check!(feature(metadata_version::METADATA_DOWNGRADE_CAPABILITY_FEATURE).is_none());
    }

    #[test]
    fn metadata_version_level_name() {
        let f = feature("metadata.version").unwrap();
        for (_case, level, want) in [
            ("latest level", 34, Some("4.5-IV0")),
            ("controller unregistration level", 33, Some("4.4-IV2")),
            ("CIDR ACL level", 32, Some("4.4-IV1")),
            ("cordoned log dirs level", 30, Some("4.3-IV0")),
            ("earliest level", 7, Some("3.3-IV3")),
            ("unknown level", 99, None),
        ] {
            assert2::assert!(f.level_name(level) == want);
        }
    }

    #[test]
    fn group_version_registered_with_range() {
        let f = feature("group.version").expect("registered");
        assert2::assert!(f.supported_range() == (0, 1));
    }

    #[test]
    fn registry_feature_contracts_are_pinned() {
        let image = MetadataImage::new(uuid::Uuid::nil());
        let expected = [
            ("metadata.version", (7, 34), 25, 7),
            ("group.version", (0, 1), 1, 0),
            ("transaction.version", (0, 2), 2, 0),
            ("share.version", (0, 2), 0, 0),
            ("streams.version", (0, 1), 0, 0),
            ("eligible.leader.replicas.version", (0, 1), 0, 0),
            // Registered like the rest, though it is finalized by a KRaft
            // control record rather than by UpdateFeatures. Leaving it out of
            // this table left its name, range and default entirely unasserted.
            ("kraft.version", (0, 1), 0, 0),
            ("krabka.version", (0, 3), 0, 0),
        ];

        for (name, range, default_at_25, floor) in expected {
            let f = feature(name).expect("registered");
            check!(
                (
                    f.supported_range(),
                    f.default_level(25),
                    f.min_required_floor(&image),
                ) == (range, default_at_25, floor),
                "feature {name}"
            );
        }
    }

    #[test]
    fn group_version_default_follows_release() {
        let f = feature("group.version").unwrap();
        for (_case, bootstrap, want) in [
            (
                "before GA threshold",
                group_version::GROUP_VERSION_GA_METADATA_LEVEL - 1,
                0,
            ),
            (
                "at GA threshold",
                group_version::GROUP_VERSION_GA_METADATA_LEVEL,
                1,
            ),
            ("after GA threshold", 25, 1),
        ] {
            assert2::assert!(f.default_level(bootstrap) == want);
        }
    }

    #[test]
    fn group_version_declares_no_hard_dependencies() {
        let f = feature("group.version").unwrap();
        assert2::assert!(
            (f.dependencies(0).is_empty(), f.dependencies(1).is_empty()) == (true, true)
        );
    }

    #[test]
    fn transaction_version_registered() {
        let f = feature("transaction.version").expect("registered");
        assert2::assert!(f.supported_range() == (0, 2));
    }

    #[test]
    fn streams_version_registered_with_no_dependencies() {
        let f = feature("streams.version").expect("registered");
        check!((f.supported_range(), f.dependencies(1).is_empty()) == ((0, 1), true));
    }

    /// Kafka's `Feature.defaultLevel`: the highest feature level whose
    /// bootstrap metadata version is at or below the bootstrap level.
    #[test]
    fn opt_in_feature_defaults_follow_kafka_bootstrap_levels() {
        use crate::metadata_version::{
            ELR_DEFAULT_METADATA_LEVEL, METADATA_VERSION_MAX,
            SHARE_VERSION_2_DEFAULT_METADATA_LEVEL, SHARE_VERSION_DEFAULT_METADATA_LEVEL,
            STREAMS_VERSION_DEFAULT_METADATA_LEVEL,
        };
        for (name, bootstrap, want) in [
            ("eligible.leader.replicas.version", 25, 0),
            (
                "eligible.leader.replicas.version",
                ELR_DEFAULT_METADATA_LEVEL - 1,
                0,
            ),
            (
                "eligible.leader.replicas.version",
                ELR_DEFAULT_METADATA_LEVEL,
                1,
            ),
            ("eligible.leader.replicas.version", METADATA_VERSION_MAX, 1),
            ("share.version", SHARE_VERSION_DEFAULT_METADATA_LEVEL - 1, 0),
            ("share.version", SHARE_VERSION_DEFAULT_METADATA_LEVEL, 1),
            (
                "share.version",
                SHARE_VERSION_2_DEFAULT_METADATA_LEVEL - 1,
                1,
            ),
            ("share.version", SHARE_VERSION_2_DEFAULT_METADATA_LEVEL, 2),
            ("share.version", METADATA_VERSION_MAX, 2),
            (
                "streams.version",
                STREAMS_VERSION_DEFAULT_METADATA_LEVEL - 1,
                0,
            ),
            ("streams.version", STREAMS_VERSION_DEFAULT_METADATA_LEVEL, 1),
            ("streams.version", METADATA_VERSION_MAX, 1),
        ] {
            check!(
                feature(name).unwrap().default_level(bootstrap) == want,
                "feature {name} at bootstrap level {bootstrap}"
            );
        }
    }

    #[test]
    fn elr_version_is_opt_in_and_depends_on_the_elr_metadata_level() {
        let f = feature("eligible.leader.replicas.version").expect("registered");
        // ELRV_1 bootstraps at 4.1-IV0, so the latest level enables it.
        check!(
            (
                f.supported_range(),
                f.default_level(metadata_version::METADATA_VERSION_MAX),
                f.dependencies(0),
                f.dependencies(1),
            ) == (
                (0, 1),
                1,
                &[][..],
                &[(
                    metadata_version::METADATA_VERSION_FEATURE,
                    metadata_version::ELR_MIN_LEVEL,
                )][..],
            )
        );
    }

    #[test]
    fn elr_version_level_one_requires_the_elr_metadata_version() {
        for (_case, metadata_level, want_err) in [
            ("below 4.0-IV1", metadata_version::ELR_MIN_LEVEL - 1, true),
            ("at 4.0-IV1", metadata_version::ELR_MIN_LEVEL, false),
            (
                "above 4.0-IV1",
                metadata_version::METADATA_VERSION_MAX,
                false,
            ),
        ] {
            let resolved = BTreeMap::from([
                (
                    metadata_version::METADATA_VERSION_FEATURE.to_owned(),
                    metadata_level,
                ),
                (metadata_version::ELR_VERSION_FEATURE.to_owned(), 1),
            ]);
            check!(validate_feature_dependencies(&resolved).is_err() == want_err);
        }
    }

    #[test]
    fn transaction_version_default_jumps_to_two_at_4_0_iv2() {
        let f = feature("transaction.version").unwrap();
        for (_case, bootstrap, want) in [
            ("below activation", 23, 0),
            ("at activation", 24, 2),
            ("after activation", 25, 2),
        ] {
            assert2::assert!(f.default_level(bootstrap) == want);
        }
    }

    #[test]
    fn transaction_version_declares_no_hard_dependencies() {
        let f = feature("transaction.version").unwrap();
        for level in [0, 1, 2] {
            assert2::assert!(f.dependencies(level).is_empty());
        }
    }

    #[test]
    fn transaction_version_ga_threshold_is_4_0_iv2() {
        // Anchor the bare 24 to the metadata.version table (mirrors the
        // group.version GA-threshold anchor test).
        assert2::assert!(
            metadata_version::from_feature_level(transaction_version::TV2_METADATA_LEVEL)
                .unwrap()
                .ivn()
                == "4.0-IV2"
        );
    }

    fn levels_of(recs: &[crate::MetadataRecord]) -> BTreeMap<String, i16> {
        recs.iter()
            .filter_map(|r| match r {
                crate::MetadataRecord::V1FeatureLevel(f) => Some((f.name.clone(), f.level)),
                _ => None,
            })
            .collect()
    }

    #[test]
    fn bootstrap_omits_level_zero_features() {
        // share.version / streams.version / ELR default to 0 at 4.0-IV3 → no
        // record emitted (level 0 = absent = disabled, like Kafka's format).
        let levels = levels_of(&bootstrap_feature_records(25));
        assert2::assert!(
            levels
                == BTreeMap::from([
                    ("metadata.version".to_owned(), 25),
                    ("group.version".to_owned(), 1),
                    ("transaction.version".to_owned(), 2),
                ])
        );
    }

    /// The whole record set, in registry order. `krabka.version` defaults to
    /// 0, so it earns no record, as Kafka writes none for a level-0 feature.
    #[test]
    fn bootstrap_at_the_latest_level_enables_every_released_feature() {
        let want: Vec<crate::MetadataRecord> = [
            ("metadata.version", 34),
            ("group.version", 1),
            ("transaction.version", 2),
            ("share.version", 2),
            ("streams.version", 1),
            ("eligible.leader.replicas.version", 1),
        ]
        .into_iter()
        .map(|(name, level)| {
            crate::MetadataRecord::V1FeatureLevel(crate::FeatureLevelRecord {
                name: name.to_owned(),
                level,
            })
        })
        .collect();
        check!(bootstrap_feature_records(metadata_version::METADATA_VERSION_MAX) == want);
    }

    #[test]
    fn bootstrap_with_overrides_applies_explicit_levels() {
        // group.version overridden down to 0 → omitted; the rest follow mv=25.
        let mut ov = BTreeMap::new();
        ov.insert("group.version".to_owned(), 0i16);
        let levels = levels_of(&bootstrap_feature_records_with_overrides(25, &ov));
        assert2::assert!(
            levels
                == BTreeMap::from([
                    ("metadata.version".to_owned(), 25),
                    ("transaction.version".to_owned(), 2),
                ])
        );
    }

    #[test]
    fn bootstrap_override_can_enable_an_opt_in_feature() {
        // streams.version is 0 by default below 4.2-IV1; an explicit override
        // turns it on and earns a record.
        let mut ov = BTreeMap::new();
        ov.insert("streams.version".to_owned(), 1i16);
        let levels = levels_of(&bootstrap_feature_records_with_overrides(25, &ov));
        assert2::assert!(levels.get("streams.version") == Some(&1));
    }

    #[test]
    fn bootstrap_unlisted_feature_follows_bootstrap_mv() {
        // mv=23: at/above group GA (22) but below txn GA (24) → group=1,
        // transaction omitted (default 0), metadata.version mirrors mv.
        let levels = levels_of(&bootstrap_feature_records_with_overrides(
            23,
            &BTreeMap::new(),
        ));
        assert2::assert!(
            levels
                == BTreeMap::from([
                    ("metadata.version".to_owned(), 23),
                    ("group.version".to_owned(), 1),
                ])
        );
    }

    #[test]
    fn validate_dependencies_ok_for_real_registry() {
        // The real registry declares no dependencies, so any resolved set passes.
        let mut resolved = BTreeMap::new();
        resolved.insert("metadata.version".to_owned(), 25i16);
        resolved.insert("group.version".to_owned(), 1i16);
        resolved.insert("transaction.version".to_owned(), 2i16);
        assert2::assert!(validate_feature_dependencies(&resolved).is_ok());
    }

    #[test]
    fn check_deps_enforces_minimum_dependency_levels() {
        // Synthetic dependency source: "b" at level 1 requires "a" >= 2.
        fn deps_of(name: &str, level: i16) -> &'static [(&'static str, i16)] {
            match (name, level) {
                ("b", 1) => &[("a", 2)],
                _ => &[],
            }
        }
        let mut ok = BTreeMap::new();
        ok.insert("a".to_owned(), 2i16);
        ok.insert("b".to_owned(), 1i16);
        assert2::assert!(check_deps(&ok, deps_of).is_ok());

        let mut too_low = BTreeMap::new();
        too_low.insert("a".to_owned(), 1i16);
        too_low.insert("b".to_owned(), 1i16);
        assert2::assert!(check_deps(&too_low, deps_of).is_err());

        // dependency feature absent entirely (treated as level 0) → rejected.
        let mut missing = BTreeMap::new();
        missing.insert("b".to_owned(), 1i16);
        assert2::assert!(check_deps(&missing, deps_of).is_err());
    }

    #[test]
    fn group_version_ga_threshold_is_4_0_iv0() {
        // Anchor the bare `22` to the metadata.version table so the
        // bootstrap-default threshold can't silently drift from `4.0-IV0`
        // (mirrors metadata_version's `gate_level_constants`).
        assert2::assert!(
            metadata_version::from_feature_level(group_version::GROUP_VERSION_GA_METADATA_LEVEL)
                .unwrap()
                .ivn()
                == "4.0-IV0"
        );
    }

    // --- mutation-coverage tests --------------------------------------------
    //
    // Exercise the per-feature accessors the suite above never reads directly.
    // Some sub-mutants are equivalent and intentionally left: a feature whose
    // supported min is 0 makes "replace floor with 0" a no-op; group/streams
    // `supported_range` and `default_level` mutated to their real `(0,1)` / `0`;
    // `dependencies` `&[]` vs a leaked empty slice; and `validate_feature_
    // dependencies -> Ok(())` (the live registry declares no dependencies).

    #[test]
    fn default_min_required_floor_is_supported_min() {
        // group.version uses the default min_required_floor -> supported min (0).
        let img = MetadataImage::new(uuid::Uuid::nil());
        assert2::assert!(feature("group.version").unwrap().min_required_floor(&img) == 0);
    }

    #[test]
    fn metadata_version_min_required_floor_tracks_image() {
        // The override returns the image's min_required_metadata_version.
        let img = MetadataImage::new(uuid::Uuid::nil());
        let mv = feature("metadata.version").unwrap();
        // Empty image floor is METADATA_VERSION_MIN (7), distinct from 0/1/-1.
        assert2::assert!(mv.min_required_floor(&img) == 7);
        assert2::assert!(img.min_required_metadata_version() == 7);
    }

    #[test]
    fn plain_features_have_no_level_name() {
        // Integer features use the default level_name -> None.
        assert2::assert!(feature("group.version").unwrap().level_name(1).is_none());
        assert2::assert!(
            feature("transaction.version")
                .unwrap()
                .level_name(2)
                .is_none()
        );
    }

    #[test]
    fn share_version_accessors() {
        let f = feature("share.version").expect("registered");
        // Opt-in below 4.2-IV0, where Kafka's SV_1 bootstraps.
        check!(
            (f.name(), f.supported_range(), f.default_level(25)) == ("share.version", (0, 2), 0)
        );
    }
}