maglev-hash 0.1.0

//! Maglev consistent hashing with top-K preference lists.
//!
//! This crate implements the lookup table construction algorithm from the
//! [Google Maglev paper][maglev], extended with a **top-K preference list**
//! for replica-aware routing.
//!
//! [maglev]: https://research.google/pubs/maglev-a-fast-and-reliable-software-network-load-balancer/
//!
//! # Maglev consistent hashing
//!
//! Maglev hashing assigns keys to nodes via a fixed-size lookup table whose
//! size is a prime number. Each node generates a permutation of table slots;
//! the table is filled round-robin from these permutations. This achieves:
//!
//! - **O(1) lookups** — a single hash indexes into the table.
//! - **Minimal disruption** — when a node is added or removed, only a small
//!   fraction of keys are remapped (close to the theoretical 1/n optimum).
//! - **Uniform distribution** — load is spread evenly across nodes.
//!
//! # Top-K preference lists (crate extension)
//!
//! **Note:** top-K preference lists are *not* part of the original Maglev
//! paper, which defines only a single-node lookup. This crate extends the
//! algorithm with `get_top_k()` and `is_match()` for replica-aware routing.
//!
//! `get_top_k()` returns an ordered preference list of up to K distinct
//! nodes for a given key by walking the lookup table from the hashed slot
//! and collecting unique nodes. The first element always agrees with
//! `get()`. This is useful for replica fallback: if the primary node is
//! unavailable, the caller can try the next node in the list.
//!
//! `is_match()` checks whether a specific node appears in the top-K
//! preference list for a key, useful for determining if a node should
//! accept a given request when running with replication.
//!
//! ## Tradeoffs vs. suffix hashing
//!
//! An alternative approach is to hash the key with different suffixes
//! (e.g., `get("key-0")`, `get("key-1")`) and deduplicate. Compared to
//! that approach, the table-walk method:
//!
//! - **Guarantees** K distinct nodes in a single pass (no dedup retries).
//! - **Preserves consistency**: `get_top_k(key, 1)[0] == get(key)`.
//! - **Couples ranks**: removing a node at rank 1 can cascade and shift
//!   ranks 2+, whereas suffix hashing gives each rank independent
//!   disruption.
//!
//! # Example
//!
//! ```
//! use maglev_hash::MaglevTable;
//!
//! let nodes = vec!["node-a".to_string(), "node-b".to_string(), "node-c".to_string()];
//! let table = MaglevTable::new(nodes);
//!
//! // Primary lookup — O(1)
//! let primary = table.get(&"my-key").unwrap();
//! println!("primary node: {primary}");
//!
//! // Top-2 preference list (first element == primary)
//! let top2 = table.get_top_k(&"my-key", 2);
//! assert_eq!(top2[0], primary);
//! println!("fallback node: {}", top2[1]);
//!
//! // Check if a specific node is in the top-2 for this key
//! let in_top2 = table.is_match(&"my-key", &"node-a".to_string(), 2);
//! println!("node-a in top-2: {in_top2}");
//! ```

#![warn(clippy::pedantic, missing_docs)]

use std::hash::{Hash, Hasher};

use fixedbitset::FixedBitSet;

/// Compact set of node indices for deduplication during table walks.
///
/// Uses an inline `u128` bitmask for up to 128 nodes (zero heap allocation),
/// falling back to a heap-allocated [`FixedBitSet`] for larger rings.
enum NodeSet {
    /// Inline bitmask — covers node indices 0..=127 with no allocation.
    Inline(u128),
    /// Heap-allocated bitset for rings with more than 128 nodes.
    Heap(FixedBitSet),
}

impl NodeSet {
    /// Create a new empty `NodeSet` sized for `n` nodes.
    fn with_capacity(n: usize) -> Self {
        if n <= 128 {
            Self::Inline(0)
        } else {
            Self::Heap(FixedBitSet::with_capacity(n))
        }
    }

    /// Insert `bit` into the set. Returns `true` if it was already present.
    fn put(&mut self, bit: usize) -> bool {
        match self {
            Self::Inline(bits) => {
                let mask = 1u128 << bit;
                let was_set = *bits & mask != 0;
                *bits |= mask;
                was_set
            }
            Self::Heap(set) => set.put(bit),
        }
    }
}

