Load account state

Streams tell you what changes; to index a program's accounts you first need their current state. Carbon's state loaders read every account a program owns, decode it with your account decoder, and stop. This guide loads Meteora DLMM pools with getProgramAccounts on the public endpoint, then shows Helius gPA v2 for large programs and incremental syncs.

You need: Rust 1.96.1 or newer. rpc-gpa works with any RPC endpoint, including https://api.mainnet-beta.solana.com when you filter the request. Helius gPA v2 needs a Helius API key.

cargo new dlmm-state
cd dlmm-state

[package]
name = "dlmm-state"
version = "0.1.0"
edition = "2024"

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
carbon-rpc-gpa-datasource = "2.0.0"
env_logger = "0.11.11"
log = "0.4.35"
solana-client = "=4.2.2"
solana-pubkey = "4.2.1"
tokio = { version = "1.54.1", features = ["full"] }

use {
    carbon_core::{
        account::AccountProcessorInputType, error::CarbonResult, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_meteora_dlmm_decoder::{accounts::MeteoraDlmmAccount, MeteoraDlmmDecoder, PROGRAM_ID},
    carbon_rpc_gpa_datasource::GpaDatasource,
    solana_client::{
        rpc_config::{CommitmentConfig, RpcAccountInfoConfig, RpcProgramAccountsConfig},
        rpc_filter::{Memcmp, RpcFilterType},
    },
    solana_pubkey::{pubkey, Pubkey},
};

const LB_PAIR_DISCRIMINATOR: [u8; 8] = [33, 11, 49, 98, 181, 101, 177, 13];
const WSOL: Pubkey = pubkey!("So11111111111111111111111111111111111111112");
const USDC: Pubkey = pubkey!("EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v");

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

    // SOL/USDC `LbPair` accounts: the account discriminator at byte 0,
    // token X mint at byte 88 and token Y mint at byte 120.
    let config = RpcProgramAccountsConfig {
        filters: Some(vec![
            RpcFilterType::Memcmp(Memcmp::new_raw_bytes(0, LB_PAIR_DISCRIMINATOR.to_vec())),
            RpcFilterType::Memcmp(Memcmp::new_raw_bytes(88, WSOL.to_bytes().to_vec())),
            RpcFilterType::Memcmp(Memcmp::new_raw_bytes(120, USDC.to_bytes().to_vec())),
        ]),
        account_config: RpcAccountInfoConfig {
            commitment: Some(CommitmentConfig::confirmed()),
            ..Default::default()
        },
        ..Default::default()
    };

    let datasource = GpaDatasource::new_with_config(
        "https://api.mainnet-beta.solana.com".to_string(),
        PROGRAM_ID,
        config,
    );

    Pipeline::builder()
        .datasource(datasource)
        .account(MeteoraDlmmDecoder, PairLogger)
        .build()?
        .run()
        .await
}

struct PairLogger;

impl Processor<AccountProcessorInputType<'_, MeteoraDlmmAccount>> for PairLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, MeteoraDlmmAccount>,
    ) -> CarbonResult<()> {
        if let MeteoraDlmmAccount::LbPair(pair) = &input.decoded_account.data {
            println!(
                "slot {} pair {}: bin_step={} active_id={}",
                input.metadata.slot, input.metadata.pubkey, pair.bin_step, pair.active_id
            );
        }
        Ok(())
    }
}

RUST_LOG=info cargo run

[2026-11-09T00:13:42Z INFO  carbon_core::pipeline] starting pipeline. num_datasources: 1, num_exporters: 0, num_account_pipes: 1, num_account_deletion_pipes: 0, num_instruction_pipes: 0, num_transaction_pipes: 0
[2026-11-09T00:13:43Z INFO  carbon_rpc_gpa_datasource] Fetched 113 accounts for program LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo (slot: 467112209)
slot 467112209 pair 1jw5fDodwGEGBVqNXsx2eqiLgNmgMDEeXWSbrTreLCM: bin_step=10 active_id=-2056
slot 467112209 pair 5rCf1DM8LjKTw4YqhnoLcngyZYeNnQqztScTogYHAS6: bin_step=4 active_id=-6110
slot 467112209 pair BGm1tav58oGcsQJehL9WXBFXF7D27vZsKefj4xJKD5Y: bin_step=10 active_id=-2445
…
slot 467112209 pair Hk96GjGpo1VMotNCC5WAZ6CcNfvgPVaD4zv1mDVx47nt: bin_step=250 active_id=-90
[2026-11-09T00:13:43Z INFO  carbon_core::pipeline] update_receiver closed, shutting down.
[2026-11-09T00:13:43Z INFO  carbon_core::pipeline] pipeline shutdown complete.

