Postgres API

Every public type in carbon_core::postgres — processors, row types, CRUD traits, migrations, metadata columns and sqlx primitives.

carbon_core::postgres persists decoded accounts and instructions to PostgreSQL through sqlx. Use this page when you write a Postgres-backed processor, query the tables Carbon creates, or implement your own row type. For a walkthrough, see Store data in Postgres.

The module is behind the postgres feature of carbon-core:

Cargo.toml

[dependencies]
carbon-core = { version = "2.0.0", features = ["postgres"] }
# A generated decoder with its own typed tables:
carbon-meteora-dlmm-decoder = { version = "2.0.0", features = ["postgres"] }
sqlx = { version = "0.8.6", features = ["runtime-tokio-rustls", "postgres"] }
sqlx_migrator = { version = "0.17.0", features = ["postgres"] }
async-trait = "0.1.89"

The module has five submodules:

Module	Contents
`processors`	Four `Processor` implementations that upsert every decoded update.
`rows`	`AccountRow<T>` and `InstructionRow<T>` (JSONB tables) and their migrations.
`operations`	The `Table`, `Insert`, `Upsert`, `Delete` and `Lookup` traits.
`metadata`	The `__`-prefixed metadata columns every row carries.
`primitives`	sqlx newtypes for `Pubkey` and unsigned and 128-bit integers.

Two storage models#

Carbon offers two ways to store data. Pick one per pipeline.

	Typed tables	Generic JSONB tables
