refactor(taiko-client-rs, preconfirmation-p2p): simplify codebase by consolidating single-file submodules (#21142)

Author: davidtaikocha

packages/preconfirmation-p2p/Cargo.lock

@@ -15,7 +15,6 @@ quic-transport = ["libp2p/quic"]
 alloy-primitives.workspace = true
 anyhow.workspace = true
 async-trait.workspace = true
-clap = { version = "4", features = ["derive"] }
 crossbeam-skiplist.workspace = true
 dashmap.workspace = true
 discv5.workspace = true
@@ -49,4 +48,5 @@ workspace = true
 workspace = true
 
 [dev-dependencies]
+clap = { version = "4", features = ["derive"] }
 tokio = { workspace = true, features = ["full"] }

packages/preconfirmation-p2p/crates/net/Cargo.toml

@@ -139,16 +139,13 @@ impl NetBehaviour {
             reqresp_cfg,
         );
 
-        let (pend_in, pend_out, est_in, est_out, est_total, per_peer, _dial_factor) =
-            cfg.resolve_connection_caps();
-
-        let mut limits = ConnectionLimits::default();
-        limits = limits.with_max_pending_incoming(pend_in);
-        limits = limits.with_max_pending_outgoing(pend_out);
-        limits = limits.with_max_established_incoming(est_in);
-        limits = limits.with_max_established_outgoing(est_out);
-        limits = limits.with_max_established(est_total);
-        limits = limits.with_max_established_per_peer(per_peer);
+        let limits = ConnectionLimits::default()
+            .with_max_pending_incoming(cfg.max_pending_incoming)
+            .with_max_pending_outgoing(cfg.max_pending_outgoing)
+            .with_max_established_incoming(cfg.max_established_incoming)
+            .with_max_established_outgoing(cfg.max_established_outgoing)
+            .with_max_established(cfg.max_established_total)
+            .with_max_established_per_peer(cfg.max_established_per_peer);
 
         let block_list = BlockListBehaviour::<BlockedPeers>::default();
         let conn_limits = ConnectionLimitsBehaviour::new(limits);

packages/preconfirmation-p2p/crates/net/src/behaviour.rs

@@ -88,13 +88,9 @@ pub fn build_transport_and_behaviour(cfg: &NetworkConfig) -> anyhow::Result<Buil
         libp2p::gossipsub::IdentTopic::new(preconfirmation_types::topic_raw_txlists(cfg.chain_id)),
     );
     let protocols = crate::codec::Protocols {
-        commitments: crate::codec::SszProtocol(
-            preconfirmation_types::protocol_get_commitments_by_number(cfg.chain_id),
-        ),
-        raw_txlists: crate::codec::SszProtocol(preconfirmation_types::protocol_get_raw_txlist(
-            cfg.chain_id,
-        )),
-        head: crate::codec::SszProtocol(preconfirmation_types::protocol_get_head(cfg.chain_id)),
+        commitments: preconfirmation_types::protocol_get_commitments_by_number(cfg.chain_id),
+        raw_txlists: preconfirmation_types::protocol_get_raw_txlist(cfg.chain_id),
+        head: preconfirmation_types::protocol_get_head(cfg.chain_id),
     };
     let behaviour = NetBehaviour::new(keypair.clone(), topics.clone(), protocols, cfg)?;
 

packages/preconfirmation-p2p/crates/net/src/builder.rs

@@ -74,31 +74,15 @@ const HEAD_REQ_MAX_BYTES: usize = 128;
 /// Maximum encoded size for head response frames.
 const HEAD_RESP_MAX_BYTES: usize = 64 * 1024;
 
-#[derive(Clone)]
 /// Holds the protocol IDs for various request-response protocols.
+#[derive(Clone)]
 pub struct Protocols {
     /// Protocol ID for commitments request-response.
-    pub commitments: SszProtocol,
+    pub commitments: String,
     /// Protocol ID for raw transaction list request-response.
-    pub raw_txlists: SszProtocol,
+    pub raw_txlists: String,
     /// Protocol ID for get_head request-response.
-    pub head: SszProtocol,
-}
-
-#[derive(Clone)]
-/// A wrapper for a protocol ID string.
-///
-/// Implements `AsRef<str>` to allow easy conversion to `&str`.
-pub struct SszProtocol(
-    /// Protocol identifier string.
-    pub String,
-);
-
-impl AsRef<str> for SszProtocol {
-    /// Return the protocol identifier as a string slice.
-    fn as_ref(&self) -> &str {
-        &self.0
-    }
+    pub head: String,
 }
 
 #[async_trait]
