init data-gatherer

Cargo.toml

@@ -41,6 +41,7 @@ ethereum_ssz_derive = "0.9"
 ethers = "2.0.14"
 eyre = "0.6.12"
 aws-sdk-s3 = "=1.35.0"
+clickhouse = "0.14.2"
 flate2 = "1.0"
 
 flux = { git = "https://github.com/gattaca-com/flux", rev = "732c996c13f8fb5e2988dbf949e1d8a74261fca4" }

crates/common/src/config.rs

@@ -43,10 +43,8 @@ pub struct RelayConfig {
     /// Configuration for block merging parameters.
     #[serde(default)]
     pub block_merging_config: BlockMergingConfig,
-    #[serde(default)]
     pub primev_config: Option<PrimevConfig>,
     pub discord_webhook_url: Option<Url>,
-    #[serde(default)]
     pub alerts_config: Option<AlertsConfig>,
     pub inclusion_list: Option<InclusionListConfig>,
     pub is_submission_instance: bool,
@@ -63,10 +61,17 @@ pub struct RelayConfig {
     pub tcp_port: u16,
     #[serde(default = "default_usize::<512>")]
     pub tcp_max_connections: usize,
-    #[serde(default)]
     pub s3_config: Option<S3Config>,
     /// Directory for local cache snapshots (bincode). Enables fast startup.
     pub snapshot_dir: Option<PathBuf>,
+    pub clickhouse: Option<ClickhouseConfig>,
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub struct ClickhouseConfig {
+    pub url: String,
+    pub database: String,
+    pub user: String,
 }
 
 impl RelayConfig {
@@ -100,13 +105,15 @@ impl RelayConfig {
                 decoder: vec![4],
                 simulator: 5,
                 top_bid: 1,
+                data_gatherer: 3,
             },
             gossip_payload_on_header: false,
             api_port: 4040,
             tcp_port: 4041,
             tcp_max_connections: 512,
             s3_config: None,
             snapshot_dir: None,
+            clickhouse: None,
         }
     }
 }
@@ -153,6 +160,8 @@ pub struct CoresConfig {
     pub simulator: usize,
     #[serde(default)]
     pub top_bid: usize,
+    #[serde(default)]
+    pub data_gatherer: usize,
 }
 
 impl Default for WebsiteConfig {
@@ -261,7 +270,6 @@ pub struct SimulatorConfig {
     #[serde(default = "default_usize::<32>")]
     pub max_concurrent_tasks: usize,
     /// If set, use the SSZ binary endpoint at this URL instead of JSON-RPC
-    #[serde(default)]
     pub ssz_url: Option<String>,
 }
 

crates/common/src/validator_preferences/mod.rs

@@ -8,15 +8,13 @@ pub struct ValidatorPreferences {
     /// An optional list of BuilderIDs. If this is set, the relay will only accept
     /// submissions from builders whose public keys are linked to the IDs in this list.
     /// This allows for limiting submissions to a trusted set of builders.
-    #[serde(default)]
     pub trusted_builders: Option<Vec<String>>,
 
     /// Allows validators to express a preference for whether a delay should be applied to get
     /// headers or not.
     #[serde(default = "default_bool::<true>")]
     pub header_delay: bool,
 
-    #[serde(default)]
     pub delay_ms: Option<u64>,
 
     #[serde(default)]

crates/relay/Cargo.toml

@@ -53,6 +53,7 @@ reqwest.workspace = true
 reqwest-eventsource.workspace = true
 rustc-hash.workspace = true
 bincode.workspace = true
+clickhouse.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 serial_test.workspace = true

crates/relay/src/auctioneer/get_header.rs

@@ -57,7 +57,7 @@ impl<B: BidAdjustor> Context<B> {
                     .with_label_values(&[strategy])
                     .observe(start.elapsed().as_micros() as f64);
 
-                self.store_data(adjusted_bid.clone(), sim_request.is_optimistic);
+                self.store_data(adjusted_bid.clone(), sim_request.is_optimistic, producers);
                 self.send_to_sim(sim_request, true, producers);
 
                 if is_adjustable_slot {

crates/relay/src/auctioneer/submit_block.rs

@@ -22,7 +22,7 @@ use crate::{
     simulator::{SimRequest, ValidationRequest, tile::ValidationResult},
     spine::{
         HelixSpineProducers,
-        messages::{ToSimKind, ToSimMsg},
+        messages::{BidEvent, BidUpdate, ToSimKind, ToSimMsg},
     },
 };
 
@@ -122,7 +122,7 @@ impl<B: BidAdjustor> Context<B> {
         );
 
         self.try_adjustments_dry_run(&entry, slot_data, producers);
-        self.store_data(entry, is_optimistic);
+        self.store_data(entry, is_optimistic, producers);
         self.try_merge_block(merging_data, producers);
     }
 
@@ -164,7 +164,7 @@ impl<B: BidAdjustor> Context<B> {
                     .with_label_values(&[strategy])
                     .observe(start.elapsed().as_micros());
 
-                self.store_data(adjusted_block, sim_request.is_optimistic);
+                self.store_data(adjusted_block, sim_request.is_optimistic, producers);
                 self.send_to_sim(sim_request, true, producers);
             }
         }
@@ -217,6 +217,7 @@ impl<B: BidAdjustor> Context<B> {
                 self.request_merged_block(producers);
 
                 if need_send_result {
+                    producers.produce(BidUpdate { block_hash, event: BidEvent::Live });
                     self.db.update_block_submission_live_ts(block_hash, Nanos::now().0);
                     send_submission_result(
                         producers,
@@ -231,7 +232,12 @@ impl<B: BidAdjustor> Context<B> {
         need_send_result
     }
 
-    pub fn store_data(&mut self, entry: PayloadEntry, is_optimistic: bool) {
+    pub fn store_data(
+        &mut self,
+        entry: PayloadEntry,
+        is_optimistic: bool,
+        producers: &mut HelixSpineProducers,
+    ) {
         let block_hash = *entry.block_hash();
         let is_adjusted = entry.is_adjusted();
 
@@ -243,7 +249,12 @@ impl<B: BidAdjustor> Context<B> {
             };
             // For optimistic submissions the bid is live as soon as it is stored.
             // For non-optimistic, live_ts is updated when the simulation result arrives.
-            let live_ts = if is_optimistic { Some(Nanos::now().0) } else { None };
+            let live_ts = if is_optimistic {
+                producers.produce(BidUpdate { block_hash, event: BidEvent::Live });
+                Some(Nanos::now().0)
+            } else {
+                None
+            };
             self.db.store_block_submission(
                 s.signed_bid_submission.clone(),
                 s.submission_trace,

crates/relay/src/auctioneer/validation.rs

@@ -36,11 +36,11 @@ impl<B: BidAdjustor> Context<B> {
             });
         };
 
-        if let helix_types::Submission::Dehydrated(ref dehydrated) = *submission
-            && !self.hydration_cache.can_hydrate(dehydrated, self.chain_info.max_blobs_per_block())
-            {
-                return Err(BlockValidationError::CannotHydrate);
-            }
+        if let helix_types::Submission::Dehydrated(ref dehydrated) = *submission &&
+            !self.hydration_cache.can_hydrate(dehydrated, self.chain_info.max_blobs_per_block())
+        {
+            return Err(BlockValidationError::CannotHydrate);
+        }
 
         self.staleness_check(submission.builder_pubkey(), submission_data.version)?;
         self.validate_submission_data(