Processors	`PostgresAccountProcessor`, `PostgresInstructionProcessor`	`PostgresJsonAccountProcessor`, `PostgresJsonInstructionProcessor`
Tables	One per account type (`<name>_account`) and instruction type (`<name>_instruction`), with one column per field	`accounts` and `instructions`, decoded body in a `JSONB` column
Migrations	`<Program>AccountsMigration`, `<Program>InstructionsMigration` from the decoder crate	`GenericAccountsMigration`, `GenericInstructionMigration`
Requires	Decoder built with its `postgres` feature	Decoded type implements `serde::Serialize` and `Deserialize` (the decoder's `serde` feature)
Works with GraphQL	Yes, the decoder's generated `QueryRoot` reads these tables	No generated schema; write your own queries

Processors#

carbon_core::postgres::processors. Source: processors.rs.

Every processor:

calls upsert once per decoded update and returns its error, so a failed write is logged by the pipeline and counted in carbon_updates_failed_total; the pipeline keeps running;
counts successes and failures and records the upsert duration in milliseconds. See Postgres metrics for the metric names;
holds a sqlx::PgPool (cheap to clone; pass pool.clone() to each processor).

`PostgresAccountProcessor`#

Upserts each decoded account through a wrapper type W that knows how to write it.

Rust

pub struct PostgresAccountProcessor<T, W> { /* private fields */ }

impl<T, W> PostgresAccountProcessor<T, W> {
    pub fn new(pool: sqlx::PgPool) -> Self
}

impl<'a, T, W> Processor<AccountProcessorInputType<'a, T>> for PostgresAccountProcessor<T, W>
where
    T: Clone + Send + Sync + 'static,
    W: From<(T, AccountMetadata)> + Upsert + Send + 'static,

Details. For every update the processor clones decoded_account.data and the AccountMetadata, builds W::from((data, metadata)) and calls W::upsert(&pool). Lamports, owner and the raw account bytes are not passed to W.

Type parameter	Description
`T`	The decoder's account type, for example `MeteoraDlmmAccount`.
`W`	The row wrapper. Generated decoders provide `<Program>AccountWithMetadata`, which dispatches to one typed row per account variant.

Example. Register a generated decoder's typed account tables:

Rust

.account(
    MeteoraDlmmDecoder,
    PostgresAccountProcessor::<MeteoraDlmmAccount, MeteoraDlmmAccountWithMetadata>::new(pool.clone()),
)

See the complete program and a hand-written wrapper.

`PostgresInstructionProcessor`#

Upserts each decoded instruction through a wrapper type W.

Rust

pub struct PostgresInstructionProcessor<T, W> { /* private fields */ }

impl<T, W> PostgresInstructionProcessor<T, W> {
    pub fn new(pool: sqlx::PgPool) -> Self
}

impl<'a, T, W> Processor<InstructionProcessorInputType<'a, T>> for PostgresInstructionProcessor<T, W>
where
    T: Clone + Send + Sync + 'static,
    W: From<(T, InstructionMetadata, Vec<AccountMeta>)> + Upsert + Send + 'static,

Details. Builds W::from((decoded_instruction, metadata, raw_instruction.accounts)) and calls upsert. The third tuple element is the instruction's full Vec<solana_instruction::AccountMeta>; generated rows store it in an __accounts JSONB column. Nested instructions are not passed to W; each inner instruction reaches the processor as its own update.

Type parameter	Description
`T`	The decoder's instruction enum, for example `MeteoraDlmmInstruction`.
`W`	The row wrapper, for example the generated `MeteoraDlmmInstructionWithMetadata`.

Example.

Rust

.instruction(
    MeteoraDlmmDecoder,
    PostgresInstructionProcessor::<MeteoraDlmmInstruction, MeteoraDlmmInstructionWithMetadata>::new(pool.clone()),
)

`PostgresJsonAccountProcessor`#

Upserts each decoded account into the generic accounts table as JSONB.

Rust

pub struct PostgresJsonAccountProcessor<T> { /* private fields */ }

impl<T> PostgresJsonAccountProcessor<T> {
    pub fn new(pool: sqlx::PgPool) -> Self
}

impl<'a, T> Processor<AccountProcessorInputType<'a, T>> for PostgresJsonAccountProcessor<T>
where
    T: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone + Send + Sync + Unpin + 'static,

Details. Builds an AccountRow<T> and calls its upsert, which runs INSERT INTO accounts (__pubkey, __slot, data) … ON CONFLICT (__pubkey) DO UPDATE. Each pubkey keeps one row holding its latest state; the update with the higher slot does not win automatically, the last one processed does. Create the table first with GenericAccountsMigration. Generated account enums serialize as {"type": "<Variant>", "data": {…}}, so filter with data->>'type' = 'LbPair'. Pubkey fields inside the JSON are arrays of 32 numbers.

Example.

Rust

.account(
    TokenProgramDecoder,
    PostgresJsonAccountProcessor::<TokenProgramAccount>::new(pool.clone()),
)

See also. AccountRow, Two storage models.

`PostgresJsonInstructionProcessor`#

Upserts each decoded instruction into the generic instructions table as JSONB.

Rust

pub struct PostgresJsonInstructionProcessor<T> { /* private fields */ }

impl<T> PostgresJsonInstructionProcessor<T> {
    pub fn new(pool: sqlx::PgPool) -> Self
}

impl<'a, T> Processor<InstructionProcessorInputType<'a, T>> for PostgresJsonInstructionProcessor<T>
where
    T: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone + Send + Sync + Unpin + 'static,

Details. Builds an InstructionRow<T> and upserts it on (__signature, __instruction_index, __stack_height). The instruction's accounts are not stored separately; they are in the JSON only if the decoded type includes them (generated instruction enums do, under accounts). Create the table with GenericInstructionMigration. See the complete program.

Example.

Rust

.instruction(
    MeteoraDlmmDecoder,
    PostgresJsonInstructionProcessor::<MeteoraDlmmInstruction>::new(pool.clone()),
)

See also. Instruction primary key, InstructionRow.

`register_postgres_metrics`#

Registers the six Postgres metrics with the global MetricsRegistry.

Rust

pub fn register_postgres_metrics()

Details. PipelineBuilder::build calls it whenever carbon-core is compiled with postgres, so you never call it yourself. The metrics therefore appear in every exporter's output when the feature is on, even if no Postgres processor is registered. Calling it again registers the metrics a second time and duplicates them in snapshots.

Example.

Rust

// Only needed if you read MetricsRegistry::global() without building a pipeline.
carbon_core::postgres::processors::register_postgres_metrics();

See also. Metrics reference.

Rows#

carbon_core::postgres::rows. Source: rows.rs.

The generic tables these rows map to:

SQL

CREATE TABLE IF NOT EXISTS accounts (
    __pubkey BYTEA NOT NULL,
    __slot NUMERIC,
    data JSONB NOT NULL,
    PRIMARY KEY (__pubkey)
);

CREATE TABLE IF NOT EXISTS instructions (
    __signature TEXT NOT NULL,
    __instruction_index BIGINT NOT NULL,
    __stack_height BIGINT NOT NULL,
    __slot NUMERIC,
    decoded_instruction JSONB NOT NULL,
    PRIMARY KEY (__signature, __instruction_index, __stack_height)
);

`AccountRow`#

One row of the generic accounts table.

Rust

#[derive(sqlx::FromRow, Debug, Clone)]
pub struct AccountRow<T>
where
    T: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone + Send + Sync + Unpin + 'static,
{
    #[sqlx(flatten)]
    pub metadata: AccountRowMetadata,
    pub data: sqlx::types::Json<T>,
}

impl<T> AccountRow<T> {
    pub fn from_parts(source: T, metadata: AccountMetadata) -> Self
}

Details.

Field	Type	Column	Description
`metadata`	`AccountRowMetadata`	`__pubkey`, `__slot`	Account address and slot.
`data`	`sqlx::types::Json<T>`	`data`	The decoded account, serialized with `serde_json`.

Trait implementations:

Trait	SQL	Key
`Insert`	`INSERT INTO accounts …`; fails on an existing pubkey	—
`Upsert`	`… ON CONFLICT (__pubkey) DO UPDATE SET __slot = $2, data = $3`	—
`Delete`	`DELETE FROM accounts WHERE __pubkey = $1`	`primitives::Pubkey`
`Lookup`	`SELECT * FROM accounts WHERE __pubkey = $1`	`primitives::Pubkey`

Database errors are returned as Error::Custom with the sqlx message.

Example. Read back what PostgresJsonAccountProcessor wrote:

Rust

use carbon_core::postgres::{operations::Lookup, primitives::Pubkey, rows::AccountRow};

let row = AccountRow::<MeteoraDlmmAccount>::lookup(Pubkey(pubkey), &pool).await?;
if let Some(row) = row {
    println!("slot={:?} data={:?}", row.metadata.slot, row.data.0);
}

`InstructionRow`#

One row of the generic instructions table.

Rust

#[derive(sqlx::FromRow, Debug, Clone)]
pub struct InstructionRow<T>
where
    T: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone + Send + Sync + Unpin + 'static,
{
    #[sqlx(flatten)]
    pub metadata: InstructionRowMetadata,
    pub decoded_instruction: sqlx::types::Json<T>,
}

impl<T> InstructionRow<T> {
    pub fn from_parts(decoded_instruction: T, metadata: InstructionMetadata) -> Self
}

Details.

Field	Type	Column	Description
`metadata`	`InstructionRowMetadata`	`__signature`, `__instruction_index`, `__stack_height`, `__slot`	Where the instruction sits.
`decoded_instruction`	`sqlx::types::Json<T>`	`decoded_instruction`	The decoded instruction as JSON.

Insert, Upsert (ON CONFLICT (__signature, __instruction_index, __stack_height) DO UPDATE SET __slot = $4, decoded_instruction = $5), Delete and Lookup are implemented. The Delete and Lookup key is (String, U32, U32): signature in base58, instruction index, stack height.

Example.

Rust

use carbon_core::postgres::{operations::Lookup, rows::InstructionRow};

// Top-level instruction 0 of a transaction.
let key = (signature.to_string(), 0u32.into(), 1u32.into());
let row = InstructionRow::<MeteoraDlmmInstruction>::lookup(key, &pool).await?;

Instruction primary key#

Instruction tables, generic and generated, use the primary key (__signature, __instruction_index, __stack_height). __instruction_index comes from InstructionMetadata::index, which for an inner instruction is the index of its top-level instruction, not its position among siblings.

`GenericAccountsMigration`#

sqlx_migrator migration that creates the generic accounts table.

Rust

pub struct GenericAccountsMigration;

impl sqlx_migrator::Migration<sqlx::Postgres> for GenericAccountsMigration

Details. App "carbon_core", name "generic_accounts", no parents, one operation: AccountRowMigrationOperation. Rolling it back drops the table.

Example.

Rust

use sqlx_migrator::{Info, Migrate, Plan};

let mut migrator = sqlx_migrator::Migrator::default();
migrator
    .add_migration(Box::new(GenericAccountsMigration))
    .map_err(|e| Error::Custom(e.to_string()))?;
let mut conn = pool.acquire().await.map_err(|e| Error::Custom(e.to_string()))?;
migrator
    .run(&mut *conn, &Plan::apply_all())
    .await
    .map_err(|e| Error::Custom(e.to_string()))?;

See also. GenericInstructionMigration.

`GenericInstructionMigration`#

sqlx_migrator migration that creates the generic instructions table.

Rust

pub struct GenericInstructionMigration;

impl sqlx_migrator::Migration<sqlx::Postgres> for GenericInstructionMigration

Details. App "carbon_core", name "generic_instructions", no parents, one operation: InstructionRowMigrationOperation.

Example.

Rust

migrator
    .add_migration(Box::new(GenericInstructionMigration))
    .map_err(|e| Error::Custom(e.to_string()))?;

See also. GenericAccountsMigration.

`AccountRowMigrationOperation`#

The sqlx_migrator::Operation behind GenericAccountsMigration.

Rust

pub struct AccountRowMigrationOperation;

impl sqlx_migrator::Operation<sqlx::Postgres> for AccountRowMigrationOperation

Details. up runs CREATE TABLE IF NOT EXISTS accounts (…) (see Rows); down runs DROP TABLE IF EXISTS accounts. Use it to include the table in a migration of your own.

Example.

Rust

fn operations(&self) -> Vec<Box<dyn sqlx_migrator::Operation<sqlx::Postgres>>> {
    vec![Box::new(AccountRowMigrationOperation)]
}

See also. InstructionRowMigrationOperation.

`InstructionRowMigrationOperation`#

The sqlx_migrator::Operation behind GenericInstructionMigration.

Rust

pub struct InstructionRowMigrationOperation;

impl sqlx_migrator::Operation<sqlx::Postgres> for InstructionRowMigrationOperation

Details. up creates the instructions table; down runs DROP TABLE IF EXISTS instructions.

Example.

Rust

fn operations(&self) -> Vec<Box<dyn sqlx_migrator::Operation<sqlx::Postgres>>> {
    vec![Box::new(InstructionRowMigrationOperation)]
}

See also. AccountRowMigrationOperation.

Operations#

carbon_core::postgres::operations. Source: operations.rs. Insert, Upsert, Delete and Lookup are #[async_trait] traits: annotate your impls with #[async_trait::async_trait].

`Table`#

Static table name and column list.

Rust

pub trait Table {
    fn table() -> &'static str;
    fn columns() -> Vec<&'static str>;
}

Details. Nothing in carbon-core calls it. Generated decoders implement it on each decoded type (not the row), for example Swap::table() returns "swap_instruction". Use it to build queries without hard-coding names.

Example.

Rust

use carbon_core::postgres::operations::Table;
use carbon_meteora_dlmm_decoder::instructions::swap::Swap;

let sql = format!("SELECT count(*) FROM {}", Swap::table());

See also. Generated decoder types.

`Insert`#

Inserts a row, failing if the primary key exists.

Rust

#[async_trait::async_trait]
pub trait Insert {
    async fn insert(&self, pool: &sqlx::PgPool) -> CarbonResult<()>;
}

Details. No built-in processor calls insert; they all use Upsert. Implemented by AccountRow, InstructionRow and every generated row and wrapper.

Example.

Rust

AccountRow::from_parts(decoded, metadata).insert(&pool).await?;

See also. Upsert.

`Upsert`#

Inserts a row or updates it on primary-key conflict.

Rust

#[async_trait::async_trait]
pub trait Upsert {
    async fn upsert(&self, pool: &sqlx::PgPool) -> CarbonResult<()>;
}

Details. The bound on W in PostgresAccountProcessor and PostgresInstructionProcessor. Return Ok(()) without writing to skip an update.

Example. See a custom row.

See also. Insert.

`Delete`#

Deletes the row matching a key.

Rust

#[async_trait::async_trait]
pub trait Delete {
    type Key;

    async fn delete(key: Self::Key, pool: &sqlx::PgPool) -> CarbonResult<()>;
}

Details. An associated function, not a method: call AccountRow::<T>::delete(key, &pool). No built-in processor deletes rows; account closures arrive as AccountDeletion updates, which you handle with a account_deletions processor that calls delete.

Example.

Rust

use carbon_core::postgres::{operations::Delete, rows::AccountRow};

AccountRow::<MeteoraDlmmAccount>::delete(deletion.pubkey.into(), &pool).await?;

See also. Lookup.

`Lookup`#

Selects the row matching a key, or None.

Rust

#[async_trait::async_trait]
pub trait Lookup: Sized {
    type Key;

    async fn lookup(key: Self::Key, pool: &sqlx::PgPool) -> CarbonResult<Option<Self>>;
}

Details. Generated account rows implement it with Key = primitives::Pubkey; the generated GraphQL QueryRoot uses it for single-account fields.

Example. See AccountRow.

See also. Delete.

Metadata#

carbon_core::postgres::metadata. Source: metadata.rs. Row types embed these with #[sqlx(flatten)]; the __ prefix keeps metadata columns from colliding with program field names.

`AccountRowMetadata`#

Account metadata columns.

Rust

#[derive(Debug, Clone, PartialEq, sqlx::FromRow, serde::Serialize, serde::Deserialize)]
pub struct AccountRowMetadata {
    #[sqlx(rename = "__pubkey")]
    pub pubkey: Pubkey,
    #[sqlx(rename = "__slot")]
    pub slot: Option<U64>,
}

impl From<AccountMetadata> for AccountRowMetadata

Details.

Field	Type	Column	Postgres type	Description
`pubkey`	`Pubkey`	`__pubkey`	`BYTEA` (32 bytes)	Account address.
`slot`	`Option<U64>`	`__slot`	`NUMERIC`	Slot of the update. Always `Some` when converted from `AccountMetadata`.

AccountMetadata::transaction_signature is not stored.

Example.

Rust

let columns: AccountRowMetadata = input.metadata.clone().into();

See also. InstructionRowMetadata.

`InstructionRowMetadata`#

Instruction metadata columns.

Rust

#[derive(Debug, Clone, PartialEq, sqlx::FromRow, serde::Serialize, serde::Deserialize)]
pub struct InstructionRowMetadata {
    #[sqlx(rename = "__signature")]
    pub signature: String,
    #[sqlx(rename = "__instruction_index")]
    pub instruction_index: U32,
    #[sqlx(rename = "__stack_height")]
    pub stack_height: U32,
    #[sqlx(rename = "__slot")]
    pub slot: Option<U64>,
}

impl From<InstructionMetadata> for InstructionRowMetadata

Details.

Field	Type	Column	Postgres type	Description
`signature`	`String`	`__signature`	`TEXT`	Transaction signature, base58.
`instruction_index`	`U32`	`__instruction_index`	`BIGINT`	`InstructionMetadata::index`: the top-level instruction's index, also for inner instructions.
`stack_height`	`U32`	`__stack_height`	`BIGINT`	`1` for top-level instructions, `2`+ for CPIs.
`slot`	`Option<U64>`	`__slot`	`NUMERIC`	Transaction slot. Always `Some` when converted.

absolute_path, block time and the rest of TransactionMetadata are not stored.

Example.

Rust

let columns: InstructionRowMetadata = input.metadata.clone().into();

See also. Instruction primary key.

Primitives#

carbon_core::postgres::primitives. Source: primitives.rs. Postgres has no unsigned integers and no 32-byte key type, so these newtypes implement sqlx Type, Encode, Decode and PgHasArrayType with a lossless mapping. Each derefs to its inner value, is a single public tuple field, and serializes transparently with serde.

Type	Wraps	Postgres type	Inner value
`Pubkey`	`solana_pubkey::Pubkey`	`BYTEA`	`solana_pubkey::Pubkey`
`U8`	`u8`	`INT2`	`i16`
`U16`	`u16`	`INT4`	`i32`
`U32`	`u32`	`INT8`	`i64`
`U64`	`u64`	`NUMERIC`	`u64`
`U128`	`u128`	`NUMERIC`	`u128`
`I128`	`i128`	`NUMERIC`	`i128`

Signed types up to 64 bits use sqlx's own i16/i32/i64 mappings directly.

`Pubkey`#

A Solana public key stored as 32 raw bytes.

Rust

#[derive(Clone, Copy, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
#[serde(transparent)]
pub struct Pubkey(pub solana_pubkey::Pubkey);

impl From<solana_pubkey::Pubkey> for Pubkey
impl From<Vec<u8>> for Pubkey
impl From<&[u8]> for Pubkey
impl Deref for Pubkey { type Target = solana_pubkey::Pubkey; }

Details. Encodes as BYTEA, arrays as BYTEA[]. Decoding fails if the column does not hold exactly 32 bytes. From<Vec<u8>> and From<&[u8]> return the all-zero key 11111111111111111111111111111111 for input of the wrong length instead of failing. It is a different type from carbon_core::graphql::primitives::Pubkey.

Example.

Rust

let key: carbon_core::postgres::primitives::Pubkey = solana_pubkey::Pubkey::new_unique().into();
println!("{}", *key); // Deref to solana_pubkey::Pubkey

See also. AccountRowMetadata.

`U8`, `U16`, `U32`#

Unsigned integers stored in the next wider signed column.

Rust

pub struct U8(pub i16);   // INT2
pub struct U16(pub i32);  // INT4
pub struct U32(pub i64);  // INT8

// For each of (U8, u8, i16), (U16, u16, i32), (U32, u32, i64):
impl From<u8> for U8
impl From<i16> for U8
impl From<&i16> for U8
impl TryFrom<U8> for u8 { type Error = carbon_core::error::Error; }
impl Deref for U8 { type Target = i16; }

Details. The inner field is the signed wider type, so *U32 is an i64. TryFrom back to the unsigned type returns Error::Custom("out of range") if the stored value does not fit, which only happens when the column was written by something other than Carbon. Derives Clone, Copy, PartialEq, Eq, Debug.

Example.

Rust

let decimals: U8 = 255u8.into(); // stored as INT2 255
let back: u8 = decimals.try_into()?;

See also. U64, U128.

`U64`, `U128`#

Unsigned 64- and 128-bit integers stored as NUMERIC.

Rust

pub struct U64(pub u64);
pub struct U128(pub u128);

// For each of (U64, u64), (U128, u128):
impl From<u64> for U64
impl TryFrom<U64> for u64 { type Error = carbon_core::error::Error; } // always Ok
impl Deref for U64 { type Target = u64; }
impl From<U64> for sqlx::types::Decimal
impl From<sqlx::types::Decimal> for U64
impl From<&sqlx::types::Decimal> for U64

Details. Encodes through BigDecimal as a binary NUMERIC, so the full range survives (u64::MAX round-trips). Decoding fails with Error::Custom when the stored value has a fractional part or a minus sign ("invalid sign/scale for unsigned integer") or exceeds the type's maximum ("value exceeds maximum for target type"). The From<Decimal> conversions are infallible: values that are negative, fractional or out of range become 0 silently. From<U64> for Decimal also returns 0 for values a rust_decimal::Decimal cannot hold (above about 7.9 × 10²⁸, only reachable with U128). Derives Clone, Debug, PartialEq, Eq (not Copy; bind with &row.field).

Example.

Rust

let supply: U64 = u64::MAX.into();
let raw: u64 = *supply;

See also. I128.

`I128`#

Signed 128-bit integer stored as NUMERIC.

Rust

pub struct I128(pub i128);

impl From<i128> for I128
impl Deref for I128 { type Target = i128; }

Details. Same NUMERIC encoding as U128. Decoding fails with Error::Custom("invalid scale for i128") for fractional values and "invalid numeric string" when out of range. Unlike U64/U128, there is no TryFrom<I128> for i128 and no Decimal conversion; use *value or .0.

Example.

Rust

let delta = I128::from(-1i128);

See also. U64, U128.

Generated decoder types#

A decoder built with its postgres feature (for example carbon-meteora-dlmm-decoder, generated by the CLI) adds these public items. Names use the program's PascalCase name (MeteoraDlmm) and each type's snake_case name.

Item	Path	Description
`<Program>AccountWithMetadata`	`accounts::postgres`	Tuple struct `(Account enum, AccountMetadata)`. `From`, `Insert`, `Upsert`; dispatches to the variant's row. Use as `W` in `PostgresAccountProcessor`.
`<Program>AccountsMigration`	`accounts::postgres`	`sqlx_migrator` migration creating every `<account>_account` table. App name is the program's kebab-case name.
`<Account>Row`	`accounts::postgres`	Typed row: flattened `AccountRowMetadata` plus one column per field. `from_parts(account, metadata)`, `TryFrom<Row> for <Account>`, `Insert`, `Upsert`, `Delete` and `Lookup` keyed by `Pubkey`.
`<Account>MigrationOperation`	`accounts::postgres`	Creates one account table.
`<Program>InstructionWithMetadata`	`instructions::postgres`	Tuple struct `(Instruction enum, InstructionMetadata, Vec<AccountMeta>)`. Use as `W` in `PostgresInstructionProcessor`.
`<Program>InstructionsMigration`	`instructions::postgres`	Creates every `<instruction>_instruction` table.
`<Instruction>Row`	`instructions::postgres`	Flattened `InstructionRowMetadata`, one column per argument, and `__accounts JSONB`.
`<Instruction>MigrationOperation`	`instructions::postgres`	Creates one instruction table.
`impl Table for <Type>`	on the decoded type	Table name (`swap_instruction`, `lb_pair_account`) and column list.

Generated columns use NUMERIC(20) for u64, NUMERIC(39) for u128, NUMERIC(38) for i128 (values with 39 digits, above 10³⁸, fail to insert), INT2/INT4/INT8 for u8/u16/u32, BYTEA for pubkeys and byte arrays, and JSONB for nested structs and enums.

Examples#

Example: typed tables#

Indexes Meteora DLMM accounts and instructions into per-type tables. It uses the public RPC transaction crawler, so only DATABASE_URL is required.

Cargo.toml

[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.8"
sqlx = { version = "0.8.6", features = ["runtime-tokio-rustls", "postgres"] }
sqlx_migrator = { version = "0.17.0", features = ["postgres"] }
tokio = { version = "1.48.0", features = ["full"] }

src/main.rs

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

#[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 = sqlx::PgPool::connect(&database_url)
        .await
        .map_err(|e| Error::Custom(e.to_string()))?;

    // Create one table per account type and per instruction type.
    let mut migrator = sqlx_migrator::Migrator::default();
    migrator
        .add_migration(Box::new(MeteoraDlmmAccountsMigration))
        .map_err(|e| Error::Custom(e.to_string()))?;
    migrator
        .add_migration(Box::new(MeteoraDlmmInstructionsMigration))
        .map_err(|e| Error::Custom(e.to_string()))?;
    let mut conn = pool.acquire().await.map_err(|e| Error::Custom(e.to_string()))?;
    migrator
        .run(&mut *conn, &Plan::apply_all())
        .await
        .map_err(|e| Error::Custom(e.to_string()))?;

    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()))
        .account(
            MeteoraDlmmDecoder,
            PostgresAccountProcessor::<MeteoraDlmmAccount, MeteoraDlmmAccountWithMetadata>::new(
                pool.clone(),
            ),
        )
        .instruction(
            MeteoraDlmmDecoder,
            PostgresInstructionProcessor::<
                MeteoraDlmmInstruction,
                MeteoraDlmmInstructionWithMetadata,
            >::new(pool.clone()),
        )
        .build()?
        .run()
        .await
}

