crdt.rs

//! Conflict-free Replicated Data Types (CRDTs) for Nulang.
//!
//! CRDTs enable distributed actors to share mutable state without locks,
//! consensus, or coordination. All changes merge automatically and converge
//! to the same result.
//!
//! This module provides the core [`Crdt`] trait and five concrete
//! implementations:
//!
//! | Type          | Operations         | Semantics                          |
//! |---------------|--------------------|-------------------------------------|
//! | [`GCounter`]  | increment          | Grow-only counter (monotonic)      |
//! | [`PNCounter`] | increment, decrement | Counter that can go negative       |
//! | [`GSet`]      | insert             | Grow-only set                      |
//! | [`ORSet`]     | add, remove        | Add-wins observed-remove set       |
//! | [`AWORSet`]   | add, remove        | Timestamp-based add-wins OR set    |
//!
//! # Mathematical Foundations
//!
//! Every CRDT satisfies three algebraic properties:
//!
//! **Associativity** — `merge(a, merge(b, c)) == merge(merge(a, b), c)`
//! Merging can be grouped arbitrarily; the order of pairwise merges does not
//! matter. This allows tree-structured or pipelined replication topologies.
//!
//! **Commutativity** — `merge(a, b) == merge(b, a)`
//! The merge order is irrelevant. This is the key property that eliminates
//! the need for consensus: replicas can merge in any order and still agree.
//!
//! **Idempotency** — `merge(a, a) == a`
//! Merging a replica with itself is a no-op. This makes retries safe and
//! deduplication unnecessary.
//!
//! Together, these properties form a **join-semilattice** where `merge` is the
//! least-upper-bound (LUB) operator and the CRDT state is ordered by the
//! "happens-before" relation. The LUB of any two states always exists and is
//! unique, guaranteeing convergence.
//!
//! # Delta-state replication
//!
//! Every CRDT additionally provides `delta_since(&self, base) -> Option<Self>`:
//! the minimal state that, when merged into any replica that already contains
//! `base`, produces the same result as merging this full state. Formally, for
//! any replica `r` with `base ⊑ r`:
//!
//! ```text
//! r.merge(self.delta_since(base)) == r.merge(self)
//! ```
//!
//! A delta is itself a valid CRDT state (serialized with the usual
//! `to_bytes`), so receivers merge it through the same code path as a
//! full-state merge. `None` means nothing changed since `base`. Deltas are
//! only meaningful to a peer that already holds `base`; a peer seeing an
//! entry for the first time must receive a full-state snapshot instead
//! (the join fallback, see `CrdtManager::generate_delta_sync_ops`).

use std::collections::{HashMap, HashSet};

// =============================================================================
// Core CRDT Trait
// =============================================================================

/// The core trait for all Conflict-free Replicated Data Types.
///
/// Every CRDT must support:
/// - [`merge`](Crdt::merge): combine two replicas into one (must be associative, commutative, idempotent)
/// - [`value`](Crdt::value): read the current logical value
/// - [`delta_since`](Crdt::delta_since): compute a minimal delta-state relative to a base
/// - [`to_bytes`](Crdt::to_bytes) / [`from_bytes`](Crdt::from_bytes): serialize for network transmission
///
/// # Type Parameters
///
/// - `Value`: The logical (user-facing) type that this CRDT represents.
///   This is often different from the internal replica state. For example,
///   a `PNCounter` internally stores two `GCounter`s but its logical value
///   is `i64`.
pub trait Crdt: Clone {
    /// The logical value type that this CRDT represents.
    ///
    /// This is the type returned by [`value`](Crdt::value) and is the
    /// user-facing abstraction over the internal replicated state.
    type Value;

    /// Merge another replica into this one.
    ///
    /// After `self.merge(other)`, `self` contains the combined state
    /// of both replicas. This operation must be:
    ///
    /// - **Associative**: `a.merge(b.merge(c)) == a.merge(b).merge(c)`
    /// - **Commutative**: `a.merge(b) == b.merge(a)`
    /// - **Idempotent**: `a.merge(a) == a`
    fn merge(&mut self, other: &Self);

    /// Read the current logical value.
    fn value(&self) -> Self::Value;

    /// Compute the delta-state relative to `base`.
    ///
    /// Returns the minimal state that, when merged into any replica that
    /// already contains `base`, produces the same result as merging the full
    /// current state. `None` when nothing changed since `base`.
    fn delta_since(&self, base: &Self) -> Option<Self>;

    /// Serialize this CRDT to bytes for network transmission.
    fn to_bytes(&self) -> Vec<u8>;

    /// Deserialize a CRDT from bytes.
    fn from_bytes(bytes: &[u8]) -> Option<Self>;
}

// =============================================================================
// Serialization Helpers
// =============================================================================

#[inline]
fn push_u64(buf: &mut Vec<u8>, v: u64) {
    buf.extend_from_slice(&v.to_be_bytes());
}

#[inline]
fn push_u32(buf: &mut Vec<u8>, v: u32) {
    buf.extend_from_slice(&v.to_be_bytes());
}

#[inline]
fn read_u64(bytes: &[u8], pos: usize) -> Option<(u64, usize)> {
    let end = pos.checked_add(8)?;
    if end > bytes.len() {
        return None;
    }
    let mut arr = [0u8; 8];
    arr.copy_from_slice(&bytes[pos..end]);
    Some((u64::from_be_bytes(arr), end))
}

#[inline]
fn read_u32(bytes: &[u8], pos: usize) -> Option<(u32, usize)> {
    let end = pos.checked_add(4)?;
    if end > bytes.len() {
        return None;
    }
    let mut arr = [0u8; 4];
    arr.copy_from_slice(&bytes[pos..end]);
    Some((u32::from_be_bytes(arr), end))
}

#[inline]
fn push_string(buf: &mut Vec<u8>, s: &str) {
    push_u32(buf, s.len() as u32);
    buf.extend_from_slice(s.as_bytes());
}

#[inline]
fn read_string(bytes: &[u8], pos: usize) -> Option<(String, usize)> {
    let (len, pos) = read_u32(bytes, pos)?;
    let len = len as usize;
    let end = pos.checked_add(len)?;
    if end > bytes.len() {
        return None;
    }
    let s = String::from_utf8(bytes[pos..end].to_vec()).ok()?;
    Some((s, end))
}

// =============================================================================
// GCounter
// =============================================================================

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GCounter {
    pub counts: HashMap<u64, u64>,
    pub node_id: u64,
}

