feat(code/test): Add channel-based test app for integration tests (#747)

* feat(test): Add test app based on the example app

* Improve logging in tests

* Better error handling

* Formatting

* Fix test suite

* Implement `RestreamProposal` in example and test app

* Disable `proposal_only` consensus mode tests

* Start from latest height in store plus one

* Do not run MBT tests in integration test suite

* Fix missing logs

* Store synced values in database for later retrieval, increase history length to 100

* Formatting

* Run integration tests for test app in their own job

* Run test app tests with `nextest` for now

* Do not fail CI for test app integration test failure

* Go back to running tests with maelstrom

* Exclude test app from code coverage for now

* Ignore unsupported tests

* Add back full nodes tests

* Wait a bit until starting the first height

* Fix off-by-one error in starting height of test and example apps

* Fix typo

* Slow down test app to ease debugging

* Update state.rs

Co-authored-by: Anca Zamfir <[email protected]>
Signed-off-by: Romain Ruetschi <[email protected]>

* feat(code/app/starknet): Adapt Starknet app to latest P2P protos (#819)

* Update Starknet protos to their latest version

* Adapt Starknet app to new protos

* Use height and round as stream id in example app

* Use `ConsensusStreamId` proto message for stream id in Starknet app

* Fix formatting of stream ids

* Update protos for interop

* Add `ProposalCommitment` type

* Add `BlockInfo` type

* Update proposal building and assembly from parts to account for BlockInfo and ProposalCommitment

* Update protos

* Adapt to latest proto changes

* Post-merge fixes

* Disable integration tests with proposal mode, keep parts only

* Set parts-only in spwan.bash

* Add detail to panic when explicit proposal

* Fix fmt

---------

Co-authored-by: Anca Zamfir <[email protected]>

* chore(code): Some WAL entries fail to decode (#840)

* Add test that shows: failed to fill whole buffer

* Add WAL replay error when unable to decode all entries

* Don't add tag w/o value for skipped timers or empty buf to wal

* Move new test to wal

* Logging cleanup

* Fix missing info in logs

* Add error detail to WalReplayError event

* Add tag to wal only if vote encoding is succesful

* Do not instruct WAL to persist step timeouts

* Move WalReplayError detection to background task

* WAL writes all timeouts it is given, leave choice up to consensus

---------

Co-authored-by: Romain Ruetschi <[email protected]>

* Post-merge fixes

* Refactor parts signature verification

* Add parts signature verification to test app

* Restream proposal parts when reproposing a value

* Formatting

* Remove decided proposal from undecided proposals table

* Remove duplicated code

* Include test app tests in code coverage

* Abort whole test on failure

* Fix full node test

* Show log when request sync value

---------

Signed-off-by: Romain Ruetschi <[email protected]>
Co-authored-by: Anca Zamfir <[email protected]>
Co-authored-by: Anca Zamfir <[email protected]>

.github/workflows/rust.yml

@@ -66,6 +66,7 @@ jobs:
             --all-features \
             --no-fail-fast \
             --failure-output final \
+            --exclude informalsystems-malachitebft-test \
             --exclude informalsystems-malachitebft-starknet-test \
             --exclude informalsystems-malachitebft-starknet-test-mbt \
             --exclude informalsystems-malachitebft-discovery-test
@@ -106,9 +107,45 @@ jobs:
         run: |
           cargo maelstrom \
             --slots 16 \
-            --include 'package.match(informalsystems-malachitebft-starknet-test) || package.match(informalsystems-malachitebft-discovery-test)' \
+            --include 'package.equals(informalsystems-malachitebft-starknet-test) || package.equals(informalsystems-malachitebft-discovery-test)' \
             --exclude 'package.match(informalsystems-malachitebft-starknet-test-mbt)'
 
+  integration_test_app:
+    name: Integration Tests (test app)
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: code
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Install Protoc
+        uses: arduino/setup-protoc@v3
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+      - name: Setup Node
+        uses: actions/setup-node@v3
+        with:
+          node-version: "18"
+      - name: Install Quint
+        run: npm install -g @informalsystems/quint
+      - name: Setup Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+          cache-workspaces: "code"
+      - name: Install cargo-maelstrom
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-maelstrom
+      - name: Disable apparmor container restrictions
+        run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0
+      - name: Run integration tests
+        continue-on-error: true
+        run: |
+          cargo maelstrom \
+            --slots 8 \
+            --include 'package.equals(informalsystems-malachitebft-test)'
+
   no_std:
     name: no_std compatibility
     needs: changes

code/Cargo.toml

@@ -24,9 +24,11 @@ members = [
 
   # Test
   "crates/test",
+  "crates/test/app",
   "crates/test/cli",
   "crates/test/mbt",
   "crates/test/mempool",
+  "crates/test/framework",
   "crates/network/test",
 
   # Starknet
@@ -64,7 +66,6 @@ unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage_nightly)'] }
 malachitebft-engine             = { version = "0.0.1", package = "informalsystems-malachitebft-engine", path = "crates/engine" }
 malachitebft-app                = { version = "0.0.1", package = "informalsystems-malachitebft-app", path = "crates/app" }
 malachitebft-app-channel        = { version = "0.0.1", package = "informalsystems-malachitebft-app-channel", path = "crates/app-channel" }
-malachitebft-test-cli           = { version = "0.0.1", package = "informalsystems-malachitebft-test-cli", path = "crates/test/cli" }
 malachitebft-codec              = { version = "0.0.1", package = "informalsystems-malachitebft-codec", path = "crates/codec" }
 malachitebft-config             = { version = "0.0.1", package = "informalsystems-malachitebft-config", path = "crates/config" }
 malachitebft-core-consensus     = { version = "0.0.1", package = "informalsystems-malachitebft-core-consensus", path = "crates/core-consensus" }
@@ -83,8 +84,11 @@ malachitebft-wal                = { version = "0.0.1", package = "informalsystem
 
 # Test
 malachitebft-test                   = { version = "0.0.1", package = "informalsystems-malachitebft-test", path = "crates/test" }
+malachitebft-test-app               = { version = "0.0.1", package = "informalsystems-malachitebft-test-app", path = "crates/test/app" }
+malachitebft-test-cli               = { version = "0.0.1", package = "informalsystems-malachitebft-test-cli", path = "crates/test/cli" }
 malachitebft-test-mbt               = { version = "0.0.1", package = "informalsystems-malachitebft-test-mbt", path = "crates/test/mbt" }
 malachitebft-test-mempool           = { version = "0.0.1", package = "informalsystems-malachitebft-test-mempool", path = "crates/test/mempool" }
+malachitebft-test-framework         = { version = "0.0.1", package = "informalsystems-malachitebft-test-framework", path = "crates/test/framework" }
 malachitebft-discovery-test         = { version = "0.0.1", package = "informalsystems-malachitebft-discovery-test", path = "crates/network/test" }
 
 # Starknet

code/crates/app-channel/src/lib.rs

@@ -19,4 +19,4 @@ mod msgs;
 pub use msgs::{AppMsg, Channels, ConsensusMsg, NetworkMsg, Reply};
 
 mod run;
-pub use run::run;
+pub use run::{start_engine, EngineHandle};

code/crates/app-channel/src/msgs.rs

@@ -2,12 +2,13 @@ use std::time::Duration;
 
 use bytes::Bytes;
 use derive_where::derive_where;
-use malachitebft_app::consensus::VoteExtensionError;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 
+use malachitebft_app::consensus::VoteExtensionError;
 use malachitebft_engine::consensus::Msg as ConsensusActorMsg;
 use malachitebft_engine::network::Msg as NetworkActorMsg;
+use malachitebft_engine::util::events::TxEvent;
 
 use crate::app::types::core::{CommitCertificate, Context, Round, ValueId, VoteExtensions};
 use crate::app::types::streaming::StreamMessage;
@@ -22,6 +23,8 @@ pub struct Channels<Ctx: Context> {
     pub consensus: mpsc::Receiver<AppMsg<Ctx>>,
     /// Channel for sending messages to the networking layer
     pub network: mpsc::Sender<NetworkMsg<Ctx>>,
+    /// Receiver of events, call `subscribe` to receive them
+    pub events: TxEvent<Ctx>,
 }
 
 /// Messages sent from consensus to the application.

code/crates/app-channel/src/run.rs

@@ -1,30 +1,37 @@
 //! Run Malachite consensus with the given configuration and context.
 //! Provides the application with a channel for receiving messages from consensus.
 
-use std::path::PathBuf;
-
 use eyre::Result;
 
+use malachitebft_app::spawn::{
+    spawn_consensus_actor, spawn_node_actor, spawn_sync_actor, spawn_wal_actor,
+};
+use malachitebft_engine::node::NodeRef;
+use malachitebft_engine::util::events::TxEvent;
+use tokio::task::JoinHandle;
+
+use crate::app;
 use crate::app::types::codec::{ConsensusCodec, SyncCodec, WalCodec};
 use crate::app::types::config::Config as NodeConfig;
 use crate::app::types::core::Context;
 use crate::app::types::metrics::{Metrics, SharedRegistry};
 use crate::spawn::{spawn_host_actor, spawn_network_actor};
-use crate::{app, Channels};
+use crate::Channels;
 
-use malachitebft_app::{spawn_consensus_actor, spawn_sync_actor, spawn_wal_actor};
-use malachitebft_engine::util::events::TxEvent;
+pub struct EngineHandle {
+    pub actor: NodeRef,
+    pub handle: JoinHandle<()>,
+}
 
 #[tracing::instrument("node", skip_all, fields(moniker = %cfg.moniker))]
-pub async fn run<Node, Ctx, Codec>(
+pub async fn start_engine<Node, Ctx, Codec>(
     ctx: Ctx,
     codec: Codec,
     node: Node,
     cfg: NodeConfig,
-    private_key_file: PathBuf,
     start_height: Option<Ctx::Height>,
     initial_validator_set: Ctx::ValidatorSet,
-) -> Result<Channels<Ctx>>
+) -> Result<(Channels<Ctx>, EngineHandle)>
 where
     Ctx: Context,
     Node: app::Node<Context = Ctx>,
@@ -37,21 +44,21 @@ where
     let registry = SharedRegistry::global().with_moniker(cfg.moniker.as_str());
     let metrics = Metrics::register(&registry);
 
-    let private_key_file = node.load_private_key_file(private_key_file)?;
+    let private_key_file = node.load_private_key_file()?;
     let private_key = node.load_private_key(private_key_file);
     let public_key = node.get_public_key(&private_key);
     let address = node.get_address(&public_key);
     let keypair = node.get_keypair(private_key.clone());
     let signing_provider = node.get_signing_provider(private_key);
 
     // Spawn consensus gossip
-    let (network, network_tx) =
+    let (network, tx_network) =
         spawn_network_actor(&cfg, keypair, &registry, codec.clone()).await?;
 
     let wal = spawn_wal_actor(&ctx, codec, &node.get_home_dir(), &registry).await?;
 
     // Spawn the host actor
-    let (connector, consensus_rx) = spawn_host_actor(metrics.clone()).await?;
+    let (connector, rx_consensus) = spawn_host_actor(metrics.clone()).await?;
 
     let sync = spawn_sync_actor(
         ctx.clone(),
@@ -62,25 +69,37 @@ where
     )
     .await?;
 
+    let tx_event = TxEvent::new();
+
     // Spawn consensus
-    let _consensus = spawn_consensus_actor(
+    let consensus = spawn_consensus_actor(
         start_height,
         initial_validator_set,
         address,
-        ctx,
+        ctx.clone(),
         cfg,
         Box::new(signing_provider),
-        network,
-        connector,
-        wal,
+        network.clone(),
+        connector.clone(),
+        wal.clone(),
         sync.clone(),
         metrics,
-        TxEvent::new(),
+        tx_event.clone(),
     )
     .await?;
 
-    Ok(Channels {
-        consensus: consensus_rx,
-        network: network_tx,
-    })
+    let (node, handle) = spawn_node_actor(ctx, network, consensus, wal, sync, connector).await?;
+
+    let channels = Channels {
+        consensus: rx_consensus,
+        network: tx_network,
+        events: tx_event,
+    };
+
+    let handle = EngineHandle {
+        actor: node,
+        handle,
+    };
+
+    Ok((channels, handle))
 }

code/crates/app-channel/src/spawn.rs

@@ -3,6 +3,7 @@
 use eyre::Result;
 use tokio::sync::mpsc;
 
+use malachitebft_app as app;
 use malachitebft_app::types::metrics::SharedRegistry;
 use malachitebft_app::types::Keypair;
 use malachitebft_config::Config as NodeConfig;
@@ -40,7 +41,7 @@ where
 {
     let (tx, mut rx) = mpsc::channel::<NetworkMsg<Ctx>>(1);
 
-    let actor_ref = malachitebft_app::spawn_network_actor(cfg, keypair, registry, codec).await?;
+    let actor_ref = app::spawn::spawn_network_actor(cfg, keypair, registry, codec).await?;
 
     tokio::spawn({
         let actor_ref = actor_ref.clone();

code/crates/app/Cargo.toml

@@ -13,22 +13,24 @@ readme = "../../../README.md"
 all-features = true
 
 [dependencies]
-malachitebft-engine.workspace = true
 malachitebft-codec.workspace = true
-malachitebft-core-types.workspace = true
 malachitebft-config.workspace = true
 malachitebft-core-consensus.workspace = true
-malachitebft-network.workspace = true
+malachitebft-core-types.workspace = true
+malachitebft-engine.workspace = true
 malachitebft-metrics.workspace = true
+malachitebft-network.workspace = true
 malachitebft-peer.workspace = true
 malachitebft-sync.workspace = true
 
 async-trait = { workspace = true }
 derive-where = { workspace = true }
 eyre = { workspace = true }
 libp2p-identity = { workspace = true }
+ractor = { workspace = true }
 rand = { workspace = true }
 serde = { workspace = true }
+tokio = { workspace = true }
 tracing = { workspace = true }
 
 [lints]

code/crates/app/src/lib.rs

@@ -12,10 +12,12 @@ mod node;
 pub use node::Node;
 
 pub mod part_store;
+pub mod spawn;
 pub mod types;
 
-mod spawn;
-pub use spawn::{spawn_consensus_actor, spawn_network_actor, spawn_sync_actor, spawn_wal_actor};
+pub mod events {
+    pub use malachitebft_engine::util::events::TxEvent;
+}
 
 pub mod streaming {
     pub use malachitebft_engine::util::streaming::*;

code/crates/app/src/node.rs

@@ -31,7 +31,7 @@ pub trait Node {
 
     fn load_private_key(&self, file: Self::PrivateKeyFile) -> PrivateKey<Self::Context>;
 
-    fn load_private_key_file(&self, path: impl AsRef<Path>) -> io::Result<Self::PrivateKeyFile>;
+    fn load_private_key_file(&self) -> io::Result<Self::PrivateKeyFile>;
 
     fn make_private_key_file(&self, private_key: PrivateKey<Self::Context>)
         -> Self::PrivateKeyFile;