# Test decoders and processors (Carbon 2.0)

> Unit-test Carbon decoders with carbon-test-utils fixtures captured from mainnet, and run processors through a real pipeline in tests without a network.

A Carbon indexer has two things worth testing: that the **decoder** turns real bytes into the right types, and that your **processor** does the right thing with them. This guide tests both offline, from fixtures captured once from mainnet:

- Decoder tests with [`carbon-test-utils`](https://carbon.sevenlabs.org/reference/test-utils.md): instruction data from an explorer, an instruction fixture, and an account fixture.
- Pipeline tests that replay a saved transaction through `Pipeline::run` with a fixture datasource and assert on what the processor recorded.

The code under test is `SwapStats`, a processor that sums Meteora DLMM swap volume per pool. The fixture is a real Jupiter route that swaps through two DLMM pools in one transaction.

**You need:** Rust 1.96.1 or newer, plus `curl` and `jq` to capture fixtures. Tests themselves need no network.

## How the Carbon repository tests

- `carbon-core` keeps real transaction metadata under `crates/core/tests/fixtures/*.json` (with the Solscan links in a README) and loads them with `carbon_test_utils::read_transaction_meta` in unit tests of its transformers.
- Decoders get unit tests next to the code: `carbon-token-program-decoder` packs SPL accounts with `solana_program_pack::Pack` and decodes the bytes; `carbon-token-2022-decoder` builds mints with extensions in `tests/`.
- Generated decoders add `carbon-test-utils` as a dev-dependency so you can add fixture tests the same way.

`carbon-test-utils` provides:

| Item                               | Use                                                                                                                                                           |
| ---------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `read_instruction(path)`           | Load an `Instruction` from JSON: `program_id`, `accounts` (`pubkey`, `is_signer`, `is_writable`) and hex `data`.                                              |
| `read_account(path)`               | Load an `Account` from JSON: base64 `data`, `executable`, `lamports`, `owner`, `rent_epoch`.                                                                  |
| `read_transaction_meta(path)`      | Load a `UiTransactionStatusMeta` (the `meta` object of a `getTransaction` response). Convert it with `transformers::transaction_metadata_from_original_meta`. |
| `base58_deserialize::ix_data(str)` | Decode base58 instruction data, as explorers display it, into bytes.                                                                                          |

All three readers return `anyhow::Result`. The full formats are in the [test-utils reference](https://carbon.sevenlabs.org/reference/test-utils.md).

### Step 1: Create the crate

```sh
cargo new --lib swap-stats
cd swap-stats
mkdir -p tests/fixtures src/bin
```

`Cargo.toml`:

```toml
[package]
name = "swap-stats"
version = "0.1.0"
edition = "2021"

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
hex = "0.4.3"
serde_json = "1.0.145"
solana-pubkey = "4.2.1"
solana-transaction-status = "=4.2.2"

[dev-dependencies]
async-trait = "0.1.89"
carbon-test-utils = "2.0.0"
solana-transaction-error = "3.3.2"
tokio = { version = "1.48.0", features = ["full"] }
tokio-util = "0.7.17"
```

`hex` and `serde_json` are used by the fixture tool below; the rest of the library needs only Carbon and the two Solana crates.

### Step 2: Write the code under test

`SwapStats` keeps its totals behind an `Arc<Mutex<..>>`, so a test can keep a clone and read them after the pipeline has finished. `load_transaction` turns a saved `getTransaction` response into the `TransactionUpdate` a datasource would send; it's what the RPC datasources do with a live response.

`src/lib.rs`:

```rust
use {
    carbon_core::{
        datasource::TransactionUpdate,
        error::{CarbonResult, Error},
        instruction::InstructionProcessorInputType,
        processor::Processor,
        transformers::transaction_metadata_from_original_meta,
    },
    carbon_meteora_dlmm_decoder::instructions::MeteoraDlmmInstruction,
    solana_pubkey::Pubkey,
    solana_transaction_status::EncodedConfirmedTransactionWithStatusMeta,
    std::{
        collections::HashMap,
        path::Path,
        sync::{Arc, Mutex},
    },
};

#[derive(Debug, Default, Clone, PartialEq)]
pub struct PairTotals {
    pub swaps: u64,
    pub amount_in: u128,
}

/// The pool and input amount of a swap instruction, if it is one.
pub fn swap_amount(instruction: &MeteoraDlmmInstruction) -> Option<(Pubkey, u64)> {
    match instruction {
        MeteoraDlmmInstruction::Swap { data, accounts, .. } => Some((accounts.lb_pair, data.amount_in)),
        MeteoraDlmmInstruction::Swap2 { data, accounts, .. } => {
            Some((accounts.lb_pair, data.amount_in))
        }
        _ => None,
    }
}

/// Sums swap volume per DLMM pool. Clone it to keep a handle on the totals.
#[derive(Clone, Default)]
pub struct SwapStats {
    pub totals: Arc<Mutex<HashMap<Pubkey, PairTotals>>>,
}

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for SwapStats {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        if input.metadata.transaction_metadata.meta.status.is_err() {
            return Ok(()); // failed transactions moved no funds
        }
        if let Some((pair, amount_in)) = swap_amount(input.decoded_instruction) {
            let mut totals = self.totals.lock().map_err(|e| Error::Custom(e.to_string()))?;
            let entry = totals.entry(pair).or_default();
            entry.swaps += 1;
            entry.amount_in += u128::from(amount_in);
        }
        Ok(())
    }
}

/// Loads a `getTransaction` result saved as JSON (Base64 encoding) into a `TransactionUpdate`.
pub fn load_transaction(path: impl AsRef<Path>) -> TransactionUpdate {
    let path = path.as_ref();
    let json = std::fs::read(path).unwrap_or_else(|e| panic!("read {}: {e}", path.display()));
    let fetched: EncodedConfirmedTransactionWithStatusMeta =
        serde_json::from_slice(&json).expect("a getTransaction result");

    let transaction = fetched
        .transaction
        .transaction
        .decode()
        .expect("Base64-encoded transaction");
    let meta = transaction_metadata_from_original_meta(fetched.transaction.meta.expect("meta"))
        .expect("convertible meta");

    TransactionUpdate {
        signature: transaction.signatures[0],
        transaction,
        meta,
        is_vote: false,
        slot: fetched.slot,
        index: None,
        block_time: fetched.block_time,
        block_hash: None,
    }
}
```

### Step 3: Capture a transaction

Save the `getTransaction` result for one signature. Request `base64` encoding: Carbon can't rebuild a transaction from `json` or `jsonParsed`.

```sh
SIG=SZE5CFK6gkG91yUWfovgDWd7XYFT9DFaHphCnQq2TbghcBY1c1uSzdksAxsayxgh27aBPhrvbBAahUbUNiNB757
curl -s https://api.mainnet-beta.solana.com -H 'content-type: application/json' \
  -d "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"getTransaction\",\"params\":[\"$SIG\",{\"encoding\":\"base64\",\"maxSupportedTransactionVersion\":1}]}" \
  | jq '.result' > tests/fixtures/route.json
jq '.slot, (.meta.logMessages | map(select(contains("Instruction: Swap"))))' tests/fixtures/route.json
```

Output:

```text
467107994
[
  "Program log: Instruction: Swap2",
  "Program log: Instruction: Swap2",
  "Program log: Instruction: SwapV2"
]
```

Two `Swap2` instructions are Meteora DLMM CPIs from Jupiter; `SwapV2` belongs to another program. If `route.json` contains `null`, the RPC rejected the request (rate limit or unknown signature): run `curl` without `jq` to see the error.

### Step 4: Extract instruction fixtures

`read_instruction` expects resolved accounts with signer and writable flags, which a raw transaction doesn't contain. This tool extracts them with Carbon's own instruction extraction and writes one fixture per DLMM instruction:

`src/bin/extract-instructions.rs`:

```rust
//! Writes every Meteora DLMM instruction of a saved transaction as a
//! `carbon-test-utils` instruction fixture.
use {
    carbon_core::{transaction::TransactionMetadata, transformers::extract_instructions_with_metadata},
    carbon_meteora_dlmm_decoder::PROGRAM_ID,
    serde_json::json,
    std::sync::Arc,
};

fn main() {
    let path = std::env::args().nth(1).expect("usage: extract-instructions <tx.json>");
    let update = swap_stats::load_transaction(&path);
    let metadata = Arc::new(TransactionMetadata::try_from(update.clone()).expect("metadata"));

    let instructions = extract_instructions_with_metadata(&metadata, &update).expect("instructions");
    for (n, (meta, ix)) in instructions.iter().enumerate() {
        if ix.program_id != PROGRAM_ID {
            continue;
        }
        let fixture = json!({
            "program_id": ix.program_id.to_string(),
            "accounts": ix.accounts.iter().map(|a| json!({
                "pubkey": a.pubkey.to_string(),
                "is_signer": a.is_signer,
                "is_writable": a.is_writable,
            })).collect::<Vec<_>>(),
            "data": hex::encode(&ix.data),
        });
        let out = format!("{}.ix{n}.json", path.trim_end_matches(".json"));
        std::fs::write(&out, serde_json::to_string_pretty(&fixture).unwrap()).unwrap();
        println!("{out}: path={:?} stack_height={}", meta.absolute_path, meta.stack_height);
    }
}
```

```sh
cargo run --bin extract-instructions -- tests/fixtures/route.json
```

Output:

```text
tests/fixtures/route.ix9.json: path=[4, 0] stack_height=2
tests/fixtures/route.ix10.json: path=[4, 0, 0] stack_height=3
tests/fixtures/route.ix13.json: path=[4, 0, 3] stack_height=3
tests/fixtures/route.ix14.json: path=[4, 1] stack_height=2
tests/fixtures/route.ix15.json: path=[4, 1, 0] stack_height=3
tests/fixtures/route.ix18.json: path=[4, 1, 3] stack_height=3
```

The two stack-height-2 entries are the swaps; the others are the DLMM program's self-CPI events. Keep the swaps under readable names and delete the rest:

```sh
mv tests/fixtures/route.ix9.json tests/fixtures/swap2_first.json
mv tests/fixtures/route.ix14.json tests/fixtures/swap2_second.json
rm tests/fixtures/route.ix*.json
```

### Step 5: Capture an account

`read_account` expects the account fields with `data` as a base64 string. `getAccountInfo` with `base64` encoding returns `data` as `[string, "base64"]`, so reshape it with `jq`:

```sh
curl -s https://api.mainnet-beta.solana.com -H 'content-type: application/json' \
  -d '{"jsonrpc":"2.0","id":1,"method":"getAccountInfo","params":["Bnoev1Gp1o247d7zTwSbznRwqjkRmbrx3R6Qt4P16YbQ",{"encoding":"base64"}]}' \
  | jq '.result.value | {data: .data[0], executable, lamports, owner, rent_epoch: .rentEpoch}' \
  > tests/fixtures/lb_pair.json
jq '{executable, lamports, owner, rent_epoch}' tests/fixtures/lb_pair.json
```

Output:

```text
{
  "executable": false,
  "lamports": 6535656,
  "owner": "LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo",
  "rent_epoch": 18446744073709551615
}
```

`rent_epoch` is `u64::MAX` for rent-exempt accounts. Use jq 1.7 or newer, which keeps large integers exact; older versions round it to a float and `read_account` fails to parse it.

### Step 6: Write decoder tests

The first test decodes instruction data copied from an explorer, without any fixture file. The others load the fixtures and go through `MeteoraDlmmDecoder`, so they also check account arrangement and account discriminators.

`tests/decoder.rs`:

```rust
use {
    carbon_core::{account::AccountDecoder, instruction::InstructionDecoder},
    carbon_meteora_dlmm_decoder::{
        accounts::MeteoraDlmmAccount,
        instructions::{MeteoraDlmmInstruction, Swap2},
        MeteoraDlmmDecoder,
    },
    carbon_test_utils::{base58_deserialize, read_account, read_instruction},
    solana_pubkey::Pubkey,
};

const POOL: Pubkey = Pubkey::from_str_const("Bnoev1Gp1o247d7zTwSbznRwqjkRmbrx3R6Qt4P16YbQ");
const WSOL: Pubkey = Pubkey::from_str_const("So11111111111111111111111111111111111111112");

#[test]
fn decodes_swap2_data_from_base58() {
    // Instruction data as an explorer shows it (base58).
    let data = base58_deserialize::ix_data("fx9RHbGFfZA1Se5ZmKqQdyyUnuKrrydfLZcDLj");

    let swap = Swap2::decode(&data).expect("Swap2 data");
    assert_eq!(swap.amount_in, 91_444_688);
    assert_eq!(swap.min_amount_out, 0);
}

#[test]
fn decodes_swap2_instruction_fixture() {
    let instruction = read_instruction("tests/fixtures/swap2_first.json").expect("fixture");

    let decoded = MeteoraDlmmDecoder
        .decode_instruction(&instruction)
        .expect("a DLMM instruction");

    let MeteoraDlmmInstruction::Swap2 { data, accounts, .. } = decoded else {
        panic!("expected Swap2, got {decoded:?}");
    };
    assert_eq!(data.amount_in, 91_444_688);
    assert_eq!(accounts.lb_pair, POOL);
    assert_eq!(accounts.token_y_mint, WSOL);
}

#[test]
fn decodes_lb_pair_account_fixture() {
    let account = read_account("tests/fixtures/lb_pair.json").expect("fixture");

    let decoded = MeteoraDlmmDecoder.decode_account(&account).expect("a DLMM account");

    let MeteoraDlmmAccount::LbPair(pair) = decoded.data else {
        panic!("expected LbPair");
    };
    assert_eq!(pair.token_y_mint, WSOL);
}
```

### Step 7: Write pipeline tests

`FixtureDatasource` sends saved updates and returns. Once it returns, the pipeline processes what's queued and `run()` returns, so the test can assert right after. The second test edits the saved transaction to check that failed transactions are ignored.

`tests/pipeline.rs`:

```rust
use {
    async_trait::async_trait,
    carbon_core::{
        datasource::{Datasource, DatasourceId, Update, UpdateType},
        error::{CarbonResult, Error},
        pipeline::Pipeline,
    },
    carbon_meteora_dlmm_decoder::MeteoraDlmmDecoder,
    solana_pubkey::Pubkey,
    swap_stats::{load_transaction, PairTotals, SwapStats},
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Sends saved updates once, then returns. The pipeline stops when it has processed them.
struct FixtureDatasource(Vec<Update>);

#[async_trait]
impl Datasource for FixtureDatasource {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        _cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        for update in &self.0 {
            sender
                .send((update.clone(), id.clone()))
                .await
                .map_err(|e| Error::Custom(e.to_string()))?;
        }
        Ok(())
    }

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

#[tokio::test]
async fn sums_both_swaps_of_a_routed_transaction() {
    let transaction = load_transaction("tests/fixtures/route.json");
    let stats = SwapStats::default();

    Pipeline::builder()
        .datasource(FixtureDatasource(vec![Update::Transaction(Box::new(transaction))]))
        .instruction(MeteoraDlmmDecoder, stats.clone())
        .build()
        .unwrap()
        .run()
        .await
        .unwrap();

    let totals = stats.totals.lock().unwrap();
    assert_eq!(totals.len(), 2);
    assert_eq!(
        totals[&Pubkey::from_str_const("Bnoev1Gp1o247d7zTwSbznRwqjkRmbrx3R6Qt4P16YbQ")],
        PairTotals { swaps: 1, amount_in: 91_444_688 }
    );
    assert_eq!(
        totals[&Pubkey::from_str_const("EZyszDEx1LZDt7TsSFV8xdPi49sDKC3mdfv2MVMEQLtU")],
        PairTotals { swaps: 1, amount_in: 405_807_562 }
    );
}

#[tokio::test]
async fn ignores_failed_transactions() {
    let mut transaction = load_transaction("tests/fixtures/route.json");
    transaction.meta.status = Err(solana_transaction_error::TransactionError::AccountInUse);
    let stats = SwapStats::default();

    Pipeline::builder()
        .datasource(FixtureDatasource(vec![Update::Transaction(Box::new(transaction))]))
        .instruction(MeteoraDlmmDecoder, stats.clone())
        .build()
        .unwrap()
        .run()
        .await
        .unwrap();

    assert!(stats.totals.lock().unwrap().is_empty());
}
```

### Step 8: Run the tests

```sh
cargo test
```

Output:

```text
     Running tests/decoder.rs (target/debug/deps/decoder-e08750c32c07deb0)

running 3 tests
test decodes_swap2_instruction_fixture ... ok
test decodes_swap2_data_from_base58 ... ok
test decodes_lb_pair_account_fixture ... ok

test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.03s

     Running tests/pipeline.rs (target/debug/deps/pipeline-f12b833c4b961edf)

running 2 tests
test sums_both_swaps_of_a_routed_transaction ... ok
test ignores_failed_transactions ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.04s
…
```

## What to test

- **Every variant you handle.** One fixture per instruction and account type your processors match on. A decoder regenerated from a newer IDL that changes a layout then fails a test instead of returning `None` in production.
- **CPIs, not only top-level calls.** Most DEX volume arrives through aggregators. The route fixture above has both swaps at stack height 2; a processor that only looked at top-level instructions would count zero.
- **Rejections.** Failed transactions, other programs' instructions, truncated data. See the rejection test in [Write a decoder by hand](https://carbon.sevenlabs.org/guides/custom-decoder.md#test-it).
- **Your write path.** For processors that write to a database, run the pipeline test against a disposable database and query it afterwards. With the [`postgres` feature](https://carbon.sevenlabs.org/guides/postgres.md), check that every row you expect is there: sibling CPIs can [share a primary key](https://carbon.sevenlabs.org/reference/postgres.md#instruction-primary-key).

## Test a processor without a pipeline

`process` is an ordinary async method, but its input borrows metadata, the decoded value, the nested instructions and the raw instruction. Building that by hand is more work than running the pipeline, so prefer the fixture datasource above. Logic that doesn't need the input, like `swap_amount` here, is easiest to test as a plain function.

## Common problems

### `Couldn't read fixture: No such file or directory`

Paths in tests are relative to the crate root (`CARGO_MANIFEST_DIR`), where `cargo test` runs them. Use `tests/fixtures/…`, or build the path with `concat!(env!("CARGO_MANIFEST_DIR"), "/tests/fixtures/route.json")` for tests that may run from elsewhere.

### `Couldn't deserialize fixture: invalid type: floating point …, expected u64`

`rent_epoch` was rounded by an old `jq`. Upgrade to jq 1.7 or newer, or edit the value back to `18446744073709551615`.

### `a getTransaction result: Error("invalid type: null …")`

The saved file is `null`: the RPC returned an error instead of the transaction. Capture it again and check the raw response.

### `Base64-encoded transaction` panic in `load_transaction`

The transaction was captured with `json` or `jsonParsed` encoding. Capture it with `"encoding":"base64"`.

### The pipeline test hangs

The datasource never returns, or keeps a clone of `sender` alive in a spawned task. The pipeline only stops when every sender is dropped or the cancellation token is cancelled.

### Decoding the account fixture returns `None`

The account was captured from a different program, or after it was closed and reopened with another layout. Check `owner` in the fixture and the first 8 bytes of `data` against the decoder's discriminator.

## Next steps

- [test-utils reference](https://carbon.sevenlabs.org/reference/test-utils.md): Every fixture type and reader.
- [Write a decoder by hand](https://carbon.sevenlabs.org/guides/custom-decoder.md): Decoders for programs without an IDL, with unit tests.
- [Write a custom datasource](https://carbon.sevenlabs.org/guides/custom-datasource.md): The same Datasource trait the fixture datasource implements.

---

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