/// Maglev consistent hashing table.
///
/// Generic over the node type `N` which must be `Hash + Clone + Eq`.
/// The table provides O(1) primary lookups and preference-list queries
/// for replica-aware routing.
pub struct MaglevTable<N> {
    nodes: Vec<N>,
    table: Vec<usize>,
    capacity: usize,
}

impl<N: Hash + Clone + Eq> MaglevTable<N> {
    /// Build a new Maglev table with default capacity (`nodes.len() * 100`,
    /// rounded up to the nearest prime).
    #[must_use]
    pub fn new(nodes: Vec<N>) -> Self {
        let raw = if nodes.is_empty() {
            1
        } else {
            nodes.len() * 100
        };
        Self::with_capacity(nodes, raw)
    }

    /// Build a new Maglev table with the given minimum capacity.
    ///
    /// The actual table size will be the smallest prime ≥ `min_capacity`.
    /// When comparing tables across node set changes (e.g., measuring
    /// disruption), use the same capacity for both to ensure keys hash
    /// to the same slots.
    #[must_use]
    // u64 % usize always fits in usize
    #[allow(clippy::cast_possible_truncation)]
    pub fn with_capacity(nodes: Vec<N>, min_capacity: usize) -> Self {
        let m = next_prime(min_capacity.max(1));

        if nodes.is_empty() {
            return Self {
                nodes,
                table: Vec::new(),
                capacity: m,
            };
        }

        let n = nodes.len();
        let mut offsets = Vec::with_capacity(n);
        let mut skips = Vec::with_capacity(n);

        for node in &nodes {
            let offset = hash_with_seed(node, 0) % (m as u64);
            let skip = hash_with_seed(node, 1) % ((m - 1) as u64) + 1;
            offsets.push(offset);
            skips.push(skip);
        }

        let mut table = vec![usize::MAX; m];
        let mut next = vec![0u64; n];

        let mut filled = 0usize;
        while filled < m {
            for i in 0..n {
                let mut c = (offsets[i] + next[i] * skips[i]) % (m as u64);
                while table[c as usize] != usize::MAX {
                    next[i] += 1;
                    c = (offsets[i] + next[i] * skips[i]) % (m as u64);
                }
                table[c as usize] = i;
                next[i] += 1;
                filled += 1;
                if filled == m {
                    break;
                }
            }
        }

        Self {
            nodes,
            table,
            capacity: m,
        }
    }

    /// O(1) lookup of the primary node for a key.
    #[must_use]
    // u64 % usize always fits in usize
    #[allow(clippy::cast_possible_truncation)]
    pub fn get<K: Hash>(&self, key: &K) -> Option<&N> {
        if self.nodes.is_empty() {
            return None;
        }
        let slot = (hash_with_seed(key, 2) % (self.capacity as u64)) as usize;
        Some(&self.nodes[self.table[slot]])
    }

    /// Return up to `k` distinct nodes in preference order for the given key.
    ///
    /// The first element always matches `get(key)`. Preference order is
    /// determined by walking the lookup table from the hashed slot and
    /// collecting unique nodes encountered. This ensures consistency
    /// with the primary lookup.
    #[must_use]
    // u64 % usize always fits in usize
    #[allow(clippy::cast_possible_truncation)]
    pub fn get_top_k<K: Hash>(&self, key: &K, k: usize) -> Vec<&N> {
        if self.nodes.is_empty() {
            return Vec::new();
        }
        let k = k.min(self.nodes.len());
        let m = self.capacity;
        let start = (hash_with_seed(key, 2) % (m as u64)) as usize;

        let mut result = Vec::with_capacity(k);
        let mut seen = NodeSet::with_capacity(self.nodes.len());

        for slot in (start..m).chain(0..start) {
            let node_idx = self.table[slot];
            if !seen.put(node_idx) {
                result.push(&self.nodes[node_idx]);
                if result.len() == k {
                    return result;
                }
            }
        }

        result
    }

