Merge branch 'main' into ma/binary-upgrade-test-scenarios

Author: sveitser

.config/nextest.toml

@@ -68,7 +68,6 @@ report-name = "nextest"
 filter = """
 test(test_success_with_async_delay_with_epochs) |
 test(test_success_with_async_delay_2_with_epochs) |
-test(test_vid2_success) |
 test(test_combined_network_half_dc) |
 test(test_staggered_restart_first_block) |
 test(/test_staggered_restart$/) |

Cargo.lock

@@ -1,10 +1,4 @@
-use std::{
-    collections::HashMap,
-    sync::{
-        Arc,
-        atomic::{AtomicBool, Ordering},
-    },
-};
+use std::{collections::HashMap, sync::Arc};
 
 use async_broadcast::{InactiveReceiver, Sender};
 use async_lock::RwLock;
@@ -13,8 +7,12 @@ use futures::{FutureExt, StreamExt, future::BoxFuture, stream::BoxStream};
 use hotshot::{traits::NodeImplementation, types::SystemContextHandle};
 use hotshot_new_protocol::{
     client::ClientApi,
-    consensus::ConsensusOutput,
+    consensus::{ConsensusOutput, PreCutoverSeed},
     coordinator::{Coordinator, CoordinatorOutput, error::Severity},
+    cutover::{
+        CutoverGate, extract_pre_cutover_seed, forward_legacy_epoch_changes,
+        forward_legacy_timeout_votes,
+    },
     network::Network,
     state::UpdateLeaf,
     storage::NewProtocolStorage,
@@ -78,15 +76,21 @@ where
                 cert2: cert2.clone(),
             })
         },
-        ConsensusOutput::ViewChanged(view, _epoch) => {
-            Some(CoordinatorEvent::ViewChanged { view_number: *view })
-        },
         ConsensusOutput::ProposalValidated { proposal, sender } => {
             Some(CoordinatorEvent::QuorumProposal {
                 proposal: proposal.clone(),
                 sender: sender.clone(),
             })
         },
+        ConsensusOutput::BlockPayloadReconstructed {
+            view,
+            header,
+            payload,
+        } => Some(CoordinatorEvent::BlockPayloadReconstructed {
+            view: *view,
+            header: header.clone(),
+            payload: payload.clone(),
+        }),
         _ => None,
     }
 }
@@ -116,7 +120,7 @@ pub struct ConsensusHandle<T: NodeType, I: NodeImplementation<T>> {
     client_api: ClientApi<T>,
     coordinator_task: AbortOnDropHandle<()>,
     epoch_height: u64,
-    new_protocol_active: AtomicBool,
+    cutover_gate: CutoverGate,
     legacy_event_rx: InactiveReceiver<Event<T>>,
     event_rx: InactiveReceiver<CoordinatorEvent<T>>,
 }
@@ -147,17 +151,32 @@ where
         let coordinator_task =
             AbortOnDropHandle::new(spawn(run_coordinator(coordinator, event_tx)));
 
+        spawn(forward_legacy_timeout_votes(
+            legacy_event_rx.clone(),
+            client_api.clone(),
+        ));
+        spawn(forward_legacy_epoch_changes(
+            legacy_event_rx.clone(),
+            client_api.clone(),
+            epoch_height,
+        ));
+
         Self {
             legacy_handle,
             client_api,
             coordinator_task,
             epoch_height,
-            new_protocol_active: AtomicBool::new(false),
+            cutover_gate: CutoverGate::new(),
             legacy_event_rx,
             event_rx: event_rx.deactivate(),
         }
     }
 
+    pub async fn extract_pre_cutover_seed(&self) -> Option<PreCutoverSeed<T>> {
+        let legacy = self.legacy_handle.read().await;
+        extract_pre_cutover_seed(&legacy).await
+    }
+
     pub fn legacy_consensus(&self) -> Arc<RwLock<SystemContextHandle<T, I>>> {
         self.legacy_handle.clone()
     }
@@ -166,7 +185,11 @@ where
         &self.client_api
     }
 
-    async fn new_protocol_at(&self, view: ViewNumber) -> bool {
+    /// Whether `view` is at or past the new-protocol upgrade boundary,
+    /// according to the legacy upgrade lock. This is a stateless version
+    /// check — use it when routing per-view queries like `state(view)`.
+    /// For "should we route to the coordinator?" use [`cutover_active`](Self::cutover_active).
+    async fn at_or_past_cutover(&self, view: ViewNumber) -> bool {
         self.legacy_handle
             .read()
             .await
@@ -176,16 +199,16 @@ where
             >= NEW_PROTOCOL_VERSION
     }
 
-    async fn new_protocol(&self) -> bool {
-        if self.new_protocol_active.load(Ordering::Relaxed) {
+    /// Whether the cutover has happened — the gate has latched on this
+    /// node. Stateful: the first call after legacy crosses the cutover
+    /// view triggers seed extraction + dispatch. Use this for "should
+    /// we route to the coordinator?" decisions.
+    pub async fn cutover_active(&self) -> bool {
+        if self.cutover_gate.is_active() {
             return true;
         }
-        let view = self.legacy_handle.read().await.cur_view().await;
-        let active = self.new_protocol_at(view).await;
-        if active {
-            self.new_protocol_active.store(true, Ordering::Relaxed);
-        }
-        active
+        let legacy = self.legacy_handle.read().await;
+        self.cutover_gate.check(&legacy, &self.client_api).await
     }
 
     pub fn event_stream(&self) -> BoxStream<'static, CoordinatorEvent<T>> {
@@ -200,7 +223,7 @@ where
     }
 
     pub async fn current_view(&self) -> ViewNumber {
-        if self.new_protocol().await {
+        if self.cutover_active().await {
             return self
                 .client_api
                 .current_view()
@@ -211,7 +234,7 @@ where
     }
 
     pub async fn decided_leaf(&self) -> Leaf2<T> {
-        if self.new_protocol().await {
+        if self.cutover_active().await {
             return self
                 .client_api
                 .decided_leaf()
@@ -222,7 +245,7 @@ where
     }
 
     pub async fn decided_state(&self) -> Option<Arc<T::ValidatedState>> {
-        if self.new_protocol().await {
+        if self.cutover_active().await {
             return self
                 .client_api
                 .decided_state()
@@ -233,7 +256,7 @@ where
     }
 
     pub async fn state(&self, view: ViewNumber) -> Option<Arc<T::ValidatedState>> {
-        if self.new_protocol_at(view).await {
+        if self.at_or_past_cutover(view).await {
             return self
                 .client_api
                 .state(view)
@@ -244,7 +267,7 @@ where
     }
 
     pub async fn state_and_delta(&self, view: ViewNumber) -> StateAndDelta<T> {
-        if self.new_protocol_at(view).await {
+        if self.at_or_past_cutover(view).await {
             return self
                 .client_api
                 .state_and_delta(view)
@@ -262,7 +285,7 @@ where
     }
 
     pub async fn undecided_leaves(&self) -> Vec<Leaf2<T>> {
-        if self.new_protocol().await {
+        if self.cutover_active().await {
             return self
                 .client_api
                 .undecided_leaves()
@@ -280,7 +303,7 @@ where
     }
 
     pub async fn current_epoch(&self) -> Option<EpochNumber> {
-        if self.new_protocol().await {
+        if self.cutover_active().await {
             return self
                 .client_api
                 .current_epoch()
@@ -291,7 +314,7 @@ where
     }
 
     pub async fn epoch_height(&self) -> u64 {
-        if self.new_protocol().await {
+        if self.cutover_active().await {
             return self.epoch_height;
         }
         self.legacy_handle.read().await.epoch_height
@@ -372,7 +395,7 @@ where
     ) -> anyhow::Result<
         BoxFuture<'static, anyhow::Result<SignedProposal<T, QuorumProposalWrapper<T>>>>,
     > {
-        if self.new_protocol_at(view).await {
+        if self.at_or_past_cutover(view).await {
             let client_api = self.client_api.clone();
             return Ok(async move {
                 client_api
@@ -395,7 +418,7 @@ where
 
     pub async fn submit_transaction(&self, tx: T::Transaction) -> anyhow::Result<()> {
         let view = self.current_view().await;
-        if self.new_protocol_at(view).await {
+        if self.at_or_past_cutover(view).await {
             return self
                 .client_api
                 .submit_transaction(tx)
@@ -417,7 +440,7 @@ where
         delta: Option<Arc<<T::ValidatedState as ValidatedState<T>>::Delta>>,
     ) -> anyhow::Result<()> {
         let view = leaf.view_number();
-        if self.new_protocol_at(view).await {
+        if self.at_or_past_cutover(view).await {
             return self
                 .client_api
                 .update_leaf(UpdateLeaf {
@@ -441,7 +464,7 @@ where
     }
 
     pub async fn start_consensus(&self) {
-        if self.new_protocol().await {
+        if self.cutover_active().await {
             // New protocol consensus is already running via the coordinator task.
             // Don't start legacy HotShot consensus tasks.
             return;

crates/espresso/node/src/consensus_handle.rs

@@ -565,6 +565,8 @@ async fn handle_events<N, P>(
                 {
                     tracing::warn!("Failed to handle legacy external message: {:?}", err);
                 }
+                // Check if we're ready to start the new protocol
+                consensus_handle.cutover_active().await;
             },
             CoordinatorEvent::ExternalMessageReceived { data, .. } => {
                 if let Err(err) = external_event_handler.handle_event(data).await {

crates/espresso/node/src/context.rs

@@ -37,6 +37,8 @@ time = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
+url = { workspace = true }
+vec1 = { workspace = true }
 versions = { workspace = true }
 
 [lints]

crates/hotshot/new-protocol/Cargo.toml

@@ -152,7 +152,7 @@ impl<T: NodeType> BlockBuilder<T> {
             // sleep so the coordinator's event queue doesnot overflow
             // because if there are no transactions then the block production is way too fast
             if buffer.is_empty() {
-                sleep(Duration::from_millis(100)).await;
+                sleep(Duration::from_millis(500)).await;
             }
             let (hashes, txs): (Vec<_>, Vec<_>) = buffer.into_iter().unzip();
             let manifest = DedupManifest {

crates/hotshot/new-protocol/src/block.rs

@@ -5,12 +5,16 @@ use committable::Commitment;
 use hotshot_types::{
     data::{EpochNumber, Leaf2, ViewNumber},
     message::Proposal as SignedProposal,
+    simple_vote::TimeoutVote2,
     traits::{leaf_fetcher_network::LeafFetcherNetwork, node_implementation::NodeType},
     utils::StateAndDelta,
 };
 use tokio::sync::{mpsc, oneshot};
 
-use crate::{coordinator::error::CoordinatorError, message::Proposal, state::UpdateLeaf};
+use crate::{
+    consensus::PreCutoverSeed, coordinator::error::CoordinatorError, message::Proposal,
+    state::UpdateLeaf,
+};
 
 #[derive(Clone)]
 pub struct ClientApi<T: NodeType> {
@@ -112,6 +116,28 @@ impl<T: NodeType> ClientApi<T> {
         .await?
     }
 
+    /// Forward a legacy `TimeoutVote2` into the new-protocol timeout collectors.
+    pub async fn submit_timeout_vote(&self, vote: TimeoutVote2<T>) -> Result<(), QueryError> {
+        let (respond, rx) = oneshot::channel();
+        self.call(ClientRequest::SubmitTimeoutVote { vote, respond }, rx)
+            .await
+    }
+
+    /// Refresh the coordinator network's peer set for `epoch`.
+    pub async fn bump_network_epoch(&self, epoch: EpochNumber) -> Result<(), QueryError> {
+        let (respond, rx) = oneshot::channel();
+        self.call(ClientRequest::BumpNetworkEpoch { epoch, respond }, rx)
+            .await
+    }
+
+    /// Bridge legacy state into the coordinator at the cutover.
+    /// Idempotent at the consensus layer.
+    pub async fn seed_pre_cutover(&self, seed: PreCutoverSeed<T>) -> Result<(), QueryError> {
+        let (respond, rx) = oneshot::channel();
+        self.call(ClientRequest::SeedPreCutover { seed, respond }, rx)
+            .await
+    }
+
     async fn call<A>(
         &self,
         request: ClientRequest<T>,
@@ -192,6 +218,18 @@ pub(crate) enum ClientRequest<T: NodeType> {
         recipient: T::SignatureKey,
         respond: oneshot::Sender<Result<(), QueryError>>,
     },
+    SeedPreCutover {
+        seed: PreCutoverSeed<T>,
+        respond: oneshot::Sender<()>,
+    },
+    SubmitTimeoutVote {
+        vote: TimeoutVote2<T>,
+        respond: oneshot::Sender<()>,
+    },
+    BumpNetworkEpoch {
+        epoch: EpochNumber,
+        respond: oneshot::Sender<()>,
+    },
 }
 
 #[derive(Debug, thiserror::Error)]