use chrono::{DateTime, Utc};
use offline_protocol_core::MessageId;
use std::collections::HashMap;
#[derive(Debug, Clone)]
pub struct DeduplicatorConfig {
pub max_tracked_messages: usize,
pub retention_time_secs: u64,
pub use_bloom_filter: bool,
pub bloom_filter_bits: usize,
pub bloom_hash_count: usize,
pub bloom_filter_count: usize,
pub bloom_rotation_secs: u64,
}
impl Default for DeduplicatorConfig {
fn default() -> Self {
Self {
max_tracked_messages: 1000,
retention_time_secs: 3600, use_bloom_filter: false,
bloom_filter_bits: 1 << 20, bloom_hash_count: 7, bloom_filter_count: 4, bloom_rotation_secs: 900, }
}
}
#[derive(Clone)]
struct BloomFilter {
bits: Vec<u64>,
bit_count: usize,
hash_count: usize,
items_added: usize,
}
impl BloomFilter {
fn new(bit_count: usize, hash_count: usize) -> Self {
let word_count = bit_count.div_ceil(64);
Self {
bits: vec![0u64; word_count],
bit_count,
hash_count,
items_added: 0,
}
}
fn hashes(&self, item: &str) -> Vec<usize> {
let mut hashes = Vec::with_capacity(self.hash_count);
let h1 = fnv1a_hash(item.as_bytes());
let h2 = fnv1a_hash_seeded(item.as_bytes(), 0x5f356495) | 1;
for i in 0..self.hash_count {
let combined = h1.wrapping_add((i as u64).wrapping_mul(h2));
hashes.push((combined as usize) % self.bit_count);
}
hashes
}
fn insert(&mut self, item: &str) {
for bit_index in self.hashes(item) {
let word_index = bit_index / 64;
let bit_offset = bit_index % 64;
self.bits[word_index] |= 1u64 << bit_offset;
}
self.items_added += 1;
}
fn contains(&self, item: &str) -> bool {
for bit_index in self.hashes(item) {
let word_index = bit_index / 64;
let bit_offset = bit_index % 64;
if (self.bits[word_index] & (1u64 << bit_offset)) == 0 {
return false;
}
}
true
}
fn clear(&mut self) {
self.bits.fill(0);
self.items_added = 0;
}
fn items_count(&self) -> usize {
self.items_added
}
fn estimated_false_positive_rate(&self) -> f64 {
if self.items_added == 0 {
return 0.0;
}
let m = self.bit_count as f64;
let n = self.items_added as f64;
let k = self.hash_count as f64;
(1.0 - (-k * n / m).exp()).powf(k)
}
}
fn fnv1a_hash(data: &[u8]) -> u64 {
fnv1a_hash_seeded(data, 0xcbf29ce484222325)
}
fn fnv1a_hash_seeded(data: &[u8], seed: u64) -> u64 {
const FNV_PRIME: u64 = 0x100000001b3;
let mut hash = seed;
for byte in data {
hash ^= *byte as u64;
hash = hash.wrapping_mul(FNV_PRIME);
}
hash
}
struct RotatingBloomFilter {
filters: Vec<BloomFilter>,
current_index: usize,
last_rotation: DateTime<Utc>,
rotation_interval_secs: u64,
}
impl RotatingBloomFilter {
fn new(filter_count: usize, bit_count: usize, hash_count: usize, rotation_secs: u64) -> Self {
let filters = (0..filter_count)
.map(|_| BloomFilter::new(bit_count, hash_count))
.collect();
Self {
filters,
current_index: 0,
last_rotation: Utc::now(),
rotation_interval_secs: rotation_secs,
}
}
fn maybe_rotate(&mut self, now: DateTime<Utc>) {
let elapsed = now.signed_duration_since(self.last_rotation);
if elapsed.num_seconds() >= self.rotation_interval_secs as i64 {
self.current_index = (self.current_index + 1) % self.filters.len();
self.filters[self.current_index].clear();
self.last_rotation = now;
}
}
fn insert(&mut self, item: &str) {
self.maybe_rotate(Utc::now());
self.filters[self.current_index].insert(item);
}
fn contains(&mut self, item: &str) -> bool {
self.maybe_rotate(Utc::now());
self.filters.iter().any(|f| f.contains(item))
}
fn total_items(&self) -> usize {
self.filters.iter().map(|f| f.items_count()).sum()
}
fn clear(&mut self) {
for filter in &mut self.filters {
filter.clear();
}
self.current_index = 0;
self.last_rotation = Utc::now();
}
fn average_false_positive_rate(&self) -> f64 {
let rates: Vec<f64> = self
.filters
.iter()
.map(|f| f.estimated_false_positive_rate())
.collect();
if rates.is_empty() {
return 0.0;
}
rates.iter().sum::<f64>() / rates.len() as f64
}
}
#[derive(Debug, Clone)]
struct SeenEntry {
seen_at: DateTime<Utc>,
last_accessed: DateTime<Utc>,
}
pub struct Deduplicator {
config: DeduplicatorConfig,
seen_messages: HashMap<String, SeenEntry>,
bloom_filter: Option<RotatingBloomFilter>,
}
impl Deduplicator {
pub fn new() -> Self {
Self::with_config(DeduplicatorConfig::default())
}
pub fn with_config(config: DeduplicatorConfig) -> Self {
let valid_bloom_config = config.bloom_filter_bits > 0
&& config.bloom_hash_count > 0
&& config.bloom_filter_count > 0
&& config.bloom_rotation_secs > 0;
let bloom_filter = if config.use_bloom_filter && valid_bloom_config {
Some(RotatingBloomFilter::new(
config.bloom_filter_count,
config.bloom_filter_bits,
config.bloom_hash_count,
config.bloom_rotation_secs,
))
} else {
None
};
Self {
config,
seen_messages: HashMap::new(),
bloom_filter,
}
}
pub fn is_exact(&self) -> bool {
self.bloom_filter.is_none()
}
pub fn is_duplicate(&self, message_id: &MessageId) -> bool {
let msg_id_str = message_id.as_str();
if let Some(ref bloom) = self.bloom_filter {
bloom.filters.iter().any(|f| f.contains(&msg_id_str))
} else {
self.seen_messages.contains_key(&msg_id_str)
}
}
pub fn is_duplicate_mut(&mut self, message_id: &MessageId) -> bool {
let msg_id_str = message_id.as_str();
if let Some(ref mut bloom) = self.bloom_filter {
bloom.contains(&msg_id_str)
} else if let Some(entry) = self.seen_messages.get_mut(&msg_id_str) {
entry.last_accessed = Utc::now();
true
} else {
false
}
}
pub fn mark_seen(&mut self, message_id: MessageId) -> bool {
let msg_id_str = message_id.as_str();
if let Some(ref mut bloom) = self.bloom_filter {
if bloom.contains(&msg_id_str) {
return false;
}
bloom.insert(&msg_id_str);
true
} else {
if self.seen_messages.contains_key(&msg_id_str) {
return false;
}
if self.seen_messages.len() >= self.config.max_tracked_messages {
self.cleanup_expired();
if self.seen_messages.len() >= self.config.max_tracked_messages {
if let Some((lru_id, _)) = self
.seen_messages
.iter()
.min_by_key(|(_, entry)| entry.last_accessed)
.map(|(id, entry)| (id.clone(), entry.clone()))
{
self.seen_messages.remove(&lru_id);
}
}
}
let now = Utc::now();
self.seen_messages.insert(
msg_id_str,
SeenEntry {
seen_at: now,
last_accessed: now,
},
);
true
}
}
pub fn unmark_seen(&mut self, message_id: &MessageId) -> bool {
if self.bloom_filter.is_some() {
return false;
}
self.seen_messages.remove(&message_id.as_str()).is_some()
}
pub fn cleanup_expired(&mut self) -> usize {
if self.bloom_filter.is_some() {
return 0;
}
let now = Utc::now();
let retention = chrono::Duration::seconds(self.config.retention_time_secs as i64);
let cutoff = now - retention;
let before_count = self.seen_messages.len();
self.seen_messages.retain(|_, entry| entry.seen_at > cutoff);
before_count - self.seen_messages.len()
}
pub fn tracked_count(&self) -> usize {
if let Some(ref bloom) = self.bloom_filter {
bloom.total_items()
} else {
self.seen_messages.len()
}
}
pub fn clear(&mut self) {
if let Some(ref mut bloom) = self.bloom_filter {
bloom.clear();
}
self.seen_messages.clear();
}
pub fn is_tracked(&self, message_id: &MessageId) -> bool {
self.is_duplicate(message_id)
}
pub fn is_bloom_filter_mode(&self) -> bool {
self.bloom_filter.is_some()
}
pub fn stats(&self) -> DeduplicatorStats {
if let Some(ref bloom) = self.bloom_filter {
let total = bloom.total_items();
let estimated_capacity = self.config.bloom_filter_bits / self.config.bloom_hash_count;
let capacity_percent =
((total as f32 / estimated_capacity as f32) * 100.0).min(100.0) as u8;
DeduplicatorStats {
total_tracked: total,
recent_tracked: total, capacity_used_percent: capacity_percent,
false_positive_rate: Some(bloom.average_false_positive_rate()),
mode: DeduplicatorMode::BloomFilter,
}
} else {
let now = Utc::now();
let recent_cutoff = now - chrono::Duration::seconds(60);
let recent_count = self
.seen_messages
.values()
.filter(|entry| entry.seen_at > recent_cutoff)
.count();
DeduplicatorStats {
total_tracked: self.seen_messages.len(),
recent_tracked: recent_count,
capacity_used_percent: (self.seen_messages.len() as f32
/ self.config.max_tracked_messages as f32
* 100.0) as u8,
false_positive_rate: None,
mode: DeduplicatorMode::HashMap,
}
}
}
pub fn estimated_memory_bytes(&self) -> usize {
if self.bloom_filter.is_some() {
let filter_bytes = self.config.bloom_filter_bits.div_ceil(8);
filter_bytes * self.config.bloom_filter_count
+ std::mem::size_of::<RotatingBloomFilter>()
} else {
self.seen_messages.len() * 50 + std::mem::size_of::<HashMap<String, SeenEntry>>()
}
}
}
impl Default for Deduplicator {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeduplicatorMode {
HashMap,
BloomFilter,
}
#[derive(Debug, Clone)]
pub struct DeduplicatorStats {
pub total_tracked: usize,
pub recent_tracked: usize,
pub capacity_used_percent: u8,
pub false_positive_rate: Option<f64>,
pub mode: DeduplicatorMode,
}
#[cfg(test)]
mod tests {
use super::*;
use std::thread;
use std::time::Duration;
fn hashmap_config() -> DeduplicatorConfig {
DeduplicatorConfig {
use_bloom_filter: false,
..Default::default()
}
}
#[test]
fn test_default_config_is_hashmap_and_capacity_1000() {
let config = DeduplicatorConfig::default();
assert!(
!config.use_bloom_filter,
"default should be exact-match HashMap to avoid false positives"
);
assert_eq!(config.max_tracked_messages, 1000);
let dedup = Deduplicator::new();
assert!(!dedup.is_bloom_filter_mode());
let stats = dedup.stats();
assert_eq!(stats.mode, DeduplicatorMode::HashMap);
}
#[test]
fn test_invalid_bloom_config_falls_back_without_panicking() {
let config = DeduplicatorConfig {
use_bloom_filter: true,
bloom_filter_bits: 0,
..Default::default()
};
let mut dedup = Deduplicator::with_config(config);
let message_id = MessageId::new();
assert!(!dedup.is_bloom_filter_mode());
assert!(dedup.mark_seen(message_id.clone()));
assert!(dedup.is_duplicate(&message_id));
}
#[test]
fn test_default_hashmap_no_false_positives() {
let mut dedup = Deduplicator::new();
let seen = MessageId::new();
dedup.mark_seen(seen.clone());
for _ in 0..20 {
let fresh = MessageId::new();
assert!(
!dedup.is_duplicate(&fresh),
"fresh message ID must not be false positive"
);
}
assert!(dedup.is_duplicate(&seen));
}
#[test]
fn test_basic_deduplication() {
let mut dedup = Deduplicator::new();
let msg_id = MessageId::new();
assert!(!dedup.is_duplicate(&msg_id));
assert!(dedup.mark_seen(msg_id.clone()));
assert!(dedup.is_duplicate(&msg_id));
assert!(!dedup.mark_seen(msg_id.clone()));
}
#[test]
fn test_unmark_seen_hashmap_mode() {
let mut dedup = Deduplicator::with_config(hashmap_config());
let msg_id = MessageId::new();
dedup.mark_seen(msg_id.clone());
assert!(dedup.is_duplicate(&msg_id));
assert!(dedup.unmark_seen(&msg_id), "tracked id must be removed");
assert!(
!dedup.is_duplicate(&msg_id),
"unmarked message must be processed fresh on replay"
);
assert!(
!dedup.unmark_seen(&msg_id),
"second unmark of the same id is a no-op"
);
}
#[test]
fn test_unmark_seen_noop_in_bloom_mode() {
let config = DeduplicatorConfig {
use_bloom_filter: true,
bloom_filter_bits: 1 << 16,
bloom_hash_count: 5,
bloom_filter_count: 2,
bloom_rotation_secs: 60,
..Default::default()
};
let mut dedup = Deduplicator::with_config(config);
let msg_id = MessageId::new();
dedup.mark_seen(msg_id.clone());
assert!(
!dedup.unmark_seen(&msg_id),
"bloom filters cannot remove entries"
);
assert!(
dedup.is_duplicate(&msg_id),
"bloom mode keeps the id until its window rotates"
);
}
#[test]
fn test_basic_deduplication_hashmap_mode() {
let mut dedup = Deduplicator::with_config(hashmap_config());
let msg_id = MessageId::new();
assert!(!dedup.is_duplicate(&msg_id));
assert!(dedup.mark_seen(msg_id.clone()));
assert!(dedup.is_duplicate(&msg_id));
assert!(!dedup.mark_seen(msg_id.clone()));
}
#[test]
fn test_multiple_messages() {
let mut dedup = Deduplicator::new();
let msg1 = MessageId::new();
let msg2 = MessageId::new();
let msg3 = MessageId::new();
assert!(dedup.mark_seen(msg1.clone()));
assert!(dedup.mark_seen(msg2.clone()));
assert!(dedup.mark_seen(msg3.clone()));
assert_eq!(dedup.tracked_count(), 3);
assert!(dedup.is_duplicate(&msg1));
assert!(dedup.is_duplicate(&msg2));
assert!(dedup.is_duplicate(&msg3));
}
#[test]
fn test_capacity_limit() {
let config = DeduplicatorConfig {
max_tracked_messages: 3,
retention_time_secs: 3600,
use_bloom_filter: false,
..Default::default()
};
let mut dedup = Deduplicator::with_config(config);
let msg1 = MessageId::new();
let msg2 = MessageId::new();
let msg3 = MessageId::new();
let msg4 = MessageId::new();
dedup.mark_seen(msg1.clone());
dedup.mark_seen(msg2.clone());
dedup.mark_seen(msg3.clone());
assert_eq!(dedup.tracked_count(), 3);
dedup.mark_seen(msg4.clone());
assert_eq!(dedup.tracked_count(), 3);
assert!(!dedup.is_duplicate(&msg1));
assert!(dedup.is_duplicate(&msg4));
}
#[test]
fn test_cleanup_expired() {
let config = DeduplicatorConfig {
max_tracked_messages: 100,
retention_time_secs: 1,
use_bloom_filter: false,
..Default::default()
};
let mut dedup = Deduplicator::with_config(config);
let msg1 = MessageId::new();
let msg2 = MessageId::new();
dedup.mark_seen(msg1.clone());
thread::sleep(Duration::from_millis(500));
dedup.mark_seen(msg2.clone());
thread::sleep(Duration::from_millis(600));
let removed = dedup.cleanup_expired();
assert_eq!(removed, 1);
assert_eq!(dedup.tracked_count(), 1);
assert!(!dedup.is_duplicate(&msg1));
assert!(dedup.is_duplicate(&msg2));
}
#[test]
fn test_clear() {
let mut dedup = Deduplicator::new();
dedup.mark_seen(MessageId::new());
dedup.mark_seen(MessageId::new());
dedup.mark_seen(MessageId::new());
assert_eq!(dedup.tracked_count(), 3);
dedup.clear();
assert_eq!(dedup.tracked_count(), 0);
}
#[test]
fn test_is_tracked() {
let mut dedup = Deduplicator::new();
let msg_id = MessageId::new();
assert!(!dedup.is_tracked(&msg_id));
dedup.mark_seen(msg_id.clone());
assert!(dedup.is_tracked(&msg_id));
}
#[test]
fn test_stats() {
let mut dedup = Deduplicator::new();
for _ in 0..5 {
dedup.mark_seen(MessageId::new());
}
let stats = dedup.stats();
assert_eq!(stats.total_tracked, 5);
}
#[test]
fn test_stats_hashmap_mode() {
let mut dedup = Deduplicator::with_config(hashmap_config());
for _ in 0..5 {
dedup.mark_seen(MessageId::new());
}
let stats = dedup.stats();
assert_eq!(stats.total_tracked, 5);
assert_eq!(stats.recent_tracked, 5);
assert!(stats.capacity_used_percent < 1);
assert_eq!(stats.mode, DeduplicatorMode::HashMap);
}
#[test]
fn test_capacity_percent() {
let config = DeduplicatorConfig {
max_tracked_messages: 10,
retention_time_secs: 3600,
use_bloom_filter: false,
..Default::default()
};
let mut dedup = Deduplicator::with_config(config);
for _ in 0..5 {
dedup.mark_seen(MessageId::new());
}
let stats = dedup.stats();
assert_eq!(stats.capacity_used_percent, 50);
}
#[test]
fn test_bloom_filter_mode() {
let config = DeduplicatorConfig {
use_bloom_filter: true,
bloom_filter_bits: 1 << 16, bloom_hash_count: 5,
bloom_filter_count: 2,
bloom_rotation_secs: 60,
..Default::default()
};
let mut dedup = Deduplicator::with_config(config);
assert!(dedup.is_bloom_filter_mode());
let msg1 = MessageId::new();
let msg2 = MessageId::new();
assert!(!dedup.is_duplicate(&msg1));
dedup.mark_seen(msg1.clone());
assert!(dedup.is_duplicate(&msg1));
assert!(!dedup.is_duplicate(&msg2));
dedup.mark_seen(msg2.clone());
assert!(dedup.is_duplicate(&msg2));
let stats = dedup.stats();
assert_eq!(stats.mode, DeduplicatorMode::BloomFilter);
assert!(stats.false_positive_rate.is_some());
}
#[test]
fn test_bloom_filter_constant_memory() {
let config = DeduplicatorConfig {
use_bloom_filter: true,
bloom_filter_bits: 1 << 16,
bloom_hash_count: 5,
bloom_filter_count: 2,
bloom_rotation_secs: 3600,
..Default::default()
};
let dedup = Deduplicator::with_config(config);
let initial_memory = dedup.estimated_memory_bytes();
let expected_filter_bytes = (1 << 16) / 8 * 2; assert!(initial_memory >= expected_filter_bytes);
assert!(initial_memory < expected_filter_bytes * 2); }
#[test]
fn test_bloom_filter_high_volume() {
let config = DeduplicatorConfig {
use_bloom_filter: true,
bloom_filter_bits: 1 << 20, bloom_hash_count: 7,
bloom_filter_count: 4,
bloom_rotation_secs: 3600,
..Default::default()
};
let mut dedup = Deduplicator::with_config(config);
let mut inserted = 0usize;
for _ in 0..10_000 {
let msg = MessageId::new();
if dedup.mark_seen(msg) {
inserted += 1;
}
}
assert!(
inserted >= 9_990,
"too many bloom false positives: {}",
10_000 - inserted
);
assert_eq!(dedup.tracked_count(), inserted);
let stats = dedup.stats();
assert!(stats.false_positive_rate.unwrap() < 0.02);
}
#[test]
fn test_lru_eviction_preserves_recently_accessed() {
let config = DeduplicatorConfig {
max_tracked_messages: 3,
retention_time_secs: 3600,
use_bloom_filter: false,
..Default::default()
};
let mut dedup = Deduplicator::with_config(config);
let msg_a = MessageId::new();
let msg_b = MessageId::new();
let msg_c = MessageId::new();
dedup.mark_seen(msg_a.clone());
thread::sleep(Duration::from_millis(10));
dedup.mark_seen(msg_b.clone());
thread::sleep(Duration::from_millis(10));
dedup.mark_seen(msg_c.clone());
assert!(dedup.is_duplicate_mut(&msg_a));
let msg_d = MessageId::new();
dedup.mark_seen(msg_d.clone());
assert_eq!(dedup.tracked_count(), 3);
assert!(
dedup.is_duplicate(&msg_a),
"A was recently accessed, should survive"
);
assert!(
!dedup.is_duplicate(&msg_b),
"B should have been evicted (LRU)"
);
assert!(dedup.is_duplicate(&msg_c), "C should survive");
assert!(dedup.is_duplicate(&msg_d), "D was just inserted");
}
}