Filters reference

Filter, FilterContext, FilterResult and the built-in DeduplicationFilter, DatasourceFilter and SlotRangeFilter in Carbon 2.0.0, with exact semantics.

Filters decide, per pipe, which updates reach the processor. This page documents the carbon_core::filter module in 2.0.0: the trait, its context and result types, and the three built-in filters, with the exact rules each one applies. For when to use filters, see Filters.

Source: crates/core/src/filter.rs.

How filters are evaluated#

You attach filters to one pipe with the *_with_filters builder methods. Each pipe owns its filters; two pipes never share filter state, even if you create them with the same arguments.
For each update and pipe, the pipeline calls the hook that matches the pipe kind on each filter in order, and stops at the first Reject: the filters are combined with AND. A rejected update is skipped for that pipe only.
Hooks run before decoding. Instruction filters see every instruction of every transaction, including instructions of other programs.
Because evaluation stops at the first Reject, a stateful filter later in the list only sees updates the earlier filters accepted. Put DeduplicationFilter last.

Pipe registered with	Hook called	Called once per
`account_with_filters`	`filter_account`	account update
`instruction_with_filters`	`filter_instruction`	instruction (top-level and inner)
`transaction_with_filters`	`filter_transaction`	transaction
`account_deletions_with_filters`	`filter_account_deletion`	account deletion
`block_details_with_filters`	`filter_block_details`	block

Built-in filters at a glance#

Filter	Account	Instruction	Transaction	Deletion	Block details
`DeduplicationFilter`	Deduplicated by `(signature, pubkey)`	Deduplicated by `(signature, absolute_path)`	Always accepted	Always accepted	Always accepted
`DatasourceFilter`	Checked	Checked	Checked	Checked	Checked
`SlotRangeFilter`	Slot only	Slot and transaction index	Slot and transaction index	Slot only	Slot only

`Filter`#

The trait every filter implements.

Rust

pub trait Filter: Send + Sync {
    fn filter_account(
        &self,
        _context: &FilterContext,
        _account_metadata: &AccountMetadata,
        _account: &solana_account::Account,
    ) -> FilterResult {
        FilterResult::Accept
    }

    fn filter_instruction(
        &self,
        _context: &FilterContext,
        _nested_instruction: &NestedInstruction,
    ) -> FilterResult {
        FilterResult::Accept
    }

    fn filter_transaction(
        &self,
        _context: &FilterContext,
        _transaction_metadata: &TransactionMetadata,
        _nested_instructions: &NestedInstructions,
    ) -> FilterResult {
        FilterResult::Accept
    }

    fn filter_account_deletion(
        &self,
        _context: &FilterContext,
        _account_deletion: &AccountDeletion,
    ) -> FilterResult {
        FilterResult::Accept
    }

    fn filter_block_details(
        &self,
        _context: &FilterContext,
        _block_details: &BlockDetails,
    ) -> FilterResult {
        FilterResult::Accept
    }
}

Details.

Hook	Receives
`filter_account`	`AccountMetadata` and the raw `Account`
`filter_instruction`	The `NestedInstruction`: metadata, raw instruction, and its CPIs
`filter_transaction`	`TransactionMetadata` and the whole transaction's `NestedInstructions` tree
`filter_account_deletion`	The `AccountDeletion`
`filter_block_details`	The `BlockDetails`

All hooks default to Accept. Hooks take &self, so keep state behind atomics or locks. Pipes store filters as Box<dyn Filter + 'static>, so build the list as Vec<Box<dyn Filter>>.

Example.

Rust

use {
    carbon_core::{
        account::AccountMetadata,
        filter::{Filter, FilterContext, FilterResult},
    },
    solana_account::Account,
};

/// Accepts accounts whose data is at least `min_len` bytes.
pub struct MinDataLen(pub usize);

impl Filter for MinDataLen {
    fn filter_account(
        &self,
        _context: &FilterContext,
        _account_metadata: &AccountMetadata,
        account: &Account,
    ) -> FilterResult {
        if account.data.len() >= self.0 {
            FilterResult::Accept
        } else {
            FilterResult::Reject
        }
    }
}

