tonic と gRPC

gRPC はサービス定義を使用して RPC メソッドとメッセージ型を記述し、tonic は Rust 用の非同期実装を提供します。一般的な組み合わせは Tokio、Tower、Prost です。生成コードにより手動でのシリアライズやインターフェースのボイラープレートコードが削減されますが、ネットワークの両端では、片方が .proto ファイルを変更しただけでは自動的に再コンパイルされません。

プロトコルのコンパイル時の型チェック、オンライン時のエンコーディング互換性、ビジネスセマンティクスは、3つの異なるレイヤーです。HTTP API でも OpenAPI などの記述とコード生成を使用できます。gRPC を選択する理由は、呼び出しモデル、ストリーミング機能、およびエコシステムの要件から来るべきものであり、別の種類のインターフェースを一律に「契約がない」と断じてはいけません。

4つのRPC形式をまず理解する

双方向ストリームは、リクエストを受信するたびにレスポンスを送信することを要求するものではなく、双方向に1対1の対応関係があることも保証しません。各方向内のメッセージには順序がありますが、アプリケーション側でメッセージID、完了条件、および失敗時のリカバリ方法を定義する必要があります。

最小限の完全なプロジェクト：生成ツールのバージョンを明確にする

以下では tonic 0.14 シリーズを使用します。このシリーズの Prost 統合は tonic-prost と tonic-prost-build にあります。古いチュートリアルの tonic_build::compile_protos を直接使用して混在させないでください。ローカル環境に protoc がインストールされている必要があり、実際の解析バージョンを記録するために Cargo.lock を保存してください。tonic、tonic-prost-build

Cargo.toml：

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

[dependencies]
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
tonic = "0.14"
tonic-prost = "0.14"
prost = "0.14"

[build-dependencies]
tonic-prost-build = "0.14"

proto/greeter.proto：

syntax = "proto3";
package greeter;
service Greeter {
  rpc SayHello(HelloRequest) returns (HelloReply);
}
message HelloRequest { string name = 1; }
message HelloReply { string message = 1; }

build.rs：

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("cargo:rerun-if-changed=proto/greeter.proto");
    tonic_prost_build::configure()
        .compile_protos(&["proto/greeter.proto"], &["proto"])?;
    Ok(())
}

src/lib.rs：

pub mod greeter {
    tonic::include_proto!("greeter");
}

src/bin/server.rs：

use grpc_demo::greeter::{HelloReply, HelloRequest};
use grpc_demo::greeter::greeter_server::{Greeter, GreeterServer};
use tonic::{Request, Response, Status};

#[derive(Default)]
struct HelloService;

#[tonic::async_trait]
impl Greeter for HelloService {
    async fn say_hello(&self, request: Request<HelloRequest>)
        -> Result<Response<HelloReply>, Status>
    {
        let name = request.into_inner().name;
        if name.trim().is_empty() {
            return Err(Status::invalid_argument("name is required"));
        }
        Ok(Response::new(HelloReply { message: format!("Hello, {name}!") }))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic::transport::Server::builder()
        .add_service(GreeterServer::new(HelloService))
        .serve("127.0.0.1:50051".parse()?)
        .await?;
    Ok(())
}

src/bin/client.rs：

use grpc_demo::greeter::{HelloRequest, greeter_client::GreeterClient};
use std::time::Duration;
use tonic::Request;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut client = GreeterClient::connect("http://127.0.0.1:50051").await?;
    let mut request = Request::new(HelloRequest { name: "Rust".into() });
    request.set_timeout(Duration::from_secs(2));
    let reply = client.say_hello(request).await?.into_inner();
    println!("{}", reply.message);
    Ok(())
}

2つのターミナルでそれぞれ cargo run --bin server と cargo run --bin client を実行すると、クライアントは Hello, Rust! を出力するはずです。ここではリクエストのチェーンを観察するためにローカルで平文リスニングを使用しています。実際のデプロイでは、環境に応じて TLS、認証、リソース制限を設定してください。

#[tonic::async_trait] は、このシリーズで生成されるトレイトの形式に合わせるために使用されます。Rust 1.75 以降ではトレイト内のネイティブな async fn がサポートされています。マクロを残しているからといって、言語が「まだサポートしていない」と解釈してはいけません。ネイティブな async trait、ダイナミックディスパッチ、および生成ライブラリが採用している規約は、それぞれ別々に考慮する必要があります。

.proto を修正しても旧クライアントが必ずしも壊れない理由

Protobuf のバイナリエンコーディングはフィールド番号に依存します。互換性のあるフィールドを追加する場合、旧クライアントは自分が認識しているフィールドを正常に読み続ける可能性があります。ローカルの Rust コードを修正する必要があるかどうかは、再生成を行うかどうか、および型をどのように構築するかによって異なります。オンラインでの互換性は、単一のリポジトリでのコンパイル成功によって証明されるものではありません。