impl GCounter {
    pub fn new(node_id: u64) -> Self {
        Self {
            counts: HashMap::new(),
            node_id,
        }
    }

    pub fn increment(&mut self) {
        *self.counts.entry(self.node_id).or_insert(0) += 1;
    }

    pub fn increment_by(&mut self, delta: u64) {
        *self.counts.entry(self.node_id).or_insert(0) += delta;
    }

    pub fn value(&self) -> u64 {
        self.counts.values().sum()
    }

    /// Delta relative to `base`: only the per-node entries that grew since
    /// `base`. `None` when no entry changed.
    pub fn delta_since(&self, base: &Self) -> Option<Self> {
        let mut counts = HashMap::new();
        for (node_id, count) in &self.counts {
            let base_count = base.counts.get(node_id).copied().unwrap_or(0);
            if *count > base_count {
                counts.insert(*node_id, *count);
            }
        }
        if counts.is_empty() {
            None
        } else {
            Some(Self {
                counts,
                node_id: self.node_id,
            })
        }
    }
}

impl Crdt for GCounter {
    type Value = u64;

    fn merge(&mut self, other: &Self) {
        for (node_id, count) in &other.counts {
            let entry = self.counts.entry(*node_id).or_insert(0);
            *entry = (*entry).max(*count);
        }
    }

    fn value(&self) -> Self::Value {
        self.value()
    }

    fn delta_since(&self, base: &Self) -> Option<Self> {
        GCounter::delta_since(self, base)
    }

    fn to_bytes(&self) -> Vec<u8> {
        let mut buf = Vec::new();
        push_u64(&mut buf, self.node_id);
        push_u32(&mut buf, self.counts.len() as u32);
        for (node_id, count) in &self.counts {
            push_u64(&mut buf, *node_id);
            push_u64(&mut buf, *count);
        }
        buf
    }

    fn from_bytes(bytes: &[u8]) -> Option<Self> {
        let (node_id, pos) = read_u64(bytes, 0)?;
        let (num_entries, mut pos) = read_u32(bytes, pos)?;
        let mut counts = HashMap::new();
        for _ in 0..num_entries {
            let (nid, p) = read_u64(bytes, pos)?;
            let (cnt, p) = read_u64(bytes, p)?;
            counts.insert(nid, cnt);
            pos = p;
        }
        Some(Self { counts, node_id })
    }
}

// =============================================================================
// PNCounter
// =============================================================================

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PNCounter {
    pub increments: GCounter,
    pub decrements: GCounter,
}

impl PNCounter {
    pub fn new(node_id: u64) -> Self {
        Self {
            increments: GCounter::new(node_id),
            decrements: GCounter::new(node_id),
        }
    }

    pub fn increment(&mut self) {
        self.increments.increment();
    }
    pub fn decrement(&mut self) {
        self.decrements.increment();
    }
    pub fn increment_by(&mut self, delta: u64) {
        self.increments.increment_by(delta);
    }
    pub fn decrement_by(&mut self, delta: u64) {
        self.decrements.increment_by(delta);
    }

    pub fn value(&self) -> i64 {
        self.increments.value() as i64 - self.decrements.value() as i64
    }

    /// Delta relative to `base`: the deltas of both underlying GCounters.
    /// `None` when neither side changed.
    pub fn delta_since(&self, base: &Self) -> Option<Self> {
        match (
            self.increments.delta_since(&base.increments),
            self.decrements.delta_since(&base.decrements),
        ) {
            (None, None) => None,
            (inc, dec) => Some(Self {
                increments: inc.unwrap_or_else(|| GCounter::new(self.increments.node_id)),
                decrements: dec.unwrap_or_else(|| GCounter::new(self.decrements.node_id)),
            }),
        }
    }
}

impl Crdt for PNCounter {
    type Value = i64;

    fn merge(&mut self, other: &Self) {
        self.increments.merge(&other.increments);
        self.decrements.merge(&other.decrements);
    }

    fn value(&self) -> Self::Value {
        self.value()
    }

    fn delta_since(&self, base: &Self) -> Option<Self> {
        PNCounter::delta_since(self, base)
    }

    fn to_bytes(&self) -> Vec<u8> {
        let mut buf = self.increments.to_bytes();
        buf.extend_from_slice(&self.decrements.to_bytes());
        buf
    }

    fn from_bytes(bytes: &[u8]) -> Option<Self> {
        let (node_id, pos) = read_u64(bytes, 0)?;
        let (num_entries, _) = read_u32(bytes, pos)?;
        // `num_entries` comes off the wire: validate the split point before
        // slicing so a crafted payload cannot panic the scheduler thread.
        // The decrement half needs no explicit bound — once `increment_len`
        // is in range the second slice is valid, and `GCounter::from_bytes`
        // bounds-checks its own contents.
        let increment_len = (num_entries as usize).checked_mul(16)?.checked_add(12)?;
        if increment_len > bytes.len() {
            return None;
        }
        let inc_bytes = &bytes[0..increment_len];
        let dec_bytes = &bytes[increment_len..];
        let increments = GCounter::from_bytes(inc_bytes)?;
        let decrements = GCounter::from_bytes(dec_bytes)?;
        if increments.node_id != node_id || decrements.node_id != node_id {
            return None;
        }
        Some(Self {
            increments,
            decrements,
        })
    }
}

// =============================================================================
// GSet
// =============================================================================

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GSet<T: Clone + Eq + std::hash::Hash> {
    pub elements: HashSet<T>,
}

impl<T: Clone + Eq + std::hash::Hash> GSet<T> {
    pub fn new() -> Self {
        Self {
            elements: HashSet::new(),
        }
    }
    pub fn insert(&mut self, element: T) -> bool {
        self.elements.insert(element)
    }
    pub fn contains(&self, element: &T) -> bool {
        self.elements.contains(element)
    }
    pub fn len(&self) -> usize {
        self.elements.len()
    }
    pub fn is_empty(&self) -> bool {
        self.elements.is_empty()
    }
    pub fn value(&self) -> &HashSet<T> {
        &self.elements
    }

    /// Delta relative to `base`: the elements added since `base`.
    /// `None` when the set did not grow.
    pub fn delta_since(&self, base: &Self) -> Option<Self> {
        let elements: HashSet<T> = self.elements.difference(&base.elements).cloned().collect();
        if elements.is_empty() {
            None
        } else {
            Some(Self { elements })
        }
    }
}

