-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat!: use gossipsub for consensus broadcasts (#1156)
Description
---
* Both the mempool and HotStuff send broadcast messages using the same
gossipsub infrastructure.
* The networking layer allows to specify a map
(`tx_gossip_messages_by_topic`) that maps topic prefixes to a channel.
On startup we register a channel for mempool (topic prefix
`transactions`) and another for consensus (topic prefix `consensus`).
Then, on gossip message received, the networking layer simply relays it
to the appropriate channel based on the topic prefix. This design allows
us to easily use gossip for other purposes in the future.
* Message encoding and decoding is done in each service (mempool or
consensus) separately.
* New `ConsensusGossipService` in the Validator node, that listens for
epoch events and subscribes to the appropriate gossip topics. It also
does message encoding/decoding.
* Updated the `MempoolGossip` module to adapt it to the new gossip
design, by implementing message encoding/decoding and to receive the
messages from networking.
* The consensus layer is independent of how communication is done.
* `ConsensusInboundMessaging` now also listens to consensus messages
coming from the new `ConsensusGossipService` .
* `ConsensusOutboundMessaging` uses the new `ConsensusGossipService` for
broadcasting.
* The `OutboundMessaging` trait for the `multicast` function now expects
a `ShardGroup` instead of a committee.
Motivation and Context
---
Hotstuff and cerberus are message based protocols. Currently we
implement a message protocol that requires nodes to connect to every
other node in the local shard. For cross shard messaging, we implement a
strategy that limits the number of messages sent but relies on multiple
connections per peer across shards.
We want to leverage libp2p's gossipsub for all consensus broadcasts to
local/foreign shards.
* Each shard subscribes to their topic `consensus-{start}-{end}`
(`start` and `end` are the start/end shards in the `ShardGroup` type,
similar to the mempool service)
* Ambient peer discovery required by gossipsub is already performed by
the Tari-implemented peer sync protocol and L1 registrations
How Has This Been Tested?
---
Manually by starting a local network using `tari_spawn`, performing
transactions and inspecting the logs.
What process can a PR reviewer use to test or verify this change?
---
See previous section
Breaking Changes
---
- [ ] None
- [ ] Requires data directory to be deleted
- [x] Other - Requires network reset, as multicast communications
between VNs are now done via gossip- Loading branch information
Showing
34 changed files
with
774 additions
and
235 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
51 changes: 51 additions & 0 deletions
51
applications/tari_validator_node/src/p2p/services/consensus_gossip/error.rs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } | ||
| } |
83 changes: 83 additions & 0 deletions
83
applications/tari_validator_node/src/p2p/services/consensus_gossip/handle.rs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| // 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>>, | ||
| }, | ||
| GetLocalShardGroup { | ||
| reply: oneshot::Sender<Result<Option<ShardGroup>, 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? | ||
| } | ||
|
|
||
| pub async fn get_local_shard_group(&self) -> Result<Option<ShardGroup>, ConsensusGossipError> { | ||
| let (tx, rx) = oneshot::channel(); | ||
| self.tx_consensus_request | ||
| .send(ConsensusGossipRequest::GetLocalShardGroup { reply: tx }) | ||
| .await?; | ||
|
|
||
| rx.await? | ||
| } | ||
| } |
60 changes: 60 additions & 0 deletions
60
applications/tari_validator_node/src/p2p/services/consensus_gossip/initializer.rs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| // 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 libp2p::{gossipsub, PeerId}; | ||
| use log::*; | ||
| use tari_dan_common_types::PeerAddress; | ||
| use tari_dan_p2p::{proto, 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>, | ||
| rx_gossip: mpsc::UnboundedReceiver<(PeerId, gossipsub::Message)>, | ||
| ) -> ( | ||
| ConsensusGossipHandle, | ||
| JoinHandle<anyhow::Result<()>>, | ||
| mpsc::Receiver<(PeerId, proto::consensus::HotStuffMessage)>, | ||
| ) { | ||
| let (tx_consensus_request, rx_consensus_request) = mpsc::channel(10); | ||
| let (tx_consensus_gossip, rx_consensus_gossip) = mpsc::channel(10); | ||
|
|
||
| let consensus_gossip = ConsensusGossipService::new( | ||
| rx_consensus_request, | ||
| epoch_manager, | ||
| networking, | ||
| rx_gossip, | ||
| tx_consensus_gossip, | ||
| ); | ||
| 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, rx_consensus_gossip) | ||
| } |
33 changes: 33 additions & 0 deletions
33
applications/tari_validator_node/src/p2p/services/consensus_gossip/mod.rs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| // 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; | ||
| pub use service::TOPIC_PREFIX; |
Oops, something went wrong.