    /// Returns `true` if `node` appears in the top-`top_k` entries of the
    /// preference list for `key`.
    ///
    /// Note: when `top_k >= self.len()` the result is always `true` (every
    /// node appears exactly once in the full preference list). This method
    /// is most useful with small `top_k` values (e.g., checking if a node
    /// is one of the top-3 replicas for a key).
    #[must_use]
    // u64 % usize always fits in usize
    #[allow(clippy::cast_possible_truncation)]
    pub fn is_match<K: Hash>(&self, key: &K, node: &N, top_k: usize) -> bool {
        if self.nodes.is_empty() {
            return false;
        }
        let top_k = top_k.min(self.nodes.len());
        let m = self.capacity;
        let start = (hash_with_seed(key, 2) % (m as u64)) as usize;

        let mut seen = NodeSet::with_capacity(self.nodes.len());
        let mut count = 0usize;

        for slot in (start..m).chain(0..start) {
            let node_idx = self.table[slot];
            if !seen.put(node_idx) {
                if self.nodes[node_idx] == *node {
                    return true;
                }
                count += 1;
                if count == top_k {
                    return false;
                }
            }
        }

        false
    }

    /// Return all nodes in the table.
    #[must_use]
    pub fn nodes(&self) -> &[N] {
        &self.nodes
    }

    /// Return the table size (the prime M).
    #[must_use]
    pub fn capacity(&self) -> usize {
        self.capacity
    }

    /// Return the number of nodes.
    #[must_use]
    pub fn len(&self) -> usize {
        self.nodes.len()
    }

    /// Return whether the table has no nodes.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.nodes.is_empty()
    }
}

/// Hash a value with a given seed using `DefaultHasher`.
fn hash_with_seed<T: Hash>(val: &T, seed: u64) -> u64 {
    let mut hasher = std::hash::DefaultHasher::new();
    seed.hash(&mut hasher);
    val.hash(&mut hasher);
    hasher.finish()
}

/// Return the smallest prime ≥ `n`.
fn next_prime(n: usize) -> usize {
    if n <= 2 {
        return 2;
    }
    primal::Primes::all()
        .find(|&p| p >= n)
        .expect("primal yields unbounded primes")
}

#[cfg(test)]
#[allow(
    clippy::cast_precision_loss,
    clippy::cast_lossless,
    clippy::similar_names
)]
mod tests {
    use super::*;
    use std::collections::HashMap;
    use std::collections::HashSet;

    #[test]
    fn test_empty_table() {
        let table: MaglevTable<String> = MaglevTable::new(vec![]);
        assert!(table.is_empty());
        assert_eq!(table.len(), 0);
        assert!(table.get(&"any-key").is_none());
        assert!(table.get_top_k(&"any-key", 5).is_empty());
        assert!(!table.is_match(&"any-key", &"node".to_string(), 3));
    }

    #[test]
    fn test_single_node() {
        let table = MaglevTable::new(vec!["only-node".to_string()]);
        assert_eq!(table.len(), 1);
        for i in 0..100 {
            let key = format!("key-{i}");
            assert_eq!(table.get(&key).unwrap(), "only-node");
        }
    }

    #[test]
    fn test_deterministic() {
        let nodes: Vec<String> = (0..5).map(|i| format!("node-{i}")).collect();
        let t1 = MaglevTable::new(nodes.clone());
        let t2 = MaglevTable::new(nodes);

        for i in 0..200 {
            let key = format!("key-{i}");
            assert_eq!(t1.get(&key), t2.get(&key));
        }
    }

    #[test]
    fn test_uniform_distribution() {
        let nodes: Vec<String> = (0..3).map(|i| format!("node-{i}")).collect();
        let table = MaglevTable::new(nodes.clone());

        let mut counts: HashMap<&str, usize> = HashMap::new();
        let num_keys = 10_000;
        for i in 0..num_keys {
            let key = format!("key-{i}");
            let node = table.get(&key).unwrap();
            *counts.entry(node.as_str()).or_default() += 1;
        }

        for node in &nodes {
            let count = counts.get(node.as_str()).copied().unwrap_or(0);
            let ratio = count as f64 / num_keys as f64;
            assert!(
                ratio > 0.20 && ratio < 0.40,
                "node {node} has ratio {ratio:.3}, expected between 0.20 and 0.40"
            );
        }
    }

