net-mesh 0.35.0

//! Generic in-memory state for a [`Fold<K>`](super::Fold).
//!
//! The fold runtime is parameterized by a single `FoldKind` trait
//! implementor (capability / routing / reservation / ...); this
//! module hosts the runtime-shared data structures: the per-key
//! entry record, the key→entry primary store, the node_id→keys
//! reverse index used by [`super::Fold::evict_node`], the merge
//! action enum that `FoldKind::merge` returns, the transition
//! enum that drives audit emission, and the [`FoldIndex`] trait
//! domain-specific secondary indices implement.
//!
//! Nothing in this module knows anything about wire format,
//! signature verification, channels, or audit chains — those
//! belong to the dispatch layer and the runtime layer
//! ([`super`]).

use std::collections::{HashMap, HashSet};
use std::time::Instant;

use super::wire::SignedAnnouncement;
use super::FoldKind;

/// Publisher's routing-layer identity, matching
/// [`behavior::placement::NodeId`](super::super::placement::NodeId).
/// The fold layer indexes by this `u64` rather than the 32-byte
/// cryptographic node identity because every query surface
/// (capability, routing, reservation) addresses nodes by their
/// routing id, and the wire envelope already commits a separate
/// [`SignedAnnouncement::signature`] to the publisher's
/// cryptographic identity.
pub type NodeId = u64;

/// Fast multiplicative mixer for hash keys built entirely out of
/// `u64`s that are already well-distributed — fold ids
/// ([`NodeId`], [`IslandId`](super::IslandId)) and the capability
/// index's `(u64, u64)` keys.
///
/// Those ids are derived from already-hashed identity bytes, so
/// collision resistance exists at construction and SipHash's DoS
/// resistance adds nothing — it just charges ~15-25 ns of mixing per
/// probe (PERF_AUDIT §4.6, PERF_AUDIT_2026_07_31_GANG_SCHEDULER §7).
/// What makes that worth removing is the probe *count*: the gang
/// matcher's `HostedByAny` scan probes once per topology entry, not
/// once per candidate host.
///
/// # What a key type must satisfy to use this
///
/// **Well-distributed in its LOW bits specifically** — not merely
/// collision-free. There is no finalizer: for the single-write case
/// (which is every real use here), [`Hasher::finish`](std::hash::Hasher::finish) returns
/// `v.wrapping_mul(FX_SEED)`, and multiplication only propagates
/// entropy *upward* — bit `k` of the product depends only on bits
/// `0..=k` of the input. `hashbrown` derives the bucket index from
/// the low bits of the hash, so low-bit quality of the *input* is
/// what carries the whole table.
///
/// That holds for the ids here because they are already digests
/// (`NodeId` from identity bytes, `IslandId` = `hash(host, domain)`),
/// where every bit is equidistributed. It would NOT hold for a
/// counter, a left-shifted composite, a pointer, or anything with
/// structural zeroes low down — those want a finalizer or a
/// different hasher. This is the same trade `rustc-hash` makes, and
/// the same caveat applies.
///
/// **One implementation, two aliases.** [`BuildU64Hasher`] here and
/// `capability::BuildU64TupleHasher` both build this type; the
/// arity lives in the alias names, not in the mixer, which only ever
/// sees a sequence of `write_u64` calls. They were briefly two
/// byte-identical copies — same constant, same fallback — which meant
/// a correction to one (this note being the obvious candidate) would
/// silently miss the other. Per-site rationale belongs on the aliases.
#[derive(Default, Clone)]
pub struct FxU64Hasher(u64);

impl std::hash::Hasher for FxU64Hasher {
    #[inline]
    fn finish(&self) -> u64 {
        self.0
    }

    #[inline]
    fn write_u64(&mut self, v: u64) {
        // FxHash-style step: rotate, xor, multiply by a large odd
        // constant. Well-distributed for already-hashed input.
        const FX_SEED: u64 = 0x51_7c_c1_b7_27_22_0a_95;
        self.0 = (self.0.rotate_left(5) ^ v).wrapping_mul(FX_SEED);
    }

    /// Defensive byte fallback — `Hash for u64` calls `write_u64`
    /// directly, but routing an unexpected key type through here
    /// must still mix rather than silently collapse.
    ///
    /// Two properties a caller arriving here should know, both
    /// harmless for the `u64`-keyed sets this serves and neither
    /// reachable from them:
    ///
    /// - An **empty slice** mixes nothing — `chunks(8)` yields no
    ///   chunks, so the state stays at its `Default` of 0.
    /// - `write(&[0u8])` produces exactly the state `write_u64(0)`
    ///   does, because the short chunk is zero-padded. So this
    ///   fallback does not domain-separate by length, and a key type
    ///   that mixes byte writes with `u64` writes could collide
    ///   across the two.
    #[inline]
    fn write(&mut self, bytes: &[u8]) {
        for chunk in bytes.chunks(8) {
            let mut buf = [0u8; 8];
            buf[..chunk.len()].copy_from_slice(chunk);
            self.write_u64(u64::from_le_bytes(buf));
        }
    }
}

/// [`BuildHasher`](std::hash::BuildHasher) for [`FxU64Hasher`] over
/// single-`u64` fold ids.
pub type BuildU64Hasher = std::hash::BuildHasherDefault<FxU64Hasher>;

/// A set of [`NodeId`]s hashed with [`FxU64Hasher`] — the
/// candidate-host set the gang matcher builds and then probes once per
/// topology entry.
pub type NodeIdSet = HashSet<NodeId, BuildU64Hasher>;

/// One entry in a fold: the payload most recently accepted for
/// its key, plus the bookkeeping the runtime needs to expire,
/// merge, and audit further announcements.
///
/// `K::Payload` is owned, not borrowed — folds are eventually
/// consistent state caches, not view layers over a foreign
/// authority.
#[derive(Debug, Clone)]
pub struct FoldEntry<K: FoldKind> {
    /// Domain-specific payload accepted at this key.
    pub payload: K::Payload,
    /// Publisher of the announcement that produced this entry.
    /// Used to populate `state.by_node` for
    /// [`super::Fold::evict_node`] and to gate owner-only
    /// transitions in folds that enforce per-publisher state
    /// machines (e.g. [`super::ReservationFold`]).
    pub node_id: NodeId,
    /// Monotonic counter per `(node_id, kind, class)`, copied
    /// from the announcement. The default [`FoldKind::merge`]
    /// rejects any incoming announcement whose generation is
    /// `<=` the stored generation — this is the wire-level
    /// anti-reorder mechanism.
    pub generation: u64,
    /// Wall-clock instant at which the runtime accepted the
    /// announcement that produced this entry. Used by metrics +
    /// snapshot diagnostics; NOT used for expiry (see
    /// `expires_at`).
    pub received_at: Instant,
    /// Wall-clock instant at which this entry becomes stale.
    /// Computed at apply time as
    /// `received_at + ann.ttl_secs.unwrap_or(K::DEFAULT_TTL)`.
    /// The background expiry sweeper removes entries past this
    /// time.
    pub expires_at: Instant,
}

/// In-memory store backing a single [`Fold<K>`](super::Fold).
///
/// Public fields are read by [`FoldKind::query`] (and by tests),
/// but mutation flows exclusively through
/// [`super::Fold::apply`] / [`super::Fold::evict_node`] /
/// [`super::Fold::restore`] so the [`super::FoldMetrics`] counters
/// and `by_node` reverse index stay coherent with `entries`.
///
/// The container is held inside an `RwLock` on the
/// [`Fold<K>`](super::Fold) struct; this type is purely the data
/// shape, not the synchronization primitive.
#[derive(Debug)]
pub struct FoldState<K: FoldKind> {
    /// Primary store: `K::Key → FoldEntry<K>`. The
    /// [`FoldKind::key_for`] function is the only sanctioned
    /// way to derive keys from announcements; the apply path
    /// uses it to look up + replace existing entries.
    pub entries: HashMap<K::Key, FoldEntry<K>>,
    /// Reverse index: `node_id → keys it owns`. Populated on
    /// every accepted apply; consulted on
    /// [`super::Fold::evict_node`] to drop every entry attached
    /// to a node in O(keys_for_that_node) instead of O(entries).
    /// At 50K-100K node scale (the plan's targeted operating
    /// range), the average node owns a handful of keys; the
    /// reverse index is the difference between "evict in
    /// microseconds" and "evict in seconds."
    pub by_node: HashMap<NodeId, HashSet<K::Key>>,
}

