# Traits (Carbon 2.0)

> Every trait you implement or call in Carbon 2.0.0, with required methods, associated types and a compiled implementation example.

Carbon is built from a few traits: datasources produce updates, decoders turn raw bytes into types, processors handle the result, filters gate pipes, and exporters publish metrics. This page lists each trait in `carbon-core` 2.0.0 with its exact definition and an implementation that compiles against the published crates.

Examples on this page assume these dependencies:

`Cargo.toml`:

```toml
[dependencies]
async-trait = "0.1"
borsh = { version = "1.5.7", features = ["derive"] }
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = { version = "2.0.0", features = ["serde"] }
carbon-token-program-decoder = { version = "2.0.0", features = ["serde"] }
serde = { version = "1", features = ["derive"] }
solana-account = "4.3.2"
solana-instruction = "3.4.1"
solana-pubkey = "4.2.1"
tokio = { version = "1", features = ["full"] }
tokio-util = "0.7"
```

| Trait                                                           | Module                     | You implement it to                                          |
| --------------------------------------------------------------- | -------------------------- | ------------------------------------------------------------ |
| [`Datasource`](#datasource)                                     | `carbon_core::datasource`  | Feed updates from a new source                               |
| [`Processor`](#processor)                                       | `carbon_core::processor`   | Handle decoded data                                          |
| [`AccountDecoder`](#accountdecoder)                             | `carbon_core::account`     | Decode accounts of a program without a generated decoder     |
| [`InstructionDecoder`](#instructiondecoder)                     | `carbon_core::instruction` | Decode instructions of a program without a generated decoder |
| [`CarbonDeserialize`](#carbondeserialize)                       | `carbon_core::deserialize` | Decode log events with `decode_log_events`                   |
| [`ArrangeAccounts`](#arrangeaccounts)                           | `carbon_core::deserialize` | Map an instruction's account list to named fields            |
| [`InstructionDecoderCollection`](#instructiondecodercollection) | `carbon_core::collection`  | Use transaction pipes over one or more programs              |
| [`Filter`](#filter)                                             | `carbon_core::filter`      | Gate which updates reach a pipe                              |
| [`MetricsExporter`](#metricsexporter)                           | `carbon_core::metrics`     | Send metrics to a backend                                    |
| [`Metric`](#metric)                                             | `carbon_core::metrics`     | Read a metric's name and help text                           |

## `Datasource`

Produces updates and sends them into the pipeline.

```rust
#[async_trait]
pub trait Datasource: Send + Sync {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: tokio::sync::mpsc::Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()>;

    fn update_types(&self) -> Vec<UpdateType>;
}
```

**Details.** `Datasource` uses `#[async_trait]` because the pipeline stores datasources as `Arc<dyn Datasource>`; put `#[async_trait::async_trait]` on your `impl` too.

| Method         | Required | Description                                                                                                                                                                                                                                                                                                                                                                                   |
| -------------- | -------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `consume`      | Yes      | Called once by `Pipeline::run`, in its own Tokio task. Send each update as `(update, id.clone())` on `sender`. Return when `cancellation_token` is cancelled. Returning (with `Ok` or `Err`) ends this datasource; the pipeline stops when all datasources have returned and the channel is empty. An `Err` is logged as `error consuming datasource: …` and does not stop other datasources. |
| `update_types` | Yes      | The [`UpdateType`](https://carbon.sevenlabs.org/reference/updates.md#updatetype)s this datasource emits. Not used by the pipeline in 2.0.0, but return an accurate list for readers and future versions.                                                                                                                                                                                      |

Use `sender.send(...).await` to wait when the pipeline is behind (backpressure), or `try_send` to drop updates instead. Build account updates with [`AccountUpdate::into_update`](https://carbon.sevenlabs.org/reference/updates.md#accountupdateinto_update) so zero-lamport accounts become deletions. `&self` is shared, so keep mutable state inside `consume` or behind a lock.

**Example.**

`src/ticker.rs`:

```rust
use {
    async_trait::async_trait,
    carbon_core::{
        datasource::{AccountUpdate, Datasource, DatasourceId, Update, UpdateType},
        error::{CarbonResult, Error},
    },
    solana_account::Account,
    solana_pubkey::Pubkey,
    std::time::Duration,
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Emits one synthetic account update per second until cancelled.
pub struct Ticker {
    pub owner: Pubkey,
}

#[async_trait]
impl Datasource for Ticker {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        let mut slot: u64 = 0;
        loop {
            tokio::select! {
                _ = cancellation_token.cancelled() => return Ok(()),
                _ = tokio::time::sleep(Duration::from_secs(1)) => {}
            }
            slot += 1;

            let mut data = b"counter!".to_vec();
            data.extend_from_slice(&slot.to_le_bytes());

            let update = AccountUpdate {
                pubkey: Pubkey::new_unique(),
                account: Account {
                    lamports: 1_000_000,
                    data,
                    owner: self.owner,
                    executable: false,
                    rent_epoch: 0,
                },
                slot,
                transaction_signature: None,
            }
            .into_update();

            sender
                .send((update, id.clone()))
                .await
                .map_err(|e| Error::FailedToConsumeDatasource(e.to_string()))?;
        }
    }

    fn update_types(&self) -> Vec<UpdateType> {
        vec![UpdateType::AccountUpdate]
    }
}
```

**See also.** [Custom datasource](https://carbon.sevenlabs.org/guides/custom-datasource.md), [Updates](https://carbon.sevenlabs.org/reference/updates.md), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/datasource/trait.Datasource.html).

## `Processor`

Handles one input value at a time.

```rust
pub trait Processor<T>
where
    T: Sync,
{
    fn process(&mut self, data: &T) -> impl Future<Output = CarbonResult<()>> + Send;
}
```

**Details.** Implement it with a native `async fn process(&mut self, data: &T) -> CarbonResult<()>`; no `#[async_trait]` is needed. `T` is the pipe's input type, listed in [Processor inputs](https://carbon.sevenlabs.org/reference/processor-inputs.md). For input types with a lifetime, write `impl Processor<InstructionProcessorInputType<'_, MyInstruction>> for MyProcessor`; the builder requires the impl for every lifetime, which `'_` provides.

The pipeline owns the processor and calls `process` sequentially, so `&mut self` state needs no locking. The returned future must be `Send`: don't hold a `std::sync::MutexGuard` or `Rc` across an `.await`. Returning `Err` logs the error, increments `carbon_updates_failed_total`, skips the remaining pipes for that update and continues with the next update.

**Example.**

`src/deletions.rs`:

```rust
use carbon_core::{datasource::AccountDeletion, error::CarbonResult, processor::Processor};

#[derive(Default)]
pub struct DeletionCounter {
    pub seen: u64,
}

impl Processor<AccountDeletion> for DeletionCounter {
    async fn process(&mut self, deletion: &AccountDeletion) -> CarbonResult<()> {
        self.seen += 1;
        println!("closed {} at slot {} ({} so far)", deletion.pubkey, deletion.slot, self.seen);
        Ok(())
    }
}
```

**See also.** [Processors](https://carbon.sevenlabs.org/concepts/processors.md), [Processor inputs](https://carbon.sevenlabs.org/reference/processor-inputs.md), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/processor/trait.Processor.html).

## `AccountDecoder`

Turns a raw account into a typed value.

```rust
pub trait AccountDecoder<'a> {
    type AccountType;

    fn decode_account(
        &self,
        account: &'a solana_account::Account,
    ) -> Option<DecodedAccount<Self::AccountType>>;
}
```

**Details.**

| Item               | Description                                                                                               |
| ------------------ | --------------------------------------------------------------------------------------------------------- |
| `type AccountType` | What your processor receives in `decoded_account.data`. Must be `Send + Sync + 'static` to register.      |
| `decode_account`   | Return `None` for accounts you don't handle; the pipe then skips the update. Check `account.owner` first. |

Generated decoders implement this trait on their decoder struct (`MeteoraDlmmDecoder`, `TokenProgramDecoder`, …): they check the owner, then try each account type's discriminator in turn. Implement it by hand for programs without an IDL. Register the decoder with [`account`](https://carbon.sevenlabs.org/reference/pipeline-builder.md#account), which requires the impl for every lifetime (`impl<'a> AccountDecoder<'a> for …`).

**Example.**

`src/counter.rs`:

```rust
use {
    carbon_core::account::{AccountDecoder, DecodedAccount},
    solana_account::Account,
    solana_pubkey::Pubkey,
};

pub const COUNTER_PROGRAM_ID: Pubkey =
    Pubkey::from_str_const("Counter111111111111111111111111111111111111");

#[derive(Debug, Clone)]
pub struct CounterAccount {
    pub count: u64,
}

pub struct CounterDecoder;

impl<'a> AccountDecoder<'a> for CounterDecoder {
    type AccountType = CounterAccount;

    fn decode_account(&self, account: &'a Account) -> Option<DecodedAccount<CounterAccount>> {
        if account.owner != COUNTER_PROGRAM_ID {
            return None;
        }
        let payload = account.data.strip_prefix(b"counter!")?;
        let count = u64::from_le_bytes(payload.get(..8)?.try_into().ok()?);

        Some(DecodedAccount {
            lamports: account.lamports,
            data: CounterAccount { count },
            owner: account.owner,
            executable: account.executable,
            rent_epoch: account.rent_epoch,
        })
    }
}
```

**See also.** [`DecodedAccount`](https://carbon.sevenlabs.org/reference/processor-inputs.md#decodedaccount), [Custom decoder](https://carbon.sevenlabs.org/guides/custom-decoder.md), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/account/trait.AccountDecoder.html).

## `InstructionDecoder`

Turns a raw instruction into a typed value.

```rust
pub trait InstructionDecoder<'a> {
    type InstructionType;

    fn decode_instruction(
        &self,
        instruction: &'a solana_instruction::Instruction,
    ) -> Option<Self::InstructionType>;
}
```

**Details.**

| Item                   | Description                                                                                                                                                  |
| ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `type InstructionType` | What your processor receives in `decoded_instruction`. Must be `Send + Sync + 'static` to register.                                                          |
| `decode_instruction`   | Called for every instruction of every transaction, top-level and inner. Return `None` quickly for other programs by checking `instruction.program_id` first. |

The `instruction` has resolved `AccountMeta`s, so you can read `is_signer` and `is_writable`. Generated decoders return an enum with one struct variant per instruction (`{ program_id, data, accounts }`) built with [`try_decode_instructions!`](https://carbon.sevenlabs.org/reference/macros.md#try_decode_instructions).

**Example.**

`src/memo.rs`:

```rust
use {
    carbon_core::instruction::InstructionDecoder,
    solana_instruction::Instruction,
    solana_pubkey::Pubkey,
};

pub const MEMO_PROGRAM_ID: Pubkey =
    Pubkey::from_str_const("MemoSq4gqABAXKb96qnH8TysNcWxMyWCqXgDLGmfcHr");

#[derive(Debug, Clone)]
pub struct Memo {
    pub text: String,
    pub signers: Vec<Pubkey>,
}

pub struct MemoDecoder;

impl<'a> InstructionDecoder<'a> for MemoDecoder {
    type InstructionType = Memo;

    fn decode_instruction(&self, instruction: &'a Instruction) -> Option<Memo> {
        if instruction.program_id != MEMO_PROGRAM_ID {
            return None;
        }
        Some(Memo {
            text: String::from_utf8(instruction.data.clone()).ok()?,
            signers: instruction
                .accounts
                .iter()
                .filter(|meta| meta.is_signer)
                .map(|meta| meta.pubkey)
                .collect(),
        })
    }
}
```

**See also.** [Instructions](https://carbon.sevenlabs.org/concepts/instructions.md), [Custom decoder](https://carbon.sevenlabs.org/guides/custom-decoder.md), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/instruction/trait.InstructionDecoder.html).

## `CarbonDeserialize`

Discriminator-prefixed Borsh deserialization.

```rust
pub trait CarbonDeserialize
where
    Self: Sized + borsh::BorshDeserialize,
{
    const DISCRIMINATOR: &'static [u8];

    fn deserialize(data: &[u8]) -> Option<Self>;
}
```

**Details.**

| Item                  | Description                                                                                                           |
| --------------------- | --------------------------------------------------------------------------------------------------------------------- |
| `const DISCRIMINATOR` | The prefix that identifies this type, usually 8 bytes for Anchor. Not read by Carbon itself; use it in `deserialize`. |
| `deserialize`         | Check and strip the discriminator, then Borsh-decode the rest. Return `None` on mismatch.                             |

In 2.0.0 the only consumer is [`InstructionMetadata::decode_log_events`](https://carbon.sevenlabs.org/reference/processor-inputs.md#instructionmetadatadecode_log_events). Generated decoders do **not** implement this trait: their account, instruction and event types have an inherent `decode(&[u8]) -> Option<Self>` method instead. To decode a generated event type from logs, wrap it or implement the trait on your own type. `carbon_core::deserialize::extract_discriminator(length, data)` splits `data` into `(discriminator, rest)` and returns `None` when `data` is shorter than `length`.

**Example.**

`src/events.rs`:

```rust
use {
    borsh::BorshDeserialize,
    carbon_core::{
        deserialize::CarbonDeserialize, error::CarbonResult,
        instruction::InstructionProcessorInputType, processor::Processor,
    },
    carbon_meteora_dlmm_decoder::instructions::MeteoraDlmmInstruction,
    solana_pubkey::Pubkey,
};

/// An Anchor `emit!` event: 8-byte discriminator, then the Borsh body.
#[derive(Debug, BorshDeserialize)]
pub struct MyEvent {
    pub pool: Pubkey,
    pub amount: u64,
}

impl CarbonDeserialize for MyEvent {
    const DISCRIMINATOR: &'static [u8] = &[0x1d, 0x2c, 0x3b, 0x4a, 0x59, 0x68, 0x77, 0x86];

    fn deserialize(data: &[u8]) -> Option<Self> {
        let body = data.strip_prefix(Self::DISCRIMINATOR)?;
        borsh::from_slice(body).ok()
    }
}

pub struct EventLogger;

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for EventLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        for event in input.metadata.decode_log_events::<MyEvent>() {
            println!("{} moved {}", event.pool, event.amount);
        }
        Ok(())
    }
}
```

**See also.** [Instructions](https://carbon.sevenlabs.org/concepts/instructions.md), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/deserialize/trait.CarbonDeserialize.html).

## `ArrangeAccounts`

Maps an instruction's positional accounts to a struct.

```rust
pub trait ArrangeAccounts {
    type ArrangedAccounts: Clone + Send + Sync + std::fmt::Debug;

    fn arrange_accounts(
        accounts: &[solana_instruction::AccountMeta],
    ) -> Option<Self::ArrangedAccounts>;
}
```

**Details.** An associated function (no `self`), implemented on the instruction's data type. Return `None` when required accounts are missing; [`try_decode_instructions!`](https://carbon.sevenlabs.org/reference/macros.md#try_decode_instructions) then rejects the instruction. Generated decoders implement it for every instruction, treat optional accounts as `Option<Pubkey>`, and collect extra accounts into `remaining: Vec<AccountMeta>`. `carbon_core::account_utils::next_account(&mut iter)` returns the next account's `Pubkey`, or `None` at the end.

**Example.**

`src/transfer.rs`:

```rust
use {
    carbon_core::{account_utils::next_account, deserialize::ArrangeAccounts},
    solana_instruction::AccountMeta,
    solana_pubkey::Pubkey,
};

pub struct Transfer;

#[derive(Debug, Clone)]
pub struct TransferAccounts {
    pub source: Pubkey,
    pub destination: Pubkey,
    pub authority: Pubkey,
    pub remaining: Vec<AccountMeta>,
}

impl ArrangeAccounts for Transfer {
    type ArrangedAccounts = TransferAccounts;

    fn arrange_accounts(accounts: &[AccountMeta]) -> Option<TransferAccounts> {
        let mut iter = accounts.iter();
        let source = next_account(&mut iter)?;
        let destination = next_account(&mut iter)?;
        let authority = next_account(&mut iter)?;

        Some(TransferAccounts {
            source,
            destination,
            authority,
            remaining: iter.as_slice().to_vec(),
        })
    }
}
```

**See also.** [`try_decode_instructions!`](https://carbon.sevenlabs.org/reference/macros.md#try_decode_instructions), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/deserialize/trait.ArrangeAccounts.html).

## `InstructionDecoderCollection`

A single enum over the instructions of several programs, used by transaction pipes.

```rust
pub trait InstructionDecoderCollection:
    Clone + std::fmt::Debug + Send + Sync + Eq + std::hash::Hash + Serialize + 'static
{
    type InstructionType: Clone + std::fmt::Debug + PartialEq + Eq + Send + Sync + 'static;

    fn parse_instruction(instruction: &solana_instruction::Instruction) -> Option<Self>;

    fn get_type(&self) -> Self::InstructionType;
}
```

**Details.**

| Item                   | Description                                                                                                                                               |
| ---------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `type InstructionType` | A tag enum identifying the variant. Carbon does not read it in 2.0.0.                                                                                     |
| `parse_instruction`    | Associated function (no decoder instance). Return the variant for the instruction's program, or `None`. Called on every instruction of every transaction. |
| `get_type`             | Return the tag for `self`.                                                                                                                                |

The supertraits include `Eq`, `Hash` and `serde::Serialize`. Generated decoder enums derive only `Debug`, `Clone` and `PartialEq` (plus `Serialize`/`Deserialize` with the `serde` feature), so wrap them in your own enum and implement `Eq` and `Hash` by hand, as below. The [`instruction_decoder_collection!`](https://carbon.sevenlabs.org/reference/macros.md#instruction_decoder_collection) macro generates this trait, but only for instruction enums that derive `Hash`, `Eq` and `InstructionType`, which generated decoders don't.

**Example.**

`src/collection.rs`:

```rust
use {
    carbon_core::{collection::InstructionDecoderCollection, instruction::InstructionDecoder},
    carbon_meteora_dlmm_decoder::{instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder},
    carbon_token_program_decoder::{instructions::TokenProgramInstruction, TokenProgramDecoder},
    solana_instruction::Instruction,
    std::hash::{Hash, Hasher},
};

#[derive(Debug, Clone, PartialEq, serde::Serialize)]
pub enum AllInstructions {
    MeteoraDlmm(MeteoraDlmmInstruction),
    Token(TokenProgramInstruction),
}

// Generated instruction enums derive `PartialEq` only, so provide `Eq` and
// `Hash` by hand. Hashing the variant alone is consistent with `PartialEq`.
impl Eq for AllInstructions {}

impl Hash for AllInstructions {
    fn hash<H: Hasher>(&self, state: &mut H) {
        std::mem::discriminant(self).hash(state);
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AllInstructionTypes {
    MeteoraDlmm,
    Token,
}

impl InstructionDecoderCollection for AllInstructions {
    type InstructionType = AllInstructionTypes;

    fn parse_instruction(instruction: &Instruction) -> Option<Self> {
        if let Some(ix) = MeteoraDlmmDecoder.decode_instruction(instruction) {
            return Some(Self::MeteoraDlmm(ix));
        }
        TokenProgramDecoder
            .decode_instruction(instruction)
            .map(Self::Token)
    }

    fn get_type(&self) -> AllInstructionTypes {
        match self {
            Self::MeteoraDlmm(_) => AllInstructionTypes::MeteoraDlmm,
            Self::Token(_) => AllInstructionTypes::Token,
        }
    }
}
```

Register it with `Pipeline::builder().transaction::<AllInstructions, _>(processor)`.

**See also.** [Transactions](https://carbon.sevenlabs.org/concepts/transactions.md), [`TransactionProcessorInputType`](https://carbon.sevenlabs.org/reference/processor-inputs.md#transactionprocessorinputtype), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/collection/trait.InstructionDecoderCollection.html).

## `Filter`

Decides whether an update reaches a pipe.

```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.** Every method has a default that returns `Accept`, so implement only the hooks for the pipe kinds you filter. Hooks take `&self`: use atomics or a lock for state, as `DeduplicationFilter` does. See the [filters reference](https://carbon.sevenlabs.org/reference/filters.md#filter) for when each hook runs.

**Example.**

`src/filters.rs`:

```rust
use {
    carbon_core::{
        filter::{Filter, FilterContext, FilterResult},
        instruction::{NestedInstruction, NestedInstructions},
        transaction::TransactionMetadata,
    },
};

/// Drops failed transactions and every instruction inside them.
pub struct SuccessfulOnly;

impl Filter for SuccessfulOnly {
    fn filter_instruction(
        &self,
        _context: &FilterContext,
        nested_instruction: &NestedInstruction,
    ) -> FilterResult {
        if nested_instruction.metadata.transaction_metadata.meta.status.is_ok() {
            FilterResult::Accept
        } else {
            FilterResult::Reject
        }
    }

    fn filter_transaction(
        &self,
        _context: &FilterContext,
        transaction_metadata: &TransactionMetadata,
        _nested_instructions: &NestedInstructions,
    ) -> FilterResult {
        if transaction_metadata.meta.status.is_ok() {
            FilterResult::Accept
        } else {
            FilterResult::Reject
        }
    }
}
```

**See also.** [Filters reference](https://carbon.sevenlabs.org/reference/filters.md), [Filters](https://carbon.sevenlabs.org/concepts/filters.md), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/filter/trait.Filter.html).

## `MetricsExporter`

Publishes metric snapshots to a backend.

```rust
pub trait MetricsExporter: Send + Sync {
    fn initialize(self: Arc<Self>) -> CarbonResult<()> {
        let _ = self;
        Ok(())
    }

    fn export(&self, snapshot: &MetricsSnapshot) -> CarbonResult<()>;

    fn shutdown(&self) -> CarbonResult<()> {
        Ok(())
    }
}
```

**Details.**

| Method       | Required | When the pipeline calls it                                                                                                                                                                |
| ------------ | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `initialize` | No       | Once, at the start of `run`, with an `Arc` of the exporter. Start periodic work here (for example `tokio::spawn` a loop that calls `self.export(&MetricsRegistry::global().snapshot())`). |
| `export`     | Yes      | Once, when the pipeline stops, with a final snapshot. Not called while the pipeline runs.                                                                                                 |
| `shutdown`   | No       | Once, after the final `export`.                                                                                                                                                           |

An `Err` from any of these makes `Pipeline::run` return that error. Register with [`PipelineBuilder::metrics`](https://carbon.sevenlabs.org/reference/pipeline-builder.md#metrics).

**Example.**

`src/exporter.rs`:

```rust
use {
    carbon_core::{
        error::CarbonResult,
        metrics::{MetricsExporter, MetricsRegistry, MetricsSnapshot},
    },
    std::{sync::Arc, time::Duration},
};

/// Prints every counter every 10 seconds, and once more at shutdown.
pub struct StdoutExporter;

impl MetricsExporter for StdoutExporter {
    fn initialize(self: Arc<Self>) -> CarbonResult<()> {
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(10));
            loop {
                interval.tick().await;
                let _ = self.export(&MetricsRegistry::global().snapshot());
            }
        });
        Ok(())
    }

    fn export(&self, snapshot: &MetricsSnapshot) -> CarbonResult<()> {
        for (name, _help, value) in &snapshot.counters {
            println!("{name} {value}");
        }
        Ok(())
    }
}
```

**See also.** [Metrics reference](https://carbon.sevenlabs.org/reference/metrics.md), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/metrics/trait.MetricsExporter.html).

## `Metric`

Name and help text shared by `Counter`, `Gauge` and `Histogram`.

```rust
pub trait Metric: Send + Sync {
    fn name(&self) -> &'static str;
    fn help(&self) -> &'static str {
        ""
    }
}
```

**Details.** Implemented by the three metric types in `carbon_core::metrics`. `MetricsRegistry` only stores `&'static Counter`, `&'static Gauge` and `&'static Histogram`, so implementing `Metric` on your own type does not make it exportable. To add a metric, declare a `static` `Counter`, `Gauge` or `Histogram` and register it.

**Example.**

`src/metrics.rs`:

```rust
use carbon_core::metrics::{Counter, Metric, MetricsRegistry};

static SWAPS_SEEN: Counter = Counter::new("my_indexer_swaps_seen_total", "Swaps decoded");

pub fn register() {
    MetricsRegistry::global().register_counter(&SWAPS_SEEN);
    assert_eq!(SWAPS_SEEN.name(), "my_indexer_swaps_seen_total");
}
```

**See also.** [Metrics reference](https://carbon.sevenlabs.org/reference/metrics.md), [docs.rs](https://docs.rs/carbon-core/2.0.0/carbon_core/metrics/trait.Metric.html).

## Complete example

This program implements a datasource and an account decoder, runs them for 3.5 seconds and needs no network.

`src/main.rs`:

```rust
use {
    async_trait::async_trait,
    carbon_core::{
        account::{AccountDecoder, AccountProcessorInputType, DecodedAccount},
        datasource::{AccountUpdate, Datasource, DatasourceId, Update, UpdateType},
        error::{CarbonResult, Error},
        pipeline::Pipeline,
        processor::Processor,
    },
    solana_account::Account,
    solana_pubkey::Pubkey,
    std::time::Duration,
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

const COUNTER_PROGRAM_ID: Pubkey =
    Pubkey::from_str_const("Counter111111111111111111111111111111111111");

/// Emits one synthetic account update per second until cancelled.
struct Ticker;

#[async_trait]
impl Datasource for Ticker {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        let mut slot: u64 = 0;
        loop {
            tokio::select! {
                _ = cancellation_token.cancelled() => return Ok(()),
                _ = tokio::time::sleep(Duration::from_secs(1)) => {}
            }
            slot += 1;

            let mut data = b"counter!".to_vec();
            data.extend_from_slice(&slot.to_le_bytes());

            let update = AccountUpdate {
                pubkey: Pubkey::new_unique(),
                account: Account {
                    lamports: 1_000_000,
                    data,
                    owner: COUNTER_PROGRAM_ID,
                    executable: false,
                    rent_epoch: 0,
                },
                slot,
                transaction_signature: None,
            }
            .into_update();

            sender
                .send((update, id.clone()))
                .await
                .map_err(|e| Error::FailedToConsumeDatasource(e.to_string()))?;
        }
    }

    fn update_types(&self) -> Vec<UpdateType> {
        vec![UpdateType::AccountUpdate]
    }
}

#[derive(Debug, Clone)]
struct CounterAccount {
    count: u64,
}

struct CounterDecoder;

impl<'a> AccountDecoder<'a> for CounterDecoder {
    type AccountType = CounterAccount;

    fn decode_account(&self, account: &'a Account) -> Option<DecodedAccount<CounterAccount>> {
        if account.owner != COUNTER_PROGRAM_ID {
            return None;
        }
        let payload = account.data.strip_prefix(b"counter!")?;
        let count = u64::from_le_bytes(payload.get(..8)?.try_into().ok()?);

        Some(DecodedAccount {
            lamports: account.lamports,
            data: CounterAccount { count },
            owner: account.owner,
            executable: account.executable,
            rent_epoch: account.rent_epoch,
        })
    }
}

struct CounterLogger;

impl Processor<AccountProcessorInputType<'_, CounterAccount>> for CounterLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, CounterAccount>,
    ) -> CarbonResult<()> {
        println!(
            "slot {}: {} count={}",
            input.metadata.slot, input.metadata.pubkey, input.decoded_account.data.count
        );
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    let token = CancellationToken::new();
    let stop = token.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(3_500)).await;
        stop.cancel();
    });

    Pipeline::builder()
        .datasource(Ticker)
        .account(CounterDecoder, CounterLogger)
        .datasource_cancellation_token(token)
        .build()?
        .run()
        .await
}
```

```sh
cargo run
```

Output:

```text
slot 1: 11157t3sqMV725NVRLrVQbAu98Jjfk1uCKehJnXXQs count=1
slot 2: 1117mWrzzrZr312ebPDHu8tbfMwFNvCvMbr6WepCNG count=2
slot 3: 1119DWteoLSdjvrT6g6L8C2PfDD2faiTQUpsjY2RiF count=3
```

---

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