Update CDN in ab/fee-upgrade (#1606)

* update the CDN

* update comment

* fix latency calculation

* query service update

builder/src/lib.rs

@@ -9,9 +9,7 @@ use futures::{
     stream::{Stream, StreamExt},
 };
 use hotshot::{
-    traits::{
-        election::static_committee::GeneralStaticCommittee, implementations::NetworkingMetricsValue,
-    },
+    traits::election::static_committee::GeneralStaticCommittee,
     types::{SignatureKey, SystemContextHandle},
     HotShotInitializer, Memberships, Networks, SystemContext,
 };
@@ -379,8 +377,7 @@ pub mod testing {
 
             let network = Arc::new(MemoryNetwork::new(
                 config.my_own_validator_config.public_key,
-                NetworkingMetricsValue::new(metrics),
-                self.master_map.clone(),
+                &self.master_map,
                 None,
             ));
             let networks = Networks {

builder/src/permissioned.rs

@@ -12,8 +12,8 @@ use hotshot::{
     traits::{
         election::static_committee::GeneralStaticCommittee,
         implementations::{
-            derive_libp2p_peer_id, CombinedNetworks, KeyPair, Libp2pNetwork,
-            NetworkingMetricsValue, PushCdnNetwork, Topic, WrappedSignatureKey,
+            derive_libp2p_peer_id, CdnMetricsValue, CombinedNetworks, KeyPair, Libp2pNetwork,
+            PushCdnNetwork, Topic, WrappedSignatureKey,
         },
     },
     types::{SignatureKey, SystemContextHandle},
@@ -209,6 +209,7 @@ pub async fn init_node<P: SequencerPersistence, Ver: StaticVersionType + 'static
             public_key: WrappedSignatureKey(my_config.public_key),
             private_key: my_config.private_key.clone(),
         },
+        CdnMetricsValue::new(metrics),
     )
     .with_context(|| "Failed to create CDN network")?;
 
@@ -221,6 +222,7 @@ pub async fn init_node<P: SequencerPersistence, Ver: StaticVersionType + 'static
         // We need the private key so we can derive our Libp2p keypair
         // (using https://docs.rs/blake3/latest/blake3/fn.derive_key.html)
         &my_config.private_key,
+        hotshot::traits::implementations::Libp2pMetricsValue::new(metrics),
     )
     .await
     .with_context(|| "Failed to create libp2p network")?;
@@ -252,12 +254,6 @@ pub async fn init_node<P: SequencerPersistence, Ver: StaticVersionType + 'static
         _pd: Default::default(),
     };
 
