feat!: use gossipsub for consensus broadcasts

applications/tari_validator_node/src/bootstrap.rs

@@ -106,8 +106,9 @@ use crate::{
     p2p::{
         create_tari_validator_node_rpc_service,
         services::{
+            consensus_gossip::{self, ConsensusGossipHandle},
             mempool::{self, MempoolHandle},
-            messaging::{ConsensusInboundMessaging, ConsensusOutboundMessaging, Gossip},
+            messaging::{ConsensusInboundMessaging, ConsensusOutboundMessaging},
         },
         NopLogger,
     },
@@ -137,7 +138,8 @@ pub async fn spawn_services(
 
     // Networking
     let (tx_consensus_messages, rx_consensus_messages) = mpsc::unbounded_channel();
-    let (tx_gossip_messages, rx_gossip_messages) = mpsc::unbounded_channel();
+    let (tx_transaction_gossip_messages, rx_transaction_gossip_messages) = mpsc::unbounded_channel();
+    let (tx_consensus_gossip_messages, rx_consensus_gossip_messages) = mpsc::unbounded_channel();
     let identity = identity::Keypair::sr25519_from_bytes(keypair.secret_key().as_bytes().to_vec()).map_err(|e| {
         ExitError::new(
             ExitCode::ConfigError,
@@ -161,7 +163,8 @@ pub async fn spawn_services(
         identity,
         MessagingMode::Enabled {
             tx_messages: tx_consensus_messages,
-            tx_gossip_messages,
+            tx_transaction_gossip_messages,
+            tx_consensus_gossip_messages,
         },
         tari_networking::Config {
             listener_port: config.validator_node.p2p.listener_port,
@@ -239,18 +242,27 @@ pub async fn spawn_services(
         per_log_cost: 1,
     };
 
+    // Consensus gossip
+    let (consensus_gossip_service, join_handle) = consensus_gossip::spawn(epoch_manager.clone(), networking.clone());
+    handles.push(join_handle);
+
     // Messaging
     let message_logger = NopLogger; // SqliteMessageLogger::new(config.validator_node.data_dir.join("message_log.sqlite"));
     let local_address = PeerAddress::from(keypair.public_key().clone());
     let (loopback_sender, loopback_receiver) = mpsc::unbounded_channel();
     let inbound_messaging = ConsensusInboundMessaging::new(
         local_address,
         rx_consensus_messages,
+        rx_consensus_gossip_messages,
         loopback_receiver,
         message_logger.clone(),
     );
-    let outbound_messaging =
-        ConsensusOutboundMessaging::new(loopback_sender, networking.clone(), message_logger.clone());
+    let outbound_messaging = ConsensusOutboundMessaging::new(
+        loopback_sender,
+        consensus_gossip_service.clone(),
+        networking.clone(),
+        message_logger.clone(),
+    );
 
     // Consensus
     let payload_processor = TariDanTransactionProcessor::new(config.network, template_manager.clone(), fee_table);
@@ -284,15 +296,14 @@ pub async fn spawn_services(
     .await;
     handles.push(consensus_join_handle);
 
-    let gossip = Gossip::new(networking.clone(), rx_gossip_messages);
-
     let (mempool, join_handle) = mempool::spawn(
         consensus_constants.num_preshards,
-        gossip,
         epoch_manager.clone(),
         create_mempool_transaction_validator(template_manager.clone()),
         state_store.clone(),
         consensus_handle.clone(),
+        networking.clone(),
+        rx_transaction_gossip_messages,
         #[cfg(feature = "metrics")]
         metrics_registry,
     );
@@ -363,6 +374,7 @@ pub async fn spawn_services(
         dry_run_transaction_processor,
         handles,
         validator_node_client_factory,
+        consensus_gossip_service,
     })
 }
 
@@ -414,6 +426,7 @@ pub struct Services {
     pub global_db: GlobalDb<SqliteGlobalDbAdapter<PeerAddress>>,
     pub dry_run_transaction_processor: DryRunTransactionProcessor,
     pub validator_node_client_factory: TariValidatorNodeRpcClientFactory,
+    pub consensus_gossip_service: ConsensusGossipHandle,
     pub state_store: SqliteStateStore<PeerAddress>,
 
     pub handles: Vec<JoinHandle<Result<(), anyhow::Error>>>,

applications/tari_validator_node/src/p2p/services/consensus_gossip/error.rs

@@ -0,0 +1,51 @@
+//  Copyright 2024. The Tari Project
+//
+//  Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+//  following conditions are met:
+//
+//  1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+//  disclaimer.
+//
+//  2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+//  following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+//  3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+//  products derived from this software without specific prior written permission.
+//
+//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+//  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+//  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+//  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+//  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+//  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+//  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+use tari_epoch_manager::EpochManagerError;
+use tari_networking::NetworkingError;
+use tokio::sync::{mpsc, oneshot};
+
+use super::ConsensusGossipRequest;
+
+#[derive(thiserror::Error, Debug)]
+pub enum ConsensusGossipError {
+    #[error("Invalid message: {0}")]
+    InvalidMessage(#[from] anyhow::Error),
+    #[error("Epoch Manager Error: {0}")]
+    EpochManagerError(#[from] EpochManagerError),
+    #[error("Internal service request cancelled")]
+    RequestCancelled,
+    #[error("Network error: {0}")]
+    NetworkingError(#[from] NetworkingError),
+}
+
+impl From<mpsc::error::SendError<ConsensusGossipRequest>> for ConsensusGossipError {
+    fn from(_: mpsc::error::SendError<ConsensusGossipRequest>) -> Self {
+        Self::RequestCancelled
+    }
+}
+
+impl From<oneshot::error::RecvError> for ConsensusGossipError {
+    fn from(_: oneshot::error::RecvError) -> Self {
+        Self::RequestCancelled
+    }
+}

applications/tari_validator_node/src/p2p/services/consensus_gossip/handle.rs

@@ -0,0 +1,71 @@
+//   Copyright 2024. The Tari Project
+//
+//   Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+//   following conditions are met:
+//
+//   1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+//   disclaimer.
+//
+//   2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+//   following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+//   3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+//   products derived from this software without specific prior written permission.
+//
+//   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+//   INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+//   DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+//   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+//   SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+//   WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+//   USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+use tari_consensus::messages::HotstuffMessage;
+use tari_dan_common_types::ShardGroup;
+use tokio::sync::{mpsc, oneshot};
+
+use super::ConsensusGossipError;
+
+pub enum ConsensusGossipRequest {
+    Multicast {
+        shard_group: ShardGroup,
+        message: HotstuffMessage,
+        reply: oneshot::Sender<Result<(), ConsensusGossipError>>,
+    },
+}
+
+#[derive(Debug)]
+pub struct ConsensusGossipHandle {
+    tx_consensus_request: mpsc::Sender<ConsensusGossipRequest>,
+}
+
+impl Clone for ConsensusGossipHandle {
+    fn clone(&self) -> Self {
+        ConsensusGossipHandle {
+            tx_consensus_request: self.tx_consensus_request.clone(),
+        }
+    }
+}
+
+impl ConsensusGossipHandle {
+    pub(super) fn new(tx_consensus_request: mpsc::Sender<ConsensusGossipRequest>) -> Self {
+        Self { tx_consensus_request }
+    }
+
+    pub async fn multicast(
+        &self,
+        shard_group: ShardGroup,
+        message: HotstuffMessage,
+    ) -> Result<(), ConsensusGossipError> {
+        let (tx, rx) = oneshot::channel();
+        self.tx_consensus_request
+            .send(ConsensusGossipRequest::Multicast {
+                shard_group,
+                message,
+                reply: tx,
+            })
+            .await?;
+
+        rx.await?
+    }
+}

applications/tari_validator_node/src/p2p/services/consensus_gossip/initializer.rs

@@ -0,0 +1,47 @@
+//  Copyright 2024. The Tari Project
+//
+//  Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+//  following conditions are met:
+//
+//  1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+//  disclaimer.
+//
+//  2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+//  following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+//  3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+//  products derived from this software without specific prior written permission.
+//
+//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+//  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+//  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+//  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+//  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+//  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+//  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+use log::*;
+use tari_dan_common_types::PeerAddress;
+use tari_dan_p2p::TariMessagingSpec;
+use tari_epoch_manager::base_layer::EpochManagerHandle;
+use tari_networking::NetworkingHandle;
+use tokio::{sync::mpsc, task, task::JoinHandle};
+
+use crate::p2p::services::consensus_gossip::{service::ConsensusGossipService, ConsensusGossipHandle};
+
+const LOG_TARGET: &str = "tari::dan::validator_node::mempool";
+
+pub fn spawn(
+    epoch_manager: EpochManagerHandle<PeerAddress>,
+    networking: NetworkingHandle<TariMessagingSpec>,
+) -> (ConsensusGossipHandle, JoinHandle<anyhow::Result<()>>) {
+    let (tx_consensus_request, rx_consensus_request) = mpsc::channel(10);
+
+    let consensus_gossip = ConsensusGossipService::new(rx_consensus_request, epoch_manager, networking);
+    let handle = ConsensusGossipHandle::new(tx_consensus_request);
+
+    let join_handle = task::spawn(consensus_gossip.run());
+    debug!(target: LOG_TARGET, "Spawning consensus gossip service (task: {:?})", join_handle);
+
+    (handle, join_handle)
+}

applications/tari_validator_node/src/p2p/services/consensus_gossip/mod.rs

@@ -0,0 +1,32 @@
+//  Copyright 2024. The Tari Project
+//
+//  Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+//  following conditions are met:
+//
+//  1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+//  disclaimer.
+//
+//  2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+//  following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+//  3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+//  products derived from this software without specific prior written permission.
+//
+//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+//  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+//  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+//  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+//  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+//  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+//  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+mod error;
+pub use error::*;
+
+mod handle;
+pub use handle::{ConsensusGossipHandle, ConsensusGossipRequest};
+
+mod initializer;
+pub use initializer::spawn;
+
+mod service;