🗄️ Kết nối database, test và quan sát lỗi

By Everwise Intern

Bài trước bạn đã chốt phần validation và tính nhất quán của luồng giao dịch, cùng viết test cho validate. Giờ giao dịch đó cần chỗ lưu thật: Postgres. Và khi có tiền thật chạy qua, câu hỏi không còn là "code có chạy không" mà là "khi nó hỏng, tôi có biết hỏng ở đâu trong 5 phút không".

Đó là toàn bộ nội dung bài này: một pool connection tử tế, integration test cho luồng chuyển tiền, và log có cấu trúc đủ để truy vết.

Service phải sống sót khi DB chậm, không chỉ khi DB khỏe

Ba thứ tôi luôn kiểm tra đầu tiên khi review một service banking viết bằng Rust:

  • Pool có giới hạn và có timeout. Không có acquire_timeout thì request treo vô hạn khi pool cạn, và bạn sẽ không thấy lỗi, chỉ thấy latency tăng.
  • Mọi thao tác tiền nằm trong transaction, và transaction đó rollback đúng cách khi có lỗi giữa đường.
  • Lỗi khi ra khỏi process phải là log có cấu trúc, không phải println!.
use sqlx::postgres::{PgPoolOptions, PgPool};
use std::time::Duration;

pub async fn connect(database_url: &str) -> anyhow::Result<PgPool> {
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .acquire_timeout(Duration::from_secs(3))
        .idle_timeout(Duration::from_secs(600))
        .connect(database_url)
        .await?;
    sqlx::migrate!("./migrations").run(&pool).await?;
    Ok(pool)
}

Số dư lưu bằng BIGINT theo đơn vị nhỏ nhất, không dùng float. Với VND, đơn vị nhỏ nhất chính là đồng:

