rialo-feature-management-program-interface 0.11.0

// Copyright (c) Subzero Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

//! Feature state management
//!
//! On-chain state for the feature management program. Activation is
//! **presence-based**: a feature is active iff its name is in `entries`. No
//! per-entry payload, no clock. See `runtime/execution/NORTHSTAR-protocol-upgrades.md`.

extern crate alloc;

#[cfg(test)]
use alloc::string::ToString;
use alloc::{
    collections::{BTreeMap, BTreeSet},
    string::String,
    vec::Vec,
};

use borsh::{BorshDeserialize, BorshSerialize};
use rialo_s_pubkey::Pubkey;

use crate::error::FeatureManagementError;

/// A deferred feature-activation request, recorded by `ScheduleEnable` and
/// keyed in [`FeaturesState::pending`] by its `request_id`.
///
/// The actual firing is handled out-of-band: `ScheduleEnable` registers a
/// one-shot subscription whose `timestamp_range` predicate fires at
/// `fire_at_ms` and invokes `FireScheduledEnable { request_id }`, which reads
/// `names` from this record, activates them, and drains the entry. This record
/// exists so the pending set is enumerable (tooling / `Cancel`) until it fires
/// or is cancelled.
#[derive(Clone, Debug, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
pub struct ScheduledRequest {
    /// Feature names to enable when the schedule fires.
    pub names: Vec<String>,
    /// Wall-clock time (ms since the Unix epoch) at which the features
    /// activate.
    pub fire_at_ms: u64,
}

/// The program's global state.
///
/// Stored under the `STORAGE_ACCOUNT_SEED` PDA owned by the feature
/// management program. Carries the authority pubkey, an optional pending
/// authority for the two-step transfer handshake, and the set of feature
/// names that are currently active.
#[derive(Clone, Debug, PartialEq, Eq, BorshSerialize, BorshDeserialize)]
pub struct FeaturesState {
    /// Authority pubkey.
    authority: Pubkey,
    /// Pubkey of the proposed next authority while a two-step transfer is
    /// in flight. `None` while no transfer is pending.
    ///
    /// Set by `ProposeAuthorityTransfer`, cleared by `CancelAuthorityTransfer`,
    /// and consumed by `AcceptAuthorityTransfer` (which promotes it to
    /// `authority`). The single-step `UpdateAuthority` path sets `authority`
    /// directly and leaves this slot untouched.
    pending_authority: Option<Pubkey>,
    /// Active feature names. Presence is the activation; there is no
    /// per-entry payload.
    ///
    /// `pub(crate)` so external callers cannot bypass `enable`'s validation
    /// by writing directly. Read access through `Self::entries()`; tests
    /// construct fixtures via `Self::insert_for_test()`.
    pub(crate) entries: BTreeSet<String>,
    /// Deferred activation requests, keyed by `request_id`. Populated by
    /// `ScheduleEnable` and drained by `Cancel` or by the one-shot
    /// subscription firing — which invokes `FireScheduledEnable`, whose
    /// `fire_scheduled` removes the entry as it activates.
    ///
    /// **Stale-entry caveat.** An entry is drained only by a *successful* fire
    /// or by `Cancel`. A one-shot is removed from the matcher when it matches,
    /// and its `Destroy` is the last instruction of the fired transaction — so
    /// if `FireScheduledEnable` fails (compute budget, `MAX_FEATURE_COUNT` at
    /// fire time, …) the transaction reverts, the entry is not drained, and the
    /// subscription does not retry. The slot is reclaimable by `Cancel` (the
    /// subscription account still exists), except after an authority transfer
    /// ([[SUB-2605]]). Automatic reaping / retry (reification) is tracked as
    /// SUB-2608; until then a wedged `pending` is cleared by the authority via
    /// `Cancel`.
    ///
    /// `pub(crate)` for the same reason as `entries` — mutated only through
    /// `schedule` / `cancel` / `fire_scheduled`.
    pub(crate) pending: BTreeMap<u64, ScheduledRequest>,
}

#[cfg(test)]
pub const DETERMINISTIC_TEST_KEYPAIR: &str =
    "57Vqb7tHij5NhQnTgrgYXA19pC8ZVHQoCHpapSiQ8LJaeUvTcSBzKoB6CazhR6VtxmyVAbWnoeDSzD1Vm672NaKp";

impl FeaturesState {
    /// Create a new state with the given authority, no pending transfer, and
    /// an empty entry set.
    ///
    /// Genesis intentionally pre-seeds no features — features active from
    /// genesis are part of the implementation, not feature flags.
    pub fn new(authority: Pubkey) -> Self {
        Self {
            authority,
            pending_authority: None,
            entries: BTreeSet::new(),
            pending: BTreeMap::new(),
        }
    }

    #[cfg(test)]
    pub fn new_for_test() -> Self {
        use rialo_s_keypair::Keypair;
        use rialo_s_signer::Signer;

        let pubkey: Pubkey = Keypair::from_base58_string(DETERMINISTIC_TEST_KEYPAIR)
            .try_pubkey()
            .expect("Failed to get pubkey from deterministic test keypair");
        Self::new(pubkey)
    }

    pub fn get_authority(&self) -> &Pubkey {
        &self.authority
    }

    pub fn set_authority(&mut self, new_authority: Pubkey) {
        self.authority = new_authority;
    }

    pub fn pending_authority(&self) -> Option<&Pubkey> {
        self.pending_authority.as_ref()
    }

    pub fn set_pending_authority(&mut self, pending: Option<Pubkey>) {
        self.pending_authority = pending;
    }

    /// Propose a two-step authority transfer.
    ///
    /// Sets `pending_authority` to `Some(new_authority)`. Returns
    /// `PendingTransferExists` if a previous proposal is still
    /// outstanding — callers must `cancel_transfer` first. Returns
    /// `InvalidTransferTarget` if `new_authority` equals the current
    /// authority (degenerate target; distinct from a signer mismatch,
    /// which is the processor's `Unauthorized`).
    ///
    /// Caller is responsible for verifying the current-authority signature
    /// before invoking this.
    pub fn propose_transfer(
        &mut self,
        new_authority: Pubkey,
    ) -> Result<(), FeatureManagementError> {
        if new_authority == self.authority {
            return Err(FeatureManagementError::InvalidTransferTarget);
        }
        if self.pending_authority.is_some() {
            return Err(FeatureManagementError::PendingTransferExists);
        }
        self.pending_authority = Some(new_authority);
        Ok(())
    }

