Filters

A filter decides, per pipe, whether an update reaches the pipe's decoder and processor. Use filters to send each datasource's updates to different processors, to drop duplicates when two sources overlap, or to process only a slot range. Filters belong to one pipe; other pipes still see everything.

Adding filters to a pipe#

Every pipe method has a *_with_filters variant that takes a Vec<Box<dyn Filter>>:

Pipeline::builder()
    .instruction_with_filters(
        MeteoraDlmmDecoder,
        SwapLogger,
        vec![
            Box::new(DatasourceFilter::new(DatasourceId::new_named("primary"))),
            Box::new(SlotRangeFilter::from(370_000_000, None)),
        ],
    )

The pipeline runs a pipe's filters in order, before decoding, and stops at the first Reject. The pipe runs only if every filter accepts. Because evaluation stops early, a filter later in the list never sees updates an earlier one rejected; this matters for stateful filters such as DeduplicationFilter.

The Filter trait#

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
    }
}

Every method defaults to Accept, so implement only the ones for the pipes you attach the filter to. Each pipe calls one method: account pipes call filter_account, instruction pipes call filter_instruction once per instruction (nested ones included), transaction pipes call filter_transaction once per transaction, and so on.

FilterContext carries the DatasourceId of the source the update came from:

pub struct FilterContext<'a> {
    pub datasource_id: &'a DatasourceId,
}

pub enum FilterResult {
    Accept,
    Reject,
}

Filter methods take &self and are synchronous. Keep them fast: they run in the pipeline loop for every update. A filter that keeps state needs interior mutability, such as a Mutex or atomics.

Built-in filters#

DatasourceFilter#

Accepts updates only from the listed datasources, for every update type.

DatasourceFilter::new(DatasourceId::new_named("primary"))
DatasourceFilter::new_many(vec![DatasourceId::new_named("a"), DatasourceId::new_named("b")])

Register the datasources with datasource_with_id so that you know their IDs. Use it to send a backfill crawler's transactions to one processor and a live stream's to another.

DeduplicationFilter#

Drops updates it has already seen within a time window:

DeduplicationFilter::new(Duration::from_secs(60))

Entries older than the TTL are purged, at most once a minute, when the filter is called. The filter is in memory and per instance: each pipe needs its own instance, and nothing survives a restart. Put it first in the list so that it records every update, and give it a TTL longer than the maximum delay between your redundant sources.

SlotRangeFilter#

Accepts updates whose slot is inside a range:

SlotRangeFilter::from(370_000_000, None)               // slot ≥ 370,000,000
SlotRangeFilter::to(371_000_000, None)                 // slot ≤ 371,000,000
SlotRangeFilter::between(370_000_000, None, 371_000_000, None)

Both ends are inclusive when you pass None for the transaction index. The optional index narrows the boundary slots for transactions and instructions, using the transaction's position in its block: at from_slot, transactions with an index below from_transaction_index are rejected; at to_slot, transactions with an index at or above to_transaction_index are rejected. Updates without an index, such as account updates and transactions from the RPC transaction crawler, are compared by slot alone. The filter applies to every update type.

Filter by program#

You rarely need a filter to keep one program's data: decoders already do it. A generated decoder returns nothing for an instruction whose program_id isn't its PROGRAM_ID, and nothing for an account whose owner isn't, so its processor only ever sees that program. That includes the program's instructions nested inside other programs' calls.

What decides how much data reaches the pipeline is the datasource. Narrow it at the source so you don't download and discard the rest of Solana:

Use a Filter only for what neither can express, such as routing one datasource's updates to one processor (DatasourceFilter).

Decode or skip failed transactions#

Solana records transactions that fail during execution, with their instructions and logs. Whether your processors see them depends on the datasource:

A failed transaction's instructions decode normally, because decoding reads the instruction bytes, not the result. To ignore failures from a source that includes them, check the status in your processor:

if input.metadata.transaction_metadata.meta.status.is_err() {
    return Ok(()); // the transaction failed on chain; nothing it did took effect
}

"Decode failed" is a different problem: an instruction from your program that the decoder doesn't recognize, usually after a program upgrade changed its IDL. See Versioned decoders and Troubleshooting.

