# Write a custom datasource (Carbon 2.0)

> Implement the Datasource trait to feed Carbon from any source, emit transactions and accounts, honour cancellation, or adapt a channel with StreamMessageClient.

A datasource is anything that implements `carbon_core::datasource::Datasource`: it runs in its own task, turns data from somewhere into [`Update`](https://carbon.sevenlabs.org/concepts/updates.md)s and sends them into the pipeline. Write one when no [built-in datasource](https://carbon.sevenlabs.org/datasources.md) reads from your source, for example a provider API, a queue, files on disk, or an RPC method Carbon doesn't wrap.

This guide builds three, all runnable against Solana's public RPC endpoint:

- [`SignaturePoller`](#write-a-transaction-datasource), which polls a program's signatures and emits each transaction.
- [`AccountPoller`](#emit-account-updates), which watches a list of accounts and emits account updates and deletions.
- An [adapter over a channel](#adapt-a-channel-with-streammessageclient) with `carbon-stream-message-datasource`, when you already have a producer and only need to hand its output to Carbon.

**You need:** Rust 1.96.1 or newer. Read [Datasources](https://carbon.sevenlabs.org/concepts/datasources.md) for how the pipeline runs them.

## The trait

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

`Datasource` uses `#[async_trait]` (the pipeline stores datasources as `dyn Datasource`), so annotate your `impl` with `#[async_trait]` from the `async-trait` crate. `Processor`, by contrast, uses a plain `async fn`.

The contract:

| Part                 | What you do                                                                                                                                                                                                                                             |
| -------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `consume`            | Runs once, in its own Tokio task, when `Pipeline::run` starts. Produce updates and send `(update, id.clone())` on `sender`.                                                                                                                             |
| `id`                 | The datasource's identity. Send it back unchanged with every update; [`DatasourceFilter`](https://carbon.sevenlabs.org/concepts/filters.md) uses it.                                                                                                    |
| `sender`             | Bounded channel into the pipeline ([`channel_buffer_size`](https://carbon.sevenlabs.org/reference/pipeline-builder.md#channel_buffer_size), 1,000 by default). `send(..).await` waits while it's full. An `Err` means the pipeline has stopped: return. |
| `cancellation_token` | Cancelled on ctrl-C or when your own token passed to `datasource_cancellation_token` is cancelled. Return promptly when it fires, including while you wait on the network or a timer.                                                                   |
| Return value         | `Ok(())` when the source is exhausted. An `Err` is logged as `error consuming datasource: …`. Either way the task ends, and once every datasource task has ended and dropped its sender, the pipeline drains the channel and `run()` returns.           |
| `update_types`       | The `UpdateType`s you emit. The 2.0.0 pipeline doesn't check it; keep it accurate as documentation.                                                                                                                                                     |

## Write a transaction datasource

`SignaturePoller` calls `getSignaturesForAddress` for one address, fetches each new transaction with `getTransaction`, and sends it as an `Update::Transaction`. It's a smaller version of the repository's [`custom-datasource` example](https://github.com/sevenlabs-hq/carbon/tree/v2.0.0/examples/custom-datasource) and of what [`RpcTransactionCrawler`](https://carbon.sevenlabs.org/datasources/rpc-transaction-crawler.md) does with more concurrency and retries.

### Step 1: Create the project

```sh
cargo new poller
cd poller
cargo add carbon-core@2.0.0 carbon-meteora-dlmm-decoder@2.0.0 carbon-token-program-decoder@2.0.0 carbon-stream-message-datasource@2.0.0
cargo add solana-client@=4.2.2 solana-transaction-status@=4.2.2 solana-commitment-config@3.1.1 solana-pubkey@4.2.1 solana-signature@3.4.1
cargo add async-trait tokio-util env_logger log
cargo add tokio --features full
```

`Cargo.toml`:

```toml
[dependencies]
async-trait = "0.1.92"
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
carbon-stream-message-datasource = "2.0.0"
carbon-token-program-decoder = "2.0.0"
env_logger = "0.11.11"
log = "0.4.35"
solana-client = "=4.2.2"
solana-commitment-config = "3.1.1"
solana-pubkey = "4.2.1"
solana-signature = "3.4.1"
solana-transaction-status = "=4.2.2"
tokio = { version = "1.54.1", features = ["full"] }
tokio-util = "0.7.20"
```

Patch versions of the unpinned crates may be newer. Use the Solana crate versions Carbon 2.0 depends on (the Agave 4.2 cohort), so each type resolves to one version. A second `solana-pubkey` or `solana-transaction-status` in the graph produces "expected `Pubkey`, found `Pubkey`" errors.

### Step 2: Write the datasource

`src/main.rs`:

```rust
use {
    async_trait::async_trait,
    carbon_core::{
        datasource::{Datasource, DatasourceId, TransactionUpdate, Update, UpdateType},
        error::CarbonResult,
        instruction::InstructionProcessorInputType,
        pipeline::Pipeline,
        processor::Processor,
        transformers::transaction_metadata_from_original_meta,
    },
    carbon_meteora_dlmm_decoder::{
        instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID,
    },
    solana_client::{
        nonblocking::rpc_client::RpcClient, rpc_client::GetConfirmedSignaturesForAddress2Config,
        rpc_config::RpcTransactionConfig,
    },
    solana_commitment_config::CommitmentConfig,
    solana_pubkey::Pubkey,
    solana_signature::Signature,
    solana_transaction_status::UiTransactionEncoding,
    std::{str::FromStr, time::Duration},
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Polls `getSignaturesForAddress` for one program and emits each new transaction.
pub struct SignaturePoller {
    pub rpc_url: String,
    pub address: Pubkey,
    pub poll_interval: Duration,
    /// Maximum signatures fetched per poll. Anything older than the newest
    /// `max_per_poll` signatures since the last poll is skipped.
    pub max_per_poll: usize,
}

#[async_trait]
impl Datasource for SignaturePoller {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        let rpc = RpcClient::new_with_commitment(self.rpc_url.clone(), CommitmentConfig::confirmed());
        let mut newest_seen: Option<Signature> = None;

        loop {
            let config = GetConfirmedSignaturesForAddress2Config {
                before: None,
                until: newest_seen,
                limit: Some(self.max_per_poll),
                commitment: Some(CommitmentConfig::confirmed()),
            };

            // Stop waiting on the RPC as soon as the pipeline shuts down.
            let result = tokio::select! {
                _ = cancellation_token.cancelled() => return Ok(()),
                result = rpc.get_signatures_for_address_with_config(&self.address, config) => result,
            };

            match result {
                Ok(signatures) => {
                    if let Some(first) = signatures.first() {
                        newest_seen = Signature::from_str(&first.signature).ok();
                    }

                    // The RPC returns newest first; emit in chain order.
                    for info in signatures.iter().rev() {
                        if cancellation_token.is_cancelled() {
                            return Ok(());
                        }
                        if info.err.is_some() {
                            continue; // skip failed transactions
                        }
                        let Ok(signature) = Signature::from_str(&info.signature) else {
                            continue;
                        };
                        let index = info.transaction_index.map(u64::from);
                        let Some(update) = fetch_transaction(&rpc, signature, index).await else {
                            continue;
                        };
                        // `send` waits when the pipeline is busy instead of dropping the update.
                        if sender.send((update, id.clone())).await.is_err() {
                            return Ok(()); // the pipeline has stopped
                        }
                    }
                }
                Err(error) => log::warn!("getSignaturesForAddress failed: {error}"),
            }

            tokio::select! {
                _ = cancellation_token.cancelled() => return Ok(()),
                _ = tokio::time::sleep(self.poll_interval) => {}
            }
        }
    }

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

async fn fetch_transaction(
    rpc: &RpcClient,
    signature: Signature,
    index: Option<u64>,
) -> Option<Update> {
    let config = RpcTransactionConfig {
        encoding: Some(UiTransactionEncoding::Base64), // binary encoding is required
        commitment: Some(CommitmentConfig::confirmed()),
        max_supported_transaction_version: Some(1),     // accept legacy, V0 and V1
    };
    let fetched = match rpc.get_transaction_with_config(&signature, config).await {
        Ok(fetched) => fetched,
        Err(error) => {
            log::warn!("getTransaction {signature} failed: {error}");
            return None;
        }
    };

    let transaction = fetched.transaction.transaction.decode()?;
    let meta = transaction_metadata_from_original_meta(fetched.transaction.meta?).ok()?;

    Some(Update::Transaction(Box::new(TransactionUpdate {
        signature,
        transaction,
        meta,
        is_vote: false,
        slot: fetched.slot,
        index, // position in the block, from getSignaturesForAddress
        block_time: fetched.block_time,
        block_hash: None,
    })))
}

struct InstructionLogger;

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for InstructionLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        let debug = format!("{:?}", input.decoded_instruction);
        let name = debug.split([' ', '{', '(']).next().unwrap_or_default();
        let metadata = &input.metadata.transaction_metadata;
        println!("slot={} {name} {}", metadata.slot, metadata.signature);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();

    let datasource = SignaturePoller {
        rpc_url: "https://api.mainnet-beta.solana.com".to_string(),
        address: PROGRAM_ID,
        poll_interval: Duration::from_secs(5),
        max_per_poll: 10,
    };

    Pipeline::builder()
        .datasource(datasource)
        .instruction(MeteoraDlmmDecoder, InstructionLogger)
        .build()?
        .run()
        .await
}
```

What each part does:

- **Cancellation.** Every wait (the signatures request and the sleep) sits in a `tokio::select!` with `cancellation_token.cancelled()`, and the inner loop checks `is_cancelled()` between transactions. Without that, ctrl-C waits for the current RPC call and the full poll interval.
- **Building the update.** `getTransaction` must use a binary encoding (`Base64`) and `max_supported_transaction_version: Some(1)`. `decode()` rebuilds the `VersionedTransaction`, and `transaction_metadata_from_original_meta` converts the RPC's `UiTransactionStatusMeta` into the `TransactionStatusMeta` Carbon expects. Carbon can't rebuild a transaction from `Json` or `JsonParsed` responses.
- **Metadata.** `index` is the transaction's position in its block, which `getSignaturesForAddress` returns on current RPC nodes; [`SlotRangeFilter`](https://carbon.sevenlabs.org/reference/filters.md) uses it. `block_hash` stays `None` because `getTransaction` doesn't return it.
- **Backpressure.** `sender.send(..).await` waits for room in the channel, so a slow processor slows the poller down instead of losing updates. Several built-in streaming datasources use `try_send` and drop updates instead; see [Run Carbon in production](https://carbon.sevenlabs.org/guides/production.md#backpressure-and-the-channel).
- **Order.** The RPC lists signatures newest first; the loop reverses them so processors see chain order.

### Step 3: Run it

```sh
cargo run
```

Output:

```text
slot=467109371 Swap2 2EWz1tSnoreYujahprX6jibXvRXCcmXJCKeoVWzGgSSf8iFVGmh5yYZPQXd3zbX7d2usEPpWyhYv9Z5aoV8TVsxf
slot=467109371 CpiEvent 2EWz1tSnoreYujahprX6jibXvRXCcmXJCKeoVWzGgSSf8iFVGmh5yYZPQXd3zbX7d2usEPpWyhYv9Z5aoV8TVsxf
slot=467109371 CpiEvent 2EWz1tSnoreYujahprX6jibXvRXCcmXJCKeoVWzGgSSf8iFVGmh5yYZPQXd3zbX7d2usEPpWyhYv9Z5aoV8TVsxf
slot=467109396 RemoveLiquidityByRange2 5mZ9yjJxJHBKee7fHZC51ejcq2JgEDd1nZCVskHFQnM6Tm6evV3cacEhaJRpDkHs6CEXzSDmgHXEaeYGSTLZ9efL
slot=467109396 ClaimFee2 5mZ9yjJxJHBKee7fHZC51ejcq2JgEDd1nZCVskHFQnM6Tm6evV3cacEhaJRpDkHs6CEXzSDmgHXEaeYGSTLZ9efL
slot=467109396 ClosePositionIfEmpty 5mZ9yjJxJHBKee7fHZC51ejcq2JgEDd1nZCVskHFQnM6Tm6evV3cacEhaJRpDkHs6CEXzSDmgHXEaeYGSTLZ9efL
slot=467109396 Swap 4sCsDTY5X3RyFBSiAihqutGYwF1N2pyxbnAe9wg5ZitCnde8RZqyXb2HsAgfeMA2yjyXrw1akUaa4cRaLwwdiZPi
…
```

`CpiEvent` lines are the program's self-CPI events, decoded as instructions. Press ctrl-C to stop:

Output:

```text
[… INFO  carbon_core::pipeline] shutting down the pipeline after processing pending updates.
[… INFO  carbon_core::pipeline] pipeline shutdown complete.
```

The public endpoint rate-limits; with `RUST_LOG=solana_rpc_client=debug` you see `Too many requests: server responded with … status: 429` while the client backs off and retries. Use your own RPC URL for anything beyond a test.

This poller is deliberately simple. It skips a transaction whose `getTransaction` call fails, and it skips older signatures when more than `max_per_poll` arrive between two polls. A production version retries failed fetches and pages backwards with `before` until it reaches `until`.

## Emit account updates

For accounts, build an `AccountUpdate` and send `AccountUpdate::into_update()`, not `Update::Account(..)`. `into_update` returns `Update::AccountDeletion` for an account with zero lamports and `Update::Account` otherwise, the same rule every built-in datasource applies. Account decoders then never see closed accounts, and `.account_deletions(..)` processors receive them with the account's final state.

Add this file next to `main.rs`. It polls `getMultipleAccounts` for the USDC and USDT mints and sends an update whenever an account changes.

`src/bin/accounts.rs`:

```rust
use {
    async_trait::async_trait,
    carbon_core::{
        account::AccountProcessorInputType,
        datasource::{AccountDeletion, AccountUpdate, Datasource, DatasourceId, Update, UpdateType},
        error::CarbonResult,
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_token_program_decoder::{accounts::TokenProgramAccount, TokenProgramDecoder},
    solana_client::nonblocking::rpc_client::RpcClient,
    solana_commitment_config::CommitmentConfig,
    solana_pubkey::Pubkey,
    std::{collections::HashMap, time::Duration},
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Polls `getMultipleAccounts` and emits an update whenever an account changes.
pub struct AccountPoller {
    pub rpc_url: String,
    pub accounts: Vec<Pubkey>,
    pub poll_interval: Duration,
}

#[async_trait]
impl Datasource for AccountPoller {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        let rpc = RpcClient::new(self.rpc_url.clone());
        let mut last_seen = HashMap::new();

        loop {
            let result = tokio::select! {
                _ = cancellation_token.cancelled() => return Ok(()),
                result = rpc.get_multiple_accounts_with_commitment(
                    &self.accounts,
                    CommitmentConfig::confirmed(),
                ) => result,
            };

            match result {
                Ok(response) => {
                    let slot = response.context.slot;
                    for (pubkey, account) in self.accounts.iter().zip(response.value) {
                        // A missing account is reported with zero lamports, so
                        // `into_update` turns it into an `AccountDeletion`.
                        let account = account.unwrap_or_default();
                        if last_seen.get(pubkey) == Some(&account) {
                            continue; // unchanged since the last poll
                        }
                        last_seen.insert(*pubkey, account.clone());

                        let update = AccountUpdate {
                            pubkey: *pubkey,
                            account,
                            slot,
                            transaction_signature: None,
                        }
                        .into_update();

                        if sender.send((update, id.clone())).await.is_err() {
                            return Ok(());
                        }
                    }
                }
                Err(error) => log::warn!("getMultipleAccounts failed: {error}"),
            }

            tokio::select! {
                _ = cancellation_token.cancelled() => return Ok(()),
                _ = tokio::time::sleep(self.poll_interval) => {}
            }
        }
    }

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

struct MintLogger;

impl Processor<AccountProcessorInputType<'_, TokenProgramAccount>> for MintLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, TokenProgramAccount>,
    ) -> CarbonResult<()> {
        if let TokenProgramAccount::Mint(mint) = &input.decoded_account.data {
            println!(
                "slot={} mint={} supply={} decimals={}",
                input.metadata.slot, input.metadata.pubkey, mint.supply, mint.decimals
            );
        }
        Ok(())
    }
}

struct DeletionLogger;

impl Processor<AccountDeletion> for DeletionLogger {
    async fn process(&mut self, deletion: &AccountDeletion) -> CarbonResult<()> {
        println!("slot={} closed {}", deletion.slot, deletion.pubkey);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();

    let datasource = AccountPoller {
        rpc_url: "https://api.mainnet-beta.solana.com".to_string(),
        accounts: vec![
            Pubkey::from_str_const("EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"), // USDC mint
            Pubkey::from_str_const("Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB"), // USDT mint
        ],
        poll_interval: Duration::from_secs(2),
    };

    Pipeline::builder()
        .datasource(datasource)
        .account(TokenProgramDecoder, MintLogger)
        .account_deletions(DeletionLogger)
        .build()?
        .run()
        .await
}
```

```sh
cargo run --bin accounts
```

Output:

```text
slot=467109863 mint=EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v supply=8450898376555615 decimals=6
slot=467109863 mint=Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB supply=3839903059562649 decimals=6
slot=467109958 mint=EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v supply=8450897696579348 decimals=6
…
```

The first poll emits every account; later polls emit only the ones whose data or lamports changed. Set `transaction_signature` when your source knows which transaction wrote the account: [`DeduplicationFilter`](https://carbon.sevenlabs.org/concepts/filters.md) uses it to drop duplicate account updates and accepts every update without one.

## Adapt a channel with `StreamMessageClient`

If you already have code that produces Carbon updates (a Kafka consumer, a websocket client, a replay from files), you don't need to implement `Datasource`. `carbon-stream-message-datasource` wraps a `tokio::sync::mpsc::Receiver<UnifiedMessage>`:

```rust
pub enum UnifiedMessage {
    Account(AccountUpdate),
    Transaction(Box<TransactionUpdate>),
}
```

`StreamMessageClient::new(receiver)` forwards each message to the pipeline, calling `into_update()` on accounts. It waits for room in the pipeline's channel, and it ends when the sender side is dropped, which stops the pipeline.

`src/bin/stream.rs`:

```rust
use {
    carbon_core::{
        account::AccountProcessorInputType,
        datasource::AccountUpdate,
        error::CarbonResult,
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_stream_message_datasource::{StreamMessageClient, UnifiedMessage},
    carbon_token_program_decoder::{accounts::TokenProgramAccount, TokenProgramDecoder},
    solana_client::nonblocking::rpc_client::RpcClient,
    solana_pubkey::Pubkey,
};

const USDC_MINT: Pubkey = Pubkey::from_str_const("EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v");

struct MintLogger;

impl Processor<AccountProcessorInputType<'_, TokenProgramAccount>> for MintLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, TokenProgramAccount>,
    ) -> CarbonResult<()> {
        if let TokenProgramAccount::Mint(mint) = &input.decoded_account.data {
            println!("slot={} supply={}", input.metadata.slot, mint.supply);
        }
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();

    let (tx, rx) = tokio::sync::mpsc::channel::<UnifiedMessage>(1_000);

    // Your producer: a Kafka consumer, a websocket, a file replay... Here, three RPC reads.
    tokio::spawn(async move {
        let rpc = RpcClient::new("https://api.mainnet-beta.solana.com".to_string());
        for _ in 0..3 {
            if let Ok(response) = rpc
                .get_account_with_commitment(&USDC_MINT, rpc.commitment())
                .await
            {
                if let Some(account) = response.value {
                    let update = AccountUpdate {
                        pubkey: USDC_MINT,
                        account,
                        slot: response.context.slot,
                        transaction_signature: None,
                    };
                    if tx.send(UnifiedMessage::Account(update)).await.is_err() {
                        break;
                    }
                }
            }
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        }
        // Dropping `tx` closes the stream; the pipeline then finishes.
    });

    Pipeline::builder()
        .datasource(StreamMessageClient::new(rx))
        .account(TokenProgramDecoder, MintLogger)
        .build()?
        .run()
        .await
}
```

```sh
cargo run --bin stream
```

Output:

```text
slot=467109975 supply=8450897696579348
slot=467109985 supply=8450897592422230
slot=467109993 supply=8450897592422230
```

The program exits by itself after the third message. A `StreamMessageClient` can be consumed once: its receiver moves into the pipeline on the first `consume` call. It doesn't carry `BlockDetails`; implement `Datasource` if you need them.

## Add metrics to your datasource

Define `Counter`s or `Histogram`s as statics and register them in `MetricsRegistry::global()` at the start of `consume`, as the built-in datasources do. They appear in [`LogMetrics`](https://carbon.sevenlabs.org/concepts/metrics.md) output and on a [Prometheus endpoint](https://carbon.sevenlabs.org/guides/prometheus.md):

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

static FETCH_FAILURES: Counter = Counter::new(
    "poller_fetch_failures_total",
    "getTransaction calls that failed",
);

// first line of `consume`:
MetricsRegistry::global().register_counter(&FETCH_FAILURES);

// where a fetch fails:
FETCH_FAILURES.inc();
```

## Common problems

### ``lifetime parameters or bounds on method `consume` do not match the trait declaration``

The `impl Datasource` block is missing `#[async_trait]`. Add `async-trait` to `Cargo.toml` and put `#[async_trait]` on the `impl`.

### ctrl-C doesn't stop the program until a request finishes

`consume` awaits something without also watching the token. Wrap every long wait in `tokio::select!` with `cancellation_token.cancelled()`.

### The pipeline exits right after starting

`consume` returned. That's correct for a finite source, but a stream must loop until it's cancelled. Also check the log for `error consuming datasource:`: an `Err` ends the task the same way. If `consume` spawns a task and returns `Ok(())`, move the `sender` into that task; the pipeline keeps running while a sender is alive.

### Processors never see a closed account

You sent `Update::Account` for a zero-lamport account. Use `AccountUpdate { .. }.into_update()` and register `.account_deletions(..)`.

### `error processing update (…): MissingFeePayer`

The `VersionedTransaction` you sent has no account keys, usually because it was built by hand. Every transaction needs its message's account keys; the first is the fee payer.

### Every transaction is skipped

`fetched.transaction.transaction.decode()` returns `None` for transactions requested as `Json` or `JsonParsed`, so `fetch_transaction` returns early. Request `UiTransactionEncoding::Base64`.

### Top-level instructions decode but CPIs never do

The update's `meta.inner_instructions` is `None` or empty. Carbon builds the CPI tree from it, so a source that strips inner instructions only ever yields top-level instructions. Keep the full metadata when you build `TransactionUpdate`.

## Next steps

- [Datasources](https://carbon.sevenlabs.org/concepts/datasources.md): How the pipeline runs datasources.
- [Updates](https://carbon.sevenlabs.org/concepts/updates.md): Every field of TransactionUpdate, AccountUpdate and the other updates.
- [Test decoders and processors](https://carbon.sevenlabs.org/guides/testing.md): Feed saved transactions through a pipeline without a network.
- [Run Carbon in production](https://carbon.sevenlabs.org/guides/production.md): Backpressure, reconnects and shutdown.

---

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