📁 Rust Backend with Axum · Part 33 of 42

Rust Web Server with Axum 0.8: Health Checks and Prometheus Metrics

November 3, 2026

Episode 33 of the Rust Backend series, where we build the backend of a blog step by step. After episode 32 slow requests time out and restarts are graceful. But how do we know our server is fine? Today we let it tell us: health checks for the machines that run it, and metrics that count everything it does. All the code below is the final, tested code from the video, ready to copy.

New to Rust? Start with the beginner series.
The full backend series playlist is here.

We continue in the same blog_api project from episode 32. New crates today: metrics and metrics-exporter-prometheus.

1. Live and ready

Two kinds of health:

live: the process runs and answers. If not, restart it.
ready: it can do real work, so the database and Redis answer too. If not, stop sending it traffic for now.

A load balancer, Docker or Kubernetes asks these all the time.

2. The two routes

Live just says ok:

rust

pub async fn health_live() -> &'static str {
    "ok"
}

Ready runs SELECT 1 on the database and a PING on Redis. If one fails: 503 Service Unavailable, and which part failed:

rust

pub async fn health_ready<R: Repo>(
    State(repo): State<R>,
    Extension(redis): Extension<ConnectionManager>,
) -> (StatusCode, Json<Value>) {
    let db = repo.ping().await.is_ok();
    let redis_ok = redis::cmd("PING").query_async::<String>(&mut redis.clone()).await.is_ok();
    let status = if db && redis_ok { StatusCode::OK } else { StatusCode::SERVICE_UNAVAILABLE };
    (status, Json(json!({ "database": if db { "up" } else { "down" }, "redis": if redis_ok { "up" } else { "down" } })))
}

rust

async fn ping(&self) -> sqlx::Result<()> {
    sqlx::query("SELECT 1").execute(&self.pool).await?;
    Ok(())
}

rust

.route("/health/live", get(handlers::health_live))
.route("/health/ready", get(handlers::health_ready::<R>))

3. Redis goes down

bash

cargo run
curl -i localhost:8080/health/live
curl -i localhost:8080/health/ready

200 and 200. Now stop Redis:

bash

sudo systemctl stop redis-server
curl -i localhost:8080/health/live
curl -i localhost:8080/health/ready

Live: still 200. Ready:

text

HTTP/1.1 503 Service Unavailable

{"database":"up","redis":"down"}

A load balancer stops sending us traffic, but nobody restarts a process that is fine.

4. Metrics and Prometheus

Metrics are numbers over time: how many requests, how many errors, how long they take. Prometheus is the common format: plain text at /metrics, which a Prometheus server reads every few seconds.

bash

sudo systemctl start redis-server
cargo add metrics metrics-exporter-prometheus

At start we install the recorder. It returns a handle that renders all metrics as text, with our own histogram buckets (in seconds):

rust

let handle = PrometheusBuilder::new().set_buckets(&[0.005, 0.01, 0.05, 0.1, 0.5, 1.0]).unwrap().install_recorder().unwrap();

The app gets the handle as an Extension, and /metrics returns the text:

rust

pub async fn metrics_text(Extension(handle): Extension<PrometheusHandle>) -> String {
    handle.render()
}

rust

.route("/metrics", get(handlers::metrics_text))

5. A middleware that counts every request

A middleware function runs around every request. It starts a timer, runs the request with next.run, then counts it with method, path and status, and records how long it took:

rust

use axum::extract::Request;
use axum::middleware::{self, Next};
use axum::response::Response;

async fn track(req: Request, next: Next) -> Response {
    let start = std::time::Instant::now();
    let method = req.method().to_string();
    let path = req.uri().path().to_string();
    let response = next.run(req).await;
    let status = response.status().as_u16().to_string();
    let labels = [("method", method), ("path", path), ("status", status)];
    metrics::counter!("http_requests_total", &labels).increment(1);
    metrics::histogram!("http_request_duration_seconds", &labels).record(start.elapsed().as_secs_f64());
    response
}

rust

.layer(middleware::from_fn(track))

6. One line per post

bash

cargo run
for i in 1 2 3; do curl -s -o /dev/null localhost:8080/posts/$i; done; curl -s localhost:8080/metrics | grep http_requests_total

text

http_requests_total{method="GET",path="/posts/1",status="200"} 1
http_requests_total{method="GET",path="/posts/2",status="200"} 1
http_requests_total{method="GET",path="/posts/3",status="200"} 1

Look at the path label: one line per post. With a million posts, a million time series. That can take down Prometheus.

We want the route, not the URL. axum's MatchedPath gives the pattern, /posts/{id}. It is only set after routing, so it exists inside a route_layer, and the middleware moves there:

rust

let path = req
    .extensions()
    .get::<MatchedPath>()
    .map(|p| p.as_str().to_string())
    .unwrap_or_else(|| "unmatched".to_string());

rust

.merge(streaming)
.route_layer(middleware::from_fn(track))
.with_state(repo)

