From c842ed3ea8813eda69e64749014524b41b222be9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Miko=C5=82aj=20Florkiewicz?= <mikolaj@florkiewicz.me>
Date: Mon, 1 Jul 2024 14:52:17 +0200
Subject: [PATCH] feat(node): Add syncing window for header sync (#309)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Signed-off-by: Mikołaj Florkiewicz <mikolaj@florkiewicz.me>
Co-authored-by: Maciej Zwoliński <mac.zwolinski@gmail.com>
---
 node/src/p2p/header_session.rs | 50 +++++++++++++----
 node/src/store/utils.rs        | 99 +++++++++++++++++++++++-----------
 node/src/syncer.rs             | 74 +++++++++++++++++++++++++
 types/src/test_utils.rs        | 61 ++++++++++++++++-----
 4 files changed, 231 insertions(+), 53 deletions(-)

diff --git a/node/src/p2p/header_session.rs b/node/src/p2p/header_session.rs
index 1591f653..9e1ccb6d 100644
--- a/node/src/p2p/header_session.rs
+++ b/node/src/p2p/header_session.rs
@@ -9,7 +9,7 @@ use crate::p2p::{P2pCmd, P2pError};
 use crate::store::header_ranges::{HeaderRange, RangeLengthExt};
 
 const MAX_AMOUNT_PER_REQ: u64 = 64;
-const MAX_CONCURRENT_REQS: usize = 1;
+const MAX_CONCURRENT_REQS: usize = 8;
 
 type Result<T, E = P2pError> = std::result::Result<T, E>;
 
@@ -139,16 +139,18 @@ impl HeaderSession {
     }
 }
 
+/// take a next batch of up to `limit` headers from the front of the `range_to_fetch`
 fn take_next_batch(range_to_fetch: &mut Option<HeaderRange>, limit: u64) -> Option<HeaderRange> {
-    // calculate potential end before we modify range_to_fetch
+    // calculate potential end offset before we modify range_to_fetch
     let end_offset = limit.checked_sub(1)?;
 
     let to_fetch = range_to_fetch.take()?;
     if to_fetch.len() <= limit {
         Some(to_fetch)
     } else {
-        let _ = range_to_fetch.insert(*to_fetch.start() + limit..=*to_fetch.end());
-        Some(*to_fetch.start()..=*to_fetch.start() + end_offset)
+        // to_fetch.len() > limit, we shouldn't underflow here
+        let _ = range_to_fetch.insert(*to_fetch.start()..=*to_fetch.end() - limit);
+        Some(*to_fetch.end() - end_offset..=*to_fetch.end())
     }
 }
 
@@ -201,10 +203,10 @@ mod tests {
             result_tx.send(res).unwrap();
         });
 
