Top down discussion

Cargo.toml

@@ -9,6 +9,7 @@ members = [
   "fendermint/testing",
   "fendermint/testing/*-test",
   "fendermint/vm/*",
+  "fendermint/ipc",
 ]
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -56,7 +57,7 @@ serde_tuple = "0.5"
 serde_with = "2.3" 
 tempfile = "3.3" 
 thiserror = "1" 
-tokio = {version = "1", features = ["rt-multi-thread", "macros"]} 
+tokio = {version = "1", features = ["rt-multi-thread", "macros", "time"]}
 tracing = "0.1" 
 tracing-subscriber = "0.3" 
 
@@ -104,3 +105,4 @@ tendermint-proto = { version = "0.31" }
 
 # Using the IPC SDK without the `fil-actor` feature so as not to depend on the actor `Runtime`.
 ipc-sdk = { git = "https://github.com/consensus-shipyard/ipc-actors.git", default-features = false, branch = "main" }
+ipc-gateway = { git = "https://github.com/consensus-shipyard/ipc-actors.git", features = [] }

fendermint/app/Cargo.toml

@@ -41,6 +41,7 @@ fendermint_vm_core = { path = "../vm/core" }
 fendermint_vm_interpreter = { path = "../vm/interpreter", features = ["bundle"] }
 fendermint_vm_message = { path = "../vm/message" }
 fendermint_vm_genesis = { path = "../vm/genesis" }
+fendermint_ipc = { path = "../ipc" }
 
 cid = { workspace = true }
 fvm = { workspace = true }

fendermint/app/src/app.rs

@@ -6,8 +6,12 @@ use std::sync::{Arc, Mutex};
 
 use anyhow::{anyhow, Context, Result};
 use async_trait::async_trait;
+use bytes::Bytes;
 use cid::Cid;
 use fendermint_abci::{AbciResult, Application};
+use fendermint_ipc::parent::PollingParentSyncer;
+use fendermint_ipc::pof::ParentFinalityProvider;
+use fendermint_ipc::IPCMessage;
 use fendermint_storage::{
     Codec, Encode, KVCollection, KVRead, KVReadable, KVStore, KVWritable, KVWrite,
 };
@@ -24,6 +28,7 @@ use fendermint_vm_interpreter::signed::InvalidSignature;
 use fendermint_vm_interpreter::{
     CheckInterpreter, ExecInterpreter, GenesisInterpreter, QueryInterpreter,
 };
+use fendermint_vm_message::chain::ChainMessage;
 use fvm::engine::MultiEngine;
 use fvm_ipld_blockstore::Blockstore;
 use fvm_shared::chainid::ChainID;
@@ -125,6 +130,8 @@ where
     ///
     /// Zero means unlimited.
     state_hist_size: u64,
+    /// The top down parent finality.
+    parent_finality: ParentFinalityProvider<PollingParentSyncer>,
 }
 
 impl<DB, SS, S, I> App<DB, SS, S, I>
@@ -157,6 +164,7 @@ where
             interpreter: Arc::new(interpreter),
             exec_state: Arc::new(Mutex::new(None)),
             check_state: Arc::new(tokio::sync::Mutex::new(None)),
+            parent_finality: (),
         };
         app.init_committed_state()?;
         Ok(app)
@@ -291,6 +299,37 @@ where
         let state = self.committed_state()?;
         Ok((state.state_params, state.block_height))
     }
+
+    fn verify_ipc_message(&self, ipc_message: IPCMessage) -> bool {
+        match ipc_message {
+            IPCMessage::TopDown(finality) => self.parent_finality.check_finality(&finality),
+            IPCMessage::BottomUp => unimplemented!(),
+        }
+    }
+
+    /// Verifies the list of transactions to see if they are valid. Returns true if all the txns
+    /// are valid and false if any one of the txns is invalid.
+    fn verify_messages(&self, txns: &Vec<Bytes>) -> bool {
+        for tx in txns {
+            let is_ok = match serde_json::from_slice::<ChainMessage>(tx) {
+                Err(_) => {
+                    // FIXME: maybe we can reject the proposal when the transaction cannot be deserialized
+                    tracing::info!("cannot deserialized transaction: {tx:?}");
+                    continue;
+                }
+                Ok(message) => match message {
+                    ChainMessage::IPC(ipc) => self.verify_ipc_message(ipc),
+                    _ => continue,
+                },
+            };
+
+            if !is_ok {
+                return false;
+            }
+        }
+
+        true
+    }
 }
 
 // NOTE: The `Application` interface doesn't allow failures at the moment. The protobuf
