use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};
use melin_raft::recency::{JournalTip, PeerTips};
use melin_transport_core::health::RaftStatus;
use tracing::{info, warn};
use melin_journal::AckPolicyByte;
use crate::ack_policy::AckPolicy;
use crate::policy_control::AckPolicyControl;
use crate::replication::ReplicaControlPlane;
const POLL_INTERVAL: Duration = Duration::from_millis(100);
const PRIMARY_DOWN_GRACE: Duration = Duration::from_secs(3);
const PEER_TIP_GRACE: Duration = Duration::from_millis(1500);
const TIP_NOT_READY: &str = "journal recovery has not seeded this node's tip yet";
const FENCED: &str = "node is fenced (superseded by a newer primary)";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PeerTipObservation {
Fresh(JournalTip),
Silent { last: Option<JournalTip> },
Unknown,
}
fn observe_peer(
peer_tips: &PeerTips,
peer: u64,
leader_for: Duration,
now: Instant,
) -> PeerTipObservation {
match peer_tips.sample(peer, now) {
Some((tip, age)) if age <= PEER_TIP_GRACE => PeerTipObservation::Fresh(tip),
Some((tip, _)) => PeerTipObservation::Silent { last: Some(tip) },
None if leader_for > PEER_TIP_GRACE => PeerTipObservation::Silent { last: None },
None => PeerTipObservation::Unknown,
}
}
fn effective_ack_policy(
lineage: Option<AckPolicyByte>,
local_fallback: impl FnOnce() -> AckPolicy,
) -> Option<AckPolicy> {
match lineage {
Some(byte) => AckPolicy::from_byte(byte),
None => Some(local_fallback()),
}
}
struct AutoPromotionInputs {
tip_ready: bool,
fenced: bool,
ack_policy: Option<AckPolicy>,
primary_link_up: bool,
primary_link_down_for: Duration,
primary_observed: bool,
local_tip: JournalTip,
peers: Vec<(u64, PeerTipObservation)>,
term: u64,
fence_epoch: u64,
}
fn auto_promotion_decision(inputs: &AutoPromotionInputs) -> Result<(), &'static str> {
if !inputs.tip_ready {
return Err(TIP_NOT_READY);
}
if inputs.fenced {
return Err(FENCED);
}
if inputs.primary_link_up {
return Err("replication link to the primary is up — refusing to depose a live primary");
}
if inputs.primary_link_down_for < PRIMARY_DOWN_GRACE {
return Err(
"the primary link dropped only moments ago — waiting out the grace period to tell \
a transient blip from a real failure before failover",
);
}
if !inputs.primary_observed && inputs.local_tip.last_sequence == 0 {
return Err(
"no primary has been observed since boot and the local journal is empty — a blank \
genesis node must not depose a primary that may still be starting; bring the \
primary up first, or promote manually",
);
}
for (_peer, obs) in &inputs.peers {
match obs {
PeerTipObservation::Fresh(tip) if tip.is_ahead_of(inputs.local_tip) => {
return Err(
"a live peer advertises a journal tip ahead of ours — it holds acked events \
this node lacks; refusing to promote so the caught-up peer can take over \
(it campaigns on its own once it sees this leadership)",
);
}
PeerTipObservation::Fresh(_) | PeerTipObservation::Silent { .. } => {}
PeerTipObservation::Unknown => {
return Err(
"a configured peer's journal tip is still unknown — waiting up to one \
peer-tip grace to learn whether it holds acked events this node lacks",
);
}
}
}
match inputs.ack_policy {
Some(AckPolicy::Disk) => {
return Err(
"the primary acks under the `disk` policy — an election win cannot prove \
this node holds every acked event; promote manually if the lag is acceptable",
);
}
None => return Err("ack policy is unrecognised"),
Some(AckPolicy::DiskAndRam | AckPolicy::TwoDisks | AckPolicy::Ram) => {}
}
if inputs.term <= inputs.fence_epoch {
return Err(
"election term is not above the fencing epoch (manual promotions outran raft \
terms) — promote manually; the alignment heals as terms advance",
);
}
Ok(())
}
#[allow(clippy::too_many_arguments)] fn consider_auto_promotion(
status: &RaftStatus,
control: &ReplicaControlPlane,
fence_state: &melin_transport_core::fence::FenceState,
ack_policy: &AckPolicyControl,
peer_tips: &PeerTips,
peer_ids: &[u64],
primary_link_down_for: Duration,
leader_for: Duration,
last_refused: &mut Option<(u64, &'static str)>,
) {
if !status.running.load(Ordering::Relaxed)
|| status.role.load(Ordering::Relaxed) != RaftStatus::ROLE_LEADER
|| control.promote.is_requested()
{
return;
}
let term = status.term.load(Ordering::Relaxed);
let fence_epoch = fence_state.epoch();
let last_sequence = control.journal_tip.load().get();
let lineage_policy = control.lineage_ack_policy.load().map(|p| p.policy());
let now = Instant::now();
let inputs = AutoPromotionInputs {
tip_ready: control.tip_ready.load(Ordering::Acquire),
fenced: fence_state.is_fenced(),
ack_policy: effective_ack_policy(lineage_policy, || ack_policy.requested()),
primary_link_up: control.primary_link_up.load(Ordering::Acquire),
primary_link_down_for,
primary_observed: control.primary_observed.load(Ordering::Acquire),
local_tip: JournalTip {
epoch: fence_epoch,
last_sequence,
},
peers: peer_ids
.iter()
.map(|&p| (p, observe_peer(peer_tips, p, leader_for, now)))
.collect(),
term,
fence_epoch,
};
match auto_promotion_decision(&inputs) {
Ok(()) => {
for (peer, obs) in &inputs.peers {
if let PeerTipObservation::Silent { last: Some(tip) } = obs
&& tip.is_ahead_of(inputs.local_tip)
{
warn!(
node_id = status.node_id,
peer,
peer_tip = ?tip,
local_tip = ?inputs.local_tip,
"promoting although an unreachable peer last advertised a journal tip \
ahead of ours — events acked past our tip exist only on that node; \
reconcile from its journal when it returns"
);
}
}
if control.promote.request(term) {
info!(
node_id = status.node_id,
term, "elected leader — auto-promoting this replica"
);
}
}
Err(reason) => {
if *last_refused != Some((term, reason)) {
*last_refused = Some((term, reason));
warn!(
node_id = status.node_id,
term,
reason,
peers = ?inputs.peers,
local_tip = ?inputs.local_tip,
"elected leader but refusing auto-promotion"
);
}
}
}
}
fn challenger_should_campaign(
tip_ready: bool,
fenced: bool,
primary_link_up: bool,
primary_link_down_for: Duration,
local_tip: JournalTip,
leader_tip: Option<JournalTip>,
) -> bool {
if !tip_ready || fenced || primary_link_up || primary_link_down_for < PRIMARY_DOWN_GRACE {
return false;
}
matches!(leader_tip, Some(tip) if local_tip.is_ahead_of(tip))
}
const CHALLENGER_HOLD_DOWN: Duration = Duration::from_millis(1500);
fn consider_challenger_nudge(
status: &RaftStatus,
control: &ReplicaControlPlane,
fence_state: &melin_transport_core::fence::FenceState,
peer_tips: &PeerTips,
elect_requested: &AtomicBool,
primary_link_down_for: Duration,
behind_since: &mut Option<(u64, u64, Instant)>,
last_nudged_term: &mut u64,
) {
if !status.running.load(Ordering::Relaxed)
|| status.role.load(Ordering::Relaxed) == RaftStatus::ROLE_LEADER
|| control.promote.is_requested()
{
*behind_since = None;
return;
}
let leader_id = status.leader_id.load(Ordering::Relaxed);
if leader_id == 0 || leader_id == status.node_id {
*behind_since = None;
return;
}
let term = status.term.load(Ordering::Relaxed);
let local_tip = JournalTip {
epoch: fence_state.epoch(),
last_sequence: control.journal_tip.load().get(),
};
let leader_tip = peer_tips
.sample(leader_id, Instant::now())
.filter(|(_, age)| *age <= PEER_TIP_GRACE)
.map(|(tip, _)| tip);
if !challenger_should_campaign(
control.tip_ready.load(Ordering::Acquire),
fence_state.is_fenced(),
control.primary_link_up.load(Ordering::Acquire),
primary_link_down_for,
local_tip,
leader_tip,
) {
*behind_since = None;
return;
}
let since = match *behind_since {
Some((l, t, s)) if l == leader_id && t == term => s,
_ => {
*behind_since = Some((leader_id, term, Instant::now()));
return;
}
};
if since.elapsed() < CHALLENGER_HOLD_DOWN || *last_nudged_term == term {
return;
}
*last_nudged_term = term;
elect_requested.store(true, Ordering::Release);
info!(
node_id = status.node_id,
leader_id,
term,
?local_tip,
?leader_tip,
"control-plane leader has been behind this node's journal tip through the hold-down — \
campaigning so promotion lands on the caught-up node"
);
}
const STAND_DOWN_GRACE: Duration = Duration::from_secs(5);
fn election_should_stand_down(
tip_ready: bool,
fenced: bool,
local_tip: JournalTip,
peer_samples: &[(JournalTip, Duration)],
) -> Option<&'static str> {
if !tip_ready {
return Some(TIP_NOT_READY);
}
if fenced {
return Some(FENCED);
}
peer_samples
.iter()
.any(|(tip, age)| *age <= STAND_DOWN_GRACE && tip.is_ahead_of(local_tip))
.then_some("a recently heard peer advertises a journal tip ahead of ours")
}
fn consider_election_stand_down(
status: &RaftStatus,
control: &ReplicaControlPlane,
fence_state: &melin_transport_core::fence::FenceState,
peer_tips: &PeerTips,
peer_ids: &[u64],
elect_enabled: &AtomicBool,
last_reason: &mut Option<&'static str>,
) {
let now = Instant::now();
let local_tip = JournalTip {
epoch: fence_state.epoch(),
last_sequence: control.journal_tip.load().get(),
};
let peer_samples: Vec<(JournalTip, Duration)> = peer_ids
.iter()
.filter_map(|&p| peer_tips.sample(p, now))
.collect();
let stand_down = election_should_stand_down(
control.tip_ready.load(Ordering::Acquire),
fence_state.is_fenced(),
local_tip,
&peer_samples,
);
elect_enabled.store(stand_down.is_none(), Ordering::Release);
if *last_reason != stand_down {
match stand_down {
Some(reason) => info!(
node_id = status.node_id,
reason,
?local_tip,
"standing down from control-plane elections — this node will not campaign \
on a timeout while the condition holds"
),
None => info!(
node_id = status.node_id,
?local_tip,
"resuming control-plane elections — the stand-down condition no longer holds"
),
}
*last_reason = stand_down;
}
}
#[allow(clippy::too_many_arguments)] pub(crate) fn spawn_auto_promotion(
status: Arc<RaftStatus>,
control: ReplicaControlPlane,
fence_state: Arc<melin_transport_core::fence::FenceState>,
ack_policy: Arc<AckPolicyControl>,
peer_tips: Arc<PeerTips>,
peer_ids: Vec<u64>,
elect_requested: Arc<AtomicBool>,
elect_enabled: Arc<AtomicBool>,
shutdown: Arc<AtomicBool>,
) -> std::thread::JoinHandle<()> {
std::thread::Builder::new()
.name("raft-promotion".into())
.spawn(move || {
let mut last_refused: Option<(u64, &'static str)> = None;
let mut last_nudged_term = 0u64;
let mut stand_down_reason: Option<&'static str> = None;
let mut behind_since: Option<(u64, u64, Instant)> = None;
let mut primary_link_down_since: Option<Instant> = None;
let mut leader_since: Option<(u64, Instant)> = None;
while !shutdown.load(Ordering::Relaxed) {
if control.primary_link_up.load(Ordering::Acquire) {
primary_link_down_since = None;
} else if primary_link_down_since.is_none() {
primary_link_down_since = Some(Instant::now());
}
let primary_link_down_for = primary_link_down_since
.map(|since| since.elapsed())
.unwrap_or(Duration::ZERO);
let term = status.term.load(Ordering::Relaxed);
let is_leader = status.role.load(Ordering::Relaxed) == RaftStatus::ROLE_LEADER;
if !is_leader {
leader_since = None;
} else if leader_since.is_none_or(|(t, _)| t != term) {
leader_since = Some((term, Instant::now()));
}
let leader_for = leader_since
.map(|(_, since)| since.elapsed())
.unwrap_or(Duration::ZERO);
consider_auto_promotion(
&status,
&control,
&fence_state,
&ack_policy,
&peer_tips,
&peer_ids,
primary_link_down_for,
leader_for,
&mut last_refused,
);
consider_challenger_nudge(
&status,
&control,
&fence_state,
&peer_tips,
&elect_requested,
primary_link_down_for,
&mut behind_since,
&mut last_nudged_term,
);
consider_election_stand_down(
&status,
&control,
&fence_state,
&peer_tips,
&peer_ids,
&elect_enabled,
&mut stand_down_reason,
);
if control.promote.is_requested() {
break;
}
std::thread::sleep(POLL_INTERVAL);
}
elect_enabled.store(true, Ordering::Release);
})
.expect("failed to spawn raft-promotion thread")
}
#[cfg(test)]
mod tests {
use super::*;
fn tip(epoch: u64, seq: u64) -> JournalTip {
JournalTip {
epoch,
last_sequence: seq,
}
}
fn ok_inputs() -> AutoPromotionInputs {
AutoPromotionInputs {
tip_ready: true,
fenced: false,
ack_policy: Some(AckPolicy::DiskAndRam),
primary_link_up: false,
primary_link_down_for: PRIMARY_DOWN_GRACE * 10,
primary_observed: true,
local_tip: tip(3, 100),
peers: vec![
(1, PeerTipObservation::Silent { last: None }),
(3, PeerTipObservation::Fresh(tip(3, 100))),
],
term: 5,
fence_epoch: 3,
}
}
#[test]
fn healthy_leader_promotes() {
assert!(auto_promotion_decision(&ok_inputs()).is_ok());
}
#[test]
fn refuses_before_tip_ready() {
let inputs = AutoPromotionInputs {
tip_ready: false,
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs)
.unwrap_err()
.contains("tip")
);
}
#[test]
fn refuses_when_fenced() {
let inputs = AutoPromotionInputs {
fenced: true,
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs)
.unwrap_err()
.contains("fenced")
);
}
#[test]
fn refuses_while_primary_link_is_up() {
let inputs = AutoPromotionInputs {
primary_link_up: true,
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs)
.unwrap_err()
.contains("live primary")
);
}
#[test]
fn refuses_on_a_transient_primary_link_blip() {
let inputs = AutoPromotionInputs {
primary_link_up: false,
primary_link_down_for: PRIMARY_DOWN_GRACE - Duration::from_millis(1),
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs)
.unwrap_err()
.contains("moments ago"),
"a sub-grace outage must be refused as a possible blip"
);
let inputs = AutoPromotionInputs {
primary_link_up: false,
primary_link_down_for: PRIMARY_DOWN_GRACE,
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs).is_ok(),
"a sustained outage past the grace must promote"
);
}
#[test]
fn refuses_a_blank_genesis_node_but_not_a_restarted_replica() {
let inputs = AutoPromotionInputs {
primary_observed: false,
local_tip: tip(3, 0),
peers: vec![
(1, PeerTipObservation::Silent { last: None }),
(3, PeerTipObservation::Silent { last: None }),
],
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs)
.unwrap_err()
.contains("journal is empty")
);
let inputs = AutoPromotionInputs {
primary_observed: false,
..ok_inputs()
};
assert!(auto_promotion_decision(&inputs).is_ok());
let inputs = AutoPromotionInputs {
primary_observed: true,
local_tip: tip(3, 0),
peers: vec![
(1, PeerTipObservation::Silent { last: None }),
(3, PeerTipObservation::Fresh(tip(3, 0))),
],
..ok_inputs()
};
assert!(auto_promotion_decision(&inputs).is_ok());
}
#[test]
fn refuses_disk_policy_and_unknown_policies() {
let inputs = AutoPromotionInputs {
ack_policy: Some(AckPolicy::Disk),
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs)
.unwrap_err()
.contains("`disk`")
);
let inputs = AutoPromotionInputs {
ack_policy: None,
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs)
.unwrap_err()
.contains("unrecognised")
);
}
#[test]
fn promotion_qualifies_exactly_on_policies_that_acked_on_a_second_node() {
use clap::ValueEnum as _;
for &policy in AckPolicy::value_variants() {
let waits_for_a_second_node = policy.to_policy().clauses().iter().any(|c| c.count >= 2);
let inputs = AutoPromotionInputs {
ack_policy: Some(policy),
..ok_inputs()
};
let decision = auto_promotion_decision(&inputs);
assert_eq!(
decision.is_ok(),
waits_for_a_second_node,
"policy `{policy}` requires a second node: {waits_for_a_second_node}, \
but the promotion decision was {decision:?}"
);
}
}
#[test]
fn refuses_when_epochs_outran_terms() {
for (term, fence_epoch) in [(3, 3), (2, 3)] {
let inputs = AutoPromotionInputs {
term,
fence_epoch,
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs)
.unwrap_err()
.contains("term"),
"term {term} epoch {fence_epoch}"
);
}
let inputs = AutoPromotionInputs {
term: 4,
fence_epoch: 3,
..ok_inputs()
};
assert!(auto_promotion_decision(&inputs).is_ok());
}
#[test]
fn refuses_while_a_live_peer_is_ahead() {
let inputs = AutoPromotionInputs {
local_tip: tip(3, 100),
peers: vec![
(1, PeerTipObservation::Silent { last: None }),
(3, PeerTipObservation::Fresh(tip(3, 101))),
],
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs)
.unwrap_err()
.contains("ahead of ours")
);
}
#[test]
fn epoch_dominates_sequence_in_the_veto() {
let inputs = AutoPromotionInputs {
local_tip: tip(3, 100),
peers: vec![(3, PeerTipObservation::Fresh(tip(2, 900)))],
..ok_inputs()
};
assert!(auto_promotion_decision(&inputs).is_ok());
let inputs = AutoPromotionInputs {
local_tip: tip(3, 100),
peers: vec![(3, PeerTipObservation::Fresh(tip(4, 1)))],
..ok_inputs()
};
assert!(auto_promotion_decision(&inputs).is_err());
}
#[test]
fn refuses_while_a_peer_tip_is_still_unknown() {
let inputs = AutoPromotionInputs {
peers: vec![
(1, PeerTipObservation::Silent { last: None }),
(3, PeerTipObservation::Unknown),
],
..ok_inputs()
};
assert!(
auto_promotion_decision(&inputs)
.unwrap_err()
.contains("unknown")
);
}
#[test]
fn silent_peers_do_not_block_promotion() {
let inputs = AutoPromotionInputs {
local_tip: tip(3, 100),
peers: vec![
(1, PeerTipObservation::Silent { last: None }),
(
3,
PeerTipObservation::Silent {
last: Some(tip(3, 101)),
},
),
],
..ok_inputs()
};
assert!(auto_promotion_decision(&inputs).is_ok());
}
#[test]
fn a_changed_refusal_reason_is_reported_within_the_same_term() {
let status = RaftStatus::new(2);
status
.role
.store(RaftStatus::ROLE_LEADER, Ordering::Relaxed);
status.term.store(7, Ordering::Relaxed);
let control = crate::replication::ReplicaControlPlane::new();
control.tip_ready.store(true, Ordering::Release);
control
.journal_tip
.advance(melin_transport_core::WireSeq::new(100));
control.lineage_ack_policy.observe(
melin_transport_core::WireSeq::new(90),
streamed_change(AckPolicy::Ram),
);
let fence = melin_transport_core::fence::FenceState::new(0);
let policy = crate::policy_control::test_support::replicated(AckPolicy::Ram);
let peer_tips = PeerTips::new();
peer_tips.record(3, tip(0, 101));
let peer_ids = [1u64, 3];
let leader_for = PEER_TIP_GRACE * 2;
let mut last_refused = None;
control.primary_link_up.store(true, Ordering::Release);
let refuse = |last: &mut Option<(u64, &'static str)>, down_for| {
consider_auto_promotion(
&status, &control, &fence, &policy, &peer_tips, &peer_ids, down_for, leader_for,
last,
);
};
refuse(&mut last_refused, Duration::ZERO);
let (term, first) = last_refused.expect("a refusal was recorded");
assert_eq!(term, 7);
assert!(first.contains("link to the primary is up"), "{first}");
control.primary_link_up.store(false, Ordering::Release);
refuse(&mut last_refused, PRIMARY_DOWN_GRACE);
let (term, second) = last_refused.expect("a refusal was recorded");
assert_eq!(term, 7, "same leadership, same term");
assert!(second.contains("ahead of ours"), "{second}");
refuse(&mut last_refused, PRIMARY_DOWN_GRACE);
assert_eq!(last_refused, Some((7, second)));
}
#[test]
fn auto_promotion_judges_by_the_lineage_policy_not_the_nodes_own() {
let status = RaftStatus::new(2);
status
.role
.store(RaftStatus::ROLE_LEADER, Ordering::Relaxed);
status.term.store(7, Ordering::Relaxed);
let control = crate::replication::ReplicaControlPlane::new();
control.tip_ready.store(true, Ordering::Release);
control.primary_observed.store(true, Ordering::Release);
control
.journal_tip
.advance(melin_transport_core::WireSeq::new(100));
let fence = melin_transport_core::fence::FenceState::new(0);
let peer_tips = PeerTips::new();
let consider = |requested: AckPolicy, last: &mut Option<(u64, &'static str)>| {
consider_auto_promotion(
&status,
&control,
&fence,
&crate::policy_control::test_support::replicated(requested),
&peer_tips,
&[],
PRIMARY_DOWN_GRACE,
PEER_TIP_GRACE * 2,
last,
);
};
let observe = |wire_seq, policy| {
control.lineage_ack_policy.observe(
melin_transport_core::WireSeq::new(wire_seq),
streamed_change(policy),
);
};
observe(80, AckPolicy::DiskAndRam);
observe(90, AckPolicy::Disk);
let mut last_refused = None;
consider(AckPolicy::DiskAndRam, &mut last_refused);
let (_, reason) = last_refused.expect("refused");
assert!(reason.contains("`disk`"), "{reason}");
assert!(!control.promote.is_requested());
observe(95, AckPolicy::DiskAndRam);
consider(AckPolicy::Disk, &mut last_refused);
assert!(control.promote.is_requested(), "promotion filed");
}
fn streamed_change(policy: AckPolicy) -> melin_journal::AckPolicyChange {
melin_journal::AckPolicyChange {
policy: policy.to_byte(),
unwitnessed: false,
}
}
#[test]
fn observe_peer_digests_freshness_correctly() {
let tips = PeerTips::new();
let t0 = Instant::now();
assert_eq!(
observe_peer(&tips, 2, Duration::ZERO, t0),
PeerTipObservation::Unknown
);
assert_eq!(
observe_peer(&tips, 2, PEER_TIP_GRACE * 2, t0),
PeerTipObservation::Silent { last: None }
);
tips.record_at(2, tip(3, 7), t0);
assert_eq!(
observe_peer(&tips, 2, Duration::ZERO, t0 + PEER_TIP_GRACE),
PeerTipObservation::Fresh(tip(3, 7))
);
assert_eq!(
observe_peer(&tips, 2, Duration::ZERO, t0 + PEER_TIP_GRACE * 2),
PeerTipObservation::Silent {
last: Some(tip(3, 7))
}
);
}
#[test]
fn challenger_campaigns_only_against_a_fresh_behind_leader() {
let ahead = tip(3, 101);
let behind = tip(3, 100);
assert!(challenger_should_campaign(
true,
false,
false,
PRIMARY_DOWN_GRACE,
ahead,
Some(behind)
));
assert!(!challenger_should_campaign(
true,
false,
false,
PRIMARY_DOWN_GRACE,
ahead,
Some(ahead)
));
assert!(!challenger_should_campaign(
true,
false,
false,
PRIMARY_DOWN_GRACE,
ahead,
None
));
}
#[test]
fn challenger_never_fires_outside_a_failover_window() {
let ahead = tip(3, 101);
let behind = tip(3, 100);
assert!(!challenger_should_campaign(
true,
false,
true,
Duration::ZERO,
ahead,
Some(behind)
));
assert!(!challenger_should_campaign(
true,
false,
false,
PRIMARY_DOWN_GRACE - Duration::from_millis(1),
ahead,
Some(behind)
));
assert!(!challenger_should_campaign(
false,
false,
false,
PRIMARY_DOWN_GRACE,
ahead,
Some(behind)
));
assert!(!challenger_should_campaign(
true,
true,
false,
PRIMARY_DOWN_GRACE,
ahead,
Some(behind)
));
}
#[test]
fn stand_down_only_while_a_recent_peer_tip_is_ahead() {
let local = tip(3, 100);
let reason =
election_should_stand_down(true, false, local, &[(tip(3, 101), Duration::ZERO)])
.expect("stands down");
assert!(reason.contains("ahead of ours"), "{reason}");
assert!(
election_should_stand_down(true, false, local, &[(tip(3, 101), STAND_DOWN_GRACE)])
.is_some()
);
assert!(
election_should_stand_down(
true,
false,
local,
&[(tip(3, 100), Duration::ZERO), (tip(3, 99), Duration::ZERO)],
)
.is_none()
);
assert!(
election_should_stand_down(
true,
false,
local,
&[(tip(3, 200), STAND_DOWN_GRACE + Duration::from_millis(1))],
)
.is_none()
);
assert!(election_should_stand_down(true, false, local, &[]).is_none());
}
#[test]
fn stand_down_is_unconditional_when_fenced_or_tip_unready() {
let local = tip(3, 100);
let fenced = election_should_stand_down(true, true, local, &[]).expect("stands down");
assert!(fenced.contains("fenced"), "{fenced}");
let unready = election_should_stand_down(false, false, local, &[]).expect("stands down");
assert!(unready.contains("recovery"), "{unready}");
let unready =
election_should_stand_down(false, false, local, &[(tip(3, 999), Duration::ZERO)])
.expect("stands down");
assert!(unready.contains("recovery"), "{unready}");
}
#[test]
fn effective_ack_policy_falls_back_and_refuses_garbage() {
assert_eq!(
effective_ack_policy(None, || AckPolicy::DiskAndRam),
Some(AckPolicy::DiskAndRam)
);
assert_eq!(
effective_ack_policy(Some(AckPolicyByte(0x7F)), || unreachable!(
"the lineage holds a change"
)),
None
);
}
}