impl<T: Clone + Eq + std::hash::Hash> Default for GSet<T> {
    fn default() -> Self {
        Self::new()
    }
}

impl Crdt for GSet<String> {
    type Value = HashSet<String>;

    fn merge(&mut self, other: &Self) {
        self.elements.extend(other.elements.iter().cloned());
    }

    fn value(&self) -> Self::Value {
        self.elements.clone()
    }

    fn delta_since(&self, base: &Self) -> Option<Self> {
        GSet::delta_since(self, base)
    }

    fn to_bytes(&self) -> Vec<u8> {
        let mut buf = Vec::new();
        push_u32(&mut buf, self.elements.len() as u32);
        for elem in &self.elements {
            push_string(&mut buf, elem);
        }
        buf
    }

    fn from_bytes(bytes: &[u8]) -> Option<Self> {
        let (num_elements, mut pos) = read_u32(bytes, 0)?;
        let mut elements = HashSet::new();
        for _ in 0..num_elements {
            let (s, p) = read_string(bytes, pos)?;
            elements.insert(s);
            pos = p;
        }
        Some(Self { elements })
    }
}

// =============================================================================
// ORSet
// =============================================================================

#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Tag {
    pub node_id: u32,
    pub counter: u32,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ORSet<T: Clone + Eq + std::hash::Hash> {
    pub entries: HashMap<T, HashSet<Tag>>,
    /// Tags removed by `remove`, each stamped with the removing node's
    /// logical time (a `LamportTime{counter, node_id}` where `counter` comes
    /// from the removing node's matrix clock). Subtracted on merge so
    /// removals replicate and removed elements never resurrect (same role as
    /// `AWORSet::removed`). The stamp is what lets the causal-stability
    /// watermark decide when the tombstone may be garbage-collected.
    pub removed: HashMap<Tag, LamportTime>,
    pub tag_counter: u64,
    pub node_id: u64,
    /// Per-replica observation matrix used to compute the causal-stability
    /// watermark for tombstone GC.
    pub matrix: MatrixClock,
}

impl<T: Clone + Eq + std::hash::Hash> ORSet<T> {
    pub fn new(node_id: u64) -> Self {
        Self {
            entries: HashMap::new(),
            removed: HashMap::new(),
            tag_counter: 0,
            node_id,
            matrix: MatrixClock::new(node_id),
        }
    }

    fn fresh_tag(&mut self) -> Tag {
        let tag = Tag {
            node_id: self.node_id as u32,
            counter: self.tag_counter as u32,
        };
        self.tag_counter = self.tag_counter.wrapping_add(1);
        tag
    }

    pub fn add(&mut self, element: T) {
        let tag = self.fresh_tag();
        self.entries
            .entry(element)
            .or_insert_with(HashSet::new)
            .insert(tag);
    }

    pub fn remove(&mut self, element: &T) {
        // Tombstone the observed tags instead of forgetting them: on merge,
        // replicas that still hold these tags will subtract them. The removal
        // is a distinct causal event, so every tombstoned tag is stamped with
        // one fresh matrix tick — without this the removal would be invisible
        // to observation vectors (the tags are reused, not re-created).
        if let Some(tags) = self.entries.remove(element) {
            let rt = self.matrix.tick_local();
            for tag in tags {
                self.removed.insert(
                    tag,
                    LamportTime {
                        counter: rt,
                        node_id: self.node_id,
                    },
                );
            }
        }
    }

    pub fn contains(&self, element: &T) -> bool {
        self.entries
            .get(element)
            .map_or(false, |tags| !tags.is_empty())
    }

    pub fn value(&self) -> HashSet<T> {
        self.entries
            .iter()
            .filter(|(_, tags)| !tags.is_empty())
            .map(|(elem, _)| elem.clone())
            .collect()
    }

    pub fn len(&self) -> usize {
        self.value().len()
    }
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Delta relative to `base`: for each element, the tags not present in
    /// `base`, plus the tombstones not present in `base`. Elements with no
    /// new tags are omitted. `None` when neither tags nor tombstones
    /// changed. The current `tag_counter` rides along so merging the delta
    /// advances the receiver's counter exactly like a full-state merge.
    pub fn delta_since(&self, base: &Self) -> Option<Self> {
        let mut entries = HashMap::new();
        for (element, tags) in &self.entries {
            let new_tags: HashSet<Tag> = tags
                .iter()
                .filter(|t| base.entries.get(element).map_or(true, |bt| !bt.contains(t)))
                .copied()
                .collect();
            if !new_tags.is_empty() {
                entries.insert(element.clone(), new_tags);
            }
        }
        let removed: HashMap<Tag, LamportTime> = self
            .removed
            .iter()
            .filter(|(tag, _)| !base.removed.contains_key(tag))
            .map(|(tag, rt)| (*tag, *rt))
            .collect();
        if entries.is_empty() && removed.is_empty() {
            None
        } else {
            Some(Self {
                entries,
                removed,
                tag_counter: self.tag_counter,
                node_id: self.node_id,
                matrix: self.matrix.clone(),
            })
        }
    }

    /// Garbage collect tombstones that are causally stable: a tombstone
    /// stamped `(counter: t, node_id: n)` is dropped once every healthy
    /// replica has observed logical time `t` from `n` (i.e.
    /// `watermark[n] >= t`).
    pub fn gc_tombstones(&mut self, watermark: &HashMap<u64, u64>) {
        self.removed
            .retain(|_, rt| rt.counter > watermark.get(&rt.node_id).copied().unwrap_or(0));
    }
}

impl Crdt for ORSet<String> {
    type Value = HashSet<String>;

    fn merge(&mut self, other: &Self) {
        // Tombstones first: a tag removed on either side must not survive.
        for tags in self.entries.values_mut() {
            tags.retain(|t| !other.removed.contains_key(t));
        }
        // Merge tombstones, keeping the latest removal stamp per tag.
        for (tag, rt) in &other.removed {
            match self.removed.get(tag) {
                Some(existing) if *rt <= *existing => {}
                _ => {
                    self.removed.insert(*tag, *rt);
                }
            }
        }
        for (element, tags) in &other.entries {
            let entry = self
                .entries
                .entry(element.clone())
                .or_insert_with(HashSet::new);
            for tag in tags {
                if !self.removed.contains_key(tag) {
                    entry.insert(*tag);
                }
            }
        }
        self.tag_counter = self.tag_counter.max(other.tag_counter);
        self.matrix.merge(&other.matrix);
    }

