fix: wait for proposal timestamps to enter future window

application/src/actor.rs

@@ -36,6 +36,33 @@ use summit_types::{Block, BlockAuxData, Digest, EngineClient};
 /// How long to wait before retrying check_payload when Reth returns SYNCING during certify.
 const CERTIFY_SYNCING_RETRY: Duration = Duration::from_millis(100);
 
+fn proposal_timestamp_wait(
+    now_millis: u64,
+    min_child_timestamp: u64,
+    allowed_timestamp_future_ms: u64,
+) -> Duration {
+    // Verifiers accept timestamps up to `now + allowed_timestamp_future_ms`.
+    // If the minimum monotonic child timestamp is beyond that bound, wait just
+    // long enough for it to enter the verifier's future window.
+    let max_allowed_timestamp = now_millis.saturating_add(allowed_timestamp_future_ms);
+    Duration::from_millis(min_child_timestamp.saturating_sub(max_allowed_timestamp))
+}
+
+fn select_proposal_timestamp(now_millis: u64, min_child_timestamp: u64) -> u64 {
+    // Once `min_child_timestamp` is inside the verifier future window, choose
+    // the later of local time and `parent.timestamp + 1` to preserve monotonicity.
+    now_millis.max(min_child_timestamp)
+}
+
+/// Whether the wait required to bring the monotonic child timestamp into the
+/// verifier future window exceeds the leader window. When true the block could
+/// not be notarized before the leader rotates anyway, so the proposal should be
+/// abandoned rather than waited out (which would also tie up the application
+/// actor, delaying verify/certify handling).
+fn proposal_wait_exceeds_leader_window(wait: Duration, leader_timeout: Duration) -> bool {
+    wait > leader_timeout
+}
+
 pub struct Actor<
     R: Storage + Metrics + Clock + Spawner + governor::clock::Clock + Rng,
     C: EngineClient,
@@ -51,6 +78,7 @@ pub struct Actor<
     genesis_hash: [u8; 32],
     epocher: ES,
     cancellation_token: CancellationToken,
+    leader_timeout: Duration,
     #[cfg(feature = "permissioned")]
     paused: Arc<AtomicBool>,
     _scheme_marker: PhantomData<S>,
@@ -82,6 +110,7 @@ impl<
                 genesis_hash,
                 epocher: cfg.epocher,
                 cancellation_token: cfg.cancellation_token,
+                leader_timeout: cfg.leader_timeout,
                 #[cfg(feature = "permissioned")]
                 paused: cfg.paused,
                 _scheme_marker: PhantomData,
@@ -648,11 +677,53 @@ impl<
         let pending_withdrawals = aux_data.withdrawals;
         let checkpoint_hash = aux_data.checkpoint_hash;
 
