FM-184: Proposal interpreter

fendermint/abci/src/application.rs

@@ -11,6 +11,8 @@ use tendermint::abci::{request, response, Request, Response};
 use tower::Service;
 use tower_abci::BoxError;
 
+use crate::util::take_until_max_size;
+
 /// Allow returning a result from the methods, so the [`Application`]
 /// implementation doesn't have to be full of `.expect("failed...")`
 /// or `.unwrap()` calls. It is still good practice to use for example
@@ -62,20 +64,12 @@ pub trait Application {
         &self,
         request: request::PrepareProposal,
     ) -> AbciResult<response::PrepareProposal> {
-        let max_tx_bytes: usize = request.max_tx_bytes.try_into().unwrap();
-        let mut size: usize = 0;
-        let mut txs = Vec::new();
-        for tx in request.txs {
-            if size.saturating_add(tx.len()) > max_tx_bytes {
-                break;
-            }
-            size += tx.len();
-            txs.push(tx);
-        }
+        let txs = take_until_max_size(request.txs, request.max_tx_bytes.try_into().unwrap());
+
         Ok(response::PrepareProposal { txs })
     }
 
-    /// Opporunity for the application to inspect the proposal before voting on it.
+    /// Opportunity for the application to inspect the proposal before voting on it.
     ///
     /// The application should accept the proposal unless there's something wrong with it.
     ///

fendermint/abci/src/lib.rs

@@ -3,3 +3,4 @@
 mod application;
 
 pub use application::{AbciResult, Application, ApplicationService};
+pub mod util;

fendermint/abci/src/util.rs

@@ -0,0 +1,19 @@
+// Copyright 2022-2023 Protocol Labs
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+/// Take the first transactions until the first one that would exceed the maximum limit.
+///
+/// The function does not skip or reorder transaction even if a later one would stay within the limit.
+pub fn take_until_max_size<T: AsRef<[u8]>>(txs: Vec<T>, max_tx_bytes: usize) -> Vec<T> {
+    let mut size: usize = 0;
+    let mut out = Vec::new();
+    for tx in txs {
+        let bz: &[u8] = tx.as_ref();
+        if size.saturating_add(bz.len()) > max_tx_bytes {
+            break;
+        }
+        size += bz.len();
+        out.push(tx);
+    }
+    out
+}

fendermint/app/src/app.rs

@@ -7,6 +7,7 @@ use std::sync::{Arc, Mutex};
 use anyhow::{anyhow, Context, Result};
 use async_trait::async_trait;
 use cid::Cid;
