krabka_metadata/

image.rs

//! Immutable snapshot of the cluster's metadata state. Only
//! [`MetadataImage::apply`] mutates it, and the Raft state machine calls that
//! method. Everywhere else reads it through shared references and `Arc` clones.

mod acl_index;

use std::{
    collections::{BTreeMap, HashMap, HashSet},
    ops::Bound,
};

use krabka_security::{KafkaPrincipal, SaslMechanism, ScramCredential};
use krabka_units::{ByteRate, convert::ByteRateExt};
use uuid::Uuid;

use self::acl_index::AclIndex;
use crate::{
    acl::{AclEntry, PatternType, ResourceType},
    break_glass::{BreakGlassApproval, BreakGlassProposalRecord},
    directory_id::MIGRATING,
    error::MetadataError,
    metadata_version,
    records::{
        BrokerConfigRecord, BrokerRegistrationChangeRecord, BrokerRegistrationRecord,
        ClientMetricsConfigRecord, ClientQuotaRecord, ControllerRegistrationRecord,
        DEFAULT_BROKER_CONFIG_NODE_ID, DelegationTokenRecord, FeatureLevelRecord,
        FeaturesEpochRecord, GroupConfigRecord, KRaftVersionRecord, LeaderRecoveryState,
        MetadataRecord, NodeId, PartitionDirAssignmentRecord, PartitionElrRecord,
        PartitionOffsetAdvanceRecord, PartitionRecord, PartitionRecoveryRecord, ProducerIdsRecord,
        QuotaEntity, ScramCredentialRecord, TopicConfigRecord, TopicRecord, UnregisterBrokerRecord,
        VotersRecord,
    },
    voters::VoterSet,
    write_freeze::TopicFreezeRecord,
};

pub type EntityKey = Vec<(String, Option<String>)>;

/// Eligible replicas and the last known ELR for one partition.
type PartitionElrState = (Vec<NodeId>, Vec<NodeId>);

/// In-memory image type for a single delegation
/// token (KIP-48). Mirrors [`DelegationTokenRecord`] without any tombstone
/// concerns. The apply path handles tombstones as removals.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DelegationToken {
    pub token_id: String,
    pub owner: KafkaPrincipal,
    pub requester: KafkaPrincipal,
    pub issue_timestamp_ms: i64,
    pub expiry_timestamp_ms: i64,
    pub max_timestamp_ms: i64,
    pub renewers: Vec<KafkaPrincipal>,
}

impl DelegationToken {
    #[must_use]
    pub fn from_record(rec: &DelegationTokenRecord) -> Self {
        Self {
            token_id: rec.token_id.clone(),
            owner: rec.owner.clone(),
            requester: rec.requester.clone(),
            issue_timestamp_ms: rec.issue_timestamp_ms,
            expiry_timestamp_ms: rec.expiry_timestamp_ms,
            max_timestamp_ms: rec.max_timestamp_ms,
            renewers: rec.renewers.clone(),
        }
    }

    /// The record that recreates this token.
    #[must_use]
    pub fn to_record(&self) -> DelegationTokenRecord {
        DelegationTokenRecord {
            token_id: self.token_id.clone(),
            owner: self.owner.clone(),
            requester: self.requester.clone(),
            issue_timestamp_ms: self.issue_timestamp_ms,
            expiry_timestamp_ms: self.expiry_timestamp_ms,
            max_timestamp_ms: self.max_timestamp_ms,
            renewers: self.renewers.clone(),
        }
    }

    /// Kafka's `TokenInformation.ownerOrRenewer`: whether `principal` is the
    /// token's owner, the principal that created it, or a listed renewer.
    #[must_use]
    pub fn owner_or_renewer(&self, principal: &KafkaPrincipal) -> bool {
        &self.owner == principal
            || &self.requester == principal
            || self.renewers.contains(principal)
    }
}

#[must_use]
pub fn canonicalize_entity(mut tuple: Vec<(String, Option<String>)>) -> EntityKey {
    tuple.sort_by(|a, b| a.0.cmp(&b.0));
    tuple
}

/// Static discriminant name for a [`MetadataRecord`], used as tracing span
/// context. The span skips the record itself to keep spans cheap and free of
/// PII.
fn record_variant(rec: &MetadataRecord) -> &'static str {
    match rec {
        MetadataRecord::V1Topic(_) => "V1Topic",
        MetadataRecord::V1Partition(_) => "V1Partition",
        MetadataRecord::V1BrokerRegistration(_) => "V1BrokerRegistration",
        MetadataRecord::V1DeleteTopic(_) => "V1DeleteTopic",
        MetadataRecord::V1TopicConfig(_) => "V1TopicConfig",
        MetadataRecord::V1ScramCredential(_) => "V1ScramCredential",
        MetadataRecord::V1DeleteScramCredential(_) => "V1DeleteScramCredential",
        MetadataRecord::V1AccessControlEntry(_) => "V1AccessControlEntry",
        MetadataRecord::V1DeleteAccessControlEntry(_) => "V1DeleteAccessControlEntry",
        MetadataRecord::V1BrokerConfig(_) => "V1BrokerConfig",
        MetadataRecord::V1ClientQuota(_) => "V1ClientQuota",
        MetadataRecord::V1ProducerIds(_) => "V1ProducerIds",
        MetadataRecord::V1DelegationToken(_) => "V1DelegationToken",
        MetadataRecord::V1DeleteDelegationToken(_) => "V1DeleteDelegationToken",
        MetadataRecord::V1UnregisterBroker(_) => "V1UnregisterBroker",
        MetadataRecord::V1KRaftVersion(_) => "V1KRaftVersion",
        MetadataRecord::V1Voters(_) => "V1Voters",
        MetadataRecord::V1FeatureLevel(_) => "V1FeatureLevel",
        MetadataRecord::V1ClientMetricsConfig(_) => "V1ClientMetricsConfig",
        MetadataRecord::V1FeaturesEpoch(_) => "V1FeaturesEpoch",
        MetadataRecord::V1PartitionDirAssignment(_) => "V1PartitionDirAssignment",
        MetadataRecord::V1PartitionOffsetAdvance(_) => "V1PartitionOffsetAdvance",
        MetadataRecord::V1GroupConfig(_) => "V1GroupConfig",
        MetadataRecord::V1ControllerRegistration(_) => "V1ControllerRegistration",
        MetadataRecord::V1TopicFreeze(_) => "V1TopicFreeze",
        MetadataRecord::V1BreakGlassProposal(_) => "V1BreakGlassProposal",
        MetadataRecord::V1DeleteBreakGlassProposal(_) => "V1DeleteBreakGlassProposal",
        MetadataRecord::V1PartitionElr(_) => "V1PartitionElr",
        MetadataRecord::V1PartitionRecovery(_) => "V1PartitionRecovery",
        MetadataRecord::V1PartitionUpdate(_) => "V1PartitionUpdate",
        MetadataRecord::V1BrokerRegistrationChange(_) => "V1BrokerRegistrationChange",
        MetadataRecord::V1UnregisterController(_) => "V1UnregisterController",
    }
}

#[derive(Debug, Clone, Default, PartialEq)]
pub struct MetadataImage {
    cluster_id: Uuid,
    topics: HashMap<String, TopicRecord>,
    /// KIP-516 reverse index: topic UUID -> topic name. `apply()` maintains it
    /// next to `topics`. A snapshot replay rebuilds it, because every record
    /// flows through `apply()`, including snapshot installs.
    topic_ids: HashMap<Uuid, String>,
    /// Partition records grouped by owning topic. The by-topic grouping is
    /// the scaling-critical index: [`Self::partitions_of`] resolves a topic
    /// in O(1) and touches only that topic's partitions. Per-topic hot paths
    /// such as Metadata responses and reconcile loops therefore stay
    /// O(total partitions) and not O(topics × partitions). The inner
    /// `BTreeMap` keys by partition index, so per-topic iteration runs in
    /// ascending-index order.
    partitions: HashMap<String, BTreeMap<i32, PartitionRecord>>,
    // Group auxiliary state by topic too: deleting a topic never scans other topics.
    partition_next_offsets: HashMap<String, BTreeMap<i32, i64>>,
    partition_elr: HashMap<String, BTreeMap<i32, PartitionElrState>>,
    partition_recovery: HashMap<String, BTreeMap<i32, LeaderRecoveryState>>,
    brokers: HashMap<NodeId, BrokerRegistrationRecord>,
    controllers: HashMap<NodeId, ControllerRegistrationRecord>,
    topic_configs: HashMap<String, BTreeMap<String, String>>,
    broker_configs: HashMap<NodeId, BTreeMap<String, String>>,
    client_metrics_configs: HashMap<String, BTreeMap<String, String>>,
    group_configs: HashMap<String, BTreeMap<String, String>>,
    scram_credentials: HashMap<(String, SaslMechanism), ScramCredential>,
    acls: AclIndex,
    /// KFC-9 write-freeze registry, split by pattern type. The prefixed side
    /// is a `BTreeMap`, because the resolve runs one hop from the produce
    /// path and needs the ordered reverse walk that finds the longest
    /// matching prefix.
    freezes_literal: HashMap<String, TopicFreezeRecord>,
    freezes_prefixed: BTreeMap<String, TopicFreezeRecord>,
    /// KFC-9 break-glass proposals, keyed by proposal id.
    break_glass: HashMap<Uuid, BreakGlassProposalRecord>,
    client_quotas: HashMap<EntityKey, BTreeMap<String, f64>>,
    producer_ids: Option<ProducerIdsRecord>,
    delegation_tokens: HashMap<String, DelegationToken>,
    kraft_version: u16,
    voters: VoterSet,
    feature_levels: BTreeMap<String, i16>,
    /// KIP-584 finalized-features epoch. It is `-1` until the first
    /// `V1FeatureLevel` record applies, and it then increases monotonically,
    /// once per applied record. It is deterministic across replicas, because
    /// records apply in committed-log order on every node.
    features_epoch: i64,
}

/// True when `candidate` is `stored` plus zero or more appended entries.
///
/// This is the concurrent-approval rule of KFC-9 in one predicate: an
/// approver may add its own entry to the list it read, and may do nothing
/// else to that list.
fn extends(stored: &[BreakGlassApproval], candidate: &[BreakGlassApproval]) -> bool {
    candidate.starts_with(stored)
}

/// Selects which broker byte-rate config key to read.
#[derive(Debug, Clone, Copy)]
pub enum ThrottleKind {
    Leader,
    Follower,
    AlterLogDirs,
}

impl MetadataImage {
    #[must_use]
    pub fn new(cluster_id: Uuid) -> Self {
        Self {
            cluster_id,
            topics: HashMap::new(),
            topic_ids: HashMap::new(),
            partitions: HashMap::new(),
            partition_next_offsets: HashMap::new(),
            partition_elr: HashMap::new(),
            partition_recovery: HashMap::new(),
            brokers: HashMap::new(),
            controllers: HashMap::new(),
            topic_configs: HashMap::new(),
            broker_configs: HashMap::new(),
            client_metrics_configs: HashMap::new(),
            group_configs: HashMap::new(),
            scram_credentials: HashMap::new(),
            acls: AclIndex::default(),
            freezes_literal: HashMap::new(),
            freezes_prefixed: BTreeMap::new(),
            break_glass: HashMap::new(),
            client_quotas: HashMap::new(),
            producer_ids: None,
            delegation_tokens: HashMap::new(),
            kraft_version: 0,
            voters: VoterSet::default(),
            feature_levels: BTreeMap::new(),
            features_epoch: -1,
        }
    }

    #[must_use]
    pub fn cluster_id(&self) -> Uuid {
        self.cluster_id
    }

    pub fn topics(&self) -> impl Iterator<Item = &TopicRecord> {
        self.topics.values()
    }

    #[tracing::instrument(level = "debug", skip_all, fields(topic = %name))]
    #[must_use]
    pub fn topic(&self, name: &str) -> Option<&TopicRecord> {
        self.topics.get(name)
    }

    /// KIP-516: resolve a topic by its UUID. O(1) through the `topic_ids` index.
    #[tracing::instrument(level = "debug", skip_all, fields(topic_id = %id))]
    #[must_use]
    pub fn topic_by_id(&self, id: &Uuid) -> Option<&TopicRecord> {
        let name = self.topic_ids.get(id)?;
        self.topics.get(name)
    }

    /// KIP-516: resolve a topic name by its UUID.
    #[must_use]
    pub fn topic_name_by_id(&self, id: &Uuid) -> Option<&str> {
        self.topic_ids.get(id).map(String::as_str)
    }

    #[tracing::instrument(level = "debug", skip_all, fields(topic = %topic, partition = idx))]
    #[must_use]
    pub fn partition(&self, topic: &str, idx: i32) -> Option<&PartitionRecord> {
        self.partitions.get(topic)?.get(&idx)
    }

    /// All partitions of `topic`, in ascending partition-index order.
    /// The by-topic index gives an O(1) topic lookup and an
    /// O(own partitions) iteration. The method never scans the partitions of
    /// other topics, so a per-topic caller stays proportional to the topic it
    /// asks about, even in images with tens of thousands of topics.
    pub fn partitions_of(&self, topic: &str) -> impl Iterator<Item = &PartitionRecord> {
        self.partitions
            .get(topic)
            .into_iter()
            .flat_map(BTreeMap::values)
    }

    /// The committed next-offset for a diskless partition, if any advance has
    /// been applied.
    #[must_use]
    pub fn partition_next_offset(&self, topic: &str, partition: i32) -> Option<i64> {
        self.partition_next_offsets
            .get(topic)
            .and_then(|partitions| partitions.get(&partition))
            .copied()
    }

    #[must_use]
    pub fn partition_elr(&self, topic: &str, partition: i32) -> (&[NodeId], &[NodeId]) {
        self.partition_elr
            .get(topic)
            .and_then(|partitions| partitions.get(&partition))
            .map_or((&[], &[]), |(eligible, last_known)| {
                (eligible.as_slice(), last_known.as_slice())
            })
    }

    #[must_use]
    pub fn leader_recovery_state(&self, topic: &str, partition: i32) -> LeaderRecoveryState {
        self.partition_recovery
            .get(topic)
            .and_then(|partitions| partitions.get(&partition))
            .copied()
            .unwrap_or_default()
    }

    /// The live partition count for `topic`, derived from the partitions
    /// map and not from the stored `TopicRecord.partitions` field. This is
    /// the authoritative count for the KIP-631 round-trip, because the
    /// KIP-631 `TopicRecord` carries no partition count. It is also the
    /// authoritative count for the `validate` partition-count-grew check.
    /// Returns 0 for an unknown topic, or for a topic with no applied
    /// partition records. O(1).
    #[must_use]
    pub fn topic_partition_count(&self, topic: &str) -> i32 {
        self.partitions
            .get(topic)
            .map_or(0, |parts| i32::try_from(parts.len()).unwrap_or(i32::MAX))
    }

    /// Total partition count across every topic. O(topics).
    fn total_partitions(&self) -> usize {
        self.partitions.values().map(BTreeMap::len).sum()
    }

    /// Single-pass iterator over every partition in the image. Each record
    /// carries its own `topic` and `partition` fields. Topic order is
    /// unspecified. Within a topic the order is ascending partition index.
    /// The iterator is O(P) in total partition count. The cluster-wide
    /// maintenance loops for failover, rebalance, reassignment, and metrics
    /// use this instead of a per-topic loop.
    pub fn all_partitions(&self) -> impl Iterator<Item = &PartitionRecord> {
        self.partitions.values().flat_map(BTreeMap::values)
    }

