loro 1.17.1

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

//! A user callback that panics must not break the document: the panic reaches
//! the caller, and the doc, its locks, and its subscriptions keep working.

use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{mpsc, Arc};
use std::time::Duration;

use loro::{ExportMode, LoroDoc, LoroValue};

fn panic_message(payload: Box<dyn std::any::Any + Send>) -> String {
    payload
        .downcast_ref::<&str>()
        .map(|s| s.to_string())
        .or_else(|| payload.downcast_ref::<String>().cloned())
        .unwrap_or_default()
}

#[test]
fn pre_commit_callback_panic_finishes_the_commit_and_keeps_the_doc_usable() {
    let doc = LoroDoc::new();
    doc.set_peer_id(1).unwrap();
    let calls = Arc::new(AtomicUsize::new(0));
    let calls_in_callback = calls.clone();
    let _sub = doc.subscribe_pre_commit(Box::new(move |_| {
        if calls_in_callback.fetch_add(1, Ordering::SeqCst) == 0 {
            panic!("pre-commit bug");
        }
        true
    }));

    let result = catch_unwind(AssertUnwindSafe(|| {
        doc.get_text("t").insert(0, "a").unwrap();
        doc.commit();
    }));
    assert_eq!(panic_message(result.unwrap_err()), "pre-commit bug");

    // The change was committed, and the txn lock is not poisoned.
    assert_eq!(doc.oplog_vv().get(&1).copied(), Some(1));
    doc.get_text("t").insert(1, "b").unwrap();
    doc.commit();
    assert_eq!(calls.load(Ordering::SeqCst), 2);
    assert_eq!(doc.get_text("t").to_string(), "ab");

    let replay = LoroDoc::new();
    replay
        .import(&doc.export(ExportMode::Snapshot).unwrap())
        .unwrap();
    assert_eq!(replay.get_deep_value(), doc.get_deep_value());
}

#[test]
fn subscriber_panic_keeps_the_subscriptions_working() {
    let doc = LoroDoc::new();
    let calls = Arc::new(AtomicUsize::new(0));
    let calls_in_callback = calls.clone();
    let _panicking = doc.subscribe_root(Arc::new(move |_| {
        if calls_in_callback.fetch_add(1, Ordering::SeqCst) == 0 {
            panic!("subscriber bug");
        }
    }));
    let other_calls = Arc::new(AtomicUsize::new(0));
    let other_calls_in_callback = other_calls.clone();
    let _other = doc.subscribe_root(Arc::new(move |_| {
        other_calls_in_callback.fetch_add(1, Ordering::SeqCst);
    }));
    let updates = Arc::new(AtomicUsize::new(0));
    let updates_in_callback = updates.clone();
    let _local = doc.subscribe_local_update(Box::new(move |_| {
        if updates_in_callback.fetch_add(1, Ordering::SeqCst) == 0 {
            panic!("local update bug");
        }
        true
    }));

    let result = catch_unwind(AssertUnwindSafe(|| {
        doc.get_text("t").insert(0, "a").unwrap();
        doc.commit();
    }));
    assert!(result.is_err());
    // The other subscriber still got the event.
    assert_eq!(other_calls.load(Ordering::SeqCst), 1);

    doc.get_text("t").insert(0, "b").unwrap();
    doc.commit();
    assert_eq!(calls.load(Ordering::SeqCst), 2);
    assert_eq!(other_calls.load(Ordering::SeqCst), 2);
    assert_eq!(updates.load(Ordering::SeqCst), 2);

    // Another thread can still emit on these subscriber sets: it used to wait
    // forever for the subscribers the panic had taken.
    let (tx, rx) = mpsc::channel();
    let doc_in_thread = doc.clone();
    std::thread::spawn(move || {
        doc_in_thread.get_text("t").insert(0, "c").unwrap();
        doc_in_thread.commit();
        tx.send(()).unwrap();
    });
    rx.recv_timeout(Duration::from_secs(10))
        .expect("an emit from another thread hung");
    assert_eq!(other_calls.load(Ordering::SeqCst), 3);
    assert_eq!(
        doc.get_deep_value(),
        LoroValue::from(std::collections::HashMap::from([(
            "t".to_string(),
            LoroValue::from("cba")
        )]))
    );
}

#[test]
fn subscriber_panic_during_import_keeps_the_doc_editable() {
    let source = LoroDoc::new();
    source.get_text("t").insert(0, "remote").unwrap();
    let update = source.export(ExportMode::all_updates()).unwrap();

    let doc = LoroDoc::new();
    let calls = Arc::new(AtomicUsize::new(0));
    let calls_in_callback = calls.clone();
    let _sub = doc.subscribe_root(Arc::new(move |_| {
        if calls_in_callback.fetch_add(1, Ordering::SeqCst) == 0 {
            panic!("subscriber bug");
        }
    }));

    // Import events are emitted while the import holds the txn lock.
    let result = catch_unwind(AssertUnwindSafe(|| doc.import(&update)));
    assert_eq!(panic_message(result.unwrap_err()), "subscriber bug");
    assert_eq!(doc.get_text("t").to_string(), "remote");

    doc.get_text("t").insert(0, "local ").unwrap();
    doc.commit();
    assert_eq!(doc.get_text("t").to_string(), "local remote");
    assert_eq!(calls.load(Ordering::SeqCst), 2);
}