# Filters (Carbon 2.0)

> Route, deduplicate and window updates per pipe with the Filter trait, DeduplicationFilter, DatasourceFilter, SlotRangeFilter and your own 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>>`:

| Without filters                   | With filters                                            |
| --------------------------------- | ------------------------------------------------------- |
| `account(decoder, processor)`     | `account_with_filters(decoder, processor, filters)`     |
| `instruction(decoder, processor)` | `instruction_with_filters(decoder, processor, filters)` |
| `transaction::<C, _>(processor)`  | `transaction_with_filters::<C, _>(processor, filters)`  |
| `account_deletions(processor)`    | `account_deletions_with_filters(processor, filters)`    |
| `block_details(processor)`        | `block_details_with_filters(processor, filters)`        |

```rust
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

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

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`](https://carbon.sevenlabs.org/concepts/datasources.md#datasource-ids) of the source the update came from:

```rust
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.

```rust
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`](https://carbon.sevenlabs.org/reference/pipeline-builder.md) 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:

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

| Pipe                                 | Key                                      | Notes                                                           |
| ------------------------------------ | ---------------------------------------- | --------------------------------------------------------------- |
| Instruction                          | (transaction signature, `absolute_path`) | Each instruction of a transaction is tracked separately.        |
| Account                              | (transaction signature, pubkey)          | Accounts without a `transaction_signature` are always accepted. |
| Transaction, deletion, block details | none                                     | Always accepted; the filter doesn't implement these methods.    |

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:

```rust
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:

| Datasource                                                                                                                                                                                                                                           | Where you choose the program                                                                    |
| ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------- |
| [Yellowstone gRPC](https://carbon.sevenlabs.org/datasources/yellowstone-grpc.md), [Helius LaserStream](https://carbon.sevenlabs.org/datasources/helius-laserstream.md)                                                                               | `account_include` / `account_required` in the transaction filter, `owner` in the account filter |
| [RPC transaction crawler](https://carbon.sevenlabs.org/datasources/rpc-transaction-crawler.md), [Helius GTFA](https://carbon.sevenlabs.org/datasources/helius-gtfa.md)                                                                               | The address you pass to the constructor                                                         |
| [RPC blockSubscribe](https://carbon.sevenlabs.org/datasources/rpc-block-subscribe.md)                                                                                                                                                                | `RpcBlockSubscribeFilter::MentionsAccountOrProgram(program_id)`                                 |
| [RPC programSubscribe](https://carbon.sevenlabs.org/datasources/rpc-program-subscribe.md), [RPC getProgramAccounts](https://carbon.sevenlabs.org/datasources/rpc-gpa.md), [Helius gPA v2](https://carbon.sevenlabs.org/datasources/helius-gpa-v2.md) | The program ID in `Filters`                                                                     |
| [RPC block crawler](https://carbon.sevenlabs.org/datasources/rpc-block-crawler.md), [validator snapshot](https://carbon.sevenlabs.org/datasources/validator-snapshot.md)                                                                             | No program filter: every transaction or account in range arrives, and decoders pick out yours   |

Use a [`Filter`](#write-a-custom-filter) only for what neither can express, such as routing one datasource's updates to one processor ([`DatasourceFilter`](#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:

| Datasource                                                                                 | Failed transactions                                                                                                                                  |
| ------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------- |
| RPC transaction crawler, RPC block crawler, RPC blockSubscribe, Helius enhanced WebSockets | Always skipped                                                                                                                                       |
| Yellowstone gRPC, Helius LaserStream                                                       | Included unless your transaction filter sets `failed: Some(false)`; for block subscriptions, `BlockFilters { failed_transactions: Some(false), .. }` |
| Helius GTFA                                                                                | Chosen with `TransactionStatusFilter` (`Succeeded`, `Failed` or `Any`)                                                                               |

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:

```rust
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](https://carbon.sevenlabs.org/guides/versioned-decoders.md) and [Troubleshooting](https://carbon.sevenlabs.org/troubleshooting.md).

## Write a custom filter

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

```rust
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.

```sh
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
```

`src/main.rs`:

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

```sh
cargo run
```

Output:

```text
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

- [Use multiple datasources](https://carbon.sevenlabs.org/guides/multiple-datasources.md): Redundant and complementary sources with routing.
- [Filters reference](https://carbon.sevenlabs.org/reference/filters.md): Every filter constructor and method.
- [Metrics](https://carbon.sevenlabs.org/concepts/metrics.md): Count what your pipeline and filters let through.

---

Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/concepts/filters · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt