Tower とミドルウェア

Tower は非同期リクエストを Service として抽象化し、「サービスをラップする」設定を Layer として抽象化します。タイムアウト、並行制限、バッファリング、リトライは、同じインターフェースの周りで組み合わせることができます。リクエストが必ず HTTP である必要はないこと、またビジネスロジックの冪等性やあらゆる形態のバックプレッシャーを自動的に解決するわけではないことに注意してください。

本稿では Tower 0.5 API を例に挙げます。ミドルウェアを読む際は、それがリクエストの受付、待機キュー、レスポンス Future、またはレスポンスボディの継続的なフローのいずれを制御しているかをまず特定してください。これらの時間範囲は通常異なります。

Service には受付と呼び出しの2つのフェーズがある

コアインターフェースの形状は以下の通りです：

use std::future::Future;
use std::task::{Context, Poll};

pub trait Service<Request> {
    type Response;
    type Error;
    type Future: Future<Output = Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>)
        -> Poll<Result<(), Self::Error>>;
    fn call(&mut self, request: Request) -> Self::Future;
}

poll_ready のレシーバーも &mut self です。これは権限を予約したり内部状態を更新したりする必要がある場合があり、単に「現在の負荷を」読み取り専用で「見る」だけではありません。Pending の場合は、状態変化後のウェイクアップをスケジュールし、呼び出し元が待機してから再ポーリングします。Ready(Ok(())) のみが今回の呼び出しが可能であることを示し、Ready(Err(e)) はサービスエラーです。

呼び出し元は readiness 契約に従わなければならず、1回の成功を無制限の呼び出し権限と見なしてはいけません。一部のサービスでは、ready ではない状態で呼び出すとパニックします。また、インスタンス A で ready を待ってから B を clone し、B が A が予約した能力を継承すると仮定してはいけません。Service ドキュメント

call の可変借用は関連する Future を作成するためのものであり、その Future が完了するまでサービス全体を借用するわけではありません。サービスは複数の独立した在途の Future を生成できる場合があります。より多くのリクエストを受け入れるかどうかは、readiness と実装によって決まります。したがって、Buffer は「Rust の本来の直列制限を解除する」ための必須要素ではありません。

権限を使って ready な状態の変化を観察する

以下の完全なプログラムは、このページの後ろにある Cargo の依存関係を使用します。内部 Future は常に Pending であるため、状態の変化は権限の保持と破棄によってのみ決定され、タイマーを待つ必要はありません。

use std::{convert::Infallible, future::pending, task::{Context, Poll, Waker}};
use tower::{Service, ServiceBuilder, service_fn};
fn main() {
    let inner = service_fn(|(): ()| pending::<Result<(), Infallible>>());
    let mut service = ServiceBuilder::new().concurrency_limit(1).service(inner);
    let mut cx = Context::from_waker(Waker::noop());
    assert!(matches!(service.poll_ready(&mut cx), Poll::Ready(Ok(()))));
    let first = service.call(());
    assert!(service.poll_ready(&mut cx).is_pending());
    drop(first);
    assert!(matches!(service.poll_ready(&mut cx), Poll::Ready(Ok(()))));
}

first が決して poll されなかったとしても、それは readiness から移行した権限を保持しています。破棄されると権限が解放され、次の poll_ready で ready が回復します。これは、受付リソースの起点が内部ビジネスロジックの最初の実行より早い可能性があることを示しています。大量のレスポンス Future を作成して保存しておくと、サービスの容量を占有する可能性があります。

ここで Waker::noop を使用するのは、手動で即座に観察可能な状態をチェックするためであり、状態が変更されるたびにプログラムが能動的に再 poll します。実際の待機コードは実行器と ready().await に任せるべきであり、無効な waker でビジーウェイトループを書いてはいけません。Service 就绪契約

Layer はラッパーを構築するためのインターフェース

Layer::layer は内部サービスを受け取り、新しいサービスを返します。ServiceBuilder は宣言された順序で、先に追加された層を外側に配置します：A、B、S は A(B(S)) を構成します。リクエストはまず A を経由し、次に B を経由します。完了した結果は逆方向に呼び出し元に戻ります。ServiceBuilder

この順序はセマンティクスを変更します。A がリトライ、B がタイムアウトであると仮定します：Retry(Timeout(S)) は各試行に個別のタイムアウトを持たせることができます。Timeout(Retry(S)) は、複数の試行を含むレスポンス Future を制約します。両方とも、readiness を待機する時間が予算に組み込まれるかどうかを解釈する必要があります。

実行可能な例：まず ready、次に call

空の Cargo バイナリプロジェクトで以下を使用します：

[package]
name = "tower-demo"
version = "0.1.0"
edition = "2024"

