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
|
//! Telling a peer which of the changes it brought were refused (REFUSALS), so a node
//! that took a change the swarm's rules refuse finds out and sets itself right.
//!
//! A node takes a change the others refuse only if it didn't judge it as they do: under
//! a policy it didn't know yet (it receives without judging then, `access::policy`), or
//! one that came after. Everything it writes from then on builds on that change and is
//! refused too, and nobody tells it so. Hence: a node that refuses changes a peer
//! brought tells it which broke a rule (`tell`), at most every `TELL_EVERY` and
//! `MAX_LISTED` of them. The peer doesn't take the word for it (a member could tell
//! anything): it judges each it holds itself, as its rules have them now
//! (`received`). If its rules refuse one too, it took it unjudged, and drops it with
//! everything built on it: its document as it would be without them, its own writes
//! among them quarantined (`AppState::drop_taken`). If its rules take it, the two
//! policies disagree, which it logs; finding out which is right is left
//! (`future-work.md`, Write protection by checking changes).
use std::time::Duration;
use tracing::{debug, warn};
use crate::document::replica::Epoch;
use crate::document::{ChangeHash, ReadDoc};
use crate::state::AppState;
use crate::transport::{ParsedMessage, RefusalsMessage};
/// Changes listed in one REFUSALS, at most.
pub(crate) const MAX_LISTED: usize = 64;
/// How often a peer is told, at most.
pub(crate) const TELL_EVERY: Duration = Duration::from_secs(30);
/// Tells `peer` that `changes` it brought, of our epoch, broke the swarm's rules.
pub(crate) async fn tell(
state: &AppState,
peer: &str,
epoch: Epoch,
epoch_seq: u32,
mut changes: Vec<ChangeHash>,
) {
changes.truncate(MAX_LISTED);
let message = ParsedMessage::Refusals(RefusalsMessage {
epoch,
epoch_seq,
changes,
});
if let Err(error) = state.transport.send_message(peer, &message).await {
// Told again with the next refusal.
debug!(peer = %peer, error = %format!("{error:#}"), "Couldn't tell refusals");
}
}
/// REFUSALS from `peer`, of our epoch: each change listed that we hold judged as our
/// rules have it now; if they refuse one, we took it unjudged, and drop it.
pub(crate) async fn received(state: &AppState, peer: &str, message: RefusalsMessage) {
let doc = std::sync::Arc::clone(&state.doc);
let changes: Vec<ChangeHash> = message.changes.into_iter().take(MAX_LISTED).collect();
let judged = tokio::task::spawn_blocking(move || {
let replica = doc.blocking_read();
if replica.epoch != message.epoch {
return None;
}
let policy = crate::access::policy::of(&replica.doc);
if !policy.enforced() || matches!(policy, crate::access::policy::Policy::Unknown(_)) {
// We can't judge as they do: receiving without judging until upgraded.
return None;
}
let mut refused = None;
let mut taken = Vec::new();
for hash in changes {
match judge_held(&replica.doc, replica.epoch, &hash) {
Some(Err(why)) => {
refused = Some((hash, why));
break;
}
Some(Ok(())) => taken.push(hash),
None => {}
}
}
Some((refused, taken))
})
.await
.ok()
.flatten();
let Some((refused, taken)) = judged else {
return;
};
if let Some((hash, why)) = refused {
warn!(peer = %peer, change = %hash, why = %why, "We hold a change the swarm's rules refuse, taken unjudged: dropping it and what builds on it (our writes among them go to the quarantine)");
state.drop_taken(hash).await;
} else if !taken.is_empty() {
warn!(peer = %peer, changes = ?taken, "A peer refuses changes our rules take: its policy or ours is off");
}
}
/// `hash`, which `doc` (of `epoch`) holds, judged by its rules now: why they refuse it,
/// if they do; None if `doc` doesn't hold it.
fn judge_held(
doc: &crate::document::Document,
epoch: Epoch,
hash: &ChangeHash,
) -> Option<Result<(), String>> {
let change = doc.get_change_by_hash(hash)?;
let decoded = change.decode();
let ops = crate::document::inspect::Inspector::new(doc)
.ops(std::slice::from_ref(&decoded))
.pop()
.unwrap_or_default();
let paths = ops.iter().map(|op| op.path.clone()).collect();
let committer = doc
.get_author_for_actor(change.actor_id())
.and_then(|author| {
std::str::from_utf8(author.as_bytes())
.ok()
.map(str::to_owned)
});
let user = change
.message()
.and_then(crate::document::replica::user_of)
.map(str::to_owned);
let deps = change.deps().to_vec();
let judged = crate::access::rules::judge(
doc,
epoch,
&crate::access::rules::Change {
hash: *hash,
deps: &deps,
seen: &deps,
committer: committer.as_deref(),
user: user.as_deref(),
paths: &paths,
ops: &ops,
},
);
Some(match judged {
Ok(()) => Ok(()),
Err(crate::access::rules::Refusal::Rule(why)) => Err(why),
// Its dependencies are in the document: it is.
Err(crate::access::rules::Refusal::Unready) => Ok(()),
})
}
|
