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
|
//! 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;
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 crate::access::rules::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");
}
}
|
