feat(code): Add actor integrated application that uses proposal only

code/Cargo.toml

@@ -37,6 +37,8 @@ members = [
 
   # Examples
   "examples/channel",
+  "examples/actor/proposal",
+  "examples/actor/test",
 ]
 
 [workspace.package]

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

@@ -200,6 +200,28 @@ where
                 }
             }
 
+            HostMsg::ReceivedProposal {
+                proposer,
+                height,
+                round,
+                value,
+                reply_to,
+            } => {
+                let (reply, rx) = oneshot::channel();
+
+                self.sender
+                    .send(AppMsg::ReceivedProposal {
+                        proposer,
+                        height,
+                        round,
+                        value,
+                        reply,
+                    })
+                    .await?;
+
+                reply_to.send(rx.await?)?;
+            }
+
             HostMsg::GetValidatorSet { height, reply_to } => {
                 let (reply, rx) = oneshot::channel();
 

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

@@ -141,6 +141,22 @@ pub enum AppMsg<Ctx: Context> {
         reply: Reply<Option<ProposedValue<Ctx>>>,
     },
 
+    /// Notifies the application that consensus has received a proposal over the network.
+    ///
+    /// The application MUST respond with the complete proposed value.
+    ReceivedProposal {
+        /// Peer whom the proposal was received from
+        proposer: Ctx::Address,
+        /// Height of the proposal
+        height: Ctx::Height,
+        /// Round of the proposal
+        round: Round,
+        /// Value of the proposal
+        value: Ctx::Value,
+        /// Channel for returning validity of the proposal
+        reply: Reply<ProposedValue<Ctx>>,
+    },
+
     /// Requests the validator set for a specific height
     GetValidatorSet {
         /// Height of the validator set to retrieve

code/crates/core-consensus/src/handle/proposal.rs

@@ -3,7 +3,7 @@ use crate::handle::signature::verify_signature;
 use crate::handle::validator_set::get_validator_set;
 use crate::input::Input;
 use crate::prelude::*;
-use crate::types::{ConsensusMsg, ProposedValue, SignedConsensusMsg, WalEntry};
+use crate::types::{ConsensusMsg, SignedConsensusMsg, WalEntry};
 use crate::util::pretty::PrettyProposal;
 
 /// Handles an incoming consensus proposal message.
@@ -100,21 +100,6 @@ where
         );
     }
 
-    if state.params.value_payload.proposal_only() {
-        // TODO - pass the received value up to the host that will verify and give back validity and extension.
-        // Currently starknet Context defines value as BlockHash, we need a PoC app for this.
-        let new_value = ProposedValue {
-            height: signed_proposal.height(),
-            round: signed_proposal.round(),
-            valid_round: signed_proposal.pol_round(),
-            proposer: signed_proposal.validator_address().clone(),
-            value: signed_proposal.value().clone(),
-            validity: Validity::Valid,
-        };
-
-        state.store_value(&new_value);
-    }
-
     if let Some(full_proposal) = state.full_proposal_at_round_and_value(
         &proposal_height,
         proposal_round,

code/crates/engine/src/consensus.rs

@@ -569,12 +569,35 @@ where
                             return Ok(());
                         }
 
+                        let proposal_clone = proposal.clone();
+
                         if let Err(e) = self
                             .process_input(&myself, state, ConsensusInput::Proposal(proposal))
                             .await
                         {
                             error!(%from, "Error when processing proposal: {e}");
                         }
+
+                        if state.consensus.params.value_payload.proposal_only() {
+                            self.host
+                                .call_and_forward(
+                                    |reply_to| HostMsg::ReceivedProposal {
+                                        proposer: proposal_clone.validator_address().clone(),
+                                        height: proposal_clone.height(),
+                                        round: proposal_clone.round(),
+                                        value: proposal_clone.value().clone(),
+                                        reply_to,
+                                    },
+                                    &myself,
+                                    |value| {
+                                        Msg::ReceivedProposedValue(value, ValueOrigin::Consensus)
+                                    },
+                                    None,
+                                )
+                                .map_err(|e| {
+                                    eyre!("Error when forwarding proposal to host: {e}")
+                                })?;
+                        }
                     }
 
                     NetworkEvent::PolkaCertificate(from, certificate) => {

code/crates/engine/src/host.rs

@@ -136,7 +136,16 @@ pub enum HostMsg<Ctx: Context> {
         reply_to: RpcReplyPort<ProposedValue<Ctx>>,
     },
 
-    /// Requests the validator set for a specific height
+    /// Received a proposal from a validator
+    ReceivedProposal {
+        proposer: Ctx::Address,
+        height: Ctx::Height,
+        round: Round,
+        value: Ctx::Value,
+        reply_to: RpcReplyPort<ProposedValue<Ctx>>,
+    },
+
+    /// Get the validator set at a given height
     GetValidatorSet {
         height: Ctx::Height,
         reply_to: RpcReplyPort<Option<Ctx::ValidatorSet>>,

code/crates/starknet/host/src/actor.rs

@@ -208,6 +208,10 @@ impl Host {
                 value_bytes,
                 reply_to,
             } => on_process_synced_value(value_bytes, height, round, proposer, reply_to),
+
+            HostMsg::ReceivedProposal { .. } => {
+                panic!("ProposalOnly not supported");
+            }
         }
     }
 }