Write a custom filter#

Implement Filter on your own type and override the methods you need. This filter keeps only accounts above a balance:

struct MinLamports {
    min: u64,
}

impl Filter for MinLamports {
    fn filter_account(
        &self,
        _context: &FilterContext,
        _metadata: &AccountMetadata,
        account: &Account,
    ) -> FilterResult {
        if account.lamports >= self.min {
            FilterResult::Accept
        } else {
            FilterResult::Reject
        }
    }
}

A filter sees raw data, before decoding. To filter on decoded fields, check them in the processor instead.

Example#

Two redundant datasources deliver the same three account writes. The pipe drops the duplicates, then the writes outside slots 0 to 25, then the accounts with less than 1,000 lamports.

cargo add carbon-core@2.0.0 carbon-stream-message-datasource@2.0.0
cargo add solana-account@4.3.2 solana-pubkey@4.2.1 solana-signature@3.4.1
cargo add tokio --features full

use {
    carbon_core::{
        account::{AccountDecoder, AccountMetadata, AccountProcessorInputType, DecodedAccount},
        datasource::{AccountUpdate, DatasourceId},
        error::CarbonResult,
        filter::{DeduplicationFilter, Filter, FilterContext, FilterResult, SlotRangeFilter},
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_stream_message_datasource::{StreamMessageClient, UnifiedMessage},
    solana_account::Account,
    solana_pubkey::Pubkey,
    solana_signature::Signature,
    std::time::Duration,
};

/// Custom filter: only accounts holding at least `min` lamports.
struct MinLamports {
    min: u64,
}

impl Filter for MinLamports {
    fn filter_account(
        &self,
        _context: &FilterContext,
        _metadata: &AccountMetadata,
        account: &Account,
    ) -> FilterResult {
        if account.lamports >= self.min {
            FilterResult::Accept
        } else {
            FilterResult::Reject
        }
    }
}

struct AnyAccount;

impl<'a> AccountDecoder<'a> for AnyAccount {
    type AccountType = ();

    fn decode_account(&self, account: &'a Account) -> Option<DecodedAccount<()>> {
        Some(DecodedAccount {
            lamports: account.lamports,
            data: (),
            owner: account.owner,
            executable: account.executable,
            rent_epoch: account.rent_epoch,
        })
    }
}

struct Print;

impl Processor<AccountProcessorInputType<'_, ()>> for Print {
    async fn process(&mut self, input: &AccountProcessorInputType<'_, ()>) -> CarbonResult<()> {
        println!("slot {} lamports {}", input.metadata.slot, input.raw_account.lamports);
        Ok(())
    }
}

/// Builds a finite source that replays the same three account writes.
async fn replica() -> StreamMessageClient {
    let (tx, rx) = tokio::sync::mpsc::channel(16);
    for (i, (slot, lamports)) in [(10, 5_000), (20, 50), (30, 7_000)].into_iter().enumerate() {
        let update = AccountUpdate {
            pubkey: Pubkey::new_from_array([i as u8; 32]),
            account: Account { lamports, ..Account::default() },
            slot,
            transaction_signature: Some(Signature::from([i as u8 + 1; 64])),
        };
        tx.send(UnifiedMessage::Account(update)).await.unwrap();
    }
    StreamMessageClient::new(rx)
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    Pipeline::builder()
        // Two redundant sources deliver the same writes.
        .datasource_with_id(replica().await, DatasourceId::new_named("primary"))
        .datasource_with_id(replica().await, DatasourceId::new_named("backup"))
        .account_with_filters(
            AnyAccount,
            Print,
            vec![
                Box::new(DeduplicationFilter::new(Duration::from_secs(60))),
                Box::new(SlotRangeFilter::between(0, None, 25, None)),
                Box::new(MinLamports { min: 1_000 }),
            ],
        )
        .build()?
        .run()
        .await
}

cargo run

slot 10 lamports 5000

Six updates arrive. DeduplicationFilter rejects the second copy of each write, SlotRangeFilter rejects slot 30, and MinLamports rejects the 50-lamport account.

Next steps#