-        let mut current = self.context.current().epoch_millis();
-        if current <= parent_block.timestamp() {
-            current = parent_block.timestamp() + 1;
+        let min_child_timestamp = parent_block
+            .timestamp()
+            .checked_add(1)
+            .ok_or_else(|| anyhow!("parent timestamp overflow"))?;
+
+        // If the wait needed to bring `min_child_timestamp` into the verifier
+        // future window exceeds the leader window, abandon the proposal now. The
+        // block could not be notarized before the leader rotates, and sleeping
+        // that long inside the actor loop would also delay verify/certify
+        // handling. This abandons proposal creation only; it never bypasses the
+        // timestamp wait below — when we do proceed, we still build only once
+        // `parent.timestamp() + 1` is inside the verifier future window.
+        let initial_wait = proposal_timestamp_wait(
+            self.context.current().epoch_millis(),
+            min_child_timestamp,
+            aux_data.allowed_timestamp_future_ms,
+        );
+        if proposal_wait_exceeds_leader_window(initial_wait, self.leader_timeout) {
+            return Err(anyhow!(
+                "proposal timestamp wait {}ms exceeds leader timeout {}ms for round {round}; \
+                 abandoning proposal",
+                initial_wait.as_millis(),
+                self.leader_timeout.as_millis()
+            ));
         }
 
+        let current = loop {
+            // Do not ask the engine to build a payload until the timestamp we
+            // must use to be greater than the parent is also acceptable to peers.
+            let now_millis = self.context.current().epoch_millis();
+            let wait = proposal_timestamp_wait(
+                now_millis,
+                min_child_timestamp,
+                aux_data.allowed_timestamp_future_ms,
+            );
+            if wait.is_zero() {
+                break select_proposal_timestamp(now_millis, min_child_timestamp);
+            }
+            debug!(
+                ?round,
+                parent_height,
+                wait_ms = wait.as_millis(),
+                "waiting for proposal timestamp to enter verifier future window"
+            );
+            self.context.sleep(wait).await;
+        };
+
         // STEP 4: Start building block (Engine Client)
         debug!(
             ?round,
@@ -1664,4 +1735,66 @@ mod tests {
             header_timestamp
         );
     }
+
+    #[test]
+    fn proposal_timestamp_waits_until_parent_child_timestamp_is_in_future_window() {
+        let now_millis = 1_000_000;
+        let allowed_timestamp_future_ms = 1_000;
+        let min_child_timestamp = now_millis + allowed_timestamp_future_ms + 1;
+
+        let wait =
+            proposal_timestamp_wait(now_millis, min_child_timestamp, allowed_timestamp_future_ms);
+        assert_eq!(wait, Duration::from_millis(1));
+
+        let after_wait = now_millis + wait.as_millis() as u64;
+        let selected = select_proposal_timestamp(after_wait, min_child_timestamp);
+
+        assert_eq!(selected, min_child_timestamp);
+        assert!(selected <= after_wait + allowed_timestamp_future_ms);
+    }
+
+    #[test]
+    fn proposal_timestamp_does_not_wait_when_local_time_is_valid() {
+        let now_millis = 1_000_000;
+        let min_child_timestamp = now_millis - 10;
+        let allowed_timestamp_future_ms = 1_000;
+
+        let wait =
+            proposal_timestamp_wait(now_millis, min_child_timestamp, allowed_timestamp_future_ms);
+        assert!(wait.is_zero());
+        assert_eq!(
+            select_proposal_timestamp(now_millis, min_child_timestamp),
+            now_millis
+        );
+    }
+
+    #[test]
+    fn proposal_aborts_when_wait_exceeds_leader_window() {
+        // Parent timestamp so far ahead that the wait to enter the verifier
+        // window is longer than the leader window: the proposal must be abandoned.
+        let now_millis = 1_000_000;
+        let allowed_timestamp_future_ms = 1_000;
+        let leader_timeout = Duration::from_millis(2_000);
+        let min_child_timestamp = now_millis + allowed_timestamp_future_ms + 5_000;
+
+        let wait =
+            proposal_timestamp_wait(now_millis, min_child_timestamp, allowed_timestamp_future_ms);
+        assert_eq!(wait, Duration::from_millis(5_000));
+        assert!(proposal_wait_exceeds_leader_window(wait, leader_timeout));
+    }
+
+    #[test]
+    fn proposal_does_not_abort_when_wait_within_leader_window() {
+        // The required wait is inside the leader window, so the proposal proceeds
+        // (and waits) rather than aborting.
+        let now_millis = 1_000_000;
+        let allowed_timestamp_future_ms = 1_000;
+        let leader_timeout = Duration::from_millis(2_000);
+        let min_child_timestamp = now_millis + allowed_timestamp_future_ms + 1;
+
+        let wait =
+            proposal_timestamp_wait(now_millis, min_child_timestamp, allowed_timestamp_future_ms);
+        assert_eq!(wait, Duration::from_millis(1));
+        assert!(!proposal_wait_exceeds_leader_window(wait, leader_timeout));
+    }
 }

application/src/config.rs

@@ -3,6 +3,7 @@ use commonware_consensus::types::Epocher;
 use std::sync::Arc;
 #[cfg(feature = "permissioned")]
 use std::sync::atomic::AtomicBool;
+use std::time::Duration;
 use summit_types::EngineClient;
 use tokio_util::sync::CancellationToken;
 
@@ -23,6 +24,11 @@ pub struct ApplicationConfig<C: EngineClient, ES: Epocher> {
 
     pub cancellation_token: CancellationToken,
 
+    /// Consensus leader timeout. A proposal whose timestamp cannot enter the
+    /// verifier future window within this window is abandoned rather than waited
+    /// out, since it could not be notarized before the leader rotates.
+    pub leader_timeout: Duration,
+
     /// When true, the node will not participate in consensus
     /// (skip proposals, reject verifications, skip broadcasts).
     #[cfg(feature = "permissioned")]

node/src/engine.rs

@@ -214,6 +214,7 @@ where
                 genesis_hash: cfg.genesis_hash,
                 epocher: epocher.clone(),
                 cancellation_token: cancellation_token.clone(),
+                leader_timeout: cfg.leader_timeout,
                 #[cfg(feature = "permissioned")]
                 paused: paused.clone(),
             },