impl<K: FoldKind> FoldState<K> {
    /// Build an empty state.
    pub fn new() -> Self {
        Self {
            entries: HashMap::new(),
            by_node: HashMap::new(),
        }
    }

    /// Total entry count. Cheap O(1) read off the primary store.
    /// Mirrors what the [`super::FoldMetrics::entries`] gauge
    /// reports; tests and the [`super::Fold::snapshot`] header
    /// read it without acquiring the metrics layer.
    pub fn len(&self) -> usize {
        self.entries.len()
    }

    /// Whether the state is empty.
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    /// Look up the entry for `key`. Borrowed access; the caller
    /// already holds the state guard via
    /// [`FoldKind::query`]'s `state: &FoldState<Self>` parameter.
    pub fn get(&self, key: &K::Key) -> Option<&FoldEntry<K>> {
        self.entries.get(key)
    }
}

impl<K: FoldKind> Default for FoldState<K> {
    fn default() -> Self {
        Self::new()
    }
}

/// Verdict from [`FoldKind::merge`] for a new announcement
/// against the current state at its key. The runtime translates
/// the verdict into a concrete state mutation in
/// [`super::Fold::apply`].
///
/// Carries the announcement payload by reference on the runtime
/// side (the apply path passes `&SignedAnnouncement` into
/// `merge`); this enum is the *decision* shape, so it doesn't
/// embed the payload again — the runtime already has it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MergeAction {
    /// No existing entry at this key. Runtime inserts the
    /// announcement's payload as a fresh [`FoldEntry`].
    Insert,
    /// Existing entry is older / out-ranked. Runtime evicts the
    /// old entry (updating `by_node` for both old and new
    /// owners) and inserts the new payload.
    Replace,
    /// Existing entry wins. Runtime drops the announcement and
    /// bumps the rejected-applies metric.
    Reject,
}

/// Transition shape passed to [`FoldKind::audit_event`] when an
/// applied announcement produces an audit-worthy state change.
/// Per the plan's audit-integration section, the defaults emit
/// `FoldEntryCreated` / `FoldEntryReplaced` / `FoldEntryExpired`
/// / `FoldEntryEvicted` / `FoldEntryRejected`; fold authors
/// match on the variant they care about.
#[derive(Debug)]
pub enum EntryTransition<'a, K: FoldKind> {
    /// First-time insert at this key. `new` is the freshly-
    /// applied entry.
    Created {
        /// Key that received the new entry.
        key: &'a K::Key,
        /// The freshly-applied entry.
        new: &'a FoldEntry<K>,
    },
    /// Replacement at this key. Both `old` (about to be dropped)
    /// and `new` (about to be installed) are visible so audit
    /// records can carry generation deltas.
    Replaced {
        /// Key whose entry was replaced.
        key: &'a K::Key,
        /// Entry that was just evicted (the loser of the merge).
        old: &'a FoldEntry<K>,
        /// Entry that replaced it.
        new: &'a FoldEntry<K>,
    },
    /// Announcement was rejected per [`MergeAction::Reject`].
    /// `existing` is the entry that wins; `incoming` is the
    /// raw announcement that lost.
    Rejected {
        /// Key the rejected announcement targeted.
        key: &'a K::Key,
        /// Current entry at the key, if any — the merge winner.
        existing: Option<&'a FoldEntry<K>>,
        /// The losing announcement.
        incoming: &'a SignedAnnouncement<K::Payload>,
    },
    /// Entry was force-removed via [`super::Fold::evict_node`].
    /// `reason` is the operator-visible string for the audit
    /// record (e.g. "SWIM declared node dead").
    Evicted {
        /// Key whose entry was evicted.
        key: &'a K::Key,
        /// Entry that was removed.
        old: &'a FoldEntry<K>,
        /// Operator-supplied reason string for the audit log.
        reason: &'a str,
    },
    /// Entry was removed by the TTL sweeper because
    /// `expires_at < now`.
    Expired {
        /// Key whose entry expired.
        key: &'a K::Key,
        /// Entry that was removed.
        old: &'a FoldEntry<K>,
    },
}