    fn value(&self) -> Self::Value {
        self.value()
    }

    fn delta_since(&self, base: &Self) -> Option<Self> {
        ORSet::delta_since(self, base)
    }

    fn to_bytes(&self) -> Vec<u8> {
        let mut buf = Vec::new();
        push_u64(&mut buf, self.node_id);
        push_u64(&mut buf, self.tag_counter);
        push_u32(&mut buf, self.entries.len() as u32);
        for (element, tags) in &self.entries {
            push_string(&mut buf, element);
            push_u32(&mut buf, tags.len() as u32);
            for tag in tags {
                push_u64(
                    &mut buf,
                    ((tag.node_id as u64) << 32) | (tag.counter as u64),
                );
            }
        }
        push_u32(&mut buf, self.removed.len() as u32);
        for (tag, rt) in &self.removed {
            push_u64(
                &mut buf,
                ((tag.node_id as u64) << 32) | (tag.counter as u64),
            );
            push_u64(&mut buf, rt.counter);
            push_u64(&mut buf, rt.node_id);
        }
        self.matrix.to_bytes(&mut buf);
        buf
    }

    fn from_bytes(bytes: &[u8]) -> Option<Self> {
        let (node_id, pos) = read_u64(bytes, 0)?;
        let (tag_counter, pos) = read_u64(bytes, pos)?;
        let (num_elements, mut pos) = read_u32(bytes, pos)?;
        let mut entries = HashMap::new();
        for _ in 0..num_elements {
            let (element, p) = read_string(bytes, pos)?;
            let (tag_count, mut p) = read_u32(bytes, p)?;
            let mut tags = HashSet::new();
            for _ in 0..tag_count {
                let (tag_val, p2) = read_u64(bytes, p)?;
                tags.insert(Tag {
                    node_id: (tag_val >> 32) as u32,
                    counter: tag_val as u32,
                });
                p = p2;
            }
            entries.insert(element, tags);
            pos = p;
        }
        let (num_removed, mut pos) = read_u32(bytes, pos)?;
        let mut removed = HashMap::new();
        for _ in 0..num_removed {
            let (tag_val, p) = read_u64(bytes, pos)?;
            pos = p;
            let (cnt, p) = read_u64(bytes, pos)?;
            pos = p;
            let (nid, p) = read_u64(bytes, pos)?;
            pos = p;
            removed.insert(
                Tag {
                    node_id: (tag_val >> 32) as u32,
                    counter: tag_val as u32,
                },
                LamportTime {
                    counter: cnt,
                    node_id: nid,
                },
            );
        }
        let matrix = MatrixClock::from_bytes(bytes, &mut pos)?;
        Some(Self {
            entries,
            removed,
            tag_counter,
            node_id,
            matrix,
        })
    }
}

// =============================================================================
// LamportTime + LamportClock
// =============================================================================

#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct LamportTime {
    pub counter: u64,
    pub node_id: u64,
}

impl LamportTime {
    pub fn new(counter: u64, node_id: u64) -> Self {
        Self { counter, node_id }
    }

    pub fn is_greater_than(&self, other: &LamportTime) -> bool {
        self.counter > other.counter
            || (self.counter == other.counter && self.node_id > other.node_id)
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct LamportClock {
    pub node_id: u64,
    pub counter: u64,
}

impl LamportClock {
    pub fn new(node_id: u64) -> Self {
        Self {
            node_id,
            counter: 0,
        }
    }

    pub fn tick(&mut self) -> LamportTime {
        self.counter += 1;
        LamportTime {
            counter: self.counter,
            node_id: self.node_id,
        }
    }
}

// =============================================================================
// MatrixClock — causal-stability watermark for tombstone GC
// =============================================================================

/// A per-CRDT matrix clock used to compute a causal-stability watermark for
/// tombstone garbage collection.
///
/// `matrix[P][N]` is the maximum logical time that node `P` has observed from
/// source node `N`; its own entry `matrix[L][L]` advances on every **local
/// mutation** — adds *and* removes — so removals are causally visible as
/// distinct events (an ORSet removal reuses existing tags, so it must be
/// timestamped or it is invisible to any observation vector).
///
/// The clock travels inside the CRDT state that is synced, so a replica that
/// merges state from `P` both learns what `P` itself observed (the `matrix[P]`
/// row) and absorbs `P`'s observations into its own row.
///
/// A tombstone created by node `N` at logical time `t` is **causally stable**
/// once every healthy replica's row satisfies `matrix[R][N] >= t`; only then
/// may it be dropped without risking element resurrection.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct MatrixClock {
    pub local_node: u64,
    /// peer node -> (source node -> max logical time observed).
    pub matrix: HashMap<u64, HashMap<u64, u64>>,
}

impl MatrixClock {
    pub fn new(local_node: u64) -> Self {
        Self {
            local_node,
            matrix: HashMap::new(),
        }
    }

    /// Advance the local node's own counter (one tick per local mutation) and
    /// return the new value.
    pub fn tick_local(&mut self) -> u64 {
        let row = self.matrix.entry(self.local_node).or_default();
        let e = row.entry(self.local_node).or_insert(0);
        *e = e.wrapping_add(1);
        *e
    }

    /// Merge another replica's matrix clock into ours.
    pub fn merge(&mut self, other: &Self) {
        for (peer, row) in &other.matrix {
            {
                let local = self.matrix.entry(self.local_node).or_default();
                for (src, t) in row {
                    let e = local.entry(*src).or_insert(0);
                    *e = (*e).max(*t);
                }
            }
            {
                let dest = self.matrix.entry(*peer).or_default();
                for (src, t) in row {
                    let e = dest.entry(*src).or_insert(0);
                    *e = (*e).max(*t);
                }
            }
        }
    }

    /// The causal-stability watermark: for each source node `N`, the maximum
    /// logical time from `N` that **every** replica in `{local} ∪ healthy`
    /// has observed. A healthy replica with no recorded row is treated as
    /// having observed nothing (so a fresh/partitioned peer blocks GC), and
    /// the local node is always considered.
    pub fn watermark(&self, healthy: &[u64]) -> HashMap<u64, u64> {
        let mut sources: Vec<u64> = Vec::new();
        for row in self.matrix.values() {
            for s in row.keys() {
                if !sources.contains(s) {
                    sources.push(*s);
                }
            }
        }
        let mut wm = HashMap::new();
        for src in sources {
            let local_t = self
                .matrix
                .get(&self.local_node)
                .and_then(|r| r.get(&src))
                .copied()
                .unwrap_or(0);
            let mut min = local_t;
            for peer in healthy {
                let t = self
                    .matrix
                    .get(peer)
                    .and_then(|r| r.get(&src))
                    .copied()
                    .unwrap_or(0);
                min = min.min(t);
            }
            wm.insert(src, min);
        }
        wm
    }

