feat(consensus): create SingleHeightConsensus struct

This involved some updating to the ConsensusBlock trait, mostly to better explain the design.
We implemented just the milestone 1 proposal flow for SHC.

crates/sequencing/papyrus_consensus/Cargo.toml

@@ -9,4 +9,9 @@ description = "Reach consensus for Starknet"
 
 [dependencies]
 async-trait.workspace = true
-tokio = { workspace = true, features = ["sync"] }
+futures.workspace = true
+starknet_api.workspace = true
+tokio = { workspace = true, features = ["full"] }
+
+[dev-dependencies]
+mockall.workspace = true

crates/sequencing/papyrus_consensus/src/lib.rs

@@ -1 +1,5 @@
+// TODO: Remove dead code allowance at the end of milestone 1.
+#[allow(dead_code)]
+pub mod single_height_consensus;
+#[allow(dead_code)]
 pub mod types;

crates/sequencing/papyrus_consensus/src/single_height_consensus.rs

@@ -0,0 +1,69 @@
+#[cfg(test)]
+#[path = "single_height_consensus_test.rs"]
+mod single_height_consensus_test;
+
+use std::sync::Arc;
+
+use futures::channel::{mpsc, oneshot};
+use futures::SinkExt;
+use starknet_api::block::BlockNumber;
+
+use crate::types::{
+    ConsensusBlock,
+    ConsensusContext,
+    NodeId,
+    PeeringConsensusMessage,
+    ProposalInit,
+};
+
+pub(crate) struct SingleHeightConsensus<BlockT>
+where
+    BlockT: ConsensusBlock,
+{
+    height: BlockNumber,
+    context: Arc<dyn ConsensusContext<Block = BlockT>>,
+    validators: Vec<NodeId>,
+    id: NodeId,
+    to_peering_sender: mpsc::Sender<PeeringConsensusMessage<BlockT::ProposalChunk>>,
+    from_peering_receiver: mpsc::Receiver<PeeringConsensusMessage<BlockT::ProposalChunk>>,
+}
+
+impl<BlockT> SingleHeightConsensus<BlockT>
+where
+    BlockT: ConsensusBlock,
+{
+    pub(crate) async fn new(
+        height: BlockNumber,
+        context: Arc<dyn ConsensusContext<Block = BlockT>>,
+        id: NodeId,
+        to_peering_sender: mpsc::Sender<PeeringConsensusMessage<BlockT::ProposalChunk>>,
+        from_peering_receiver: mpsc::Receiver<PeeringConsensusMessage<BlockT::ProposalChunk>>,
+    ) -> Self {
+        let validators = context.validators(height).await;
+        Self { height, context, validators, id, to_peering_sender, from_peering_receiver }
+    }
+
+    pub(crate) async fn run(mut self) -> BlockT {
+        // TODO: In the future this logic will be encapsulated in the state machine, and SHC will
+        // await a signal from SHC to propose.
+        let proposer_id = self.context.proposer(&self.validators, self.height);
+        if proposer_id == self.id {
+            self.propose().await
+        } else {
+            todo!("run as validator");
+        }
+    }
+
+    async fn propose(&mut self) -> BlockT {
+        let (content_receiver, block_receiver) = self.context.build_proposal(self.height).await;
+        let (block_hash_sender, block_hash_receiver) = oneshot::channel();
+        let init = ProposalInit { height: self.height, proposer: self.id };
+        self.to_peering_sender
+            .send(PeeringConsensusMessage::Proposal((init, content_receiver, block_hash_receiver)))
+            .await
+            .expect("failed to send proposal to peering");
+        let block = block_receiver.await.expect("failed to build block");
+        block_hash_sender.send(block.id()).expect("failed to send block hash");
+        block
+    }
+}

crates/sequencing/papyrus_consensus/src/single_height_consensus_test.rs

