2 giờ sáng, service crash vì “too many connections”
Mình vẫn nhớ cái đêm đó rõ mồn một — 2 giờ sáng, Slack nhảy liên tục, MySQL trả về ERROR 1040: Too many connections. Service Rust đang chạy production, mỗi request tự mở một connection MySQL mới và không đóng lại. 800 Tokio task, mỗi task giữ một connection mở. Database sập hết.
Sau sự cố đó, mình refactor toàn bộ database layer sang sqlx với connection pool đúng nghĩa. Bài này ghi lại cái setup đó — đủ để bạn tránh vết xe đổ.
Quick Start: Chạy trong 5 phút
1. Thêm dependencies vào Cargo.toml
[dependencies]
sqlx = { version = "0.7", features = ["runtime-tokio-rustls", "mysql", "macros", "migrate"] }
tokio = { version = "1", features = ["full"] }
dotenvy = "0.15"
macros bật query! macro — cái kiểm tra SQL ngay lúc compile, không phải runtime. Còn migrate cần thiết để sqlx-cli chạy migration về sau.
2. Tạo pool và chạy query đầu tiên
use sqlx::mysql::MySqlPoolOptions;
use sqlx::MySqlPool;
#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
dotenvy::dotenv().ok();
let database_url = std::env::var("DATABASE_URL")
.expect("DATABASE_URL must be set");
let pool: MySqlPool = MySqlPoolOptions::new()
.max_connections(20)
.connect(&database_url)
.await?;
let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM users")
.fetch_one(&pool)
.await?;
println!("Total users: {}", row.0);
Ok(())
}
File .env:
DATABASE_URL=mysql://user:password@localhost:3306/mydb
cargo run
In ra được số users là connect thành công. Tiếp theo là phần quan trọng hơn: pool sizing và type-safe query.
Giải thích chi tiết: Connection Pool và Async Query
Tại sao pool quan trọng?
MySQL mặc định giới hạn max_connections = 151. Mỗi connection tốn RAM (~8MB) và thời gian TLS handshake (~10–50ms). Nếu mỗi request mở connection mới là bạn sẽ gặp đúng sự cố mình gặp lúc 2 giờ sáng đó.
sqlx pool hoạt động kiểu: tạo sẵn N connections, tái sử dụng. Request đến → mượn connection → trả lại sau khi xong. Không mở/đóng liên tục.
Cấu hình pool đúng cho production
use std::time::Duration;
use sqlx::mysql::MySqlPoolOptions;
async fn create_pool(database_url: &str) -> Result<MySqlPool, sqlx::Error> {
MySqlPoolOptions::new()
.max_connections(20) // Tổng connection tối đa
.min_connections(5) // Giữ sẵn 5 connection idle
.acquire_timeout(Duration::from_secs(3)) // Timeout khi chờ connection rảnh
.idle_timeout(Duration::from_secs(600)) // Đóng connection idle > 10 phút
.max_lifetime(Duration::from_secs(1800)) // Tạo lại connection sau 30 phút
.connect(database_url)
.await
}
Kinh nghiệm thực tế: max_connections ≈ (số CPU core × 2) + số disk. VPS 4 core thì đặt 10–15 là ổn. Đặt quá cao không giúp gì thêm — MySQL vẫn bị bottleneck ở I/O disk.
Type-safe query với macro query!
Đây mới là điểm khiến sqlx khác hẳn các thư viện khác. Macro query! kết nối thẳng vào database lúc compile để kiểm tra schema:
#[derive(Debug, sqlx::FromRow)]
struct User {
id: i64,
username: String,
email: String,
created_at: chrono::NaiveDateTime,
}
async fn get_user_by_id(
pool: &MySqlPool,
user_id: i64,
) -> Result<Option<User>, sqlx::Error> {
sqlx::query_as!(
User,
"SELECT id, username, email, created_at FROM users WHERE id = ?",
user_id
)
.fetch_optional(pool)
.await
}
Viết sai tên cột, sai kiểu dữ liệu → compiler báo lỗi ngay, không phải runtime. Đây là type-safety thật sự, không phải ORM giả vờ.
Cần set DATABASE_URL lúc build để sqlx kết nối DB kiểm tra schema:
export DATABASE_URL=mysql://user:password@localhost:3306/mydb
cargo build
Các fetch method thường dùng
// Lấy đúng 1 row — lỗi nếu không có hoặc nhiều hơn 1
let user = sqlx::query_as!(User, "SELECT ...").fetch_one(pool).await?;
// Lấy 1 row nếu có, None nếu không
let user = sqlx::query_as!(User, "SELECT ...").fetch_optional(pool).await?;
// Lấy tất cả
let users = sqlx::query_as!(User, "SELECT ...").fetch_all(pool).await?;
// Stream — xử lý từng row, tiết kiệm RAM khi dataset lớn
use futures::TryStreamExt;
let mut stream = sqlx::query_as!(User, "SELECT ...").fetch(pool);
while let Some(user) = stream.try_next().await? {
process_user(user).await;
}
Nâng cao: Migration An toàn với sqlx-cli
Cài sqlx-cli
cargo install sqlx-cli --no-default-features --features rustls,mysql
Tạo và viết migration
sqlx migrate add create_users_table
# Tạo ra: migrations/20240101000000_create_users_table.sql
-- migrations/20240101000000_create_users_table.sql
CREATE TABLE users (
id BIGINT NOT NULL AUTO_INCREMENT,
username VARCHAR(100) NOT NULL UNIQUE,
email VARCHAR(255) NOT NULL UNIQUE,
password_hash VARCHAR(255) NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (id),
INDEX idx_email (email)
);
# Chạy tất cả migration pending
sqlx migrate run
# Kiểm tra trạng thái
sqlx migrate info
Tự động migrate khi service khởi động
Cách mình làm cho microservices — tự migrate khi start, không cần CI/CD step riêng:
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let pool = create_pool(&database_url).await?;
sqlx::migrate!("./migrations")
.run(&pool)
.await
.expect("Failed to run database migrations");
start_http_server(pool).await?;
Ok(())
}
sqlx ghi lại migration đã chạy vào bảng _sqlx_migrations. Deploy nhiều instance cùng lúc không lo chạy trùng — có distributed lock xử lý rồi. Idempotent hoàn toàn.
Transaction chuẩn
async fn transfer_credits(
pool: &MySqlPool,
from_id: i64,
to_id: i64,
amount: i64,
) -> Result<(), sqlx::Error> {
let mut tx = pool.begin().await?;
sqlx::query!(
"UPDATE wallets SET balance = balance - ? WHERE user_id = ?",
amount, from_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE wallets SET balance = balance + ? WHERE user_id = ?",
amount, to_id
)
.execute(&mut *tx)
.await?;
tx.commit().await?
// Nếu function trả Err hoặc tx bị drop trước commit → rollback tự động
}
Tips thực tế từ production
SQLX_OFFLINE cho CI/CD không có DB
# Trên máy local — generate cache
cargo sqlx prepare
# Commit vào git
git add .sqlx/
git commit -m "chore: update sqlx offline cache"
# Trong CI pipeline
SQLX_OFFLINE=true cargo build
Monitor pool health
tokio::spawn({
let pool = pool.clone();
async move {
loop {
tokio::time::sleep(Duration::from_secs(30)).await;
log::info!(
"DB Pool — size: {}, idle: {}",
pool.size(),
pool.num_idle()
);
}
}
});
Nếu idle = 0 liên tục — pool đang bị saturation, cần tăng max_connections hoặc tối ưu query.
Xử lý lỗi pool gracefully
use sqlx::Error as SqlxError;
match get_user_by_id(&pool, user_id).await {
Ok(Some(user)) => handle_user(user),
Ok(None) => return Err(AppError::NotFound),
Err(SqlxError::PoolTimedOut) => {
log::error!("DB pool exhausted — tăng max_connections hoặc tối ưu query");
return Err(AppError::ServiceUnavailable);
}
Err(e) => {
log::error!("DB error: {:?}", e);
return Err(AppError::Internal);
}
}
Backup trước khi migrate — bắt buộc
Mình từng gặp database corruption lúc 3 giờ sáng. Mất gần 4 tiếng restore từ backup cũ nhất tìm được. Từ đó, backup trước mỗi lần migrate là quy tắc không thương lượng — dù chỉ thêm một cột:
#!/bin/bash
# pre-migrate.sh
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
mysqldump -u"$DB_USER" -p"$DB_PASS" "$DB_NAME" > "backup_${TIMESTAMP}.sql"
echo "Backup saved: backup_${TIMESTAMP}.sql"
sqlx migrate run
Bulk insert hiệu quả với QueryBuilder
async fn bulk_insert_users(
pool: &MySqlPool,
users: &[NewUser],
) -> Result<(), sqlx::Error> {
let mut builder = sqlx::QueryBuilder::new(
"INSERT INTO users (username, email) "
);
builder.push_values(users.iter(), |mut b, u| {
b.push_bind(&u.username).push_bind(&u.email);
});
builder.build().execute(pool).await?;
Ok(())
}
Với 1000 records, cách này nhanh hơn 50–100x so với insert từng cái một trong vòng lặp.
Khi nào dùng sqlx thay ORM như Diesel hay SeaORM?
- sqlx: SQL thật, type-safe tại compile time, kiểm soát tuyệt đối query, async native. Phù hợp microservices cần hiệu năng cao.
- Diesel: ORM đầy đủ, sync (async hạn chế), compile-time schema check kiểu khác. Tốt cho app phức tạp cần abstraction cao.
- SeaORM: ORM async, API giống ActiveRecord, ít boilerplate hơn sqlx. Phù hợp khi cần tốc độ phát triển hơn fine-grained control.
Mình chọn sqlx cho tất cả service cần chịu tải cao — query được optimize thủ công, không có magic query generation ẩn bên dưới gây surprise lúc production.
