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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
use super::*;
impl Node {
/// Promote a connection to active peer after successful authentication.
///
/// Handles cross-connection detection and resolution using tie-breaker rules.
pub(in crate::node) fn promote_connection(
&mut self,
link_id: LinkId,
verified_identity: PeerIdentity,
current_time_ms: u64,
) -> Result<PromotionResult, NodeError> {
// Remove the connection from pending
let mut connection = self
.peers
.remove_connection(&link_id)
.ok_or(NodeError::ConnectionNotFound(link_id))?;
// Verify handshake is complete and extract session
if !connection.has_session() {
return Err(NodeError::HandshakeIncomplete(link_id));
}
let noise_session = connection
.take_session()
.ok_or(NodeError::NoSession(link_id))?;
let our_index = connection
.our_index()
.ok_or_else(|| NodeError::PromotionFailed {
link_id,
reason: "missing our_index".into(),
})?;
let their_index = connection
.their_index()
.ok_or_else(|| NodeError::PromotionFailed {
link_id,
reason: "missing their_index".into(),
})?;
let transport_id = connection
.transport_id()
.ok_or_else(|| NodeError::PromotionFailed {
link_id,
reason: "missing transport_id".into(),
})?;
let observed_addr = connection
.source_addr()
.ok_or_else(|| NodeError::PromotionFailed {
link_id,
reason: "missing source_addr".into(),
})?
.clone();
let link_stats = connection.link_stats().clone();
let remote_epoch = connection.remote_epoch();
let peer_node_addr = *verified_identity.node_addr();
let is_outbound = connection.is_outbound();
let current_addr = observed_addr;
let discovery_fallback_transit_allowed =
self.discovery_fallback_transit_for_promotion(&peer_node_addr);
// Check for cross-connection
if let Some(existing_peer) = self.peers.get(&peer_node_addr) {
let existing_link_id = existing_peer.link_id();
let existing_path_unusable = !existing_peer.is_healthy() || !existing_peer.can_send();
let outbound_alternate_path = is_outbound
&& (existing_peer.transport_id() != Some(transport_id)
|| existing_peer.current_addr() != Some(¤t_addr));
let remote_epoch_changed = matches!((existing_peer.remote_epoch(), remote_epoch), (Some(old), Some(new)) if old != new);
let existing_path_unusable = existing_path_unusable
|| self.session_direct_path_blocks_direct_payload(&peer_node_addr, current_time_ms);
let outbound_alternate_path_wins = outbound_alternate_path
&& self.alternate_path_priority_allows_replace(
&peer_node_addr,
transport_id,
¤t_addr,
);
// Determine which connection wins. A peer restart (different
// startup epoch) is not a normal cross-connection: the old link
// and FSP sessions are cryptographically stale, so the freshly
// authenticated connection must replace them regardless of the
// tie-breaker direction.
//
// Likewise, a link-dead path is kept as a reconnecting peer so
// higher-level sessions and routes survive. A fresh authenticated
// connection is proof of a usable replacement path, so it should
// win instead of applying the simultaneous-handshake tie-breaker to
// a path we already marked unusable.
//
// A completed outbound handshake on a different transport tuple is
// also not a symmetric cross-connection. It is an explicit
// alternate-path refresh we initiated after learning a candidate;
// successful authentication is enough proof only when the
// candidate is at least as preferred as the current healthy path.
let this_wins = remote_epoch_changed
|| existing_path_unusable
|| if outbound_alternate_path {
outbound_alternate_path_wins
} else {
cross_connection_winner(self.identity.node_addr(), &peer_node_addr, is_outbound)
};
if this_wins {
// This connection wins, replace the existing peer
let old_peer = self.peers.remove(&peer_node_addr).unwrap();
let loser_link_id = old_peer.link_id();
// Clean up old peer's index from active peer registry session-index dispatch
if let (Some(old_tid), Some(old_idx)) =
(old_peer.transport_id(), old_peer.our_index())
{
self.deregister_session_index((old_tid, old_idx.as_u32()));
let _ = self.index_allocator.free(old_idx);
}
if remote_epoch_changed {
self.unregister_decrypt_worker_fsp_session(&peer_node_addr);
if self.sessions.remove(&peer_node_addr).is_some() {
debug!(
peer = %self.peer_display_name(&peer_node_addr),
"Cleared stale FSP session after peer restart during promotion"
);
}
info!(
peer = %self.peer_display_name(&peer_node_addr),
winner_link = %link_id,
loser_link = %loser_link_id,
"Peer restart detected during promotion, replacing stale active peer"
);
}
self.seed_path_mtu_for_link_peer(&peer_node_addr, transport_id, ¤t_addr);
let mut new_peer = ActivePeer::with_session(
verified_identity,
link_id,
current_time_ms,
noise_session,
our_index,
their_index,
transport_id,
current_addr,
link_stats,
is_outbound,
&self.config.node.mmp,
remote_epoch,
);
new_peer.set_tree_announce_min_interval_ms(
self.config.node.tree.announce_min_interval_ms,
);
let inserted = self
.peers
.insert_with_current_session_index(peer_node_addr, new_peer);
self.log_active_peer_insert_result(
&peer_node_addr,
&inserted,
"cross_connection_won",
);
self.clear_session_direct_path_degraded(&peer_node_addr);
self.clear_retry_unless_direct_refresh_needed(&peer_node_addr);
self.set_discovery_fallback_transit_allowed(
peer_node_addr,
discovery_fallback_transit_allowed,
);
self.register_identity(peer_node_addr, verified_identity.pubkey_full());
// Hand the new FMP recv state to the decrypt-worker
// shard. The sibling "no existing peer" branch below
// already does this on initial promotion; the
// existing-peer replace branch was missing it, so a
// cross-connection winner ended up never registered
// with the worker and silently fell back to the
// in-line decrypt path for the lifetime of the peer.
self.register_decrypt_worker_session(&peer_node_addr);
debug!(
peer = %self.peer_display_name(&peer_node_addr),
winner_link = %link_id,
loser_link = %loser_link_id,
"Cross-connection resolved: this connection won"
);
Ok(PromotionResult::CrossConnectionWon {
loser_link_id,
node_addr: peer_node_addr,
})
} else {
// This connection loses, keep existing
// Free the index we allocated
let _ = self.index_allocator.free(our_index);
debug!(
peer = %self.peer_display_name(&peer_node_addr),
winner_link = %existing_link_id,
loser_link = %link_id,
"Cross-connection resolved: this connection lost"
);
Ok(PromotionResult::CrossConnectionLost {
winner_link_id: existing_link_id,
})
}
} else {
// No existing promoted peer. There may be a pending outbound
// connection to the same peer (cross-connection in progress).
// Do NOT clean it up yet — we need the outbound to stay alive
// so that when the peer's msg2 arrives, we can learn the peer's
// inbound session index and update their_index on the promoted
// peer. The outbound will be cleaned up in handle_msg2 or by
// the 30s handshake timeout.
let pending_to_same_peer: Vec<LinkId> = self
.peers
.connection_iter()
.filter(|(_, conn)| {
conn.expected_identity()
.map(|id| *id.node_addr() == peer_node_addr)
.unwrap_or(false)
})
.map(|(lid, _)| *lid)
.collect();
for pending_link_id in &pending_to_same_peer {
debug!(
peer = %self.peer_display_name(&peer_node_addr),
pending_link_id = %pending_link_id,
promoted_link_id = %link_id,
"Deferring cleanup of pending outbound (awaiting msg2 for index update)"
);
}
// Normal promotion
if self.max_peers > 0 && self.peers.len() >= self.max_peers {
let _ = self.index_allocator.free(our_index);
return Err(NodeError::MaxPeersExceeded {
max: self.max_peers,
});
}
// Preserve tree announce rate-limit state from old peer (if reconnecting).
// Without this, reconnection resets the rate limit window to zero,
// allowing an immediate announce that can feed an announce loop.
let old_announce_ts = self
.peers
.get(&peer_node_addr)
.map(|p| p.last_tree_announce_sent_ms());
self.seed_path_mtu_for_link_peer(&peer_node_addr, transport_id, ¤t_addr);
let mut new_peer = ActivePeer::with_session(
verified_identity,
link_id,
current_time_ms,
noise_session,
our_index,
their_index,
transport_id,
current_addr,
link_stats,
is_outbound,
&self.config.node.mmp,
remote_epoch,
);
new_peer
.set_tree_announce_min_interval_ms(self.config.node.tree.announce_min_interval_ms);
if let Some(ts) = old_announce_ts {
new_peer.set_last_tree_announce_sent_ms(ts);
}
let inserted = self
.peers
.insert_with_current_session_index(peer_node_addr, new_peer);
self.log_active_peer_insert_result(&peer_node_addr, &inserted, "promoted");
self.clear_session_direct_path_degraded(&peer_node_addr);
self.clear_retry_unless_direct_refresh_needed(&peer_node_addr);
self.set_discovery_fallback_transit_allowed(
peer_node_addr,
discovery_fallback_transit_allowed,
);
self.register_identity(peer_node_addr, verified_identity.pubkey_full());
// Eagerly hand the FMP recv state to the decrypt-worker
// shard. From this point on the shard is the
// authoritative FMP-replay-window writer for this peer;
// rx_loop's in-line decrypt path is no longer used.
self.register_decrypt_worker_session(&peer_node_addr);
info!(
peer = %self.peer_display_name(&peer_node_addr),
link_id = %link_id,
our_index = %our_index,
their_index = %their_index,
"Connection promoted to active peer"
);
Ok(PromotionResult::Promoted(peer_node_addr))
}
}
}