#560 take two, now that #560 has been reverted (#561)

* Clear upons upon round, not block

* Cache the proposal for a round

* Rebase onto develop, which reverted this PR, and re-apply this PR

* Set participation upon participation instead of constantly recalculating

* Cache message instances

* Add missing txn commit

Identified by @akildemir.

* Correct clippy lint identified upon rebase

* Fix tendermint chain sync (#581)

* fix p2p Reqres protocol

* stabilize tributary chain sync

* fix pr comments

---------

Co-authored-by: akildemir <34187742+akildemir@users.noreply.github.com>

common/db/src/parity_db.rs

@@ -4,6 +4,7 @@ pub use ::parity_db::{Options, Db as ParityDb};
 
 use crate::*;
 
+#[must_use]
 pub struct Transaction<'a>(&'a Arc<ParityDb>, Vec<(u8, Vec<u8>, Option<Vec<u8>>)>);
 
 impl Get for Transaction<'_> {

common/db/src/rocks.rs

@@ -7,6 +7,7 @@ use rocksdb::{
 
 use crate::*;
 
+#[must_use]
 pub struct Transaction<'a, T: ThreadMode>(
   RocksTransaction<'a, OptimisticTransactionDB<T>>,
   &'a OptimisticTransactionDB<T>,

coordinator/src/p2p.rs

@@ -9,7 +9,7 @@ use std::{
 use async_trait::async_trait;
 use rand_core::{RngCore, OsRng};
 
-use scale::Encode;
+use scale::{Decode, Encode};
 use borsh::{BorshSerialize, BorshDeserialize};
 use serai_client::{primitives::NetworkId, validator_sets::primitives::ValidatorSet, Serai};
 
@@ -29,7 +29,7 @@ use libp2p::{
   noise, yamux,
   request_response::{
     Codec as RrCodecTrait, Message as RrMessage, Event as RrEvent, Config as RrConfig,
-    Behaviour as RrBehavior,
+    Behaviour as RrBehavior, ProtocolSupport,
   },
   gossipsub::{
     IdentTopic, FastMessageId, MessageId, MessageAuthenticity, ValidationMode, ConfigBuilder,
@@ -45,9 +45,20 @@ pub(crate) use tributary::{ReadWrite, P2p as TributaryP2p};
 use crate::{Transaction, Block, Tributary, ActiveTributary, TributaryEvent};
 
 // Block size limit + 1 KB of space for signatures/metadata
-const MAX_LIBP2P_MESSAGE_SIZE: usize = tributary::BLOCK_SIZE_LIMIT + 1024;
+const MAX_LIBP2P_GOSSIP_MESSAGE_SIZE: usize = tributary::BLOCK_SIZE_LIMIT + 1024;
+
+const MAX_LIBP2P_REQRES_MESSAGE_SIZE: usize =
+  (tributary::BLOCK_SIZE_LIMIT * BLOCKS_PER_BATCH) + 1024;
+
 const LIBP2P_TOPIC: &str = "serai-coordinator";
 
+// Amount of blocks in a minute
+// We can't use tendermint::TARGET_BLOCK_TIME here to calculate this since that is a u32.
+const BLOCKS_PER_MINUTE: usize = 10;
+
+// Maximum amount of blocks to send in a batch
+const BLOCKS_PER_BATCH: usize = BLOCKS_PER_MINUTE + 1;
+
 #[derive(Clone, Copy, PartialEq, Eq, Hash, Debug, BorshSerialize, BorshDeserialize)]
 pub struct CosignedBlock {
   pub network: NetworkId,
@@ -173,6 +184,18 @@ pub struct Message<P: P2p> {
   pub msg: Vec<u8>,
 }
 
+#[derive(Clone, Debug, Encode, Decode)]
+pub struct BlockCommit {
+  pub block: Vec<u8>,
+  pub commit: Vec<u8>,
+}
+
+#[derive(Clone, Debug, Encode, Decode)]
+pub struct HeartbeatBatch {
+  pub blocks: Vec<BlockCommit>,
+  pub timestamp: u64,
+}
+
 #[async_trait]
 pub trait P2p: Send + Sync + Clone + fmt::Debug + TributaryP2p {
   type Id: Send + Sync + Clone + Copy + fmt::Debug;
@@ -228,8 +251,8 @@ impl RrCodecTrait for RrCodec {
     let mut len = [0; 4];
     io.read_exact(&mut len).await?;
     let len = usize::try_from(u32::from_le_bytes(len)).expect("not a 32-bit platform?");
-    if len > MAX_LIBP2P_MESSAGE_SIZE {
-      Err(io::Error::other("request length exceeded MAX_LIBP2P_MESSAGE_SIZE"))?;
+    if len > MAX_LIBP2P_REQRES_MESSAGE_SIZE {
+      Err(io::Error::other("request length exceeded MAX_LIBP2P_REQRES_MESSAGE_SIZE"))?;
     }
     // This may be a non-trivial allocation easily causable
     // While we could chunk the read, meaning we only perform the allocation as bandwidth is used,
@@ -297,7 +320,7 @@ impl LibP2p {
     let throwaway_key_pair = Keypair::generate_ed25519();
 
     let behavior = Behavior {
-      reqres: { RrBehavior::new([], RrConfig::default()) },
+      reqres: { RrBehavior::new([("/coordinator", ProtocolSupport::Full)], RrConfig::default()) },
       gossipsub: {
         let heartbeat_interval = tributary::tendermint::LATENCY_TIME / 2;
         let heartbeats_per_block =
@@ -308,7 +331,7 @@ impl LibP2p {
           .heartbeat_interval(Duration::from_millis(heartbeat_interval.into()))
           .history_length(heartbeats_per_block * 2)
           .history_gossip(heartbeats_per_block)
-          .max_transmit_size(MAX_LIBP2P_MESSAGE_SIZE)
+          .max_transmit_size(MAX_LIBP2P_GOSSIP_MESSAGE_SIZE)
           // We send KeepAlive after 80s
           .idle_timeout(Duration::from_secs(85))
           .validation_mode(ValidationMode::Strict)
@@ -348,10 +371,11 @@ impl LibP2p {
       .with_tcp(TcpConfig::default().nodelay(true), noise::Config::new, || {
         let mut config = yamux::Config::default();
         // 1 MiB default + max message size
-        config.set_max_buffer_size((1024 * 1024) + MAX_LIBP2P_MESSAGE_SIZE);
+        config.set_max_buffer_size((1024 * 1024) + MAX_LIBP2P_REQRES_MESSAGE_SIZE);
         // 256 KiB default + max message size
-        config
-          .set_receive_window_size(((256 * 1024) + MAX_LIBP2P_MESSAGE_SIZE).try_into().unwrap());
+        config.set_receive_window_size(
+          ((256 * 1024) + MAX_LIBP2P_REQRES_MESSAGE_SIZE).try_into().unwrap(),
+        );
         config
       })
       .unwrap()
@@ -868,7 +892,7 @@ pub async fn handle_p2p_task<D: Db, P: P2p>(
               let p2p = p2p.clone();
               async move {
                 loop {
-                  let Some(mut msg) = recv.recv().await else {
+                  let Some(msg) = recv.recv().await else {
                     // Channel closure happens when the tributary retires
                     break;
                   };
@@ -913,35 +937,54 @@ pub async fn handle_p2p_task<D: Db, P: P2p>(
                           latest = next;
                         }
                         if to_send.len() > 3 {
-                          for next in to_send {
-                            let mut res = reader.block(&next).unwrap().serialize();
-                            res.extend(reader.commit(&next).unwrap());
-                            // Also include the timestamp used within the Heartbeat
-                            res.extend(&msg.msg[32 .. 40]);
-                            p2p.send(msg.sender, ReqResMessageKind::Block(genesis), res).await;
+                          // prepare the batch to sends
+                          let mut blocks = vec![];
+                          for (i, next) in to_send.iter().enumerate() {
+                            if i >= BLOCKS_PER_BATCH {
+                              break;
                             }
+
+                            blocks.push(BlockCommit {
+                              block: reader.block(next).unwrap().serialize(),
+                              commit: reader.commit(next).unwrap(),
+                            });
+                          }
+                          let batch = HeartbeatBatch { blocks, timestamp: msg_time };
+
+                          p2p
+                            .send(msg.sender, ReqResMessageKind::Block(genesis), batch.encode())
+                            .await;
                         }
                       });
                     }
 
                     P2pMessageKind::ReqRes(ReqResMessageKind::Block(msg_genesis)) => {
                       assert_eq!(msg_genesis, genesis);
-                      let mut msg_ref: &[u8] = msg.msg.as_ref();
-                      let Ok(block) = Block::<Transaction>::read(&mut msg_ref) else {
+                      // decode the batch
+                      let Ok(batch) = HeartbeatBatch::decode(&mut msg.msg.as_ref()) else {
+                        log::error!(
+                          "received HeartBeatBatch message with an invalidly serialized batch"
+                        );
+                        continue;
+                      };
+
+                      // sync blocks
+                      for bc in batch.blocks {
+                        // TODO: why do we use ReadWrite instead of Encode/Decode for blocks?
+                        // Should we use the same for batches so we can read both at the same time?
+                        let Ok(block) = Block::<Transaction>::read(&mut bc.block.as_slice()) else {
                           log::error!("received block message with an invalidly serialized block");
                           continue;
                         };
-                      // Get just the commit
-                      msg.msg.drain(.. (msg.msg.len() - msg_ref.len()));
-                      msg.msg.drain((msg.msg.len() - 8) ..);
 
-                      let res = tributary.tributary.sync_block(block, msg.msg).await;
+                        let res = tributary.tributary.sync_block(block, bc.commit).await;
                         log::debug!(
                           "received block from {:?}, sync_block returned {}",
                           msg.sender,
                           res
                         );
                       }
+                    }
 
                     P2pMessageKind::Gossip(GossipMessageKind::Tributary(msg_genesis)) => {
                       assert_eq!(msg_genesis, genesis);

coordinator/tributary/src/lib.rs

@@ -1,5 +1,5 @@
 use core::{marker::PhantomData, fmt::Debug};
-use std::{sync::Arc, io, collections::VecDeque};
+use std::{sync::Arc, io};
 
 use async_trait::async_trait;
 
@@ -154,14 +154,6 @@ pub struct Tributary<D: Db, T: TransactionTrait, P: P2p> {
   synced_block: Arc<RwLock<SyncedBlockSender<TendermintNetwork<D, T, P>>>>,
   synced_block_result: Arc<RwLock<SyncedBlockResultReceiver>>,
   messages: Arc<RwLock<MessageSender<TendermintNetwork<D, T, P>>>>,
-
-  p2p_meta_task_handle: Arc<tokio::task::AbortHandle>,
-}
-
-impl<D: Db, T: TransactionTrait, P: P2p> Drop for Tributary<D, T, P> {
-  fn drop(&mut self) {
-    self.p2p_meta_task_handle.abort();
-  }
 }
 
 impl<D: Db, T: TransactionTrait, P: P2p> Tributary<D, T, P> {
@@ -193,28 +185,7 @@ impl<D: Db, T: TransactionTrait, P: P2p> Tributary<D, T, P> {
     );
     let blockchain = Arc::new(RwLock::new(blockchain));
 
-    let to_rebroadcast = Arc::new(RwLock::new(VecDeque::new()));
-    // Actively rebroadcast consensus messages to ensure they aren't prematurely dropped from the
-    // P2P layer
-    let p2p_meta_task_handle = Arc::new(
-      tokio::spawn({
-        let to_rebroadcast = to_rebroadcast.clone();
-        let p2p = p2p.clone();
-        async move {
-          loop {
-            let to_rebroadcast = to_rebroadcast.read().await.clone();
-            for msg in to_rebroadcast {
-              p2p.broadcast(genesis, msg).await;
-            }
-            tokio::time::sleep(core::time::Duration::from_secs(60)).await;
-          }
-        }
-      })
-      .abort_handle(),
-    );
-
-    let network =
-      TendermintNetwork { genesis, signer, validators, blockchain, to_rebroadcast, p2p };
+    let network = TendermintNetwork { genesis, signer, validators, blockchain, p2p };
 
     let TendermintHandle { synced_block, synced_block_result, messages, machine } =
       TendermintMachine::new(
@@ -235,7 +206,6 @@ impl<D: Db, T: TransactionTrait, P: P2p> Tributary<D, T, P> {
       synced_block: Arc::new(RwLock::new(synced_block)),
       synced_block_result: Arc::new(RwLock::new(synced_block_result)),
       messages: Arc::new(RwLock::new(messages)),
-      p2p_meta_task_handle,
     })
   }
 

coordinator/tributary/src/tendermint/mod.rs

@@ -1,8 +1,5 @@
 use core::ops::Deref;
-use std::{
-  sync::Arc,
-  collections::{VecDeque, HashMap},
-};
+use std::{sync::Arc, collections::HashMap};
 
 use async_trait::async_trait;
 
@@ -270,8 +267,6 @@ pub struct TendermintNetwork<D: Db, T: TransactionTrait, P: P2p> {
   pub(crate) validators: Arc<Validators>,
   pub(crate) blockchain: Arc<RwLock<Blockchain<D, T>>>,
 
-  pub(crate) to_rebroadcast: Arc<RwLock<VecDeque<Vec<u8>>>>,
-
   pub(crate) p2p: P,
 }
 
@@ -308,26 +303,6 @@ impl<D: Db, T: TransactionTrait, P: P2p> Network for TendermintNetwork<D, T, P>
   async fn broadcast(&mut self, msg: SignedMessageFor<Self>) {
     let mut to_broadcast = vec![TENDERMINT_MESSAGE];
     to_broadcast.extend(msg.encode());
-
-    // Since we're broadcasting a Tendermint message, set it to be re-broadcasted every second
-    // until the block it's trying to build is complete
-    // If the P2P layer drops a message before all nodes obtained access, or a node had an
-    // intermittent failure, this will ensure reconcilliation
-    // This is atrocious if there's no content-based deduplication protocol for messages actively
-    // being gossiped
-    // LibP2p, as used by Serai, is configured to content-based deduplicate
-    {
-      let mut to_rebroadcast_lock = self.to_rebroadcast.write().await;
-      to_rebroadcast_lock.push_back(to_broadcast.clone());
-      // We should have, ideally, 3 * validators messages within a round
-      // Therefore, this should keep the most recent 2-rounds
-      // TODO: This isn't perfect. Each participant should just rebroadcast their latest round of
-      // messages
-      while to_rebroadcast_lock.len() > (6 * self.validators.weights.len()) {
-        to_rebroadcast_lock.pop_front();
-      }
-    }
-
     self.p2p.broadcast(self.genesis, to_broadcast).await
   }
 
@@ -366,7 +341,7 @@ impl<D: Db, T: TransactionTrait, P: P2p> Network for TendermintNetwork<D, T, P>
     }
   }
 
-  async fn validate(&mut self, block: &Self::Block) -> Result<(), TendermintBlockError> {
+  async fn validate(&self, block: &Self::Block) -> Result<(), TendermintBlockError> {
     let block =
       Block::read::<&[u8]>(&mut block.0.as_ref()).map_err(|_| TendermintBlockError::Fatal)?;
     self
@@ -428,9 +403,6 @@ impl<D: Db, T: TransactionTrait, P: P2p> Network for TendermintNetwork<D, T, P>
       }
     }
 
-    // Since we've added a valid block, clear to_rebroadcast
-    *self.to_rebroadcast.write().await = VecDeque::new();
-
     Some(TendermintBlock(
       self.blockchain.write().await.build_block::<Self>(&self.signature_scheme()).serialize(),
     ))

coordinator/tributary/tendermint/src/block.rs

@@ -3,7 +3,6 @@ use std::{
   collections::{HashSet, HashMap},
 };
 
-use parity_scale_codec::Encode;
 use serai_db::{Get, DbTxn, Db};
 
 use crate::{
@@ -20,7 +19,7 @@ pub(crate) struct BlockData<N: Network> {
 
   pub(crate) number: BlockNumber,
   pub(crate) validator_id: Option<N::ValidatorId>,
-  pub(crate) proposal: Option<N::Block>,
+  pub(crate) our_proposal: Option<N::Block>,
 
   pub(crate) log: MessageLog<N>,
   pub(crate) slashes: HashSet<N::ValidatorId>,
@@ -43,7 +42,7 @@ impl<N: Network> BlockData<N> {
     weights: Arc<N::Weights>,
     number: BlockNumber,
     validator_id: Option<N::ValidatorId>,
-    proposal: Option<N::Block>,
+    our_proposal: Option<N::Block>,
   ) -> BlockData<N> {
     BlockData {
       db,
@@ -51,7 +50,7 @@ impl<N: Network> BlockData<N> {
 
       number,
       validator_id,
-      proposal,
+      our_proposal,
 
       log: MessageLog::new(weights),
       slashes: HashSet::new(),
@@ -108,17 +107,17 @@ impl<N: Network> BlockData<N> {
       self.populate_end_time(round);
     }
 
-    // 11-13
+    // L11-13
     self.round = Some(RoundData::<N>::new(
       round,
       time.unwrap_or_else(|| self.end_time[&RoundNumber(round.0 - 1)]),
     ));
     self.end_time.insert(round, self.round().end_time());
 
-    // 14-21
+    // L14-21
     if Some(proposer) == self.validator_id {
       let (round, block) = self.valid.clone().unzip();
-      block.or_else(|| self.proposal.clone()).map(|block| Data::Proposal(round, block))
+      block.or_else(|| self.our_proposal.clone()).map(|block| Data::Proposal(round, block))
     } else {
       self.round_mut().set_timeout(Step::Propose);
       None
@@ -198,8 +197,8 @@ impl<N: Network> BlockData<N> {
         assert!(!new_round);
         None?;
       }
-      // Put this message to the DB
-      txn.put(&msg_key, res.encode());
+      // Put that we're sending this message to the DB
+      txn.put(&msg_key, []);
 
       txn.commit();
     }

coordinator/tributary/tendermint/src/ext.rs

@@ -288,7 +288,7 @@ pub trait Network: Sized + Send + Sync {
   async fn slash(&mut self, validator: Self::ValidatorId, slash_event: SlashEvent);
 
   /// Validate a block.
-  async fn validate(&mut self, block: &Self::Block) -> Result<(), BlockError>;
+  async fn validate(&self, block: &Self::Block) -> Result<(), BlockError>;
 
   /// Add a block, returning the proposal for the next one.
   ///

coordinator/tributary/tendermint/src/message_log.rs

@@ -2,21 +2,30 @@ use std::{sync::Arc, collections::HashMap};
 
 use parity_scale_codec::Encode;
 
-use crate::{ext::*, RoundNumber, Step, DataFor, TendermintError, SignedMessageFor, Evidence};
+use crate::{ext::*, RoundNumber, Step, DataFor, SignedMessageFor, Evidence};
 
 type RoundLog<N> = HashMap<<N as Network>::ValidatorId, HashMap<Step, SignedMessageFor<N>>>;
 pub(crate) struct MessageLog<N: Network> {
   weights: Arc<N::Weights>,
+  round_participation: HashMap<RoundNumber, u64>,
+  participation: HashMap<(RoundNumber, Step), u64>,
+  message_instances: HashMap<(RoundNumber, DataFor<N>), u64>,
   pub(crate) log: HashMap<RoundNumber, RoundLog<N>>,
 }
 
 impl<N: Network> MessageLog<N> {
   pub(crate) fn new(weights: Arc<N::Weights>) -> MessageLog<N> {
-    MessageLog { weights, log: HashMap::new() }
+    MessageLog {
+      weights,
+      round_participation: HashMap::new(),
+      participation: HashMap::new(),
+      message_instances: HashMap::new(),
+      log: HashMap::new(),
+    }
   }
 
   // Returns true if it's a new message
-  pub(crate) fn log(&mut self, signed: SignedMessageFor<N>) -> Result<bool, TendermintError<N>> {
+  pub(crate) fn log(&mut self, signed: SignedMessageFor<N>) -> Result<bool, Evidence> {
     let msg = &signed.msg;
     // Clarity, and safety around default != new edge cases
     let round = self.log.entry(msg.round).or_insert_with(HashMap::new);
@@ -30,69 +39,36 @@ impl<N: Network> MessageLog<N> {
           target: "tendermint",
           "Validator sent multiple messages for the same block + round + step"
         );
-        Err(TendermintError::Malicious(
-          msg.sender,
-          Some(Evidence::ConflictingMessages(existing.encode(), signed.encode())),
-        ))?;
+        Err(Evidence::ConflictingMessages(existing.encode(), signed.encode()))?;
       }
       return Ok(false);
     }
 
+    // Since we have a new message, update the participation
+    let sender_weight = self.weights.weight(msg.sender);
+    if msgs.is_empty() {
+      *self.round_participation.entry(msg.round).or_insert_with(|| 0) += sender_weight;
+    }
+    *self.participation.entry((msg.round, step)).or_insert_with(|| 0) += sender_weight;
+    *self.message_instances.entry((msg.round, msg.data.clone())).or_insert_with(|| 0) +=
+      sender_weight;
+
     msgs.insert(step, signed);
     Ok(true)
   }
 
-  // For a given round, return the participating weight for this step, and the weight agreeing with
-  // the data.
-  pub(crate) fn message_instances(&self, round: RoundNumber, data: &DataFor<N>) -> (u64, u64) {
-    let mut participating = 0;
-    let mut weight = 0;
-    for (participant, msgs) in &self.log[&round] {
-      if let Some(msg) = msgs.get(&data.step()) {
-        let validator_weight = self.weights.weight(*participant);
-        participating += validator_weight;
-        if data == &msg.msg.data {
-          weight += validator_weight;
-        }
-      }
-    }
-    (participating, weight)
-  }
-
   // Get the participation in a given round
   pub(crate) fn round_participation(&self, round: RoundNumber) -> u64 {
-    let mut weight = 0;
-    if let Some(round) = self.log.get(&round) {
-      for participant in round.keys() {
-        weight += self.weights.weight(*participant);
-      }
-    };
-    weight
+    *self.round_participation.get(&round).unwrap_or(&0)
   }
 
   // Check if a supermajority of nodes have participated on a specific step
   pub(crate) fn has_participation(&self, round: RoundNumber, step: Step) -> bool {
-    let mut participating = 0;
-    for (participant, msgs) in &self.log[&round] {
-      if msgs.get(&step).is_some() {
-        participating += self.weights.weight(*participant);
-      }
-    }
-    participating >= self.weights.threshold()
+    *self.participation.get(&(round, step)).unwrap_or(&0) >= self.weights.threshold()
   }
 
   // Check if consensus has been reached on a specific piece of data
   pub(crate) fn has_consensus(&self, round: RoundNumber, data: &DataFor<N>) -> bool {
-    let (_, weight) = self.message_instances(round, data);
-    weight >= self.weights.threshold()
-  }
-
-  pub(crate) fn get(
-    &self,
-    round: RoundNumber,
-    sender: N::ValidatorId,
-    step: Step,
-  ) -> Option<&SignedMessageFor<N>> {
-    self.log.get(&round).and_then(|round| round.get(&sender).and_then(|msgs| msgs.get(&step)))
+    *self.message_instances.get(&(round, data.clone())).unwrap_or(&0) >= self.weights.threshold()
   }
 }

coordinator/tributary/tendermint/src/round.rs

@@ -57,6 +57,7 @@ impl<N: Network> RoundData<N> {
 
   // Poll all set timeouts, returning the Step whose timeout has just expired
   pub(crate) async fn timeout_future(&self) -> Step {
+    /*
     let now = Instant::now();
     log::trace!(
       target: "tendermint",
@@ -64,6 +65,7 @@ impl<N: Network> RoundData<N> {
       self.step,
       self.timeouts.iter().map(|(k, v)| (k, v.duration_since(now))).collect::<HashMap<_, _>>()
     );
+    */
 
     let timeout_future = |step| {
       let timeout = self.timeouts.get(&step).copied();

coordinator/tributary/tendermint/tests/ext.rs

@@ -145,7 +145,7 @@ impl Network for TestNetwork {
     println!("Slash for {id} due to {event:?}");
   }
 
-  async fn validate(&mut self, block: &TestBlock) -> Result<(), BlockError> {
+  async fn validate(&self, block: &TestBlock) -> Result<(), BlockError> {
     block.valid
   }