[dependencies]
tokio = { version = "1", features = ["macros", "rt", "time"] }
tower = { version = "0.5", features = ["util", "limit", "timeout"] }

src/main.rs：

use std::convert::Infallible;
use std::time::Duration;
use tower::{Service, ServiceBuilder, ServiceExt, service_fn};

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), tower::BoxError> {
    let inner = service_fn(|delay_ms: u64| async move {
        tokio::time::sleep(Duration::from_millis(delay_ms)).await;
        Ok::<_, Infallible>(delay_ms)
    });
    let mut service = ServiceBuilder::new()
        .timeout(Duration::from_millis(50))
        .concurrency_limit(1)
        .service(inner);

    let value = service.ready().await?.call(0).await?;
    println!("completed: {value}");

    let result = service.ready().await?.call(500).await;
    match result {
        Ok(value) => println!("completed: {value}"),
        Err(error) => println!("failed: {error}"),
    }
    Ok(())
}

cargo run を実行します。通常、最初の項目は完了し、2番目の項目はタイムアウトします。マシンが深刻にブロックされている場合、スケジューリングがウォッチクロックの観察に影響を与えるため、このプログラムで正確な遅延を校正することはできません。

ここでは Timeout が ConcurrencyLimit の外側にありますが、ready().await は call の前に発生するため、50ms にはその待機時間は含まれません。エンドツーエンドの予算が必要な場合は、呼び出し元で ready と call の非同期プロセス全体を絶対デッドラインで囲み、残りの予算をダウンストリームに渡す必要があります。リトライも同じ総予算を消費する必要があります。

絶対期限を ready と call の外側に配置する

use std::{convert::Infallible, future::pending, time::Duration};
use tower::{Service, ServiceBuilder, ServiceExt, service_fn};
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), tower::BoxError> {
    let inner = service_fn(|(): ()| pending::<Result<(), Infallible>>());
    let mut service = ServiceBuilder::new().concurrency_limit(1).service(inner);
    let first = service.ready().await?.call(());
    let deadline = tokio::time::Instant::now() + Duration::from_millis(1);
    let second = tokio::time::timeout_at(deadline, async {
        service.ready().await?.call(()).await
    }).await;
    assert!(second.is_err()); // 受付を待っているだけで予算を使い果たし、2回目の call はまだ発行されていない
    drop(first);
    let ready = service.ready().await?;
    drop(ready.call(()));
    Ok(())
}

この例では、最初のリクエストが完了しないように意図的に設定し、権限を待っている間も外側の期限制約を受けることを検証しています。1ms は単に非常に短いテスト予算であり、スレッドがこの精度でスケジュールされると保証するものではありません。本番環境のチェーンでは、同じリクエスト期限を選択し、ダウンストリームで待機またはリトライするたびに残り時間を計算し、層を通過するたびに完全な予算を再取得してはいけません。

内部でレスポンスオブジェクトを取得した後にボディの読み取りを継続する場合、上記の期限は Response を取得する段階のみをカバーします。フルストリームの期限が必要な場合は、ストリームの次の項目の待機に元の期限を引き続き使用する必要があります。「アイドルタイムアウト」が必要な場合は、最後の有効な進捗に基づいて別のタイマーを更新します。これらの2つのタイムアウトは異なる問題を解決します。

タイムアウトの予算はどこから始まるか

接納待ちと応答 Future を分けます。Timeout が制限するのは call の Future なので、40 ms の待機と 30 ms の処理が合計 70 ms で成功し得ます。入口から 50 ms にしたいなら ready と call を同じ絶対期限で囲みます。同時刻の競合は実際のランタイムで判断します。

バックプレッシャーには呼び出しチェーン全体の連携が必要

ConcurrencyLimit は権限を使用して在途のレスポンス Future を制御します。上限に達すると readiness を待機しますが、呼び出し元が無限に待機タスクの作成に切り替えると、システムは依然として大量のメモリを消費する可能性があります。バックプレッシャーはアップストリームに継続的に伝播するか、有界キューと拒否戦略と組み合わせて初めて、全体の容量制御を形成します。

Buffer は有界チャネルとバックグラウンドワーカータスクを使用してリクエストを転送し、クローン可能なエントリポイントを提供し、専用の OS スレッドを作成する必要はありません。両側の readiness を分離し、キュー待機も導入します。アップストリームが取得した ready はキューの位置を予約している可能性があるため、保持していない権限を保持しないよう、速やかに call する必要があります。Buffer

RateLimit は一定時間内に処理できるリクエスト数を制約しますが、それは並行制限ではありません。実装を確認せずに、トークンを滑らかに補充する汎用トークンバケットと称してはいけません。LoadShed は、一時的に ready ではないものをエラーに変換し、継続的なキューイングではなく明確な拒否に適した境界です。RateLimit

