timvw-rust-runtime 0.2.0

# timvw-rust-runtime

The runtime of Rust clients generated by
[openapi-generator-timvw-rust](https://github.com/timvw/openapi-generator-timvw-rust) (`-g timvw-rust`, with the
default `httpStack=tower`). A generated client sends every request through a `Client`: a
[tower](https://docs.rs/tower) service from `http::Request<Body>` to `http::Response<Body>`. That makes everything
around sending a tower layer: retries, timeouts, logging, headers, authentication, rate limits. Use the layers this
crate ships, any layer from the tower ecosystem, or write your own.

- [Quick start](#quick-start)
- [Retries](#retries)
- [Timeouts](#timeouts)
- [Logging](#logging)
- [Middleware: headers, authentication, layers of your own](#middleware)
- [Concurrency and rate limits](#concurrency-and-rate-limits)
- [Choosing the HTTP client, and tests without a network](#choosing-the-http-client)
- [Your own types, and streams of server-sent events](#your-own-types-and-server-sent-events)
- [Credentials](#credentials)
- [Headers and query parameters for one call](#headers-and-query-parameters-for-one-call)
- [Errors](#errors)
- [Features](#features)

## Quick start

A generated client keeps its `Client` in `Configuration.client`, which starts as `Client::default()`: reqwest, with
[retries](#retries) of `429`, `408`, `5xx` and failed connections. `layer` wraps it in more:

```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::trace::TraceLayer;
use timvw_rust_runtime::Client;
use tower::timeout::TimeoutLayer;

let client = Client::default() // reqwest, with retries
    .layer(TraceLayer::new()) // logs each call, and each retry
    .layer(TimeoutLayer::new(Duration::from_secs(120))); // bounds the whole call, retries included

// In a generated client:
// let mut configuration = my_client::apis::configuration::Configuration::new();
// configuration.client = client;
```

**Layers apply from the inside out: the last one added sees a request first.** Layers added to `Client::default()`
wrap its retries. To put a layer inside the retries (a timeout or a log line per attempt), or to retry differently,
build the stack from the bare service, `Client::from_service(ReqwestService::default())`:

```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::retry::{RetryLayer, RetryPolicy};
use timvw_rust_runtime::trace::TraceLayer;
use timvw_rust_runtime::{Client, ReqwestService};
use tower::timeout::TimeoutLayer;

let client = Client::from_service(ReqwestService::default()) // reqwest, no retries
    .layer(TimeoutLayer::new(Duration::from_secs(60))) // each attempt
    .layer(TraceLayer::new()) // logs each attempt
    .layer(RetryLayer::new(RetryPolicy::default())); // retries
```

A `Client` is cheap to clone, and clones share their layers' state (a rate limit, a token cache).

## Retries

`Client::default()`, `Client::new()` and `Client::from_reqwest(..)` retry with
`retry::RetryLayer::new(RetryPolicy::default())` (feature `retry`, on by default). `retry::RetryPolicy` is a
`tower::retry::Policy` for HTTP APIs:

- It retries `429 Too Many Requests`, `408 Request Timeout`, every `5xx` and failures to connect. Any other response
  or error is returned at once.
- It waits as long as the server asks, with `retry-after-ms` (milliseconds, as OpenAI sends) or `Retry-After`
  (seconds), up to `max_retry_after` (60 s by default). A response that asks for longer is returned at once instead
  of retried: waiting less than asked would likely fail again, and an hour-long wait is rarely what a caller wants.
  (`Retry-After` as an HTTP date is ignored, and the policy backs off.)
- Otherwise it backs off exponentially: 200 ms, 400 ms, 800 ms, ..., capped at `max_delay` (8 s). Each delay is
  jittered, picked at random between half and all of it, so that clients that failed together do not retry together;
  `.jitter(false)` turns that off. A server's `Retry-After` is not jittered, nor capped by `max_delay`.
- It retries at most 3 times by default.
- It does not retry a request whose body is a stream (a streamed file upload), since that cannot be sent twice. JSON,
  form, text and in-memory multipart bodies are sent again.

```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::retry::{RetryLayer, RetryPolicy};
use timvw_rust_runtime::{Client, ReqwestService};

let policy = RetryPolicy::new(5)
    .base_delay(Duration::from_millis(500))
    .max_delay(Duration::from_secs(30)) // the backoff
    .max_retry_after(Duration::from_secs(120)); // the longest Retry-After to wait for
let client = Client::from_service(ReqwestService::default()).layer(RetryLayer::new(policy));
```

`retry::RetryLayer::new(policy)` retries just like `tower::retry::RetryLayer::new(policy)`, which works too.

### Permanent `429`s: an exhausted quota

Some APIs answer `429` both for a rate limit, which passes, and for a quota or credit that is used up, which does not:
OpenAI's `insufficient_quota`. Only the body tells them apart. `RetryLayer::permanent_429` reads the body of each
`429` (up to `body_limit`, 64 KiB by default) and gives it to a function; when that returns `true`, the response is
returned at once, marked with the `retry::Permanent` extension. The body is put back, so the caller, and a generated
client's error, still read all of it. `retry::openai_insufficient_quota` is that function for OpenAI (an `error.code`
or `error.type` of `insufficient_quota`):

```rust,no_run
use timvw_rust_runtime::retry::{openai_insufficient_quota, RetryLayer, RetryPolicy};
use timvw_rust_runtime::{Client, ReqwestService};

let client = Client::from_service(ReqwestService::default()).layer(
    RetryLayer::new(RetryPolicy::default())
        .permanent_429(openai_insufficient_quota)
        .body_limit(16 * 1024),
);

// Any API: a function of the body's bytes.
let client = Client::from_service(ReqwestService::default()).layer(
    RetryLayer::default().permanent_429(|body: &[u8]| body.windows(14).any(|w| w == b"quota_exceeded")),
);
```

`Client::default()` does not read `429` bodies: what makes a `429` permanent differs between APIs, so the runtime
leaves it to the client of each API to turn on. A body longer than the limit, or one that fails to read, is treated as
a rate limit. A layer of your own inside the retries can mark any response not to retry:
`response.extensions_mut().insert(retry::Permanent)`.

Without retries: `Client::from_service(ReqwestService::default())`, or build without the `retry` feature.

The policy retries every method, `POST` included: for most APIs a `429` or `503` means the request was not processed.
If a retried `POST` could act twice on your API, send an idempotency key the API understands, or write a
`tower::retry::Policy` of your own: the trait has two methods, `retry` (whether to try again and after which delay)
and `clone_request`. `Body::try_clone` clones a body held in memory.

## Timeouts

`tower::timeout::TimeoutLayer` fails a request that takes too long. Where it sits decides what it bounds (see
[Quick start](#quick-start)). A timeout is an `Error` whose `is_timeout()` is true:

```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::{Client, Method};

async fn list_items(client: &Client) -> Result<(), timvw_rust_runtime::Error> {
    match client.request(Method::GET, "https://api.example.com/items").send().await {
        Err(error) if error.is_timeout() => eprintln!("too slow, try again later"),
        response => println!("{}", response?.status()),
    }
    Ok(())
}

let client = Client::default().layer(tower::timeout::TimeoutLayer::new(Duration::from_secs(30)));
```

reqwest's own timeouts (connect timeout, read timeout) are set on a `reqwest::Client`, see
[Choosing the HTTP client](#choosing-the-http-client). For a response streamed for minutes (server-sent events), a
total timeout ends the stream; use a connect timeout instead, or none.

## Logging

`trace::TraceLayer` (feature `tracing`) logs every request with [`tracing`](https://docs.rs/tracing):

- a `http.request` span with `http.request.method`, `url.full` (without the query, which can hold an API key) and
  `http.response.status_code`;
- an event when the response arrives, with `status` and `elapsed_ms`: `DEBUG` below `500`, `WARN` for a `5xx` or a
  failure to send;
- with `RetryPolicy`, a `DEBUG` event `retrying` with `attempt`, `max_retries`, `delay_ms` and `status`, and a
  `DEBUG` event `not retrying: ...` for a `Retry-After` beyond `max_retry_after` or a permanent `429`.

Headers and bodies are never logged. The target is `timvw_rust_runtime::trace`:

```rust,no_run
use timvw_rust_runtime::trace::TraceLayer;
use timvw_rust_runtime::Client;

// RUST_LOG=timvw_rust_runtime=debug shows every request, RUST_LOG=timvw_rust_runtime=warn only failures.
tracing_subscriber::fmt()
    .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
    .init();
let client = Client::default().layer(TraceLayer::new());
```

```text
 WARN http.request{http.request.method=POST url.full=https://api.example.com/v1/items http.response.status_code=503}: timvw_rust_runtime::trace: response status=503 elapsed_ms=41
DEBUG timvw_rust_runtime::trace: retrying attempt=1 max_retries=3 delay_ms=200 status=503
DEBUG http.request{http.request.method=POST url.full=https://api.example.com/v1/items http.response.status_code=200}: timvw_rust_runtime::trace: response status=200 elapsed_ms=380
```

reqwest's and hyper's own `DEBUG` events (connecting, pooling) are logged inside the `http.request` span too.

For more (headers, custom spans and fields, OpenTelemetry), use [`tower-http`](https://docs.rs/tower-http)'s
`TraceLayer`; it fits the same way:

```rust,no_run
let client = timvw_rust_runtime::Client::default().layer(tower_http::trace::TraceLayer::new_for_http());
```

## Middleware

Anything that changes a request or a response is a layer. Three ways to write one, from the shortest:

**A header on every request**: `tower::util::MapRequestLayer` with a function, or tower-http's
`SetRequestHeaderLayer`:

```rust,no_run
use timvw_rust_runtime::http::{self, HeaderName, HeaderValue};
use timvw_rust_runtime::{Body, Client};

let client = Client::default()
    .layer(tower::util::MapRequestLayer::new(|mut request: http::Request<Body>| {
        request.headers_mut().insert("x-team", HeaderValue::from_static("search"));
        request
    }))
    .layer(tower_http::set_header::SetRequestHeaderLayer::if_not_present(
        HeaderName::from_static("x-request-source"),
        HeaderValue::from_static("batch"),
    ));
```

**Several layers at once**: `tower::ServiceBuilder` composes layers top to bottom (the first one sees a request
first), the reverse of chaining `Client::layer`:

```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::retry::RetryLayer;
use timvw_rust_runtime::trace::TraceLayer;
use timvw_rust_runtime::{Client, ReqwestService};

let client = Client::from_service(ReqwestService::default()).layer(
    tower::ServiceBuilder::new()
        .layer(RetryLayer::default())
        .layer(TraceLayer::new())
        .timeout(Duration::from_secs(60)),
);
```

**A layer of your own**, for work that needs state or `await`, such as fetching and caching an access token. A layer
makes the service; the service changes the request and calls the service inside it:

```rust,no_run
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use timvw_rust_runtime::http::{self, header::AUTHORIZATION, HeaderValue};
use timvw_rust_runtime::{Body, Client};
use tower::{Layer, Service};

#[derive(Clone)]
struct AccessToken {
    cached: Arc<tokio::sync::Mutex<Option<String>>>,
}

impl<S> Layer<S> for AccessToken {
    type Service = WithAccessToken<S>;
    fn layer(&self, inner: S) -> Self::Service {
        WithAccessToken { inner, cached: self.cached.clone() }
    }
}

#[derive(Clone)]
struct WithAccessToken<S> {
    inner: S,
    cached: Arc<tokio::sync::Mutex<Option<String>>>,
}

impl<S> Service<http::Request<Body>> for WithAccessToken<S>
where
    S: Service<http::Request<Body>> + Clone + Send + 'static,
    S::Future: Send,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = Pin<Box<dyn Future<Output = Result<S::Response, S::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut request: http::Request<Body>) -> Self::Future {
        // Call the service that was polled ready, and keep a fresh clone for the next request.
        let fresh = self.inner.clone();
        let mut inner = std::mem::replace(&mut self.inner, fresh);
        let cached = self.cached.clone();
        Box::pin(async move {
            let mut cached = cached.lock().await;
            if cached.is_none() {
                *cached = Some(fetch_token().await);
            }
            let value = HeaderValue::from_str(&format!("Bearer {}", cached.as_deref().unwrap())).unwrap();
            drop(cached);
            request.headers_mut().insert(AUTHORIZATION, value);
            inner.call(request).await
        })
    }
}

async fn fetch_token() -> String {
    "token".to_owned() // ask your identity provider
}

let client = Client::default().layer(AccessToken { cached: Arc::default() });
```

A layer's service must be `Clone + Send`; it need not be `Sync`. Its response body can be any `http_body::Body`.

## Concurrency and rate limits

`tower::limit::ConcurrencyLimitLayer` caps requests in flight; the clones of a `Client` share the cap.
`RateLimit` cannot be cloned, so put a `Buffer` in front of it: the buffer shares one limiter between all clones. A
buffer starts a tokio task, so build that client inside a tokio runtime.

```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::Client;

#[tokio::main]
async fn main() {
    let client = Client::default()
        .layer(tower::limit::ConcurrencyLimitLayer::new(8))
        .layer(tower::ServiceBuilder::new().buffer(256).rate_limit(50, Duration::from_secs(1)));
}
```

These layers need tower's `limit` and `buffer` features in your `Cargo.toml`, like `timeout` for `TimeoutLayer`:
`tower = { version = "0.5", features = ["limit", "buffer", "timeout"] }`.

## Choosing the HTTP client

`Client::from_reqwest` sends with a `reqwest::Client` you built (a proxy, a custom TLS setup, connect and read timeouts,
pool settings), with the default retries; `Client::from_service(ReqwestService::new(reqwest_client))` without them:

```rust,no_run
use std::time::Duration;
use timvw_rust_runtime::Client;

let reqwest_client = reqwest::Client::builder()
    .connect_timeout(Duration::from_secs(5))
    .proxy(reqwest::Proxy::all("http://proxy.internal:3128").unwrap())
    .build()
    .unwrap();
let client = Client::from_reqwest(reqwest_client);
```

`Client::from_service` sends with any tower service from `http::Request<Body>` to `http::Response<B>`, as is, with no
retries added: a hyper client, a client for another runtime, or a function that answers without a network, for tests:

```rust,no_run
use timvw_rust_runtime::{http, Body, BoxError, Client};

let client = Client::from_service(tower::service_fn(|request: http::Request<Body>| async move {
    assert_eq!(request.uri().path(), "/v1/items");
    let page = r#"{"data":[{"id":"1","name":"Lamp"}],"has_more":false}"#;
    Ok::<_, BoxError>(
        http::Response::builder()
            .header("content-type", "application/json")
            .body(Body::from(page))?,
    )
}));
```

Without the `reqwest` feature (and `rustls` or `native-tls`), the crate does not depend on reqwest, and
`Client::default()` fails every request: set a service with `Client::from_service`.

## Your own types and server-sent events

A generated function decodes to the spec's models. To send and decode types of your own (fewer fields, a provider
whose replies differ, `serde_json::Value`), call the client directly: the same layers apply (retries, logging).
`Response::events()` reads a `text/event-stream` body, and `.json::<T>()` decodes each event's `data:`:

```rust,no_run
use futures_util::StreamExt;
use serde::Deserialize;
use timvw_rust_runtime::{Client, ExposeSecret, Method, SecretString};

#[derive(Deserialize)]
struct Delta {
    id: String,
}

async fn run(client: &Client, api_key: &SecretString) -> Result<(), timvw_rust_runtime::Error> {
    let request = serde_json::json!({"model": "some-model", "stream": true});
    let response = client
        .request(Method::POST, "https://api.example.com/v1/chat/completions")
        .bearer_auth(api_key.expose_secret())
        .json(&request)
        .send()
        .await?;
    let mut deltas = response.events().json::<Delta>();
    while let Some(delta) = deltas.next().await {
        println!("{}", delta?.id);
    }
    Ok(())
}
```

`data: [DONE]` or the end of the body ends the stream. A payload that does not decode yields an error whose
`is_decode()` is true, and the stream goes on. `events()` alone gives each `sse::Event` undecoded (`event:`, `data:`,
`id:`), and `sse::Decoder` is the parser, for a body read some other way. A generated client's typed event streams use
the same parser.

## Credentials

A generated `Configuration` keeps its tokens, basic-auth password and api key in
[`secrecy`](https://docs.rs/secrecy)'s `SecretString`, which this crate re-exports (`timvw_rust_runtime::secrecy`,
`SecretString` and `ExposeSecret`). A `SecretString` is made from a `String` or a `&str`, its `Debug` is redacted,
and it is zeroed when dropped. `bearer_auth`, `basic_auth` and `sensitive_header` take the exposed value and mark the
header sensitive, so the `Debug` of the request's headers leaves it out:

```rust
use timvw_rust_runtime::{Client, ExposeSecret, Method, SecretString};

let key = SecretString::from("sk-123");
assert_eq!(format!("{key:?}"), "SecretBox<str>([REDACTED])");

let request = Client::default()
    .request(Method::GET, "https://api.example.com/v1/models")
    .sensitive_header("x-api-key", key.expose_secret())
    .build()?;
assert!(request.headers()["x-api-key"].is_sensitive());
assert!(!format!("{:?}", request.headers()).contains("sk-123"));
# Ok::<(), timvw_rust_runtime::Error>(())
```

## Headers and query parameters for one call

A layer changes every request. `RequestOptions` holds extra headers and query parameters for one call; every
operation of a generated client takes it, `.options(&options)` on its builder, and `RequestBuilder::options` applies it
to a request of your own:

```rust,no_run
use timvw_rust_runtime::{Client, Method, RequestOptions};

async fn run(client: &Client) -> Result<(), timvw_rust_runtime::Error> {
    let options = RequestOptions::new()
        .header("x-request-source", "batch")?
        .query(&[("trace", "1")])?;
    // generated: items_api::list_items().configuration(&configuration).options(&options).call().await
    let response = client
        .request(Method::GET, "https://api.example.com/v1/items?limit=10")
        .bearer_auth("secret")
        .options(&options)
        .send()
        .await?; // GET /v1/items?limit=10&trace=1, with x-request-source: batch
    Ok(())
}
```

- A header of the options replaces every value the request has under its name, whatever set it: a generated header
  parameter, `Authorization`, `Content-Type`, `Accept`. Generated clients apply the options last, after everything
  else. `RequestOptions::header` sets one value (a second call for the same name replaces it); `headers` merges a
  `HeaderMap`, keeping several values of one name.
- Query parameters are appended after the request's own: a name the request already has is sent twice.
- `header` and `query` check their input at once and return an `Error` whose `is_builder()` is true for an invalid
  header name or value, or a query that is not a flat list of pairs.
- The options do not change the path or the body. One value can be cloned, or reused for many calls.
- A header value marked with `HeaderValue::set_sensitive` stays sensitive (hidden from `Debug` output).

## Errors

A request that fails to send is an `Error`:

| Method | True when |
| --- | --- |
| `is_builder()` | the request could not be built (an invalid URL, header or body) |
| `is_request()` | the service failed; `is_timeout()` and `is_connect()` narrow it down |
| `is_body()` | reading the response body failed |
| `is_decode()` | the response could not be decoded (JSON) |

`source()` is the error of the service or layer that failed: a `reqwest::Error`, tower's `Elapsed`, or your own. A
generated client wraps it as `apis::Error::Http(error)`. A response with an error status is not an `Error` here; the
generated client turns it into `apis::Error::ResponseError`.

## Features

| Feature | Default | What it adds |
| --- | --- | --- |
| `rustls` | yes | reqwest with rustls, as `Client::default()` |
| `native-tls` | no | reqwest with the platform's TLS |
| `reqwest` | via `rustls` or `native-tls` | `ReqwestService`, `Client::from_reqwest` |
| `retry` | yes | `retry::RetryLayer` and `retry::RetryPolicy` (sleep with tokio), and retries in `Client::default()` |
| `tracing` | no | `trace::TraceLayer`, and a log event per retry |

A generated crate forwards `native-tls`, `rustls`, `retry` and `tracing` under the same names.

## What the generated code uses

The generated functions build requests with `RequestBuilder` (`query`, `header`, `sensitive_header`, `bearer_auth`,
`basic_auth`, `json`, `form`, `body`, `multipart`, `options`, `build`) and read responses with `Response` (`status`, `headers`, `text`, `bytes`,
`bytes_stream`): the methods of reqwest's types of the same names. `multipart::Form` encodes multipart bodies.

## License

MIT OR Apache-2.0.