    /// Re-root this clock to a different local node id.
    ///
    /// Used when a replica is created from an incoming payload (or a persisted
    /// snapshot): the parsed clock's `local_node` is the *sender's* id. After
    /// re-rooting, the new node absorbs every observation the payload carried
    /// into its own row and keeps each peer's row, so its view is attributed
    /// to the correct node from then on.
    pub fn reroot(&mut self, new_local: u64) {
        if self.local_node == new_local {
            return;
        }
        let old = std::mem::replace(self, MatrixClock::new(new_local));
        self.merge(&old);
    }

    /// Drop rows for peers that are no longer healthy (keeps the clock from
    /// growing without bound as nodes churn). The local row is always kept.
    pub fn prune(&mut self, healthy: &[u64]) {
        self.matrix
            .retain(|peer, _| *peer == self.local_node || healthy.contains(peer));
    }

    pub(crate) fn to_bytes(&self, buf: &mut Vec<u8>) {
        push_u64(buf, self.local_node);
        let mut peers: Vec<u64> = self.matrix.keys().copied().collect();
        peers.sort_unstable();
        push_u32(buf, peers.len() as u32);
        for peer in peers {
            let row = &self.matrix[&peer];
            push_u64(buf, peer);
            let mut srcs: Vec<u64> = row.keys().copied().collect();
            srcs.sort_unstable();
            push_u32(buf, srcs.len() as u32);
            for src in srcs {
                push_u64(buf, src);
                push_u64(buf, row[&src]);
            }
        }
    }

    pub(crate) fn from_bytes(bytes: &[u8], pos: &mut usize) -> Option<Self> {
        let (local_node, p) = read_u64(bytes, *pos)?;
        *pos = p;
        let (num_peers, p) = read_u32(bytes, *pos)?;
        *pos = p;
        let mut matrix = HashMap::new();
        for _ in 0..num_peers {
            let (peer, p) = read_u64(bytes, *pos)?;
            *pos = p;
            let (num_srcs, p) = read_u32(bytes, *pos)?;
            *pos = p;
            let mut row = HashMap::new();
            for _ in 0..num_srcs {
                let (src, p) = read_u64(bytes, *pos)?;
                *pos = p;
                let (t, p) = read_u64(bytes, *pos)?;
                *pos = p;
                row.insert(src, t);
            }
            matrix.insert(peer, row);
        }
        Some(Self { local_node, matrix })
    }
}

// =============================================================================
// AWORSet
// =============================================================================

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct AWORSet<T: Clone + Eq + std::hash::Hash> {
    pub entries: HashMap<T, LamportTime>,
    pub removed: HashMap<T, LamportTime>,
    pub clock: LamportClock,
    /// Per-replica observation matrix used to compute the causal-stability
    /// watermark for tombstone GC. Local add/remove timestamps are drawn
    /// from it so removals are causally comparable to the watermark.
    pub matrix: MatrixClock,
}

impl<T: Clone + Eq + std::hash::Hash> AWORSet<T> {
    pub fn new(node_id: u64) -> Self {
        Self {
            entries: HashMap::new(),
            removed: HashMap::new(),
            clock: LamportClock::new(node_id),
            matrix: MatrixClock::new(node_id),
        }
    }

    pub fn add(&mut self, element: T) {
        let ts = self.matrix.tick_local();
        self.clock.counter = self.clock.counter.max(ts);
        self.entries.insert(
            element,
            LamportTime {
                counter: ts,
                node_id: self.clock.node_id,
            },
        );
    }

    pub fn remove(&mut self, element: &T) {
        let ts = self.matrix.tick_local();
        self.clock.counter = self.clock.counter.max(ts);
        self.removed.insert(
            element.clone(),
            LamportTime {
                counter: ts,
                node_id: self.clock.node_id,
            },
        );
    }

    pub fn contains(&self, element: &T) -> bool {
        match (self.entries.get(element), self.removed.get(element)) {
            (Some(add_ts), Some(rem_ts)) => add_ts.is_greater_than(rem_ts),
            (Some(_), None) => true,
            (None, _) => false,
        }
    }

    pub fn value(&self) -> HashSet<T> {
        self.entries
            .keys()
            .filter(|e| self.contains(e))
            .cloned()
            .collect()
    }

    pub fn len(&self) -> usize {
        self.value().len()
    }
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Delta relative to `base`: add/remove timestamps strictly newer than
    /// `base`'s for the same element. `None` when nothing changed. The
    /// current clock rides along so merging the delta advances the
    /// receiver's clock exactly like a full-state merge.
    pub fn delta_since(&self, base: &Self) -> Option<Self> {
        let mut entries = HashMap::new();
        for (element, ts) in &self.entries {
            if base.entries.get(element).map_or(true, |bts| ts > bts) {
                entries.insert(element.clone(), *ts);
            }
        }
        let mut removed = HashMap::new();
        for (element, ts) in &self.removed {
            if base.removed.get(element).map_or(true, |bts| ts > bts) {
                removed.insert(element.clone(), *ts);
            }
        }
        if entries.is_empty() && removed.is_empty() {
            None
        } else {
            Some(Self {
                entries,
                removed,
                clock: self.clock,
                matrix: self.matrix.clone(),
            })
        }
    }

    /// Garbage collect tombstones that are causally stable: a removal
    /// stamped `(counter: t, node_id: n)` is dropped once every healthy
    /// replica has observed logical time `t` from `n` (watermark[n] >= t).
    pub fn gc_tombstones(&mut self, watermark: &HashMap<u64, u64>) {
        self.removed
            .retain(|_, ts| ts.counter > watermark.get(&ts.node_id).copied().unwrap_or(0));
    }
}

impl Crdt for AWORSet<String> {
    type Value = HashSet<String>;

