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,
)
| 143 | /// `JoinResponse` or an error describing the last failure in this |
| 144 | /// attempt. |
| 145 | async 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, |
no test coverage detected