Skip to main content

beam/
utils.rs

1//! Utility functions and data structures for BEAM.
2//!
3//! This module provides:
4//!
5//! - [`random_string`] — a cryptographically random alphanumeric string
6//!   generator used for message IDs and actor addresses.
7//! - [`BoundedHashMap`] — a fixed-capacity FIFO-eviction map used for
8//!   deduplication tracking (seen message IDs, etc.).
9//!
10//! ## Security Note
11//!
12//! [`random_string`] uses [`rand::thread_rng`] which is backed by the OS
13//! CSPRNG. This is suitable for message IDs and session tokens but is
14//! **not** suitable for cryptographic key generation — use the [`crate::sea`]
15//! module's `generate_pair()` for key generation.
16
17use rand::distr::Alphanumeric;
18use rand::{Rng, rng};
19use std::collections::VecDeque;
20use std::hash::BuildHasherDefault;
21
22/// Default hasher for non-cryptographic HashMaps/HashSets in BEAM.
23///
24/// Uses FxHash — the same hasher used by the Rust compiler itself.
25/// FxHash is ~3-4× faster than SipHash (std default) but is NOT
26/// HashDoS-resistant. This is acceptable for BEAM's P2P relay context
27/// where peers are semi-trusted (must connect via WebSocket first).
28///
29/// For cryptographic contexts (e.g. `sea::session`), use std's default
30/// `HashMap` with `RandomState` (SipHash) instead.
31pub type DefaultHasher = BuildHasherDefault<rustc_hash::FxHasher>;
32
33// Re-export FxHashMap and FxHashSet for convenient use across all BEAM modules.
34// These use FxHash (non-cryptographic, fast) — see `DefaultHasher` docs.
35pub use rustc_hash::{FxHashMap, FxHashSet};
36
37/// Generates a random alphanumeric string of the given length.
38///
39/// Uses [`rand::thread_rng`] (OS CSPRNG) and the [`Alphanumeric`]
40/// distribution (a–z, A–Z, 0–9). Each character provides ~5.95 bits of
41/// entropy.
42///
43/// # Example
44///
45/// ```ignore
46/// // Module is crate-private; use from within beam.
47/// let id = beam::utils::random_string(32);
48/// assert_eq!(id.len(), 32);
49/// ```
50pub fn random_string(len: usize) -> String {
51    rng()
52        .sample_iter(&Alphanumeric)
53        .take(len)
54        .map(char::from)
55        .collect()
56}
57
58/// A fixed-capacity hash map that evicts the oldest entries when full.
59///
60/// When the map is at capacity, each `insert` pushes out the oldest entry
61/// (FIFO eviction order). This is used to track recently-seen message IDs
62/// for deduplication, preventing unbounded memory growth in long-running
63/// nodes.
64///
65/// # Example
66///
67/// ```ignore
68/// // Module is crate-private; use from within beam.
69/// use beam::utils::BoundedHashMap;
70/// let mut map = BoundedHashMap::new(2);
71/// map.insert("a", 1);
72/// map.insert("b", 2);
73/// ```
74pub struct BoundedHashMap<K, V, S = DefaultHasher> {
75    map: FxHashMap<K, V>,
76    queue: VecDeque<K>,
77    max_entries: usize,
78    _marker: std::marker::PhantomData<S>,
79}
80
81impl<K: Clone + std::hash::Hash + std::cmp::Eq, V> BoundedHashMap<K, V> {
82    /// Creates a new `BoundedHashMap` with the given maximum capacity.
83    ///
84    /// Uses the default FxHash hasher for non-cryptographic hashing.
85    ///
86    /// # Panics
87    ///
88    /// Does not panic; a capacity of 0 will simply evict on every insert.
89    pub fn new(max_entries: usize) -> Self {
90        BoundedHashMap {
91            map: FxHashMap::default(),
92            queue: VecDeque::new(),
93            max_entries,
94            _marker: std::marker::PhantomData,
95        }
96    }
97
98    /// Inserts a key-value pair, evicting the oldest entry if at capacity.
99    ///
100    /// If the key already exists, the value is updated in place and the
101    /// eviction queue is not modified (the key's position is preserved).
102    /// If capacity is 0, the insert is silently dropped.
103    pub fn insert(&mut self, key: K, value: V) {
104        if self.max_entries == 0 {
105            return;
106        }
107        if self.queue.len() >= self.max_entries {
108            if let Some(removed) = self.queue.pop_back() {
109                self.map.remove(&removed);
110            }
111        }
112        if !self.map.contains_key(&key) {
113            self.queue.push_front(key.clone());
114        }
115        self.map.insert(key, value);
116    }
117
118    /// Returns a mutable reference to the value for the given key, or `None`.
119    pub fn get_mut(&mut self, key: &K) -> Option<&mut V> {
120        self.map.get_mut(key)
121    }
122
123    /// Returns a reference to the value for the given key, or `None`.
124    #[allow(dead_code)] // Public API for Step 3 (router); module not yet `pub`
125    pub fn get(&self, key: &K) -> Option<&V> {
126        self.map.get(key)
127    }
128
129    /// Returns the number of entries currently stored.
130    #[allow(dead_code)] // Public API for Step 3 (router); module not yet `pub`
131    pub fn len(&self) -> usize {
132        self.map.len()
133    }
134
135    /// Returns `true` if the map contains no entries.
136    #[allow(dead_code)] // Public API for Step 3 (router); module not yet `pub`
137    pub fn is_empty(&self) -> bool {
138        self.map.is_empty()
139    }
140
141    /// Returns the maximum number of entries before eviction begins.
142    #[allow(dead_code)] // Public API for Step 3 (router); module not yet `pub`
143    pub fn capacity(&self) -> usize {
144        self.max_entries
145    }
146
147    /// Removes and returns the value for the given key, or `None`.
148    ///
149    /// Also removes the key from the eviction queue to prevent it from being
150    /// re-inserted as a stale entry on the next FIFO eviction. If you re-insert
151    /// the same key later, it goes to the front of the queue (most-recently-used).
152    pub fn take(&mut self, key: &K) -> Option<V> {
153        self.queue.retain(|k| k != key);
154        self.map.remove(key)
155    }
156
157    /// Iterator over all (key, value) pairs.
158    ///
159    /// Used by periodic cleanup tasks (e.g., the quorum reaper) that need to
160    /// scan all entries for expiration. Order is unspecified — typically the
161    /// `HashMap`'s random iteration order. For FIFO-scoped iteration, callers
162    /// should combine with `take()` to evict expired entries.
163    ///
164    /// # Examples
165    ///
166    /// ```ignore
167    /// for (key, value) in map.iter() {
168    ///     if should_evict(&value) {
169    ///         map.take(&key);
170    ///     }
171    /// }
172    /// ```
173    pub fn iter(&self) -> impl Iterator<Item = (&K, &V)> {
174        self.map.iter()
175    }
176}
177
178impl<K: Clone + std::hash::Hash + std::cmp::Eq, V> Default for BoundedHashMap<K, V> {
179    fn default() -> Self {
180        Self::new(1024)
181    }
182}
183
184// === Fire-and-forget observability (Follow-up B) ==========================
185
186use crate::message::Message;
187use crate::metrics::Metrics;
188
189/// Try to send a message to an actor address. On full mailbox, increment
190/// the [`Metrics::dropped_sends`] counter and log at debug level instead
191/// of silently dropping the error.
192///
193/// This is the canonical BEAM pattern for fire-and-forget sends. It
194/// converts the previously invisible `let _ = addr.send(msg)` pattern
195/// into observable behavior without introducing a new abstraction layer.
196///
197/// # When to use
198///
199/// Use `try_send_or_log` when:
200///
201/// - The caller does **not** need ack confirmation
202/// - The caller can tolerate message loss under actor back-pressure
203/// - Observability of dropped messages is desirable (production telemetry)
204///
205/// Use `addr.send(msg).expect(...)` (or pattern-match on `Result`) when:
206///
207/// - The caller requires delivery confirmation
208/// - The caller can handle back-pressure by retrying or propagating the error
209/// - A silent drop would cause data loss (e.g. critical storage writes)
210///
211/// # Performance
212///
213/// On success this is a single `Addr::send` call. On failure it adds one
214/// atomic increment and one `tracing::debug!` event — both lock-free and
215/// negligible cost relative to the message-send attempt itself.
216///
217/// [`Metrics::dropped_sends`]: crate::metrics::Metrics::dropped_sends
218pub(crate) fn try_send_or_log(
219    addr: &crate::actor::Addr,
220    msg: Message,
221    metrics: &Metrics,
222    ctx: &'static str,
223) {
224    if addr.send(msg).is_err() {
225        metrics.record_dropped_send();
226        log::debug!(target: "beam::send", "actor mailbox full or closed, dropped message (context={})", ctx);
227    }
228}
229
230#[cfg(test)]
231mod tests {
232    use super::*;
233
234    // ── random_string ──
235
236    #[test]
237    fn test_random_string_length() {
238        let s = random_string(32);
239        assert_eq!(s.len(), 32);
240    }
241
242    #[test]
243    fn test_random_string_empty() {
244        let s = random_string(0);
245        assert_eq!(s.len(), 0);
246    }
247
248    #[test]
249    fn test_random_string_alphanumeric() {
250        let s = random_string(100);
251        assert!(s.chars().all(|c| c.is_ascii_alphanumeric()));
252    }
253
254    #[test]
255    fn test_random_string_uniqueness() {
256        let a = random_string(32);
257        let b = random_string(32);
258        // Astronomically unlikely to collide for 32 chars
259        assert_ne!(a, b);
260    }
261
262    // ── BoundedHashMap ──
263
264    #[test]
265    fn test_bounded_insert_and_get() {
266        let mut map = BoundedHashMap::new(10);
267        map.insert("a", 1);
268        map.insert("b", 2);
269        assert_eq!(map.get(&"a"), Some(&1));
270        assert_eq!(map.get(&"b"), Some(&2));
271        assert_eq!(map.len(), 2);
272    }
273
274    #[test]
275    fn test_bounded_eviction_fifo() {
276        let mut map = BoundedHashMap::new(2);
277        map.insert("a", 1);
278        map.insert("b", 2);
279        assert_eq!(map.len(), 2);
280        map.insert("c", 3); // should evict "a" (oldest)
281        assert_eq!(map.get(&"a"), None);
282        assert_eq!(map.get(&"b"), Some(&2));
283        assert_eq!(map.get(&"c"), Some(&3));
284        assert_eq!(map.len(), 2);
285    }
286
287    #[test]
288    fn test_bounded_update_existing_key() {
289        let mut map = BoundedHashMap::new(2);
290        map.insert("a", 1);
291        map.insert("a", 99);
292        assert_eq!(map.get(&"a"), Some(&99));
293        assert_eq!(map.len(), 1);
294    }
295
296    #[test]
297    fn test_bounded_get_mut() {
298        let mut map = BoundedHashMap::new(10);
299        map.insert("a", 1);
300        if let Some(v) = map.get_mut(&"a") {
301            *v = 42;
302        }
303        assert_eq!(map.get(&"a"), Some(&42));
304    }
305
306    #[test]
307    fn test_bounded_is_empty() {
308        let map: BoundedHashMap<&str, i32> = BoundedHashMap::new(10);
309        assert!(map.is_empty());
310    }
311
312    #[test]
313    fn test_bounded_capacity() {
314        let map: BoundedHashMap<&str, i32> = BoundedHashMap::new(42);
315        assert_eq!(map.capacity(), 42);
316    }
317
318    #[test]
319    fn test_bounded_default() {
320        let map: BoundedHashMap<&str, i32> = BoundedHashMap::default();
321        assert_eq!(map.capacity(), 1024);
322    }
323
324    #[test]
325    fn test_bounded_zero_capacity() {
326        let mut map = BoundedHashMap::new(0);
327        map.insert("a", 1);
328        assert_eq!(map.get(&"a"), None); // immediately evicted
329    }
330}
331
332// Unit tests for `try_send_or_log` deferred to Phase 3 e2e integration
333// tests in tests/send_metrics_e2e.rs. The helper is a thin wrapper around
334// `Addr::send` + metric increment — full coverage via realistic actor
335// scenarios is more valuable than mocked unit tests.