Store decoded data in Postgres

Carbon's postgres feature gives you processors that write every decoded instruction or account to PostgreSQL, the migrations that create the tables, and row types for reading them back. This guide stores Meteora DLMM instructions from mainnet, first in typed tables (one table per instruction, one column per field), then in a single generic JSONB table, and shows what each looks like in the database.

You need:

• Rust 1.96.1 or newer.
• PostgreSQL. This guide was run against PostgreSQL 16.4.
• A decoder crate with a postgres feature. Every decoder Carbon publishes has one, and the CLI generates it unless you pass --with-postgres false.

Choose a storage model#

Both models upsert on a primary key, so writing the same update twice leaves one row.

docker run -d --name carbon-pg -e POSTGRES_USER=carbon -e POSTGRES_PASSWORD=carbon -e POSTGRES_DB=carbon -p 5432:5432 postgres:16
export DATABASE_URL=postgres://carbon:carbon@localhost:5432/carbon

cargo new dlmm-postgres
cd dlmm-postgres
cargo add carbon-core@2.0.0 --features postgres
cargo add carbon-meteora-dlmm-decoder@2.0.0 --features postgres
cargo add carbon-rpc-transaction-crawler-datasource@2.0.0 carbon-log-metrics@2.0.0
cargo add sqlx@0.8.6 --features postgres,runtime-tokio-rustls
cargo add sqlx_migrator@0.17.0 --features postgres
cargo add tokio --features full
cargo add env_logger log

[dependencies]
carbon-core = { version = "2.0.0", features = ["postgres"] }
carbon-log-metrics = "2.0.0"
carbon-meteora-dlmm-decoder = { version = "2.0.0", features = ["postgres"] }
carbon-rpc-transaction-crawler-datasource = "2.0.0"
env_logger = "0.11.11"
log = "0.4.35"
sqlx = { version = "0.8.6", features = ["postgres", "runtime-tokio-rustls"] }
sqlx_migrator = { version = "0.17.0", features = ["postgres"] }
tokio = { version = "1.54.1", features = ["full"] }

use {
    carbon_core::{
        error::{CarbonResult, Error},
        pipeline::Pipeline,
        postgres::processors::PostgresInstructionProcessor,
    },
    carbon_log_metrics::LogMetrics,
    carbon_meteora_dlmm_decoder::{
        instructions::{
            postgres::{MeteoraDlmmInstructionWithMetadata, MeteoraDlmmInstructionsMigration},
            MeteoraDlmmInstruction,
        },
        MeteoraDlmmDecoder, PROGRAM_ID,
    },
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    sqlx::postgres::PgPoolOptions,
    sqlx_migrator::{Info, Migrate, Migrator, Plan},
    std::sync::Arc,
};

fn db_error(e: impl std::fmt::Display) -> Error {
    Error::Custom(format!("postgres: {e}"))
}

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

    let database_url = std::env::var("DATABASE_URL")
        .map_err(|_| Error::Custom("DATABASE_URL must be set".to_string()))?;
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .map_err(db_error)?;

    // Create one table per instruction (swap_instruction, swap2_instruction, ...).
    let mut migrator = Migrator::default();
    migrator
        .add_migration(Box::new(MeteoraDlmmInstructionsMigration))
        .map_err(db_error)?;
    let mut conn = pool.acquire().await.map_err(db_error)?;
    migrator
        .run(&mut *conn, &Plan::apply_all())
        .await
        .map_err(db_error)?;
    drop(conn);

    let datasource = RpcTransactionCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        PROGRAM_ID,
        ConnectionConfig::default(),
        Filters::new(None, None, None),
        None,
    );

    Pipeline::builder()
        .datasource(datasource)
        .metrics(Arc::new(LogMetrics::new()))
        .instruction(
            MeteoraDlmmDecoder,
            PostgresInstructionProcessor::<MeteoraDlmmInstruction, MeteoraDlmmInstructionWithMetadata>::new(pool),
        )
        .build()?
        .run()
        .await
}