    /// All partitions where a reassignment is currently in flight, that is,
    /// where `adding_replicas` or `removing_replicas` is non-empty.
    pub fn reassignments_in_flight(&self) -> impl Iterator<Item = &PartitionRecord> + '_ {
        self.all_partitions()
            .filter(|p| !p.adding_replicas.is_empty() || !p.removing_replicas.is_empty())
    }

    /// Currently-effective config overrides for `topic`, or `None` if no
    /// `V1TopicConfig` record has applied for this topic since the last
    /// `V1DeleteTopic`, or since image creation.
    #[must_use]
    pub fn topic_config(&self, topic: &str) -> Option<&BTreeMap<String, String>> {
        self.topic_configs.get(topic)
    }

    /// Per-broker config overrides for `node_id`, or `None` if no
    /// `V1BrokerConfig` record has been applied for this broker.
    #[must_use]
    pub fn broker_config(&self, node_id: NodeId) -> Option<&BTreeMap<String, String>> {
        self.broker_configs.get(&node_id)
    }

    /// Cluster-wide dynamic broker-config defaults, or `None` when no default
    /// broker config has been applied.
    #[must_use]
    pub fn default_broker_config(&self) -> Option<&BTreeMap<String, String>> {
        self.broker_config(DEFAULT_BROKER_CONFIG_NODE_ID)
    }

    /// Returns the KIP-73 throttle rate for `node_id` and `kind`.
    /// Returns `None` if the config key is absent, unparseable, or negative.
    /// `-1` is Kafka's convention for "disabled" or "unlimited".
    #[must_use]
    pub fn broker_throttle_rate(&self, node_id: NodeId, kind: ThrottleKind) -> Option<ByteRate> {
        let key = match kind {
            ThrottleKind::Leader => "leader.replication.throttled.rate",
            ThrottleKind::Follower => "follower.replication.throttled.rate",
            ThrottleKind::AlterLogDirs => "replica.alter.log.dirs.io.max.bytes.per.second",
        };
        let raw = self
            .broker_config(node_id)
            .and_then(|configs| configs.get(key))
            .or_else(|| self.default_broker_config()?.get(key))?;
        let v: i64 = raw.parse().ok()?;
        (v >= 0).then(|| ByteRate::from_bytes_per_sec(v))
    }

    /// Override map for a single KIP-714 client-metrics subscription.
    #[must_use]
    pub fn client_metrics_config(&self, name: &str) -> Option<&BTreeMap<String, String>> {
        self.client_metrics_configs.get(name)
    }

    /// All configured client-metrics subscriptions, `(name, overrides)`.
    pub fn client_metrics_subscriptions(
        &self,
    ) -> impl Iterator<Item = (&String, &BTreeMap<String, String>)> {
        self.client_metrics_configs.iter()
    }

    /// Dynamic KIP-1071 overrides for one GROUP config resource.
    #[must_use]
    pub fn group_config(&self, group_id: &str) -> Option<&BTreeMap<String, String>> {
        self.group_configs.get(group_id)
    }

    /// All configured GROUP resources.
    pub fn group_configs(&self) -> impl Iterator<Item = (&String, &BTreeMap<String, String>)> {
        self.group_configs.iter()
    }

    #[must_use]
    pub fn client_quotas(&self) -> &HashMap<EntityKey, BTreeMap<String, f64>> {
        &self.client_quotas
    }

    #[must_use]
    pub fn scram_credential(
        &self,
        user: &str,
        mechanism: SaslMechanism,
    ) -> Option<&ScramCredential> {
        self.scram_credentials.get(&(user.to_owned(), mechanism))
    }

    /// All distinct users with at least one SCRAM credential. Order is
    /// unspecified.
    #[must_use]
    pub fn scram_credentials_users(&self) -> Vec<String> {
        self.scram_credentials
            .keys()
            .map(|(u, _)| u.clone())
            .collect::<HashSet<_>>()
            .into_iter()
            .collect()
    }

    /// All `(mechanism, iterations)` pairs for `user`. Empty if user has
    /// no SCRAM credentials. Order is unspecified.
    #[must_use]
    pub fn scram_credentials_for_user(&self, user: &str) -> Vec<(SaslMechanism, u32)> {
        self.scram_credentials
            .iter()
            .filter(|((u, _), _)| u == user)
            .map(|((_, mech), cred)| (*mech, cred.iterations))
            .collect()
    }

    #[must_use]
    pub fn broker(&self, node_id: NodeId) -> Option<&BrokerRegistrationRecord> {
        self.brokers.get(&node_id)
    }

    #[must_use]
    pub fn controller(&self, node_id: NodeId) -> Option<&ControllerRegistrationRecord> {
        self.controllers.get(&node_id)
    }

    /// KIP-903: the broker epoch (registration commit offset) for `node_id`,
    /// or `None` if the broker is not registered in this image.
    #[must_use]
    pub fn broker_epoch(&self, node_id: NodeId) -> Option<i64> {
        self.brokers.get(&node_id).map(|b| b.broker_epoch)
    }

    /// First producer ID not covered by any committed allocation block.
    #[must_use]
    pub fn next_producer_id(&self) -> i64 {
        self.producer_ids.as_ref().map_or(0, |r| r.next_producer_id)
    }

    pub fn brokers(&self) -> impl Iterator<Item = &BrokerRegistrationRecord> {
        self.brokers.values()
    }

    pub fn controllers(&self) -> impl Iterator<Item = &ControllerRegistrationRecord> {
        self.controllers.values()
    }

    #[must_use]
    pub fn kraft_version(&self) -> u16 {
        self.kraft_version
    }

    #[must_use]
    pub fn voters(&self) -> &VoterSet {
        &self.voters
    }

    /// Iterate every ACL that could possibly match `(rt, rn)`:
    /// - all literal entries at `(rt, rn)`
    /// - all literal entries at `(rt, "*")`, the `WILDCARD_RESOURCE`, which
    ///   matches every resource of that type. See Kafka's `AclAuthorizer` and
    ///   `StandardAuthorizer`. This is what `kafka-acls --topic '*'` produces.
    /// - all prefixed entries whose `resource_name` is a prefix of `rn`
    ///
    /// Each matching entry is yielded once, also when `rn` is `"*"`. The
    /// lookup allocates nothing, and finds the prefixed entries with one
    /// hash probe per distinct prefix length, not a scan of every prefixed
    /// entry. The order of the entries is unspecified.
    pub fn matching_acls<'a>(
        &'a self,
        rt: ResourceType,
        rn: &'a str,
    ) -> impl Iterator<Item = &'a AclEntry> + 'a {
        self.acls.matching(rt, rn)
    }

    /// All ACL entries, both literal and prefixed, across all resource types,
    /// in unspecified order. `DescribeAcls` uses this.
    pub fn all_acls(&self) -> impl Iterator<Item = &AclEntry> {
        self.acls.iter()
    }

    /// The write-freeze entry that covers `topic`, or `None` when the topic
    /// accepts writes (KFC-9).
    ///
    /// A cluster with no freeze costs two `is_empty` tests, because the
    /// produce path calls this once per topic per request. A literal entry
    /// beats every prefix entry, and the longest matching prefix wins among
    /// the prefix entries.
    ///
    /// An internal topic is never frozen. The `__` name convention is the
    /// test, so a prefix scope of `""` or `"_"` cannot take
    /// `__consumer_offsets` down with the rest of the cluster.
    #[must_use]
    pub fn topic_freeze(&self, topic: &str) -> Option<&TopicFreezeRecord> {
        if self.freezes_literal.is_empty() && self.freezes_prefixed.is_empty() {
            return None;
        }
        if topic.starts_with("__") {
            return None;
        }
        if let Some(freeze) = self.freezes_literal.get(topic) {
            return Some(freeze);
        }
        // Keys that sort at or below `topic`, walked downwards. Among the
        // prefixes of `topic`, the longest one sorts highest, so the first
        // match is the winner. The bound pair is what keeps the lookup key a
        // borrowed `str`: `RangeBounds<str>` has no impl for `..=&str`.
        self.freezes_prefixed
            .range::<str, _>((Bound::Unbounded, Bound::Included(topic)))
            .rev()
            .find(|(scope, _)| topic.starts_with(scope.as_str()))
            .map(|(_, freeze)| freeze)
    }

    /// Every live write-freeze entry, literal and prefixed (KFC-9). A thawed
    /// entry is removed on apply, so every entry here has `frozen: true`.
    /// Order is unspecified.
    pub fn topic_freezes(&self) -> impl Iterator<Item = &TopicFreezeRecord> {
        self.freezes_literal
            .values()
            .chain(self.freezes_prefixed.values())
    }

    /// Look up a break-glass proposal by its id (KFC-9).
    #[must_use]
    pub fn break_glass_proposal(&self, id: Uuid) -> Option<&BreakGlassProposalRecord> {
        self.break_glass.get(&id)
    }

    /// Every break-glass proposal still in the image (KFC-9), including the
    /// consumed and withdrawn ones that the expiry sweep has not yet removed.
    /// Order is unspecified.
    pub fn break_glass_proposals(&self) -> impl Iterator<Item = &BreakGlassProposalRecord> {
        self.break_glass.values()
    }

    /// Look up a delegation token by its `token_id` (KIP-48).
    #[must_use]
    pub fn delegation_token_by_id(&self, token_id: &str) -> Option<&DelegationToken> {
        self.delegation_tokens.get(token_id)
    }

    /// All tokens owned by `owner` (KIP-48). The match on the owning
    /// [`KafkaPrincipal`] is exact. Order is unspecified.
    #[must_use]
    pub fn delegation_tokens_by_owner(&self, owner: &KafkaPrincipal) -> Vec<&DelegationToken> {
        self.delegation_tokens
            .values()
            .filter(|t| &t.owner == owner)
            .collect()
    }

    /// Tokens that `principal` may see through `DescribeDelegationToken`
    /// without `DescribeToken` permission: those it owns, created or may
    /// renew ([`DelegationToken::owner_or_renewer`]). Order is unspecified.
    #[must_use]
    pub fn delegation_tokens_visible_to(
        &self,
        principal: &KafkaPrincipal,
    ) -> Vec<&DelegationToken> {
        self.delegation_tokens
            .values()
            .filter(|t| t.owner_or_renewer(principal))
            .collect()
    }

    /// Every delegation token currently in the
    /// image (KIP-48). `DescribeDelegationToken` uses this for callers with
    /// `DescribeToken` permission on the cluster.
    pub fn all_delegation_tokens(&self) -> impl Iterator<Item = &DelegationToken> {
        self.delegation_tokens.values()
    }

    /// KIP-48: the delegation token whose HMAC under `secret_key` is `hmac`.
    ///
    /// `RenewDelegationToken` and `ExpireDelegationToken` identify a token by
    /// HMAC on the wire and not by `token_id`. The image keeps no HMAC, as
    /// Kafka's metadata does not, so each token's HMAC is recomputed with
    /// [`krabka_security::compute_token_hmac`]. The scan is linear over the
    /// small in-memory token map.
    #[must_use]
    pub fn delegation_token_by_hmac(
        &self,
        secret_key: &[u8],
        hmac: &[u8],
    ) -> Option<&DelegationToken> {
        self.delegation_tokens
            .values()
            .find(|t| krabka_security::compute_token_hmac(secret_key, &t.token_id) == hmac)
    }

    /// KIP-584: finalized feature levels, keyed by feature name. Empty
    /// until an `UpdateFeatures` call lands a `V1FeatureLevel` record.
    #[must_use]
    pub fn finalized_features(&self) -> &BTreeMap<String, i16> {
        &self.feature_levels
    }

    /// KIP-584 finalized-features epoch. `-1` ("unknown") until the first
    /// feature is finalized.
    #[must_use]
    pub fn finalized_features_epoch(&self) -> i64 {
        self.features_epoch
    }

    /// The finalized `metadata.version` level, or `None` if no
    /// `V1FeatureLevel` for `metadata.version` has applied. That is a
    /// pre-bootstrap or legacy image, which is `MetadataVersion.UNKNOWN`.
    #[must_use]
    pub fn finalized_metadata_version(&self) -> Option<i16> {
        self.feature_levels
            .get(metadata_version::METADATA_VERSION_FEATURE)
            .copied()
    }

    /// KIP-584: the finalized level for an arbitrary feature, or `None` if it
    /// has not been finalized. Generic counterpart to
    /// [`Self::finalized_metadata_version`].
    #[must_use]
    pub fn finalized_feature(&self, name: &str) -> Option<i16> {
        self.feature_levels.get(name).copied()
    }

    /// The minimum `metadata.version` level that the live image needs: the
    /// floor that a downgrade must not drop below. It rises with feature-gated
    /// state present in the image: `KRaft` SCRAM creds, delegation tokens, and
    /// KIP-1276 CIDR ACL hosts. The baseline is `METADATA_VERSION_MIN`.
    #[must_use]
    pub fn min_required_metadata_version(&self) -> i16 {
        use crate::metadata_version::{
            CIDR_ACL_MIN_LEVEL, DELEGATION_TOKEN_MIN_LEVEL, METADATA_VERSION_MIN, SCRAM_MIN_LEVEL,
        };
        let mut floor = METADATA_VERSION_MIN;
        if !self.scram_credentials.is_empty() {
            floor = floor.max(SCRAM_MIN_LEVEL);
        }
        if !self.delegation_tokens.is_empty() {
            floor = floor.max(DELEGATION_TOKEN_MIN_LEVEL);
        }
        if self.all_acls().any(|acl| is_cidr_acl_host(&acl.host)) {
            floor = floor.max(CIDR_ACL_MIN_LEVEL);
        }
        floor
    }

    /// Metadata removals needed to materialize this image at `target`.
    /// An empty result means the downgrade is lossless. The controller emits
    /// these records before the trailing metadata.version feature record only
    /// for an explicitly unsafe downgrade.
    #[must_use]
    pub fn metadata_version_downgrade_records(&self, target: i16) -> Vec<MetadataRecord> {
        use crate::{
            metadata_version::{
                DELEGATION_TOKEN_MIN_LEVEL, DIRECTORY_ASSIGNMENT_MIN_LEVEL, SCRAM_MIN_LEVEL,
            },
            records::{DeleteDelegationTokenRecord, DeleteScramCredentialRecord},
        };

        let mut records = Vec::new();
        if target < metadata_version::CIDR_ACL_MIN_LEVEL {
            let mut cidr_acls: Vec<&AclEntry> = self
                .all_acls()
                .filter(|acl| is_cidr_acl_host(&acl.host))
                .collect();
            cidr_acls.sort_by(|a, b| {
                (&a.resource_name, &a.principal, &a.host).cmp(&(
                    &b.resource_name,
                    &b.principal,
                    &b.host,
                ))
            });
            records.extend(cidr_acls.into_iter().map(|acl| {
                MetadataRecord::V1DeleteAccessControlEntry(crate::AclEntryFilter {
                    resource_type: Some(acl.resource_type),
                    resource_name: Some(acl.resource_name.clone()),
                    pattern_type: Some(acl.pattern_type),
                    principal: Some(acl.principal.clone()),
                    host: Some(acl.host.clone()),
                    operation: Some(acl.operation),
                    permission_type: Some(acl.permission_type),
                })
            }));
        }
        if target < SCRAM_MIN_LEVEL {
            records.extend(self.scram_credentials.keys().map(|(user, mechanism)| {
                MetadataRecord::V1DeleteScramCredential(DeleteScramCredentialRecord {
                    user: user.clone(),
                    mechanism: *mechanism,
                })
            }));
        }
        if target < DELEGATION_TOKEN_MIN_LEVEL {
            records.extend(self.delegation_tokens.keys().map(|token_id| {
                MetadataRecord::V1DeleteDelegationToken(DeleteDelegationTokenRecord {
                    token_id: token_id.clone(),
                })
            }));
        }
        records.extend(self.brokers.values().filter_map(|broker| {
            let mut projected = broker.clone();
            if target < DIRECTORY_ASSIGNMENT_MIN_LEVEL {
                projected.log_dirs.clear();
            }
            if target < metadata_version::CORDONED_LOG_DIRS_MIN_LEVEL {
                projected.cordoned_log_dirs = None;
            }
            (projected != *broker).then_some(MetadataRecord::V1BrokerRegistration(projected))
        }));
        // Below `3.7-IV2` a partition record carries no directories, and Kafka
        // reads it back as every replica `MIGRATING`.
        if target < DIRECTORY_ASSIGNMENT_MIN_LEVEL {
            for partition in self.all_partitions() {
                records.extend(
                    partition
                        .replicas
                        .iter()
                        .zip(&partition.directories)
                        .filter(|(_, directory)| **directory != MIGRATING)
                        .map(|(replica, _)| {
                            MetadataRecord::V1PartitionDirAssignment(PartitionDirAssignmentRecord {
                                topic: partition.topic.clone(),
                                partition: partition.partition,
                                replica: *replica,
                                directory: MIGRATING,
                            })
                        }),
                );
            }
        }
        records
    }

    /// Apply one record. Returns the previous record, for `V1Topic` and
    /// `V1BrokerRegistration`, so the caller can observe overwrite cases.
    /// The method is infallible. The controller pre-validates against the
    /// current image before it submits to Raft. Apply must never fail on a
    /// committed entry.
    #[tracing::instrument(level = "info", skip_all, fields(record = record_variant(rec)))]
    // exhaustive match over MetadataRecord
    pub fn apply(&mut self, rec: &MetadataRecord) {
        match rec {
            MetadataRecord::V1Topic(t) => {
                // If a topic with this name already exists under a different
                // id, drop the stale id entry before re-indexing.
                if let Some(prev) = self.topics.get(&t.name)
                    && prev.topic_id != t.topic_id
                {
                    self.topic_ids.remove(&prev.topic_id);
                }
                self.topic_ids.insert(t.topic_id, t.name.clone());
                // The denormalized `partitions` count is a derived cache of the
                // partitions map (KIP-631 `TopicRecord` carries no partition
                // count), maintained incrementally as `V1Partition` records
                // apply. A brand-new topic starts at 0 and the partition records
                // that follow it in log order restore the real count; a
                // re-registration of an existing topic keeps its partitions in
                // the map, so preserve the running count rather than trusting
                // the (KIP-631-zeroed) record field.
                let mut rec = t.clone();
                rec.partitions = self.topics.get(&t.name).map_or(0, |prev| prev.partitions);
                self.topics.insert(t.name.clone(), rec);
            }
            MetadataRecord::V1Partition(p) => self.apply_partition(p),
            MetadataRecord::V1BrokerRegistration(b) => {
                self.brokers.insert(b.node_id, b.clone());
            }
            MetadataRecord::V1BrokerRegistrationChange(change) => {
                self.apply_registration_change(change);
            }
            MetadataRecord::V1UnregisterBroker(rec) => self.apply_unregister_broker(rec),
            MetadataRecord::V1ControllerRegistration(c) => {
                self.controllers.insert(c.node_id, c.clone());
            }
            MetadataRecord::V1UnregisterController(rec) => {
                self.controllers.remove(&rec.node_id);
            }
            MetadataRecord::V1DeleteTopic(d) => {
                if let Some(prev) = self.topics.get(&d.name) {
                    self.topic_ids.remove(&prev.topic_id);
                }
                self.topics.remove(&d.name);
                self.partitions.remove(&d.name);
                self.partition_next_offsets.remove(&d.name);
                self.partition_elr.remove(&d.name);
                self.partition_recovery.remove(&d.name);
                self.topic_configs.remove(&d.name);
            }
            MetadataRecord::V1TopicConfig(c) => {
                if c.overrides.is_empty() {
                    self.topic_configs.remove(&c.topic);
                } else {
                    self.topic_configs
                        .insert(c.topic.clone(), c.overrides.clone());
                }
            }
            MetadataRecord::V1ScramCredential(r) => {
                self.scram_credentials.insert(
                    (r.user.clone(), r.mechanism),
                    ScramCredential {
                        mechanism: r.mechanism,
                        salt: r.salt.clone(),
                        stored_key: r.stored_key.clone(),
                        server_key: r.server_key.clone(),
                        iterations: r.iterations,
                    },
                );
            }
            MetadataRecord::V1DeleteScramCredential(r) => {
                self.scram_credentials
                    .remove(&(r.user.clone(), r.mechanism));
            }
            MetadataRecord::V1AccessControlEntry(entry) => self.acls.insert(entry),
            MetadataRecord::V1DeleteAccessControlEntry(filter) => {
                self.acls.remove_matching(filter);
            }
            MetadataRecord::V1BrokerConfig(rec) => {
                let entry = self.broker_configs.entry(rec.node_id).or_default();
                match &rec.config_value {
                    Some(v) => {
                        entry.insert(rec.config_name.clone(), v.clone());
                    }
                    None => {
                        entry.remove(&rec.config_name);
                    }
                }
            }
            MetadataRecord::V1ClientQuota(rec) => {
                let key = canonicalize_entity(
                    rec.entity
                        .iter()
                        .map(|e| (e.entity_type.clone(), e.entity_name.clone()))
                        .collect(),
                );
                let configs = self.client_quotas.entry(key).or_default();
                match rec.config_value {
                    Some(v) => {
                        configs.insert(rec.config_key.clone(), v);
                    }
                    None => {
                        configs.remove(&rec.config_key);
                    }
                }
            }
            MetadataRecord::V1ProducerIds(rec) => {
                if rec.next_producer_id > self.next_producer_id() {
                    self.producer_ids = Some(*rec);
                } else {
                    tracing::error!(
                        current = self.next_producer_id(),
                        proposed = rec.next_producer_id,
                        "ignoring non-monotonic producer ID allocation record"
                    );
                }
            }
            // Replacement semantics (KIP-48) — same
            // `token_id` overwrites the prior entry (used by Create
            // and Renew). Tombstone removes by `token_id`.
            MetadataRecord::V1DelegationToken(rec) => {
                self.delegation_tokens
                    .insert(rec.token_id.clone(), DelegationToken::from_record(rec));
            }
            MetadataRecord::V1DeleteDelegationToken(rec) => {
                self.delegation_tokens.remove(&rec.token_id);
            }
            MetadataRecord::V1KRaftVersion(r) => {
                self.kraft_version = r.kraft_version;
            }
            MetadataRecord::V1Voters(r) => {
                self.voters = r.voters.clone();
            }
            MetadataRecord::V1FeatureLevel(rec) => {
                if rec.level == 0 {
                    self.feature_levels.remove(&rec.name);
                } else {
                    self.feature_levels.insert(rec.name.clone(), rec.level);
                }
                // Monotonic epoch: -1 -> 0 on the first record, then +1.
                self.features_epoch = self.features_epoch.saturating_add(1).max(0);
            }
            MetadataRecord::V1ClientMetricsConfig(c) => {
                if c.configs.is_empty() {
                    self.client_metrics_configs.remove(&c.name);
                } else {
                    self.client_metrics_configs
                        .insert(c.name.clone(), c.configs.clone());
                }
            }
            MetadataRecord::V1GroupConfig(c) => {
                if c.configs.is_empty() {
                    self.group_configs.remove(&c.group_id);
                } else {
                    self.group_configs
                        .insert(c.group_id.clone(), c.configs.clone());
                }
            }
            // Snapshot-only: restore the epoch verbatim rather than bumping.
            // Emitted last by `to_records` so it overrides the value the
            // preceding `V1FeatureLevel` replays accumulated. Never reaches
            // the live log (not a controller-submittable change).
            MetadataRecord::V1FeaturesEpoch(rec) => {
                self.features_epoch = rec.epoch;
            }
            // KIP-858 directory-assignment delta: merge ONLY the reporting
            // replica's `directories` slot. Never touches leader/isr/replicas/
            // adding/removing, so a concurrent reassignment or ISR change is
            // preserved regardless of commit order.
            MetadataRecord::V1PartitionDirAssignment(r) => {
                if let Some(pr) = self
                    .partitions
                    .get_mut(&r.topic)
                    .and_then(|parts| parts.get_mut(&r.partition))
                    && let Some(slot) = pr.replicas.iter().position(|n| *n == r.replica)
                {
                    // An empty list is every replica `MIGRATING`, as Kafka's
                    // `PartitionRegistration.defaultToMigrating` reads it.
                    if pr.directories.len() < pr.replicas.len() {
                        pr.directories.resize(pr.replicas.len(), MIGRATING);
                    }
                    if let Some(directory) = pr.directories.get_mut(slot) {
                        *directory = r.directory;
                    }
                }
            }
            // Diskless offset delta: bump the partition's committed next-offset.
            MetadataRecord::V1PartitionOffsetAdvance(r) => {
                *self
                    .partition_next_offsets
                    .entry(r.topic.clone())
                    .or_default()
                    .entry(r.partition)
                    .or_insert(0) += r.count;
            }
            MetadataRecord::V1PartitionElr(r) => self.apply_partition_elr(r),
            MetadataRecord::V1PartitionRecovery(r) => {
                self.apply_partition_recovery(&r.topic, r.partition, r.state);
            }
            MetadataRecord::V1PartitionUpdate(r) => self.apply_partition_update(r),
            // KFC-9 write-freeze registry.
            MetadataRecord::V1TopicFreeze(r) => self.apply_topic_freeze(r),
            // KFC-9 break-glass proposal. Replacement semantics on the id:
            // an approval and a consumption each write the whole record back.
            // `validate` is what stops a replacement from losing an approval.
            MetadataRecord::V1BreakGlassProposal(r) => {
                self.break_glass.insert(r.proposal_id, r.clone());
            }
            MetadataRecord::V1DeleteBreakGlassProposal(id) => {
                self.break_glass.remove(id);
            }
        }
    }

    /// Kafka's `ClusterControlManager.replayRegistrationChange` applies the
    /// change only to the registration at the record's epoch. An unknown
    /// broker or another epoch is a no-op.
    fn apply_registration_change(&mut self, change: &BrokerRegistrationChangeRecord) {
        if let Some(broker) = self.brokers.get_mut(&change.node_id)
            && broker.broker_epoch == change.broker_epoch
        {
            *broker = change.applied_to(broker);
        }
    }

    /// Kafka's `ClusterControlManager.replay(UnregisterBrokerRecord)` removes
    /// only the registration at the record's epoch. An unknown broker or
    /// another epoch is a no-op.
    fn apply_unregister_broker(&mut self, rec: &UnregisterBrokerRecord) {
        if self.broker_epoch(rec.node_id) == Some(rec.broker_epoch) {
            self.brokers.remove(&rec.node_id);
        }
    }

    fn apply_partition(&mut self, partition: &PartitionRecord) {
        let is_new = self
            .partitions
            .entry(partition.topic.clone())
            .or_default()
            .insert(partition.partition, partition.clone())
            .is_none();
        if let Some(topic) = self.topics.get_mut(&partition.topic) {
            if is_new {
                topic.partitions = topic.partitions.saturating_add(1);
            }
            topic.replication_factor =
                i16::try_from(partition.replicas.len()).unwrap_or(topic.replication_factor);
        }
    }

    fn apply_partition_update(&mut self, update: &crate::PartitionUpdateRecord) {
        let key = (update.partition.topic.clone(), update.partition.partition);
        self.apply_partition(&update.partition);
        if let (Some(eligible), Some(last_known)) =
            (&update.eligible_leader_replicas, &update.last_known_elr)
        {
            self.apply_partition_elr(&PartitionElrRecord {
                topic: key.0.clone(),
                partition: key.1,
                eligible_leader_replicas: eligible.clone(),
                last_known_elr: last_known.clone(),
            });
        }
        if let Some(state) = update.recovery_state {
            self.apply_partition_recovery(&key.0, key.1, state);
        }
    }

    fn apply_partition_elr(&mut self, record: &PartitionElrRecord) {
        if record.eligible_leader_replicas.is_empty() && record.last_known_elr.is_empty() {
            if let Some(partitions) = self.partition_elr.get_mut(&record.topic) {
                partitions.remove(&record.partition);
                if partitions.is_empty() {
                    self.partition_elr.remove(&record.topic);
                }
            }
        } else {
            self.partition_elr
                .entry(record.topic.clone())
                .or_default()
                .insert(
                    record.partition,
                    (
                        record.eligible_leader_replicas.clone(),
                        record.last_known_elr.clone(),
                    ),
                );
        }
    }

    fn apply_partition_recovery(
        &mut self,
        topic: &str,
        partition: i32,
        state: LeaderRecoveryState,
    ) {
        if state == LeaderRecoveryState::Recovered {
            if let Some(partitions) = self.partition_recovery.get_mut(topic) {
                partitions.remove(&partition);
                if partitions.is_empty() {
                    self.partition_recovery.remove(topic);
                }
            }
        } else {
            self.partition_recovery
                .entry(topic.to_owned())
                .or_default()
                .insert(partition, state);
        }
    }

    /// Upsert or remove one write-freeze registry entry (KFC-9).
    ///
    /// `frozen` is the sentinel: a `true` record upserts the entry into the
    /// registry that its pattern type selects, and a `false` record removes
    /// it. That keeps the thaw attributed in the raft log and needs no second
    /// record type.
    fn apply_topic_freeze(&mut self, rec: &TopicFreezeRecord) {
        match (rec.pattern_type, rec.frozen) {
            (PatternType::Literal, true) => {
                self.freezes_literal.insert(rec.scope.clone(), rec.clone());
            }
            (PatternType::Literal, false) => {
                self.freezes_literal.remove(&rec.scope);
            }
            (PatternType::Prefixed, true) => {
                self.freezes_prefixed.insert(rec.scope.clone(), rec.clone());
            }
            (PatternType::Prefixed, false) => {
                self.freezes_prefixed.remove(&rec.scope);
            }
        }
    }

    /// Serialize the image into the minimal sequence of [`MetadataRecord`]s
    /// whose in-order [`apply`](Self::apply) reproduces this image exactly.
    /// This is the inverse of `apply` over a sequence of non-tombstone
    /// records: each stored entry maps to the record that would create it.
    ///
    /// The method deliberately never emits tombstone or removal records:
    /// `V1DeleteTopic`, `V1DeleteScramCredential`,
    /// `V1DeleteAccessControlEntry`, `V1DeleteDelegationToken`, and
    /// `V1UnregisterBroker`. A snapshot captures resulting *state*, not
    /// deletion history. `V1BrokerConfig` deletes vanish in the same way. The
    /// method emits only the surviving key and value pairs, as `Some(value)`
    /// sets.
    ///
    /// The method emits records in dependency order, so a fresh image that
    /// `apply`s them never sees a dangling reference: brokers,
    /// broker-configs, topics, partitions, topic-configs, SCRAM creds,
    /// ACLs, client quotas, delegation tokens.
    #[tracing::instrument(
        level = "debug",
        skip_all,
        fields(topics = self.topics.len(), partitions = self.total_partitions())
    )]
    #[must_use]
    pub fn to_records(&self) -> Vec<MetadataRecord> {
        let mut out = Vec::new();

        // Brokers before their configs.
        for b in self.brokers.values() {
            out.push(MetadataRecord::V1BrokerRegistration(b.clone()));
        }
        for controller in self.controllers.values() {
            out.push(MetadataRecord::V1ControllerRegistration(controller.clone()));
        }
        for (node_id, configs) in &self.broker_configs {
            for (config_name, config_value) in configs {
                out.push(MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
                    node_id: *node_id,
                    config_name: config_name.clone(),
                    config_value: Some(config_value.clone()),
                }));
            }
        }

        // Topics before partitions and topic-configs (apply-side validate
        // and the snapshot's own DeleteTopic-clears-configs invariant both
        // assume the topic exists first).
        for t in self.topics.values() {
            out.push(MetadataRecord::V1Topic(t.clone()));
        }
        for p in self.all_partitions() {
            out.push(MetadataRecord::V1Partition(p.clone()));
        }
        self.push_partition_offset_advances(&mut out);
        for (topic, partition, (eligible, last_known)) in
            self.partition_elr.iter().flat_map(|(topic, partitions)| {
                partitions
                    .iter()
                    .map(move |(partition, value)| (topic, partition, value))
            })
        {
            out.push(MetadataRecord::V1PartitionElr(PartitionElrRecord {
                topic: topic.clone(),
                partition: *partition,
                eligible_leader_replicas: eligible.clone(),
                last_known_elr: last_known.clone(),
            }));
        }
        for (topic, partition, state) in
            self.partition_recovery
                .iter()
                .flat_map(|(topic, partitions)| {
                    partitions
                        .iter()
                        .map(move |(partition, value)| (topic, partition, value))
                })
        {
            out.push(MetadataRecord::V1PartitionRecovery(
                PartitionRecoveryRecord {
                    topic: topic.clone(),
                    partition: *partition,
                    state: *state,
                },
            ));
        }
        for (topic, overrides) in &self.topic_configs {
            out.push(MetadataRecord::V1TopicConfig(TopicConfigRecord {
                topic: topic.clone(),
                overrides: overrides.clone(),
            }));
        }

        // SCRAM credentials.
        for ((user, mechanism), cred) in &self.scram_credentials {
            out.push(MetadataRecord::V1ScramCredential(ScramCredentialRecord {
                user: user.clone(),
                mechanism: *mechanism,
                salt: cred.salt.clone(),
                stored_key: cred.stored_key.clone(),
                server_key: cred.server_key.clone(),
                iterations: cred.iterations,
            }));
        }

        // ACLs (literal + prefixed). Each stored entry already carries its
        // own pattern_type, so re-applying routes it back to the right
        // bucket.
        for entry in self.all_acls() {
            out.push(MetadataRecord::V1AccessControlEntry(entry.clone()));
        }

        // Client quotas. The key is the canonicalized entity tuple; rebuild
        // a QuotaEntity list from it and emit one record per config pair.
        // apply re-canonicalizes, so the round-trip is stable.
        for (entity_key, configs) in &self.client_quotas {
            let entity: Vec<QuotaEntity> = entity_key
                .iter()
                .map(|(entity_type, entity_name)| QuotaEntity {
                    entity_type: entity_type.clone(),
                    entity_name: entity_name.clone(),
                })
                .collect();
            for (config_key, config_value) in configs {
                out.push(MetadataRecord::V1ClientQuota(ClientQuotaRecord {
                    entity: entity.clone(),
                    config_key: config_key.clone(),
                    config_value: Some(*config_value),
                }));
            }
        }

        if let Some(record) = self.producer_ids {
            out.push(MetadataRecord::V1ProducerIds(record));
        }

        // Delegation tokens.
        for tok in self.delegation_tokens.values() {
            out.push(MetadataRecord::V1DelegationToken(DelegationTokenRecord {
                token_id: tok.token_id.clone(),
                owner: tok.owner.clone(),
                requester: tok.requester.clone(),
                issue_timestamp_ms: tok.issue_timestamp_ms,
                expiry_timestamp_ms: tok.expiry_timestamp_ms,
                max_timestamp_ms: tok.max_timestamp_ms,
                renewers: tok.renewers.clone(),
            }));
        }

        // Client-metrics subscriptions (KIP-714).
        for (name, configs) in &self.client_metrics_configs {
            out.push(MetadataRecord::V1ClientMetricsConfig(
                ClientMetricsConfigRecord {
                    name: name.clone(),
                    configs: configs.clone(),
                },
            ));
        }

        // KIP-1071 dynamic GROUP resources.
        for (group_id, configs) in &self.group_configs {
            out.push(MetadataRecord::V1GroupConfig(GroupConfigRecord {
                group_id: group_id.clone(),
                configs: configs.clone(),
            }));
        }

        // KFC-9 write-freeze registry and break-glass proposals. Both are
        // independent of every other record, and both are lost without this:
        // a snapshot install would thaw the cluster and drop every pending
        // approval. Every stored freeze has `frozen: true`, so re-applying it
        // routes it back to the same registry.
        for freeze in self.topic_freezes() {
            out.push(MetadataRecord::V1TopicFreeze(freeze.clone()));
        }
        for proposal in self.break_glass.values() {
            out.push(MetadataRecord::V1BreakGlassProposal(proposal.clone()));
        }

        // KIP-584 finalized features: one record per live feature, in the
        // BTreeMap's deterministic key order. (level == 0 is the delete
        // sentinel and is never stored, so every emitted level is >= 1 and
        // re-applies as a set.) The trailing V1FeaturesEpoch then pins the
        // epoch verbatim — the preceding feature records each bump it, but the
        // bumped value reflects the live feature count, not the original apply
        // history, so it must be overwritten. Emitted whenever any feature was
        // ever finalized (epoch left -1 only on a pristine image), so an image
        // that finalized then removed every feature still restores its epoch.
        for (name, level) in &self.feature_levels {
            out.push(MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
                name: name.clone(),
                level: *level,
            }));
        }
        if self.features_epoch >= 0 {
            out.push(MetadataRecord::V1FeaturesEpoch(FeaturesEpochRecord {
                epoch: self.features_epoch,
            }));
        }

        // KIP-853 controller-quorum state. Both MUST be emitted or a snapshot
        // recovery / learner install silently drops them: the voter set goes
        // empty (breaking `DescribeQuorum` and KIP-853 auto-join) and the
        // cluster `kraft.version` reverts to 0. The cluster `kraft_version`
        // in particular has no other persistence path — unlike the openraft
        // membership (restored from `SnapshotMeta.last_membership`), it is not
        // carried anywhere outside the metadata records, so the image is its
        // only source of truth. Independent of every other record, so order
        // here is irrelevant. The reconfig coordinator commits the openraft
        // membership change and the authoritative `V1Voters` record under one
        // lock, so the emitted voter set mirrors `last_membership` for the
        // same committed prefix (any transient one-reconfig skew self-heals as
        // the trailing entry replays from the retained log — the same
        // eventual lockstep that holds in live operation).
        if self.kraft_version != 0 {
            out.push(MetadataRecord::V1KRaftVersion(KRaftVersionRecord {
                kraft_version: self.kraft_version,
            }));
        }
        if !self.voters.is_empty() {
            out.push(MetadataRecord::V1Voters(VotersRecord {
                voters: self.voters.clone(),
            }));
        }

        out
    }

    fn push_partition_offset_advances(&self, out: &mut Vec<MetadataRecord>) {
        for (topic, partition, count) in
            self.partition_next_offsets
                .iter()
                .flat_map(|(topic, partitions)| {
                    partitions
                        .iter()
                        .map(move |(partition, value)| (topic, partition, value))
                })
        {
            out.push(MetadataRecord::V1PartitionOffsetAdvance(
                PartitionOffsetAdvanceRecord {
                    topic: topic.clone(),
                    partition: *partition,
                    count: *count,
                },
            ));
        }
    }

    /// Reconstruct an image from a `cluster_id` and a record sequence,
    /// typically the output of [`Self::to_records`] read back from a snapshot.
    /// The method `new`s an empty image and `apply`s each record in order.
    #[tracing::instrument(level = "info", skip_all, fields(records = records.len()))]
    #[must_use]
    pub fn from_records(cluster_id: Uuid, records: &[MetadataRecord]) -> Self {
        let mut image = Self::new(cluster_id);
        for rec in records {
            image.apply(rec);
        }
        image
    }

    /// Synchronous pre-validation: returns `Ok` if the record would be a
    /// no-conflict apply, otherwise the appropriate error.
    /// `Controller::submit_change` uses this before it forwards to openraft.
    ///
    /// # Errors
    /// Returns the conflict that would prevent the record from being applied.
    #[tracing::instrument(level = "debug", skip_all, fields(record = record_variant(rec)), err)]
    pub fn validate(&self, rec: &MetadataRecord) -> Result<(), MetadataError> {
        match rec {
            MetadataRecord::V1Topic(t) => {
                if let Some(existing) = self.topics.get(&t.name) {
                    // Updating an existing topic is allowed only if it's a
                    // strict partition-count expansion that preserves
                    // identity: same topic_id, same replication_factor,
                    // partitions strictly growing. CreatePartitions emits
                    // exactly this. Identical re-submits stay rejected so
                    // CreateTopics' idempotency contract still holds.
                    // Partition count derives from the partitions map (the
                    // KIP-631 round-trip does not carry the stored
                    // `TopicRecord.partitions`), so the expansion check reads
                    // the live count rather than `existing.partitions`.
                    if existing.topic_id != t.topic_id
                        || existing.replication_factor != t.replication_factor
                        || t.partitions <= self.topic_partition_count(&t.name)
                    {
                        return Err(MetadataError::TopicExists(t.name.clone()));
                    }
                    return Ok(());
                }
                // A new topic is valid regardless of the `TopicRecord.partitions`
                // count: KIP-631 framing does not carry that field, so a
                // freshly-decoded `V1Topic` round-trips back as partitions=0 with
                // the real count arriving via the following `V1Partition` records.
                // Partition-count validity (`> 0`) is enforced at the CreateTopics
                // request boundary, not on the metadata-record apply path.
                Ok(())
            }
            MetadataRecord::V1Partition(p) => {
                if !self.topics.contains_key(&p.topic) {
                    return Err(MetadataError::UnknownTopic(p.topic.clone()));
                }
                Ok(())
            }
            MetadataRecord::V1PartitionUpdate(update) => {
                if !self.topics.contains_key(&update.partition.topic) {
                    return Err(MetadataError::UnknownTopic(update.partition.topic.clone()));
                }
                Ok(())
            }
            MetadataRecord::V1PartitionElr(record) => {
                if self.partition(&record.topic, record.partition).is_none() {
                    return Err(MetadataError::UnknownTopic(record.topic.clone()));
                }
                Ok(())
            }
            MetadataRecord::V1PartitionRecovery(record) => {
                if self.partition(&record.topic, record.partition).is_none() {
                    return Err(MetadataError::UnknownTopic(record.topic.clone()));
                }
                Ok(())
            }
            MetadataRecord::V1DeleteTopic(d) => {
                if !self.topics.contains_key(&d.name) {
                    return Err(MetadataError::UnknownTopic(d.name.clone()));
                }
                Ok(())
            }
            MetadataRecord::V1TopicConfig(c) => {
                if !self.topics.contains_key(&c.topic) {
                    return Err(MetadataError::UnknownTopic(c.topic.clone()));
                }
                Ok(())
            }
            // Kafka's `ClusterControlManager` refuses to replay a registration
            // change or an unregistration unless the broker is registered at
            // the record's epoch, and a controller unregistration unless the
            // controller is registered. Apply treats those as no-ops, so they
            // are refused here, where the submitter still learns of it.
            MetadataRecord::V1BrokerRegistrationChange(change) => {
                if self.broker_epoch(change.node_id) != Some(change.broker_epoch) {
                    return Err(MetadataError::InvalidRecord(
                        "broker registration change for an epoch the broker is not registered at",
                    ));
                }
                Ok(())
            }
            MetadataRecord::V1UnregisterBroker(rec) => {
                if self.broker_epoch(rec.node_id) != Some(rec.broker_epoch) {
                    return Err(MetadataError::InvalidRecord(
                        "broker unregistration for an epoch the broker is not registered at",
                    ));
                }
                Ok(())
            }
            MetadataRecord::V1UnregisterController(rec) => {
                if !self.controllers.contains_key(&rec.node_id) {
                    return Err(MetadataError::InvalidRecord(
                        "controller unregistration for a controller that is not registered",
                    ));
                }
                Ok(())
            }
            MetadataRecord::V1BrokerRegistration(_)
            | MetadataRecord::V1ControllerRegistration(_)
            | MetadataRecord::V1ScramCredential(_)
            | MetadataRecord::V1DeleteScramCredential(_)
            | MetadataRecord::V1AccessControlEntry(_)
            | MetadataRecord::V1DeleteAccessControlEntry(_)
            | MetadataRecord::V1BrokerConfig(_)
            | MetadataRecord::V1ClientQuota(_)
            // Delegation-token records have no
            // topic-store concerns and admission is gated by the
            // handler-side checks (KIP-48 §protocol errors), so the
            // image-level validate is unconditional Ok.
            | MetadataRecord::V1DelegationToken(_)
            | MetadataRecord::V1DeleteDelegationToken(_)
            // KIP-853: voter-set / kraft.version records are validated by
            // the reconfiguration coordinator before submission; the
            // image-level apply is an unconditional replacement.
            | MetadataRecord::V1KRaftVersion(_)
            | MetadataRecord::V1Voters(_)
            // KIP-584: feature-level admission is fully gated by the
            // UpdateFeatures handler (supported-range + downgrade checks);
            // image-level apply is an idempotent map upsert.
            | MetadataRecord::V1FeatureLevel(_)
            // KIP-714: client-metrics subscription config. The
            // IncrementalAlterConfigs handler validates the subscription
            // name and config keys before submission; image-level apply
            // is an idempotent map upsert (or removal on empty map).
            | MetadataRecord::V1ClientMetricsConfig(_)
            // KIP-1071: GROUP config keys and values are validated by the
            // IncrementalAlterConfigs handler before this whole-map upsert.
            | MetadataRecord::V1GroupConfig(_)
            // Snapshot-only epoch carrier: only ever produced by `to_records`
            // and replayed on snapshot install, never submitted as a change.
            // Validated permissively for match-exhaustiveness.
            | MetadataRecord::V1FeaturesEpoch(_)
            // KIP-858 directory-assignment delta: a merge into one replica's
            // `directories` slot. The handler already resolved topic/partition/
            // replica against the image; apply is an idempotent slot upsert
            // (a no-op if the partition or replica is unknown).
            | MetadataRecord::V1PartitionDirAssignment(_)
            // Diskless offset-sequencer delta. The handler supplies a positive
            // count; image-level apply is an unconditional increment.
            | MetadataRecord::V1PartitionOffsetAdvance(_)
            // KFC-9: the expiry sweep removes a proposal by id, and apply is
            // idempotent against an id that is already gone.
            | MetadataRecord::V1DeleteBreakGlassProposal(_) => Ok(()),
            // KFC-9: a freeze scope must not be empty, and that is the only
            // check. The scope deliberately does NOT have to name an existing
            // topic: a prefix scope names none, and a freeze put on a
            // namespace BEFORE a restore writes into it is the
            // disaster-recovery case the feature exists for.
            MetadataRecord::V1TopicFreeze(freeze) => {
                if freeze.scope.is_empty() {
                    return Err(MetadataError::EmptyFreezeScope);
                }
                Ok(())
            }
            // KFC-9: the concurrent-approval rule. Two approvers read the
            // same image and each submits the approval list it read plus its
            // own entry. Without this check the second write overwrites the
            // first, and the proposal never reaches two principals. So an
            // incoming list must be the stored list plus zero or more
            // appended entries, and a settled proposal takes no change at
            // all.
            MetadataRecord::V1BreakGlassProposal(proposal) => {
                let Some(stored) = self.break_glass.get(&proposal.proposal_id) else {
                    return Ok(());
                };
                if stored.consumed_at_ms != 0 {
                    return Err(MetadataError::BreakGlassProposalConsumed(
                        proposal.proposal_id,
                    ));
                }
                if stored.withdrawn {
                    return Err(MetadataError::BreakGlassProposalWithdrawn(
                        proposal.proposal_id,
                    ));
                }
                if !extends(&stored.approvals, &proposal.approvals) {
                    return Err(MetadataError::BreakGlassApprovalsNotAnExtension(
                        proposal.proposal_id,
                    ));
                }
                Ok(())
            }
            MetadataRecord::V1ProducerIds(record) => {
                if record.next_producer_id <= self.next_producer_id() {
                    return Err(MetadataError::InvalidRecord(
                        "next producer ID must advance monotonically",
                    ));
                }
                if self.broker_epoch(record.broker_id) != Some(record.broker_epoch) {
                    return Err(MetadataError::InvalidRecord(
                        "producer ID allocation has a stale broker epoch",
                    ));
                }
                Ok(())
            }
        }
    }
}

