Index a program across an IDL upgrade

Decode a program before and after a breaking IDL change by registering one generated decoder per version, each limited to its slot range.

When a program upgrade changes an instruction's layout, one decoder can't read both the old and the new data. You generate one decoder crate per IDL version, register both with the pipeline, and give each a SlotRangeFilter so it only sees the slots where its layout applied. A single processor turns both versions into one output type.

This guide uses a two-version program, simple_dex, whose swap instruction gains a fee_tier: u8 argument in V2. The pipeline replays four swaps around the upgrade from memory, so it runs without any network access. To index a real program, swap the replay datasource for a real one.

You need:

Rust 1.96.1 or newer.
Node.js 20.18.0 or newer, to run the Carbon CLI (@sevenlabs-hq/carbon-cli@2.0.0).
The IDL of each program version. Keep the old one when you upgrade; you can't fetch it from chain afterwards, because the on-chain IDL account only holds the latest version.
The last slot that ran the old program. See Find the upgrade slot.

How it works#

Every instruction reaches both pipes. Each pipe's filter runs before its decoder, so an instruction from slot 500 is decoded only as V1 and one from slot 501 only as V2. The filter is required, not an optimisation: generated decoders ignore trailing bytes, so the V1 decoder also decodes V2 data and returns a Swap without fee_tier. Without filters, every V2 swap would reach your processor twice.

Create the project and save both IDLs#

Terminal

cargo new simple-dex-indexer
cd simple-dex-indexer
mkdir idls

Save the IDL of each version. These are Anchor IDLs; the only change is the new fee_tier argument.

idls/simple_dex_v1.json

{
  "address": "Dex1111111111111111111111111111111111111111",
  "metadata": { "name": "simple_dex", "version": "0.1.0", "spec": "0.1.0" },
  "instructions": [
    {
      "name": "swap",
      "discriminator": [248, 198, 158, 145, 225, 117, 135, 200],
      "accounts": [
        { "name": "pool", "writable": true },
        { "name": "user", "signer": true }
      ],
      "args": [
        { "name": "amount_in", "type": "u64" },
        { "name": "min_amount_out", "type": "u64" }
      ]
    }
  ]
}

idls/simple_dex_v2.json

{
  "address": "Dex1111111111111111111111111111111111111111",
  "metadata": { "name": "simple_dex", "version": "0.2.0", "spec": "0.1.0" },
  "instructions": [
    {
      "name": "swap",
      "discriminator": [248, 198, 158, 145, 225, 117, 135, 200],
      "accounts": [
        { "name": "pool", "writable": true },
        { "name": "user", "signer": true }
      ],
      "args": [
        { "name": "amount_in", "type": "u64" },
        { "name": "min_amount_out", "type": "u64" },
        { "name": "fee_tier", "type": "u8" }
      ]
    }
  ]
}

Generate one decoder per version#

Run parse once per IDL. --version-name appends -v1 or -v2 to the crate name, so both crates can live in one dependency graph. The output directory is deleted before generation, so give each version its own directory.

Terminal

npx -y @sevenlabs-hq/carbon-cli@2.0.0 parse --idl idls/simple_dex_v1.json --out-dir decoders/simple-dex-v1 --name simple-dex --version-name v1 --with-postgres false --with-graphql false
npx -y @sevenlabs-hq/carbon-cli@2.0.0 parse --idl idls/simple_dex_v2.json --out-dir decoders/simple-dex-v2 --name simple-dex --version-name v2 --with-postgres false --with-graphql false

Terminal

pnpm dlx @sevenlabs-hq/carbon-cli@2.0.0 parse --idl idls/simple_dex_v1.json --out-dir decoders/simple-dex-v1 --name simple-dex --version-name v1 --with-postgres false --with-graphql false
pnpm dlx @sevenlabs-hq/carbon-cli@2.0.0 parse --idl idls/simple_dex_v2.json --out-dir decoders/simple-dex-v2 --name simple-dex --version-name v2 --with-postgres false --with-graphql false

Terminal