@@ -108,20 +92,20 @@ where
     Req: SimpleSerialize + Clone + Send + Sync + 'static,
     Resp: SimpleSerialize + Clone + Send + Sync + 'static,
 {
-    type Protocol = SszProtocol;
+    type Protocol = String;
     type Request = Req;
     type Response = Resp;
 
     /// Read and decode a request frame.
-    async fn read_request<R>(&mut self, _: &SszProtocol, io: &mut R) -> io::Result<Self::Request>
+    async fn read_request<R>(&mut self, _: &String, io: &mut R) -> io::Result<Self::Request>
     where
         R: AsyncRead + Unpin + Send,
     {
         read_ssz(io, MAX_REQ).await
     }
 
     /// Read and decode a response frame.
-    async fn read_response<R>(&mut self, _: &SszProtocol, io: &mut R) -> io::Result<Self::Response>
+    async fn read_response<R>(&mut self, _: &String, io: &mut R) -> io::Result<Self::Response>
     where
         R: AsyncRead + Unpin + Send,
     {
@@ -131,7 +115,7 @@ where
     /// Encode and write a request frame.
     async fn write_request<W>(
         &mut self,
-        _: &SszProtocol,
+        _: &String,
         io: &mut W,
         req: Self::Request,
     ) -> io::Result<()>
@@ -144,7 +128,7 @@ where
     /// Encode and write a response frame.
     async fn write_response<W>(
         &mut self,
-        _: &SszProtocol,
+        _: &String,
         io: &mut W,
         res: Self::Response,
     ) -> io::Result<()>
@@ -160,7 +144,7 @@ where
 ///
 /// The function reads a u32 length encoded as unsigned-varint, rejects frames larger than
 /// `max_len`, then reads the payload and attempts SSZ deserialization.
-async fn read_ssz<T: SimpleSerialize + Default, R: AsyncRead + Unpin + Send>(
+async fn read_ssz<T: SimpleSerialize, R: AsyncRead + Unpin + Send>(
     io: &mut R,
     max_len: usize,
 ) -> io::Result<T> {
@@ -219,7 +203,7 @@ mod tests {
     #[tokio::test]
     async fn commitments_codec_roundtrip() {
         let mut codec = CommitmentsCodec::default();
-        let proto = SszProtocol("/taiko/test/commitments/1".into());
+        let proto = "/taiko/test/commitments/1".to_string();
 
         let req =
             GetCommitmentsByNumberRequest { start_block_number: Uint256::from(1u64), max_count: 5 };
@@ -241,7 +225,7 @@ mod tests {
     #[tokio::test]
     async fn oversized_response_is_rejected_on_read() {
         let mut codec = RawTxListCodec::default();
-        let proto = SszProtocol("/taiko/test/rawtx/1".into());
+        let proto = "/taiko/test/rawtx/1".to_string();
 
         // Craft a frame with a length one byte over the allowed bound.
         let over = RAW_TXLIST_RESP_MAX_BYTES + 1;
@@ -267,7 +251,7 @@ mod tests {
             16,
         >;
         let mut codec = TinyCodec::default();
-        let proto = SszProtocol("/taiko/test/rawtx/1".into());
+        let proto = "/taiko/test/rawtx/1".to_string();
 
         let txlist = preconfirmation_types::TxListBytes::try_from(vec![0u8; 32]).unwrap();
         let resp = GetRawTxListResponse {

packages/preconfirmation-p2p/crates/net/src/codec.rs

@@ -193,32 +193,7 @@ impl Default for NetworkConfig {
     }
 }
 
-/// Logical grouping for the resolved connection caps and dial factor returned by
-/// `NetworkConfig::resolve_connection_caps`.
-pub(crate) type ConnectionCaps = (
-    Option<u32>, // pending inbound cap
-    Option<u32>, // pending outbound cap
-    Option<u32>, // established inbound cap
-    Option<u32>, // established outbound cap
-    Option<u32>, // total established cap
-    Option<u32>, // established per peer cap
-    u8,          // dial concurrency factor
-);
-
 impl NetworkConfig {
-    /// Resolve connection caps and dial factor.
-    pub(crate) fn resolve_connection_caps(&self) -> ConnectionCaps {
-        (
-            self.max_pending_incoming,
-            self.max_pending_outgoing,
-            self.max_established_incoming,
-            self.max_established_outgoing,
-            self.max_established_total,
-            self.max_established_per_peer,
-            self.dial_concurrency_factor,
-        )
-    }
-
     /// Ensure rate-limit parameters are sane before constructing a limiter.
     pub(crate) fn validate_request_rate_limits(&self) {
         debug_assert!(self.request_window > Duration::ZERO, "request_window must be > 0");

packages/preconfirmation-p2p/crates/net/src/config.rs

@@ -12,34 +12,24 @@ use super::*;
 /// Identifies which gossip topic payload is being processed.
 #[derive(Clone, Copy)]
 enum GossipKind {
-    /// Signed commitment payloads on the commitments topic.
     Commitment,
-    /// Raw transaction list payloads on the raw txlists topic.
     RawTxList,
 }
 
 impl GossipKind {
-    /// Returns the label used for metrics for this gossip kind.
+    /// Returns the label used for metrics.
     fn label(self) -> &'static str {
         match self {
-            GossipKind::Commitment => "commitment",
-            GossipKind::RawTxList => "raw_txlists",
+            Self::Commitment => "commitment",
+            Self::RawTxList => "raw_txlists",
         }
     }
 
-    /// Returns the error string used when decoding fails.
-    fn decode_error(self) -> &'static str {
+    /// Returns the error string used for decode/validation failures.
+    fn error_msg(self) -> &'static str {
         match self {
-            GossipKind::Commitment => "invalid signed commitment gossip",
-            GossipKind::RawTxList => "invalid raw txlist gossip",
-        }
-    }
-
-    /// Returns the error string used when validation fails.
-    fn validation_error(self) -> &'static str {
-        match self {
-            GossipKind::Commitment => "invalid signed commitment gossip",
-            GossipKind::RawTxList => "invalid raw txlist gossip",
+            Self::Commitment => "invalid signed commitment gossip",
+            Self::RawTxList => "invalid raw txlist gossip",
         }
     }
 }
@@ -88,7 +78,7 @@ impl NetworkDriver {
                     );
                     metrics::counter!("p2p_gossip_invalid", "kind" => kind.label(), "reason" => "validation").increment(1);
                     self.apply_reputation(propagation_source, PeerAction::GossipInvalid);
-                    self.emit_error(NetworkErrorKind::GossipValidation, kind.validation_error());
+                    self.emit_error(NetworkErrorKind::GossipValidation, kind.error_msg());
                 }
             }
             Err(_) => {
@@ -100,7 +90,7 @@ impl NetworkDriver {
                 metrics::counter!("p2p_gossip_invalid", "kind" => kind.label(), "reason" => "decode")
                     .increment(1);
                 self.apply_reputation(propagation_source, PeerAction::GossipInvalid);
-                self.emit_error(NetworkErrorKind::GossipDecode, kind.decode_error());
+                self.emit_error(NetworkErrorKind::GossipDecode, kind.error_msg());
             }
         }
     }

packages/preconfirmation-p2p/crates/net/src/driver/gossip.rs

@@ -111,10 +111,8 @@ impl NetworkDriver {
         validator: Box<dyn ValidationAdapter>,
         storage: Option<Arc<dyn PreconfStorage>>,
     ) -> anyhow::Result<(Self, NetworkHandle)> {
-        let dial_factor = {
-            let (_, _, _, _, _, _, dial) = cfg.resolve_connection_caps();
-            NonZeroU8::new(dial).unwrap_or_else(|| NonZeroU8::new(1).unwrap())
-        };
+        let dial_factor =
+            NonZeroU8::new(cfg.dial_concurrency_factor).unwrap_or(NonZeroU8::new(1).unwrap());
 
         let parts = build_transport_and_behaviour(&cfg)?;
         let peer_id = parts.keypair.public().to_peer_id();