Oregami
Repositories/oxedyne/fe2o3

oxedyne/fe2o3/fe2o3_o3db_sync/tests/dist_hotstuff.rs

15.7 KiB, 91 runs

created by r1870400018:11230, which is this file's identity for as long as the history lasts, whatever it is later renamed to

download · who wrote it · its history

1#![cfg(feature = "dist")]
2//! Integration tests for the HotStuff primitive.
3//!
4//! [Written with AI entirely](https://need2know.ai/entirely-ai/code)\
5//! Anthropic Claude
6
7use oxedyne_fe2o3_core::prelude::*;
8
9use oxedyne_fe2o3_o3db_sync::dist::hotstuff::{
10 replica::{
11 Command,
12 Config,
13 Replica,
14 },
15 types::{
16 BLOCK_HASH_LEN,
17 BlockHash,
18 NewView,
19 Phase,
20 Proposal,
21 Qc,
22 ReplicaId,
23 Vote,
24 },
25};
26
27use std::collections::VecDeque;
28
29
30#[derive(Clone, Debug)]
31enum Delivery {
32 Proposal(Proposal),
33 Vote { to: ReplicaId, vote: Vote },
34 NewView { to: ReplicaId, new_view: NewView },
35}
36
37/// Routes each emitted command into a FIFO and runs until the queue drains or
38/// every replica has decided.
39struct Driver {
40 replicas: Vec<Replica>,
41 queue: VecDeque<Delivery>,
42 decided: Vec<Option<Vec<u8>>>,
43}
44
45impl Driver {
46 fn new(cohort_size: usize, f: usize) -> Outcome<Self> {
47 let mut replicas = Vec::with_capacity(cohort_size);
48 for id in 0..cohort_size {
49 let cfg = Config {
50 cohort_size,
51 f,
52 self_id: id as ReplicaId,
53 };
54 replicas.push(res!(Replica::new(cfg)));
55 }
56 Ok(Self {
57 replicas,
58 queue: VecDeque::new(),
59 decided: vec![None; cohort_size],
60 })
61 }
62
63 fn handle_commands(&mut self, from: ReplicaId, cmds: Vec<Command>) {
64 for cmd in cmds {
65 match cmd {
66 Command::BroadcastProposal(p) => {
67 self.queue.push_back(Delivery::Proposal(p));
68 },
69 Command::SendVote { to, vote } => {
70 self.queue.push_back(Delivery::Vote { to, vote });
71 },
72 Command::SendNewView { to, new_view } => {
73 self.queue.push_back(Delivery::NewView { to, new_view });
74 },
75 Command::Decide { block, .. } => {
76 self.decided[from as usize] = Some(block);
77 },
78 }
79 }
80 }
81
82 fn run(&mut self) -> Outcome<()> {
83 let max_steps = 10_000;
84 let mut steps = 0;
85 while let Some(d) = self.queue.pop_front() {
86 steps += 1;
87 if steps > max_steps {
88 return Err(err!(
89 "Driver exceeded {} steps without terminating.", max_steps;
90 Bug, Timeout));
91 }
92 ok!(self.deliver(d));
93 if self.decided.iter().all(|d| d.is_some()) {
94 break;
95 }
96 }
97 Ok(())
98 }
99
100 fn deliver(&mut self, d: Delivery) -> Outcome<()> {
101 match d {
102 Delivery::Proposal(p) => {
103 let cohort = self.replicas.len();
104 for id in 0..cohort {
105 let cmds = res!(self.replicas[id].on_proposal(p.clone()));
106 self.handle_commands(id as ReplicaId, cmds);
107 }
108 },
109 Delivery::Vote { to, vote } => {
110 let cmds = res!(self.replicas[to as usize].on_vote(vote));
111 self.handle_commands(to, cmds);
112 },
113 Delivery::NewView { to, new_view } => {
114 let cmds = res!(self.replicas[to as usize].on_new_view(new_view));
115 self.handle_commands(to, cmds);
116 },
117 }
118 Ok(())
119 }
120
121 /// Simulates a cohort-wide timeout: every replica that has not yet decided
122 /// fires, and the resulting NewView commands go into the queue.
123 fn timeout_all(&mut self) -> Outcome<()> {
124 let cohort = self.replicas.len();
125 for id in 0..cohort {
126 if self.decided[id].is_some() {
127 continue;
128 }
129 let cmds = res!(self.replicas[id].on_timeout());
130 self.handle_commands(id as ReplicaId, cmds);
131 }
132 Ok(())
133 }
134}
135
136fn fixed_block_hash(seed: u8) -> BlockHash {
137 let mut h = [0u8; BLOCK_HASH_LEN];
138 for (i, cell) in h.iter_mut().enumerate() {
139 *cell = seed.wrapping_add(i as u8);
140 }
141 h
142}
143
144
145// --- configuration ---------------------------------------------------------
146
147#[test]
148fn config_rejects_degenerate() -> Outcome<()> {
149 assert!(Replica::new(Config { cohort_size: 0, f: 0, self_id: 0 }).is_err());
150 assert!(Replica::new(Config { cohort_size: 3, f: 1, self_id: 0 }).is_err());
151 assert!(Replica::new(Config { cohort_size: 4, f: 1, self_id: 4 }).is_err());
152 Ok(())
153}
154
155#[test]
156fn config_quorum_follows_lambda_minus_z() -> Outcome<()> {
157 let cfg5 = Config { cohort_size: 5, f: 1, self_id: 0 };
158 let cfg7 = Config { cohort_size: 7, f: 2, self_id: 0 };
159 let cfg9 = Config { cohort_size: 9, f: 2, self_id: 0 };
160 assert_eq!(cfg5.quorum(), 4);
161 assert_eq!(cfg7.quorum(), 5);
162 assert_eq!(cfg9.quorum(), 7);
163 Ok(())
164}
165
166#[test]
167fn leader_rotates_round_robin() -> Outcome<()> {
168 let cfg = Config { cohort_size: 5, f: 1, self_id: 0 };
169 assert_eq!(cfg.leader_for(1), 0);
170 assert_eq!(cfg.leader_for(2), 1);
171 assert_eq!(cfg.leader_for(3), 2);
172 assert_eq!(cfg.leader_for(6), 0); // wraps
173 assert!(cfg.is_leader_for(1));
174 assert!(!cfg.is_leader_for(2));
175 Ok(())
176}
177
178
179// --- happy path ------------------------------------------------------------
180
181#[test]
182fn propose_rejected_on_non_leader() -> Outcome<()> {
183 let cfg = Config { cohort_size: 5, f: 1, self_id: 1 };
184 let mut r = res!(Replica::new(cfg));
185 assert!(r.propose(b"x".to_vec(), fixed_block_hash(1)).is_err());
186 Ok(())
187}
188
189#[test]
190fn five_replica_cohort_reaches_decide() -> Outcome<()> {
191 let mut drv = res!(Driver::new(5, 1));
192 let block = b"consensus input".to_vec();
193 let h = fixed_block_hash(7);
194 let cmds = res!(drv.replicas[0].propose(block.clone(), h));
195 drv.handle_commands(0, cmds);
196 res!(drv.run());
197 for (i, d) in drv.decided.iter().enumerate() {
198 let got = match d {
199 Some(b) => b,
200 None => return Err(err!("replica {} did not decide", i; Bug, Fatal)),
201 };
202 assert_eq!(got, &block);
203 }
204 Ok(())
205}
206
207#[test]
208fn seven_replica_cohort_reaches_decide() -> Outcome<()> {
209 let mut drv = res!(Driver::new(7, 2));
210 let block = b"more replicas".to_vec();
211 let h = fixed_block_hash(3);
212 let cmds = res!(drv.replicas[0].propose(block.clone(), h));
213 drv.handle_commands(0, cmds);
214 res!(drv.run());
215 for d in &drv.decided {
216 assert_eq!(d.as_deref(), Some(block.as_slice()));
217 }
218 Ok(())
219}
220
221#[test]
222fn nine_replica_cohort_reaches_decide() -> Outcome<()> {
223 let mut drv = res!(Driver::new(9, 2));
224 let block = b"nine nodes".to_vec();
225 let h = fixed_block_hash(12);
226 let cmds = res!(drv.replicas[0].propose(block.clone(), h));
227 drv.handle_commands(0, cmds);
228 res!(drv.run());
229 for d in &drv.decided {
230 assert_eq!(d.as_deref(), Some(block.as_slice()));
231 }
232 Ok(())
233}
234
235#[test]
236fn duplicate_votes_are_idempotent() -> Outcome<()> {
237 let mut drv = res!(Driver::new(5, 1));
238 let block = b"test duplicates".to_vec();
239 let h = fixed_block_hash(5);
240 let cmds = res!(drv.replicas[0].propose(block.clone(), h));
241 drv.handle_commands(0, cmds);
242 let first = match drv.queue.pop_front() {
243 Some(Delivery::Proposal(p)) => p,
244 _ => return Err(err!("expected initial Prepare proposal"; Bug)),
245 };
246 let mut first_vote: Option<Vote> = None;
247 for id in 0..drv.replicas.len() {
248 let out = res!(drv.replicas[id].on_proposal(first.clone()));
249 for cmd in out {
250 if let Command::SendVote { vote, .. } = &cmd {
251 if first_vote.is_none() {
252 first_vote = Some(vote.clone());
253 }
254 }
255 drv.handle_commands(id as ReplicaId, vec![cmd]);
256 }
257 }
258 let v = match first_vote {
259 Some(v) => v,
260 None => return Err(err!("no vote captured"; Bug)),
261 };
262 let _ = res!(drv.replicas[0].on_vote(v.clone()));
263 let _ = res!(drv.replicas[0].on_vote(v));
264 res!(drv.run());
265 for d in &drv.decided {
266 assert_eq!(d.as_deref(), Some(block.as_slice()));
267 }
268 Ok(())
269}
270
271
272// --- structural rejection --------------------------------------------------
273
274#[test]
275fn out_of_range_voter_rejected() -> Outcome<()> {
276 let cfg = Config { cohort_size: 5, f: 1, self_id: 0 };
277 let mut leader = res!(Replica::new(cfg));
278 let bogus = Vote {
279 view: 1,
280 phase: Phase::Prepare,
281 block_hash: fixed_block_hash(9),
282 voter: 99,
283 signature: Vec::new(),
284 };
285 assert!(leader.on_vote(bogus).is_err());
286 Ok(())
287}
288
289#[test]
290fn prepare_without_block_is_rejected() -> Outcome<()> {
291 let cfg = Config { cohort_size: 5, f: 1, self_id: 1 };
292 let mut r = res!(Replica::new(cfg));
293 let p = Proposal {
294 view: 1,
295 phase: Phase::Prepare,
296 block_hash: fixed_block_hash(1),
297 block: None,
298 justify: None,
299 };
300 assert!(r.on_proposal(p).is_err());
301 Ok(())
302}
303
304#[test]
305fn precommit_without_justify_is_rejected() -> Outcome<()> {
306 let cfg = Config { cohort_size: 5, f: 1, self_id: 1 };
307 let mut r = res!(Replica::new(cfg));
308 let p = Proposal {
309 view: 1,
310 phase: Phase::PreCommit,
311 block_hash: fixed_block_hash(1),
312 block: None,
313 justify: None,
314 };
315 assert!(r.on_proposal(p).is_err());
316 Ok(())
317}
318
319
320// --- QC validation ---------------------------------------------------------
321
322#[test]
323fn qc_validate_catches_duplicate_voter() -> Outcome<()> {
324 let qc = Qc {
325 view: 1,
326 phase: Phase::Prepare,
327 block_hash: fixed_block_hash(1),
328 signatures: vec![(0, vec![]), (2, vec![]), (2, vec![]), (3, vec![])],
329 };
330 assert!(qc.validate(1, Phase::Prepare, &fixed_block_hash(1), 3, 5).is_err());
331 Ok(())
332}
333
334#[test]
335fn qc_validate_catches_out_of_order_voter() -> Outcome<()> {
336 let qc = Qc {
337 view: 1,
338 phase: Phase::Prepare,
339 block_hash: fixed_block_hash(1),
340 signatures: vec![(2, vec![]), (0, vec![]), (3, vec![])],
341 };
342 assert!(qc.validate(1, Phase::Prepare, &fixed_block_hash(1), 3, 5).is_err());
343 Ok(())
344}
345
346#[test]
347fn qc_validate_catches_insufficient_quorum() -> Outcome<()> {
348 let qc = Qc {
349 view: 1,
350 phase: Phase::Prepare,
351 block_hash: fixed_block_hash(1),
352 signatures: vec![(0, vec![]), (1, vec![])],
353 };
354 assert!(qc.validate(1, Phase::Prepare, &fixed_block_hash(1), 4, 5).is_err());
355 Ok(())
356}
357
358
359// --- view change -----------------------------------------------------------
360
361#[test]
362fn silent_leader_recovers_via_view_change() -> Outcome<()> {
363 // Leader 0 stays silent; everyone times out; leader 1 takes over with
364 // a fresh block.
365 let mut drv = res!(Driver::new(5, 1));
366 res!(drv.timeout_all());
367 res!(drv.run());
368 // After timeout, queue has 5 NewView deliveries to leader 1. Leader 1
369 // collects quorum, has no prepare_qc from anyone, is awaiting a fresh
370 // block.
371 assert!(drv.replicas[1].awaiting_fresh_block(),
372 "leader 1 should be awaiting a fresh block after timeout_all");
373 assert_eq!(drv.replicas[1].view(), 2);
374
375 // Leader 1 now proposes a fresh block.
376 let block = b"second view input".to_vec();
377 let h = fixed_block_hash(22);
378 let cmds = res!(drv.replicas[1].propose(block.clone(), h));
379 drv.handle_commands(1, cmds);
380 res!(drv.run());
381 for d in &drv.decided {
382 assert_eq!(d.as_deref(), Some(block.as_slice()));
383 }
384 Ok(())
385}
386
387#[test]
388fn view_change_preserves_pinned_block_when_prepare_qc_exists() -> Outcome<()> {
389 // Leader 0 proposes and everyone reaches PreCommit stage (i.e. prepare
390 // QC is formed and distributed). Then they time out. Leader 1 takes
391 // over; because a prepare QC existed, it MUST re-propose the same block.
392 let mut drv = res!(Driver::new(5, 1));
393 let block = b"pinned block".to_vec();
394 let h = fixed_block_hash(44);
395 let cmds = res!(drv.replicas[0].propose(block.clone(), h));
396 drv.handle_commands(0, cmds);
397 // Drive until a PreCommit proposal has been broadcast (but not further).
398 // The PreCommit proposal distributes the Prepare QC, populating each
399 // replica's prepare_qc.
400 let mut saw_precommit = false;
401 while let Some(d) = drv.queue.pop_front() {
402 let is_precommit = matches!(&d,
403 Delivery::Proposal(p) if p.phase == Phase::PreCommit);
404 res!(drv.deliver(d));
405 if is_precommit {
406 saw_precommit = true;
407 break;
408 }
409 }
410 assert!(saw_precommit, "expected a PreCommit proposal in the queue");
411 // Drop the remainder of the queue (simulating the leader going silent
412 // mid-PreCommit before any Commit happened), and time out.
413 drv.queue.clear();
414 assert!(drv.replicas.iter().all(|r| r.prepare_qc().is_some()),
415 "every replica should hold a prepare_qc after PreCommit proposal");
416 assert!(drv.replicas.iter().all(|r| r.locked_qc().is_none()),
417 "no replica should be locked yet (no Commit proposal seen)");
418 res!(drv.timeout_all());
419 res!(drv.run());
420 // Leader 1 collected NewViews with prepare_qc pinning block `h`, so it
421 // must have broadcast a Prepare for `block` without a fresh propose().
422 for d in &drv.decided {
423 assert_eq!(d.as_deref(), Some(block.as_slice()),
424 "view-changed cohort must decide the pinned block");
425 }
426 Ok(())
427}
428
429#[test]
430fn view_change_rejects_unsafe_proposal() -> Outcome<()> {
431 // A replica that is LOCKED on block B (saw a Commit proposal for B in
432 // view 1) must reject a Prepare proposal in view 2 for a different
433 // block whose justify is from view 1.
434 let cfg = Config { cohort_size: 5, f: 1, self_id: 1 };
435 let mut victim = res!(Replica::new(cfg));
436
437 // Construct a synthetic PreCommit QC on block B, view 1, which is what
438 // a Commit proposal's justify would carry. Feeding this through the
439 // Commit code path installs locked_qc on the victim.
440 let block_b = b"locked block".to_vec();
441 let h_b = fixed_block_hash(100);
442 // First: feed a Prepare for B so that blocks[h_b] is populated and
443 // last_voted tracks Prepare.
444 let p1 = Proposal {
445 view: 1, phase: Phase::Prepare,
446 block_hash: h_b, block: Some(block_b.clone()), justify: None,
447 };
448 let _ = res!(victim.on_proposal(p1));
449 // Synthesise a Prepare QC for (view=1, phase=Prepare, hash=h_b) with
450 // quorum=4 votes.
451 let prepare_qc = Qc {
452 view: 1, phase: Phase::Prepare, block_hash: h_b,
453 signatures: (0..4).map(|i| (i as ReplicaId, vec![i as u8])).collect(),
454 };
455 let p2 = Proposal {
456 view: 1, phase: Phase::PreCommit,
457 block_hash: h_b, block: None, justify: Some(prepare_qc),
458 };
459 let _ = res!(victim.on_proposal(p2));
460 let precommit_qc = Qc {
461 view: 1, phase: Phase::PreCommit, block_hash: h_b,
462 signatures: (0..4).map(|i| (i as ReplicaId, vec![i as u8])).collect(),
463 };
464 let p3 = Proposal {
465 view: 1, phase: Phase::Commit,
466 block_hash: h_b, block: None, justify: Some(precommit_qc),
467 };
468 let _ = res!(victim.on_proposal(p3));
469 assert!(victim.locked_qc().is_some(), "victim should be locked on B");
470 assert_eq!(victim.locked_qc().unwrap().block_hash, h_b);
471
472 // Now simulate view change to view 2.
473 res!(victim.on_timeout());
474 assert_eq!(victim.view(), 2);
475
476 // Byzantine leader proposes a DIFFERENT block C in view 2, with a
477 // fabricated justify from view 1 (same view as the victim's lock).
478 // safeBlock says: justify.block_hash != locked.block_hash AND
479 // justify.view == locked.view, so REJECT.
480 let block_c = b"attacker block".to_vec();
481 let h_c = fixed_block_hash(77);
482 let evil_justify = Qc {
483 view: 1, phase: Phase::Prepare, block_hash: h_c,
484 signatures: (0..4).map(|i| (i as ReplicaId, vec![i as u8])).collect(),
485 };
486 let attack = Proposal {
487 view: 2, phase: Phase::Prepare,
488 block_hash: h_c, block: Some(block_c), justify: Some(evil_justify),
489 };
490 let outcome = victim.on_proposal(attack);
491 assert!(outcome.is_err(),
492 "locked victim must reject an unsafe view-change proposal");
493 Ok(())
494}
495
496#[test]
497fn view_change_accepts_later_view_proposal() -> Outcome<()> {
498 // A replica that was ONLY prepared (not locked) in view 1 may accept a
499 // different block in view 2 if its justify.view > locked_qc.view --
500 // but here locked_qc is None, so anything with a valid prepare QC
501 // passes safeBlock trivially.
502 let cfg = Config { cohort_size: 5, f: 1, self_id: 1 };
503 let mut r = res!(Replica::new(cfg));
504 let block_a = b"view-1 block".to_vec();
505 let h_a = fixed_block_hash(50);
506 let _ = res!(r.on_proposal(Proposal {
507 view: 1, phase: Phase::Prepare,
508 block_hash: h_a, block: Some(block_a), justify: None,
509 }));
510 let prepare_qc_a = Qc {
511 view: 1, phase: Phase::Prepare, block_hash: h_a,
512 signatures: (0..4).map(|i| (i as ReplicaId, vec![i as u8])).collect(),
513 };
514 let _ = res!(r.on_proposal(Proposal {
515 view: 1, phase: Phase::PreCommit,
516 block_hash: h_a, block: None, justify: Some(prepare_qc_a),
517 }));
518 assert!(r.prepare_qc().is_some());
519 assert!(r.locked_qc().is_none());
520
521 // Time out to view 2, then feed a Prepare for a different block whose
522 // justify is from view 1 -- legal because locked_qc is None.
523 res!(r.on_timeout());
524 let block_b = b"view-2 block".to_vec();
525 let h_b = fixed_block_hash(60);
526 let qc_b = Qc {
527 view: 1, phase: Phase::Prepare, block_hash: h_b,
528 signatures: (0..4).map(|i| (i as ReplicaId, vec![i as u8])).collect(),
529 };
530 let proposal = Proposal {
531 view: 2, phase: Phase::Prepare,
532 block_hash: h_b, block: Some(block_b), justify: Some(qc_b),
533 };
534 let out = res!(r.on_proposal(proposal));
535 assert!(out.iter().any(|c| matches!(c, Command::SendVote { .. })),
536 "replica without a lock should vote on a valid view-2 Prepare");
537 Ok(())
538}