clawft_kernel/node_facts/
cache.rs

1//! TTL cache of verified node facts, held by
2//! [`crate::cluster::ClusterMembership`] (ADR-099 section 2).
3//!
4//! Only verified facts enter. An entry is served while its facts are
5//! within TTL; expired entries are invisible to readers and removed by
6//! [`NodeFactsCache::evict_expired`]. A node's facts are replaced only by a
7//! higher `seq` (or the same `seq` re-issued with identical content, which
8//! refreshes nothing), so a replayed older block cannot roll a node back.
9//! Deltas must be signed by the same key as the cached base, name its
10//! `seq`, and carry a strictly increasing delta `seq`.
11
12use dashmap::DashMap;
13
14use clawft_types::placement::{
15    Capability, FactsError, NodeFacts, NodeLoad, Provenance, TrustTier,
16};
17
18use crate::node_facts_advert::{
19    NodeFactsAdvertError, SignedFactsDelta, SignedNodeFacts, verify_facts_delta, verify_node_facts,
20};
21
22/// Most nodes held.
23pub const MAX_CACHED_NODES: usize = 4096;
24/// Most mesh-derived `Discovered` entries held at once, whatever the total.
25pub const MAX_DISCOVERED: usize = 512;
26
27/// Test hook: pause an insert of `(node_id, seq)` after its staleness check
28/// has passed and before it writes, so a test can force the interleave.
29#[cfg(test)]
30pub(crate) mod pause {
31    use std::sync::Mutex;
32    use std::sync::atomic::{AtomicBool, Ordering};
33    pub(crate) static TARGET: Mutex<Option<(String, u64)>> = Mutex::new(None);
34    pub(crate) static PAUSED: AtomicBool = AtomicBool::new(false);
35    pub(crate) static RELEASE: AtomicBool = AtomicBool::new(false);
36    pub(crate) fn here(node_id: &str, seq: u64) {
37        let hit = TARGET.lock().unwrap().as_ref().is_some_and(|(n, s)| n == node_id && *s == seq);
38        if hit {
39            PAUSED.store(true, Ordering::SeqCst);
40            while !RELEASE.load(Ordering::SeqCst) {
41                std::thread::sleep(std::time::Duration::from_millis(1));
42            }
43        }
44    }
45}
46
47/// Where a node's tier came from. Only an operator-set tier is sticky: a
48/// mesh-derived tier follows the current connection (admission) every frame.
49#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
50#[serde(rename_all = "snake_case")]
51pub enum TierSource {
52    /// Derived from how the node's connection was admitted.
53    Mesh,
54    /// Set by the operator (or the local node itself).
55    Operator,
56}
57
58/// One cached, verified facts block plus what the receiver knows about it.
59#[derive(Debug, Clone, PartialEq)]
60pub struct CachedNodeFacts {
61    /// Current facts (base plus applied deltas).
62    pub facts: NodeFacts,
63    /// The signed base as received (re-shareable, re-verifiable).
64    pub signed: SignedNodeFacts,
65    /// Trust tier assigned by this node.
66    pub trust_tier: TrustTier,
67    /// Where the tier came from.
68    pub tier_source: TierSource,
69    /// Connection that supplied the facts (0 = none), for per-connection quotas.
70    pub origin: u64,
71    /// When this node accepted the base, unix seconds.
72    pub received_at: u64,
73    /// Last applied delta `seq` (0 = none).
74    pub delta_seq: u64,
75    /// Signing key of the node.
76    pub public_key: [u8; 32],
77}
78
79impl CachedNodeFacts {
80    /// Node id.
81    pub fn node_id(&self) -> &str {
82        self.facts.node_id()
83    }
84    /// Advertised capabilities.
85    pub fn capabilities(&self) -> &[Capability] {
86        self.facts.capabilities()
87    }
88    /// Receiver-assigned trust tier.
89    pub fn trust_tier(&self) -> TrustTier {
90        self.trust_tier
91    }
92    /// Busy count and free memory.
93    pub fn load(&self) -> NodeLoad {
94        self.facts.load()
95    }
96    /// Facts still within TTL at `now`.
97    pub fn is_fresh(&self, now: u64) -> bool {
98        self.facts.is_fresh(now)
99    }
100}
101
102/// What an insert did.
103#[derive(Debug, Clone, Copy, PartialEq, Eq)]
104pub enum InsertOutcome {
105    /// First facts for this node.
106    Added,
107    /// Replaced older facts.
108    Replaced,
109    /// Same `seq` and content as held; nothing changed.
110    Unchanged,
111}
112
113/// Why the cache refused something.
114#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
115pub enum CacheError {
116    /// Signature, binding, structure or TTL failure.
117    #[error(transparent)]
118    Verify(#[from] NodeFactsAdvertError),
119    /// Older (or conflicting same-`seq`) facts than those held.
120    #[error("stale facts for {node_id}: seq {got} <= held {held}")]
121    Stale {
122        /// Node.
123        node_id: String,
124        /// Offered seq.
125        got: u64,
126        /// Held seq.
127        held: u64,
128    },
129    /// No fresh base for a delta.
130    #[error("no fresh facts cached for {0}")]
131    UnknownNode(String),
132    /// Delta signed by a different key than the base.
133    #[error("delta key does not match the cached facts key")]
134    KeyMismatch,
135    /// Cache at [`MAX_CACHED_NODES`].
136    #[error("node facts cache full")]
137    Full,
138}
139
140/// Verified node facts with TTL.
141#[derive(Debug, Default)]
142pub struct NodeFactsCache {
143    entries: DashMap<String, CachedNodeFacts>,
144}
145
146impl NodeFactsCache {
147    /// Empty cache.
148    pub fn new() -> Self {
149        Self::default()
150    }
151
152    /// Verify and cache a signed facts block with a receiver-assigned tier.
153    pub fn insert(
154        &self,
155        signed: SignedNodeFacts,
156        trust_tier: TrustTier,
157        now: u64,
158    ) -> Result<InsertOutcome, CacheError> {
159        self.insert_capped(signed, trust_tier, now, None, 0)
160    }
161
162    /// [`Self::insert`] for facts a *peer* sent. Every capability's
163    /// provenance is capped at `max_provenance` in the cached copy: the
164    /// receiver did not probe or measure remote data, so a peer's
165    /// `measured` claim is never held as `measured`. The signed envelope is
166    /// kept as received.
167    ///
168    /// `mesh_tier` is derived from the connection that sent the facts and is
169    /// re-derived on every frame; a tier the operator set is kept. `origin`
170    /// names the connection (see [`Self::discovered_from`]). When the cache
171    /// is full, the oldest mesh-derived `Discovered` entry makes room before
172    /// the insert is refused, so `Paired` and `Pinned` entries are never
173    /// pushed out by `Discovered` ones.
174    pub fn insert_remote(
175        &self,
176        signed: SignedNodeFacts,
177        mesh_tier: TrustTier,
178        max_provenance: Provenance,
179        origin: u64,
180        now: u64,
181    ) -> Result<InsertOutcome, CacheError> {
182        self.insert_capped(signed, mesh_tier, now, Some(max_provenance), origin)
183    }
184
185    fn insert_capped(
186        &self,
187        signed: SignedNodeFacts,
188        trust_tier: TrustTier,
189        now: u64,
190        cap: Option<Provenance>,
191        origin: u64,
192    ) -> Result<InsertOutcome, CacheError> {
193        let remote = cap.is_some();
194        let mut facts = verify_node_facts(&signed, now)?;
195        if let Some(cap) = cap {
196            for c in &mut facts.capabilities {
197                c.provenance = c.provenance.min(cap);
198            }
199        }
200        let mut public_key = [0u8; 32];
201        public_key.copy_from_slice(&signed.public_key);
202        let node_id = facts.node_id.clone();
203        let mut entry = CachedNodeFacts {
204            facts,
205            signed,
206            trust_tier,
207            tier_source: if remote { TierSource::Mesh } else { TierSource::Operator },
208            origin,
209            received_at: now,
210            delta_seq: 0,
211            public_key,
212        };
213        // Capacity first, for a node not held yet (soft: a racing insert of
214        // another node may still land between this check and the write).
215        if !self.entries.contains_key(&node_id) {
216            // Unverified claims are capped as a class, so they cannot crowd
217            // out the cache; the heaviest connection pays first.
218            if remote
219                && entry.tier_source == TierSource::Mesh
220                && entry.trust_tier == TrustTier::Discovered
221                && self.count_discovered() >= MAX_DISCOVERED
222                && !self.evict_discovered_by_weight()
223            {
224                return Err(CacheError::Full);
225            }
226            if self.entries.len() >= MAX_CACHED_NODES {
227                self.evict_expired(now);
228                if self.entries.len() >= MAX_CACHED_NODES && !self.evict_discovered_by_weight() {
229                    return Err(CacheError::Full);
230                }
231            }
232        }
233        // The held-versus-new decision and the write happen under the entry's
234        // lock, so two concurrent inserts for one node cannot both pass the
235        // staleness check and let the older `seq` land last.
236        use dashmap::mapref::entry::Entry;
237        match self.entries.entry(node_id.clone()) {
238            Entry::Vacant(v) => {
239                v.insert(entry);
240                Ok(InsertOutcome::Added)
241            }
242            Entry::Occupied(mut o) => {
243                // Expired entries do not block a node that restarted its seq.
244                if o.get().is_fresh(now) {
245                    let held = o.get_mut();
246                    if remote && held.tier_source == TierSource::Operator {
247                        entry.trust_tier = held.trust_tier;
248                        entry.tier_source = TierSource::Operator;
249                    }
250                    if held.signed.payload == entry.signed.payload {
251                        if remote && held.tier_source == TierSource::Mesh {
252                            // Same facts, but the connection's tier is current.
253                            held.trust_tier = trust_tier;
254                            held.origin = origin;
255                        }
256                        return Ok(InsertOutcome::Unchanged);
257                    }
258                    if entry.facts.seq <= held.facts.seq {
259                        return Err(CacheError::Stale {
260                            node_id,
261                            got: entry.facts.seq,
262                            held: held.facts.seq,
263                        });
264                    }
265                }
266                #[cfg(test)]
267                pause::here(&node_id, entry.facts.seq);
268                o.insert(entry);
269                Ok(InsertOutcome::Replaced)
270            }
271        }
272    }
273
274    fn count_discovered(&self) -> usize {
275        self.entries
276            .iter()
277            .filter(|e| e.tier_source == TierSource::Mesh && e.trust_tier == TrustTier::Discovered)
278            .count()
279    }
280
281    /// Drop one mesh-derived `Discovered` entry, chosen by weight: the oldest
282    /// entry of the connection that holds the most of them, so a connection
283    /// that fills its quota loses its own entries before anyone else's, and a
284    /// peer with one entry outlives an attacker with several. (A connection
285    /// that reconnects gets a fresh id and a fresh quota; connections with
286    /// equally many entries lose the oldest first. There is no remote address
287    /// to charge: see the swarm doc.) False if there is none to drop.
288    fn evict_discovered_by_weight(&self) -> bool {
289        let mut by_origin: std::collections::HashMap<u64, (usize, u64, String)> = Default::default();
290        for e in self
291            .entries
292            .iter()
293            .filter(|e| e.tier_source == TierSource::Mesh && e.trust_tier == TrustTier::Discovered)
294        {
295            // Entries with no connection (origin 0) each count as their own group.
296            let group = if e.origin == 0 { u64::MAX - (by_origin.len() as u64) } else { e.origin };
297            let slot = by_origin
298                .entry(group)
299                .or_insert((0, u64::MAX, String::new()));
300            slot.0 += 1;
301            if (e.received_at, e.key().clone()) < (slot.1, slot.2.clone()) || slot.2.is_empty() {
302                slot.1 = e.received_at;
303                slot.2 = e.key().clone();
304            }
305        }
306        let victim = by_origin
307            .values()
308            .max_by(|a, b| a.0.cmp(&b.0).then(b.1.cmp(&a.1)))
309            .map(|v| v.2.clone());
310        match victim {
311            Some(k) => self.entries.remove(&k).is_some(),
312            None => false,
313        }
314    }
315
316    /// Fresh mesh-derived `Discovered` entries that connection `origin` supplied.
317    pub fn discovered_from(&self, origin: u64, now: u64) -> usize {
318        if origin == 0 {
319            return 0;
320        }
321        self.entries
322            .iter()
323            .filter(|e| {
324                e.origin == origin
325                    && e.tier_source == TierSource::Mesh
326                    && e.trust_tier == TrustTier::Discovered
327                    && e.is_fresh(now)
328            })
329            .count()
330    }
331
332    /// Lower a mesh-derived tier to `Discovered` (the node's admission was
333    /// revoked or its verified connection is gone). An operator-set tier is
334    /// left alone. True if the tier changed.
335    pub fn demote_mesh_tier(&self, node_id: &str) -> bool {
336        match self.entries.get_mut(node_id) {
337            Some(mut e)
338                if e.tier_source == TierSource::Mesh && e.trust_tier != TrustTier::Discovered =>
339            {
340                e.trust_tier = TrustTier::Discovered;
341                true
342            }
343            _ => false,
344        }
345    }
346
347    /// Verify a signed delta and apply it to the cached base.
348    pub fn apply_delta(&self, signed: &SignedFactsDelta, now: u64) -> Result<(), CacheError> {
349        let (delta, pk) = verify_facts_delta(signed, now)?;
350        let mut entry = self
351            .entries
352            .get_mut(&delta.node_id)
353            .filter(|e| e.is_fresh(now))
354            .ok_or_else(|| CacheError::UnknownNode(delta.node_id.clone()))?;
355        if entry.public_key != pk {
356            return Err(CacheError::KeyMismatch);
357        }
358        if delta.seq <= entry.delta_seq {
359            return Err(NodeFactsAdvertError::Facts(FactsError::Delta(format!(
360                "delta seq {} <= applied {}",
361                delta.seq, entry.delta_seq
362            )))
363            .into());
364        }
365        entry
366            .facts
367            .apply_delta(&delta)
368            .map_err(NodeFactsAdvertError::Facts)?;
369        entry.delta_seq = delta.seq;
370        Ok(())
371    }
372
373    /// Re-assign the receiver-side trust tier of a held node (an operator
374    /// changed it; the signed facts are unchanged). False if not held.
375    pub fn set_trust_tier(&self, node_id: &str, trust_tier: TrustTier) -> bool {
376        match self.entries.get_mut(node_id) {
377            Some(mut e) => {
378                e.trust_tier = trust_tier;
379                e.tier_source = TierSource::Operator;
380                true
381            }
382            None => false,
383        }
384    }
385
386    /// Fresh facts for one node.
387    pub fn get(&self, node_id: &str, now: u64) -> Option<CachedNodeFacts> {
388        self.entries
389            .get(node_id)
390            .filter(|e| e.is_fresh(now))
391            .map(|e| e.clone())
392    }
393
394    /// All fresh facts, sorted by node id.
395    pub fn list(&self, now: u64) -> Vec<CachedNodeFacts> {
396        let mut v: Vec<CachedNodeFacts> = self
397            .entries
398            .iter()
399            .filter(|e| e.is_fresh(now))
400            .map(|e| e.clone())
401            .collect();
402        v.sort_by(|a, b| a.node_id().cmp(b.node_id()));
403        v
404    }
405
406    /// Drop expired entries; returns how many were removed.
407    pub fn evict_expired(&self, now: u64) -> usize {
408        let before = self.entries.len();
409        self.entries.retain(|_, e| e.is_fresh(now));
410        before - self.entries.len()
411    }
412
413    /// Entries held (fresh or not yet evicted).
414    pub fn len(&self) -> usize {
415        self.entries.len()
416    }
417
418    /// True if nothing is held.
419    pub fn is_empty(&self) -> bool {
420        self.entries.is_empty()
421    }
422}
423
424#[cfg(test)]
425#[path = "cache_tests.rs"]
426mod tests;
clawft_kernel/node_facts/cache.rs

clawft_kernel/node_facts/
cache.rs
