loro 1.12.0

Loro is a high-performance CRDTs framework. Make your app collaborative efforlessly.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300

#[cfg(loom)]
mod loom_test {

    use loom::{explore, stop_exploring, sync::atomic::AtomicUsize, thread::yield_now};
    use loro::{ExportMode, LoroDoc, UpdateOptions};
    use std::{sync::atomic::Ordering, thread::sleep, time::Duration};

    #[test]
    fn concurrently_inserting_text_content() {
        loom::model(|| {
            let doc = LoroDoc::new();
            let doc1 = doc.clone();
            let doc2 = doc.clone();
            let h0 = loom::thread::spawn(move || {
                doc1.get_text("text").insert(0, "1").unwrap();
            });
            let h1 = loom::thread::spawn(move || {
                for _ in 0..2 {
                    doc2.get_text("text").insert(0, "2").unwrap();
                }
            });

            if let Err(e) = h0.join() {
                eprintln!("Thread h0 failed: {:?}", e);
                panic!("Thread h0 failed: {:?}", e);
            }

            if let Err(e) = h1.join() {
                eprintln!("Thread h1 failed: {:?}", e);
                panic!("Thread h1 failed: {:?}", e);
            }
            let text = doc.get_text("text");
            assert_eq!(text.len_utf8(), 3);
            dbg!("{}", text.to_string());
        });
    }

    #[test]
    fn concurrently_creating_events_with_subscriber() {
        loom::model(|| {
            let doc = LoroDoc::new();
            let count = std::sync::Arc::new(AtomicUsize::new(0));
            let count_clone = count.clone();
            let _sub = doc.subscribe_root(std::sync::Arc::new(move |e| {
                stop_exploring();
                for e in e.events {
                    let v = e.diff.as_text().unwrap();
                    for v in v {
                        match &v {
                            loro::TextDelta::Retain { .. } => unreachable!(),
                            loro::TextDelta::Delete { .. } => unreachable!(),
                            loro::TextDelta::Insert { insert, .. } => {
                                count_clone.fetch_add(insert.len(), Ordering::SeqCst);
                            }
                        }
                    }
                }
                explore();
            }));
            let doc1 = doc.clone();
            let doc2 = doc.clone();
            let h0 = loom::thread::spawn(move || {
                doc1.get_text("text").insert(0, "1").unwrap();
                doc1.commit();
            });
            let h1 = loom::thread::spawn(move || {
                doc2.get_text("text").insert(0, "2").unwrap();
                doc2.commit();
            });

            h0.join().unwrap();
            h1.join().unwrap();
            let text = doc.get_text("text");
            assert_eq!(text.len_utf8(), 2);
            assert_eq!(count.load(Ordering::SeqCst), 2);
        });
    }

    #[test]
    fn concurrent_callbacks_modifying_same_doc() {
        loom::model(|| {
            let doc = LoroDoc::new();
            let text_id = "shared_text";

            // Set up a condition variable to control thread execution
            let pair =
                std::sync::Arc::new((loom::sync::Mutex::new(false), loom::sync::Condvar::new()));
            let pair_clone1 = pair.clone();
            let pair_clone2 = pair.clone();

            // Create a counter to track callbacks
            let counter = std::sync::Arc::new(AtomicUsize::new(0));
            let counter_clone = counter.clone();

            // First subscription modifies the document in its callback
            let doc1 = doc.clone();
            let _sub1 = doc.subscribe_root(std::sync::Arc::new(move |_| {
                let count = counter_clone.fetch_add(1, Ordering::SeqCst);

                // Only react to the first few events to avoid infinite loops
                if count < 2 {
                    // This will trigger another event
                    doc1.get_text(text_id).insert(0, "A").unwrap();
                    doc1.commit();

                    // Signal the other thread
                    let (lock, cvar) = &*pair_clone1;
                    let mut started = lock.lock().unwrap();
                    *started = true;
                    cvar.notify_one();
                }
            }));

            // Second thread also has a subscription that modifies the document
            let doc2 = doc.clone();
            let doc3 = doc.clone();

            let h = loom::thread::spawn(move || {
                // Wait for the first thread's callback to trigger
                let (lock, cvar) = &*pair_clone2;
                let mut started = lock.lock().unwrap();
                while !*started {
                    // Wait until the condition variable is signaled
                    started = cvar.wait(started).unwrap();
                }

                drop(started);
                // Now both threads are modifying the document based on events
                doc2.get_text(text_id).insert(0, "B").unwrap();
                doc2.commit();

                doc2.commit();
            });

            // Start the chain of events
            doc3.get_text(text_id).insert(0, "Start").unwrap();
            doc3.commit();

            h.join().unwrap();
        });
    }

