Implement Commit message

core/src/consensus/tendermint/message.rs

@@ -79,6 +79,8 @@ const MESSAGE_ID_PROPOSAL_BLOCK: u8 = 0x02;
 const MESSAGE_ID_STEP_STATE: u8 = 0x03;
 const MESSAGE_ID_REQUEST_MESSAGE: u8 = 0x04;
 const MESSAGE_ID_REQUEST_PROPOSAL: u8 = 0x05;
+const MESSAGE_ID_REQUEST_COMMIT: u8 = 0x06;
+const MESSAGE_ID_COMMIT: u8 = 0x07;
 
 #[derive(Debug, PartialEq)]
 pub enum TendermintMessage {
@@ -102,6 +104,13 @@ pub enum TendermintMessage {
         height: Height,
         view: View,
     },
+    RequestCommit {
+        height: Height,
+    },
+    Commit {
+        block: Bytes,
+        votes: Vec<ConsensusMessage>,
+    },
 }
 
 impl Encodable for TendermintMessage {
@@ -160,6 +169,22 @@ impl Encodable for TendermintMessage {
                 s.append(height);
                 s.append(view);
             }
+            TendermintMessage::RequestCommit {
+                height,
+            } => {
+                s.begin_list(2);
+                s.append(&MESSAGE_ID_REQUEST_COMMIT);
+                s.append(height);
+            }
+            TendermintMessage::Commit {
+                block,
+                votes,
+            } => {
+                s.begin_list(3);
+                s.append(&MESSAGE_ID_COMMIT);
+                s.append(block);
+                s.append_list(votes);
+            }
         }
     }
 }
@@ -253,6 +278,34 @@ impl Decodable for TendermintMessage {
                     view,
                 }
             }
+            MESSAGE_ID_REQUEST_COMMIT => {
+                let item_count = rlp.item_count()?;
+                if item_count != 2 {
+                    return Err(DecoderError::RlpIncorrectListLen {
+                        got: item_count,
+                        expected: 2,
+                    })
+                }
+                let height = rlp.at(1)?.as_val()?;
+                TendermintMessage::RequestCommit {
+                    height,
+                }
+            }
+            MESSAGE_ID_COMMIT => {
+                let item_count = rlp.item_count()?;
+                if item_count != 3 {
+                    return Err(DecoderError::RlpIncorrectListLen {
+                        got: item_count,
+                        expected: 3,
+                    })
+                }
+                let block = rlp.at(1)?.as_val()?;
+                let votes = rlp.at(2)?.as_list()?;
+                TendermintMessage::Commit {
+                    block,
+                    votes,
+                }
+            }
             _ => return Err(DecoderError::Custom("Unknown message id detected")),
         })
     }
@@ -408,6 +461,42 @@ mod tests {
         });
     }
 
+    #[test]
+    fn encode_and_decode_tendermint_message_6() {
+        rlp_encode_and_decode_test!(TendermintMessage::RequestCommit {
+            height: 3,
+        });
+    }
+
+    #[test]
+    fn encode_and_decode_tendermint_message_7() {
+        rlp_encode_and_decode_test!(TendermintMessage::Commit {
+            block: vec![1u8, 2u8],
+            votes: vec![
+                ConsensusMessage {
+                    signature: SchnorrSignature::random(),
+                    signer_index: 0x1234,
+                    on: VoteOn {
+                        step: VoteStep::new(2, 3, Step::Commit),
+                        block_hash: Some(H256::from(
+                            "07feab4c39250abf60b77d7589a5b61fdf409bd837e936376381d19db1e1f050"
+                        )),
+                    },
+                },
+                ConsensusMessage {
+                    signature: SchnorrSignature::random(),
+                    signer_index: 0x1235,
+                    on: VoteOn {
+                        step: VoteStep::new(2, 3, Step::Commit),
+                        block_hash: Some(H256::from(
+                            "07feab4c39250abf60b77d7589a5b61fdf409bd837e936376381d19db1e1f050"
+                        )),
+                    },
+                }
+            ]
+        });
+    }
+
     #[test]
     fn encode_and_decode_consensus_message_1() {
         let message = ConsensusMessage::default();

core/src/consensus/tendermint/network.rs

@@ -352,6 +352,43 @@ impl NetworkExtension<Event> for TendermintExtension {
                     self.send_votes(token, votes);
                 }
             }
+            Ok(TendermintMessage::RequestCommit {
+                height,
+            }) => {
+                ctrace!(ENGINE, "Received RequestCommit for {} from {:?}", height, token);
+                let (result, receiver) = crossbeam::bounded(1);
+                self.inner
+                    .send(worker::Event::RequestCommit {
+                        height,
+                        result,
+                    })
+                    .unwrap();
+
+                if let Ok(message) = receiver.recv() {
+                    ctrace!(ENGINE, "Send commit for {} to {:?}", height, token);
+                    self.api.send(token, Arc::new(message));
+                }
+            }
+            Ok(TendermintMessage::Commit {
+                block,
+                votes,
+            }) => {
+                ctrace!(ENGINE, "Received Commit from {:?}", token);
+                let (result, receiver) = crossbeam::bounded(1);
+                self.inner
+                    .send(worker::Event::GetCommit {
+                        block: block.clone(),
+                        votes,
+                        result,
+                    })
+                    .unwrap();
+
+                if let Some(c) = receiver.recv().unwrap() {
+                    if let Err(e) = c.import_block(block) {
+                        cinfo!(ENGINE, "Failed to import committed block {:?}", e);
+                    }
+                }
+            }
             _ => cinfo!(ENGINE, "Invalid message from peer {}", token),
         }
     }