Expand timestamp liveness check

node/narwhal/src/helpers/proposal.rs

@@ -145,6 +145,7 @@ impl<N: Network> Proposal<N> {
         signer: Address<N>,
         signature: Signature<N>,
         timestamp: i64,
+        previous_timestamp: i64,
         committee: &Committee<N>,
     ) -> Result<()> {
         // Ensure the signer is in the committee.
@@ -161,7 +162,7 @@ impl<N: Network> Proposal<N> {
             bail!("Signature verification failed")
         }
         // Check the timestamp for liveness.
-        check_timestamp_for_liveness(timestamp)?;
+        check_timestamp_for_liveness(timestamp, previous_timestamp)?;
         // Insert the signature.
         self.signatures.insert(signature, timestamp);
         Ok(())

node/narwhal/src/helpers/storage.rs

@@ -305,6 +305,21 @@ impl<N: Network> Storage<N> {
         }
     }
 
+    /// Returns the median timestamp for the given `round`.
+    pub fn median_timestamp_for_round(&self, round: u64) -> i64 {
+        let previous_certificates = self.get_certificates_for_round(round);
+        if previous_certificates.is_empty() {
+            return 0;
+        }
+
+        let mut timestamps = previous_certificates
+            .into_iter()
+            .map(|batch_certificate| batch_certificate.median_timestamp())
+            .collect::<Vec<_>>();
+        timestamps.sort_unstable();
+        timestamps[timestamps.len() / 2]
+    }
+
     /// Checks the given `batch_header` for validity, returning the missing transmissions from storage.
     ///
     /// This method ensures the following invariants:
@@ -344,7 +359,7 @@ impl<N: Network> Storage<N> {
         }
 
         // Check the timestamp for liveness.
-        check_timestamp_for_liveness(batch_header.timestamp())?;
+        check_timestamp_for_liveness(batch_header.timestamp(), self.median_timestamp_for_round(round - 1))?;
 
         // Initialize a list for the missing transmissions from storage.
         let mut missing_transmissions = HashMap::new();
@@ -452,7 +467,7 @@ impl<N: Network> Storage<N> {
         // Iterate over the timestamps.
         for timestamp in certificate.timestamps() {
             // Check the timestamp for liveness.
-            check_timestamp_for_liveness(timestamp)?;
+            check_timestamp_for_liveness(timestamp, self.median_timestamp_for_round(round - 1))?;
         }
 
         // Retrieve the previous committee for the batch round.

node/narwhal/src/helpers/timestamp.rs

@@ -23,16 +23,19 @@ pub fn now() -> i64 {
 }
 
 /// Sanity checks the timestamp for liveness.
-pub fn check_timestamp_for_liveness(timestamp: i64) -> Result<()> {
+pub fn check_timestamp_for_liveness(timestamp: i64, previous_timestamp: i64) -> Result<()> {
     // Ensure the timestamp is within range.
     if timestamp > (now() + MAX_TIMESTAMP_DELTA_IN_SECS) {
         bail!("Timestamp {timestamp} is too far in the future")
     }
-    // TODO (howardwu): Ensure the timestamp is after the previous timestamp. (Needs Bullshark committee)
-    // // Ensure the timestamp is after the previous timestamp.
-    // if timestamp <= committee.previous_timestamp() {
-    //     bail!("Timestamp {timestamp} for the proposed batch must be after the previous round timestamp")
-    // }
+
+    // Ensure the timestamp is after the previous timestamp.
+    if timestamp < previous_timestamp {
+        bail!(
+            "Timestamp {timestamp} for the proposed batch must be after the previous round timestamp {previous_timestamp}"
+        )
+    }
+
     Ok(())
 }
 
@@ -52,15 +55,34 @@ mod prop_tests {
         (Just(now()), MAX_TIMESTAMP_DELTA_IN_SECS..).prop_map(|(now, delta)| now + delta).boxed()
     }
 
+    fn any_previous_timestamp() -> BoxedStrategy<i64> {
+        (0..now()).prop_map(|timestamp| timestamp).boxed()
+    }
+
     #[proptest]
-    fn test_check_timestamp_for_liveness(#[strategy(any_valid_timestamp())] timestamp: i64) {
-        let result = check_timestamp_for_liveness(timestamp);
+    fn test_check_timestamp_for_liveness(
+        #[strategy(any_valid_timestamp())] timestamp: i64,
+        #[strategy(any_previous_timestamp())] previous_timestamp: i64,
+    ) {
+        let result = check_timestamp_for_liveness(timestamp, previous_timestamp);
         assert!(result.is_ok());
     }
 
     #[proptest]
-    fn test_check_timestamp_for_liveness_too_far_in_future(#[strategy(any_invalid_timestamp())] timestamp: i64) {
-        let result = check_timestamp_for_liveness(timestamp);
+    fn test_check_timestamp_for_liveness_too_far_in_future(
+        #[strategy(any_invalid_timestamp())] timestamp: i64,
+        #[strategy(any_previous_timestamp())] previous_timestamp: i64,
+    ) {
+        let result = check_timestamp_for_liveness(timestamp, previous_timestamp);
+        assert!(result.is_err());
+    }
+
+    #[proptest]
+    fn test_check_timestamp_for_liveness_too_far_in_past(
+        #[strategy(any_previous_timestamp())] previous_timestamp: i64,
+        #[strategy(0..#previous_timestamp)] timestamp: i64,
+    ) {
+        let result = check_timestamp_for_liveness(timestamp, previous_timestamp);
         assert!(result.is_err());
     }
 }

node/narwhal/src/primary.rs

@@ -541,7 +541,13 @@ impl<N: Network> Primary<N> {
                     // Retrieve the address of the peer.
                     match self.gateway.resolver().get_address(peer_ip) {
                         // Add the signature to the batch.
-                        Some(signer) => proposal.add_signature(signer, signature, timestamp, &previous_committee)?,
+                        Some(signer) => proposal.add_signature(
+                            signer,
+                            signature,
+                            timestamp,
+                            self.storage.median_timestamp_for_round(self.current_round() - 1),
+                            &previous_committee,
+                        )?,
                         None => bail!("Signature is from a disconnected peer"),
                     };
                     info!("Received a batch signature for round {} from '{peer_ip}'", proposal.round());