    /// Commit a previously-proposed authority transfer.
    ///
    /// On success the authority moves to the pending value and
    /// `pending_authority` clears. Returns `NoPendingTransfer` if nothing
    /// is pending.
    ///
    /// **Contract:** the caller MUST have verified that the transaction
    /// signer equals `pending_authority()` before invoking this. This
    /// method does not re-check; the processor proves signer == pending
    /// via `verify_authority(&pending, ...)` and then commits here. Code
    /// outside the processor that calls this without the signer check
    /// would let any keypair commit the pending authority.
    pub fn accept_transfer(&mut self) -> Result<(), FeatureManagementError> {
        let Some(pending) = self.pending_authority else {
            return Err(FeatureManagementError::NoPendingTransfer);
        };
        self.authority = pending;
        self.pending_authority = None;
        Ok(())
    }

    /// Cancel a previously-proposed authority transfer.
    ///
    /// Clears `pending_authority`. Returns `NoPendingTransfer` if nothing
    /// is pending. Caller is responsible for verifying the
    /// current-authority signature before invoking this.
    pub fn cancel_transfer(&mut self) -> Result<(), FeatureManagementError> {
        if self.pending_authority.is_none() {
            return Err(FeatureManagementError::NoPendingTransfer);
        }
        self.pending_authority = None;
        Ok(())
    }

    pub fn serialize(&self) -> Result<Vec<u8>, borsh::io::Error> {
        borsh::to_vec(self)
    }

    pub fn deserialize(data: &[u8]) -> Result<Self, borsh::io::Error> {
        borsh::from_slice(data)
    }

    /// Whether `feature_name` is active.
    ///
    /// Activation is membership: a feature is active iff its name is in
    /// `entries`. No clock consult.
    pub fn is_active(&self, feature_name: &str) -> bool {
        self.entries.contains(feature_name)
    }

    /// Read-only view of the active entry set. Use `enable` to mutate so the
    /// `MAX_FEATURE_COUNT` cap stays enforced.
    pub fn entries(&self) -> &BTreeSet<String> {
        &self.entries
    }

    /// Test-only direct insert that skips `enable`'s validation. Use to
    /// seed fixtures that need specific names. Never call from production
    /// paths.
    ///
    /// Not `cfg(test)`-gated because cross-crate test crates (e.g.
    /// `svm-execution`'s bank tests) need to construct `FeaturesState`
    /// fixtures, and `cfg(test)` from the declaring crate is invisible to
    /// dependent test crates. Documented as fixtures-only and never invoked
    /// by production paths.
    pub fn insert_for_test(&mut self, name: String) {
        self.entries.insert(name);
    }

    /// Add one or more feature names to the active set.
    ///
    /// Idempotent: re-submitting an existing name is a no-op (presence is
    /// the activation, so a name that is already present is already
    /// active). Enforces:
    ///
    /// * `MAX_NAMES_PER_BATCH` per-instruction cap → `TooManyNames`.
    /// * `validate_feature_name` on every name (length, allowed charset,
    ///   no leading/trailing whitespace) → `InvalidFeatureName`. Owning
    ///   the check here keeps it the single source of truth — callers
    ///   reaching `enable` outside the processor path cannot smuggle in
    ///   malformed names.
    /// * `MAX_FEATURE_COUNT` cap on the resulting set size — checked
    ///   against the post-insert count of distinct new names so a single
    ///   batch cannot push the registry past the cap →
    ///   `MaxFeatureCountExceeded`.
    pub fn enable(&mut self, names: Vec<String>) -> Result<(), FeatureManagementError> {
        if names.len() > crate::MAX_NAMES_PER_BATCH {
            return Err(FeatureManagementError::TooManyNames);
        }
        for name in &names {
            if !crate::validate_feature_name(name) {
                return Err(FeatureManagementError::InvalidFeatureName);
            }
        }
        // Note: `enable` is permissive on `KNOWN_FEATURES` membership. The
        // on-chain program does not (and cannot, deterministically) reject
        // a name absent from a given binary's `KNOWN_FEATURES`; that's a
        // binary-level concept. An active feature unknown to the running
        // binary is caught by `Bank::can_process_block`, which halts the
        // node at the activation block. Operators staging a name that no
        // released binary yet recognises is therefore on the hook to ship
        // the binary first.
        // Count *distinct* names absent from the existing set. Without the
        // collect-into-set, intra-batch duplicates (`["x", "x"]`) would
        // double-count and reject a request that BTreeSet semantics would
        // otherwise accept.
        let new_distinct = names
            .iter()
            .filter(|n| !self.entries.contains(*n))
            .collect::<BTreeSet<_>>()
            .len();
        if self.entries.len().saturating_add(new_distinct) > crate::MAX_FEATURE_COUNT {
            return Err(FeatureManagementError::MaxFeatureCountExceeded);
        }
        for name in names {
            self.entries.insert(name);
        }
        Ok(())
    }

    /// Record a deferred activation request under the caller-chosen
    /// `request_id`.
    ///
    /// Validates `names` per-name (charset/length) and per-batch
    /// (`MAX_NAMES_PER_BATCH`) at schedule time, enforces `MAX_PENDING_REQUESTS`
    /// on the pending entry count, and rejects a request that would push the
    /// serialized state over `MAX_FEATURES_STATE_SIZE` with
    /// `PendingStateTooLarge`. It does **not** enforce the `MAX_FEATURE_COUNT`
    /// registry cap here — that cap is checked when the schedule fires, via
    /// [`fire_scheduled`](Self::fire_scheduled) → [`enable`](Self::enable). The
    /// `fire_at_ms` future / horizon check is the **caller's** responsibility —
    /// it needs the clock, which this deliberately clock-free state type does
    /// not consult. The `request_id` is supplied by the caller (it is the
    /// cancel handle and the subscription nonce); a `request_id` that already
    /// has a pending entry is rejected with `RequestAlreadyExists`.
    ///
    /// Note the count cap and the byte cap interact: `MAX_PENDING_REQUESTS`
    /// (100) is only reachable with small requests. A request carrying near-max
    /// batches (≈51 KB serialized) exhausts the `MAX_FEATURES_STATE_SIZE`
    /// (100 KB) budget after one or two entries, at which point
    /// `PendingStateTooLarge` — not `TooManyPendingRequests` — is the operative
    /// rejection. The explicit byte check here means that surfaces as a clear
    /// error rather than an opaque `SerializationError` at save time.
    pub fn schedule(
        &mut self,
        request_id: u64,
        names: Vec<String>,
        fire_at_ms: u64,
    ) -> Result<(), FeatureManagementError> {
        if names.len() > crate::MAX_NAMES_PER_BATCH {
            return Err(FeatureManagementError::TooManyNames);
        }
        for name in &names {
            if !crate::validate_feature_name(name) {
                return Err(FeatureManagementError::InvalidFeatureName);
            }
        }
        // Duplicate-id check before the capacity check: a re-used `request_id`
        // is a deterministic client error regardless of how full the map is, so
        // report `RequestAlreadyExists` rather than masking it with
        // `TooManyPendingRequests` (which would wrongly suggest retrying under a
        // different id). A duplicate also would not grow the map, so the
        // capacity guard does not apply to it.
        if self.pending.contains_key(&request_id) {
            return Err(FeatureManagementError::RequestAlreadyExists);
        }
        if self.pending.len() >= crate::MAX_PENDING_REQUESTS {
            return Err(FeatureManagementError::TooManyPendingRequests);
        }
        self.pending
            .insert(request_id, ScheduledRequest { names, fire_at_ms });
        // Reject (and roll back) a request that would exceed the on-chain byte
        // budget, so the operator sees `PendingStateTooLarge` here rather than a
        // bare `SerializationError` when the processor tries to persist. Mirrors
        // the `save_state` guard so the two never disagree.
        let too_large = self
            .serialize()
            .map(|bytes| bytes.len() > crate::MAX_FEATURES_STATE_SIZE)
            .unwrap_or(true);
        if too_large {
            self.pending.remove(&request_id);
            return Err(FeatureManagementError::PendingStateTooLarge);
        }
        Ok(())
    }

