SQLx connection pool tuning Postgres

Sqlx default config OK untuk dev, tapi kurang di production. Pool 10 connection di service yang dapat 1000 req/s = bottleneck instant. Plus, AWS RDS punya idle timeout 5 menit — kalau lifetime tidak diset, query gagal dengan “connection reset”. Snippet ini tuning lengkap dengan pattern repository.

Kode

// Cargo.toml
// [dependencies]
// sqlx = { version = "0.8", features = ["postgres", "runtime-tokio", "tls-rustls", "chrono", "uuid", "macros"] }
// tokio = { version = "1.40", features = ["full"] }
// chrono = { version = "0.4", features = ["serde"] }
// uuid = { version = "1.10", features = ["v4", "serde"] }
// thiserror = "1"

// db.rs
use std::time::Duration;

use sqlx::postgres::{PgConnectOptions, PgPoolOptions, PgSslMode};
use sqlx::{PgPool, Postgres};
use tracing::info;

pub struct DbConfig {
    pub url: String,
    pub max_connections: u32,
    pub min_connections: u32,
    pub application_name: String,
}

impl Default for DbConfig {
    fn default() -> Self {
        Self {
            url: std::env::var("DATABASE_URL")
                .unwrap_or_else(|_| "postgres://postgres:postgres@localhost/app".to_string()),
            max_connections: 30,
            min_connections: 5,
            application_name: "tokopedia-api".to_string(),
        }
    }
}

pub async fn create_pool(config: DbConfig) -> Result<PgPool, sqlx::Error> {
    info!(
        "Creating Postgres pool: max={}, min={}",
        config.max_connections, config.min_connections
    );

    let connect_opts: PgConnectOptions = config.url.parse::<PgConnectOptions>()?
        // SSL — wajib untuk cloud DB
        .ssl_mode(PgSslMode::Require)
        // Application name muncul di pg_stat_activity untuk debug
        .application_name(&config.application_name)
        // Disable statement logging dari sqlx (kalau noisy)
        .log_statements(tracing::log::LevelFilter::Debug)
        .log_slow_statements(
            tracing::log::LevelFilter::Warn,
            Duration::from_millis(500),
        );

    let pool = PgPoolOptions::new()
        // ==========================================
        // POOL SIZE
        // ==========================================
        // Total connection ke DB = max_connections * jumlah_instance
        // Cek Postgres max_connections: SHOW max_connections;
        // Default 100 — bagi sesuai jumlah instance
        .max_connections(config.max_connections)
        .min_connections(config.min_connections)

        // ==========================================
        // TIMEOUT
        // ==========================================
        // Max wait untuk dapat connection dari pool
        .acquire_timeout(Duration::from_secs(3))

        // ==========================================
        // CONNECTION LIFECYCLE
        // ==========================================
        // Connection idle ditutup setelah X — clean up resources
        .idle_timeout(Some(Duration::from_secs(600)))   // 10 menit

        // Max lifetime — workaround untuk firewall / load balancer
        // yang putus koneksi idle > 30 menit (AWS NLB, GCP, dll)
        // Set lebih pendek dari server idle_in_transaction
        .max_lifetime(Some(Duration::from_secs(1800)))  // 30 menit

        // ==========================================
        // HEALTH CHECK
        // ==========================================
        // Test connection sebelum kasih ke caller — pricier tapi reliable
        // Set false di high-throughput production, true di staging
        .test_before_acquire(true)

        // Connect dengan options yang sudah dibuat
        .connect_with(connect_opts)
        .await?;

    // Verify pool dengan smoke test query
    sqlx::query("SELECT 1").execute(&pool).await?;

    info!("Pool ready");
    Ok(pool)
}

/// Helper untuk graceful shutdown
pub async fn close_pool(pool: PgPool) {
    info!("Closing pool, drain active connections");
    pool.close().await;
    info!("Pool closed");
}

// ==========================================
// Repository pattern
// ==========================================
pub mod produk {
    use chrono::{DateTime, Utc};
    use sqlx::PgPool;
    use uuid::Uuid;

    #[derive(Debug, sqlx::FromRow)]
    pub struct Produk {
        pub id: Uuid,
        pub nama: String,
        pub harga: i64,
        pub stok: i32,
        pub created_at: DateTime<Utc>,
    }

    pub async fn find_by_id(pool: &PgPool, id: Uuid) -> Result<Option<Produk>, sqlx::Error> {
        sqlx::query_as::<_, Produk>(
            "SELECT id, nama, harga, stok, created_at FROM produk WHERE id = $1"
        )
        .bind(id)
        .fetch_optional(pool)
        .await
    }

