//! T M10.2: CRDT-backed buffer state. //! //! This module implements the wrapper layer that mediates between the //! editor's existing rope contract and the underlying CRDT library //! (loro 1.12, selected in M10.1). //! //! # Architecture (per M10.1 Decision section, §sec:m10-crdt-choice) //! //! The CRDT lives on the main thread. Workers consume a *rope //! projection* materialized from the CRDT state via the existing //! [`crate::rope::Rope`] type and the existing //! [`crate::buffer::Buffer::snapshot_rope`] API. Workers never see //! the CRDT directly; the rope-projection redirect preserves the //! v0.1 worker contract. //! //! Two cost-model paths: //! //! * **Hot path**: per-edit incremental projection updates. Each //! CRDT op produces an [`crate::rope::Edit`] description that //! updates the rope projection at O(log n) via the existing rope //! edit path. This is the cost shape the editor already pays. //! * **Cold path**: attach-time / catastrophic-divergence full //! re-materialization. O(n); paid once per attach, dispatched to a //! foreground worker for documents above ~1MB. //! //! # Feature gating //! //! The whole module is gated behind the `crdt` Cargo feature so v0.1 //! builds carry zero CRDT overhead — the `loro` dependency isn't //! pulled in, no field on the [`crate::buffer::Buffer`] struct //! layout, no branch on `apply_edit`. v1.0 builds enable `crdt`. //! //! # Day 1 scope //! //! M10.2 Day 1 builds the minimal `CrdtState` wrapper: insert/delete //! operations against a [`loro::LoroDoc`], a `from_bytes` constructor //! that seeds a `CrdtState` from existing rope contents, and a //! [`CrdtState::materialize_string`] projection extractor. Subsequent days wire this //! into [`crate::buffer::Buffer`] and add the per-edit incremental //! propagation, the optional `crdt_op` field on [`crate::rope::Edit`], //! and the convergence proptest. use std::sync::{ Arc, Mutex, atomic::{AtomicBool, Ordering}, }; use loro::{ ContainerTrait, ExportMode, LoroDoc, LoroEncodeError, LoroResult, TextDelta, UndoManager, VersionVector, }; type TextDeltaBatches = Arc>>>; type TextDeltaSubscription = (TextDeltaBatches, Arc, loro::Subscription); /// The CRDT-backed buffer state. /// /// Owns a [`LoroDoc`] with a single text container (named `"body"`). /// Edit operations route through the loro APIs; the editor consumes /// projections via [`Self::materialize_string`] for the rope materialization. /// /// `Send`-but-not-`Sync`: a single-threaded owner (the main thread) /// holds it; workers consume the rope projection, not the CRDT. The /// `Send` bound is needed for the M10.2 Day 7 cold-path attach /// (foreground worker materializes the initial projection). pub struct CrdtState { doc: LoroDoc, /// Text projection deltas captured only while a remote import is /// active. Keeping the subscription alive avoids registering and /// dropping one callback for every typed character. text_delta_batches: TextDeltaBatches, text_delta_capture_enabled: Arc, _text_delta_subscription: loro::Subscription, /// T M10.4: per-peer undo machinery. Bound to `doc`'s `peer_id` /// at construction; produces inverse ops attributed to that peer. /// /// Loro's `UndoManager`: /// - Local-only: undoes the bound peer's most recent change, not /// the document's most recent change (the M10.4 collaborative /// semantics that "undo my edits, not theirs" is provided by /// loro's underlying design, not by pmacs). /// - Inverse ops interact with concurrent remote ops via the /// CRDT's normal convergence rules — the M10.4 acceptance /// criterion "B's edit lands on whatever surrounding text /// remains" is loro's intrinsic behavior. /// - Default max undo steps: 100. Pmacs raises this to `10_000` to /// match v0.1's effectively-unbounded undo stack semantics. /// /// `UndoManager` is `!Send + !Sync` internally; `CrdtState` is /// main-thread-only, matching the M10.1 rope-projection-redirect /// constraint (workers consume the rope, not the CRDT). undo: std::cell::RefCell, } impl CrdtState { /// Construct an empty CRDT state. /// /// The `peer_id` is the producing-frontend identity for ops this /// state generates. M10.4 (per-frontend undo) consumes this as /// the "is this op mine?" filter; M10.5 (wire protocol) embeds it /// into broadcast `CrdtOp` messages. pub fn new(peer_id: u64) -> LoroResult { let doc = LoroDoc::new(); doc.set_peer_id(peer_id)?; // Ensure the "body" text container exists by creating the // handle. Loro creates containers lazily on first access; the // explicit get here ensures the container is registered before // any read or write. let _ = doc.get_text("body"); let (text_delta_batches, text_delta_capture_enabled, text_delta_subscription) = Self::subscribe_text_deltas(&doc); let undo = Self::create_undo_manager(&doc); Ok(Self { doc, text_delta_batches, text_delta_capture_enabled, _text_delta_subscription: text_delta_subscription, undo: std::cell::RefCell::new(undo), }) } fn subscribe_text_deltas(doc: &LoroDoc) -> TextDeltaSubscription { let text = doc.get_text("body"); let batches = Arc::new(Mutex::new(Vec::>::new())); let capture_enabled = Arc::new(AtomicBool::new(false)); let captured_batches = Arc::clone(&batches); let captured_enabled = Arc::clone(&capture_enabled); let subscription = doc.subscribe( &text.id(), Arc::new(move |event| { if !captured_enabled.load(Ordering::Relaxed) { return; } let mut guard = captured_batches .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); for event in event.events { if let Some(delta) = event.diff.as_text() && !delta.is_empty() { guard.push(delta.clone()); } } }), ); (batches, capture_enabled, subscription) } /// T M10.4: construct a fresh `UndoManager` bound to the given doc. /// Extracted as a helper because `from_bytes` constructs it AFTER /// the initial seed insert (so the seed isn't observable as an /// undoable op), while `new` constructs it before any ops happen /// (same effect). fn create_undo_manager(doc: &LoroDoc) -> UndoManager { let mut undo = UndoManager::new(doc); // Raise max undo steps from loro's default 100 to 10_000. // v0.1's undo stack is effectively unbounded; 10k is the // realistic ceiling for human-driven editing and gives users // ample undo depth without unreasonable memory cost. undo.set_max_undo_steps(10_000); undo } /// Construct a CRDT state seeded with the given bytes. /// /// Used by the cold-path attach: an existing rope's contents are /// loaded into a fresh CRDT state. The result has one initial op /// (the bulk insert at byte 0) attributed to `peer_id`. pub fn from_bytes(peer_id: u64, bytes: &[u8]) -> LoroResult { let doc = LoroDoc::new(); doc.set_peer_id(peer_id)?; let _ = doc.get_text("body"); if !bytes.is_empty() { // Loro's text API takes &str; we accept `&[u8]` and decode // as UTF-8 with replacement to match the existing rope's // permissive byte-level semantics. Files that aren't valid // UTF-8 round-trip via the projection but lose the // ill-formed-byte detail; v0.1 has the same limitation // for any operation that reads bytes as text. let text = String::from_utf8_lossy(bytes); doc.get_text("body").insert(0, &text)?; } // T M10.4: construct UndoManager AFTER the initial seed // insert so the seed is not observable as an undoable op. // The buffer's starting contents (from file load, scratch // initial text, etc.) shouldn't be undoable from the user's // perspective; only post-construction edits are. let (text_delta_batches, text_delta_capture_enabled, text_delta_subscription) = Self::subscribe_text_deltas(&doc); let undo = Self::create_undo_manager(&doc); Ok(Self { doc, text_delta_batches, text_delta_capture_enabled, _text_delta_subscription: text_delta_subscription, undo: std::cell::RefCell::new(undo), }) } /// Return the producing-frontend identity for this state. pub fn peer_id(&self) -> u64 { self.doc.peer_id() } /// Insert `s` at byte position `pos` in the body text. /// /// Routes to loro's byte-native `insert_utf8` per the M10.2 Day 2 /// morning audit (Q1 verification). Mid-codepoint positions are /// rejected with a clear error from loro; callers should ensure /// `pos` aligns to a UTF-8 codepoint boundary. pub fn insert(&self, pos: usize, s: &str) -> LoroResult<()> { self.doc.get_text("body").insert_utf8(pos, s) } /// Delete `len` bytes starting at byte position `pos`. Same /// codepoint-boundary requirement as [`Self::insert`]. pub fn delete(&self, pos: usize, len: usize) -> LoroResult<()> { self.doc.get_text("body").delete_utf8(pos, len) } /// Length of the body text in bytes (UTF-8 length). pub fn len_utf8(&self) -> usize { self.doc.get_text("body").len_utf8() } /// Length of the body text in unicode code points. Used by tests /// that assert on codepoint-vs-byte distinctions; production /// callers should prefer [`Self::len_utf8`] which matches the /// rope's byte-length contract. pub fn len_unicode(&self) -> usize { self.doc.get_text("body").len_unicode() } /// Materialize the rope projection as a `String`. /// /// O(n). Day 1 ships the trivial projection (full string /// extraction); Day 2's Buffer integration replaces this with the /// per-edit-incremental path for hot-path edits and reserves /// this method for the cold-path attach. /// /// Named `materialize_string` rather than `to_string` to avoid /// shadowing the [`std::fmt::Display`]-derived `to_string` /// (clippy `inherent_to_string`); semantics are explicit at the /// call site (it's a projection materialization, not a display /// conversion). pub fn materialize_string(&self) -> String { self.doc.get_text("body").to_string() } /// T M10.2 Day 3: capture the current oplog frontier. /// /// Used as the `from` argument to a subsequent /// [`Self::export_updates_since`] call to capture the wire bytes /// for ops produced between the two version captures. Cheap; the /// version vector is a small data structure that loro maintains /// alongside its op log. pub fn version(&self) -> VersionVector { self.doc.oplog_vv() } /// T M11.2 — the oplog version projected to a single monotonic /// scalar: the sum of every peer's op counter in the version /// vector. /// /// This is the `generation` anchor for the semantic projection /// (`InstanceMessage::StyleSpans::generation`). A loro counter is /// per-peer non-decreasing and only ever grows as ops accrue, so /// the sum is non-decreasing for the document as a whole — a /// frontend can compare a received `generation` against the one /// it computed locally and discard styling that predates an edit /// it already applied optimistically. It is deliberately *not* a /// causal clock: equal scalars do not imply equal states across /// divergent replicas. It is only ever compared against itself on /// one replica (the frontend's own mirror vs. the instance's /// authoritative doc), where it is monotone, which is all the /// staleness check needs. #[must_use] pub fn version_scalar(&self) -> u64 { self.doc .oplog_vv() .values() .map(|counter| u64::try_from(*counter).unwrap_or(0)) .sum() } /// T M10.2 Day 3: export wire-format bytes for ops added since /// `from`. /// /// Used by [`crate::buffer::Buffer`] to capture the per-edit op /// delta that populates [`crate::rope::Edit::crdt_op`]. The bytes /// are loro's incremental-update format; M10.5 (wire protocol) /// sends them across the wire to remote frontends, which import /// via [`Self::import_snapshot`] (or its updates-shaped variant). /// /// The capture-apply-export idiom is: /// 1. `let pre = state.version();` /// 2. apply ops via [`Self::insert`] / [`Self::delete`] / etc. /// 3. `let bytes = state.export_updates_since(&pre)?;` /// /// Loro's transactional model gives a consistent before/after /// pair: the version captured before the op does not include the /// op's effect, and the export from that version captures /// exactly the ops that were applied after the capture (the /// per-edit delta the M10.2 Day 3 framing requires). /// /// Returns empty bytes if no ops have been applied since `from` /// (no-op case detection is the caller's responsibility — empty /// bytes still has loro's structural overhead, so byte-length /// alone isn't a reliable empty-check; the caller pre-checks /// the `EditOp` variants instead). pub fn export_updates_since(&self, from: &VersionVector) -> Result, LoroEncodeError> { self.doc.export(ExportMode::updates(from)) } /// Export a wire-format snapshot of the entire CRDT state. /// /// Used by M10.5 wire-protocol serialization for full-state-sync /// to reconnecting frontends. Loro's snapshot is run-encoded /// (~0.8% of source at 1MB+ per the M10.1 measurements), so the /// wire bandwidth cost is small. pub fn export_snapshot(&self) -> Result, LoroEncodeError> { self.doc.export(ExportMode::Snapshot) } /// Import a wire-format snapshot into this state. /// /// Inverse of [`Self::export_snapshot`]. Used by M10.5 attach /// flow: a frontend receives a snapshot from the instance and /// constructs its local state from it. pub fn import_snapshot(&self, bytes: &[u8]) -> LoroResult<()> { self.doc.import(bytes).map(|_| ()) } /// T M10.2 Day 4: import incremental update bytes from a remote /// peer. /// /// Inverse of [`Self::export_updates_since`]. Used by M10.5's /// wire-protocol layer when a frontend receives a `CrdtOp` /// message broadcast by the instance — the bytes are the delta /// the originating frontend produced; this method merges them /// into the local CRDT. /// /// Loro's underlying `doc.import` accepts both full snapshots /// AND incremental updates (the format is universal); the /// separate method exists to make call sites self-documenting: /// /// * [`Self::import_snapshot`] — full-state replacement (attach /// path); call site signals "I'm receiving the whole state." /// * [`Self::import_updates`] — partial delta merge (per-edit /// path); call site signals "I'm receiving incremental ops." /// /// Loro's CRDT semantics handle the merge: concurrent ops from /// different peers converge regardless of import order, which is /// the property M10.2 Day 4's convergence proptest verifies. pub fn import_updates(&self, bytes: &[u8]) -> LoroResult<()> { self.doc.import(bytes).map(|_| ()) } /// Import remote updates and capture Loro's text projection deltas. /// /// The import callback runs synchronously before `doc.import` /// returns. Buffer's hot path uses the captured single-insert shape /// to update its rope projection without materializing the whole /// document. pub fn import_updates_with_text_deltas(&self, bytes: &[u8]) -> LoroResult>> { self.text_delta_batches .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .clear(); self.text_delta_capture_enabled .store(true, Ordering::Relaxed); let import_result = self.doc.import(bytes).map(|_| ()); self.text_delta_capture_enabled .store(false, Ordering::Relaxed); import_result?; let mut guard = self .text_delta_batches .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); Ok(std::mem::take(&mut *guard)) } /// Convert a Unicode scalar offset in the current text projection /// to its UTF-8 byte offset. #[must_use] pub fn unicode_to_utf8_pos(&self, pos: usize) -> Option { self.doc.get_text("body").convert_pos( pos, loro::cursor::PosType::Unicode, loro::cursor::PosType::Bytes, ) } /// T M10.10 post-audit-round-4 F26 — validate that importing /// the wire bytes `bytes` would attribute every new op to /// `expected_peer_id`. /// /// Imports into a forked clone so the live state isn't mutated. /// Compares the oplog version vector before and after import: /// every peer whose counter advances must be `expected_peer_id`, /// or the bytes carry ops produced under a different identity /// than the wire wrapper's `op.peer_id`. /// /// Returns: /// - `Ok(())` — every advancing peer matches `expected_peer_id`, /// OR the loro decoder rejected the bytes (the real import /// will surface the same decode error to the caller; we don't /// reject on validation grounds here). /// - `Err(actual)` — the first peer whose counter advanced and /// doesn't match `expected_peer_id`. The caller (the daemon's /// `validate_remote_crdt_op`) treats this as a protocol /// violation and drops the op. /// /// Cost: doubles the import cost. Loro's `fork()` is shallow; /// the real import reapplies. Acceptable for v1.0 throughput; /// v0.2+ may add a peek API to loro that avoids the second /// import. pub fn validate_update_peer_ids(&self, expected_peer_id: u64, bytes: &[u8]) -> Result<(), u64> { let fork = self.doc.fork(); let before = fork.oplog_vv(); if fork.import(bytes).is_err() { // Decode error: let the real import surface it. return Ok(()); } let after = fork.oplog_vv(); for (peer, after_counter) in after.iter() { let before_counter = before.get(peer).copied().unwrap_or(0); if *after_counter > before_counter && *peer != expected_peer_id { return Err(*peer); } } Ok(()) } /// T M10.4: undo the bound peer's most recent change. /// /// Returns `true` if an undo was performed, `false` if there was /// nothing to undo (the undo stack was empty). Inverse ops are /// applied to the doc; the projection (`materialize_string`) /// reflects the post-undo state immediately. Callers needing the /// wire-format bytes for the inverse should use the /// `version()` → `undo()` → `export_updates_since()` pattern, /// mirroring the `apply_edit` path. /// /// Loro's `UndoManager`: /// - Affects only the bound peer's ops; remote ops are unchanged /// - Inverse interacts with concurrent remote ops via CRDT /// convergence (M10.4 acceptance: "B's edit lands on /// whatever surrounding text remains") /// /// `&self` (not `&mut self`) via interior mutability: the /// underlying `UndoManager` needs `&mut` but pmacs's call sites /// hold `CrdtState` by reference. `RefCell` gates this safely; /// the main-thread-only constraint means there's no contention. pub fn undo(&self) -> LoroResult { self.undo.borrow_mut().undo() } /// T M10.4: redo the most-recently-undone change by the bound peer. /// /// Symmetric to [`Self::undo`]. Returns `true` if a redo was /// performed, `false` if the redo stack was empty. pub fn redo(&self) -> LoroResult { self.undo.borrow_mut().redo() } /// T M10.4: whether the bound peer has anything to undo. pub fn can_undo(&self) -> bool { self.undo.borrow().can_undo() } /// T M10.4: whether the bound peer has anything to redo. pub fn can_redo(&self) -> bool { self.undo.borrow().can_redo() } /// T M10.4: record an undo checkpoint. /// /// Pmacs's `apply_edit` semantics is "each successful forward edit /// is its own undo unit." Loro's `UndoManager` groups ops into /// undo units by merge interval; default 0 means no merging, /// which matches pmacs's per-edit semantics naturally. The /// explicit checkpoint method is exposed for v0.2+ batch-op /// coalescing (Day 7 mitigation B) where multiple CRDT ops /// should group into one undo unit. pub fn record_checkpoint(&self) -> LoroResult<()> { self.undo.borrow_mut().record_new_checkpoint() } } /// T M10.3: map a [`crate::protocol::FrontendId`] to the loro `PeerID` /// (u64) used as the producing-frontend identity in CRDT ops. /// /// The mapping is **identity**: `FrontendId(n)` maps to `n`. Both /// types are `u64` by design (`FrontendId(pub u64)`, loro `PeerID` /// is `u64`-wrapped). The single invariant the mapping must /// preserve is non-zero: /// /// * `FrontendId::LOCAL = FrontendId(1)` — the v0.1 default; non-zero /// * Multi-frontend allocations from M5's counter start above `LOCAL` /// * Loro's `PeerID` is internally `NonZeroU64` (default feature) — zero /// would panic at `LoroDoc::set_peer_id` /// /// The identity mapping preserves the non-zero invariant by /// construction: every `FrontendId` produced by pmacs's frontend /// machinery is non-zero, and `FrontendId(0)` is reserved as the /// "no frontend" sentinel (not allocated to any real attach). The /// mapping is a thin wrapper rather than implicit conversion so /// call sites are explicit about which value-space they're in: /// the wire-protocol layer uses `FrontendId`; the CRDT layer uses /// `PeerID`; the boundary is this function. /// /// M10.4's per-frontend undo reads this back from `CrdtOp.peer_id` /// to decide whose ops to undo; M10.5's wire protocol embeds it /// in `InstanceMessage::CrdtOp` broadcasts. #[must_use] pub fn peer_id_from_frontend(frontend_id: crate::protocol::FrontendId) -> u64 { // The non-zero invariant is a precondition; callers must not // pass FrontendId(0). Debug-assertion guards the assumption; // in release builds we trust callers and the identity mapping // returns 0 which loro will reject with a clear error at the // CrdtState::new boundary. debug_assert!( frontend_id.0 != 0, "FrontendId(0) is the no-frontend sentinel; loro PeerID requires non-zero \ (NonZeroU64 internally). Caller passed a sentinel-valued FrontendId, \ which is not a real frontend identity." ); frontend_id.0 } #[cfg(test)] mod m10_3_tests { use super::*; use crate::protocol::FrontendId; #[test] fn peer_id_from_frontend_is_identity_for_local() { assert_eq!(peer_id_from_frontend(FrontendId::LOCAL), 1); } #[test] fn peer_id_from_frontend_is_identity_for_arbitrary_ids() { for raw in [2u64, 3, 100, 1_000_000, u64::MAX] { assert_eq!(peer_id_from_frontend(FrontendId(raw)), raw); } } #[test] fn peer_id_from_frontend_round_trips_through_crdt_state() { // The identity mapping is round-trip-safe: a CrdtState // constructed with peer_id_from_frontend(fid) returns the // same fid back when its peer_id is read. let fid = FrontendId(42); let state = CrdtState::new(peer_id_from_frontend(fid)).expect("new"); assert_eq!(state.peer_id(), 42); assert_eq!(state.peer_id(), fid.0); } } // --------------------------------------------------------------------------- // Smoke tests // --------------------------------------------------------------------------- #[cfg(test)] mod tests { use super::*; #[test] fn new_state_is_empty() { let s = CrdtState::new(1).expect("new"); assert_eq!(s.len_unicode(), 0); assert_eq!(s.materialize_string(), ""); assert_eq!(s.peer_id(), 1); } #[test] fn version_scalar_is_monotonic_non_decreasing() { // T M11.2 — the semantic projection's `generation` anchor. // Empty doc is 0; each applied op only grows the scalar; a // no-op delete does not shrink it. let s = CrdtState::new(1).expect("new"); assert_eq!(s.version_scalar(), 0, "empty doc has generation 0"); s.insert(0, "hello").expect("insert"); let g1 = s.version_scalar(); assert!(g1 > 0, "an applied op must advance the generation"); s.insert(5, " world").expect("insert"); let g2 = s.version_scalar(); assert!(g2 >= g1, "generation must not decrease across ops"); s.delete(0, 1).expect("delete"); let g3 = s.version_scalar(); assert!( g3 >= g2, "a delete is still an op — the version vector only grows" ); } #[test] fn insert_round_trip_ascii() { let s = CrdtState::new(1).expect("new"); s.insert(0, "hello").expect("insert"); assert_eq!(s.materialize_string(), "hello"); s.insert(5, " world").expect("insert at end"); assert_eq!(s.materialize_string(), "hello world"); s.insert(5, ",").expect("insert in middle"); assert_eq!(s.materialize_string(), "hello, world"); } #[test] fn delete_removes_range() { let s = CrdtState::new(1).expect("new"); s.insert(0, "hello, world").expect("seed"); s.delete(5, 2).expect("delete ', '"); assert_eq!(s.materialize_string(), "helloworld"); } #[test] fn from_bytes_seeds_initial_content() { let s = CrdtState::from_bytes(42, b"the quick brown fox").expect("from_bytes"); assert_eq!(s.materialize_string(), "the quick brown fox"); assert_eq!(s.peer_id(), 42); // Subsequent edits compose with the seed. s.insert(19, " jumps").expect("append"); assert_eq!(s.materialize_string(), "the quick brown fox jumps"); } #[test] fn from_bytes_empty_yields_empty_state() { let s = CrdtState::from_bytes(7, b"").expect("from_bytes empty"); assert_eq!(s.len_unicode(), 0); assert_eq!(s.materialize_string(), ""); } #[test] fn snapshot_round_trips_through_export_import() { let source = CrdtState::new(1).expect("source"); source.insert(0, "snapshot me").expect("seed"); let bytes = source.export_snapshot().expect("export"); // Independent state on a different peer imports the bytes // and observes the same content. let restored = CrdtState::new(2).expect("restored"); restored.import_snapshot(&bytes).expect("import"); assert_eq!(restored.materialize_string(), "snapshot me"); // Restored state can continue editing under its own peer ID. restored.insert(11, "!").expect("post-restore edit"); assert_eq!(restored.materialize_string(), "snapshot me!"); } // ----------------------------------------------------------------- // M10.2 Day 2 morning audit — Q1 sub-checks for byte-native loro. // // Findings recorded here so a future loro upgrade or wrapper change // doesn't silently shift the byte/unicode boundary semantics. Loro // 1.12 provides parallel utf8-suffixed methods (`insert_utf8`, // `delete_utf8`, `mark_utf8`) for byte-native operation rather // than a global `OffsetKind` config. The default `insert` / `delete` // are unicode-native; we use the `_utf8` variants exclusively in // the wrapper to match pmacs's byte-position rope API. // // Test string: "héllo" — h=0x68 (1B), é=0xC3 0xA9 (2B), l=0x6C // (1B), l=0x6C (1B), o=0x6F (1B). Total 6 bytes, 5 codepoints. #[test] fn q1_sub_check_1_byte_native_insertion() { // We don't (yet) expose `insert_utf8` on `CrdtState` — Day 2's // afternoon adds it. This test pins the underlying loro // behaviour we'll wrap. let state = CrdtState::new(1).expect("new"); // Using the underlying loro doc directly to verify the byte- // native method exists and behaves as expected. The wrapper // method (`insert_bytes` / similar) will route to this. let text = state.doc.get_text("body"); text.insert_utf8(0, "héllo").expect("seed"); // Insert "X" at byte offset 3 (after "hé"). Byte-native: // result is "héXllo". Unicode-native: would be "hélXlo". text.insert_utf8(3, "X").expect("byte-native insert"); let s = text.to_string(); assert_eq!( s, "héXllo", "loro insert_utf8 must be byte-native (got {s:?})" ); } #[test] fn q1_sub_check_2_byte_native_deletion() { let state = CrdtState::new(1).expect("new"); let text = state.doc.get_text("body"); text.insert_utf8(0, "héllo world").expect("seed"); // Delete bytes [1, 3) — the é (2-byte codepoint). Byte-native: // "hllo world". Unicode-native (codepoints [1, 3) = "él"): // "hlo world". The two interpretations differ unambiguously. text.delete_utf8(1, 2).expect("byte-native delete"); let s = text.to_string(); assert_eq!(s, "hllo world", "loro delete_utf8 must be byte-native"); } #[test] fn q1_sub_check_3_mid_codepoint_rejection_on_delete() { // Trying to delete a range that splits a multi-byte codepoint // must fail rather than silently corrupt. The é at bytes 1..3: // delete_utf8(2, 1) starts mid-codepoint and should error. let state = CrdtState::new(1).expect("new"); let text = state.doc.get_text("body"); text.insert_utf8(0, "héllo").expect("seed"); let r = text.delete_utf8(2, 1); assert!( r.is_err(), "delete_utf8 with mid-codepoint range must reject (got {r:?})" ); // The buffer should be unchanged after the rejected delete. assert_eq!(text.to_string(), "héllo"); } #[test] fn q1_sub_check_3_mid_codepoint_on_insert() { // insert_utf8 at a mid-codepoint position. Behaviour pinned; // either reject or split-with-replacement. The wrapper layer // can normalize whichever it is to a uniform error shape, but // we need to know which one loro picks. let state = CrdtState::new(1).expect("new"); let text = state.doc.get_text("body"); text.insert_utf8(0, "héllo").expect("seed"); let r = text.insert_utf8(2, "X"); match r { Ok(()) => { let s = text.to_string(); // Document the observed behaviour rather than asserting // a specific one — the wrapper layer adapts. eprintln!("loro insert_utf8 at mid-codepoint accepted: result = {s:?}"); // Sanity: result must not be UB-shaped; should still // be valid UTF-8 and contain at least the inserted X. assert!(s.contains('X')); } Err(e) => { eprintln!("loro insert_utf8 at mid-codepoint rejected: {e}"); assert_eq!(text.to_string(), "héllo"); } } } #[test] fn q1_sub_check_cursor_position_default_kind() { // get_cursor takes a unicode position by default and // get_cursor_pos returns AbsolutePosition.pos in some kind. // We need to know which kind so the wrapper can route // correctly. use loro::cursor::Side; let state = CrdtState::new(1).expect("new"); let text = state.doc.get_text("body"); text.insert_utf8(0, "héllo").expect("seed"); // Place a cursor with input pos=2 (unicode). Multi-byte case // makes the byte/unicode answer differ: // unicode pos 2 = after "hé" (2 codepoints in) // byte pos 2 = mid-é (between é's two bytes — illegal but // informative if loro uses it) let cursor = text.get_cursor(2, Side::Middle).expect("cursor"); let pos = state.doc.get_cursor_pos(&cursor).expect("query"); eprintln!( "get_cursor(2, Middle) on \"héllo\" -> current.pos = {} (unicode 2 = after 'hé', byte 2 = mid-é)", pos.current.pos ); // Loro's default get_cursor / get_cursor_pos use unicode // positions per the docs; the wrapper layer will translate // when exposing cursor positions. Pin the expectation: assert_eq!( pos.current.pos, 2, "loro cursor positions are unicode by default (input pos round-trips)" ); } #[test] fn from_bytes_handles_non_utf8_via_replacement() { // The byte 0xFF is not valid UTF-8. The editor's rope is // permissive at the byte level; the CRDT-backed mode loses // the ill-formed-byte detail (replaced with U+FFFD) per the // documented from_bytes caveat. Pin the behaviour. let s = CrdtState::from_bytes(1, b"ok\xFFok").expect("from_bytes"); // The replacement-character round-trip preserves length-3 // structure: "ok" + U+FFFD + "ok" — verify text. assert!(s.materialize_string().contains("ok")); assert!(s.materialize_string().contains('\u{FFFD}')); } // ----------------------------------------------------------------- // T M10.2 Day 4 — multi-peer convergence smoke tests. // // Targeted tests for the canonical CRDT convergence claim: // concurrent ops from N peers, when merged in any order, produce // identical final states on every peer. The proptest version is // in `crdt::proptests` below; these smoke tests verify the // smallest interesting cases first so a proptest failure is // easier to debug. // // Per the Q2 design: `import_updates` is the per-edit-delta // path (Day 4 + M10.5); `import_snapshot` is the full-state // attach path. Both wrap loro's universal `doc.import`. // ----------------------------------------------------------------- /// Helper: peer-to-peer sync. Each peer captures its updates since /// the empty version vector (which is "all my updates") and the /// other peer imports them. After this call, both peers should be /// converged. #[cfg(test)] fn sync_pair(a: &CrdtState, b: &CrdtState) { use loro::VersionVector; let zero = VersionVector::default(); let a_updates = a.export_updates_since(&zero).expect("a export"); let b_updates = b.export_updates_since(&zero).expect("b export"); a.import_updates(&b_updates).expect("a import b"); b.import_updates(&a_updates).expect("b import a"); } #[test] fn two_peers_converge_with_concurrent_inserts_at_zero() { // Both peers start empty. // Peer A inserts "AAA" at position 0. // Peer B inserts "BBB" at position 0. // After bidirectional sync, both peers must hold the same // final string. Loro's tie-breaking yields one of "AAABBB" // or "BBBAAA" deterministically by peer ID; the test asserts // the peers agree, not a specific outcome. let a = CrdtState::new(1).expect("peer a"); let b = CrdtState::new(2).expect("peer b"); a.insert(0, "AAA").expect("a insert"); b.insert(0, "BBB").expect("b insert"); let pre_a = a.materialize_string(); let pre_b = b.materialize_string(); assert_eq!(pre_a, "AAA", "peer A sees only its own op pre-sync"); assert_eq!(pre_b, "BBB", "peer B sees only its own op pre-sync"); sync_pair(&a, &b); let post_a = a.materialize_string(); let post_b = b.materialize_string(); assert_eq!( post_a, post_b, "peers must converge after bidirectional sync" ); // Sanity: both edits are represented in the converged state. assert!(post_a.contains("AAA")); assert!(post_a.contains("BBB")); assert_eq!(post_a.len(), 6, "no duplication, no loss"); } #[test] fn two_peers_converge_independent_of_import_order() { use loro::VersionVector; // Apply identical ops to two peer pairs but in different // import orders; the converged states must match. This is // the smallest "different sync ordering, same final state" // smoke test — the canonical CRDT property. let make_pair = || { let p1 = CrdtState::new(1).unwrap(); let p2 = CrdtState::new(2).unwrap(); p1.insert(0, "hello").unwrap(); p2.insert(0, "world").unwrap(); (p1, p2) }; // Ordering 1: peer 1 imports first, then peer 2. let (a1, a2) = make_pair(); let zero = VersionVector::default(); let a2_bytes = a2.export_updates_since(&zero).unwrap(); a1.import_updates(&a2_bytes).unwrap(); let a1_bytes = a1.export_updates_since(&zero).unwrap(); a2.import_updates(&a1_bytes).unwrap(); // Ordering 2: peer 2 imports first, then peer 1. let (b1, b2) = make_pair(); let b1_bytes = b1.export_updates_since(&zero).unwrap(); b2.import_updates(&b1_bytes).unwrap(); let b2_bytes = b2.export_updates_since(&zero).unwrap(); b1.import_updates(&b2_bytes).unwrap(); // Both orderings produce the same final state across all peers. let final_state = a1.materialize_string(); assert_eq!(a1.materialize_string(), final_state); assert_eq!(a2.materialize_string(), final_state); assert_eq!(b1.materialize_string(), final_state); assert_eq!(b2.materialize_string(), final_state); } // ----------------------------------------------------------------- // T M10.2 Day 4 — convergence proptest. // // Generates N peers (2-4), each with a random sequence of // local ops, applied independently (each peer sees only its own // ops until sync). After full sync via one of five hand-picked // patterns, all peers must converge to the same projection // string. // // Five sync orderings: // * Sequential — peer-0 → peer-1 → peer-2 → ... (each peer // receives every prior peer's updates) // * Star — peer-0 receives all others' updates, then // broadcasts the combined state // * Pairwise — peer-0 syncs with peer-1, then both sync // with peer-2, etc. // * DelayedJoin — first half of peers sync each other; second // half joins later and pulls combined state // * Reverse — like Sequential but in reverse peer order // // Hand-picked patterns beat random permutations here: realistic // multi-frontend topologies have structural meaning that uniform // random permutation dilutes. Loro's convergence is proved // upstream; what's being tested is the wrapper's correctness // under realistic patterns. // ----------------------------------------------------------------- mod proptests { use super::*; use loro::VersionVector; use proptest::prelude::*; const ALPHABET: &[&str] = &["a", "b", "c", " ", "\n"]; #[derive(Clone, Debug)] enum PeerOp { Insert(usize, String), Delete(usize, usize), Replace(usize, usize, String), } #[derive(Clone, Debug)] enum SyncPattern { Sequential, Star, Pairwise, DelayedJoin, Reverse, } fn gen_payload() -> impl Strategy { prop::collection::vec(prop::sample::select(ALPHABET.to_vec()), 1..6) .prop_map(|parts| parts.concat()) } fn gen_op() -> impl Strategy { prop_oneof![ 3 => (any::(), gen_payload()).prop_map(|(p, s)| PeerOp::Insert(p as usize, s)), 2 => (any::(), any::()).prop_map(|(p, l)| PeerOp::Delete(p as usize, l as usize)), 1 => (any::(), any::(), gen_payload()) .prop_map(|(p, l, s)| PeerOp::Replace(p as usize, l as usize, s)), ] } /// Apply an op to a peer, clamping positions to the peer's /// local state at op-generation time. The op's recorded /// position refers to that state; loro's CRDT handles /// translation when the op is applied on a peer with /// different local state. We're testing that this translation /// produces convergent results, not that loro rejects /// invalid positions — so the clamp ensures the generated /// ops are *valid for the local peer*; convergence handles /// the rest. fn apply_to_peer(peer: &CrdtState, op: &PeerOp) -> Result<(), loro::LoroError> { let len = peer.len_utf8(); match op { PeerOp::Insert(pos, s) => { let pos = (*pos).min(len); peer.insert(pos, s) } PeerOp::Delete(pos, l) => { let pos = (*pos).min(len); let l = (*l).min(len.saturating_sub(pos)); if l == 0 { return Ok(()); } peer.delete(pos, l) } PeerOp::Replace(pos, l, s) => { let pos = (*pos).min(len); let l = (*l).min(len.saturating_sub(pos)); if l > 0 { peer.delete(pos, l)?; } peer.insert(pos, s) } } } /// Run a sync pattern against a list of peers. After this /// call, every peer must hold the same converged state. /// /// Each pattern is implemented as a sequence of pairwise /// imports: each invocation transfers one peer's complete /// update set (since the empty version vector) to another /// peer. Loro's import handles concurrent merges via the /// underlying CRDT semantics; this driver just orchestrates /// who-imports-from-whom in the chosen topology. /// /// The explicit `for i in 0..n / for j in 0..n` index loops /// (vs `.iter().enumerate()`) make the topology semantics /// readable: "every peer imports from every other peer" is /// a 2D index pattern, not a transformation chain. #[allow( clippy::needless_range_loop, reason = "index loops express topology more clearly than iterator chains for the i != j cross-product pattern" )] fn run_sync(peers: &[CrdtState], pattern: &SyncPattern) { let zero = VersionVector::default(); let n = peers.len(); let exports: Vec> = peers .iter() .map(|p| p.export_updates_since(&zero).expect("export")) .collect(); match pattern { SyncPattern::Sequential => { // Each peer i imports from every prior peer 0..i, // and forward-shares to peer i+1. for i in 0..n { for j in 0..n { if i != j { peers[i].import_updates(&exports[j]).expect("import"); } } } } SyncPattern::Reverse => { // Same as sequential but processed in reverse. for i in (0..n).rev() { for j in (0..n).rev() { if i != j { peers[i].import_updates(&exports[j]).expect("import"); } } } } SyncPattern::Star => { // peer 0 collects from all; then re-exports to all. for j in 1..n { peers[0].import_updates(&exports[j]).expect("import"); } let combined = peers[0].export_updates_since(&zero).expect("re-export"); for i in 1..n { peers[i].import_updates(&combined).expect("import"); } } SyncPattern::Pairwise => { // Peer 0 syncs with 1, then (0,1) with 2, etc. // Each step pulls the cumulative state to the // joining peer. for i in 1..n { let combined = peers[i - 1] .export_updates_since(&zero) .expect("export combined"); peers[i].import_updates(&combined).expect("import"); // Reverse direction so the older peers also // see the new one's ops. let back = peers[i].export_updates_since(&zero).expect("back-export"); for j in 0..i { peers[j].import_updates(&back).expect("back-import"); } } } SyncPattern::DelayedJoin => { // First half syncs each other; second half joins // later and pulls combined state. let half = n / 2; if half >= 2 { for i in 0..half { for j in 0..half { if i != j { peers[i].import_updates(&exports[j]).expect("first-half"); } } } } let combined = peers[0] .export_updates_since(&zero) .expect("first-half combined"); for i in half..n { peers[i].import_updates(&combined).expect("delayed import"); // Late joiners also share with the first half. let late = peers[i].export_updates_since(&zero).expect("late export"); for j in 0..half { peers[j].import_updates(&late).expect("late-receive"); } } // Final round: everyone imports everyone (idempotent // for already-merged ops). let final_exports: Vec> = peers .iter() .map(|p| p.export_updates_since(&zero).expect("final-export")) .collect(); for i in 0..n { for j in 0..n { if i != j { peers[i].import_updates(&final_exports[j]).expect("final"); } } } } } } proptest! { // 32 cases for CI; bump locally (e.g. 256+) when validating // before declaring Day 4 done. 32 cases passing means "no // failure surfaced in this sample," not exhaustive proof. #![proptest_config(ProptestConfig::with_cases(32))] #[test] fn peers_converge_under_arbitrary_sync_order( peer_count in 2usize..=4, op_seqs in prop::collection::vec( prop::collection::vec(gen_op(), 1..=15), 2..=4, ), sync_pattern in prop::sample::select(vec![ SyncPattern::Sequential, SyncPattern::Reverse, SyncPattern::Star, SyncPattern::Pairwise, SyncPattern::DelayedJoin, ]), ) { // The strategy may generate more op-seqs than peer_count; // truncate to peer_count so the sizes match. let op_seqs: Vec<_> = op_seqs.into_iter().take(peer_count).collect(); // The strategy guarantees at least 2 op-seqs (vec size // bound), so peer_count<=op_seqs.len() may not hold if // peer_count is 4 but the inner vec generated only 2. // Re-derive the actual peer count from op_seqs.len(). let actual_peer_count = op_seqs.len(); // Create N peers with unique peer IDs (loro requires // NonZeroU64, so we use 1..=N). let peers: Vec = (1..=actual_peer_count as u64) .map(|id| CrdtState::new(id).expect("peer")) .collect(); // Each peer applies its own op sequence in isolation // (no cross-peer visibility yet). Some ops may be // benign no-ops after clamping; that's fine — the // convergence claim holds across any valid op set. for (peer, ops) in peers.iter().zip(op_seqs.iter()) { for op in ops { let _ = apply_to_peer(peer, op); } } // Capture pre-sync states for diagnostic output on // failure. The shrinker output without these is hard // to interpret; with them, the wrapper bug (if any) // is visible. let pre_sync: Vec = peers .iter() .map(CrdtState::materialize_string) .collect(); // Capture logical version vectors (not byte encodings) // — encoding may not canonicalize entry order, so // byte-comparison would false-positive divergence; // VersionVector has PartialEq that compares logical // content. let pre_versions: Vec<_> = peers.iter().map(CrdtState::version).collect(); run_sync(&peers, &sync_pattern); let post_sync: Vec = peers .iter() .map(CrdtState::materialize_string) .collect(); let post_versions: Vec<_> = peers.iter().map(CrdtState::version).collect(); // Convergence: all peers' projections match peer 0's. for i in 1..actual_peer_count { prop_assert_eq!( &post_sync[0], &post_sync[i], "peers diverged after {:?} sync\n \ peer 0 ops: {:?}\n \ peer {} ops: {:?}\n \ peer 0 pre-sync: {:?}\n \ peer {} pre-sync: {:?}\n \ peer 0 post-sync: {:?}\n \ peer {} post-sync: {:?}\n \ peer 0 pre-version: {:?}\n \ peer {} pre-version: {:?}\n \ peer 0 post-version: {:?}\n \ peer {} post-version: {:?}", sync_pattern, op_seqs[0], i, op_seqs[i], pre_sync[0], i, pre_sync[i], post_sync[0], i, post_sync[i], pre_versions[0], i, pre_versions[i], post_versions[0], i, post_versions[i] ); } // Also assert all peers' version vectors are identical // post-sync. This is stronger than projection-string // equality: if peers agree on the string but disagree // on the version vector, future remote ops might cause // them to diverge. Loro's CRDT contract guarantees // both; pin both. for i in 1..actual_peer_count { prop_assert_eq!( &post_versions[0], &post_versions[i], "peer version vectors diverged" ); } // Avoid the unused-variable warning when peer_count is // referenced only via op_seqs.len(). let _ = peer_count; } } } // ----------------------------------------------------------------- // T M10.4 — per-frontend undo acceptance tests. // // Five tests covering the spec's three criteria plus two extras // surfaced during the framing pass (concurrent-without-sync, // region-overlap): // // 1. (covered elsewhere) Single-frontend identical to v0.1 // — dual-mode buffer tests already verify this // 2. Concurrent inserts with intervening sync: A inserts / sync // / B inserts / sync / A undoes / B's insert remains // 3. Concurrent inserts WITHOUT intervening sync: A inserts / // B inserts (both unaware) / sync both ways / A undoes / // B's insert remains // 4. B edits A's region: A inserts / sync / B edits within / // sync / A undoes A's insert / verify loro's region- // overlap behavior // 5. Redo symmetry: each pattern with redo applied, state // returns to pre-undo // ----------------------------------------------------------------- #[test] fn m10_4_concurrent_with_sync_a_undoes_b_remains() { use loro::VersionVector; // A inserts / sync / B inserts / sync / A undoes / verify // B's insert remains, A's insert is gone. let a = CrdtState::new(1).expect("A"); let b = CrdtState::new(2).expect("B"); // Round 1: A inserts "AA" at 0. a.insert(0, "AA").expect("A insert"); // Sync A → B. let zero = VersionVector::default(); b.import_updates(&a.export_updates_since(&zero).unwrap()) .unwrap(); assert_eq!(b.materialize_string(), "AA"); // Round 2: B inserts "BB" at end (position 2). b.insert(b.len_utf8(), "BB").expect("B insert"); // Sync B → A. a.import_updates(&b.export_updates_since(&zero).unwrap()) .unwrap(); assert_eq!(a.materialize_string(), "AABB"); assert_eq!(b.materialize_string(), "AABB"); // A undoes its own insert. B's "BB" must remain. let undid = a.undo().expect("A undo"); assert!(undid); let after_undo = a.materialize_string(); assert_eq!( after_undo, "BB", "A's undo must remove only A's insert; B's must remain" ); // Sync A's undo back to B; B's projection must match A's. b.import_updates(&a.export_updates_since(&zero).unwrap()) .unwrap(); assert_eq!(b.materialize_string(), "BB"); } #[test] fn m10_4_concurrent_without_sync_a_undoes_b_remains() { use loro::VersionVector; // A inserts / B inserts (both unaware of each other) / sync // bidirectionally / A undoes / verify B's insert remains. let a = CrdtState::new(1).expect("A"); let b = CrdtState::new(2).expect("B"); // Both peers insert concurrently with no intervening sync. a.insert(0, "AAA").expect("A insert"); b.insert(0, "BBB").expect("B insert"); // Now sync bidirectionally. let zero = VersionVector::default(); let a_bytes = a.export_updates_since(&zero).unwrap(); let b_bytes = b.export_updates_since(&zero).unwrap(); a.import_updates(&b_bytes).unwrap(); b.import_updates(&a_bytes).unwrap(); // Converged state contains both edits. assert_eq!(a.materialize_string(), b.materialize_string()); let converged = a.materialize_string(); assert!(converged.contains("AAA")); assert!(converged.contains("BBB")); assert_eq!(converged.len(), 6); // A undoes its insert. B's "BBB" must remain. a.undo().expect("A undo"); let after_undo = a.materialize_string(); assert_eq!( after_undo, "BBB", "A's undo removes A's insert; B's BBB remains regardless of sync order" ); // Sync undo to B; convergence holds. b.import_updates(&a.export_updates_since(&zero).unwrap()) .unwrap(); assert_eq!(b.materialize_string(), "BBB"); } #[test] fn m10_4_b_edits_a_region_then_a_undoes() { use loro::VersionVector; // A inserts "hello" / sync / B edits within (inserts "X" at // position 2, between "he" and "llo") / sync / A undoes / // verify loro's behavior for region-overlap undo. // // Loro's UndoManager produces an inverse op that removes the // bytes A originally inserted. B's "X" insert was at a // position WITHIN A's "hello" block; after A's undo, what // happens to B's "X"? // // CRDT semantics: B's "X" insert referenced A's "hello" block // structurally (insert-between-codepoints). When A's hello is // removed, B's X has no anchor — but it persists because // loro's tombstone preserves the insertion point. // // This test PINS whatever behavior loro produces; the spec // language "B's edit lands on whatever surrounding text // remains" is loro's intrinsic behavior, not ours to design. let a = CrdtState::new(1).expect("A"); let b = CrdtState::new(2).expect("B"); a.insert(0, "hello").expect("A insert"); let zero = VersionVector::default(); b.import_updates(&a.export_updates_since(&zero).unwrap()) .unwrap(); assert_eq!(b.materialize_string(), "hello"); // B inserts "X" at position 2 ("he" + "X" + "llo" -> "heXllo"). b.insert(2, "X").expect("B insert in middle"); a.import_updates(&b.export_updates_since(&zero).unwrap()) .unwrap(); assert_eq!(a.materialize_string(), "heXllo"); assert_eq!(b.materialize_string(), "heXllo"); // A undoes "hello". B's "X" remains; the surrounding "hello" // bytes attributed to A are removed. a.undo().expect("A undo"); let after_undo = a.materialize_string(); assert_eq!( after_undo, "X", "A's undo removes A's hello; B's X remains on surrounding text \ (which is now empty)" ); // Sync to B; convergence holds. b.import_updates(&a.export_updates_since(&zero).unwrap()) .unwrap(); assert_eq!(b.materialize_string(), "X"); } #[test] fn m10_4_redo_symmetric_after_undo() { use loro::VersionVector; // Verify redo correctly reverses undo across the multi-peer // patterns. Same setup as test 2 (concurrent-with-sync); A // undoes / verifies / A redoes / verifies state returns to // pre-undo. let a = CrdtState::new(1).expect("A"); let b = CrdtState::new(2).expect("B"); a.insert(0, "AA").expect("A insert"); let zero = VersionVector::default(); b.import_updates(&a.export_updates_since(&zero).unwrap()) .unwrap(); b.insert(b.len_utf8(), "BB").expect("B insert"); a.import_updates(&b.export_updates_since(&zero).unwrap()) .unwrap(); let pre_undo = a.materialize_string(); assert_eq!(pre_undo, "AABB"); a.undo().expect("undo"); assert_eq!(a.materialize_string(), "BB"); a.redo().expect("redo"); assert_eq!( a.materialize_string(), pre_undo, "redo must restore pre-undo state across multi-peer scenarios" ); } #[test] fn m10_4_undo_is_local_only_across_peers() { // Verify that A.undo() doesn't affect B's local view. // (B's view only changes when A's inverse op is synced to B // via import_updates.) Pins the "local-only" undo semantics // loro's UndoManager promises. use loro::VersionVector; let a = CrdtState::new(1).expect("A"); let b = CrdtState::new(2).expect("B"); a.insert(0, "hello").expect("A insert"); let zero = VersionVector::default(); b.import_updates(&a.export_updates_since(&zero).unwrap()) .unwrap(); assert_eq!(b.materialize_string(), "hello"); // A undoes. B's view is NOT updated until B explicitly imports. a.undo().expect("A undo"); assert_eq!(a.materialize_string(), ""); assert_eq!( b.materialize_string(), "hello", "B's view unchanged until B imports A's inverse op" ); // Now B imports; convergence. b.import_updates(&a.export_updates_since(&zero).unwrap()) .unwrap(); assert_eq!(b.materialize_string(), ""); } #[test] fn delayed_join_peer_converges_with_others() { use loro::VersionVector; // The cold-attach pattern: peers A and B exchange ops over // time, then peer C joins much later and syncs from both. // After C catches up, all three peers must converge. let a = CrdtState::new(1).expect("peer a"); let b = CrdtState::new(2).expect("peer b"); // Round 1: a and b exchange. a.insert(0, "round1-a ").unwrap(); b.insert(0, "round1-b ").unwrap(); sync_pair(&a, &b); // Round 2: more ops, more exchange. let a_len = a.len_utf8(); let b_len = b.len_utf8(); a.insert(a_len, "round2-a ").unwrap(); b.insert(b_len, "round2-b ").unwrap(); sync_pair(&a, &b); // Now peer C joins, importing from both a and b. let c = CrdtState::new(3).expect("peer c"); let zero = VersionVector::default(); c.import_updates(&a.export_updates_since(&zero).unwrap()) .unwrap(); c.import_updates(&b.export_updates_since(&zero).unwrap()) .unwrap(); // C's state matches a's and b's. All three converged. assert_eq!(c.materialize_string(), a.materialize_string()); assert_eq!(c.materialize_string(), b.materialize_string()); } }