@@ -0,0 +1,102 @@
+use std::sync::Arc;
+
+use futures::channel::{mpsc, oneshot};
+use futures::StreamExt;
+use starknet_api::block::{BlockHash, BlockNumber};
+use starknet_api::hash::StarkFelt;
+use tokio;
+
+use super::SingleHeightConsensus;
+use crate::types::{
+    ConsensusBlock,
+    MockConsensusBlock,
+    MockTestContext,
+    NodeId,
+    PeeringConsensusMessage,
+    ProposalInit,
+};
+
+type Shc = SingleHeightConsensus<MockConsensusBlock>;
+type ProposalChunk = <MockConsensusBlock as ConsensusBlock>::ProposalChunk;
+type PeeringMessage = PeeringConsensusMessage<ProposalChunk>;
+
+struct TestFields {
+    pub context: MockTestContext,
+    pub shc_to_peering_sender: mpsc::Sender<PeeringConsensusMessage<u32>>,
+    pub shc_to_peering_receiver: mpsc::Receiver<PeeringConsensusMessage<u32>>,
+    pub peering_to_shc_sender: mpsc::Sender<PeeringConsensusMessage<u32>>,
+    pub peering_to_shc_receiver: mpsc::Receiver<PeeringConsensusMessage<u32>>,
+}
+
+impl TestFields {
+    async fn new_shc(
+        self,
+        height: BlockNumber,
+        id: NodeId,
+    ) -> (
+        Shc,
+        mpsc::Receiver<PeeringConsensusMessage<u32>>,
+        mpsc::Sender<PeeringConsensusMessage<u32>>,
+    ) {
+        let shc = Shc::new(
+            height,
+            Arc::new(self.context),
+            id,
+            self.shc_to_peering_sender,
+            self.peering_to_shc_receiver,
+        )
+        .await;
+        (shc, self.shc_to_peering_receiver, self.peering_to_shc_sender)
+    }
+}
+
+fn setup() -> TestFields {
+    let (shc_to_peering_sender, shc_to_peering_receiver) = mpsc::channel(1);
+    let (peering_to_shc_sender, peering_to_shc_receiver) = mpsc::channel(1);
+    let context = MockTestContext::new();
+    TestFields {
+        context,
+        shc_to_peering_sender,
+        shc_to_peering_receiver,
+        peering_to_shc_sender,
+        peering_to_shc_receiver,
+    }
+}
+
+#[tokio::test]
+async fn propose() {
+    let mut test_fields = setup();
+    let node_id: NodeId = 1;
+    let block_id = BlockHash(StarkFelt::try_from(0 as u128).unwrap());
+    // Set expectations for how the test should run:
+    test_fields.context.expect_validators().returning(move |_| vec![node_id, 2, 3, 4]);
+    test_fields.context.expect_proposer().returning(move |_, _| node_id);
+    let block_id_clone = block_id.clone();
+    test_fields.context.expect_build_proposal().returning(move |_| {
+        // SHC doesn't actually handle the content, so ignore for unit tests.
+        let (_, content_receiver) = mpsc::channel(1);
+        let (block_sender, block_receiver) = oneshot::channel();
+
+        // Create the mock block.
+        let mut block = MockConsensusBlock::new();
+        let block_id = block_id_clone.clone();
+        block.expect_id().returning(move || block_id.clone());
+
+        block_sender.send(block).unwrap();
+        (content_receiver, block_receiver)
+    });
+
+    // Creation calls to `context.validators`.
+    let (shc, mut shc_to_peering_receiver, _) = test_fields.new_shc(BlockNumber(0), node_id).await;
+
+    // This calls to `context.proposer` and `context.build_proposal`.
+    let block = shc.run().await;
+    assert_eq!(block.id(), block_id);
+
+    // Check what was sent to peering. We don't check the content stream as that is filled by
+    // ConsensusContext, not SHC.
+    let PeeringMessage::Proposal((init, _, block_hash_receiver)) =
+        shc_to_peering_receiver.next().await.unwrap();
+    assert_eq!(init, ProposalInit { height: BlockNumber(0), proposer: node_id });
+    assert_eq!(block_hash_receiver.await.unwrap(), block_id);
+}

crates/sequencing/papyrus_consensus/src/types.rs

@@ -3,33 +3,35 @@
 mod types_test;
 
 use async_trait::async_trait;
-use tokio::sync::{mpsc, oneshot};
+use futures::channel::{mpsc, oneshot};
+#[cfg(test)]
+use mockall::{automock, mock};
+use starknet_api::block::{BlockHash, BlockNumber};
 
 /// Used to identify the node by consensus.
 /// 1. This ID is derived from the id registered with Starknet's L2 staking contract.
 /// 2. We must be able to derive the public key associated with this ID for the sake of validating
 ///    signatures.
-// TODO: Determine the actual type of NodeId.
+// TODO(matan): Determine the actual type of NodeId.
 pub type NodeId = u64;
 
 /// Interface that any concrete block type must implement to be used by consensus.
