Improve catchup (#2375)

* Limit time for each catchup request to defend against malicious peers

* Rank catchup peers by reliability; try most reliable peers first

* Add metrics exposing a node's idea of its peers' catchup reliability

* cargo sort

Cargo.toml

@@ -126,6 +126,7 @@ thiserror = "1.0.69"
 tracing = "0.1"
 bytesize = "1.3"
 itertools = "0.12"
+priority-queue = "2"
 rand_chacha = "0.3"
 rand_distr = "0.4"
 reqwest = "0.12"

builder/src/non_permissioned.rs

@@ -19,6 +19,7 @@ use hotshot_types::{
     data::{fake_commitment, ViewNumber},
     traits::{
         block_contents::{vid_commitment, GENESIS_VID_NUM_STORAGE_NODES},
+        metrics::NoMetrics,
         node_implementation::Versions,
         EncodeBytes,
     },
@@ -53,6 +54,7 @@ pub async fn build_instance_state<V: Versions>(
         Arc::new(StatePeers::<SequencerApiVersion>::from_urls(
             state_peers,
             Default::default(),
+            &NoMetrics,
         )),
         V::Base::VERSION,
     );

marketplace-builder/src/builder.rs

@@ -30,6 +30,7 @@ use hotshot_types::{
     data::{fake_commitment, Leaf, ViewNumber},
     traits::{
         block_contents::{vid_commitment, Transaction as _, GENESIS_VID_NUM_STORAGE_NODES},
+        metrics::NoMetrics,
         node_implementation::{ConsensusTime, NodeType, Versions},
         EncodeBytes,
     },
@@ -74,6 +75,7 @@ pub async fn build_instance_state<V: Versions>(
         Arc::new(StatePeers::<SequencerApiVersion>::from_urls(
             state_peers,
             Default::default(),
+            &NoMetrics,
         )),
         V::Base::version(),
     );

sequencer/Cargo.toml

@@ -95,6 +95,7 @@ marketplace-solver = { path = "../marketplace-solver" }
 num_enum = "0.7"
 parking_lot = "0.12"
 portpicker = { workspace = true }
+priority-queue = { workspace = true }
 rand = { workspace = true }
 rand_chacha = { workspace = true }
 rand_distr = { workspace = true }

sequencer/src/api.rs

@@ -1527,6 +1527,7 @@ mod test {
                 StatePeers::<StaticVersion<0, 1>>::from_urls(
                     vec![format!("http://localhost:{port}").parse().unwrap()],
                     Default::default(),
+                    &NoMetrics,
                 )
             }))
             .build();
@@ -1571,6 +1572,7 @@ mod test {
                 StatePeers::<StaticVersion<0, 1>>::from_urls(
                     vec![format!("http://localhost:{port}").parse().unwrap()],
                     Default::default(),
+                    &NoMetrics,
                 ),
                 &NoMetrics,
                 test_helpers::STAKE_TABLE_CAPACITY_FOR_TEST,
@@ -1636,6 +1638,7 @@ mod test {
                 StatePeers::<StaticVersion<0, 1>>::from_urls(
                     vec![format!("http://localhost:{port}").parse().unwrap()],
                     Default::default(),
+                    &NoMetrics,
                 )
             }))
             .network_config(TestConfigBuilder::default().l1_url(l1).build())
@@ -1713,6 +1716,7 @@ mod test {
                 StatePeers::<StaticVersion<0, 1>>::from_urls(
                     vec![format!("http://localhost:{port}").parse().unwrap()],
                     Default::default(),
+                    &NoMetrics,
                 )
             }))
             .network_config(TestConfigBuilder::default().l1_url(l1).build())
@@ -1773,13 +1777,15 @@ mod test {
             StatePeers::<SequencerApiVersion>::from_urls(
                 vec![format!("http://localhost:{port}").parse().unwrap()],
                 BackoffParams::default(),
+                &NoMetrics,
             )
         });
 
         // one of the node has dishonest peer. This list of peers is for node#1
         peers[2] = StatePeers::<SequencerApiVersion>::from_urls(
             vec![url.clone()],
             BackoffParams::default(),
+            &NoMetrics,
         );
 
         let config = TestNetworkConfigBuilder::<NUM_NODES, _, _>::with_num_nodes()
@@ -1801,13 +1807,16 @@ mod test {
         // The catchup should successfully retrieve the correct chain config.
         let node = &network.peers[0];
         let peers = node.node_state().peers;
-        peers.try_fetch_chain_config(cf.commit()).await.unwrap();
+        peers.try_fetch_chain_config(0, cf.commit()).await.unwrap();
 
         // Test a catchup request for node #1, which is connected to a dishonest peer.
         // This request will result in an error due to the malicious chain config provided by the peer.
         let node = &network.peers[1];
         let peers = node.node_state().peers;
-        peers.try_fetch_chain_config(cf.commit()).await.unwrap_err();
+        peers
+            .try_fetch_chain_config(0, cf.commit())
+            .await
+            .unwrap_err();
 
         network.server.shut_down().await;
         handle.abort();
@@ -1963,6 +1972,7 @@ mod test {
                 StatePeers::<SequencerApiVersion>::from_urls(
                     vec![format!("http://localhost:{port}").parse().unwrap()],
                     Default::default(),
+                    &NoMetrics,
                 )
             }))
             .network_config(
@@ -2136,6 +2146,7 @@ mod test {
                 StatePeers::<StaticVersion<0, 1>>::from_urls(
                     vec![format!("http://localhost:{port}").parse().unwrap()],
                     Default::default(),
+                    &NoMetrics,
                 )
             }))
             .network_config(TestConfigBuilder::default().l1_url(l1).build())
@@ -2200,6 +2211,7 @@ mod test {
         let peers = StatePeers::<StaticVersion<0, 1>>::from_urls(
             vec!["https://notarealnode.network".parse().unwrap(), url],
             Default::default(),
+            &NoMetrics,
         );
 
         // Fetch the config from node 1, a different node than the one running the service.