Reduce network by only fetching block meta instead of whole block

cluster-endpoints/src/grpc_multiplex.rs

@@ -1,13 +1,13 @@
 use crate::grpc_stream_utils::channelize_stream;
 use crate::grpc_subscription::map_block_update;
+use futures::StreamExt;
 use geyser_grpc_connector::grpc_subscription_autoreconnect::{
-    create_geyser_reconnecting_stream, GeyserFilter, GrpcConnectionTimeouts, GrpcSourceConfig,
+    create_geyser_reconnecting_stream, GeyserFilter, GrpcSourceConfig,
 };
 use geyser_grpc_connector::grpcmultiplex_fastestwins::{
     create_multiplexed_stream, FromYellowstoneExtractor,
 };
-use log::info;
-use merge_streams::MergeStreams;
+use log::{debug, info, trace, warn};
 use solana_lite_rpc_core::structures::produced_block::ProducedBlock;
 use solana_lite_rpc_core::structures::slot_notification::SlotNotification;
 use solana_lite_rpc_core::AnyhowJoinHandle;
@@ -36,6 +36,21 @@ impl FromYellowstoneExtractor for BlockExtractor {
     }
 }
 
+struct BlockMetaHashExtractor(CommitmentConfig);
+
+impl FromYellowstoneExtractor for BlockMetaHashExtractor {
+    type Target = String;
+    fn map_yellowstone_update(&self, update: SubscribeUpdate) -> Option<(u64, String)> {
+        match update.update_oneof {
+            Some(UpdateOneof::BlockMeta(block_meta)) => {
+                Some((block_meta.slot, block_meta.blockhash))
+            }
+            _ => None,
+        }
+    }
+}
+
+/// connect to multiple grpc sources to consume confirmed blocks and block status update
 pub fn create_grpc_multiplex_blocks_subscription(
     grpc_sources: Vec<GrpcSourceConfig>,
 ) -> (Receiver<ProducedBlock>, AnyhowJoinHandle) {
@@ -47,13 +62,7 @@ pub fn create_grpc_multiplex_blocks_subscription(
         info!("- connection to {}", grpc_source);
     }
 
-    let _timeouts = GrpcConnectionTimeouts {
-        connect_timeout: Duration::from_secs(5),
-        request_timeout: Duration::from_secs(5),
-        subscribe_timeout: Duration::from_secs(5),
-    };
-
-    let multiplex_stream_confirmed = {
+    let confirmed_blocks_stream = {
         let commitment_config = CommitmentConfig::confirmed();
 
         let mut streams = Vec::new();
@@ -68,28 +77,77 @@ pub fn create_grpc_multiplex_blocks_subscription(
         create_multiplexed_stream(streams, BlockExtractor(commitment_config))
     };
 
-    let multiplex_stream_finalized = {
+    let finalized_blockmeta_stream = {
         let commitment_config = CommitmentConfig::finalized();
 
         let mut streams = Vec::new();
         for grpc_source in &grpc_sources {
             let stream = create_geyser_reconnecting_stream(
                 grpc_source.clone(),
-                GeyserFilter(commitment_config).blocks_and_txs(),
+                GeyserFilter(commitment_config).blocks_meta(),
             );
             streams.push(stream);
         }
-
-        create_multiplexed_stream(streams, BlockExtractor(commitment_config))
+        create_multiplexed_stream(streams, BlockMetaHashExtractor(commitment_config))
     };
 
-    let merged_stream_confirmed_finalize =
-        (multiplex_stream_confirmed, multiplex_stream_finalized).merge();
-
-    let (multiplexed_finalized_blocks, jh_channelizer) =
-        channelize_stream(merged_stream_confirmed_finalize);
+    // return value is the broadcast receiver
+    let (producedblock_sender, blocks_output_stream) =
+        tokio::sync::broadcast::channel::<ProducedBlock>(1000);
+
+    let jh_block_emitter_task = {
+        tokio::task::spawn(async move {
+            // by blockhash
+            let mut recent_confirmed_blocks = HashMap::<String, ProducedBlock>::new();
+            let mut confirmed_blocks_stream = std::pin::pin!(confirmed_blocks_stream);
+            let mut finalized_blockmeta_stream = std::pin::pin!(finalized_blockmeta_stream);
+
+            let sender = producedblock_sender;
+            let mut cleanup_tick = tokio::time::interval(Duration::from_secs(5));
+            let mut last_finalized_slot: Slot = 0;
+            loop {
+                tokio::select! {
+                    confirmed_block = confirmed_blocks_stream.next() => {
+                        let confirmed_block = confirmed_block.expect("confirmed block from stream");
+                        trace!("got confirmed block {} with blockhash {}",
+                            confirmed_block.slot, confirmed_block.blockhash.clone());
+                        if let Err(e) = sender.send(confirmed_block.clone()) {
+                            warn!("Confirmed block channel has no receivers {e:?}");
+                            continue
+                        }
+                        recent_confirmed_blocks.insert(confirmed_block.blockhash.clone(), confirmed_block);
+                    },
+                    meta_finalized = finalized_blockmeta_stream.next() => {
+                        let blockhash = meta_finalized.expect("finalized block meta from stream");
+                        if let Some(cached_confirmed_block) = recent_confirmed_blocks.remove(&blockhash) {
+                            let finalized_block = cached_confirmed_block.to_finalized_block();
+                            last_finalized_slot = finalized_block.slot;
+                            debug!("got finalized blockmeta {} with blockhash {}",
+                                finalized_block.slot, finalized_block.blockhash.clone());
+                            if let Err(e) = sender.send(finalized_block) {
+                                warn!("Finalized block channel has no receivers {e:?}");
+                                continue;
+                            }
+                        } else {
+                            debug!("finalized block meta received for blockhash {} which was never seen or already emitted", blockhash);
+                        }
+                    },
+                    _ = cleanup_tick.tick() => {
+                        let size_before = recent_confirmed_blocks.len();
+                        recent_confirmed_blocks.retain(|_blockhash, block| {
+                            last_finalized_slot == 0 || block.slot > last_finalized_slot - 100
+                        });
+                        let cnt_cleaned = size_before - recent_confirmed_blocks.len();
+                        if cnt_cleaned > 0 {
+                            debug!("cleaned {} confirmed blocks from cache", cnt_cleaned);
+                        }
+                    }
+                }
+            }
+        })
+    };
 
-    (multiplexed_finalized_blocks, jh_channelizer)
+    (blocks_output_stream, jh_block_emitter_task)
 }
 
 struct SlotExtractor {}

core/src/structures/produced_block.rs

@@ -165,4 +165,12 @@ impl ProducedBlock {
             rewards,
         }
     }
+
+    /// moving commitment level to finalized
+    pub fn to_finalized_block(&self) -> Self {
+        ProducedBlock {
+            commitment_config: CommitmentConfig::finalized(),
+            ..self.clone()
+        }
+    }
 }