Terminal

DATABASE_URL=postgres://postgres:postgres@localhost:5432/carbon cargo run

The transaction crawler emits transactions only, so the account tables stay empty here; pair the same processors with an account datasource such as rpc-program-subscribe to fill them. Then query a table:

SQL

SELECT __signature, amount_in, min_amount_out FROM swap_instruction ORDER BY __slot DESC LIMIT 5;

Example: generic JSONB tables#

The same pipeline writing into the two generic tables. Replace the migrations and the processor:

src/main.rs

use {
    carbon_core::{
        error::{CarbonResult, Error},
        pipeline::Pipeline,
        postgres::{
            processors::PostgresJsonInstructionProcessor,
            rows::{GenericAccountsMigration, GenericInstructionMigration},
        },
    },
    carbon_log_metrics::LogMetrics,
    carbon_meteora_dlmm_decoder::{
        instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID,
    },
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    sqlx_migrator::{Info, Migrate, Plan},
    std::sync::Arc,
};

#[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 = sqlx::PgPool::connect(&database_url)
        .await
        .map_err(|e| Error::Custom(e.to_string()))?;

    // Creates the generic `accounts` and `instructions` tables.
    let mut migrator = sqlx_migrator::Migrator::default();
    migrator
        .add_migration(Box::new(GenericAccountsMigration))
        .map_err(|e| Error::Custom(e.to_string()))?;
    migrator
        .add_migration(Box::new(GenericInstructionMigration))
        .map_err(|e| Error::Custom(e.to_string()))?;
    let mut conn = pool.acquire().await.map_err(|e| Error::Custom(e.to_string()))?;
    migrator
        .run(&mut *conn, &Plan::apply_all())
        .await
        .map_err(|e| Error::Custom(e.to_string()))?;

    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,
            PostgresJsonInstructionProcessor::<MeteoraDlmmInstruction>::new(pool.clone()),
        )
        .build()?
        .run()
        .await
}

