reddb_server/cluster/
hot_mirror_failover.rs

1//! Operator-invoked hot mirror failover workflow.
2//!
3//! This is the narrow planned/zero-RPO path: choose a current range replica whose
4//! applied log covers the range commit watermark, promote it through the ordinary
5//! ownership transition machine, and return evidence that distinguishes this
6//! from archive-based recovery.
7
8use super::identity::NodeIdentity;
9use super::ownership::{
10    CatalogVersion, CollectionId, OwnershipEpoch, RangeId, ShardOwnershipCatalog,
11};
12use super::ownership_transition::{
13    run_transition, CatchUpEvidence, CommitWatermark, TransitionError, TransitionKind,
14    TransitionOutcome, TransitionRequest,
15};
16
17/// Inputs the planner reads for one range failover.
18#[derive(Debug)]
19pub struct HotMirrorInputs<'a> {
20    catalog: &'a ShardOwnershipCatalog,
21    collection: CollectionId,
22    range_id: RangeId,
23    watermark: CommitWatermark,
24    catch_up: Vec<CatchUpEvidence>,
25}
26
27impl<'a> HotMirrorInputs<'a> {
28    pub fn new(
29        catalog: &'a ShardOwnershipCatalog,
30        collection: CollectionId,
31        range_id: RangeId,
32        watermark: CommitWatermark,
33    ) -> Self {
34        Self {
35            catalog,
36            collection,
37            range_id,
38            watermark,
39            catch_up: Vec::new(),
40        }
41    }
42
43    pub fn with_catch_up(mut self, evidence: CatchUpEvidence) -> Self {
44        self.catch_up.push(evidence);
45        self
46    }
47}
48
49/// Whether a mirror covers the range commit watermark.
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub enum WatermarkOutcome {
52    Covered,
53    Behind { applied_term: u64, applied_lsn: u64 },
54}
55
56/// A hot mirror considered by the failover planner.
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct HotMirrorFailoverCandidate {
59    pub candidate: NodeIdentity,
60    pub watermark_outcome: WatermarkOutcome,
61}
62
63/// The range identity carried in operator-facing failover evidence.
64#[derive(Debug, Clone, PartialEq, Eq)]
65pub struct HotMirrorRangeEvidence {
66    pub collection: CollectionId,
67    pub range_id: RangeId,
68}
69
70/// The ownership epoch boundary crossed by a successful promotion.
71#[derive(Debug, Clone, Copy, PartialEq, Eq)]
72pub struct HotMirrorEpochEvidence {
73    pub previous: OwnershipEpoch,
74    pub new: OwnershipEpoch,
75}
76
77/// Operator-facing result evidence for a successful hot mirror promotion.
78#[derive(Debug, Clone, PartialEq, Eq)]
79pub struct HotMirrorFailoverEvidence {
80    pub workflow: &'static str,
81    pub range: HotMirrorRangeEvidence,
82    pub previous_owner: NodeIdentity,
83    pub promoted_owner: NodeIdentity,
84    pub epoch: HotMirrorEpochEvidence,
85    pub watermark: CommitWatermark,
86    pub watermark_outcome: WatermarkOutcome,
87}
88
89impl HotMirrorFailoverEvidence {
90    fn from_transition(outcome: TransitionOutcome) -> Self {
91        Self {
92            workflow: "hot-mirror-zero-rpo",
93            range: HotMirrorRangeEvidence {
94                collection: outcome.collection,
95                range_id: outcome.range_id,
96            },
97            previous_owner: outcome.previous_owner,
98            promoted_owner: outcome.new_owner,
99            epoch: HotMirrorEpochEvidence {
100                previous: outcome.previous_epoch,
101                new: outcome.new_epoch,
102            },
103            watermark: outcome.watermark,
104            watermark_outcome: WatermarkOutcome::Covered,
105        }
106    }
107}
108
109/// A validated plan for one hot mirror failover.
110#[derive(Debug, Clone, PartialEq, Eq)]
111pub struct HotMirrorFailoverPlan {
112    collection: CollectionId,
113    range_id: RangeId,
114    previous_owner: NodeIdentity,
115    expected_epoch: OwnershipEpoch,
116    expected_version: CatalogVersion,
117    watermark: CommitWatermark,
118    current_replicas: Vec<NodeIdentity>,
119    eligible_candidates: Vec<NodeIdentity>,
120    eligible_evidence: Vec<CatchUpEvidence>,
121    ineligible_candidates: Vec<HotMirrorFailoverCandidate>,
122}
123
124impl HotMirrorFailoverPlan {
125    pub fn previous_owner(&self) -> &NodeIdentity {
126        &self.previous_owner
127    }
128
129    pub fn eligible_candidates(&self) -> &[NodeIdentity] {
130        &self.eligible_candidates
131    }
132
133    pub fn ineligible_candidates(&self) -> &[HotMirrorFailoverCandidate] {
134        &self.ineligible_candidates
135    }
136
137    fn evidence_for(&self, candidate: &NodeIdentity) -> Option<CatchUpEvidence> {
138        self.eligible_evidence
139            .iter()
140            .find(|evidence| evidence.candidate == *candidate)
141            .cloned()
142    }
143}
144
145/// Why planning or executing a hot mirror failover failed.
146#[derive(Debug, Clone, PartialEq, Eq)]
147pub enum HotMirrorFailoverError {
148    UnknownRange {
149        collection: CollectionId,
150        range_id: RangeId,
151    },
152    CandidateNotEligible {
153        collection: CollectionId,
154        range_id: RangeId,
155        candidate: NodeIdentity,
156    },
157    Transition(TransitionError),
158}
159
160impl std::fmt::Display for HotMirrorFailoverError {
161    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
162        match self {
163            Self::UnknownRange {
164                collection,
165                range_id,
166            } => write!(
167                f,
168                "no range {collection}/{range_id} for hot mirror failover"
169            ),
170            Self::CandidateNotEligible {
171                collection,
172                range_id,
173                candidate,
174            } => write!(
175                f,
176                "candidate {candidate} is not an eligible hot mirror for {collection}/{range_id}"
177            ),
178            Self::Transition(err) => write!(f, "{err}"),
179        }
180    }
181}
182
183impl std::error::Error for HotMirrorFailoverError {}
184
185/// Identify current replica mirrors that can be promoted without crossing the
186/// range commit watermark safety line.
187pub fn plan_hot_mirror_failover(
188    inputs: &HotMirrorInputs<'_>,
189) -> Result<HotMirrorFailoverPlan, HotMirrorFailoverError> {
190    let range = inputs
191        .catalog
192        .range(&inputs.collection, inputs.range_id)
193        .ok_or_else(|| HotMirrorFailoverError::UnknownRange {
194            collection: inputs.collection.clone(),
195            range_id: inputs.range_id,
196        })?;
197
198    let mut eligible_candidates = Vec::new();
199    let mut eligible_evidence = Vec::new();
200    let mut ineligible_candidates = Vec::new();
201
202    for replica in range.replicas() {
203        let Some(evidence) = inputs
204            .catch_up
205            .iter()
206            .find(|evidence| evidence.candidate == *replica)
207        else {
208            ineligible_candidates.push(HotMirrorFailoverCandidate {
209                candidate: replica.clone(),
210                watermark_outcome: WatermarkOutcome::Behind {
211                    applied_term: 0,
212                    applied_lsn: 0,
213                },
214            });
215            continue;
216        };
217
218        if evidence.covers(inputs.watermark) {
219            eligible_candidates.push(replica.clone());
220            eligible_evidence.push(evidence.clone());
221        } else {
222            ineligible_candidates.push(HotMirrorFailoverCandidate {
223                candidate: replica.clone(),
224                watermark_outcome: WatermarkOutcome::Behind {
225                    applied_term: evidence.applied_term,
226                    applied_lsn: evidence.applied_lsn,
227                },
228            });
229        }
230    }
231
232    Ok(HotMirrorFailoverPlan {
233        collection: inputs.collection.clone(),
234        range_id: inputs.range_id,
235        previous_owner: range.owner().clone(),
236        expected_epoch: range.epoch(),
237        expected_version: range.version(),
238        watermark: inputs.watermark,
239        current_replicas: range.replicas().to_vec(),
240        eligible_candidates,
241        eligible_evidence,
242        ineligible_candidates,
243    })
244}
245
246/// Promote one eligible hot mirror and return the operator-facing evidence.
247pub fn execute_hot_mirror_failover(
248    catalog: &mut ShardOwnershipCatalog,
249    plan: &HotMirrorFailoverPlan,
250    candidate: &NodeIdentity,
251) -> Result<HotMirrorFailoverEvidence, HotMirrorFailoverError> {
252    let evidence = plan.evidence_for(candidate).ok_or_else(|| {
253        HotMirrorFailoverError::CandidateNotEligible {
254            collection: plan.collection.clone(),
255            range_id: plan.range_id,
256            candidate: candidate.clone(),
257        }
258    })?;
259    let request = TransitionRequest::new(
260        TransitionKind::Promote,
261        plan.collection.clone(),
262        plan.range_id,
263        plan.previous_owner.clone(),
264        plan.expected_epoch,
265        plan.expected_version,
266        candidate.clone(),
267        plan.watermark,
268    )
269    .with_evidence(evidence)
270    .with_replicas(
271        plan.current_replicas
272            .iter()
273            .filter(|replica| *replica != candidate)
274            .cloned(),
275    );
276
277    run_transition(catalog, &request)
278        .map(HotMirrorFailoverEvidence::from_transition)
279        .map_err(HotMirrorFailoverError::Transition)
280}
281
282#[cfg(test)]
283mod tests {
284    use super::*;
285    use crate::cluster::identity::NodeIdentity;
286    use crate::cluster::ownership::{
287        CollectionId, PlacementMetadata, RangeBounds, RangeId, RangeOwnership, ShardKeyMode,
288        ShardOwnershipCatalog,
289    };
290    use crate::cluster::ownership_lease::{
291        admit_durable_write, DurableWriteReject, LeasedOwner, OwnershipLease, SupervisorTerm,
292    };
293    use crate::cluster::ownership_transition::{CatchUpEvidence, CommitWatermark};
294
295    fn collection(name: &str) -> CollectionId {
296        CollectionId::new(name).unwrap()
297    }
298
299    fn ident(cn: &str) -> NodeIdentity {
300        NodeIdentity::from_certificate_subject(cn).unwrap()
301    }
302
303    fn catalog_with(owner: &str, replicas: &[&str]) -> (ShardOwnershipCatalog, CollectionId) {
304        let orders = collection("orders");
305        let mut catalog = ShardOwnershipCatalog::new();
306        catalog
307            .apply_update(RangeOwnership::establish(
308                orders.clone(),
309                RangeId::new(1),
310                ShardKeyMode::Hash,
311                RangeBounds::full(),
312                ident(owner),
313                replicas.iter().map(|r| ident(r)).collect::<Vec<_>>(),
314                PlacementMetadata::with_replication_factor(3),
315            ))
316            .unwrap();
317        (catalog, orders)
318    }
319
320    #[test]
321    fn hot_mirror_failover_promotes_only_a_candidate_covering_the_watermark() {
322        let (mut catalog, orders) = catalog_with("CN=node-a", &["CN=node-b", "CN=node-c"]);
323        let watermark = CommitWatermark::new(7, 400);
324        let inputs = HotMirrorInputs::new(&catalog, orders.clone(), RangeId::new(1), watermark)
325            .with_catch_up(CatchUpEvidence::new(ident("CN=node-b"), 7, 400))
326            .with_catch_up(CatchUpEvidence::new(ident("CN=node-c"), 7, 399));
327
328        let plan = plan_hot_mirror_failover(&inputs).expect("range is known");
329        assert_eq!(plan.previous_owner(), &ident("CN=node-a"));
330        assert_eq!(plan.eligible_candidates(), &[ident("CN=node-b")]);
331        assert_eq!(
332            plan.ineligible_candidates()[0].candidate,
333            ident("CN=node-c")
334        );
335
336        let stale = execute_hot_mirror_failover(&mut catalog, &plan, &ident("CN=node-c"))
337            .expect_err("behind candidate is refused");
338        assert!(matches!(
339            stale,
340            HotMirrorFailoverError::CandidateNotEligible { .. }
341        ));
342        assert_eq!(
343            catalog.range(&orders, RangeId::new(1)).unwrap().owner(),
344            &ident("CN=node-a")
345        );
346
347        let old_epoch = catalog.range(&orders, RangeId::new(1)).unwrap().epoch();
348        let old_owner_lease = LeasedOwner::with_lease(OwnershipLease::grant(
349            SupervisorTerm::genesis(),
350            orders.clone(),
351            RangeId::new(1),
352            ident("CN=node-a"),
353            old_epoch,
354            0,
355            60_000,
356        ));
357
358        let evidence =
359            execute_hot_mirror_failover(&mut catalog, &plan, &ident("CN=node-b")).unwrap();
360        assert_eq!(evidence.range.collection, orders);
361        assert_eq!(evidence.previous_owner, ident("CN=node-a"));
362        assert_eq!(evidence.promoted_owner, ident("CN=node-b"));
363        assert_eq!(evidence.watermark, watermark);
364        assert_eq!(evidence.watermark_outcome, WatermarkOutcome::Covered);
365        assert!(evidence.epoch.new > evidence.epoch.previous);
366
367        let reject = admit_durable_write(
368            &catalog,
369            &old_owner_lease,
370            &ident("CN=node-a"),
371            &orders,
372            b"customer-1",
373            SupervisorTerm::genesis(),
374            1,
375        )
376        .unwrap_err();
377        assert!(matches!(reject, DurableWriteReject::StaleOwnership { .. }));
378    }
379}
reddb_server/cluster/hot_mirror_failover.rs

reddb_server/cluster/
hot_mirror_failover.rs