    #[test]
    fn concurrent_document_checkout_with_modifications() {
        let mut builder = loom::model::Builder::new();
        builder.max_branches = 3000;
        builder.check(|| {
            let doc = LoroDoc::new();
            doc.set_detached_editing(true);

            // Set up multiple frontiers to test checkout
            doc.get_text("text").insert(0, "Initial state").unwrap();
            doc.commit();
            let initial_frontier = doc.state_frontiers();

            doc.get_text("text")
                .insert(doc.get_text("text").len_utf8(), " - First update")
                .unwrap();
            doc.commit();
            let second_frontier = doc.state_frontiers();

            doc.get_text("text")
                .insert(doc.get_text("text").len_utf8(), " - Second update")
                .unwrap();
            doc.commit();
            let third_frontier = doc.state_frontiers();

            // Now create threads that will concurrently checkout and modify
            let doc1 = doc.clone();
            let doc2 = doc.clone();
            let initial_clone = initial_frontier.clone();
            let second_clone = second_frontier.clone();
            let third_clone = third_frontier.clone();

            let h1 = loom::thread::spawn(move || {
                // First checkout to an earlier state
                doc1.checkout(&initial_clone).unwrap();

                // Modify the document
                doc1.get_text("text")
                    .insert(doc1.get_text("text").len_utf8(), " - Thread 1 modification")
                    .unwrap();
                doc1.commit();

                // Sleep to increase chance of race condition
                yield_now();

                // Checkout to a later state
                doc1.checkout(&third_clone).unwrap();

                // Modify again
                doc1.get_text("text")
                    .insert(0, " - Thread 1 after checkout")
                    .unwrap();
                doc1.commit();
            });

            let h2 = loom::thread::spawn(move || {
                yield_now();

                // Checkout to the middle state
                doc2.checkout(&second_clone).unwrap();

                // Modify
                doc2.get_text("text")
                    .insert(doc2.get_text("text").len_utf8(), " - Thread 2 modification")
                    .unwrap();
                doc2.commit();

                yield_now();

                // Checkout to latest
                doc2.checkout_to_latest();

                // Modify again
                doc2.get_text("text")
                    .insert(0, " - Thread 2 after checkout")
                    .unwrap();
                doc2.commit();
            });

            h1.join().unwrap();
            h2.join().unwrap();
        });
    }

    #[test]
    fn concurrently_import_export() {
        let mut builder = loom::model::Builder::new();
        builder.max_branches = 2000;
        builder.check(|| {
            let doc1 = LoroDoc::new();
            let doc1_clone = doc1.clone();
            let doc1_clone2 = doc1.clone();
            let doc2 = LoroDoc::new();
            let doc2_clone = doc2.clone();
            let doc2_clone2 = doc2.clone();

            let mut handlers = vec![];
            handlers.push(loom::thread::spawn(move || {
                doc1_clone.get_text("text").insert(0, "1").unwrap();
                doc1_clone.commit();
                doc1_clone.get_text("text").insert(0, "1").unwrap();
                doc1_clone.commit();
            }));
            handlers.push(loom::thread::spawn(move || {
                doc2_clone.get_text("text").insert(0, "2").unwrap();
                doc2_clone.commit();
            }));
            handlers.push(loom::thread::spawn(move || {
                let e = &doc1_clone2
                    .export(ExportMode::updates(&doc2_clone2.oplog_vv()))
                    .unwrap();
                doc2_clone2.import(e).unwrap();
                yield_now();
                let e = &doc2_clone2
                    .export(ExportMode::updates(&doc1_clone2.oplog_vv()))
                    .unwrap();
                doc1_clone2.import(e).unwrap();
            }));

            for h in handlers {
                h.join().unwrap();
            }
        });
    }

    #[test]
    fn local_edits_during_batch_import() {
        let mut builder = loom::model::Builder::new();
        builder.max_branches = 3000;
        builder.check(move || {
            let doc = LoroDoc::new();
            doc.get_text("text").insert(0, "hello").unwrap();
            let update_a = doc.export(ExportMode::all_updates()).unwrap();
            doc.get_text("text")
                .update("yo! hello", UpdateOptions::default())
                .unwrap();
            let update_b = doc.export(ExportMode::all_updates()).unwrap();

            let mut handlers = vec![];
            let doc = LoroDoc::new();
            let doc_clone = doc.clone();

            handlers.push(loom::thread::spawn(move || {
                doc_clone.get_text("text").insert(0, "1").unwrap();
                doc_clone.commit();
                doc_clone.get_text("text").insert(0, "1").unwrap();
                doc_clone.commit();
            }));

            handlers.push(loom::thread::spawn(move || {
                doc.import_batch(&[update_a, update_b]).unwrap();
            }));
            for h in handlers {
                h.join().unwrap();
            }
        });
    }
}