MCPcopy Create free account
hub / github.com/NodeDB-Lab/nodedb / try_join_once

Function try_join_once

nodedb-cluster/src/bootstrap/join.rs:145–225  ·  view source on GitHub ↗

One pass over the seed list plus up to `MAX_REDIRECTS_PER_ATTEMPT` leader-redirect hops. Returns `Ok(state)` on the first successful `JoinResponse` or an error describing the last failure in this attempt.

(
    config: &ClusterConfig,
    catalog: &ClusterCatalog,
    transport: &NexarTransport,
    req_template: &JoinRequest,
)

Source from the content-addressed store, hash-verified

143/// `JoinResponse` or an error describing the last failure in this
144/// attempt.
145async fn try_join_once(
146 config: &ClusterConfig,
147 catalog: &ClusterCatalog,
148 transport: &NexarTransport,
149 req_template: &JoinRequest,
150) -> Result<ClusterState> {
151 // Work list: try seeds in sorted order so the lexicographically
152 // smallest address — the designated bootstrapper under the
153 // single-elected-bootstrapper rule — is contacted first. This is
154 // critical during the initial 5-node race: every other seed points
155 // at a node that is itself still joining, so asking them first
156 // eats the full RPC timeout per non-bootstrapper before we reach
157 // the one peer that can actually answer. `HashSet` deduplicates
158 // so a redirect loop can't consume all attempts against the same
159 // address.
160 let mut work: std::collections::VecDeque<SocketAddr> =
161 config.seed_nodes.iter().copied().collect();
162 {
163 // Sort so the designated bootstrapper surfaces first. Leader
164 // redirects get prepended with push_front below, keeping the
165 // "most likely to answer" candidate at the head.
166 let mut sorted: Vec<SocketAddr> = work.drain(..).collect();
167 sorted.sort();
168 work.extend(sorted);
169 }
170 let mut visited: HashSet<SocketAddr> = HashSet::new();
171 let mut redirects: u32 = 0;
172 let mut last_err: Option<ClusterError> = None;
173
174 while let Some(addr) = work.pop_front() {
175 if !visited.insert(addr) {
176 continue;
177 }
178
179 let rpc = RaftRpc::JoinRequest(req_template.clone());
180 match transport.send_rpc_to_addr(addr, rpc).await {
181 Ok(RaftRpc::JoinResponse(resp)) => {
182 if resp.success {
183 return apply_join_response(config, catalog, transport, &resp);
184 }
185 // Rejected — is it a leader redirect we can follow?
186 if let Some(leader) = parse_leader_hint(&resp.error) {
187 if redirects < MAX_REDIRECTS_PER_ATTEMPT && !visited.contains(&leader) {
188 info!(
189 node_id = config.node_id,
190 from = %addr,
191 to = %leader,
192 "following leader redirect"
193 );
194 redirects += 1;
195 work.push_front(leader);
196 continue;
197 }
198 debug!(
199 node_id = config.node_id,
200 from = %addr,
201 leader = %leader,
202 redirects,

Callers 1

joinFunction · 0.85

Calls 13

JoinRequestClass · 0.85
apply_join_responseFunction · 0.85
parse_leader_hintFunction · 0.85
collectMethod · 0.80
pop_frontMethod · 0.80
send_rpc_to_addrMethod · 0.80
push_frontMethod · 0.80
iterMethod · 0.45
drainMethod · 0.45
extendMethod · 0.45
insertMethod · 0.45
cloneMethod · 0.45

Tested by

no test coverage detected