bash

cargo run
for i in 1 2 3; do curl -s -o /dev/null localhost:8080/posts/$i; done; curl -s localhost:8080/metrics | grep http_requests_total

text

http_requests_total{method="GET",path="/posts/{id}",status="200"} 3

One line for all posts.

7. Histograms and our own counters

The duration is a histogram: buckets counting how many requests took under 5 ms, under 10 ms, and so on:

bash

curl -s localhost:8080/metrics | grep http_request_duration_seconds_bucket | head -7

Each _bucket line has an le="..." ("less or equal") label. From these, Prometheus computes percentiles, like how slow the slowest 1% of requests are.

We can count our own things too. Every new post adds one, in create_post:

rust

metrics::counter!("posts_created_total").increment(1);

Metrics show how our system works inside, so /metrics should not be public. In production it listens on a separate internal port, or is blocked at the proxy.

The final code

src/handlers.rs

rust

use axum::extract::{rejection::QueryRejection, Extension, Multipart, Path, Query, State};
use axum::{http::{header, StatusCode}, response::{IntoResponse, Response}, Json};

use crate::error::ApiError;
use crate::extract::{AdminUser, AuthUser, Role, ValidJson};
use crate::models::{Comment, Hit, Image, ListFilter, ListParams, NewComment, NewPost, NewUser, Page, Post, PostV2, PostWithTags, SearchParams, User};
use crate::repo::Repo;
use crate::auth;
use crate::config::config;
use serde::Deserialize;
use serde_json::{json, Value};
use axum_extra::extract::cookie::{Cookie, CookieJar, SameSite};
use uuid::Uuid;
use crate::storage::Storage;
use std::sync::Arc;
use std::io;
use crate::jobs::Job;
use tokio::sync::mpsc;
use crate::cache;
use redis::aio::ConnectionManager;
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use tokio::sync::broadcast;
use tokio::sync::broadcast::error::RecvError;
use axum::response::sse::{Event, KeepAlive, Sse};
use std::convert::Infallible;
use tokio_stream::{wrappers::BroadcastStream, Stream, StreamExt};
use metrics_exporter_prometheus::PrometheusHandle;

#[tracing::instrument(skip_all)]
pub async fn create_post<R: Repo>(
    State(repo): State<R>,
    user: AuthUser,
    ValidJson(new): ValidJson<NewPost>,
) -> Result<(StatusCode, Json<PostWithTags>), ApiError> {
    let post = repo.create(user.id, new).await.map_err(|err| match err.as_database_error() {
        Some(e) if e.is_unique_violation() => ApiError::Conflict("a post with this title already exists".into()),
        _ => ApiError::Database(err),
    })?;
    tracing::debug!(id = post.post.id, "created");
    metrics::counter!("posts_created_total").increment(1);
    Ok((StatusCode::CREATED, Json(post)))
}

pub async fn get_post<R: Repo>(
    State(repo): State<R>,
    Extension(redis): Extension<ConnectionManager>,
    Path(id): Path<i64>,
) -> Result<Json<Post>, ApiError> {
    if let Some(post) = cache::get(&redis, id).await {
        tracing::debug!(id, "cache hit");
        return Ok(Json(post));
    }
    tracing::debug!(id, "cache miss");
    let post = repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    cache::put(&redis, &post).await;
    Ok(Json(post))
}

pub async fn list_posts<R: Repo>(
    State(repo): State<R>,
    params: Result<Query<ListParams>, QueryRejection>,
) -> Result<Json<Page<Post>>, ApiError> {
    let Query(p) = params?;
    let page = p.page.unwrap_or(1).max(1);
    let per_page = p.per_page.unwrap_or(10).clamp(1, 100);
    let (items, total) = repo.list(ListFilter { page, per_page, tag: p.tag, sort: p.sort }).await?;
    Ok(Json(Page { items, page, per_page, total }))
}

pub async fn update_post<R: Repo>(
    State(repo): State<R>,
    Extension(redis): Extension<ConnectionManager>,
    Path(id): Path<i64>,
    user: AuthUser,
    ValidJson(new): ValidJson<NewPost>,
) -> Result<Json<Post>, ApiError> {
    let post = repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    if !can_edit(&user, &post) {
        return Err(ApiError::Forbidden);
    }
    let post = repo.update(id, new).await?.ok_or(ApiError::NotFound(id))?;
    cache::del(&redis, id).await;
    Ok(Json(post))
}

pub async fn delete_post<R: Repo>(State(repo): State<R>, Extension(redis): Extension<ConnectionManager>, Path(id): Path<i64>, user: AuthUser) -> Result<StatusCode, ApiError> {
    let post = repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    if !can_edit(&user, &post) {
        return Err(ApiError::Forbidden);
    }
    repo.delete(id).await?;
    cache::del(&redis, id).await;
    Ok(StatusCode::NO_CONTENT)
}

