feat(new-protocol): tear down legacy stack after cutover decides

crates/espresso/node/src/api.rs

@@ -5186,6 +5186,102 @@ mod test {
         Ok(())
     }
 
+    /// Tear down the legacy consensus stack (tasks + network) on every node
+    /// mid-run — what the decide-count trigger in `handle_events` does after
+    /// `LEGACY_SHUTDOWN_DECIDE_COUNT` new-protocol decides — and verify the
+    /// network keeps deciding across epoch boundaries: DRB computations on
+    /// the shared membership coordinator must survive the teardown.
+    #[test_log::test(tokio::test(flavor = "multi_thread"))]
+    async fn test_new_protocol_survives_legacy_shutdown() -> anyhow::Result<()> {
+        const EPOCH_HEIGHT: u64 = 20;
+        const NUM_NODES: usize = 5;
+        const SHUTDOWN_HEIGHT: u64 = 10;
+        const TARGET_BLOCK_HEIGHT: u64 = 50;
+
+        const NEW_PROTOCOL: Upgrade = Upgrade::trivial(NEW_PROTOCOL_VERSION);
+
+        let network_config = TestConfigBuilder::default()
+            .epoch_height(EPOCH_HEIGHT)
+            .epoch_start_block(0)
+            .build();
+
+        let api_port = reserve_tcp_port().expect("No ports free for query service");
+
+        let storage = join_all((0..NUM_NODES).map(|_| SqlDataSource::create_storage())).await;
+        let persistence: [_; NUM_NODES] = storage
+            .iter()
+            .map(<SqlDataSource as TestableSequencerDataSource>::persistence_options)
+            .collect::<Vec<_>>()
+            .try_into()
+            .unwrap();
+
+        let config = TestNetworkConfigBuilder::<NUM_NODES, _, _>::with_num_nodes()
+            .api_config(SqlDataSource::options(
+                &storage[0],
+                Options::with_port(api_port),
+            ))
+            .network_config(network_config)
+            .persistences(persistence)
+            .catchups(std::array::from_fn(|_| {
+                StatePeers::<SequencerApiVersion>::from_urls(
+                    vec![format!("http://localhost:{api_port}").parse().unwrap()],
+                    Default::default(),
+                    Duration::from_secs(2),
+                    &NoMetrics,
+                )
+            }))
+            .pos_hook(
+                DelegationConfig::MultipleDelegators,
+                StakeTableContractVersion::V3,
+                NEW_PROTOCOL,
+            )
+            .await
+            .unwrap()
+            .build();
+
+        let network = TestNetwork::new(config, NEW_PROTOCOL).await;
+
+        let client: Client<ServerError, SequencerApiVersion> =
+            Client::new(format!("http://localhost:{api_port}").parse().unwrap());
+        client.connect(Some(Duration::from_secs(30))).await;
+
+        let mut leaves = client
+            .socket("availability/stream/leaves/0")
+            .subscribe::<LeafQueryData<SeqTypes>>()
+            .await
+            .expect("subscribe to leaf stream");
+
+        // Let the new protocol decide a few blocks first.
+        let mut height = 0;
+        while height < SHUTDOWN_HEIGHT {
+            let leaf = leaves
+                .next()
+                .await
+                .expect("leaf stream ended early")
+                .expect("leaf stream yielded an error");
+            height = leaf.header().height();
+        }
+
+        // Tear down the legacy stack on every node.
+        network.server.consensus_handle().shut_down_legacy().await;
+        for peer in &network.peers {
+            peer.consensus_handle().shut_down_legacy().await;
+        }
+
+        // The chain must keep growing across the epoch boundaries at 20 and
+        // 40 purely on the new protocol.
+        while height < TARGET_BLOCK_HEIGHT {
+            let leaf = leaves
+                .next()
+                .await
+                .expect("leaf stream ended early")
+                .expect("leaf stream yielded an error");
+            height = leaf.header().height();
+        }
+
+        Ok(())
+    }
+
     /// Full-application epoch-boundary committee change: a validator sends a
     /// real `deregisterValidator` transaction to the StakeTable contract on
     /// L1 mid-run, drops out of the consensus committee at the epoch boundary

