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
|
//! Dispatch of inbound messages.
use std::time::Duration;
use anyhow::{Result, anyhow, bail};
use tokio::time::Instant;
use tracing::{debug, info, warn};
use super::admit::admit;
use super::epoch::observe_peer_epoch;
use super::handshake::{
discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply,
};
use super::keepalive::{heal_if_diverged, local_keepalive, note_keepalive};
use crate::state::AppState;
use crate::swarm::sync::{SyncEvent, handle_sync};
use crate::transport::{self, ParsedMessage};
pub(crate) async fn handle_message(
state: &AppState,
peer: String,
message: transport::ParsedMessage,
) -> Result<()> {
if peer == state.local_id() {
debug!("Ignoring message from self");
return Ok(());
}
// A reply to a HELLO from an earlier run of ours (late, or queued in Mycelium while
// we were down) says nothing about now; the peer answers our current HELLO.
if let transport::ParsedMessage::HelloReply(reply) = &message {
if reply.run_id != state.run_id {
state.stats.stale_hello_reply();
debug!(peer = %peer, "Ignoring a HELLO_REPLY to a HELLO from an earlier run");
return Ok(());
}
}
if !admit(state, &peer, &message).await {
return Ok(());
}
remember_peer(state, &peer).await;
// Any inbound message proves liveness. When we heard from it before tells a
// restarted peer how long it was away (in our HELLO_REPLY).
let (reactivated, previously_seen) = {
let peer_state_mutex = state.get_or_create_peer_state(&peer).await;
let mut peer_state = peer_state_mutex.lock().await;
let previously_seen = peer_state.last_seen;
if peer_state.unreachable_since.is_some() {
// Reachable again (`mark_seen` clears it).
state.topology_changed.notify_one();
}
(peer_state.mark_seen(Instant::now()), previously_seen)
};
crate::swarm::source::heard_from(state, &peer);
if reactivated {
info!(peer = %peer, "Dormant peer is active again; re-handshaking");
}
let run = match &message {
transport::ParsedMessage::Hello(hello) => Some(hello.run_id),
transport::ParsedMessage::HelloReply(reply) => Some(reply.sender_run_id),
_ => None,
};
let same_epoch =
observe_peer_epoch(state, &peer, message.epoch(), message.epoch_seq(), run).await;
//debug!(from = %peer, message = ?message, "Received message");
match message {
transport::ParsedMessage::Hello(hello) => {
// A HELLO means the peer is (re)starting its handshake with us, e.g. after a
// restart. It becomes a member right away; our reply makes us one for it.
let Some(swarm_id) = state.swarm_id().await else {
// We are joining ourselves; the peer retries until we know our swarm.
debug!(peer = %peer, "HELLO before we know our swarm; not answering yet");
return Ok(());
};
if let Some(theirs) = &hello.swarm_id {
if theirs != &swarm_id {
bail!("Incompatible swarm at {peer}: ours is {swarm_id}, it claims {theirs}");
}
}
establish_peer(state, &peer, swarm_id, true, hello.ready).await?;
let away = previously_seen.map(|seen| seen.elapsed());
if let Err(error) = send_hello_reply(state, &peer, hello.run_id, away).await {
// The peer retries its HELLO.
warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed");
}
let _ = state
.sync_events
.send(SyncEvent::PeerValidated(peer.clone()));
}
transport::ParsedMessage::HelloReply(reply) => {
state.lock_rtt().hello_reply_received(&peer, Instant::now());
crate::swarm::stability::note_away(state, reply.away_ms.map(Duration::from_millis));
establish_peer(state, &peer, reply.swarm_id.clone(), false, reply.ready).await?;
discover_peers(state, &peer, &reply.peers).await;
state
.sync_events
.send(SyncEvent::PeerValidated(peer.clone()))
.map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?;
}
transport::ParsedMessage::Sync(envelope) => {
if reactivated {
ensure_handshake(state, &peer).await;
}
if !same_epoch {
debug!(peer = %peer, epoch = %envelope.epoch, "Dropping sync message from another epoch");
return Ok(());
}
crate::swarm::source::synced_from(state, &peer);
handle_sync(state, &peer, envelope).await?;
}
transport::ParsedMessage::Blob(blob) => {
crate::swarm::blob::handle(state, &peer, blob).await;
}
transport::ParsedMessage::Ping(remote) => {
let local = on_keepalive(state, &peer, &remote, reactivated).await;
if let Err(error) = state
.transport
.send_message(&peer, &ParsedMessage::Pong(local))
.await
{
debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed");
}
}
transport::ParsedMessage::Pong(remote) => {
// It answers our PING: a round-trip sample.
state.lock_rtt().pong_received(&peer, Instant::now());
on_keepalive(state, &peer, &remote, reactivated).await;
}
}
Ok(())
}
/// What a PING and a PONG have in common: a handshake if needed, the peer's keepalive
/// noted, and the heal if our documents differ. Returns our keepalive for the peer.
async fn on_keepalive(
state: &AppState,
peer: &str,
remote: &transport::Keepalive,
reactivated: bool,
) -> transport::Keepalive {
if reactivated || peer_swarm(state, peer).await.is_none() {
ensure_handshake(state, peer).await;
}
note_keepalive(state, peer, remote).await;
let local = local_keepalive(state, peer).await;
heal_if_diverged(state, peer, &local, remote).await;
local
}
|