pub async fn list_comments<R: Repo>(
    State(repo): State<R>,
    Path(id): Path<i64>,
) -> Result<Json<Vec<Comment>>, ApiError> {
    repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    Ok(Json(repo.list_comments(id).await?))
}

pub async fn add_comment<R: Repo>(
    State(repo): State<R>,
    Path(id): Path<i64>,
    Extension(live): Extension<broadcast::Sender<Comment>>,
    _user: AuthUser,
    ValidJson(new): ValidJson<NewComment>,
) -> Result<(StatusCode, Json<Comment>), ApiError> {
    let comment = repo.add_comment(id, new).await.map_err(|err| {
        match err.as_database_error() {
            Some(db_err) if db_err.is_foreign_key_violation() => ApiError::NotFound(id),
            _ => ApiError::Database(err),
        }
    })?;
    let _ = live.send(comment.clone());
    Ok((StatusCode::CREATED, Json(comment)))
}

pub async fn search_posts<R: Repo>(
    State(repo): State<R>,
    Query(p): Query<SearchParams>,
) -> Result<Json<Vec<Hit>>, ApiError> {
    Ok(Json(repo.search(&p.q).await?))
}

pub async fn restore_post<R: Repo>(State(repo): State<R>, Path(id): Path<i64>) -> Result<StatusCode, ApiError> {
    if !repo.restore(id).await? {
        return Err(ApiError::NotFound(id));
    }
    Ok(StatusCode::OK)
}

pub async fn signup<R: Repo>(
    State(repo): State<R>,
    Extension(jobs): Extension<mpsc::Sender<Job>>,
    ValidJson(new): ValidJson<NewUser>,
) -> Result<(StatusCode, Json<User>), ApiError> {
    let hash = crate::auth::hash_password(new.password).await?;
    let user = repo.create_user(&new.email, &hash).await.map_err(|err| match err.as_database_error() {
        Some(e) if e.is_unique_violation() => ApiError::Conflict("email already registered".into()),
        _ => ApiError::Database(err),
    })?;
    if jobs.send(Job::Welcome { email: user.email.clone() }).await.is_err() {
        tracing::error!("job queue closed");
    }
    Ok((StatusCode::CREATED, Json(user)))
}

#[derive(Deserialize)]
pub struct Login {
    pub email: String,
    pub password: String,
}

fn refresh_cookie(token: String) -> Cookie<'static> {
    Cookie::build(("refresh", token))
        .http_only(true)
        .secure(true)
        .same_site(SameSite::Strict)
        .path("/auth")
        .max_age(time::Duration::days(30))
        .build()
}

pub async fn login<R: Repo>(
    State(repo): State<R>,
    jar: CookieJar,
    Json(body): Json<Login>,
) -> Result<(CookieJar, Json<Value>), ApiError> {
    let user = repo.find_user_by_email(&body.email).await?.ok_or(ApiError::Unauthorized)?;
    if !auth::verify_password(body.password, user.password_hash).await? {
        return Err(ApiError::Unauthorized);
    }
    let access = auth::make_token(user.id, &user.role, &config().jwt_secret)?;
    let refresh = auth::new_refresh_token();
    repo.store_refresh(&auth::hash_token(&refresh), user.id, auth::now() + auth::REFRESH_SECS).await?;
    Ok((jar.add(refresh_cookie(refresh)), Json(json!({ "token": access }))))
}

pub async fn me<R: Repo>(State(repo): State<R>, user: AuthUser) -> Result<Json<User>, ApiError> {
    let user = repo.get_user(user.id).await?.ok_or(ApiError::Unauthorized)?;
    Ok(Json(user))
}

pub async fn refresh<R: Repo>(State(repo): State<R>, jar: CookieJar) -> Result<(CookieJar, Json<Value>), ApiError> {
    let old = jar.get("refresh").ok_or(ApiError::Unauthorized)?.value().to_string();
    let user_id = repo.take_refresh(&auth::hash_token(&old), auth::now()).await?.ok_or(ApiError::Unauthorized)?;
    let user = repo.get_user(user_id).await?.ok_or(ApiError::Unauthorized)?;
    let access = auth::make_token(user.id, &user.role, &config().jwt_secret)?;
    let new = auth::new_refresh_token();
    repo.store_refresh(&auth::hash_token(&new), user.id, auth::now() + auth::REFRESH_SECS).await?;
    Ok((jar.add(refresh_cookie(new)), Json(json!({ "token": access }))))
}

pub async fn logout<R: Repo>(State(repo): State<R>, jar: CookieJar) -> Result<(CookieJar, StatusCode), ApiError> {
    if let Some(cookie) = jar.get("refresh") {
        repo.delete_refresh(&auth::hash_token(cookie.value())).await?;
    }
    Ok((jar.remove(Cookie::build("refresh").path("/auth")), StatusCode::NO_CONTENT))
}

pub fn can_edit(user: &AuthUser, post: &Post) -> bool {
    user.role == Role::Admin || post.author_id == Some(user.id)
}