    /// Fire a previously-scheduled request: drain its pending entry and
    /// activate its names.
    ///
    /// Removes `pending[request_id]` (returning `RequestNotFound` if absent),
    /// then runs [`enable`](Self::enable) on the recorded names — so the same
    /// `MAX_FEATURE_COUNT` / charset / per-batch validation that a direct
    /// `Enable` enforces is applied at fire time, not just at schedule time.
    ///
    /// Atomicity: the processor saves state after this returns and a failed
    /// `enable` aborts the whole transaction, rolling back the `remove`. So the
    /// drain-and-enable is all-or-nothing — a fired schedule never half-drains.
    pub fn fire_scheduled(&mut self, request_id: u64) -> Result<(), FeatureManagementError> {
        let req = self
            .pending
            .remove(&request_id)
            .ok_or(FeatureManagementError::RequestNotFound)?;
        self.enable(req.names)
    }

    /// Remove a pending scheduled request by id, returning the withdrawn entry.
    ///
    /// Returns `RequestNotFound` if no pending request has that id — already
    /// fired (the one-shot self-removes), already cancelled, or never existed.
    /// Activation is append-only, so a request that has already fired cannot
    /// be undone here.
    pub fn cancel(&mut self, request_id: u64) -> Result<ScheduledRequest, FeatureManagementError> {
        self.pending
            .remove(&request_id)
            .ok_or(FeatureManagementError::RequestNotFound)
    }

    /// Read-only view of the pending scheduled requests, keyed by `request_id`.
    pub fn pending(&self) -> &BTreeMap<u64, ScheduledRequest> {
        &self.pending
    }
}

#[cfg(test)]
mod tests {
    use alloc::{format, vec};

    use super::*;

    #[test]
    fn test_new_has_empty_set() {
        let state = FeaturesState::new_for_test();
        assert!(state.entries().is_empty());
    }

    #[test]
    fn test_enable_inserts_name() {
        let mut state = FeaturesState::new_for_test();
        state.enable(vec!["feat_a".to_string()]).unwrap();
        assert!(state.is_active("feat_a"));
        assert!(!state.is_active("feat_b"));
    }

    #[test]
    fn test_enable_is_idempotent() {
        let mut state = FeaturesState::new_for_test();
        state.enable(vec!["feat_a".to_string()]).unwrap();
        state.enable(vec!["feat_a".to_string()]).unwrap();
        assert_eq!(state.entries().len(), 1);
    }

    #[test]
    fn test_enable_multi_names() {
        let mut state = FeaturesState::new_for_test();
        state
            .enable(vec!["a".to_string(), "b".to_string(), "c".to_string()])
            .unwrap();
        assert_eq!(state.entries().len(), 3);
        assert!(state.is_active("a"));
        assert!(state.is_active("b"));
        assert!(state.is_active("c"));
    }

    #[test]
    fn test_enable_dedup_within_batch() {
        let mut state = FeaturesState::new_for_test();
        state
            .enable(vec!["a".to_string(), "a".to_string(), "b".to_string()])
            .unwrap();
        assert_eq!(state.entries().len(), 2);
    }

    /// Helper: fill state by submitting MAX_NAMES_PER_BATCH-sized chunks
    /// until it carries `target` names.
    fn fill_to(state: &mut FeaturesState, target: usize) {
        let batch = crate::MAX_NAMES_PER_BATCH;
        let mut filled = 0;
        while filled < target {
            let take = (target - filled).min(batch);
            let names: Vec<String> = (filled..filled + take).map(|i| format!("f{i}")).collect();
            state.enable(names).unwrap();
            filled += take;
        }
        assert_eq!(state.entries().len(), target);
    }

    #[test]
    fn test_enable_max_count_enforced() {
        let mut state = FeaturesState::new_for_test();
        fill_to(&mut state, crate::MAX_FEATURE_COUNT);
        assert_eq!(
            state.enable(vec!["overflow".to_string()]),
            Err(FeatureManagementError::MaxFeatureCountExceeded)
        );
    }

    #[test]
    fn test_enable_intra_batch_dup_at_boundary() {
        // At `MAX_FEATURE_COUNT - 1`, enabling `["x", "x"]` must succeed:
        // BTreeSet inserts a single new name, so the post-insert count is
        // exactly `MAX_FEATURE_COUNT`. A naive occurrence count would
        // wrongly reject this.
        let mut state = FeaturesState::new_for_test();
        fill_to(&mut state, crate::MAX_FEATURE_COUNT - 1);
        state
            .enable(vec!["x".to_string(), "x".to_string()])
            .expect("intra-batch duplicate must not double-count against the cap");
        assert_eq!(state.entries().len(), crate::MAX_FEATURE_COUNT);
    }

    #[test]
    fn test_set_authority_works() {
        let mut state = FeaturesState::new_for_test();
        let new = Pubkey::new_from_array([7u8; 32]);
        state.set_authority(new);
        assert_eq!(state.get_authority(), &new);
    }

    #[test]
    fn test_borsh_round_trip_empty() {
        let state = FeaturesState::new_for_test();
        let bytes = state.serialize().unwrap();
        let back = FeaturesState::deserialize(&bytes).unwrap();
        assert_eq!(state, back);
    }

    #[test]
    fn test_borsh_round_trip_with_entries() {
        let mut state = FeaturesState::new_for_test();
        state
            .enable(vec!["alpha".to_string(), "beta".to_string()])
            .unwrap();
        let bytes = state.serialize().unwrap();
        let back = FeaturesState::deserialize(&bytes).unwrap();
        assert_eq!(state, back);
    }

