Wait for the semantic goal of "learner is now tracked in `routing.group_info(group_id).learners`", polling every [`CONF_CHANGE_POLL_INTERVAL`] up to `deadline`. This is the post-apply condition that `apply_conf_change` writes once a committed `AddLearner` entry has been applied to the local state. Polling this rather than the raw `commit_index` is what lets the join flow stay correct on multi-vot
(
&self,
group_id: u64,
learner_id: u64,
log_index: u64,
deadline: Instant,
)
| 311 | /// flow can match the cause and so the crate's central |
| 312 | /// error enum owns the human-readable rendering. |
| 313 | async fn wait_for_learner_applied( |
| 314 | &self, |
| 315 | group_id: u64, |
| 316 | learner_id: u64, |
| 317 | log_index: u64, |
| 318 | deadline: Instant, |
| 319 | ) -> Result<()> { |
| 320 | loop { |
| 321 | let applied = { |
| 322 | let mr = self.multi_raft.lock().unwrap_or_else(|p| p.into_inner()); |
| 323 | mr.routing() |
| 324 | .group_info(group_id) |
| 325 | .map(|info| info.learners.contains(&learner_id)) |
| 326 | }; |
| 327 | match applied { |
| 328 | Some(true) => return Ok(()), |
| 329 | Some(false) => {} |
| 330 | None => return Err(ClusterError::JoinGroupDisappeared { group_id }), |
| 331 | } |
| 332 | if Instant::now() >= deadline { |
| 333 | return Err(ClusterError::JoinCommitTimeout { |
| 334 | group_id, |
| 335 | log_index, |
| 336 | }); |
| 337 | } |
| 338 | tokio::time::sleep(CONF_CHANGE_POLL_INTERVAL).await; |
| 339 | } |
| 340 | } |
| 341 | |
| 342 | /// Build a `JoinResponse` snapshotting the current topology |
| 343 | /// and routing. Used both by the happy-path return and by the |