Generated instruction enums serialize as {"type": "<Variant>", "data": {"program_id": …, "data": {…}, "accounts": {…}}}. Pubkeys inside the JSON are arrays of 32 numbers, not base58 strings.

SQL

SELECT __signature, decoded_instruction->'data'->'data'->>'amount_in' AS amount_in
FROM instructions
WHERE decoded_instruction->>'type' = 'Swap'
LIMIT 5;

Example: custom row#

A hand-written wrapper that stores only SPL Token mints in a table of your own. It shows Table, Upsert, Lookup, Delete and the primitives together. Create the table yourself:

SQL

CREATE TABLE mints (pubkey BYTEA PRIMARY KEY, slot NUMERIC NOT NULL, supply NUMERIC NOT NULL, decimals INT2 NOT NULL);

src/mints.rs

use {
    carbon_core::{
        account::AccountMetadata,
        error::{CarbonResult, Error},
        postgres::{
            operations::{Delete, Lookup, Table, Upsert},
            primitives::{Pubkey, U64, U8},
            processors::PostgresAccountProcessor,
        },
    },
    carbon_token_program_decoder::accounts::TokenProgramAccount,
};

/// One row of a `mints` table. Non-mint accounts map to `None` and are skipped.
#[derive(sqlx::FromRow, Debug, Clone)]
pub struct MintRow {
    pub pubkey: Pubkey,
    pub slot: U64,
    pub supply: U64,
    pub decimals: U8,
}

