use std::collections::{BTreeMap, HashMap};
use crate::stable_id::Block;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct OpId {
counter: u64,
peer: u64,
}
impl OpId {
pub fn counter(&self) -> u64 {
self.counter
}
pub fn peer(&self) -> u64 {
self.peer
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct TextOp {
pub id: OpId,
pub ch: char,
pub origin: Option<OpId>,
pub deleted: Option<OpId>,
}
pub type TextVersionVector = BTreeMap<u64, u64>;
#[derive(Debug, Clone)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
struct Elem {
ch: char,
origin: Option<OpId>,
deleted: Option<OpId>,
}
#[derive(Debug, Clone)]
pub struct TextCrdt {
elems: HashMap<OpId, Elem>,
peer: u64,
counter: u64,
}
impl TextCrdt {
pub fn new(peer: u64) -> Self {
Self {
elems: HashMap::new(),
peer,
counter: 0,
}
}
pub fn from_str(peer: u64, s: &str) -> Self {
let mut t = Self::new(peer);
t.append_root_chain(s);
t
}
pub fn fork(&self, peer: u64) -> Self {
Self {
elems: self.elems.clone(),
peer,
counter: self.counter,
}
}
fn next_id(&mut self) -> OpId {
self.counter += 1;
OpId {
counter: self.counter,
peer: self.peer,
}
}
pub fn clock(&self) -> OpId {
OpId {
counter: self.counter,
peer: self.peer,
}
}
pub fn insert(&mut self, index: usize, ch: char) {
let visible = self.ordered_ids(false);
let origin = if index == 0 {
None
} else {
visible.get(index - 1).copied()
};
let id = self.next_id();
self.elems.insert(
id,
Elem {
ch,
origin,
deleted: None,
},
);
}
pub fn insert_str(&mut self, index: usize, s: &str) {
for (i, ch) in s.chars().enumerate() {
self.insert(index + i, ch);
}
}
fn append_root_chain(&mut self, s: &str) {
let mut origin = None;
for ch in s.chars() {
let id = self.next_id();
self.elems.insert(
id,
Elem {
ch,
origin,
deleted: None,
},
);
origin = Some(id);
}
}
pub fn replace_all(&mut self, s: &str) {
let visible_ids: Vec<OpId> = self
.elems
.iter()
.filter_map(|(id, elem)| elem.deleted.is_none().then_some(*id))
.collect();
for id in visible_ids {
let deleted = self.next_id();
if let Some(elem) = self.elems.get_mut(&id)
&& elem.deleted.is_none()
{
elem.deleted = Some(deleted);
}
}
self.append_root_chain(s);
}
pub fn delete(&mut self, index: usize) {
let visible = self.ordered_ids(false);
if let Some(id) = visible.get(index).copied() {
let del = self.next_id();
if let Some(e) = self.elems.get_mut(&id)
&& e.deleted.is_none()
{
e.deleted = Some(del);
}
}
}
pub fn text(&self) -> String {
self.ordered_ids(false)
.into_iter()
.filter_map(|id| self.elems.get(&id).map(|e| e.ch))
.collect()
}
pub fn len(&self) -> usize {
self.elems.values().filter(|e| e.deleted.is_none()).count()
}
pub fn tombstone_count(&self) -> usize {
self.elems.values().filter(|e| e.deleted.is_some()).count()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
fn ordered_ids(&self, include_deleted: bool) -> Vec<OpId> {
let mut children: HashMap<Option<OpId>, Vec<OpId>> = HashMap::new();
for (id, e) in &self.elems {
children.entry(e.origin).or_default().push(*id);
}
for list in children.values_mut() {
list.sort_unstable_by(|a, b| b.cmp(a));
}
let mut out = Vec::with_capacity(self.elems.len());
let mut stack: Vec<OpId> = children
.get(&None)
.cloned()
.unwrap_or_default()
.into_iter()
.rev()
.collect();
while let Some(id) = stack.pop() {
let e = &self.elems[&id];
if include_deleted || e.deleted.is_none() {
out.push(id);
}
if let Some(kids) = children.get(&Some(id)) {
for &k in kids.iter().rev() {
stack.push(k);
}
}
}
out
}
pub fn merge(&mut self, other: &TextCrdt) -> bool {
let before = self.text();
for (id, oe) in &other.elems {
self.counter = self.counter.max(id.counter);
if let Some(d) = oe.deleted {
self.counter = self.counter.max(d.counter);
}
match self.elems.get_mut(id) {
Some(e) => {
e.deleted = match (e.deleted, oe.deleted) {
(Some(a), Some(b)) => Some(a.min(b)),
(a, b) => a.or(b),
};
}
None => {
self.elems.insert(*id, oe.clone());
}
}
}
self.text() != before
}
pub fn gc_with(&mut self, is_stable: impl Fn(OpId) -> bool) -> usize {
let mut removed = 0;
loop {
let referenced: std::collections::HashSet<OpId> =
self.elems.values().filter_map(|e| e.origin).collect();
let collectable: Vec<OpId> = self
.elems
.iter()
.filter(|(id, e)| e.deleted.is_some_and(&is_stable) && !referenced.contains(id))
.map(|(id, _)| *id)
.collect();
if collectable.is_empty() {
break;
}
for id in collectable {
self.elems.remove(&id);
removed += 1;
}
}
removed
}
}
impl TextCrdt {
pub fn version_vector(&self) -> TextVersionVector {
let mut vv = TextVersionVector::new();
let mut bump = |id: OpId| {
let slot = vv.entry(id.peer()).or_insert(0);
*slot = (*slot).max(id.counter());
};
for (id, elem) in &self.elems {
bump(*id);
if let Some(d) = elem.deleted {
bump(d);
}
}
vv
}
pub fn delta_since(&self, their_vv: &TextVersionVector) -> Vec<TextOp> {
let seen = |id: OpId| id.counter() <= their_vv.get(&id.peer()).copied().unwrap_or(0);
self.elems
.iter()
.filter_map(|(id, elem)| {
let insert_new = !seen(*id);
let delete_new = elem.deleted.is_some_and(|d| !seen(d));
(insert_new || delete_new).then_some(TextOp {
id: *id,
ch: elem.ch,
origin: elem.origin,
deleted: elem.deleted,
})
})
.collect()
}
pub fn apply_delta(&mut self, ops: &[TextOp]) -> bool {
let before = self.text();
for op in ops {
self.counter = self.counter.max(op.id.counter());
if let Some(d) = op.deleted {
self.counter = self.counter.max(d.counter());
}
match self.elems.get_mut(&op.id) {
Some(e) => {
e.deleted = match (e.deleted, op.deleted) {
(Some(a), Some(b)) => Some(a.min(b)),
(a, b) => a.or(b),
};
}
None => {
self.elems.insert(
op.id,
Elem {
ch: op.ch,
origin: op.origin,
deleted: op.deleted,
},
);
}
}
}
self.text() != before
}
}
pub fn parse_blocks(text: &str) -> Vec<Block> {
text.split("\n\n")
.map(str::trim)
.filter(|p| !p.is_empty())
.map(Block::text)
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn delta_sync_converges_two_replicas() {
let base = TextCrdt::from_str(0, "hello\n");
let mut a = base.fork(1);
a.insert_str(a.len(), "world\n"); let mut b = base.fork(2);
b.delete(0);
let a_delta = a.delta_since(&b.version_vector());
let b_delta = b.delta_since(&a.version_vector());
assert!(a.apply_delta(&b_delta));
b.apply_delta(&a_delta);
assert_eq!(a.text(), b.text(), "replicas converge after delta exchange");
assert_eq!(a.text(), "ello\nworld\n");
}
#[test]
fn full_snapshot_delta_reconstructs_a_mergeable_replica() {
let mut canonical = TextCrdt::from_str(1, "base\n");
let snapshot = canonical.delta_since(&TextVersionVector::new());
let mut member = TextCrdt::new(2);
member.apply_delta(&snapshot);
assert_eq!(member.text(), "base\n");
canonical.insert_str(canonical.len(), "A\n");
member.insert_str(member.len(), "B\n");
let to_member = canonical.delta_since(&member.version_vector());
let to_canonical = member.delta_since(&canonical.version_vector());
canonical.apply_delta(&to_canonical);
member.apply_delta(&to_member);
assert_eq!(
canonical.text(),
member.text(),
"shared-identity convergence"
);
}
#[test]
fn from_str_seeds_a_large_buffer_as_one_linear_chain() {
let text = "0123456789abcdef\n".repeat(512);
let t = TextCrdt::from_str(7, &text);
assert_eq!(t.text(), text);
assert_eq!(t.elems.len(), text.chars().count());
assert_eq!(
t.version_vector().get(&7).copied(),
Some(t.elems.len() as u64)
);
}
#[test]
fn whole_document_replace_deltas_converge_without_duplication() {
let base = TextCrdt::from_str(1, "old heading\nold body\n");
let mut canonical = base.clone();
let mut member = TextCrdt::new(2);
member.apply_delta(&base.delta_since(&TextVersionVector::new()));
canonical.replace_all("new heading\nnew body\n");
let delta = canonical.delta_since(&member.version_vector());
assert!(member.apply_delta(&delta));
assert_eq!(canonical.text(), "new heading\nnew body\n");
assert_eq!(member.text(), canonical.text());
}
#[test]
fn delta_apply_is_idempotent() {
let a = TextCrdt::from_str(1, "abc\n");
let mut b = TextCrdt::new(2);
let delta = a.delta_since(&TextVersionVector::new());
assert!(b.apply_delta(&delta));
assert!(!b.apply_delta(&delta), "re-applying a delta is a no-op");
assert_eq!(b.text(), a.text());
}
#[test]
fn local_insert_and_delete() {
let mut t = TextCrdt::from_str(1, "helo");
t.insert(3, 'l'); assert_eq!(t.text(), "hello");
t.insert(5, '!'); assert_eq!(t.text(), "hello!");
t.delete(0); assert_eq!(t.text(), "ello!");
assert_eq!(t.len(), 5);
}
#[test]
fn concurrent_inserts_converge_keeping_both() {
let mut a = TextCrdt::from_str(1, "hi");
let mut b = a.fork(2);
a.insert(2, '!'); b.insert(0, 'O'); let changed = a.merge(&b);
b.merge(&a);
assert!(changed);
assert_eq!(a.text(), b.text(), "replicas converge");
assert!(a.text().contains('!') && a.text().contains('O'));
assert_eq!(a.text().len(), 4);
}
#[test]
fn concurrent_inserts_at_same_spot_converge_deterministically() {
let mut a = TextCrdt::from_str(1, "XY");
let mut b = a.fork(2);
a.insert(1, 'a'); b.insert(1, 'b'); a.merge(&b);
b.merge(&a);
assert_eq!(a.text(), b.text());
assert_eq!(a.text().len(), 4);
assert!(a.text().starts_with('X') && a.text().ends_with('Y'));
}
#[test]
fn concurrent_insert_and_delete_merge() {
let mut a = TextCrdt::from_str(1, "abc");
let mut b = a.fork(2);
a.delete(1); b.insert(3, 'd'); a.merge(&b);
b.merge(&a);
assert_eq!(a.text(), b.text());
assert_eq!(a.text(), "acd"); }
#[test]
fn merge_is_idempotent_and_commutative() {
let mut a = TextCrdt::from_str(1, "one");
let mut b = a.fork(2);
a.insert(3, 'X');
b.insert(0, 'Y');
let mut ab = a.clone();
ab.merge(&b);
ab.merge(&b); let mut ba = b.clone();
ba.merge(&a);
assert_eq!(ab.text(), ba.text(), "commutative");
let once = {
let mut x = a.clone();
x.merge(&b);
x.text()
};
assert_eq!(ab.text(), once, "idempotent");
}
#[test]
fn reparse_projects_merged_text_onto_keyed_blocks() {
use crate::stable_id::assign_stable_keys;
let old_text = "first paragraph\n\nsecond paragraph";
let old_blocks = parse_blocks(old_text);
assert_eq!(old_blocks.len(), 2);
let mut a = TextCrdt::from_str(1, old_text);
let mut b = a.fork(2);
a.insert_str(a.len(), "\n\nthird paragraph");
b.insert_str(5, " EDITED"); a.merge(&b);
let new_blocks = parse_blocks(&a.text());
assert_eq!(new_blocks.len(), 3, "merged text re-parses to 3 paragraphs");
let keys = assign_stable_keys(&old_blocks, &new_blocks);
let old_keys: Vec<String> = old_blocks
.iter()
.map(|bl| crate::stable_id::block_key(bl).as_string())
.collect();
assert!(
keys.contains(&old_keys[1]),
"unchanged paragraph keeps identity across the text-CRDT merge"
);
}
#[test]
fn gc_collects_a_stable_deleted_leaf() {
let mut t = TextCrdt::from_str(1, "abc");
t.delete(2); assert_eq!(t.text(), "ab");
assert_eq!(t.tombstone_count(), 1);
assert_eq!(t.gc_with(|_| false), 0);
assert_eq!(t.tombstone_count(), 1);
assert_eq!(t.gc_with(|_| true), 1);
assert_eq!(t.tombstone_count(), 0);
assert_eq!(t.text(), "ab");
}
#[test]
fn gc_keeps_a_referenced_tombstone_then_collects_bottom_up() {
let mut t = TextCrdt::from_str(1, "abc");
t.delete(1); assert_eq!(t.text(), "ac");
assert_eq!(
t.gc_with(|_| true),
0,
"referenced tombstone is not collected"
);
assert_eq!(t.tombstone_count(), 1);
assert_eq!(
t.text(),
"ac",
"live text intact while tombstone is retained"
);
t.delete(1); assert_eq!(t.text(), "a");
assert_eq!(
t.gc_with(|_| true),
2,
"both tombstones collected bottom-up"
);
assert_eq!(t.tombstone_count(), 0);
assert_eq!(t.text(), "a");
}
#[test]
fn concurrent_deletes_of_same_char_converge() {
let mut a = TextCrdt::from_str(1, "abc");
let mut b = a.fork(2);
a.delete(1); b.delete(1); let mut ab = a.clone();
ab.merge(&b);
let mut ba = b.clone();
ba.merge(&a);
assert_eq!(ab.text(), "ac");
assert_eq!(ba.text(), "ac");
assert_eq!(ab.tombstone_count(), ba.tombstone_count());
ab.merge(&ba);
ba.merge(&ab);
assert_eq!(ab.text(), ba.text());
}
}