    fn merge(&mut self, other: &Self) {
        for (element, ts) in &other.entries {
            match self.entries.get(element) {
                Some(existing) if *ts <= *existing => {}
                _ => {
                    self.entries.insert(element.clone(), *ts);
                }
            }
        }
        for (element, ts) in &other.removed {
            match self.removed.get(element) {
                Some(existing) if *ts <= *existing => {}
                _ => {
                    self.removed.insert(element.clone(), *ts);
                }
            }
        }
        self.clock.counter = self.clock.counter.max(other.clock.counter);
        self.matrix.merge(&other.matrix);
    }

    fn value(&self) -> Self::Value {
        self.value()
    }

    fn delta_since(&self, base: &Self) -> Option<Self> {
        AWORSet::delta_since(self, base)
    }

    fn to_bytes(&self) -> Vec<u8> {
        let mut buf = Vec::new();
        push_u64(&mut buf, self.clock.node_id);
        push_u64(&mut buf, self.clock.counter);
        push_u32(&mut buf, self.entries.len() as u32);
        for (element, ts) in &self.entries {
            push_string(&mut buf, element);
            push_u64(&mut buf, ts.counter);
            push_u64(&mut buf, ts.node_id);
        }
        push_u32(&mut buf, self.removed.len() as u32);
        for (element, ts) in &self.removed {
            push_string(&mut buf, element);
            push_u64(&mut buf, ts.counter);
            push_u64(&mut buf, ts.node_id);
        }
        self.matrix.to_bytes(&mut buf);
        buf
    }

    fn from_bytes(bytes: &[u8]) -> Option<Self> {
        let (node_id, pos) = read_u64(bytes, 0)?;
        let (counter, pos) = read_u64(bytes, pos)?;
        let (num_entries, mut pos) = read_u32(bytes, pos)?;
        let mut entries = HashMap::new();
        for _ in 0..num_entries {
            let (element, p) = read_string(bytes, pos)?;
            let (cnt, p) = read_u64(bytes, p)?;
            let (nid, p) = read_u64(bytes, p)?;
            entries.insert(
                element,
                LamportTime {
                    counter: cnt,
                    node_id: nid,
                },
            );
            pos = p;
        }
        let (num_removed, mut pos) = read_u32(bytes, pos)?;
        let mut removed = HashMap::new();
        for _ in 0..num_removed {
            let (element, p) = read_string(bytes, pos)?;
            let (cnt, p) = read_u64(bytes, p)?;
            let (nid, p) = read_u64(bytes, p)?;
            removed.insert(
                element,
                LamportTime {
                    counter: cnt,
                    node_id: nid,
                },
            );
            pos = p;
        }
        let matrix = MatrixClock::from_bytes(bytes, &mut pos)?;
        Some(Self {
            entries,
            removed,
            clock: LamportClock { node_id, counter },
            matrix,
        })
    }
}

// =============================================================================
// Tests
// =============================================================================

#[cfg(test)]
mod tests {
    use super::*;

    // ---- GCounter ----

    #[test]
    fn test_gcounter_increment() {
        let mut c = GCounter::new(1);
        c.increment();
        assert_eq!(c.value(), 1);
        c.increment();
        assert_eq!(c.value(), 2);
    }

    #[test]
    fn test_gcounter_value() {
        let mut c = GCounter::new(1);
        c.increment_by(5);
        assert_eq!(c.value(), 5);
    }

    #[test]
    fn test_gcounter_merge() {
        let mut a = GCounter::new(1);
        a.increment_by(3);
        let mut b = GCounter::new(2);
        b.increment_by(5);
        a.merge(&b);
        assert_eq!(a.value(), 8);
    }

    #[test]
    fn test_gcounter_merge_commutative() {
        let mut a = GCounter::new(1);
        a.increment_by(3);
        let b_orig = a.clone();
        let mut b = GCounter::new(2);
        b.increment_by(5);
        let c_orig = b.clone();
        a.merge(&c_orig);
        let mut b = b_orig;
        b.merge(&c_orig);
        // They both merged with the same state
        let mut a2 = GCounter::new(1);
        a2.increment_by(3);
        let mut b2 = GCounter::new(2);
        b2.increment_by(5);
        a2.merge(&b2);
        b2.merge(&a2);
        assert_eq!(a2.value(), b2.value());
    }

    #[test]
    fn test_gcounter_merge_idempotent() {
        let mut a = GCounter::new(1);
        a.increment_by(10);
        let snapshot = a.clone();
        a.merge(&snapshot);
        assert_eq!(a.value(), 10);
    }

    #[test]
    fn test_gcounter_serialize_roundtrip() {
        let mut c = GCounter::new(42);
        c.increment_by(100);
        let bytes = c.to_bytes();
        let restored = GCounter::from_bytes(&bytes).unwrap();
        assert_eq!(c, restored);
        assert_eq!(restored.value(), 100);
    }

    // ---- PNCounter ----

    #[test]
    fn test_pncounter_increment() {
        let mut c = PNCounter::new(1);
        c.increment();
        c.increment();
        assert_eq!(c.value(), 2);
    }

    #[test]
    fn test_pncounter_decrement() {
        let mut c = PNCounter::new(1);
        c.increment_by(5);
        c.decrement_by(2);
        assert_eq!(c.value(), 3);
    }

    #[test]
    fn test_pncounter_net_negative() {
        let mut c = PNCounter::new(1);
        c.decrement_by(3);
        assert_eq!(c.value(), -3);
    }

    #[test]
    fn test_pncounter_merge() {
        let mut a = PNCounter::new(1);
        a.increment_by(3);
        let mut b = PNCounter::new(2);
        b.increment_by(5);
        b.decrement_by(2);
        a.merge(&b);
        assert_eq!(a.value(), 6);
    }

    #[test]
    fn test_pncounter_merge_commutative() {
        let mut a = PNCounter::new(1);
        a.increment_by(3);
        let mut b = PNCounter::new(2);
        b.increment_by(5);
        let a_snap = a.clone();
        let b_snap = b.clone();
        a.merge(&b_snap);
        let mut b = b_snap.clone();
        b.merge(&a_snap);
        assert_eq!(a.value(), b.value());
    }

    #[test]
    fn test_pncounter_serialize_roundtrip() {
        let mut c = PNCounter::new(42);
        c.increment_by(10);
        c.decrement_by(3);
        let bytes = c.to_bytes();
        let restored = PNCounter::from_bytes(&bytes).unwrap();
        assert_eq!(c.value(), restored.value());
    }