+use fendermint_abci::util::take_until_max_size;
 use fendermint_abci::{AbciResult, Application};
 use fendermint_storage::{
     Codec, Encode, KVCollection, KVRead, KVReadable, KVStore, KVWritable, KVWrite,
@@ -22,7 +23,7 @@ use fendermint_vm_interpreter::fvm::state::{
 use fendermint_vm_interpreter::fvm::{FvmApplyRet, FvmGenesisOutput};
 use fendermint_vm_interpreter::signed::InvalidSignature;
 use fendermint_vm_interpreter::{
-    CheckInterpreter, ExecInterpreter, GenesisInterpreter, QueryInterpreter,
+    CheckInterpreter, ExecInterpreter, GenesisInterpreter, ProposalInterpreter, QueryInterpreter,
 };
 use fvm::engine::MultiEngine;
 use fvm_ipld_blockstore::Blockstore;
@@ -309,6 +310,15 @@ where
     S::Namespace: Sync + Send,
     DB: KVWritable<S> + KVReadable<S> + Clone + Send + Sync + 'static,
     SS: Blockstore + Clone + Send + Sync + 'static,
+    I: GenesisInterpreter<
+        State = FvmGenesisState<SS>,
+        Genesis = Vec<u8>,
+        Output = FvmGenesisOutput,
+    >,
+    I: ProposalInterpreter<
+        State = (), // TODO
+        Message = Vec<u8>,
+    >,
     I: ExecInterpreter<
         State = FvmExecState<SS>,
         Message = Vec<u8>,
@@ -326,11 +336,6 @@ where
         Query = BytesMessageQuery,
         Output = BytesMessageQueryRet,
     >,
-    I: GenesisInterpreter<
-        State = FvmGenesisState<SS>,
-        Genesis = Vec<u8>,
-        Output = FvmGenesisOutput,
-    >,
 {
     /// Provide information about the ABCI application.
     async fn info(&self, _request: request::Info) -> AbciResult<response::Info> {
@@ -490,6 +495,45 @@ where
         Ok(response)
     }
 
+    /// Amend which transactions to put into the next block proposal.
+    async fn prepare_proposal(
+        &self,
+        request: request::PrepareProposal,
+    ) -> AbciResult<response::PrepareProposal> {
+        let txs = request.txs.into_iter().map(|tx| tx.to_vec()).collect();
+
+        let txs = self
+            .interpreter
+            .prepare((), txs)
+            .await
+            .context("failed to prepare proposal")?;
+
+        let txs = txs.into_iter().map(bytes::Bytes::from).collect();
+        let txs = take_until_max_size(txs, request.max_tx_bytes.try_into().unwrap());
+
+        Ok(response::PrepareProposal { txs })
+    }
+
+    /// Inspect a proposal and decide whether to vote on it.
+    async fn process_proposal(
+        &self,
+        request: request::ProcessProposal,
+    ) -> AbciResult<response::ProcessProposal> {
+        let txs = request.txs.into_iter().map(|tx| tx.to_vec()).collect();
+
+        let accept = self
+            .interpreter
+            .process((), txs)
+            .await
+            .context("failed to process proposal")?;
+
+        if accept {
+            Ok(response::ProcessProposal::Accept)
+        } else {
+            Ok(response::ProcessProposal::Reject)
+        }
+    }
+
     /// Signals the beginning of a new block, prior to any `DeliverTx` calls.
     async fn begin_block(&self, request: request::BeginBlock) -> AbciResult<response::BeginBlock> {
         let db = self.state_store_clone();

fendermint/app/src/cmd/run.rs

@@ -6,7 +6,9 @@ use fendermint_abci::ApplicationService;
 use fendermint_app::{App, AppStore};
 use fendermint_rocksdb::{blockstore::NamespaceBlockstore, namespaces, RocksDb, RocksDbConfig};
 use fendermint_vm_interpreter::{
-    bytes::BytesMessageInterpreter, chain::ChainMessageInterpreter, fvm::FvmMessageInterpreter,
+    bytes::{BytesMessageInterpreter, ProposalPrepareMode},
+    chain::ChainMessageInterpreter,
+    fvm::FvmMessageInterpreter,
     signed::SignedMessageInterpreter,
 };
 use tracing::info;
@@ -27,7 +29,8 @@ async fn run(settings: Settings) -> anyhow::Result<()> {
     );
     let interpreter = SignedMessageInterpreter::new(interpreter);
     let interpreter = ChainMessageInterpreter::new(interpreter);
-    let interpreter = BytesMessageInterpreter::new(interpreter);
+    let interpreter =
+        BytesMessageInterpreter::new(interpreter, ProposalPrepareMode::AppendOnly, false);
 
     let ns = Namespaces::default();
     let db = open_db(&settings, &ns).context("error opening DB")?;

fendermint/vm/interpreter/src/bytes.rs

@@ -1,6 +1,6 @@
 // Copyright 2022-2023 Protocol Labs
 // SPDX-License-Identifier: Apache-2.0, MIT
-use anyhow::anyhow;
+use anyhow::{anyhow, Context};
 use async_trait::async_trait;
 use cid::Cid;
 use fendermint_vm_genesis::Genesis;
@@ -9,7 +9,7 @@ use fendermint_vm_message::chain::ChainMessage;
 use crate::{
     chain::{ChainMessageApplyRet, ChainMessageCheckRet},
     fvm::{FvmQuery, FvmQueryRet},
-    CheckInterpreter, ExecInterpreter, GenesisInterpreter, QueryInterpreter,
+    CheckInterpreter, ExecInterpreter, GenesisInterpreter, ProposalInterpreter, QueryInterpreter,
 };
 
 pub type BytesMessageApplyRet = Result<ChainMessageApplyRet, fvm_ipld_encoding::Error>;
@@ -19,15 +19,124 @@ pub type BytesMessageQueryRet = Result<FvmQueryRet, fvm_ipld_encoding::Error>;
 /// Close to what the ABCI sends: (Path, Bytes).
 pub type BytesMessageQuery = (String, Vec<u8>);
 
+/// Behavour of proposal preparation. It's an optimisation to cut down needless serialization
+/// when we know we aren't doing anything with the messages.
+#[derive(Debug, Default, Clone)]
+pub enum ProposalPrepareMode {
+    /// Deserialize all messages and pass them to the inner interpreter.
+    #[default]
+    PassThrough,
+    /// Does not pass messages to the inner interpreter, only appends what is returned from it.
+    AppendOnly,
+    /// Does not pass messages to the inner interpreter, only prepends what is returned from it.
+    PrependOnly,
+}
+
 /// Interpreter working on raw bytes.
 #[derive(Clone)]
 pub struct BytesMessageInterpreter<I> {
     inner: I,
+    /// Should we parse and pass on all messages during prepare.
+    prepare_mode: ProposalPrepareMode,
+    /// Should we reject proposals with transactions we cannot parse.
+    reject_malformed_proposal: bool,
 }
 
 impl<I> BytesMessageInterpreter<I> {
-    pub fn new(inner: I) -> Self {
-        Self { inner }
+    pub fn new(
+        inner: I,
+        prepare_mode: ProposalPrepareMode,
+        reject_malformed_proposal: bool,
+    ) -> Self {
+        Self {
+            inner,
+            prepare_mode,
+            reject_malformed_proposal,
+        }
+    }
+}
+
+#[async_trait]
+impl<I> ProposalInterpreter for BytesMessageInterpreter<I>
+where
+    I: ProposalInterpreter<Message = ChainMessage>,
+{
+    type State = I::State;
+    type Message = Vec<u8>;
+
+    /// Parse messages in the mempool and pass them into the inner `ChainMessage` interpreter.
+    async fn prepare(
+        &self,
+        state: Self::State,
+        msgs: Vec<Self::Message>,
+    ) -> anyhow::Result<Vec<Self::Message>> {
+        // Collect the messages to pass to the inner interpreter.
+        let chain_msgs = match self.prepare_mode {
+            ProposalPrepareMode::PassThrough => {
+                let mut chain_msgs = Vec::new();
+                for msg in msgs.iter() {
+                    match fvm_ipld_encoding::from_slice::<ChainMessage>(msg) {
+                        Err(e) => {
+                            // This should not happen because the `CheckInterpreter` implementation below would
+                            // have rejected any such user transaction.
+                            tracing::warn!(
+                                error = e.to_string(),
+                                "failed to decode message in mempool as ChainMessage"
+                            );
+                        }
+                        Ok(msg) => chain_msgs.push(msg),
+                    }
+                }
+                chain_msgs
+            }
+            ProposalPrepareMode::AppendOnly | ProposalPrepareMode::PrependOnly => Vec::new(),
+        };
+
+        let chain_msgs = self.inner.prepare(state, chain_msgs).await?;
+
+        let chain_msgs = chain_msgs
+            .into_iter()
+            .map(|msg| {
+                fvm_ipld_encoding::to_vec(&msg).context("failed to encode ChainMessage as IPLD")
+            })
+            .collect::<anyhow::Result<Vec<Self::Message>>>()?;
+
+        match self.prepare_mode {
+            ProposalPrepareMode::PassThrough => Ok(chain_msgs),
+            ProposalPrepareMode::AppendOnly => Ok(vec![msgs, chain_msgs].concat()),
+            ProposalPrepareMode::PrependOnly => Ok(vec![chain_msgs, msgs].concat()),
+        }
+    }
+
+    /// Parse messages in the block, reject if unknown format. Pass the rest to the inner `ChainMessage` interpreter.
+    async fn process(&self, state: Self::State, msgs: Vec<Self::Message>) -> anyhow::Result<bool> {
+        let mut chain_msgs = Vec::new();
+
+        for msg in msgs {
+            match fvm_ipld_encoding::from_slice::<ChainMessage>(&msg) {
+                Err(e) => {
+                    // If we cannot parse a message, then either:
+                    // * The proposer is Byzantine - as an attack this isn't very effective as they could just not send a proposal and cause a timeout.
+                    // * Our or the proposer node have different versions, or contain bugs
+                    // We can either vote for it or not:
+                    // * If we accept, we can punish the validator during block execution, and if it turns out we had a bug, we will have a consensus failure.
+                    // * If we accept, then the serialization error will become visible in the transaction results through RPC.
+                    // * If we reject, the majority can still accept the block, which indicates we had the bug (that way we might even panic during delivery, since we know it got voted on),
+                    //   but a buggy transaction format that fails for everyone would cause liveness issues.
+                    // * If we reject, then the serialization error will only be visible in the logs (and potentially earlier check_tx results).
+                    tracing::warn!(
+                        error = e.to_string(),
+                        "failed to decode message in proposal as ChainMessage"
+                    );
+                    if self.reject_malformed_proposal {
+                        return Ok(false);
+                    }
+                }
+                Ok(msg) => chain_msgs.push(msg),
+            }
+        }
+
+        self.inner.process(state, chain_msgs).await
     }
 }
 
@@ -53,6 +162,14 @@ where
             // There is always the possibility that our codebase is incompatible,
             // but then we'll have a consensus failure later when we don't agree on the ledger.
             {
+                if self.reject_malformed_proposal {
+                    // We could consider panicking here, otherwise if the majority executes this transaction (they voted for it)
+                    // then we will just get a consensu failure after the block.
+                    tracing::warn!(
+                        error = e.to_string(),
+                        "failed to decode delivered message as ChainMessage; we did not vote for it, maybe our node is buggy?"
+                    );
+                }
                 Ok((state, Err(e)))
             }
             Ok(msg) => {

fendermint/vm/interpreter/src/chain.rs

@@ -7,7 +7,7 @@ use fendermint_vm_message::{chain::ChainMessage, signed::SignedMessage};
 
 use crate::{
     signed::{SignedMessageApplyRet, SignedMessageCheckRet},
-    CheckInterpreter, ExecInterpreter, GenesisInterpreter, QueryInterpreter,
+    CheckInterpreter, ExecInterpreter, GenesisInterpreter, ProposalInterpreter, QueryInterpreter,
 };
 
 /// A message a user is not supposed to send.
@@ -34,6 +34,40 @@ impl<I> ChainMessageInterpreter<I> {
     }
 }
 
+#[async_trait]
+impl<I> ProposalInterpreter for ChainMessageInterpreter<I>
+where
+    I: Sync + Send,
+{
+    // TODO: The state can include the IPLD Resolver mempool, for example by using STM
+    // to implement a shared memory space.
+    type State = ();
+    type Message = ChainMessage;
+
+    /// Check whether there are any "ready" messages in the IPLD resolution mempool which can be appended to the proposal.
+    ///
+    /// We could also use this to select the most profitable user transactions, within the gas limit. We can also take into
+    /// account the transactions which are part of top-down or bottom-up checkpoints, to stay within gas limits.
+    async fn prepare(
+        &self,
+        _state: Self::State,
+        msgs: Vec<Self::Message>,
+    ) -> anyhow::Result<Vec<Self::Message>> {
+        // For now this is just a placeholder.
+        Ok(msgs)
+    }
+
+    /// Perform finality checks on top-down transactions and availability checks on bottom-up transactions.
+    async fn process(
+        &self,
+        _state: Self::State,
+        _msgs: Vec<Self::Message>,
+    ) -> anyhow::Result<bool> {
+        // For now this is just a placeholder.
+        Ok(true)
+    }
+}
+
 #[async_trait]
 impl<I> ExecInterpreter for ChainMessageInterpreter<I>
 where