firehose connection loadbalance

Author: mangas

graph/src/firehose/endpoints.rs

@@ -11,7 +11,7 @@ use futures03::StreamExt;
 use http::uri::{Scheme, Uri};
 use rand::prelude::IteratorRandom;
 use slog::Logger;
-use std::{collections::BTreeMap, fmt::Display, iter, sync::Arc, time::Duration};
+use std::{collections::BTreeMap, fmt::Display, sync::Arc, time::Duration};
 use tonic::{
     metadata::MetadataValue,
     transport::{Channel, ClientTlsConfig},
@@ -42,7 +42,6 @@ impl FirehoseEndpoint {
         token: Option<String>,
         filters_enabled: bool,
         compression_enabled: bool,
-        conn_pool_size: u16,
     ) -> Self {
         let uri = url
             .as_ref()
@@ -75,8 +74,7 @@ impl FirehoseEndpoint {
             // Timeout on each request, so the timeout to estabilish each 'Blocks' stream.
             .timeout(Duration::from_secs(120));
 
-        // Load balancing on a same endpoint is useful because it creates a connection pool.
-        let channel = Channel::balance_list(iter::repeat(endpoint).take(conn_pool_size as usize));
+        let channel = Channel::balance_list(vec![endpoint].into_iter());
 
         FirehoseEndpoint {
             provider: provider.as_ref().to_string(),
@@ -267,10 +265,12 @@ impl FirehoseEndpoints {
         self.0.len()
     }
 
+    // selects the FirehoseEndpoint with the lest amount of references, which will help with spliting
+    // the load naively across the entire list.
     pub fn random(&self) -> Option<&Arc<FirehoseEndpoint>> {
-        // Select from the matching adapters randomly
-        let mut rng = rand::thread_rng();
-        self.0.iter().choose(&mut rng)
+        self.0
+            .iter()
+            .min_by(|x, y| Arc::strong_count(x).cmp(&Arc::strong_count(y)))
     }
 
     pub fn remove(&mut self, provider: &str) {

node/src/chain.rs

@@ -124,19 +124,22 @@ pub fn create_substreams_networks(
                     "provider" => &provider.label,
                 );
 
-                let endpoint = FirehoseEndpoint::new(
-                    &provider.label,
-                    &firehose.url,
-                    firehose.token.clone(),
-                    firehose.filters_enabled(),
-                    firehose.compression_enabled(),
-                    firehose.conn_pool_size,
-                );
-
                 let parsed_networks = networks_by_kind
                     .entry(chain.protocol)
                     .or_insert_with(|| FirehoseNetworks::new());
-                parsed_networks.insert(name.to_string(), Arc::new(endpoint));
+
+                for i in 0..firehose.conn_pool_size {
+                    parsed_networks.insert(
+                        name.to_string(),
+                        Arc::new(FirehoseEndpoint::new(
+                            &format!("{}-{}", provider.label, i),
+                            &firehose.url,
+                            firehose.token.clone(),
+                            firehose.filters_enabled(),
+                            firehose.compression_enabled(),
+                        )),
+                    );
+                }
             }
         }
     }
@@ -166,19 +169,21 @@ pub fn create_firehose_networks(
                     "provider" => &provider.label,
                 );
 
-                let endpoint = FirehoseEndpoint::new(
-                    &provider.label,
-                    &firehose.url,
-                    firehose.token.clone(),
-                    firehose.filters_enabled(),
-                    firehose.compression_enabled(),
-                    firehose.conn_pool_size,
-                );
-
                 let parsed_networks = networks_by_kind
                     .entry(chain.protocol)
                     .or_insert_with(|| FirehoseNetworks::new());
-                parsed_networks.insert(name.to_string(), Arc::new(endpoint));
+                for i in 0..firehose.conn_pool_size {
+                    parsed_networks.insert(
+                        name.to_string(),
+                        Arc::new(FirehoseEndpoint::new(
+                            &format!("{}-{}", provider.label, i),
+                            &firehose.url,
+                            firehose.token.clone(),
+                            firehose.filters_enabled(),
+                            firehose.compression_enabled(),
+                        )),
+                    );
+                }
             }
         }
     }

tests/src/fixture/ethereum.rs

@@ -34,7 +34,6 @@ pub async fn chain(blocks: Vec<BlockWithTriggers<Chain>>, stores: &Stores) -> Ch
         None,
         true,
         false,
-        0,
     ))]
     .into();