@@ -490,6 +529,47 @@ where
         Ok(response)
     }
 
+    /// 1. Prepare txns
+    /// 2. Prepare parent finality proposal
+    async fn prepare_proposal(
+        &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);
+        }
+
+        if let Some(finality_proposal) = self
+            .parent_finality
+            .next_finality_proposal()
+            .map_err(|e| anyhow!("cannot get next finality to propose due to: {e}"))?
+        {
+            let bytes =
+                serde_json::to_vec(&ChainMessage::IPC(IPCMessage::TopDown(finality_proposal)))?;
+            txs.push(bytes.into());
+        }
+
+        Ok(response::PrepareProposal { txs })
+    }
+
+    async fn process_proposal(
+        &self,
+        request: request::ProcessProposal,
+    ) -> AbciResult<response::ProcessProposal> {
+        if !self.verify_messages(&request.txs) {
+            Ok(response::ProcessProposal::Reject)
+        } else {
+            Ok(response::ProcessProposal::Accept)
+        }
+    }
+
     /// 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();
@@ -538,6 +618,7 @@ where
                     );
                     to_deliver_tx(ret)
                 }
+                _ => todo!(),
             },
         };
 

fendermint/ipc/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "fendermint_ipc"
+version = "0.1.0"
+edition = "2021"
+description = "Interfacing with IPC"
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+tokio = { workspace = true }
+anyhow = { workspace = true }
+ipc-sdk = { workspace = true }
+async-trait = { workspace = true }
+tracing = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+cid = { workspace = true }
+fvm_ipld_encoding = { workspace = true }
+num-traits = { workspace = true }

fendermint/ipc/src/agent.rs

@@ -0,0 +1,29 @@
+// Copyright 2022-2023 Protocol Labs
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use crate::BlockHeight;
+
+/// The ipc agent proxy. Use this struct to interface with the running ipc-agent.
+pub(crate) struct AgentProxy {}
+
+impl AgentProxy {
+    pub async fn get_chain_head_height(&self) -> anyhow::Result<BlockHeight> {
+        todo!()
+    }
+
+    pub async fn get_block_hash(&self, _height: BlockHeight) -> anyhow::Result<Vec<u8>> {
+        todo!()
+    }
+
+    // pub async fn get_top_down_msgs(
+    //     &self,
+    //     _height: BlockHeight,
+    //     _nonce: u64,
+    // ) -> anyhow::Result<Vec<CrossMsg>> {
+    //     todo!()
+    // }
+    //
+    // pub async fn get_membership(&self) -> anyhow::Result<ValidatorSet> {
+    //     todo!()
+    // }
+}

fendermint/ipc/src/cache.rs

@@ -0,0 +1,104 @@
+// Copyright 2022-2023 Protocol Labs
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use num_traits::{PrimInt};
+use std::cmp::{max, min};
+use std::collections::HashMap;
+use std::hash::Hash;
+
+type Bounds<T> = (T, T);
+
+/// The key value cache. The key must be numeric and falls within a range/bound.
+pub struct RangeKeyCache<Key, Value> {
+    /// Stores the data in a hashmap.
+    data: HashMap<Key, Value>,
+    /// The lower and upper bound of keys stored in data
+    bounds: Option<Bounds<Key>>,
+}
+
+impl<Key: PrimInt + Hash, Value> RangeKeyCache<Key, Value> {
+    pub fn new() -> Self {
+        Self {
+            data: Default::default(),
+            bounds: None,
+        }
+    }
+
+    pub fn upper_bound(&self) -> Option<Key> {
+        self.within_bounds(|(_, upper)| *upper)
+    }
+
+    pub fn get_value(&self, key: Key) -> Option<&Value> {
+        self.within_bounds(|(lower_bound, upper_bound)| {
+            if *lower_bound > key || *upper_bound < key {
+                return None;
+            }
+            return self.data.get(&key);
+        })
+        .flatten()
+    }
+
+    pub fn values_within_range(&self, start: Key, end: Option<Key>) -> Vec<&Value> {
+        self.within_bounds(|(lower_bound, upper_bound)| {
+            let start = max(*lower_bound, start);
+            let end = min(*upper_bound, end.unwrap_or(*upper_bound));
+
+            let mut r = vec![];
+            let mut i = start;
+            while i <= end {
+                if let Some(v) = self.get_value(i) {
+                    r.push(v);
+                }
+
+                i = i + Key::one();
+            }
+
+            r
+        })
+        .unwrap_or(vec![])
+    }
+
+    /// Removes the block hashes stored till the specified height, exclusive.
+    pub fn remove_key_till(&mut self, key: Key) {
+        if let Some((lower_bound, upper_bound)) = self.bounds.as_mut() {
+            if *lower_bound > key || *upper_bound < key {
+                return;
+            }
+
+            let mut i = *lower_bound;
+            while i < key {
+                self.data.remove(&i);
+                i = i + Key::one();
+            }
+
+            *lower_bound = key;
+        }
+    }
+
+    /// Insert the block hash at the next height
+    pub fn insert_after_lower_bound(&mut self, key: Key, val: Value) {
+        match &mut self.bounds {
+            None => {
+                self.data.insert(key, val);
+                self.bounds.replace((key, key));
+            }
+            Some((upper, lower)) => {
+                if *lower > key {
+                    return;
+                }
+
+                self.data.insert(key, val);
+                if *upper < key {
+                    *upper = key;
+                }
+            }
+        }
+    }
+
+    fn within_bounds<F, T>(&self, f: F) -> Option<T>
+    where
+        F: Fn(&(Key, Key)) -> T,
+    {
+        self.bounds.as_ref().map(f)
+    }
+}