RUST_LOG=info cargo run

[… INFO  carbon_core::pipeline] starting pipeline. num_datasources: 1, num_exporters: 1, num_account_pipes: 0, num_account_deletion_pipes: 0, num_instruction_pipes: 1, num_transaction_pipes: 0
…
[… INFO  carbon_log_metrics] 00:01:55 (+4.999493119s) | 42 processed (100%), 42 successful, 0 failed (0%), 0 in queue
…
[… INFO  carbon_log_metrics]   postgres_instructions_upsert_upserted: 12
[… INFO  carbon_log_metrics]   postgres_instructions_upsert_failed: 0

psql "$DATABASE_URL" -c "SELECT left(__signature, 12) AS signature, __instruction_index, __stack_height, __slot, amount_in, min_amount_out FROM swap2_instruction ORDER BY __slot DESC LIMIT 3"

  signature   | __instruction_index | __stack_height |  __slot   | amount_in | min_amount_out 
--------------+---------------------+----------------+-----------+-----------+----------------
 57gWxR71yzrh |                   2 |              2 | 467111933 |  24356122 |              0
 Ez2XgG9LDAvj |                   2 |              2 | 467111933 | 124337400 |              0
 UMQXvmmhKaHw |                   0 |              2 | 467111932 |  26993569 |              0
(3 rows)

psql "$DATABASE_URL" -c '\d swap2_instruction'

                     Table "public.swap2_instruction"
         Column          |     Type      | Collation | Nullable | Default 
-------------------------+---------------+-----------+----------+---------
 amount_in               | numeric(20,0) |           | not null | 
 min_amount_out          | numeric(20,0) |           | not null | 
 remaining_accounts_info | jsonb         |           | not null | 
 __signature             | text          |           | not null | 
 __instruction_index     | bigint        |           | not null | 
 __stack_height          | bigint        |           | not null | 
 __slot                  | numeric(20,0) |           |          | 
 __accounts              | jsonb         |           | not null | 
Indexes:
    "swap2_instruction_pkey" PRIMARY KEY, btree (__signature, __instruction_index, __stack_height)

Store everything in one JSONB table#

Swap the processor and the migration to write every instruction of every program to one instructions table. The decoder's instruction enum must implement serde::Serialize and Deserialize; its postgres feature turns on serde.

use {
    carbon_core::{
        error::{CarbonResult, Error},
        pipeline::Pipeline,
        postgres::{processors::PostgresJsonInstructionProcessor, rows::GenericInstructionMigration},
    },
    carbon_meteora_dlmm_decoder::{instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID},
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    sqlx::postgres::PgPoolOptions,
    sqlx_migrator::{Info, Migrate, Migrator, Plan},
};

fn db_error(e: impl std::fmt::Display) -> Error {
    Error::Custom(format!("postgres: {e}"))
}

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

    let database_url = std::env::var("DATABASE_URL")
        .map_err(|_| Error::Custom("DATABASE_URL must be set".to_string()))?;
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await
        .map_err(db_error)?;

    // One generic `instructions` table with a JSONB column.
    let mut migrator = Migrator::default();
    migrator
        .add_migration(Box::new(GenericInstructionMigration))
        .map_err(db_error)?;
    let mut conn = pool.acquire().await.map_err(db_error)?;
    migrator
        .run(&mut *conn, &Plan::apply_all())
        .await
        .map_err(db_error)?;
    drop(conn);

    let datasource = RpcTransactionCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        PROGRAM_ID,
        ConnectionConfig::default(),
        Filters::new(None, None, None),
        None,
    );

    Pipeline::builder()
        .datasource(datasource)
        .instruction(
            MeteoraDlmmDecoder,
            PostgresJsonInstructionProcessor::<MeteoraDlmmInstruction>::new(pool),
        )
        .build()?
        .run()
        .await
}

