feat : add block subscriber to vixen and fix few bugs

Author: Nagaprasadvr

crates/core/src/lib.rs

@@ -26,9 +26,11 @@ use std::{
 
 use serde::Deserialize;
 use yellowstone_grpc_proto::geyser::{
-    self, SubscribeRequest, SubscribeRequestFilterAccounts, SubscribeRequestFilterBlocksMeta,
-    SubscribeRequestFilterSlots, SubscribeRequestFilterTransactions, SubscribeUpdateAccount,
-    SubscribeUpdateBlockMeta, SubscribeUpdateSlot, SubscribeUpdateTransaction,
+    self, SubscribeRequest, SubscribeRequestFilterAccounts, SubscribeRequestFilterBlocks,
+    SubscribeRequestFilterBlocksMeta, SubscribeRequestFilterSlots,
+    SubscribeRequestFilterTransactions, SubscribeUpdateAccount, SubscribeUpdateAccountInfo,
+    SubscribeUpdateBlock, SubscribeUpdateBlockMeta, SubscribeUpdateSlot,
+    SubscribeUpdateTransaction,
 };
 
 pub extern crate bs58;
@@ -62,10 +64,14 @@ pub type ParseResult<T> = Result<T, ParseError>;
 
 /// An account update from Yellowstone.
 pub type AccountUpdate = SubscribeUpdateAccount;
+/// An account update from Yellowstone.
+pub type AccountUpdateInfo = SubscribeUpdateAccountInfo;
 /// A transaction update from Yellowstone.
 pub type TransactionUpdate = SubscribeUpdateTransaction;
 /// A block meta update from Yellowstone.
 pub type BlockMetaUpdate = SubscribeUpdateBlockMeta;
+/// A block update from Yellowstone.
+pub type BlockUpdate = SubscribeUpdateBlock;
 /// A slot update from Yellowstone.
 pub type SlotUpdate = SubscribeUpdateSlot;
 
@@ -158,6 +164,8 @@ pub struct Prefilter {
     pub transaction: Option<TransactionPrefilter>,
     /// Filters for block meta updates.
     pub block_meta: Option<BlockMetaPrefilter>,
+    /// Filters for block updates.
+    pub block: Option<BlockPrefilter>,
     /// Filters for slot updates.
     pub slot: Option<SlotPrefilter>,
 }
@@ -182,11 +190,13 @@ impl Prefilter {
             account,
             transaction,
             block_meta,
+            block,
             slot,
         } = self;
         merge_opt(account, other.account, AccountPrefilter::merge);
         merge_opt(transaction, other.transaction, TransactionPrefilter::merge);
         merge_opt(block_meta, other.block_meta, BlockMetaPrefilter::merge);
+        merge_opt(block, other.block, BlockPrefilter::merge);
         merge_opt(slot, other.slot, SlotPrefilter::merge);
     }
 }
@@ -259,6 +269,36 @@ impl BlockMetaPrefilter {
     pub fn merge(_lhs: &mut Self, _rhs: Self) {}
 }
 
+/// A prefilter for matching block updates.
+#[derive(Debug, Default, Clone, PartialEq)]
+pub struct BlockPrefilter {
+    /// filter transactions and accounts that use any account from the list
+    pub accounts_include: HashSet<Pubkey>,
+    /// include all transactions
+    pub include_transactions: bool,
+    /// include all account updates
+    pub include_accounts: bool,
+    /// include all entries
+    pub include_entries: bool,
+}
+
+impl BlockPrefilter {
+    /// Merge another block prefilter into this one.
+    pub fn merge(&mut self, other: BlockPrefilter) {
+        let Self {
+            accounts_include,
+            include_transactions,
+            include_accounts,
+            include_entries,
+        } = self;
+
+        accounts_include.extend(other.accounts_include);
+        *include_accounts = other.include_accounts;
+        *include_transactions = other.include_transactions;
+        *include_entries = other.include_entries;
+    }
+}
+
 /// A prefilter for matching slot updates updates.
 #[derive(Debug, Default, Clone, PartialEq, Copy)]
 pub struct SlotPrefilter {}
