dittolive-ditto 5.0.3

Ditto is a peer to peer cross-platform database that allows mobile, web, IoT and server apps to sync with or without an internet connection.
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

use std::path::PathBuf;

use dittolive_ditto::prelude::*;

mod common;
use common::*;

/// Test that peers will recover and sync if a persistence directory is copied
/// and those copies are used on two devices simultaneously.
///
/// This is not recommended or supported by Ditto. If it happens anyway, upon
/// connecting to each other the two clones should recognise that their Peer Key
/// is no longer private to them. Both sides will decide to create a new key and
/// restart sync automatically.
///
/// The result from the user's perspective is that sync "just works".
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn recovery_from_duplicated_persistence_dir() {
    const COMPONENT_1: &str = "recovery_from_duplicated_persistence_dir_1";
    const COMPONENT_2: &str = "recovery_from_duplicated_persistence_dir_2";

    let database_id = ::uuid::Uuid::new_v4().to_string();

    let root = ::tempfile::tempdir().unwrap();
    let mut path_1 = root.path().to_path_buf();
    path_1.push(COMPONENT_1);
    let mut path_2 = root.path().to_path_buf();
    path_2.push(COMPONENT_2);

    let ditto_1 = Ditto::open_sync(
        DittoConfig::new(
            database_id.clone(),
            DittoConfigConnect::SmallPeersOnly { private_key: None },
        )
        .with_persistence_directory(path_1.clone()),
    )
    .unwrap();
    ditto_1.set_license_from_env("DITTO_LICENSE").ok();
    ditto_1.close();

    // Duplicate the persistence directory
    for entry in ::walkdir::WalkDir::new(path_1.clone()) {
        let entry = entry.unwrap();
        if !entry.file_type().is_file() {
            continue;
        }
        let path = entry.path();
        let dest = PathBuf::from(
            path.as_os_str()
                .to_string_lossy()
                .replace(COMPONENT_1, COMPONENT_2),
        );
        #[allow(clippy::disallowed_methods)]
        let _ = std::fs::create_dir_all(dest.parent().unwrap());
        std::fs::copy(path, dest).unwrap();
    }

    let ditto_1 = Ditto::open_sync(
        DittoConfig::new(
            database_id.clone(),
            DittoConfigConnect::SmallPeersOnly { private_key: None },
        )
        .with_persistence_directory(path_1),
    )
    .unwrap();
    {
        let mut config = TransportConfig::new();
        config.listen.tcp.enabled = true;
        config.listen.tcp.interface_ip = "127.0.0.1".to_owned();
        config.listen.tcp.port = 30001;
        ditto_1.set_transport_config(config);
    }
    let ditto_2 = Ditto::open_sync(
        DittoConfig::new(
            database_id.clone(),
            DittoConfigConnect::SmallPeersOnly { private_key: None },
        )
        .with_persistence_directory(path_2),
    )
    .unwrap();
    {
        let mut config = TransportConfig::new();
        config
            .connect
            .tcp_servers
            .insert("127.0.0.1:30001".to_owned());
        ditto_2.set_transport_config(config);
    }
    ditto_1.set_license_from_env("DITTO_LICENSE").ok();
    ditto_2.set_license_from_env("DITTO_LICENSE").ok();

    ditto_1.set_device_name("ditto_1");
    ditto_2.set_device_name("ditto_2");

    let (seen_2_tx, mut seen_2_rx) = ::tokio::sync::mpsc::unbounded_channel();
    let presence_observer_1 = ditto_1.presence().register_observer(move |graph| {
        if let Some(_seen_ditto_2) = graph
            .remote_peers
            .iter()
            .find(|peer| peer.device_name == "ditto_2")
        {
            seen_2_tx.send(()).ok();
        }
    });

    let (seen_1_tx, mut seen_1_rx) = ::tokio::sync::mpsc::unbounded_channel();
    let presence_observer_2 = ditto_2.presence().register_observer(move |graph| {
        if let Some(_seen_ditto_1) = graph
            .remote_peers
            .iter()
            .find(|peer| peer.device_name == "ditto_1")
        {
            seen_1_tx.send(()).ok();
        }
    });

    ditto_1.sync().start().unwrap();
    ditto_2.sync().start().unwrap();

    ::tokio::time::timeout(::std::time::Duration::from_secs(30), async {
        seen_1_rx.recv().await.unwrap();
        seen_2_rx.recv().await.unwrap();
    })
    .await
    .unwrap();

    ditto_1.sync().stop();
    ditto_2.sync().stop();
    drop(presence_observer_1);
    drop(presence_observer_2);
}