bunx @sevenlabs-hq/carbon-cli@2.0.0 parse --idl idls/simple_dex_v1.json --out-dir decoders/simple-dex-v1 --name simple-dex --version-name v1 --with-postgres false --with-graphql false
bunx @sevenlabs-hq/carbon-cli@2.0.0 parse --idl idls/simple_dex_v2.json --out-dir decoders/simple-dex-v2 --name simple-dex --version-name v2 --with-postgres false --with-graphql false

Terminal

yarn dlx @sevenlabs-hq/carbon-cli@2.0.0 parse --idl idls/simple_dex_v1.json --out-dir decoders/simple-dex-v1 --name simple-dex --version-name v1 --with-postgres false --with-graphql false
yarn dlx @sevenlabs-hq/carbon-cli@2.0.0 parse --idl idls/simple_dex_v2.json --out-dir decoders/simple-dex-v2 --name simple-dex --version-name v2 --with-postgres false --with-graphql false

Output

✔ Decoder generated
✔ Code formatted successfully

✨ Decoder Generated Successfully! ✨
  • Output: /home/you/simple-dex-indexer/decoders/simple-dex-v1
  • Source: file (anchor)
…

Each crate is named carbon-simple-dex-decoder-v1 or -v2 (Rust paths carbon_simple_dex_decoder_v1 and _v2). Only the crate name changes: both contain the same type names, SimpleDexDecoder, SimpleDexInstruction and Swap, so you rename them on import. Each generated Cargo.toml has its own empty [workspace] table, which lets you use it as a path dependency as-is.

Add the dependencies#

Cargo.toml

[package]
name = "simple-dex-indexer"
version = "0.1.0"
edition = "2021"

[dependencies]
carbon-core = "2.0.0"
carbon-simple-dex-decoder-v1 = { path = "decoders/simple-dex-v1" }
carbon-simple-dex-decoder-v2 = { path = "decoders/simple-dex-v2" }

async-trait = "0.1.89"
env_logger = "0.11.8"
log = "0.4.28"
solana-hash = "4.5.0"
solana-message = "4.4.1"
solana-pubkey = "4.2.1"
solana-signature = { version = "3.4.1", features = ["rand"] }
solana-transaction = "4.1.6"
solana-transaction-status = "=4.2.2"
tokio = { version = "1.48.0", features = ["full"] }
tokio-util = "0.7.17"

The solana-* crates and async-trait are only needed by the replay datasource at the bottom of main.rs.

Write the pipeline#

LAST_V1_SLOT is the boundary. The V1 pipe accepts slots up to and including it, the V2 pipe every slot after it. SwapProcessor implements Processor once per decoder's instruction type and converts both into the same Swap struct, with fee_tier: None for V1.

src/main.rs

use {
    async_trait::async_trait,
    carbon_core::{
        datasource::{Datasource, DatasourceId, TransactionUpdate, Update, UpdateType},
        error::{CarbonResult, Error},
        filter::SlotRangeFilter,
        instruction::InstructionProcessorInputType,
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_simple_dex_decoder_v1::{
        instructions::SimpleDexInstruction as V1Instruction, SimpleDexDecoder as V1Decoder,
        PROGRAM_ID,
    },
    carbon_simple_dex_decoder_v2::{
        instructions::SimpleDexInstruction as V2Instruction, SimpleDexDecoder as V2Decoder,
    },
    solana_hash::Hash,
    solana_message::{
        compiled_instruction::CompiledInstruction, legacy::Message, MessageHeader, VersionedMessage,
    },
    solana_pubkey::Pubkey,
    solana_signature::Signature,
    solana_transaction::versioned::VersionedTransaction,
    solana_transaction_status::TransactionStatusMeta,
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Last slot that ran the V1 program. Every transaction from the next slot on uses V2.
const LAST_V1_SLOT: u64 = 500;

/// One swap, whichever program version produced it.
#[derive(Debug)]
struct Swap {
    signature: Signature,
    slot: u64,
    pool: Pubkey,
    amount_in: u64,
    min_amount_out: u64,
    fee_tier: Option<u8>,
}

struct SwapProcessor;

impl SwapProcessor {
    fn handle(&mut self, swap: Swap) -> CarbonResult<()> {
        log::info!(
            "slot={} sig={}… pool={}… amount_in={} min_out={} fee_tier={:?}",
            swap.slot,
            &swap.signature.to_string()[..8],
            &swap.pool.to_string()[..8],
            swap.amount_in,
            swap.min_amount_out,
            swap.fee_tier,
        );
        Ok(())
    }
}

impl Processor<InstructionProcessorInputType<'_, V1Instruction>> for SwapProcessor {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, V1Instruction>,
    ) -> CarbonResult<()> {
        let V1Instruction::Swap { data, accounts, .. } = input.decoded_instruction;
        self.handle(Swap {
            signature: input.metadata.transaction_metadata.signature,
            slot: input.metadata.transaction_metadata.slot,
            pool: accounts.pool,
            amount_in: data.amount_in,
            min_amount_out: data.min_amount_out,
            fee_tier: None,
        })
    }
}