code/crates/starknet/test/src/lib.rs

@@ -49,7 +49,7 @@ pub struct TestRunner {
 fn temp_dir(id: NodeId) -> PathBuf {
     TempDir::with_prefix(format!("malachitebft-test-app-{id}-"))
         .unwrap()
-        .into_path()
+        .keep()
 }
 
 #[async_trait]

code/crates/test/app/src/app.rs

@@ -363,6 +363,10 @@ pub async fn run(
                     error!("Failed to send VerifyVoteExtension reply");
                 }
             }
+
+            AppMsg::ReceivedProposal { .. } => {
+                panic!("ReceivedProposal should not be called in test app");
+            }
         }
     }
 

code/crates/test/framework/src/node.rs

@@ -160,6 +160,26 @@ where
         })
     }
 
+    pub fn expect_wal_replay_at_crash_height<F>(&mut self, get_height: F) -> &mut Self
+    where
+        F: Fn(&State) -> Option<u64> + Send + Sync + 'static,
+    {
+        self.on_event(move |event, state| {
+            let Event::WalReplayBegin(height, count) = event else {
+                return Ok(HandlerResult::WaitForNextEvent);
+            };
+
+            let expected_height = get_height(state).expect("Should have recorded crash height");
+            info!("Replaying WAL at height {height} with {count} messages");
+
+            if height.as_u64() != expected_height {
+                bail!("Unexpected WAL replay at height {height}, expected {expected_height}")
+            }
+
+            Ok(HandlerResult::ContinueTest)
+        })
+    }
+
     pub fn expect_vote_rebroadcast(
         &mut self,
         at_height: u64,

code/crates/test/tests/it/main.rs

@@ -56,7 +56,7 @@ pub struct TestRunner {
 fn temp_dir(id: NodeId) -> PathBuf {
     TempDir::with_prefix(format!("malachitebft-test-app-{id}"))
         .unwrap()
-        .into_path()
+        .keep()
 }
 
 #[derive(Clone)]

code/examples/actor/proposal/Cargo.toml

@@ -0,0 +1,61 @@
+[package]
+name = "informalsystems_malachitebft_actor_app_proposal"
+version = "0.3.0-pre"
+edition = "2021"
+publish = false
+
+[[bin]]
+name = "actor-app-proposal"
+path = "app/src/main.rs"
+
+[dependencies]
+malachitebft-test-cli = { workspace = true }
+malachitebft-engine = { workspace = true }
+malachitebft-app = { workspace = true }
+malachitebft-codec = { workspace = true }
+malachitebft-config = { workspace = true }
+malachitebft-core-consensus = { workspace = true, features = ["debug"] }
+malachitebft-core-types = { workspace = true }
+malachitebft-network = { workspace = true }
+malachitebft-metrics = { workspace = true }
+malachitebft-proto = { workspace = true }
+malachitebft-signing-ed25519 = { workspace = true, features = ["serde", "rand"] }
+malachitebft-sync = { workspace = true }
+malachitebft-test-mempool = { workspace = true }
+malachitebft-wal = { workspace = true }
+
+async-trait = { workspace = true }
+bytes = { workspace = true, features = ["serde"] }
+bytesize = { workspace = true }
+color-eyre = { workspace = true }
+config = { workspace = true }
+derive-where = { workspace = true }
+eyre = { workspace = true }
+itertools = { workspace = true }
+libp2p-identity = { workspace = true }
+prost = { workspace = true }
+ractor = { workspace = true }
+rand = { workspace = true }
+redb = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+sha3 = { workspace = true }
+toml = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+hex = { version = "0.4", features = ["serde"] }
+
+[lints]
+workspace = true
+
+[dev-dependencies]
+tempfile = { workspace = true }
+clap = { workspace = true }
+
+[target.'cfg(target_os = "linux")'.dependencies]
+tikv-jemallocator = { workspace = true, features = ["background_threads"] }
+
+[build-dependencies]
+prost-build = { workspace = true }
+protox = { workspace = true }

code/examples/actor/proposal/README.md

@@ -0,0 +1,55 @@
+
+# Example actor-based app
+
+This is an example application using the actor based integration with malachite
+
+## Run a local testnet
+
+### Prerequisites
+
+Before running the examples, make sure you have the required development environment setup as specified in this [Setup](../../../CONTRIBUTING_CODE.md#setup) section.
+
+### Compile the app
+
+```
+cargo build
+```
+
+### Setup the testnet
+
+Generate configuration and genesis for three nodes using the `testnet` command:
+
+```
+cargo run --bin actor-app-proposal -- testnet --nodes 3 --home nodes
+```
+
+This will create the configuration for three nodes in the `nodes` folder. Feel free to inspect this folder and look at the generated files.
+
+### Spawn the nodes
+
+```
+bash ./examples/actor/proposal/spawn.bash --nodes 3 --home nodes
+```
+
+If successful, the logs for each node can then be found at `nodes/X/logs/node.log`.
+
+```
+tail -f nodes/0/logs/node.log
+```
+
+Check the metrics
+
+For the block time:
+
+```
+curl -s localhost:29000/metrics | grep 'time_per_block_[sum|count]'
+```
+
+For the number of rounds per block:
+
+```
+curl -s localhost:29000/metrics | grep consensus_round
+```
+
+Press `Ctrl-C` to stop all the nodes.
+