    /// Golden wire-format test. Pins the exact borsh byte layout so any
    /// accidental field reorder, type swap, or new field added without a
    /// schema bump trips here rather than silently mutating the on-chain
    /// shape.
    ///
    /// Layout (borsh):
    /// * `authority`: 32 bytes (Pubkey, raw)
    /// * `pending_authority`: 1-byte tag (0 = None, 1 = Some) + 32 bytes when Some
    /// * `entries`: `BTreeSet<String>` — 4-byte LE length, then each item as
    ///   4-byte LE length + utf-8 bytes, in sorted order
    #[test]
    fn test_wire_format_golden_none_pending() {
        let mut state = FeaturesState::new(Pubkey::new_from_array([1u8; 32]));
        state.insert_for_test("a".to_string());
        state.insert_for_test("bb".to_string());
        let bytes = state.serialize().unwrap();

        let mut expected = vec![0u8; 0];
        expected.extend_from_slice(&[1u8; 32]); // authority
        expected.push(0); // pending_authority = None tag
        expected.extend_from_slice(&2u32.to_le_bytes()); // entries len
        expected.extend_from_slice(&1u32.to_le_bytes()); // "a" len
        expected.extend_from_slice(b"a");
        expected.extend_from_slice(&2u32.to_le_bytes()); // "bb" len
        expected.extend_from_slice(b"bb");
        expected.extend_from_slice(&0u32.to_le_bytes()); // pending (scheduled) len = 0

        assert_eq!(bytes, expected, "wire format drift detected");
        let back = FeaturesState::deserialize(&bytes).unwrap();
        assert_eq!(state, back, "golden bytes must round-trip");
    }

    #[test]
    fn test_wire_format_golden_some_pending() {
        let mut state = FeaturesState::new(Pubkey::new_from_array([1u8; 32]));
        state.set_pending_authority(Some(Pubkey::new_from_array([2u8; 32])));
        state.insert_for_test("a".to_string());
        let bytes = state.serialize().unwrap();

        let mut expected = vec![0u8; 0];
        expected.extend_from_slice(&[1u8; 32]); // authority
        expected.push(1); // pending_authority = Some tag
        expected.extend_from_slice(&[2u8; 32]); // pending_authority payload
        expected.extend_from_slice(&1u32.to_le_bytes()); // entries len
        expected.extend_from_slice(&1u32.to_le_bytes()); // "a" len
        expected.extend_from_slice(b"a");
        expected.extend_from_slice(&0u32.to_le_bytes()); // pending (scheduled) len = 0

        assert_eq!(bytes, expected, "wire format drift detected");
        let back = FeaturesState::deserialize(&bytes).unwrap();
        assert_eq!(state, back, "golden bytes must round-trip");
    }

    /// Golden wire-format test with a non-empty `pending` map. Pins the exact
    /// borsh layout of a single scheduled request so a field reorder / type
    /// swap inside `ScheduledRequest` (or a change to how `pending` is encoded)
    /// trips here.
    ///
    /// Layout of the `pending` `BTreeMap<u64, ScheduledRequest>`:
    /// * 4-byte LE entry count
    /// * per entry: `request_id` (u64 LE), then `ScheduledRequest` borsh =
    ///   `names` `Vec<String>` (4-byte LE count + per-name 4-byte LE len + utf8)
    ///   followed by `fire_at_ms` (u64 LE)
    #[test]
    fn test_wire_format_golden_with_pending() {
        let mut state = FeaturesState::new(Pubkey::new_from_array([1u8; 32]));
        state
            .schedule(9, vec!["feat_x".to_string()], 1_700_000_000_000)
            .expect("schedule must succeed");
        let bytes = state.serialize().unwrap();

        let mut expected = vec![0u8; 0];
        expected.extend_from_slice(&[1u8; 32]); // authority
        expected.push(0); // pending_authority = None tag
        expected.extend_from_slice(&0u32.to_le_bytes()); // entries len = 0
        expected.extend_from_slice(&1u32.to_le_bytes()); // pending (scheduled) count = 1
        expected.extend_from_slice(&9u64.to_le_bytes()); // request_id key
                                                         // ScheduledRequest.names: Vec<String>
        expected.extend_from_slice(&1u32.to_le_bytes()); // names len = 1
        expected.extend_from_slice(&6u32.to_le_bytes()); // "feat_x" len
        expected.extend_from_slice(b"feat_x");
        // ScheduledRequest.fire_at_ms: u64 LE
        expected.extend_from_slice(&1_700_000_000_000u64.to_le_bytes());

        assert_eq!(bytes, expected, "wire format drift detected");
        let back = FeaturesState::deserialize(&bytes).unwrap();
        assert_eq!(state, back, "golden bytes must round-trip");
    }

    /// Golden wire-format test pinning the FULL-shape byte layout that the
    /// other `test_wire_format_golden_*` fixtures don't cover together:
    /// `pending_authority = Some(..)` AND a non-empty `entries` set AND a
    /// non-empty `pending` map, in one record. A field reorder, type swap, or
    /// new field added without a schema bump trips this rather than silently
    /// mutating the on-chain shape.
    ///
    /// Full borsh layout, in declaration order:
    /// * `authority`: 32 raw bytes (Pubkey)
    /// * `pending_authority`: 1-byte Option tag (1 = Some) + 32 bytes payload
    /// * `entries`: `BTreeSet<String>` — 4-byte LE count, then each name as
    ///   4-byte LE len + utf-8, in sorted order ("a" < "bb")
    /// * `pending`: `BTreeMap<u64, ScheduledRequest>` — 4-byte LE count, then
    ///   per entry (ascending key): `request_id` (u64 LE) + `ScheduledRequest`
    ///   = `names` `Vec<String>` (4-byte LE count + per-name 4-byte LE len +
    ///   utf-8) + `fire_at_ms` (u64 LE)
    #[test]
    fn test_wire_format_golden_full_shape() {
        let mut state = FeaturesState::new(Pubkey::new_from_array([7u8; 32]));
        state.set_pending_authority(Some(Pubkey::new_from_array([8u8; 32])));
        state.insert_for_test("a".to_string());
        state.insert_for_test("bb".to_string());
        // request_id 1 < 2, so BTreeMap iterates 1 then 2.
        state
            .schedule(1, vec!["x".to_string()], 1_000)
            .expect("schedule id 1");
        state
            .schedule(2, vec!["y".to_string(), "z".to_string()], 2_000)
            .expect("schedule id 2");
        let bytes = state.serialize().unwrap();

        let mut expected = vec![0u8; 0];
        expected.extend_from_slice(&[7u8; 32]); // authority
        expected.push(1); // pending_authority = Some tag
        expected.extend_from_slice(&[8u8; 32]); // pending_authority payload
        expected.extend_from_slice(&2u32.to_le_bytes()); // entries count = 2
        expected.extend_from_slice(&1u32.to_le_bytes()); // "a" len
        expected.extend_from_slice(b"a");
        expected.extend_from_slice(&2u32.to_le_bytes()); // "bb" len
        expected.extend_from_slice(b"bb");
        expected.extend_from_slice(&2u32.to_le_bytes()); // pending count = 2
                                                         // pending[1]
        expected.extend_from_slice(&1u64.to_le_bytes()); // request_id key
        expected.extend_from_slice(&1u32.to_le_bytes()); // names count = 1
        expected.extend_from_slice(&1u32.to_le_bytes()); // "x" len
        expected.extend_from_slice(b"x");
        expected.extend_from_slice(&1_000u64.to_le_bytes()); // fire_at_ms
                                                             // pending[2]
        expected.extend_from_slice(&2u64.to_le_bytes()); // request_id key
        expected.extend_from_slice(&2u32.to_le_bytes()); // names count = 2
        expected.extend_from_slice(&1u32.to_le_bytes()); // "y" len
        expected.extend_from_slice(b"y");
        expected.extend_from_slice(&1u32.to_le_bytes()); // "z" len
        expected.extend_from_slice(b"z");
        expected.extend_from_slice(&2_000u64.to_le_bytes()); // fire_at_ms

        assert_eq!(bytes, expected, "wire format drift detected");
        let back = FeaturesState::deserialize(&bytes).unwrap();
        assert_eq!(state, back, "golden bytes must round-trip");
    }