    #[test]
    fn test_pncounter_from_bytes_rejects_malformed_payloads() {
        // Regression: `num_entries` comes off the wire. A crafted packet
        // claiming far more entries than the payload holds must return
        // None instead of panicking the scheduler thread.
        let mut crafted = Vec::new();
        crafted.extend_from_slice(&1u64.to_be_bytes()); // node_id
        crafted.extend_from_slice(&u32::MAX.to_be_bytes()); // bogus entry count
        assert_eq!(PNCounter::from_bytes(&crafted), None);

        // Every strict prefix of a valid payload must also fail cleanly
        // (valid increments with truncated decrements included).
        let mut c = PNCounter::new(1);
        c.increment_by(2);
        c.decrement_by(1);
        let bytes = c.to_bytes();
        for len in 0..bytes.len() {
            assert_eq!(
                PNCounter::from_bytes(&bytes[..len]),
                None,
                "prefix of len {len} must not parse"
            );
        }
        // And the full payload still parses.
        assert!(PNCounter::from_bytes(&bytes).is_some());
    }

    // ---- GSet ----

    #[test]
    fn test_gset_insert() {
        let mut s = GSet::<String>::new();
        assert!(s.insert("a".to_string()));
        assert!(!s.insert("a".to_string()));
        assert!(s.contains(&"a".to_string()));
    }

    #[test]
    fn test_gset_merge() {
        let mut a = GSet::<String>::new();
        a.insert("x".to_string());
        let mut b = GSet::<String>::new();
        b.insert("y".to_string());
        a.merge(&b);
        assert!(a.contains(&"x".to_string()));
        assert!(a.contains(&"y".to_string()));
    }

    #[test]
    fn test_gset_merge_commutative() {
        let mut a = GSet::<String>::new();
        a.insert("x".to_string());
        let mut b = GSet::<String>::new();
        b.insert("y".to_string());
        let a_snap = a.clone();
        let b_snap = b.clone();
        a.merge(&b_snap);
        let mut b = b_snap.clone();
        b.merge(&a_snap);
        assert_eq!(a.value(), b.value());
    }

    #[test]
    fn test_gset_serialize_roundtrip() {
        let mut s = GSet::<String>::new();
        s.insert("hello".to_string());
        s.insert("world".to_string());
        let bytes = s.to_bytes();
        let restored = GSet::<String>::from_bytes(&bytes).unwrap();
        assert_eq!(s.value(), restored.value());
    }

    // ---- ORSet ----

    #[test]
    fn test_orset_add() {
        let mut s = ORSet::<String>::new(1);
        s.add("a".to_string());
        assert!(s.contains(&"a".to_string()));
    }

    #[test]
    fn test_orset_remove() {
        let mut s = ORSet::<String>::new(1);
        s.add("a".to_string());
        s.remove(&"a".to_string());
        assert!(!s.contains(&"a".to_string()));
    }

    #[test]
    fn test_orset_add_wins() {
        let mut a = ORSet::<String>::new(1);
        a.add("x".to_string());
        let mut b = a.clone();
        a.add("x".to_string());
        b.remove(&"x".to_string());
        a.merge(&b);
        assert!(a.contains(&"x".to_string()), "add should win");
    }

    #[test]
    fn test_orset_merge() {
        let mut a = ORSet::<String>::new(1);
        a.add("x".to_string());
        let mut b = ORSet::<String>::new(2);
        b.add("y".to_string());
        a.merge(&b);
        assert!(a.contains(&"x".to_string()));
        assert!(a.contains(&"y".to_string()));
    }

    #[test]
    fn test_orset_merge_commutative() {
        let mut a = ORSet::<String>::new(1);
        a.add("x".to_string());
        let mut b = ORSet::<String>::new(2);
        b.add("y".to_string());
        let b_snap = b.clone();
        a.merge(&b_snap);
        let mut b = b_snap.clone();
        b.merge(&a.clone());
        assert_eq!(a.value(), b.value());
    }

    #[test]
    fn test_orset_serialize_roundtrip() {
        let mut s = ORSet::<String>::new(1);
        s.add("hello".to_string());
        s.add("world".to_string());
        s.remove(&"hello".to_string()); // exercise the tombstone section
        let bytes = s.to_bytes();
        let restored = ORSet::<String>::from_bytes(&bytes).unwrap();
        assert_eq!(s, restored);
    }

    #[test]
    fn test_orset_remove_then_merge_stays_removed() {
        // Regression: a removed element must not resurrect when merging with
        // a replica that still holds it.
        let mut a = ORSet::<String>::new(1);
        a.add("x".to_string());
        let mut b = a.clone();
        b.remove(&"x".to_string());

        a.merge(&b);
        assert!(!a.contains(&"x".to_string()), "removed element resurrected");

        // Merge order must not matter (commutativity).
        let mut a = ORSet::<String>::new(1);
        a.add("x".to_string());
        let mut b = a.clone();
        b.remove(&"x".to_string());
        b.merge(&a);
        assert!(!b.contains(&"x".to_string()), "removed element resurrected");
    }

    #[test]
    fn test_orset_remove_replicates_via_delta() {
        // Regression: removals must ride in `delta_since`, otherwise they
        // never propagate over delta-state replication.
        let mut base = ORSet::<String>::new(1);
        base.add("x".to_string());
        let mut full = base.clone();
        full.remove(&"x".to_string());

        let delta = full
            .delta_since(&base)
            .expect("removal must produce a delta");
        assert!(delta.entries.is_empty());
        assert_eq!(delta.removed.len(), 1);

        let mut via_delta = base.clone();
        via_delta.merge(&delta);
        assert!(!via_delta.contains(&"x".to_string()));

        let mut via_full = base.clone();
        via_full.merge(&full);
        assert_eq!(via_delta, via_full);
    }

    #[test]
    fn test_orset_tag_counter_wraparound_does_not_panic() {
        let mut s = ORSet::<String>::new(1);
        s.tag_counter = u64::MAX;
        s.add("should-not-panic".to_string());
        assert!(s.contains(&"should-not-panic".to_string()));
    }

    // ---- AWORSet ----

    #[test]
    fn test_aworset_add() {
        let mut s = AWORSet::<String>::new(1);
        s.add("a".to_string());
        assert!(s.contains(&"a".to_string()));
    }

    #[test]
    fn test_aworset_remove() {
        let mut s = AWORSet::<String>::new(1);
        s.add("a".to_string());
        s.remove(&"a".to_string());
        assert!(!s.contains(&"a".to_string()));
    }