pub struct MaybeMint(Option<MintRow>);

impl From<(TokenProgramAccount, AccountMetadata)> for MaybeMint {
    fn from((account, metadata): (TokenProgramAccount, AccountMetadata)) -> Self {
        match account {
            TokenProgramAccount::Mint(mint) => MaybeMint(Some(MintRow {
                pubkey: metadata.pubkey.into(),
                slot: metadata.slot.into(),
                supply: mint.supply.into(),
                decimals: mint.decimals.into(),
            })),
            _ => MaybeMint(None),
        }
    }
}

impl Table for MintRow {
    fn table() -> &'static str {
        "mints"
    }

    fn columns() -> Vec<&'static str> {
        vec!["pubkey", "slot", "supply", "decimals"]
    }
}

#[async_trait::async_trait]
impl Upsert for MaybeMint {
    async fn upsert(&self, pool: &sqlx::PgPool) -> CarbonResult<()> {
        let Some(row) = &self.0 else { return Ok(()) };
        sqlx::query(
            "INSERT INTO mints (pubkey, slot, supply, decimals) VALUES ($1, $2, $3, $4)
             ON CONFLICT (pubkey) DO UPDATE SET slot = $2, supply = $3, decimals = $4",
        )
        .bind(row.pubkey)
        .bind(&row.slot)
        .bind(&row.supply)
        .bind(row.decimals)
        .execute(pool)
        .await
        .map_err(|e| Error::Custom(e.to_string()))?;
        Ok(())
    }
}

