armdb 0.11.2

//! The engine-owned row and owner types for a unique namespace (T10).
//!
//! A unique namespace is `key -> owner`: [`UniqueRow`] is the 80-byte
//! zerocopy value every namespace collection stores; [`UniqueOwner`] is its
//! typed, decoded form — the source collection that holds the key (via its
//! 64-bit name tag) plus that collection's own row key, and the deadline
//! when the hold is a reservation rather than a permanent claim.

use armour_core::GetType;
use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};

/// Owner-key capacity. A format constant: the widest key in the measured
/// consumer is 48 bytes, this leaves one 8-byte component of headroom.
pub const UNIQUE_OWNER_KEY_CAP: usize = 56;

/// The engine-owned value of every unique namespace row.
///
/// `expires_at == 0` is a permanent owner; any other value is a reservation
/// deadline in milliseconds since the Unix epoch, which is also what the TTL
/// policy reads (a permanent row has no deadline and the sweeper never sees it).
#[derive(
    Clone, Copy, Debug, PartialEq, Eq, GetType, FromBytes, IntoBytes, Immutable, KnownLayout,
)]
#[repr(C)]
pub struct UniqueRow {
    expires_at: u64,
    collection_id: u64,
    key_len: u8,
    _pad: [u8; 7],
    key: [u8; UNIQUE_OWNER_KEY_CAP],
}

const _: () = assert!(size_of::<UniqueRow>() == 80);

/// FNV-1a over the collection name. `const` so a declaration can compute the
/// tag of `T::NAME` at compile time.
pub const fn collection_tag(name: &str) -> u64 {
    let bytes = name.as_bytes();
    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
    let mut i = 0;
    while i < bytes.len() {
        hash ^= bytes[i] as u64;
        hash = hash.wrapping_mul(0x1000_0000_01b3);
        i += 1;
    }
    hash
}

impl UniqueRow {
    /// Build a row from an owner — the resolver's [`ClaimDecision::Write`]
    /// arm (`unique::resolve::decide`, T10 Task 7) is the production caller.
    ///
    /// [`ClaimDecision::Write`]: crate::unique::resolve::ClaimDecision::Write
    pub(crate) fn new(owner: UniqueOwner) -> Self {
        Self {
            expires_at: owner.expires_at,
            collection_id: owner.tag,
            key_len: owner.key_len,
            _pad: [0; 7],
            key: owner.key,
        }
    }

    pub(crate) fn owner(&self) -> UniqueOwner {
        UniqueOwner {
            tag: self.collection_id,
            key_len: self.key_len,
            key: self.key,
            expires_at: self.expires_at,
        }
    }

    /// A row whose `key_len` cannot address its own buffer is corrupt; the
    /// validator reports it and no decode is attempted.
    pub(crate) fn is_well_formed(&self) -> bool {
        (self.key_len as usize) <= UNIQUE_OWNER_KEY_CAP
    }

    /// The row's TTL deadline: `None` for a permanent owner (`expires_at ==
    /// 0`), `Some(expires_at)` for a reservation.
    ///
    /// `pub`, not `pub(crate)`: the [`crate::unique_index!`] macro is
    /// `#[macro_export]`ed and expands in the declaring crate (application
    /// code, or this crate's own integration tests) — code it generates can
    /// only call methods this crate exposes across the crate boundary, so
    /// the row's own [`CollectionMeta::TTL`](crate::CollectionMeta::TTL)
    /// policy reads the deadline through this accessor rather than a
    /// crate-private one.
    pub fn deadline(&self) -> Option<u64> {
        (self.expires_at != 0).then_some(self.expires_at)
    }
}

/// Who holds a key: a source collection and its row key, plus the deadline
/// when the hold is a reservation.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct UniqueOwner {
    tag: u64,
    key_len: u8,
    key: [u8; UNIQUE_OWNER_KEY_CAP],
    expires_at: u64,
}

impl UniqueOwner {
    /// Build an owner from a source collection's name tag and key bytes.
    /// Returns `None` when the key does not fit — callers that can prove the
    /// fit at compile time use the `const` assert in the declaration instead.
    pub(crate) fn new(tag: u64, key: &[u8], expires_at: u64) -> Option<Self> {
        if key.len() > UNIQUE_OWNER_KEY_CAP {
            return None;
        }
        let mut buf = [0u8; UNIQUE_OWNER_KEY_CAP];
        buf[..key.len()].copy_from_slice(key);
        Some(Self {
            tag,
            key_len: key.len() as u8,
            key: buf,
            expires_at,
        })
    }

    pub fn collection_id(&self) -> u64 {
        self.tag
    }

    pub fn is_reservation(&self) -> bool {
        self.expires_at != 0
    }

    pub fn expires_at(&self) -> Option<u64> {
        (self.expires_at != 0).then_some(self.expires_at)
    }