    #[test]
    fn test_pending_authority_round_trip() {
        let mut state = FeaturesState::new_for_test();
        assert!(state.pending_authority().is_none());
        let p = Pubkey::new_from_array([3u8; 32]);
        state.set_pending_authority(Some(p));
        assert_eq!(state.pending_authority(), Some(&p));
        state.set_pending_authority(None);
        assert!(state.pending_authority().is_none());
    }

    #[test]
    fn test_propose_then_accept_transfers_authority() {
        let mut state = FeaturesState::new_for_test();
        let original = *state.get_authority();
        let new = Pubkey::new_from_array([4u8; 32]);

        state.propose_transfer(new).unwrap();
        assert_eq!(state.pending_authority(), Some(&new));
        assert_eq!(state.get_authority(), &original);

        state.accept_transfer().unwrap();
        assert_eq!(state.get_authority(), &new);
        assert!(state.pending_authority().is_none());
    }

    #[test]
    fn test_propose_to_self_rejected() {
        let mut state = FeaturesState::new_for_test();
        let current = *state.get_authority();
        assert_eq!(
            state.propose_transfer(current),
            Err(FeatureManagementError::InvalidTransferTarget)
        );
        assert!(state.pending_authority().is_none());
    }

    #[test]
    fn test_propose_with_pending_rejected() {
        let mut state = FeaturesState::new_for_test();
        state.propose_transfer(Pubkey::new_unique()).unwrap();
        assert_eq!(
            state.propose_transfer(Pubkey::new_unique()),
            Err(FeatureManagementError::PendingTransferExists)
        );
    }

    #[test]
    fn test_accept_with_no_pending_rejected() {
        let mut state = FeaturesState::new_for_test();
        assert_eq!(
            state.accept_transfer(),
            Err(FeatureManagementError::NoPendingTransfer)
        );
    }

    #[test]
    fn test_cancel_clears_pending() {
        let mut state = FeaturesState::new_for_test();
        let original = *state.get_authority();
        state.propose_transfer(Pubkey::new_unique()).unwrap();
        state.cancel_transfer().unwrap();
        assert!(state.pending_authority().is_none());
        assert_eq!(state.get_authority(), &original);
    }

    #[test]
    fn test_cancel_with_no_pending_rejected() {
        let mut state = FeaturesState::new_for_test();
        assert_eq!(
            state.cancel_transfer(),
            Err(FeatureManagementError::NoPendingTransfer)
        );
    }

    #[test]
    fn test_full_cycle_propose_cancel_propose_accept() {
        // Operator cancels a proposal and replaces it with a different
        // target; the new target accepts. Guards against a cancelled
        // proposal leaking into the post-accept authority.
        let mut state = FeaturesState::new_for_test();
        let original = *state.get_authority();
        let first = Pubkey::new_from_array([10u8; 32]);
        let second = Pubkey::new_from_array([20u8; 32]);

        state.propose_transfer(first).unwrap();
        state.cancel_transfer().unwrap();
        assert_eq!(state.get_authority(), &original);
        assert!(state.pending_authority().is_none());

        state.propose_transfer(second).unwrap();
        state.accept_transfer().unwrap();
        assert_eq!(state.get_authority(), &second);
        assert!(state.pending_authority().is_none());
    }

    #[test]
    fn test_accept_then_propose_works_with_new_authority() {
        // After a transfer commits, the new authority can propose
        // another transfer — the pending slot is back to None
        // post-accept so no `PendingTransferExists` leakage.
        let mut state = FeaturesState::new_for_test();
        let mid = Pubkey::new_from_array([11u8; 32]);
        let final_target = Pubkey::new_from_array([22u8; 32]);

        state.propose_transfer(mid).unwrap();
        state.accept_transfer().unwrap();
        assert_eq!(state.get_authority(), &mid);

        state.propose_transfer(final_target).unwrap();
        assert_eq!(state.pending_authority(), Some(&final_target));
    }

    #[test]
    fn test_propose_does_not_touch_authority() {
        // A proposal in flight must not modify the active authority field
        // — only Accept commits.
        let mut state = FeaturesState::new_for_test();
        let original = *state.get_authority();
        state
            .propose_transfer(Pubkey::new_from_array([33u8; 32]))
            .unwrap();
        assert_eq!(state.get_authority(), &original);
    }

    #[test]
    fn test_single_step_clear_pending_blocks_stale_accept() {
        // Models the processor's `UpdateAuthority` sequence:
        // `set_authority(new)` + `set_pending_authority(None)`. After
        // that sequence the previously-pending party cannot displace
        // the just-set authority via `accept_transfer`. Guards the
        // displacement vector flagged in PR3790 review.
        let mut state = FeaturesState::new_for_test();
        let pending = Pubkey::new_from_array([44u8; 32]);
        let single_step_target = Pubkey::new_from_array([55u8; 32]);

        state.propose_transfer(pending).unwrap();
        assert_eq!(state.pending_authority(), Some(&pending));

        // Operator picks the single-step path instead of accepting.
        state.set_authority(single_step_target);
        state.set_pending_authority(None);

        // Stale pending party cannot displace the new authority.
        assert_eq!(
            state.accept_transfer(),
            Err(FeatureManagementError::NoPendingTransfer)
        );
        assert_eq!(state.get_authority(), &single_step_target);
    }

