truffle_core/crdt_doc/
mod.rs

1//! CrdtDoc — CRDT document synchronization across the mesh.
2//!
3//! Wraps a [Loro](https://loro.dev) `LoroDoc` with automatic peer-to-peer
4//! sync via the truffle mesh. Each document has a unique ID and syncs on
5//! namespace `"crdt:{doc_id}"`.
6//!
7//! # Quick start
8//!
9//! ```ignore
10//! use std::sync::Arc;
11//! use truffle_core::crdt_doc::CrdtDoc;
12//!
13//! let node = Arc::new(node);
14//! let doc = CrdtDoc::new(node, "my-doc").await.unwrap();
15//!
16//! // Get a Map container and insert a value.
17//! let map = doc.map("root");
18//! doc.with_doc(|d| {
19//!     map.insert_with_doc(d, "key", "value").unwrap();
20//! });
21//! doc.commit();
22//! ```
23
24pub mod backend;
25pub mod sync;
26pub mod types;
27
28#[cfg(test)]
29mod tests;
30
31pub use backend::{CrdtBackend, CrdtFileBackend, MemoryCrdtBackend};
32pub use types::{CrdtDocError, CrdtDocEvent, CrdtSyncMessage};
33
34use std::sync::Arc;
35
36use loro::LoroDoc;
37use tokio::sync::{broadcast, mpsc};
38use tokio::task::JoinHandle;
39
40use crate::network::NetworkProvider;
41use crate::node::Node;
42
43// ---------------------------------------------------------------------------
44// CrdtDoc — public API
45// ---------------------------------------------------------------------------
46
47/// A CRDT document synchronized across the mesh.
48///
49/// Wraps a Loro `LoroDoc` with automatic background sync. Local changes
50/// are broadcast to peers; remote changes are merged automatically via
51/// Loro's CRDT algorithms.
52///
53/// Created via [`CrdtDoc::new`] or [`Node::crdt_doc`].
54pub struct CrdtDoc {
55    /// Document identifier.
56    doc_id: String,
57    /// The underlying Loro document, behind a std::sync::Mutex.
58    ///
59    /// We use `std::sync::Mutex` (not `tokio::sync::Mutex`) because
60    /// `LoroDoc` is not `Send` across `.await` points — we always lock,
61    /// operate, and drop the guard synchronously before any `.await`.
62    doc: Arc<std::sync::Mutex<LoroDoc>>,
63    /// Event broadcast channel.
64    event_tx: broadcast::Sender<CrdtDocEvent>,
65    /// Handle to the background sync task.
66    task_handle: tokio::sync::Mutex<Option<JoinHandle<()>>>,
67    /// Persistence backend.
68    backend: Arc<dyn CrdtBackend>,
69    /// Channel to notify the sync task that a manual compaction occurred,
70    /// so it can reset `accumulated_update_bytes`.
71    compact_tx: mpsc::UnboundedSender<()>,
72    /// Subscription from Loro's `subscribe_local_update`. MUST be kept
73    /// alive for the duration of this CrdtDoc — dropping it unsubscribes.
74    _local_update_sub: loro::Subscription,
75}
76
77impl CrdtDoc {
78    /// Create a new CrdtDoc and start the background sync task.
79    ///
80    /// Uses the default [`MemoryCrdtBackend`] (no persistence).
81    pub async fn new<N: NetworkProvider + 'static>(
82        node: Arc<Node<N>>,
83        doc_id: &str,
84    ) -> Result<Arc<Self>, CrdtDocError> {
85        Self::new_with_backend(node, doc_id, Arc::new(MemoryCrdtBackend)).await
86    }
87
88    /// Create a new CrdtDoc with a custom persistence backend.
89    ///
90    /// On startup, attempts to restore from the backend: first loads the
91    /// snapshot (if any), then replays incremental updates on top.
92    pub async fn new_with_backend<N: NetworkProvider + 'static>(
93        node: Arc<Node<N>>,
94        doc_id: &str,
95        backend: Arc<dyn CrdtBackend>,
96    ) -> Result<Arc<Self>, CrdtDocError> {
97        let doc = LoroDoc::new();
98
99        // Restore from backend if available.
100        if let Some(snapshot) = backend.load_snapshot(doc_id) {
101            doc.import(&snapshot)
102                .map_err(|e| CrdtDocError::Decode(format!("failed to import snapshot: {e}")))?;
103        }
104        for update in backend.load_updates(doc_id) {
105            doc.import(&update)
106                .map_err(|e| CrdtDocError::Decode(format!("failed to import update: {e}")))?;
107        }
108
109        let doc = Arc::new(std::sync::Mutex::new(doc));
110        let (event_tx, _) = broadcast::channel(256);
111        let (local_update_tx, local_update_rx) = mpsc::unbounded_channel::<Vec<u8>>();
112        let (compact_tx, compact_rx) = mpsc::unbounded_channel::<()>();
113
114        // Subscribe to local updates from Loro. The callback sends bytes
115        // into the mpsc channel which the sync task consumes.
116        let sub = {
117            let d = doc.lock().unwrap();
118            d.subscribe_local_update(Box::new(move |bytes| {
119                let _ = local_update_tx.send(bytes.to_vec());
120                true // return true to keep the subscription alive
121            }))
122        };
123
124        // Spawn the background sync task.
125        let task_handle = sync::spawn_sync_task(
126            node,
127            doc.clone(),
128            doc_id.to_string(),
129            backend.clone(),
130            event_tx.clone(),
131            local_update_rx,
132            compact_rx,
133        );
134
135        Ok(Arc::new(Self {
136            doc_id: doc_id.to_string(),
137            doc,
138            event_tx,
139            task_handle: tokio::sync::Mutex::new(Some(task_handle)),
140            backend,
141            compact_tx,
142            _local_update_sub: sub,
143        }))
144    }
145
146    // ── Container accessors ────────────────────────────────────────────
147
148    /// Get a `LoroMap` container by name.
149    pub fn map(&self, name: &str) -> loro::LoroMap {
150        let doc = self.doc.lock().unwrap();
151        doc.get_map(name)
152    }
153
154    /// Get a `LoroList` container by name.
155    pub fn list(&self, name: &str) -> loro::LoroList {
156        let doc = self.doc.lock().unwrap();
157        doc.get_list(name)
158    }
159
160    /// Get a `LoroText` container by name.
161    pub fn text(&self, name: &str) -> loro::LoroText {
162        let doc = self.doc.lock().unwrap();
163        doc.get_text(name)
164    }
165
166    /// Get a `LoroTree` container by name.
167    pub fn tree(&self, name: &str) -> loro::LoroTree {
168        let doc = self.doc.lock().unwrap();
169        doc.get_tree(name)
170    }
171
172    /// Get a `LoroMovableList` container by name.
173    pub fn movable_list(&self, name: &str) -> loro::LoroMovableList {
174        let doc = self.doc.lock().unwrap();
175        doc.get_movable_list(name)
176    }
177
178    /// Get a `LoroCounter` container by name.
179    pub fn counter(&self, name: &str) -> loro::LoroCounter {
180        let doc = self.doc.lock().unwrap();
181        doc.get_counter(name)
182    }
183
184    // ── Document operations ────────────────────────────────────────────
185
186    /// Get the deep JSON value of the entire document.
187    pub fn get_deep_value(&self) -> loro::LoroValue {
188        let doc = self.doc.lock().unwrap();
189        doc.get_deep_value()
190    }
191
192    /// Commit pending changes. This triggers the `subscribe_local_update`
193    /// callback, which feeds the sync task.
194    pub fn commit(&self) {
195        let doc = self.doc.lock().unwrap();
196        doc.commit();
197    }
198
199    /// Export the document as a full snapshot.
200    pub fn export_snapshot(&self) -> Result<Vec<u8>, CrdtDocError> {
201        let doc = self.doc.lock().unwrap();
202        doc.export(loro::ExportMode::Snapshot)
203            .map_err(|e| CrdtDocError::Encode(format!("snapshot export failed: {e}")))
204    }
205
206    /// Export incremental updates since the given version vector.
207    pub fn export_updates_since(&self, vv: &loro::VersionVector) -> Result<Vec<u8>, CrdtDocError> {
208        let doc = self.doc.lock().unwrap();
209        doc.export(loro::ExportMode::updates(vv))
210            .map_err(|e| CrdtDocError::Encode(format!("updates export failed: {e}")))
211    }
212
213    /// Import binary data (snapshot or updates) into the document.
214    pub fn import(&self, data: &[u8]) -> Result<(), CrdtDocError> {
215        let doc = self.doc.lock().unwrap();
216        doc.import(data)
217            .map(|_| ())
218            .map_err(|e| CrdtDocError::Decode(format!("import failed: {e}")))
219    }
220
221    /// Get the current version vector of the document.
222    pub fn version_vector(&self) -> loro::VersionVector {
223        let doc = self.doc.lock().unwrap();
224        doc.state_vv()
225    }
226
227    /// Subscribe to document events.
228    pub fn subscribe(&self) -> broadcast::Receiver<CrdtDocEvent> {
229        self.event_tx.subscribe()
230    }
231
232    /// The document identifier.
233    pub fn doc_id(&self) -> &str {
234        &self.doc_id
235    }
236
237    /// Compact the document: export a snapshot and save it to the backend,
238    /// clearing incremental updates.
239    pub fn compact(&self) -> Result<(), CrdtDocError> {
240        let snapshot = {
241            let doc = self.doc.lock().unwrap();
242            doc.export(loro::ExportMode::Snapshot)
243                .map_err(|e| CrdtDocError::Encode(format!("compact snapshot failed: {e}")))?
244        };
245        self.backend.save_snapshot(&self.doc_id, &snapshot);
246        // Notify the sync task so it resets accumulated_update_bytes.
247        let _ = self.compact_tx.send(());
248        Ok(())
249    }
250
251    /// Checkout the document to a specific version (frontiers).
252    ///
253    /// The document becomes "detached" — local edits are not allowed until
254    /// [`checkout_to_latest`](Self::checkout_to_latest) is called.
255    pub fn checkout(&self, frontiers: &loro::Frontiers) -> Result<(), CrdtDocError> {
256        let doc = self.doc.lock().unwrap();
257        doc.checkout(frontiers)
258            .map_err(|e| CrdtDocError::Loro(format!("checkout failed: {e}")))
259    }
260
261    /// Return to the latest version after a `checkout`.
262    pub fn checkout_to_latest(&self) {
263        let doc = self.doc.lock().unwrap();
264        doc.checkout_to_latest();
265    }
266
267    /// Whether the document is in "detached" mode (checked out to a
268    /// historical version).
269    pub fn is_detached(&self) -> bool {
270        let doc = self.doc.lock().unwrap();
271        doc.is_detached()
272    }
273
274    /// Fork the document — create an independent `LoroDoc` clone.
275    pub fn fork(&self) -> Result<LoroDoc, CrdtDocError> {
276        let doc = self.doc.lock().unwrap();
277        Ok(doc.fork())
278    }
279
280    /// Access the underlying LoroDoc behind the mutex.
281    ///
282    /// The closure runs with the mutex held. Do NOT call any async
283    /// functions inside the closure.
284    pub fn with_doc<F, R>(&self, f: F) -> R
285    where
286        F: FnOnce(&LoroDoc) -> R,
287    {
288        let doc = self.doc.lock().unwrap();
289        f(&doc)
290    }
291
292    // ── Lifecycle ──────────────────────────────────────────────────────
293
294    /// Stop the background sync task.
295    pub async fn stop(&self) {
296        let mut handle = self.task_handle.lock().await;
297        if let Some(h) = handle.take() {
298            h.abort();
299            tracing::info!(doc_id = self.doc_id.as_str(), "crdt_doc: stopped");
300        }
301    }
302}
303
304impl Drop for CrdtDoc {
305    fn drop(&mut self) {
306        // Best-effort abort on drop.
307        if let Ok(mut handle) = self.task_handle.try_lock() {
308            if let Some(h) = handle.take() {
309                h.abort();
310            }
311        }
312    }
313}
truffle_core/crdt_doc/mod.rs

truffle_core/crdt_doc/
mod.rs