fendermint/ipc/src/lib.rs

@@ -0,0 +1,58 @@
+// Copyright 2022-2023 Protocol Labs
+// SPDX-License-Identifier: Apache-2.0, MIT
+//! Interfacing with IPC, provides utility functions
+
+mod agent;
+mod cache;
+mod message;
+pub mod parent;
+pub mod pof;
+
+use crate::pof::IPCParentFinality;
+use async_trait::async_trait;
+use ipc_sdk::cross::CrossMsg;
+use ipc_sdk::ValidatorSet;
+pub use message::IPCMessage;
+
+#[derive(Debug, Clone)]
+pub struct Config {
+    /// The number of blocks to delay reporting when creating the pof
+    chain_head_delay: BlockHeight,
+    /// The lower bound for the chain head height in parent view
+    chain_head_lower_bound: BlockHeight,
+
+    /// Parent syncing cron period, in seconds
+    polling_interval: u64,
+}
+
+type BlockHeight = u64;
+type Nonce = u64;
+
+/// Obtain the latest state information required for IPC from the parent subnet
+pub trait ParentViewProvider {
+    /// Fetch the latest chain head
+    fn latest_height(&self) -> Option<BlockHeight>;
+    /// Fetch the block hash at target height
+    fn block_hash(&self, height: BlockHeight) -> Option<Vec<u8>>;
+    /// Get the top down messages from the nonce of a height
+    fn top_down_msgs(&self, height: BlockHeight, nonce: u64) -> Vec<CrossMsg>;
+    /// Get the latest membership information
+    fn membership(&self) -> Option<ValidatorSet>;
+
+    /// Called when finality is committed
+    fn on_finality_committed(&self, finality: &IPCParentFinality);
+}
+
+/// Obtain the latest state information required for IPC from the parent subnet
+#[async_trait]
+pub trait ParentViewQuery {
+    /// Fetch the latest chain head
+    async fn latest_height(&self) -> anyhow::Result<BlockHeight>;
+    /// Fetch the block hash at target height
+    async fn block_hash(&self, height: BlockHeight) -> anyhow::Result<Option<Vec<u8>>>;
+    /// Get the top down messages from the nonce of a height
+    async fn top_down_msgs(&self, height: BlockHeight, nonce: u64)
+        -> anyhow::Result<Vec<CrossMsg>>;
+    /// Get the latest membership information
+    async fn membership(&self) -> anyhow::Result<ValidatorSet>;
+}

fendermint/ipc/src/message.rs

@@ -0,0 +1,19 @@
+// Copyright 2022-2023 Protocol Labs
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+//! IPC messages
+
+use crate::pof::IPCParentFinality;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub enum IPCMessage {
+    TopDown(IPCParentFinality),
+    BottomUp,
+}
+
+impl From<IPCMessage> for Vec<u8> {
+    fn from(value: IPCMessage) -> Self {
+        serde_json::to_vec(&value).expect("should not happen")
+    }
+}