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

rust 
use serde::Deserialize;
 
#[derive(Debug, Deserialize)]
#[serde(untagged)]

Parsing flexible JSON shapes with serde

deserialization enums json
Intermediate 6 steps
python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
rust 
use std::cmp::{Ordering, Reverse};
use std::collections::BinaryHeap;
use std::fs::File;
use std::io::{self, BufRead, BufReader, BufWriter, Lines, Write};

K-way merge of sorted logs in Rust

binary-heap k-way-merge streaming-io
Intermediate 8 steps
rust 
use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel
Advanced 7 steps
rust 
use std::f64::consts::PI;
 
#[derive(Debug, Clone, Copy)]
pub struct LatLng {

Building geographic bounding boxes in Rust

geospatial value-types option
Intermediate 7 steps
javascript 
import { useState, useEffect, useCallback, useRef } from 'react';
 
const cache = new Map();
const inflight = new Map();

Building a stale-while-revalidate hook in React

caching request-deduplication custom-hooks
Advanced 10 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