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
|
//! A swarm whose policy has every node judge every change (`access::policy`,
//! `access::screen`), with signatures required: honest writes go everywhere, a joining
//! node registers, a write a node lets through for a user the registry doesn't have
//! stays where it was made, and a remake carries the swarm over without refusing
//! anything.
use std::time::Duration;
use serde_json::json;
use super::{Sim, path};
use crate::document::signatures::Sign;
use crate::document::store;
use crate::state::AppState;
use crate::swarm::bootstrap::Start;
const SOON: Duration = Duration::from_secs(30);
/// Writes `value` at `at` on `node` for `user`, as the API does after its checks, or
/// without them.
async fn write_as(node: &AppState, user: &str, at: &str, value: serde_json::Value) {
let at = path(at);
let sign = Sign::Write {
path: at.clone(),
asked: false,
};
node.mutate_doc_for(Some(user.to_owned()), sign, move |doc| {
store::write_path(doc, &at, &value)
})
.await
.expect("write");
}
async fn holds(node: &AppState, at: &str) -> bool {
let at = path(at);
let replica = node.doc.read().await;
store::read_path(&replica.doc, &at).ok().flatten().is_some()
}
async fn refused(node: &AppState) -> usize {
use crate::locks::MutexExt;
node.screening.lock_or_recover().refused().count()
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_swarm_judging_every_change() {
let sim = Sim::with_identity(71);
let a = sim.start_with("a", &[], Start::Genesis, |config| {
config.first_users = [("alice".to_owned(), Default::default())].into();
});
let b = sim.start("b", &["a"], Start::Join);
let c = sim.start("c", &["a"], Start::Join);
sim.wait_ready(&[&a, &b, &c], SOON).await;
write_as(
&a,
"alice",
"_continuum/access/policy",
json!({"version": 1, "signatures": "required"}),
)
.await;
sim.wait_converged(&[&a, &b, &c], SOON).await;
// Honest writes, on any node: everywhere.
write_as(&b, "alice", "notes/x", json!(1)).await;
write_as(&c, "alice", "notes/y", json!(2)).await;
sim.wait_converged(&[&a, &b, &c], SOON).await;
for node in [&a, &b, &c] {
assert!(holds(node, "notes/x").await && holds(node, "notes/y").await);
}
// A node joining under the policy: it registers, and its writes go everywhere.
let d = sim.start("d", &["a"], Start::Join);
sim.wait_ready(&[&d], SOON).await;
write_as(&d, "alice", "notes/z", json!(3)).await;
sim.wait_converged(&[&a, &b, &c, &d], SOON).await;
for node in [&a, &b, &c, &d] {
assert_eq!(refused(node).await, 0, "an honest change was refused");
assert!(holds(node, "notes/z").await);
}
// A write c lets through for a user the registry doesn't have: refused by the
// others, and so is what c builds on it.
write_as(&c, "mallory", "notes/forged", json!(4)).await;
for node in [&a, &b, &d] {
sim.wait_until("the forged write is refused", SOON, move || async move {
refused(node).await > 0
})
.await;
assert!(!holds(node, "notes/forged").await);
}
}
/// A compaction under the policy: every node switches, the data and what was written
/// while it ran come along (replayed, signed at once after the switch), and nothing an
/// honest node wrote is refused, before or after.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_remake_under_the_policy() {
let sim = Sim::with_identity(72);
let a = sim.start_with("a", &[], Start::Genesis, |config| {
config.first_users = [("alice".to_owned(), Default::default())].into();
});
let b = sim.start("b", &["a"], Start::Join);
let c = sim.start("c", &["a"], Start::Join);
sim.wait_ready(&[&a, &b, &c], SOON).await;
let nodes = [a.clone(), b.clone(), c.clone()];
sim.wait_until("every node is registered", SOON, || {
let nodes = nodes.clone();
async move {
for node in &nodes {
if crate::swarm::members::members(&node.doc.read().await.doc).len() < 3 {
return false;
}
}
true
}
})
.await;
write_as(
&a,
"alice",
"_continuum/access/policy",
json!({"version": 1, "signatures": "required"}),
)
.await;
write_as(&b, "alice", "notes/before", json!(1)).await;
sim.wait_converged(&[&a, &b, &c], SOON).await;
let (status, report) =
crate::remake::testing::start(a.clone(), json!({ "kind": "compact" }), None).await;
assert!(status.is_success(), "proposal refused: {status} {report}");
// While it runs: replayed into the new epoch.
write_as(&c, "alice", "notes/during", json!(2)).await;
for node in [&a, &b, &c] {
assert_eq!(refused(node).await, 0, "an honest change was refused");
}
sim.wait_until(
"every node is on generation 1",
Duration::from_secs(60),
|| {
let nodes = nodes.clone();
async move {
for node in &nodes {
if node.local_epoch().generation != 1 || !node.is_ready().await {
return false;
}
}
true
}
},
)
.await;
write_as(&b, "alice", "notes/after", json!(3)).await;
for node in [&a, &b, &c] {
sim.wait_until("every write is everywhere", SOON, move || async move {
holds(node, "notes/before").await
&& holds(node, "notes/during").await
&& holds(node, "notes/after").await
})
.await;
assert_eq!(refused(node).await, 0, "an honest change was refused");
}
}
|