/// KIP-1276: Kafka's `AclControlManager.validateHostPattern` treats any ACL
/// host containing `/` as a CIDR block, which needs `metadata.version`
/// `4.4-IV1`.
fn is_cidr_acl_host(host: &str) -> bool {
    host.contains('/')
}

#[cfg(test)]
mod tests {
    use krabka_units::bytes_per_sec;

    use super::*;

    /// `record_variant` maps each enum variant to its exact discriminant
    /// name. Assertions on concrete mappings kill whole-fn replacements, for
    /// example a return of `""` or of a constant.
    /// Three readers on the image had no caller in any test: a
    /// `group_configs` or `controllers` replaced by an empty iterator, and a
    /// `controller` replaced by `None`, all describe a cluster with nothing
    /// registered -- which is what an image looks like before any record is
    /// applied, so nothing else notices.
    #[test]
    fn group_config_and_controller_readers_report_applied_records() {
        let mut image = MetadataImage::new(Uuid::nil());
        image.apply(&MetadataRecord::V1GroupConfig(GroupConfigRecord {
            group_id: "streams-app".into(),
            configs: BTreeMap::from([("streams.num.standby.replicas".into(), "1".into())]),
        }));
        image.apply(&MetadataRecord::V1ControllerRegistration(
            ControllerRegistrationRecord {
                node_id: NodeId(3),
                incarnation_id: Uuid::from_u128(7),
                zk_migration_ready: false,
                endpoints: Vec::new(),
                features: BTreeMap::new(),
            },
        ));

        let groups: Vec<_> = image
            .group_configs()
            .map(|(id, cfg)| (id.clone(), cfg.clone()))
            .collect();
        check!(
            groups
                == vec![(
                    "streams-app".to_owned(),
                    BTreeMap::from([("streams.num.standby.replicas".to_owned(), "1".to_owned())])
                )]
        );

        let ids: Vec<_> = image.controllers().map(|c| c.node_id).collect();
        check!(ids == vec![NodeId(3)]);

        // Present and absent, so a reader stuck on `None` is not merely right
        // about the node that is missing.
        check!(image.controller(NodeId(3)).map(|c| c.incarnation_id) == Some(Uuid::from_u128(7)));
        check!(image.controller(NodeId(99)).is_none());
    }

