Helius gPA v2

HeliusGpaV2Datasource calls Helius' getProgramAccountsV2 repeatedly, following paginationKey until the last page, and emits every account as an account update. Pages keep memory flat on programs with millions of accounts, and changedSinceSlot lets you fetch only accounts modified since your last load. Use it instead of GpaDatasource when you're on Helius or the program is large.

Quick example#

This program loads every Orca Whirlpool pool account, 5,000 per page, and prints each pool's mints. It compiles as shown; to run it, set HELIUS_API_KEY.

cargo new helius-gpa && cd helius-gpa
cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-orca-whirlpool-decoder@2.0.0 carbon-helius-gpa-v2-datasource@2.0.0
cargo add solana-client@=4.2.2
cargo add tokio@1 --features full
cargo add env_logger@0.11 log@0.4

use {
    carbon_core::{
        account::AccountProcessorInputType, error::CarbonResult, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_helius_gpa_v2_datasource::{HeliusGpaV2Config, HeliusGpaV2Datasource},
    carbon_log_metrics::LogMetrics,
    carbon_orca_whirlpool_decoder::{
        accounts::OrcaWhirlpoolAccount, OrcaWhirlpoolDecoder, PROGRAM_ID,
    },
    solana_client::{rpc_config::RpcProgramAccountsConfig, rpc_filter::RpcFilterType},
    std::{env, sync::Arc},
};

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

    let config = HeliusGpaV2Config::new(
        Some(RpcProgramAccountsConfig {
            // Only `Whirlpool` pool accounts (653 bytes).
            filters: Some(vec![RpcFilterType::DataSize(653)]),
            ..Default::default()
        }),
        None,       // changed_since_slot: return every matching account
        Some(5000), // accounts per page (max 10000)
    );

    let datasource = HeliusGpaV2Datasource::new_with_config(
        format!(
            "https://mainnet.helius-rpc.com/?api-key={}",
            env::var("HELIUS_API_KEY").expect("HELIUS_API_KEY must be set")
        ),
        PROGRAM_ID,
        config,
    );

    Pipeline::builder()
        .datasource(datasource)
        .metrics(Arc::new(LogMetrics::new()))
        .account(OrcaWhirlpoolDecoder, PoolLogger)
        .build()?
        .run()
        .await
}

struct PoolLogger;

impl Processor<AccountProcessorInputType<'_, OrcaWhirlpoolAccount>> for PoolLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, OrcaWhirlpoolAccount>,
    ) -> CarbonResult<()> {
        if let OrcaWhirlpoolAccount::Whirlpool(pool) = &input.decoded_account.data {
            println!(
                "{} {} / {} liquidity={}",
                input.metadata.pubkey, pool.token_mint_a, pool.token_mint_b, pool.liquidity
            );
        }
        Ok(())
    }
}

HELIUS_API_KEY=your-helius-api-key RUST_LOG=info cargo run

The program prints one line per pool, logs Reached end of pagination, and exits. With an invalid key it stops at once:

[ERROR carbon_core::pipeline] error consuming datasource: FailedToConsumeDatasource("Failed to parse response: error decoding response body")

Configuration#

Constructors#

pub fn new(helius_rpc_url: String, program_id: Pubkey) -> Self
pub fn new_with_config(helius_rpc_url: String, program_id: Pubkey, config: HeliusGpaV2Config) -> Self

helius_rpc_url is the full URL including ?api-key=. new uses HeliusGpaV2Config::default().

HeliusGpaV2Config#

pub fn new(
    program_accounts_config: Option<RpcProgramAccountsConfig>,
    changed_since_slot: Option<u64>,
    limit: Option<u32>,
) -> Self

Filters are the standard RpcFilterType values from solana_client::rpc_filter: DataSize, Memcmp, TokenAccountState.

Behaviour#

• Paging. Pages are fetched one after another. Each account update gets the slot of its page's response, so accounts from different pages can carry different slots.
• Backpressure. Updates are sent with send().await; nothing is dropped when the pipeline is busy.
• Finishing. consume returns after the page with no paginationKey. A pipeline with only this datasource then shuts down.
• Errors. There are no retries. An HTTP failure, a body that isn't JSON, an RPC error (RPC error: … (code: …)) or a response without context/value stops the datasource with FailedToConsumeDatasource; accounts from earlier pages have already been emitted. Accounts whose pubkey, owner or data can't be parsed are skipped without a log line.
• Cancellation. Checked before each page and between accounts.
• Metrics. helius_gpa_v2_pages_fetched_total, helius_gpa_v2_accounts_processed_total (counters), helius_gpa_v2_fetch_duration_milliseconds (histogram).

Troubleshooting#

Failed to parse response: error decoding response body#

Helius answered with something other than JSON-RPC, usually 401 Unauthorized for a missing or invalid API key. Check that the URL ends in ?api-key=<your key>.

RPC error: … (code: …)#

Helius rejected the request, for example an invalid filter or a plan without getProgramAccountsV2. The message is Helius' own.

Incremental loads miss accounts#

A getProgramAccounts query only returns accounts that still exist, so accounts closed since your last load never appear. Combine it with a stream (for example LaserStream account filters) to see deletions.

See also#

• Load account state
• RPC getProgramAccounts
• Helius GTFA for transaction history
• Datasources concept and update types