    pub async fn list_paginated(
        pool: &PgPool,
        limit: i64,
        offset: i64,
    ) -> Result<Vec<Produk>, sqlx::Error> {
        sqlx::query_as::<_, Produk>(
            "SELECT id, nama, harga, stok, created_at FROM produk
             ORDER BY created_at DESC LIMIT $1 OFFSET $2"
        )
        .bind(limit)
        .bind(offset)
        .fetch_all(pool)
        .await
    }

    pub async fn insert(
        pool: &PgPool,
        nama: &str,
        harga: i64,
        stok: i32,
    ) -> Result<Uuid, sqlx::Error> {
        let row: (Uuid,) = sqlx::query_as(
            "INSERT INTO produk (nama, harga, stok) VALUES ($1, $2, $3) RETURNING id"
        )
        .bind(nama)
        .bind(harga)
        .bind(stok)
        .fetch_one(pool)
        .await?;

        Ok(row.0)
    }

    /// Transaction pattern — batch update dalam satu tx
    pub async fn update_stok_batch(
        pool: &PgPool,
        updates: &[(Uuid, i32)],
    ) -> Result<usize, sqlx::Error> {
        let mut tx = pool.begin().await?;
        let mut affected = 0;

        for (id, delta) in updates {
            let result = sqlx::query(
                "UPDATE produk SET stok = stok + $1 WHERE id = $2 AND stok + $1 >= 0"
            )
            .bind(delta)
            .bind(id)
            .execute(&mut *tx)
            .await?;

            if result.rows_affected() == 0 {
                tx.rollback().await?;
                return Err(sqlx::Error::RowNotFound);
            }

            affected += 1;
        }

        tx.commit().await?;
        Ok(affected)
    }
}

Pemakaian

// main.rs
use tracing::info;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt::init();

    let pool = db::create_pool(db::DbConfig::default()).await?;

    // Insert produk
    let id = db::produk::insert(&pool, "Indomie Goreng", 3500, 100).await?;
    info!("Inserted produk: {id}");

    // Query
    if let Some(produk) = db::produk::find_by_id(&pool, id).await? {
        info!("Produk: {:?}", produk);
    }

    // List dengan pagination
    let items = db::produk::list_paginated(&pool, 20, 0).await?;
    info!("Total list: {}", items.len());

    // Batch update dalam transaction
    db::produk::update_stok_batch(&pool, &[(id, -5)]).await?;

    // Graceful shutdown
    db::close_pool(pool).await;
    Ok(())
}

-- Cek pool dari Postgres side
SELECT application_name, state, COUNT(*)
FROM pg_stat_activity
WHERE application_name = 'tokopedia-api'
GROUP BY application_name, state;

-- application_name | state               | count
-- tokopedia-api    | active              | 8
-- tokopedia-api    | idle                | 22
-- tokopedia-api    | idle in transaction | 0    -- HARUS 0

# Monitor pool metric — sqlx 0.8 belum native expose metric
# Workaround: implement custom metric exporter

# Alternative: check via pg_stat_activity periodic
psql -c "
SELECT
  COUNT(*) FILTER (WHERE state = 'active') AS active,
  COUNT(*) FILTER (WHERE state = 'idle') AS idle,
  COUNT(*) FILTER (WHERE state = 'idle in transaction') AS idle_in_tx
FROM pg_stat_activity
WHERE application_name = 'tokopedia-api';
"

Kapan dipakai

• Production HTTP API yang back by Postgres.
• ETL job yang butuh long-running connection.
• Background worker dengan transaction kompleks.
• Multi-instance deployment di Kubernetes / Cloud Run.

Catatan

• max_connections × instance_count — total connection ke DB. Cek SHOW max_connections; di Postgres. Default 100, sering kurang.
• idle in transaction = 0 wajib — kalau ada > 0, ada code yang lupa commit/rollback. Slot connection ke-lock terus.
• max_lifetime < server idle limit — AWS RDS / Cloud SQL punya 30 menit. Set 25-28 menit supaya rotate dulu.
• test_before_acquire — di staging set true (catch connection issue). Production set false untuk performance, andalkan acquire_timeout untuk fail fast.
• log_slow_statements — log query > 500ms ke tracing. Bisa dipakai untuk identify query yang perlu optimasi.
• Pool clone cheap — pool.clone() adalah Arc, bisa pass-by-value ke setiap handler.
• Connection per task — sqlx jangan share connection antar task secara manual. Pool handle ini otomatis.

sqlx::query_as! macro lebih aman karena compile-time check terhadap schema. Tapi butuh DATABASE_URL valid saat compile. Untuk CI/CD, pakai cargo sqlx prepare dan commit query metadata.