See also. Filter in traits, docs.rs.

`FilterContext`#

Read-only context passed to every hook.

Rust

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FilterContext<'a> {
    pub datasource_id: &'a DatasourceId,
}

Details.

Field	Type	Meaning
`datasource_id`	`&DatasourceId`	ID of the datasource that sent the update: the one you passed to `datasource_with_id`, or a random UUID for `datasource`.

Example.

Rust

fn filter_block_details(&self, context: &FilterContext, _block: &BlockDetails) -> FilterResult {
    if *context.datasource_id == DatasourceId::new_named("primary") {
        FilterResult::Accept
    } else {
        FilterResult::Reject
    }
}

See also. DatasourceId, docs.rs.

`FilterResult`#

A hook's decision.

Rust

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FilterResult {
    Accept,
    Reject,
}

Details. Accept passes the update to the next filter, or to the pipe if this was the last one. Reject skips the update for this pipe; other pipes are unaffected.

Example.

Rust

let result = if slot >= 400_000_000 { FilterResult::Accept } else { FilterResult::Reject };

See also. docs.rs.

`DeduplicationFilter`#

Drops instructions and accounts already seen within a time window.

Rust

pub struct DeduplicationFilter { /* private fields */ }

impl DeduplicationFilter {
    pub fn new(ttl: Duration) -> Self;
    pub fn cleanup_expired(&self);
}

Details. Use it when two datasources deliver the same data, for example two Yellowstone endpoints for redundancy, or a crawler that overlaps a stream.

Update	Key	Behaviour
Instruction	`(transaction signature, absolute_path)`	First occurrence is accepted and recorded; any later occurrence of the same key is rejected while the entry is stored.
Account	`(transaction_signature, pubkey)`	Same, but only when `AccountMetadata::transaction_signature` is `Some`. Accounts without a signature (RPC GPA, program subscribe, snapshots, Atlas) are always accepted.
Transaction, deletion, block details	none	Always accepted. Transaction pipes are not deduplicated.

The key does not include the datasource or the slot, so the same instruction from two different sources is a duplicate, which is the point. Entries record when they were first seen; a duplicate does not refresh the timestamp.

TTL and memory. Expired entries are removed lazily: on each hook call, if at least 60 seconds have passed since the last cleanup, the filter removes every entry older than ttl. An entry therefore lives between ttl and ttl + 60 s, or longer if no updates arrive. Memory grows with the number of distinct keys seen in that window: every instruction the pipe sees is recorded, including instructions of other programs (the filter runs before the decoder). A duplicate arriving after its entry was removed is accepted again. cleanup_expired runs the cleanup immediately.

If the internal lock is poisoned, the filter accepts everything instead of panicking.

Example.

Rust

let filters: Vec<Box<dyn Filter>> = vec![
    Box::new(DeduplicationFilter::new(Duration::from_secs(300))),
];
Pipeline::builder()
    .datasource_with_id(primary, DatasourceId::new_named("primary"))
    .datasource_with_id(backup, DatasourceId::new_named("backup"))
    .instruction_with_filters(MeteoraDlmmDecoder, SwapLogger, filters)

See also. Multiple datasources, docs.rs.

`DatasourceFilter`#

Accepts only updates from the listed datasources.

Rust

pub struct DatasourceFilter {
    pub allowed_datasources: Vec<DatasourceId>,
}

impl DatasourceFilter {
    pub fn new(datasource_id: DatasourceId) -> Self;
    pub fn new_many(datasource_ids: Vec<DatasourceId>) -> Self;
}

Details. Implements all five hooks: an update is accepted when context.datasource_id is in allowed_datasources (compared with ==), rejected otherwise. new allows one ID, new_many several. An empty list rejects everything. Register the datasources with datasource_with_id; IDs from datasource are random and can't be matched.

Example.

Rust

let geyser = DatasourceId::new_named("geyser");
let snapshot = DatasourceId::new_named("snapshot");

Pipeline::builder()
    .datasource_with_id(geyser_client, geyser.clone())
    .datasource_with_id(snapshot_loader, snapshot.clone())
    .account_with_filters(
        MeteoraDlmmDecoder,
        PoolLogger,
        vec![Box::new(DatasourceFilter::new_many(vec![geyser, snapshot]))],
    )