pub async fn list_users<R: Repo>(State(repo): State<R>, _admin: AdminUser) -> Result<Json<Vec<User>>, ApiError> {
    Ok(Json(repo.list_users().await?))
}

pub async fn upload_image<R: Repo, S: Storage>(
    State(repo): State<R>,
    user: AuthUser,
    Path(id): Path<i64>,
    Extension(store): Extension<Arc<S>>,
    mut multipart: Multipart,
) -> Result<(StatusCode, Json<Image>), ApiError> {
    let post = repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    if !can_edit(&user, &post) {
        return Err(ApiError::Forbidden);
    }
    while let Some(field) = multipart.next_field().await? {
        if field.name() != Some("file") {
            continue;
        }
        let bytes = field.bytes().await?;
        let ext = image_kind(&bytes).ok_or(ApiError::UnsupportedMedia)?;
        let name = format!("{}.{ext}", Uuid::new_v4());
        store.put(&name, &bytes).await?;
        let image = repo.add_image(id, &format!("/files/{name}")).await?;
        return Ok((StatusCode::CREATED, Json(image)));
    }
    Err(ApiError::MissingFile)
}

fn image_kind(bytes: &[u8]) -> Option<&'static str> {
    if bytes.starts_with(&[0x89, b'P', b'N', b'G']) {
        Some("png")
    } else if bytes.starts_with(&[0xFF, 0xD8, 0xFF]) {
        Some("jpg")
    } else {
        None
    }
}

pub async fn get_file<S: Storage>(Extension(store): Extension<Arc<S>>, Path(name): Path<String>) -> Result<Response, ApiError> {
    let bytes = store.get(&name).await.map_err(|err| match err.kind() {
        io::ErrorKind::NotFound | io::ErrorKind::InvalidInput => ApiError::FileNotFound,
        _ => ApiError::Io(err),
    })?;
    let content_type = if name.ends_with(".png") { "image/png" } else { "image/jpeg" };
    Ok(([(header::CONTENT_TYPE, content_type)], bytes).into_response())
}

pub async fn live(
    Extension(live): Extension<broadcast::Sender<Comment>>,
    Path(post_id): Path<i64>,
    ws: WebSocketUpgrade,
) -> Response {
    let rx = live.subscribe();
    ws.on_upgrade(move |socket| live_socket(socket, rx, post_id))
}

async fn live_socket(mut socket: WebSocket, mut rx: broadcast::Receiver<Comment>, post_id: i64) {
    loop {
        tokio::select! {
            event = rx.recv() => match event {
                Ok(c) => {
                    if c.post_id != post_id {
                        continue;
                    }
                    let text = serde_json::to_string(&c).unwrap();
                    if socket.send(Message::Text(text.into())).await.is_err() {
                        break;
                    }
                }
                Err(RecvError::Lagged(n)) => {
                    let _ = socket.send(Message::Text(format!("{{\"missed\":{n}}}").into())).await;
                }
                Err(RecvError::Closed) => break,
            },
            msg = socket.recv() => match msg {
                None | Some(Ok(Message::Close(_))) | Some(Err(_)) => break,
                Some(Ok(_)) => {}
            },
        }
    }
    tracing::info!(post_id, "live socket closed");
}

pub async fn events(
    Extension(live): Extension<broadcast::Sender<Comment>>,
    Path(post_id): Path<i64>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let stream = BroadcastStream::new(live.subscribe()).filter_map(move |msg| match msg {
        Ok(c) if c.post_id == post_id => Some(Ok(Event::default().event("comment").json_data(&c).unwrap())),
        _ => None,
    });
    Sse::new(stream).keep_alive(KeepAlive::default())
}

pub async fn get_post_v2<R: Repo>(State(repo): State<R>, Path(id): Path<i64>) -> Result<Json<PostV2>, ApiError> {
    let post = repo.get(id).await?.ok_or(ApiError::NotFound(id))?;
    Ok(Json(post.into()))
}

#[derive(Deserialize)]
pub struct SlowParams {
    secs: Option<u64>,
}

pub async fn slow(Query(p): Query<SlowParams>) -> &'static str {
    tokio::time::sleep(std::time::Duration::from_secs(p.secs.unwrap_or(30))).await;
    "finally done"
}

pub async fn health_live() -> &'static str {
    "ok"
}

pub async fn health_ready<R: Repo>(
    State(repo): State<R>,
    Extension(redis): Extension<ConnectionManager>,
) -> (StatusCode, Json<Value>) {
    let db = repo.ping().await.is_ok();
    let redis_ok = redis::cmd("PING").query_async::<String>(&mut redis.clone()).await.is_ok();
    let status = if db && redis_ok { StatusCode::OK } else { StatusCode::SERVICE_UNAVAILABLE };
    (status, Json(json!({ "database": if db { "up" } else { "down" }, "redis": if redis_ok { "up" } else { "down" } })))
}

