use std::sync::Arc;
use std::time::Duration;
use maplit::btreeset;
#[allow(unused_imports)]
use pretty_assertions::assert_eq;
#[allow(unused_imports)]
use pretty_assertions::assert_ne;
#[allow(unused_imports)]
use pretty_assertions::assert_str_eq;
use crate::Membership;
use crate::MembershipState;
use crate::Vote;
use crate::batch::Batch;
use crate::engine::Command;
use crate::engine::Engine;
use crate::engine::TargetProgress;
use crate::engine::leader_log_ids::LeaderLogIds;
use crate::engine::testing::UTConfig;
use crate::engine::testing::log_id;
use crate::entry::RaftEntry;
use crate::entry::payload::EntryPayload;
use crate::log_id_range::LogIdRange;
use crate::progress::entry::ProgressEntry;
use crate::progress::inflight_id::InflightId;
use crate::progress::stream_id::StreamId;
use crate::raft_state::IOId;
use crate::raft_state::LogStateReader;
use crate::replication::replicate::Replicate;
use crate::testing::blank_ent;
use crate::type_config::TypeConfigExt;
use crate::type_config::alias::EffectiveMembershipOf;
use crate::type_config::alias::EntryOf;
use crate::utime::Leased;
use crate::vote::raft_vote::RaftVoteExt;
fn committed_leader_id(term: u64, node_id: u64) -> crate::type_config::alias::CommittedLeaderIdOf<UTConfig> {
*log_id(term, node_id, 0).committed_leader_id()
}
fn m01() -> Membership<u64, ()> {
Membership::<u64, ()>::new_with_defaults(vec![btreeset! {0,1}], [])
}
fn m1() -> Membership<u64, ()> {
Membership::<u64, ()>::new_with_defaults(vec![btreeset! {1}], [])
}
fn m1_2() -> Membership<u64, ()> {
Membership::<u64, ()>::new_with_defaults(vec![btreeset! {1}], btreeset! {2})
}
fn m23() -> Membership<u64, ()> {
Membership::<u64, ()>::new_with_defaults(vec![btreeset! {2,3}], btreeset! {1,2,3})
}
fn eng() -> Engine<UTConfig> {
let mut eng = Engine::testing_default(0);
eng.state.enable_validation(false);
eng.config.id = 1;
eng.state.apply_progress_mut().accept(log_id(0, 1, 0));
eng.state.vote = Leased::new(
UTConfig::<()>::now(),
Duration::from_millis(500),
Vote::new_committed(3, 1),
);
eng.state.log_ids.append(log_id(1, 1, 1));
eng.state.log_ids.append(log_id(2, 1, 3));
eng.state.membership_state = MembershipState::new(
Arc::new(EffectiveMembershipOf::<UTConfig>::new(Some(log_id(1, 1, 1)), m01())),
Arc::new(EffectiveMembershipOf::<UTConfig>::new(Some(log_id(2, 1, 3)), m23())),
);
eng.testing_new_leader();
eng.state.server_state = eng.calc_server_state();
eng
}
#[test]
fn test_leader_append_entries_empty() -> anyhow::Result<()> {
let mut eng = eng();
eng.output.take_commands();
let got = eng.try_leader_handler()?.leader_append_entries([]);
assert_eq!(None, got, "empty entries should return None");
assert_eq!(
None,
eng.state.accepted_log_io(),
"no accepted log updated for empty entries"
);
assert_eq!(None, eng.state.log_ids.purged());
assert_eq!(
&[
log_id(1, 1, 1), log_id(2, 1, 3),
],
eng.state.log_ids.key_log_ids()
);
assert_eq!(Some(&log_id(2, 1, 3)), eng.state.last_log_id());
assert_eq!(
MembershipState::new(
Arc::new(EffectiveMembershipOf::<UTConfig>::new(Some(log_id(1, 1, 1)), m01())),
Arc::new(EffectiveMembershipOf::<UTConfig>::new(Some(log_id(2, 1, 3)), m23())),
),
eng.state.membership_state
);
assert_eq!(eng.output.take_commands(), vec![]);
Ok(())
}
#[test]
fn test_leader_append_entries_normal() -> anyhow::Result<()> {
let mut eng = eng();
eng.output.take_commands();
let got = eng.try_leader_handler()?.leader_append_entries([
EntryPayload::Blank,
EntryPayload::Blank,
EntryPayload::Blank,
]);
assert_eq!(
Some(LeaderLogIds::new(committed_leader_id(3, 1), 4, 6)),
got,
"should return log ids for 3 entries"
);
assert_eq!(
Some(&IOId::new_log_io(
Vote::new(3, 1).into_committed(),
Some(log_id(3, 1, 6))
)),
eng.state.accepted_log_io()
);
assert_eq!(None, eng.state.log_ids.purged());
assert_eq!(
&[
log_id(1, 1, 1), log_id(2, 1, 3),
log_id(3, 1, 6),
],
eng.state.log_ids.key_log_ids()
);
assert_eq!(Some(&log_id(3, 1, 6)), eng.state.last_log_id());
assert_eq!(
MembershipState::new(
Arc::new(EffectiveMembershipOf::<UTConfig>::new(Some(log_id(1, 1, 1)), m01())),
Arc::new(EffectiveMembershipOf::<UTConfig>::new(Some(log_id(2, 1, 3)), m23())),
),
eng.state.membership_state
);
assert_eq!(
vec![
Command::AppendEntries {
committed_vote: Vote::new(3, 1).into_committed(),
entries: Batch::of([
blank_ent::<UTConfig>(3, 1, 4), blank_ent::<UTConfig>(3, 1, 5),
blank_ent::<UTConfig>(3, 1, 6),
])
},
Command::Replicate {
target: 2,
req: Replicate::new_logs(LogIdRange::new(None, Some(log_id(3, 1, 6))), InflightId::new(1)),
},
Command::Replicate {
target: 3,
req: Replicate::new_logs(LogIdRange::new(None, Some(log_id(3, 1, 6))), InflightId::new(2)),
},
],
eng.output.take_commands()
);
Ok(())
}
#[test]
fn test_leader_append_entries_single_node_leader() -> anyhow::Result<()> {
let mut eng = eng();
eng.state.membership_state.set_effective(Arc::new(EffectiveMembershipOf::<UTConfig>::new(
Some(log_id(2, 1, 3)),
m1(),
)));
eng.testing_new_leader();
eng.output.clear_commands();
let got = eng.try_leader_handler()?.leader_append_entries([
EntryPayload::Blank,
EntryPayload::Blank,
EntryPayload::Blank,
]);
assert_eq!(
Some(LeaderLogIds::new(committed_leader_id(3, 1), 4, 6)),
got,
"should return log ids for 3 entries"
);
assert_eq!(
Some(&IOId::new_log_io(
Vote::new(3, 1).into_committed(),
Some(log_id(3, 1, 6))
)),
eng.state.accepted_log_io()
);
assert_eq!(None, eng.state.log_ids.purged());
assert_eq!(
&[
log_id(1, 1, 1), log_id(2, 1, 3),
log_id(3, 1, 6),
],
eng.state.log_ids.key_log_ids()
);
assert_eq!(Some(&log_id(3, 1, 6)), eng.state.last_log_id());
assert_eq!(
MembershipState::new(
Arc::new(EffectiveMembershipOf::<UTConfig>::new(Some(log_id(1, 1, 1)), m01())),
Arc::new(EffectiveMembershipOf::<UTConfig>::new(Some(log_id(2, 1, 3)), m1())),
),
eng.state.membership_state
);
assert_eq!(Some(&log_id(0, 1, 0)), eng.state.committed());
assert_eq!(
vec![Command::AppendEntries {
committed_vote: Vote::new(3, 1).into_committed(),
entries: Batch::of([
blank_ent::<UTConfig>(3, 1, 4), blank_ent::<UTConfig>(3, 1, 5),
blank_ent::<UTConfig>(3, 1, 6),
])
},],
eng.output.take_commands()
);
Ok(())
}
#[test]
fn test_leader_append_entries_with_membership_log() -> anyhow::Result<()> {
let mut eng = eng();
eng.state.membership_state.set_effective(Arc::new(EffectiveMembershipOf::<UTConfig>::new(
Some(log_id(2, 1, 3)),
m1(),
)));
eng.testing_new_leader();
eng.state.server_state = eng.calc_server_state();
eng.output.clear_commands();
let got = eng.try_leader_handler()?.leader_append_entries([
EntryPayload::Blank,
EntryPayload::Membership(m1_2()),
EntryPayload::Blank,
]);
assert_eq!(
Some(LeaderLogIds::new(committed_leader_id(3, 1), 4, 6)),
got,
"should return log ids for 3 entries"
);
assert_eq!(
Some(&IOId::new_log_io(
Vote::new(3, 1).into_committed(),
Some(log_id(3, 1, 6))
)),
eng.state.accepted_log_io()
);
assert_eq!(None, eng.state.log_ids.purged());
assert_eq!(
&[
log_id(1, 1, 1), log_id(2, 1, 3),
log_id(3, 1, 6),
],
eng.state.log_ids.key_log_ids()
);
assert_eq!(Some(&log_id(3, 1, 6)), eng.state.last_log_id());
assert_eq!(
MembershipState::new(
Arc::new(EffectiveMembershipOf::<UTConfig>::new(Some(log_id(2, 1, 3)), m1())),
Arc::new(EffectiveMembershipOf::<UTConfig>::new(Some(log_id(3, 1, 5)), m1_2())),
),
eng.state.membership_state
);
assert_eq!(Some(&log_id(0, 1, 0)), eng.state.committed());
assert_eq!(
vec![
Command::AppendEntries {
committed_vote: Vote::new(3, 1).into_committed(),
entries: Batch::of([
blank_ent::<UTConfig>(3, 1, 4), EntryOf::<UTConfig>::new_membership(log_id(3, 1, 5), m1_2()),
blank_ent::<UTConfig>(3, 1, 6),
])
},
Command::RebuildReplicationStreams {
leader_vote: Vote::new(3, 1).into_committed(),
targets: vec![TargetProgress {
target: 2,
target_node: (),
progress: ProgressEntry::empty(StreamId::new(8), 7),
}],
close_old_streams: false,
},
Command::Replicate {
target: 2,
req: Replicate::new_logs(LogIdRange::new(None, Some(log_id(3, 1, 6))), InflightId::new(1))
},
],
eng.output.take_commands()
);
Ok(())
}