    /// Two owners are the same holder when the collection and key match — the
    /// deadline is state, not identity, which is what makes "my own
    /// reservation" and "my own permanent row" the same branch.
    ///
    /// The resolver's rule (`unique::resolve::decide`, T10 Task 7) is the
    /// production caller: a current owner that is not `same_holder` as the
    /// claimant is a conflict for every [`crate::txn::participant::ClaimKind`].
    pub(crate) fn same_holder(&self, other: &UniqueOwner) -> bool {
        self.tag == other.tag && self.key_len == other.key_len && self.key == other.key
    }

    /// A copy of this owner as a reservation expiring at `deadline_ms` —
    /// used by [`crate::unique::resolve::decide`]'s `Reserve` arms.
    /// `deadline_ms == 0` would read back as permanent
    /// ([`UniqueRow::deadline`]), so a caller must never pass it —
    /// `resolve_claims` (T10 Task 8) never does, since it stamps a real
    /// absolute deadline before `Reserve` ever reaches `decide`.
    pub(crate) fn with_deadline(&self, deadline_ms: u64) -> Self {
        Self {
            expires_at: deadline_ms,
            ..*self
        }
    }

    pub(crate) fn key_bytes(&self) -> &[u8] {
        &self.key[..self.key_len as usize]
    }

    /// Decode the owner key as `T`'s key type.
    ///
    /// Both the collection tag **and** the encoded length must match before any
    /// decode is attempted, guarding against a corrupt or foreign row.
    ///
    /// This does **not** go through `Key::from_bytes` (a typed `ptr::read` that
    /// requires the source bytes to be aligned to `align_of::<T::SelfId>()`).
    /// `self.key` is a `[u8; 56]` field inside `UniqueOwner`, whose own
    /// alignment is 8 — a key type with alignment greater than 8 (e.g. a
    /// `u128` field, or an explicit `#[repr(align(16))]`) would make that read
    /// unaligned, which is undefined behaviour even though this is a safe
    /// public method (T10 review F1). Building through `T::SelfId::zeroed()`
    /// and copying bytes into it instead routes the copy through a properly
    /// aligned local, at the cost of one extra memcpy over every other key
    /// type — not a real cost against an 80-byte row.
    pub fn belongs_to<T: crate::CollectionMeta>(&self) -> Option<T::SelfId> {
        if self.tag != collection_tag(T::NAME) {
            return None;
        }
        if self.key_len as usize != size_of::<T::SelfId>() {
            return None;
        }
        let mut out = <T::SelfId as crate::Key>::zeroed();
        <T::SelfId as crate::Key>::as_bytes_mut(&mut out).copy_from_slice(self.key_bytes());
        Some(out)
    }
}

/// Build a [`UniqueOwner`] as if `T` had claimed `key_bytes` permanently (no
/// reservation deadline) — for integration tests that need a decodable owner
/// without running a real projection. Reachable at
/// `armdb::testing::unique_owner_for`; not part of the public API.
#[doc(hidden)]
pub fn unique_owner_for<T: crate::CollectionMeta>(key_bytes: &[u8]) -> UniqueOwner {
    UniqueOwner::new(collection_tag(T::NAME), key_bytes, 0)
        .expect("unique_owner_for: key_bytes must fit UNIQUE_OWNER_KEY_CAP")
}