pub async fn metrics_text(Extension(handle): Extension<PrometheusHandle>) -> String {
    handle.render()
}

src/main.rs

rust

mod auth;
mod config;
mod error;
mod extract;
mod handlers;
mod models;
mod repo;
mod storage;
mod mail;
mod jobs;
mod cache;

use axum::{routing::{get, post, put}, Extension, Router};
use repo::{Repo, SqliteRepo};
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
use std::str::FromStr;
use tower_http::trace::TraceLayer;
use tracing_subscriber::EnvFilter;
use axum::extract::DefaultBodyLimit;
use axum::http::{header, HeaderName, HeaderValue, Method, StatusCode};
use tower_http::cors::CorsLayer;
use tower_http::set_header::SetResponseHeaderLayer;
use tower_http::timeout::TimeoutLayer;
use std::sync::Arc;
use tower_governor::{governor::GovernorConfigBuilder, GovernorLayer};
use std::net::SocketAddr;
use storage::{LocalStorage, MemStorage, Storage};
use mail::Mailer;
use jobs::Job;
use tokio::sync::mpsc;
use tokio::time::Duration;
use redis::aio::{ConnectionManager, ConnectionManagerConfig};
use tokio::sync::broadcast;
use models::Comment;
use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
use axum::extract::{MatchedPath, Request};
use axum::middleware::{self, Next};
use axum::response::Response;

pub fn app<R: Repo, S: Storage>(repo: R, store: S, jobs: mpsc::Sender<Job>, cache: ConnectionManager, live: broadcast::Sender<Comment>, handle: PrometheusHandle) -> Router {
    let cors = CorsLayer::new()
        .allow_origin(HeaderValue::from_static("http://localhost:5500"))
        .allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE])
        .allow_headers([header::AUTHORIZATION, header::CONTENT_TYPE])
        .allow_credentials(true);
    let governor = Arc::new(GovernorConfigBuilder::default().per_second(2).burst_size(5).finish().unwrap());
    let auth_routes = Router::new()
        .route("/signup", post(handlers::signup::<R>))
        .route("/login", post(handlers::login::<R>))
        .route("/auth/refresh", post(handlers::refresh::<R>))
        .layer(GovernorLayer::new(governor));
    let streaming = Router::new()
        .route("/posts/{id}/live", get(handlers::live))
        .route("/posts/{id}/events", get(handlers::events));
    Router::new()
        .nest("/api/v1", posts_v1())
        .nest("/api/v2", posts_v2())
        .merge(posts_v1())
        .route("/posts/{id}/restore", post(handlers::restore_post::<R>))
        .route("/me", get(handlers::me::<R>))
        .route("/admin/users", get(handlers::list_users::<R>))
        .route("/auth/logout", post(handlers::logout::<R>))
        .route(
            "/posts/{id}/images",
            post(handlers::upload_image::<R, S>).layer(DefaultBodyLimit::max(10 * 1024 * 1024)),
        )
        .route("/files/{name}", get(handlers::get_file::<S>))
        .route("/debug/slow", get(handlers::slow))
        .route("/health/live", get(handlers::health_live))
        .route("/health/ready", get(handlers::health_ready::<R>))
        .route("/metrics", get(handlers::metrics_text))
        .merge(auth_routes)
        .layer(TimeoutLayer::with_status_code(StatusCode::REQUEST_TIMEOUT, Duration::from_secs(10)))
        .merge(streaming)
        .route_layer(middleware::from_fn(track))
        .with_state(repo)
        .layer(Extension(Arc::new(store)))
        .layer(Extension(jobs))
        .layer(Extension(cache))
        .layer(Extension(live))
        .layer(Extension(handle))
        .layer(cors)
        .layer(SetResponseHeaderLayer::if_not_present(header::X_CONTENT_TYPE_OPTIONS, HeaderValue::from_static("nosniff")))
        .layer(SetResponseHeaderLayer::if_not_present(header::X_FRAME_OPTIONS, HeaderValue::from_static("DENY")))
        .layer(SetResponseHeaderLayer::if_not_present(header::REFERRER_POLICY, HeaderValue::from_static("no-referrer")))
        .layer(TraceLayer::new_for_http())
}

