Complete a waiter after the entry has been committed and executed. If the proposer has already called `register()`, the result is sent immediately. If not, the result is stored so the next `register()` call picks it up without waiting. Returns true if a live waiter was found and notified, false otherwise.
(
&self,
group_id: u64,
log_index: u64,
applied_key: u64,
result: ProposeResult,
)
| 123 | /// |
| 124 | /// Returns true if a live waiter was found and notified, false otherwise. |
| 125 | pub fn complete( |
| 126 | &self, |
| 127 | group_id: u64, |
| 128 | log_index: u64, |
| 129 | applied_key: u64, |
| 130 | result: ProposeResult, |
| 131 | ) -> bool { |
| 132 | // Bump the per-group apply watermark. Bumping unconditionally |
| 133 | // (success and error) keeps the watcher monotonic with raft's |
| 134 | // commit progression — a data-plane error means "the entry |
| 135 | // could not be applied" but the entry IS committed and Raft |
| 136 | // has advanced its applied index. Tests waiting on |
| 137 | // visibility care about the success path; liveness on the |
| 138 | // error path requires the bump too. |
| 139 | if let Some(w) = &self.group_watchers { |
| 140 | w.bump(group_id, log_index); |
| 141 | } |
| 142 | |
| 143 | let mut slots = self.slots.lock().unwrap_or_else(|p| p.into_inner()); |
| 144 | match slots.entry((group_id, log_index)) { |
| 145 | Entry::Vacant(e) => { |
| 146 | // No waiter yet — store result for the upcoming register(). |
| 147 | e.insert(TrackerSlot::Completed(result)); |
| 148 | false |
| 149 | } |
| 150 | Entry::Occupied(e) => { |
| 151 | match e.get() { |
| 152 | TrackerSlot::Waiting { expected_key, .. } => { |
| 153 | // Idempotency-key gate: the entry that committed |
| 154 | // at this (group_id, log_index) must be the one |
| 155 | // the proposer reserved. If the keys disagree, |
| 156 | // a leader change overwrote the proposer's entry |
| 157 | // with a different one — surface the retryable |
| 158 | // signal instead of the (success-shaped) result |
| 159 | // that belongs to a different proposer. A zero |
| 160 | // applied_key means "no key carried" (empty |
| 161 | // entry / legacy); a zero expected_key means the |
| 162 | // registration is wildcard (legacy callers). |
| 163 | let mismatch = |
| 164 | applied_key != 0 && *expected_key != 0 && applied_key != *expected_key; |
| 165 | let final_result = if mismatch { |
| 166 | tracing::warn!( |
| 167 | group_id, |
| 168 | log_index, |
| 169 | applied_key, |
| 170 | expected_key = *expected_key, |
| 171 | "raft entry at proposer's index was overwritten by \ |
| 172 | a different proposal (idempotency_key mismatch); \ |
| 173 | surfacing RetryableLeaderChange" |
| 174 | ); |
| 175 | Err(crate::Error::RetryableLeaderChange { |
| 176 | group_id, |
| 177 | log_index, |
| 178 | }) |
| 179 | } else { |
| 180 | result |
| 181 | }; |
| 182 | if let TrackerSlot::Waiting { tx, .. } = e.remove() { |
no test coverage detected