#[async_trait::async_trait]
impl Lookup for MintRow {
    type Key = Pubkey;

    async fn lookup(key: Pubkey, pool: &sqlx::PgPool) -> CarbonResult<Option<Self>> {
        sqlx::query_as("SELECT * FROM mints WHERE pubkey = $1")
            .bind(key)
            .fetch_optional(pool)
            .await
            .map_err(|e| Error::Custom(e.to_string()))
    }
}

#[async_trait::async_trait]
impl Delete for MintRow {
    type Key = Pubkey;

    async fn delete(key: Pubkey, pool: &sqlx::PgPool) -> CarbonResult<()> {
        sqlx::query("DELETE FROM mints WHERE pubkey = $1")
            .bind(key)
            .execute(pool)
            .await
            .map_err(|e| Error::Custom(e.to_string()))?;
        Ok(())
    }
}

pub fn mint_processor(pool: sqlx::PgPool) -> PostgresAccountProcessor<TokenProgramAccount, MaybeMint> {
    PostgresAccountProcessor::new(pool)
}

Register it with .account(TokenProgramDecoder, mints::mint_processor(pool.clone())). It needs carbon-token-program-decoder = "2.0.0" and async-trait = "0.1.89" in addition to the dependencies above.

Two storage models#

Processors#

PostgresAccountProcessor#