@@ -467,7 +507,19 @@ pub struct PrefilterBuilder {
     error: Option<PrefilterError>,
     slots: bool,
     block_metas: bool,
+    /// Matching [`BlockPrefilter::accounts`]
+    block_accounts_include: Option<HashSet<Pubkey>>,
+    /// Matching [`BlockPrefilter::include_accounts`]
+    block_include_accounts: bool,
+    /// Matching [`BlockPrefilter::include_transactions`]
+    block_include_transactions: bool,
+    /// Matching [`BlockPrefilter::include_entries`]
+    block_include_entries: bool,
+    /// Including all accounts  
+    accounts_include_all: bool,
+    /// Matching [`AccountPrefilter::accounts`]
     accounts: Option<HashSet<Pubkey>>,
+    /// Matching [`AccountPrefilter::account_owners`]
     account_owners: Option<HashSet<Pubkey>>,
     /// Matching [`TransactionPrefilter::accounts_include`]
     transaction_accounts_include: Option<HashSet<Pubkey>>,
@@ -504,21 +556,26 @@ impl PrefilterBuilder {
     pub fn build(self) -> Result<Prefilter, PrefilterError> {
         let PrefilterBuilder {
             error,
+            accounts_include_all,
             accounts,
             account_owners,
             slots,
             block_metas,
+            block_accounts_include,
+            block_include_accounts,
+            block_include_entries,
+            block_include_transactions,
             transaction_accounts_include,
             transaction_accounts_required,
         } = self;
         if let Some(err) = error {
             return Err(err);
         }
 
-        let account = account_owners.map(|owners| AccountPrefilter {
+        let account = AccountPrefilter {
             accounts: accounts.unwrap_or_default(),
-            owners,
-        });
+            owners: account_owners.unwrap_or_default(),
+        };
 
         let transaction = TransactionPrefilter {
             accounts_include: transaction_accounts_include.unwrap_or_default(),
@@ -527,12 +584,26 @@ impl PrefilterBuilder {
 
         let block_meta = BlockMetaPrefilter {};
 
+        let block = BlockPrefilter {
+            accounts_include: block_accounts_include.unwrap_or_default(),
+            include_accounts: block_include_accounts,
+            include_transactions: block_include_transactions,
+            include_entries: block_include_entries,
+        };
+
         let slot = SlotPrefilter {};
 
+        let account = if accounts_include_all {
+            Some(AccountPrefilter::default())
+        } else {
+            (account != AccountPrefilter::default()).then_some(account)
+        };
+
         Ok(Prefilter {
             account,
             transaction: (transaction != TransactionPrefilter::default()).then_some(transaction),
             block_meta: block_metas.then_some(block_meta),
+            block: (block != BlockPrefilter::default()).then_some(block),
             slot: slots.then_some(slot),
         })
     }
@@ -553,6 +624,22 @@ impl PrefilterBuilder {
         })
     }
 
+    /// Set prefilter will request block_metas updates.
+    pub fn block_metas(self) -> Self {
+        self.mutate(|this| {
+            this.block_metas = true;
+            Ok(())
+        })
+    }
+
+    /// Set accounts_include_all filter
+    pub fn accounts_include_all(self) -> Self {
+        self.mutate(|this| {
+            this.accounts_include_all = true;
+            Ok(())
+        })
+    }
+
     /// Set the accounts that this prefilter will match.
     pub fn accounts<I: IntoIterator>(self, it: I) -> Self
     where I::Item: AsRef<[u8]> {
@@ -601,6 +688,42 @@ impl PrefilterBuilder {
             )
         })
     }
+
+    /// Set the included accounts for this block prefilter.
+    pub fn block_accounts_include<I: IntoIterator>(self, it: I) -> Self
+    where I::Item: AsRef<[u8]> {
+        self.mutate(|this| {
+            set_opt(
+                &mut this.block_accounts_include,
+                "block_accounts_include",
+                collect_pubkeys(it)?,
+            )
+        })
+    }
+
+    /// Set the include_accounts flag for this block prefilter.
+    pub fn block_include_accounts(self) -> Self {
+        self.mutate(|this| {
+            this.block_include_accounts = true;
+            Ok(())
+        })
+    }
+
+    /// Set the include_transactions flag for this block prefilter.
+    pub fn block_include_transactions(self) -> Self {
+        self.mutate(|this| {
+            this.block_include_transactions = true;
+            Ok(())
+        })
+    }
+
+    /// Set the include_entries flag for this block prefilter.
+    pub fn block_include_entries(self) -> Self {
+        self.mutate(|this| {
+            this.block_include_entries = true;
+            Ok(())
+        })
+    }
 }
 
 /// A collection of filters for a Vixen subscription.
