rust 66 lines · 8 steps

Request coalescing in an Axum handler

A cache-and-Notify pattern lets concurrent requests for the same profile trigger just one upstream fetch.

Explained by highlit
Axum Advanced caching concurrency request-coalescing shared-state async
1use axum::{extract::{Path, State}, http::StatusCode, Json};
2use dashmap::DashMap;
3use serde::Serialize;
4use std::sync::Arc;
5use tokio::sync::Notify;
6 
7#[derive(Clone, Serialize)]
8pub struct Profile {
9 id: u64,
10 display_name: String,
11}
12 
13#[derive(Clone)]
14pub struct AppState {
15 cache: Arc<DashMap<u64, Profile>>,
16 inflight: Arc<DashMap<u64, Arc<Notify>>>,
17 upstream: reqwest::Client,
18}
19 
20pub async fn get_profile(
21 State(state): State<AppState>,
22 Path(id): Path<u64>,
23) -> Result<Json<Profile>, StatusCode> {
24 loop {
25 if let Some(hit) = state.cache.get(&id) {
26 return Ok(Json(hit.clone()));
27 }
28 
29 let notify = match state.inflight.entry(id) {
30 dashmap::mapref::entry::Entry::Occupied(e) => {
31 let notify = e.get().clone();
32 drop(e);
33 notify.notified().await;
34 continue;
35 }
36 dashmap::mapref::entry::Entry::Vacant(e) => {
37 let notify = Arc::new(Notify::new());
38 e.insert(notify.clone());
39 notify
40 }
41 };
42 
43 let result = fetch_upstream(&state.upstream, id).await;
44 
45 if let Ok(profile) = &result {
46 state.cache.insert(id, profile.clone());
47 }
48 state.inflight.remove(&id);
49 notify.notify_waiters();
50 
51 return match result {
52 Ok(profile) => Ok(Json(profile)),
53 Err(_) => Err(StatusCode::BAD_GATEWAY),
54 };
55 }
56}
57 
58async fn fetch_upstream(client: &reqwest::Client, id: u64) -> Result<Profile, reqwest::Error> {
59 client
60 .get(format!("https://accounts.internal/profiles/{id}"))
61 .send()
62 .await?
63 .error_for_status()?
64 .json::<Profile>()
65 .await
66}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Tracking in-flight work in a separate map lets duplicate requests wait instead of stampeding the upstream.
  2. 2A Notify handle shared through an Arc turns 'wake everyone waiting on this key' into a single call.
  3. 3Looping after being woken re-checks the cache, so waiters serve the freshly fetched result rather than refetching.

Related explainers

go 
package config
 
import (
	"fmt"

A thread-safe config singleton in Go

singleton concurrency environment-variables
Intermediate 7 steps
python 
from functools import wraps
import asyncio
 
from fastapi import APIRouter, FastAPI, Request

Per-route request timeouts in FastAPI

decorators async timeouts
Intermediate 6 steps
rust 
use std::net::Ipv4Addr;
use std::str::FromStr;
 
#[derive(Debug, Clone, Copy)]

Parsing and matching IPv4 CIDR ranges in Rust

bitwise-operations parsing error-handling
Intermediate 8 steps
rust 
use std::sync::OnceLock;
 
static CRC_TABLE: OnceLock<[u32; 256]> = OnceLock::new();
 

Table-driven CRC32 with lazy init in Rust

checksums lazy-initialization lookup-tables
Intermediate 8 steps
rust 
use axum::{
    body::Body,
    extract::Path,
    http::{header, HeaderMap, HeaderValue, StatusCode},

HTTP range requests for video streaming in Axum

http-range-requests streaming async-io
Advanced 8 steps
rust 
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
 

Running work with a timeout in Rust

concurrency channels timeout
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

Request coalescing in an Axum handler — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code