cargo run --bin jsonb
psql "$DATABASE_URL" -c "SELECT left(__signature, 12) AS signature, __instruction_index, __stack_height, decoded_instruction->>'type' AS type FROM instructions ORDER BY __slot DESC LIMIT 4"

  signature   | __instruction_index | __stack_height |   type   
--------------+---------------------+----------------+----------
 SjGeh3aLxKbB |                   3 |              2 | Swap2
 SjGeh3aLxKbB |                   3 |              3 | CpiEvent
 4AaBRNtD8ZV2 |                   0 |              2 | Swap
 4AaBRNtD8ZV2 |                   0 |              3 | CpiEvent
(4 rows)

decoded_instruction is the enum serialized with serde as {"type": "<Variant>", "data": {"program_id": …, "data": {…}, "accounts": {…}}}, so the arguments are under data.data:

psql "$DATABASE_URL" -c "SELECT left(__signature, 12) AS signature, (decoded_instruction->'data'->'data'->>'amount_in')::numeric AS amount_in FROM instructions WHERE decoded_instruction->>'type' IN ('Swap', 'Swap2') ORDER BY __slot DESC LIMIT 3"

  signature   | amount_in 
--------------+-----------
 SjGeh3aLxKbB | 169262461
 4AaBRNtD8ZV2 |  13025584
 2ya9GepZqk9Y |  99050000
(3 rows)

The table has no index on the JSON. Add one for the fields you filter on, for example CREATE INDEX ON instructions ((decoded_instruction->>'type'));.

Store accounts#

Accounts work the same way with the account processors and migrations. Register them on a pipeline whose datasource emits account updates, such as RpcProgramSubscribe, RpcGpa or Yellowstone; see Load account state. The transaction crawler above never emits them.

use carbon_meteora_dlmm_decoder::accounts::{
    postgres::{MeteoraDlmmAccountWithMetadata, MeteoraDlmmAccountsMigration},
    MeteoraDlmmAccount,
};

migrator
    .add_migration(Box::new(MeteoraDlmmAccountsMigration))
    .map_err(db_error)?;

// on the pipeline builder:
.account(
    MeteoraDlmmDecoder,
    PostgresAccountProcessor::<MeteoraDlmmAccount, MeteoraDlmmAccountWithMetadata>::new(pool.clone()),
)

PostgresAccountProcessor comes from carbon_core::postgres::processors. Account tables (lb_pair_account, …) are keyed by __pubkey, so each account keeps its latest state, with __slot set to the slot of the last update written. For the generic model, use PostgresJsonAccountProcessor::<MeteoraDlmmAccount> with carbon_core::postgres::rows::GenericAccountsMigration (table accounts).

Account processors don't handle closures. To delete closed accounts, add .account_deletions(..) with a processor that deletes the row, for example with the generated row type's Delete implementation (LbPairRow::delete(pubkey.into(), &pool) deletes from lb_pair_account).

What the decoder's postgres module provides#

With the feature on, a generated decoder adds, under accounts::postgres and instructions::postgres:

• <Program>AccountWithMetadata and <Program>InstructionWithMetadata: the W type for the typed processors. Their Upsert and Insert match on the variant and write the matching row.
• <Program>AccountsMigration and <Program>InstructionsMigration: sqlx_migrator migrations that create every table (CREATE TABLE IF NOT EXISTS).
• One <Name>Row per account and instruction, with from_parts, TryFrom<Row> back to the decoded type, and Insert, Upsert, Delete and Lookup from carbon_core::postgres::operations.

Use the row types to read data back in your own code:

use carbon_core::postgres::operations::Lookup;
use carbon_meteora_dlmm_decoder::accounts::postgres::LbPairRow;

let row = LbPairRow::lookup(pubkey.into(), &pool).await?; // Option<LbPairRow>

The full list, column types and the generic row types are in the Postgres reference.