    #[test]
    fn test_enable_rejects_oversized_batch() {
        let mut state = FeaturesState::new_for_test();
        let names: Vec<String> = (0..=crate::MAX_NAMES_PER_BATCH)
            .map(|i| format!("f{i}"))
            .collect();
        assert_eq!(
            state.enable(names),
            Err(FeatureManagementError::TooManyNames)
        );
    }

    #[test]
    fn test_enable_rejects_name_above_max_length() {
        let mut state = FeaturesState::new_for_test();
        let long_name = "a".repeat(crate::MAX_FEATURE_NAME_LENGTH + 1);
        assert_eq!(
            state.enable(vec![long_name]),
            Err(FeatureManagementError::InvalidFeatureName)
        );
    }

    #[test]
    fn test_enable_accepts_name_at_exact_max_length() {
        let mut state = FeaturesState::new_for_test();
        let at_max = "a".repeat(crate::MAX_FEATURE_NAME_LENGTH);
        state
            .enable(vec![at_max])
            .expect("name at exact MAX_FEATURE_NAME_LENGTH must be accepted");
        assert_eq!(state.entries().len(), 1);
    }

    #[test]
    fn test_enable_rejects_one_long_name_in_otherwise_valid_batch() {
        // The check is per-name, not per-batch: a single oversized name
        // among well-formed ones must still trip the rejection.
        let mut state = FeaturesState::new_for_test();
        let names = vec![
            "shortish".to_string(),
            "a".repeat(crate::MAX_FEATURE_NAME_LENGTH + 1),
            "also_shortish".to_string(),
        ];
        assert_eq!(
            state.enable(names),
            Err(FeatureManagementError::InvalidFeatureName)
        );
        assert!(
            state.entries().is_empty(),
            "rejected enable must leave state untouched"
        );
    }

    #[test]
    fn test_enable_rejects_empty_name() {
        let mut state = FeaturesState::new_for_test();
        assert_eq!(
            state.enable(vec![String::new()]),
            Err(FeatureManagementError::InvalidFeatureName)
        );
    }

    #[test]
    fn test_enable_rejects_invalid_charset() {
        // `validate_feature_name` allows alphanumeric + `_` + `-` only.
        let mut state = FeaturesState::new_for_test();
        assert_eq!(
            state.enable(vec!["has space".to_string()]),
            Err(FeatureManagementError::InvalidFeatureName)
        );
        assert_eq!(
            state.enable(vec!["dot.name".to_string()]),
            Err(FeatureManagementError::InvalidFeatureName)
        );
    }

    #[test]
    fn schedule_rejects_duplicate_request_id() {
        let mut state = FeaturesState::new_for_test();
        state
            .schedule(7, vec!["feat_a".to_string()], 1_000)
            .expect("first schedule with id 7 must succeed");
        assert_eq!(
            state.schedule(7, vec!["feat_b".to_string()], 2_000),
            Err(FeatureManagementError::RequestAlreadyExists)
        );
        // The original entry is untouched by the rejected duplicate.
        assert_eq!(state.pending().len(), 1);
        let entry = state.pending().get(&7).expect("entry must exist");
        assert_eq!(entry.names, vec!["feat_a".to_string()]);
        assert_eq!(entry.fire_at_ms, 1_000);
    }

    #[test]
    fn schedule_duplicate_id_at_capacity_reports_duplicate_not_full() {
        // Fill the pending map to MAX_PENDING_REQUESTS, then resubmit an id that
        // already exists. The duplicate-id check runs before the capacity check,
        // so the caller gets RequestAlreadyExists (the precise error) rather than
        // TooManyPendingRequests masking the collision.
        let mut state = FeaturesState::new_for_test();
        for id in 0..crate::MAX_PENDING_REQUESTS as u64 {
            state
                .schedule(id, vec!["feat".to_string()], 1_000)
                .expect("fill under cap");
        }
        assert_eq!(state.pending().len(), crate::MAX_PENDING_REQUESTS);
        assert_eq!(
            state.schedule(0, vec!["feat".to_string()], 2_000),
            Err(FeatureManagementError::RequestAlreadyExists),
            "a duplicate id at capacity is RequestAlreadyExists, not TooManyPendingRequests"
        );
        // A new id at capacity still correctly reports the capacity error.
        assert_eq!(
            state.schedule(
                crate::MAX_PENDING_REQUESTS as u64,
                vec!["feat".to_string()],
                2_000
            ),
            Err(FeatureManagementError::TooManyPendingRequests)
        );
    }

    #[test]
    fn schedule_rejects_oversized_pending_state() {
        // A batch of MAX_NAMES_PER_BATCH names each at MAX_FEATURE_NAME_LENGTH
        // serializes to ≈51 KB, so two of them blow past MAX_FEATURES_STATE_SIZE
        // (100 KB). This pins the real binding cap: the byte budget, not the
        // MAX_PENDING_REQUESTS count.
        let max_batch =
            || vec!["a".repeat(crate::MAX_FEATURE_NAME_LENGTH); crate::MAX_NAMES_PER_BATCH];

        let mut state = FeaturesState::new_for_test();
        // One near-max request fits within the byte budget.
        state
            .schedule(1, max_batch(), 1_000)
            .expect("one max-size request must fit");
        assert!(
            state.serialize().expect("serialize").len() <= crate::MAX_FEATURES_STATE_SIZE,
            "a single max-size request must stay within MAX_FEATURES_STATE_SIZE"
        );

        // A second pushes the serialized state past the cap and is rejected
        // with the explicit error — not a bare SerializationError at save time.
        assert_eq!(
            state.schedule(2, max_batch(), 2_000),
            Err(FeatureManagementError::PendingStateTooLarge)
        );
        // The rejected request is rolled back; only the first entry remains.
        assert_eq!(state.pending().len(), 1);
        assert!(state.pending().get(&2).is_none());
        // The byte budget tripped well before the count cap (100).
        assert!(state.pending().len() < crate::MAX_PENDING_REQUESTS);
    }

