feat!: use gossipsub for consensus broadcasts

[mrnaveira]

Description

• New ConsensusGossipService in the Validator node, that listens for epoch events and subscribes to the appropriate gossip topics

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
• Peers should maintain enough connections (recommended 4/5) to local nodes and at least one connection to each foreign shard

How Has This Been Tested?

TBD

What process can a PR reviewer use to test or verify this change?

TBD

Breaking Changes

• None
• Requires data directory to be deleted
• Other - Multicast communications between VNs are now done via gossip

[github-actions]

Test Results (CI)

549 tests   - 18   549 ✅  - 18   1h 23m 20s ⏱️ - 19m 25s
 53 suites  -  9     0 💤 ± 0 
  1 files    -  1     0 ❌ ± 0 

Results for commit 622718a. ± Comparison against base commit 49c82f7.

This pull request removes 18 tests.

Scenario: Claim and transfer confidential assets via wallet daemon: tests/features/wallet_daemon.feature:89:3
Scenario: Claim base layer burn funds with wallet daemon: tests/features/claim_burn.feature:7:3
Scenario: Confidential transfer to unexisting account: tests/features/transfer.feature:162:3
Scenario: Counter template registration and invocation: tests/features/counter.feature:8:3
Scenario: Create account and transfer faucets via wallet daemon: tests/features/wallet_daemon.feature:8:3
Scenario: Create and mint account NFT: tests/features/wallet_daemon.feature:132:3
Scenario: Create resource and mint in one transaction: tests/features/nft.feature:82:3
Scenario: Double Claim base layer burn funds with wallet daemon. should fail: tests/features/claim_burn.feature:45:3
Scenario: Indexer GraphQL filtering and pagination of events: tests/features/indexer.feature:136:3
Scenario: Indexer GraphQL requests events over network substate indexing: tests/features/indexer.feature:103:3
…

♻️ This comment has been updated with latest results.

[sdbondi]

LGTM will test when we are out of draft.

For performance and potential issues with incorrect decoding I think we need to use the topic to determine which message type we have

[sdbondi]

Maybe we should inspect the topic prefix rather than trying to decode and failing - as this might be slow or even potentially erroneously succeed for the wrong message type (because protobuf works on optional fields and ignores unknown fields).

We could "register" channels for each topic or topic prefix.

e.g.

networking.gossipsub_subscribe("consensus-0-31", codec).await?;

or maybe register a channel for sending/receiving messages on the topic

let channel = message_channel::channel::<HotstuffMessage>::(100);
networking.gossipsub_subscribe("consensus-0-31", channel).await?;

Channel could be something like this

pub fn channel<T>(capacity: usize) -> (MessageChannel<T>, MessageChannel<T>) {
    let (ltr_tx, ltr_rx) = mpsc::channel(capacity);
    let (rtl_tx, rtl_rx) = mpsc::channel(capacity);
    (
        MessageChannel {
            sender: ltr_tx,
            receiver: rtl_rx,
        },
        MessageChannel {
            sender: rtl_tx,
            receiver: ltr_rx,
        },
    )
}

#[derive(Debug)]
pub struct MessageChannel<T> {
    sender: mpsc::Sender<T>,
    receiver: mpsc::Receiver<T>,
}

impl<T> MessageChannel<T> {
    pub async fn recv(&mut self) -> Option<T> {
        self.receiver.recv().await
    }

    pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
        self.receiver.try_recv()
    }

    pub async fn send(&self, value: T) -> Result<(), SendError<T>> {
        self.sender.send(value).await
    }

    pub fn try_send(&self, value: T) -> Result<(), TrySendError<T>> {
        self.sender.try_send(value)
    }

  pub fn split(self) -> (mpsc::Receiver<T>, mpsc::Sender<T>) {
        (self.receiver, self.sender)
   } 
}