See also. Multiple datasources, docs.rs.

`SlotRangeFilter`#

Accepts only updates inside a slot range, optionally refined by transaction index.

Rust

#[derive(Debug, Clone)]
pub struct SlotRangeFilter { /* private fields */ }

impl SlotRangeFilter {
    pub fn from(slot: u64, transaction_index: Option<u64>) -> Self;
    pub fn to(slot: u64, transaction_index: Option<u64>) -> Self;
    pub fn between(
        from_slot: u64,
        from_transaction_index: Option<u64>,
        to_slot: u64,
        to_transaction_index: Option<u64>,
    ) -> Self;
    pub fn contains(&self, slot: u64, index: Option<u64>) -> bool;
}

Details. from sets only a lower bound, to only an upper bound, between both. Every hook calls contains(slot, index):

Update	`slot`	`index`
Instruction	`transaction_metadata.slot`	`transaction_metadata.index`
Transaction	`TransactionMetadata::slot`	`TransactionMetadata::index`
Account, deletion, block details	the update's `slot`	`None`

contains applies these rules:

Bound	Rejects when
Lower (`from_slot`, `from_transaction_index`)	`slot < from_slot`, or `slot == from_slot` and both indexes are `Some` and `index < from_transaction_index`
Upper (`to_slot`, `to_transaction_index`)	`slot > to_slot`, or `slot == to_slot` and both indexes are `Some` and `index >= to_transaction_index`

So the lower bound is inclusive. The upper slot is inclusive unless you give a to_transaction_index and the update has an index, in which case the range stops just before that transaction. When the update has no index (accounts, or transactions from the RPC transaction crawler and Helius Atlas), the index bounds are ignored and only slots are compared. The source code comment describes the range as half-open [from, to); that is only true at the transaction-index level.

Example. This program prints how contains treats boundary cases:

src/main.rs

use carbon_core::filter::SlotRangeFilter;

fn main() {
    let to = SlotRangeFilter::to(100, None);
    let to_idx = SlotRangeFilter::to(100, Some(5));
    let from_idx = SlotRangeFilter::from(100, Some(5));
    let between = SlotRangeFilter::between(100, None, 200, None);

    println!("to(100, None)        slot 100      -> {}", to.contains(100, None));
    println!("to(100, None)        slot 101      -> {}", to.contains(101, None));
    println!("to(100, Some(5))     slot 100 idx 4 -> {}", to_idx.contains(100, Some(4)));
    println!("to(100, Some(5))     slot 100 idx 5 -> {}", to_idx.contains(100, Some(5)));
    println!("to(100, Some(5))     slot 100 None -> {}", to_idx.contains(100, None));
    println!("from(100, Some(5))   slot 100 idx 4 -> {}", from_idx.contains(100, Some(4)));
    println!("from(100, Some(5))   slot 100 None -> {}", from_idx.contains(100, None));
    println!("between(100, 200)    slot 200      -> {}", between.contains(200, None));
}

Terminal

cargo run

Output

to(100, None)        slot 100      -> true
to(100, None)        slot 101      -> false
to(100, Some(5))     slot 100 idx 4 -> true
to(100, Some(5))     slot 100 idx 5 -> false
to(100, Some(5))     slot 100 None -> true
from(100, Some(5))   slot 100 idx 4 -> false
from(100, Some(5))   slot 100 None -> true
between(100, 200)    slot 200      -> true

To register it on a pipe:

Rust

Pipeline::builder().instruction_with_filters(
    MeteoraDlmmDecoder,
    SwapLogger,
    vec![Box::new(SlotRangeFilter::between(400_000_000, None, 400_100_000, None))],
)

See also. Backfill transactions, docs.rs.

How filters are evaluated#

Built-in filters at a glance#

Filter#

FilterContext#

FilterResult#

DeduplicationFilter#

DatasourceFilter#

SlotRangeFilter#

`Filter`#

`FilterContext`#

`FilterResult`#

`DeduplicationFilter`#

`DatasourceFilter`#

`SlotRangeFilter`#