Aggregate SecretConnection chunks with unmarshal protobuf retry

src/rpc.rs

@@ -7,12 +7,9 @@ use crate::privval::SignableMsg;
 use prost::Message as _;
 use std::io::Read;
 use tendermint::{chain, Proposal, Vote};
+use tendermint_p2p::secret_connection::DATA_MAX_SIZE;
 use tendermint_proto as proto;
 
-// TODO(tarcieri): use `tendermint_p2p::secret_connection::DATA_MAX_SIZE`
-// See informalsystems/tendermint-rs#1356
-const DATA_MAX_SIZE: usize = 262144;
-
 use crate::{
     error::{Error, ErrorKind},
     prelude::*,
@@ -31,12 +28,36 @@ pub enum Request {
 impl Request {
     /// Read a request from the given readable.
     pub fn read(conn: &mut impl Read, expected_chain_id: &chain::Id) -> Result<Self, Error> {
-        let msg_bytes = read_msg(conn)?;
-
-        // Parse Protobuf-encoded request message
-        let msg = proto::privval::Message::decode_length_delimited(msg_bytes.as_ref())
-            .map_err(|e| format_err!(ErrorKind::ProtocolError, "malformed message packet: {}", e))?
-            .sum;
+        let mut msg_bytes: Vec<u8> = vec![];
+        let msg;
+
+        // fix for Sei: collect incoming bytes of Protobuf from incoming msg
+        loop {
+            let mut msg_chunk = read_msg(conn)?;
+            let chunk_len = msg_chunk.len();
+            msg_bytes.append(&mut msg_chunk);
+
+            // if we can decode it, great, break the loop
+            match proto::privval::Message::decode_length_delimited(msg_bytes.as_ref()) {
+                Ok(m) => {
+                    msg = m.sum;
+                    break;
+                }
+                Err(e) => {
+                    // if chunk_len < DATA_MAX_SIZE (1024) we assume it was the end of the message and it is malformed
+                    if chunk_len < DATA_MAX_SIZE {
+                        return Err(format_err!(
+                            ErrorKind::ProtocolError,
+                            "malformed message packet: {}",
+                            e
+                        )
+                        .into());
+                    }
+                    // otherwise, we go to start of the loop assuming next chunk(s)
+                    // will fill the message
+                }
+            }
+        }
 
         let (req, chain_id) = match msg {
             Some(proto::privval::message::Sum::SignVoteRequest(

tests/integration.rs

@@ -5,6 +5,7 @@ use chrono::{DateTime, Utc};
 use prost::Message;
 use rand::Rng;
 use signature::Verifier;
+use std::fs::File;
 use std::{
     fs,
     io::{self, Cursor, Read, Write},
@@ -617,6 +618,20 @@ fn test_handle_and_sign_ping_pong() {
     });
 }
 
+#[test]
+fn test_buffer_underflow_sign_proposal() {
+    let key_type = KeyType::Consensus;
+    ProtocolTester::apply(&key_type, |mut pt| {
+        send_buffer_underflow_request(&mut pt);
+        let response: Result<(), ()> = match read_response(&mut pt) {
+            proto::privval::message::Sum::SignedProposalResponse(_) => Ok(()),
+            other => panic!("unexpected message type in response: {other:?}"),
+        };
+
+        assert!(response.is_ok());
+    });
+}
+
 /// Encode request as a Protobuf message
 fn send_request(request: proto::privval::message::Sum, pt: &mut ProtocolTester) {
     let mut buf = vec![];
@@ -627,6 +642,15 @@ fn send_request(request: proto::privval::message::Sum, pt: &mut ProtocolTester)
     pt.write_all(&buf).unwrap();
 }
 
+/// Opens a binary file with big proposal (> 1024 bytes, from Sei network)
+/// and sends via protocol tester
+fn send_buffer_underflow_request(pt: &mut ProtocolTester) {
+    let mut file = File::open("tests/support/buffer-underflow-proposal.bin").unwrap();
+    let mut buf = Vec::<u8>::new();
+    file.read_to_end(&mut buf).unwrap();
+    pt.write_all(&buf).unwrap();
+}
+
 /// Read the response as a Protobuf message
 fn read_response(pt: &mut ProtocolTester) -> proto::privval::message::Sum {
     let mut resp_buf = vec![0u8; 4096];