/// Secondary index maintained alongside the primary
/// `key → entry` store. Domain-specific: capability uses a
/// tag-inverted lookup, reservation uses a "currently free" set,
/// routing uses no extra index (uses the primary store
/// directly).
///
/// The runtime calls `on_insert` / `on_remove` on every accepted
/// apply, before / after the primary-store mutation respectively
/// so the index sees the same `(key, payload)` shape the entry
/// is built from. [`FoldKind::query`] reads the index by
/// reference; it does NOT mutate.
pub trait FoldIndex<K: FoldKind>: Send + Sync {
    /// Called after an [`MergeAction::Insert`] or
    /// [`MergeAction::Replace`] commits to the primary store.
    /// For `Replace`, the previous payload was already passed
    /// to [`Self::on_remove`].
    fn on_insert(&mut self, key: &K::Key, payload: &K::Payload);

    /// Called before an [`MergeAction::Replace`] or an
    /// [`super::Fold::evict_node`] eviction drops the entry
    /// from the primary store, with the payload that's about
    /// to be removed.
    fn on_remove(&mut self, key: &K::Key, payload: &K::Payload);

    /// Drop every cached relation. Called by
    /// [`super::Fold::restore`] before re-populating from a
    /// snapshot.
    fn clear(&mut self);

    /// Returns `true` when the two payloads index identically —
    /// i.e. [`Self::on_remove`] + [`Self::on_insert`] against
    /// these two payloads would net to a no-op on every
    /// dimension this index maintains.
    ///
    /// The runtime's `MergeAction::Replace` arm consults this
    /// before paying the index churn: when an announcement
    /// refreshes generation/TTL without changing the tags /
    /// region / state the index keys on (the steady-state
    /// republish case), the index dance is pure waste. Per
    /// PERF_AUDIT §4.5 — pre-fix the refresh always re-walked
    /// every tag bucket, re-derived synthetic indexes, and
    /// re-allocated the `entry().or_default()` HashSets even
    /// when nothing changed, all under the writer lock.
    ///
    /// Default `false` keeps the safe pre-fix behavior for
    /// indexes that don't implement a content-aware equality
    /// check.
    fn index_payload_equivalent(_old: &K::Payload, _new: &K::Payload) -> bool {
        false
    }
}

/// Default no-op secondary index. Folds that don't need a
/// secondary lookup use this as their `K::Index` so the runtime
/// still has a uniformly-typed hook to call.
#[derive(Debug, Default)]
pub struct NoIndex;

impl<K: FoldKind> FoldIndex<K> for NoIndex {
    fn on_insert(&mut self, _key: &K::Key, _payload: &K::Payload) {}
    fn on_remove(&mut self, _key: &K::Key, _payload: &K::Payload) {}
    fn clear(&mut self) {}
}

/// Outcome of a single [`super::Fold::apply`] call. Mirrors
/// [`MergeAction`] but carries the entry that produced the
/// audit event (if any) so the runtime can hand it to
/// [`FoldKind::audit_event`] without re-locking.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApplyOutcome {
    /// New entry was created at the key.
    Inserted,
    /// Existing entry was replaced.
    Replaced,
    /// Existing entry wins; announcement dropped.
    Rejected,
}

/// Errors the runtime returns from the apply / snapshot path.
/// Dispatch-layer errors (bad signature, unknown kind) flow
/// through [`super::WireError`] / [`super::DispatchError`]
/// instead.
#[derive(Debug, thiserror::Error)]
pub enum FoldError {
    /// Apply rejected because the announcement's generation is
    /// `0`, which the wire format reserves as the "uninitialized"
    /// sentinel. A legitimate publisher always starts at `1`.
    #[error("invalid generation 0 from publisher {node_id}")]
    InvalidGeneration {
        /// Publisher whose announcement carried generation 0.
        node_id: NodeId,
    },
    /// Restore was called on a non-empty fold without the
    /// `force` flag. The runtime refuses to merge a snapshot
    /// over a live state — operators who really want this pass
    /// `force: true` to [`super::Fold::restore`].
    #[error("restore refused: fold is non-empty (len={current_len})")]
    RestoreOverLiveState {
        /// Current entry count of the live fold.
        current_len: usize,
    },
}