1pub mod elector;
690pub mod scheme;
691pub mod types;
692
693use crate::types::{TermLength, View, ViewDelta};
694
695#[derive(Clone, Copy)]
709#[cfg_attr(target_arch = "wasm32", allow(dead_code))]
710pub(crate) struct Lookahead {
711 pub term_length: TermLength,
713 pub optimistic_views: ViewDelta,
717}
718
719#[cfg_attr(target_arch = "wasm32", allow(dead_code))]
720impl Lookahead {
721 pub(crate) const fn new(terms: &elector::Terms) -> Self {
723 Self {
724 term_length: terms.length(),
725 optimistic_views: terms.optimistic_views(),
726 }
727 }
728
729 pub fn in_admission_window(&self, current: View, pending: View) -> bool {
733 current < pending && pending <= self.admission_limit(current)
734 }
735
736 pub fn admits(&self, current: View, pending: View) -> bool {
749 if current.admits(pending, self.term_length) || self.in_admission_window(current, pending) {
750 return true;
751 }
752 let next_start = current.next_term_start(self.term_length);
753 let lane = self
754 .optimistic_views
755 .get()
756 .max(1)
757 .min(self.term_length.get())
758 .saturating_add(1);
759 pending > next_start && pending <= next_start.saturating_add(ViewDelta::new(lane))
760 }
761
762 pub fn admits_outbound(&self, current: View, pending: View) -> bool {
767 current.admits(pending, self.term_length)
768 || (pending <= current.saturating_add(self.optimistic_views)
769 && self.in_admission_window(current, pending))
770 }
771
772 pub fn admission_limit(&self, current: View) -> View {
776 let window = self
777 .optimistic_views
778 .get()
779 .saturating_mul(2)
780 .saturating_add(1);
781 current
782 .term_end(self.term_length)
783 .min(current.saturating_add(ViewDelta::new(window)))
784 }
785
786 pub const fn issuance_floor(&self, view: View) -> Option<View> {
799 if self.optimistic_views.is_zero() || view.is_term_start(self.term_length) {
800 return None;
801 }
802 Some(
803 view.saturating_sub(self.optimistic_views)
804 .saturating_sub(ViewDelta::new(1)),
805 )
806 }
807}
808
809cfg_if::cfg_if! {
810 if #[cfg(not(target_arch = "wasm32"))] {
811 use crate::types::Round;
812 use commonware_cryptography::PublicKey;
813 use commonware_p2p::Recipients;
814
815 mod actors;
816 pub mod config;
817 pub use config::{Config, Floor, ForwardPolicy, SkipBudget, SkipPolicy};
818 mod engine;
819 pub use engine::Engine;
820 mod limits;
821 pub use limits::max_payload_size;
822 mod metrics;
823
824 #[derive(Clone, Copy)]
827 pub(crate) struct Viewport {
828 pub finalized: View,
830 pub current: View,
832 pub view_retention: ViewDelta,
834 pub lookahead: Lookahead,
836 }
837
838 impl Viewport {
839 pub const fn floor(&self) -> View {
841 self.finalized.saturating_sub(self.view_retention)
842 }
843
844 pub const fn retains(&self, view: View) -> bool {
849 !view.is_zero() && view.get() >= self.floor().get()
850 }
851
852 pub fn admits_vote(&self, pending: View) -> bool {
856 self.retains(pending) && self.lookahead.admits(self.current, pending)
857 }
858
859 pub const fn admits_certificate(&self, pending: View) -> bool {
864 self.retains(pending)
865 }
866 }
867
868 pub enum Plan<P: PublicKey> {
870 Prepare {
878 round: Round,
880 },
881 Propose {
892 round: Round,
894 },
895 Forward {
902 round: Round,
904 recipients: Recipients<P>,
906 },
907 }
908 }
909}
910
911#[cfg(any(test, feature = "mocks"))]
912pub mod mocks;
913
914#[cfg(test)]
916pub(crate) fn quorum(n: u32) -> u32 {
917 use commonware_utils::{Faults, N3f1};
918
919 N3f1::quorum(n)
920}
921
922#[cfg(test)]
923mod tests {
924 use super::*;
925 use crate::{
926 Handoff, Monitor, Viewable,
927 simplex::{
928 elector::{
929 Config as _, Elector, Random, RandomVersion, RoundRobin, RoundRobinElector,
930 Scheduled,
931 },
932 mocks::{
933 scheme as scheme_mocks,
934 twins::{self, Elector as TwinsElector},
935 wrapped,
936 },
937 scheme::{
938 Scheme, bls12381_multisig,
939 bls12381_threshold::{
940 standard as bls12381_threshold_std,
941 vrf::{self as bls12381_threshold_vrf, Seedable},
942 },
943 ed25519, secp256r1,
944 },
945 types::{
946 Certificate, Finalization as TFinalization, Finalize as TFinalize,
947 Notarization as TNotarization, Notarize as TNotarize,
948 Nullification as TNullification, Nullify as TNullify, Proposal, Vote,
949 },
950 },
951 types::{Epoch, Participant, Round, TermLength, View, ViewDelta},
952 };
953 use commonware_codec::{Decode, DecodeExt, Encode};
954 use commonware_cryptography::{
955 Hasher as _, Sha256, Signer as _,
956 bls12381::primitives::variant::{MinPk, MinSig, Variant},
957 certificate::mocks::Fixture,
958 ed25519::{PrivateKey, PublicKey},
959 sha256::{Digest as Sha256Digest, Digest as D},
960 };
961 use commonware_macros::{select, test_group, test_traced};
962 use commonware_p2p::{
963 Manager as _, Recipients, Sender as _, TrackedPeers, max_message_size,
964 simulated::{Config, Link, Network, Oracle, Receiver, Sender, SplitOrigin},
965 utils::mocks::inert_channel,
966 };
967 use commonware_parallel::{Sequential, Strategy};
968 use commonware_runtime::{
969 Clock, IoBuf, Metrics as _, Quota, Runner, Spawner, Strategizer as _, Supervisor as _,
970 buffer::paged::CacheRef, deterministic, telemetry::metrics::count_running_tasks,
971 };
972 use commonware_utils::{
973 Faults, N3f1, NZDuration, NZU16, NZU32, NZUsize, TestRng, Widen,
974 channel::{fallible::OneshotExt as _, oneshot},
975 non_empty,
976 ordered::Set,
977 probability,
978 sync::Mutex,
979 test_rng,
980 };
981 use engine::Engine;
982 use futures::future::join_all;
983 use rand::{RngExt as _, SeedableRng, rngs::StdRng};
984 use rand_core::CryptoRng;
985 use std::{
986 collections::{BTreeMap, HashMap, HashSet},
987 num::{NonZeroU16, NonZeroUsize},
988 sync::{
989 Arc,
990 atomic::{AtomicUsize, Ordering},
991 },
992 time::Duration,
993 };
994 use tracing::{debug, info, warn};
995 use types::Activity;
996
997 macro_rules! for_each_fixture {
1000 ($cb:ident!($($args:tt)*)) => {
1001 $cb!($($args)*, bls12381_threshold_vrf_min_pk, Random, bls12381_threshold_vrf::fixture::<MinPk, _>, Random::new(RandomVersion::V1));
1002 $cb!($($args)*, bls12381_threshold_vrf_min_sig, Random, bls12381_threshold_vrf::fixture::<MinSig, _>, Random::new(RandomVersion::V1));
1003 $cb!($($args)*, bls12381_threshold_std_min_pk, RoundRobin, bls12381_threshold_std::fixture::<MinPk, _>, RoundRobin::default());
1004 $cb!($($args)*, bls12381_threshold_std_min_sig, RoundRobin, bls12381_threshold_std::fixture::<MinSig, _>, RoundRobin::default());
1005 $cb!($($args)*, bls12381_multisig_min_pk, RoundRobin, bls12381_multisig::fixture::<MinPk, _>, RoundRobin::default());
1006 $cb!($($args)*, bls12381_multisig_min_sig, RoundRobin, bls12381_multisig::fixture::<MinSig, _>, RoundRobin::default());
1007 $cb!($($args)*, ed25519, RoundRobin, ed25519::fixture, RoundRobin::default());
1008 $cb!($($args)*, secp256r1, RoundRobin, secp256r1::fixture, RoundRobin::default());
1009 };
1010 }
1011
1012 macro_rules! test_for_all_fixtures {
1024 ($callee:ident) => {
1025 for_each_fixture!(test_for_all_fixtures!(@emit [test_traced] $callee [] []));
1026 };
1027 ($callee:ident, level = $level:literal) => {
1028 for_each_fixture!(test_for_all_fixtures!(@emit [test_traced($level)] $callee [] []));
1029 };
1030 ($callee:ident, seeds = $n:expr) => {
1031 for_each_fixture!(test_for_all_fixtures!(@seeded $n, $callee));
1032 };
1033 ($callee:ident, $arg:expr, level = $level:literal) => {
1034 for_each_fixture!(test_for_all_fixtures!(@emit [test_traced($level)] $callee [, _] [, $arg]));
1035 };
1036 ($callee:ident, $arg:expr) => {
1037 for_each_fixture!(test_for_all_fixtures!(@emit [test_traced] $callee [, _] [, $arg]));
1038 };
1039 (@emit [$traced:meta] $callee:ident [$($generics:tt)*] [$($args:tt)*], $suffix:ident, $elector:ty, $fixture:expr, $elector_config:expr) => {
1040 paste::paste! {
1041 #[test_group("slow")]
1042 #[$traced]
1043 fn [<test_ $callee _ $suffix>]() {
1044 $callee::<_, _, $elector $($generics)*>($fixture, $elector_config $($args)*);
1045 }
1046 }
1047 };
1048 (@seeded $n:expr, $callee:ident, $suffix:ident, $elector:ty, $fixture:expr, $elector_config:expr) => {
1049 paste::paste! {
1050 #[test_group("slow")]
1051 #[test_traced]
1052 fn [<test_ $callee _ $suffix>]() {
1053 for seed in 0..$n {
1054 $callee::<_, _, $elector>(seed, $fixture, $elector_config);
1055 }
1056 }
1057 }
1058 };
1059 }
1060
1061 const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
1062 const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);
1063 const TEST_QUOTA: Quota = Quota::per_second(NZU32!(1_024));
1064
1065 type TestChannel = (
1066 Sender<PublicKey, deterministic::Context>,
1067 Receiver<PublicKey>,
1068 );
1069 type TestRegistration = (TestChannel, TestChannel, TestChannel);
1070 type TestRegistrations = HashMap<PublicKey, TestRegistration>;
1071
1072 fn test_lookahead(optimistic_views: u64) -> Lookahead {
1074 Lookahead {
1075 term_length: TermLength::new(NZU32!(5)),
1076 optimistic_views: ViewDelta::new(optimistic_views),
1077 }
1078 }
1079
1080 #[test]
1081 fn test_lookahead_admits() {
1082 let current = View::new(6);
1083
1084 assert!(test_lookahead(0).admits(current, current.next()));
1086
1087 let next_term_start = current.next_term_start(test_lookahead(0).term_length);
1089 assert!(test_lookahead(0).admits(current, next_term_start));
1090
1091 assert!(test_lookahead(1).admits(current, View::new(9)));
1095 assert!(!test_lookahead(1).admits(current, View::new(10)));
1096
1097 assert!(!test_lookahead(10).admits(current, View::new(100)));
1099
1100 assert!(test_lookahead(1).admits_outbound(current, View::new(7)));
1102 assert!(!test_lookahead(1).admits_outbound(current, View::new(8)));
1103 }
1104
1105 #[test]
1106 fn test_large_lookahead_admission_is_bounded_by_term() {
1107 let lookahead = test_lookahead(100);
1108 let current = View::new(9);
1109 let next_term_start = current.next_term_start(lookahead.term_length);
1110
1111 assert!(lookahead.admits(current, View::new(10)));
1113
1114 assert!(lookahead.admits(current, next_term_start));
1117 assert!(lookahead.admits(current, View::new(17)));
1118 assert!(!lookahead.admits(current, View::new(18)));
1119 }
1120
1121 #[test]
1122 fn test_admission_limit() {
1123 let current = View::new(6);
1125
1126 assert_eq!(test_lookahead(0).admission_limit(current), View::new(7));
1128
1129 assert_eq!(test_lookahead(1).admission_limit(current), View::new(9));
1131
1132 assert_eq!(test_lookahead(100).admission_limit(current), View::new(10));
1134
1135 assert_eq!(
1137 test_lookahead(3).admission_limit(View::new(10)),
1138 View::new(10)
1139 );
1140 }
1141
1142 #[test]
1143 fn test_issuance_floor() {
1144 assert!(test_lookahead(0).issuance_floor(View::new(7)).is_none());
1146
1147 assert!(test_lookahead(100).issuance_floor(View::new(11)).is_none());
1149
1150 assert_eq!(
1153 test_lookahead(2).issuance_floor(View::new(9)),
1154 Some(View::new(6))
1155 );
1156
1157 assert_eq!(
1159 test_lookahead(2).issuance_floor(View::new(2)),
1160 Some(View::zero())
1161 );
1162 }
1163
1164 async fn register_validator(
1166 oracle: &mut Oracle<PublicKey, deterministic::Context>,
1167 validator: PublicKey,
1168 ) -> TestRegistration {
1169 let control = oracle.control(validator.clone());
1170 let (vote_sender, vote_receiver) = control.register(0, TEST_QUOTA).await.unwrap();
1171 let (certificate_sender, certificate_receiver) =
1172 control.register(1, TEST_QUOTA).await.unwrap();
1173 let (resolver_sender, resolver_receiver) = control.register(2, TEST_QUOTA).await.unwrap();
1174 (
1175 (vote_sender, vote_receiver),
1176 (certificate_sender, certificate_receiver),
1177 (resolver_sender, resolver_receiver),
1178 )
1179 }
1180
1181 async fn register_validators(
1183 oracle: &mut Oracle<PublicKey, deterministic::Context>,
1184 validators: &[PublicKey],
1185 ) -> TestRegistrations {
1186 let mut registrations = HashMap::new();
1187 for validator in validators.iter() {
1188 let registration = register_validator(oracle, validator.clone()).await;
1189 registrations.insert(validator.clone(), registration);
1190 }
1191 registrations
1192 }
1193
1194 async fn start_test_network_with_peers<S, I>(
1197 context: deterministic::Context,
1198 peers: I,
1199 disconnect_on_block: bool,
1200 ) -> Oracle<PublicKey, deterministic::Context>
1201 where
1202 S: Scheme<Sha256Digest>,
1203 I: IntoIterator<Item = PublicKey>,
1204 {
1205 let peers: Vec<_> = peers.into_iter().collect();
1206 let (network, oracle) = Network::new_with_peers(
1207 context.child("network"),
1208 Config {
1209 max_size: max_message_size(&[max_payload_size::<S, Sha256Digest>(
1210 u32::try_from(peers.len()).unwrap(),
1211 )]),
1212 max_peers_per_set: NZUsize!(peers.len() + 1),
1214 disconnect_on_block,
1215 tracked_peer_sets: NZUsize!(1),
1216 },
1217 peers,
1218 )
1219 .await;
1220 network.start();
1221 oracle
1222 }
1223
1224 async fn start_test_network_with_split_peers<S, I, J>(
1227 context: deterministic::Context,
1228 primary: I,
1229 secondary: J,
1230 disconnect_on_block: bool,
1231 ) -> Oracle<PublicKey, deterministic::Context>
1232 where
1233 S: Scheme<Sha256Digest>,
1234 I: IntoIterator<Item = PublicKey>,
1235 J: IntoIterator<Item = PublicKey>,
1236 {
1237 let primary: Vec<_> = primary.into_iter().collect();
1238 let secondary: Vec<_> = secondary.into_iter().collect();
1239 let (network, oracle) = Network::new_with_split_peers(
1240 context.child("network"),
1241 Config {
1242 max_size: max_message_size(&[max_payload_size::<S, Sha256Digest>(
1243 u32::try_from(primary.len()).unwrap(),
1244 )]),
1245 max_peers_per_set: NZUsize!(primary.len() + secondary.len()),
1246 disconnect_on_block,
1247 tracked_peer_sets: NZUsize!(1),
1248 },
1249 primary,
1250 secondary,
1251 )
1252 .await;
1253 network.start();
1254 oracle
1255 }
1256
1257 enum Action {
1259 Link(Link),
1260 Update(Link), Unlink,
1262 }
1263
1264 async fn link_validators(
1270 oracle: &mut Oracle<PublicKey, deterministic::Context>,
1271 validators: &[PublicKey],
1272 action: Action,
1273 restrict_to: Option<fn(usize, usize, usize) -> bool>,
1274 ) {
1275 for (i1, v1) in validators.iter().enumerate() {
1276 for (i2, v2) in validators.iter().enumerate() {
1277 if v2 == v1 {
1279 continue;
1280 }
1281
1282 if let Some(f) = restrict_to
1284 && !f(validators.len(), i1, i2)
1285 {
1286 continue;
1287 }
1288
1289 match action {
1291 Action::Update(_) | Action::Unlink => {
1292 oracle.remove_link(v1.clone(), v2.clone()).await.unwrap();
1293 }
1294 _ => {}
1295 }
1296
1297 match action {
1299 Action::Link(ref link) | Action::Update(ref link) => {
1300 oracle
1301 .add_link(v1.clone(), v2.clone(), link.clone())
1302 .await
1303 .unwrap();
1304 }
1305 _ => {}
1306 }
1307 }
1308 }
1309 }
1310
1311 fn count_nonzero_metric_lines(encoded: &str, patterns: &[&str]) -> u32 {
1313 encoded
1314 .lines()
1315 .filter(|line| patterns.iter().all(|p| line.contains(p)))
1316 .filter(|line| {
1317 line.split_whitespace()
1318 .last()
1319 .and_then(|s| s.parse::<u64>().ok())
1320 .is_some_and(|n| n > 0)
1321 })
1322 .count() as u32
1323 }
1324
1325 fn all_online<S, F, L, T>(
1326 mut fixture: F,
1327 elector: L,
1328 strategy: impl FnOnce(&mut deterministic::Context) -> T + Send + 'static,
1329 ) where
1330 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
1331 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
1332 L: elector::Config<S>,
1333 T: Strategy,
1334 {
1335 let n = 5;
1337 let quorum = quorum(n) as usize;
1338 let required_containers = View::new(100);
1339 let view_retention = ViewDelta::new(10);
1340 let skip_timeout = Duration::from_secs(12);
1341 let namespace = b"consensus".to_vec();
1342 let executor = deterministic::Runner::timed(Duration::from_secs(300));
1343 executor.start(|mut context| async move {
1344 let Fixture {
1346 participants,
1347 schemes,
1348 ..
1349 } = fixture(&mut context, &namespace, n);
1350 let strategy = strategy(&mut context);
1351 let mut oracle = start_test_network_with_peers::<S, _>(
1352 context.child("network"),
1353 participants.clone(),
1354 true,
1355 )
1356 .await;
1357 let mut registrations = register_validators(&mut oracle, &participants).await;
1358
1359 let link = Link {
1365 latency: Duration::from_millis(200),
1366 jitter: Duration::from_millis(1),
1367 success_rate: probability!(1.0),
1368 };
1369 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
1370
1371 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
1373 let mut reporters = Vec::new();
1374 let mut engine_handlers = Vec::new();
1375 for (idx, validator) in participants.iter().enumerate() {
1376 let context = context
1378 .child("validator")
1379 .with_attribute("public_key", validator);
1380
1381 let reporter_config = mocks::reporter::Config {
1383 participants: participants.clone().try_into().unwrap(),
1384 scheme: schemes[idx].clone(),
1385 elector: elector.clone(),
1386 };
1387 let reporter =
1388 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
1389 reporters.push(reporter.clone());
1390 let application_cfg = mocks::application::Config::<Sha256, _> {
1391 relay: relay.clone(),
1392 me: validator.clone(),
1393 propose_latency: (10.0, 5.0),
1394 verify_latency: (10.0, 5.0),
1395 certify_latency: (10.0, 5.0),
1396 should_certify: mocks::application::Certifier::Always,
1397 };
1398 let (actor, application) = mocks::application::Application::new(
1399 context.child("application"),
1400 application_cfg,
1401 );
1402 actor.start();
1403 let blocker = oracle.control(validator.clone());
1404 let cfg = config::Config {
1405 scheme: schemes[idx].clone(),
1406 elector: elector.clone(),
1407 blocker,
1408 automaton: application.clone(),
1409 relay: application.clone(),
1410 reporter: reporter.clone(),
1411 strategy: strategy.clone(),
1412 partition: validator.to_string(),
1413 mailbox_size: NZUsize!(1024),
1414 epoch: Epoch::new(333),
1415 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
1416 Epoch::new(333),
1417 )),
1418 leader_timeout: Duration::from_secs(2),
1419 certification_timeout: Duration::from_secs(3),
1420 timeout_retry: Duration::from_secs(10),
1421 fetch_timeout: NZDuration!(Duration::from_secs(1)),
1422 fetch_retry_delay: Duration::from_millis(100),
1423 view_retention,
1424 skip: SkipPolicy::Enabled {
1425 timeout: skip_timeout,
1426 budget: SkipBudget::Participants,
1427 },
1428 replay_buffer: NZUsize!(1024 * 1024),
1429 write_buffer: NZUsize!(1024 * 1024),
1430 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
1431 forward: ForwardPolicy::Disabled,
1432 track_historical_votes: false,
1433 };
1434 let engine = Engine::new(context.child("engine"), cfg);
1435
1436 let (pending, recovered, resolver) = registrations
1438 .remove(validator)
1439 .expect("validator should be registered");
1440 engine_handlers.push(engine.start(pending, recovered, resolver));
1441 }
1442
1443 let mut finalizers = Vec::new();
1445 for reporter in reporters.iter_mut() {
1446 let (mut latest, mut monitor) = reporter.subscribe().await;
1447 finalizers.push(context.child("finalizer").spawn(move |_| async move {
1448 while latest < required_containers {
1449 latest = monitor.recv().await.expect("event missing");
1450 }
1451 }));
1452 }
1453 join_all(finalizers).await;
1454
1455 let latest_complete = required_containers.saturating_sub(view_retention);
1457 for reporter in reporters.iter() {
1458 reporter.assert_no_faults();
1460
1461 reporter.assert_no_invalid();
1463
1464 {
1466 let certified = reporter.certified.lock();
1467 for view in View::range(View::new(1), latest_complete) {
1468 if !certified.contains(&view) {
1470 panic!("view: {view}");
1471 }
1472 }
1473 }
1474
1475 let mut notarized = HashMap::new();
1477 let mut finalized = HashMap::new();
1478 {
1479 let notarizes = reporter.notarizes.lock();
1480 for view in View::range(View::new(1), latest_complete) {
1481 let Some(payloads) = notarizes.get(&view) else {
1483 continue;
1484 };
1485 if payloads.len() > 1 {
1486 panic!("view: {view}");
1487 }
1488 let (digest, notarizers) = payloads.iter().next().unwrap();
1489 notarized.insert(view, *digest);
1490
1491 if notarizers.len() < quorum {
1492 panic!("view: {view}");
1495 }
1496 }
1497 }
1498 {
1499 let notarizations = reporter.notarizations.lock();
1500 for view in View::range(View::new(1), latest_complete) {
1501 let Some(notarization) = notarizations.get(&view) else {
1503 continue;
1504 };
1505 let Some(digest) = notarized.get(&view) else {
1506 continue;
1507 };
1508 assert_eq!(¬arization.proposal.payload, digest);
1509 }
1510 }
1511 {
1512 let finalizes = reporter.finalizes.lock();
1513 for view in View::range(View::new(1), latest_complete) {
1514 let Some(payloads) = finalizes.get(&view) else {
1516 continue;
1517 };
1518 if payloads.len() > 1 {
1519 panic!("view: {view}");
1520 }
1521 let (digest, finalizers) = payloads.iter().next().unwrap();
1522 finalized.insert(view, *digest);
1523
1524 if view > latest_complete {
1526 continue;
1527 }
1528
1529 if finalizers.len() < quorum {
1531 panic!("view: {view}");
1534 }
1535
1536 let nullifies = reporter.nullifies.lock();
1538 let Some(nullifies) = nullifies.get(&view) else {
1539 continue;
1540 };
1541 for finalizers in payloads.values() {
1542 for finalizer in finalizers.iter() {
1543 if nullifies.contains(finalizer) {
1544 panic!("should not nullify and finalize at same view");
1545 }
1546 }
1547 }
1548 }
1549 }
1550 {
1551 let finalizations = reporter.finalizations.lock();
1552 for view in View::range(View::new(1), latest_complete) {
1553 let Some((finalization, _)) = finalizations.get(&view) else {
1555 continue;
1556 };
1557 let Some(digest) = finalized.get(&view) else {
1558 continue;
1559 };
1560 assert_eq!(&finalization.proposal.payload, digest);
1561 }
1562 }
1563 }
1564
1565 let blocked = oracle.blocked().await.unwrap();
1567 assert!(blocked.is_empty());
1568 });
1569 }
1570
1571 test_for_all_fixtures!(all_online, |_| Sequential);
1572
1573 #[test_group("slow")]
1574 #[test_traced]
1575 fn test_all_online_rayon_bls12381_threshold_vrf_min_pk() {
1576 all_online::<_, _, Random, _>(
1577 bls12381_threshold_vrf::fixture::<MinPk, _>,
1578 Random::new(RandomVersion::V1),
1579 |context| context.strategy(NZUsize!(2)),
1580 );
1581 }
1582
1583 fn non_genesis_floor_joiner_catches_tip<S, F, L>(fixture: F, elector: L)
1584 where
1585 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
1586 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
1587 L: elector::Config<S>,
1588 {
1589 non_genesis_floor_joiner_catches_tip_with_term::<S, F, L>(elector, fixture);
1590 }
1591
1592 fn non_genesis_floor_joiner_catches_tip_with_term<S, F, L>(elector: L, mut fixture: F)
1593 where
1594 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
1595 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
1596 L: elector::Config<S>,
1597 {
1598 let n = 5;
1601 let active_count = quorum(n) as usize;
1602 let initial_tip_target = View::new(15);
1603 let view_retention = ViewDelta::new(10);
1604 let skip_timeout = Duration::from_secs(5);
1605 let timeout_retry = Duration::from_secs(1);
1606 let namespace = b"consensus".to_vec();
1607 let executor = deterministic::Runner::timed(Duration::from_secs(300));
1608 executor.start(|mut context| async move {
1609 let Fixture {
1610 participants,
1611 schemes,
1612 ..
1613 } = fixture(&mut context, &namespace, n);
1614 let mut oracle = start_test_network_with_peers::<S, _>(
1615 context.child("network"),
1616 participants.clone(),
1617 true,
1618 )
1619 .await;
1620
1621 let active = &participants[..active_count];
1622 let joiner_idx = active_count;
1623 let joiner = participants[joiner_idx].clone();
1624
1625 let link = Link {
1626 latency: Duration::from_millis(10),
1627 jitter: Duration::from_millis(1),
1628 success_rate: probability!(1.0),
1629 };
1630 link_validators(&mut oracle, active, Action::Link(link.clone()), None).await;
1631
1632 let term_length = elector
1633 .clone()
1634 .build(schemes[0].participants())
1635 .terms()
1636 .length();
1637 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
1638 let mut reporters = Vec::new();
1639 let mut engine_handlers = Vec::new();
1640
1641 for (idx, validator) in active.iter().enumerate() {
1642 let validator_context = context
1643 .child("validator")
1644 .with_attribute("public_key", validator);
1645
1646 let reporter_config = mocks::reporter::Config {
1647 participants: participants.clone().try_into().unwrap(),
1648 scheme: schemes[idx].clone(),
1649 elector: elector.clone(),
1650 };
1651 let reporter = mocks::reporter::Reporter::new(
1652 validator_context.child("reporter"),
1653 reporter_config,
1654 );
1655 reporters.push(reporter.clone());
1656
1657 let application_cfg = mocks::application::Config::<Sha256, _> {
1658 relay: relay.clone(),
1659 me: validator.clone(),
1660 propose_latency: (10.0, 5.0),
1661 verify_latency: (10.0, 5.0),
1662 certify_latency: (10.0, 5.0),
1663 should_certify: mocks::application::Certifier::Always,
1664 };
1665 let (actor, application) = mocks::application::Application::new(
1666 validator_context.child("application"),
1667 application_cfg,
1668 );
1669 actor.start();
1670
1671 let cfg = config::Config {
1672 scheme: schemes[idx].clone(),
1673 elector: elector.clone(),
1674 blocker: oracle.control(validator.clone()),
1675 automaton: application.clone(),
1676 relay: application.clone(),
1677 reporter: reporter.clone(),
1678 strategy: Sequential,
1679 partition: format!("joiner_catches_tip_{validator}"),
1680 mailbox_size: NZUsize!(1024),
1681 epoch: Epoch::new(333),
1682 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
1683 Epoch::new(333),
1684 )),
1685 leader_timeout: Duration::from_secs(1),
1686 certification_timeout: Duration::from_secs(2),
1687 timeout_retry,
1688 fetch_timeout: NZDuration!(Duration::from_secs(1)),
1689 fetch_retry_delay: Duration::from_millis(100),
1690 view_retention,
1691 skip: SkipPolicy::Enabled {
1692 timeout: skip_timeout,
1693 budget: SkipBudget::Participants,
1694 },
1695 replay_buffer: NZUsize!(1024 * 1024),
1696 write_buffer: NZUsize!(1024 * 1024),
1697 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
1698 forward: ForwardPolicy::Disabled,
1699 track_historical_votes: false,
1700 };
1701 let engine = Engine::new(validator_context.child("engine"), cfg);
1702 let (pending, recovered, resolver) =
1703 register_validator(&mut oracle, validator.clone()).await;
1704 engine_handlers.push(engine.start(pending, recovered, resolver));
1705 }
1706
1707 let mut finalizers = Vec::new();
1708 for reporter in reporters.iter_mut() {
1709 let (mut latest, mut monitor) = reporter.subscribe().await;
1710 finalizers.push(
1711 context
1712 .child("initial_finalizer")
1713 .spawn(move |_| async move {
1714 while latest < initial_tip_target {
1715 latest = monitor.recv().await.expect("event missing");
1716 }
1717 latest
1718 }),
1719 );
1720 }
1721 let tip_at_join = join_all(finalizers)
1722 .await
1723 .into_iter()
1724 .map(|result| result.expect("initial finalizer failed"))
1725 .min()
1726 .expect("initial validators missing");
1727
1728 let (floor_view, floor_finalization) = {
1732 let finalizations = reporters[0].finalizations.lock();
1733 let mut eligible: Vec<_> = finalizations
1734 .iter()
1735 .filter(|(view, _)| **view > View::zero() && **view < tip_at_join)
1736 .collect();
1737 eligible.sort_by_key(|(view, _)| view.get());
1738 eligible
1739 .iter()
1740 .find(|(view, _)| !view.is_term_start(term_length))
1741 .or_else(|| eligible.first())
1742 .map(|(view, (finalization, _))| (**view, finalization.clone()))
1743 .expect("non-genesis floor finalization missing")
1744 };
1745 assert!(floor_view > View::zero());
1746 assert!(floor_view < tip_at_join);
1747 if term_length.get() > 1 {
1748 assert!(
1749 !floor_view.is_term_start(term_length),
1750 "expected a mid-term floor at view {floor_view}"
1751 );
1752 }
1753
1754 for validator in active.iter() {
1757 oracle
1758 .add_link(joiner.clone(), validator.clone(), link.clone())
1759 .await
1760 .unwrap();
1761 oracle
1762 .add_link(validator.clone(), joiner.clone(), link.clone())
1763 .await
1764 .unwrap();
1765 }
1766
1767 let joiner_context = context
1768 .child("validator")
1769 .with_attribute("public_key", &joiner);
1770 let reporter_config = mocks::reporter::Config {
1771 participants: participants.clone().try_into().unwrap(),
1772 scheme: schemes[joiner_idx].clone(),
1773 elector: elector.clone(),
1774 };
1775 let mut joiner_reporter =
1776 mocks::reporter::Reporter::new(joiner_context.child("reporter"), reporter_config);
1777 reporters.push(joiner_reporter.clone());
1778
1779 let application_cfg = mocks::application::Config::<Sha256, _> {
1780 relay: relay.clone(),
1781 me: joiner.clone(),
1782 propose_latency: (10.0, 5.0),
1783 verify_latency: (10.0, 5.0),
1784 certify_latency: (10.0, 5.0),
1785 should_certify: mocks::application::Certifier::Always,
1786 };
1787 let (actor, application) = mocks::application::Application::new(
1788 joiner_context.child("application"),
1789 application_cfg,
1790 );
1791 actor.start();
1792
1793 let cfg = config::Config {
1794 scheme: schemes[joiner_idx].clone(),
1795 elector,
1796 blocker: oracle.control(joiner.clone()),
1797 automaton: application.clone(),
1798 relay: application.clone(),
1799 reporter: joiner_reporter.clone(),
1800 strategy: Sequential,
1801 partition: format!("joiner_catches_tip_{joiner}"),
1802 mailbox_size: NZUsize!(1024),
1803 epoch: Epoch::new(333),
1804 floor: config::Floor::Finalized(floor_finalization),
1805 leader_timeout: Duration::from_secs(1),
1806 certification_timeout: Duration::from_secs(2),
1807 timeout_retry,
1808 fetch_timeout: NZDuration!(Duration::from_secs(1)),
1809 fetch_retry_delay: Duration::from_millis(100),
1810 view_retention,
1811 skip: SkipPolicy::Enabled {
1812 timeout: skip_timeout,
1813 budget: SkipBudget::Participants,
1814 },
1815 replay_buffer: NZUsize!(1024 * 1024),
1816 write_buffer: NZUsize!(1024 * 1024),
1817 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
1818 forward: ForwardPolicy::Disabled,
1819 track_historical_votes: false,
1820 };
1821 let engine = Engine::new(joiner_context.child("engine"), cfg);
1822 let (pending, recovered, resolver) = register_validator(&mut oracle, joiner).await;
1823 engine_handlers.push(engine.start(pending, recovered, resolver));
1824
1825 let (mut joiner_latest, mut joiner_monitor) = joiner_reporter.subscribe().await;
1826 while joiner_latest < tip_at_join {
1827 joiner_latest = joiner_monitor.recv().await.expect("event missing");
1828 }
1829
1830 let post_join_target = tip_at_join.saturating_add(ViewDelta::new(5));
1831 while joiner_latest < post_join_target {
1832 joiner_latest = joiner_monitor.recv().await.expect("event missing");
1833 }
1834
1835 for reporter in reporters.iter() {
1836 reporter.assert_no_faults();
1837 reporter.assert_no_invalid();
1838 }
1839
1840 let blocked = oracle.blocked().await.unwrap();
1841 assert!(blocked.is_empty());
1842 });
1843 }
1844
1845 test_for_all_fixtures!(non_genesis_floor_joiner_catches_tip);
1846
1847 #[test_group("slow")]
1848 #[test_traced]
1849 fn test_non_genesis_floor_joiner_catches_tip_stable_leader() {
1850 non_genesis_floor_joiner_catches_tip_with_term::<_, _, RoundRobin>(
1851 RoundRobin::default().with_term(
1852 TermLength::new(NZU32!(3)),
1853 Duration::from_secs(12),
1854 ViewDelta::new(0),
1855 ),
1856 scheme_mocks::fixture,
1857 );
1858 }
1859
1860 fn dishonest_leader_certification_rejected<S, F>(mut fixture: F)
1875 where
1876 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
1877 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
1878 RoundRobin: elector::Config<S>,
1879 <RoundRobin as elector::Config<S>>::Elector: Elector<S, Mode = Scheduled>,
1880 {
1881 let n = 5;
1882 let required_containers = View::new(50);
1883 let view_retention = ViewDelta::new(10);
1884 let skip_timeout = Duration::from_secs(12);
1885 let namespace = b"consensus".to_vec();
1886 let executor = deterministic::Runner::timed(Duration::from_secs(300));
1887 executor.start(|mut context| async move {
1888 let Fixture {
1889 participants,
1890 schemes,
1891 ..
1892 } = fixture(&mut context, &namespace, n);
1893 let mut oracle = start_test_network_with_peers::<S, _>(
1894 context.child("network"),
1895 participants.clone(),
1896 true,
1897 )
1898 .await;
1899 let mut registrations = register_validators(&mut oracle, &participants).await;
1900
1901 let link = Link {
1902 latency: Duration::from_millis(10),
1903 jitter: Duration::from_millis(1),
1904 success_rate: probability!(1.0),
1905 };
1906 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
1907
1908 let elector = RoundRobin::default();
1909 let participants_set: Set<S::PublicKey> = participants.clone().try_into().unwrap();
1910 let built_elector = elector.clone().build(&participants_set);
1911 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
1912 let mut reporters = Vec::new();
1913 let mut engine_handlers = Vec::new();
1914 let dishonest = Participant::new(0);
1915 for (idx, validator) in participants.iter().enumerate() {
1916 let context = context
1917 .child("validator")
1918 .with_attribute("public_key", validator);
1919 let reporter_config = mocks::reporter::Config {
1920 participants: participants.clone().try_into().unwrap(),
1921 scheme: schemes[idx].clone(),
1922 elector: elector.clone(),
1923 };
1924 let reporter =
1925 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
1926 reporters.push(reporter.clone());
1927
1928 let application_cfg = mocks::application::Config::<Sha256, _> {
1929 relay: relay.clone(),
1930 me: validator.clone(),
1931 propose_latency: (10.0, 5.0),
1932 verify_latency: (10.0, 5.0),
1933 certify_latency: (10.0, 5.0),
1934 should_certify: mocks::application::Certifier::Custom(Box::new({
1935 let built_elector_clone = built_elector.clone();
1936 move |round, _| built_elector_clone.elect(round, ()) != dishonest
1937 })),
1938 };
1939 let (actor, application) = mocks::application::Application::new(
1940 context.child("application"),
1941 application_cfg,
1942 );
1943 actor.start();
1944
1945 let blocker = oracle.control(validator.clone());
1946 let cfg = config::Config {
1947 scheme: schemes[idx].clone(),
1948 elector: elector.clone(),
1949 blocker,
1950 automaton: application.clone(),
1951 relay: application.clone(),
1952 reporter: reporter.clone(),
1953 strategy: Sequential,
1954 partition: validator.to_string(),
1955 mailbox_size: NZUsize!(1024),
1956 epoch: Epoch::new(333),
1957 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
1958 Epoch::new(333),
1959 )),
1960 leader_timeout: Duration::from_secs(1),
1961 certification_timeout: Duration::from_secs(2),
1962 timeout_retry: Duration::from_secs(10),
1963 fetch_timeout: NZDuration!(Duration::from_secs(1)),
1964 fetch_retry_delay: Duration::from_millis(100),
1965 view_retention,
1966 skip: SkipPolicy::Enabled {
1967 timeout: skip_timeout,
1968 budget: SkipBudget::Participants,
1969 },
1970 replay_buffer: NZUsize!(1024 * 1024),
1971 write_buffer: NZUsize!(1024 * 1024),
1972 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
1973 forward: ForwardPolicy::Disabled,
1974 track_historical_votes: false,
1975 };
1976 let engine = Engine::new(context.child("engine"), cfg);
1977 let (pending, recovered, resolver) = registrations
1978 .remove(validator)
1979 .expect("validator should be registered");
1980 engine_handlers.push(engine.start(pending, recovered, resolver));
1981 }
1982
1983 let mut finalizers = Vec::new();
1984 for reporter in reporters.iter_mut() {
1985 let (mut latest, mut monitor) = reporter.subscribe().await;
1986 finalizers.push(context.child("finalizer").spawn(move |_| async move {
1987 while latest < required_containers {
1988 latest = monitor.recv().await.expect("event missing");
1989 }
1990 }));
1991 }
1992 join_all(finalizers).await;
1993
1994 for reporter in reporters.iter() {
1995 reporter.assert_no_faults();
1996 reporter.assert_no_invalid();
1997 }
1998 });
1999 }
2000
2001 #[test_group("slow")]
2002 #[test_traced]
2003 fn test_dishonest_leader_certification_rejected() {
2004 dishonest_leader_certification_rejected::<_, _>(
2005 bls12381_threshold_vrf::fixture::<MinPk, _>,
2006 );
2007 dishonest_leader_certification_rejected::<_, _>(
2008 bls12381_threshold_vrf::fixture::<MinSig, _>,
2009 );
2010 dishonest_leader_certification_rejected::<_, _>(
2011 bls12381_threshold_std::fixture::<MinPk, _>,
2012 );
2013 dishonest_leader_certification_rejected::<_, _>(
2014 bls12381_threshold_std::fixture::<MinSig, _>,
2015 );
2016 dishonest_leader_certification_rejected::<_, _>(bls12381_multisig::fixture::<MinPk, _>);
2017 dishonest_leader_certification_rejected::<_, _>(bls12381_multisig::fixture::<MinSig, _>);
2018 dishonest_leader_certification_rejected::<_, _>(ed25519::fixture);
2019 dishonest_leader_certification_rejected::<_, _>(secp256r1::fixture);
2020 }
2021
2022 type RoundRobinReporter = mocks::reporter::Reporter<
2024 deterministic::Context,
2025 ed25519::Scheme,
2026 RoundRobin<Sha256>,
2027 Sha256Digest,
2028 >;
2029
2030 const CLUSTER_LEADER_TIMEOUT: Duration = Duration::from_millis(1_500);
2032
2033 const CLUSTER_CERTIFICATION_TIMEOUT: Duration = Duration::from_millis(3_500);
2035
2036 async fn setup_round_robin_cluster(
2045 context: &mut deterministic::Context,
2046 namespace: &[u8],
2047 link: Link,
2048 elector: RoundRobin<Sha256>,
2049 propose_latency: (f64, f64),
2050 certify_latency: fn(usize) -> (f64, f64),
2051 handoff: Handoff<()>,
2052 ) -> (
2053 Vec<RoundRobinReporter>,
2054 usize,
2055 Oracle<PublicKey, deterministic::Context>,
2056 ) {
2057 let epoch = Epoch::new(333);
2058 let Fixture {
2059 participants,
2060 schemes,
2061 ..
2062 } = ed25519::fixture(context, namespace, 5);
2063 let mut oracle = start_test_network_with_peers::<ed25519::Scheme, _>(
2064 context.child("network"),
2065 participants.clone(),
2066 true,
2067 )
2068 .await;
2069 let mut registrations = register_validators(&mut oracle, &participants).await;
2070 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
2071
2072 let relay = Arc::new(mocks::relay::Relay::new());
2073 let mut reporters = Vec::new();
2074
2075 for (idx, validator) in participants.iter().enumerate() {
2076 let context = context
2077 .child("validator")
2078 .with_attribute("public_key", validator);
2079 let reporter_config = mocks::reporter::Config {
2080 participants: participants.clone().try_into().unwrap(),
2081 scheme: schemes[idx].clone(),
2082 elector: elector.clone(),
2083 };
2084 let reporter =
2085 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
2086 reporters.push(reporter.clone());
2087
2088 let application_cfg = mocks::application::Config::<Sha256, _> {
2089 relay: relay.clone(),
2090 me: validator.clone(),
2091 propose_latency,
2092 verify_latency: (1.0, 0.0),
2093 certify_latency: certify_latency(idx),
2094 should_certify: mocks::application::Certifier::Always,
2095 };
2096 let (mut actor, application) =
2097 mocks::application::Application::new(context.child("application"), application_cfg);
2098 actor.set_handoff(handoff);
2099 actor.start();
2100
2101 let blocker = oracle.control(validator.clone());
2102 let cfg = config::Config {
2103 scheme: schemes[idx].clone(),
2104 elector: elector.clone(),
2105 blocker,
2106 automaton: application.clone(),
2107 relay: application.clone(),
2108 reporter: reporter.clone(),
2109 strategy: Sequential,
2110 partition: validator.to_string(),
2111 mailbox_size: NZUsize!(1024),
2112 epoch,
2113 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(epoch)),
2114 leader_timeout: CLUSTER_LEADER_TIMEOUT,
2115 certification_timeout: CLUSTER_CERTIFICATION_TIMEOUT,
2116 timeout_retry: Duration::from_secs(10),
2117 fetch_timeout: NZDuration!(Duration::from_secs(1)),
2118 fetch_retry_delay: Duration::from_millis(100),
2119 view_retention: ViewDelta::new(10),
2120 skip: SkipPolicy::Enabled {
2121 timeout: Duration::from_secs(12),
2122 budget: SkipBudget::Participants,
2123 },
2124 replay_buffer: NZUsize!(1024 * 1024),
2125 write_buffer: NZUsize!(1024 * 1024),
2126 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
2127 forward: ForwardPolicy::Disabled,
2128 track_historical_votes: false,
2129 };
2130 let engine = Engine::new(context.child("engine"), cfg);
2131 let (pending, recovered, resolver) = registrations
2132 .remove(validator)
2133 .expect("validator should be registered");
2134 engine.start(pending, recovered, resolver);
2135 }
2136
2137 let participants_set = participants.clone().try_into().unwrap();
2138 let built_elector: RoundRobinElector<ed25519::Scheme> = elector.build(&participants_set);
2139 let leader_idx = usize::from(built_elector.elect(Round::new(epoch, View::new(1)), ()));
2140
2141 (reporters, leader_idx, oracle)
2142 }
2143
2144 #[test_traced]
2145 fn test_stable_leader_optimistic_blocks_faster_than_network_latency() {
2146 let required_containers = View::new(100);
2147 let link_latency = Duration::from_millis(100);
2148 let executor = deterministic::Runner::timed(Duration::from_secs(30));
2149 executor.start(|mut context| async move {
2150 let (reporters, leader_idx, _oracle) = setup_round_robin_cluster(
2151 &mut context,
2152 b"consensus_stable_leader_high_latency",
2153 Link {
2154 latency: link_latency,
2155 jitter: Duration::from_millis(0),
2156 success_rate: probability!(1.0),
2157 },
2158 RoundRobin::<Sha256>::default().with_term(
2159 TermLength::new(NZU32!(128)),
2160 Duration::from_secs(20),
2161 ViewDelta::new(128),
2162 ),
2163 (10.0, 0.0),
2164 |_| (1.0, 0.0),
2165 Handoff::Wait,
2166 )
2167 .await;
2168
2169 let leader_reporter = reporters[leader_idx].clone();
2170 let start = context.current();
2171 while !leader_reporter
2172 .notarizes
2173 .lock()
2174 .contains_key(&required_containers)
2175 {
2176 context.sleep(Duration::from_millis(10)).await;
2177 }
2178 let elapsed = context.current().duration_since(start).unwrap_or_default();
2179 let average_block_time_s = elapsed.as_secs_f64() / required_containers.get() as f64;
2180 let network_latency_s = link_latency.as_secs_f64();
2181 assert!(
2182 average_block_time_s < network_latency_s,
2183 "expected average optimistic block time ({:.3}ms) to be below network latency ({:.3}ms); elapsed {:?} for {} views",
2184 average_block_time_s * 1000.0,
2185 network_latency_s * 1000.0,
2186 elapsed,
2187 required_containers
2188 );
2189
2190 for reporter in reporters.iter() {
2191 reporter.assert_no_invalid();
2192 }
2193 });
2194 }
2195
2196 #[test_traced]
2203 fn test_pipelined_handoff_reduces_boundary_latency() {
2204 let measured_views = 10u64..=100;
2205 let link_latency = Duration::from_millis(100);
2206 let term_length = TermLength::new(NZU32!(2));
2207 let executor = deterministic::Runner::timed(Duration::from_secs(30));
2208 executor.start(|mut context| async move {
2209 let (reporters, _, _oracle) = setup_round_robin_cluster(
2210 &mut context,
2211 b"consensus_pipelined_handoff_boundary_latency",
2212 Link {
2213 latency: link_latency,
2214 jitter: Duration::from_millis(0),
2215 success_rate: probability!(1.0),
2216 },
2217 RoundRobin::<Sha256>::default().with_term(
2218 term_length,
2219 Duration::from_secs(20),
2220 ViewDelta::new(4),
2221 ),
2222 (10.0, 0.0),
2223 |_| (1.0, 0.0),
2224 Handoff::Vote(()),
2225 )
2226 .await;
2227
2228 let reporter = reporters[0].clone();
2230 let mut observed_at = Vec::new();
2231 for view in measured_views.clone() {
2232 while !reporter.notarizations.lock().contains_key(&View::new(view)) {
2233 context.sleep(Duration::from_millis(1)).await;
2234 }
2235 observed_at.push(context.current());
2236 }
2237
2238 for (view, window) in measured_views.skip(1).zip(observed_at.windows(2)) {
2241 if !View::new(view).is_term_start(term_length) {
2242 continue;
2243 }
2244 let gap = window[1].duration_since(window[0]).unwrap_or_default();
2245 assert!(
2246 gap < 2 * link_latency,
2247 "expected pipelined boundary view {view} within two link latencies of its \
2248 parent, got {gap:?}"
2249 );
2250 }
2251
2252 for reporter in reporters.iter() {
2253 reporter.assert_no_invalid();
2254 reporter.assert_no_faults();
2255 assert!(
2256 reporter.nullifies.lock().is_empty(),
2257 "expected nullification-free boundaries"
2258 );
2259 }
2260 });
2261 }
2262
2263 #[test_traced]
2266 fn test_pipelined_handoff_halves_rotating_view_time() {
2267 let measured_views = 10u64..=100;
2268 let link_latency = Duration::from_millis(100);
2269 let executor = deterministic::Runner::timed(Duration::from_secs(60));
2270 executor.start(|mut context| async move {
2271 let (reporters, _, _oracle) = setup_round_robin_cluster(
2272 &mut context,
2273 b"consensus_pipelined_handoff_rotating",
2274 Link {
2275 latency: link_latency,
2276 jitter: Duration::from_millis(0),
2277 success_rate: probability!(1.0),
2278 },
2279 RoundRobin::<Sha256>::default(),
2280 (10.0, 0.0),
2281 |_| (1.0, 0.0),
2282 Handoff::Vote(()),
2283 )
2284 .await;
2285
2286 let reporter = reporters[0].clone();
2288 let start = View::new(*measured_views.start());
2289 while !reporter.notarizations.lock().contains_key(&start) {
2290 context.sleep(Duration::from_millis(1)).await;
2291 }
2292 let first = context.current();
2293 let end = View::new(*measured_views.end());
2294 while !reporter.notarizations.lock().contains_key(&end) {
2295 context.sleep(Duration::from_millis(1)).await;
2296 }
2297 let elapsed = context.current().duration_since(first).unwrap_or_default();
2298 let views = measured_views.end() - measured_views.start();
2299 let average = elapsed / u32::try_from(views).unwrap();
2300 assert!(
2301 average < 3 * link_latency / 2,
2302 "expected pipelined rotating views near one link latency, got {average:?}"
2303 );
2304
2305 for reporter in reporters.iter() {
2306 reporter.assert_no_invalid();
2307 reporter.assert_no_faults();
2308 assert!(
2309 reporter.nullifies.lock().is_empty(),
2310 "expected nullification-free views"
2311 );
2312 }
2313 });
2314 }
2315
2316 #[test_traced]
2322 fn test_pipelined_handoff_parent_nullify_releases_blocked_certification() {
2323 let n = 4;
2324 let epoch = Epoch::new(333);
2325 let required = View::new(10);
2326 let executor = deterministic::Runner::timed(Duration::from_secs(120));
2327 executor.start(|mut context| async move {
2328 let Fixture {
2329 participants,
2330 schemes,
2331 ..
2332 } = ed25519::fixture(&mut context, b"pipelined_handoff_blocked_certification", n);
2333 let mut oracle = start_test_network_with_peers::<ed25519::Scheme, _>(
2334 context.child("network"),
2335 participants.clone(),
2336 true,
2337 )
2338 .await;
2339 let mut registrations = register_validators(&mut oracle, &participants).await;
2340 let link = Link {
2341 latency: Duration::from_millis(10),
2342 jitter: Duration::from_millis(1),
2343 success_rate: probability!(1.0),
2344 };
2345 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
2346
2347 let elector = RoundRobin::<Sha256>::default().with_term(
2350 TermLength::new(NZU32!(2)),
2351 Duration::from_secs(30),
2352 ViewDelta::new(1),
2353 );
2354 let participant_set: Set<PublicKey> = participants.clone().try_into().unwrap();
2355 let schedule: RoundRobinElector<ed25519::Scheme> =
2356 elector.clone().build(&participant_set);
2357 let leader_of =
2358 |view: u64| usize::from(schedule.elect(Round::new(epoch, View::new(view)), ()));
2359 let incoming = leader_of(3);
2360 let offline = leader_of(5);
2361 assert!(![leader_of(1), incoming].contains(&offline));
2362
2363 let build: Arc<Mutex<Option<oneshot::Sender<Handoff<D>>>>> = Arc::default();
2366 let certification: Arc<Mutex<Option<oneshot::Sender<bool>>>> = Arc::default();
2367 let relay = Arc::new(mocks::relay::Relay::new());
2368 let mut reporters = Vec::new();
2369 for (idx, validator) in participants.iter().enumerate() {
2370 if idx == offline {
2371 continue;
2372 }
2373 let context = context
2374 .child("validator")
2375 .with_attribute("public_key", validator);
2376 let reporter_config = mocks::reporter::Config {
2377 participants: participants.clone().try_into().unwrap(),
2378 scheme: schemes[idx].clone(),
2379 elector: elector.clone(),
2380 };
2381 let reporter =
2382 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
2383 reporters.push(reporter.clone());
2384 let should_certify = if idx == incoming {
2385 let certification = certification.clone();
2386 mocks::application::Certifier::Controlled(Box::new(
2387 move |round, _, response| {
2388 if round.view() == View::new(2) {
2389 *certification.lock() = Some(response);
2390 } else {
2391 response.send_lossy(true);
2392 }
2393 },
2394 ))
2395 } else {
2396 mocks::application::Certifier::Always
2397 };
2398 let application_cfg = mocks::application::Config::<Sha256, _> {
2399 relay: relay.clone(),
2400 me: validator.clone(),
2401 propose_latency: (1.0, 0.0),
2402 verify_latency: (1.0, 0.0),
2403 certify_latency: (1.0, 0.0),
2404 should_certify,
2405 };
2406 let (mut actor, application) = mocks::application::Application::new(
2407 context.child("application"),
2408 application_cfg,
2409 );
2410 if idx == incoming {
2411 actor.set_handoff(Handoff::Stage(()));
2412 let build = build.clone();
2413 actor.set_handoff_controller(Box::new(move |round, proposal, response| {
2414 if round.view() == View::new(3) {
2415 *build.lock() = Some(response);
2416 } else {
2417 response.send_lossy(proposal);
2418 }
2419 }));
2420 }
2421 actor.start();
2422
2423 let cfg = config::Config {
2424 scheme: schemes[idx].clone(),
2425 elector: elector.clone(),
2426 blocker: oracle.control(validator.clone()),
2427 automaton: application.clone(),
2428 relay: application.clone(),
2429 reporter: reporter.clone(),
2430 strategy: Sequential,
2431 partition: validator.to_string(),
2432 mailbox_size: NZUsize!(1024),
2433 epoch,
2434 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(epoch)),
2435 leader_timeout: Duration::from_secs(1),
2436 certification_timeout: Duration::from_secs(2),
2437 timeout_retry: Duration::from_secs(10),
2438 fetch_timeout: NZDuration!(Duration::from_secs(1)),
2439 fetch_retry_delay: Duration::from_millis(100),
2440 view_retention: ViewDelta::new(10),
2441 skip: SkipPolicy::Enabled {
2442 timeout: Duration::from_secs(11),
2443 budget: SkipBudget::Participants,
2444 },
2445 replay_buffer: NZUsize!(1024 * 1024),
2446 write_buffer: NZUsize!(1024 * 1024),
2447 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
2448 forward: ForwardPolicy::Disabled,
2449 track_historical_votes: false,
2450 };
2451 let engine = Engine::new(context.child("engine"), cfg);
2452 let (pending, recovered, resolver) = registrations
2453 .remove(validator)
2454 .expect("validator should be registered");
2455 engine.start(pending, recovered, resolver);
2456 }
2457
2458 while !reporters.iter().all(|reporter| {
2461 reporter
2462 .finalizations
2463 .lock()
2464 .keys()
2465 .any(|view| *view >= required)
2466 }) {
2467 if build.lock().as_ref().is_some_and(|build| build.is_closed())
2468 && let Some(response) = certification.lock().take()
2469 {
2470 response.send_lossy(true);
2471 }
2472 context.sleep(Duration::from_millis(10)).await;
2473 }
2474 for reporter in reporters.iter() {
2475 reporter.assert_no_faults();
2476 reporter.assert_no_invalid();
2477 }
2478 });
2479 }
2480
2481 #[test_group("slow")]
2482 #[test]
2483 fn test_stable_leader_finalizes_full_term_without_nullification() {
2484 let required_view = View::new(1000);
2485 let executor = deterministic::Runner::timed(Duration::from_secs(40));
2486 executor.start(|mut context| async move {
2487 let (reporters, leader_idx, oracle) = setup_round_robin_cluster(
2488 &mut context,
2489 b"consensus_stable_leader_full_term_no_nullify",
2490 Link {
2494 latency: Duration::from_millis(1_000),
2495 jitter: Duration::from_millis(1),
2496 success_rate: probability!(1.0),
2497 },
2498 RoundRobin::<Sha256>::default().with_term(
2499 TermLength::new(NZU32!(1000)),
2500 Duration::from_secs(6),
2501 ViewDelta::new(100),
2502 ),
2503 (1.0, 0.0),
2504 |_| (1.0, 0.0),
2505 Handoff::Wait,
2506 )
2507 .await;
2508
2509 let start = context.current();
2510 let deadline = start + Duration::from_secs(25);
2511 while !reporters
2512 .iter()
2513 .all(|reporter| reporter.finalizations.lock().contains_key(&required_view))
2514 {
2515 if context.current() >= deadline {
2516 let progress: Vec<_> = reporters
2517 .iter()
2518 .map(|reporter| {
2519 let finalized = reporter
2520 .finalizations
2521 .lock()
2522 .keys()
2523 .copied()
2524 .max()
2525 .unwrap_or(View::zero());
2526 (finalized, reporter.nullifications.lock().len())
2527 })
2528 .collect();
2529 panic!(
2530 "expected all validators to finalize view {required_view} before {deadline:?}; (max finalized, nullifications) per reporter: {progress:?}",
2531 );
2532 }
2533 context.sleep(Duration::from_millis(10)).await;
2534 }
2535
2536 let leader_reporter = &reporters[leader_idx];
2537 assert!(
2538 leader_reporter.notarizes.lock().contains_key(&required_view),
2539 "stable leader must notarize through full term ending at view {}",
2540 required_view
2541 );
2542 let leader_tip_notarization = leader_reporter
2543 .notarizations
2544 .lock()
2545 .get(&required_view)
2546 .cloned()
2547 .expect("leader reporter missing tip notarization");
2548 assert_eq!(
2549 leader_tip_notarization.proposal.parent,
2550 required_view.previous().unwrap_or(View::zero()),
2551 "unexpected parent for leader tip notarization"
2552 );
2553
2554 for (idx, reporter) in reporters.iter().enumerate() {
2555 reporter.assert_no_invalid();
2556 reporter.assert_no_faults();
2557
2558 assert!(
2559 reporter.nullifies.lock().is_empty(),
2560 "reporter {} observed unexpected nullify votes",
2561 idx
2562 );
2563 assert!(
2564 reporter.nullifications.lock().is_empty(),
2565 "reporter {} observed unexpected nullification certificates",
2566 idx
2567 );
2568
2569 let finalization = reporter
2570 .finalizations
2571 .lock()
2572 .get(&required_view)
2573 .map(|(finalization, _)| finalization.clone())
2574 .unwrap_or_else(|| panic!("reporter {idx} missing tip finalization"));
2575 assert_eq!(
2576 finalization.proposal.round.view(),
2577 required_view,
2578 "reporter {idx} has mismatched tip finalization round"
2579 );
2580 assert_eq!(
2581 finalization.proposal.parent,
2582 required_view.previous().unwrap_or(View::zero()),
2583 "reporter {idx} has non-chain tip finalization parent"
2584 );
2585 }
2586
2587 let blocked = oracle.blocked().await.unwrap();
2588 assert!(blocked.is_empty());
2589 });
2590 }
2591
2592 fn observer<S, F, L>(mut fixture: F, elector: L)
2593 where
2594 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
2595 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
2596 L: elector::Config<S>,
2597 {
2598 let n_active = 5;
2600 let required_containers = View::new(100);
2601 let view_retention = ViewDelta::new(10);
2602 let skip_timeout = Duration::from_secs(12);
2603 let namespace = b"consensus".to_vec();
2604 let executor = deterministic::Runner::timed(Duration::from_secs(300));
2605 executor.start(|mut context| async move {
2606 let Fixture {
2608 participants,
2609 schemes,
2610 verifier,
2611 ..
2612 } = fixture(&mut context, &namespace, n_active);
2613
2614 let private_key_observer = PrivateKey::from_seed(n_active as u64);
2616 let public_key_observer = private_key_observer.public_key();
2617
2618 let mut oracle = start_test_network_with_split_peers::<S, _, _>(
2619 context.child("network"),
2620 participants.clone(),
2621 [public_key_observer.clone()],
2622 true,
2623 )
2624 .await;
2625
2626 let mut all_validators = participants.clone();
2628 all_validators.push(public_key_observer.clone());
2629 all_validators.sort();
2630 let mut registrations = register_validators(&mut oracle, &all_validators).await;
2631
2632 let link = Link {
2634 latency: Duration::from_millis(10),
2635 jitter: Duration::from_millis(1),
2636 success_rate: probability!(1.0),
2637 };
2638 link_validators(&mut oracle, &all_validators, Action::Link(link), None).await;
2639
2640 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
2642 let mut reporters = Vec::new();
2643
2644 for (idx, validator) in participants.iter().enumerate() {
2645 let is_observer = *validator == public_key_observer;
2646
2647 let context = context
2649 .child("validator")
2650 .with_attribute("public_key", validator);
2651
2652 let signing = if is_observer {
2654 verifier.clone()
2655 } else {
2656 schemes[idx].clone()
2657 };
2658 let reporter_config = mocks::reporter::Config {
2659 participants: participants.clone().try_into().unwrap(),
2660 scheme: signing.clone(),
2661 elector: elector.clone(),
2662 };
2663 let reporter =
2664 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
2665 reporters.push(reporter.clone());
2666 let application_cfg = mocks::application::Config::<Sha256, _> {
2667 relay: relay.clone(),
2668 me: validator.clone(),
2669 propose_latency: (10.0, 5.0),
2670 verify_latency: (10.0, 5.0),
2671 certify_latency: (10.0, 5.0),
2672 should_certify: mocks::application::Certifier::Always,
2673 };
2674 let (actor, application) = mocks::application::Application::new(
2675 context.child("application"),
2676 application_cfg,
2677 );
2678 actor.start();
2679 let blocker = oracle.control(validator.clone());
2680 let cfg = config::Config {
2681 scheme: signing.clone(),
2682 elector: elector.clone(),
2683 blocker,
2684 automaton: application.clone(),
2685 relay: application.clone(),
2686 reporter: reporter.clone(),
2687 strategy: Sequential,
2688 partition: validator.to_string(),
2689 mailbox_size: NZUsize!(1024),
2690 epoch: Epoch::new(333),
2691 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
2692 Epoch::new(333),
2693 )),
2694 leader_timeout: Duration::from_secs(1),
2695 certification_timeout: Duration::from_secs(2),
2696 timeout_retry: Duration::from_secs(10),
2697 fetch_timeout: NZDuration!(Duration::from_secs(1)),
2698 fetch_retry_delay: Duration::from_millis(100),
2699 view_retention,
2700 skip: SkipPolicy::Enabled {
2701 timeout: skip_timeout,
2702 budget: SkipBudget::Participants,
2703 },
2704 replay_buffer: NZUsize!(1024 * 1024),
2705 write_buffer: NZUsize!(1024 * 1024),
2706 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
2707 forward: ForwardPolicy::Disabled,
2708 track_historical_votes: false,
2709 };
2710 let engine = Engine::new(context.child("engine"), cfg);
2711
2712 let (pending, recovered, resolver) = registrations
2714 .remove(validator)
2715 .expect("validator should be registered");
2716 engine.start(pending, recovered, resolver);
2717 }
2718
2719 let mut finalizers = Vec::new();
2721 for reporter in reporters.iter_mut() {
2722 let (mut latest, mut monitor) = reporter.subscribe().await;
2723 finalizers.push(context.child("finalizer").spawn(move |_| async move {
2724 while latest < required_containers {
2725 latest = monitor.recv().await.expect("event missing");
2726 }
2727 }));
2728 }
2729 join_all(finalizers).await;
2730
2731 for reporter in reporters.iter() {
2734 reporter.assert_no_faults();
2736 reporter.assert_no_invalid();
2737
2738 let blocked = oracle.blocked().await.unwrap();
2740 assert!(blocked.is_empty());
2741 }
2742 });
2743 }
2744
2745 test_for_all_fixtures!(observer);
2746
2747 fn unclean_shutdown<S, F, L>(fixture: F, elector: L)
2748 where
2749 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
2750 F: FnMut(&mut TestRng, &[u8], u32) -> Fixture<S>,
2751 L: elector::Config<S>,
2752 {
2753 unclean_shutdown_with_term::<S, F, L>(elector, fixture);
2754 }
2755
2756 fn unclean_shutdown_with_term<S, F, L>(elector: L, mut fixture: F)
2757 where
2758 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
2759 F: FnMut(&mut TestRng, &[u8], u32) -> Fixture<S>,
2760 L: elector::Config<S>,
2761 {
2762 let n = 5;
2764 let required_containers = View::new(100);
2765 let view_retention = ViewDelta::new(10);
2766 let skip_timeout = Duration::from_secs(12);
2767 let namespace = b"consensus".to_vec();
2768
2769 let shutdowns: Arc<Mutex<u64>> = Arc::new(Mutex::new(0));
2771 let supervised = Arc::new(Mutex::new(Vec::new()));
2772 let mut prev_checkpoint = None;
2773
2774 let mut rng = test_rng();
2776 let Fixture {
2777 participants,
2778 schemes,
2779 ..
2780 } = fixture(&mut rng, &namespace, n);
2781 let reporter_seed: [u8; 32] = rng.random();
2782
2783 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, S::PublicKey>::new());
2785
2786 loop {
2787 let participants = participants.clone();
2788 let schemes = schemes.clone();
2789 let shutdowns = shutdowns.clone();
2790 let supervised = supervised.clone();
2791 let relay = relay.clone();
2792 let elector = elector.clone();
2793 relay.deregister_all(); let f = |mut context: deterministic::Context| async move {
2796 let mut oracle = start_test_network_with_peers::<S, _>(
2798 context.child("network"),
2799 participants.clone(),
2800 true,
2801 )
2802 .await;
2803 let mut registrations = register_validators(&mut oracle, &participants).await;
2804
2805 let link = Link {
2807 latency: Duration::from_millis(50),
2808 jitter: Duration::from_millis(50),
2809 success_rate: probability!(1.0),
2810 };
2811 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
2812
2813 let elector = elector.clone();
2815 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
2816 let mut reporters = HashMap::new();
2817 let mut engine_handlers = Vec::new();
2818 for (idx, validator) in participants.iter().enumerate() {
2819 let context = context
2821 .child("validator")
2822 .with_attribute("public_key", validator);
2823
2824 let reporter_config = mocks::reporter::Config {
2826 participants: participants.clone().try_into().unwrap(),
2827 scheme: schemes[idx].clone(),
2828 elector: elector.clone(),
2829 };
2830 let reporter_rng = StdRng::from_seed(reporter_seed);
2831 let reporter = mocks::reporter::Reporter::new(reporter_rng, reporter_config);
2832 reporters.insert(validator.clone(), reporter.clone());
2833 let application_cfg = mocks::application::Config::<Sha256, _> {
2834 relay: relay.clone(),
2835 me: validator.clone(),
2836 propose_latency: (10.0, 5.0),
2837 verify_latency: (10.0, 5.0),
2838 certify_latency: (10.0, 5.0),
2839 should_certify: mocks::application::Certifier::Always,
2840 };
2841 let (actor, application) = mocks::application::Application::new(
2842 context.child("application"),
2843 application_cfg,
2844 );
2845 actor.start();
2846 let blocker = oracle.control(validator.clone());
2847 let cfg = config::Config {
2848 scheme: schemes[idx].clone(),
2849 elector: elector.clone(),
2850 blocker,
2851 automaton: application.clone(),
2852 relay: application.clone(),
2853 reporter: reporter.clone(),
2854 strategy: Sequential,
2855 partition: validator.to_string(),
2856 mailbox_size: NZUsize!(1024),
2857 epoch: Epoch::new(333),
2858 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
2859 Epoch::new(333),
2860 )),
2861 leader_timeout: Duration::from_millis(500),
2864 certification_timeout: Duration::from_secs(1),
2865 timeout_retry: Duration::from_millis(500),
2866 fetch_timeout: NZDuration!(Duration::from_secs(1)),
2867 fetch_retry_delay: Duration::from_millis(100),
2868 view_retention,
2869 skip: SkipPolicy::Enabled {
2870 timeout: skip_timeout,
2871 budget: SkipBudget::Participants,
2872 },
2873 replay_buffer: NZUsize!(1024 * 1024),
2874 write_buffer: NZUsize!(1024 * 1024),
2875 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
2876 forward: ForwardPolicy::Disabled,
2877 track_historical_votes: false,
2878 };
2879 let engine = Engine::new(context.child("engine"), cfg);
2880
2881 let (pending, recovered, resolver) = registrations
2883 .remove(validator)
2884 .expect("validator should be registered");
2885 engine_handlers.push(engine.start(pending, recovered, resolver));
2886 }
2887
2888 let mut finalizers = Vec::new();
2890 for reporter in reporters.values_mut() {
2891 let (mut latest, mut monitor) = reporter.subscribe().await;
2892 finalizers.push(context.child("finalizer").spawn(move |_| async move {
2893 while latest < required_containers {
2894 latest = monitor.recv().await.expect("event missing");
2895 }
2896 }));
2897 }
2898
2899 let wait =
2901 context.random_range(Duration::from_millis(100)..Duration::from_millis(2_000));
2902 let result = select! {
2903 _ = context.sleep(wait) => {
2904 {
2906 let mut shutdowns = shutdowns.lock();
2907 debug!(shutdowns = *shutdowns, elapsed = ?wait, "restarting");
2908 *shutdowns += 1;
2909 }
2910 supervised.lock().push(reporters);
2911 false
2912 },
2913 _ = join_all(finalizers) => {
2914 let supervised = supervised.lock();
2916 for reporters in supervised.iter() {
2917 for reporter in reporters.values() {
2918 reporter.assert_no_faults();
2919 }
2920 }
2921 true
2922 },
2923 };
2924
2925 let blocked = oracle.blocked().await.unwrap();
2927 assert!(blocked.is_empty());
2928
2929 result
2930 };
2931
2932 let (complete, checkpoint) = prev_checkpoint
2933 .map_or_else(
2934 || deterministic::Runner::timed(Duration::from_secs(180)),
2935 deterministic::Runner::from,
2936 )
2937 .start_and_recover(f);
2938
2939 if complete {
2941 break;
2942 }
2943
2944 prev_checkpoint = Some(checkpoint);
2945 }
2946 }
2947
2948 test_for_all_fixtures!(unclean_shutdown);
2949
2950 #[test_group("slow")]
2958 #[test_traced]
2959 fn test_unclean_shutdown_stable_leader_optimistic() {
2960 unclean_shutdown_with_term::<_, _, RoundRobin>(
2961 RoundRobin::default().with_term(
2962 TermLength::new(NZU32!(5)),
2963 Duration::from_secs(13),
2964 ViewDelta::new(2),
2965 ),
2966 ed25519::fixture,
2967 );
2968 }
2969
2970 fn backfill<S, F, L>(mut fixture: F, elector: L)
2971 where
2972 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
2973 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
2974 L: elector::Config<S>,
2975 {
2976 let n = 4;
2978 let required_containers = View::new(100);
2979 let view_retention = ViewDelta::new(10);
2980 let skip_timeout = Duration::from_secs(11);
2981 let namespace = b"consensus".to_vec();
2982 let executor = deterministic::Runner::timed(Duration::from_secs(240));
2983 executor.start(|mut context| async move {
2984 let Fixture {
2986 participants,
2987 schemes,
2988 ..
2989 } = fixture(&mut context, &namespace, n);
2990 let mut oracle = start_test_network_with_peers::<S, _>(
2991 context.child("network"),
2992 participants.clone(),
2993 true,
2994 )
2995 .await;
2996 let mut registrations = register_validators(&mut oracle, &participants).await;
2997
2998 let link = Link {
3000 latency: Duration::from_millis(10),
3001 jitter: Duration::from_millis(1),
3002 success_rate: probability!(1.0),
3003 };
3004 link_validators(
3005 &mut oracle,
3006 &participants,
3007 Action::Link(link),
3008 Some(|_, i, j| ![i, j].contains(&0usize)),
3009 )
3010 .await;
3011
3012 let elector = elector.clone();
3014 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
3015 let mut reporters = Vec::new();
3016 let mut engine_handlers = Vec::new();
3017 for (idx_scheme, validator) in participants.iter().enumerate() {
3018 if idx_scheme == 0 {
3020 continue;
3021 }
3022
3023 let context = context
3025 .child("validator")
3026 .with_attribute("public_key", validator);
3027
3028 let reporter_config = mocks::reporter::Config {
3030 participants: participants.clone().try_into().unwrap(),
3031 scheme: schemes[idx_scheme].clone(),
3032 elector: elector.clone(),
3033 };
3034 let reporter =
3035 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
3036 reporters.push(reporter.clone());
3037 let application_cfg = mocks::application::Config::<Sha256, _> {
3038 relay: relay.clone(),
3039 me: validator.clone(),
3040 propose_latency: (10.0, 5.0),
3041 verify_latency: (10.0, 5.0),
3042 certify_latency: (10.0, 5.0),
3043 should_certify: mocks::application::Certifier::Always,
3044 };
3045 let (actor, application) = mocks::application::Application::new(
3046 context.child("application"),
3047 application_cfg,
3048 );
3049 actor.start();
3050 let blocker = oracle.control(validator.clone());
3051 let cfg = config::Config {
3052 scheme: schemes[idx_scheme].clone(),
3053 elector: elector.clone(),
3054 blocker,
3055 automaton: application.clone(),
3056 relay: application.clone(),
3057 reporter: reporter.clone(),
3058 strategy: Sequential,
3059 partition: validator.to_string(),
3060 mailbox_size: NZUsize!(1024),
3061 epoch: Epoch::new(333),
3062 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
3063 Epoch::new(333),
3064 )),
3065 leader_timeout: Duration::from_secs(1),
3066 certification_timeout: Duration::from_secs(2),
3067 timeout_retry: Duration::from_secs(10),
3068 fetch_timeout: NZDuration!(Duration::from_secs(1)),
3069 fetch_retry_delay: Duration::from_millis(100),
3070 view_retention,
3071 skip: SkipPolicy::Enabled {
3072 timeout: skip_timeout,
3073 budget: SkipBudget::Participants,
3074 },
3075 replay_buffer: NZUsize!(1024 * 1024),
3076 write_buffer: NZUsize!(1024 * 1024),
3077 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
3078 forward: ForwardPolicy::Disabled,
3079 track_historical_votes: false,
3080 };
3081 let engine = Engine::new(context.child("engine"), cfg);
3082
3083 let (pending, recovered, resolver) = registrations
3085 .remove(validator)
3086 .expect("validator should be registered");
3087 engine_handlers.push(engine.start(pending, recovered, resolver));
3088 }
3089
3090 let mut finalizers = Vec::new();
3092 for reporter in reporters.iter_mut() {
3093 let (mut latest, mut monitor) = reporter.subscribe().await;
3094 finalizers.push(context.child("finalizer").spawn(move |_| async move {
3095 while latest < required_containers {
3096 latest = monitor.recv().await.expect("event missing");
3097 }
3098 }));
3099 }
3100 join_all(finalizers).await;
3101
3102 let link = Link {
3104 latency: Duration::from_secs(3),
3105 jitter: Duration::from_millis(0),
3106 success_rate: probability!(1.0),
3107 };
3108 link_validators(
3109 &mut oracle,
3110 &participants,
3111 Action::Update(link.clone()),
3112 Some(|_, i, j| ![i, j].contains(&0usize)),
3113 )
3114 .await;
3115
3116 context.sleep(Duration::from_secs(60)).await;
3118
3119 link_validators(
3121 &mut oracle,
3122 &participants,
3123 Action::Unlink,
3124 Some(|_, i, j| [i, j].contains(&1usize) && ![i, j].contains(&0usize)),
3125 )
3126 .await;
3127
3128 let me = participants[0].clone();
3130 let context = context.child("validator").with_attribute("public_key", &me);
3131
3132 link_validators(
3134 &mut oracle,
3135 &participants,
3136 Action::Link(link),
3137 Some(|_, i, j| [i, j].contains(&0usize) && ![i, j].contains(&1usize)),
3138 )
3139 .await;
3140
3141 let link = Link {
3143 latency: Duration::from_millis(10),
3144 jitter: Duration::from_millis(3),
3145 success_rate: probability!(1.0),
3146 };
3147 link_validators(
3148 &mut oracle,
3149 &participants,
3150 Action::Update(link),
3151 Some(|_, i, j| ![i, j].contains(&1usize)),
3152 )
3153 .await;
3154
3155 let reporter_config = mocks::reporter::Config {
3157 participants: participants.clone().try_into().unwrap(),
3158 scheme: schemes[0].clone(),
3159 elector: elector.clone(),
3160 };
3161 let mut reporter =
3162 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
3163 reporters.push(reporter.clone());
3164 let application_cfg = mocks::application::Config::<Sha256, _> {
3165 relay: relay.clone(),
3166 me: me.clone(),
3167 propose_latency: (10.0, 5.0),
3168 verify_latency: (10.0, 5.0),
3169 certify_latency: (10.0, 5.0),
3170 should_certify: mocks::application::Certifier::Always,
3171 };
3172 let (actor, application) =
3173 mocks::application::Application::new(context.child("application"), application_cfg);
3174 actor.start();
3175 let blocker = oracle.control(me.clone());
3176 let cfg = config::Config {
3177 scheme: schemes[0].clone(),
3178 elector: elector.clone(),
3179 blocker,
3180 automaton: application.clone(),
3181 relay: application.clone(),
3182 reporter: reporter.clone(),
3183 strategy: Sequential,
3184 partition: me.to_string(),
3185 mailbox_size: NZUsize!(1024),
3186 epoch: Epoch::new(333),
3187 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(Epoch::new(
3188 333,
3189 ))),
3190 leader_timeout: Duration::from_secs(1),
3191 certification_timeout: Duration::from_secs(2),
3192 timeout_retry: Duration::from_secs(10),
3193 fetch_timeout: NZDuration!(Duration::from_secs(1)),
3194 fetch_retry_delay: Duration::from_millis(100),
3195 view_retention,
3196 skip: SkipPolicy::Enabled {
3197 timeout: skip_timeout,
3198 budget: SkipBudget::Participants,
3199 },
3200 replay_buffer: NZUsize!(1024 * 1024),
3201 write_buffer: NZUsize!(1024 * 1024),
3202 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
3203 forward: ForwardPolicy::Disabled,
3204 track_historical_votes: false,
3205 };
3206 let engine = Engine::new(context.child("engine"), cfg);
3207
3208 let (pending, recovered, resolver) = registrations
3210 .remove(&me)
3211 .expect("validator should be registered");
3212 engine_handlers.push(engine.start(pending, recovered, resolver));
3213
3214 let (mut latest, mut monitor) = reporter.subscribe().await;
3216 while latest < required_containers {
3217 latest = monitor.recv().await.expect("event missing");
3218 }
3219
3220 let blocked = oracle.blocked().await.unwrap();
3222 assert!(blocked.is_empty());
3223 });
3224 }
3225
3226 test_for_all_fixtures!(backfill);
3227
3228 #[test_group("slow")]
3229 #[test_traced]
3230 fn test_backfill_stable_leader_optimistic() {
3231 backfill::<_, _, RoundRobin>(
3232 ed25519::fixture,
3233 RoundRobin::default().with_term(
3236 TermLength::new(NZU32!(5)),
3237 Duration::from_secs(51),
3238 ViewDelta::new(2),
3239 ),
3240 );
3241 }
3242
3243 fn one_offline<S, F, L>(fixture: F, elector: L)
3244 where
3245 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
3246 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
3247 L: elector::Config<S>,
3248 {
3249 one_offline_with_term::<S, F, L>(elector, fixture);
3250 }
3251
3252 fn one_offline_with_term<S, F, L>(elector: L, mut fixture: F)
3253 where
3254 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
3255 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
3256 L: elector::Config<S>,
3257 {
3258 let n = 5;
3260 let quorum = quorum(n) as usize;
3261 let required_containers = View::new(100);
3262 let view_retention = ViewDelta::new(10);
3263 let skip_timeout = Duration::from_secs(11);
3264 let max_exceptions = 10;
3265 let namespace = b"consensus".to_vec();
3266 let executor = deterministic::Runner::timed(Duration::from_secs(300));
3267 executor.start(|mut context| async move {
3268 let Fixture {
3270 participants,
3271 schemes,
3272 ..
3273 } = fixture(&mut context, &namespace, n);
3274 let mut oracle = start_test_network_with_peers::<S, _>(
3275 context.child("network"),
3276 participants.clone(),
3277 true,
3278 )
3279 .await;
3280 let mut registrations = register_validators(&mut oracle, &participants).await;
3281
3282 let link = Link {
3284 latency: Duration::from_millis(10),
3285 jitter: Duration::from_millis(1),
3286 success_rate: probability!(1.0),
3287 };
3288 link_validators(
3289 &mut oracle,
3290 &participants,
3291 Action::Link(link),
3292 Some(|_, i, j| ![i, j].contains(&0usize)),
3293 )
3294 .await;
3295
3296 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
3298 let mut reporters = Vec::new();
3299 let mut engine_handlers = Vec::new();
3300 for (idx_scheme, validator) in participants.iter().enumerate() {
3301 if idx_scheme == 0 {
3303 continue;
3304 }
3305
3306 let context = context
3308 .child("validator")
3309 .with_attribute("public_key", validator);
3310
3311 let reporter_config = mocks::reporter::Config {
3313 participants: participants.clone().try_into().unwrap(),
3314 scheme: schemes[idx_scheme].clone(),
3315 elector: elector.clone(),
3316 };
3317 let reporter =
3318 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
3319 reporters.push(reporter.clone());
3320 let application_cfg = mocks::application::Config::<Sha256, _> {
3321 relay: relay.clone(),
3322 me: validator.clone(),
3323 propose_latency: (10.0, 5.0),
3324 verify_latency: (10.0, 5.0),
3325 certify_latency: (10.0, 5.0),
3326 should_certify: mocks::application::Certifier::Always,
3327 };
3328 let (actor, application) = mocks::application::Application::new(
3329 context.child("application"),
3330 application_cfg,
3331 );
3332 actor.start();
3333 let blocker = oracle.control(validator.clone());
3334 let cfg = config::Config {
3335 scheme: schemes[idx_scheme].clone(),
3336 elector: elector.clone(),
3337 blocker,
3338 automaton: application.clone(),
3339 relay: application.clone(),
3340 reporter: reporter.clone(),
3341 strategy: Sequential,
3342 partition: validator.to_string(),
3343 mailbox_size: NZUsize!(1024),
3344 epoch: Epoch::new(333),
3345 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
3346 Epoch::new(333),
3347 )),
3348 leader_timeout: Duration::from_secs(1),
3349 certification_timeout: Duration::from_secs(2),
3350 timeout_retry: Duration::from_secs(10),
3351 fetch_timeout: NZDuration!(Duration::from_secs(1)),
3352 fetch_retry_delay: Duration::from_millis(100),
3353 view_retention,
3354 skip: SkipPolicy::Enabled {
3355 timeout: skip_timeout,
3356 budget: SkipBudget::Participants,
3357 },
3358 replay_buffer: NZUsize!(1024 * 1024),
3359 write_buffer: NZUsize!(1024 * 1024),
3360 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
3361 forward: ForwardPolicy::Disabled,
3362 track_historical_votes: false,
3363 };
3364 let engine = Engine::new(context.child("engine"), cfg);
3365
3366 let (pending, recovered, resolver) = registrations
3368 .remove(validator)
3369 .expect("validator should be registered");
3370 engine_handlers.push(engine.start(pending, recovered, resolver));
3371 }
3372
3373 let mut finalizers = Vec::new();
3375 for reporter in reporters.iter_mut() {
3376 let (mut latest, mut monitor) = reporter.subscribe().await;
3377 finalizers.push(context.child("finalizer").spawn(move |_| async move {
3378 while latest < required_containers {
3379 latest = monitor.recv().await.expect("event missing");
3380 }
3381 }));
3382 }
3383 join_all(finalizers).await;
3384
3385 let exceptions = 0;
3387 let offline = &participants[0];
3388 for reporter in reporters.iter() {
3389 reporter.assert_no_faults();
3391
3392 reporter.assert_no_invalid();
3394
3395 let mut exceptions = 0;
3397 {
3398 let notarizes = reporter.notarizes.lock();
3399 for (view, payloads) in notarizes.iter() {
3400 for participants in payloads.values() {
3401 if participants.contains(offline) {
3402 panic!("view: {view}");
3403 }
3404 }
3405 }
3406 }
3407 {
3408 let nullifies = reporter.nullifies.lock();
3409 for (view, participants) in nullifies.iter() {
3410 if participants.contains(offline) {
3411 panic!("view: {view}");
3412 }
3413 }
3414 }
3415 {
3416 let finalizes = reporter.finalizes.lock();
3417 for (view, payloads) in finalizes.iter() {
3418 for finalizers in payloads.values() {
3419 if finalizers.contains(offline) {
3420 panic!("view: {view}");
3421 }
3422 }
3423 }
3424 }
3425
3426 let mut offline_views = Vec::new();
3428 {
3429 let leaders = reporter.leaders.lock();
3430 for (view, leader) in leaders.iter() {
3431 if leader == offline {
3432 offline_views.push(*view);
3433 }
3434 }
3435 }
3436 assert!(!offline_views.is_empty());
3437
3438 {
3440 let nullifies = reporter.nullifies.lock();
3441 for view in offline_views.iter() {
3442 let nullifies = nullifies.get(view).map_or(0, |n| n.len());
3443 if nullifies < quorum {
3444 warn!("missing expected view nullifies: {}", view);
3445 exceptions += 1;
3446 }
3447 }
3448 }
3449 {
3450 let nullifications = reporter.nullifications.lock();
3451 for view in offline_views.iter() {
3452 if !nullifications.contains_key(view) {
3453 warn!("missing expected view nullifies: {}", view);
3454 exceptions += 1;
3455 }
3456 }
3457 }
3458
3459 assert!(exceptions <= max_exceptions);
3461 }
3462 assert!(exceptions <= max_exceptions);
3463
3464 let blocked = oracle.blocked().await.unwrap();
3466 assert!(blocked.is_empty());
3467
3468 let encoded = context.encode();
3470 let leader_label = format!("leader=\"{}\"", offline);
3471 assert!(
3472 count_nonzero_metric_lines(&encoded, &["_timeouts", &leader_label]) >= n - 1,
3473 "expected timeout metrics for offline leader"
3474 );
3475 assert_eq!(
3476 count_nonzero_metric_lines(&encoded, &["_nullifications", &leader_label]),
3477 n - 1,
3478 "expected all online nodes to record _nullifications for offline leader"
3479 );
3480 });
3481 }
3482
3483 test_for_all_fixtures!(one_offline);
3484
3485 #[test_group("slow")]
3486 #[test_traced]
3487 fn test_one_offline_stable_leader() {
3488 one_offline_with_term::<_, _, RoundRobin>(
3489 RoundRobin::default().with_term(
3490 TermLength::new(NZU32!(3)),
3491 Duration::from_secs(12),
3492 ViewDelta::new(0),
3493 ),
3494 scheme_mocks::fixture,
3495 );
3496 }
3497
3498 fn slow_validator<S, F, L>(mut fixture: F, elector: L)
3499 where
3500 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
3501 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
3502 L: elector::Config<S>,
3503 {
3504 let n = 5;
3506 let required_containers = View::new(50);
3507 let view_retention = ViewDelta::new(10);
3508 let skip_timeout = Duration::from_secs(11);
3509 let namespace = b"consensus".to_vec();
3510 let executor = deterministic::Runner::timed(Duration::from_secs(300));
3511 executor.start(|mut context| async move {
3512 let Fixture {
3514 participants,
3515 schemes,
3516 ..
3517 } = fixture(&mut context, &namespace, n);
3518 let mut oracle = start_test_network_with_peers::<S, _>(
3519 context.child("network"),
3520 participants.clone(),
3521 true,
3522 )
3523 .await;
3524 let mut registrations = register_validators(&mut oracle, &participants).await;
3525
3526 let link = Link {
3528 latency: Duration::from_millis(10),
3529 jitter: Duration::from_millis(1),
3530 success_rate: probability!(1.0),
3531 };
3532 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
3533
3534 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
3536 let mut reporters = Vec::new();
3537 let mut engine_handlers = Vec::new();
3538 for (idx_scheme, validator) in participants.iter().enumerate() {
3539 let context = context
3541 .child("validator")
3542 .with_attribute("public_key", validator);
3543
3544 let reporter_config = mocks::reporter::Config {
3546 participants: participants.clone().try_into().unwrap(),
3547 scheme: schemes[idx_scheme].clone(),
3548 elector: elector.clone(),
3549 };
3550 let reporter =
3551 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
3552 reporters.push(reporter.clone());
3553 let application_cfg = if idx_scheme == 0 {
3554 mocks::application::Config::<Sha256, _> {
3555 relay: relay.clone(),
3556 me: validator.clone(),
3557 propose_latency: (10_000.0, 0.0),
3558 verify_latency: (10_000.0, 5.0),
3559 certify_latency: (10_000.0, 5.0),
3560 should_certify: mocks::application::Certifier::Always,
3561 }
3562 } else {
3563 mocks::application::Config::<Sha256, _> {
3564 relay: relay.clone(),
3565 me: validator.clone(),
3566 propose_latency: (10.0, 5.0),
3567 verify_latency: (10.0, 5.0),
3568 certify_latency: (10.0, 5.0),
3569 should_certify: mocks::application::Certifier::Always,
3570 }
3571 };
3572 let (actor, application) = mocks::application::Application::new(
3573 context.child("application"),
3574 application_cfg,
3575 );
3576 actor.start();
3577 let blocker = oracle.control(validator.clone());
3578 let cfg = config::Config {
3579 scheme: schemes[idx_scheme].clone(),
3580 elector: elector.clone(),
3581 blocker,
3582 automaton: application.clone(),
3583 relay: application.clone(),
3584 reporter: reporter.clone(),
3585 strategy: Sequential,
3586 partition: validator.to_string(),
3587 mailbox_size: NZUsize!(1024),
3588 epoch: Epoch::new(333),
3589 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
3590 Epoch::new(333),
3591 )),
3592 leader_timeout: Duration::from_secs(1),
3593 certification_timeout: Duration::from_secs(2),
3594 timeout_retry: Duration::from_secs(10),
3595 fetch_timeout: NZDuration!(Duration::from_secs(1)),
3596 fetch_retry_delay: Duration::from_millis(100),
3597 view_retention,
3598 skip: SkipPolicy::Enabled {
3599 timeout: skip_timeout,
3600 budget: SkipBudget::Participants,
3601 },
3602 replay_buffer: NZUsize!(1024 * 1024),
3603 write_buffer: NZUsize!(1024 * 1024),
3604 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
3605 forward: ForwardPolicy::Disabled,
3606 track_historical_votes: false,
3607 };
3608 let engine = Engine::new(context.child("engine"), cfg);
3609
3610 let (pending, recovered, resolver) = registrations
3612 .remove(validator)
3613 .expect("validator should be registered");
3614 engine_handlers.push(engine.start(pending, recovered, resolver));
3615 }
3616
3617 let mut finalizers = Vec::new();
3619 for reporter in reporters.iter_mut() {
3620 let (mut latest, mut monitor) = reporter.subscribe().await;
3621 finalizers.push(context.child("finalizer").spawn(move |_| async move {
3622 while latest < required_containers {
3623 latest = monitor.recv().await.expect("event missing");
3624 }
3625 }));
3626 }
3627 join_all(finalizers).await;
3628
3629 let slow = &participants[0];
3631 for reporter in reporters.iter() {
3632 reporter.assert_no_faults();
3634
3635 reporter.assert_no_invalid();
3637
3638 {
3641 let notarizes = reporter.notarizes.lock();
3642 assert!(notarizes.values().all(|payloads| {
3643 payloads
3644 .values()
3645 .all(|participants| !participants.contains(slow))
3646 }));
3647 }
3648 {
3649 let finalizes = reporter.finalizes.lock();
3650 assert!(finalizes.values().all(|payloads| {
3651 payloads
3652 .values()
3653 .all(|participants| !participants.contains(slow))
3654 }));
3655 }
3656
3657 {
3659 let finalizations = reporter.finalizations.lock();
3660 assert!(
3661 finalizations
3662 .keys()
3663 .any(|view| *view >= required_containers)
3664 );
3665 }
3666 }
3667
3668 let blocked = oracle.blocked().await.unwrap();
3670 assert!(blocked.is_empty());
3671 });
3672 }
3673
3674 test_for_all_fixtures!(slow_validator);
3675
3676 fn all_recovery<S, F, L>(mut fixture: F, elector: L)
3677 where
3678 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
3679 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
3680 L: elector::Config<S>,
3681 {
3682 let n = 5;
3684 let required_containers = View::new(100);
3685 let view_retention = ViewDelta::new(10);
3686 let skip_timeout = Duration::from_secs(11);
3687 let namespace = b"consensus".to_vec();
3688 let executor = deterministic::Runner::timed(Duration::from_secs(1800));
3689 executor.start(|mut context| async move {
3690 let Fixture {
3692 participants,
3693 schemes,
3694 ..
3695 } = fixture(&mut context, &namespace, n);
3696 let mut oracle = start_test_network_with_peers::<S, _>(
3697 context.child("network"),
3698 participants.clone(),
3699 true,
3700 )
3701 .await;
3702 let mut registrations = register_validators(&mut oracle, &participants).await;
3703
3704 let link = Link {
3706 latency: Duration::from_secs(3),
3707 jitter: Duration::from_millis(0),
3708 success_rate: probability!(1.0),
3709 };
3710 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
3711
3712 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
3714 let mut reporters = Vec::new();
3715 let mut engine_handlers = Vec::new();
3716 for (idx, validator) in participants.iter().enumerate() {
3717 let context = context
3719 .child("validator")
3720 .with_attribute("public_key", validator);
3721
3722 let reporter_config = mocks::reporter::Config {
3724 participants: participants.clone().try_into().unwrap(),
3725 scheme: schemes[idx].clone(),
3726 elector: elector.clone(),
3727 };
3728 let reporter =
3729 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
3730 reporters.push(reporter.clone());
3731 let application_cfg = mocks::application::Config::<Sha256, _> {
3732 relay: relay.clone(),
3733 me: validator.clone(),
3734 propose_latency: (10.0, 5.0),
3735 verify_latency: (10.0, 5.0),
3736 certify_latency: (10.0, 5.0),
3737 should_certify: mocks::application::Certifier::Always,
3738 };
3739 let (actor, application) = mocks::application::Application::new(
3740 context.child("application"),
3741 application_cfg,
3742 );
3743 actor.start();
3744 let blocker = oracle.control(validator.clone());
3745 let cfg = config::Config {
3746 scheme: schemes[idx].clone(),
3747 elector: elector.clone(),
3748 blocker,
3749 automaton: application.clone(),
3750 relay: application.clone(),
3751 reporter: reporter.clone(),
3752 strategy: Sequential,
3753 partition: validator.to_string(),
3754 mailbox_size: NZUsize!(1024),
3755 epoch: Epoch::new(333),
3756 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
3757 Epoch::new(333),
3758 )),
3759 leader_timeout: Duration::from_secs(1),
3760 certification_timeout: Duration::from_secs(2),
3761 timeout_retry: Duration::from_secs(10),
3762 fetch_timeout: NZDuration!(Duration::from_secs(1)),
3763 fetch_retry_delay: Duration::from_millis(100),
3764 view_retention,
3765 skip: SkipPolicy::Enabled {
3766 timeout: skip_timeout,
3767 budget: SkipBudget::Participants,
3768 },
3769 replay_buffer: NZUsize!(1024 * 1024),
3770 write_buffer: NZUsize!(1024 * 1024),
3771 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
3772 forward: ForwardPolicy::Disabled,
3773 track_historical_votes: false,
3774 };
3775 let engine = Engine::new(context.child("engine"), cfg);
3776
3777 let (pending, recovered, resolver) = registrations
3779 .remove(validator)
3780 .expect("validator should be registered");
3781 engine_handlers.push(engine.start(pending, recovered, resolver));
3782 }
3783
3784 let mut finalizers = Vec::new();
3786 for reporter in reporters.iter_mut() {
3787 let (_, mut monitor) = reporter.subscribe().await;
3788 finalizers.push(context.child("finalizer").spawn(move |context| async move {
3789 select! {
3790 _timeout = context.sleep(Duration::from_secs(60)) => {},
3791 _done = monitor.recv() => {
3792 panic!("engine should not notarize or finalize anything");
3793 },
3794 }
3795 }));
3796 }
3797 join_all(finalizers).await;
3798
3799 link_validators(&mut oracle, &participants, Action::Unlink, None).await;
3801
3802 context.sleep(Duration::from_secs(60)).await;
3804
3805 let mut latest = View::zero();
3807 for reporter in reporters.iter() {
3808 let nullifies = reporter.nullifies.lock();
3809 let max = nullifies.keys().max().unwrap();
3810 if *max > latest {
3811 latest = *max;
3812 }
3813 }
3814
3815 let link = Link {
3817 latency: Duration::from_millis(10),
3818 jitter: Duration::from_millis(1),
3819 success_rate: probability!(1.0),
3820 };
3821 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
3822
3823 let mut finalizers = Vec::new();
3825 for reporter in reporters.iter_mut() {
3826 let (mut latest, mut monitor) = reporter.subscribe().await;
3827 finalizers.push(context.child("finalizer").spawn(move |_| async move {
3828 while latest < required_containers {
3829 latest = monitor.recv().await.expect("event missing");
3830 }
3831 }));
3832 }
3833 join_all(finalizers).await;
3834
3835 for reporter in reporters.iter() {
3837 reporter.assert_no_faults();
3839
3840 reporter.assert_no_invalid();
3842
3843 {
3848 let mut found = 0;
3852 let notarizations = reporter.notarizations.lock();
3853 for view in View::range(latest, latest.saturating_add(view_retention)) {
3854 if notarizations.contains_key(&view) {
3855 found += 1;
3856 }
3857 }
3858 let tolerated_missing = 3;
3862 assert!(
3863 found >= view_retention.get().saturating_sub(tolerated_missing),
3864 "found: {found}"
3865 );
3866 }
3867 }
3868
3869 let blocked = oracle.blocked().await.unwrap();
3871 assert!(blocked.is_empty());
3872 });
3873 }
3874
3875 test_for_all_fixtures!(all_recovery);
3876
3877 fn all_crash_after_nullify<S, F, L>(mut fixture: F, elector: L)
3878 where
3879 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
3880 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
3881 L: elector::Config<S>,
3882 {
3883 let n = 4;
3885 let required_containers = View::new(10);
3886 let view_retention = ViewDelta::new(10);
3887 let skip_timeout = Duration::from_secs(11);
3888 let namespace = b"consensus".to_vec();
3889 let executor = deterministic::Runner::timed(Duration::from_secs(3600));
3890 executor.start(|mut context| async move {
3891 let Fixture {
3893 participants,
3894 schemes,
3895 ..
3896 } = fixture(&mut context, &namespace, n);
3897 let mut oracle = start_test_network_with_peers::<S, _>(
3898 context.child("network"),
3899 participants.clone(),
3900 true,
3901 )
3902 .await;
3903 let mut registrations = register_validators(&mut oracle, &participants).await;
3904
3905 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
3908 let mut reporters = Vec::new();
3909 let mut engine_handlers = Vec::new();
3910 for (idx_scheme, validator) in participants.iter().enumerate() {
3911 if idx_scheme == 0 {
3913 continue;
3914 }
3915
3916 let context = context
3918 .child("validator")
3919 .with_attribute("public_key", validator);
3920
3921 let reporter_config = mocks::reporter::Config {
3923 participants: participants.clone().try_into().unwrap(),
3924 scheme: schemes[idx_scheme].clone(),
3925 elector: elector.clone(),
3926 };
3927 let reporter =
3928 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
3929 reporters.push(reporter.clone());
3930 let application_cfg = mocks::application::Config::<Sha256, _> {
3931 relay: relay.clone(),
3932 me: validator.clone(),
3933 propose_latency: (10.0, 5.0),
3934 verify_latency: (10.0, 5.0),
3935 certify_latency: (10.0, 5.0),
3936 should_certify: mocks::application::Certifier::Always,
3937 };
3938 let (actor, application) = mocks::application::Application::new(
3939 context.child("application"),
3940 application_cfg,
3941 );
3942 actor.start();
3943 let blocker = oracle.control(validator.clone());
3944 let cfg = config::Config {
3945 scheme: schemes[idx_scheme].clone(),
3946 elector: elector.clone(),
3947 blocker,
3948 automaton: application.clone(),
3949 relay: application.clone(),
3950 reporter: reporter.clone(),
3951 strategy: Sequential,
3952 partition: validator.to_string(),
3953 mailbox_size: NZUsize!(1024),
3954 epoch: Epoch::new(333),
3955 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
3956 Epoch::new(333),
3957 )),
3958 leader_timeout: Duration::from_secs(1),
3959 certification_timeout: Duration::from_secs(2),
3960 timeout_retry: Duration::from_secs(10),
3961 fetch_timeout: NZDuration!(Duration::from_secs(1)),
3962 fetch_retry_delay: Duration::from_millis(100),
3963 view_retention,
3964 skip: SkipPolicy::Enabled {
3965 timeout: skip_timeout,
3966 budget: SkipBudget::Participants,
3967 },
3968 replay_buffer: NZUsize!(1024 * 1024),
3969 write_buffer: NZUsize!(1024 * 1024),
3970 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
3971 forward: ForwardPolicy::Disabled,
3972 track_historical_votes: false,
3973 };
3974 let engine = Engine::new(context.child("engine"), cfg);
3975
3976 let (pending, recovered, resolver) = registrations
3978 .remove(validator)
3979 .expect("validator should be registered");
3980 engine_handlers.push(engine.start(pending, recovered, resolver));
3981 }
3982
3983 let stalled = View::new(1);
3986 loop {
3987 let nullified = reporters.iter().zip(participants.iter().skip(1)).all(
3988 |(reporter, validator)| {
3989 reporter
3990 .nullifies
3991 .lock()
3992 .get(&stalled)
3993 .is_some_and(|nullifiers| nullifiers.contains(validator))
3994 },
3995 );
3996 if nullified {
3997 break;
3998 }
3999 context.sleep(Duration::from_millis(100)).await;
4000 }
4001
4002 context.sleep(Duration::from_secs(1)).await;
4006
4007 for handle in engine_handlers.drain(..) {
4010 handle.abort();
4011 let _ = handle.await;
4012 }
4013 relay.deregister_all();
4014
4015 let link = Link {
4017 latency: Duration::from_millis(10),
4018 jitter: Duration::from_millis(1),
4019 success_rate: probability!(1.0),
4020 };
4021 link_validators(
4022 &mut oracle,
4023 &participants,
4024 Action::Link(link),
4025 Some(|_, i, j| ![i, j].contains(&0usize)),
4026 )
4027 .await;
4028
4029 let mut reporters = Vec::new();
4031 for (idx_scheme, validator) in participants.iter().enumerate() {
4032 if idx_scheme == 0 {
4034 continue;
4035 }
4036
4037 let context = context
4039 .child("validator_restarted")
4040 .with_attribute("public_key", validator);
4041
4042 let reporter_config = mocks::reporter::Config {
4044 participants: participants.clone().try_into().unwrap(),
4045 scheme: schemes[idx_scheme].clone(),
4046 elector: elector.clone(),
4047 };
4048 let reporter =
4049 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
4050 reporters.push(reporter.clone());
4051 let application_cfg = mocks::application::Config::<Sha256, _> {
4052 relay: relay.clone(),
4053 me: validator.clone(),
4054 propose_latency: (10.0, 5.0),
4055 verify_latency: (10.0, 5.0),
4056 certify_latency: (10.0, 5.0),
4057 should_certify: mocks::application::Certifier::Always,
4058 };
4059 let (actor, application) = mocks::application::Application::new(
4060 context.child("application"),
4061 application_cfg,
4062 );
4063 actor.start();
4064 let blocker = oracle.control(validator.clone());
4065 let cfg = config::Config {
4066 scheme: schemes[idx_scheme].clone(),
4067 elector: elector.clone(),
4068 blocker,
4069 automaton: application.clone(),
4070 relay: application.clone(),
4071 reporter: reporter.clone(),
4072 strategy: Sequential,
4073 partition: validator.to_string(),
4074 mailbox_size: NZUsize!(1024),
4075 epoch: Epoch::new(333),
4076 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
4077 Epoch::new(333),
4078 )),
4079 leader_timeout: Duration::from_secs(1),
4080 certification_timeout: Duration::from_secs(2),
4081 timeout_retry: Duration::from_secs(10),
4082 fetch_timeout: NZDuration!(Duration::from_secs(1)),
4083 fetch_retry_delay: Duration::from_millis(100),
4084 view_retention,
4085 skip: SkipPolicy::Enabled {
4086 timeout: skip_timeout,
4087 budget: SkipBudget::Participants,
4088 },
4089 replay_buffer: NZUsize!(1024 * 1024),
4090 write_buffer: NZUsize!(1024 * 1024),
4091 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
4092 forward: ForwardPolicy::Disabled,
4093 track_historical_votes: false,
4094 };
4095 let engine = Engine::new(context.child("engine"), cfg);
4096
4097 let (pending, recovered, resolver) =
4099 register_validator(&mut oracle, validator.clone()).await;
4100 engine.start(pending, recovered, resolver);
4101 }
4102
4103 let mut finalizers = Vec::new();
4107 for reporter in reporters.iter_mut() {
4108 let (mut latest, mut monitor) = reporter.subscribe().await;
4109 finalizers.push(context.child("finalizer").spawn(move |_| async move {
4110 while latest < required_containers {
4111 latest = monitor.recv().await.expect("event missing");
4112 }
4113 }));
4114 }
4115 join_all(finalizers).await;
4116 });
4117 }
4118
4119 test_for_all_fixtures!(all_crash_after_nullify);
4120
4121 fn partition<S, F, L>(fixture: F, elector: L)
4122 where
4123 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
4124 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
4125 L: elector::Config<S>,
4126 {
4127 partition_with_term::<S, F, L>(elector, fixture);
4128 }
4129
4130 fn partition_with_term<S, F, L>(elector: L, mut fixture: F)
4131 where
4132 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
4133 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
4134 L: elector::Config<S>,
4135 {
4136 let n = 10;
4138 let required_containers = View::new(50);
4139 let view_retention = ViewDelta::new(10);
4140 let skip_timeout = Duration::from_secs(11);
4141 let namespace = b"consensus".to_vec();
4142 let executor = deterministic::Runner::timed(Duration::from_secs(900));
4143 executor.start(|mut context| async move {
4144 let Fixture {
4146 participants,
4147 schemes,
4148 ..
4149 } = fixture(&mut context, &namespace, n);
4150 let mut oracle = start_test_network_with_peers::<S, _>(
4151 context.child("network"),
4152 participants.clone(),
4153 true,
4154 )
4155 .await;
4156 let mut registrations = register_validators(&mut oracle, &participants).await;
4157
4158 let link = Link {
4160 latency: Duration::from_millis(10),
4161 jitter: Duration::from_millis(1),
4162 success_rate: probability!(1.0),
4163 };
4164 link_validators(&mut oracle, &participants, Action::Link(link.clone()), None).await;
4165
4166 let elector = elector.clone();
4168 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
4169 let mut reporters = Vec::new();
4170 let mut engine_handlers = Vec::new();
4171 for (idx, validator) in participants.iter().enumerate() {
4172 let context = context
4174 .child("validator")
4175 .with_attribute("public_key", validator);
4176
4177 let reporter_config = mocks::reporter::Config {
4179 participants: participants.clone().try_into().unwrap(),
4180 scheme: schemes[idx].clone(),
4181 elector: elector.clone(),
4182 };
4183 let reporter =
4184 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
4185 reporters.push(reporter.clone());
4186 let application_cfg = mocks::application::Config::<Sha256, _> {
4187 relay: relay.clone(),
4188 me: validator.clone(),
4189 propose_latency: (10.0, 5.0),
4190 verify_latency: (10.0, 5.0),
4191 certify_latency: (10.0, 5.0),
4192 should_certify: mocks::application::Certifier::Always,
4193 };
4194 let (actor, application) = mocks::application::Application::new(
4195 context.child("application"),
4196 application_cfg,
4197 );
4198 actor.start();
4199 let blocker = oracle.control(validator.clone());
4200 let cfg = config::Config {
4201 scheme: schemes[idx].clone(),
4202 elector: elector.clone(),
4203 blocker,
4204 automaton: application.clone(),
4205 relay: application.clone(),
4206 reporter: reporter.clone(),
4207 strategy: Sequential,
4208 partition: validator.to_string(),
4209 mailbox_size: NZUsize!(1024),
4210 epoch: Epoch::new(333),
4211 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
4212 Epoch::new(333),
4213 )),
4214 leader_timeout: Duration::from_secs(1),
4215 certification_timeout: Duration::from_secs(2),
4216 timeout_retry: Duration::from_secs(10),
4217 fetch_timeout: NZDuration!(Duration::from_secs(1)),
4218 fetch_retry_delay: Duration::from_millis(100),
4219 view_retention,
4220 skip: SkipPolicy::Enabled {
4221 timeout: skip_timeout,
4222 budget: SkipBudget::Participants,
4223 },
4224 replay_buffer: NZUsize!(1024 * 1024),
4225 write_buffer: NZUsize!(1024 * 1024),
4226 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
4227 forward: ForwardPolicy::Disabled,
4228 track_historical_votes: false,
4229 };
4230 let engine = Engine::new(context.child("engine"), cfg);
4231
4232 let (pending, recovered, resolver) = registrations
4234 .remove(validator)
4235 .expect("validator should be registered");
4236 engine_handlers.push(engine.start(pending, recovered, resolver));
4237 }
4238
4239 let mut finalizers = Vec::new();
4241 for reporter in reporters.iter_mut() {
4242 let (mut latest, mut monitor) = reporter.subscribe().await;
4243 finalizers.push(context.child("finalizer").spawn(move |_| async move {
4244 while latest < required_containers {
4245 latest = monitor.recv().await.expect("event missing");
4246 }
4247 }));
4248 }
4249 join_all(finalizers).await;
4250
4251 fn separated(n: usize, a: usize, b: usize) -> bool {
4253 let m = n / 2;
4254 (a < m && b >= m) || (a >= m && b < m)
4255 }
4256 link_validators(&mut oracle, &participants, Action::Unlink, Some(separated)).await;
4257
4258 context.sleep(Duration::from_secs(10)).await;
4260
4261 let mut finalizers = Vec::new();
4263 for reporter in reporters.iter_mut() {
4264 let (_, mut monitor) = reporter.subscribe().await;
4265 finalizers.push(context.child("finalizer").spawn(move |context| async move {
4266 select! {
4267 _timeout = context.sleep(Duration::from_secs(60)) => {},
4268 _done = monitor.recv() => {
4269 panic!("engine should not notarize or finalize anything");
4270 },
4271 }
4272 }));
4273 }
4274 join_all(finalizers).await;
4275
4276 link_validators(
4278 &mut oracle,
4279 &participants,
4280 Action::Link(link),
4281 Some(separated),
4282 )
4283 .await;
4284
4285 let mut finalizers = Vec::new();
4287 for reporter in reporters.iter_mut() {
4288 let (mut latest, mut monitor) = reporter.subscribe().await;
4289 let required = latest.saturating_add(ViewDelta::new(required_containers.get()));
4290 finalizers.push(context.child("finalizer").spawn(move |_| async move {
4291 while latest < required {
4292 latest = monitor.recv().await.expect("event missing");
4293 }
4294 }));
4295 }
4296 join_all(finalizers).await;
4297
4298 for reporter in reporters.iter() {
4300 reporter.assert_no_faults();
4302
4303 reporter.assert_no_invalid();
4305 }
4306
4307 let blocked = oracle.blocked().await.unwrap();
4309 assert!(blocked.is_empty());
4310 });
4311 }
4312
4313 test_for_all_fixtures!(partition);
4314
4315 #[test_group("slow")]
4316 #[test_traced]
4317 fn test_partition_stable_leader_optimistic() {
4318 partition_with_term::<_, _, RoundRobin>(
4319 RoundRobin::default().with_term(
4320 TermLength::new(NZU32!(5)),
4321 Duration::from_secs(13),
4322 ViewDelta::new(2),
4323 ),
4324 ed25519::fixture,
4325 );
4326 }
4327
4328 fn slow_and_lossy_links_seeded<S, F, L>(seed: u64, fixture: F, elector: L) -> String
4329 where
4330 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
4331 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
4332 L: elector::Config<S>,
4333 {
4334 slow_and_lossy_links_seeded_with_term::<S, F, L>(elector, seed, fixture)
4335 }
4336
4337 fn slow_and_lossy_links_seeded_with_term<S, F, L>(
4338 elector: L,
4339 seed: u64,
4340 mut fixture: F,
4341 ) -> String
4342 where
4343 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
4344 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
4345 L: elector::Config<S>,
4346 {
4347 let n = 5;
4349 let required_containers = View::new(50);
4350 let view_retention = ViewDelta::new(10);
4351 let skip_timeout = Duration::from_secs(11);
4352 let namespace = b"consensus".to_vec();
4353 let cfg = deterministic::Config::new()
4354 .with_seed(seed)
4355 .with_timeout(Some(Duration::from_secs(5_000)));
4356 let executor = deterministic::Runner::new(cfg);
4357 executor.start(|mut context| async move {
4358 let Fixture {
4360 participants,
4361 schemes,
4362 ..
4363 } = fixture(&mut context, &namespace, n);
4364 let mut oracle = start_test_network_with_peers::<S, _>(
4365 context.child("network"),
4366 participants.clone(),
4367 true,
4368 )
4369 .await;
4370 let mut registrations = register_validators(&mut oracle, &participants).await;
4371
4372 let degraded_link = Link {
4374 latency: Duration::from_millis(200),
4375 jitter: Duration::from_millis(150),
4376 success_rate: probability!(0.5),
4377 };
4378 link_validators(
4379 &mut oracle,
4380 &participants,
4381 Action::Link(degraded_link),
4382 None,
4383 )
4384 .await;
4385
4386 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
4388 let mut reporters = Vec::new();
4389 let mut engine_handlers = Vec::new();
4390 for (idx, validator) in participants.iter().enumerate() {
4391 let context = context
4393 .child("validator")
4394 .with_attribute("public_key", validator);
4395
4396 let reporter_config = mocks::reporter::Config {
4398 participants: participants.clone().try_into().unwrap(),
4399 scheme: schemes[idx].clone(),
4400 elector: elector.clone(),
4401 };
4402 let reporter =
4403 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
4404 reporters.push(reporter.clone());
4405 let application_cfg = mocks::application::Config::<Sha256, _> {
4406 relay: relay.clone(),
4407 me: validator.clone(),
4408 propose_latency: (10.0, 5.0),
4409 verify_latency: (10.0, 5.0),
4410 certify_latency: (10.0, 5.0),
4411 should_certify: mocks::application::Certifier::Always,
4412 };
4413 let (actor, application) = mocks::application::Application::new(
4414 context.child("application"),
4415 application_cfg,
4416 );
4417 actor.start();
4418 let blocker = oracle.control(validator.clone());
4419 let cfg = config::Config {
4420 scheme: schemes[idx].clone(),
4421 elector: elector.clone(),
4422 blocker,
4423 automaton: application.clone(),
4424 relay: application.clone(),
4425 reporter: reporter.clone(),
4426 strategy: Sequential,
4427 partition: validator.to_string(),
4428 mailbox_size: NZUsize!(1024),
4429 epoch: Epoch::new(333),
4430 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
4431 Epoch::new(333),
4432 )),
4433 leader_timeout: Duration::from_secs(1),
4434 certification_timeout: Duration::from_secs(2),
4435 timeout_retry: Duration::from_secs(10),
4436 fetch_timeout: NZDuration!(Duration::from_secs(1)),
4437 fetch_retry_delay: Duration::from_millis(100),
4438 view_retention,
4439 skip: SkipPolicy::Enabled {
4440 timeout: skip_timeout,
4441 budget: SkipBudget::Participants,
4442 },
4443 replay_buffer: NZUsize!(1024 * 1024),
4444 write_buffer: NZUsize!(1024 * 1024),
4445 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
4446 forward: ForwardPolicy::Disabled,
4447 track_historical_votes: false,
4448 };
4449 let engine = Engine::new(context.child("engine"), cfg);
4450
4451 let (pending, recovered, resolver) = registrations
4453 .remove(validator)
4454 .expect("validator should be registered");
4455 engine_handlers.push(engine.start(pending, recovered, resolver));
4456 }
4457
4458 let mut finalizers = Vec::new();
4460 for reporter in reporters.iter_mut() {
4461 let (mut latest, mut monitor) = reporter.subscribe().await;
4462 finalizers.push(context.child("finalizer").spawn(move |_| async move {
4463 while latest < required_containers {
4464 latest = monitor.recv().await.expect("event missing");
4465 }
4466 }));
4467 }
4468 join_all(finalizers).await;
4469
4470 for reporter in reporters.iter() {
4472 reporter.assert_no_faults();
4474
4475 reporter.assert_no_invalid();
4477 }
4478
4479 let blocked = oracle.blocked().await.unwrap();
4481 assert!(blocked.is_empty());
4482
4483 context.auditor().state()
4484 })
4485 }
4486
4487 fn slow_and_lossy_links<S, F, L>(fixture: F, elector: L) -> String
4488 where
4489 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
4490 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
4491 L: elector::Config<S>,
4492 {
4493 slow_and_lossy_links_seeded::<_, _, L>(6, fixture, elector)
4494 }
4495
4496 test_for_all_fixtures!(slow_and_lossy_links);
4497
4498 #[test_group("slow")]
4499 #[test_traced]
4500 fn test_slow_and_lossy_links_stable_leader_optimistic() {
4501 slow_and_lossy_links_seeded_with_term::<_, _, RoundRobin>(
4502 RoundRobin::default().with_term(
4503 TermLength::new(NZU32!(5)),
4504 Duration::from_secs(13),
4505 ViewDelta::new(2),
4506 ),
4507 6,
4508 ed25519::fixture,
4509 );
4510 }
4511
4512 fn determinism<S, F, L>(seed: u64, fixture: F, elector: L)
4513 where
4514 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
4515 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S> + Copy,
4516 L: elector::Config<S>,
4517 {
4518 assert_eq!(
4521 slow_and_lossy_links_seeded::<_, _, L>(seed, fixture, elector.clone()),
4522 slow_and_lossy_links_seeded::<_, _, L>(seed, fixture, elector),
4523 );
4524 }
4525
4526 test_for_all_fixtures!(determinism, seeds = 5);
4527
4528 #[test_group("slow")]
4529 #[test_traced]
4530 fn test_distinct_states() {
4531 macro_rules! collect {
4533 ($vec:ident, $suffix:ident, $elector:ty, $fixture:expr, $elector_config:expr) => {
4534 $vec.push((
4535 stringify!($suffix),
4536 slow_and_lossy_links_seeded::<_, _, $elector>(7, $fixture, $elector_config),
4537 ));
4538 };
4539 }
4540 let mut states = Vec::new();
4541 for_each_fixture!(collect!(states));
4542 for pair in states.windows(2) {
4543 assert_ne!(
4544 pair[0].1, pair[1].1,
4545 "state {} equals state {}",
4546 pair[0].0, pair[1].0
4547 );
4548 }
4549 }
4550
4551 fn conflicter<S, F, L>(seed: u64, mut fixture: F, elector: L)
4552 where
4553 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
4554 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
4555 L: elector::Config<S>,
4556 {
4557 let n = 4;
4559 let required_containers = View::new(50);
4560 let view_retention = ViewDelta::new(10);
4561 let skip_timeout = Duration::from_secs(11);
4562 let namespace = b"consensus".to_vec();
4563 let cfg = deterministic::Config::new()
4564 .with_seed(seed)
4565 .with_timeout(Some(Duration::from_secs(30)));
4566 let executor = deterministic::Runner::new(cfg);
4567 executor.start(|mut context| async move {
4568 let Fixture {
4570 participants,
4571 schemes,
4572 ..
4573 } = fixture(&mut context, &namespace, n);
4574 let mut oracle = start_test_network_with_peers::<S, _>(
4575 context.child("network"),
4576 participants.clone(),
4577 true,
4578 )
4579 .await;
4580 let mut registrations = register_validators(&mut oracle, &participants).await;
4581
4582 let link = Link {
4584 latency: Duration::from_millis(10),
4585 jitter: Duration::from_millis(1),
4586 success_rate: probability!(1.0),
4587 };
4588 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
4589
4590 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
4592 let mut reporters = Vec::new();
4593 for (idx_scheme, validator) in participants.iter().enumerate() {
4594 let context = context
4596 .child("validator")
4597 .with_attribute("public_key", validator);
4598
4599 let reporter_config = mocks::reporter::Config {
4601 participants: participants.clone().try_into().unwrap(),
4602 scheme: schemes[idx_scheme].clone(),
4603 elector: elector.clone(),
4604 };
4605 let reporter =
4606 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
4607 let (pending, recovered, resolver) = registrations
4608 .remove(validator)
4609 .expect("validator should be registered");
4610 if idx_scheme == 0 {
4611 let cfg = mocks::conflicter::Config {
4612 scheme: schemes[idx_scheme].clone(),
4613 };
4614
4615 let engine: mocks::conflicter::Conflicter<_, _, Sha256> =
4616 mocks::conflicter::Conflicter::new(context.child("byzantine_engine"), cfg);
4617 engine.start(pending);
4618 } else {
4619 reporters.push(reporter.clone());
4620 let application_cfg = mocks::application::Config::<Sha256, _> {
4621 relay: relay.clone(),
4622 me: validator.clone(),
4623 propose_latency: (10.0, 5.0),
4624 verify_latency: (10.0, 5.0),
4625 certify_latency: (10.0, 5.0),
4626 should_certify: mocks::application::Certifier::Always,
4627 };
4628 let (actor, application) = mocks::application::Application::new(
4629 context.child("application"),
4630 application_cfg,
4631 );
4632 actor.start();
4633 let blocker = oracle.control(validator.clone());
4634 let cfg = config::Config {
4635 scheme: schemes[idx_scheme].clone(),
4636 elector: elector.clone(),
4637 blocker,
4638 automaton: application.clone(),
4639 relay: application.clone(),
4640 reporter: reporter.clone(),
4641 strategy: Sequential,
4642 partition: validator.to_string(),
4643 mailbox_size: NZUsize!(1024),
4644 epoch: Epoch::new(333),
4645 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
4646 Epoch::new(333),
4647 )),
4648 leader_timeout: Duration::from_secs(1),
4649 certification_timeout: Duration::from_secs(2),
4650 timeout_retry: Duration::from_secs(10),
4651 fetch_timeout: NZDuration!(Duration::from_secs(1)),
4652 fetch_retry_delay: Duration::from_millis(100),
4653 view_retention,
4654 skip: SkipPolicy::Enabled {
4655 timeout: skip_timeout,
4656 budget: SkipBudget::Participants,
4657 },
4658 replay_buffer: NZUsize!(1024 * 1024),
4659 write_buffer: NZUsize!(1024 * 1024),
4660 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
4661 forward: ForwardPolicy::Disabled,
4662 track_historical_votes: true,
4663 };
4664 let engine = Engine::new(context.child("engine"), cfg);
4665 engine.start(pending, recovered, resolver);
4666 }
4667 }
4668
4669 let mut finalizers = Vec::new();
4671 for reporter in reporters.iter_mut() {
4672 let (mut latest, mut monitor) = reporter.subscribe().await;
4673 finalizers.push(context.child("finalizer").spawn(move |_| async move {
4674 while latest < required_containers {
4675 latest = monitor.recv().await.expect("event missing");
4676 }
4677 }));
4678 }
4679 join_all(finalizers).await;
4680
4681 let byz = &participants[0];
4683 let mut count_conflicting = 0;
4684 for reporter in reporters.iter() {
4685 {
4687 let faults = reporter.faults.lock();
4688 assert_eq!(faults.len(), 1);
4689 let faulter = faults.get(byz).expect("byzantine party is not faulter");
4690 for faults in faulter.values() {
4691 for fault in faults.iter() {
4692 match fault {
4693 Activity::ConflictingNotarize(_) => {
4694 count_conflicting += 1;
4695 }
4696 Activity::ConflictingFinalize(_) => {
4697 count_conflicting += 1;
4698 }
4699 _ => panic!("unexpected fault: {fault:?}"),
4700 }
4701 }
4702 }
4703 }
4704
4705 reporter.assert_no_invalid();
4707 }
4708 assert!(count_conflicting > 0);
4709
4710 let blocked = oracle.blocked().await.unwrap();
4712 assert!(!blocked.is_empty());
4713 for (a, b) in blocked {
4714 assert_ne!(&a, byz);
4715 assert_eq!(&b, byz);
4716 }
4717 });
4718 }
4719
4720 test_for_all_fixtures!(conflicter, seeds = 5);
4721
4722 fn invalid<S, F, L>(seed: u64, mut fixture: F, elector: L)
4723 where
4724 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
4725 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
4726 L: elector::Config<S>,
4727 {
4728 let n = 4;
4730 let required_containers = View::new(50);
4731 let view_retention = ViewDelta::new(10);
4732 let skip_timeout = Duration::from_secs(11);
4733 let namespace = b"consensus".to_vec();
4734 let cfg = deterministic::Config::new()
4735 .with_seed(seed)
4736 .with_timeout(Some(Duration::from_secs(30)));
4737 let executor = deterministic::Runner::new(cfg);
4738 executor.start(|mut context| async move {
4739 let Fixture {
4741 participants,
4742 schemes,
4743 ..
4744 } = fixture(&mut context, &namespace, n);
4745
4746 let schemes: Vec<_> = schemes
4747 .into_iter()
4748 .enumerate()
4749 .map(|(idx, scheme)| {
4750 let is_byzantine = idx == 0;
4751 let behavior = if is_byzantine {
4752 wrapped::Behavior::CorruptSignature
4753 } else {
4754 wrapped::Behavior::Honest
4755 };
4756 wrapped::Scheme::new(scheme, behavior)
4757 })
4758 .collect();
4759
4760 let mut oracle = start_test_network_with_peers::<S, _>(
4761 context.child("network"),
4762 participants.clone(),
4763 true,
4764 )
4765 .await;
4766 let mut registrations = register_validators(&mut oracle, &participants).await;
4767
4768 let link = Link {
4770 latency: Duration::from_millis(10),
4771 jitter: Duration::from_millis(1),
4772 success_rate: probability!(1.0),
4773 };
4774 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
4775
4776 let elector = wrapped::Config(elector);
4778 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
4779 let mut reporters = Vec::new();
4780 for (idx_scheme, validator) in participants.iter().enumerate() {
4781 let context = context
4783 .child("validator")
4784 .with_attribute("public_key", validator);
4785
4786 let reporter_config = mocks::reporter::Config {
4787 participants: participants.clone().try_into().unwrap(),
4788 scheme: schemes[idx_scheme].clone(),
4789 elector: elector.clone(),
4790 };
4791 let reporter =
4792 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
4793 reporters.push(reporter.clone());
4794
4795 let application_cfg = mocks::application::Config::<Sha256, _> {
4796 relay: relay.clone(),
4797 me: validator.clone(),
4798 propose_latency: (10.0, 5.0),
4799 verify_latency: (10.0, 5.0),
4800 certify_latency: (10.0, 5.0),
4801 should_certify: mocks::application::Certifier::Always,
4802 };
4803 let (actor, application) = mocks::application::Application::new(
4804 context.child("application"),
4805 application_cfg,
4806 );
4807 actor.start();
4808 let blocker = oracle.control(validator.clone());
4809 let cfg = config::Config {
4810 scheme: schemes[idx_scheme].clone(),
4811 elector: elector.clone(),
4812 blocker,
4813 automaton: application.clone(),
4814 relay: application.clone(),
4815 reporter: reporter.clone(),
4816 strategy: Sequential,
4817 partition: validator.clone().to_string(),
4818 mailbox_size: NZUsize!(1024),
4819 epoch: Epoch::new(333),
4820 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
4821 Epoch::new(333),
4822 )),
4823 leader_timeout: Duration::from_secs(1),
4824 certification_timeout: Duration::from_secs(2),
4825 timeout_retry: Duration::from_secs(10),
4826 fetch_timeout: NZDuration!(Duration::from_secs(1)),
4827 fetch_retry_delay: Duration::from_millis(100),
4828 view_retention,
4829 skip: SkipPolicy::Enabled {
4830 timeout: skip_timeout,
4831 budget: SkipBudget::Participants,
4832 },
4833 replay_buffer: NZUsize!(1024 * 1024),
4834 write_buffer: NZUsize!(1024 * 1024),
4835 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
4836 forward: ForwardPolicy::Disabled,
4837 track_historical_votes: false,
4838 };
4839 let engine = Engine::new(context.child("engine"), cfg);
4840 let (pending, recovered, resolver) = registrations
4841 .remove(validator)
4842 .expect("validator should be registered");
4843 engine.start(pending, recovered, resolver);
4844 }
4845
4846 let mut finalizers = Vec::new();
4850 for reporter in reporters.iter_mut().skip(1) {
4851 let (mut latest, mut monitor) = reporter.subscribe().await;
4852 finalizers.push(context.child("finalizer").spawn(move |_| async move {
4853 while latest < required_containers {
4854 latest = monitor.recv().await.expect("event missing");
4855 }
4856 }));
4857 }
4858 join_all(finalizers).await;
4859
4860 for (i, reporter) in reporters.iter().enumerate() {
4862 reporter.assert_no_faults();
4864
4865 assert!(*reporter.invalid_votes.lock() > 0);
4868
4869 let is_byzantine = i == 0;
4872 if is_byzantine {
4873 assert!(*reporter.invalid_certificates.lock() > 0);
4874 } else {
4875 assert_eq!(*reporter.invalid_certificates.lock(), 0);
4876 }
4877 }
4878
4879 let blocked = oracle.blocked().await.unwrap();
4882 assert!(blocked.len() >= participants.len() - 1);
4883 let byz = &participants[0];
4884 for (_, b) in blocked {
4885 assert_eq!(&b, byz);
4887 }
4888 });
4889 }
4890
4891 test_for_all_fixtures!(invalid, seeds = 5);
4892
4893 fn received_certificates_are_reported<S, F, L>(mut fixture: F, elector: L)
4895 where
4896 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
4897 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
4898 L: elector::Config<S>,
4899 {
4900 let n = 4;
4901 let required_containers = View::new(10);
4902 let view_retention = ViewDelta::new(10);
4903 let skip_timeout = Duration::from_secs(11);
4904 let namespace = b"consensus".to_vec();
4905 let cfg = deterministic::Config::new()
4906 .with_seed(0)
4907 .with_timeout(Some(Duration::from_secs(30)));
4908 let executor = deterministic::Runner::new(cfg);
4909 executor.start(|mut context| async move {
4910 let Fixture {
4911 participants,
4912 schemes,
4913 ..
4914 } = fixture(&mut context, &namespace, n);
4915
4916 let mut oracle = start_test_network_with_peers::<S, _>(
4917 context.child("network"),
4918 participants.clone(),
4919 false,
4920 )
4921 .await;
4922 let mut registrations = register_validators(&mut oracle, &participants).await;
4923
4924 let link = Link {
4929 latency: Duration::from_millis(100),
4930 jitter: Duration::from_millis(1),
4931 success_rate: probability!(1.0),
4932 };
4933 fn link_graph(_: usize, i: usize, j: usize) -> bool {
4934 if i == 0 || j == 0 {
4935 return i == 1 || j == 1;
4936 }
4937 true
4938 }
4939 link_validators(
4940 &mut oracle,
4941 &participants,
4942 Action::Link(link),
4943 Some(link_graph),
4944 )
4945 .await;
4946
4947 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
4948 let mut reporters = Vec::new();
4949 for (idx_scheme, validator) in participants.iter().enumerate() {
4950 let context = context
4951 .child("validator")
4952 .with_attribute("public_key", validator);
4953 let reporter_config = mocks::reporter::Config {
4954 participants: participants.clone().try_into().unwrap(),
4955 scheme: schemes[idx_scheme].clone(),
4956 elector: elector.clone(),
4957 };
4958 let reporter =
4959 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
4960 reporters.push(reporter.clone());
4961
4962 let application_cfg = mocks::application::Config::<Sha256, _> {
4963 relay: relay.clone(),
4964 me: validator.clone(),
4965 propose_latency: (10.0, 5.0),
4966 verify_latency: (10.0, 5.0),
4967 certify_latency: (10.0, 5.0),
4968 should_certify: mocks::application::Certifier::Always,
4969 };
4970 let (actor, application) = mocks::application::Application::new(
4971 context.child("application"),
4972 application_cfg,
4973 );
4974 actor.start();
4975 let blocker = oracle.control(validator.clone());
4976 let cfg = config::Config {
4977 scheme: schemes[idx_scheme].clone(),
4978 elector: elector.clone(),
4979 blocker,
4980 automaton: application.clone(),
4981 relay: application.clone(),
4982 reporter: reporter.clone(),
4983 strategy: Sequential,
4984 partition: validator.clone().to_string(),
4985 mailbox_size: NZUsize!(1024),
4986 epoch: Epoch::new(333),
4987 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
4988 Epoch::new(333),
4989 )),
4990 leader_timeout: Duration::from_secs(1),
4991 certification_timeout: Duration::from_secs(2),
4992 timeout_retry: Duration::from_secs(10),
4993 fetch_timeout: NZDuration!(Duration::from_secs(1)),
4994 fetch_retry_delay: Duration::from_millis(100),
4995 view_retention,
4996 skip: SkipPolicy::Enabled {
4997 timeout: skip_timeout,
4998 budget: SkipBudget::Participants,
4999 },
5000 replay_buffer: NZUsize!(1024 * 1024),
5001 write_buffer: NZUsize!(1024 * 1024),
5002 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
5003 forward: ForwardPolicy::Disabled,
5004 track_historical_votes: false,
5005 };
5006 let engine = Engine::new(context.child("engine"), cfg);
5007 let (pending, recovered, resolver) = registrations
5008 .remove(validator)
5009 .expect("validator should be registered");
5010 engine.start(pending, recovered, resolver);
5011 }
5012 let excluded_reporter = reporters[0].clone();
5014 let mut honest_reporter = reporters[1].clone();
5015 let (mut honest_latest, mut honest_monitor) = honest_reporter.subscribe().await;
5016 while honest_latest < required_containers {
5017 honest_latest = honest_monitor.recv().await.expect("event missing");
5018 }
5019
5020 context.sleep(Duration::from_secs(1)).await;
5022
5023 let honest_notarized = {
5026 let notarizations = honest_reporter.notarizations.lock();
5027 View::range(View::new(1), required_containers.next())
5028 .filter(|view| notarizations.contains_key(view))
5029 .count()
5030 };
5031 let excluded_notarized = {
5032 let notarizations = excluded_reporter.notarizations.lock();
5033 View::range(View::new(1), required_containers.next())
5034 .filter(|view| notarizations.contains_key(view))
5035 .count()
5036 };
5037 assert!(
5038 excluded_notarized >= honest_notarized.saturating_sub(2),
5039 "honest_notarized: {honest_notarized}, excluded_notarized: {excluded_notarized}"
5040 );
5041
5042 let honest_finalized = {
5043 let finalizations = honest_reporter.finalizations.lock();
5044 View::range(View::new(1), required_containers.next())
5045 .filter(|view| finalizations.contains_key(view))
5046 .count()
5047 };
5048 let excluded_finalized = {
5049 let finalizations = excluded_reporter.finalizations.lock();
5050 View::range(View::new(1), required_containers.next())
5051 .filter(|view| finalizations.contains_key(view))
5052 .count()
5053 };
5054 assert!(
5055 excluded_finalized >= honest_finalized.saturating_sub(2),
5056 "honest_finalized: {honest_finalized}, excluded_finalized: {excluded_finalized}"
5057 );
5058 });
5059 }
5060
5061 test_for_all_fixtures!(received_certificates_are_reported);
5062
5063 fn survives_burst<S, F, L>(mut fixture: F, elector: L)
5064 where
5065 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
5066 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
5067 L: elector::Config<S>,
5068 {
5069 let n = 4;
5070 let epoch = Epoch::new(333);
5071 let namespace = b"mailbox_size_one_certificate_burst".to_vec();
5072 let executor = deterministic::Runner::default();
5073 executor.start(|mut context| async move {
5074 let Fixture {
5075 participants,
5076 schemes,
5077 ..
5078 } = fixture(&mut context, &namespace, n);
5079 let me = participants[0].clone();
5080 let mut oracle = start_test_network_with_peers::<S, _>(
5081 context.child("network"),
5082 participants.clone(),
5083 true,
5084 )
5085 .await;
5086 let (pending, recovered, resolver) = register_validator(&mut oracle, me.clone()).await;
5087
5088 let injector_pk = PrivateKey::from_seed(9_000_000).public_key();
5089 let (mut injector_sender, _injector_receiver) = oracle
5090 .control(injector_pk.clone())
5091 .register(1, TEST_QUOTA)
5092 .await
5093 .unwrap();
5094 let link = Link {
5095 latency: Duration::from_millis(0),
5096 jitter: Duration::from_millis(0),
5097 success_rate: probability!(1.0),
5098 };
5099 oracle
5100 .add_link(injector_pk.clone(), me.clone(), link)
5101 .await
5102 .unwrap();
5103 oracle.manager().track(
5104 1,
5105 TrackedPeers::new(
5106 Set::from_iter_dedup(std::iter::once(me.clone())),
5107 Set::from_iter_dedup(std::iter::once(injector_pk.clone())),
5108 ),
5109 );
5110 context.sleep(Duration::from_millis(1)).await;
5111
5112 let quorum = quorum(n) as usize;
5113 let notarization = |view: View, parent: View, payload: &[u8]| {
5114 let proposal =
5115 Proposal::new(Round::new(epoch, view), parent, Sha256::hash(&[payload]));
5116 let votes: Vec<_> = schemes
5117 .iter()
5118 .take(quorum)
5119 .map(|scheme| TNotarize::sign(scheme, proposal.clone()).unwrap())
5120 .collect();
5121 TNotarization::from_notarizes(&schemes[0], non_empty![@votes.iter()], &Sequential)
5122 .expect("notarization requires quorum")
5123 };
5124 let finalization = |view: View, parent: View, payload: &[u8]| {
5125 let proposal =
5126 Proposal::new(Round::new(epoch, view), parent, Sha256::hash(&[payload]));
5127 let votes: Vec<_> = schemes
5128 .iter()
5129 .take(quorum)
5130 .map(|scheme| TFinalize::sign(scheme, proposal.clone()).unwrap())
5131 .collect();
5132 TFinalization::from_finalizes(&schemes[0], non_empty![@votes.iter()], &Sequential)
5133 .expect("finalization requires quorum")
5134 };
5135
5136 for certificate in [
5138 Certificate::Notarization(notarization(View::new(1), View::zero(), b"payload-1")),
5139 Certificate::Notarization(notarization(View::new(2), View::new(1), b"payload-2")),
5140 Certificate::Notarization(notarization(View::new(3), View::new(2), b"payload-3")),
5141 Certificate::Finalization(finalization(View::new(3), View::new(2), b"payload-3")),
5142 ] {
5143 injector_sender.send(Recipients::One(me.clone()), certificate.encode(), true);
5144 }
5145
5146 let reporter_config = mocks::reporter::Config {
5147 participants: participants.clone().try_into().unwrap(),
5148 scheme: schemes[0].clone(),
5149 elector: elector.clone(),
5150 };
5151 let reporter =
5152 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
5153 let mut monitor_reporter = reporter.clone();
5154 let (mut latest, mut monitor) = monitor_reporter.subscribe().await;
5155
5156 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
5157 let application_cfg = mocks::application::Config::<Sha256, _> {
5158 relay: relay.clone(),
5159 me: me.clone(),
5160 propose_latency: (0.0, 0.0),
5161 verify_latency: (0.0, 0.0),
5162 certify_latency: (0.0, 0.0),
5163 should_certify: mocks::application::Certifier::Always,
5164 };
5165 let (mut application_actor, application) =
5166 mocks::application::Application::new(context.child("application"), application_cfg);
5167 application_actor.set_stall_proposals(true);
5168 application_actor.start();
5169
5170 let cfg = config::Config {
5171 scheme: schemes[0].clone(),
5172 elector,
5173 blocker: oracle.control(me.clone()),
5174 automaton: application.clone(),
5175 relay: application,
5176 reporter,
5177 strategy: Sequential,
5178 partition: me.to_string(),
5179 mailbox_size: NZUsize!(1),
5180 epoch,
5181 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(epoch)),
5182 leader_timeout: Duration::from_secs(1),
5183 certification_timeout: Duration::from_secs(2),
5184 timeout_retry: Duration::from_secs(10),
5185 fetch_timeout: NZDuration!(Duration::from_secs(1)),
5186 fetch_retry_delay: Duration::from_millis(100),
5187 view_retention: ViewDelta::new(10),
5188 skip: SkipPolicy::Enabled {
5189 timeout: Duration::from_secs(11),
5190 budget: SkipBudget::Participants,
5191 },
5192 replay_buffer: NZUsize!(1024 * 1024),
5193 write_buffer: NZUsize!(1024 * 1024),
5194 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
5195 forward: ForwardPolicy::Disabled,
5196 track_historical_votes: false,
5197 };
5198 let engine = Engine::new(context.child("engine"), cfg);
5199 engine.start(pending, recovered, resolver);
5200
5201 while latest < View::new(3) {
5202 latest = monitor.recv().await.expect("finalization event missing");
5203 }
5204 });
5205 }
5206
5207 test_for_all_fixtures!(survives_burst);
5208
5209 fn impersonator<S, F, L>(seed: u64, mut fixture: F, elector: L)
5210 where
5211 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
5212 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
5213 L: elector::Config<S>,
5214 {
5215 let n = 4;
5217 let required_containers = View::new(50);
5218 let view_retention = ViewDelta::new(10);
5219 let skip_timeout = Duration::from_secs(11);
5220 let namespace = b"consensus".to_vec();
5221 let cfg = deterministic::Config::new()
5222 .with_seed(seed)
5223 .with_timeout(Some(Duration::from_secs(30)));
5224 let executor = deterministic::Runner::new(cfg);
5225 executor.start(|mut context| async move {
5226 let Fixture {
5228 participants,
5229 schemes,
5230 ..
5231 } = fixture(&mut context, &namespace, n);
5232 let mut oracle = start_test_network_with_peers::<S, _>(
5233 context.child("network"),
5234 participants.clone(),
5235 true,
5236 )
5237 .await;
5238 let mut registrations = register_validators(&mut oracle, &participants).await;
5239
5240 let link = Link {
5242 latency: Duration::from_millis(10),
5243 jitter: Duration::from_millis(1),
5244 success_rate: probability!(1.0),
5245 };
5246 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
5247
5248 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
5250 let mut reporters = Vec::new();
5251 for (idx_scheme, validator) in participants.iter().enumerate() {
5252 let context = context
5254 .child("validator")
5255 .with_attribute("public_key", validator);
5256
5257 let reporter_config = mocks::reporter::Config {
5259 participants: participants.clone().try_into().unwrap(),
5260 scheme: schemes[idx_scheme].clone(),
5261 elector: elector.clone(),
5262 };
5263 let reporter =
5264 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
5265 let (pending, recovered, resolver) = registrations
5266 .remove(validator)
5267 .expect("validator should be registered");
5268 if idx_scheme == 0 {
5269 let cfg = mocks::impersonator::Config {
5270 scheme: schemes[idx_scheme].clone(),
5271 };
5272
5273 let engine: mocks::impersonator::Impersonator<_, _, Sha256> =
5274 mocks::impersonator::Impersonator::new(
5275 context.child("byzantine_engine"),
5276 cfg,
5277 );
5278 engine.start(pending);
5279 } else {
5280 reporters.push(reporter.clone());
5281 let application_cfg = mocks::application::Config::<Sha256, _> {
5282 relay: relay.clone(),
5283 me: validator.clone(),
5284 propose_latency: (10.0, 5.0),
5285 verify_latency: (10.0, 5.0),
5286 certify_latency: (10.0, 5.0),
5287 should_certify: mocks::application::Certifier::Always,
5288 };
5289 let (actor, application) = mocks::application::Application::new(
5290 context.child("application"),
5291 application_cfg,
5292 );
5293 actor.start();
5294 let blocker = oracle.control(validator.clone());
5295 let cfg = config::Config {
5296 scheme: schemes[idx_scheme].clone(),
5297 elector: elector.clone(),
5298 blocker,
5299 automaton: application.clone(),
5300 relay: application.clone(),
5301 reporter: reporter.clone(),
5302 strategy: Sequential,
5303 partition: validator.clone().to_string(),
5304 mailbox_size: NZUsize!(1024),
5305 epoch: Epoch::new(333),
5306 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
5307 Epoch::new(333),
5308 )),
5309 leader_timeout: Duration::from_secs(1),
5310 certification_timeout: Duration::from_secs(2),
5311 timeout_retry: Duration::from_secs(10),
5312 fetch_timeout: NZDuration!(Duration::from_secs(1)),
5313 fetch_retry_delay: Duration::from_millis(100),
5314 view_retention,
5315 skip: SkipPolicy::Enabled {
5316 timeout: skip_timeout,
5317 budget: SkipBudget::Participants,
5318 },
5319 replay_buffer: NZUsize!(1024 * 1024),
5320 write_buffer: NZUsize!(1024 * 1024),
5321 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
5322 forward: ForwardPolicy::Disabled,
5323 track_historical_votes: false,
5324 };
5325 let engine = Engine::new(context.child("engine"), cfg);
5326 engine.start(pending, recovered, resolver);
5327 }
5328 }
5329
5330 let mut finalizers = Vec::new();
5332 for reporter in reporters.iter_mut() {
5333 let (mut latest, mut monitor) = reporter.subscribe().await;
5334 finalizers.push(context.child("finalizer").spawn(move |_| async move {
5335 while latest < required_containers {
5336 latest = monitor.recv().await.expect("event missing");
5337 }
5338 }));
5339 }
5340 join_all(finalizers).await;
5341
5342 let byz = &participants[0];
5344 for reporter in reporters.iter() {
5345 reporter.assert_no_faults();
5347
5348 reporter.assert_no_invalid();
5350 }
5351
5352 let blocked = oracle.blocked().await.unwrap();
5354 assert!(!blocked.is_empty());
5355 for (a, b) in blocked {
5356 assert_ne!(&a, byz);
5357 assert_eq!(&b, byz);
5358 }
5359 });
5360 }
5361
5362 test_for_all_fixtures!(impersonator, seeds = 5);
5363
5364 fn equivocator_seeded<S, F, L>(seed: u64, fixture: F, elector: L) -> bool
5365 where
5366 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
5367 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
5368 L: elector::Config<S>,
5369 {
5370 equivocator_seeded_with_term::<S, F, L>(seed, elector, fixture)
5371 }
5372
5373 fn equivocator_seeded_with_term<S, F, L>(seed: u64, elector: L, mut fixture: F) -> bool
5374 where
5375 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
5376 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
5377 L: elector::Config<S>,
5378 {
5379 let n = 7;
5381 let required_containers = View::new(50);
5382 let view_retention = ViewDelta::new(10);
5383 let skip_timeout = Duration::from_secs(11);
5384 let namespace = b"consensus".to_vec();
5385 let cfg = deterministic::Config::new()
5386 .with_seed(seed)
5387 .with_timeout(Some(Duration::from_secs(60)));
5388 let executor = deterministic::Runner::new(cfg);
5389 executor.start(|mut context| async move {
5390 let Fixture {
5392 participants,
5393 schemes,
5394 ..
5395 } = fixture(&mut context, &namespace, n);
5396 let mut oracle = start_test_network_with_peers::<S, _>(
5397 context.child("network"),
5398 participants.clone(),
5399 true,
5400 )
5401 .await;
5402 let mut registrations = register_validators(&mut oracle, &participants).await;
5403
5404 let link = Link {
5406 latency: Duration::from_millis(10),
5407 jitter: Duration::from_millis(1),
5408 success_rate: probability!(1.0),
5409 };
5410 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
5411
5412 let elector = elector.clone();
5414 let mut engines = Vec::new();
5415 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
5416 let mut reporters = Vec::new();
5417 for (idx_scheme, validator) in participants.iter().enumerate() {
5418 let context = context
5420 .child("validator")
5421 .with_attribute("public_key", validator);
5422
5423 let reporter_config = mocks::reporter::Config {
5425 participants: participants.clone().try_into().unwrap(),
5426 scheme: schemes[idx_scheme].clone(),
5427 elector: elector.clone(),
5428 };
5429 let reporter =
5430 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
5431 reporters.push(reporter.clone());
5432 let (pending, recovered, resolver) = registrations
5433 .remove(validator)
5434 .expect("validator should be registered");
5435 if idx_scheme == 0 {
5436 let cfg = mocks::equivocator::Config::<_, _, Sha256> {
5437 scheme: schemes[idx_scheme].clone(),
5438 epoch: Epoch::new(333),
5439 relay: relay.clone(),
5440 elector: elector.clone(),
5441 };
5442
5443 let engine = mocks::equivocator::Equivocator::new(
5444 context.child("byzantine_engine"),
5445 cfg,
5446 );
5447 engines.push(engine.start(pending, recovered));
5448 } else {
5449 let application_cfg = mocks::application::Config::<Sha256, _> {
5450 relay: relay.clone(),
5451 me: validator.clone(),
5452 propose_latency: (10.0, 5.0),
5453 verify_latency: (10.0, 5.0),
5454 certify_latency: (10.0, 5.0),
5455 should_certify: mocks::application::Certifier::Always,
5456 };
5457 let (actor, application) = mocks::application::Application::new(
5458 context.child("application"),
5459 application_cfg,
5460 );
5461 actor.start();
5462 let blocker = oracle.control(validator.clone());
5463 let cfg = config::Config {
5464 scheme: schemes[idx_scheme].clone(),
5465 elector: elector.clone(),
5466 blocker,
5467 automaton: application.clone(),
5468 relay: application.clone(),
5469 reporter: reporter.clone(),
5470 strategy: Sequential,
5471 partition: validator.to_string(),
5472 mailbox_size: NZUsize!(1024),
5473 epoch: Epoch::new(333),
5474 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
5475 Epoch::new(333),
5476 )),
5477 leader_timeout: Duration::from_secs(1),
5478 certification_timeout: Duration::from_secs(2),
5479 timeout_retry: Duration::from_secs(10),
5480 fetch_timeout: NZDuration!(Duration::from_secs(1)),
5481 fetch_retry_delay: Duration::from_millis(100),
5482 view_retention,
5483 skip: SkipPolicy::Enabled {
5484 timeout: skip_timeout,
5485 budget: SkipBudget::Participants,
5486 },
5487 replay_buffer: NZUsize!(1024 * 1024),
5488 write_buffer: NZUsize!(1024 * 1024),
5489 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
5490 forward: ForwardPolicy::Disabled,
5491 track_historical_votes: false,
5492 };
5493 let engine = Engine::new(context.child("engine"), cfg);
5494 engines.push(engine.start(pending, recovered, resolver));
5495 }
5496 }
5497
5498 let mut finalizers = Vec::new();
5500 for reporter in reporters.iter_mut().skip(1) {
5501 let (mut latest, mut monitor) = reporter.subscribe().await;
5502 finalizers.push(context.child("finalizer").spawn(move |_| async move {
5503 while latest < required_containers {
5504 latest = monitor.recv().await.expect("event missing");
5505 }
5506 }));
5507 }
5508 join_all(finalizers).await;
5509
5510 let idx = context.random_range(1..engines.len()); let validator = &participants[idx];
5513 let handle = engines.remove(idx);
5514 handle.abort();
5515 let _ = handle.await;
5516 reporters.remove(idx);
5517 info!(idx, ?validator, "aborted validator");
5518
5519 let mut finalizers = Vec::new();
5521 for reporter in reporters.iter_mut().skip(1) {
5522 let (mut latest, mut monitor) = reporter.subscribe().await;
5523 finalizers.push(context.child("finalizer").spawn(move |_| async move {
5524 while latest < View::new(required_containers.get() * 2) {
5525 latest = monitor.recv().await.expect("event missing");
5526 }
5527 }));
5528 }
5529 join_all(finalizers).await;
5530
5531 info!(idx, ?validator, "restarting validator");
5533 let context = context
5534 .child("validator_restarted")
5535 .with_attribute("public_key", validator);
5536
5537 let reporter_config = mocks::reporter::Config {
5539 participants: participants.clone().try_into().unwrap(),
5540 scheme: schemes[idx].clone(),
5541 elector: elector.clone(),
5542 };
5543 let reporter =
5544 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
5545 let (pending, recovered, resolver) =
5546 register_validator(&mut oracle, validator.clone()).await;
5547 reporters.push(reporter.clone());
5548 let application_cfg = mocks::application::Config::<Sha256, _> {
5549 relay: relay.clone(),
5550 me: validator.clone(),
5551 propose_latency: (10.0, 5.0),
5552 verify_latency: (10.0, 5.0),
5553 certify_latency: (10.0, 5.0),
5554 should_certify: mocks::application::Certifier::Always,
5555 };
5556 let (actor, application) =
5557 mocks::application::Application::new(context.child("application"), application_cfg);
5558 actor.start();
5559 let blocker = oracle.control(validator.clone());
5560 let cfg = config::Config {
5561 scheme: schemes[idx].clone(),
5562 elector: elector.clone(),
5563 blocker,
5564 automaton: application.clone(),
5565 relay: application.clone(),
5566 reporter: reporter.clone(),
5567 strategy: Sequential,
5568 partition: validator.to_string(),
5569 mailbox_size: NZUsize!(1024),
5570 epoch: Epoch::new(333),
5571 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(Epoch::new(
5572 333,
5573 ))),
5574 leader_timeout: Duration::from_secs(1),
5575 certification_timeout: Duration::from_secs(2),
5576 timeout_retry: Duration::from_secs(10),
5577 fetch_timeout: NZDuration!(Duration::from_secs(1)),
5578 fetch_retry_delay: Duration::from_millis(100),
5579 view_retention,
5580 skip: SkipPolicy::Enabled {
5581 timeout: skip_timeout,
5582 budget: SkipBudget::Participants,
5583 },
5584 replay_buffer: NZUsize!(1024 * 1024),
5585 write_buffer: NZUsize!(1024 * 1024),
5586 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
5587 forward: ForwardPolicy::Disabled,
5588 track_historical_votes: false,
5589 };
5590 let engine = Engine::new(context.child("engine"), cfg);
5591 engine.start(pending, recovered, resolver);
5592
5593 let mut finalizers = Vec::new();
5595 for reporter in reporters.iter_mut().skip(1) {
5596 let (mut latest, mut monitor) = reporter.subscribe().await;
5597 finalizers.push(context.child("finalizer").spawn(move |_| async move {
5598 while latest < View::new(required_containers.get() * 3) {
5599 latest = monitor.recv().await.expect("event missing");
5600 }
5601 }));
5602 }
5603 join_all(finalizers).await;
5604
5605 let byz = &participants[0];
5609 let blocked = oracle.blocked().await.unwrap();
5610 for (a, b) in &blocked {
5611 assert_ne!(a, byz);
5612 assert_eq!(b, byz);
5613 }
5614 !blocked.is_empty()
5615 })
5616 }
5617
5618 fn equivocator<S, F, L>(fixture: F, elector: L)
5619 where
5620 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
5621 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S> + Copy,
5622 L: elector::Config<S>,
5623 {
5624 let detected =
5625 (0..5).any(|seed| equivocator_seeded::<_, _, L>(seed, fixture, elector.clone()));
5626 assert!(
5627 detected,
5628 "expected at least one seed to detect equivocation"
5629 );
5630 }
5631
5632 test_for_all_fixtures!(equivocator);
5633
5634 #[test_group("slow")]
5635 #[test_traced]
5636 fn test_equivocator_stable_leader_optimistic() {
5637 let detected = (0..5).any(|seed| {
5638 equivocator_seeded_with_term::<_, _, RoundRobin>(
5639 seed,
5640 RoundRobin::default().with_term(
5641 TermLength::new(NZU32!(5)),
5642 Duration::from_secs(13),
5643 ViewDelta::new(2),
5644 ),
5645 ed25519::fixture,
5646 )
5647 });
5648 assert!(
5649 detected,
5650 "expected at least one seed to detect equivocation"
5651 );
5652 }
5653
5654 fn reconfigurer<S, F, L>(seed: u64, mut fixture: F, elector: L)
5655 where
5656 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
5657 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
5658 L: elector::Config<S>,
5659 {
5660 let n = 4;
5662 let required_containers = View::new(50);
5663 let view_retention = ViewDelta::new(10);
5664 let skip_timeout = Duration::from_secs(11);
5665 let namespace = b"consensus".to_vec();
5666 let cfg = deterministic::Config::new()
5667 .with_seed(seed)
5668 .with_timeout(Some(Duration::from_secs(30)));
5669 let executor = deterministic::Runner::new(cfg);
5670 executor.start(|mut context| async move {
5671 let Fixture {
5673 participants,
5674 schemes,
5675 ..
5676 } = fixture(&mut context, &namespace, n);
5677 let mut oracle = start_test_network_with_peers::<S, _>(
5678 context.child("network"),
5679 participants.clone(),
5680 true,
5681 )
5682 .await;
5683 let mut registrations = register_validators(&mut oracle, &participants).await;
5684
5685 let link = Link {
5687 latency: Duration::from_millis(10),
5688 jitter: Duration::from_millis(1),
5689 success_rate: probability!(1.0),
5690 };
5691 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
5692
5693 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
5695 let mut reporters = Vec::new();
5696 for (idx_scheme, validator) in participants.iter().enumerate() {
5697 let context = context
5699 .child("validator")
5700 .with_attribute("public_key", validator);
5701
5702 let reporter_config = mocks::reporter::Config {
5704 participants: participants.clone().try_into().unwrap(),
5705 scheme: schemes[idx_scheme].clone(),
5706 elector: elector.clone(),
5707 };
5708 let reporter =
5709 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
5710 let (pending, recovered, resolver) = registrations
5711 .remove(validator)
5712 .expect("validator should be registered");
5713 if idx_scheme == 0 {
5714 let cfg = mocks::reconfigurer::Config {
5715 scheme: schemes[idx_scheme].clone(),
5716 };
5717 let engine: mocks::reconfigurer::Reconfigurer<_, _, Sha256> =
5718 mocks::reconfigurer::Reconfigurer::new(
5719 context.child("byzantine_engine"),
5720 cfg,
5721 );
5722 engine.start(pending);
5723 } else {
5724 reporters.push(reporter.clone());
5725 let application_cfg = mocks::application::Config::<Sha256, _> {
5726 relay: relay.clone(),
5727 me: validator.clone(),
5728 propose_latency: (10.0, 5.0),
5729 verify_latency: (10.0, 5.0),
5730 certify_latency: (10.0, 5.0),
5731 should_certify: mocks::application::Certifier::Always,
5732 };
5733 let (actor, application) = mocks::application::Application::new(
5734 context.child("application"),
5735 application_cfg,
5736 );
5737 actor.start();
5738 let blocker = oracle.control(validator.clone());
5739 let cfg = config::Config {
5740 scheme: schemes[idx_scheme].clone(),
5741 elector: elector.clone(),
5742 blocker,
5743 automaton: application.clone(),
5744 relay: application.clone(),
5745 reporter: reporter.clone(),
5746 strategy: Sequential,
5747 partition: validator.to_string(),
5748 mailbox_size: NZUsize!(1024),
5749 epoch: Epoch::new(333),
5750 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
5751 Epoch::new(333),
5752 )),
5753 leader_timeout: Duration::from_secs(1),
5754 certification_timeout: Duration::from_secs(2),
5755 timeout_retry: Duration::from_secs(10),
5756 fetch_timeout: NZDuration!(Duration::from_secs(1)),
5757 fetch_retry_delay: Duration::from_millis(100),
5758 view_retention,
5759 skip: SkipPolicy::Enabled {
5760 timeout: skip_timeout,
5761 budget: SkipBudget::Participants,
5762 },
5763 replay_buffer: NZUsize!(1024 * 1024),
5764 write_buffer: NZUsize!(1024 * 1024),
5765 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
5766 forward: ForwardPolicy::Disabled,
5767 track_historical_votes: false,
5768 };
5769 let engine = Engine::new(context.child("engine"), cfg);
5770 engine.start(pending, recovered, resolver);
5771 }
5772 }
5773
5774 let mut finalizers = Vec::new();
5776 for reporter in reporters.iter_mut() {
5777 let (mut latest, mut monitor) = reporter.subscribe().await;
5778 finalizers.push(context.child("finalizer").spawn(move |_| async move {
5779 while latest < required_containers {
5780 latest = monitor.recv().await.expect("event missing");
5781 }
5782 }));
5783 }
5784 join_all(finalizers).await;
5785
5786 let byz = &participants[0];
5788 for reporter in reporters.iter() {
5789 reporter.assert_no_faults();
5791
5792 reporter.assert_no_invalid();
5794 }
5795
5796 let blocked = oracle.blocked().await.unwrap();
5798 assert!(!blocked.is_empty());
5799 for (a, b) in blocked {
5800 assert_ne!(&a, byz);
5801 assert_eq!(&b, byz);
5802 }
5803 });
5804 }
5805
5806 test_for_all_fixtures!(reconfigurer, seeds = 5);
5807
5808 fn nuller<S, F, L>(seed: u64, mut fixture: F, elector: L)
5809 where
5810 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
5811 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
5812 L: elector::Config<S>,
5813 {
5814 let n = 4;
5816 let required_containers = View::new(50);
5817 let view_retention = ViewDelta::new(10);
5818 let skip_timeout = Duration::from_secs(11);
5819 let namespace = b"consensus".to_vec();
5820 let cfg = deterministic::Config::new()
5821 .with_seed(seed)
5822 .with_timeout(Some(Duration::from_secs(30)));
5823 let executor = deterministic::Runner::new(cfg);
5824 executor.start(|mut context| async move {
5825 let Fixture {
5827 participants,
5828 schemes,
5829 ..
5830 } = fixture(&mut context, &namespace, n);
5831 let mut oracle = start_test_network_with_peers::<S, _>(
5832 context.child("network"),
5833 participants.clone(),
5834 true,
5835 )
5836 .await;
5837 let mut registrations = register_validators(&mut oracle, &participants).await;
5838
5839 let link = Link {
5841 latency: Duration::from_millis(10),
5842 jitter: Duration::from_millis(1),
5843 success_rate: probability!(1.0),
5844 };
5845 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
5846
5847 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
5849 let mut reporters = Vec::new();
5850 for (idx_scheme, validator) in participants.iter().enumerate() {
5851 let context = context
5853 .child("validator")
5854 .with_attribute("public_key", validator);
5855
5856 let reporter_config = mocks::reporter::Config {
5858 participants: participants.clone().try_into().unwrap(),
5859 scheme: schemes[idx_scheme].clone(),
5860 elector: elector.clone(),
5861 };
5862 let reporter =
5863 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
5864 let (pending, recovered, resolver) = registrations
5865 .remove(validator)
5866 .expect("validator should be registered");
5867 if idx_scheme == 0 {
5868 let cfg = mocks::nuller::Config {
5869 scheme: schemes[idx_scheme].clone(),
5870 };
5871 let engine: mocks::nuller::Nuller<_, _, Sha256> =
5872 mocks::nuller::Nuller::new(context.child("byzantine_engine"), cfg);
5873 engine.start(pending);
5874 } else {
5875 reporters.push(reporter.clone());
5876 let application_cfg = mocks::application::Config::<Sha256, _> {
5877 relay: relay.clone(),
5878 me: validator.clone(),
5879 propose_latency: (10.0, 5.0),
5880 verify_latency: (10.0, 5.0),
5881 certify_latency: (10.0, 5.0),
5882 should_certify: mocks::application::Certifier::Always,
5883 };
5884 let (actor, application) = mocks::application::Application::new(
5885 context.child("application"),
5886 application_cfg,
5887 );
5888 actor.start();
5889 let blocker = oracle.control(validator.clone());
5890 let cfg = config::Config {
5891 scheme: schemes[idx_scheme].clone(),
5892 elector: elector.clone(),
5893 blocker,
5894 automaton: application.clone(),
5895 relay: application.clone(),
5896 reporter: reporter.clone(),
5897 strategy: Sequential,
5898 partition: validator.clone().to_string(),
5899 mailbox_size: NZUsize!(1024),
5900 epoch: Epoch::new(333),
5901 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
5902 Epoch::new(333),
5903 )),
5904 leader_timeout: Duration::from_secs(1),
5905 certification_timeout: Duration::from_secs(2),
5906 timeout_retry: Duration::from_secs(10),
5907 fetch_timeout: NZDuration!(Duration::from_secs(1)),
5908 fetch_retry_delay: Duration::from_millis(100),
5909 view_retention,
5910 skip: SkipPolicy::Enabled {
5911 timeout: skip_timeout,
5912 budget: SkipBudget::Participants,
5913 },
5914 replay_buffer: NZUsize!(1024 * 1024),
5915 write_buffer: NZUsize!(1024 * 1024),
5916 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
5917 forward: ForwardPolicy::Disabled,
5918 track_historical_votes: true,
5919 };
5920 let engine = Engine::new(context.child("engine"), cfg);
5921 engine.start(pending, recovered, resolver);
5922 }
5923 }
5924
5925 let mut finalizers = Vec::new();
5927 for reporter in reporters.iter_mut() {
5928 let (mut latest, mut monitor) = reporter.subscribe().await;
5929 finalizers.push(context.child("finalizer").spawn(move |_| async move {
5930 while latest < required_containers {
5931 latest = monitor.recv().await.expect("event missing");
5932 }
5933 }));
5934 }
5935 join_all(finalizers).await;
5936
5937 let byz = &participants[0];
5939 let mut count_nullify_and_finalize = 0;
5940 for reporter in reporters.iter() {
5941 {
5943 let faults = reporter.faults.lock();
5944 assert_eq!(faults.len(), 1);
5945 let faulter = faults.get(byz).expect("byzantine party is not faulter");
5946 for faults in faulter.values() {
5947 for fault in faults.iter() {
5948 match fault {
5949 Activity::NullifyFinalize(_) => {
5950 count_nullify_and_finalize += 1;
5951 }
5952 _ => panic!("unexpected fault: {fault:?}"),
5953 }
5954 }
5955 }
5956 }
5957
5958 reporter.assert_no_invalid();
5960 }
5961 assert!(count_nullify_and_finalize > 0);
5962
5963 let blocked = oracle.blocked().await.unwrap();
5965 assert!(!blocked.is_empty());
5966 for (a, b) in blocked {
5967 assert_ne!(&a, byz);
5968 assert_eq!(&b, byz);
5969 }
5970 });
5971 }
5972
5973 test_for_all_fixtures!(nuller, seeds = 5);
5974
5975 fn outdated<S, F, L>(seed: u64, mut fixture: F, elector: L)
5976 where
5977 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
5978 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
5979 L: elector::Config<S>,
5980 {
5981 let n = 4;
5983 let required_containers = View::new(100);
5984 let view_retention = ViewDelta::new(10);
5985 let skip_timeout = Duration::from_secs(11);
5986 let namespace = b"consensus".to_vec();
5987 let cfg = deterministic::Config::new()
5988 .with_seed(seed)
5989 .with_timeout(Some(Duration::from_secs(60)));
5990 let executor = deterministic::Runner::new(cfg);
5991 executor.start(|mut context| async move {
5992 let Fixture {
5994 participants,
5995 schemes,
5996 ..
5997 } = fixture(&mut context, &namespace, n);
5998 let mut oracle = start_test_network_with_peers::<S, _>(
5999 context.child("network"),
6000 participants.clone(),
6001 true,
6002 )
6003 .await;
6004 let mut registrations = register_validators(&mut oracle, &participants).await;
6005
6006 let link = Link {
6008 latency: Duration::from_millis(10),
6009 jitter: Duration::from_millis(1),
6010 success_rate: probability!(1.0),
6011 };
6012 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
6013
6014 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
6016 let mut reporters = Vec::new();
6017 for (idx_scheme, validator) in participants.iter().enumerate() {
6018 let context = context
6020 .child("validator")
6021 .with_attribute("public_key", validator);
6022
6023 let reporter_config = mocks::reporter::Config {
6025 participants: participants.clone().try_into().unwrap(),
6026 scheme: schemes[idx_scheme].clone(),
6027 elector: elector.clone(),
6028 };
6029 let reporter =
6030 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
6031 let (pending, recovered, resolver) = registrations
6032 .remove(validator)
6033 .expect("validator should be registered");
6034 if idx_scheme == 0 {
6035 let cfg = mocks::outdated::Config {
6036 scheme: schemes[idx_scheme].clone(),
6037 view_delta: ViewDelta::new(view_retention.get().saturating_mul(4)),
6038 };
6039 let engine: mocks::outdated::Outdated<_, _, Sha256> =
6040 mocks::outdated::Outdated::new(context.child("byzantine_engine"), cfg);
6041 engine.start(pending);
6042 } else {
6043 reporters.push(reporter.clone());
6044 let application_cfg = mocks::application::Config::<Sha256, _> {
6045 relay: relay.clone(),
6046 me: validator.clone(),
6047 propose_latency: (10.0, 5.0),
6048 verify_latency: (10.0, 5.0),
6049 certify_latency: (10.0, 5.0),
6050 should_certify: mocks::application::Certifier::Always,
6051 };
6052 let (actor, application) = mocks::application::Application::new(
6053 context.child("application"),
6054 application_cfg,
6055 );
6056 actor.start();
6057 let blocker = oracle.control(validator.clone());
6058 let cfg = config::Config {
6059 scheme: schemes[idx_scheme].clone(),
6060 elector: elector.clone(),
6061 blocker,
6062 automaton: application.clone(),
6063 relay: application.clone(),
6064 reporter: reporter.clone(),
6065 strategy: Sequential,
6066 partition: validator.clone().to_string(),
6067 mailbox_size: NZUsize!(1024),
6068 epoch: Epoch::new(333),
6069 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
6070 Epoch::new(333),
6071 )),
6072 leader_timeout: Duration::from_secs(1),
6073 certification_timeout: Duration::from_secs(2),
6074 timeout_retry: Duration::from_secs(10),
6075 fetch_timeout: NZDuration!(Duration::from_secs(1)),
6076 fetch_retry_delay: Duration::from_millis(100),
6077 view_retention,
6078 skip: SkipPolicy::Enabled {
6079 timeout: skip_timeout,
6080 budget: SkipBudget::Participants,
6081 },
6082 replay_buffer: NZUsize!(1024 * 1024),
6083 write_buffer: NZUsize!(1024 * 1024),
6084 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
6085 forward: ForwardPolicy::Disabled,
6086 track_historical_votes: false,
6087 };
6088 let engine = Engine::new(context.child("engine"), cfg);
6089 engine.start(pending, recovered, resolver);
6090 }
6091 }
6092
6093 let mut finalizers = Vec::new();
6095 for reporter in reporters.iter_mut() {
6096 let (mut latest, mut monitor) = reporter.subscribe().await;
6097 finalizers.push(context.child("finalizer").spawn(move |_| async move {
6098 while latest < required_containers {
6099 latest = monitor.recv().await.expect("event missing");
6100 }
6101 }));
6102 }
6103 join_all(finalizers).await;
6104
6105 for reporter in reporters.iter() {
6107 reporter.assert_no_faults();
6109
6110 reporter.assert_no_invalid();
6112 }
6113
6114 let blocked = oracle.blocked().await.unwrap();
6116 assert!(blocked.is_empty());
6117 });
6118 }
6119
6120 test_for_all_fixtures!(outdated, seeds = 5);
6121
6122 fn run_1k<S, F, L>(mut fixture: F, elector: L)
6123 where
6124 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
6125 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
6126 L: elector::Config<S>,
6127 {
6128 let n = 10;
6130 let required_containers = View::new(1_000);
6131 let view_retention = ViewDelta::new(10);
6132 let skip_timeout = Duration::from_secs(11);
6133 let namespace = b"consensus".to_vec();
6134 let cfg = deterministic::Config::new();
6135 let executor = deterministic::Runner::new(cfg);
6136 executor.start(|mut context| async move {
6137 let Fixture {
6139 participants,
6140 schemes,
6141 ..
6142 } = fixture(&mut context, &namespace, n);
6143 let mut oracle = start_test_network_with_peers::<S, _>(
6144 context.child("network"),
6145 participants.clone(),
6146 true,
6147 )
6148 .await;
6149 let mut registrations = register_validators(&mut oracle, &participants).await;
6150
6151 let link = Link {
6153 latency: Duration::from_millis(80),
6154 jitter: Duration::from_millis(10),
6155 success_rate: probability!(0.98),
6156 };
6157 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
6158
6159 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
6161 let mut reporters = Vec::new();
6162 let mut engine_handlers = Vec::new();
6163 for (idx, validator) in participants.iter().enumerate() {
6164 let context = context
6166 .child("validator")
6167 .with_attribute("public_key", validator);
6168
6169 let reporter_config = mocks::reporter::Config {
6171 participants: participants.clone().try_into().unwrap(),
6172 scheme: schemes[idx].clone(),
6173 elector: elector.clone(),
6174 };
6175 let reporter =
6176 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
6177 reporters.push(reporter.clone());
6178 let application_cfg = mocks::application::Config::<Sha256, _> {
6179 relay: relay.clone(),
6180 me: validator.clone(),
6181 propose_latency: (100.0, 50.0),
6182 verify_latency: (50.0, 40.0),
6183 certify_latency: (50.0, 40.0),
6184 should_certify: mocks::application::Certifier::Always,
6185 };
6186 let (actor, application) = mocks::application::Application::new(
6187 context.child("application"),
6188 application_cfg,
6189 );
6190 actor.start();
6191 let blocker = oracle.control(validator.clone());
6192 let cfg = config::Config {
6193 scheme: schemes[idx].clone(),
6194 elector: elector.clone(),
6195 blocker,
6196 automaton: application.clone(),
6197 relay: application.clone(),
6198 reporter: reporter.clone(),
6199 strategy: Sequential,
6200 partition: validator.to_string(),
6201 mailbox_size: NZUsize!(1024),
6202 epoch: Epoch::new(333),
6203 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
6204 Epoch::new(333),
6205 )),
6206 leader_timeout: Duration::from_secs(1),
6207 certification_timeout: Duration::from_secs(2),
6208 timeout_retry: Duration::from_secs(10),
6209 fetch_timeout: NZDuration!(Duration::from_secs(1)),
6210 fetch_retry_delay: Duration::from_millis(100),
6211 view_retention,
6212 skip: SkipPolicy::Enabled {
6213 timeout: skip_timeout,
6214 budget: SkipBudget::Participants,
6215 },
6216 replay_buffer: NZUsize!(1024 * 1024),
6217 write_buffer: NZUsize!(1024 * 1024),
6218 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
6219 forward: ForwardPolicy::Disabled,
6220 track_historical_votes: false,
6221 };
6222 let engine = Engine::new(context.child("engine"), cfg);
6223
6224 let (pending, recovered, resolver) = registrations
6226 .remove(validator)
6227 .expect("validator should be registered");
6228 engine_handlers.push(engine.start(pending, recovered, resolver));
6229 }
6230
6231 let mut finalizers = Vec::new();
6233 for reporter in reporters.iter_mut() {
6234 let (mut latest, mut monitor) = reporter.subscribe().await;
6235 finalizers.push(context.child("finalizer").spawn(move |_| async move {
6236 while latest < required_containers {
6237 latest = monitor.recv().await.expect("event missing");
6238 }
6239 }));
6240 }
6241 join_all(finalizers).await;
6242
6243 for reporter in reporters.iter() {
6245 reporter.assert_no_faults();
6247
6248 reporter.assert_no_invalid();
6250 }
6251
6252 let blocked = oracle.blocked().await.unwrap();
6254 assert!(blocked.is_empty());
6255 })
6256 }
6257
6258 #[test_group("slow")]
6259 #[test_traced]
6260 fn test_1k() {
6261 run_1k::<_, _, RoundRobin>(scheme_mocks::fixture, RoundRobin::default());
6262 }
6263
6264 fn engine_shutdown<S, F, L>(seed: u64, mut fixture: F, elector: L, graceful: bool)
6265 where
6266 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
6267 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
6268 L: elector::Config<S>,
6269 {
6270 let n = 1;
6271 let namespace = b"consensus".to_vec();
6272 let cfg = deterministic::Config::default()
6273 .with_seed(seed)
6274 .with_timeout(Some(Duration::from_secs(10)));
6275 let executor = deterministic::Runner::new(cfg);
6276 executor.start(|mut context| async move {
6277 let Fixture {
6279 participants,
6280 schemes,
6281 ..
6282 } = fixture(&mut context, &namespace, n);
6283 let mut oracle = start_test_network_with_peers::<S, _>(
6284 context.child("network"),
6285 participants.clone(),
6286 true,
6287 )
6288 .await;
6289 let mut registrations = register_validators(&mut oracle, &participants).await;
6290
6291 let link = Link {
6293 latency: Duration::from_millis(1),
6294 jitter: Duration::from_millis(0),
6295 success_rate: probability!(1.0),
6296 };
6297 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
6298
6299 let reporter_config = mocks::reporter::Config {
6301 participants: participants.clone().try_into().unwrap(),
6302 scheme: schemes[0].clone(),
6303 elector: elector.clone(),
6304 };
6305 let reporter =
6306 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
6307 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
6308 let application_cfg = mocks::application::Config::<Sha256, _> {
6309 relay: relay.clone(),
6310 me: participants[0].clone(),
6311 propose_latency: (1.0, 0.0),
6312 verify_latency: (1.0, 0.0),
6313 certify_latency: (1.0, 0.0),
6314 should_certify: mocks::application::Certifier::Always,
6315 };
6316 let (actor, application) =
6317 mocks::application::Application::new(context.child("application"), application_cfg);
6318 actor.start();
6319 let blocker = oracle.control(participants[0].clone());
6320 let cfg = config::Config {
6321 scheme: schemes[0].clone(),
6322 elector: elector.clone(),
6323 blocker,
6324 automaton: application.clone(),
6325 relay: application.clone(),
6326 reporter: reporter.clone(),
6327 strategy: Sequential,
6328 partition: participants[0].clone().to_string(),
6329 mailbox_size: NZUsize!(64),
6330 epoch: Epoch::new(333),
6331 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(Epoch::new(
6332 333,
6333 ))),
6334 leader_timeout: Duration::from_millis(50),
6335 certification_timeout: Duration::from_millis(100),
6336 timeout_retry: Duration::from_millis(250),
6337 fetch_timeout: NZDuration!(Duration::from_millis(50)),
6338 fetch_retry_delay: Duration::from_millis(50),
6339 view_retention: ViewDelta::new(4),
6340 skip: SkipPolicy::Enabled {
6341 timeout: Duration::from_secs(2),
6342 budget: SkipBudget::Participants,
6343 },
6344 replay_buffer: NZUsize!(1024 * 16),
6345 write_buffer: NZUsize!(1024 * 16),
6346 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
6347 forward: ForwardPolicy::Disabled,
6348 track_historical_votes: false,
6349 };
6350 let engine = Engine::new(context.child("engine"), cfg);
6351
6352 let (pending, recovered, resolver) = registrations
6354 .remove(&participants[0])
6355 .expect("validator should be registered");
6356 let handle = engine.start(pending, recovered, resolver);
6357
6358 context.sleep(Duration::from_millis(1000)).await;
6360
6361 let running_before = count_running_tasks(&context, "engine");
6363 assert!(
6364 running_before > 0,
6365 "at least one engine task should be running"
6366 );
6367
6368 context.sleep(Duration::from_millis(1500)).await;
6370 assert!(
6371 count_running_tasks(&context, "engine") > 0,
6372 "engine tasks should still be running"
6373 );
6374
6375 let running_after = if graceful {
6377 let result = context
6378 .child("stop")
6379 .stop(0, Some(Duration::from_secs(5)))
6380 .await;
6381 assert!(
6382 result.is_ok(),
6383 "graceful shutdown should complete: {result:?}"
6384 );
6385 count_running_tasks(&context, "engine")
6386 } else {
6387 handle.abort();
6388 let _ = handle.await; context.sleep(Duration::from_millis(1000)).await;
6392 count_running_tasks(&context, "engine")
6393 };
6394 assert_eq!(
6395 running_after, 0,
6396 "all engine tasks should be stopped, but {running_after} still running"
6397 );
6398 });
6399 }
6400
6401 fn children_shutdown_on_engine_abort<S, F, L>(seed: u64, fixture: F, elector: L)
6402 where
6403 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
6404 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
6405 L: elector::Config<S>,
6406 {
6407 engine_shutdown::<S, F, L>(seed, fixture, elector, false);
6408 }
6409
6410 test_for_all_fixtures!(children_shutdown_on_engine_abort, seeds = 10);
6411
6412 fn graceful_shutdown<S, F, L>(seed: u64, fixture: F, elector: L)
6413 where
6414 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
6415 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
6416 L: elector::Config<S>,
6417 {
6418 engine_shutdown::<S, F, L>(seed, fixture, elector, true);
6419 }
6420
6421 test_for_all_fixtures!(graceful_shutdown, seeds = 10);
6422
6423 fn attributable_reporter_filtering<S, F, L>(mut fixture: F, elector: L)
6424 where
6425 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
6426 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
6427 L: elector::Config<S>,
6428 {
6429 let n = 3;
6430 let required_containers = View::new(10);
6431 let view_retention = ViewDelta::new(10);
6432 let skip_timeout = Duration::from_secs(11);
6433 let namespace = b"consensus".to_vec();
6434 let executor = deterministic::Runner::timed(Duration::from_secs(30));
6435 executor.start(|mut context| async move {
6436 let Fixture {
6438 participants,
6439 schemes,
6440 ..
6441 } = fixture(&mut context, &namespace, n);
6442 let mut oracle = start_test_network_with_peers::<S, _>(
6443 context.child("network"),
6444 participants.clone(),
6445 false,
6446 )
6447 .await;
6448 let mut registrations = register_validators(&mut oracle, &participants).await;
6449
6450 let link = Link {
6452 latency: Duration::from_millis(10),
6453 jitter: Duration::from_millis(1),
6454 success_rate: probability!(1.0),
6455 };
6456 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
6457
6458 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
6460 let mut reporters = Vec::new();
6461 for (idx, validator) in participants.iter().enumerate() {
6462 let context = context
6463 .child("validator")
6464 .with_attribute("public_key", validator);
6465
6466 let reporter_config = mocks::reporter::Config {
6467 participants: participants.clone().try_into().unwrap(),
6468 scheme: schemes[idx].clone(),
6469 elector: elector.clone(),
6470 };
6471 let mock_reporter =
6472 mocks::reporter::Reporter::new(context.child("mock_reporter"), reporter_config);
6473
6474 let attributable_reporter = scheme::reporter::AttributableReporter::new(
6476 context.child("rng"),
6477 schemes[idx].clone(),
6478 mock_reporter.clone(),
6479 Sequential,
6480 true, );
6482 reporters.push(mock_reporter.clone());
6483
6484 let application_cfg = mocks::application::Config::<Sha256, _> {
6485 relay: relay.clone(),
6486 me: validator.clone(),
6487 propose_latency: (10.0, 5.0),
6488 verify_latency: (10.0, 5.0),
6489 certify_latency: (10.0, 5.0),
6490 should_certify: mocks::application::Certifier::Always,
6491 };
6492 let (actor, application) = mocks::application::Application::new(
6493 context.child("application"),
6494 application_cfg,
6495 );
6496 actor.start();
6497 let blocker = oracle.control(validator.clone());
6498 let cfg = config::Config {
6499 scheme: schemes[idx].clone(),
6500 elector: elector.clone(),
6501 blocker,
6502 automaton: application.clone(),
6503 relay: application.clone(),
6504 reporter: attributable_reporter,
6505 strategy: Sequential,
6506 partition: validator.to_string(),
6507 mailbox_size: NZUsize!(1024),
6508 epoch: Epoch::new(333),
6509 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
6510 Epoch::new(333),
6511 )),
6512 leader_timeout: Duration::from_secs(1),
6513 certification_timeout: Duration::from_secs(2),
6514 timeout_retry: Duration::from_secs(10),
6515 fetch_timeout: NZDuration!(Duration::from_secs(1)),
6516 fetch_retry_delay: Duration::from_millis(100),
6517 view_retention,
6518 skip: SkipPolicy::Enabled {
6519 timeout: skip_timeout,
6520 budget: SkipBudget::Participants,
6521 },
6522 replay_buffer: NZUsize!(1024 * 1024),
6523 write_buffer: NZUsize!(1024 * 1024),
6524 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
6525 forward: ForwardPolicy::Disabled,
6526 track_historical_votes: false,
6527 };
6528 let engine = Engine::new(context.child("engine"), cfg);
6529
6530 let (pending, recovered, resolver) = registrations
6532 .remove(validator)
6533 .expect("validator should be registered");
6534 engine.start(pending, recovered, resolver);
6535 }
6536
6537 let mut finalizers = Vec::new();
6539 for reporter in reporters.iter_mut() {
6540 let (mut latest, mut monitor) = reporter.subscribe().await;
6541 finalizers.push(context.child("finalizer").spawn(move |_| async move {
6542 while latest < required_containers {
6543 latest = monitor.recv().await.expect("event missing");
6544 }
6545 }));
6546 }
6547 join_all(finalizers).await;
6548
6549 for reporter in reporters.iter() {
6551 reporter.assert_no_faults();
6553
6554 reporter.assert_no_invalid();
6556
6557 {
6559 let notarizations = reporter.notarizations.lock();
6560 let finalizations = reporter.finalizations.lock();
6561 assert!(
6562 !notarizations.is_empty() || !finalizations.is_empty(),
6563 "Certificates should be reported"
6564 );
6565 }
6566
6567 let notarizes = reporter.notarizes.lock();
6569 let last_view = notarizes.keys().max().cloned().unwrap_or_default();
6570 for (view, payloads) in notarizes.iter() {
6571 if *view == last_view {
6572 continue; }
6574
6575 let signers: usize = payloads.values().map(|signers| signers.len()).sum();
6576
6577 if S::is_attributable() {
6579 assert!(signers > 1, "view {view}: {signers}");
6580 } else {
6581 assert_eq!(signers, 0);
6583 }
6584 }
6585
6586 let finalizes = reporter.finalizes.lock();
6588 for payloads in finalizes.values() {
6589 let signers: usize = payloads.values().map(|signers| signers.len()).sum();
6590
6591 if S::is_attributable() {
6593 assert!(signers > 1);
6594 } else {
6595 assert_eq!(signers, 0);
6597 }
6598 }
6599 }
6600
6601 let blocked = oracle.blocked().await.unwrap();
6603 assert!(blocked.is_empty());
6604 });
6605 }
6606
6607 test_for_all_fixtures!(attributable_reporter_filtering);
6608
6609 fn split_views_no_lockup<S, F, L>(mut fixture: F, elector: L)
6610 where
6611 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
6612 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
6613 L: elector::Config<S>,
6614 {
6615 enum ParticipantType {
6628 Group1, Group2, Ignorant, Byzantine, }
6633 let get_type = |idx: usize| -> ParticipantType {
6634 match idx {
6635 0..3 => ParticipantType::Group1,
6636 3..6 => ParticipantType::Group2,
6637 6 => ParticipantType::Ignorant,
6638 7..10 => ParticipantType::Byzantine,
6639 _ => unreachable!(),
6640 }
6641 };
6642
6643 let n = 10;
6645 let quorum = quorum(n) as usize;
6646 assert_eq!(quorum, 7);
6647 let view_retention = ViewDelta::new(10);
6648 let skip_timeout = Duration::from_secs(12);
6649 let namespace = b"consensus".to_vec();
6650 let executor = deterministic::Runner::timed(Duration::from_secs(300));
6651 executor.start(|mut context| async move {
6652 let Fixture {
6654 participants,
6655 schemes,
6656 ..
6657 } = fixture(&mut context, &namespace, n);
6658 let mut oracle = start_test_network_with_peers::<S, _>(
6659 context.child("network"),
6660 participants.clone(),
6661 false,
6662 )
6663 .await;
6664 let mut registrations = register_validators(&mut oracle, &participants).await;
6665
6666 let build_finalization = |proposal: &Proposal<D>| -> TFinalization<_, D> {
6668 let votes: Vec<_> = (0..=quorum)
6669 .map(|i| TFinalize::sign(&schemes[i], proposal.clone()).unwrap())
6670 .collect();
6671 TFinalization::from_finalizes(&schemes[0], non_empty![@&votes], &Sequential)
6672 .expect("finalization quorum")
6673 };
6674 let build_notarization = |proposal: &Proposal<D>| -> TNotarization<_, D> {
6676 let votes: Vec<_> = (0..=quorum)
6677 .map(|i| TNotarize::sign(&schemes[i], proposal.clone()).unwrap())
6678 .collect();
6679 TNotarization::from_notarizes(&schemes[0], non_empty![@&votes], &Sequential)
6680 .expect("notarization quorum")
6681 };
6682 let build_nullification = |round: Round| -> TNullification<_> {
6683 let votes: Vec<_> = (0..=quorum)
6684 .map(|i| TNullify::sign::<D>(&schemes[i], round).unwrap())
6685 .collect();
6686 TNullification::from_nullifies(&schemes[0], non_empty![@&votes], &Sequential)
6687 .expect("nullification quorum")
6688 };
6689 let f_view = 1;
6691 let round_f = Round::new(Epoch::new(333), View::new(f_view));
6692 let payload_b0 = Sha256::hash(&[b"B_F"]);
6693 let proposal_b0 = Proposal::new(round_f, View::new(f_view - 1), payload_b0);
6694 let payload_b1a = Sha256::hash(&[b"B_G1"]);
6695 let proposal_b1a = Proposal::new(
6696 Round::new(Epoch::new(333), View::new(f_view + 1)),
6697 View::new(f_view),
6698 payload_b1a,
6699 );
6700 let payload_b1b = Sha256::hash(&[b"B_G2"]);
6701 let proposal_b1b = Proposal::new(
6702 Round::new(Epoch::new(333), View::new(f_view + 2)),
6703 View::new(f_view),
6704 payload_b1b,
6705 );
6706
6707 let b0_notarization = build_notarization(&proposal_b0);
6709 let b0_finalization = build_finalization(&proposal_b0);
6710 let b1a_notarization = build_notarization(&proposal_b1a);
6712 let b1b_notarization = build_notarization(&proposal_b1b);
6713 let null_a = build_nullification(Round::new(Epoch::new(333), View::new(f_view + 1)));
6715 let null_b = build_nullification(Round::new(Epoch::new(333), View::new(f_view + 2)));
6716
6717 let link = Link {
6719 latency: Duration::from_millis(10),
6720 jitter: Duration::from_millis(0),
6721 success_rate: probability!(1.0),
6722 };
6723 let mut injector_sender =
6724 start_certificate_injector(&context, &mut oracle, &participants, &link).await;
6725
6726 let msg = Certificate::<_, D>::Notarization(b0_notarization).encode();
6728 injector_sender.send(Recipients::All, msg, true);
6729 let msg = Certificate::<_, D>::Finalization(b0_finalization).encode();
6730 injector_sender.send(Recipients::All, msg, true);
6731 let notarization_msg = Certificate::<_, D>::Notarization(b1a_notarization);
6733 let nullification_msg = Certificate::<_, D>::Nullification(null_a.clone());
6734 for (i, participant) in participants.iter().enumerate() {
6735 let recipient = Recipients::One(participant.clone());
6736 let msg = match get_type(i) {
6737 ParticipantType::Group1 => notarization_msg.encode(),
6738 _ => nullification_msg.encode(),
6739 };
6740 injector_sender.send(recipient, msg, true);
6741 }
6742 let notarization_msg = Certificate::<_, D>::Notarization(b1b_notarization);
6744 let nullification_msg = Certificate::<_, D>::Nullification(null_b.clone());
6745 for (i, participant) in participants.iter().enumerate() {
6746 let recipient = Recipients::One(participant.clone());
6747 let msg = match get_type(i) {
6748 ParticipantType::Group2 => notarization_msg.encode(),
6749 _ => nullification_msg.encode(),
6750 };
6751 injector_sender.send(recipient, msg, true);
6752 }
6753
6754 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
6758 let mut honest_reporters = Vec::new();
6759 for (idx, validator) in participants.iter().enumerate() {
6760 let (pending, recovered, resolver) = registrations
6761 .remove(validator)
6762 .expect("validator should be registered");
6763 let participant_type = get_type(idx);
6764 if matches!(participant_type, ParticipantType::Byzantine) {
6765 let cfg = mocks::nullify_only::Config {
6767 scheme: schemes[idx].clone(),
6768 };
6769 let engine: mocks::nullify_only::NullifyOnly<_, _, Sha256> =
6770 mocks::nullify_only::NullifyOnly::new(
6771 context
6772 .child("byzantine")
6773 .with_attribute("public_key", validator),
6774 cfg,
6775 );
6776 engine.start(pending);
6777 drop(recovered);
6779 drop(resolver);
6780 } else {
6781 let reporter_config = mocks::reporter::Config {
6783 participants: participants.clone().try_into().unwrap(),
6784 scheme: schemes[idx].clone(),
6785 elector: elector.clone(),
6786 };
6787 let reporter = mocks::reporter::Reporter::new(
6788 context
6789 .child("reporter")
6790 .with_attribute("public_key", validator),
6791 reporter_config,
6792 );
6793 honest_reporters.push(reporter.clone());
6794
6795 let application_cfg = mocks::application::Config::<Sha256, _> {
6796 relay: relay.clone(),
6797 me: validator.clone(),
6798 propose_latency: (250.0, 50.0), verify_latency: (10.0, 5.0),
6800 certify_latency: (10.0, 5.0),
6801 should_certify: mocks::application::Certifier::Always,
6802 };
6803 let (actor, application) = mocks::application::Application::new(
6804 context
6805 .child("application")
6806 .with_attribute("public_key", validator),
6807 application_cfg,
6808 );
6809 actor.start();
6810 let blocker = oracle.control(validator.clone());
6811 let cfg = config::Config {
6812 scheme: schemes[idx].clone(),
6813 elector: elector.clone(),
6814 blocker,
6815 automaton: application.clone(),
6816 relay: application.clone(),
6817 reporter: reporter.clone(),
6818 strategy: Sequential,
6819 partition: validator.to_string(),
6820 mailbox_size: NZUsize!(1024),
6821 epoch: Epoch::new(333),
6822 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
6823 Epoch::new(333),
6824 )),
6825 leader_timeout: Duration::from_secs(10),
6826 certification_timeout: Duration::from_secs(11),
6827 timeout_retry: Duration::from_secs(10),
6828 fetch_timeout: NZDuration!(Duration::from_secs(1)),
6829 fetch_retry_delay: Duration::from_millis(100),
6830 view_retention,
6831 skip: SkipPolicy::Enabled {
6832 timeout: skip_timeout,
6833 budget: SkipBudget::Participants,
6834 },
6835 replay_buffer: NZUsize!(1024 * 1024),
6836 write_buffer: NZUsize!(1024 * 1024),
6837 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
6838 forward: ForwardPolicy::Disabled,
6839 track_historical_votes: false,
6840 };
6841 let engine = Engine::new(
6842 context
6843 .child("engine")
6844 .with_attribute("public_key", validator),
6845 cfg,
6846 );
6847 engine.start(pending, recovered, resolver);
6848 }
6849 }
6850
6851 context.sleep(Duration::from_secs(2)).await;
6853
6854 let view = View::new(f_view);
6857 for reporter in honest_reporters.iter() {
6858 let finalizations = reporter.finalizations.lock();
6859 assert!(finalizations.contains_key(&view));
6860 }
6861
6862 let view = View::new(f_view + 1);
6866 for (i, reporter) in honest_reporters.iter().enumerate() {
6867 let finalizations = reporter.finalizations.lock();
6868 assert!(!finalizations.contains_key(&view));
6869 let nullifications = reporter.nullifications.lock();
6870 let notarizations = reporter.notarizations.lock();
6871 match get_type(i) {
6872 ParticipantType::Group1 => {
6873 assert!(notarizations.contains_key(&view));
6874 assert!(!nullifications.contains_key(&view));
6875 }
6876 _ => {
6877 assert!(nullifications.contains_key(&view));
6878 assert!(!notarizations.contains_key(&view));
6879 }
6880 }
6881 }
6882
6883 let view = View::new(f_view + 2);
6887 for (i, reporter) in honest_reporters.iter().enumerate() {
6888 let finalizations = reporter.finalizations.lock();
6889 assert!(!finalizations.contains_key(&view));
6890 let nullifications = reporter.nullifications.lock();
6891 let notarizations = reporter.notarizations.lock();
6892 match get_type(i) {
6893 ParticipantType::Group2 => {
6894 assert!(notarizations.contains_key(&view));
6895 assert!(!nullifications.contains_key(&view));
6896 }
6897 _ => {
6898 assert!(nullifications.contains_key(&view));
6899 assert!(!notarizations.contains_key(&view));
6900 }
6901 }
6902 }
6903
6904 let next_view = View::new(f_view + 3);
6906 for (i, reporter) in honest_reporters.iter().enumerate() {
6907 let nullifies = reporter.nullifies.lock();
6908 assert!(!nullifies.contains_key(&next_view), "reporter {i}");
6909 }
6910
6911 link_validators(&mut oracle, &participants, Action::Link(link.clone()), None).await;
6913
6914 {
6916 let target = View::new(f_view + 3);
6917 let mut finalizers = Vec::new();
6918 for reporter in honest_reporters.iter_mut() {
6919 let (mut latest, mut monitor) = reporter.subscribe().await;
6920 finalizers.push(
6921 context
6922 .child("resume_finalizer")
6923 .spawn(move |_| async move {
6924 while latest < target {
6925 latest = monitor.recv().await.expect("event missing");
6926 }
6927 }),
6928 );
6929 }
6930 join_all(finalizers).await;
6931 }
6932
6933 for reporter in honest_reporters.iter() {
6935 reporter.assert_no_faults();
6936 reporter.assert_no_invalid();
6937 }
6938 let blocked = oracle.blocked().await.unwrap();
6939 assert!(blocked.is_empty(), "blocked peers: {blocked:?}");
6940 });
6941 }
6942
6943 test_for_all_fixtures!(split_views_no_lockup);
6944
6945 type CertifiedSplitReporter<S, L> =
6946 mocks::reporter::Reporter<deterministic::Context, S, L, Sha256Digest>;
6947
6948 struct CertifiedSplitEngineConfig<'a, S, L> {
6949 oracle: &'a Oracle<PublicKey, deterministic::Context>,
6950 participants: &'a [PublicKey],
6951 schemes: &'a [S],
6952 registrations: &'a mut TestRegistrations,
6953 silent: usize,
6954 elector: &'a L,
6955 epoch: Epoch,
6956 view_retention: ViewDelta,
6957 skip_timeout: Duration,
6958 }
6959
6960 async fn start_certificate_injector(
6963 context: &deterministic::Context,
6964 oracle: &mut Oracle<PublicKey, deterministic::Context>,
6965 participants: &[PublicKey],
6966 link: &Link,
6967 ) -> Sender<PublicKey, deterministic::Context> {
6968 let injector = PrivateKey::from_seed(1_000_000).public_key();
6969 let (sender, _receiver) = oracle
6970 .control(injector.clone())
6971 .register(1, TEST_QUOTA)
6972 .await
6973 .unwrap();
6974 for participant in participants {
6975 oracle
6976 .add_link(injector.clone(), participant.clone(), link.clone())
6977 .await
6978 .unwrap();
6979 }
6980 oracle.manager().track(
6981 1,
6982 TrackedPeers::new(
6983 Set::from_iter_dedup(participants.iter().cloned()),
6984 Set::from_iter_dedup(std::iter::once(injector)),
6985 ),
6986 );
6987 context.sleep(Duration::from_millis(10)).await;
6988 sender
6989 }
6990
6991 fn start_certified_split_engines<S, L>(
6994 context: &deterministic::Context,
6995 cfg: CertifiedSplitEngineConfig<'_, S, L>,
6996 ) -> HashMap<usize, CertifiedSplitReporter<S, L>>
6997 where
6998 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
6999 L: elector::Config<S>,
7000 {
7001 let CertifiedSplitEngineConfig {
7002 oracle,
7003 participants,
7004 schemes,
7005 registrations,
7006 silent,
7007 elector,
7008 epoch,
7009 view_retention,
7010 skip_timeout,
7011 } = cfg;
7012 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
7013 let mut reporters = HashMap::new();
7014
7015 for (idx, validator) in participants.iter().enumerate() {
7016 let registration = registrations
7017 .remove(validator)
7018 .expect("validator should be registered");
7019 if idx == silent {
7020 drop(registration);
7021 continue;
7022 }
7023
7024 let reporter_config = mocks::reporter::Config {
7025 participants: participants.to_vec().try_into().unwrap(),
7026 scheme: schemes[idx].clone(),
7027 elector: elector.clone(),
7028 };
7029 let reporter = mocks::reporter::Reporter::new(
7030 context
7031 .child("reporter")
7032 .with_attribute("public_key", validator),
7033 reporter_config,
7034 );
7035 reporters.insert(idx, reporter.clone());
7036
7037 let application_cfg = mocks::application::Config::<Sha256, _> {
7038 relay: relay.clone(),
7039 me: validator.clone(),
7040 propose_latency: (250.0, 50.0), verify_latency: (10.0, 5.0),
7042 certify_latency: (10.0, 5.0),
7043 should_certify: mocks::application::Certifier::Always,
7044 };
7045 let (actor, application) = mocks::application::Application::new(
7046 context
7047 .child("application")
7048 .with_attribute("public_key", validator),
7049 application_cfg,
7050 );
7051 actor.start();
7052
7053 let cfg = config::Config {
7054 scheme: schemes[idx].clone(),
7055 elector: elector.clone(),
7056 blocker: oracle.control(validator.clone()),
7057 automaton: application.clone(),
7058 relay: application.clone(),
7059 reporter: reporter.clone(),
7060 strategy: Sequential,
7061 partition: validator.to_string(),
7062 mailbox_size: NZUsize!(1024),
7063 epoch,
7064 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(epoch)),
7065 leader_timeout: Duration::from_secs(10),
7066 certification_timeout: Duration::from_secs(11),
7067 timeout_retry: Duration::from_secs(10),
7068 fetch_timeout: NZDuration!(Duration::from_secs(1)),
7069 fetch_retry_delay: Duration::from_millis(100),
7070 view_retention,
7071 skip: SkipPolicy::Enabled {
7072 timeout: skip_timeout,
7073 budget: SkipBudget::Participants,
7074 },
7075 replay_buffer: NZUsize!(1024 * 1024),
7076 write_buffer: NZUsize!(1024 * 1024),
7077 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
7078 forward: ForwardPolicy::Disabled,
7079 track_historical_votes: false,
7080 };
7081 let engine = Engine::new(
7082 context
7083 .child("engine")
7084 .with_attribute("public_key", validator),
7085 cfg,
7086 );
7087 let (pending, recovered, resolver) = registration;
7088 engine.start(pending, recovered, resolver);
7089 }
7090
7091 reporters
7092 }
7093
7094 fn certified_split_heals_in_group_led_view<S, F, L>(mut fixture: F, elector: L)
7101 where
7102 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
7103 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
7104 L: elector::Config<S>,
7105 L::Elector: Elector<S, Mode = Scheduled>,
7106 {
7107 let n = 4;
7108 let quorum = quorum(n) as usize;
7109 assert_eq!(quorum, 3);
7110 let view_retention = ViewDelta::new(10);
7111 let skip_timeout = Duration::from_secs(12);
7112 let namespace = b"consensus".to_vec();
7113 let executor = deterministic::Runner::timed(Duration::from_secs(300));
7114 executor.start(|mut context| async move {
7115 let Fixture {
7116 participants,
7117 schemes,
7118 ..
7119 } = fixture(&mut context, &namespace, n);
7120 let mut oracle = start_test_network_with_peers::<S, _>(
7121 context.child("network"),
7122 participants.clone(),
7123 false,
7124 )
7125 .await;
7126 let mut registrations = register_validators(&mut oracle, &participants).await;
7127
7128 let epoch = Epoch::new(333);
7132 let participant_set: Set<PublicKey> = participants.clone().try_into().unwrap();
7133 let schedule = elector.clone().build(&participant_set);
7134 let leader_of =
7135 |view: u64| usize::from(schedule.elect(Round::new(epoch, View::new(view)), ()));
7136 let byzantine = leader_of(10);
7137 let group = [leader_of(11), leader_of(12)];
7138 let lone = leader_of(13);
7139 assert!(!group.contains(&byzantine) && !group.contains(&lone) && byzantine != lone);
7140
7141 let build_notarization = |proposal: &Proposal<D>| -> TNotarization<_, D> {
7143 let votes: Vec<_> = (0..quorum)
7144 .map(|i| TNotarize::sign(&schemes[i], proposal.clone()).unwrap())
7145 .collect();
7146 TNotarization::from_notarizes(&schemes[0], non_empty![@&votes], &Sequential)
7147 .expect("notarization quorum")
7148 };
7149 let build_finalization = |proposal: &Proposal<D>| -> TFinalization<_, D> {
7150 let votes: Vec<_> = (0..quorum)
7151 .map(|i| TFinalize::sign(&schemes[i], proposal.clone()).unwrap())
7152 .collect();
7153 TFinalization::from_finalizes(&schemes[0], non_empty![@&votes], &Sequential)
7154 .expect("finalization quorum")
7155 };
7156 let build_nullification = |view: u64| -> TNullification<_> {
7157 let round = Round::new(epoch, View::new(view));
7158 let votes: Vec<_> = (0..quorum)
7159 .map(|i| TNullify::sign::<D>(&schemes[i], round).unwrap())
7160 .collect();
7161 TNullification::from_nullifies(&schemes[0], non_empty![@&votes], &Sequential)
7162 .expect("nullification quorum")
7163 };
7164 let payload_b2 = Sha256::hash(&[b"B_2"]);
7165 let proposal_b2 =
7166 Proposal::new(Round::new(epoch, View::new(2)), View::new(1), payload_b2);
7167 let payload_b3 = Sha256::hash(&[b"B_3"]);
7168 let proposal_b3 =
7169 Proposal::new(Round::new(epoch, View::new(3)), View::new(2), payload_b3);
7170 let b2_notarization = build_notarization(&proposal_b2);
7171 let b2_finalization = build_finalization(&proposal_b2);
7172 let b3_notarization = build_notarization(&proposal_b3);
7173 let null_3 = build_nullification(3);
7174
7175 let link = Link {
7176 latency: Duration::from_millis(10),
7177 jitter: Duration::from_millis(0),
7178 success_rate: probability!(1.0),
7179 };
7180 let mut injector_sender =
7181 start_certificate_injector(&context, &mut oracle, &participants, &link).await;
7182
7183 let msg = Certificate::<_, D>::Notarization(b2_notarization).encode();
7185 injector_sender.send(Recipients::All, msg, true);
7186 let msg = Certificate::<_, D>::Finalization(b2_finalization).encode();
7187 injector_sender.send(Recipients::All, msg, true);
7188 let msg = Certificate::<_, D>::Notarization(b3_notarization).encode();
7189 injector_sender.send(Recipients::One(participants[lone].clone()), msg, true);
7190 let msg = Certificate::<_, D>::Nullification(null_3).encode();
7191 for idx in group {
7192 injector_sender.send(
7193 Recipients::One(participants[idx].clone()),
7194 msg.clone(),
7195 true,
7196 );
7197 }
7198 for view in 4..=9 {
7199 let msg = Certificate::<_, D>::Nullification(build_nullification(view)).encode();
7200 injector_sender.send(Recipients::All, msg, true);
7201 }
7202
7203 let mut honest_reporters = start_certified_split_engines(
7205 &context,
7206 CertifiedSplitEngineConfig {
7207 oracle: &oracle,
7208 participants: &participants,
7209 schemes: &schemes,
7210 registrations: &mut registrations,
7211 silent: byzantine,
7212 elector: &elector,
7213 epoch,
7214 view_retention,
7215 skip_timeout,
7216 },
7217 );
7218
7219 context.sleep(Duration::from_secs(2)).await;
7221
7222 let view_2 = View::new(2);
7224 let view_3 = View::new(3);
7225 for (idx, reporter) in honest_reporters.iter() {
7226 assert!(
7227 reporter.finalizations.lock().contains_key(&view_2),
7228 "reporter {idx} missing finalization for view 2"
7229 );
7230 let notarizations = reporter.notarizations.lock();
7231 let nullifications = reporter.nullifications.lock();
7232 if *idx == lone {
7233 assert!(notarizations.contains_key(&view_3));
7234 assert!(!nullifications.contains_key(&view_3));
7235 assert!(reporter.certifications.lock().contains_key(&view_3));
7236 } else {
7237 assert!(nullifications.contains_key(&view_3), "reporter {idx}");
7238 assert!(!notarizations.contains_key(&view_3), "reporter {idx}");
7239 }
7240 }
7241
7242 link_validators(&mut oracle, &participants, Action::Link(link.clone()), None).await;
7244
7245 {
7246 let target = View::new(3);
7247 let mut finalizers = Vec::new();
7248 for reporter in honest_reporters.values_mut() {
7249 let (mut latest, mut monitor) = reporter.subscribe().await;
7250 finalizers.push(
7251 context
7252 .child("resume_finalizer")
7253 .spawn(move |_| async move {
7254 while latest < target {
7255 latest = monitor.recv().await.expect("event missing");
7256 }
7257 }),
7258 );
7259 }
7260 join_all(finalizers).await;
7261 }
7262
7263 for (idx, reporter) in honest_reporters.iter() {
7265 let first = {
7266 let finalizations = reporter.finalizations.lock();
7267 finalizations
7268 .keys()
7269 .filter(|view| **view > view_2)
7270 .min()
7271 .copied()
7272 .expect("no finalization past the preload")
7273 };
7274 assert_eq!(
7275 first,
7276 View::new(11),
7277 "reporter {idx} did not finalize the first honest-led view"
7278 );
7279 }
7280 assert!(
7282 honest_reporters[&lone]
7283 .nullifications
7284 .lock()
7285 .contains_key(&view_3),
7286 "lone did not resolve Nullification(3) from the group leader"
7287 );
7288
7289 for reporter in honest_reporters.values() {
7290 reporter.assert_no_faults();
7291 reporter.assert_no_invalid();
7292 }
7293 let blocked = oracle.blocked().await.unwrap();
7294 assert!(blocked.is_empty(), "blocked peers: {blocked:?}");
7295 });
7296 }
7297
7298 #[test_group("slow")]
7299 #[test_traced]
7300 fn test_certified_split_heals_in_group_led_view() {
7301 certified_split_heals_in_group_led_view::<_, _, RoundRobin>(
7302 ed25519::fixture,
7303 RoundRobin::default(),
7304 );
7305 }
7306
7307 fn certified_split_heals_when_lone_holder_leads_first<S, F, L>(mut fixture: F, elector: L)
7312 where
7313 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
7314 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
7315 L: elector::Config<S>,
7316 L::Elector: Elector<S, Mode = Scheduled>,
7317 {
7318 let n = 4;
7319 let quorum = quorum(n) as usize;
7320 assert_eq!(quorum, 3);
7321 let view_retention = ViewDelta::new(10);
7322 let skip_timeout = Duration::from_secs(12);
7323 let namespace = b"consensus".to_vec();
7324 let executor = deterministic::Runner::timed(Duration::from_secs(300));
7325 executor.start(|mut context| async move {
7326 let Fixture {
7327 participants,
7328 schemes,
7329 ..
7330 } = fixture(&mut context, &namespace, n);
7331 let mut oracle = start_test_network_with_peers::<S, _>(
7332 context.child("network"),
7333 participants.clone(),
7334 false,
7335 )
7336 .await;
7337 let mut registrations = register_validators(&mut oracle, &participants).await;
7338
7339 let epoch = Epoch::new(333);
7343 let participant_set: Set<PublicKey> = participants.clone().try_into().unwrap();
7344 let schedule = elector.clone().build(&participant_set);
7345 let leader_of =
7346 |view: u64| usize::from(schedule.elect(Round::new(epoch, View::new(view)), ()));
7347 let byzantine = leader_of(10);
7348 let lone = leader_of(11);
7349 let group = [leader_of(12), leader_of(13)];
7350 assert!(!group.contains(&byzantine) && !group.contains(&lone) && byzantine != lone);
7351
7352 let build_notarization = |proposal: &Proposal<D>| -> TNotarization<_, D> {
7354 let votes: Vec<_> = (0..quorum)
7355 .map(|i| TNotarize::sign(&schemes[i], proposal.clone()).unwrap())
7356 .collect();
7357 TNotarization::from_notarizes(&schemes[0], non_empty![@&votes], &Sequential)
7358 .expect("notarization quorum")
7359 };
7360 let build_finalization = |proposal: &Proposal<D>| -> TFinalization<_, D> {
7361 let votes: Vec<_> = (0..quorum)
7362 .map(|i| TFinalize::sign(&schemes[i], proposal.clone()).unwrap())
7363 .collect();
7364 TFinalization::from_finalizes(&schemes[0], non_empty![@&votes], &Sequential)
7365 .expect("finalization quorum")
7366 };
7367 let build_nullification = |view: u64| -> TNullification<_> {
7368 let round = Round::new(epoch, View::new(view));
7369 let votes: Vec<_> = (0..quorum)
7370 .map(|i| TNullify::sign::<D>(&schemes[i], round).unwrap())
7371 .collect();
7372 TNullification::from_nullifies(&schemes[0], non_empty![@&votes], &Sequential)
7373 .expect("nullification quorum")
7374 };
7375 let payload_b2 = Sha256::hash(&[b"B_2"]);
7376 let proposal_b2 =
7377 Proposal::new(Round::new(epoch, View::new(2)), View::new(1), payload_b2);
7378 let payload_b3 = Sha256::hash(&[b"B_3"]);
7379 let proposal_b3 =
7380 Proposal::new(Round::new(epoch, View::new(3)), View::new(2), payload_b3);
7381 let b2_notarization = build_notarization(&proposal_b2);
7382 let b2_finalization = build_finalization(&proposal_b2);
7383 let b3_notarization = build_notarization(&proposal_b3);
7384 let null_3 = build_nullification(3);
7385
7386 let link = Link {
7387 latency: Duration::from_millis(10),
7388 jitter: Duration::from_millis(0),
7389 success_rate: probability!(1.0),
7390 };
7391 let mut injector_sender =
7392 start_certificate_injector(&context, &mut oracle, &participants, &link).await;
7393
7394 let msg = Certificate::<_, D>::Notarization(b2_notarization).encode();
7396 injector_sender.send(Recipients::All, msg, true);
7397 let msg = Certificate::<_, D>::Finalization(b2_finalization).encode();
7398 injector_sender.send(Recipients::All, msg, true);
7399 let msg = Certificate::<_, D>::Notarization(b3_notarization).encode();
7400 injector_sender.send(Recipients::One(participants[lone].clone()), msg, true);
7401 let msg = Certificate::<_, D>::Nullification(null_3).encode();
7402 for idx in group {
7403 injector_sender.send(
7404 Recipients::One(participants[idx].clone()),
7405 msg.clone(),
7406 true,
7407 );
7408 }
7409 for view in 4..=9 {
7410 let msg = Certificate::<_, D>::Nullification(build_nullification(view)).encode();
7411 injector_sender.send(Recipients::All, msg, true);
7412 }
7413
7414 let mut honest_reporters = start_certified_split_engines(
7416 &context,
7417 CertifiedSplitEngineConfig {
7418 oracle: &oracle,
7419 participants: &participants,
7420 schemes: &schemes,
7421 registrations: &mut registrations,
7422 silent: byzantine,
7423 elector: &elector,
7424 epoch,
7425 view_retention,
7426 skip_timeout,
7427 },
7428 );
7429
7430 context.sleep(Duration::from_secs(2)).await;
7432
7433 let view_2 = View::new(2);
7435 let view_3 = View::new(3);
7436 for (idx, reporter) in honest_reporters.iter() {
7437 assert!(
7438 reporter.finalizations.lock().contains_key(&view_2),
7439 "reporter {idx} missing finalization for view 2"
7440 );
7441 let notarizations = reporter.notarizations.lock();
7442 let nullifications = reporter.nullifications.lock();
7443 if *idx == lone {
7444 assert!(notarizations.contains_key(&view_3));
7445 assert!(!nullifications.contains_key(&view_3));
7446 assert!(reporter.certifications.lock().contains_key(&view_3));
7447 } else {
7448 assert!(nullifications.contains_key(&view_3), "reporter {idx}");
7449 assert!(!notarizations.contains_key(&view_3), "reporter {idx}");
7450 }
7451 }
7452
7453 link_validators(&mut oracle, &participants, Action::Link(link.clone()), None).await;
7455
7456 {
7457 let target = View::new(3);
7458 let mut finalizers = Vec::new();
7459 for reporter in honest_reporters.values_mut() {
7460 let (mut latest, mut monitor) = reporter.subscribe().await;
7461 finalizers.push(
7462 context
7463 .child("resume_finalizer")
7464 .spawn(move |_| async move {
7465 while latest < target {
7466 latest = monitor.recv().await.expect("event missing");
7467 }
7468 }),
7469 );
7470 }
7471 join_all(finalizers).await;
7472 }
7473
7474 for (idx, reporter) in honest_reporters.iter() {
7476 let first = {
7477 let finalizations = reporter.finalizations.lock();
7478 finalizations
7479 .keys()
7480 .filter(|view| **view > view_2)
7481 .min()
7482 .copied()
7483 .expect("no finalization past the preload")
7484 };
7485 assert_eq!(
7486 first, view_3,
7487 "reporter {idx} did not finalize the recovered parent view"
7488 );
7489 }
7490
7491 for idx in group {
7493 let reporter = &honest_reporters[&idx];
7494 assert!(
7495 reporter.notarizations.lock().contains_key(&view_3),
7496 "group reporter {idx} did not resolve Notarization(3) from lone"
7497 );
7498 assert!(
7499 reporter.certifications.lock().contains_key(&view_3),
7500 "group reporter {idx} did not certify Notarization(3)"
7501 );
7502 }
7503
7504 for reporter in honest_reporters.values() {
7505 reporter.assert_no_faults();
7506 reporter.assert_no_invalid();
7507 }
7508 let blocked = oracle.blocked().await.unwrap();
7509 assert!(blocked.is_empty(), "blocked peers: {blocked:?}");
7510 });
7511 }
7512
7513 #[test_group("slow")]
7514 #[test_traced]
7515 fn test_certified_split_heals_when_lone_holder_leads_first() {
7516 certified_split_heals_when_lone_holder_leads_first::<_, _, RoundRobin>(
7517 ed25519::fixture,
7518 RoundRobin::default(),
7519 );
7520 }
7521
7522 fn certified_split_heals_with_displaced_certified_view<S, F, L>(mut fixture: F, elector: L)
7528 where
7529 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
7530 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
7531 L: elector::Config<S>,
7532 L::Elector: Elector<S, Mode = Scheduled>,
7533 {
7534 let n = 4;
7535 let quorum = quorum(n) as usize;
7536 assert_eq!(quorum, 3);
7537 let view_retention = ViewDelta::new(10);
7538 let skip_timeout = Duration::from_secs(12);
7539 let namespace = b"consensus".to_vec();
7540 let executor = deterministic::Runner::timed(Duration::from_secs(300));
7541 executor.start(|mut context| async move {
7542 let Fixture {
7543 participants,
7544 schemes,
7545 ..
7546 } = fixture(&mut context, &namespace, n);
7547 let mut oracle = start_test_network_with_peers::<S, _>(
7548 context.child("network"),
7549 participants.clone(),
7550 false,
7551 )
7552 .await;
7553 let mut registrations = register_validators(&mut oracle, &participants).await;
7554
7555 let epoch = Epoch::new(333);
7559 let participant_set: Set<PublicKey> = participants.clone().try_into().unwrap();
7560 let schedule = elector.clone().build(&participant_set);
7561 let leader_of =
7562 |view: u64| usize::from(schedule.elect(Round::new(epoch, View::new(view)), ()));
7563 let byzantine = leader_of(10);
7564 let group = [leader_of(11), leader_of(12)];
7565 let lone = leader_of(13);
7566 assert!(!group.contains(&byzantine) && !group.contains(&lone) && byzantine != lone);
7567
7568 let build_notarization = |proposal: &Proposal<D>| -> TNotarization<_, D> {
7570 let votes: Vec<_> = (0..quorum)
7571 .map(|i| TNotarize::sign(&schemes[i], proposal.clone()).unwrap())
7572 .collect();
7573 TNotarization::from_notarizes(&schemes[0], non_empty![@&votes], &Sequential)
7574 .expect("notarization quorum")
7575 };
7576 let build_finalization = |proposal: &Proposal<D>| -> TFinalization<_, D> {
7577 let votes: Vec<_> = (0..quorum)
7578 .map(|i| TFinalize::sign(&schemes[i], proposal.clone()).unwrap())
7579 .collect();
7580 TFinalization::from_finalizes(&schemes[0], non_empty![@&votes], &Sequential)
7581 .expect("finalization quorum")
7582 };
7583 let build_nullification = |view: u64| -> TNullification<_> {
7584 let round = Round::new(epoch, View::new(view));
7585 let votes: Vec<_> = (0..quorum)
7586 .map(|i| TNullify::sign::<D>(&schemes[i], round).unwrap())
7587 .collect();
7588 TNullification::from_nullifies(&schemes[0], non_empty![@&votes], &Sequential)
7589 .expect("nullification quorum")
7590 };
7591 let payload_b2 = Sha256::hash(&[b"B_2"]);
7592 let proposal_b2 =
7593 Proposal::new(Round::new(epoch, View::new(2)), View::new(1), payload_b2);
7594 let payload_b5 = Sha256::hash(&[b"B_5"]);
7595 let proposal_b5 =
7596 Proposal::new(Round::new(epoch, View::new(5)), View::new(2), payload_b5);
7597 let b2_notarization = build_notarization(&proposal_b2);
7598 let b2_finalization = build_finalization(&proposal_b2);
7599 let b5_notarization = build_notarization(&proposal_b5);
7600
7601 let link = Link {
7602 latency: Duration::from_millis(10),
7603 jitter: Duration::from_millis(0),
7604 success_rate: probability!(1.0),
7605 };
7606 let mut injector_sender =
7607 start_certificate_injector(&context, &mut oracle, &participants, &link).await;
7608
7609 let msg = Certificate::<_, D>::Notarization(b2_notarization).encode();
7611 injector_sender.send(Recipients::All, msg, true);
7612 let msg = Certificate::<_, D>::Finalization(b2_finalization).encode();
7613 injector_sender.send(Recipients::All, msg, true);
7614 let msg = Certificate::<_, D>::Notarization(b5_notarization).encode();
7615 injector_sender.send(Recipients::One(participants[lone].clone()), msg, true);
7616 for view in 3..=5 {
7617 let msg = Certificate::<_, D>::Nullification(build_nullification(view)).encode();
7618 for idx in group {
7619 injector_sender.send(
7620 Recipients::One(participants[idx].clone()),
7621 msg.clone(),
7622 true,
7623 );
7624 }
7625 }
7626 for view in 6..=9 {
7627 let msg = Certificate::<_, D>::Nullification(build_nullification(view)).encode();
7628 injector_sender.send(Recipients::All, msg, true);
7629 }
7630
7631 let mut honest_reporters = start_certified_split_engines(
7633 &context,
7634 CertifiedSplitEngineConfig {
7635 oracle: &oracle,
7636 participants: &participants,
7637 schemes: &schemes,
7638 registrations: &mut registrations,
7639 silent: byzantine,
7640 elector: &elector,
7641 epoch,
7642 view_retention,
7643 skip_timeout,
7644 },
7645 );
7646
7647 context.sleep(Duration::from_secs(2)).await;
7649
7650 let view_2 = View::new(2);
7652 let view_5 = View::new(5);
7653 for (idx, reporter) in honest_reporters.iter() {
7654 assert!(
7655 reporter.finalizations.lock().contains_key(&view_2),
7656 "reporter {idx} missing finalization for view 2"
7657 );
7658 let notarizations = reporter.notarizations.lock();
7659 let nullifications = reporter.nullifications.lock();
7660 if *idx == lone {
7661 assert!(notarizations.contains_key(&view_5));
7662 assert!(reporter.certifications.lock().contains_key(&view_5));
7663 for view in 3..=5 {
7664 assert!(!nullifications.contains_key(&View::new(view)));
7665 }
7666 } else {
7667 for view in 3..=5 {
7668 assert!(
7669 nullifications.contains_key(&View::new(view)),
7670 "reporter {idx}"
7671 );
7672 }
7673 assert!(!notarizations.contains_key(&view_5), "reporter {idx}");
7674 }
7675 }
7676
7677 link_validators(&mut oracle, &participants, Action::Link(link.clone()), None).await;
7679
7680 {
7681 let target = View::new(3);
7682 let mut finalizers = Vec::new();
7683 for reporter in honest_reporters.values_mut() {
7684 let (mut latest, mut monitor) = reporter.subscribe().await;
7685 finalizers.push(
7686 context
7687 .child("resume_finalizer")
7688 .spawn(move |_| async move {
7689 while latest < target {
7690 latest = monitor.recv().await.expect("event missing");
7691 }
7692 }),
7693 );
7694 }
7695 join_all(finalizers).await;
7696 }
7697
7698 for (idx, reporter) in honest_reporters.iter() {
7700 let first = {
7701 let finalizations = reporter.finalizations.lock();
7702 finalizations
7703 .keys()
7704 .filter(|view| **view > view_2)
7705 .min()
7706 .copied()
7707 .expect("no finalization past the preload")
7708 };
7709 assert_eq!(
7710 first,
7711 View::new(11),
7712 "reporter {idx} did not finalize the first honest-led view"
7713 );
7714 }
7715 let lone_reporter = &honest_reporters[&lone];
7716 for view in 3..=5 {
7717 assert!(
7718 lone_reporter
7719 .nullifications
7720 .lock()
7721 .contains_key(&View::new(view)),
7722 "lone did not resolve Nullification({view}) from the group leader"
7723 );
7724 }
7725
7726 for reporter in honest_reporters.values() {
7727 reporter.assert_no_faults();
7728 reporter.assert_no_invalid();
7729 }
7730 let blocked = oracle.blocked().await.unwrap();
7731 assert!(blocked.is_empty(), "blocked peers: {blocked:?}");
7732 });
7733 }
7734
7735 #[test_group("slow")]
7736 #[test_traced]
7737 fn test_certified_split_heals_with_displaced_certified_view() {
7738 certified_split_heals_with_displaced_certified_view::<_, _, RoundRobin>(
7739 ed25519::fixture,
7740 RoundRobin::default(),
7741 );
7742 }
7743
7744 fn stable_leader_cross_term_certified_split<S, F, L>(
7765 mut fixture: F,
7766 elector: L,
7767 suppress_backfill: bool,
7768 ) where
7769 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
7770 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
7771 L: elector::Config<S>,
7772 L::Elector: Elector<S, Mode = Scheduled>,
7773 {
7774 let n = 4;
7775 let quorum = quorum(n) as usize;
7776 assert_eq!(quorum, 3);
7777 let view_retention = ViewDelta::new(10);
7778 let skip_timeout = Duration::from_secs(12);
7779 let namespace = b"consensus".to_vec();
7780 let executor = deterministic::Runner::timed(Duration::from_secs(300));
7781 executor.start(|mut context| async move {
7782 let Fixture {
7783 participants,
7784 schemes,
7785 ..
7786 } = fixture(&mut context, &namespace, n);
7787 let mut oracle = start_test_network_with_peers::<S, _>(
7788 context.child("network"),
7789 participants.clone(),
7790 false,
7791 )
7792 .await;
7793 let mut registrations = register_validators(&mut oracle, &participants).await;
7794
7795 let epoch = Epoch::new(2);
7797 let participant_set: Set<PublicKey> = participants.clone().try_into().unwrap();
7798 let schedule = elector.clone().build(&participant_set);
7799 let leader_of =
7800 |view: u64| usize::from(schedule.elect(Round::new(epoch, View::new(view)), ()));
7801 let offline = leader_of(1);
7802 assert_eq!(offline, leader_of(21), "offline must lead terms 1 and 5");
7803 let honest: Vec<usize> = (0..n as usize).filter(|idx| *idx != offline).collect();
7804 let (a, b, c) = (honest[0], honest[1], honest[2]);
7805 for view in [6u64, 11, 16, 26, 31, 36] {
7806 assert_ne!(leader_of(view), offline, "term start must be honest");
7807 }
7808 assert_ne!(
7810 leader_of(26),
7811 c,
7812 "view-26 leader must hold certification 22"
7813 );
7814 assert_eq!(leader_of(41), offline, "term 9 must return to offline");
7816
7817 let build_notarization =
7818 |proposal: &Proposal<D>, signers: [usize; 3]| -> TNotarization<_, D> {
7819 let votes: Vec<_> = signers
7820 .iter()
7821 .map(|i| TNotarize::sign(&schemes[*i], proposal.clone()).unwrap())
7822 .collect();
7823 TNotarization::from_notarizes(&schemes[0], non_empty![@&votes], &Sequential)
7824 .expect("notarization quorum")
7825 };
7826 let build_nullification = |view: u64, signers: [usize; 3]| -> TNullification<_> {
7827 let round = Round::new(epoch, View::new(view));
7828 let votes: Vec<_> = signers
7829 .iter()
7830 .map(|i| TNullify::sign::<D>(&schemes[*i], round).unwrap())
7831 .collect();
7832 TNullification::from_nullifies(&schemes[0], non_empty![@&votes], &Sequential)
7833 .expect("nullification quorum")
7834 };
7835
7836 let payload_1 = Sha256::hash(&[b"V1"]);
7838 let proposal_1 =
7839 Proposal::new(Round::new(epoch, View::new(1)), View::new(0), payload_1);
7840 let payload_2 = Sha256::hash(&[b"V2"]);
7841 let proposal_2 =
7842 Proposal::new(Round::new(epoch, View::new(2)), View::new(1), payload_2);
7843 let payload_21 = Sha256::hash(&[b"V21"]);
7847 let proposal_21 =
7848 Proposal::new(Round::new(epoch, View::new(21)), View::new(0), payload_21);
7849 let payload_22 = Sha256::hash(&[b"V22"]);
7850 let proposal_22 =
7851 Proposal::new(Round::new(epoch, View::new(22)), View::new(21), payload_22);
7852
7853 let link = Link {
7854 latency: Duration::from_millis(10),
7855 jitter: Duration::from_millis(0),
7856 success_rate: probability!(1.0),
7857 };
7858 let mut injector_sender =
7859 start_certificate_injector(&context, &mut oracle, &participants, &link).await;
7860
7861 {
7862 let mut send_to = |certificate: Certificate<S, D>, targets: &[usize]| {
7863 let msg = certificate.encode();
7864 for idx in targets {
7865 injector_sender.send(
7866 Recipients::One(participants[*idx].clone()),
7867 msg.clone(),
7868 true,
7869 );
7870 }
7871 };
7872
7873 send_to(
7876 Certificate::Notarization(build_notarization(&proposal_1, [b, c, offline])),
7877 &[b, c],
7878 );
7879 send_to(
7880 Certificate::Notarization(build_notarization(&proposal_2, [b, c, offline])),
7881 &[b, c],
7882 );
7883 send_to(
7884 Certificate::Nullification(build_nullification(3, [b, c, offline])),
7885 &[b, c],
7886 );
7887 let a_skip = if suppress_backfill { 1 } else { 3 };
7888 send_to(
7889 Certificate::Nullification(build_nullification(a_skip, [a, c, offline])),
7890 &[a, b],
7891 );
7892
7893 for view in [6u64, 11, 16] {
7895 send_to(
7896 Certificate::Nullification(build_nullification(view, [a, b, c])),
7897 &[a, b, c],
7898 );
7899 }
7900
7901 send_to(
7904 Certificate::Notarization(build_notarization(&proposal_21, [a, b, offline])),
7905 &[a, b],
7906 );
7907 send_to(
7908 Certificate::Notarization(build_notarization(&proposal_22, [a, b, offline])),
7909 &[a, b],
7910 );
7911 let c_skip = if suppress_backfill { 21 } else { 23 };
7912 send_to(
7913 Certificate::Nullification(build_nullification(c_skip, [a, c, offline])),
7914 &[c],
7915 );
7916 send_to(
7917 Certificate::Nullification(build_nullification(23, [a, b, offline])),
7918 &[a, b],
7919 );
7920 }
7921
7922 let mut honest_reporters = start_certified_split_engines(
7925 &context,
7926 CertifiedSplitEngineConfig {
7927 oracle: &oracle,
7928 participants: &participants,
7929 schemes: &schemes,
7930 registrations: &mut registrations,
7931 silent: offline,
7932 elector: &elector,
7933 epoch,
7934 view_retention,
7935 skip_timeout,
7936 },
7937 );
7938
7939 context.sleep(Duration::from_secs(2)).await;
7941
7942 for (idx, reporter) in honest_reporters.iter() {
7944 let certifications = reporter.certifications.lock();
7945 let has_term1 = certifications.contains_key(&View::new(2));
7946 let has_term5 = certifications.contains_key(&View::new(22));
7947 if *idx == a {
7948 assert!(!has_term1 && has_term5, "a: {has_term1} {has_term5}");
7949 } else if *idx == b {
7950 assert!(has_term1 && has_term5, "b: {has_term1} {has_term5}");
7951 } else {
7952 assert!(has_term1 && !has_term5, "c: {has_term1} {has_term5}");
7953 }
7954 assert!(
7955 reporter.finalizations.lock().is_empty(),
7956 "reporter {idx} finalized during preload"
7957 );
7958 }
7959
7960 link_validators(&mut oracle, &participants, Action::Link(link.clone()), None).await;
7962
7963 let target = View::new(26);
7965 let mut finalizers = Vec::new();
7966 for (idx, reporter) in honest_reporters.iter_mut() {
7967 let (mut latest, mut monitor) = reporter.subscribe().await;
7968 finalizers.push(
7969 context
7970 .child("finalizer")
7971 .with_attribute("index", idx)
7972 .spawn(move |_| async move {
7973 while latest < target {
7974 latest = monitor.recv().await.expect("finalization missing");
7975 }
7976 }),
7977 );
7978 }
7979 let progress_reporters: Vec<_> = honest_reporters.values().cloned().collect();
7982 let next_byzantine_term =
7983 context
7984 .child("next_byzantine_term")
7985 .spawn(move |context| async move {
7986 loop {
7987 let reached = progress_reporters.iter().all(|reporter| {
7988 reporter
7989 .nullifications
7990 .lock()
7991 .keys()
7992 .any(|view| *view >= View::new(41))
7993 });
7994 if reached {
7995 return;
7996 }
7997 context.sleep(Duration::from_secs(1)).await;
7998 }
7999 });
8000 let finalized = select! {
8001 _ = join_all(finalizers) => true,
8002 _ = next_byzantine_term => false,
8003 };
8004
8005 for reporter in honest_reporters.values() {
8006 reporter.assert_no_faults();
8007 reporter.assert_no_invalid();
8008 }
8009 let blocked = oracle.blocked().await.unwrap();
8010 assert!(blocked.is_empty(), "blocked peers: {blocked:?}");
8011
8012 if !finalized {
8013 let a_reporter = &honest_reporters[&a];
8017 assert!(
8018 a_reporter.notarizations.lock().contains_key(&View::new(2)),
8019 "halted, but a is missing notarization(2)"
8020 );
8021 assert!(
8022 !a_reporter.notarizations.lock().contains_key(&View::new(1)),
8023 "halted, but a repaired notarization(1)"
8024 );
8025 assert!(
8026 !a_reporter.certifications.lock().contains_key(&View::new(2)),
8027 "halted, but a certified view 2"
8028 );
8029
8030 let c_reporter = &honest_reporters[&c];
8031 assert!(
8032 c_reporter.notarizations.lock().contains_key(&View::new(22)),
8033 "halted, but c is missing notarization(22)"
8034 );
8035 assert!(
8036 !c_reporter.notarizations.lock().contains_key(&View::new(21)),
8037 "halted, but c repaired notarization(21)"
8038 );
8039 assert!(
8040 !c_reporter.certifications.lock().contains_key(&View::new(22)),
8041 "halted, but c certified view 22"
8042 );
8043
8044 panic!(
8045 "all honest leaders exhausted a term, but recursive parent repair never finalized"
8046 );
8047 }
8048 for (idx, reporter) in honest_reporters.iter() {
8049 assert!(
8050 reporter
8051 .finalizations
8052 .lock()
8053 .keys()
8054 .any(|view| *view >= target),
8055 "reporter {idx} never finalized after GST"
8056 );
8057 }
8058 let c_certifications = honest_reporters[&c].certifications.lock();
8059 for view in [View::new(21), View::new(22)] {
8060 assert!(
8061 c_certifications.contains_key(&view),
8062 "c did not recover and certify view {view} from an honest validator"
8063 );
8064 }
8065 });
8066 }
8067
8068 #[test_group("slow")]
8069 #[test_traced]
8070 fn test_stable_leader_cross_term_certified_split_recovers() {
8071 stable_leader_cross_term_certified_split::<_, _, RoundRobin>(
8072 ed25519::fixture,
8073 RoundRobin::default().with_term(
8074 TermLength::new(NZU32!(5)),
8075 Duration::from_secs(12),
8076 ViewDelta::new(0),
8077 ),
8078 true,
8079 );
8080 }
8081
8082 #[test_group("slow")]
8083 #[test_traced]
8084 fn test_stable_leader_cross_term_certified_split_backfillable() {
8085 stable_leader_cross_term_certified_split::<_, _, RoundRobin>(
8086 ed25519::fixture,
8087 RoundRobin::default().with_term(
8088 TermLength::new(NZU32!(5)),
8089 Duration::from_secs(12),
8090 ViewDelta::new(0),
8091 ),
8092 false,
8093 );
8094 }
8095
8096 fn tle<V, L>(elector: L)
8097 where
8098 V: Variant,
8099 L: elector::Config<bls12381_threshold_vrf::Scheme<PublicKey, V>>,
8100 {
8101 let n = 4;
8103 let namespace = b"consensus".to_vec();
8104 let view_retention = ViewDelta::new(100);
8105 let skip_timeout = Duration::from_secs(50);
8106 let executor = deterministic::Runner::timed(Duration::from_secs(30));
8107 executor.start(|mut context| async move {
8108 let Fixture {
8110 participants,
8111 schemes,
8112 ..
8113 } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, &namespace, n);
8114 let mut oracle = start_test_network_with_peers::<
8115 bls12381_threshold_vrf::Scheme<PublicKey, V>,
8116 _,
8117 >(context.child("network"), participants.clone(), true)
8118 .await;
8119 let mut registrations = register_validators(&mut oracle, &participants).await;
8120
8121 let link = Link {
8123 latency: Duration::from_millis(10),
8124 jitter: Duration::from_millis(5),
8125 success_rate: probability!(1.0),
8126 };
8127 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
8128
8129 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
8131 let mut reporters = Vec::new();
8132 let mut engine_handlers = Vec::new();
8133 let monitor_reporter = Arc::new(Mutex::new(None));
8134 for (idx, validator) in participants.iter().enumerate() {
8135 let context = context
8137 .child("validator")
8138 .with_attribute("public_key", validator);
8139
8140 let reporter_config = mocks::reporter::Config {
8142 participants: participants.clone().try_into().unwrap(),
8143 scheme: schemes[idx].clone(),
8144 elector: elector.clone(),
8145 };
8146 let reporter =
8147 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
8148 reporters.push(reporter.clone());
8149 if idx == 0 {
8150 *monitor_reporter.lock() = Some(reporter.clone());
8151 }
8152
8153 let application_cfg = mocks::application::Config::<Sha256, _> {
8155 relay: relay.clone(),
8156 me: validator.clone(),
8157 propose_latency: (10.0, 5.0),
8158 verify_latency: (10.0, 5.0),
8159 certify_latency: (10.0, 5.0),
8160 should_certify: mocks::application::Certifier::Always,
8161 };
8162 let (actor, application) = mocks::application::Application::new(
8163 context.child("application"),
8164 application_cfg,
8165 );
8166 actor.start();
8167 let blocker = oracle.control(validator.clone());
8168 let cfg = config::Config {
8169 scheme: schemes[idx].clone(),
8170 elector: elector.clone(),
8171 blocker,
8172 automaton: application.clone(),
8173 relay: application.clone(),
8174 reporter: reporter.clone(),
8175 strategy: Sequential,
8176 partition: validator.to_string(),
8177 mailbox_size: NZUsize!(1024),
8178 epoch: Epoch::new(333),
8179 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
8180 Epoch::new(333),
8181 )),
8182 leader_timeout: Duration::from_millis(100),
8183 certification_timeout: Duration::from_millis(200),
8184 timeout_retry: Duration::from_millis(500),
8185 fetch_timeout: NZDuration!(Duration::from_millis(100)),
8186 fetch_retry_delay: Duration::from_millis(100),
8187 view_retention,
8188 skip: SkipPolicy::Enabled {
8189 timeout: skip_timeout,
8190 budget: SkipBudget::Participants,
8191 },
8192 replay_buffer: NZUsize!(1024 * 1024),
8193 write_buffer: NZUsize!(1024 * 1024),
8194 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
8195 forward: ForwardPolicy::Disabled,
8196 track_historical_votes: false,
8197 };
8198 let engine = Engine::new(context.child("engine"), cfg);
8199
8200 let (pending, recovered, resolver) = registrations
8202 .remove(validator)
8203 .expect("validator should be registered");
8204 engine_handlers.push(engine.start(pending, recovered, resolver));
8205 }
8206
8207 let target = Round::new(Epoch::new(333), View::new(10)); let message = b"Secret message for future view10"; let ciphertext = schemes[0]
8213 .encrypt(&mut context, target, *message)
8214 .expect("valid TLE encryption inputs");
8215
8216 let reporter = monitor_reporter.lock().clone().unwrap();
8218 loop {
8219 context.sleep(Duration::from_millis(100)).await;
8221 let notarizations = reporter.notarizations.lock();
8222 let Some(notarization) = notarizations.get(&target.view()) else {
8223 continue;
8224 };
8225
8226 let seed = notarization.seed();
8228 let decrypted = seed
8229 .decrypt(&ciphertext)
8230 .expect("Decryption should succeed with valid seed signature");
8231 assert_eq!(
8232 message,
8233 decrypted.as_ref(),
8234 "Decrypted message should match original message"
8235 );
8236 break;
8237 }
8238 });
8239 }
8240
8241 #[test_traced]
8242 fn test_tle() {
8243 tle::<MinPk, Random>(Random::new(RandomVersion::V1));
8244 tle::<MinSig, Random>(Random::new(RandomVersion::V1));
8245 }
8246
8247 fn run_hailstorm<S, F, L>(
8248 seed: u64,
8249 shutdowns: usize,
8250 interval: ViewDelta,
8251 elector: L,
8252 mut fixture: F,
8253 ) -> String
8254 where
8255 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
8256 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
8257 L: elector::Config<S>,
8258 {
8259 let n = 5;
8261 let view_retention = ViewDelta::new(10);
8262 let skip_timeout = Duration::from_secs(11);
8263 let namespace = b"consensus".to_vec();
8264 let cfg = deterministic::Config::new().with_seed(seed);
8265 let executor = deterministic::Runner::new(cfg);
8266 executor.start(|mut context| async move {
8267 let Fixture {
8269 participants,
8270 schemes,
8271 ..
8272 } = fixture(&mut context, &namespace, n);
8273 let mut oracle = start_test_network_with_peers::<S, _>(
8274 context.child("network"),
8275 participants.clone(),
8276 true,
8277 )
8278 .await;
8279 let mut registrations = register_validators(&mut oracle, &participants).await;
8280
8281 let link = Link {
8283 latency: Duration::from_millis(10),
8284 jitter: Duration::from_millis(1),
8285 success_rate: probability!(1.0),
8286 };
8287 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
8288
8289 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
8291 let mut reporters = BTreeMap::new();
8292 let mut engine_handlers = BTreeMap::new();
8293 for (idx, validator) in participants.iter().enumerate() {
8294 let context = context
8296 .child("validator")
8297 .with_attribute("public_key", validator);
8298
8299 let reporter_config = mocks::reporter::Config {
8301 participants: participants.clone().try_into().unwrap(),
8302 scheme: schemes[idx].clone(),
8303 elector: elector.clone(),
8304 };
8305 let reporter =
8306 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
8307 reporters.insert(idx, reporter.clone());
8308 let application_cfg = mocks::application::Config::<Sha256, _> {
8309 relay: relay.clone(),
8310 me: validator.clone(),
8311 propose_latency: (10.0, 5.0),
8312 verify_latency: (10.0, 5.0),
8313 certify_latency: (10.0, 5.0),
8314 should_certify: mocks::application::Certifier::Always,
8315 };
8316 let (actor, application) = mocks::application::Application::new(
8317 context.child("application"),
8318 application_cfg,
8319 );
8320 actor.start();
8321 let blocker = oracle.control(validator.clone());
8322 let cfg = config::Config {
8323 scheme: schemes[idx].clone(),
8324 elector: elector.clone(),
8325 blocker,
8326 automaton: application.clone(),
8327 relay: application.clone(),
8328 reporter: reporter.clone(),
8329 strategy: Sequential,
8330 partition: validator.to_string(),
8331 mailbox_size: NZUsize!(1024),
8332 epoch: Epoch::new(333),
8333 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
8334 Epoch::new(333),
8335 )),
8336 leader_timeout: Duration::from_secs(1),
8337 certification_timeout: Duration::from_secs(2),
8338 timeout_retry: Duration::from_secs(10),
8339 fetch_timeout: NZDuration!(Duration::from_secs(1)),
8340 fetch_retry_delay: Duration::from_millis(100),
8341 view_retention,
8342 skip: SkipPolicy::Enabled {
8343 timeout: skip_timeout,
8344 budget: SkipBudget::Participants,
8345 },
8346 replay_buffer: NZUsize!(1024 * 1024),
8347 write_buffer: NZUsize!(1024 * 1024),
8348 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
8349 forward: ForwardPolicy::Disabled,
8350 track_historical_votes: false,
8351 };
8352 let engine = Engine::new(context.child("engine"), cfg);
8353
8354 let (pending, recovered, resolver) = registrations
8356 .remove(validator)
8357 .expect("validator should be registered");
8358 engine_handlers.insert(idx, engine.start(pending, recovered, resolver));
8359 }
8360
8361 let mut target = View::zero();
8363 for i in 0..shutdowns {
8364 target = target.saturating_add(interval);
8366
8367 let mut finalizers = Vec::new();
8369 for reporter in reporters.values_mut() {
8370 let (mut latest, mut monitor) = reporter.subscribe().await;
8371 finalizers.push(context.child("finalizer").spawn(move |_| async move {
8372 while latest < target {
8373 latest = monitor.recv().await.expect("event missing");
8374 }
8375 }));
8376 }
8377 join_all(finalizers).await;
8378 target = target.saturating_add(interval);
8379
8380 let idx = context.random_range(0..engine_handlers.len());
8382 let validator = &participants[idx];
8383 let handle = engine_handlers.remove(&idx).unwrap();
8384 handle.abort();
8385 let _ = handle.await;
8386 let selected_reporter = reporters.remove(&idx).unwrap();
8387 info!(idx, ?validator, "shutdown validator");
8388
8389 let mut finalizers = Vec::new();
8391 for reporter in reporters.values_mut() {
8392 let (mut latest, mut monitor) = reporter.subscribe().await;
8393 finalizers.push(context.child("finalizer").spawn(move |_| async move {
8394 while latest < target {
8395 latest = monitor.recv().await.expect("event missing");
8396 }
8397 }));
8398 }
8399 join_all(finalizers).await;
8400 target = target.saturating_add(interval);
8401
8402 info!(idx, ?validator, "restarting validator");
8404 let context = context
8405 .child("validator_restarted")
8406 .with_attribute("public_key", validator)
8407 .with_attribute("restart", i);
8408
8409 let (pending, recovered, resolver) =
8411 register_validator(&mut oracle, validator.clone()).await;
8412 let application_cfg = mocks::application::Config::<Sha256, _> {
8413 relay: relay.clone(),
8414 me: validator.clone(),
8415 propose_latency: (10.0, 5.0),
8416 verify_latency: (10.0, 5.0),
8417 certify_latency: (10.0, 5.0),
8418 should_certify: mocks::application::Certifier::Always,
8419 };
8420 let (actor, application) = mocks::application::Application::new(
8421 context.child("application"),
8422 application_cfg,
8423 );
8424 actor.start();
8425 reporters.insert(idx, selected_reporter.clone());
8426 let blocker = oracle.control(validator.clone());
8427 let cfg = config::Config {
8428 scheme: schemes[idx].clone(),
8429 elector: elector.clone(),
8430 blocker,
8431 automaton: application.clone(),
8432 relay: application.clone(),
8433 reporter: selected_reporter,
8434 strategy: Sequential,
8435 partition: validator.to_string(),
8436 mailbox_size: NZUsize!(1024),
8437 epoch: Epoch::new(333),
8438 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
8439 Epoch::new(333),
8440 )),
8441 leader_timeout: Duration::from_secs(1),
8442 certification_timeout: Duration::from_secs(2),
8443 timeout_retry: Duration::from_secs(10),
8444 fetch_timeout: NZDuration!(Duration::from_secs(1)),
8445 fetch_retry_delay: Duration::from_millis(100),
8446 view_retention,
8447 skip: SkipPolicy::Enabled {
8448 timeout: skip_timeout,
8449 budget: SkipBudget::Participants,
8450 },
8451 replay_buffer: NZUsize!(1024 * 1024),
8452 write_buffer: NZUsize!(1024 * 1024),
8453 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
8454 forward: ForwardPolicy::Disabled,
8455 track_historical_votes: false,
8456 };
8457 let engine = Engine::new(context.child("engine"), cfg);
8458 engine_handlers.insert(idx, engine.start(pending, recovered, resolver));
8459
8460 let mut finalizers = Vec::new();
8462 for reporter in reporters.values_mut() {
8463 let (mut latest, mut monitor) = reporter.subscribe().await;
8464 finalizers.push(context.child("finalizer").spawn(move |_| async move {
8465 while latest < target {
8466 latest = monitor.recv().await.expect("event missing");
8467 }
8468 }));
8469 }
8470 join_all(finalizers).await;
8471 info!(idx, ?validator, "validator recovered");
8472 }
8473
8474 let latest_complete = target.saturating_sub(view_retention);
8476 for reporter in reporters.values() {
8477 reporter.assert_no_faults();
8479
8480 reporter.assert_no_invalid();
8482
8483 let mut notarized = HashMap::new();
8485 let mut finalized = HashMap::new();
8486 {
8487 let notarizes = reporter.notarizes.lock();
8488 for view in View::range(View::new(1), latest_complete) {
8489 let Some(payloads) = notarizes.get(&view) else {
8491 continue;
8492 };
8493 if payloads.len() > 1 {
8494 panic!("view: {view}");
8495 }
8496 let (digest, _) = payloads.iter().next().unwrap();
8497 notarized.insert(view, *digest);
8498 }
8499 }
8500 {
8501 let notarizations = reporter.notarizations.lock();
8502 for view in View::range(View::new(1), latest_complete) {
8503 let Some(notarization) = notarizations.get(&view) else {
8505 continue;
8506 };
8507 let Some(digest) = notarized.get(&view) else {
8508 continue;
8509 };
8510 assert_eq!(¬arization.proposal.payload, digest);
8511 }
8512 }
8513 {
8514 let finalizes = reporter.finalizes.lock();
8515 for view in View::range(View::new(1), latest_complete) {
8516 let Some(payloads) = finalizes.get(&view) else {
8518 continue;
8519 };
8520 if payloads.len() > 1 {
8521 panic!("view: {view}");
8522 }
8523 let (digest, _) = payloads.iter().next().unwrap();
8524 finalized.insert(view, *digest);
8525
8526 if view > latest_complete {
8528 continue;
8529 }
8530
8531 let nullifies = reporter.nullifies.lock();
8533 let Some(nullifies) = nullifies.get(&view) else {
8534 continue;
8535 };
8536 for finalizers in payloads.values() {
8537 for finalizer in finalizers.iter() {
8538 if nullifies.contains(finalizer) {
8539 panic!("should not nullify and finalize at same view");
8540 }
8541 }
8542 }
8543 }
8544 }
8545 {
8546 let finalizations = reporter.finalizations.lock();
8547 for view in View::range(View::new(1), latest_complete) {
8548 let Some((finalization, _)) = finalizations.get(&view) else {
8550 continue;
8551 };
8552 let Some(digest) = finalized.get(&view) else {
8553 continue;
8554 };
8555 assert_eq!(&finalization.proposal.payload, digest);
8556 }
8557 }
8558 }
8559
8560 let blocked = oracle.blocked().await.unwrap();
8562 assert!(blocked.is_empty());
8563
8564 context.auditor().state()
8566 })
8567 }
8568
8569 fn hailstorm<S, F, L>(fixture: F, elector: L)
8572 where
8573 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
8574 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S> + Copy,
8575 L: elector::Config<S>,
8576 {
8577 assert_eq!(
8578 run_hailstorm::<_, _, L>(0, 10, ViewDelta::new(15), elector.clone(), fixture,),
8579 run_hailstorm::<_, _, L>(0, 10, ViewDelta::new(15), elector, fixture),
8580 );
8581 }
8582
8583 test_for_all_fixtures!(hailstorm);
8584
8585 #[test_group("slow")]
8586 #[test_traced]
8587 fn test_hailstorm_stable_leader_ed25519() {
8588 assert_eq!(
8589 run_hailstorm::<_, _, RoundRobin>(
8590 0,
8591 10,
8592 ViewDelta::new(15),
8593 RoundRobin::default().with_term(
8594 TermLength::new(NZU32!(3)),
8595 Duration::from_secs(12),
8596 ViewDelta::new(0)
8597 ),
8598 ed25519::fixture
8599 ),
8600 run_hailstorm::<_, _, RoundRobin>(
8601 0,
8602 10,
8603 ViewDelta::new(15),
8604 RoundRobin::default().with_term(
8605 TermLength::new(NZU32!(3)),
8606 Duration::from_secs(12),
8607 ViewDelta::new(0)
8608 ),
8609 ed25519::fixture
8610 )
8611 );
8612 }
8613
8614 #[derive(Clone, Copy, Debug)]
8659 struct TwinsCampaign {
8660 n: u32,
8661 rounds: usize,
8662 mode: twins::Mode,
8663 max_cases: usize,
8664 trailing_finalizations: usize,
8665 handoffs: bool,
8666 }
8667
8668 fn configure_twins_handoff(
8673 actor: &mut mocks::application::Application<deterministic::Context, Sha256, PublicKey>,
8674 prefix_end: View,
8675 returned: Arc<AtomicUsize>,
8676 ) {
8677 actor.set_handoff(Handoff::Vote(()));
8678 actor.set_handoff_controller(Box::new(move |round, proposal, response| {
8679 if round.view() <= prefix_end {
8680 returned.fetch_add(1, Ordering::Relaxed);
8681 response.send_lossy(proposal);
8682 } else {
8683 response.send_lossy(Handoff::Wait);
8684 }
8685 }));
8686 }
8687
8688 fn twins_campaign<S, F, L>(
8689 rng: &mut impl CryptoRng,
8690 campaign: TwinsCampaign,
8691 elector: L,
8692 link: Link,
8693 mut fixture: F,
8694 ) where
8695 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
8696 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
8697 L: elector::Config<S>,
8698 {
8699 let honest_handoffs = Arc::new(AtomicUsize::new(0));
8700 let twin_handoffs = Arc::new(AtomicUsize::new(0));
8701 let early_vote_series = Arc::new(Mutex::new(0u64));
8702 let n = campaign.n;
8703 let faults = N3f1::max_faults(n) as usize;
8704 let cases = twins::cases(
8705 rng,
8706 twins::Framework {
8707 participants: n as usize,
8708 faults,
8709 rounds: campaign.rounds,
8710 mode: campaign.mode,
8711 max_cases: campaign.max_cases,
8712 },
8713 );
8714 assert!(
8715 !cases.is_empty(),
8716 "twins campaign should generate at least one case"
8717 );
8718 for case in cases {
8719 let scenario = case.scenario.clone();
8720 let twin_indices = case.compromised.clone();
8721 assert_eq!(
8722 twin_indices.len(),
8723 faults,
8724 "unexpected twins count for n={n} (expected f={faults})",
8725 );
8726
8727 let view_retention = ViewDelta::new(10);
8728 let skip_timeout = Duration::from_secs(11);
8729 let namespace = b"consensus".to_vec();
8730 let link = link.clone();
8731 let trailing_finalizations = campaign.trailing_finalizations;
8732 let elector = elector.clone();
8733 let honest_handoffs = honest_handoffs.clone();
8734 let twin_handoffs = twin_handoffs.clone();
8735 let early_vote_series = early_vote_series.clone();
8736 let mut case_fixture =
8737 |ctx: &mut deterministic::Context, ns: &[u8], n: u32| fixture(ctx, ns, n);
8738 let rng: deterministic::BoxDynRng = Box::new(StdRng::from_rng(&mut *rng));
8739 let cfg = deterministic::Config::new().with_rng(rng);
8740 let executor = deterministic::Runner::new(cfg);
8741 executor.start(|mut context| async move {
8742 let Fixture {
8743 participants,
8744 schemes,
8745 ..
8746 } = case_fixture(&mut context, &namespace, n);
8747 let participants: Arc<[_]> = participants.into();
8748 let mut oracle = start_test_network_with_peers::<S, _>(
8749 context.child("network"),
8750 participants.iter().cloned(),
8751 false,
8752 )
8753 .await;
8754 let mut registrations = register_validators(&mut oracle, &participants).await;
8755 link_validators(&mut oracle, &participants, Action::Link(link), None).await;
8756
8757 let term_length = elector
8761 .clone()
8762 .build(schemes[0].participants())
8763 .terms()
8764 .length();
8765 let elector = TwinsElector::new(
8766 elector.clone(),
8767 &scenario,
8768 n as usize,
8769 );
8770
8771 let prefix_end = View::new(Widen::widen(scenario.rounds().len()) * term_length.get());
8774
8775 let relay = Arc::new(mocks::relay::Relay::<Sha256Digest, _>::new());
8776 let mut reporters = Vec::new();
8777 let mut engine_handlers = Vec::new();
8778 let twin_index_set: HashSet<usize> = twin_indices.iter().copied().collect();
8779
8780 for idx in twin_indices.iter().copied() {
8782 let validator = &participants[idx];
8783 let (
8784 (vote_sender, vote_receiver),
8785 (certificate_sender, certificate_receiver),
8786 (_resolver_sender, _resolver_receiver),
8787 ) = registrations
8788 .remove(validator)
8789 .expect("validator should be registered");
8790
8791 let make_vote_forwarder = || {
8792 let participants = participants.clone();
8793 let scenario = scenario.clone();
8794 move |origin: SplitOrigin, _: &Recipients<_>, message: &IoBuf| {
8795 let msg: Vote<S, D> = Vote::decode(message.clone()).unwrap();
8796 let (primary, secondary) =
8797 scenario.partitions(msg.view(), term_length, participants.as_ref());
8798 match origin {
8799 SplitOrigin::Primary => Some(Recipients::Some(primary)),
8800 SplitOrigin::Secondary => Some(Recipients::Some(secondary)),
8801 }
8802 }
8803 };
8804 let make_certificate_forwarder = || {
8805 let codec = schemes[idx].certificate_codec_config();
8806 let participants = participants.clone();
8807 let scenario = scenario.clone();
8808 move |origin: SplitOrigin, _: &Recipients<_>, message: &IoBuf| {
8809 let msg: Certificate<S, D> =
8810 Certificate::decode_cfg(message.clone(), &codec).unwrap();
8811 let (primary, secondary) =
8812 scenario.partitions(msg.view(), term_length, participants.as_ref());
8813 match origin {
8814 SplitOrigin::Primary => Some(Recipients::Some(primary)),
8815 SplitOrigin::Secondary => Some(Recipients::Some(secondary)),
8816 }
8817 }
8818 };
8819 let make_vote_router = || {
8820 let participants = participants.clone();
8821 let scenario = scenario.clone();
8822 move |(sender, message): &(_, IoBuf)| {
8823 let msg: Vote<S, D> = Vote::decode(message.clone()).unwrap();
8824 scenario.route(msg.view(), term_length, sender, participants.as_ref())
8825 }
8826 };
8827 let make_certificate_router = || {
8828 let codec = schemes[idx].certificate_codec_config();
8829 let participants = participants.clone();
8830 let scenario = scenario.clone();
8831 move |(sender, message): &(_, IoBuf)| {
8832 let msg: Certificate<S, D> =
8833 Certificate::decode_cfg(message.clone(), &codec).unwrap();
8834 scenario.route(msg.view(), term_length, sender, participants.as_ref())
8835 }
8836 };
8837 let (vote_sender_primary, vote_sender_secondary) =
8838 vote_sender.split_with(make_vote_forwarder());
8839 let (vote_receiver_primary, vote_receiver_secondary) = vote_receiver
8840 .split_with(
8841 context.child("pending_split").with_attribute("index", idx),
8842 make_vote_router(),
8843 );
8844 let (certificate_sender_primary, certificate_sender_secondary) =
8845 certificate_sender.split_with(make_certificate_forwarder());
8846 let (certificate_receiver_primary, certificate_receiver_secondary) =
8847 certificate_receiver.split_with(
8848 context
8849 .child("recovered_split")
8850 .with_attribute("index", idx),
8851 make_certificate_router(),
8852 );
8853
8854 for (twin_label, pending, recovered) in [
8855 (
8856 "primary",
8857 (vote_sender_primary, vote_receiver_primary),
8858 (certificate_sender_primary, certificate_receiver_primary),
8859 ),
8860 (
8861 "secondary",
8862 (vote_sender_secondary, vote_receiver_secondary),
8863 (certificate_sender_secondary, certificate_receiver_secondary),
8864 ),
8865 ] {
8866 let partition = format!("twin_{idx}_{twin_label}");
8867 let context = context
8868 .child("twin")
8869 .with_attribute("index", idx)
8870 .with_attribute("side", twin_label);
8871
8872 let reporter_config = mocks::reporter::Config {
8873 participants: participants.as_ref().try_into().unwrap(),
8874 scheme: schemes[idx].clone(),
8875 elector: elector.clone(),
8876 };
8877 let reporter = mocks::reporter::Reporter::new(
8878 context.child("reporter"),
8879 reporter_config,
8880 );
8881 reporters.push(reporter.clone());
8882
8883 let application_cfg = mocks::application::Config::<Sha256, _> {
8884 relay: relay.clone(),
8885 me: validator.clone(),
8886 propose_latency: (10.0, 5.0),
8887 verify_latency: (10.0, 5.0),
8888 certify_latency: (10.0, 5.0),
8889 should_certify: mocks::application::Certifier::Always,
8890 };
8891 let (mut actor, application) = mocks::application::Application::new(
8892 context.child("application"),
8893 application_cfg,
8894 );
8895 if campaign.handoffs {
8896 configure_twins_handoff(&mut actor, prefix_end, twin_handoffs.clone());
8897 }
8898 actor.start();
8899
8900 let blocker = oracle.control(validator.clone());
8901 let cfg = config::Config {
8902 scheme: schemes[idx].clone(),
8903 elector: elector.clone(),
8904 blocker,
8905 automaton: application.clone(),
8906 relay: application.clone(),
8907 reporter: reporter.clone(),
8908 strategy: Sequential,
8909 partition,
8910 mailbox_size: NZUsize!(1024),
8911 epoch: Epoch::new(333),
8912 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
8913 Epoch::new(333),
8914 )),
8915 leader_timeout: Duration::from_secs(1),
8916 certification_timeout: Duration::from_millis(1_500),
8917 timeout_retry: Duration::from_secs(10),
8918 fetch_timeout: NZDuration!(Duration::from_secs(1)),
8919 fetch_retry_delay: Duration::from_millis(100),
8920 view_retention,
8921 skip: SkipPolicy::Enabled {
8922 timeout: skip_timeout,
8923 budget: SkipBudget::Participants,
8924 },
8925 replay_buffer: NZUsize!(1024 * 1024),
8926 write_buffer: NZUsize!(1024 * 1024),
8927 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
8928 forward: ForwardPolicy::Disabled,
8929 track_historical_votes: false,
8930 };
8931 let engine = Engine::new(context.child("engine"), cfg);
8932 engine_handlers.push(engine.start(
8933 pending,
8934 recovered,
8935 inert_channel(participants.as_ref()),
8936 ));
8937 }
8938 }
8939
8940 let honest_start = reporters.len();
8942 for (idx, validator) in participants.iter().enumerate() {
8943 if twin_index_set.contains(&idx) {
8944 continue;
8945 }
8946
8947 let partition = format!("honest_{idx}");
8948 let context = context.child("honest").with_attribute("index", idx);
8949
8950 let reporter_config = mocks::reporter::Config {
8951 participants: participants.as_ref().try_into().unwrap(),
8952 scheme: schemes[idx].clone(),
8953 elector: elector.clone(),
8954 };
8955 let reporter =
8956 mocks::reporter::Reporter::new(context.child("reporter"), reporter_config);
8957 reporters.push(reporter.clone());
8958
8959 let application_cfg = mocks::application::Config::<Sha256, _> {
8960 relay: relay.clone(),
8961 me: validator.clone(),
8962 propose_latency: (10.0, 5.0),
8963 verify_latency: (10.0, 5.0),
8964 certify_latency: (10.0, 5.0),
8965 should_certify: mocks::application::Certifier::Always,
8966 };
8967 let (mut actor, application) = mocks::application::Application::new(
8968 context.child("application"),
8969 application_cfg,
8970 );
8971 if campaign.handoffs {
8972 configure_twins_handoff(&mut actor, prefix_end, honest_handoffs.clone());
8973 }
8974 actor.start();
8975
8976 let blocker = oracle.control(validator.clone());
8977 let cfg = config::Config {
8978 scheme: schemes[idx].clone(),
8979 elector: elector.clone(),
8980 blocker,
8981 automaton: application.clone(),
8982 relay: application.clone(),
8983 reporter: reporter.clone(),
8984 strategy: Sequential,
8985 partition,
8986 mailbox_size: NZUsize!(1024),
8987 epoch: Epoch::new(333),
8988 floor: config::Floor::Genesis(mocks::application::genesis::<Sha256>(
8989 Epoch::new(333),
8990 )),
8991 leader_timeout: Duration::from_secs(1),
8992 certification_timeout: Duration::from_millis(1_500),
8993 timeout_retry: Duration::from_secs(10),
8994 fetch_timeout: NZDuration!(Duration::from_secs(1)),
8995 fetch_retry_delay: Duration::from_millis(100),
8996 view_retention,
8997 skip: SkipPolicy::Enabled {
8998 timeout: skip_timeout,
8999 budget: SkipBudget::Participants,
9000 },
9001 replay_buffer: NZUsize!(1024 * 1024),
9002 write_buffer: NZUsize!(1024 * 1024),
9003 page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
9004 forward: ForwardPolicy::Disabled,
9005 track_historical_votes: false,
9006 };
9007 let engine = Engine::new(context.child("engine"), cfg);
9008
9009 let (
9010 (pending_sender, pending_receiver),
9011 (recovered_sender, recovered_receiver),
9012 _,
9013 ) = registrations
9014 .remove(validator)
9015 .expect("validator should be registered");
9016 engine_handlers.push(engine.start(
9017 (pending_sender, pending_receiver),
9018 (recovered_sender, recovered_receiver),
9019 inert_channel(participants.as_ref()),
9020 ));
9021 }
9022
9023 let mut finalizers = Vec::new();
9033 for (i, reporter) in reporters.iter_mut().skip(honest_start).enumerate() {
9034 let (_latest, mut monitor) = reporter.subscribe().await;
9035 let required = trailing_finalizations;
9036 finalizers.push(context.child("finalizer").with_attribute("index", i).spawn(
9037 move |_| async move {
9038 let mut count = 0usize;
9039 while count < required {
9040 let view = monitor.recv().await.expect("event missing");
9041 if view > prefix_end {
9042 count += 1;
9043 }
9044 }
9045 },
9046 ));
9047 }
9048 join_all(finalizers).await;
9049 if campaign.handoffs {
9050 *early_vote_series.lock() += u64::from(count_nonzero_metric_lines(
9051 &context.encode(),
9052 &[
9053 "_handoff_events_total{",
9054 "event=\"VotedBeforeCertification\"",
9055 ],
9056 ));
9057 }
9058
9059 let mut finalized_at_view: BTreeMap<View, D> = BTreeMap::new();
9061 for reporter in reporters.iter().skip(honest_start) {
9062 let finalizations = reporter.finalizations.lock();
9063 for (view, (finalization, _)) in finalizations.iter() {
9064 let digest = finalization.proposal.payload;
9065 if let Some(existing) = finalized_at_view.get(view) {
9066 assert_eq!(
9067 existing, &digest,
9068 "safety violation: conflicting finalizations at view {view}"
9069 );
9070 } else {
9071 finalized_at_view.insert(*view, digest);
9072 }
9073 }
9074 }
9075
9076 for reporter in reporters.iter().skip(honest_start) {
9078 reporter.assert_no_invalid();
9079 }
9080
9081 let twin_keys: HashSet<_> = twin_indices
9083 .iter()
9084 .map(|idx| participants[*idx].clone())
9085 .collect();
9086 let mut notarized_by_signer: BTreeMap<View, HashMap<PublicKey, D>> =
9087 BTreeMap::new();
9088 let mut finalized_by_signer: BTreeMap<View, HashMap<PublicKey, D>> =
9089 BTreeMap::new();
9090 for reporter in reporters.iter().skip(honest_start) {
9091 let notarizes = reporter.notarizes.lock();
9092 for (view, payloads) in notarizes.iter() {
9093 let signers = notarized_by_signer.entry(*view).or_default();
9094 for (digest, payload_signers) in payloads.iter() {
9095 for signer in payload_signers.iter() {
9096 if twin_keys.contains(signer) {
9097 continue;
9098 }
9099 if let Some(existing) = signers.insert(signer.clone(), *digest) {
9100 assert_eq!(
9101 existing, *digest,
9102 "honest signer produced conflicting notarizes at view {view}"
9103 );
9104 }
9105 }
9106 }
9107 }
9108
9109 let finalizes = reporter.finalizes.lock();
9110 for (view, payloads) in finalizes.iter() {
9111 let signers = finalized_by_signer.entry(*view).or_default();
9112 for (digest, payload_signers) in payloads.iter() {
9113 for signer in payload_signers.iter() {
9114 if twin_keys.contains(signer) {
9115 continue;
9116 }
9117 if let Some(existing) = signers.insert(signer.clone(), *digest) {
9118 assert_eq!(
9119 existing, *digest,
9120 "honest signer produced conflicting finalizes at view {view}"
9121 );
9122 }
9123 }
9124 }
9125 }
9126 }
9127
9128 for reporter in reporters.iter().skip(honest_start) {
9130 let faults = reporter.faults.lock();
9131 for faulter in faults.keys() {
9132 assert!(
9133 twin_keys.contains(faulter),
9134 "fault from non-twin participant"
9135 );
9136 }
9137 }
9138
9139 let blocked = oracle.blocked().await.unwrap();
9140 for (_, faulter) in blocked {
9141 assert!(
9142 twin_keys.contains(&faulter),
9143 "blocked peer attributed to non-twin participant"
9144 );
9145 }
9146 });
9147 }
9148 if campaign.handoffs {
9149 assert!(
9150 *early_vote_series.lock() > 0,
9151 "campaign must vote for prefix handoffs before certification"
9152 );
9153 assert!(
9154 honest_handoffs.load(Ordering::Relaxed) > 0,
9155 "honest apps must return handoffs during the adversarial prefix"
9156 );
9157 assert!(
9158 twin_handoffs.load(Ordering::Relaxed) > 0,
9159 "twin apps must return handoffs during the adversarial prefix"
9160 );
9161 }
9162 }
9163
9164 const TWINS_CAMPAIGN: TwinsCampaign = TwinsCampaign {
9165 n: 5,
9166 rounds: 3,
9167 mode: twins::Mode::Sampled,
9168 max_cases: 20,
9169 trailing_finalizations: 10,
9170 handoffs: false,
9171 };
9172
9173 const TWINS_LINK: Link = Link {
9174 latency: Duration::from_millis(500),
9175 jitter: Duration::from_millis(500),
9176 success_rate: probability!(1.0),
9177 };
9178
9179 fn twins_campaign_all_links(campaign: TwinsCampaign, elector: RoundRobin) {
9181 for link in [
9182 Link {
9183 latency: Duration::from_millis(10),
9184 jitter: Duration::from_millis(10),
9185 success_rate: probability!(1.0),
9186 },
9187 TWINS_LINK,
9188 ] {
9189 twins_campaign::<_, _, RoundRobin>(
9190 &mut test_rng(),
9191 campaign,
9192 elector.clone(),
9193 link,
9194 scheme_mocks::fixture,
9195 );
9196 }
9197 }
9198
9199 #[test_group("slow")]
9200 #[test_traced("INFO")]
9201 fn test_twins_sampled() {
9202 twins_campaign_all_links(TWINS_CAMPAIGN, RoundRobin::default());
9203 }
9204
9205 #[test_group("slow")]
9206 #[test_traced("INFO")]
9207 fn test_twins_pipelined_handoff() {
9208 twins_campaign_all_links(
9209 TwinsCampaign {
9210 handoffs: true,
9211 ..TWINS_CAMPAIGN
9212 },
9213 RoundRobin::default(),
9214 );
9215 }
9216
9217 #[test_group("slow")]
9218 #[test_traced("INFO")]
9219 fn test_twins_sustained() {
9220 twins_campaign_all_links(
9221 TwinsCampaign {
9222 mode: twins::Mode::Sustained,
9223 ..TWINS_CAMPAIGN
9224 },
9225 RoundRobin::default(),
9226 );
9227 }
9228
9229 #[test_group("slow")]
9230 #[test_traced("INFO")]
9231 fn test_twins_stable_leader() {
9232 twins_campaign_all_links(
9233 TWINS_CAMPAIGN,
9234 RoundRobin::default().with_term(
9235 TermLength::new(NZU32!(3)),
9236 Duration::from_secs(12),
9237 ViewDelta::new(0),
9238 ),
9239 );
9240 }
9241
9242 #[test_group("slow")]
9243 #[test_traced("INFO")]
9244 fn test_twins_stable_leader_optimistic() {
9245 twins_campaign_all_links(
9246 TWINS_CAMPAIGN,
9247 RoundRobin::default().with_term(
9248 TermLength::new(NZU32!(3)),
9249 Duration::from_secs(12),
9250 ViewDelta::new(2),
9251 ),
9252 );
9253 }
9254
9255 #[test_group("slow")]
9256 #[test_traced("INFO")]
9257 fn test_twins_large_sampled() {
9258 let campaign = TwinsCampaign {
9259 n: 10,
9260 rounds: 5,
9261 ..TWINS_CAMPAIGN
9262 };
9263 twins_campaign::<_, _, RoundRobin>(
9264 &mut test_rng(),
9265 campaign,
9266 RoundRobin::default(),
9267 TWINS_LINK,
9268 scheme_mocks::fixture,
9269 );
9270 }
9271
9272 #[test_group("slow")]
9273 #[test_traced("INFO")]
9274 fn test_twins_large_sustained() {
9275 let campaign = TwinsCampaign {
9276 n: 10,
9277 rounds: 5,
9278 mode: twins::Mode::Sustained,
9279 ..TWINS_CAMPAIGN
9280 };
9281 twins_campaign::<_, _, RoundRobin>(
9282 &mut test_rng(),
9283 campaign,
9284 RoundRobin::default(),
9285 TWINS_LINK,
9286 scheme_mocks::fixture,
9287 );
9288 }
9289
9290 fn twins<S, F, L>(fixture: F, elector: L)
9291 where
9292 S: Scheme<Sha256Digest, PublicKey = PublicKey>,
9293 F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
9294 L: elector::Config<S>,
9295 {
9296 twins_campaign::<_, _, L>(
9297 &mut test_rng(),
9298 TWINS_CAMPAIGN,
9299 elector,
9300 TWINS_LINK,
9301 fixture,
9302 );
9303 }
9304
9305 test_for_all_fixtures!(twins, level = "INFO");
9306
9307 #[test]
9308 fn test_viewport() {
9309 let viewport = Viewport {
9310 finalized: View::new(20),
9311 current: View::new(25),
9312 view_retention: ViewDelta::new(10),
9313 lookahead: Lookahead {
9314 term_length: TermLength::new(commonware_utils::NZU32!(10)),
9315 optimistic_views: ViewDelta::new(0),
9316 },
9317 };
9318
9319 assert!(!viewport.retains(View::zero()));
9321
9322 assert_eq!(viewport.floor(), View::new(10));
9324 assert!(!viewport.retains(View::new(9)));
9325 assert!(viewport.retains(View::new(10)));
9326
9327 assert!(viewport.admits_vote(View::new(10)));
9330 assert!(viewport.admits_vote(View::new(25)));
9331 assert!(viewport.admits_vote(View::new(26)));
9332 assert!(viewport.admits_vote(View::new(31)));
9333 assert!(viewport.admits_vote(View::new(33)));
9334 assert!(!viewport.admits_vote(View::new(5)));
9335 assert!(!viewport.admits_vote(View::new(27)));
9336 assert!(!viewport.admits_vote(View::new(34)));
9337
9338 assert!(!viewport.admits_certificate(View::new(9)));
9341 assert!(viewport.admits_certificate(View::new(10_000)));
9342
9343 let optimistic = Viewport {
9347 lookahead: Lookahead {
9348 optimistic_views: ViewDelta::new(2),
9349 ..viewport.lookahead
9350 },
9351 ..viewport
9352 };
9353 assert!(optimistic.admits_vote(View::new(26)));
9354 assert!(optimistic.admits_vote(View::new(30)));
9355 assert!(optimistic.admits_vote(View::new(34)));
9356 assert!(!optimistic.admits_vote(View::new(35)));
9357
9358 let term_edge = Viewport {
9360 current: View::new(29),
9361 lookahead: Lookahead {
9362 optimistic_views: ViewDelta::new(5),
9363 ..viewport.lookahead
9364 },
9365 ..viewport
9366 };
9367 assert!(term_edge.admits_vote(View::new(30)));
9368 assert!(term_edge.admits_vote(View::new(37)));
9369 assert!(!term_edge.admits_vote(View::new(38)));
9370 }
9371
9372 fn slow_certifiers_finalize_without_nullification(
9379 namespace: &'static [u8],
9380 elector: RoundRobin<Sha256>,
9381 handoff: Handoff<()>,
9382 ) {
9383 let executor = deterministic::Runner::timed(Duration::from_secs(120));
9384 executor.start(|mut context| async move {
9385 let (reporters, _, oracle) = setup_round_robin_cluster(
9387 &mut context,
9388 namespace,
9389 Link {
9390 latency: Duration::from_millis(10),
9391 jitter: Duration::from_millis(1),
9392 success_rate: probability!(1.0),
9393 },
9394 elector,
9395 (10.0, 5.0),
9396 |idx| if idx >= 3 { (80.0, 5.0) } else { (10.0, 5.0) },
9397 handoff,
9398 )
9399 .await;
9400 context.sleep(Duration::from_secs(30)).await;
9401
9402 for (idx, reporter) in reporters.iter().enumerate() {
9404 reporter.assert_no_invalid();
9405 reporter.assert_no_faults();
9406 assert!(
9407 reporter.nullifications.lock().is_empty(),
9408 "reporter {idx} observed nullifications"
9409 );
9410 let finalized = reporter.finalizations.lock().len();
9411 assert!(
9412 finalized >= 200,
9413 "reporter {idx} finalized only {finalized} views"
9414 );
9415 }
9416 let blocked = oracle.blocked().await.unwrap();
9417 assert!(blocked.is_empty());
9418 });
9419 }
9420
9421 #[test_traced]
9424 fn test_stable_leader_optimistic_tolerates_slow_certifiers() {
9425 slow_certifiers_finalize_without_nullification(
9426 b"consensus_stable_leader_slow_certifiers",
9427 RoundRobin::<Sha256>::default().with_term(
9428 TermLength::new(NZU32!(50)),
9429 Duration::from_secs(30),
9430 ViewDelta::new(2),
9431 ),
9432 Handoff::Wait,
9433 );
9434 }
9435
9436 #[test_traced]
9439 fn test_stable_leader_handoff_vote_tolerates_slow_certifiers() {
9440 slow_certifiers_finalize_without_nullification(
9441 b"consensus_stable_leader_handoff_slow_certifiers",
9442 RoundRobin::<Sha256>::default().with_term(
9443 TermLength::new(NZU32!(7)),
9444 Duration::from_secs(30),
9445 ViewDelta::new(2),
9446 ),
9447 Handoff::Vote(()),
9448 );
9449 }
9450
9451 #[test_traced]
9452 fn test_rotating_handoff_vote_tolerates_slow_certifiers() {
9453 slow_certifiers_finalize_without_nullification(
9454 b"consensus_rotating_handoff_slow_certifiers",
9455 RoundRobin::<Sha256>::default(),
9456 Handoff::Vote(()),
9457 );
9458 }
9459
9460 #[test_traced]
9464 fn test_short_term_handoff_vote_tolerates_slow_certifiers() {
9465 slow_certifiers_finalize_without_nullification(
9466 b"consensus_short_term_handoff_slow_certifiers",
9467 RoundRobin::<Sha256>::default().with_term(
9468 TermLength::new(NZU32!(3)),
9469 Duration::from_secs(30),
9470 ViewDelta::new(2),
9471 ),
9472 Handoff::Vote(()),
9473 );
9474 }
9475
9476 #[test]
9477 fn test_lookahead_admits_next_term_lane() {
9478 let lookahead = test_lookahead(2);
9483 let current = View::new(9);
9484 assert!(lookahead.admits(current, View::new(11)));
9485 assert!(lookahead.admits(current, View::new(14)));
9486 assert!(!lookahead.admits(current, View::new(15)));
9487
9488 let rotating = Lookahead {
9492 term_length: TermLength::ONE,
9493 optimistic_views: ViewDelta::zero(),
9494 };
9495 assert!(rotating.admits(View::new(6), View::new(8)));
9496 assert!(rotating.admits(View::new(25), View::new(28)));
9497 assert!(!rotating.admits(View::new(25), View::new(29)));
9498 }
9499}