-    // The web server network doesn't have any metrics. By creating and dropping a
-    // `NetworkingMetricsValue`, we ensure the networking metrics are created, but just not
-    // populated, so that monitoring software built to work with network-related metrics doesn't
-    // crash horribly just because we're not using the P2P network yet.
-    let _ = NetworkingMetricsValue::new(metrics);
-
     let mut genesis_state = ValidatedState {
         chain_config: genesis.chain_config.into(),
         ..Default::default()

sequencer/Cargo.toml

@@ -86,6 +86,7 @@ jf-utils = { workspace = true } # TODO temporary: used only for test_rng()
 jf-vid = { workspace = true }
 libp2p = { workspace = true } 
 num-traits = "0.2.18"
+num_enum = "0.7"
 portpicker = { workspace = true }
 rand = "0.8.5"
 rand_chacha = { workspace = true }
@@ -95,6 +96,7 @@ serde = { workspace = true }
 serde_json = "^1.0.113"
 sha2 = "0.10" # TODO temporary, used only for VID, should be set in hotshot
 snafu = { workspace = true }
+static_assertions = "1"
 strum = { workspace = true }
 surf-disco = { workspace = true }
 tagged-base64 = { workspace = true }

sequencer/src/bin/cdn-broker.rs

@@ -7,6 +7,7 @@ use clap::Parser;
 use hotshot_types::traits::node_implementation::NodeType;
 use hotshot_types::traits::signature_key::SignatureKey;
 use sequencer::network::cdn::{ProductionDef, WrappedSignatureKey};
+use sequencer::options::parse_size;
 use sequencer::SeqTypes;
 use sha2::Digest;
 use tracing_subscriber::EnvFilter;
@@ -72,6 +73,17 @@ struct Args {
     /// The seed for broker key generation
     #[arg(short, long, default_value_t = 0, env = "ESPRESSO_CDN_BROKER_KEY_SEED")]
     key_seed: u64,
+
+    /// The size of the global memory pool. This is the maximum number of bytes that
+    /// can be allocated at once for all connections. A connection will block if it
+    /// tries to allocate more than this amount until some memory is freed.
+    #[arg(
+        long,
+        default_value = "1GB",
+        value_parser = parse_size,
+        env = "ESPRESSO_CDN_BROKER_GLOBAL_MEMORY_POOL_SIZE"
+    )]
+    global_memory_pool_size: usize,
 }
 #[async_std::main]
 async fn main() -> Result<()> {
@@ -111,6 +123,7 @@ async fn main() -> Result<()> {
         public_advertise_endpoint: args.public_advertise_endpoint,
         private_bind_endpoint: args.private_bind_endpoint,
         private_advertise_endpoint: args.private_advertise_endpoint,
+        global_memory_pool_size: Some(args.global_memory_pool_size),
     };
 
     // Create new `Broker`

sequencer/src/bin/cdn-marshal.rs

@@ -4,7 +4,7 @@
 use anyhow::Result;
 use cdn_marshal::{Config, Marshal};
 use clap::Parser;
-use sequencer::{network::cdn::ProductionDef, SeqTypes};
+use sequencer::{network::cdn::ProductionDef, options::parse_size, SeqTypes};
 use tracing_subscriber::EnvFilter;
 
 #[derive(Parser, Debug)]
@@ -38,6 +38,17 @@ struct Args {
     /// If not provided, a local, pinned CA is used
     #[arg(long, env = "ESPRESSO_CDN_MARSHAL_CA_KEY_PATH")]
     ca_key_path: Option<String>,
+
+    /// The size of the global memory pool. This is the maximum number of bytes that
+    /// can be allocated at once for all connections. A connection will block if it
+    /// tries to allocate more than this amount until some memory is freed.
+    #[arg(
+        long,
+        default_value = "1GB",
+        value_parser = parse_size,
+        env = "ESPRESSO_CDN_MARSHAL_GLOBAL_MEMORY_POOL_SIZE"
+    )]
+    global_memory_pool_size: usize,
 }
 
 #[async_std::main]
@@ -64,6 +75,7 @@ async fn main() -> Result<()> {
         metrics_bind_endpoint: args.metrics_bind_endpoint,
         ca_cert_path: args.ca_cert_path,
         ca_key_path: args.ca_key_path,
+        global_memory_pool_size: Some(args.global_memory_pool_size),
     };
 
     // Create new `Marshal` from the config

sequencer/src/bin/dev-cdn.rs

@@ -75,6 +75,7 @@ async fn main() -> Result<()> {
 
         ca_cert_path: None,
         ca_key_path: None,
+        global_memory_pool_size: Some(1024 * 1024 * 1024),
     };
 
     // Configure the marshal
@@ -84,6 +85,7 @@ async fn main() -> Result<()> {
         discovery_endpoint: discovery_endpoint.clone(),
         ca_cert_path: None,
         ca_key_path: None,
+        global_memory_pool_size: Some(1024 * 1024 * 1024),
     };
 
     // Create a new `Broker`

sequencer/src/lib.rs

@@ -42,8 +42,8 @@ use hotshot::{
     traits::{
         election::static_committee::GeneralStaticCommittee,
         implementations::{
-            derive_libp2p_peer_id, KeyPair, MemoryNetwork, NetworkingMetricsValue, PushCdnNetwork,
-            Topic, WrappedSignatureKey,
+            derive_libp2p_peer_id, CdnMetricsValue, KeyPair, MemoryNetwork, PushCdnNetwork, Topic,
+            WrappedSignatureKey,
         },
     },
     types::SignatureKey,
@@ -445,6 +445,7 @@ pub async fn init_node<P: PersistenceOptions, Ver: StaticVersionType + 'static>(
             public_key: WrappedSignatureKey(my_config.public_key),
             private_key: my_config.private_key.clone(),
         },
+        CdnMetricsValue::new(metrics),
     )
     .with_context(|| "Failed to create CDN network")?;
 
@@ -458,6 +459,7 @@ pub async fn init_node<P: PersistenceOptions, Ver: StaticVersionType + 'static>(
             // We need the private key so we can derive our Libp2p keypair
             // (using https://docs.rs/blake3/latest/blake3/fn.derive_key.html)
             &my_config.private_key,
