Add direct offer w/ running census process to state bridge (#1422)

* feat(portal-bridge): add direct offer / census feature to state bridge

Cargo.toml

@@ -87,6 +87,7 @@ async-trait = "0.1.68"
 bytes = "1.3.0"
 chrono = "0.4.38"
 clap = { version = "4.2.1", features = ["derive"] }
+delay_map = "0.4.0"
 directories = "3.0"
 discv5 = { version = "0.4.1", features = ["serde"] }
 e2store = { path = "e2store" }

portal-bridge/Cargo.toml

@@ -18,6 +18,7 @@ anyhow.workspace = true
 async-trait.workspace = true
 chrono.workspace = true
 clap.workspace = true
+delay_map.workspace = true
 discv5.workspace = true
 e2store.workspace = true
 eth_trie.workspace = true

portal-bridge/src/bridge/era1.rs

@@ -49,7 +49,9 @@ pub struct Era1Bridge {
     pub era1_files: Vec<String>,
     pub http_client: Client,
     pub metrics: BridgeMetricsReporter,
-    pub gossip_limit: usize,
+    // Semaphore used to limit the amount of active gossip transfers
+    // to make sure we don't overwhelm the trin client
+    pub gossip_semaphore: Arc<Semaphore>,
     pub execution_api: ExecutionApi,
 }
 
@@ -69,6 +71,7 @@ impl Era1Bridge {
             .try_into()?;
         let era1_files = get_shuffled_era1_files(&http_client).await?;
         let metrics = BridgeMetricsReporter::new("era1".to_string(), &format!("{mode:?}"));
+        let gossip_semaphore = Arc::new(Semaphore::new(gossip_limit));
         Ok(Self {
             mode,
             portal_client,
@@ -77,7 +80,7 @@ impl Era1Bridge {
             era1_files,
             http_client,
             metrics,
-            gossip_limit,
+            gossip_semaphore,
             execution_api,
         })
     }
@@ -228,9 +231,6 @@ impl Era1Bridge {
 
     async fn gossip_era1(&self, era1_path: String, gossip_range: Option<Range<u64>>, hunt: bool) {
         info!("Processing era1 file at path: {era1_path:?}");
-        // We are using a semaphore to limit the amount of active gossip transfers to make sure
-        // we don't overwhelm the trin client
-        let gossip_send_semaphore = Arc::new(Semaphore::new(self.gossip_limit));
 
         let raw_era1 = self
             .http_client
@@ -263,7 +263,8 @@ impl Era1Bridge {
                     continue;
                 }
             }
-            let permit = gossip_send_semaphore
+            let permit = self
+                .gossip_semaphore
                 .clone()
                 .acquire_owned()
                 .await

portal-bridge/src/bridge/history.rs

@@ -42,7 +42,9 @@ pub struct HistoryBridge {
     pub header_oracle: HeaderOracle,
     pub epoch_acc_path: PathBuf,
     pub metrics: BridgeMetricsReporter,
-    pub gossip_limit: usize,
+    // Semaphore used to limit the amount of active gossip transfers
+    // to make sure we don't overwhelm the trin client
+    pub gossip_semaphore: Arc<Semaphore>,
 }
 
 impl HistoryBridge {
@@ -55,14 +57,15 @@ impl HistoryBridge {
         gossip_limit: usize,
     ) -> Self {
         let metrics = BridgeMetricsReporter::new("history".to_string(), &format!("{mode:?}"));
+        let gossip_semaphore = Arc::new(Semaphore::new(gossip_limit));
         Self {
             mode,
             portal_client,
             execution_api,
             header_oracle,
             epoch_acc_path,
             metrics,
-            gossip_limit,
+            gossip_semaphore,
         }
     }
 }
@@ -177,10 +180,6 @@ impl HistoryBridge {
         // epoch_acc gets set on the first iteration of the loop
         let mut current_epoch_index = u64::MAX;
 
-        // We are using a semaphore to limit the amount of active gossip transfers to make sure we
-        // don't overwhelm the trin client
-        let gossip_send_semaphore = Arc::new(Semaphore::new(self.gossip_limit));
-
         info!("fetching headers in range: {gossip_range:?}");
         let mut epoch_acc = None;
         let mut serve_full_block_handles = vec![];
@@ -203,9 +202,7 @@ impl HistoryBridge {
             } else if height > MERGE_BLOCK_NUMBER {
                 epoch_acc = None;
             }
-            let permit = gossip_send_semaphore.clone().acquire_owned().await.expect(
-                "acquire_owned() can only error on semaphore close, this should be impossible",
-            );
+            let permit = self.acquire_gossip_permit().await;
             self.metrics.report_current_block(height as i64);
             serve_full_block_handles.push(Self::spawn_serve_full_block(
                 height,
@@ -408,4 +405,12 @@ impl HistoryBridge {
         metrics.stop_process_timer(timer);
         result
     }
+
+    async fn acquire_gossip_permit(&self) -> OwnedSemaphorePermit {
+        self.gossip_semaphore
+            .clone()
+            .acquire_owned()
+            .await
+            .expect("to be able to acquire semaphore")
+    }
 }