hive_router_plan_executor/executors/
dedupe.rs

1use ahash::AHasher;
2use ahash::RandomState;
3use std::collections::BTreeMap;
4use std::hash::{BuildHasherDefault, Hash, Hasher};
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::sync::OnceLock;
7use xxhash_rust::xxh3::Xxh3;
8
9use crate::executors::http::SendRequestOpts;
10
11impl Hash for SendRequestOpts<'_> {
12    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
13        // BTreeMap to ensure case-insensitivity and consistent order for hashing
14        let mut headers = BTreeMap::new();
15        for (header_name, header_value) in self.headers.iter() {
16            if let Ok(value_str) = header_value.to_str() {
17                headers.insert(header_name.as_str(), value_str);
18            }
19        }
20
21        self.method.hash(state);
22        self.endpoint.hash(state);
23        headers.hash(state);
24        self.body.hash(state);
25    }
26}
27
28impl SendRequestOpts<'_> {
29    /// Generate a hash for the request options, used for deduplication.
30    pub fn fingerprint(&self) -> u64 {
31        let mut hasher = Xxh3::new();
32        self.hash(&mut hasher);
33        hasher.finish()
34    }
35}
36
37pub type ABuildHasher = BuildHasherDefault<AHasher>;
38
39static LEADER_COUNTER: AtomicU64 = AtomicU64::new(1);
40static LEADER_SALT: OnceLock<u64> = OnceLock::new();
41
42/// Generate a unique fingerprint for the current leader.
43/// This is used to identify the leader in a distributed system.
44pub fn unique_leader_fingerprint() -> u64 {
45    let idx = LEADER_COUNTER.fetch_add(1, Ordering::Relaxed);
46    let salt =
47        LEADER_SALT.get_or_init(|| RandomState::new().hash_one(b"unique-leader-fingerprint"));
48    idx ^ salt
49}
hive_router_plan_executor/executors/dedupe.rs

hive_router_plan_executor/executors/
dedupe.rs