タイムアウトは何をキャンセルするか

Timeout はレスポンス Future の返却を終了した後、呼び出し元は通常それをドロップし、内部の Future は随之解放されます。ポーリングによって進行する通常の非同期ロジックの場合、これによりその後の進行を停止できます。しかし、送信済みのネットワークリクエスト、コミット済みのデータベーストランザクション、独立して spawn されたタスクは、これによって自動的に取り消されるわけではありません。

したがって、「タイムアウトは内部実行を完全に停止しない」とも、「タイムアウトはすべての副作用を取り消す」とも書けません。キャンセルが安全かどうかは、Future の Drop、リソース所有権、および基盤となる操作セマンティクスと組み合わせて判断する必要があります。タイムアウトエラーは、遠隔地のビジネスロジックが成功しなかったことを証明するものではありません。

リトライとストリーミングレスポンスの2つの境界

Retry は戦略に基づいて再試行するかどうかを判断し、リクエストの再構築またはクローンが必要です。再試行可能性はビジネス効果と失敗フェーズに由来します：クエリは再試行しやすいですが、書き込みは冪等キーと永続的な重複排除を活用できます。HTTP メソッド名だけを根拠に安全性を保証してはいけません。また、障害時に負荷をさらに増幅しないよう、回数、バックオフ、総時間を制限する必要があります。

Service の Response には、まだ消費されていないストリームが含まれている可能性があります。レスポンス Future が完了したからといって、ストリームが終了したわけではありません。通常の Timeout や ConcurrencyLimit がカバーするライフサイクルはすでに終了している可能性がありますが、クライアントはメッセージの継続的な受信を続けている可能性があります。完全なストリームを制限する場合は、権限、タイムアウト、およびクリーンアップ戦略をストリームのライフサイクルに拡張する必要があります。

遅延消費可能な Response を使って権限解放の位置を明確にする

use std::convert::Infallible;
use tower::{Service, ServiceBuilder, ServiceExt, service_fn};
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), tower::BoxError> {
    let inner = service_fn(|(): ()| async {
        Ok::<_, Infallible>(vec![1, 2, 3].into_iter())
    });
    let mut service = ServiceBuilder::new().concurrency_limit(1).service(inner);
    let mut response = service.ready().await?.call(()).await?;
    // 最初の Response はまだ消費されていないが、次の呼び出しを受け入れることができる。
    let second = service.ready().await?.call(()).await?;
    assert_eq!(response.next(), Some(1));
    assert_eq!(second.collect::<Vec<_>>(), [1, 2, 3]);
    Ok(())
}

ここでは、通常のイテレータを使用して「オブジェクトを返した後も継続して消費できる」所有関係を示していますが、ネットワークストリームをシミュレートしているわけではありません。Tower の汎用制限器は、任意の Response 型から、いつが本当に終了したのかを推測することはできません。制限対象のリソースがストリームに属している場合、ストリームラッパーが権限を保持し、正常な終了、エラー、Drop のすべてに明確な解放パスを持たせるべきです。レスポンスヘッダーの生成のみを制限したい場合は、現在の範囲がちょうど適切かもしれません。

レベルごとにキューイングと処理時間を記録する

キューイング、タイムアウト、リトライは論理リクエスト識別子を共有すべきですが、各試行の記録も保持する必要があります。そうしないと、1つのリクエストが3回リトライされることで、呼び出し量と「成功率」の両方が同時に上昇し、ユーザーの実際の待機時間とダウンストリームの追加負荷が隠蔽される可能性があります。

Response 後もストリームは許可を持つか

Response は後から消費するストリームを所有できる値です。「Response を返した」時、通常の応答 Future の許可は解放されても Body は生存します。完全なストリームの同時数を制限するなら、終了・エラー・Drop まで所有ラッパが許可を保持します。

tonic に接続する際は、型とエラーマッピングをまず確認する

tonic はトランスポートおよび生成サービス層で Tower を使用していますが、「すべての gRPC メソッドに任意の Layer を直接適用できる」わけではありません。ミドルウェアの Request、Response、Error、Clone、Send などの制約は、挿入位置と一致する必要があります。HTTP レベルのエラーは、gRPC Status に正しくマッピングされる必要があります。

HTTP リクエストの追跡は通常、tower-http の TraceLayer を使用しますが、すべての Trace が tower 本体から来るわけではないと誤解してはいけません。Interceptor は短い同期メタデータ処理に適していますが、非同期認証やレスポンスボディの観測には、通常、より完全なミドルウェアが必要です。tonic を参照してください。