Add Commit message in Tendermint

Before this commit, Tendermint extension only requests current height
and view's votes. Handling only the current view's message makes code
simple. However, there may be a commit before the current view. This
creates a liveness problem.

After this commit, a node requests a Commit message if some of its
peer's height is higher than the node. The commit message is not
related to the node's current view. The Commit message fixes the
liveness problem.

Author: majecty

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,14 @@ pub enum TendermintMessage {
         height: Height,
         view: View,
     },
+    RequestCommit {
+        height: Height,
+    },
+    Commit {
+        height: Height,
+        block: Bytes,
+        votes: Vec<ConsensusMessage>,
+    },
 }
 
 impl Encodable for TendermintMessage {
@@ -160,6 +170,24 @@ 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 {
+                height,
+                block,
+                votes,
+            } => {
+                s.begin_list(4);
+                s.append(&MESSAGE_ID_COMMIT);
+                s.append(height);
+                s.append(block);
+                s.append_list(votes);
+            }
         }
     }
 }
@@ -253,6 +281,36 @@ 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 != 4 {
+                    return Err(DecoderError::RlpIncorrectListLen {
+                        got: item_count,
+                        expected: 4,
+                    })
+                }
+                let height = rlp.at(1)?.as_val()?;
+                let block = rlp.at(2)?.as_val()?;
+                let votes = rlp.at(3)?.as_list()?;
+                TendermintMessage::Commit {
+                    height,
+                    block,
+                    votes,
+                }
+            }
             _ => return Err(DecoderError::Custom("Unknown message id detected")),
         })
     }
@@ -408,6 +466,43 @@ 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 {
+            height: 3,
+            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,45 @@ 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::unbounded();
+                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 {
+                height,
+                block,
+                votes,
+            }) => {
+                ctrace!(ENGINE, "Received Commit for {:?} from {:?}", height, token);
+                let (result, receiver) = crossbeam::unbounded();
+                self.inner
+                    .send(worker::Event::GetCommit {
+                        height,
+                        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),
         }
     }