/// Build a row that fails [`UniqueRow::is_well_formed`] — `key_len` set past
/// its own key buffer. The schema validator's corrupt-row test needs one:
/// `UniqueIndex` has no public write path at all, so there is no way to
/// reach this state other than constructing the bytes directly (the same way
/// on-disk corruption would produce them). Field-level access to
/// [`UniqueRow`] is private to this module, so this — unlike
/// [`unique_owner_for`], which composes public-enough pieces — has to live
/// here. Reachable at `armdb::testing::corrupt_unique_row`; not part of the
/// public API.
///
/// `#[cfg(any(test, feature = "test-faults"))]` (T10 review I1): the row this
/// builds is deliberately malformed — pairing it with [`force_put`](super::force_put)
/// is how a test constructs a namespace state no live commit could ever
/// produce. Neither belongs in an ordinary build: `force_put`/
/// `corrupt_unique_row` together are the *only* bypass of "a namespace has no
/// public write API", so shipping them un-gated would leave exactly the
/// compiling violation the type's own premise rules out.
#[cfg(any(test, feature = "test-faults"))]
#[doc(hidden)]
pub fn corrupt_unique_row() -> UniqueRow {
    let mut row = UniqueRow::new(UniqueOwner::new(1, b"x", 0).expect("fits UNIQUE_OWNER_KEY_CAP"));
    row.key_len = (UNIQUE_OWNER_KEY_CAP + 1) as u8;
    row
}

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

    #[test]
    fn row_size_is_pinned() {
        assert_eq!(size_of::<UniqueRow>(), 80);
        assert_eq!(UNIQUE_OWNER_KEY_CAP, 56);
    }

    #[test]
    fn row_owner_roundtrip_preserves_fields() {
        let owner = UniqueOwner::new(0xdead_beef, b"hello", 42).unwrap();
        let row = UniqueRow::new(owner);
        assert!(row.is_well_formed());
        assert_eq!(row.deadline(), Some(42));
        let back = row.owner();
        assert_eq!(back, owner);
        assert!(back.same_holder(&owner));
    }

    #[test]
    fn permanent_row_has_no_deadline() {
        let owner = UniqueOwner::new(1, b"k", 0).unwrap();
        let row = UniqueRow::new(owner);
        assert_eq!(row.deadline(), None);
        assert!(!row.owner().is_reservation());
    }

    #[test]
    fn owner_new_rejects_oversized_key() {
        let key = vec![0u8; UNIQUE_OWNER_KEY_CAP + 1];
        assert!(UniqueOwner::new(1, &key, 0).is_none());
    }

    #[test]
    fn is_well_formed_rejects_out_of_range_key_len() {
        let owner = UniqueOwner::new(1, b"k", 0).unwrap();
        let mut row = UniqueRow::new(owner);
        row.key_len = (UNIQUE_OWNER_KEY_CAP + 1) as u8;
        assert!(!row.is_well_formed());
    }

    #[test]
    fn corrupt_unique_row_is_actually_corrupt() {
        assert!(!corrupt_unique_row().is_well_formed());
    }

    #[test]
    fn with_deadline_changes_only_the_deadline() {
        let owner = UniqueOwner::new(7, b"hello", 0).unwrap();
        let reserved = owner.with_deadline(42);
        assert_eq!(reserved.expires_at(), Some(42));
        assert!(reserved.is_reservation());
        assert!(
            reserved.same_holder(&owner),
            "a reservation is still the same holder as the permanent owner it came from"
        );
        assert_eq!(reserved.collection_id(), owner.collection_id());
    }

    #[test]
    fn collection_tag_is_const_evaluable_and_stable() {
        const TAG: u64 = collection_tag("usernames");
        assert_eq!(TAG, collection_tag("usernames"));
        assert_ne!(TAG, collection_tag("other"));
    }

    /// T10 review F1: `belongs_to` must not perform a typed, potentially
    /// unaligned read out of `self.key`. `OverAlignedKey` has an alignment
    /// (16) greater than `UniqueOwner`'s own (8), so a `ptr::read::<Self>()`
    /// off `key_bytes()` would very likely read from a misaligned address —
    /// this exercises exactly the path the fix moved off that read.
    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        Eq,
        zerocopy::FromBytes,
        zerocopy::IntoBytes,
        zerocopy::Immutable,
        zerocopy::KnownLayout,
    )]
    #[repr(C, align(16))]
    struct OverAlignedKey {
        bytes: [u8; 16],
    }
    crate::impl_key_zerocopy!(
        OverAlignedKey,
        armour_core::KeyScheme::Typed(&[armour_core::KeyType::Array(16)])
    );

    struct OverAlignedOwner;
    impl armour_core::GetType for OverAlignedOwner {
        const TYPE: armour_core::Typ = armour_core::Typ::Custom("over_aligned_owner_test", &[]);
    }
    impl crate::CollectionMeta for OverAlignedOwner {
        type SelfId = OverAlignedKey;
        const NAME: &'static str = "over_aligned_owner_test";
    }

    #[test]
    fn belongs_to_decodes_an_over_aligned_key_without_unaligned_read() {
        assert!(align_of::<OverAlignedKey>() > align_of::<UniqueOwner>());

        let key = OverAlignedKey { bytes: [7u8; 16] };
        let owner = UniqueOwner::new(
            collection_tag(OverAlignedOwner::NAME),
            crate::Key::as_bytes(&key),
            0,
        )
        .unwrap();

        let decoded = owner
            .belongs_to::<OverAlignedOwner>()
            .expect("tag and length match");
        assert_eq!(decoded, key);
    }

    #[test]
    fn belongs_to_rejects_wrong_tag_and_wrong_length() {
        let key = OverAlignedKey { bytes: [1u8; 16] };
        let wrong_tag = UniqueOwner::new(
            collection_tag("someone_else"),
            crate::Key::as_bytes(&key),
            0,
        )
        .unwrap();
        assert!(wrong_tag.belongs_to::<OverAlignedOwner>().is_none());

        let wrong_len =
            UniqueOwner::new(collection_tag(OverAlignedOwner::NAME), b"short", 0).unwrap();
        assert!(wrong_len.belongs_to::<OverAlignedOwner>().is_none());
    }
}