dson 0.3.0

A delta-state CRDT implementation
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

//! Integration tests for nested transaction API.

use dson::{
    CausalDotStore, Identifier, OrMap, crdts::mvreg::MvRegValue, transaction::MapTransaction,
};

#[test]
fn deeply_nested_map_array_map() {
    // Structure: map -> array -> map
    // Like: { "projects": [{ "name": "DSON", "tasks": [...] }] }

    let mut store = CausalDotStore::<OrMap<String>>::default();
    let id = Identifier::new(0, 0);

    {
        let mut tx = MapTransaction::new(&mut store, id);

        tx.in_array("projects", |projects_tx| {
            projects_tx.insert_map(0, |project_tx| {
                project_tx.write_register("name", MvRegValue::String("DSON".to_string()));
                project_tx.write_register("priority", MvRegValue::U64(1));

                project_tx.in_array("tasks", |tasks_tx| {
                    tasks_tx
                        .insert_register(0, MvRegValue::String("Implement nested TX".to_string()));
                    tasks_tx.insert_register(1, MvRegValue::String("Write tests".to_string()));
                });
            });
        });

        let _delta = tx.commit();
    }

    // Verify structure was created
    use dson::crdts::snapshot::ToValue;

    let projects_val = store.store.get(&"projects".to_string()).unwrap();
    assert_eq!(projects_val.array.len(), 1);

    let project = projects_val.array.get(0).unwrap();
    let name = project.map.get(&"name".to_string()).unwrap();
    assert_eq!(
        name.reg.value().unwrap(),
        &MvRegValue::String("DSON".to_string())
    );

    let tasks_val = project.map.get(&"tasks".to_string()).unwrap();
    assert_eq!(tasks_val.array.len(), 2);
}

#[test]
fn array_of_arrays() {
    // Test [[1, 2], [3, 4]]

    let mut store = CausalDotStore::<OrMap<String>>::default();
    let id = Identifier::new(0, 0);

    {
        let mut tx = MapTransaction::new(&mut store, id);

        tx.in_array("matrix", |matrix_tx| {
            matrix_tx.insert_array(0, |row_tx| {
                row_tx.insert_register(0, MvRegValue::U64(1));
                row_tx.insert_register(1, MvRegValue::U64(2));
            });

            matrix_tx.insert_array(1, |row_tx| {
                row_tx.insert_register(0, MvRegValue::U64(3));
                row_tx.insert_register(1, MvRegValue::U64(4));
            });
        });

        let _delta = tx.commit();
    }

    // Verify 2x2 matrix
    use dson::crdts::snapshot::ToValue;

    let matrix = store.store.get(&"matrix".to_string()).unwrap();
    assert_eq!(matrix.array.len(), 2);

    let row0 = matrix.array.get(0).unwrap();
    assert_eq!(row0.array.len(), 2);
    assert_eq!(
        row0.array.get(0).unwrap().reg.value().unwrap(),
        &MvRegValue::U64(1)
    );
    assert_eq!(
        row0.array.get(1).unwrap().reg.value().unwrap(),
        &MvRegValue::U64(2)
    );

    let row1 = matrix.array.get(1).unwrap();
    assert_eq!(row1.array.len(), 2);
    assert_eq!(
        row1.array.get(0).unwrap().reg.value().unwrap(),
        &MvRegValue::U64(3)
    );
    assert_eq!(
        row1.array.get(1).unwrap().reg.value().unwrap(),
        &MvRegValue::U64(4)
    );
}

#[test]
fn concurrent_nested_modifications() {
    // Two replicas modify nested structures concurrently

    let id1 = Identifier::new(0, 0);
    let id2 = Identifier::new(1, 0);

    let mut replica1 = CausalDotStore::<OrMap<String>>::default();
    let mut replica2 = CausalDotStore::<OrMap<String>>::default();

    // Both create initial structure
    let init_delta = {
        let mut tx = MapTransaction::new(&mut replica1, id1);
        tx.in_map("config", |cfg_tx| {
            cfg_tx.write_register("version", MvRegValue::U64(1));
        });
        tx.commit()
    };

    replica1.join_or_replace_with(init_delta.0.store.clone(), &init_delta.0.context);
    replica2.join_or_replace_with(init_delta.0.store, &init_delta.0.context);

    // Replica 1: adds array to config
    let delta1 = {
        let mut tx = MapTransaction::new(&mut replica1, id1);
        tx.in_map("config", |cfg_tx| {
            cfg_tx.in_array("features", |features_tx| {
                features_tx.insert_register(0, MvRegValue::String("fast".to_string()));
            });
        });
        tx.commit()
    };

    // Replica 2: updates version concurrently
    let delta2 = {
        let mut tx = MapTransaction::new(&mut replica2, id2);
        tx.in_map("config", |cfg_tx| {
            cfg_tx.write_register("version", MvRegValue::U64(2));
        });
        tx.commit()
    };

    // Exchange deltas
    replica1.join_or_replace_with(delta2.0.store, &delta2.0.context);
    replica2.join_or_replace_with(delta1.0.store, &delta1.0.context);

    // Both should converge
    assert_eq!(replica1, replica2);

    // Verify both changes present
    use dson::crdts::snapshot::ToValue;

    let config = replica1.store.get(&"config".to_string()).unwrap();
    let version = config.map.get(&"version".to_string()).unwrap();
    assert!(version.reg.value().unwrap() == &MvRegValue::U64(2));

    let features = config.map.get(&"features".to_string()).unwrap();
    assert_eq!(features.array.len(), 1);
}