Automatic yellowstone gRPC deduplication for subscribing to multiple streams

crates/core/Cargo.toml

@@ -38,6 +38,10 @@ thiserror = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
 uuid = { workspace = true, features = ["v4"] }
+ahash = "0.8"
+hashbrown = "0.14"
+lru = "0.12"
+once_cell = "1"
 
 # Optional macro dependencies
 carbon-macros = { workspace = true, optional = true }

crates/core/src/account_utils.rs

@@ -6,21 +6,45 @@ use solana_pubkey::Pubkey;
 /// # Usage
 /// - Use with `?` to indicate the account is required:
 ///   ```
-///   let required = next_account(&mut iter)?;
+///   use carbon_core::account_utils::next_account;
+///   use solana_instruction::AccountMeta;
+///   use solana_pubkey::Pubkey;
+///
+///   let accounts = vec![
+///       AccountMeta::new(Pubkey::new_unique(), false),
+///       AccountMeta::new(Pubkey::new_unique(), false),
+///   ];
+///   let mut iter = accounts.iter();
+///   let required = next_account(&mut iter).ok_or("missing account")?; // propagates None if missing
+///   Ok::<(), &'static str>(())
 ///   ```
-///   This will propagate `None` if the account is missing.
 /// - Use without `?` to handle optional accounts:
 ///   ```
-///   let optional = next_account(&mut iter);
+///   use carbon_core::account_utils::next_account;
+///   use solana_instruction::AccountMeta;
+///   use solana_pubkey::Pubkey;
+///
+///   let accounts = vec![AccountMeta::new(Pubkey::new_unique(), false)];
+///   let mut iter = accounts.iter();
+///   let optional = next_account(&mut iter); // Option<Pubkey>
+///   Ok::<(), &'static str>(())
 ///   ```
-///   This returns `Option<Pubkey>` that you can match or use directly.
 ///
 /// # Example
 /// ```
+/// use carbon_core::account_utils::next_account;
+/// use solana_instruction::AccountMeta;
+/// use solana_pubkey::Pubkey;
+///
+/// let accounts = vec![
+///     AccountMeta::new(Pubkey::new_unique(), false),
+///     AccountMeta::new(Pubkey::new_unique(), false),
+/// ];
 /// let mut iter = accounts.iter();
-/// let required = next_account(&mut iter)?;           // required account
-/// let optional = next_account(&mut iter);            // optional account
-/// `
+/// let required = next_account(&mut iter).ok_or("missing account")?;           // required account
+/// let optional = next_account(&mut iter);                                    // optional account
+/// Ok::<(), &'static str>(())
+/// ```
 pub fn next_account<'a>(iter: &mut impl Iterator<Item = &'a AccountMeta>) -> Option<Pubkey> {
     Some(iter.next()?.pubkey)
 }

crates/core/src/datasource.rs

@@ -194,6 +194,10 @@ impl DatasourceId {
     pub fn new_named(name: &str) -> Self {
         Self(name.to_string())
     }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
 }
 
 /// Represents a data update in the `carbon-core` pipeline, encompassing

crates/core/src/filter.rs

@@ -37,9 +37,13 @@
 //! Custom filter implementation:
 //! ```
 //! use carbon_core::{
-//!     datasource::{DatasourceId, BlockDetails},
+//!     account::AccountMetadata,
+//!     datasource::{AccountDeletion, DatasourceId, BlockDetails},
 //!     filter::Filter,
+//!     instruction::{NestedInstruction, NestedInstructions},
+//!     transaction::TransactionMetadata,
 //! };
+//! use solana_account::Account;
 //!
 //! struct BlockHeightFilter {
 //!     min_height: u64,
@@ -51,14 +55,19 @@
 //!         _datasource_id: &DatasourceId,
 //!         block_details: &BlockDetails,
 //!     ) -> bool {
-//!         block_details.block_height >= self.min_height
+//!         block_details.block_height.unwrap_or(0) >= self.min_height
 //!     }
 //!
 //!     // Implement other methods with default behavior
-//!     fn filter_account(&self, _: &DatasourceId, _: &_, _: &_) -> bool { true }
-//!     fn filter_instruction(&self, _: &DatasourceId, _: &_) -> bool { true }
-//!     fn filter_transaction(&self, _: &DatasourceId, _: &_, _: &_) -> bool { true }
-//!     fn filter_account_deletion(&self, _: &DatasourceId, _: &_) -> bool { true }
+//!     fn filter_account(&self, _: &DatasourceId, _: &AccountMetadata, _: &Account) -> bool { true }
+//!     fn filter_instruction(&self, _: &DatasourceId, _: &NestedInstruction) -> bool { true }
+//!     fn filter_transaction(
+//!         &self,
+//!         _: &DatasourceId,
+//!         _: &TransactionMetadata,
+//!         _: &NestedInstructions,
+//!     ) -> bool { true }
+//!     fn filter_account_deletion(&self, _: &DatasourceId, _: &AccountDeletion) -> bool { true }
 //! }
 //! ```
 
@@ -96,9 +105,13 @@ use crate::{
 /// A simple datasource-based filter:
 /// ```
 /// use carbon_core::{
-///     datasource::{DatasourceId, BlockDetails},
+///     account::AccountMetadata,
+///     datasource::{AccountDeletion, DatasourceId, BlockDetails},
 ///     filter::Filter,
+///     instruction::{NestedInstruction, NestedInstructions},
+///     transaction::TransactionMetadata,
 /// };
+/// use solana_account::Account;
 ///
 /// struct MyFilter {
 ///     allowed_datasource: DatasourceId,
@@ -114,10 +127,15 @@ use crate::{
 ///     }
 ///
 ///     // Implement other methods with default behavior
-///     fn filter_account(&self, _: &DatasourceId, _: &_, _: &_) -> bool { true }
-///     fn filter_instruction(&self, _: &DatasourceId, _: &_) -> bool { true }
-///     fn filter_transaction(&self, _: &DatasourceId, _: &_, _: &_) -> bool { true }
-///     fn filter_account_deletion(&self, _: &DatasourceId, _: &_) -> bool { true }
+///     fn filter_account(&self, _: &DatasourceId, _: &AccountMetadata, _: &Account) -> bool { true }
+///     fn filter_instruction(&self, _: &DatasourceId, _: &NestedInstruction) -> bool { true }
+///     fn filter_transaction(
+///         &self,
+///         _: &DatasourceId,
+///         _: &TransactionMetadata,
+///         _: &NestedInstructions,
+///     ) -> bool { true }
+///     fn filter_account_deletion(&self, _: &DatasourceId, _: &AccountDeletion) -> bool { true }
 /// }
 /// ```
 pub trait Filter {
@@ -260,7 +278,7 @@ pub trait Filter {
 ///
 /// Using with pipeline builders:
 /// ```
-/// use carbon_core::{datasource::DatasourceId, filter::DatasourceFilter};
+/// use carbon_core::{datasource::DatasourceId, filter::{DatasourceFilter, Filter}};
 ///
 /// let filter = DatasourceFilter::new(DatasourceId::new_named("mainnet"));
 /// let filters = vec![Box::new(filter) as Box<dyn Filter>];