    #[test]
    fn fire_scheduled_drains_pending_and_activates() {
        let mut state = FeaturesState::new_for_test();
        state
            .schedule(5, vec!["feat_a".to_string(), "feat_b".to_string()], 1_000)
            .expect("schedule must succeed");
        assert_eq!(state.pending().len(), 1);

        state.fire_scheduled(5).expect("fire must succeed");

        // Names are now active and the pending entry is drained.
        assert!(state.is_active("feat_a"));
        assert!(state.is_active("feat_b"));
        assert!(state.pending().get(&5).is_none());
        assert!(state.pending().is_empty());
    }

    #[test]
    fn fire_scheduled_unknown_request_id() {
        let mut state = FeaturesState::new_for_test();
        assert_eq!(
            state.fire_scheduled(99),
            Err(FeatureManagementError::RequestNotFound)
        );
    }

    #[test]
    fn test_enable_is_permissive_on_unknown_names() {
        // Documents the design: `state.enable` does not check
        // `KNOWN_FEATURES`. A name unknown to a binary may be stored on
        // chain; the runtime's `can_process_block` is what halts the node
        // at the activation block if the name is also active. The on-chain
        // program lacks a deterministic view of every binary's
        // `KNOWN_FEATURES`, so the check cannot live here.
        let mut state = FeaturesState::new_for_test();
        state
            .enable(vec!["totally_fabricated_name_not_in_any_binary".to_string()])
            .expect("enable must be permissive on KNOWN_FEATURES membership");
    }

    mod proptests {
        use alloc::collections::{BTreeMap, BTreeSet};

        use proptest::{prelude::*, test_runner::TestCaseError};

        use super::*;

        prop_compose! {
            /// A name that always passes `validate_feature_name`: 1..=16
            /// alphanumeric / `_` / `-` chars (capped well under
            /// `MAX_FEATURE_NAME_LENGTH` to keep generation cheap).
            fn valid_name()(s in "[a-zA-Z0-9_-]{1,16}") -> String {
                s
            }
        }

        prop_compose! {
            /// A small batch (0..=8 names) of valid feature names.
            fn valid_batch()(batch in prop::collection::vec(valid_name(), 0..=8)) -> Vec<String> {
                batch
            }
        }

        prop_compose! {
            /// An arbitrary [`FeaturesState`] built only from valid inputs:
            /// random authority bytes, an optional pending authority, a set of
            /// valid `entries`, and a `pending` map of `ScheduledRequest`.
            ///
            /// Bounds are deliberately tight (≤8 distinct entry names, ≤8
            /// pending ids each carrying ≤4 names of ≤16 chars) so the
            /// serialized form stays far below `MAX_FEATURES_STATE_SIZE`
            /// (100 KB) and generation stays cheap. `entries` / `pending` are
            /// written through the `pub(crate)` fields directly rather than via
            /// `enable` / `schedule`, so the round-trip exercises arbitrary
            /// in-memory shapes, not just validator-reachable ones.
            fn arbitrary_state()(
                authority in any::<[u8; 32]>(),
                pending_authority in proptest::option::of(any::<[u8; 32]>()),
                entries in prop::collection::btree_set(valid_name(), 0..=8),
                pending in prop::collection::btree_map(
                    any::<u64>(),
                    (prop::collection::vec(valid_name(), 0..=4), any::<u64>()),
                    0..=8,
                ),
            ) -> FeaturesState {
                let mut state = FeaturesState::new(Pubkey::new_from_array(authority));
                state.set_pending_authority(pending_authority.map(Pubkey::new_from_array));
                state.entries = entries;
                state.pending = pending
                    .into_iter()
                    .map(|(id, (names, fire_at_ms))| (id, ScheduledRequest { names, fire_at_ms }))
                    .collect::<BTreeMap<u64, ScheduledRequest>>();
                state
            }
        }