    fn registered_broker(node_id: u64, broker_epoch: i64) -> BrokerRegistrationRecord {
        BrokerRegistrationRecord {
            node_id: NodeId(node_id),
            broker_epoch,
            incarnation_id: Uuid::from_u128(u128::from(node_id)),
            host: format!("broker-{node_id}"),
            port: 9092,
            rack: None,
            endpoints: vec![],
            log_dirs: vec![],
            fenced: true,
            in_controlled_shutdown: false,
            cordoned_log_dirs: None,
            features: BTreeMap::new(),
        }
    }

    /// Kafka's `ClusterControlManager` replays a `BrokerRegistrationChangeRecord`
    /// or an `UnregisterBrokerRecord` only against the registration at the
    /// record's epoch. Validate refuses the others and apply ignores them.
    #[test]
    fn registration_change_and_unregister_apply_only_at_the_registered_epoch() {
        use crate::records::{
            BrokerRegistrationChangeRecord, FencingChange, UnregisterBrokerRecord,
        };

        let registered = registered_broker(4, 12);
        let unfenced = BrokerRegistrationRecord {
            fenced: false,
            ..registered.clone()
        };
        let unfence = |node_id, broker_epoch| {
            MetadataRecord::V1BrokerRegistrationChange(BrokerRegistrationChangeRecord {
                fenced: FencingChange::Unfence,
                ..BrokerRegistrationChangeRecord::no_change(NodeId(node_id), broker_epoch)
            })
        };
        let unregister = |node_id, broker_epoch| {
            MetadataRecord::V1UnregisterBroker(UnregisterBrokerRecord {
                node_id: NodeId(node_id),
                broker_epoch,
            })
        };
        let stale_change = Err(MetadataError::InvalidRecord(
            "broker registration change for an epoch the broker is not registered at",
        ));
        let stale_unregister = Err(MetadataError::InvalidRecord(
            "broker unregistration for an epoch the broker is not registered at",
        ));
        for (case, record, want_validate, want_broker) in [
            (
                "unfence at the epoch",
                unfence(4, 12),
                Ok(()),
                Some(unfenced),
            ),
            (
                "unfence at a stale epoch",
                unfence(4, 11),
                stale_change.clone(),
                Some(registered.clone()),
            ),
            (
                "unfence an unknown broker",
                unfence(5, 12),
                stale_change,
                Some(registered.clone()),
            ),
            ("unregister at the epoch", unregister(4, 12), Ok(()), None),
            (
                "unregister at a stale epoch",
                unregister(4, 11),
                stale_unregister.clone(),
                Some(registered.clone()),
            ),
            (
                "unregister an unknown broker",
                unregister(5, 12),
                stale_unregister,
                Some(registered.clone()),
            ),
        ] {
            let mut image = MetadataImage::new(Uuid::nil());
            image.apply(&MetadataRecord::V1BrokerRegistration(registered.clone()));
            check!(image.validate(&record) == want_validate, "{case}");
            image.apply(&record);
            check!(image.broker(NodeId(4)).cloned() == want_broker, "{case}");
        }
    }

    /// KIP-1312: `UnregisterControllerRecord` removes the controller's
    /// registration. Kafka refuses to replay it for an unregistered id.
    #[test]
    fn unregister_controller_removes_the_registration() {
        use crate::records::{ControllerRegistrationRecord, UnregisterControllerRecord};

        let mut image = MetadataImage::new(Uuid::nil());
        for node_id in [3000, 3001] {
            image.apply(&MetadataRecord::V1ControllerRegistration(
                ControllerRegistrationRecord {
                    node_id: NodeId(node_id),
                    incarnation_id: Uuid::from_u128(u128::from(node_id)),
                    zk_migration_ready: false,
                    endpoints: Vec::new(),
                    features: BTreeMap::new(),
                },
            ));
        }
        let unregister = |node_id| {
            MetadataRecord::V1UnregisterController(UnregisterControllerRecord {
                node_id: NodeId(node_id),
            })
        };

        check!(image.validate(&unregister(3000)) == Ok(()));
        image.apply(&unregister(3000));
        let ids: Vec<_> = image.controllers().map(|c| c.node_id).collect();
        check!(ids == vec![NodeId(3001)]);

        check!(
            image.validate(&unregister(3000))
                == Err(MetadataError::InvalidRecord(
                    "controller unregistration for a controller that is not registered",
                ))
        );
        image.apply(&unregister(3000));
        let ids: Vec<_> = image.controllers().map(|c| c.node_id).collect();
        check!(ids == vec![NodeId(3001)]);
    }

    /// Deleting a topic drops that topic's offset sequencer state and leaves
    /// every other topic's alone. The retain predicate is what decides which is
    /// which: inverted, deleting one topic keeps only that topic's offsets and
    /// discards the rest of the cluster's.
    #[test]
    fn deleting_a_topic_drops_only_its_own_partition_offsets() {
        let mut image = MetadataImage::new(Uuid::nil());
        for topic in ["orders", "payments"] {
            image.apply(&MetadataRecord::V1PartitionOffsetAdvance(
                PartitionOffsetAdvanceRecord {
                    topic: topic.into(),
                    partition: 0,
                    count: 5,
                },
            ));
        }
        check!(
            (
                image.partition_next_offset("orders", 0),
                image.partition_next_offset("payments", 0),
            ) == (Some(5), Some(5))
        );

        image.apply(&MetadataRecord::V1DeleteTopic(DeleteTopicRecord {
            name: "orders".into(),
        }));
        check!(
            (
                image.partition_next_offset("orders", 0),
                image.partition_next_offset("payments", 0),
            ) == (None, Some(5))
        );
    }

    fn auxiliary_records(topic: &str, partition: i32) -> [MetadataRecord; 3] {
        [
            MetadataRecord::V1PartitionOffsetAdvance(PartitionOffsetAdvanceRecord {
                topic: topic.into(),
                partition,
                count: 7,
            }),
            MetadataRecord::V1PartitionElr(PartitionElrRecord {
                topic: topic.into(),
                partition,
                eligible_leader_replicas: vec![NodeId(3), NodeId(1)],
                last_known_elr: vec![NodeId(2)],
            }),
            MetadataRecord::V1PartitionRecovery(PartitionRecoveryRecord {
                topic: topic.into(),
                partition,
                state: LeaderRecoveryState::Recovering,
            }),
        ]
    }

    #[test]
    fn deleting_topics_cleans_orphan_auxiliary_state_and_preserves_snapshot_records() {
        use serde_wincode::SerdeCompat;
        use wincode::Serialize as _;

        let mut image = MetadataImage::new(Uuid::nil());
        let mut expected = Vec::new();
        for topic in ["orders", "payments", "orphan"] {
            for partition in [7, 0, 2] {
                for record in auxiliary_records(topic, partition) {
                    image.apply(&record);
                    if topic == "payments" {
                        expected.push(record);
                    }
                }
            }
        }
        // No TopicRecord was required to create these auxiliary entries.
        let original = image.clone();
        for topic in ["unknown", "orders", "orphan", "orders"] {
            image.apply(&MetadataRecord::V1DeleteTopic(DeleteTopicRecord {
                name: topic.into(),
            }));
        }
        let records = image.to_records();
        let bytes = |records: &[MetadataRecord]| {
            let mut encoded: Vec<_> = records
                .iter()
                .map(|record| <SerdeCompat<MetadataRecord>>::serialize(record).unwrap())
                .collect();
            encoded.sort();
            encoded
        };
        // HashMap order is unspecified; every persisted record and its bytes
        // must remain identical, with the same dependency-category order.
        assert2::assert!(bytes(&records) == bytes(&expected));
        let kinds: Vec<_> = records
            .iter()
            .map(|record| match record {
                MetadataRecord::V1PartitionOffsetAdvance(_) => 0,
                MetadataRecord::V1PartitionElr(_) => 1,
                MetadataRecord::V1PartitionRecovery(_) => 2,
                _ => unreachable!(),
            })
            .collect();
        assert2::assert!(kinds == vec![0, 0, 0, 1, 1, 1, 2, 2, 2]);
        assert2::assert!(MetadataImage::from_records(Uuid::nil(), &records) == image);
        assert2::assert!(
            MetadataImage::from_records(Uuid::nil(), &original.to_records()) == original
        );
        for partition in [0, 2, 7] {
            assert2::assert!(image.partition_next_offset("orders", partition).is_none());
            assert2::assert!(image.partition_elr("orders", partition) == (&[][..], &[][..]));
            assert2::assert!(
                image.leader_recovery_state("orders", partition) == LeaderRecoveryState::Recovered
            );
            assert2::assert!(image.partition_next_offset("payments", partition) == Some(7));
            assert2::assert!(
                image.partition_elr("payments", partition)
                    == (&[NodeId(3), NodeId(1)][..], &[NodeId(2)][..])
            );
        }
        // Reusing the name starts fresh, without sequencer or ELR remnants.
        for record in auxiliary_records("orders", 0) {
            image.apply(&record);
        }
        assert2::assert!(image.partition_next_offset("orders", 0) == Some(7));
    }

    #[test]
    fn clearing_last_auxiliary_partition_leaves_no_empty_topic_groups() {
        let mut image = MetadataImage::new(Uuid::nil());
        for record in auxiliary_records("orders", 0) {
            image.apply(&record);
        }
        image.apply(&MetadataRecord::V1PartitionElr(PartitionElrRecord {
            topic: "orders".into(),
            partition: 0,
            eligible_leader_replicas: vec![],
            last_known_elr: vec![],
        }));
        image.apply(&MetadataRecord::V1PartitionRecovery(
            PartitionRecoveryRecord {
                topic: "orders".into(),
                partition: 0,
                state: LeaderRecoveryState::Recovered,
            },
        ));
        assert2::assert!(image.partition_elr.is_empty());
        assert2::assert!(image.partition_recovery.is_empty());
        assert2::assert!(MetadataImage::from_records(Uuid::nil(), &image.to_records()) == image);
    }

    #[test]
    fn partition_updates_preserve_absent_fields_and_other_auxiliary_partitions() {
        use crate::PartitionUpdateRecord;

        let mut image = MetadataImage::new(Uuid::nil());
        for partition in [0, 1] {
            for record in auxiliary_records("orders", partition) {
                image.apply(&record);
            }
        }
        let partition = PartitionRecord {
            topic: "orders".into(),
            partition: 0,
            ..PartitionRecord::default()
        };
        image.apply(&MetadataRecord::V1PartitionUpdate(PartitionUpdateRecord {
            partition: partition.clone(),
            eligible_leader_replicas: None,
            last_known_elr: None,
            recovery_state: None,
        }));
        assert2::assert!(
            image.partition_elr("orders", 0) == (&[NodeId(3), NodeId(1)][..], &[NodeId(2)][..])
        );
        assert2::assert!(
            image.leader_recovery_state("orders", 0) == LeaderRecoveryState::Recovering
        );
        image.apply(&MetadataRecord::V1PartitionUpdate(PartitionUpdateRecord {
            partition,
            eligible_leader_replicas: Some(vec![]),
            last_known_elr: Some(vec![]),
            recovery_state: Some(LeaderRecoveryState::Recovered),
        }));
        assert2::assert!(image.partition_elr("orders", 0) == (&[][..], &[][..]));
        assert2::assert!(
            image.leader_recovery_state("orders", 0) == LeaderRecoveryState::Recovered
        );
        assert2::assert!(
            image.partition_elr("orders", 1) == (&[NodeId(3), NodeId(1)][..], &[NodeId(2)][..])
        );
        assert2::assert!(
            image.leader_recovery_state("orders", 1) == LeaderRecoveryState::Recovering
        );
        assert2::assert!(image.partition_next_offset("orders", 0) == Some(7));
        assert2::assert!(MetadataImage::from_records(Uuid::nil(), &image.to_records()) == image);
    }