-// In principle Consensus does not care about the content of a block. In practice though it will
-// need to perform certain activities with blocks:
-// 1. All proposals for a given height are held by consensus for book keeping, with only the decided
-//    block returned to ConsensusContext.
-// 2. If we re-propose a block, then Consensus needs the ability to stream out the content.
-// 3. We need the ability to identify the content of a proposal.
-pub trait ConsensusBlock {
-    /// The chunks of content returned for each step when streaming out the proposal.
-    // Why do we use an associated type?
-    // 1. In principle consensus is indifferent to the actual content in a proposal, and so we don't
-    //    want consensus to be tied to a specific concrete type.
-    // 2. Why not make `proposal_stream` generic? While this would allow a given concrete block type
-    //    to stream its content in multiple forms, this would also make `ConsensusBlock` object
-    //    unsafe, which hurts ergonomics.
+///
+/// In principle Consensus does not care about the content of a block. In practice though it will
+/// need to perform certain activities with blocks:
+/// 1. All proposals for a given height are held by consensus for book keeping, with only the
+///    decided block returned to ConsensusContext.
+/// 2. Tendermint may require re-broadcasting an old proposal [Line 16 of Algorithm 1]( https://arxiv.org/pdf/1807.04938)
+// This trait was designed with the following in mind:
+// 1. It must allow `ConsensusContext` to be object safe. This precludes generics.
+// 2. Starknet blocks are expected to be quite large, and we expect consensus to hold something akin
+//    to a reference with a small stack size and cheap shallow cloning.
+#[cfg_attr(test, automock(type ProposalChunk = u32; type ProposalIter = std::vec::IntoIter<u32>;))]
+pub trait ConsensusBlock: Send {
+    /// The chunks of content returned when iterating the proposal.
     type ProposalChunk;
     /// Iterator for accessing the proposal's content.
-    // ProposalIter is used instead of returning `impl Iterator` due to object safety.
+    // An associated type is used instead of returning `impl Iterator` due to object safety.
     type ProposalIter: Iterator<Item = Self::ProposalChunk>;
 
     /// Identifies the block for the sake of Consensus voting.
@@ -41,14 +43,24 @@ pub trait ConsensusBlock {
     // Since the proposal as well as votes sign not only on the block ID but also the height at
     // which they vote, not including height poses no security risk. Including it has no impact on
     // Tendermint.
-    fn id(&self) -> u64;
-
-    /// Returns an iterator over the block's content. Since this is the proposal content we stream
-    /// out only the information used to build the proposal need be included, not the built
-    /// information (ie the transactions, not the state diff).
-    // For milestone 1 this will be used by Peering when we receive a proposal, to fake streaming
-    // this into the ConsensusContext.
-    fn proposal_stream(&self) -> Self::ProposalIter;
+    fn id(&self) -> BlockHash;
+
+    /// Returns an iterator for streaming out this block as a proposal to other nodes.
+    // Note on the ownership and lifetime model. This call is done by reference, yet the returned
+    // iterator is implicitly an owning iterator.
+    // 1. Why did we not want reference iteration? This would require a lifetime to be part of the
+    //    type definition for `ProposalIter` and therefore `ConsensusBlock`. This results in a lot
+    //    of lifetime pollution making it much harder to work with this type; attempted both options
+    //    from here:
+    //    https://stackoverflow.com/questions/33734640/how-do-i-specify-lifetime-parameters-in-an-associated-type
+    // 2. Why is owning iteration reasonable? The expected use case for this is to stream out the
+    //    proposal to other nodes, which implies ownership of data, not just a reference for
+    //    internal use. We also expect the actual object implementing this trait to be itself a
+    //    reference to the underlying data, and so returning an "owning" iterator to be relatively
+    //    cheap.
+    // TODO(matan): Consider changing ConsensusBlock to `IntoIterator + Clone` and removing
+    // `proposal_iter`.
+    fn proposal_iter(&self) -> Self::ProposalIter;
 }
 
 /// Interface for consensus to call out to the node.
@@ -60,9 +72,14 @@ pub trait ConsensusBlock {
 //    limitation of Sync to keep functions `&self` shouldn't be a problem.
 #[async_trait]
 pub trait ConsensusContext: Send + Sync {
-    // See [`ConsensusBlock::ProposalChunk`] for why we use an associated type.
+    /// The [block](`ConsensusBlock`) type built by `ConsensusContext` from a proposal.
+    // We use an associated type since consensus is indifferent to the actual content of a proposal,
+    // but we cannot use generics due to object safety.
     type Block: ConsensusBlock;
 
+    // TODO(matan): The oneshot for receiving the build block could be generalized to just be some
+    // future which returns a block.
+
     /// This function is called by consensus to request a block from the node. It expects that this
     /// call will return immediately and that consensus can then stream in the block's content in
     /// parallel to the block being built.
@@ -78,7 +95,7 @@ pub trait ConsensusContext: Send + Sync {
     ///   ConsensusContext.
     async fn build_proposal(
         &self,
-        height: u64,
+        height: BlockNumber,
     ) -> (
         mpsc::Receiver<<Self::Block as ConsensusBlock>::ProposalChunk>,
         oneshot::Receiver<Self::Block>,
@@ -98,17 +115,52 @@ pub trait ConsensusContext: Send + Sync {
     ///   dropped by ConsensusContext.
     async fn validate_proposal(
         &self,
-        height: u64,
+        height: BlockNumber,
         content: mpsc::Receiver<<Self::Block as ConsensusBlock>::ProposalChunk>,
     ) -> oneshot::Receiver<Self::Block>;
 
     /// Get the set of validators for a given height. These are the nodes that can propose and vote
     /// on blocks.
-    // TODO: We expect this to change in the future to BTreeMap. Why?
+    // TODO(matan): We expect this to change in the future to BTreeMap. Why?
     // 1. Map - The nodes will have associated information (e.g. voting weight).
     // 2. BTreeMap - We want a stable ordering of the nodes for deterministic leader selection.
-    async fn validators(&self, height: u64) -> Vec<NodeId>;
+    async fn validators(&self, height: BlockNumber) -> Vec<NodeId>;
 
     /// Calculates the ID of the Proposer based on the inputs.
-    fn proposer(&self, validators: &Vec<NodeId>, height: u64) -> NodeId;
+    fn proposer(&self, validators: &Vec<NodeId>, height: BlockNumber) -> NodeId;
+}
+
+// Cannot use automock.
+#[cfg(test)]
+mock! {
+    pub(crate) TestContext {}
+
+    #[async_trait]
+    impl ConsensusContext for TestContext {
+        type Block = MockConsensusBlock;
+
+        async fn build_proposal(&self, height: BlockNumber) -> (
+            mpsc::Receiver<<MockConsensusBlock as ConsensusBlock>::ProposalChunk>,
+            oneshot::Receiver<<Self as ConsensusContext>::Block>
+        );
+        async fn validate_proposal(
+            &self,
+            height: BlockNumber,
+            content: mpsc::Receiver<<MockConsensusBlock as ConsensusBlock>::ProposalChunk>
+        ) -> oneshot::Receiver<<Self as ConsensusContext>::Block>;
+        async fn validators(&self, height: BlockNumber) -> Vec<NodeId>;
+        fn proposer(&self, validators: &Vec<NodeId>, height: BlockNumber) -> NodeId;
+    }
+}
+
+#[derive(PartialEq, Debug)]
+pub(crate) struct ProposalInit {
+    pub height: BlockNumber,
+    pub proposer: NodeId,
+}
+
+// This type encapsulate the messages that are sent between the SingleHeightConsensus and the
+// Peering components.
+pub(crate) enum PeeringConsensusMessage<ProposalChunkT> {
+    Proposal((ProposalInit, mpsc::Receiver<ProposalChunkT>, oneshot::Receiver<BlockHash>)),
 }

crates/sequencing/papyrus_consensus/src/types_test.rs

@@ -1,17 +1,10 @@
-use crate::types::{ConsensusBlock, ConsensusContext};
+use crate::types::ConsensusContext;
 
-// This should cause compilation to fail if the traits are not object safe.
+// This should cause compilation to fail if `ConsensusContext` is not object safe. Note that
+// `ConsensusBlock` need not be object safe for this to work.
 #[test]
 fn check_object_safety() {
-    // Arbitrarily chosen types for testing.
-    type _ProposalIter = std::slice::Iter<'static, u32>;
-    type _Blk = Box<dyn ConsensusBlock<ProposalChunk = u32, ProposalIter = _ProposalIter>>;
-
-    fn _check_consensus_block() -> _Blk {
-        todo!()
-    }
-
-    fn _check_context() -> Box<dyn ConsensusContext<Block = _Blk>> {
+    fn _check_context() -> Box<dyn ConsensusContext<Block = ()>> {
         todo!()
     }
 }