add an optional start timestamp to the rpc to get blocks (#89)

In the follow blocks script, we end up getting all the blocks from the
beginning up to the latest when what we really wanna do is see the
blocks minted starting from the time when we run the script. Add an
optional start timestamp parameter to the `GetBlocks` request to support
this.

src/bin/collect_stats.rs

@@ -111,15 +111,14 @@ async fn main() {
         });
 
         let start = Instant::now();
-        let start_farcaster_time = current_time();
         let mut stats_calculation_timer = time::interval(STATS_CALCULATION_INTERVAL);
         let mut block_count = 0;
         let mut num_messages_confirmed = 0;
         let mut num_messages_submitted = 0;
         let mut pending_messages = HashSet::new();
         let mut time_to_confirmation = vec![];
         let mut block_times = vec![];
-        let mut last_block_time = start_farcaster_time;
+        let mut last_block_time = current_time();
         let mut iteration = 0;
         loop {
             select! {
@@ -129,19 +128,17 @@ async fn main() {
                 },
                 Some(block) = blocks_rx.recv() => {
                     let block_timestamp = block.header.as_ref().unwrap().timestamp;
-                    if block_timestamp > start_farcaster_time {
-                        block_count += 1;
-                        block_times.push(block_timestamp - last_block_time);
-                        last_block_time = block_timestamp;
-                        for chunk in &block.shard_chunks {
-                            for tx in &chunk.transactions {
-                                for msg in &tx.user_messages {
-                                    let msg_data = MessageData::decode(msg.data_bytes.as_ref().unwrap().as_slice()).unwrap();
-                                    let msg_timestamp = msg_data.timestamp;
-                                    time_to_confirmation.push(block_timestamp  - msg_timestamp as u64);
-                                    num_messages_confirmed += 1;
-                                    pending_messages.remove(&hex::encode(msg.hash.clone()));
-                                }
+                    block_count += 1;
+                    block_times.push(block_timestamp - last_block_time);
+                    last_block_time = block_timestamp;
+                    for chunk in &block.shard_chunks {
+                        for tx in &chunk.transactions {
+                            for msg in &tx.user_messages {
+                                let msg_data = MessageData::decode(msg.data_bytes.as_ref().unwrap().as_slice()).unwrap();
+                                let msg_timestamp = msg_data.timestamp;
+                                time_to_confirmation.push(block_timestamp  - msg_timestamp as u64);
+                                num_messages_confirmed += 1;
+                                pending_messages.remove(&hex::encode(msg.hash.clone()));
                             }
                         }
                     }

src/consensus/proposer.rs

@@ -389,9 +389,9 @@ impl Proposer for BlockProposer {
                 let destination_addr = format!("http://{}", rpc_address.clone());
                 let mut rpc_client = SnapchainServiceClient::connect(destination_addr).await?;
                 let request = Request::new(BlocksRequest {
-                    shard_id: self.shard_id.shard_id(),
                     start_block_number: prev_block_number + 1,
                     stop_block_number: None,
+                    start_timestamp: None,
                 });
                 let missing_blocks = rpc_client.get_blocks(request).await?;
                 for block in missing_blocks.get_ref().blocks.clone() {

src/network/server.rs

@@ -69,6 +69,7 @@ impl SnapchainService for MySnapchainService {
     ) -> Result<Response<BlocksResponse>, Status> {
         let start_block_number = request.get_ref().start_block_number;
         let stop_block_number = request.get_ref().stop_block_number;
+        let start_timestamp = request.get_ref().start_timestamp;
 
         info!( {start_block_number, stop_block_number}, "Received call to [get_blocks] RPC");
 
@@ -78,6 +79,13 @@ impl SnapchainService for MySnapchainService {
         {
             Err(err) => Err(Status::from_error(Box::new(err))),
             Ok(blocks) => {
+                let blocks = match start_timestamp {
+                    None => blocks,
+                    Some(start_timestamp) => blocks
+                        .into_iter()
+                        .filter(|block| block.header.as_ref().unwrap().timestamp >= start_timestamp)
+                        .collect(),
+                };
                 let response = Response::new(BlocksResponse { blocks });
                 Ok(response)
             }

src/proto/rpc.proto

@@ -6,9 +6,9 @@ import "blocks.proto";
 import "hub_event.proto";
 
 message BlocksRequest {
-  uint32 shard_id = 1;
-  uint64 start_block_number = 2;
-  optional uint64 stop_block_number = 3;
+  uint64 start_block_number = 1;
+  optional uint64 stop_block_number = 2;
+  optional uint64 start_timestamp = 3;
 }
 
 message BlocksResponse {

src/utils/cli.rs

@@ -1,3 +1,4 @@
+use crate::consensus::proposer::current_time;
 use crate::proto::msg as message;
 use crate::proto::rpc::snapchain_service_client::SnapchainServiceClient;
 use crate::proto::{rpc, snapchain::Block};
@@ -41,9 +42,9 @@ pub async fn follow_blocks(
 
     loop {
         let msg = rpc::BlocksRequest {
-            shard_id: 0,
             start_block_number: i,
             stop_block_number: Some(i + FETCH_SIZE),
+            start_timestamp: Some(current_time()),
         };
 
         let request = tonic::Request::new(msg);