フィールドを削除した後、その番号は保持し、通常は将来の誤用を防ぐために名前も保持してください。異なる意味を表現するために旧番号を再利用しないでください。フィールドタイプの変更は、ワイヤー互換性のルールに従って個別に判断する必要があります。エンコーディングがデコード可能であっても、単位が秒からミリ秒に変更されると、ビジネスセマンティクスが直接破壊される可能性があります。Proto3 更新ルール

「提供されていない」と「デフォルト値が提供されている」を区別する必要がある場合、フィールドの存在性設計を明確に使用すべきです。例えば、適切な optional フィールドを使用します。空文字列は、タイプが string だからといって自動的に検証エラーにはなりません。そのため、上記のハンドラーでは明示的に name をチェックしています。

2種類のメッセージ型を使用してフィールドの存在性と旧中継ノードを検証する

以下は独立した src/bin/compat.rs です。プロジェクトの既存の Prost 依存関係を使用し、新旧の構造を同じプログラム内で同時に観察するためにメッセージを直接宣言します。フィールド番号 1 は name を保持し、バージョン2では optional の locale フィールド番号 2 が追加されます。

use prost::Message;
#[derive(Clone, PartialEq, Message)]
struct V1 {
    #[prost(string, tag = "1")]
    name: String,
}
#[derive(Clone, PartialEq, Message)]
struct V2 {
    #[prost(string, tag = "1")]
    name: String,
    #[prost(string, optional, tag = "2")]
    locale: Option<String>,
}
fn main() {
    let absent = V2 { name: String::new(), locale: None };
    let empty = V2 { name: String::new(), locale: Some(String::new()) };
    assert!(absent.encode_to_vec().is_empty());
    assert_eq!(empty.encode_to_vec(), [0x12, 0]);

    let new = V2 { name: "Rust".into(), locale: Some("zh-CN".into()) };
    let bytes = new.encode_to_vec();
    let old = V1::decode(bytes.as_slice()).unwrap();
    assert_eq!(old.name, "Rust");
    let relayed = old.encode_to_vec();
    let decoded = V2::decode(relayed.as_slice()).unwrap();
    assert_eq!(decoded.name, "Rust");
    assert_eq!(decoded.locale, None);
}

optional は、欠落と明示的な空値をそれぞれ None と Some("") にマッピングします。この例では、0x12 はフィールド番号 2、ワイヤータイプ 2 のキーであり、後の 0 はゼロバイトの長さを表します。したがって、ビジネス上の文字列がどちらも空であっても、オンラインでの表現は異なります。Prost フィールド表現

旧 V1 は、新しいデータから認識できる name を読み取ることができますが、この Prost メッセージ構造は再エンコーディング時に未知の locale を保持しません。したがって、「旧クライアントが新しいフィールドを含むメッセージを読み取れること」と「旧ノードを介してデコードおよび再エンコードされた後、新しいフィールドが終点に到達できること」は、2つの異なる互換性の要件です。旧ノードが元のメッセージバイトを透過的に転送するだけの場合は、ここで言及されている損失のある再構築は発生していません。具体的なチェーンは、実際のトランスフォーメーションプロセスに従って検証する必要があります。

リリース前に読み書きの方向でチェックする

段階的なリリースでは、通常、読み取り側が新しい形式を受け入れるようにしてから、新しい書き込みを有効にします。ロールバックする場合は、書き込まれた新しいデータを旧バージョンが適切に処理できるかも確認する必要があります。フィールド番号を保持することでエンコーディングの誤読を防げますが、単位、権限、またはステートマシンのセマンティクスの変更を自動的に解決するわけではありません。

旧ノードは新フィールドを保持できるか

None はフィールド 2 を送らず、Some("") はキーと長さゼロを送ります。生転送は違いを保持しますが、本文の旧 Prost V1 は name しか知らず、再構築で locale を失います。旧側が復号できるかだけでなく、読み書きの全経路で互換性を確認します。

Status、デッドライン、およびリトライ

Status は RPC の結果を表し、HTTP のレスポンスステータスだけを見てはいけません。一般的な分類は以下の通りです。完全なコード定義は gRPC Status Codes を参照してください。

クライアントにはデッドライン（期限切れ時刻）を設定すべきです。総予算には、キュー待ち、接続、処理、および可能性のあるリトライのカバーが必要です。Request::set_timeout は grpc-timeout メタデータに書き込まれ、ローカルでの転送タイムアウトを設定するのとは異なる操作です。上記の設定は connect の後に発生するため、接続確立自体に2秒の上限を設定しているわけではありません。

サービスがキャンセルされた後、アプリケーションは作成したバックグラウンドワークを停止する必要があります。ダウンストリームに RPC を送信する際は、各レイヤーで完全なタイムアウトを再設定するのではなく、残りの予算を伝播してください。異なる言語やライブラリでの伝播のサポートは異なるため、明確に検証する必要があります。gRPC Deadlines

ストリーミング処理のライフサイクルはより長い

サーバーが Response<Stream> を返すのは、レスポンスストリームが確立されたことを示すだけで、すべてのビジネスメッセージが完了したことを意味しません。その後、ストリームの要素は通常 Result<Message, Status> となります。消費側は正常な終了とエラーを区別する必要があり、while let Some(Ok(...)) でエラーを静かに飲み込んではいけません。

