feat: add qbft module (#18)

* Add Nix support

* Add `typos`

* Apply copilot suggestions

* Update Cargo.lock

* Add Dockerfile and .dockerignore

* Add placeholder executable

* Include cargo tools in Nix

* Remove the need for nightly

* Fix checks on placeholder app

* Adjust target name

* Build only the CLI

* Rely on system libraries

- Cannot set up nightly without breaking stuff

* Remove duplicated inputs

* Use `rust:1.89` and `debian:trixie-slim` as images

- Bigger image (80 mb vs 14 mb)

* Use bookworm instead of trixie

* Formatting

* Revert changes to `rustfmt.toml`

* Revert changes to `pre-push`

* Expose port 3030

* Initial qbft impl

- Missing public `Run` method

* Accept references to `I`

* Define errors as structs

* Initial qbft implementation

* Rename SomeMsg/Msg

* Reduce cloning when ref is enough

* Complete `Run` method

* Use crossbeam for channels

* Fix `q_commit`

- Wrap in Option
- Make classify return None instead of empty vec

* Initialize remaining state elements

* Renaming

* Use `select!` macro

* Apply clippy suggestions

* Match reference project structure

* C

* Add qbft module docs

* Run `d.log_*` methods

* Match visibility modifiers

* Match module docs

* Regen `Cargo.lock`

* Move dependency to workspace

* Fix pre-push checks

- Code sorting

* Revert docs change

* Fix typo

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Fix equality comparison with `Default`

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Add internal tests

* Add `as_any` to `SomeMsg` trait to allow downcasting

* Make structs thread safe

* Complete `test_qbft` function

* Add `happy_0` test

* Include `rand`

* Disable test due to deadlock

* Sort deps

* Run formatter

* Apply clippy suggestions

* Fix wrong check

- Ensure `ppj_cache` is None only

* Use single `is_leader` function

* Simplify imports

* Log when all resuts are received.

* Propagate cancellation

Use `cancellation` crate

* Fix deadlock

- Main test thread was not exiting due to wrong scope

* Match original log messages

* Add working tests

* Port remaining test cases

* Add `duplicate_pre_prepare_rules` test

* Formatting

* Mutate `input_value`

Fixes multiple test cases

* Avoid f64 for time

* Add `fake_clock`

* Sort deps

* Sort module declaration

* Apply Copilot fixes/suggestions

* Use `fake_clock`

Speeds up tests but does not fix concurrency issues

* Add `cargo nextest`

* Update CI to use `nextest`

* Test for clock cancellation

* Update comments

* Use workspace `rand`

* Improve retries for flaky tests

* Apply copilot comment suggestions

* Use explicit error enum

* Remove `anyhow`

* Small suggestions

* Add `Drop` impl for `FakeClock`

* Fire and forget in `bcast`

* Ignore all tests

- Flaky tests

* Remove nextest

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: emlautarom1

Cargo.lock

@@ -24,20 +24,22 @@ publish = false
 
 [workspace.dependencies]
 axum = "0.8.6"
+blst = "0.3.13"
+cancellation = "0.1.0"
 chrono = { version = "0.4", features = ["serde"] }
+crossbeam = "0.8.4"
 hex = { version = "^0.4.3" }
-serde = { version = "1.0", features = ["derive"] }
-serde_json = { version = "^1.0" }
-tokio = { version = "1", features = ["full"] }
-libp2p = { version = "0.56", features = ["full", "secp256k1"] }
 prost = "0.14"
-prost-types = "0.14"
 prost-build = "0.14"
-blst = "0.3.13"
+prost-types = "0.14"
+rand = { version = "0.8", features = ["std_rng"] }
 rand_core = "0.6"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = { version = "^1.0" }
 thiserror = "2.0.12"
-rand = {version = "0.8", features = ["std_rng"]}
-uuid = {version = "1.16", features = ["serde", "v4"] }
+tokio = { version = "1", features = ["full"] }
+libp2p = { version = "0.56", features = ["full", "secp256k1"] }
+uuid = { version = "1.16", features = ["serde", "v4"] }
 serde_with = { version = "3", features = ["hex", "base64"] }
 base64 = { version = "0.22.1" }
 sha3 = { version = "0.10.8" }

Cargo.toml

@@ -7,8 +7,19 @@ license.workspace = true
 publish.workspace = true
 
 [dependencies]
+cancellation.workspace = true
+chrono.workspace = true
+crossbeam.workspace = true
+hex.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+thiserror.workspace = true
+libp2p.workspace = true
+prost.workspace = true
+prost-types.workspace = true
+
+[dev-dependencies]
+rand.workspace = true
 libp2p.workspace = true
 prost.workspace = true
 prost-types.workspace = true
@@ -19,4 +30,4 @@ chrono.workspace = true
 charon-build-proto.workspace = true
 
 [lints]
-workspace = true
+workspace = true

crates/charon-core/Cargo.toml

@@ -4,6 +4,7 @@
 //! This crate provides the fundamental building blocks, data structures, and
 //! core algorithms used throughout the Charon system.
 
+pub mod qbft;
 /// Types for the Charon core.
 pub mod types;
 

crates/charon-core/src/lib.rs

@@ -0,0 +1,153 @@
+use crossbeam::channel as mpmc;
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+    thread,
+    time::{Duration, Instant},
+};
+
+#[derive(Clone)]
+pub struct FakeClock {
+    inner: Arc<Mutex<FakeClockInner>>,
+}
+
+struct FakeClockInner {
+    start: Instant,
+    now: Instant,
+    last_id: usize,
+    clients: HashMap<usize, (mpmc::Sender<Instant>, Instant)>,
+}
+
+impl FakeClock {
+    pub fn new(now: Instant) -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(FakeClockInner {
+                start: now,
+                now,
+                last_id: 1,
+                clients: Default::default(),
+            })),
+        }
+    }
+
+    pub fn new_timer(
+        &self,
+        duration: Duration,
+    ) -> (
+        mpmc::Receiver<Instant>,
+        Box<dyn Fn() + Send + Sync + 'static>,
+    ) {
+        let (tx, rx) = mpmc::bounded::<Instant>(1);
+
+        let client_id = {
+            let mut inner = self.inner.lock().unwrap();
+            let id = inner.last_id;
+            let deadline = inner.now + duration;
+
+            inner.last_id += 1;
+            inner.clients.insert(id, (tx, deadline));
+
+            id
+        };
+
+        let inner = Arc::clone(&self.inner);
+        let cancel = Box::new(move || {
+            let mut inner = inner.lock().unwrap();
+            inner.clients.remove(&client_id);
+        });
+
+        (rx, cancel)
+    }
+
+    pub fn advance(&self, duration: Duration) {
+        // Advance time and collect expired senders under lock, but perform sends
+        // without holding lock.
+        let mut expired = vec![];
+
+        let now = {
+            let mut inner = self.inner.lock().unwrap();
+            inner.now += duration;
+            let now = inner.now;
+
+            for (&id, (ch, deadline)) in inner.clients.iter() {
+                if *deadline <= now {
+                    expired.push((id, ch.clone()));
+                }
+            }
+
+            for (id, _) in expired.iter() {
+                inner.clients.remove(id);
+            }
+
+            now
+        };
+
+        for (_, ch) in expired {
+            let _ = ch.send(now);
+        }
+    }
+
+    pub fn elapsed(&self) -> Duration {
+        let inner = self.inner.lock().unwrap();
+        inner.now - inner.start
+    }
+
+    pub fn cancel(&self) {
+        let mut inner = self.inner.lock().unwrap();
+        inner.clients.clear();
+    }
+}
+
+impl Drop for FakeClock {
+    fn drop(&mut self) {
+        self.cancel();
+    }
+}
+
+#[test]
+fn multiple_threads_timers() {
+    let clock = FakeClock::new(Instant::now());
+
+    let start = Instant::now();
+    thread::scope(|s| {
+        let c1 = clock.clone();
+        let (ch_1, _) = c1.new_timer(Duration::from_secs(5));
+        s.spawn(move || {
+            let _ = ch_1.recv();
+        });
+
+        let c2 = clock.clone();
+        let (ch_2, _) = c2.new_timer(Duration::from_secs(5));
+        s.spawn(move || {
+            let _ = ch_2.recv();
+        });
+
+        clock.advance(Duration::from_secs(6));
+    });
+
+    println!("start={:?}, clock={:?}", start.elapsed(), clock.elapsed());
+}
+
+#[test]
+fn multiple_threads_cancellation() {
+    let clock = FakeClock::new(Instant::now());
+
+    let start = Instant::now();
+    thread::scope(|s| {
+        let c1 = clock.clone();
+        let (ch_1, _) = c1.new_timer(Duration::from_secs(5));
+        s.spawn(move || {
+            let _ = ch_1.recv();
+        });
+
+        let c2 = clock.clone();
+        let (ch_2, _) = c2.new_timer(Duration::from_secs(5));
+        s.spawn(move || {
+            let _ = ch_2.recv();
+        });
+
+        clock.cancel();
+    });
+
+    println!("start={:?}, clock={:?}", start.elapsed(), clock.elapsed());
+}