crates/espresso/node/src/consensus_handle.rs

@@ -518,11 +518,32 @@ where
         let _ = coordinator.await;
     }
 
+    /// Permanently tear down the legacy consensus stack — its tasks and the
+    /// legacy network — once the new protocol has decided enough leaves that
+    /// nothing can still depend on it.
+    ///
+    /// The in-memory legacy consensus state stays readable for pre-cutover
+    /// queries, and in-flight DRB computations on the shared membership
+    /// coordinator keep running (the new protocol needs them for upcoming
+    /// epochs).
+    pub async fn shut_down_legacy(&self) {
+        for t in &self.tasks {
+            t.abort()
+        }
+        self.legacy_handle
+            .write()
+            .await
+            .shut_down_tasks_and_network()
+            .await;
+    }
+
     fn is_new_proto_running(&self) -> bool {
         matches!(*self.new_proto.read(), NewProtocol::Running { .. })
     }
 
-    async fn client_api(&self) -> Option<ClientApi<T>> {
+    /// The coordinator's client API, if the new protocol is running (trying
+    /// to activate it first if it isn't yet).
+    pub(crate) async fn client_api(&self) -> Option<ClientApi<T>> {
         if let NewProtocol::Running { client_api, .. } = &*self.new_proto.read() {
             return Some(client_api.clone());
         }

crates/espresso/node/src/context.rs

@@ -314,6 +314,7 @@ where
             &mut tasks,
             request_response_sender,
             outbound_message_receiver,
+            consensus_handle.clone(),
             network,
             pub_key,
         )
@@ -610,6 +611,13 @@ impl DecideProcessorMetrics {
     }
 }
 