#[tokio::main]
async fn main() {
    let cfg = config::config();
    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")))
        .init();
    let addr = format!("{}:{}", cfg.host, cfg.port);
    let options = SqliteConnectOptions::from_str(&cfg.database_url)
        .unwrap()
        .create_if_missing(true)
        .journal_mode(SqliteJournalMode::Wal);
    let pool = SqlitePoolOptions::new()
        .max_connections(5)
        .connect_with(options)
        .await
        .unwrap();
    sqlx::migrate!().run(&pool).await.unwrap();
    let repo = SqliteRepo { pool };
    let (tx, rx) = mpsc::channel::<Job>(100);
    let mailer = Mailer::new(cfg);
    tokio::spawn(jobs::worker(rx, mailer));
    tokio::spawn(jobs::cleanup(repo.clone(), Duration::from_secs(cfg.cleanup_secs)));
    let client = redis::Client::open(cfg.redis_url.as_str()).unwrap();
    let redis_config = ConnectionManagerConfig::new()
        .set_number_of_retries(1)
        .set_connection_timeout(Some(Duration::from_millis(300)))
        .set_response_timeout(Some(Duration::from_millis(300)));
    let cache = ConnectionManager::new_with_config(client, redis_config).await.unwrap();
    let (live, _) = broadcast::channel::<Comment>(100);
    let handle = PrometheusBuilder::new().set_buckets(&[0.005, 0.01, 0.05, 0.1, 0.5, 1.0]).unwrap().install_recorder().unwrap();
    match cfg.storage.as_str() {
        "memory" => serve(app(repo, MemStorage::default(), tx, cache, live, handle), &addr).await,
        _ => serve(app(repo, LocalStorage::new("uploads"), tx, cache, live, handle), &addr).await,
    }
}

async fn serve(app: Router, addr: &str) {
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    tracing::info!(%addr, "listening");
    axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>())
        .with_graceful_shutdown(shutdown_signal())
        .await
        .unwrap();
}

fn posts_common<R: Repo>() -> Router<R> {
    Router::new()
        .route("/posts", get(handlers::list_posts::<R>).post(handlers::create_post::<R>))
        .route(
            "/posts/{id}",
            put(handlers::update_post::<R>).delete(handlers::delete_post::<R>),
        )
        .route("/posts/{id}/comments", get(handlers::list_comments::<R>).post(handlers::add_comment::<R>))
        .route("/search", get(handlers::search_posts::<R>))
}