@@ -665,12 +788,13 @@ impl From<Filters> for SubscribeRequest {
                 .collect(),
             slots: value
                 .parsers_filters
-                .keys()
-                .map(|k| {
-                    (k.clone(), SubscribeRequestFilterSlots {
+                .iter()
+                .filter_map(|(k, v)| {
+                    v.slot?;
+                    Some((k.clone(), SubscribeRequestFilterSlots {
                         filter_by_commitment: Some(true),
                         interslot_updates: None,
-                    })
+                    }))
                 })
                 .collect(),
             transactions: value
@@ -699,11 +823,31 @@ impl From<Filters> for SubscribeRequest {
                 })
                 .collect(),
             transactions_status: [].into_iter().collect(),
-            blocks: [].into_iter().collect(),
+            blocks: value
+                .parsers_filters
+                .iter()
+                .filter_map(|(k, v)| {
+                    let v = v.block.as_ref()?;
+
+                    Some((k.clone(), SubscribeRequestFilterBlocks {
+                        account_include: v
+                            .accounts_include
+                            .iter()
+                            .map(ToString::to_string)
+                            .collect(),
+                        include_transactions: Some(v.include_transactions),
+                        include_accounts: Some(v.include_accounts),
+                        include_entries: Some(v.include_entries),
+                    }))
+                })
+                .collect(),
             blocks_meta: value
                 .parsers_filters
-                .keys()
-                .map(|k| (k.clone(), SubscribeRequestFilterBlocksMeta {}))
+                .iter()
+                .filter_map(|(k, v)| {
+                    v.block_meta?;
+                    Some((k.clone(), SubscribeRequestFilterBlocksMeta {}))
+                })
                 .collect(),
             entry: [].into_iter().collect(),
             commitment: None,

crates/parser/src/block_meta.rs

@@ -1,10 +1,9 @@
 use std::borrow::Cow;
 
-use yellowstone_vixen_core::{
-    BlockMetaUpdate, ParseResult, Parser, Prefilter, ProgramParser, Pubkey,
-};
+use yellowstone_grpc_proto::geyser::SubscribeUpdateBlockMeta;
+use yellowstone_vixen_core::{ParseResult, Parser, Prefilter, ProgramParser, Pubkey};
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct Reward {
     pub pubkey: String,
     pub lamports: i64,
@@ -20,7 +19,7 @@ pub struct Rewards {
 }
 
 #[derive(Debug, Clone)]
-pub struct BlockUpdate {
+pub struct BlockMetaUpdate {
     pub slot: u64,
     pub blockhash: String,
     pub rewards: Option<Rewards>,
@@ -36,14 +35,14 @@ pub struct BlockUpdate {
 pub struct BlockMetaParser;
 
 impl Parser for BlockMetaParser {
-    type Input = BlockMetaUpdate;
-    type Output = BlockUpdate;
+    type Input = SubscribeUpdateBlockMeta;
+    type Output = BlockMetaUpdate;
 
     fn id(&self) -> Cow<'static, str> { "yellowstone::BlockMetaParser".into() }
 
-    fn prefilter(&self) -> Prefilter { Prefilter::builder().build().unwrap() }
+    fn prefilter(&self) -> Prefilter { Prefilter::builder().block_metas().build().unwrap() }
 
-    async fn parse(&self, block_meta: &BlockMetaUpdate) -> ParseResult<Self::Output> {
+    async fn parse(&self, block_meta: &SubscribeUpdateBlockMeta) -> ParseResult<Self::Output> {
         let rewards = block_meta.rewards.as_ref().map(|reward| Rewards {
             rewards: reward
                 .rewards
@@ -59,7 +58,7 @@ impl Parser for BlockMetaParser {
             num_partitions: reward.num_partitions.map(|num| num.num_partitions),
         });
 
-        Ok(BlockUpdate {
+        Ok(BlockMetaUpdate {
             slot: block_meta.slot,
             blockhash: block_meta.blockhash.clone(),
             rewards,

crates/runtime/src/buffer.rs

@@ -118,6 +118,18 @@ impl<H: Send> topograph::AsyncHandler<Job, H> for Handler {
                     )
                     .await;
             },
+            UpdateOneof::Block(b) => {
+                pipelines
+                    .block
+                    .get_handlers(&filters)
+                    .run(
+                        span,
+                        &b,
+                        #[cfg(feature = "prometheus")]
+                        update_type,
+                    )
+                    .await;
+            },
             UpdateOneof::Slot(s) => {
                 pipelines
                     .slot