+/// How many new-protocol decides to wait for before tearing down the legacy
+/// consensus stack (tasks + network). Once the coordinator has decided even a
+/// single leaf the cutover boundary is final and all consensus traffic runs
+/// on the coordinator's network; the margin only gives slightly-lagging peers
+/// a window to finish crossing the boundary with legacy help.
+const LEGACY_SHUTDOWN_DECIDE_COUNT: u64 = 100;
+
 #[tracing::instrument(skip_all, fields(node_id))]
 #[allow(clippy::too_many_arguments)]
 async fn handle_events<N, P, C>(
@@ -627,10 +635,23 @@ async fn handle_events<N, P, C>(
     P: SequencerPersistence,
     C: PersistenceEventConsumer + 'static,
 {
+    let mut new_protocol_decides: u64 = 0;
+
     while let Some(event) = events.next().await {
         tracing::debug!(node_id, ?event, "consensus event");
 
         match &event {
+            CoordinatorEvent::NewDecide { .. } => {
+                new_protocol_decides += 1;
+                if new_protocol_decides == LEGACY_SHUTDOWN_DECIDE_COUNT {
+                    tracing::info!(
+                        node_id,
+                        "new protocol is live, shutting down legacy consensus and network"
+                    );
+                    let handle = consensus_handle.clone();
+                    spawn(async move { handle.shut_down_legacy().await });
+                }
+            },
             CoordinatorEvent::LegacyEvent(hotshot_event) => {
                 if let hotshot_types::event::EventType::ExternalMessageReceived { ref data, .. } =
                     hotshot_event.event

crates/espresso/node/src/external_event_handler.rs

@@ -3,8 +3,9 @@
 use std::sync::Arc;
 
 use anyhow::{Context, Result, bail};
-use espresso_types::{PubKey, SeqTypes};
+use espresso_types::{PubKey, SeqTypes, v0::traits::SequencerPersistence};
 use hotshot::types::Message;
+use hotshot_new_protocol::client::ClientApi;
 use hotshot_types::{
     message::MessageKind,
     traits::network::{BroadcastDelay, ConnectedNetwork, Topic, ViewMessage},
@@ -14,7 +15,10 @@ use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc::{Receiver, Sender, error::TrySendError};
 use vbs::{BinarySerializer, bincode_serializer::BincodeSerializer, version::StaticVersion};
 
-use crate::context::TaskList;
+use crate::{
+    consensus_handle::ConsensusHandle,
+    context::{ConsensusNode, TaskList},
+};
 
 /// An external message that can be sent to or received from a node
 #[derive(Debug, Serialize, Deserialize, Clone)]
@@ -39,17 +43,27 @@ pub enum OutboundMessage {
 
 impl ExternalEventHandler {
     /// Creates a new `ExternalEventHandler` with the given network
-    pub async fn new<N: ConnectedNetwork<PubKey>>(
+    pub async fn new<N, P>(
         tasks: &mut TaskList,
         request_response_sender: Sender<Bytes>,
         outbound_message_receiver: Receiver<OutboundMessage>,
+        consensus_handle: Arc<ConsensusHandle<SeqTypes, ConsensusNode<N, P>>>,
         network: Arc<N>,
         public_key: PubKey,
-    ) -> Result<Self> {
+    ) -> Result<Self>
+    where
+        N: ConnectedNetwork<PubKey>,
+        P: SequencerPersistence,
+    {
         // Spawn the outbound message handling loop
         tasks.spawn(
             "ExternalEventHandler",
-            Self::outbound_message_loop(outbound_message_receiver, network, public_key),
+            Self::outbound_message_loop(
+                outbound_message_receiver,
+                consensus_handle,
+                network,
+                public_key,
+            ),
         );
 
         Ok(Self {
@@ -82,12 +96,32 @@ impl ExternalEventHandler {
     }
 
     /// The main loop for sending outbound messages.
-    async fn outbound_message_loop<N: ConnectedNetwork<PubKey>>(
+    async fn outbound_message_loop<N, P>(
         mut receiver: Receiver<OutboundMessage>,
+        consensus_handle: Arc<ConsensusHandle<SeqTypes, ConsensusNode<N, P>>>,
         network: Arc<N>,
         public_key: PubKey,
-    ) {
+    ) where
+        N: ConnectedNetwork<PubKey>,
+        P: SequencerPersistence,
+    {
+        // Dropped as soon as the coordinator takes over, so the legacy
+        // network can be freed after the cutover.
+        let mut network = Some(network);
+
         while let Some(message) = receiver.recv().await {
+            // Once the coordinator is running it owns the only live network;
+            // route external messages through it. The coordinator never
+            // stops once started, so this switch is permanent.
+            if let Some(client_api) = consensus_handle.client_api().await {
+                network = None;
+                Self::send_via_coordinator(&client_api, message, public_key).await;
+                continue;
+            }
+            let Some(network) = &network else {
+                continue;
+            };
+
             // Match the message type
             match message {
                 OutboundMessage::Direct(message, recipient) => {
@@ -109,7 +143,7 @@ impl ExternalEventHandler {
                         };
 
                     // Send the message to the recipient
-                    let network = Arc::clone(&network);
+                    let network = Arc::clone(network);
                     tokio::spawn(async move {
                         if let Err(err) =
                             network.direct_message(view, message_bytes, recipient).await
@@ -148,4 +182,48 @@ impl ExternalEventHandler {
             }
         }
     }
+
+    /// Send an outbound message through the coordinator's network.
+    ///
+    /// The coordinator's network sends external payloads over the wire
+    /// verbatim, so they must be self-framing: wrap the payload in the same
+    /// versioned `Message` envelope the legacy path uses, which the receiving
+    /// side's fallback decoder recognizes and unwraps.
+    async fn send_via_coordinator(
+        client_api: &ClientApi<SeqTypes>,
+        message: OutboundMessage,
+        public_key: PubKey,
+    ) {
+        match message {
+            OutboundMessage::Direct(kind @ MessageKind::External(_), recipient) => {
+                let message = Message {
+                    sender: public_key,
+                    kind,
+                };
+                let message_bytes =
+                    match BincodeSerializer::<StaticVersion<0, 0>>::serialize(&message) {
+                        Ok(message_bytes) => message_bytes,
+                        Err(err) => {
+                            tracing::warn!("Failed to serialize direct message: {}", err);
+                            return;
+                        },
+                    };
+                if let Err(err) = client_api
+                    .send_external_message(message_bytes, recipient)
+                    .await
+                {
+                    tracing::warn!(%err, "failed to send external message via coordinator");
+                }
+            },
+            // Nothing sends these today: all catchup requests are batched
+            // direct messages, and the coordinator's network has no broadcast
+            // topic for external messages.
+            other => {
+                tracing::warn!(
+                    message = ?other,
+                    "dropping unsupported external message after cutover"
+                );
+            },
+        }
+    }
 }

crates/hotshot/hotshot/src/traits/networking/combined_network.rs

@@ -349,6 +349,11 @@ impl<TYPES: NodeType> ConnectedNetwork<TYPES::SignatureKey> for CombinedNetworks
     {
         let closure = async move {
             join!(self.primary().shut_down(), self.secondary().shut_down());
+            // The network object may be kept alive long after shutdown (e.g.
+            // the legacy stack after the new-protocol cutover); release the
+            // cached message hashes and per-view channels it still pins.
+            self.message_deduplication_cache.write().drop_contents();
+            self.delayed_tasks_channels.write().await.clear();
         };
         boxed_sync(closure)
     }
@@ -545,6 +550,13 @@ impl MessageDeduplicationCache {
         }
     }
 
+    /// Drop all cached hashes and shrink to the minimum footprint. Only
+    /// sensible after `shut_down`, when no more messages will be processed.
+    fn drop_contents(&mut self) {
+        self.primary_message_cache = LruCache::new(NonZeroUsize::MIN);
+        self.secondary_message_cache = LruCache::new(NonZeroUsize::MIN);
+    }
+
     /// Determine if a message is unique between two sources
     fn is_unique(&mut self, message: &[u8], from_primary: bool) -> bool {
         // Calculate the hash of the message

crates/hotshot/hotshot/src/types/handle.rs

@@ -273,7 +273,18 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES> + 'static> SystemContextHandl
     /// Shut down the inner hotshot and wait until all background threads are closed.
     pub async fn shut_down(&mut self) {
         self.membership_coordinator.cancel_all_drb();
+        self.shut_down_tasks_and_network().await;
+    }
 
+    /// Shut down all consensus and network tasks and the network itself, but
+    /// leave shared state alone: in-flight DRB computations on the membership
+    /// coordinator keep running, and the in-memory consensus state stays
+    /// readable.
+    ///
+    /// This is the teardown used at the new-protocol cutover, where the
+    /// `EpochMembershipCoordinator` is shared with the new-protocol
+    /// coordinator and must survive the legacy stack.
+    pub async fn shut_down_tasks_and_network(&mut self) {
         // this is required because `SystemContextHandle` holds an inactive receiver and
         // `broadcast_direct` below can wait indefinitely
         self.internal_event_stream.0.set_await_active(false);
@@ -284,13 +295,13 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES> + 'static> SystemContextHandl
             .await
             .inspect_err(|err| tracing::error!("Failed to send shutdown event: {err}"));
 
-        tracing::error!("Shutting down the network!");
+        tracing::info!("Shutting down the network!");
         self.hotshot.network.shut_down().await;
 
-        tracing::error!("Shutting down network tasks!");
+        tracing::info!("Shutting down network tasks!");
         self.network_registry.shutdown().await;
 
-        tracing::error!("Shutting down consensus!");
+        tracing::info!("Shutting down consensus!");
         self.consensus_registry.shutdown().await;
     }