    #[test]
    fn test_minimal_disruption() {
        let nodes_full: Vec<String> = (0..10).map(|i| format!("node-{i}")).collect();
        let nodes_minus_one: Vec<String> = nodes_full[..9].to_vec();

        // Use the same capacity for both tables so keys hash to the same slots.
        let cap = nodes_full.len() * 100;
        let t_full = MaglevTable::with_capacity(nodes_full, cap);
        let t_minus = MaglevTable::with_capacity(nodes_minus_one, cap);

        let num_keys = 10_000;
        let mut changed = 0usize;
        for i in 0..num_keys {
            let key = format!("key-{i}");
            let a = t_full.get(&key).unwrap();
            let b = t_minus.get(&key).unwrap();
            if a != b {
                changed += 1;
            }
        }

        let disruption = changed as f64 / num_keys as f64;
        assert!(
            disruption < 0.20,
            "disruption {disruption:.3} exceeds 0.20 (expected ~0.10 for 10→9 nodes)"
        );
    }

    #[test]
    fn test_preference_list_length() {
        let nodes: Vec<String> = (0..5).map(|i| format!("node-{i}")).collect();
        let table = MaglevTable::new(nodes.clone());
        let top = table.get_top_k(&"some-key", 5);
        assert_eq!(top.len(), 5);
    }

    #[test]
    fn test_preference_list_deterministic() {
        let nodes: Vec<String> = (0..5).map(|i| format!("node-{i}")).collect();
        let table = MaglevTable::new(nodes);

        let a = table.get_top_k(&"stable-key", 5);
        let b = table.get_top_k(&"stable-key", 5);
        assert_eq!(a, b);
    }

    #[test]
    fn test_preference_list_unique() {
        let nodes: Vec<String> = (0..7).map(|i| format!("node-{i}")).collect();
        let table = MaglevTable::new(nodes);

        for i in 0..100 {
            let key = format!("key-{i}");
            let top = table.get_top_k(&key, 7);
            let set: HashSet<&String> = top.iter().copied().collect();
            assert_eq!(
                set.len(),
                top.len(),
                "duplicate in preference list for {key}"
            );
        }
    }

    #[test]
    fn test_preference_list_primary_matches_get() {
        let nodes: Vec<String> = (0..5).map(|i| format!("node-{i}")).collect();
        let table = MaglevTable::new(nodes);

        for i in 0..200 {
            let key = format!("key-{i}");
            let primary = table.get(&key).unwrap();
            let top = table.get_top_k(&key, 1);
            assert_eq!(top[0], primary, "mismatch for {key}");
        }
    }

    #[test]
    fn test_is_match() {
        let nodes: Vec<String> = (0..5).map(|i| format!("node-{i}")).collect();
        let table = MaglevTable::new(nodes);

        let key = "test-key";
        let top3 = table.get_top_k(&key, 3);
        for node in &top3 {
            assert!(table.is_match(&key, node, 3));
        }

        let top5 = table.get_top_k(&key, 5);
        for node in &top5[3..] {
            assert!(!table.is_match(&key, node, 3));
        }
    }

    #[test]
    fn test_is_match_single_replica() {
        let nodes: Vec<String> = (0..5).map(|i| format!("node-{i}")).collect();
        let table = MaglevTable::new(nodes.clone());

        let key = "single-replica-key";
        let primary = table.get(&key).unwrap().clone();
        assert!(table.is_match(&key, &primary, 1));

        for node in &nodes {
            if node != &primary {
                assert!(
                    !table.is_match(&key, node, 1),
                    "non-primary node {node} matched with replicas=1"
                );
            }
        }
    }

    #[test]
    fn test_large_table() {
        let nodes: Vec<String> = (0..100).map(|i| format!("worker-{i}")).collect();
        let table = MaglevTable::new(nodes.clone());
        assert_eq!(table.len(), 100);

        for i in 0..500 {
            let key = format!("request-{i}");
            let node = table.get(&key).unwrap();
            assert!(nodes.contains(node));
        }
    }

    #[test]
    fn test_prime_capacity() {
        for n in [1, 2, 3, 5, 10, 50, 100] {
            let nodes: Vec<u32> = (0..n).collect();
            let table = MaglevTable::new(nodes);
            let cap = table.capacity();
            assert!(
                primal::is_prime(cap as u64),
                "capacity {cap} is not prime for {n} nodes"
            );
        }
    }

    #[test]
    fn test_preference_list_covers_all_nodes() {
        let nodes: Vec<String> = (0..7).map(|i| format!("node-{i}")).collect();
        let table = MaglevTable::new(nodes.clone());

        for i in 0..100 {
            let key = format!("key-{i}");
            let top = table.get_top_k(&key, 7);
            let set: HashSet<&String> = top.iter().copied().collect();
            assert_eq!(
                set.len(),
                7,
                "preference list for {key} did not cover all nodes"
            );
            for node in &nodes {
                assert!(set.contains(node), "node {node} missing for {key}");
            }
        }
    }

