Leaky Bucketとは

リーキーバケットはネットワークに送信するトラフィックに対して、その転送レートを制御する。名前の通り、穴の開いたバケツに相当し、様々な流量の水流がそのバケツに流れ込むと、そこまでの流量がどうであっても小さな穴からは一定の水流が流れ出す。同様にリーキーバケット・アルゴリズムは様々なバースト性のトラフィックを一定のトラフィックに変換する機構を提供する。
リーキーバケット - Wikipedia

主にレート制限の手法として活用されているようです。
つまりサーバー都合でリクエスト回数に制限を設けることにより、サービスやトラフィックに制限を加えます。

画像引用元：How we built rate limiting capable of scaling to millions of domains

なお、レート制限の文脈では下記アルゴリズムも紹介されていることが多いみたいです。
いずれも制限の対象を工夫することで、要求に合わせて最適化している印象でした。

Token Bucket
Fixed Window
Sliding Window

参考：
レート制限アルゴリズム - システム設計 - GeeksforGeeks
様々なrate limitアルゴリズム - Carpe Diem
QoS（9）帯域制御の制御方法トークンバケット/リーキーバケット/万能リーキーバケットモデルとは – Ethernet TSN がIoTを変える

leaky_bucket crateについて

Leaky BucketのRust実装であるleaky_bucket¹の動作を確認します。
このクレートはtokioの機能に依存しているのでtokioランタイム上で使用する必要があるようです。

動作確認

やや冗長ですが、axumサーバーで簡単に動作確認しました。
RateLimiterの設定項目がいくつかあり、単純な構成でレート制限する上ではこの設定項目でおおよそ十分な印象を勝手に受けました。
設定項目について詳しくはドキュメントを参照ください。

確認内容

バケツに空きがある（発行したtokenが残っている）間はリクエストを受け付けます
バケツが満杯になる（発行したtokenを全て消費する）とリクエストを拒否します²
tokenは毎秒1つ補充されるので、バケツが満杯になった（発行したtokenを全て消費した）状態から時間が経てば再度リクエストを受け付けます

use std::{sync::Arc, time::Duration};

use axum::{
    extract::State,
    http::StatusCode,
    response::{IntoResponse, Response},
    routing::{get, post},
    Router,
};
use leaky_bucket::RateLimiter;
use tracing::{debug, level_filters::LevelFilter};

struct AppState {
    tokens: Arc<RateLimiter>,
}

impl AppState {
    pub fn new() -> AppState {
        AppState {
            tokens: Arc::new(
                RateLimiter::builder()
                    .initial(5)
                    .interval(Duration::from_secs(1))
                    .refill(1)
                    .max(5)
                    .build(),
            ),
        }
    }
}

#[tokio::main]
async fn main() {
    use tracing_subscriber::prelude::*;

    tracing_subscriber::registry()
        .with(
            tracing_subscriber::EnvFilter::builder()
                .with_default_directive(LevelFilter::DEBUG.into())
                .from_env_lossy(),
        )
        .with(
            tracing_subscriber::fmt::layer()
                .with_target(false)
                .with_level(true)
                .compact(),
        )
        .init();

    let state = Arc::new(AppState::new());

    let app = Router::new()
        .route("/bucket", post(use_bucket))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

async fn use_bucket(State(state): State<Arc<AppState>>) -> Response {
    debug!("API called.");
    if !state.tokens.try_acquire(1) {
        return (StatusCode::TOO_MANY_REQUESTS, "No token left.\n").into_response();
    }
    (StatusCode::OK, "Token used.\n").into_response()
}

リクエスト例

curl -X POST http://localhost:3000/bucket
curl -X POST http://localhost:3000/bucket
curl -X POST http://localhost:3000/bucket
curl -X POST http://localhost:3000/bucket
curl -X POST http://localhost:3000/bucket
curl -X POST http://localhost:3000/bucket
sleep 2
curl -X POST http://localhost:3000/bucket
curl -X POST http://localhost:3000/bucket
curl -X POST http://localhost:3000/bucket

レスポンス出力結果

Token used.
Token used.
Token used.
Token used.
Token used.
No token left.
Token used.
Token used.
No token left.

ログ出力結果

2024-12-25T17:14:24.064404Z DEBUG API called.
2024-12-25T17:14:24.069417Z DEBUG API called.
2024-12-25T17:14:24.075361Z DEBUG API called.
2024-12-25T17:14:24.081062Z DEBUG API called.
2024-12-25T17:14:24.088231Z DEBUG API called.
2024-12-25T17:14:24.094721Z DEBUG API called.
2024-12-25T17:14:26.101068Z DEBUG API called.
2024-12-25T17:14:26.105920Z DEBUG API called.
2024-12-25T17:14:26.111101Z DEBUG API called.

バージョン情報

tokio ={ version = "1.42.0" , features = ["full"]}
leaky-bucket = "1.1.2"
axum = "0.7.9"
tracing = "0.1.41"
tracing-subscriber = {version = "0.3.19", features = ["env-filter"]}

leaky_bucketという名前について、ドキュメントにはtoken-basedと記載があるのでLeaky BucketとToken Bucketのどちらなのかという疑問が浮かびました。
内部的にtokenを利用しているだけと思いましたが、挙動的にはある程度のバーストを許容しているようにみえるのでToken Bucketなのかもしれません。 ↩
バケツから溢れたリクエストの扱いは重要だと思われました。
リクエストを即時拒否してもよいのか、それともできるだけ拒否しないように一定期間保持したいのか、今後触れる機会で意識しておきたいです。 ↩