CREATE TABLE accounts (
    id          UUID PRIMARY KEY,
    balance_vnd BIGINT NOT NULL CHECK (balance_vnd >= 0),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

Phần trừ tiền giữ nguyên tinh thần bài trước, nhưng điều kiện số dư nằm ngay trong câu UPDATE để DB tự chặn race:

let mut tx = pool.begin().await?;
let row = sqlx::query!(
    r#"UPDATE accounts
       SET balance_vnd = balance_vnd - $1, updated_at = now()
       WHERE id = $2 AND balance_vnd >= $1
       RETURNING balance_vnd"#,
    amount, account_id
)
.fetch_optional(&mut *tx)
.await?
.ok_or(DomainError::InsufficientFunds)?;
tx.commit().await?;

Điểm cần nhớ: ? ở dòng ok_or làm tx bị drop, và sqlx rollback khi Transaction bị drop mà chưa commit. Đây là hành vi đúng, không phải bug.

Bài tập 1: viết handler GET /healthz trả 200 chỉ khi pool còn query được.

Gợi ý đáp án

pub async fn healthz(pool: &PgPool) -> Result<StatusCode, StatusCode> {
    sqlx::query("SELECT 1").execute(pool).await
        .map(|_| StatusCode::OK)
        .map_err(|_| StatusCode::SERVICE_UNAVAILABLE)
}

Đừng đặt endpoint này sau lớp auth, load balancer cần gọi được nó.

Thứ tự làm việc

  1. Tạo file migration trong migrations/, chạy sqlx migrate run để có schema thật.
  2. Viết connect() với pool limit và timeout, gọi từ main.rs lúc khởi động.
  3. Chuyển các query sang sqlx::query! để có kiểm tra kiểu lúc compile.
  4. Bọc luồng chuyển tiền trong transaction, kiểm tra rollback bằng test.
  5. Viết integration test với #[sqlx::test] cho 3 case: đủ tiền, thiếu tiền, tài khoản không tồn tại.
  6. Cắm tracing với output JSON và request id, log ở biên handler.

Bước 5 là bước đáng đầu tư nhất. #[sqlx::test] tạo database riêng cho mỗi test, chạy migrations, rồi truyền PgPool vào tham số:

#[sqlx::test(migrations = "./migrations")]
async fn thieu_tien_thi_khong_tru(pool: PgPool) {
    let id = Uuid::new_v4();
    sqlx::query!("INSERT INTO accounts (id, balance_vnd) VALUES ($1, $2)", id, 100_000i64)
        .execute(&pool).await.unwrap();

    let err = debit(&pool, id, 150_000).await.unwrap_err();
    assert!(matches!(err, DomainError::InsufficientFunds));

    let row = sqlx::query!("SELECT balance_vnd FROM accounts WHERE id = $1", id)
        .fetch_one(&pool).await.unwrap();
    assert_eq!(row.balance_vnd, 100_000);
}

Bài tập 2: thêm test cho case chuyển tiền hợp lệ và kiểm tra cả hai tài khoản đều đổi số dư. Câu hỏi phụ: nếu thiếu AND balance_vnd >= $1, test nào sẽ fail?

Gợi ý đáp án

Test hợp lệ: insert hai tài khoản, gọi hàm transfer, assert balance của cả hai. Test thiếu tiền ở trên vẫn pass nếu bạn dựa vào CHECK (balance_vnd >= 0) vì DB sẽ trả lỗi constraint, nhưng lỗi đó là sqlx::Error, không phải DomainError::InsufficientFunds, nên matches! sẽ fail. Đó chính là lý do phải chặn ở tầng ứng dụng chứ không phó mặc constraint.

Nghiệm thu: khi nào coi là xong

  • connect() fail nhanh (dưới 5 giây) khi DB chết, không treo process.
  • Ba integration test chạy song song không giẫm dữ liệu lên nhau.
  • Log ra JSON, có request_id, account_id, amount, error.
  • Không có log nào chứa số tài khoản đầy đủ hay thông tin định danh khách hàng.
  • /healthz trả 503 khi DB ngắt.

Ôn nhanh các điểm dễ quên

sqlx quản lý connection qua _____ thay vì mở connection mới cho mỗi request. Macro _____ kiểm tra SQL và kiểu dữ liệu ngay lúc compile, nên cần biến môi trường DATABASE_URL hoặc thư mục .sqlx khi build offline. Nếu một Transaction bị drop mà không gọi _____, sqlx sẽ rollback. Thuộc tính _____ tạo database riêng cho từng test và chạy migrations trước khi test bắt đầu. Lỗi nghiệp vụ 4xx nên log ở mức _____, còn ERROR chỉ dành cho lỗi 5xx cần người trực xử lý.

Bẫy tôi đã gặp khi đưa service lên staging

1. query! build fail trên CI. Macro cần kết nối DB lúc compile. Cách chuẩn: chạy cargo sqlx prepare ở máy local, commit thư mục .sqlx, và đặt SQLX_OFFLINE=true trong CI. Đừng trỏ CI vào DB production, tôi đã thấy người làm vậy.

2. Log lỗi hai lần. Nếu tầng repository log rồi tầng handler log lại, một sự cố thành hai alert. Chỉ log một lần, ở biên handler, nơi bạn biết nó map ra HTTP status nào.

3. Dùng error! cho mọi lỗi. InsufficientFunds là nghiệp vụ bình thường, log ERROR sẽ làm nhiễu dashboard và khiến on-call mất niềm tin vào alert.

4. Giữ connection quá lâu. sqlx::query xen giữa một await gọi HTTP bên ngoài sẽ giữ connection của pool. Với max_connections = 10, mười request chậm là cả service đứng.

5. Log có cấu trúc nhưng lại nối chuỗi. error!("debit failed for {}", id) tạo log không parse được. Dùng field:

tracing::error!(error = %e, account_id = %account_id, amount = amount, "debit thất bại");

Bài tập 3: một giao dịch chuyển tiền thất bại với sqlx::Error::PoolTimedOut. Xác định level log và các field nên có.

Gợi ý đáp án

Level ERROR: đây là lỗi hạ tầng 5xx, người trực cần biết. Field: error = %e, request_id, account_id, amount, và operation = "transfer". Thêm metric counter transfer_failed_total{reason="pool_timeout"} để vẽ được biểu đồ, log một mình không đủ.

Front: PgPool dùng để làm gì thay vì mở connection mỗi request? Back: Tái sử dụng một tập connection có giới hạn, tránh làm sập Postgres khi tải cao.

Front: sqlx::query! khác sqlx::query() ở điểm nào? Back: query! kiểm tra SQL và kiểu dữ liệu lúc compile, query() chỉ kiểm tra lúc runtime.

Front: Build trên CI không có DATABASE_URL thì query! cần gì? Back: Chạy cargo sqlx prepare, commit thư mục .sqlx và đặt SQLX_OFFLINE=true.

Front: Transaction trong sqlx rollback khi nào, ngoài gọi rollback tường minh? Back: Khi Transaction bị drop mà chưa gọi commit, kể cả khi thoát bằng dấu ?.

Front: Vì sao truyền &mut *tx vào execute? Back: Để deref Transaction thành connection có mutable borrow đúng kiểu sqlx yêu cầu.

Front: Điều kiện AND balance_vnd >= $1 trong UPDATE có tác dụng gì? Back: Chặn race condition, chỉ trừ tiền khi đủ số dư, ngay tại tầng database.

Front: #[sqlx::test] chuẩn bị gì trước mỗi test? Back: Tạo database riêng, chạy migrations, rồi truyền PgPool vào tham số test.

Front: Số tiền nên lưu trong Postgres bằng kiểu gì? Back: BIGINT theo đơn vị nhỏ nhất hoặc NUMERIC với rust_decimal, tuyệt đối không dùng float.

Front: tracing khác println! ở điểm nào? Back: Log có cấu trúc key-value, có level, span và bật tắt được bằng filter.

Front: Lỗi InsufficientFunds nên log ở level nào? Back: WARN, vì đây là lỗi nghiệp vụ 4xx, không phải sự cố hạ tầng.

Front: Vì sao không log đầy đủ số tài khoản? Back: Đó là dữ liệu nhạy cảm, chỉ log account_id dạng UUID hoặc che bớt vài chữ số.

Front: Đặt acquire_timeout để làm gì? Back: Fail nhanh khi pool cạn thay vì treo request vô hạn.

Front: Health check nên kiểm tra gì? Back: Một query nhẹ như SELECT 1 qua pool để xác nhận DB còn sống.

Chốt lại: bạn vừa có phần khó nhất của service

Sau bài này, service của bạn có schema thật, migration chạy tự động, transaction rollback đúng khi thiếu tiền, ba integration test độc lập, và log JSON đủ để dựng alert. Đây là phần mà reviewer phỏng vấn hay hỏi sâu nhất, vì nó cho thấy bạn đã từng vận hành code chứ không chỉ viết cho nó chạy.

Bài tiếp theo là checkpoint: bạn sẽ ghép toàn bộ API, transaction và lớp quan sát này thành một service hoàn chỉnh, có README và demo luồng chuyển tiền từ đầu đến cuối.