    fn broker_at(node_id: NodeId, epoch: i64) -> MetadataRecord {
        MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
            node_id,
            broker_epoch: epoch,
            incarnation_id: Uuid::nil(),
            host: "h".into(),
            port: 9092,
            rack: None,
            log_dirs: vec![],
            fenced: false,
            in_controlled_shutdown: false,
            cordoned_log_dirs: None,
            endpoints: vec![],
            features: BTreeMap::new(),
        })
    }

    /// `validate` gates producer-ID allocation on two conditions, and both
    /// boundaries were unasserted: the id must strictly advance, and the
    /// record's broker epoch must be the epoch the image holds. Inverted, the
    /// first accepts exactly the replays it exists to reject, and the second
    /// accepts only the stale epochs.
    #[test]
    fn producer_id_allocation_must_advance_and_carry_the_live_epoch() {
        let mut image = MetadataImage::new(Uuid::nil());
        image.apply(&broker_at(NodeId(1), 11));
        image.apply(&MetadataRecord::V1ProducerIds(ProducerIdsRecord {
            broker_id: NodeId(1),
            broker_epoch: 11,
            next_producer_id: 1000,
        }));

        let allocation = |next: i64, epoch: i64| {
            MetadataRecord::V1ProducerIds(ProducerIdsRecord {
                broker_id: NodeId(1),
                broker_epoch: epoch,
                next_producer_id: next,
            })
        };

        check!(image.validate(&allocation(2000, 11)).is_ok());
        // Equal is not an advance, and neither is going backwards.
        check!(image.validate(&allocation(1000, 11)).is_err());
        check!(image.validate(&allocation(999, 11)).is_err());
        // A stale epoch is refused even when the id does advance.
        check!(image.validate(&allocation(2000, 10)).is_err());
    }

    /// `apply` keeps the higher allocation and drops the rest. At equality it
    /// keeps what it has, so a record carrying the same id from another broker
    /// must not displace the one already recorded -- which only the snapshot
    /// shows, since the id itself is unchanged either way.
    #[test]
    fn an_equal_producer_id_allocation_does_not_replace_the_recorded_one() {
        let mut image = MetadataImage::new(Uuid::nil());
        image.apply(&broker_at(NodeId(1), 11));
        image.apply(&broker_at(NodeId(2), 22));
        let original = ProducerIdsRecord {
            broker_id: NodeId(1),
            broker_epoch: 11,
            next_producer_id: 1000,
        };
        image.apply(&MetadataRecord::V1ProducerIds(original));
        image.apply(&MetadataRecord::V1ProducerIds(ProducerIdsRecord {
            broker_id: NodeId(2),
            broker_epoch: 22,
            next_producer_id: 1000,
        }));

        let recorded: Vec<_> = image
            .to_records()
            .into_iter()
            .filter(|r| matches!(r, MetadataRecord::V1ProducerIds(_)))
            .collect();
        check!(recorded == vec![MetadataRecord::V1ProducerIds(original)]);
    }

    #[test]
    fn record_variant_returns_exact_discriminant_names() {
        let topic = MetadataRecord::V1Topic(TopicRecord {
            name: "orders".into(),
            topic_id: Uuid::new_v4(),
            partitions: 3,
            replication_factor: 2,
        });
        assert2::assert!(record_variant(&topic) == "V1Topic");

        let feature = MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 1,
        });
        assert2::assert!(record_variant(&feature) == "V1FeatureLevel");
    }

    use assert2::check;

    use crate::{
        acl::{AclEntryFilter, AclOperation, PermissionType},
        break_glass::BreakGlassAction,
        records::{
            BrokerConfigRecord, BrokerEndpoint, ClientQuotaRecord, DeleteDelegationTokenRecord,
            DeleteScramCredentialRecord, DeleteTopicRecord, FeatureLevelRecord, LeaderEpoch,
            PartitionOffsetAdvanceRecord, QuotaEntity, ScramCredentialRecord,
        },
        voters::{KRaftVersionRange, Voter},
    };

    fn img() -> MetadataImage {
        MetadataImage::new(Uuid::nil())
    }

    #[test]
    fn fresh_image_has_no_features_and_unknown_epoch() {
        let m = img();
        assert2::assert!(m.finalized_features().is_empty());
        assert2::assert!(m.finalized_features_epoch() == -1);
    }

    #[test]
    fn finalized_feature_reads_arbitrary_name() {
        let mut m = MetadataImage::new(uuid::Uuid::nil());
        assert2::assert!(m.finalized_feature("transaction.version").is_none());
        m.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "transaction.version".into(),
            level: 2,
        }));
        assert2::assert!(m.finalized_feature("transaction.version") == Some(2));
    }

    #[test]
    fn apply_feature_level_sets_level_and_bumps_epoch() {
        let mut m = img();
        m.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 1,
        }));
        assert2::assert!(m.finalized_features().get("metadata.version") == Some(&1));
        assert2::assert!(m.finalized_features_epoch() == 0);

        // A second apply bumps the epoch again (monotonic).
        m.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 1,
        }));
        assert2::assert!(m.finalized_features_epoch() == 1);
    }

    #[test]
    fn apply_feature_level_zero_deletes_entry() {
        let mut m = img();
        m.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 1,
        }));
        m.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 0,
        }));
        assert2::assert!(
            (
                m.finalized_features().get("metadata.version"),
                m.finalized_features_epoch(),
            ) == (None, 1)
        );
    }

    #[test]
    fn to_records_from_records_round_trips() {
        let cid = Uuid::new_v4();
        let mut image = MetadataImage::new(cid);
        image.apply(&MetadataRecord::V1Topic(TopicRecord {
            name: "orders".into(),
            topic_id: Uuid::new_v4(),
            partitions: 3,
            replication_factor: 2,
        }));
        assert2::assert!(MetadataImage::from_records(cid, &image.to_records()) == image);
    }

    #[test]
    fn offset_advance_applies_as_monotonic_delta() {
        let mut image = img();
        let advance = |count: i64| {
            MetadataRecord::V1PartitionOffsetAdvance(PartitionOffsetAdvanceRecord {
                topic: "t".into(),
                partition: 0,
                count,
            })
        };

        image.apply(&advance(3));
        image.apply(&advance(2));
        image.apply(&MetadataRecord::V1PartitionOffsetAdvance(
            PartitionOffsetAdvanceRecord {
                topic: "t".into(),
                partition: 1,
                count: 7,
            },
        ));

        assert2::assert!(image.partition_next_offset("t", 0) == Some(5));
        assert2::assert!(image.partition_next_offset("t", 1) == Some(7));
    }

    #[test]
    fn offset_advance_survives_snapshot_round_trip() {
        let mut image = img();
        image.apply(&MetadataRecord::V1PartitionOffsetAdvance(
            PartitionOffsetAdvanceRecord {
                topic: "t".into(),
                partition: 0,
                count: 5,
            },
        ));

        let restored = MetadataImage::from_records(Uuid::nil(), &image.to_records());

        assert2::assert!(restored.partition_next_offset("t", 0) == Some(5));
    }

    fn apply_broker_fixture(image: &mut MetadataImage) {
        for record in [
            MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
                node_id: NodeId(1),
                broker_epoch: 0,
                incarnation_id: Uuid::nil(),
                host: "h1".into(),
                port: 9092,
                rack: Some("r1".into()),
                log_dirs: vec![],
                fenced: false,
                in_controlled_shutdown: false,
                cordoned_log_dirs: None,
                endpoints: vec![BrokerEndpoint {
                    name: "EXTERNAL".into(),
                    host: "ext".into(),
                    port: 9093,
                    protocol: krabka_security::ListenerProtocol::SaslSsl,
                }],
                features: BTreeMap::new(),
            }),
            MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
                node_id: NodeId(2),
                broker_epoch: 0,
                incarnation_id: Uuid::nil(),
                host: "h2".into(),
                port: 9092,
                rack: None,
                log_dirs: vec![],
                fenced: false,
                in_controlled_shutdown: false,
                cordoned_log_dirs: None,
                endpoints: vec![],
                features: BTreeMap::new(),
            }),
            MetadataRecord::V1UnregisterBroker(UnregisterBrokerRecord {
                node_id: NodeId(2),
                broker_epoch: 0,
            }),
            MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
                node_id: NodeId(1),
                config_name: "leader.replication.throttled.rate".into(),
                config_value: Some("2048".into()),
            }),
            MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
                node_id: NodeId(1),
                config_name: "follower.replication.throttled.rate".into(),
                config_value: Some("4096".into()),
            }),
            MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
                node_id: NodeId(1),
                config_name: "follower.replication.throttled.rate".into(),
                config_value: None,
            }),
        ] {
            image.apply(&record);
        }
    }

    /// Exercises every stored variant that the image can hold, that is, the 9
    /// state-producing records. It also exercises tombstones whose effects
    /// must NOT leak into the snapshot. The round-trip must reproduce the
    /// image exactly.
    #[test]
    // exhaustive fixture over every stored variant
    fn to_records_round_trips_all_variants() {
        use krabka_security::{KafkaPrincipal, SaslMechanism};

        let cid = Uuid::new_v4();
        let mut image = MetadataImage::new(cid);

        // One broker survives and one is unregistered; one config survives
        // and one is deleted back to empty.
        apply_broker_fixture(&mut image);

        // Topics + partitions (one topic deleted → topic, partitions and
        // its config must vanish).
        image.apply(&MetadataRecord::V1Topic(TopicRecord {
            name: "orders".into(),
            topic_id: Uuid::new_v4(),
            partitions: 2,
            replication_factor: 2,
        }));
        image.apply(&MetadataRecord::V1Topic(TopicRecord {
            name: "doomed".into(),
            topic_id: Uuid::new_v4(),
            partitions: 1,
            replication_factor: 1,
        }));
        image.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "orders".into(),
            partition: 0,
            leader: NodeId(1),
            replicas: vec![NodeId(1), NodeId(2)],
            isr: vec![NodeId(1), NodeId(2)],
            leader_epoch: LeaderEpoch(4),
            adding_replicas: vec![NodeId(2)],
            removing_replicas: vec![],
            directories: vec![],
            partition_epoch: 0,
        }));
        image.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "orders".into(),
            partition: 1,
            leader: NodeId(2),
            replicas: vec![NodeId(2), NodeId(1)],
            isr: vec![NodeId(2)],
            leader_epoch: LeaderEpoch(7),
            adding_replicas: vec![],
            removing_replicas: vec![NodeId(1)],
            directories: vec![],
            partition_epoch: 0,
        }));
        image.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "doomed".into(),
            partition: 0,
            leader: NodeId(1),
            replicas: vec![NodeId(1)],
            isr: vec![NodeId(1)],
            leader_epoch: LeaderEpoch(0),
            adding_replicas: vec![],
            removing_replicas: vec![],
            directories: vec![],
            partition_epoch: 0,
        }));

        // Topic config.
        let mut overrides = BTreeMap::new();
        overrides.insert("retention.ms".to_owned(), "60000".to_owned());
        overrides.insert("segment.bytes".to_owned(), "1048576".to_owned());
        image.apply(&MetadataRecord::V1TopicConfig(TopicConfigRecord {
            topic: "orders".into(),
            overrides,
        }));

        // Delete the doomed topic (clears its partition + config too).
        image.apply(&MetadataRecord::V1DeleteTopic(DeleteTopicRecord {
            name: "doomed".into(),
        }));

        // SCRAM creds (one survives, one deleted).
        image.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "alice".into(),
            mechanism: SaslMechanism::ScramSha512,
            salt: vec![1; 16],
            stored_key: vec![2; 64],
            server_key: vec![3; 64],
            iterations: 8192,
        }));
        image.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "bob".into(),
            mechanism: SaslMechanism::ScramSha256,
            salt: vec![4; 16],
            stored_key: vec![5; 32],
            server_key: vec![6; 32],
            iterations: 4096,
        }));
        image.apply(&MetadataRecord::V1DeleteScramCredential(
            DeleteScramCredentialRecord {
                user: "bob".into(),
                mechanism: SaslMechanism::ScramSha256,
            },
        ));

        // ACLs: literal + prefixed survive; a deleted one must not.
        let literal = AclEntry {
            resource_type: ResourceType::Topic,
            resource_name: "orders".into(),
            pattern_type: PatternType::Literal,
            principal: "User:alice".into(),
            host: "*".into(),
            operation: AclOperation::Read,
            permission_type: PermissionType::Allow,
        };
        let prefixed = AclEntry {
            resource_type: ResourceType::Group,
            resource_name: "cg-".into(),
            pattern_type: PatternType::Prefixed,
            principal: "User:alice".into(),
            host: "*".into(),
            operation: AclOperation::Read,
            permission_type: PermissionType::Allow,
        };
        let doomed_acl = AclEntry {
            resource_type: ResourceType::Cluster,
            resource_name: "kafka-cluster".into(),
            pattern_type: PatternType::Literal,
            principal: "User:eve".into(),
            host: "*".into(),
            operation: AclOperation::Alter,
            permission_type: PermissionType::Deny,
        };
        image.apply(&MetadataRecord::V1AccessControlEntry(literal));
        image.apply(&MetadataRecord::V1AccessControlEntry(prefixed));
        image.apply(&MetadataRecord::V1AccessControlEntry(doomed_acl));
        image.apply(&MetadataRecord::V1DeleteAccessControlEntry(
            AclEntryFilter {
                resource_type: Some(ResourceType::Cluster),
                ..AclEntryFilter::default()
            },
        ));

        // Client quotas.
        image.apply(&MetadataRecord::V1ClientQuota(ClientQuotaRecord {
            entity: vec![
                QuotaEntity {
                    entity_type: "user".into(),
                    entity_name: Some("alice".into()),
                },
                QuotaEntity {
                    entity_type: "client-id".into(),
                    entity_name: Some("app1".into()),
                },
            ],
            config_key: "producer_byte_rate".into(),
            config_value: Some(1024.0),
        }));
        image.apply(&MetadataRecord::V1ClientQuota(ClientQuotaRecord {
            entity: vec![QuotaEntity {
                entity_type: "user".into(),
                entity_name: None,
            }],
            config_key: "consumer_byte_rate".into(),
            config_value: Some(2048.0),
        }));

        // Delegation tokens (one survives, one deleted).
        let alice = KafkaPrincipal {
            principal_type: "User".into(),
            name: "alice".into(),
        };
        let bob = KafkaPrincipal {
            principal_type: "User".into(),
            name: "bob".into(),
        };
        image.apply(&MetadataRecord::V1DelegationToken(DelegationTokenRecord {
            token_id: "tok-1".into(),
            owner: alice.clone(),
            requester: alice,
            issue_timestamp_ms: 1_000,
            expiry_timestamp_ms: 5_000,
            max_timestamp_ms: 10_000,
            renewers: vec![bob.clone()],
        }));
        image.apply(&MetadataRecord::V1DelegationToken(DelegationTokenRecord {
            token_id: "tok-2".into(),
            owner: bob.clone(),
            requester: bob,
            issue_timestamp_ms: 1_000,
            expiry_timestamp_ms: 5_000,
            max_timestamp_ms: 10_000,
            renewers: vec![],
        }));
        image.apply(&MetadataRecord::V1DeleteDelegationToken(
            DeleteDelegationTokenRecord {
                token_id: "tok-2".into(),
            },
        ));

        // KIP-584 finalized features: two live features at a non-trivial epoch
        // (the metadata.version re-finalize bumps the epoch past the live
        // count), so the trailing V1FeaturesEpoch carrier is exercised too.
        image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 24,
        }));
        image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 25,
        }));
        image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "group.version".into(),
            level: 1,
        }));
        assert2::assert!(image.finalized_features_epoch() == 2);

        let rebuilt = MetadataImage::from_records(cid, &image.to_records());
        assert2::assert!(rebuilt == image);
    }

    /// Finalized features and their epoch must survive a `to_records` and
    /// `from_records` round-trip exactly. A trailing snapshot-only record
    /// carries the epoch verbatim, and apply does NOT recompute it by a bump.
    /// That epoch is the history of `UpdateFeatures` applies, so an image whose
    /// epoch exceeds its live feature count still reproduces exactly. This test
    /// guards the regression where `to_records` emitted no feature records at
    /// all and the snapshot path silently dropped every finalized feature.
    #[test]
    fn to_records_round_trips_features_and_epoch() {
        let cid = Uuid::new_v4();
        let mut image = MetadataImage::new(cid);
        // Four applies → epoch climbs -1 → 0 → 1 → 2 → 3, while only two
        // features remain live (metadata.version re-finalized once).
        image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 24,
        }));
        image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 25,
        }));
        image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "group.version".into(),
            level: 1,
        }));
        image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 25,
        }));
        assert2::assert!(image.finalized_features_epoch() == 3);

        let rebuilt = MetadataImage::from_records(cid, &image.to_records());
        assert2::assert!(rebuilt == image);
        // Belt-and-suspenders beyond derived PartialEq: the two fields the
        // snapshot path used to lose.
        check!(
            (
                rebuilt.finalized_features().get("metadata.version"),
                rebuilt.finalized_features().get("group.version"),
                rebuilt.finalized_features_epoch(),
            ) == (Some(&25), Some(&1), 3)
        );
    }

    /// The KIP-853 voter set and the finalized cluster `kraft.version` must
    /// survive `to_records` and `from_records`. If `to_records` drops both, a
    /// snapshot recovery or a learner install rebuilds the image with an EMPTY
    /// voter set and `kraft_version = 0`. That breaks `DescribeQuorum` and
    /// KIP-853 auto-join. The cluster `kraft_version` has no other persistence
    /// source, because it is not part of openraft's membership, so it can only
    /// round-trip through a `V1KRaftVersion` record.
    #[test]
    fn to_records_preserves_voters_and_kraft_version() {
        use crate::{
            records::{KRaftVersionRecord, VotersRecord},
            voters::{KRaftVersionRange, Voter, VoterEndpoint, VoterSet},
        };

        let cid = Uuid::new_v4();
        let mut image = MetadataImage::new(cid);
        image.apply(&MetadataRecord::V1KRaftVersion(KRaftVersionRecord {
            kraft_version: 1,
        }));
        let voters = VoterSet::from_voters([
            Voter {
                id: NodeId(1),
                directory_id: Uuid::from_u128(1),
                endpoints: vec![VoterEndpoint {
                    name: "CONTROLLER".into(),
                    host: "127.0.0.1".into(),
                    port: 9093,
                }],
                kraft_version: KRaftVersionRange::default(),
            },
            Voter {
                id: NodeId(2),
                directory_id: Uuid::from_u128(2),
                endpoints: vec![VoterEndpoint {
                    name: "CONTROLLER".into(),
                    host: "127.0.0.1".into(),
                    port: 9094,
                }],
                kraft_version: KRaftVersionRange { min: 0, max: 1 },
            },
        ]);
        image.apply(&MetadataRecord::V1Voters(VotersRecord { voters }));

        let rebuilt = MetadataImage::from_records(cid, &image.to_records());

        assert2::assert!(rebuilt == image);
    }

    fn topic(name: &str, partitions: i32) -> MetadataRecord {
        MetadataRecord::V1Topic(TopicRecord {
            name: name.into(),
            topic_id: Uuid::new_v4(),
            partitions,
            replication_factor: 1,
        })
    }

    #[test]
    fn apply_topic_inserts() {
        let mut m = img();
        m.apply(&topic("t", 3));
        assert2::assert!(m.topic("t").is_some());
    }

    #[test]
    fn apply_dir_assignment_merges_one_slot_without_clobbering_reassignment() {
        // Regression guard for the KIP-858 dir-assignment clobber: a directory
        // report applied as a delta must set only the reporting replica's slot
        // and leave a concurrent reassignment's `adding_replicas` intact.
        let mut m = img();
        m.apply(&topic("t", 1));
        m.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "t".into(),
            partition: 0,
            leader: NodeId(1),
            replicas: vec![NodeId(1), NodeId(2), NodeId(3)],
            isr: vec![NodeId(1), NodeId(2)],
            leader_epoch: LeaderEpoch(0),
            adding_replicas: vec![NodeId(3)],
            removing_replicas: vec![],
            directories: vec![uuid::Uuid::nil(), uuid::Uuid::nil(), uuid::Uuid::nil()],
            partition_epoch: 0,
        }));
        let dir = uuid::Uuid::from_u128(0xABCD);
        m.apply(&MetadataRecord::V1PartitionDirAssignment(
            PartitionDirAssignmentRecord {
                topic: "t".into(),
                partition: 0,
                replica: NodeId(2),
                directory: dir,
            },
        ));
        let pr = m.partition("t", 0).unwrap();
        // Slot 1 (replica 2) set; slots 0 and 2 untouched. The delta did NOT
        // clobber the in-flight reassignment.
        assert2::assert!(
            *pr == PartitionRecord {
                topic: "t".to_owned(),
                partition: 0,
                leader: NodeId(1),
                replicas: vec![NodeId(1), NodeId(2), NodeId(3)],
                isr: vec![NodeId(1), NodeId(2)],
                leader_epoch: krabka_ids::LeaderEpoch(0),
                adding_replicas: vec![NodeId(3)],
                removing_replicas: vec![],
                directories: vec![uuid::Uuid::nil(), dir, uuid::Uuid::nil()],
                partition_epoch: 0,
            }
        );
    }

    #[test]
    fn apply_delete_clears_partitions() {
        let mut m = img();
        m.apply(&topic("t", 2));
        m.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "t".into(),
            partition: 0,
            leader: NodeId(1),
            replicas: vec![NodeId(1)],
            isr: vec![NodeId(1)],
            leader_epoch: LeaderEpoch(0),
            adding_replicas: vec![],
            removing_replicas: vec![],
            directories: vec![],
            partition_epoch: 0,
        }));
        m.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "t".into(),
            partition: 1,
            leader: NodeId(1),
            replicas: vec![NodeId(1)],
            isr: vec![NodeId(1)],
            leader_epoch: LeaderEpoch(0),
            adding_replicas: vec![],
            removing_replicas: vec![],
            directories: vec![],
            partition_epoch: 0,
        }));
        assert2::assert!(m.partitions_of("t").count() == 2);
        m.apply(&MetadataRecord::V1DeleteTopic(DeleteTopicRecord {
            name: "t".into(),
        }));
        assert2::assert!(m.topic("t").is_none());
        assert2::assert!(m.partitions_of("t").count() == 0);
    }

    #[test]
    fn validate_topic_exists_rejected() {
        let mut m = img();
        m.apply(&topic("t", 1));
        let err = m.validate(&topic("t", 1)).unwrap_err();
        assert2::assert!(matches!(err, MetadataError::TopicExists(_)));
    }

    /// Apply `count` partition records, with indices `0..count`, for `topic`,
    /// so the derived partition count of the image matches a realistic topic.
    /// The `validate` partition-count check reads this derived count and not
    /// the stored `TopicRecord.partitions`.
    fn apply_partitions(m: &mut MetadataImage, topic: &str, count: i32) {
        for p in 0..count {
            m.apply(&MetadataRecord::V1Partition(PartitionRecord {
                topic: topic.into(),
                partition: p,
                leader: NodeId(1),
                replicas: vec![NodeId(1)],
                isr: vec![NodeId(1)],
                leader_epoch: LeaderEpoch(0),
                adding_replicas: vec![],
                removing_replicas: vec![],
                directories: vec![],
                partition_epoch: 0,
            }));
        }
    }

    #[test]
    fn validate_topic_partition_count_increase_allowed() {
        let mut m = img();
        m.apply(&topic("t", 1));
        apply_partitions(&mut m, "t", 1);
        let existing = m.topic("t").unwrap().clone();
        let updated = MetadataRecord::V1Topic(TopicRecord {
            name: "t".into(),
            topic_id: existing.topic_id,
            partitions: 3,
            replication_factor: existing.replication_factor,
        });
        assert2::assert!(m.validate(&updated).is_ok());
    }

    #[test]
    fn validate_topic_partition_count_decrease_rejected() {
        let mut m = img();
        m.apply(&topic("t", 3));
        apply_partitions(&mut m, "t", 3);
        let existing = m.topic("t").unwrap().clone();
        let updated = MetadataRecord::V1Topic(TopicRecord {
            name: "t".into(),
            topic_id: existing.topic_id,
            partitions: 1,
            replication_factor: existing.replication_factor,
        });
        let err = m.validate(&updated).unwrap_err();
        assert2::assert!(matches!(err, MetadataError::TopicExists(_)));
    }

    #[test]
    fn validate_delete_unknown_topic_rejected() {
        let m = img();
        let err = m
            .validate(&MetadataRecord::V1DeleteTopic(DeleteTopicRecord {
                name: "ghost".into(),
            }))
            .unwrap_err();
        assert2::assert!(matches!(err, MetadataError::UnknownTopic(_)));
    }

    #[test]
    fn validate_partition_for_unknown_topic_rejected() {
        let m = img();
        let p = MetadataRecord::V1Partition(PartitionRecord {
            topic: "ghost".into(),
            partition: 0,
            leader: NodeId(1),
            replicas: vec![NodeId(1)],
            isr: vec![NodeId(1)],
            leader_epoch: LeaderEpoch(0),
            adding_replicas: vec![],
            removing_replicas: vec![],
            directories: vec![],
            partition_epoch: 0,
        });
        let err = m.validate(&p).unwrap_err();
        assert2::assert!(matches!(err, MetadataError::UnknownTopic(_)));
    }

    #[test]
    fn broker_registration_is_idempotent() {
        let mut m = img();
        let b = MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
            node_id: NodeId(1),
            broker_epoch: 0,
            incarnation_id: Uuid::nil(),
            host: "h".into(),
            port: 9092,
            rack: None,
            log_dirs: vec![],
            fenced: false,
            in_controlled_shutdown: false,
            cordoned_log_dirs: None,
            endpoints: vec![],
            features: BTreeMap::new(),
        });
        m.apply(&b);
        m.apply(&b);
        assert2::assert!(m.brokers().count() == 1);
    }

    #[test]
    fn apply_topic_config_inserts() {
        let mut m = img();
        m.apply(&topic("t", 1));
        let mut overrides = BTreeMap::new();
        overrides.insert("retention.ms".to_owned(), "60000".to_owned());
        m.apply(&MetadataRecord::V1TopicConfig(TopicConfigRecord {
            topic: "t".into(),
            overrides: overrides.clone(),
        }));
        assert2::assert!(m.topic_config("t") == Some(&overrides));
    }

    #[test]
    fn apply_topic_config_replaces_previous() {
        let mut m = img();
        m.apply(&topic("t", 1));

        let mut first = BTreeMap::new();
        first.insert("retention.ms".to_owned(), "60000".to_owned());
        first.insert("segment.bytes".to_owned(), "1024".to_owned());
        m.apply(&MetadataRecord::V1TopicConfig(TopicConfigRecord {
            topic: "t".into(),
            overrides: first,
        }));

        let mut second = BTreeMap::new();
        second.insert("retention.ms".to_owned(), "120000".to_owned());
        m.apply(&MetadataRecord::V1TopicConfig(TopicConfigRecord {
            topic: "t".into(),
            overrides: second.clone(),
        }));

        // segment.bytes is GONE — last-write-wins is authoritative.
        assert2::assert!(m.topic_config("t") == Some(&second));
    }

    #[test]
    fn delete_topic_clears_configs() {
        let mut m = img();
        m.apply(&topic("t", 1));
        let mut overrides = BTreeMap::new();
        overrides.insert("retention.ms".to_owned(), "60000".to_owned());
        m.apply(&MetadataRecord::V1TopicConfig(TopicConfigRecord {
            topic: "t".into(),
            overrides,
        }));
        m.apply(&MetadataRecord::V1DeleteTopic(DeleteTopicRecord {
            name: "t".into(),
        }));
        assert2::assert!(m.topic_config("t").is_none());
    }

    #[test]
    fn validate_topic_config_for_unknown_topic_rejected() {
        let m = img();
        let r = MetadataRecord::V1TopicConfig(TopicConfigRecord {
            topic: "ghost".into(),
            overrides: BTreeMap::new(),
        });
        let err = m.validate(&r).unwrap_err();
        assert2::assert!(matches!(err, MetadataError::UnknownTopic(_)));
    }

    #[test]
    fn apply_scram_credential_stores() {
        let mut m = img();
        m.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "alice".into(),
            mechanism: krabka_security::SaslMechanism::ScramSha512,
            salt: vec![1; 16],
            stored_key: vec![2; 64],
            server_key: vec![3; 64],
            iterations: 4096,
        }));
        let got = m.scram_credential("alice", krabka_security::SaslMechanism::ScramSha512);
        assert2::assert!(got.map(|credential| credential.iterations) == Some(4096));
    }

    #[test]
    fn apply_scram_credential_last_write_wins() {
        let mut m = img();
        let mech = krabka_security::SaslMechanism::ScramSha512;
        m.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "alice".into(),
            mechanism: mech,
            salt: vec![1; 16],
            stored_key: vec![2; 64],
            server_key: vec![3; 64],
            iterations: 4096,
        }));
        m.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "alice".into(),
            mechanism: mech,
            salt: vec![9; 16],
            stored_key: vec![9; 64],
            server_key: vec![9; 64],
            iterations: 8192,
        }));
        let got = m.scram_credential("alice", mech).unwrap();
        assert2::assert!((got.iterations, &got.salt) == (8192, &vec![9; 16]));
    }

    #[test]
    fn delete_scram_credential_removes() {
        let mut m = img();
        let mech = krabka_security::SaslMechanism::ScramSha512;
        m.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "alice".into(),
            mechanism: mech,
            salt: vec![1; 16],
            stored_key: vec![2; 64],
            server_key: vec![3; 64],
            iterations: 4096,
        }));
        m.apply(&MetadataRecord::V1DeleteScramCredential(
            DeleteScramCredentialRecord {
                user: "alice".into(),
                mechanism: mech,
            },
        ));
        assert2::assert!(m.scram_credential("alice", mech).is_none());
    }

    fn topic_read_for_alice() -> AclEntry {
        AclEntry {
            resource_type: ResourceType::Topic,
            resource_name: "foo".into(),
            pattern_type: PatternType::Literal,
            principal: "User:alice".into(),
            host: "*".into(),
            operation: AclOperation::Read,
            permission_type: PermissionType::Allow,
        }
    }

    fn topic_prefixed_team() -> AclEntry {
        AclEntry {
            resource_type: ResourceType::Topic,
            resource_name: "team-".into(),
            pattern_type: PatternType::Prefixed,
            principal: "User:alice".into(),
            host: "*".into(),
            operation: AclOperation::Read,
            permission_type: PermissionType::Allow,
        }
    }

    #[test]
    fn apply_v1_access_control_entry_literal_stores_in_literal_map() {
        let mut m = img();
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_read_for_alice()));
        let hits: Vec<_> = m.matching_acls(ResourceType::Topic, "foo").collect();
        assert2::assert!(
            hits.iter()
                .map(|entry| entry.resource_name.as_str())
                .collect::<Vec<_>>()
                == vec!["foo"]
        );
    }

    #[test]
    fn apply_v1_access_control_entry_replaces_identical_entry() {
        let mut m = img();
        let acl = topic_read_for_alice();
        m.apply(&MetadataRecord::V1AccessControlEntry(acl.clone()));
        m.apply(&MetadataRecord::V1AccessControlEntry(acl));

        let hits = m.matching_acls(ResourceType::Topic, "foo").count();
        assert2::assert!(hits == 1);
    }

    #[test]
    fn apply_v1_access_control_entry_prefixed_stores_in_prefixed_vec() {
        let mut m = img();
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_prefixed_team()));
        let hits: Vec<_> = m.matching_acls(ResourceType::Topic, "team-foo").collect();
        assert2::assert!(
            hits.iter()
                .map(|entry| entry.resource_name.as_str())
                .collect::<Vec<_>>()
                == vec!["team-"]
        );
        // Non-matching resource: empty.
        let none = m.matching_acls(ResourceType::Topic, "other").count();
        assert2::assert!(none == 0);
    }

    #[test]
    fn matching_acls_combines_literal_and_prefixed() {
        let mut m = img();
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_read_for_alice()));
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_prefixed_team()));
        let hits_foo = m.matching_acls(ResourceType::Topic, "foo").count();
        let hits_team = m.matching_acls(ResourceType::Topic, "team-x").count();
        assert2::assert!((hits_foo, hits_team) == (1, 1));
    }

    fn topic_wildcard_allow() -> AclEntry {
        AclEntry {
            resource_type: ResourceType::Topic,
            resource_name: "*".into(),
            pattern_type: PatternType::Literal,
            principal: "User:alice".into(),
            host: "*".into(),
            operation: AclOperation::Read,
            permission_type: PermissionType::Allow,
        }
    }

    fn topic_wildcard_deny() -> AclEntry {
        AclEntry {
            resource_type: ResourceType::Topic,
            resource_name: "*".into(),
            pattern_type: PatternType::Literal,
            principal: "User:alice".into(),
            host: "*".into(),
            operation: AclOperation::Read,
            permission_type: PermissionType::Deny,
        }
    }

    #[test]
    fn matching_acls_literal_wildcard_allow_matches_arbitrary_topic() {
        let mut m = img();
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_wildcard_allow()));
        // An arbitrarily-named topic the grant never mentions by name.
        let hits: Vec<_> = m
            .matching_acls(ResourceType::Topic, "some-random-topic")
            .collect();
        assert2::assert!(hits == vec![&topic_wildcard_allow()]);
        // Requesting "*" itself must not double-count the wildcard entry.
        let star = m.matching_acls(ResourceType::Topic, "*").count();
        assert2::assert!(star == 1);
    }

    #[test]
    fn matching_acls_literal_wildcard_deny_matches_arbitrary_topic() {
        let mut m = img();
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_wildcard_deny()));
        let hits: Vec<_> = m
            .matching_acls(ResourceType::Topic, "another-random-topic")
            .collect();
        assert2::assert!(hits == vec![&topic_wildcard_deny()]);
    }

    #[test]
    fn apply_v1_delete_access_control_entry_removes_matching() {
        let mut m = img();
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_read_for_alice()));
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_prefixed_team()));
        let filter = AclEntryFilter {
            resource_type: Some(ResourceType::Topic),
            pattern_type: Some(PatternType::Literal),
            ..AclEntryFilter::default()
        };
        m.apply(&MetadataRecord::V1DeleteAccessControlEntry(filter));
        let hits_foo = m.matching_acls(ResourceType::Topic, "foo").count();
        let hits_team = m.matching_acls(ResourceType::Topic, "team-x").count();
        assert2::assert!((hits_foo, hits_team) == (0, 1));
    }

    #[test]
    fn apply_v1_delete_access_control_entry_no_match_is_noop() {
        let mut m = img();
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_read_for_alice()));
        let filter = AclEntryFilter {
            resource_type: Some(ResourceType::Group),
            ..AclEntryFilter::default()
        };
        m.apply(&MetadataRecord::V1DeleteAccessControlEntry(filter));
        let hits = m.matching_acls(ResourceType::Topic, "foo").count();
        assert2::assert!(hits == 1);
    }

    #[test]
    fn all_acls_returns_every_entry() {
        let mut m = img();
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_read_for_alice()));
        m.apply(&MetadataRecord::V1AccessControlEntry(topic_prefixed_team()));
        assert2::assert!(m.all_acls().count() == 2);
    }

    #[test]
    fn all_partitions_count_matches_map_size() {
        let mut m = img();
        m.apply(&topic("t", 3));
        for p in 0..3 {
            m.apply(&MetadataRecord::V1Partition(PartitionRecord {
                topic: "t".into(),
                partition: p,
                leader: NodeId(1),
                replicas: vec![NodeId(1)],
                isr: vec![NodeId(1)],
                leader_epoch: LeaderEpoch(0),
                adding_replicas: vec![],
                removing_replicas: vec![],
                directories: vec![],
                partition_epoch: 0,
            }));
        }
        m.apply(&topic("u", 1));
        m.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "u".into(),
            partition: 0,
            leader: NodeId(1),
            replicas: vec![NodeId(1)],
            isr: vec![NodeId(1)],
            leader_epoch: LeaderEpoch(0),
            adding_replicas: vec![],
            removing_replicas: vec![],
            directories: vec![],
            partition_epoch: 0,
        }));
        // 3 partitions for "t" + 1 for "u" = 4, one walk, no per-topic filter.
        assert2::assert!(m.all_partitions().count() == 4);
        assert2::assert!(
            m.all_partitions().count()
                == m.partitions_of("t").count() + m.partitions_of("u").count()
        );
    }

    /// `total_partitions` sums the by-topic index across every topic. It backs
    /// the `to_records` snapshot tracing field, whose span is inert under the
    /// test subscriber. This test therefore pins its value directly. Without
    /// that pin, a constant-return mutant survives and no behavioral test
    /// catches it.
    #[test]
    fn total_partitions_sums_across_topics() {
        let mut m = img();
        // Empty image: no topics, no partitions.
        check!(m.total_partitions() == 0);

        // "a" gets 2 partitions, "b" gets 1 → 3 total across the index.
        m.apply(&topic("a", 2));
        m.apply(&topic("b", 1));
        for (name, idx) in [("a", 0), ("a", 1), ("b", 0)] {
            m.apply(&MetadataRecord::V1Partition(PartitionRecord {
                topic: name.into(),
                partition: idx,
                leader: NodeId(1),
                replicas: vec![NodeId(1)],
                isr: vec![NodeId(1)],
                leader_epoch: LeaderEpoch(0),
                adding_replicas: vec![],
                removing_replicas: vec![],
                directories: vec![],
                partition_epoch: 0,
            }));
        }
        check!(m.total_partitions() == 3);
        // Matches an independent full walk, and is not the mutant constant 1.
        check!(m.total_partitions() == m.all_partitions().count());
    }

    /// Pins the `partitions_of` ordering contract: ascending partition-index
    /// order regardless of apply order. The `Metadata` and
    /// `DescribeTopicPartitions` rows and the cursor-based pagination rely on
    /// it.
    #[test]
    fn partitions_of_yields_ascending_partition_index_order() {
        let mut m = img();
        m.apply(&topic("t", 3));
        for p in [2, 0, 1] {
            m.apply(&MetadataRecord::V1Partition(PartitionRecord {
                topic: "t".into(),
                partition: p,
                leader: NodeId(1),
                replicas: vec![NodeId(1)],
                isr: vec![NodeId(1)],
                leader_epoch: LeaderEpoch(0),
                adding_replicas: vec![],
                removing_replicas: vec![],
                directories: vec![],
                partition_epoch: 0,
            }));
        }
        let idxs: Vec<i32> = m.partitions_of("t").map(|p| p.partition).collect();
        assert2::assert!(idxs == vec![0, 1, 2]);
    }

    /// A delete of one topic must not disturb the partitions of another topic
    /// in the by-topic index. `partitions_of` for the deleted topic goes
    /// empty, and its sibling stays intact.
    #[test]
    fn delete_topic_leaves_other_topics_partitions_intact() {
        let mut m = img();
        m.apply(&topic("doomed", 1));
        m.apply(&topic("survivor", 1));
        for name in ["doomed", "survivor"] {
            m.apply(&MetadataRecord::V1Partition(PartitionRecord {
                topic: name.into(),
                partition: 0,
                leader: NodeId(1),
                replicas: vec![NodeId(1)],
                isr: vec![NodeId(1)],
                leader_epoch: LeaderEpoch(0),
                adding_replicas: vec![],
                removing_replicas: vec![],
                directories: vec![],
                partition_epoch: 0,
            }));
        }
        m.apply(&MetadataRecord::V1DeleteTopic(DeleteTopicRecord {
            name: "doomed".into(),
        }));
        check!(
            (
                m.partitions_of("doomed").count(),
                m.topic_partition_count("doomed"),
                m.partitions_of("survivor").count(),
                m.partition("survivor", 0).is_some(),
                m.all_partitions().count(),
            ) == (0, 0, 1, true, 1)
        );
    }

    #[test]
    fn reassignments_in_flight_excludes_idle_partitions() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1Topic(TopicRecord {
            name: "foo".into(),
            topic_id: uuid::Uuid::nil(),
            partitions: 1,
            replication_factor: 3,
        }));
        img.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "foo".into(),
            partition: 0,
            leader: NodeId(1),
            replicas: vec![NodeId(1), NodeId(2), NodeId(3)],
            isr: vec![NodeId(1), NodeId(2), NodeId(3)],
            leader_epoch: LeaderEpoch(0),
            adding_replicas: vec![],
            removing_replicas: vec![],
            directories: vec![],
            partition_epoch: 0,
        }));
        assert2::assert!(img.reassignments_in_flight().count() == 0);
    }

    #[test]
    fn reassignments_in_flight_returns_partitions_with_adding() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1Topic(TopicRecord {
            name: "foo".into(),
            topic_id: uuid::Uuid::nil(),
            partitions: 1,
            replication_factor: 3,
        }));
        img.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "foo".into(),
            partition: 0,
            leader: NodeId(1),
            replicas: vec![NodeId(1), NodeId(2), NodeId(3), NodeId(4)],
            isr: vec![NodeId(1), NodeId(2), NodeId(3)],
            leader_epoch: LeaderEpoch(0),
            adding_replicas: vec![NodeId(4)],
            removing_replicas: vec![],
            directories: vec![],
            partition_epoch: 0,
        }));
        let rows: Vec<_> = img.reassignments_in_flight().collect();
        assert2::assert!(
            rows.iter()
                .map(|row| row.adding_replicas.clone())
                .collect::<Vec<_>>()
                == vec![vec![NodeId(4)]]
        );
    }

    #[test]
    fn reassignments_in_flight_returns_partitions_with_removing() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1Topic(TopicRecord {
            name: "foo".into(),
            topic_id: uuid::Uuid::nil(),
            partitions: 1,
            replication_factor: 3,
        }));
        img.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "foo".into(),
            partition: 0,
            leader: NodeId(1),
            replicas: vec![NodeId(1), NodeId(2), NodeId(3)],
            isr: vec![NodeId(1), NodeId(2), NodeId(3)],
            leader_epoch: LeaderEpoch(0),
            adding_replicas: vec![],
            removing_replicas: vec![NodeId(3)],
            directories: vec![],
            partition_epoch: 0,
        }));
        let rows: Vec<_> = img.reassignments_in_flight().collect();
        assert2::assert!(
            rows.iter()
                .map(|row| row.removing_replicas.clone())
                .collect::<Vec<_>>()
                == vec![vec![NodeId(3)]]
        );
    }

    #[test]
    fn reassignments_in_flight_covers_multiple_topics() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        for name in ["foo", "bar"] {
            img.apply(&MetadataRecord::V1Topic(TopicRecord {
                name: name.into(),
                topic_id: uuid::Uuid::nil(),
                partitions: 1,
                replication_factor: 3,
            }));
            img.apply(&MetadataRecord::V1Partition(PartitionRecord {
                topic: name.into(),
                partition: 0,
                leader: NodeId(1),
                replicas: vec![NodeId(1), NodeId(2), NodeId(3), NodeId(4)],
                isr: vec![NodeId(1), NodeId(2), NodeId(3)],
                leader_epoch: LeaderEpoch(0),
                adding_replicas: vec![NodeId(4)],
                removing_replicas: vec![],
                directories: vec![],
                partition_epoch: 0,
            }));
        }
        assert2::assert!(img.reassignments_in_flight().count() == 2);
    }

    #[test]
    fn broker_config_set_inserts_into_image() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
            node_id: NodeId(1),
            config_name: "leader.replication.throttled.rate".into(),
            config_value: Some("2048".into()),
        }));
        let bc = img.broker_config(NodeId(1)).expect("broker config");
        assert2::assert!(bc.get("leader.replication.throttled.rate") == Some(&"2048".to_owned()));
    }

    #[test]
    fn broker_config_delete_removes_from_image() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
            node_id: NodeId(1),
            config_name: "leader.replication.throttled.rate".into(),
            config_value: Some("2048".into()),
        }));
        img.apply(&MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
            node_id: NodeId(1),
            config_name: "leader.replication.throttled.rate".into(),
            config_value: None,
        }));
        let bc = img
            .broker_config(NodeId(1))
            .expect("broker_configs entry retained");
        assert2::assert!(bc.get("leader.replication.throttled.rate").is_none());
    }

    #[test]
    fn broker_throttle_rate_parses_positive_value() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
            node_id: NodeId(1),
            config_name: "leader.replication.throttled.rate".into(),
            config_value: Some("2048".into()),
        }));
        assert2::assert!(
            img.broker_throttle_rate(NodeId(1), ThrottleKind::Leader) == Some(bytes_per_sec(2048))
        );

        img.apply(&MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
            node_id: NodeId(1),
            config_name: "replica.alter.log.dirs.io.max.bytes.per.second".into(),
            config_value: Some("4096".into()),
        }));
        assert2::assert!(
            img.broker_throttle_rate(NodeId(1), ThrottleKind::AlterLogDirs)
                == Some(bytes_per_sec(4096))
        );
    }

    /// The config value is a `String` holding Kafka's `int64` quota, and the
    /// quantity stores `f64`. Every magnitude below 2^53 round-trips exactly,
    /// which covers the whole range the config key can express in practice.
    /// This test pins that, so the accessor is not quietly lossy at the top
    /// end.
    #[test]
    fn broker_throttle_rate_round_trips_large_quotas_exactly() {
        for raw in ["0", "1", "1073741824", "9007199254740992"] {
            let mut img = MetadataImage::new(uuid::Uuid::nil());
            img.apply(&MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
                node_id: NodeId(7),
                config_name: "follower.replication.throttled.rate".into(),
                config_value: Some(raw.into()),
            }));
            let rate = img
                .broker_throttle_rate(NodeId(7), ThrottleKind::Follower)
                .expect("configured rate");
            assert2::check!(rate.bytes_per_sec_i64().to_string() == raw);
        }
    }

    #[test]
    fn broker_throttle_rate_returns_none_for_negative_one() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
            node_id: NodeId(1),
            config_name: "leader.replication.throttled.rate".into(),
            config_value: Some("-1".into()),
        }));
        assert2::assert!(
            img.broker_throttle_rate(NodeId(1), ThrottleKind::Leader)
                .is_none()
        );
    }

    #[test]
    fn broker_throttle_rate_uses_cluster_default() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
            node_id: DEFAULT_BROKER_CONFIG_NODE_ID,
            config_name: "leader.replication.throttled.rate".into(),
            config_value: Some("2048".into()),
        }));

        assert2::assert!(
            img.default_broker_config()
                .and_then(|configs| configs.get("leader.replication.throttled.rate"))
                .map(String::as_str)
                == Some("2048")
        );
        assert2::assert!(
            img.broker_throttle_rate(NodeId(7), ThrottleKind::Leader) == Some(bytes_per_sec(2048))
        );
    }

    #[test]
    fn per_broker_disabled_rate_overrides_cluster_default() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        for (node_id, value) in [(DEFAULT_BROKER_CONFIG_NODE_ID, "2048"), (NodeId(7), "-1")] {
            img.apply(&MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
                node_id,
                config_name: "leader.replication.throttled.rate".into(),
                config_value: Some(value.into()),
            }));
        }

        assert2::assert!(
            img.broker_throttle_rate(NodeId(7), ThrottleKind::Leader)
                .is_none()
        );
    }

    #[test]
    fn client_quota_apply_inserts_canonicalized() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        // Input order: (user, client-id) — should canonicalize to (client-id, user).
        img.apply(&MetadataRecord::V1ClientQuota(ClientQuotaRecord {
            entity: vec![
                QuotaEntity {
                    entity_type: "user".into(),
                    entity_name: Some("alice".into()),
                },
                QuotaEntity {
                    entity_type: "client-id".into(),
                    entity_name: Some("app1".into()),
                },
            ],
            config_key: "producer_byte_rate".into(),
            config_value: Some(1024.0),
        }));
        let key: EntityKey = vec![
            ("client-id".into(), Some("app1".into())),
            ("user".into(), Some("alice".into())),
        ];
        let configs = img
            .client_quotas()
            .get(&key)
            .expect("entry under canonical key");
        assert2::assert!(configs.get("producer_byte_rate") == Some(&1024.0));
    }

    #[test]
    fn client_quota_apply_delete_removes_key() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1ClientQuota(ClientQuotaRecord {
            entity: vec![QuotaEntity {
                entity_type: "user".into(),
                entity_name: Some("alice".into()),
            }],
            config_key: "producer_byte_rate".into(),
            config_value: Some(1024.0),
        }));
        img.apply(&MetadataRecord::V1ClientQuota(ClientQuotaRecord {
            entity: vec![QuotaEntity {
                entity_type: "user".into(),
                entity_name: Some("alice".into()),
            }],
            config_key: "producer_byte_rate".into(),
            config_value: None,
        }));
        let key: EntityKey = vec![("user".into(), Some("alice".into()))];
        let configs = img.client_quotas().get(&key).expect("entry retained");
        assert2::assert!(configs.get("producer_byte_rate").is_none());
    }

    #[test]
    fn client_quota_default_entity_uses_none_name() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1ClientQuota(ClientQuotaRecord {
            entity: vec![QuotaEntity {
                entity_type: "user".into(),
                entity_name: None,
            }],
            config_key: "producer_byte_rate".into(),
            config_value: Some(512.0),
        }));
        let key: EntityKey = vec![("user".into(), None)];
        assert2::assert!(img.client_quotas().contains_key(&key));
    }

    #[test]
    fn canonicalize_sorts_alphabetically_by_entity_type() {
        let input = vec![
            ("user".to_owned(), Some("alice".to_owned())),
            ("client-id".to_owned(), Some("app1".to_owned())),
        ];
        let canon = canonicalize_entity(input);
        assert2::assert!(
            canon
                .iter()
                .map(|(kind, _)| kind.as_str())
                .collect::<Vec<_>>()
                == vec!["client-id", "user"]
        );
    }

    #[test]
    fn scram_credentials_users_returns_distinct_users() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "alice".into(),
            mechanism: SaslMechanism::ScramSha512,
            salt: vec![1, 2, 3],
            stored_key: vec![4, 5, 6],
            server_key: vec![7, 8, 9],
            iterations: 4096,
        }));
        img.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "bob".into(),
            mechanism: SaslMechanism::ScramSha512,
            salt: vec![1, 2, 3],
            stored_key: vec![4, 5, 6],
            server_key: vec![7, 8, 9],
            iterations: 4096,
        }));
        let mut users = img.scram_credentials_users();
        users.sort();
        assert2::assert!(users == vec!["alice".to_owned(), "bob".to_owned()]);
    }

    fn principal(pt: &str, name: &str) -> KafkaPrincipal {
        KafkaPrincipal {
            principal_type: pt.into(),
            name: name.into(),
        }
    }

    fn dt_record(
        token_id: &str,
        owner: KafkaPrincipal,
        expiry_timestamp_ms: i64,
        renewers: Vec<KafkaPrincipal>,
    ) -> MetadataRecord {
        MetadataRecord::V1DelegationToken(DelegationTokenRecord {
            token_id: token_id.into(),
            owner: owner.clone(),
            requester: owner,
            issue_timestamp_ms: 1_000,
            expiry_timestamp_ms,
            max_timestamp_ms: 10_000,
            renewers,
        })
    }

    #[test]
    fn apply_delegation_token_insert_and_replace() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        let alice = principal("User", "alice");

        img.apply(&dt_record("tok-1", alice.clone(), 5_000, vec![]));
        let got = img.delegation_token_by_id("tok-1").expect("token present");
        assert2::assert!(got.expiry_timestamp_ms == 5_000);
        assert2::assert!(&got.owner == &alice);

        // Same token_id, different expiry — replace, not duplicate.
        img.apply(&dt_record("tok-1", alice, 7_500, vec![]));
        let got = img.delegation_token_by_id("tok-1").expect("token present");
        assert2::assert!(got.expiry_timestamp_ms == 7_500);
        assert2::assert!(img.all_delegation_tokens().count() == 1);
    }

    #[test]
    fn apply_delete_delegation_token_removes_from_image() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        let alice = principal("User", "alice");

        img.apply(&dt_record("tok-1", alice, 5_000, vec![]));
        assert2::assert!(img.delegation_token_by_id("tok-1").is_some());

        img.apply(&MetadataRecord::V1DeleteDelegationToken(
            DeleteDelegationTokenRecord {
                token_id: "tok-1".into(),
            },
        ));
        assert2::assert!(img.delegation_token_by_id("tok-1").is_none());
        assert2::assert!(img.all_delegation_tokens().count() == 0);
    }

    #[test]
    fn delegation_token_by_hmac_recomputes_each_tokens_hmac() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&dt_record(
            "tok-a",
            principal("User", "alice"),
            5_000,
            vec![],
        ));
        img.apply(&dt_record("tok-b", principal("User", "bob"), 5_000, vec![]));
        let hmac_b = krabka_security::compute_token_hmac(b"key", "tok-b");

        // (secret key, presented hmac, token found)
        let cases: [(&[u8], &[u8], Option<&str>); 3] = [
            (b"key", &hmac_b, Some("tok-b")),
            // The same HMAC under another key names no token.
            (b"other", &hmac_b, None),
            (b"key", &[0xCC; 64], None),
        ];
        for (secret_key, hmac, expected) in cases {
            check!(
                img.delegation_token_by_hmac(secret_key, hmac)
                    .map(|t| t.token_id.as_str())
                    == expected
            );
        }
    }

    #[test]
    fn delegation_tokens_by_owner_filters_correctly() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        let alice = principal("User", "alice");
        let bob = principal("User", "bob");

        img.apply(&dt_record("a-1", alice.clone(), 5_000, vec![]));
        img.apply(&dt_record("a-2", alice.clone(), 6_000, vec![bob.clone()]));
        img.apply(&dt_record("b-1", bob.clone(), 7_000, vec![]));

        let alice_tokens = img.delegation_tokens_by_owner(&alice);
        assert2::assert!(alice_tokens.len() == 2);
        assert2::assert!(alice_tokens.iter().all(|token| token.owner == alice));

        let bob_tokens = img.delegation_tokens_by_owner(&bob);
        assert2::assert!(
            bob_tokens
                .iter()
                .map(|token| token.token_id.as_str())
                .collect::<Vec<_>>()
                == vec!["b-1"]
        );

        // visible_to: bob owns b-1 and is renewer on a-2 → 2 tokens.
        let bob_visible = img.delegation_tokens_visible_to(&bob);
        assert2::assert!(bob_visible.len() == 2);
        let mut ids: Vec<&str> = bob_visible.iter().map(|t| t.token_id.as_str()).collect();
        ids.sort_unstable();
        assert2::assert!(ids == vec!["a-2", "b-1"]);
    }

    /// Kafka's `TokenInformation.ownerOrRenewer` counts the requester too, so
    /// an administrator sees a token it created on another owner's behalf.
    #[test]
    fn delegation_tokens_are_visible_to_their_requester() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        let alice = principal("User", "alice");
        let admin = principal("User", "admin");
        img.apply(&MetadataRecord::V1DelegationToken(DelegationTokenRecord {
            requester: admin.clone(),
            ..match dt_record("a-1", alice.clone(), 5_000, vec![]) {
                MetadataRecord::V1DelegationToken(record) => record,
                _ => unreachable!("dt_record builds a token record"),
            }
        }));
        img.apply(&dt_record("a-2", alice.clone(), 5_000, vec![]));

        // (principal, visible token ids)
        let cases = [
            (&admin, vec!["a-1"]),
            (&alice, vec!["a-1", "a-2"]),
            (&principal("User", "carol"), vec![]),
        ];
        for (who, expected) in cases {
            let mut ids: Vec<&str> = img
                .delegation_tokens_visible_to(who)
                .iter()
                .map(|t| t.token_id.as_str())
                .collect();
            ids.sort_unstable();
            check!(ids == expected, "{who}");
        }
    }

    #[test]
    fn applies_voters_and_version() {
        let mut image = MetadataImage::new(Uuid::nil());
        image.apply(&MetadataRecord::V1KRaftVersion(KRaftVersionRecord {
            kraft_version: 1,
        }));
        image.apply(&MetadataRecord::V1Voters(VotersRecord {
            voters: VoterSet::from_voters([Voter {
                id: NodeId(1),
                directory_id: uuid::Uuid::nil(),
                endpoints: vec![],
                kraft_version: KRaftVersionRange::default(),
            }]),
        }));
        assert2::assert!(image.kraft_version() == 1);
        assert2::assert!(image.voters().contains(NodeId(1)));
    }

    #[test]
    fn scram_credentials_for_user_returns_pairs() {
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        img.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "alice".into(),
            mechanism: SaslMechanism::ScramSha512,
            salt: vec![1, 2, 3],
            stored_key: vec![4, 5, 6],
            server_key: vec![7, 8, 9],
            iterations: 8192,
        }));
        let pairs = img.scram_credentials_for_user("alice");
        assert2::assert!(pairs == vec![(SaslMechanism::ScramSha512, 8192)]);
        assert2::assert!(img.scram_credentials_for_user("ghost").is_empty());
    }

    #[test]
    fn finalized_metadata_version_reads_feature_map() {
        use crate::records::FeatureLevelRecord;
        let mut m = img();
        assert2::assert!(m.finalized_metadata_version() == None);
        m.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: "metadata.version".into(),
            level: 19,
        }));
        assert2::assert!(m.finalized_metadata_version() == Some(19));
    }

    #[test]
    fn min_required_metadata_version_baseline_is_min() {
        use crate::metadata_version::METADATA_VERSION_MIN;
        let m = img();
        assert2::assert!(m.min_required_metadata_version() == METADATA_VERSION_MIN);
    }

    #[test]
    fn client_metrics_config_apply_and_clear() {
        use crate::records::ClientMetricsConfigRecord;
        let mut img = MetadataImage::new(uuid::Uuid::nil());
        let mut cfgs = BTreeMap::new();
        cfgs.insert("interval.ms".to_owned(), "60000".to_owned());
        img.apply(&MetadataRecord::V1ClientMetricsConfig(
            ClientMetricsConfigRecord {
                name: "sub-a".into(),
                configs: cfgs,
            },
        ));
        assert2::assert!(
            img.client_metrics_config("sub-a")
                .and_then(|m| m.get("interval.ms"))
                .map(String::as_str)
                == Some("60000")
        );
        assert2::assert!(img.client_metrics_subscriptions().count() == 1);
        img.apply(&MetadataRecord::V1ClientMetricsConfig(
            ClientMetricsConfigRecord {
                name: "sub-a".into(),
                configs: BTreeMap::new(),
            },
        ));
        assert2::assert!(img.client_metrics_config("sub-a").is_none());
        assert2::assert!(img.client_metrics_subscriptions().count() == 0);
    }

    #[test]
    fn group_config_apply_snapshot_and_clear() {
        let mut image = MetadataImage::new(Uuid::nil());
        let record = MetadataRecord::V1GroupConfig(GroupConfigRecord {
            group_id: "streams-app".into(),
            configs: BTreeMap::from([("streams.num.standby.replicas".into(), "1".into())]),
        });
        image.apply(&record);
        assert2::assert!(
            image
                .group_config("streams-app")
                .and_then(|configs| configs.get("streams.num.standby.replicas"))
                .map(String::as_str)
                == Some("1")
        );
        assert2::assert!(image.to_records().iter().any(|item| item == &record));

        image.apply(&MetadataRecord::V1GroupConfig(GroupConfigRecord {
            group_id: "streams-app".into(),
            configs: BTreeMap::new(),
        }));
        assert2::assert!(image.group_config("streams-app").is_none());
    }

    #[test]
    fn producer_id_boundary_is_monotonic_and_survives_snapshot() {
        let mut image = MetadataImage::new(Uuid::nil());
        image.apply(&MetadataRecord::V1ProducerIds(ProducerIdsRecord {
            broker_id: NodeId(1),
            broker_epoch: 4,
            next_producer_id: 1_000,
        }));
        image.apply(&MetadataRecord::V1ProducerIds(ProducerIdsRecord {
            broker_id: NodeId(2),
            broker_epoch: 8,
            next_producer_id: 2_000,
        }));
        image.apply(&MetadataRecord::V1ProducerIds(ProducerIdsRecord {
            broker_id: NodeId(1),
            broker_epoch: 4,
            next_producer_id: 1_000,
        }));
        assert2::assert!(image.next_producer_id() == 2_000);

        let restored = MetadataImage::from_records(Uuid::nil(), &image.to_records());
        assert2::assert!(restored.next_producer_id() == 2_000);
    }

    /// KIP-1276: an ACL whose host is a CIDR block raises the floor to
    /// `4.4-IV1`, and a downgrade below it removes exactly those ACLs.
    #[test]
    fn cidr_acl_hosts_follow_the_kip_1276_gate() {
        use crate::{
            AclEntryFilter,
            acl::{AclOperation, PermissionType},
            metadata_version::{CIDR_ACL_MIN_LEVEL, METADATA_VERSION_MIN},
        };

        let acl = |host: &str| AclEntry {
            resource_type: ResourceType::Topic,
            resource_name: "orders".into(),
            pattern_type: PatternType::Literal,
            principal: "User:alice".into(),
            host: host.into(),
            operation: AclOperation::Read,
            permission_type: PermissionType::Allow,
        };
        let delete = |host: &str| {
            let entry = acl(host);
            MetadataRecord::V1DeleteAccessControlEntry(AclEntryFilter {
                resource_type: Some(entry.resource_type),
                resource_name: Some(entry.resource_name),
                pattern_type: Some(entry.pattern_type),
                principal: Some(entry.principal),
                host: Some(entry.host),
                operation: Some(entry.operation),
                permission_type: Some(entry.permission_type),
            })
        };
        for (case, hosts, floor, cidr_hosts) in [
            ("no ACLs", vec![], METADATA_VERSION_MIN, vec![]),
            (
                "wildcard and address",
                vec!["*", "10.0.0.1"],
                METADATA_VERSION_MIN,
                vec![],
            ),
            (
                "IPv4 CIDR",
                vec!["*", "10.0.0.0/8"],
                CIDR_ACL_MIN_LEVEL,
                vec!["10.0.0.0/8"],
            ),
            (
                "IPv6 CIDR",
                vec!["2001:db8::/32"],
                CIDR_ACL_MIN_LEVEL,
                vec!["2001:db8::/32"],
            ),
        ] {
            let mut image = img();
            for host in &hosts {
                image.apply(&MetadataRecord::V1AccessControlEntry(acl(host)));
            }
            let want_cleanup: Vec<_> = cidr_hosts.into_iter().map(delete).collect();
            assert2::check!(
                (
                    image.min_required_metadata_version(),
                    image.metadata_version_downgrade_records(CIDR_ACL_MIN_LEVEL),
                    image.metadata_version_downgrade_records(CIDR_ACL_MIN_LEVEL - 1),
                ) == (floor, vec![], want_cleanup.clone()),
                "case {case}"
            );
            for record in &want_cleanup {
                image.apply(record);
            }
            assert2::check!(
                image.min_required_metadata_version() == METADATA_VERSION_MIN,
                "case {case} after cleanup"
            );
        }
    }

    #[test]
    fn min_required_metadata_version_rises_with_scram_and_tokens() {
        use krabka_security::{KafkaPrincipal, SaslMechanism};

        use crate::metadata_version::{DELEGATION_TOKEN_MIN_LEVEL, SCRAM_MIN_LEVEL};
        let mut m = img();
        m.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "alice".into(),
            mechanism: SaslMechanism::ScramSha512,
            salt: vec![1; 16],
            stored_key: vec![2; 64],
            server_key: vec![3; 64],
            iterations: 4096,
        }));
        assert2::assert!(m.min_required_metadata_version() == SCRAM_MIN_LEVEL);
        m.apply(&MetadataRecord::V1DelegationToken(DelegationTokenRecord {
            token_id: "t1".into(),
            owner: KafkaPrincipal {
                principal_type: "User".into(),
                name: "alice".into(),
            },
            requester: KafkaPrincipal {
                principal_type: "User".into(),
                name: "alice".into(),
            },
            issue_timestamp_ms: 1,
            expiry_timestamp_ms: 5,
            max_timestamp_ms: 10,
            renewers: vec![],
        }));
        assert2::assert!(m.min_required_metadata_version() == DELEGATION_TOKEN_MIN_LEVEL);
    }

    #[test]
    fn downgrade_projection_removes_scram_and_tokens_below_their_record_gates() {
        use krabka_security::{KafkaPrincipal, SaslMechanism};

        use crate::{
            metadata_version::{METADATA_VERSION_MIN, SCRAM_MIN_LEVEL},
            records::{
                DelegationTokenRecord, DeleteDelegationTokenRecord, DeleteScramCredentialRecord,
                ScramCredentialRecord,
            },
        };

        let mut image = img();
        image.apply(&MetadataRecord::V1ScramCredential(ScramCredentialRecord {
            user: "alice".into(),
            mechanism: SaslMechanism::ScramSha512,
            salt: vec![1; 16],
            stored_key: vec![2; 64],
            server_key: vec![3; 64],
            iterations: 4096,
        }));
        image.apply(&MetadataRecord::V1DelegationToken(DelegationTokenRecord {
            token_id: "token-1".into(),
            owner: KafkaPrincipal {
                principal_type: "User".into(),
                name: "alice".into(),
            },
            requester: KafkaPrincipal {
                principal_type: "User".into(),
                name: "alice".into(),
            },
            issue_timestamp_ms: 1,
            expiry_timestamp_ms: 5,
            max_timestamp_ms: 10,
            renewers: vec![],
        }));

        // At the level that introduced each feature the state is still
        // supported, so nothing is cleaned up. Read as `target <= MIN_LEVEL`,
        // a downgrade *to* the introducing level deletes the very credentials
        // and tokens that level exists to support.
        assert2::assert!(
            image
                .metadata_version_downgrade_records(SCRAM_MIN_LEVEL)
                .iter()
                .all(|r| !matches!(r, MetadataRecord::V1DeleteScramCredential(_)))
        );
        assert2::assert!(
            image
                .metadata_version_downgrade_records(metadata_version::DELEGATION_TOKEN_MIN_LEVEL)
                .iter()
                .all(|r| !matches!(r, MetadataRecord::V1DeleteDelegationToken(_)))
        );

        let cleanup = image.metadata_version_downgrade_records(SCRAM_MIN_LEVEL - 1);
        assert2::assert!(
            cleanup
                == vec![
                    MetadataRecord::V1DeleteScramCredential(DeleteScramCredentialRecord {
                        user: "alice".into(),
                        mechanism: SaslMechanism::ScramSha512,
                    }),
                    MetadataRecord::V1DeleteDelegationToken(DeleteDelegationTokenRecord {
                        token_id: "token-1".into(),
                    }),
                ]
        );

        for record in &cleanup {
            image.apply(record);
        }
        assert2::assert!(image.scram_credentials_for_user("alice").is_empty());
        assert2::assert!(image.all_delegation_tokens().next().is_none());
        assert2::assert!(image.min_required_metadata_version() == METADATA_VERSION_MIN);
    }

    #[test]
    fn downgrade_projection_clears_directory_fields_below_kip_858() {
        use crate::metadata_version::DIRECTORY_ASSIGNMENT_MIN_LEVEL;

        let directory = Uuid::from_u128(0xD1);
        let mut image = img();
        let broker = BrokerRegistrationRecord {
            node_id: NodeId(1),
            broker_epoch: 7,
            incarnation_id: Uuid::from_u128(1),
            host: "broker-1".into(),
            port: 9092,
            rack: None,
            endpoints: vec![],
            log_dirs: vec![directory],
            fenced: false,
            in_controlled_shutdown: false,
            cordoned_log_dirs: None,
            features: BTreeMap::new(),
        };
        image.apply(&MetadataRecord::V1BrokerRegistration(broker.clone()));
        image.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "orders".into(),
            partition: 0,
            leader: NodeId(1),
            replicas: vec![NodeId(1), NodeId(2)],
            isr: vec![NodeId(1), NodeId(2)],
            directories: vec![directory, MIGRATING],
            ..Default::default()
        }));

        assert2::assert!(
            image
                .metadata_version_downgrade_records(DIRECTORY_ASSIGNMENT_MIN_LEVEL)
                .is_empty()
        );
        let cleanup = image.metadata_version_downgrade_records(DIRECTORY_ASSIGNMENT_MIN_LEVEL - 1);
        let mut projected_broker = broker;
        projected_broker.log_dirs.clear();
        assert2::assert!(
            cleanup
                == vec![
                    MetadataRecord::V1BrokerRegistration(projected_broker),
                    MetadataRecord::V1PartitionDirAssignment(PartitionDirAssignmentRecord {
                        topic: "orders".into(),
                        partition: 0,
                        replica: NodeId(1),
                        directory: MIGRATING,
                    },),
                ]
        );

        for record in &cleanup {
            image.apply(record);
        }
        assert2::assert!(
            image.partition("orders", 0).expect("partition").directories == vec![MIGRATING; 2]
        );
        assert2::assert!(image.broker(NodeId(1)).expect("broker").log_dirs.is_empty());
    }

    /// KIP-1066 cordoned log dirs exist from `4.3-IV0`: a downgrade below it
    /// re-registers the broker without them, and below `3.7-IV2` without its
    /// log dirs as well.
    #[test]
    fn downgrade_projection_clears_cordoned_log_dirs_below_kip_1066() {
        use crate::metadata_version::{
            CORDONED_LOG_DIRS_MIN_LEVEL, DIRECTORY_ASSIGNMENT_MIN_LEVEL,
        };

        let directory = Uuid::from_u128(0xD1);
        let broker = BrokerRegistrationRecord {
            node_id: NodeId(1),
            broker_epoch: 7,
            incarnation_id: Uuid::from_u128(1),
            host: "broker-1".into(),
            port: 9092,
            rack: None,
            endpoints: vec![],
            log_dirs: vec![directory],
            fenced: false,
            in_controlled_shutdown: false,
            cordoned_log_dirs: Some(vec![directory]),
            features: BTreeMap::new(),
        };
        let mut image = img();
        image.apply(&MetadataRecord::V1BrokerRegistration(broker.clone()));

        let uncordoned = BrokerRegistrationRecord {
            cordoned_log_dirs: None,
            ..broker
        };
        let without_dirs = BrokerRegistrationRecord {
            log_dirs: vec![],
            ..uncordoned.clone()
        };
        for (target, want) in [
            (CORDONED_LOG_DIRS_MIN_LEVEL, vec![]),
            (
                CORDONED_LOG_DIRS_MIN_LEVEL - 1,
                vec![MetadataRecord::V1BrokerRegistration(uncordoned)],
            ),
            (
                DIRECTORY_ASSIGNMENT_MIN_LEVEL - 1,
                vec![MetadataRecord::V1BrokerRegistration(without_dirs)],
            ),
        ] {
            assert2::check!(
                image.metadata_version_downgrade_records(target) == want,
                "target {target}"
            );
        }
    }

    #[test]
    fn topic_by_id_resolves_and_drops_on_delete() {
        use crate::records::{MetadataRecord, TopicRecord};
        let mut img = MetadataImage::new(Uuid::nil());
        let id = Uuid::from_u128(0x1234_5678_9abc_def0_1122_3344_5566_7788);
        img.apply(&MetadataRecord::V1Topic(TopicRecord {
            name: "orders".into(),
            topic_id: id,
            partitions: 1,
            replication_factor: 1,
        }));

        // Resolves by id to the same record `topic(name)` returns.
        assert2::assert!(img.topic_by_id(&id).map(|t| t.name.as_str()) == Some("orders"));
        assert2::assert!(img.topic_name_by_id(&id) == Some("orders"));

        // After delete the id no longer resolves and the name index is gone.
        img.apply(&MetadataRecord::V1DeleteTopic(DeleteTopicRecord {
            name: "orders".into(),
        }));
        assert2::assert!(img.topic_by_id(&id).is_none());
        assert2::assert!(img.topic_name_by_id(&id).is_none());
    }

    #[test]
    fn broker_epoch_reads_back_registered_epoch() {
        let mut image = MetadataImage::new(Uuid::nil());
        image.apply(&MetadataRecord::V1BrokerRegistration(
            BrokerRegistrationRecord {
                node_id: NodeId(5),
                broker_epoch: 99,
                incarnation_id: Uuid::nil(),
                host: "h".into(),
                port: 9092,
                rack: None,
                log_dirs: vec![],
                fenced: false,
                in_controlled_shutdown: false,
                cordoned_log_dirs: None,
                endpoints: vec![],
                features: BTreeMap::new(),
            },
        ));
        assert2::assert!(image.broker_epoch(NodeId(5)) == Some(99));
        assert2::assert!(image.broker_epoch(NodeId(404)) == None);
    }

    // --- mutation-coverage tests --------------------------------------------

    #[test]
    fn fresh_image_accessors() {
        let id = Uuid::from_u128(0xABCD);
        let image = MetadataImage::new(id);
        assert2::assert!(image.cluster_id() == id);
        assert2::assert!(image.kraft_version() == 0);
    }

    #[test]
    fn broker_lookup_returns_registered_record() {
        let mut image = MetadataImage::new(Uuid::nil());
        image.apply(&MetadataRecord::V1BrokerRegistration(
            BrokerRegistrationRecord {
                node_id: NodeId(7),
                broker_epoch: 1,
                incarnation_id: Uuid::nil(),
                host: "h".into(),
                port: 9092,
                rack: None,
                log_dirs: vec![],
                fenced: false,
                in_controlled_shutdown: false,
                cordoned_log_dirs: None,
                endpoints: vec![],
                features: BTreeMap::new(),
            },
        ));
        let b = image
            .broker(NodeId(7))
            .expect("registered broker resolvable");
        assert2::assert!((b.node_id, image.broker(NodeId(404)).is_none()) == (NodeId(7), true));
    }

    #[test]
    fn reapplying_topic_with_new_id_drops_stale_id_index() {
        let mut image = MetadataImage::new(Uuid::nil());
        let old_id = Uuid::from_u128(1);
        let new_id = Uuid::from_u128(2);
        let topic = |id| {
            MetadataRecord::V1Topic(TopicRecord {
                name: "t".into(),
                topic_id: id,
                partitions: 1,
                replication_factor: 1,
            })
        };
        image.apply(&topic(old_id));
        assert2::assert!(image.topic_by_id(&old_id).is_some());
        // Re-apply the same NAME with a different id: the stale id index entry
        // must be dropped (the `prev.topic_id != t.topic_id` guard).
        image.apply(&topic(new_id));
        assert2::assert!(image.topic_by_id(&new_id).is_some());
        assert2::assert!(image.topic_by_id(&old_id).is_none());
    }

    #[test]
    fn broker_throttle_rate_allows_zero() {
        // A configured rate of 0 is a real (zero) value, distinct from absent
        // or -1 (disabled). The cap check is `v < 0` (not `<=`), so 0 -> Some(0).
        let mut image = MetadataImage::new(Uuid::nil());
        image.apply(&MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
            node_id: NodeId(3),
            config_name: "leader.replication.throttled.rate".into(),
            config_value: Some("0".into()),
        }));
        assert2::assert!(
            image.broker_throttle_rate(NodeId(3), ThrottleKind::Leader)
                == Some(<ByteRate as ByteRateExt>::ZERO)
        );
    }

    #[test]
    fn dir_assignment_resizes_directories_to_replica_count() {
        // KIP-858: a dir-assignment delta for a partition whose `directories`
        // vec is shorter than `replicas` must grow it (to `replicas.len()`,
        // padded with `MIGRATING`, which is how Kafka reads an empty list)
        // before writing the reporting replica's slot. Without the resize the
        // slot write would index out of bounds.
        let mut image = MetadataImage::new(Uuid::nil());
        image.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "t".into(),
            partition: 0,
            leader: NodeId(10),
            replicas: vec![NodeId(10), NodeId(20)],
            isr: vec![NodeId(10), NodeId(20)],
            directories: vec![], // not yet reported
            ..Default::default()
        }));
        let dir = Uuid::from_u128(0xD15);
        image.apply(&MetadataRecord::V1PartitionDirAssignment(
            PartitionDirAssignmentRecord {
                topic: "t".into(),
                partition: 0,
                replica: NodeId(10),
                directory: dir,
            },
        ));
        let pr = image.partition("t", 0).expect("partition present");
        // Replica 10 is slot 0; replica 20 stays `MIGRATING`.
        assert2::assert!(pr.directories == vec![dir, MIGRATING]);
    }

    #[test]
    fn dir_assignment_updates_existing_full_length_directories() {
        let mut image = MetadataImage::new(Uuid::nil());
        let old = Uuid::from_u128(0xAAA);
        image.apply(&MetadataRecord::V1Partition(PartitionRecord {
            topic: "t".into(),
            partition: 0,
            leader: NodeId(10),
            replicas: vec![NodeId(10), NodeId(20)],
            isr: vec![NodeId(10), NodeId(20)],
            directories: vec![old, Uuid::nil()],
            ..Default::default()
        }));
        let dir = Uuid::from_u128(0xD15);
        image.apply(&MetadataRecord::V1PartitionDirAssignment(
            PartitionDirAssignmentRecord {
                topic: "t".into(),
                partition: 0,
                replica: NodeId(20),
                directory: dir,
            },
        ));

        let pr = image.partition("t", 0).expect("partition present");
        assert2::assert!(pr.directories == vec![old, dir]);
    }
    // ---------- KFC-9 write-freeze registry ----------

    fn freeze(scope: &str, pattern_type: PatternType) -> TopicFreezeRecord {
        TopicFreezeRecord {
            scope: scope.into(),
            pattern_type,
            frozen: true,
            reason: "DR cutover".into(),
            set_by: "User:alice".into(),
            set_at_ms: 1_700_000_000_000,
            proposal_id: Uuid::nil(),
            key_id: "alice-yubi".into(),
            signature: vec![0xAB; 64],
        }
    }

    #[test]
    fn an_unfrozen_cluster_resolves_no_freeze() {
        check!(img().topic_freeze("orders").is_none());
        check!(img().topic_freezes().count() == 0);
    }

    #[test]
    fn a_literal_freeze_beats_a_prefix_and_the_longest_prefix_wins() {
        let mut image = img();
        for (scope, pattern_type) in [
            ("tenant-a.", PatternType::Prefixed),
            ("tenant-a.orders.", PatternType::Prefixed),
            ("tenant-a.orders.eu", PatternType::Literal),
        ] {
            image.apply(&MetadataRecord::V1TopicFreeze(freeze(scope, pattern_type)));
        }

        for (label, topic, want) in [
            (
                "a literal entry beats every prefix entry",
                "tenant-a.orders.eu",
                Some("tenant-a.orders.eu"),
            ),
            (
                "the longest matching prefix wins",
                "tenant-a.orders.us",
                Some("tenant-a.orders."),
            ),
            (
                "a shorter prefix still matches",
                "tenant-a.billing",
                Some("tenant-a."),
            ),
            ("no scope covers the topic", "tenant-b.orders", None),
            ("a topic that sorts below every scope", "aaa", None),
        ] {
            check!(
                image.topic_freeze(topic).map(|f| f.scope.as_str()) == want,
                "{label}"
            );
        }
    }

    #[test]
    fn an_internal_topic_is_never_frozen() {
        for (label, scope, pattern_type, covered) in [
            (
                "an empty prefix scope",
                "",
                PatternType::Prefixed,
                Some("orders"),
            ),
            (
                "a single-underscore prefix scope",
                "_",
                PatternType::Prefixed,
                Some("_metrics"),
            ),
            (
                "a literal internal name",
                "__consumer_offsets",
                PatternType::Literal,
                None,
            ),
        ] {
            let mut image = img();
            image.apply(&MetadataRecord::V1TopicFreeze(freeze(scope, pattern_type)));

            check!(
                image.topic_freeze("__consumer_offsets").is_none(),
                "{label}"
            );
            if let Some(topic) = covered {
                // The registry is live, so the exemption comes from the name
                // and not from an empty registry.
                check!(image.topic_freeze(topic).is_some(), "{label}");
            }
        }
    }

    #[test]
    fn a_thaw_removes_the_registry_entry() {
        for (label, pattern_type, topic) in [
            ("a literal scope", PatternType::Literal, "orders"),
            ("a prefixed scope", PatternType::Prefixed, "orders-eu"),
        ] {
            let mut image = img();
            let frozen = freeze("orders", pattern_type);
            image.apply(&MetadataRecord::V1TopicFreeze(frozen.clone()));
            check!(image.topic_freeze(topic).is_some(), "{label}");

            image.apply(&MetadataRecord::V1TopicFreeze(TopicFreezeRecord {
                frozen: false,
                proposal_id: Uuid::from_u128(0xB1),
                ..frozen
            }));
            check!(image.topic_freeze(topic).is_none(), "{label}");
            check!(image.topic_freezes().count() == 0, "{label}");
        }
    }

    #[test]
    fn a_later_freeze_replaces_the_entry_at_the_same_scope() {
        let mut image = img();
        image.apply(&MetadataRecord::V1TopicFreeze(freeze(
            "orders",
            PatternType::Literal,
        )));
        let replacement = TopicFreezeRecord {
            reason: "tenant offboarding".into(),
            set_by: "User:bob".into(),
            set_at_ms: 1_700_000_060_000,
            ..freeze("orders", PatternType::Literal)
        };
        image.apply(&MetadataRecord::V1TopicFreeze(replacement.clone()));

        check!(image.topic_freeze("orders") == Some(&replacement));
        check!(image.topic_freezes().count() == 1);
    }

    // ---------- KFC-9 break-glass proposals ----------

    fn approval(principal: &str) -> BreakGlassApproval {
        BreakGlassApproval {
            principal: principal.into(),
            approved_at_ms: 1_700_000_060_000,
            key_id: format!("{principal}-yubi"),
            signature: vec![0xCD; 64],
        }
    }

    fn proposal(id: Uuid) -> BreakGlassProposalRecord {
        BreakGlassProposalRecord {
            proposal_id: id,
            action: BreakGlassAction::ThawTopicFreeze,
            target: "literal:orders".into(),
            proposer: "User:alice".into(),
            reason: "incident 4711 closed".into(),
            created_at_ms: 1_700_000_000_000,
            expires_at_ms: 1_700_000_600_000,
            approvals: Vec::new(),
            consumed_at_ms: 0,
            withdrawn: false,
        }
    }

    #[test]
    fn a_break_glass_proposal_upserts_and_deletes() {
        let id = Uuid::from_u128(0xB1);
        let mut image = img();
        let record = proposal(id);
        image.apply(&MetadataRecord::V1BreakGlassProposal(record.clone()));
        check!(image.break_glass_proposal(id) == Some(&record));
        check!(image.break_glass_proposals().count() == 1);

        let approved = BreakGlassProposalRecord {
            approvals: vec![approval("User:bob")],
            ..record
        };
        image.apply(&MetadataRecord::V1BreakGlassProposal(approved.clone()));
        check!(image.break_glass_proposal(id) == Some(&approved));

        image.apply(&MetadataRecord::V1DeleteBreakGlassProposal(id));
        check!(image.break_glass_proposal(id).is_none());
        check!(image.break_glass_proposals().count() == 0);
    }

    #[test]
    fn to_records_preserves_both_freeze_registries_and_the_proposals() {
        let cid = Uuid::new_v4();
        let mut image = MetadataImage::new(cid);
        let literal = freeze("orders", PatternType::Literal);
        let prefixed = freeze("tenant-a.", PatternType::Prefixed);
        let pending = BreakGlassProposalRecord {
            approvals: vec![approval("User:bob")],
            ..proposal(Uuid::from_u128(0xB1))
        };
        image.apply(&MetadataRecord::V1TopicFreeze(literal.clone()));
        image.apply(&MetadataRecord::V1TopicFreeze(prefixed.clone()));
        image.apply(&MetadataRecord::V1BreakGlassProposal(pending.clone()));

        let rebuilt = MetadataImage::from_records(cid, &image.to_records());

        check!(rebuilt == image);
        check!(rebuilt.topic_freeze("orders") == Some(&literal));
        check!(rebuilt.topic_freeze("tenant-a.billing") == Some(&prefixed));
        check!(rebuilt.break_glass_proposal(pending.proposal_id) == Some(&pending));
    }

    // ---------- KFC-9 validate ----------

    #[test]
    fn validate_rejects_only_an_empty_freeze_scope() {
        let image = img();
        for (label, scope, pattern_type, want) in [
            (
                "a literal scope naming a topic that does not exist",
                "orders",
                PatternType::Literal,
                Ok(()),
            ),
            (
                "a prefix scope, which names no topic by construction",
                "tenant-a.",
                PatternType::Prefixed,
                Ok(()),
            ),
            (
                "an empty literal scope",
                "",
                PatternType::Literal,
                Err(MetadataError::EmptyFreezeScope),
            ),
            (
                "an empty prefix scope",
                "",
                PatternType::Prefixed,
                Err(MetadataError::EmptyFreezeScope),
            ),
        ] {
            let rec = MetadataRecord::V1TopicFreeze(freeze(scope, pattern_type));
            check!(image.validate(&rec) == want, "{label}");
        }
    }

    #[test]
    fn validate_accepts_a_break_glass_proposal_that_is_not_yet_stored() {
        let image = img();
        let rec = MetadataRecord::V1BreakGlassProposal(proposal(Uuid::from_u128(0xB1)));
        check!(image.validate(&rec) == Ok(()));
    }

    #[test]
    fn validate_accepts_only_an_appended_approval() {
        let id = Uuid::from_u128(0xB1);
        let (alice, bob, carol) = (
            approval("User:alice"),
            approval("User:bob"),
            approval("User:carol"),
        );
        let mut image = img();
        let stored = BreakGlassProposalRecord {
            approvals: vec![alice.clone(), bob.clone()],
            ..proposal(id)
        };
        image.apply(&MetadataRecord::V1BreakGlassProposal(stored.clone()));

        for (label, approvals, want) in [
            (
                "the stored list unchanged, which a consumption writes back",
                vec![alice.clone(), bob.clone()],
                Ok(()),
            ),
            (
                "one appended approval",
                vec![alice.clone(), bob.clone(), carol.clone()],
                Ok(()),
            ),
            (
                "a reordered list",
                vec![bob, alice.clone()],
                Err(MetadataError::BreakGlassApprovalsNotAnExtension(id)),
            ),
            (
                "a truncated list",
                vec![alice.clone()],
                Err(MetadataError::BreakGlassApprovalsNotAnExtension(id)),
            ),
            (
                "an empty list",
                Vec::new(),
                Err(MetadataError::BreakGlassApprovalsNotAnExtension(id)),
            ),
            (
                "a replaced approval",
                vec![alice, carol],
                Err(MetadataError::BreakGlassApprovalsNotAnExtension(id)),
            ),
        ] {
            let candidate = MetadataRecord::V1BreakGlassProposal(BreakGlassProposalRecord {
                approvals,
                ..stored.clone()
            });
            check!(image.validate(&candidate) == want, "{label}");
        }
    }

    /// Two approvers read the same image and each submits the list it read
    /// plus its own entry. Without the extension rule the second write drops
    /// the first approval, and the proposal never reaches two principals.
    #[test]
    fn a_concurrent_approval_cannot_drop_the_stored_one() {
        let id = Uuid::from_u128(0xB1);
        let mut image = img();
        image.apply(&MetadataRecord::V1BreakGlassProposal(proposal(id)));

        let by_bob = BreakGlassProposalRecord {
            approvals: vec![approval("User:bob")],
            ..proposal(id)
        };
        let by_carol = BreakGlassProposalRecord {
            approvals: vec![approval("User:carol")],
            ..proposal(id)
        };
        check!(image.validate(&MetadataRecord::V1BreakGlassProposal(by_bob.clone())) == Ok(()));
        image.apply(&MetadataRecord::V1BreakGlassProposal(by_bob));

        check!(
            image.validate(&MetadataRecord::V1BreakGlassProposal(by_carol))
                == Err(MetadataError::BreakGlassApprovalsNotAnExtension(id))
        );

        let both = BreakGlassProposalRecord {
            approvals: vec![approval("User:bob"), approval("User:carol")],
            ..proposal(id)
        };
        check!(image.validate(&MetadataRecord::V1BreakGlassProposal(both)) == Ok(()));
    }

    #[test]
    fn validate_rejects_every_change_to_a_settled_proposal() {
        let id = Uuid::from_u128(0xB1);
        for (label, settled, want) in [
            (
                "a consumed proposal",
                BreakGlassProposalRecord {
                    consumed_at_ms: 1_700_000_120_000,
                    ..proposal(id)
                },
                MetadataError::BreakGlassProposalConsumed(id),
            ),
            (
                "a withdrawn proposal",
                BreakGlassProposalRecord {
                    withdrawn: true,
                    ..proposal(id)
                },
                MetadataError::BreakGlassProposalWithdrawn(id),
            ),
        ] {
            let mut image = img();
            image.apply(&MetadataRecord::V1BreakGlassProposal(settled.clone()));

            for (change_label, candidate) in [
                (
                    "an appended approval",
                    BreakGlassProposalRecord {
                        approvals: vec![approval("User:bob")],
                        ..settled.clone()
                    },
                ),
                (
                    "a second consumption",
                    BreakGlassProposalRecord {
                        consumed_at_ms: 1_700_000_180_000,
                        ..settled.clone()
                    },
                ),
                ("the record unchanged", settled.clone()),
            ] {
                let rec = MetadataRecord::V1BreakGlassProposal(candidate);
                check!(
                    image.validate(&rec) == Err(want.clone()),
                    "{label}, {change_label}"
                );
            }
        }
    }

    #[test]
    fn validate_accepts_a_break_glass_tombstone_for_any_id() {
        let image = img();
        let rec = MetadataRecord::V1DeleteBreakGlassProposal(Uuid::from_u128(0xB1));
        check!(image.validate(&rec) == Ok(()));
    }
}