Method compact

nodedb/src/event/cdc/buffer.rs:160–203

Compact the buffer: deduplicate by key field, keeping only the latest event per key value. DELETE events are retained as tombstones until they exceed `tombstone_grace_secs` age, then removed.

(&self, key_field: &str, tombstone_grace_secs: u64)

    /// event per key value. DELETE events are retained as tombstones until
    /// they exceed `tombstone_grace_secs` age, then removed.
    pub fn compact(&self, key_field: &str, tombstone_grace_secs: u64) -> u32 {
        let mut events = self.events.write().unwrap_or_else(|p| {
            tracing::warn!(stream = %self.name, "StreamBuffer RwLock poisoned during compact, recovering");
            p.into_inner()
        });
        let before = events.len();

        let now_ms = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;
        let tombstone_cutoff_ms = now_ms.saturating_sub(tombstone_grace_secs * 1000);

        let mut latest: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
        for (idx, event) in events.iter().enumerate() {
            let key_value = extract_key_value(event, key_field);
            latest.insert(key_value, idx);
        }

        let mut keep = vec![false; events.len()];
        for (idx, event) in events.iter().enumerate() {
            let key_value = extract_key_value(event, key_field);
            let is_latest = latest.get(&key_value) == Some(&idx);
            let is_tombstone = event.op == "DELETE";
            if is_latest && !(is_tombstone && event.event_time < tombstone_cutoff_ms) {
                keep[idx] = true;
            }
        }

        let mut new_events = VecDeque::with_capacity(events.len());
        for (idx, event) in events.drain(..).enumerate() {
            if keep[idx] {
                new_events.push_back(event);
            }
        }
        *events = new_events;

        let removed = (before - events.len()) as u32;
        if removed > 0 {
            self.total_evicted
                .fetch_add(removed as u64, std::sync::atomic::Ordering::Relaxed);
        }
        removed
    }

    pub fn len(&self) -> usize {
        self.events.read().unwrap_or_else(|p| p.into_inner()).len()