+            hotshot::traits::implementations::Libp2pMetricsValue::new(metrics),
         )
         .await
         .with_context(|| "Failed to create libp2p network")?;
@@ -503,12 +505,6 @@ pub async fn init_node<P: PersistenceOptions, Ver: StaticVersionType + 'static>(
         _pd: Default::default(),
     };
 
-    // The web server network doesn't have any metrics. By creating and dropping a
-    // `NetworkingMetricsValue`, we ensure the networking metrics are created, but just not
-    // populated, so that monitoring software built to work with network-related metrics doesn't
-    // crash horribly just because we're not using the P2P network yet.
-    let _ = NetworkingMetricsValue::new(metrics);
-
     let mut genesis_state = ValidatedState {
         chain_config: genesis.chain_config.into(),
         ..Default::default()
@@ -789,8 +785,7 @@ pub mod testing {
 
             let network = Arc::new(MemoryNetwork::new(
                 config.my_own_validator_config.public_key,
-                NetworkingMetricsValue::new(metrics),
-                self.master_map.clone(),
+                &self.master_map,
                 None,
             ));
             let networks = Networks {

sequencer/src/network/cdn.rs

@@ -3,16 +3,33 @@ use std::marker::PhantomData;
 
 use bincode::Options;
 use cdn_broker::reexports::{
-    connection::{
-        protocols::{Quic, Tcp},
-        NoMiddleware, TrustedMiddleware, UntrustedMiddleware,
-    },
+    connection::protocols::{Quic, Tcp},
     crypto::signature::{Serializable, SignatureScheme},
-    def::{ConnectionDef, RunDef},
+    def::{ConnectionDef, RunDef, Topic as TopicTrait},
     discovery::{Embedded, Redis},
 };
-use hotshot::{traits::implementations::Topic, types::SignatureKey};
+use hotshot::{traits::implementations::Topic as HotShotTopic, types::SignatureKey};
 use hotshot_types::{traits::node_implementation::NodeType, utils::bincode_opts};
+use num_enum::{IntoPrimitive, TryFromPrimitive};
+use static_assertions::const_assert_eq;
+
+/// The enum for the topics we can subscribe to in the Push CDN
+#[repr(u8)]
+#[derive(IntoPrimitive, TryFromPrimitive, Clone, PartialEq, Eq)]
+pub enum Topic {
+    /// The global topic
+    Global = 0,
+    /// The DA topic
+    Da = 1,
+}
+
+// Make sure the topics are the same as defined in `HotShot`.
+const_assert_eq!(Topic::Global as u8, HotShotTopic::Global as u8);
+const_assert_eq!(Topic::Da as u8, HotShotTopic::Da as u8);
+
+/// Implement the `TopicTrait` for our `Topic` enum. This lets us define
+/// compatible topics at the broker-level. Others will be rejected.
+impl TopicTrait for Topic {}
 
 /// A wrapped `SignatureKey`. We need to implement the Push CDN's `SignatureScheme`
 /// trait in order to sign and verify messages to/from the CDN.
@@ -70,7 +87,6 @@ pub struct UserDef<TYPES: NodeType>(PhantomData<TYPES>);
 impl<TYPES: NodeType> ConnectionDef for UserDef<TYPES> {
     type Scheme = WrappedSignatureKey<TYPES::SignatureKey>;
     type Protocol = Quic;
-    type Middleware = UntrustedMiddleware;
 }
 
 /// The broker definition for the Push CDN.
@@ -79,7 +95,6 @@ pub struct BrokerDef<TYPES: NodeType>(PhantomData<TYPES>);
 impl<TYPES: NodeType> ConnectionDef for BrokerDef<TYPES> {
     type Scheme = WrappedSignatureKey<TYPES::SignatureKey>;
     type Protocol = Tcp;
-    type Middleware = TrustedMiddleware;
 }
 
 /// The client definition for the Push CDN. Uses the Quic
@@ -90,7 +105,6 @@ pub struct ClientDef<TYPES: NodeType>(PhantomData<TYPES>);
 impl<TYPES: NodeType> ConnectionDef for ClientDef<TYPES> {
     type Scheme = WrappedSignatureKey<TYPES::SignatureKey>;
     type Protocol = Quic;
-    type Middleware = NoMiddleware;
 }
 
 /// The testing run definition for the Push CDN.