-        for i in 0..8 {
+        for i in (0..8).rev() {
             let (height, amount, respond_to) =
                 p2p_mock.expect_header_request_for_height_cmd().await;
-            assert_eq!(height, 1 + 64 * i);
+            assert_eq!(height, 9 + 64 * i);
             assert_eq!(amount, 64);
             let start = (height - 1) as usize;
             let end = start + amount as usize;
@@ -212,7 +214,7 @@ mod tests {
         }
 
         let (height, amount, respond_to) = p2p_mock.expect_header_request_for_height_cmd().await;
-        assert_eq!(height, 513);
+        assert_eq!(height, 1);
         assert_eq!(amount, 8);
         let start = (height - 1) as usize;
         let end = start + amount as usize;
@@ -299,8 +301,38 @@ mod tests {
     fn take_next_batch_truncated_batch() {
         let mut range_to_fetch = Some(1..=10);
         let batch = take_next_batch(&mut range_to_fetch, 5);
-        assert_eq!(batch, Some(1..=5));
-        assert_eq!(range_to_fetch, Some(6..=10));
+        assert_eq!(batch, Some(6..=10));
+        assert_eq!(range_to_fetch, Some(1..=5));
+    }
+
+    #[test]
+    fn take_next_batch_truncated_calc() {
+        let mut range_to_fetch = Some(1..=512);
+
+        let batch = take_next_batch(&mut range_to_fetch, 64);
+        assert_eq!(batch, Some(449..=512));
+        assert_eq!(range_to_fetch, Some(1..=448));
+        let batch = take_next_batch(&mut range_to_fetch, 64);
+        assert_eq!(batch, Some(385..=448));
+        assert_eq!(range_to_fetch, Some(1..=384));
+        let batch = take_next_batch(&mut range_to_fetch, 64);
+        assert_eq!(batch, Some(321..=384));
+        assert_eq!(range_to_fetch, Some(1..=320));
+        let batch = take_next_batch(&mut range_to_fetch, 64);
+        assert_eq!(batch, Some(257..=320));
+        assert_eq!(range_to_fetch, Some(1..=256));
+        let batch = take_next_batch(&mut range_to_fetch, 64);
+        assert_eq!(batch, Some(193..=256));
+        assert_eq!(range_to_fetch, Some(1..=192));
+        let batch = take_next_batch(&mut range_to_fetch, 64);
+        assert_eq!(batch, Some(129..=192));
+        assert_eq!(range_to_fetch, Some(1..=128));
+        let batch = take_next_batch(&mut range_to_fetch, 64);
+        assert_eq!(batch, Some(65..=128));
+        assert_eq!(range_to_fetch, Some(1..=64));
+        let batch = take_next_batch(&mut range_to_fetch, 64);
+        assert_eq!(batch, Some(1..=64));
+        assert_eq!(range_to_fetch, None);
     }
 
     #[test]
diff --git a/node/src/store/utils.rs b/node/src/store/utils.rs
index a3513b9a..3090c493 100644
--- a/node/src/store/utils.rs
+++ b/node/src/store/utils.rs
@@ -3,7 +3,7 @@ use std::ops::RangeInclusive;
 use celestia_types::ExtendedHeader;
 
 use crate::executor::yield_now;
-use crate::store::header_ranges::HeaderRange;
+use crate::store::header_ranges::{HeaderRange, RangeLengthExt};
 use crate::store::{Result, StoreError};
 
 pub(crate) const VALIDATIONS_PER_YIELD: usize = 4;
@@ -13,34 +13,48 @@ pub(crate) const VALIDATIONS_PER_YIELD: usize = 4;
 pub(crate) fn calculate_range_to_fetch(
     head_height: u64,
     store_headers: &[RangeInclusive<u64>],
+    syncing_window_edge: Option<u64>,
     limit: u64,
+) -> HeaderRange {
+    let mut missing_range = get_most_recent_missing_range(head_height, store_headers);
+
+    // truncate to syncing window, if height is known
+    if let Some(window_edge) = syncing_window_edge {
+        if missing_range.start() < &window_edge {
+            missing_range = window_edge + 1..=*missing_range.end();
+        }
+    }
+
+    // truncate number of headers to limit
+    if missing_range.len() > limit {
+        let end = missing_range.end();
+        let start = end.saturating_sub(limit) + 1;
+        missing_range = start..=*end;
+    }
+
+    missing_range
+}
+
+fn get_most_recent_missing_range(
+    head_height: u64,
+    store_headers: &[RangeInclusive<u64>],
 ) -> HeaderRange {
     let mut store_headers_iter = store_headers.iter().rev();
 
     let Some(store_head_range) = store_headers_iter.next() else {
-        // empty store, just fetch from head
-        return head_height.saturating_sub(limit) + 1..=head_height;
+        // empty store, we're missing everything
+        return 1..=head_height;
     };
 
-    if store_head_range.end() != &head_height {
+    if store_head_range.end() < &head_height {
         // if we haven't caught up with network head, start from there
-        let fetch_start = u64::max(
-            store_head_range.end() + 1,
-            head_height.saturating_sub(limit) + 1,
-        );
-        return fetch_start..=head_height;
+        return store_head_range.end() + 1..=head_height;
     }
 
     // there exists a range contiguous with network head. inspect previous range end
     let penultimate_range_end = store_headers_iter.next().map(|r| *r.end()).unwrap_or(0);
 
-    let fetch_end = store_head_range.start().saturating_sub(1);
-    let fetch_start = u64::max(
-        penultimate_range_end + 1,
-        fetch_end.saturating_sub(limit) + 1,
-    );
-
-    fetch_start..=fetch_end
+    penultimate_range_end + 1..=store_head_range.start().saturating_sub(1)
 }
 
 pub(crate) fn try_consolidate_ranges(
@@ -129,35 +143,43 @@ mod tests {
         let head_height = 1024;
         let ranges = [256..=512];
 
-        let fetch_range = calculate_range_to_fetch(head_height, &ranges, 16);
+        let fetch_range = calculate_range_to_fetch(head_height, &ranges, None, 16);
         assert_eq!(fetch_range, 1009..=1024);
 
-        let fetch_range = calculate_range_to_fetch(head_height, &ranges, 511);
+        let fetch_range = calculate_range_to_fetch(head_height, &ranges, None, 511);
         assert_eq!(fetch_range, 514..=1024);
-        let fetch_range = calculate_range_to_fetch(head_height, &ranges, 512);
+        let fetch_range = calculate_range_to_fetch(head_height, &ranges, None, 512);
         assert_eq!(fetch_range, 513..=1024);
-        let fetch_range = calculate_range_to_fetch(head_height, &ranges, 513);
+        let fetch_range = calculate_range_to_fetch(head_height, &ranges, None, 513);
         assert_eq!(fetch_range, 513..=1024);
 
-        let fetch_range = calculate_range_to_fetch(head_height, &ranges, 1024);
+        let fetch_range = calculate_range_to_fetch(head_height, &ranges, None, 1024);
         assert_eq!(fetch_range, 513..=1024);
+        let fetch_range = calculate_range_to_fetch(head_height, &ranges, Some(900), 1024);
+        assert_eq!(fetch_range, 901..=1024);
     }
 
     #[test]
     fn calculate_range_to_fetch_empty_store() {
-        let fetch_range = calculate_range_to_fetch(1, &[], 100);
+        let fetch_range = calculate_range_to_fetch(1, &[], None, 100);
         assert_eq!(fetch_range, 1..=1);
 
-        let fetch_range = calculate_range_to_fetch(100, &[], 10);
+        let fetch_range = calculate_range_to_fetch(100, &[], None, 10);
         assert_eq!(fetch_range, 91..=100);
+
+        let fetch_range = calculate_range_to_fetch(100, &[], Some(75), 50);
+        assert_eq!(fetch_range, 76..=100);
     }
 
     #[test]
     fn calculate_range_to_fetch_fully_synced() {
-        let fetch_range = calculate_range_to_fetch(1, &[1..=1], 100);
+        let fetch_range = calculate_range_to_fetch(1, &[1..=1], None, 100);
         assert!(fetch_range.is_empty());
 
-        let fetch_range = calculate_range_to_fetch(100, &[1..=100], 10);
+        let fetch_range = calculate_range_to_fetch(100, &[1..=100], None, 10);
+        assert!(fetch_range.is_empty());
+
+        let fetch_range = calculate_range_to_fetch(100, &[1..=100], Some(100), 10);
         assert!(fetch_range.is_empty());
     }
 
@@ -165,24 +187,39 @@ mod tests {
     fn calculate_range_to_fetch_caught_up() {
         let head_height = 4000;
 
-        let fetch_range = calculate_range_to_fetch(head_height, &[3000..=4000], 500);
+        let fetch_range = calculate_range_to_fetch(head_height, &[3000..=4000], None, 500);
         assert_eq!(fetch_range, 2500..=2999);
-        let fetch_range = calculate_range_to_fetch(head_height, &[500..=1000, 3000..=4000], 500);
+        let fetch_range = calculate_range_to_fetch(head_height, &[3000..=4000], Some(2600), 500);
+        assert_eq!(fetch_range, 2601..=2999);
+        let fetch_range =
+            calculate_range_to_fetch(head_height, &[500..=1000, 3000..=4000], None, 500);
         assert_eq!(fetch_range, 2500..=2999);
-        let fetch_range = calculate_range_to_fetch(head_height, &[2500..=2800, 3000..=4000], 500);
+        let fetch_range =
+            calculate_range_to_fetch(head_height, &[2500..=2800, 3000..=4000], None, 500);
         assert_eq!(fetch_range, 2801..=2999);
-        let fetch_range = calculate_range_to_fetch(head_height, &[300..=4000], 500);
+        let fetch_range =
+            calculate_range_to_fetch(head_height, &[2500..=2800, 3000..=4000], Some(2000), 500);
+        assert_eq!(fetch_range, 2801..=2999);
+        let fetch_range = calculate_range_to_fetch(head_height, &[300..=4000], None, 500);
         assert_eq!(fetch_range, 1..=299);
+        let fetch_range = calculate_range_to_fetch(head_height, &[300..=4000], Some(2000), 500);
+        assert!(fetch_range.is_empty());
     }
 
     #[test]
     fn calculate_range_to_fetch_catching_up() {
         let head_height = 4000;
 
-        let fetch_range = calculate_range_to_fetch(head_height, &[2000..=3000], 500);
+        let fetch_range = calculate_range_to_fetch(head_height, &[2000..=3000], None, 500);
         assert_eq!(fetch_range, 3501..=4000);
-        let fetch_range = calculate_range_to_fetch(head_height, &[1..=2998, 3000..=3800], 500);
+        let fetch_range = calculate_range_to_fetch(head_height, &[2000..=3000], Some(3600), 500);
+        assert_eq!(fetch_range, 3601..=4000);
+        let fetch_range =
+            calculate_range_to_fetch(head_height, &[1..=2998, 3000..=3800], None, 500);
         assert_eq!(fetch_range, 3801..=4000);
+        let fetch_range =
+            calculate_range_to_fetch(head_height, &[1..=2998, 3000..=3800], Some(3900), 500);
+        assert_eq!(fetch_range, 3901..=4000);
     }
 
     #[test]
diff --git a/node/src/syncer.rs b/node/src/syncer.rs
index f9e8ce01..584f48d2 100644
--- a/node/src/syncer.rs
+++ b/node/src/syncer.rs
@@ -15,6 +15,7 @@ use std::time::Duration;
 
 use backoff::backoff::Backoff;
 use backoff::ExponentialBackoffBuilder;
+use celestia_tendermint::Time;
 use celestia_types::ExtendedHeader;
 use futures::FutureExt;
 use serde::{Deserialize, Serialize};
@@ -36,6 +37,7 @@ type Result<T, E = SyncerError> = std::result::Result<T, E>;
 
 const MAX_HEADERS_IN_BATCH: u64 = 512;
 const TRY_INIT_BACKOFF_MAX_INTERVAL: Duration = Duration::from_secs(60);
+const SYNCING_WINDOW: Duration = Duration::from_secs(30 * 24 * 60 * 60); // 30 days
 
 /// Representation of all the errors that can occur when interacting with the [`Syncer`].
 #[derive(Debug, thiserror::Error)]
@@ -180,6 +182,7 @@ where
     headers_tx: mpsc::Sender<(Result<Vec<ExtendedHeader>, P2pError>, Duration)>,
     headers_rx: mpsc::Receiver<(Result<Vec<ExtendedHeader>, P2pError>, Duration)>,
     ongoing_batch: Option<Ongoing>,
+    estimated_syncing_window_end: Option<u64>,
 }
 
 struct Ongoing {
@@ -210,6 +213,7 @@ where
             headers_tx,
             headers_rx,
             ongoing_batch: None,
+            estimated_syncing_window_end: None,
         })
     }
 
@@ -466,6 +470,7 @@ where
         let next_batch = calculate_range_to_fetch(
             subjective_head_height,
             store_ranges.as_ref(),
+            self.estimated_syncing_window_end,
             MAX_HEADERS_IN_BATCH,
         );
 
@@ -474,6 +479,14 @@ where
             return;
         }
 
+        // make sure we're inside the syncing window before we start
+        if let Ok(known_header) = self.store.get_by_height(next_batch.end() + 1).await {
+            if !in_syncing_window(&known_header) {
+                self.estimated_syncing_window_end = Some(known_header.height().value());
+                return;
+            }
+        }
+
         self.event_pub.send(NodeEvent::FetchingHeadersStarted {
             from_height: *next_batch.start(),
             to_height: *next_batch.end(),
@@ -540,6 +553,15 @@ where
     }
 }
 
+fn in_syncing_window(header: &ExtendedHeader) -> bool {
+    let syncing_window_start = Time::now().checked_sub(SYNCING_WINDOW).unwrap_or_else(|| {
+        warn!("underflow when computing syncing window start, defaulting to unix epoch");
+        Time::unix_epoch()
+    });
+
+    header.time().after(syncing_window_start)
+}
+
 async fn try_init<S>(p2p: &P2p, store: &S) -> Result<u64>
 where
     S: Store,
@@ -713,6 +735,58 @@ mod tests {
         p2p_mock.expect_no_cmd().await;
     }
 
+    #[async_test]
+    async fn syncing_window_edge() {
+        let month_and_day_ago = Duration::from_secs(31 * 24 * 60 * 60);
+        let mut gen = ExtendedHeaderGenerator::new();
+        gen.set_time((Time::now() - month_and_day_ago).expect("to not underflow"));
+        let mut headers = gen.next_many(1200);
+        gen.set_time(Time::now());
+        headers.append(&mut gen.next_many(2049 - 1200));
+
+        let (syncer, store, mut p2p_mock) = initialized_syncer(headers[2048].clone()).await;
+        assert_syncing(&syncer, &store, &[2049..=2049], 2049).await;
+
+        // Syncer requested the first batch ([1537..=2048])
+        handle_session_batch(
+            &mut p2p_mock,
+            &headers,
+            vec![
+                (1537, 64),
+                (1601, 64),
+                (1665, 64),
+                (1729, 64),
+                (1793, 64),
+                (1857, 64),
+                (1921, 64),
+                (1985, 64),
+            ],
+        )
+        .await;
+        assert_syncing(&syncer, &store, &[1537..=2049], 2049).await;
+
+        // Syncer requested the second batch ([1025, 1536]) hitting the syncing window
+        handle_session_batch(
+            &mut p2p_mock,
+            &headers,
+            vec![
+                (1025, 64),
+                (1089, 64),
+                (1153, 64),
+                (1217, 64),
+                (1281, 64),
+                (1345, 64),
+                (1409, 64),
+                (1473, 64),
+            ],
+        )
+        .await;
+        assert_syncing(&syncer, &store, &[1025..=2049], 2049).await;
+
+        // Syncer is fully synced and awaiting for events
+        p2p_mock.expect_no_cmd().await;
+    }
+
     #[async_test]
     async fn start_with_filled_store() {
         let events = EventChannel::new();
diff --git a/types/src/test_utils.rs b/types/src/test_utils.rs
index d65199e7..567023ce 100644
--- a/types/src/test_utils.rs
+++ b/types/src/test_utils.rs
@@ -1,4 +1,5 @@
 //! Utilities for writing tests.
+use std::time::Duration;
 
 use celestia_tendermint::block::header::{Header, Version};
 use celestia_tendermint::block::{parts, Commit, CommitSig};
@@ -23,6 +24,7 @@ pub struct ExtendedHeaderGenerator {
     chain_id: chain::Id,
     key: SigningKey,
     current_header: Option<ExtendedHeader>,
+    spoofed_time: Option<Time>,
 }
 
 impl ExtendedHeaderGenerator {
@@ -35,6 +37,7 @@ impl ExtendedHeaderGenerator {
             chain_id,
             key,
             current_header: None,
+            spoofed_time: None,
         }
     }
 
@@ -53,7 +56,13 @@ impl ExtendedHeaderGenerator {
         gen.current_header = if prev_height == 0 {
             None
         } else {
-            Some(generate_new(prev_height, &gen.chain_id, &gen.key, None))
+            Some(generate_new(
+                prev_height,
+                &gen.chain_id,
+                Time::now(),
+                &gen.key,
+                None,
+            ))
         };
 
         gen
@@ -69,9 +78,10 @@ impl ExtendedHeaderGenerator {
     /// ```
     #[allow(clippy::should_implement_trait)]
     pub fn next(&mut self) -> ExtendedHeader {
+        let time = self.get_and_increment_time();
         let header = match self.current_header {
-            Some(ref header) => generate_next(1, header, &self.key, None),
-            None => generate_new(GENESIS_HEIGHT, &self.chain_id, &self.key, None),
+            Some(ref header) => generate_next(1, header, time, &self.key, None),
+            None => generate_new(GENESIS_HEIGHT, &self.chain_id, time, &self.key, None),
         };
 
         self.current_header = Some(header.clone());
@@ -91,9 +101,10 @@ impl ExtendedHeaderGenerator {
     /// ```
     #[allow(clippy::should_implement_trait)]
     pub fn next_with_dah(&mut self, dah: DataAvailabilityHeader) -> ExtendedHeader {
+        let time = self.get_and_increment_time();
         let header = match self.current_header {
-            Some(ref header) => generate_next(1, header, &self.key, Some(dah)),
-            None => generate_new(GENESIS_HEIGHT, &self.chain_id, &self.key, Some(dah)),
+            Some(ref header) => generate_next(1, header, time, &self.key, Some(dah)),
+            None => generate_new(GENESIS_HEIGHT, &self.chain_id, time, &self.key, Some(dah)),
         };
 
         self.current_header = Some(header.clone());
@@ -128,7 +139,8 @@ impl ExtendedHeaderGenerator {
     ///
     /// This method does not change the state of `ExtendedHeaderGenerator`.
     pub fn next_of(&self, header: &ExtendedHeader) -> ExtendedHeader {
-        generate_next(1, header, &self.key, None)
+        let time = self.spoofed_time.unwrap_or_else(Time::now);
+        generate_next(1, header, time, &self.key, None)
     }
 
     /// Generates the next header of the provided header with the given [`DataAvailabilityHeader`].
@@ -155,7 +167,8 @@ impl ExtendedHeaderGenerator {
         header: &ExtendedHeader,
         dah: DataAvailabilityHeader,
     ) -> ExtendedHeader {
-        generate_next(1, header, &self.key, Some(dah))
+        let time = self.spoofed_time.unwrap_or_else(Time::now);
+        generate_next(1, header, time, &self.key, Some(dah))
     }
 
     /// Generates the next amount of headers of the provided header.
@@ -220,9 +233,11 @@ impl ExtendedHeaderGenerator {
             return;
         }
 
+        let time = self.get_and_increment_time();
+
         let header = match self.current_header {
-            Some(ref header) => generate_next(amount, header, &self.key, None),
-            None => generate_new(amount, &self.chain_id, &self.key, None),
+            Some(ref header) => generate_next(amount, header, time, &self.key, None),
+            None => generate_new(amount, &self.chain_id, time, &self.key, None),
         };
 
         self.current_header = Some(header.clone());
@@ -250,6 +265,24 @@ impl ExtendedHeaderGenerator {
     pub fn fork(&self) -> ExtendedHeaderGenerator {
         self.clone()
     }
+
+    /// Change header generator time. Headers generated from now on will have `time` as creation
+    /// time.
+    pub fn set_time(&mut self, time: Time) {
+        self.spoofed_time = Some(time);
+    }
+
+    // private function which also increments time, since we cannot have multiple headers on the
+    // exact same timestamp
+    fn get_and_increment_time(&mut self) -> Time {
+        let Some(spoofed_time) = self.spoofed_time.take() else {
+            return Time::now();
+        };
+
+        self.spoofed_time =
+            Some((spoofed_time + Duration::from_millis(1)).expect("not to overflow"));
+        self.spoofed_time.unwrap()
+    }
 }
 
 impl Default for ExtendedHeaderGenerator {
@@ -338,6 +371,7 @@ pub(crate) fn random_bytes(len: usize) -> Vec<u8> {
 fn generate_new(
     height: u64,
     chain_id: &chain::Id,
+    time: Time,
     signing_key: &SigningKey,
     dah: Option<DataAvailabilityHeader>,
 ) -> ExtendedHeader {
@@ -365,7 +399,7 @@ fn generate_new(
             },
             chain_id: chain_id.clone(),
             height: height.try_into().unwrap(),
-            time: Time::now(),
+            time,
             last_block_id,
             last_commit_hash: Hash::default_sha256(),
             data_hash: Hash::None,
@@ -391,7 +425,7 @@ fn generate_new(
             },
             signatures: vec![CommitSig::BlockIdFlagCommit {
                 validator_address,
-                timestamp: Time::now(),
+                timestamp: time,
                 signature: None,
             }],
         },
@@ -423,6 +457,7 @@ fn generate_new(
 fn generate_next(
     increment: u64,
     current: &ExtendedHeader,
+    time: Time,
     signing_key: &SigningKey,
     dah: Option<DataAvailabilityHeader>,
 ) -> ExtendedHeader {
@@ -449,7 +484,7 @@ fn generate_next(
             version: current.header.version,
             chain_id: current.header.chain_id.clone(),
             height,
-            time: Time::now(),
+            time,
             last_block_id,
             last_commit_hash: Hash::default_sha256(),
             data_hash: Hash::None,
@@ -475,7 +510,7 @@ fn generate_next(
             },
             signatures: vec![CommitSig::BlockIdFlagCommit {
                 validator_address,
-                timestamp: Time::now(),
+                timestamp: time,
                 signature: None,
             }],
         },