    #[test]
    fn test_node_removal_minimal_disruption_preference_list() {
        let nodes_full: Vec<String> = (0..6).map(|i| format!("node-{i}")).collect();
        let nodes_minus: Vec<String> = nodes_full[..5].to_vec();
        let removed = nodes_full[5].clone();

        // Use the same capacity so slot assignments are comparable.
        let cap = nodes_full.len() * 100;
        let t_full = MaglevTable::with_capacity(nodes_full, cap);
        let t_minus = MaglevTable::with_capacity(nodes_minus, cap);

        let mut preserved = 0usize;
        let mut total_pairs = 0usize;
        let num_keys = 1_000;

        for i in 0..num_keys {
            let key = format!("key-{i}");
            let full_list: Vec<&String> = t_full
                .get_top_k(&key, 6)
                .into_iter()
                .filter(|n| **n != removed)
                .collect();
            let minus_list = t_minus.get_top_k(&key, 5);

            for a in 0..full_list.len() {
                for b in (a + 1)..full_list.len() {
                    total_pairs += 1;
                    let pos_a_minus = minus_list.iter().position(|n| *n == full_list[a]);
                    let pos_b_minus = minus_list.iter().position(|n| *n == full_list[b]);
                    if let (Some(pa), Some(pb)) = (pos_a_minus, pos_b_minus) {
                        if pa < pb {
                            preserved += 1;
                        }
                    }
                }
            }
        }

        let ratio = preserved as f64 / total_pairs as f64;
        assert!(
            ratio > 0.5,
            "pairwise ordering preservation {ratio:.3} is too low"
        );
    }

    #[test]
    fn test_capacity_auto_sizing() {
        let nodes: Vec<u32> = (0..10).collect();
        let table = MaglevTable::new(nodes);
        assert!(table.capacity() >= 1000);
        assert!(primal::is_prime(table.capacity() as u64));
    }

    #[test]
    fn test_string_nodes() {
        let nodes: Vec<String> = vec![
            "us-east-1a".into(),
            "us-west-2b".into(),
            "eu-west-1c".into(),
        ];
        let table = MaglevTable::new(nodes.clone());

        let node = table.get(&"user-session-123").unwrap();
        assert!(nodes.contains(node));

        let top = table.get_top_k(&"user-session-123", 3);
        assert_eq!(top.len(), 3);
    }

    #[test]
    fn test_get_top_k_with_k_greater_than_nodes() {
        let nodes: Vec<String> = (0..3).map(|i| format!("node-{i}")).collect();
        let table = MaglevTable::new(nodes);
        let top = table.get_top_k(&"key", 10);
        assert_eq!(top.len(), 3);
    }