fn posts_v1<R: Repo>() -> Router<R> {
    posts_common()
        .route("/posts/{id}", get(handlers::get_post::<R>))
        .layer(SetResponseHeaderLayer::overriding(HeaderName::from_static("deprecation"), HeaderValue::from_static("true")))
        .layer(SetResponseHeaderLayer::overriding(
            HeaderName::from_static("sunset"),
            HeaderValue::from_static("Wed, 30 Jun 2027 00:00:00 GMT"),
        ))
        .layer(SetResponseHeaderLayer::overriding(header::LINK, HeaderValue::from_static(r#"</api/v2>; rel="successor-version""#)))
}

fn posts_v2<R: Repo>() -> Router<R> {
    posts_common().route("/posts/{id}", get(handlers::get_post_v2::<R>))
}

async fn shutdown_signal() {
    let ctrl_c = async {
        tokio::signal::ctrl_c().await.expect("Ctrl+C handler");
    };
    let terminate = async {
        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            .expect("SIGTERM handler")
            .recv()
            .await;
    };
    tokio::select! {
        _ = ctrl_c => {},
        _ = terminate => {},
    }
    tracing::info!("shutting down");
}

async fn track(req: Request, next: Next) -> Response {
    let start = std::time::Instant::now();
    let method = req.method().to_string();
    let path = req
        .extensions()
        .get::<MatchedPath>()
        .map(|p| p.as_str().to_string())
        .unwrap_or_else(|| "unmatched".to_string());
    let response = next.run(req).await;
    let status = response.status().as_u16().to_string();
    let labels = [("method", method), ("path", path), ("status", status)];
    metrics::counter!("http_requests_total", &labels).increment(1);
    metrics::histogram!("http_request_duration_seconds", &labels).record(start.elapsed().as_secs_f64());
    response
}

src/repo.rs

rust

use std::future::Future;

use sqlx::{QueryBuilder, Sqlite, SqlitePool};

use crate::models::{slugify, Comment, Hit, Image, ListFilter, NewComment, NewPost, Post, PostWithTags, Sort, User, UserWithHash};

pub trait Repo: Clone + Send + Sync + 'static {
    fn list(&self, f: ListFilter) -> impl Future<Output = sqlx::Result<(Vec<Post>, i64)>> + Send;
    fn get(&self, id: i64) -> impl Future<Output = sqlx::Result<Option<Post>>> + Send;
    fn create(&self, author_id: i64, new: NewPost) -> impl Future<Output = sqlx::Result<PostWithTags>> + Send;
    fn update(&self, id: i64, new: NewPost) -> impl Future<Output = sqlx::Result<Option<Post>>> + Send;
    fn delete(&self, id: i64) -> impl Future<Output = sqlx::Result<bool>> + Send;
    fn restore(&self, id: i64) -> impl Future<Output = sqlx::Result<bool>> + Send;
    fn list_comments(&self, post_id: i64) -> impl Future<Output = sqlx::Result<Vec<Comment>>> + Send;
    fn add_comment(&self, post_id: i64, new: NewComment) -> impl Future<Output = sqlx::Result<Comment>> + Send;
    fn search(&self, q: &str) -> impl Future<Output = sqlx::Result<Vec<Hit>>> + Send;
    fn create_user(&self, email: &str, password_hash: &str) -> impl Future<Output = sqlx::Result<User>> + Send;
    fn find_user_by_email(&self, email: &str) -> impl Future<Output = sqlx::Result<Option<UserWithHash>>> + Send;
    fn get_user(&self, id: i64) -> impl Future<Output = sqlx::Result<Option<User>>> + Send;
    fn store_refresh(&self, hash: &str, user_id: i64, expires_at: i64) -> impl Future<Output = sqlx::Result<()>> + Send;
    fn take_refresh(&self, hash: &str, now: i64) -> impl Future<Output = sqlx::Result<Option<i64>>> + Send;
    fn delete_refresh(&self, hash: &str) -> impl Future<Output = sqlx::Result<()>> + Send;
    fn delete_expired_refresh(&self, now: i64) -> impl Future<Output = sqlx::Result<u64>> + Send;
    fn list_users(&self) -> impl Future<Output = sqlx::Result<Vec<User>>> + Send;
    fn ping(&self) -> impl Future<Output = sqlx::Result<()>> + Send;
    fn add_image(&self, post_id: i64, url: &str) -> impl Future<Output = sqlx::Result<Image>> + Send;
}

#[derive(Clone)]
pub struct SqliteRepo {
    pub pool: SqlitePool,
}

impl Repo for SqliteRepo {
    async fn list(&self, f: ListFilter) -> sqlx::Result<(Vec<Post>, i64)> {
        let mut q = QueryBuilder::<Sqlite>::new("SELECT p.id, p.title, p.body, p.slug, p.created_at, p.author_id FROM posts p WHERE p.deleted_at IS NULL");
        if let Some(tag) = &f.tag {
            q.push(" AND p.id IN (SELECT pt.post_id FROM post_tags pt JOIN tags t ON t.id = pt.tag_id WHERE t.name = ")
                .push_bind(tag.clone())
                .push(")");
        }
        q.push(match f.sort {
            Sort::Newest => " ORDER BY p.id DESC",
            Sort::Oldest => " ORDER BY p.id ASC",
            Sort::Title => " ORDER BY p.title ASC",
        });
        q.push(" LIMIT ").push_bind(f.per_page as i64);
        q.push(" OFFSET ").push_bind(((f.page - 1) * f.per_page) as i64);
        let posts = q.build_query_as::<Post>().fetch_all(&self.pool).await?;

        let mut c = QueryBuilder::<Sqlite>::new("SELECT COUNT(*) FROM posts p WHERE p.deleted_at IS NULL");
        if let Some(tag) = &f.tag {
            c.push(" AND p.id IN (SELECT pt.post_id FROM post_tags pt JOIN tags t ON t.id = pt.tag_id WHERE t.name = ")
                .push_bind(tag.clone())
                .push(")");
        }
        let total: i64 = c.build_query_scalar().fetch_one(&self.pool).await?;
        Ok((posts, total))
    }

    async fn search(&self, q: &str) -> sqlx::Result<Vec<Hit>> {
        let q = fts_query(q);
        sqlx::query_as!(
            Hit,
            r#"SELECT p.id as "id!", p.title, snippet(posts_fts, 1, '[', ']', '...', 8) as "snippet!: String"
               FROM posts_fts JOIN posts p ON p.id = posts_fts.rowid
               WHERE posts_fts MATCH ? AND p.deleted_at IS NULL
               ORDER BY bm25(posts_fts) LIMIT 20"#,
            q
        )
        .fetch_all(&self.pool)
        .await
    }

    async fn get(&self, id: i64) -> sqlx::Result<Option<Post>> {
        sqlx::query_as!(Post, "SELECT id, title, body, slug, created_at, author_id FROM posts WHERE id = ? AND deleted_at IS NULL", id)
            .fetch_optional(&self.pool)
            .await
    }

    async fn create(&self, author_id: i64, new: NewPost) -> sqlx::Result<PostWithTags> {
        let slug = slugify(&new.title);
        let mut tx = self.pool.begin().await?;
        let post = sqlx::query_as!(
            Post,
            r#"INSERT INTO posts (title, body, slug, author_id) VALUES (?, ?, ?, ?) RETURNING id, title, body, slug, created_at, author_id"#,
            new.title,
            new.body,
            slug,
            author_id
        )
        .fetch_one(&mut *tx)
        .await?;
        for tag in &new.tags {
            sqlx::query!("INSERT INTO tags (name) VALUES (?) ON CONFLICT (name) DO NOTHING", tag)
                .execute(&mut *tx)
                .await?;
            sqlx::query!(
                "INSERT INTO post_tags (post_id, tag_id) SELECT ?, id FROM tags WHERE name = ?",
                post.id,
                tag
            )
            .execute(&mut *tx)
            .await?;
        }
        tx.commit().await?;
        Ok(PostWithTags { post, tags: new.tags })
    }

    async fn update(&self, id: i64, new: NewPost) -> sqlx::Result<Option<Post>> {
        sqlx::query_as!(
            Post,
            r#"UPDATE posts SET title = ?, body = ? WHERE id = ? AND deleted_at IS NULL RETURNING id, title, body, slug, created_at, author_id"#,
            new.title,
            new.body,
            id
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn delete(&self, id: i64) -> sqlx::Result<bool> {
        let result = sqlx::query!("UPDATE posts SET deleted_at = CURRENT_TIMESTAMP WHERE id = ? AND deleted_at IS NULL", id)
            .execute(&self.pool)
            .await?;
        Ok(result.rows_affected() > 0)
    }

    async fn restore(&self, id: i64) -> sqlx::Result<bool> {
        let result = sqlx::query!("UPDATE posts SET deleted_at = NULL WHERE id = ? AND deleted_at IS NOT NULL", id)
            .execute(&self.pool)
            .await?;
        Ok(result.rows_affected() > 0)
    }

    async fn create_user(&self, email: &str, password_hash: &str) -> sqlx::Result<User> {
        sqlx::query_as!(
            User,
            r#"INSERT INTO users (email, password_hash) VALUES (?, ?) RETURNING id, email, role, created_at"#,
            email,
            password_hash
        )
        .fetch_one(&self.pool)
        .await
    }

    async fn find_user_by_email(&self, email: &str) -> sqlx::Result<Option<UserWithHash>> {
        sqlx::query_as!(
            UserWithHash,
            "SELECT id, email, password_hash, role FROM users WHERE email = ?",
            email
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn get_user(&self, id: i64) -> sqlx::Result<Option<User>> {
        sqlx::query_as!(User, "SELECT id, email, role, created_at FROM users WHERE id = ?", id)
            .fetch_optional(&self.pool)
            .await
    }

    async fn store_refresh(&self, hash: &str, user_id: i64, expires_at: i64) -> sqlx::Result<()> {
        sqlx::query!(
            "INSERT INTO refresh_tokens (token_hash, user_id, expires_at) VALUES (?, ?, ?)",
            hash,
            user_id,
            expires_at
        )
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn take_refresh(&self, hash: &str, now: i64) -> sqlx::Result<Option<i64>> {
        sqlx::query_scalar!(
            r#"DELETE FROM refresh_tokens WHERE token_hash = ? AND expires_at > ? RETURNING user_id as "user_id!""#,
            hash,
            now
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn delete_refresh(&self, hash: &str) -> sqlx::Result<()> {
        sqlx::query!("DELETE FROM refresh_tokens WHERE token_hash = ?", hash).execute(&self.pool).await?;
        Ok(())
    }

    async fn delete_expired_refresh(&self, now: i64) -> sqlx::Result<u64> {
        let done = sqlx::query!("DELETE FROM refresh_tokens WHERE expires_at < ?", now).execute(&self.pool).await?;
        Ok(done.rows_affected())
    }

    async fn list_comments(&self, post_id: i64) -> sqlx::Result<Vec<Comment>> {
        sqlx::query_as!(
            Comment,
            "SELECT id, post_id, body, created_at FROM comments WHERE post_id = ? ORDER BY id",
            post_id
        )
        .fetch_all(&self.pool)
        .await
    }

    async fn add_comment(&self, post_id: i64, new: NewComment) -> sqlx::Result<Comment> {
        sqlx::query_as!(
            Comment,
            r#"INSERT INTO comments (post_id, body) VALUES (?, ?) RETURNING id as "id!", post_id, body, created_at"#,
            post_id,
            new.body
        )
        .fetch_one(&self.pool)
        .await
    }

    async fn list_users(&self) -> sqlx::Result<Vec<User>> {
        sqlx::query_as!(User, "SELECT id, email, role, created_at FROM users ORDER BY id")
            .fetch_all(&self.pool)
            .await
    }

    async fn ping(&self) -> sqlx::Result<()> {
        sqlx::query("SELECT 1").execute(&self.pool).await?;
        Ok(())
    }

    async fn add_image(&self, post_id: i64, url: &str) -> sqlx::Result<Image> {
        sqlx::query_as!(
            Image,
            r#"INSERT INTO images (post_id, url) VALUES (?, ?) RETURNING id as "id!", post_id, url"#,
            post_id,
            url
        )
        .fetch_one(&self.pool)
        .await
    }
}

fn fts_query(input: &str) -> String {
    input
        .split_whitespace()
        .map(|word| format!("\"{}\"", word.replace('"', "\"\"")))
        .collect::<Vec<_>>()
        .join(" ")
}

Recap

/health/live says the process runs; /health/ready checks the database and Redis and answers 503 if one is down
The metrics crate counts and times; metrics-exporter-prometheus renders the text for /metrics
middleware::from_fn wraps every request: timer, next.run, then record
Label with MatchedPath (inside route_layer), never the raw URL, to keep the number of series small
Histograms give percentiles; add counters for business events too

Next episode: documentation that writes itself, OpenAPI and Swagger UI. Watch the series on YouTube.