        proptest! {
            /// Borsh round-trips any `FeaturesState`: deserializing its own
            /// serialization yields an equal value across the full shape —
            /// random authority, optional pending authority, arbitrary
            /// `entries`, and an arbitrary `pending` map. Complements the
            /// fixed-input `test_wire_format_golden_*` goldens (which pin exact
            /// bytes for hand-picked shapes) by exercising the encode/decode
            /// pair over a wide input space.
            #[test]
            fn borsh_round_trips_arbitrary_state(state in arbitrary_state()) {
                // `map_err(TestCaseError::fail)?` rather than `.expect()` so a
                // failure preserves proptest's input shrinking instead of
                // aborting the whole run with a panic.
                let bytes = state
                    .serialize()
                    .map_err(|e| TestCaseError::fail(e.to_string()))?;
                // Stay well within the on-chain byte budget for the bounds we
                // generate; a regression that bloats the encoding trips here.
                prop_assert!(
                    bytes.len() <= crate::MAX_FEATURES_STATE_SIZE,
                    "serialized {} bytes exceeds MAX_FEATURES_STATE_SIZE",
                    bytes.len()
                );
                let back = FeaturesState::deserialize(&bytes)
                    .map_err(|e| TestCaseError::fail(e.to_string()))?;
                prop_assert_eq!(state, back);
            }

            /// enable is idempotent, append-only (monotone), and the final set
            /// is exactly the union of every enabled name; `is_active` tracks
            /// membership.
            #[test]
            fn enable_is_monotone_and_unions(batches in prop::collection::vec(valid_batch(), 0..=10)) {
                let mut state = FeaturesState::new_for_test();
                let mut expected: BTreeSet<String> = BTreeSet::new();
                let mut prev_len = 0usize;
                for batch in &batches {
                    let r = state.enable(batch.clone());
                    prop_assert!(r.is_ok(), "valid batch must enable: {r:?}");
                    expected.extend(batch.iter().cloned());
                    // Append-only: the set never shrinks.
                    prop_assert!(state.entries().len() >= prev_len);
                    prev_len = state.entries().len();
                }
                prop_assert_eq!(state.entries(), &expected);
                for name in &expected {
                    prop_assert!(state.is_active(name));
                }
                // Re-enabling everything already present is a no-op.
                let all: Vec<String> = expected.iter().cloned().collect();
                for chunk in all.chunks(crate::MAX_NAMES_PER_BATCH) {
                    let r = state.enable(chunk.to_vec());
                    prop_assert!(r.is_ok(), "re-enable is a no-op: {r:?}");
                }
                prop_assert_eq!(state.entries().len(), expected.len());
            }

            /// `is_active(name)` agrees with an independent oracle (the set of
            /// names actually enabled) for any probe name. Checking against the
            /// input batch rather than `entries()` keeps this from collapsing
            /// into `entries.contains(x) == entries.contains(x)` — it would
            /// catch an `is_active` that drifted from the backing set.
            #[test]
            fn is_active_equals_membership(
                batch in valid_batch(),
                probe in valid_name(),
            ) {
                let oracle: BTreeSet<String> = batch.iter().cloned().collect();
                let mut state = FeaturesState::new_for_test();
                let r = state.enable(batch);
                prop_assert!(r.is_ok(), "valid batch must enable: {r:?}");
                prop_assert_eq!(state.is_active(&probe), oracle.contains(&probe));
            }

            /// A batch larger than `MAX_NAMES_PER_BATCH` is rejected with
            /// `TooManyNames`, leaving the set untouched.
            #[test]
            fn enable_rejects_oversized_batch(
                extra in 1usize..=20,
            ) {
                let mut state = FeaturesState::new_for_test();
                let names: Vec<String> = (0..crate::MAX_NAMES_PER_BATCH + extra)
                    .map(|i| format!("f{i}"))
                    .collect();
                prop_assert_eq!(state.enable(names), Err(FeatureManagementError::TooManyNames));
                prop_assert!(state.entries().is_empty());
            }

            /// A name that fails `validate_feature_name` (illegal char or over
            /// length) is rejected with `InvalidFeatureName`.
            #[test]
            fn enable_rejects_invalid_name(
                bad in prop_oneof![
                    "[a-z]{0,4}[ .!@/][a-z]{0,4}",
                    Just("a".repeat(crate::MAX_FEATURE_NAME_LENGTH + 1)),
                ],
            ) {
                prop_assume!(!crate::validate_feature_name(&bad));
                let mut state = FeaturesState::new_for_test();
                prop_assert_eq!(
                    state.enable(vec![bad]),
                    Err(FeatureManagementError::InvalidFeatureName)
                );
                prop_assert!(state.entries().is_empty());
            }

            /// schedule records each distinct request keyed by id with `names`
            /// and `fire_at_ms` preserved; a duplicate id is rejected; `cancel`
            /// returns and removes the entry; cancelling an unknown id is
            /// `RequestNotFound`; draining everything empties `pending`.
            #[test]
            fn schedule_cancel_round_trip(
                // Bounded well under `MAX_PENDING_REQUESTS` (100) so distinct
                // ids can never overflow `pending` and trip `TooManyPendingRequests`.
                reqs in prop::collection::vec(
                    (any::<u64>(), valid_batch(), any::<u64>()),
                    0..=50,
                ),
            ) {
                let mut state = FeaturesState::new_for_test();
                let mut recorded: Vec<(u64, Vec<String>, u64)> = Vec::new();
                for (id, names, fire_at_ms) in reqs {
                    if recorded.iter().any(|(rid, ..)| *rid == id) {
                        // Duplicate id: rejected, original untouched.
                        prop_assert_eq!(
                            state.schedule(id, names.clone(), fire_at_ms),
                            Err(FeatureManagementError::RequestAlreadyExists)
                        );
                    } else {
                        let r = state.schedule(id, names.clone(), fire_at_ms);
                        prop_assert!(r.is_ok(), "distinct id must schedule: {r:?}");
                        let entry = state.pending().get(&id);
                        prop_assert!(entry.is_some(), "entry recorded");
                        let entry = entry.unwrap();
                        prop_assert_eq!(&entry.names, &names);
                        prop_assert_eq!(entry.fire_at_ms, fire_at_ms);
                        recorded.push((id, names, fire_at_ms));
                    }
                }
                prop_assert_eq!(state.pending().len(), recorded.len());

                // Cancelling an id never scheduled is RequestNotFound.
                let mut absent = 0u64;
                while recorded.iter().any(|(rid, ..)| *rid == absent) {
                    absent = absent.wrapping_add(1);
                }
                prop_assert_eq!(
                    state.cancel(absent),
                    Err(FeatureManagementError::RequestNotFound)
                );

                // Cancel returns the entry and removes it; draining empties.
                for (id, names, fire_at_ms) in &recorded {
                    let got = state.cancel(*id);
                    prop_assert!(got.is_ok(), "recorded id must cancel: {got:?}");
                    let got = got.unwrap();
                    prop_assert_eq!(&got.names, names);
                    prop_assert_eq!(got.fire_at_ms, *fire_at_ms);
                    prop_assert_eq!(
                        state.cancel(*id),
                        Err(FeatureManagementError::RequestNotFound)
                    );
                }
                prop_assert!(state.pending().is_empty());
            }

            /// Scheduling `MAX_PENDING_REQUESTS` distinct ids fills `pending`;
            /// one more distinct id is rejected with `TooManyPendingRequests`
            /// and leaves the set at the cap.
            #[test]
            fn schedule_rejects_over_capacity(
                fire_at_ms in any::<u64>(),
            ) {
                let mut state = FeaturesState::new_for_test();
                for id in 0..crate::MAX_PENDING_REQUESTS as u64 {
                    let r = state.schedule(id, vec!["feat".to_string()], fire_at_ms);
                    prop_assert!(r.is_ok(), "id {id} under cap must schedule: {r:?}");
                }
                prop_assert_eq!(state.pending().len(), crate::MAX_PENDING_REQUESTS);
                // One past the cap, with a distinct id, is rejected.
                prop_assert_eq!(
                    state.schedule(
                        crate::MAX_PENDING_REQUESTS as u64,
                        vec!["feat".to_string()],
                        fire_at_ms,
                    ),
                    Err(FeatureManagementError::TooManyPendingRequests)
                );
                prop_assert_eq!(state.pending().len(), crate::MAX_PENDING_REQUESTS);
            }

            /// `schedule` runs the same per-batch / per-name validation as
            /// `enable`: an oversized batch is `TooManyNames` and an invalid
            /// name is `InvalidFeatureName`, in both cases leaving `pending`
            /// untouched. Symmetric with `enable_rejects_oversized_batch` /
            /// `enable_rejects_invalid_name` so a regression in either gate is
            /// caught on the schedule path too.
            #[test]
            fn schedule_rejects_invalid_input(
                extra in 1usize..=20,
                bad in prop_oneof![
                    "[a-z]{0,4}[ .!@/][a-z]{0,4}",
                    Just("a".repeat(crate::MAX_FEATURE_NAME_LENGTH + 1)),
                ],
            ) {
                prop_assume!(!crate::validate_feature_name(&bad));

                let mut state = FeaturesState::new_for_test();
                let oversized: Vec<String> = (0..crate::MAX_NAMES_PER_BATCH + extra)
                    .map(|i| format!("f{i}"))
                    .collect();
                prop_assert_eq!(
                    state.schedule(1, oversized, 1_000),
                    Err(FeatureManagementError::TooManyNames)
                );
                prop_assert_eq!(
                    state.schedule(2, vec![bad], 1_000),
                    Err(FeatureManagementError::InvalidFeatureName)
                );
                // Neither rejected request left a pending entry.
                prop_assert!(state.pending().is_empty());
            }
        }
    }
}