actordb 0.6.0

An actor-first distributed database
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

mod common;
#[path = "common/transport.rs"]
mod transport;

use std::sync::Arc;
use std::time::Duration;

use actordb::network::Network;
use actordb::node::{Node, NodeConfig};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use transport::MemoryNetwork;

const TIMEOUT: Duration = Duration::from_secs(10);

#[tokio::test]
async fn buffered_transport_preserves_bytes_eof_and_listener_ownership() {
    tokio::time::timeout(TIMEOUT, async {
        let network = MemoryNetwork::default();
        let (addr, mut acceptor) = network.bind("127.0.0.1:0".parse().unwrap()).await.unwrap();
        assert!(network.bind(addr).await.is_err());
        // Cancelling a pending accept leaves the listener usable.
        assert!(
            tokio::time::timeout(Duration::from_millis(10), acceptor.accept())
                .await
                .is_err()
        );
        let mut dialed = network.connect(addr).await.unwrap();
        let mut accepted = acceptor.accept().await.unwrap();
        let payload: Vec<u8> = (0..65536).map(|i| i as u8).collect();
        let ((), received) = tokio::join!(
            async {
                dialed.write_all(&payload).await.unwrap();
                dialed.flush().await.unwrap();
                dialed.shutdown().await.unwrap();
            },
            async {
                let mut received = Vec::new();
                accepted.read_to_end(&mut received).await.unwrap();
                received
            }
        );
        assert_eq!(received, payload);
        drop(acceptor);
        assert!(network.connect(addr).await.is_err());
        let (_, replacement) = network.bind(addr).await.unwrap();
        drop(replacement);
    })
    .await
    .unwrap();
}

#[tokio::test]
async fn mesh_replicates_and_reconnects_over_buffered_non_tcp_streams() {
    tokio::time::timeout(TIMEOUT, async {
        let network = Arc::new(MemoryNetwork::default());
        let config = NodeConfig::builder().network(network.clone()).build();
        let a = Node::start(
            "buffered-a".into(),
            "127.0.0.1:0".parse().unwrap(),
            config.clone(),
        )
        .await
        .unwrap();
        let b = Node::start(
            "buffered-b".into(),
            "127.0.0.1:0".parse().unwrap(),
            config.clone(),
        )
        .await
        .unwrap();
        let addr = b.addr();
        a.connect(addr).await.unwrap();
        common::wait_for(
            || async { a.peer_count().await == 1 && b.peer_count().await == 1 },
            TIMEOUT,
        )
        .await;
        a.put("from-a".into(), vec![7; 32768]).await.unwrap();
        b.put("from-b".into(), vec![8; 32768]).await.unwrap();
        common::wait_for(
            || async {
                b.get_all()
                    .await
                    .unwrap()
                    .get("from-a")
                    .is_some_and(|v| v.value == vec![7; 32768])
                    && a.get_all()
                        .await
                        .unwrap()
                        .get("from-b")
                        .is_some_and(|v| v.value == vec![8; 32768])
            },
            TIMEOUT,
        )
        .await;
        b.shutdown().await.unwrap();
        common::wait_for(|| async { a.peer_count().await == 0 }, TIMEOUT).await;
        a.put("offline".into(), b"catch up".to_vec()).await.unwrap();
        let b = Node::start("buffered-b".into(), addr, config)
            .await
            .unwrap();
        common::wait_for(
            || async {
                b.get_all()
                    .await
                    .unwrap()
                    .get("offline")
                    .is_some_and(|v| v.value == b"catch up")
            },
            TIMEOUT,
        )
        .await;
        a.shutdown().await.unwrap();
        b.shutdown().await.unwrap();
        // Node shutdown drops its acceptor, permitting another bind.
        let (_, replacement) = network.bind(addr).await.unwrap();
        drop(replacement);
    })
    .await
    .unwrap();
}

#[tokio::test]
async fn node_shutdown_cancels_idle_custom_streams_and_pending_accepts() {
    tokio::time::timeout(TIMEOUT, async {
        let network = Arc::new(MemoryNetwork::default());
        let node = Node::start(
            "idle".into(),
            "127.0.0.1:0".parse().unwrap(),
            NodeConfig::builder().network(network.clone()).build(),
        )
        .await
        .unwrap();
        let mut idle = network.connect(node.addr()).await.unwrap();
        // The peer Hello proves this connection was accepted; never answer it.
        let length = idle.read_u32().await.unwrap();
        let mut body = vec![0; length as usize];
        idle.read_exact(&mut body).await.unwrap();
        node.shutdown().await.unwrap();
        assert_eq!(idle.read(&mut [0]).await.unwrap(), 0);
        let (_, replacement) = network.bind(node.addr()).await.unwrap();
        drop(replacement);
    })
    .await
    .unwrap();
}