午前2時、「too many connections」でサービスがクラッシュした夜
あの夜のことは今でも鮮明に覚えている——午前2時、Slackが鳴り止まず、MySQLがERROR 1040: Too many connectionsを返し続けた。プロダクションで動いていたRustサービスは、リクエストのたびに新しいMySQLコネクションを開いては閉じずにいた。800個のTokioタスクが、それぞれコネクションを握ったまま放さない。データベースは完全に落ちた。
その障害を機に、データベース層全体をsqlxと本物のコネクションプールへとリファクタリングした。この記事はそのセットアップの記録だ——同じ轍を踏まないために。
クイックスタート：5分で動かす
1. Cargo.tomlに依存関係を追加
[dependencies]
sqlx = { version = "0.7", features = ["runtime-tokio-rustls", "mysql", "macros", "migrate"] }
tokio = { version = "1", features = ["full"] }
dotenvy = "0.15"
macrosを有効にするとquery!マクロが使えるようになる——ランタイムではなくコンパイル時にSQLを検証する機能だ。migrateは後でsqlx-cliによるマイグレーション実行に必要になる。
2. プールを作成して最初のクエリを実行
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が設定されていません");
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!("ユーザー総数: {}", row.0);
Ok(())
}
.envファイル：
DATABASE_URL=mysql://user:password@localhost:3306/mydb
cargo run
ユーザー数が出力されれば接続成功だ。次はより重要なポイント——プールのサイジングとタイプセーフクエリを見ていこう。
詳細解説：コネクションプールと非同期クエリ
なぜプールが重要なのか？
MySQLはデフォルトでmax_connections = 151に制限されている。コネクションは1本あたりRAMを約8MB消費し、TLSハンドシェイクに10〜50msかかる。リクエストのたびに新しいコネクションを開いていると、あの午前2時の悪夢が再現される。
sqlxのプールはこう動く：あらかじめN本のコネクションを確立しておき、使い回す。リクエストが来たらコネクションを借り、処理が終わったら返却する。接続・切断を繰り返さない。
プロダクション向けの正しいプール設定
use std::time::Duration;
use sqlx::mysql::MySqlPoolOptions;
async fn create_pool(database_url: &str) -> Result<MySqlPool, sqlx::Error> {
MySqlPoolOptions::new()
.max_connections(20) // コネクション上限数
.min_connections(5) // アイドル状態で維持するコネクション数
.acquire_timeout(Duration::from_secs(3)) // 空きコネクション待機タイムアウト
.idle_timeout(Duration::from_secs(600)) // 10分以上アイドルのコネクションを切断
.max_lifetime(Duration::from_secs(1800)) // 30分後にコネクションを再作成
.connect(database_url)
.await
}
実運用の経験則：max_connections ≈ (CPUコア数 × 2) + ディスク数。4コアのVPSなら10〜15で十分だ。上限を高くしても意味はない——MySQLのボトルネックはディスクI/Oにある。
query!マクロによるタイプセーフクエリ
これこそがsqlxを他のライブラリと一線を画す機能だ。query!マクロはコンパイル時にデータベースへ直接接続してスキーマを検証する：
#[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
}
カラム名のタイポや型の不一致はコンパイル時にエラーとして検出される——ランタイムではなく。これが本物のタイプ安全性であり、ORMが見せかけるそれとは違う。
スキーマ検証のため、ビルド時にDATABASE_URLを設定しておく必要がある：
export DATABASE_URL=mysql://user:password@localhost:3306/mydb
cargo build
よく使うfetchメソッド
// ちょうど1行を取得 — 0行または複数行の場合はエラー
let user = sqlx::query_as!(User, "SELECT ...").fetch_one(pool).await?;
// 1行あればSome、なければNoneを返す
let user = sqlx::query_as!(User, "SELECT ...").fetch_optional(pool).await?;
// 全行を取得
let users = sqlx::query_as!(User, "SELECT ...").fetch_all(pool).await?;
// ストリーム — 1行ずつ処理、大量データでもRAMを節約できる
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;
}
応用：sqlx-cliによる安全なマイグレーション
sqlx-cliのインストール
cargo install sqlx-cli --no-default-features --features rustls,mysql
マイグレーションの作成と記述
sqlx migrate add create_users_table
# 生成されるファイル: 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)
);
# 未適用のマイグレーションをすべて実行
sqlx migrate run
# 適用状況を確認
sqlx migrate info
サービス起動時に自動マイグレーション
マイクロサービスでの実践パターン——起動時に自動でマイグレーションを実行し、CI/CDの別ステップを不要にする：
#[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("データベースマイグレーションの実行に失敗しました");
start_http_server(pool).await?;
Ok(())
}
sqlxは適用済みのマイグレーションを_sqlx_migrationsテーブルに記録する。複数インスタンスを同時にデプロイしても二重実行の心配はない——分散ロックが処理してくれる。完全なべき等性が保証されている。
正しいトランザクションの書き方
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?
// Errを返すかcommit前にtxがdropされると自動的にロールバックされる
}
プロダクションから学んだ実践的なTips
DBなしのCI/CD環境にはSQLX_OFFLINEを使う
# ローカルマシンでキャッシュを生成
cargo sqlx prepare
# gitにコミット
git add .sqlx/
git commit -m "chore: update sqlx offline cache"
# CIパイプライン内での実行
SQLX_OFFLINE=true cargo build
プールの状態を監視する
tokio::spawn({
let pool = pool.clone();
async move {
loop {
tokio::time::sleep(Duration::from_secs(30)).await;
log::info!(
"DBプール状態 — 使用中: {}, アイドル: {}",
pool.size(),
pool.num_idle()
);
}
}
});
idle = 0が続いている場合はプールが飽和状態にある。max_connectionsを増やすか、クエリを最適化する必要がある。
プールエラーのグレースフルな処理
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プールが枯渇しました — max_connectionsを増やすかクエリを最適化してください");
return Err(AppError::ServiceUnavailable);
}
Err(e) => {
log::error!("DBエラー: {:?}", e);
return Err(AppError::Internal);
}
}
マイグレーション前のバックアップは絶対に必須
かつて午前3時にデータベースが破損した経験がある。一番古いバックアップからのリストアに4時間近くかかった。それ以来、マイグレーション前のバックアップは交渉の余地のないルールになった——カラムを1つ追加するだけでも例外はない：
#!/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_${TIMESTAMP}.sql"
sqlx migrate run
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(())
}
1000件のレコードなら、ループで1件ずつインサートするより50〜100倍高速になる。
DieselやSeaORMではなくsqlxを選ぶ判断基準
- sqlx：生SQL、コンパイル時のタイプ安全性、クエリを完全にコントロール、ネイティブ非同期。高いスループットが求められるマイクロサービスに最適。
- Diesel：フル機能のORM、同期式（非同期は限定的）、独自方式のコンパイル時スキーマチェック。高い抽象化が必要な複雑なアプリケーションに向いている。
- SeaORM：非同期ORM、ActiveRecord的なAPI、sqlxよりボイラープレートが少ない。細かい制御よりも開発速度を優先したい場合に適している。
高負荷に耐える必要があるサービスにはすべてsqlxを選んでいる——クエリを手動で最適化でき、プロダクションで驚かされるような隠れたクエリ生成が存在しない。
