# Store decoded data in Postgres (Carbon 2.0) > Persist decoded instructions and accounts to PostgreSQL with the postgres feature: generated typed tables or generic JSONB, and sqlx_migrator migrations. 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](https://carbon.sevenlabs.org/guides/generate-a-decoder.md) generates it unless you pass `--with-postgres false`. ## Choose a storage model | | Typed tables | Generic JSONB tables | | ---------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | --------------------------------------------------------------------------------------------- | | Processor | `PostgresInstructionProcessor`, `PostgresAccountProcessor` | `PostgresJsonInstructionProcessor`, `PostgresJsonAccountProcessor` | | Tables | One per instruction (`swap_instruction`) and per account type (`lb_pair_account`), created by the decoder's migrations | `instructions` and `accounts`, shared by every program, created by `carbon-core`'s migrations | | Columns | Metadata columns plus one column per field: `NUMERIC(20)` for `u64`, `BYTEA` for pubkeys, `JSONB` for nested types | Metadata columns plus one `JSONB` column with the whole decoded value | | Query with | Plain SQL on typed columns, indexes per field; the decoder's [GraphQL](https://carbon.sevenlabs.org/guides/graphql.md) module reads these tables | JSON operators (`->`, `->>`, `@>`) | | Best for | One or a few programs you query in depth | Many programs, exploration, schemas that change | Both models upsert on a primary key, so writing the same update twice leaves one row. ### Step 1: Start PostgreSQL Use any PostgreSQL you can reach. With Docker: ```sh 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 ``` The program reads the connection string from `DATABASE_URL`. Add `?sslmode=require` for a managed database that requires TLS. ### Step 2: Create the project ```sh 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 ``` `Cargo.toml`: ```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.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"] } ``` Patch versions may be newer. Enable `postgres` on both `carbon-core` and the decoder. Keep `sqlx` on 0.8 and `sqlx_migrator` on 0.17, the versions Carbon 2.0 is built with; another major version gives you a second, incompatible `PgPool` type. ### Step 3: Write the pipeline The program connects, runs the decoder's instruction migration, and writes every decoded DLMM instruction to its table. `src/main.rs`: ```rust 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::::new(pool), ) .build()? .run() .await } ``` The two type parameters of `PostgresInstructionProcessor` are the decoder's instruction enum and its generated `MeteoraDlmmInstructionWithMetadata` wrapper, which knows which table and row type each variant goes to. ### Step 4: Run it ```sh RUST_LOG=info cargo run ``` Output: ```text [… 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 ``` The first run creates 76 `*_instruction` tables. The public RPC endpoint is rate-limited, so rows arrive slowly; that's the datasource, not Postgres. Stop with ctrl-C. ### Step 5: Query the tables ```sh 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" ``` Output: ```text 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) ``` All three are CPIs (`__stack_height` 2) from aggregators. The schema of one table: ```sh psql "$DATABASE_URL" -c '\d swap2_instruction' ``` Output: ```text 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) ``` Columns starting with `__` are metadata. `__accounts` holds the instruction's accounts as JSON, with each pubkey as an array of 32 bytes: `{"pubkey": [115, 168, …], "is_signer": false, "is_writable": true}`. ## 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`. `src/bin/jsonb.rs`: ```rust 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::::new(pool), ) .build()? .run() .await } ``` ```sh 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" ``` Output: ```text 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": "", "data": {"program_id": …, "data": {…}, "accounts": {…}}}`, so the arguments are under `data.data`: ```sh 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" ``` Output: ```text 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`](https://carbon.sevenlabs.org/datasources/rpc-program-subscribe.md), [`RpcGpa`](https://carbon.sevenlabs.org/datasources/rpc-gpa.md) or [Yellowstone](https://carbon.sevenlabs.org/datasources/yellowstone-grpc.md); see [Load account state](https://carbon.sevenlabs.org/guides/load-account-state.md). The transaction crawler above never emits them. ```rust 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::::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::` 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`: - `AccountWithMetadata` and `InstructionWithMetadata`: the `W` type for the typed processors. Their `Upsert` and `Insert` match on the variant and write the matching row. - `AccountsMigration` and `InstructionsMigration`: `sqlx_migrator` migrations that create every table (`CREATE TABLE IF NOT EXISTS`). - One `Row` per account and instruction, with `from_parts`, `TryFrom` 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: ```rust use carbon_core::postgres::operations::Lookup; use carbon_meteora_dlmm_decoder::accounts::postgres::LbPairRow; let row = LbPairRow::lookup(pubkey.into(), &pool).await?; // Option ``` The full list, column types and the generic row types are in the [Postgres reference](https://carbon.sevenlabs.org/reference/postgres.md#generated-decoder-types). ## Migrations `sqlx_migrator` records applied migrations in `_sqlx_migrator_migrations` and skips them on the next run: ```sh psql "$DATABASE_URL" -c "SELECT app, name FROM _sqlx_migrator_migrations" ``` Output: ```text 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](https://carbon.sevenlabs.org/guides/graphql.md) 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](https://carbon.sevenlabs.org/guides/prometheus.md). ## 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: Output: ```text 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 `CpiEvent`s 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](https://carbon.sevenlabs.org/reference/postgres.md#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')` for `BYTEA` columns; in Rust, bind `sqlx::types::Json(pubkey.to_bytes())` against a JSONB element, as the [GraphQL guide](https://carbon.sevenlabs.org/guides/graphql.md#add-your-own-queries) 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 - [Serve it over GraphQL](https://carbon.sevenlabs.org/guides/graphql.md): Expose these tables with the decoder's generated GraphQL schema. - [Postgres reference](https://carbon.sevenlabs.org/reference/postgres.md): Processors, rows, operations, primitives and generated types. - [Run Carbon in production](https://carbon.sevenlabs.org/guides/production.md): Idempotent writes, backpressure and shutdown. --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/guides/postgres · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt