クエリとハンドル

エンティティとクエリを定義し、ケイパビリティ型付きの Db ハンドルで実行して、読み取りをレプリカへ振り分け、ハンドルをコンポーネントに注入します。

このページでは、PostgreSQL を使う小さなユーザーサービスのデータアクセスを組み立てます。クエリを定義し、プライマリ用とレプリカ用の 2 つのハンドルを作り、各クエリを適切なハンドルで実行して、ハンドルを ServiceBuilder に登録します。背景にある考え方（エンティティ、ケイパビリティ、エフェクト）は概要で説明しています。

セットアップ

Cargo.toml

[dependencies]
wakuwaku = { version = "1.0.0-rc3", features = ["sqlx-pg"] }
tokio = { version = "1", features = ["full"] }

sqlx-pg を有効にすると、PostgreSQL のデータソースを含む wakuwaku::entity が使えるようになり、sqlx が wakuwaku::deps::sqlx として再エクスポートされます。

素の i64 は、どのテーブルを指す ID なのかを表しません。PostsByAuthor { author } にはユーザー ID と同じように投稿 ID も渡せてしまいますし、EditPostTitle の id と version を取り違えてもコンパイルが通ります。主キーごとに newtype を用意すれば、こうした取り違えは型エラーになります。外部キーの意味も型に表れます。author: UserId を見ればこのフィールドが users を指すことがわかり、author を使っているどこからでも「型定義へ移動」で User エンティティのすぐ隣にある UserId へジャンプできます。ラッパーの実行時コストはありません。バインドするところで .0 で取り出し、読み取ったところで包み直します。

wakuwaku と同じバージョンの sqlx を derive feature 付きで直接依存に加えれば、#[derive(sqlx::Type)] と #[sqlx(transparent)] で newtype をそのままバインド・デコードできます。

クエリを定義する

クエリは 2 つのトレイトを実装します。

Query は何に触れるか、つまり Effect を表します。どこで実行されるかには依存しないので、ケイパビリティ検査やフックはどのデータソースでも同じように働きます。
Execute<S> はデータソース S でどう実行するか、つまり Output、Error、そしてコネクション上で動くコードを表します。移植性のある SQL で書いたクエリなら、複数のデータソースに対して実装できます。

src/queries.rs

use crate::entities::{Post, PostId, User, UserId};
use wakuwaku::deps::sqlx::{self, PgConnection};
use wakuwaku::entity::sqlx::backend::PgSource;
use wakuwaku::entity::{Execute, Query, Read, Write};

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UserRow {
    pub id: UserId,
    pub name: String,
}

/// ID でユーザーを 1 件読み込む
pub struct FindUser {
    pub id: UserId,
}

impl Query for FindUser {
    type Effect = Read<User>;
}

impl Execute<PgSource> for FindUser {
    type Output = Option<UserRow>;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<Option<UserRow>, sqlx::Error> {
        let row: Option<(i64, String)> = sqlx::query_as("SELECT id, name FROM users WHERE id = $1")
            .bind(self.id.0)
            .fetch_optional(conn)
            .await?;
        Ok(row.map(|(id, name)| UserRow { id: UserId(id), name }))
    }
}

execute は普通の &mut PgConnection を受け取り、クエリの Error を返します。データベースと同じ理由でしか失敗しないクエリは、データソースのエラー（sqlx のデータソースなら sqlx::Error）を使います。自分の判断で拒否することもあるクエリは、独自のエラー型を宣言します。クエリを単独で実行すると、コネクションはプールから取得され、自動コミットで動きます。トランザクションの中では、そのトランザクションのコネクションが渡されます。同じクエリをどちらでも使えます。

複数のエンティティに触れるクエリでは、そのすべてを列挙します。タプルは要素のエフェクトの和集合です。

src/queries.rs

/// アクティブな著者の投稿タイトル
pub struct PostsByAuthor {
    pub author: UserId,
}

impl Query for PostsByAuthor {
    // 2 つのテーブルを読む。タプルはエフェクトの和集合
    type Effect = (Read<User>, Read<Post>);
}

impl Execute<PgSource> for PostsByAuthor {
    type Output = Vec<String>;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<Vec<String>, sqlx::Error> {
        sqlx::query_scalar(
            "SELECT p.title FROM posts p JOIN users u ON u.id = p.author_id \
             WHERE u.id = $1 AND u.active ORDER BY p.created_at DESC",
        )
        .bind(self.author.0)
        .fetch_all(conn)
        .await
    }
}

書き込みには Write<E> を宣言します。E に書き込めるケイパビリティは E を読むこともできるので、users で絞り込む UPDATE でも Write<User> だけで足ります。

src/queries.rs

/// ユーザーの表示名を変更する
pub struct RenameUser {
    pub id: UserId,
    pub name: String,
}

impl Query for RenameUser {
    type Effect = Write<User>;
}

impl Execute<PgSource> for RenameUser {
    /// 影響を受けた行数
    type Output = u64;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<u64, sqlx::Error> {
        let done = sqlx::query("UPDATE users SET name = $2 WHERE id = $1")
            .bind(self.id.0)
            .bind(self.name)
            .execute(conn)
            .await?;
        Ok(done.rows_affected())
    }
}

エフェクトは正直に宣言する

コンパイラが検査するのは、書いた SQL ではなく宣言したエフェクトです。Read<User> と宣言したクエリで UPDATE を実行してもコンパイルは通ります。Effect の行は関数シグネチャと同じように扱い、SQL 文と一緒にレビューしてください。

データソース

DataSource は、コネクションプールや多重化クライアントなど、クエリを実行できるものです。クエリが受け取るコネクションの型と、データソース自身のエラー型を決めます。どのクエリの Error も、このエラー型から変換できなければなりません。

pub trait DataSource: Send + Sync + Sized {
    type Connection: Send;
    type Error: Send;

    fn run<Q: Execute<Self>>(
        &self,
        query: Q,
    ) -> impl Future<Output = Result<Q::Output, Q::Error>> + Send;
}

run はクエリ自身の結果を返します。コネクションを取得できなかったときなど、クエリが始まる前にデータソースが失敗した場合は、そのエラーが From<Self::Error> で Q::Error に変換されます。

自分で実装することはまずありません。バックエンドごとに用意されています。

データソース	フィーチャー	コネクション	データソースのエラー
`entity::sqlx::backend::PgSource`	`sqlx-pg`	`PgConnection`	`sqlx::Error`
`entity::sqlx::backend::MySqlSource`	`sqlx-mysql`	`MySqlConnection`	`sqlx::Error`
`entity::redis::RedisSource`	`redis`	Redis を参照	`redis::RedisError`
`entity::meilisearch::MeiliSource`	`meilisearch`	Meilisearch を参照	`meilisearch_sdk::errors::Error`

PgSource は SqlxSource<Postgres> の別名です。単独で実行されるクエリは、それぞれプールからコネクションを取得します。

ハンドルを作る

Db<S, C> はデータソース S を所有し、ケイパビリティ C を持ちます。ケイパビリティは型パラメータにすぎないので、ハンドルを作るときに選びます。プライマリとリードレプリカを持つサービスでは、ハンドルを 2 つ作ります。

src/wiring.rs

use crate::entities::{PostId, UserId};
use crate::queries::{
    DeactivateUsers, DormantUsers, EditPostError, EditPostTitle, FindUser, PostsByAuthor, RenameUser, UserRow,
};
use std::time::Duration;
use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::deps::sqlx::{self, postgres::PgPoolOptions};
use wakuwaku::entity::sqlx::backend::{PgSource, SqlxSource};
use wakuwaku::entity::{Db, ReadOnly, ReadWrite};
use wakuwaku::{CronJob, CronJobInterval, CronTick, ServiceBuilder};

/// サービスの 2 つのハンドル。書き込みはプライマリへ、
/// レプリケーション遅延を許容できる読み取りはレプリカへ送る
#[derive(Clone)]
pub struct Databases {
    pub primary: Db<PgSource, ReadWrite>,
    pub replica: Db<PgSource, ReadOnly>,
}

impl Databases {
    pub async fn connect(primary_url: &str, replica_url: &str) -> Result<Self, sqlx::Error> {
        let primary = PgPoolOptions::new().max_connections(10).connect(primary_url).await?;
        let replica = PgPoolOptions::new().max_connections(20).connect(replica_url).await?;
        Ok(Self {
            primary: Db::new(SqlxSource::new(primary)),
            replica: Db::new(SqlxSource::new(replica)),
        })
    }
}

型が約束になる

Db::new には、プールがどのサーバーにつながっているかが分かりません。レプリカのプールから Db<PgSource, ReadWrite> を作ってもコンパイルは通り、書き込みがレプリカに送られてしまいます。書き込めるケイパビリティは書き込み可能なデータベースに接続したデータソースからだけ作り、この組み立ては 1 か所にまとめてください。

一度ラップしたプールは、ハンドルから取り出せません。これがケイパビリティの健全性を支えています。マイグレーションなどの準備は SqlxSource::new を呼ぶ前にプールに対して行ってください。Db を clone するとデータソースも clone されますが、sqlx の場合はプールのハンドルを clone するだけなので軽量です。

クエリを実行する

Db は kanau の Processor を実装しています。トレイトをスコープに入れて process を呼びます。

src/wiring.rs

pub async fn run_one(dbs: &Databases) -> Result<(), sqlx::Error> {
    let user: Option<UserRow> = dbs.replica.process(FindUser { id: UserId(7) }).await?;
    let renamed: u64 = dbs.primary.process(RenameUser { id: UserId(7), name: "Haruki".into() }).await?;
    println!("{user:?}, {renamed} row(s) updated");
    Ok(())
}

process はクエリの Output か、クエリの Error を返します。

ハンドラはハンドルに対してジェネリックにする

ハンドラの中で Db<PgSource, ReadOnly> という型を書かないでください。代わりに、必要なクエリを Processor 境界として列挙します。

src/wiring.rs

pub struct Profile {
    pub user: UserRow,
    pub posts: Vec<String>,
}

/// 両方のクエリを実行できるハンドルなら何でも受け取る
pub async fn load_profile<D>(db: &D, id: UserId) -> Result<Option<Profile>, sqlx::Error>
where
    D: Processor<FindUser, Output = Option<UserRow>, Error = sqlx::Error>
        + Processor<PostsByAuthor, Output = Vec<String>, Error = sqlx::Error>,
{
    let Some(user) = db.process(FindUser { id }).await? else {
        return Ok(None);
    };
    let posts = db.process(PostsByAuthor { author: id }).await?;
    Ok(Some(Profile { user, posts }))
}

pub async fn rename_user<D>(db: &D, id: UserId, name: String) -> Result<bool, sqlx::Error>
where
    D: Processor<RenameUser, Output = u64, Error = sqlx::Error>,
{
    Ok(db.process(RenameUser { id, name }).await? == 1)
}

こうすると、ケイパビリティはハンドルを渡す場所で一度だけ検査されます。さらに次の利点があります。

ハンドラは、そのクエリを処理できるものなら何でも受け取ります。Db、書き込みフックでラップした Db、ユニットテスト用のフェイクのどれでも構いません。
Processor 境界が複数あるときは、上の例のように各境界で Output と Error を固定します。そうすると db.process(...) の結果の型が一意に決まります。

クエリ独自のエラー型

データベースの失敗だけでなく、自分の理由で失敗するクエリもあります。楽観ロックで守られた更新が、先に誰かが行を変更していたことに気づく場合や、出金が口座の残高を超える場合です。こうした拒否は操作の失敗なので、Output ではなくクエリの Error に入れます。

Output に入れること（Output = bool や Output = Result<Post, Conflict>）もできますが、そうすると呼び出し側は ? の後でもう一度結果を確かめなければならず、確かめ忘れてもコンパイルは通り、拒否が成功として扱われてしまいます。独自のエラー型なら、拒否もほかのエラーと同じように扱えます。? で伝播し、match で処理します。

条件は type Error: From<S::Error> だけです。データソースのエラーをその型に収められれば十分です。次の例では posts の行が version 列を持ち、編集はクライアントが最後に読んだバージョンに対してだけ適用されます。

src/queries.rs

/// Why an edit of a post failed.
#[derive(Debug)]
pub enum EditPostError {
    /// The post changed, or was deleted, since `version` was read.
    Stale,
    /// The database failed.
    Database(sqlx::Error),
}

impl From<sqlx::Error> for EditPostError {
    fn from(error: sqlx::Error) -> Self {
        Self::Database(error)
    }
}

/// Change a post's title, if nobody changed the post since `version` was read.
pub struct EditPostTitle {
    pub id: PostId,
    pub version: i64,
    pub title: String,
}

impl Query for EditPostTitle {
    type Effect = Write<Post>;
}

impl Execute<PgSource> for EditPostTitle {
    /// The new version of the post.
    type Output = i64;
    type Error = EditPostError;

    async fn execute(self, conn: &mut PgConnection) -> Result<i64, EditPostError> {
        let version: Option<i64> = sqlx::query_scalar(
            "UPDATE posts SET title = $3, version = version + 1 \
             WHERE id = $1 AND version = $2 RETURNING version",
        )
        .bind(self.id.0)
        .bind(self.version)
        .bind(self.title)
        .fetch_optional(conn)
        .await?; // sqlx::Error -> EditPostError::Database
        version.ok_or(EditPostError::Stale)
    }
}

Db::process は Result<i64, EditPostError> を返すので、それぞれの失敗が何を意味するかを呼び出し側が 1 か所で決められます。

src/wiring.rs

pub async fn save_title(dbs: &Databases, id: PostId, version: i64, title: String) -> Result<Option<i64>, sqlx::Error> {
    match dbs.primary.process(EditPostTitle { id, version, title }).await {
        Ok(new_version) => Ok(Some(new_version)),
        // Someone else won: tell the client to reload and retry.
        Err(EditPostError::Stale) => Ok(None),
        Err(EditPostError::Database(error)) => Err(error),
    }
}

EditPostError を返す関数では、同じ From 実装により、エラーが sqlx::Error のクエリにも ? を使えます。フックもこれを変えません。フックを付けたハンドルは、クエリのエラーをそのまま返します。

トランザクションの中で使う

トランザクションの中で実行するクエリのエラーは、AsSqlxError を実装している必要があります。このトレイトは、エラーがデータベースから来たもの（トランザクションを中断させた可能性がある）か、クエリ独自のもの（決して中断させない）かをトランザクションに伝えます。sqlx::Error は実装済みです。独自のエラー型では、包んでいる sqlx::Error を返します。

src/queries.rs

use wakuwaku::entity::sqlx::transaction::AsSqlxError;

impl AsSqlxError for EditPostError {
    fn as_sqlx_error(&self) -> Option<&sqlx::Error> {
        match self {
            Self::Database(error) => Some(error),
            Self::Stale => None,
        }
    }
}

トランザクション内のエラーを参照してください。

読み取りはレプリカへ、書き込みはプライマリへ

呼び出し側でハンドルを選びます。

src/wiring.rs

pub async fn handle_requests(dbs: &Databases) -> Result<(), sqlx::Error> {
    // プロフィールページ。多少古いデータでも構わない
    let _profile = load_profile(&dbs.replica, UserId(7)).await?;

    // 書き込みはプライマリへ
    rename_user(&dbs.primary, UserId(7), "Haruki".into()).await?;

    // 自分の書き込みを読む。プライマリでも読み取れる
    let _fresh = load_profile(&dbs.primary, UserId(7)).await?;
    Ok(())
}

ReadWrite はすべての読み取りを網羅するので、プライマリのハンドルでも読み取りクエリを実行できます。レプリカはプライマリより遅れることがあるため、直前の書き込みを確実に反映した読み取りが必要な場合はプライマリを使ってください。

書き込みが読み取り専用ハンドルに届いたとき

書き込むハンドラにレプリカを渡してみます。

pub async fn oops(dbs: &Databases) -> Result<bool, sqlx::Error> {
    rename_user(&dbs.replica, UserId(7), "Haruki".into()).await
}

コンパイルは通りません。

error[E0277]: the trait bound `ReadOnly: CanWrite<User>` is not satisfied
  --> src/fails.rs:18:17
   |
18 |     rename_user(&dbs.replica, UserId(7), "Haruki".into()).await
   |     ----------- ^^^^^^^^^^^^ the trait `CanWrite<User>` is not implemented for `ReadOnly`
   |     |
   |     required by a bound introduced by this call
   |
   = note: required for `ReadOnly` to implement `Covers<wakuwaku::wakuwaku_entity::Write<User>>`
   = note: required for `Db<SqlxSource<Postgres>, ReadOnly>` to implement `Processor<RenameUser>`
note: required by a bound in `rename_user`

note は下から読みます。rename_user には Processor<RenameUser> が必要です。Db がそれを実装するのはケイパビリティが Write<User> を網羅する場合だけで、ReadOnly には CanWrite<User> がありません。dbs.replica.process(RenameUser { .. }) と直接呼んでも同じエラーになります。

一部のエンティティにだけ書き込ませたいコンポーネントには、ReadWrite ではなく独自のケイパビリティを渡してください。

ハンドルをコンポーネントに注入する

ハンドルは普通の値なので、ほかの共有依存と同じように ServiceBuilder に登録できます。Db<PgSource, ReadOnly> と Db<PgSource, ReadWrite> は別の型なので、両方を登録し、各コンポーネントが必要なものを取り出します。

次の cron ジョブは、休眠ユーザーをレプリカで探し、プライマリで無効化します。

src/wiring.rs

/// 1 年間サインインしていないユーザーを無効化する
pub struct DeactivateDormantUsers {
    replica: Db<PgSource, ReadOnly>,
    primary: Db<PgSource, ReadWrite>,
}

impl Processor<CronTick> for DeactivateDormantUsers {
    type Output = ();
    type Error = sqlx::Error;

    async fn process(&self, _tick: CronTick) -> Result<(), sqlx::Error> {
        let ids = self.replica.process(DormantUsers { days: 365 }).await?;
        if !ids.is_empty() {
            self.primary.process(DeactivateUsers { ids }).await?;
        }
        Ok(())
    }
}

impl CronJob for DeactivateDormantUsers {
    const INTERVAL: CronJobInterval = CronJobInterval::AfterFinish(Duration::from_secs(60 * 60));
}

DormantUsers（Read<User>、出力は Vec<UserId>）と DeactivateUsers（Write<User>、ids: Vec<UserId> を受け取る）は、上のクエリと同じように定義します。ハンドルを登録し、その両方からジョブを組み立てます。

src/wiring.rs

pub async fn start() -> Result<(), sqlx::Error> {
    let dbs = Databases::connect(
        "postgres://app@db-primary/app",
        "postgres://app@db-replica/app",
    )
    .await?;

    let (services, cron) = ServiceBuilder::new(dbs.primary.clone())
        .push(dbs.replica.clone())
        .compose_cron_job(|s| DeactivateDormantUsers {
            replica: s.provide::<Db<PgSource, ReadOnly>, _>().clone(),
            primary: s.provide::<Db<PgSource, ReadWrite>, _>().clone(),
        })
        .start();

    // ほかのコンポーネントもビルダーからハンドルを受け取る
    let replica: &Db<PgSource, ReadOnly> = services.provide::<Db<PgSource, ReadOnly>, _>();
    let _profile = load_profile(replica, UserId(7)).await?;
    handle_requests(&dbs).await?;

    cron.shutdown().await;
    Ok(())
}

compose_cron_job はクロージャにビルダーを貸すので、コンポーネントは登録済みのハンドルをいくつでも受け取れます。ServiceCreation の実装は不要です。ハンドルを 1 つだけ必要とするコンポーネントなら、そのハンドルを Dep として ServiceCreation を実装し、cron_job で登録することもできます。gRPC サービスとイベントコンシューマーにも同じ 2 つのメソッドがあります。ServiceBuilder を参照してください。ハンドルの clone はプールのハンドルを clone するだけなので、各コンポーネントが自分のコピーを持てます。

クエリとハンドル

セットアップ

エンティティを定義する

クエリを定義する

データソース

ハンドルを作る

クエリを実行する

ハンドラはハンドルに対してジェネリックにする

クエリ独自のエラー型

読み取りはレプリカへ、書き込みはプライマリへ

書き込みが読み取り専用ハンドルに届いたとき

ハンドルをコンポーネントに注入する

次のステップ

書き込みフック

sqlx

トランザクション

目次

クエリとハンドル

セットアップリンクをコピー

エンティティを定義するリンクをコピー

クエリを定義するリンクをコピー

データソースリンクをコピー

ハンドルを作るリンクをコピー

クエリを実行するリンクをコピー

ハンドラはハンドルに対してジェネリックにするリンクをコピー

クエリ独自のエラー型リンクをコピー

読み取りはレプリカへ、書き込みはプライマリへリンクをコピー

書き込みが読み取り専用ハンドルに届いたときリンクをコピー

ハンドルをコンポーネントに注入するリンクをコピー

なぜクエリはリポジトリのメソッドではなく値なのか

なぜケイパビリティは実行時のフラグではなく型パラメータなのか

次のステップリンクをコピー

書き込みフック

sqlx

トランザクション

目次

セットアップ

エンティティを定義する

クエリを定義する

データソース

ハンドルを作る

クエリを実行する

ハンドラはハンドルに対してジェネリックにする

クエリ独自のエラー型

読み取りはレプリカへ、書き込みはプライマリへ

書き込みが読み取り専用ハンドルに届いたとき

ハンドルをコンポーネントに注入する

次のステップ