Crate timvw_rust_runtime

Expand description

§timvw-rust-runtime

The runtime of Rust clients generated by openapi-generator-timvw-rust (-g timvw-rust, with the default httpStack=tower). A generated client sends every request through a Client: a 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
Retries
Timeouts
Logging
Middleware: headers, authentication, layers of your own
Concurrency and rate limits
Choosing the HTTP client, and tests without a network
Your own types, and streams of server-sent events
Credentials
Headers and query parameters for one call
Errors
Features

§Quick start

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

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()):

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.

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):

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). A timeout is an Error whose is_timeout() is true:

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. 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:

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:

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());

 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’s TraceLayer; it fits the same way:

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:

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:

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:

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.

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:

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:

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::

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’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:

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"));

§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:

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.

Re-exports§

pub use http;
pub use secrecy;
pub use tower;

Modules§

header: HTTP header types
multipart: multipart/form-data bodies, with the API of reqwest’s multipart module that generated clients use.
retryretry: Retries for HTTP APIs: RetryPolicy, a tower retry policy, and RetryLayer, which adds to it a look at the body of a 429 to tell a rate limit (retried) from an exhausted quota (not retried).
sse: Server-sent events (text/event-stream): a parser, the events of a response, and their data: decoded as JSON.
tracetracing: Logs every request with tracing (feature tracing).

Structs§

Body: The body of a request or a response: bytes held in memory, or a stream.
Client: Sends the requests of a generated client through a tower service.
Error: An error building a request, sending it, or reading the response body.
HeaderMap: A specialized multimap for header names and values.
Method: The Request Method (VERB)
RequestBuilder: Builds a request, with the methods of reqwest’s RequestBuilder that generated clients use.
RequestOptions: Extra headers and query parameters for one call, applied after everything the generated code sets.
ReqwestServicereqwest: A tower service that sends http::Request<Body> with a reqwest::Client.
Response: A response, with the methods of reqwest’s Response that generated clients use.
StatusCode: An HTTP status code (status-code in RFC 9110 et al.).

Traits§

ExposeSecret: Expose a reference to an inner secret

Type Aliases§

BoxError: The error type services and layers return: any error that can cross threads.
HttpService: The service a Client sends requests through, with its type erased.
SecretString: Secret string type.