PostgresInstructionProcessor#

PostgresJsonAccountProcessor#

PostgresJsonInstructionProcessor#

register_postgres_metrics#

Rows#

AccountRow#

InstructionRow#

Instruction primary key#

GenericAccountsMigration#

GenericInstructionMigration#

AccountRowMigrationOperation#

InstructionRowMigrationOperation#

Operations#

Table#

Insert#

Upsert#

Delete#

Lookup#

Metadata#

AccountRowMetadata#

InstructionRowMetadata#

Primitives#

Pubkey#

U8, U16, U32#

U64, U128#

I128#

Generated decoder types#

Examples#

Example: typed tables#

Example: generic JSONB tables#

Example: custom row#

`PostgresAccountProcessor`#

`PostgresInstructionProcessor`#

`PostgresJsonAccountProcessor`#

`PostgresJsonInstructionProcessor`#

`register_postgres_metrics`#

`AccountRow`#

`InstructionRow`#

`GenericAccountsMigration`#

`GenericInstructionMigration`#

`AccountRowMigrationOperation`#

`InstructionRowMigrationOperation`#

`Table`#

`Insert`#

`Upsert`#

`Delete`#

`Lookup`#

`AccountRowMetadata`#

`InstructionRowMetadata`#

`Pubkey`#

`U8`, `U16`, `U32`#

`U64`, `U128`#

`I128`#