leaky_bucketクレートの動作確認メモ

Leaky Bucketアルゴリズム（下記Rustクレート）に遭遇したので、その動作確認メモです。

[leaky_bucket - Rust](https://docs.rs/leaky-bucket/latest/leaky_bucket/)

## Leaky Bucketとは

> リーキーバケットはネットワークに送信するトラフィックに対して、その転送レートを制御する。名前の通り、穴の開いたバケツに相当し、様々な流量の水流がそのバケツに流れ込むと、そこまでの流量がどうであっても小さな穴からは一定の水流が流れ出す。同様にリーキーバケット・アルゴリズムは様々なバースト性のトラフィックを一定のトラフィックに変換する機構を提供する。
> [リーキーバケット - Wikipedia](https://ja.wikipedia.org/wiki/%E3%83%AA%E3%83%BC%E3%82%AD%E3%83%BC%E3%83%90%E3%82%B1%E3%83%83%E3%83%88)

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2127889/2da61ab8-76ee-3b65-573a-bb5619ce6b81.png)

画像引用元：[How we built rate limiting capable of scaling to millions of domains](https://blog.cloudflare.com/counting-things-a-lot-of-different-things/)


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

- Token Bucket
- Fixed Window
- Sliding Window

参考：
[レート制限アルゴリズム - システム設計 - GeeksforGeeks](https://www.geeksforgeeks.org/rate-limiting-algorithms-system-design/)
[様々なrate limitアルゴリズム - Carpe Diem](https://christina04.hatenablog.com/entry/rate-limiting-algorithm)
[QoS（9）帯域制御の制御方法 トークンバケット/リーキーバケット/万能リーキーバケットモデルとは – Ethernet TSN がIoTを変える](https://www.n-s-c.co.jp/Ethernet-TSN/network/qos/qos-bandwidth/2453/)


## [leaky_bucket](https://docs.rs/leaky-bucket/1.1.2/leaky_bucket/index.html) crateについて

Leaky BucketのRust実装である[leaky_bucket](https://docs.rs/leaky-bucket/1.1.2/leaky_bucket/index.html)[^1]の動作を確認します。
このクレートは[tokio](https://docs.rs/tokio/1.42.0/tokio/index.html)の機能に依存しているのでtokioランタイム上で使用する必要があるようです。


### 動作確認

やや冗長ですが、[axum](https://docs.rs/axum/latest/axum/index.html)サーバーで簡単に動作確認しました。
`RateLimiter`の設定項目がいくつかあり、単純な構成でレート制限する上ではこの設定項目でおおよそ十分な印象を勝手に受けました。
設定項目について詳しくは[ドキュメント](https://docs.rs/leaky-bucket/1.1.2/leaky_bucket/struct.RateLimiter.html#method.balance)を参照ください。

確認内容

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

```rust
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()
}

```


リクエスト例
```bash
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
```

レスポンス出力結果
```bash
Token used.
Token used.
Token used.
Token used.
Token used.
No token left.
Token used.
Token used.
No token left.
```

ログ出力結果
```log
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.
```


バージョン情報
```toml
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"]}
```


[^1]: leaky_bucketという名前について、ドキュメントにはtoken-basedと記載があるのでLeaky BucketとToken Bucketのどちらなのかという疑問が浮かびました。
内部的にtokenを利用しているだけと思いましたが、挙動的にはある程度のバーストを許容しているようにみえるのでToken Bucketなのかもしれません。

<!-- なおLeaky BucketとToken Bucketの違いは下記（Wikipedia）が端的と思われました。ただしわかりづらいので、もう少し理解が進んだら整理

> 大きな違いは、リーキーバケットがデータ転送レートの上限を設定するのに対して、トークンバケットはデータ転送レートの平均に制限を課して、ある程度のバースト性を許容する。
> [トークンバケット - Wikipedia](https://ja.wikipedia.org/wiki/%E3%83%88%E3%83%BC%E3%82%AF%E3%83%B3%E3%83%90%E3%82%B1%E3%83%83%E3%83%88) -->

[^2]: バケツから溢れたリクエストの扱いは重要だと思われました。
リクエストを即時拒否してもよいのか、それともできるだけ拒否しないように一定期間保持したいのか、今後触れる機会で意識しておきたいです。