Type Source Config, Subscribe to Slots, Fumarole Source, Separate Stream Crate (#111)

---------

Co-authored-by: Kyle Espinola <[email protected]>
Co-authored-by: riprsa <[email protected]>

Author: 3 people

Cargo.lock

@@ -4,9 +4,10 @@ exclude = ["examples/streams-ts/"]
 resolver = "2"
 
 [workspace.dependencies]
-yellowstone-grpc-client = { version = "6.1.0" }
-yellowstone-grpc-proto = { version = "6.1.0", default-features = false }
+yellowstone-grpc-client = { version = "9" }
+yellowstone-grpc-proto = { version = "9", default-features = false }
 yellowstone-vixen = { path = "crates/runtime", version = "0.4.0" }
+yellowstone-vixen-stream = { path = "crates/stream", version = "0.0.0" }
 yellowstone-vixen-core = { path = "crates/core", version = "0.4.0" }
 yellowstone-vixen-mock = { path = "crates/mock", version = "0.4.0" }
 yellowstone-vixen-parser = { path = "crates/parser", version = "0.4.0" }
@@ -33,6 +34,8 @@ yellowstone-vixen-virtuals-parser = { path = "crates/virtuals-parser", version =
 # Sources
 yellowstone-vixen-solana-rpc-source = { path = "crates/solana-rpc-source", version = "0.2.0" }
 yellowstone-vixen-yellowstone-grpc-source = { path = "crates/yellowstone-grpc-source", version = "0.2.0" }
+yellowstone-vixen-yellowstone-fumarole-source = { path = "crates/yellowstone-fumarole-source", version = "0.2.0" }
+yellowstone-vixen-solana-snapshot-source = { path = "crates/solana-snapshot-source", version = "0.2.0" }
 
 
 # helps resolving zeroize dependency conflicts betweeen solana creates depending on curve25519-dalek and yellowstone-grpc 

Cargo.toml

@@ -1,7 +1,8 @@
 # This is an example TOML configuration file for Yellowstone Vixen.
 # Copy to Vixen.toml and them modify as needed.
 
-[yellowstone]
+# Only needed if you are using the Yellowstone gRPC as a source.
+[source]
 # The endpoint of the dragon's mouth stream.
 # This is the URL where the stream can be accessed.
 endpoint = "https://index.rpcpool.com"
@@ -14,6 +15,12 @@ x-token = "<X-TOKEN>"
 # This defines how long to wait for a connection before timing out.
 timeout = 60
 
+# # Only needed if you are using Fumarole as a source.
+# [source]
+# endpoint = "https://index.rpcpool.com"
+# x-token = "<X-TOKEN>"
+# subscriber-name = "my_subscribe_group"
+
 # Metrics configuration section.
 # Uncomment the following lines if you are running Prometheus 
 

Vixen.example.toml

@@ -13,6 +13,8 @@ bs58 = "0.5.1"
 thiserror = "1.0.64"
 yellowstone-grpc-proto = { workspace = true }
 yellowstone-vixen-proto = { workspace = true, optional = true }
+serde = { version = "1.0.198", features = ["derive"] }
+clap = { version = "4.5.4", features = ["derive", "cargo", "wrap_help"] }
 
 [features]
 proto = ["dep:yellowstone-vixen-proto"]

crates/core/Cargo.toml

@@ -200,6 +200,7 @@ impl InstructionUpdate {
             return_data: _,
             return_data_none: _,
             compute_units_consumed,
+            cost_units: _,
         } = meta.ok_or(Missing::TransactionMeta)?;
         let Message {
             header,

crates/core/src/instruction.rs

@@ -24,10 +24,11 @@ use std::{
     sync::Arc,
 };
 
+use serde::Deserialize;
 use yellowstone_grpc_proto::geyser::{
-    CommitmentLevel, SubscribeRequest, SubscribeRequestFilterAccounts,
-    SubscribeRequestFilterBlocksMeta, SubscribeRequestFilterTransactions, SubscribeUpdateAccount,
-    SubscribeUpdateBlockMeta, SubscribeUpdateTransaction,
+    self, SubscribeRequest, SubscribeRequestFilterAccounts, SubscribeRequestFilterBlocksMeta,
+    SubscribeRequestFilterSlots, SubscribeRequestFilterTransactions, SubscribeUpdateAccount,
+    SubscribeUpdateBlockMeta, SubscribeUpdateSlot, SubscribeUpdateTransaction,
 };
 
 pub extern crate bs58;
@@ -65,6 +66,8 @@ pub type AccountUpdate = SubscribeUpdateAccount;
 pub type TransactionUpdate = SubscribeUpdateTransaction;
 /// A block meta update from Yellowstone.
 pub type BlockMetaUpdate = SubscribeUpdateBlockMeta;
+/// A slot update from Yellowstone.
+pub type SlotUpdate = SubscribeUpdateSlot;
 
 /// Generic output type for instruction parsers that wraps shared data for all instructions
 /// in the given transaction.
@@ -154,7 +157,9 @@ pub struct Prefilter {
     /// Filters for transaction updates.
     pub transaction: Option<TransactionPrefilter>,
     /// Filters for block meta updates.
-    pub(crate) block_meta: Option<BlockMetaPrefilter>,
+    pub block_meta: Option<BlockMetaPrefilter>,
+    /// Filters for slot updates.
+    pub slot: Option<SlotPrefilter>,
 }
 
 fn merge_opt<T, F: FnOnce(&mut T, T)>(lhs: &mut Option<T>, rhs: Option<T>, f: F) {
@@ -177,10 +182,12 @@ impl Prefilter {
             account,
             transaction,
             block_meta,
+            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(slot, other.slot, SlotPrefilter::merge);
     }
 }
 
@@ -242,10 +249,23 @@ impl TransactionPrefilter {
     }
 }
 
-#[derive(Debug, Default, Clone, PartialEq)]
-pub(crate) struct BlockMetaPrefilter {}
+/// A prefilter for matching block metadata updates.
+#[derive(Debug, Default, Clone, PartialEq, Copy)]
+pub struct BlockMetaPrefilter {}
 
 impl BlockMetaPrefilter {
+    /// Merge another block metadata prefilter into this one.
+    /// This function currently does nothing as the struct has no fields.
+    pub fn merge(_lhs: &mut Self, _rhs: Self) {}
+}
+
+/// A prefilter for matching slot updates updates.
+#[derive(Debug, Default, Clone, PartialEq, Copy)]
+pub struct SlotPrefilter {}
+
+impl SlotPrefilter {
+    /// Merge another slot prefilter into this one.
+    /// This function currently does nothing as the struct has no fields.
     pub fn merge(_lhs: &mut Self, _rhs: Self) {}
 }
 
@@ -445,6 +465,8 @@ pub enum PrefilterError {
 #[must_use = "Consider calling .build() on this builder"]
 pub struct PrefilterBuilder {
     error: Option<PrefilterError>,
+    slots: bool,
+    block_metas: bool,
     accounts: Option<HashSet<Pubkey>>,
     account_owners: Option<HashSet<Pubkey>>,
     /// Matching [`TransactionPrefilter::accounts_include`]
@@ -484,6 +506,8 @@ impl PrefilterBuilder {
             error,
             accounts,
             account_owners,
+            slots,
+            block_metas,
             transaction_accounts_include,
             transaction_accounts_required,
         } = self;
@@ -503,10 +527,13 @@ impl PrefilterBuilder {
 
         let block_meta = BlockMetaPrefilter {};
 
+        let slot = SlotPrefilter {};
+
         Ok(Prefilter {
             account: (account != AccountPrefilter::default()).then_some(account),
             transaction: (transaction != TransactionPrefilter::default()).then_some(transaction),
-            block_meta: (block_meta != BlockMetaPrefilter::default()).then_some(block_meta),
+            block_meta: block_metas.then_some(block_meta),
+            slot: slots.then_some(slot),
         })
     }
 
@@ -518,6 +545,14 @@ impl PrefilterBuilder {
         self
     }
 
+    /// Set prefilter will request slot updates.
+    pub fn slots(self) -> Self {
+        self.mutate(|this| {
+            this.slots = true;
+            Ok(())
+        })
+    }
+
     /// Set the accounts that this prefilter will match.
     pub fn accounts<I: IntoIterator>(self, it: I) -> Self
     where I::Item: AsRef<[u8]> {
@@ -573,17 +608,6 @@ impl PrefilterBuilder {
 pub struct Filters {
     /// Filters for each parser.
     pub parsers_filters: HashMap<String, Prefilter>,
-    /// Global filters for the subscription.
-    pub global_filters: GlobalFilters,
-}
-
-/// A collection of global filters shared by all parsers for a Vixen subscription.
-#[derive(Debug, Clone, Default, Copy)]
-pub struct GlobalFilters {
-    /// The commitment level for the subscription.
-    pub commitment: Option<CommitmentLevel>,
-    /// The from slot filter for the subscription.
-    pub from_slot: Option<u64>,
 }
 
 impl Filters {
@@ -593,26 +617,30 @@ impl Filters {
     pub fn new(filters: HashMap<String, Prefilter>) -> Self {
         Self {
             parsers_filters: filters,
-            global_filters: GlobalFilters::default(),
         }
     }
+}
 
-    /// Set the commitment level filter.
-    #[inline]
-    #[must_use]
-    pub fn commitment(mut self, commitment: Option<CommitmentLevel>) -> Self {
-        self.global_filters.commitment = commitment;
-
-        self
-    }
-
-    /// Set the from slot filter.
-    #[inline]
-    #[must_use]
-    pub fn from_slot(mut self, from_slot: Option<u64>) -> Self {
-        self.global_filters.from_slot = from_slot;
-
-        self
+/// Type mirroring the `CommitmentLevel` enum in the `geyser` crate but serializable.
+/// Used to avoid need for custom deserialization logic.
+#[derive(Debug, Clone, Copy, Deserialize, clap::ValueEnum)]
+#[serde(rename_all = "lowercase")]
+pub enum CommitmentLevel {
+    /// Processed
+    Processed,
+    /// Confirmed
+    Confirmed,
+    /// Finalized
+    Finalized,
+}
+
+impl From<geyser::CommitmentLevel> for CommitmentLevel {
+    fn from(value: geyser::CommitmentLevel) -> Self {
+        match value {
+            geyser::CommitmentLevel::Processed => Self::Processed,
+            geyser::CommitmentLevel::Confirmed => Self::Confirmed,
+            geyser::CommitmentLevel::Finalized => Self::Finalized,
+        }
     }
 }
 
@@ -635,7 +663,16 @@ impl From<Filters> for SubscribeRequest {
                     }))
                 })
                 .collect(),
-            slots: [].into_iter().collect(),
+            slots: value
+                .parsers_filters
+                .keys()
+                .map(|k| {
+                    (k.clone(), SubscribeRequestFilterSlots {
+                        filter_by_commitment: Some(true),
+                        interslot_updates: None,
+                    })
+                })
+                .collect(),
             transactions: value
                 .parsers_filters
                 .iter()
@@ -669,13 +706,10 @@ impl From<Filters> for SubscribeRequest {
                 .map(|k| (k.clone(), SubscribeRequestFilterBlocksMeta {}))
                 .collect(),
             entry: [].into_iter().collect(),
-            commitment: value
-                .global_filters
-                .commitment
-                .map(|commitment| commitment as i32),
+            commitment: None,
             accounts_data_slice: vec![],
             ping: None,
-            from_slot: value.global_filters.from_slot,
+            from_slot: None,
         }
     }
 }

crates/core/src/lib.rs

@@ -34,6 +34,7 @@ tokio = "1.39.2"
 [features]
 default = []
 block-meta = []
+slot = []
 proto = [
   "dep:yellowstone-vixen-proto",
   "yellowstone-vixen-core/proto",

crates/parser/Cargo.toml

@@ -18,6 +18,9 @@ mod helpers;
 #[cfg(feature = "block-meta")]
 pub mod block_meta;
 
+#[cfg(feature = "slot")]
+pub mod slot;
+
 #[cfg(feature = "token-extensions")]
 pub mod token_extension_program;
 #[cfg(feature = "token-program")]

crates/parser/src/lib.rs

@@ -0,0 +1,28 @@
+use std::borrow::Cow;
+
+use yellowstone_vixen_core::{ParseResult, Parser, Prefilter, ProgramParser, Pubkey, SlotUpdate};
+
+#[derive(Debug, Clone, Copy)]
+pub struct SlotParser;
+
+impl Parser for SlotParser {
+    type Input = SlotUpdate;
+    type Output = SlotUpdate;
+
+    fn id(&self) -> Cow<str> { "yellowstone::SlotParser".into() }
+
+    fn prefilter(&self) -> Prefilter { Prefilter::builder().slots().build().unwrap() }
+
+    async fn parse(&self, slot: &SlotUpdate) -> ParseResult<Self::Output> { Ok(slot.to_owned()) }
+}
+
+impl ProgramParser for SlotParser {
+    /// "S111111111111111111111111111111111111111112"
+    #[inline]
+    fn program_id(&self) -> Pubkey {
+        Pubkey::new([
+            83, 144, 207, 224, 10, 18, 54, 164, 34, 33, 252, 0, 121, 244, 178, 50, 161, 204, 101,
+            119, 224, 0, 123, 66, 92, 89, 105, 1, 1, 1, 1, 2,
+        ])
+    }
+}

crates/parser/src/slot.rs

@@ -18,8 +18,8 @@ clap = { version = "4.5.4", default-features = false, features = [
 futures-channel = { version = "0.3.30", features = ["sink"] }
 futures-util = { version = "0.3.30", features = ["sink"] }
 opentelemetry = { version = "0.24.0", features = ["metrics"], optional = true }
-pin-project-lite = { version = "0.2.14", optional = true }
 prometheus = { version = "0.14.0", features = ["push"], optional = true }
+toml = "0.8.12"
 serde = { version = "1.0.198", features = ["derive"] }
 smallvec = "1.13.2"
 thiserror = "1.0.64"
@@ -29,13 +29,7 @@ tracing = "0.1.40"
 yellowstone-grpc-client = { workspace = true }
 yellowstone-grpc-proto = { workspace = true }
 yellowstone-vixen-core = { workspace = true }
-yellowstone-vixen-proto = { workspace = true, optional = true }
 
 [features]
 opentelemetry = ["dep:opentelemetry"]
 prometheus = ["dep:prometheus"]
-stream = [
-  "dep:pin-project-lite",
-  "dep:yellowstone-vixen-proto",
-  "yellowstone-vixen-proto/stream",
-]