    #[test]
    fn test_send_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<MaglevTable<String>>();
    }

    /// Guard against the stateful-hasher bug found in `chvish/maglev_rs`:
    /// a `DefaultHasher` that isn't reset between nodes makes each node's
    /// hash depend on all previous nodes, so input order silently changes
    /// the table. Our `hash_with_seed` creates a fresh hasher per call,
    /// so inserting an unrelated node must not change existing nodes'
    /// offset/skip values (and therefore must not change which slot an
    /// existing key maps to, given the same capacity).
    #[test]
    fn test_node_hashes_are_independent() {
        let nodes_a: Vec<String> = vec!["alpha", "beta", "gamma"]
            .into_iter()
            .map(String::from)
            .collect();
        let nodes_b: Vec<String> = vec!["alpha", "beta", "gamma", "delta"]
            .into_iter()
            .map(String::from)
            .collect();

        let cap = 10007; // fixed prime capacity
        let t_a = MaglevTable::with_capacity(nodes_a, cap);
        let t_b = MaglevTable::with_capacity(nodes_b, cap);

        // For keys that map to one of the original 3 nodes in both tables,
        // the assignment must be identical — proving that adding "delta"
        // didn't retroactively change alpha/beta/gamma's hashes.
        let mut same = 0;
        let mut comparable = 0;
        for i in 0..10_000 {
            let key = format!("key-{i}");
            let a = t_a.get(&key).unwrap();
            let b = t_b.get(&key).unwrap();
            // Only compare when t_b also picked one of the original 3
            if b == "alpha" || b == "beta" || b == "gamma" {
                comparable += 1;
                if a == b {
                    same += 1;
                }
            }
        }

        // With consistent hashing + same capacity, keys that still map to
        // an original node should get the same node. Allow for the ~25%
        // of slots that get reassigned to "delta".
        let ratio = same as f64 / comparable as f64;
        assert!(
            ratio > 0.95,
            "only {ratio:.3} of comparable keys matched — \
             node hashes may be dependent on input order"
        );
    }

    /// Verify that adding a node causes ~1/N disruption at each position
    /// in the preference list, not just the primary. Without independent
    /// hashes, adding a node would scramble the entire preference order.
    #[test]
    fn test_top_k_minimal_disruption_on_node_addition() {
        let nodes_3: Vec<String> = vec!["alpha", "beta", "gamma"]
            .into_iter()
            .map(String::from)
            .collect();
        let nodes_4: Vec<String> = vec!["alpha", "beta", "gamma", "delta"]
            .into_iter()
            .map(String::from)
            .collect();

        let cap = 10007;
        let t3 = MaglevTable::with_capacity(nodes_3, cap);
        let t4 = MaglevTable::with_capacity(nodes_4, cap);

        let num_keys = 10_000;

        // Per-position disruption: how often does each rank change?
        let mut changed_at_rank = [0usize; 3];
        for i in 0..num_keys {
            let key = format!("key-{i}");
            let top3_old = t3.get_top_k(&key, 3);
            let top3_new = t4.get_top_k(&key, 3);

            for rank in 0..3 {
                if top3_old[rank] != top3_new[rank] {
                    changed_at_rank[rank] += 1;
                }
            }
        }

        // Each rank should see roughly 1/(N+1) ≈ 0.25 disruption.
        // Rank 0 (primary) is tightest; higher ranks allow more drift
        // due to cascading slot displacement.
        let max_disruption = [0.35, 0.40, 0.55];
        for (rank, &changed) in changed_at_rank.iter().enumerate() {
            let ratio = changed as f64 / num_keys as f64;
            assert!(
                ratio < max_disruption[rank],
                "rank {rank} disruption {ratio:.3} exceeds {:.2} — \
                 preference list may be fully scrambled on node addition",
                max_disruption[rank]
            );
        }

        // Set membership: on average, ≥2.2 of the original 3 nodes
        // should survive in the new top-3 (delta takes ~0.75 slots).
        let mut total_surviving = 0usize;
        for i in 0..num_keys {
            let key = format!("key-{i}");
            let old_set: HashSet<&String> = t3.get_top_k(&key, 3).into_iter().collect();
            let new_top3 = t4.get_top_k(&key, 3);
            total_surviving += new_top3.iter().filter(|n| old_set.contains(*n)).count();
        }
        let avg_surviving = total_surviving as f64 / num_keys as f64;
        assert!(
            avg_surviving > 2.2,
            "average surviving nodes in top-3 is {avg_surviving:.2}, \
             expected > 2.2 (only ~1 slot should be taken by new node)"
        );
    }

    /// Guard against the non-deterministic seed bug found in hash-rings:
    /// random `SipHasher` keys mean two independently-constructed tables
    /// with the same nodes produce different assignments. This test
    /// simulates two processes building the table from the same node
    /// list and verifies byte-level agreement.
    #[test]
    fn test_cross_process_agreement() {
        // Simulate two independent constructions (as if in separate processes)
        let build = || {
            let nodes: Vec<String> = vec![
                "us-east-1a",
                "us-west-2b",
                "eu-west-1c",
                "ap-south-1a",
                "sa-east-1a",
            ]
            .into_iter()
            .map(String::from)
            .collect();
            MaglevTable::with_capacity(nodes, 5003)
        };

        let t1 = build();
        let t2 = build();

        // Every single key must produce the same result
        for i in 0..10_000 {
            let key = format!("request-{i}");
            assert_eq!(
                t1.get(&key),
                t2.get(&key),
                "tables disagree on key {key} — seeds may be non-deterministic"
            );
            assert_eq!(
                t1.get_top_k(&key, 5),
                t2.get_top_k(&key, 5),
                "preference lists disagree on key {key}"
            );
        }
    }
}