    #[test]
    fn test_aworset_timestamp_wins() {
        let mut a = AWORSet::<String>::new(1);
        a.add("x".to_string());
        let mut b = a.clone();
        b.remove(&"x".to_string());
        b.add("x".to_string());
        a.merge(&b);
        assert!(a.contains(&"x".to_string()), "later add should win");
    }

    #[test]
    fn test_aworset_merge() {
        let mut a = AWORSet::<String>::new(1);
        a.add("x".to_string());
        let mut b = AWORSet::<String>::new(2);
        b.add("y".to_string());
        a.merge(&b);
        assert!(a.contains(&"x".to_string()));
        assert!(a.contains(&"y".to_string()));
    }

    #[test]
    fn test_aworset_merge_commutative() {
        let mut a = AWORSet::<String>::new(1);
        a.add("x".to_string());
        let mut b = AWORSet::<String>::new(2);
        b.add("y".to_string());
        let b_snap = b.clone();
        a.merge(&b_snap);
        let mut b = b_snap.clone();
        b.merge(&a.clone());
        assert_eq!(a.value(), b.value());
    }

    #[test]
    fn test_aworset_serialize_roundtrip() {
        let mut s = AWORSet::<String>::new(1);
        s.add("hello".to_string());
        s.add("world".to_string());
        let bytes = s.to_bytes();
        let restored = AWORSet::<String>::from_bytes(&bytes).unwrap();
        assert_eq!(s.value(), restored.value());
    }

    #[test]
    fn test_lamport_time_ordering() {
        let a = LamportTime {
            counter: 1,
            node_id: 1,
        };
        let b = LamportTime {
            counter: 2,
            node_id: 1,
        };
        let c = LamportTime {
            counter: 2,
            node_id: 2,
        };
        assert!(b.is_greater_than(&a));
        assert!(c.is_greater_than(&b));
    }

    // ---- Delta-state replication ----
    //
    // For each type: merging the delta into a replica that already holds
    // `base` must produce exactly the same state as merging the full state.

    #[test]
    fn test_gcounter_delta_since_unchanged() {
        let mut c = GCounter::new(1);
        c.increment_by(5);
        assert!(c.delta_since(&c.clone()).is_none());
    }

    #[test]
    fn test_gcounter_delta_merge_equals_full_merge() {
        let mut base = GCounter::new(1);
        base.increment_by(3);
        let mut full = base.clone();
        full.increment_by(4);

        let delta = full.delta_since(&base).expect("counter grew");
        // The delta carries only the updated entry, not the whole map.
        assert_eq!(delta.counts.len(), 1);

        let mut via_delta = base.clone();
        via_delta.merge(&delta);
        let mut via_full = base.clone();
        via_full.merge(&full);
        assert_eq!(via_delta, via_full);
    }

    #[test]
    fn test_pncounter_delta_merge_equals_full_merge() {
        let mut base = PNCounter::new(1);
        base.increment_by(5);
        let mut full = base.clone();
        full.decrement_by(2);

        let delta = full.delta_since(&base).expect("decrements grew");
        assert!(delta.increments.counts.is_empty());

        let mut via_delta = base.clone();
        via_delta.merge(&delta);
        let mut via_full = base.clone();
        via_full.merge(&full);
        assert_eq!(via_delta, via_full);
    }

    #[test]
    fn test_gset_delta_merge_equals_full_merge() {
        let mut base = GSet::<String>::new();
        base.insert("a".to_string());
        let mut full = base.clone();
        full.insert("b".to_string());

        let delta = full.delta_since(&base).expect("set grew");
        assert_eq!(delta.elements.len(), 1);

        let mut via_delta = base.clone();
        via_delta.merge(&delta);
        let mut via_full = base.clone();
        via_full.merge(&full);
        assert_eq!(via_delta, via_full);
    }

    #[test]
    fn test_orset_delta_merge_equals_full_merge() {
        let mut base = ORSet::<String>::new(1);
        base.add("a".to_string());
        let mut full = base.clone();
        full.add("b".to_string());
        full.add("a".to_string()); // second tag on an existing element

        let delta = full.delta_since(&base).expect("tags added");
        // Only the new tags travel: one for "a", one for "b".
        assert_eq!(delta.entries.values().map(|t| t.len()).sum::<usize>(), 2);

        let mut via_delta = base.clone();
        via_delta.merge(&delta);
        let mut via_full = base.clone();
        via_full.merge(&full);
        assert_eq!(via_delta, via_full);
    }

    #[test]
    fn test_aworset_delta_merge_equals_full_merge() {
        let mut base = AWORSet::<String>::new(1);
        base.add("a".to_string());
        let mut full = base.clone();
        full.add("b".to_string());
        full.remove(&"a".to_string());

        let delta = full.delta_since(&base).expect("timestamps advanced");
        assert_eq!(delta.entries.len(), 1);
        assert_eq!(delta.removed.len(), 1);

        let mut via_delta = base.clone();
        via_delta.merge(&delta);
        let mut via_full = base.clone();
        via_full.merge(&full);
        assert_eq!(via_delta, via_full);
    }

    #[test]
    fn test_delta_since_empty_base_is_full_state() {
        // With an empty base every entry is new: the delta must carry the
        // complete state (this is what makes the join fallback work).
        let mut c = GCounter::new(1);
        c.increment_by(9);
        let delta = c.delta_since(&GCounter::new(1)).expect("everything is new");
        let mut merged = GCounter::new(2);
        merged.merge(&delta);
        assert_eq!(merged.value(), 9);
    }

    #[test]
    fn test_orset_tombstone_gc_respects_watermark() {
        let mut orset = ORSet::new(1);
        let item = "item".to_string();

        // Each add/remove cycle creates one tombstone stamped with an
        // increasing local matrix tick (removal times 1, 2, 3, ...).
        for _i in 0..100 {
            orset.add(item.clone());
            orset.remove(&item);
        }
        assert_eq!(orset.removed.len(), 100);

        // A watermark covering only the first 50 removal stamps reclaims
        // exactly those; the rest are not yet causally stable.
        let mut partial = HashMap::new();
        partial.insert(1u64, 50u64);
        let mut copy = orset.clone();
        copy.gc_tombstones(&partial);
        assert_eq!(copy.removed.len(), 50, "only stable tombstones purged");

        // Full observation reclaims everything.
        let mut full = HashMap::new();
        full.insert(1u64, u64::MAX);
        orset.gc_tombstones(&full);
        assert!(orset.removed.is_empty(), "all tombstones should be purged");
    }
}