Verify the load#

The datasource logs how many accounts the node returned (Fetched 113 accounts …). Count the accounts your processor handles: if it's lower, some accounts didn't decode, usually because the filter also matches other account types or the decoder is outdated for the program. Any account's current data can be checked with getAccountInfo or a block explorer.

How rpc-gpa works#

• One request. GpaDatasource sends a single getProgramAccounts call and sends every account into the pipeline. When it's done, the datasource finishes; a pipeline with no other datasource then exits.
• One slot. All accounts carry the response's context slot in input.metadata.slot. transaction_signature is None.
• Encoding and context are set for you. The datasource always requests Base64 and withContext, whatever your config says. filters, commitment and min_context_slot are passed through.
• Everything is in memory. The node builds the whole response and the datasource holds it until it's sent. Filter as tightly as you can.
• Don't use data_slice. Decoders need each account's full data. A sliced account fails to decode and is skipped without a message.

GpaDatasource::new(url, program) sends no filters at all. That works for programs with few accounts; on large programs the public endpoint refuses or times out.

Write filters#

All filters must match. The discriminator of each account type is in the generated decoder's accounts/<type>.rs (decode compares data[0..8] with it). Field offsets follow from the account's layout: 8 bytes of discriminator, then each field in order with its Borsh size.

Large programs with Helius gPA v2#

Helius's getProgramAccountsV2 returns accounts in pages and can return only the accounts that changed since a slot. The datasource fetches page after page and stops after the last one.

let datasource = GpaDatasource::new_with_config(
    "https://api.mainnet-beta.solana.com".to_string(),
    PROGRAM_ID,
    config, // RpcProgramAccountsConfig with your filters
);

[dependencies]
carbon-helius-gpa-v2-datasource = "2.0.0"

use carbon_helius_gpa_v2_datasource::{HeliusGpaV2Config, HeliusGpaV2Datasource};

let datasource = HeliusGpaV2Datasource::new_with_config(
    std::env::var("HELIUS_RPC_URL").expect("HELIUS_RPC_URL must be set"),
    PROGRAM_ID,
    HeliusGpaV2Config::new(
        Some(config), // the same RpcProgramAccountsConfig
        None,         // changed_since_slot
        Some(5_000),  // accounts per page (default 1,000, max 10,000)
    ),
);

With Helius you can drop the mint filters and page through every DLMM pool. Everything else in the program stays the same, and the processor sees the same AccountProcessorInputType. Differences from rpc-gpa:

• Each page has its own slot. input.metadata.slot is the context slot of the page the account came in, so slots increase during a long load.
• It waits for the pipeline between accounts, so memory stays bounded to one page.
• No retries. An HTTP or RPC error stops the datasource with FailedToConsumeDatasource("RPC error: …"). Restart the load, or use incremental syncs below so a restart costs less.

Sync only what changed#

changed_since_slot returns only the accounts modified since that slot. Run a full load once, store the highest slot you saw, then run incremental loads from it, for example on a schedule:

let last_synced_slot: u64 = 467_112_209; // read it from your database

let config = HeliusGpaV2Config::new(
    Some(lb_pair_filters),
    Some(last_synced_slot),
    Some(5_000),
);

After each run, store the highest input.metadata.slot your processor handled. Accounts closed since the last sync aren't returned, because they no longer exist; detect closures with a stream's account deletions or a periodic full load.

Keep the state current#

A load is a point-in-time copy. To keep it current, run it together with a stream of changes in the same pipeline, rpc-program-subscribe on any RPC or a Yellowstone account filter, and apply updates in slot order per account. Multiple datasources has a complete, runnable example. For a whole program at a past slot without any RPC load, use a validator snapshot.

Common problems#

error consuming datasource: FailedToConsumeDatasource("Failed to fetch program accounts")#

The RPC node rejected or dropped the request, and the datasource doesn't log why. Usual causes: no filters on a large program (the public endpoint refuses some programs entirely), a response too large to finish before the timeout, or rate limiting. Add filters, or use a paid endpoint or Helius gPA v2. To see the node's answer, send the same request with curl.

The processor sees fewer accounts than Fetched N accounts#

Some accounts didn't decode: other account types match the filter, a data_slice truncated the data, or the decoder doesn't match the program's current layout. Filter on the discriminator of the type you want.

The pipeline exits right after loading#

That's expected: the state loader is finite. Add a stream as a second datasource to keep running (see Keep the state current).

RPC error: … (code: …) from Helius gPA v2#

The API key or plan doesn't allow getProgramAccountsV2, or a filter is invalid. The load stops at that page; fix the cause and run it again.

Next steps#