有界の mpsc チャンネルへの send().await は、消費者の遅延を生産者にフィードバックできます。送信に失敗した場合は、受信側が消滅したことを意味するため、unwrap するのではなく対応する作業を終了すべきです。無界キューは、ネットワークのバックプレッシャーをメモリ増加に変換する可能性があります。バックグラウンドタスクは、適切なキャンセルシグナルを保持して応答する必要があります。

メッセージサイズの制限は、ストリームの総サイズ制限ではありません。長時間のストリームには、時間、総バイト数、同時実行数、またはビジネスウィンドウの制御が必要です。クライアントが一部のみを取得して破棄した場合、サーバーも関連するリソースを解放する必要があります。

プロセス内で生産者と消費者の終了プロトコルを検証する

前述の tokio 依存関係の features に "sync" を追加し、以下の src/bin/stream_state.rs を保存してください。これは、ストリーミングサービスで一般的な生産側と同じ有界チャネルを使用し、キュー内の Result を直接観察します。このステップでは、ポートのリスニングや遠端サービスの呼び出しは必要ありません。

use tokio::sync::mpsc;
use tonic::{Code, Status};
use grpc_demo::greeter::HelloReply;

#[tokio::main]
async fn main() {
    let (sender, mut receiver) = mpsc::channel::<Result<HelloReply, Status>>(1);
    let producer = tokio::spawn(async move {
        if sender.send(Ok(HelloReply { message: "part-1".into() })).await.is_err() {
            return;
        }
        let _ = sender.send(Err(Status::unavailable("source interrupted"))).await;
        // 終了エラーの後、ビジネスメッセージを生産せず、退出時に sender を解放する。
    });
    let mut received = Vec::new();
    let mut failure = None;
    while let Some(item) = receiver.recv().await {
        match item {
            Ok(reply) => received.push(reply.message),
            Err(status) => {
                failure = Some(status.code());
                break;
            }
        }
    }
    drop(receiver);
    producer.await.unwrap();
    assert_eq!(received, ["part-1"]);
    assert_eq!(failure, Some(Code::Unavailable));

    let (sender, receiver) = mpsc::channel::<Result<HelloReply, Status>>(1);
    drop(receiver);
    assert!(sender.send(Ok(HelloReply { message: "unused".into() })).await.is_err());
}

有界容量 1 は、キューに積まれるメッセージ項目数だけを制約し、1つのメッセージ内の文字列バイト数や、生産者が send の前に構築済みのオブジェクトのサイズは制限しません。生産者は受信側が離脱すると送信を停止します。データベースカーソルを保持している場合や他のタスクを開始している場合は、それらのリソースも同時に解放またはキャンセルする必要があります。

実際の tonic クライアントは、stream.message().await の Result<Option<Message>, Status> を使用して、メッセージ、正常な終了、およびエラーを別々に処理できます。Err を None として扱うことはできません。ビジネスで1回の完全なスナップショットを要求する場合、上記の例の part-1 は一時保存データに過ぎず、最終的に失敗した場合は完全な結果としてコミットすべきではありません。ビジネスが増分消費を許可する場合、再接続を受け入れるためにメッセージシーケンス番号、リカバリカーソル、および冪等処理が必要です。

ローカル検証と転送検証は異なる問題をカバーする

生成された Greeter トレイトメソッドを直接呼び出して空の name と正常なレスポンスを検証し、Prost を使用してフィールドエンコーディングをテストできます。これらのチェックは、ビジネスおよびプロトコルモデルの問題を迅速に特定します。転送層では、メタデータ、TLS、メッセージ制限、デッドライン、およびストリームの終了状態を別途確認する必要があります。ハンドラーを直接呼び出す場合、tonic の HTTP/2 エンコーディング/デコーディングおよびミドルウェアを経由しないためです。

同様に、クライアントの set_timeout によって生成されたメタデータは、サーバーと中継チェーンが正しく理解する必要があります。これは、Channel 作成前の DNS/接続待機に予算を追加するものではなく、バックグラウンドタスクが自動的に停止することを意味するものでもありません。1つのユーザー操作围绕して総期限を設定し、接続確立、RPC 待機、およびアプリケーションリソースのクリーンアップをそれぞれ検証してください。

Interceptor はどのような場面に適しているか

Interceptor のインターフェースは Request<()> を処理し、リクエストまたは Status を同期的に返します。メタデータの読み取り、軽量コンテキストの注入、および即時拒絶に適しています。リモート認証を直接 await することはできず、任意のレスポンスボディの変換機能も提供しません。Interceptor

非同期の外部クエリ、リクエストとレスポンスの完全な観測、またはより細かいフロー制御が必要な場合は、挿入に適した Tower サービスを使用するか、ビジネスサービス内で明示的に完了してください。どこに配置するかにかかわらず、エラーがどのように Status に変換され、ストリーム終了時にリソースをどのようにクリーンアップするかを明確にする必要があります。Tower とミドルウェア を参照してください。