Merge branch 'main' into deployer-tech-debt

Author: alysiahuggins

Cargo.lock

@@ -358,12 +358,11 @@ impl<TYPES: NodeType> ConnectedNetwork<TYPES::SignatureKey> for CombinedNetworks
         let secondary = self.secondary().clone();
         let primary_message = message.clone();
         let secondary_message = message.clone();
-        let topic_clone = topic.clone();
         self.send_both_networks(
             message,
             async move {
                 primary
-                    .broadcast_message(primary_message, topic_clone, BroadcastDelay::None)
+                    .broadcast_message(primary_message, topic, BroadcastDelay::None)
                     .await
             },
             async move {

crates/hotshot/hotshot/src/traits/networking/combined_network.rs

@@ -853,6 +853,15 @@ impl<T: NodeType> ConnectedNetwork<T::SignatureKey> for Libp2pNetwork<T> {
             return Err(NetworkError::NoPeersYet);
         };
 
+        // If we are subscribed to the DA topic, send the message to ourselves first
+        let topic = Topic::Da.to_string();
+        if self.inner.subscribed_topics.contains(&topic) {
+            self.inner.sender.try_send(message.clone()).map_err(|_| {
+                self.inner.metrics.num_failed_messages.add(1);
+                NetworkError::ShutDown
+            })?;
+        }
+
         let future_results = recipients
             .into_iter()
             .map(|r| self.direct_message(message.clone(), r));

crates/hotshot/hotshot/src/traits/networking/libp2p_network.rs

@@ -153,7 +153,7 @@ impl<K: SignatureKey> MemoryNetwork<K> {
         for topic in subscribed_topics {
             master_map
                 .subscribed_map
-                .entry(topic.clone())
+                .entry(*topic)
                 .or_default()
                 .push((pub_key.clone(), mn.clone()));
         }

crates/hotshot/hotshot/src/traits/networking/memory_network.rs

@@ -274,6 +274,17 @@ impl<K: SignatureKey + 'static> PushCdnNetwork<K> {
     /// - If we fail to serialize the message
     /// - If we fail to send the broadcast message.
     async fn broadcast_message(&self, message: Vec<u8>, topic: Topic) -> Result<(), NetworkError> {
+        // If the message should also go to us, add it to the internal queue
+        if self
+            .client
+            .subscribed_topics
+            .read()
+            .await
+            .contains(&(topic.clone() as u8))
+        {
+            self.internal_queue.lock().push_back(message.clone());
+        }
+
         // If we're paused, don't send the message
         #[cfg(feature = "hotshot-testing")]
         if self.is_paused.load(Ordering::Relaxed) {
@@ -517,6 +528,8 @@ impl<K: SignatureKey + 'static> ConnectedNetwork<K> for PushCdnNetwork<K> {
         if self.is_paused.load(Ordering::Relaxed) {
             return Ok(());
         }
+
+        // Broadcast the message
         self.broadcast_message(message, topic.into())
             .await
             .inspect_err(|_e| {
@@ -540,6 +553,8 @@ impl<K: SignatureKey + 'static> ConnectedNetwork<K> for PushCdnNetwork<K> {
         if self.is_paused.load(Ordering::Relaxed) {
             return Ok(());
         }
+
+        // Broadcast the message
         self.broadcast_message(message, Topic::Da)
             .await
             .inspect_err(|_e| {

crates/hotshot/hotshot/src/traits/networking/push_cdn_network.rs

@@ -37,6 +37,7 @@ displaydoc = { version = "0.2.5", default-features = false }
 dyn-clone = "1.0.17"
 either = { workspace = true }
 futures = { workspace = true, features = ["alloc"] }
+generic-array = "1"
 hotshot-utils = { workspace = true }
 jf-advz = { workspace = true }
 jf-crhf = { workspace = true }

crates/hotshot/types/Cargo.toml

@@ -38,10 +38,10 @@ pub const COMBINED_NETWORK_DELAY_DURATION: u64 = 5000;
 pub const REQUEST_DATA_DELAY: u64 = 5000;
 
 /// Default channel size for consensus event sharing
-pub const EVENT_CHANNEL_SIZE: usize = 100_000;
+pub const EVENT_CHANNEL_SIZE: usize = 1_000;
 
 /// Default channel size for HotShot -> application communication
-pub const EXTERNAL_EVENT_CHANNEL_SIZE: usize = 100_000;
+pub const EXTERNAL_EVENT_CHANNEL_SIZE: usize = 1_000;
 
 /// Default values for the upgrade constants
 pub const DEFAULT_UPGRADE_CONSTANTS: UpgradeConstants = UpgradeConstants {

crates/hotshot/types/src/constants.rs

@@ -17,7 +17,7 @@ use ark_std::{
     vec::Vec,
 };
 use bitvec::prelude::*;
-use digest::generic_array::GenericArray;
+use generic_array::GenericArray;
 use jf_signature::{AggregateableSignatureSchemes, SignatureError};
 use serde::{Deserialize, Serialize};
 use typenum::U32;

crates/hotshot/types/src/qc.rs

@@ -9,7 +9,7 @@
 use alloy::primitives::U256;
 use ark_serialize::SerializationError;
 use bitvec::{slice::BitSlice, vec::BitVec};
-use digest::generic_array::GenericArray;
+use generic_array::GenericArray;
 use jf_signature::{
     bls_over_bn254::{BLSOverBN254CurveSignatureScheme, KeyPair, SignKey, VerKey},
     SignatureError, SignatureScheme,

crates/hotshot/types/src/signature_key.rs

@@ -587,7 +587,7 @@ impl NetworkReliability for ChaosNetwork {
 }
 
 /// Used when broadcasting messages
-#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+#[derive(Clone, Debug, PartialEq, Eq, Hash, Copy)]
 pub enum Topic {
     /// The `Global` topic goes out to all nodes
     Global,