Migrations#

sqlx_migrator records applied migrations in _sqlx_migrator_migrations and skips them on the next run:

psql "$DATABASE_URL" -c "SELECT app, name FROM _sqlx_migrator_migrations"

     app      |           name            
--------------+---------------------------
 meteora-dlmm | meteora_dlmm_instructions
 meteora-dlmm | meteora_dlmm_accounts
 carbon_core  | generic_instructions

Things to know before you rely on them:

• They only create tables. A newer decoder version with a changed layout doesn't alter existing tables, and its migration has the same name, so it's skipped. Write your own migration (a type implementing sqlx_migrator::Migration) with the ALTER TABLE statements, or create the new tables in a fresh schema.
• No indexes beyond the primary key. Add the ones your queries need in your own migration.
• Run them once at startup, before the pipeline, on one connection, as in main.rs. Several replicas starting at once can race; run migrations from one replica or a separate job.

Connection settings#

• Pool size. The pipeline processes one update at a time, so a processor uses one connection at a time; max_connections(5) leaves room for other code. A GraphQL server on the same pool needs more.
• Latency. Each instruction is a separate INSERT … ON CONFLICT round trip, awaited before the next update. At 1 ms per write, a pipeline tops out near 1,000 instructions per second. Keep the database close to the indexer, watch postgres_instructions_upsert_duration_milliseconds, and batch writes in your own processor if you need more.
• Errors. A failed write returns Err, which the pipeline logs (error processing update …), counts in carbon_updates_failed_total and postgres_*_upsert_failed, and then moves on. The update is not retried. Alert on these metrics; see Monitor with Prometheus.

Common problems#

Two swaps in one transaction, but only one row#

Instruction tables are keyed by (__signature, __instruction_index, __stack_height), and __instruction_index is the index of the top-level instruction. Two CPIs of the same program at the same depth under the same top-level instruction get the same key, and the second upsert overwrites the first. This happens with every multi-hop aggregator route: replaying the Jupiter transaction SZE5CFK6… (two DLMM Swap2 CPIs under top-level instruction 4) leaves one row:

  signature   | __instruction_index | __stack_height | amount_in 
--------------+---------------------+----------------+-----------
 SZE5CFK6gkG9 |                   4 |              2 | 405807562
(1 row)

The first swap (amount_in 91444688) is lost. The same happens to the program's two CpiEvents per swap. If you need every CPI, write your own processor and key rows on InstructionMetadata::absolute_path (for example [4, 0] and [4, 1]). See Instruction primary key.

relation "lb_pair_account" does not exist#

The account migration wasn't run. Add MeteoraDlmmAccountsMigration (or GenericAccountsMigration for the generic model) to the migrator.

Error: Custom("postgres: pool timed out while waiting for an open connection")#

The program couldn't connect within sqlx's default 30-second acquire timeout: PostgreSQL isn't running, or DATABASE_URL has the wrong host or port. Test it with psql "$DATABASE_URL" -c 'select 1'. During a run, the same error means every pooled connection is busy; raise max_connections if other code shares the pool.

Error: Custom("postgres: error returned from database: role \"…\" does not exist")#

The user, password or database name in DATABASE_URL is wrong (database "…" does not exist and password authentication failed have the same cause). Create the role and database, or fix the URL.

Pubkeys in __accounts or JSONB columns are arrays of numbers#

That's how the decoders serialize Pubkey with serde. Typed pubkey columns are BYTEA. To filter, compare with the bytes: in SQL, decode('<hex>', 'hex') for BYTEA columns; in Rust, bind sqlx::types::Json(pubkey.to_bytes()) against a JSONB element, as the GraphQL guide does.

Values above 2^63 fail to read as i64#

u64 fields and __slot are NUMERIC(20). Read them as carbon_core::postgres::primitives::U64 or sqlx::types::BigDecimal, or cast in SQL.

Next steps#