impl Processor<InstructionProcessorInputType<'_, V2Instruction>> for SwapProcessor {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, V2Instruction>,
    ) -> CarbonResult<()> {
        let V2Instruction::Swap { data, accounts, .. } = input.decoded_instruction;
        self.handle(Swap {
            signature: input.metadata.transaction_metadata.signature,
            slot: input.metadata.transaction_metadata.slot,
            pool: accounts.pool,
            amount_in: data.amount_in,
            min_amount_out: data.min_amount_out,
            fee_tier: Some(data.fee_tier),
        })
    }
}

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

    Pipeline::builder()
        .datasource(ReplayDatasource::new())
        .instruction_with_filters(
            V1Decoder,
            SwapProcessor,
            vec![Box::new(SlotRangeFilter::to(LAST_V1_SLOT, None))],
        )
        .instruction_with_filters(
            V2Decoder,
            SwapProcessor,
            vec![Box::new(SlotRangeFilter::from(LAST_V1_SLOT + 1, None))],
        )
        .build()?
        .run()
        .await
}

/// Replays four swaps around the upgrade, then stops. Replace it with a real datasource.
struct ReplayDatasource {
    updates: Vec<Update>,
}

impl ReplayDatasource {
    fn new() -> Self {
        let pool = Pubkey::new_unique();
        Self {
            updates: vec![
                swap_tx(499, swap_data(1_000_000, 990_000, None), pool),
                swap_tx(500, swap_data(2_000_000, 1_980_000, None), pool),
                swap_tx(501, swap_data(3_000_000, 2_970_000, Some(5)), pool),
                swap_tx(502, swap_data(4_000_000, 3_960_000, Some(10)), pool),
            ],
        }
    }
}

#[async_trait]
impl Datasource for ReplayDatasource {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        _cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        for update in &self.updates {
            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]
    }
}

const SWAP_DISCRIMINATOR: [u8; 8] = [248, 198, 158, 145, 225, 117, 135, 200];

fn swap_data(amount_in: u64, min_amount_out: u64, fee_tier: Option<u8>) -> Vec<u8> {
    let mut data = SWAP_DISCRIMINATOR.to_vec();
    data.extend_from_slice(&amount_in.to_le_bytes());
    data.extend_from_slice(&min_amount_out.to_le_bytes());
    data.extend(fee_tier);
    data
}

fn swap_tx(slot: u64, data: Vec<u8>, pool: Pubkey) -> Update {
    let user = Pubkey::new_unique();
    let signature = Signature::new_unique();
    let message = VersionedMessage::Legacy(Message {
        header: MessageHeader {
            num_required_signatures: 1,
            num_readonly_signed_accounts: 0,
            num_readonly_unsigned_accounts: 1,
        },
        account_keys: vec![user, pool, PROGRAM_ID],
        recent_blockhash: Hash::default(),
        instructions: vec![CompiledInstruction {
            program_id_index: 2,
            accounts: vec![1, 0],
            data,
        }],
    });

    Update::Transaction(Box::new(TransactionUpdate {
        signature,
        transaction: VersionedTransaction {
            signatures: vec![signature],
            message,
        },
        meta: TransactionStatusMeta::default(),
        is_vote: false,
        slot,
        index: None,
        block_time: None,
        block_hash: None,
    }))
}

