diff --git a/crates/espresso/node/src/api.rs b/crates/espresso/node/src/api.rs index 8e128424b48..b8242869cdd 100644 --- a/crates/espresso/node/src/api.rs +++ b/crates/espresso/node/src/api.rs @@ -610,7 +610,7 @@ impl, P: SequencerPersistence> RequestResponseDataSo duration, request_response_protocol.request_indefinitely::<_, _, _>( Request::VidShare(block_number, request_id), - RequestType::Broadcast, + RequestType::Batched, move |_request, response| { let avidm_param = avidm_param.clone(); let received_shares = received_shares_clone.clone(); @@ -5186,6 +5186,104 @@ mod test { Ok(()) } + /// Run entirely without the legacy consensus stack: with base version + /// `NEW_PROTOCOL_VERSION` it is torn down at startup, and the explicit + /// mid-run `shut_down_legacy` calls below — what the decide-count trigger + /// in `handle_events` does after `LEGACY_SHUTDOWN_DECIDE_COUNT` decides + /// on an upgraded network — must be harmless to repeat. The network has + /// to keep 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(::persistence_options) + .collect::>() + .try_into() + .unwrap(); + + let config = TestNetworkConfigBuilder::::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::::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 = + 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::>() + .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 diff --git a/crates/espresso/node/src/consensus_handle.rs b/crates/espresso/node/src/consensus_handle.rs index bcd2526581e..716c82532ed 100644 --- a/crates/espresso/node/src/consensus_handle.rs +++ b/crates/espresso/node/src/consensus_handle.rs @@ -491,6 +491,10 @@ where pub async fn start_consensus(&self) { self.activate().await; if self.is_new_proto_running() { + if self.upgrade_lock.upgrade().base >= versions::NEW_PROTOCOL_VERSION { + tracing::info!("base version starts at the cutover, shutting down legacy stack"); + self.shut_down_legacy().await; + } return; } self.legacy_handle @@ -518,11 +522,29 @@ where let _ = coordinator.await; } + /// Permanently tear down the legacy consensus stack: its tasks and the + /// legacy network. + /// + /// 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> { + pub(crate) async fn client_api(&self) -> Option> { if let NewProtocol::Running { client_api, .. } = &*self.new_proto.read() { return Some(client_api.clone()); } diff --git a/crates/espresso/node/src/context.rs b/crates/espresso/node/src/context.rs index 3e6096318da..cb576a1c969 100644 --- a/crates/espresso/node/src/context.rs +++ b/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( @@ -627,10 +635,23 @@ async fn handle_events( 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 diff --git a/crates/espresso/node/src/external_event_handler.rs b/crates/espresso/node/src/external_event_handler.rs index ae1016330f3..450a6e39b91 100644 --- a/crates/espresso/node/src/external_event_handler.rs +++ b/crates/espresso/node/src/external_event_handler.rs @@ -3,8 +3,10 @@ use std::sync::Arc; use anyhow::{Context, Result, bail}; -use espresso_types::{PubKey, SeqTypes}; +use either::Either; +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 +16,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 +44,27 @@ pub enum OutboundMessage { impl ExternalEventHandler { /// Creates a new `ExternalEventHandler` with the given network - pub async fn new>( + pub async fn new( tasks: &mut TaskList, request_response_sender: Sender, outbound_message_receiver: Receiver, + consensus_handle: Arc>>, network: Arc, public_key: PubKey, - ) -> Result { + ) -> Result + where + N: ConnectedNetwork, + 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 +97,35 @@ impl ExternalEventHandler { } /// The main loop for sending outbound messages. - async fn outbound_message_loop>( + async fn outbound_message_loop( mut receiver: Receiver, + consensus_handle: Arc>>, network: Arc, public_key: PubKey, - ) { + ) where + N: ConnectedNetwork, + P: SequencerPersistence, + { + let mut network = Either::Left(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 swap the legacy network out for its + // client API for good. + if network.is_left() + && let Some(client_api) = consensus_handle.client_api().await + { + network = Either::Right(client_api); + } + let network = match &network { + Either::Right(client_api) => { + Self::send_via_coordinator(client_api, message, public_key).await; + continue; + }, + Either::Left(network) => network, + }; + // Match the message type match message { OutboundMessage::Direct(message, recipient) => { @@ -109,7 +147,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 +186,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, + 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::>::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"); + } + }, + // All request-response traffic uses batched direct messages; the + // coordinator's network has no broadcast topic for external + // messages. + other => { + tracing::warn!( + message = ?other, + "dropping unsupported external message after cutover" + ); + }, + } + } } diff --git a/crates/espresso/node/src/lib.rs b/crates/espresso/node/src/lib.rs index 10790f47fee..513cfdc32b7 100644 --- a/crates/espresso/node/src/lib.rs +++ b/crates/espresso/node/src/lib.rs @@ -826,15 +826,20 @@ where info!("Libp2p network initialized"); - tracing::warn!("Waiting for at least one connection to be initialized"); - select! { - _ = cdn_network.wait_for_ready() => { - tracing::warn!("CDN connection initialized"); - }, - _ = p2p_network.wait_for_ready() => { - tracing::warn!("P2P connection initialized"); - }, - }; + // From `NEW_PROTOCOL_VERSION` on, all consensus traffic runs on + // cliquenet and the legacy stack is torn down at startup, so don't + // hold up boot waiting for legacy connectivity. + if genesis.base_version < versions::NEW_PROTOCOL_VERSION { + tracing::warn!("Waiting for at least one connection to be initialized"); + select! { + _ = cdn_network.wait_for_ready() => { + tracing::warn!("CDN connection initialized"); + }, + _ = p2p_network.wait_for_ready() => { + tracing::warn!("P2P connection initialized"); + }, + }; + } // Combine the CDN and P2P networks CombinedNetworks::new(cdn_network, p2p_network, Some(Duration::from_secs(1))) diff --git a/crates/hotshot/hotshot/src/traits/networking/combined_network.rs b/crates/hotshot/hotshot/src/traits/networking/combined_network.rs index 54838c429f0..4836ed26dab 100644 --- a/crates/hotshot/hotshot/src/traits/networking/combined_network.rs +++ b/crates/hotshot/hotshot/src/traits/networking/combined_network.rs @@ -349,6 +349,11 @@ impl ConnectedNetwork 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 diff --git a/crates/hotshot/hotshot/src/types/handle.rs b/crates/hotshot/hotshot/src/types/handle.rs index 4e811ed6a47..0a623e1a302 100644 --- a/crates/hotshot/hotshot/src/types/handle.rs +++ b/crates/hotshot/hotshot/src/types/handle.rs @@ -273,7 +273,14 @@ impl + '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. + 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 +291,13 @@ impl + 'static> SystemContextHandl .await .inspect_err(|err| tracing::error!("Failed to send shutdown event: {err}")); - tracing::error!("Shutting down the network!"); + tracing::info!(target: "announce", "Shutting down the network!"); self.hotshot.network.shut_down().await; - tracing::error!("Shutting down network tasks!"); + tracing::info!(target: "announce", "Shutting down network tasks!"); self.network_registry.shutdown().await; - tracing::error!("Shutting down consensus!"); + tracing::info!(target: "announce", "Shutting down consensus!"); self.consensus_registry.shutdown().await; }