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
|
//! Signatures of changes sent on their own (SIGNATURES), where no sync message took
//! them: a signature made after its change went (`--sign-every`), or one received
//! without the change it vouches for, as from a peer that had it already. Every
//! `SEND_EVERY`, each active peer on our epoch is sent the signatures it wasn't sent yet
//! (`Signatures::newer_than`), if any; it checks them as it checks those sync messages
//! bring (`Signatures::offer`), and sends them on in turn: each actor's latest goes to
//! each node about once.
use std::time::Duration;
use tracing::debug;
use crate::state::AppState;
use crate::transport::{ParsedMessage, SignaturesMessage};
/// How often the signatures the peers weren't sent yet go out.
pub(crate) const SEND_EVERY: Duration = Duration::from_secs(1);
/// Sends the signatures due every `SEND_EVERY`. Runs for as long as the node does.
pub(crate) async fn run(state: AppState) {
let mut ticks = tokio::time::interval(SEND_EVERY);
ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
ticks.tick().await;
send_due(&state).await;
}
}
/// Each active peer on our epoch: the signatures it wasn't sent yet, if any.
async fn send_due(state: &AppState) {
for peer in state.active_peers_on_local_epoch().await {
let Some(peer_state) = state.get_peer_state(&peer).await else {
continue;
};
let doc = std::sync::Arc::clone(&state.doc);
// Lock order doc -> peer, as in sync generation.
let message = tokio::task::spawn_blocking(move || {
let replica = doc.blocking_read();
let mut peer_state = peer_state.blocking_lock();
let (signatures, sent) = replica.signatures.newer_than(peer_state.signatures_sent);
if signatures.is_empty() {
return None;
}
peer_state.signatures_sent = sent;
Some(ParsedMessage::Signatures(SignaturesMessage {
epoch: replica.epoch,
epoch_seq: replica.epoch_seq,
signatures: serde_bytes::ByteBuf::from(crate::document::signatures::pack(
&signatures,
)),
}))
})
.await;
if let Ok(Some(message)) = message
&& let Err(error) = state.transport.send_message(&peer, &message).await
{
// Lost, they come again with the next sync session (`PeerState::reset_sync`).
debug!(peer = %peer, error = %format!("{error:#}"), "Couldn't send signatures");
}
}
}
/// SIGNATURES from `peer`, of our epoch: checked and kept as those of sync messages.
pub(crate) async fn received(state: &AppState, message: SignaturesMessage) {
let Some(signatures) = crate::document::signatures::unpack(&message.signatures)
.filter(|signatures| signatures.len() <= crate::document::signatures::MAX_PER_MESSAGE)
else {
return;
};
let doc = std::sync::Arc::clone(&state.doc);
let screening = std::sync::Arc::clone(&state.screening);
// How many waiting changes went in, if any: published as received ones are.
let applied = tokio::task::spawn_blocking(move || {
let mut replica = doc.blocking_write();
if replica.epoch != message.epoch {
return 0;
}
{
let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
let replica = &mut *replica;
replica.signatures.offer(&replica.doc, signatures);
// With signatures required, changes waiting for them judged again
// (`access::screen`).
if crate::access::policy::of(&replica.doc).required() {
let applied = replica.logged_signed(|doc, known, log| {
let applied = crate::access::screen::apply_bytes(
doc,
epoch,
epoch_seq,
&[],
&screening,
Some(known),
log,
);
(applied, true)
});
replica.refresh_heads();
return match applied {
Ok(outcome) => outcome.applied,
Err(error) => {
tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures");
0
}
};
}
0
}
})
.await
.unwrap_or(0);
if applied > 0 {
state.publish_change();
}
}
|