Run it#

Terminal

RUST_LOG=info cargo run

Output

[… INFO  carbon_core::pipeline] starting pipeline. num_datasources: 1, num_exporters: 0, num_account_pipes: 0, num_account_deletion_pipes: 0, num_instruction_pipes: 2, num_transaction_pipes: 0
[… INFO  simple_dex_indexer] slot=499 sig=u3p7abzh… pool=11157t3s… amount_in=1000000 min_out=990000 fee_tier=None
[… INFO  simple_dex_indexer] slot=500 sig=2QXH3Ngt… pool=11157t3s… amount_in=2000000 min_out=1980000 fee_tier=None
[… INFO  simple_dex_indexer] slot=501 sig=45pyVgCR… pool=11157t3s… amount_in=3000000 min_out=2970000 fee_tier=Some(5)
[… INFO  simple_dex_indexer] slot=502 sig=5dJJo4Ni… pool=11157t3s… amount_in=4000000 min_out=3960000 fee_tier=Some(10)
[… INFO  carbon_core::pipeline] update_receiver closed, shutting down.
[… INFO  carbon_core::pipeline] pipeline shutdown complete.

Signatures and pubkeys differ on each run. The replay datasource returns after sending its four updates, so the pipeline stops on its own.

Verify the routing#

Each swap must appear exactly once. Slots up to 500 print fee_tier=None (V1), later slots print fee_tier=Some(..) (V2). If you see a slot twice, both filters accept it: check that the V2 filter starts at LAST_V1_SLOT + 1.

To check against real data, run the indexer over a range that spans the upgrade and compare the number of swaps per slot with a block explorer.

Find the upgrade slot#

A program's ProgramData account records the slot of its most recent deployment. The Solana CLI prints it as Last Deployed In Slot:

Terminal

solana program show <PROGRAM_ID> --url https://api.mainnet-beta.solana.com

That is only the latest upgrade. For an older one, find the upgrade transaction in the ProgramData account's history (its signatures are listed by getSignaturesForAddress on that address) and take its slot.

Use the upgrade transaction's slot as LAST_V1_SLOT. Agave makes upgraded program code visible from the slot after the upgrade, so every successful transaction in the upgrade slot itself still ran the old program.

Choose slot or transaction-index boundaries#

SlotRangeFilter::to(slot, index) and SlotRangeFilter::from(slot, index) also take a transaction index inside the slot. At the boundary slot, from accepts index >= from_index and to accepts index < to_index; outside the boundary slot only the slot matters. When an update has no index (None), both filters accept every transaction in the boundary slot.

Datasources differ in what they provide:

Datasource	`TransactionUpdate::index`
Yellowstone gRPC, Helius LaserStream	Position in the block
RPC block crawler	Position in the block
RPC block subscribe	Position in the list the RPC returned. With a `mentionsAccountOrProgram` filter that list is filtered, so this is not the block position.
Helius GTFA	`transactionIndex` from the API
RPC transaction crawler, Helius Atlas WS	Always `None`

Because upgraded code takes effect at a slot boundary, a slot boundary with None is exact and works with every datasource. Use a transaction index only if you need to split inside a slot for another reason, and only with a datasource that provides one.

Version accounts too#

Account layouts change in upgrades as well. If both versions keep the same discriminator, register both account decoders with the same pair of filters (account_with_filters). SlotRangeFilter compares the slot of the account update, not the slot the account was last written in the old layout, so an account that the program migrates lazily can still hold V1 bytes after the boundary. In that case, decide by content instead: try the V2 decoder first and fall back to V1, or compare account.data.len() with each layout's size in a custom AccountDecoder.

Common problems#

Every V2 swap is printed twice#

The V1 pipe has no filter, or its range overlaps V2. Generated decoders stop reading after the last field and ignore extra bytes, so V1 decodes V2 data. Give both pipes a SlotRangeFilter with ranges that don't overlap.

V2 swaps are missing for the first slots after the upgrade#

The V2 decoder rejects V1 data (it is one byte short), so this is a boundary that is too late: LAST_V1_SLOT is after the real upgrade slot. Recheck the upgrade transaction's slot.