rs-singleflight

rs-singleflight coalesces overlapping asynchronous work by key. For each active key, one background leader runs the operation while duplicate callers wait for the same result.

Results are shared only while an operation is active; this crate is not a cache.

Quick start

use std::sync::{
    Arc,
    atomic::{AtomicUsize, Ordering},
};

use rs_singleflight::{Group, Outcome};
use tokio::{sync::Notify, task::yield_now};

#[tokio::main]
async fn main() {
    let calls = Arc::new(AtomicUsize::new(0));
    let release = Arc::new(Notify::new());
    let group = Group::new({
        let calls = Arc::clone(&calls);
        let release = Arc::clone(&release);
        move |key: String| {
            let calls = Arc::clone(&calls);
            let release = Arc::clone(&release);
            async move {
                calls.fetch_add(1, Ordering::Relaxed);
                release.notified().await;
                Ok::<String, ()>(format!("value for {key}"))
            }
        }
    });

    let mut tasks = Vec::new();
    for _ in 0..8 {
        let group = group.clone();
        tasks.push(tokio::spawn(async move {
            group.run("resource".to_owned()).await
        }));
    }

    // The leader remains blocked until all seven duplicates have joined.
    while group.duplicate_count("resource") != Some(7) {
        yield_now().await;
    }
    release.notify_one();

    for task in tasks {
        match task.await.unwrap().unwrap().as_ref() {
            Outcome::Complete { result, shared } => {
                assert_eq!(result.as_ref().unwrap(), "value for resource");
                assert!(*shared);
            }
            Outcome::Canceled { reason } => panic!("flight canceled: {reason:?}"),
        }
    }

    assert_eq!(calls.load(Ordering::Relaxed), 1);
    assert_eq!(group.in_flight(), 0);
}

Group::run uses tokio::spawn, so it must be called from a Tokio runtime and its future, key, result, error, and hash builder must satisfy the documented Send/'static bounds.

Cancellation

The operation is detached from the caller that creates the flight. Timing out or aborting that caller does not cancel the operation or other waiters. The background computation continues until one of these events occurs:

• it returns Ok or Err, producing Outcome::Complete;
• it panics while unwinding is enabled, producing Outcome::Canceled { reason: CancelReason::Panicked };
• its Tokio task is aborted, including during runtime shutdown, producing CancelReason::LeaderDropped.

Dropping every Group::run future does not itself stop the background task.

HTTP request coalescing example

examples/request_coalescing.rs contains a runnable Axum service that coalesces concurrent user lookups:

cargo run --example request_coalescing

From another shell, send a burst of requests for the same user:

for request in {1..20}; do
  curl -s http://127.0.0.1:3000/users/42 &
done
wait

All overlapping responses contain the same upstream_request value, and the server logs one upstream request followed by responses with shared=true. Requests for different user IDs run independently. A later request after the flight completes starts another upstream request because singleflight does not cache results.

Low-level coordination

Use Coordinator when the caller needs to choose where the leader runs or wants to complete a flight manually. A Coordinator does not own an operation.

use rs_singleflight::{Coordinator, Entry, Outcome};

#[tokio::main]
async fn main() {
    let coordinator = Coordinator::<&'static str, usize, ()>::new();
    let leader = match coordinator.entry("key") {
        Entry::Leader(leader) => leader,
        Entry::Subscriber(_) => unreachable!(),
    };
    let subscriber = match coordinator.entry("key") {
        Entry::Subscriber(subscriber) => subscriber,
        Entry::Leader(_) => unreachable!(),
    };

    let task = tokio::spawn(async move {
        leader.complete(Ok(42));
    });

    assert!(matches!(
        subscriber.recv().await.unwrap().as_ref(),
        Outcome::Complete {
            result: Ok(42),
            shared: true
        }
    ));
    task.await.unwrap();
}

Dropping a manually managed Leader publishes a canceled outcome so its subscribers cannot hang.

Semantics

• Only calls whose lifetimes overlap are coalesced.
• Successful values and application errors are shared identically.
• shared is true if any duplicate joined the flight, even if that subscriber was later dropped.
• duplicate_count is cumulative for the current flight, not the number of subscribers currently waiting.
• forget(key) removes the current registration without canceling it. Existing subscribers still receive the old result, while the next call starts a new flight.
• A flight is removed atomically with publication. A later call either joins that flight and receives its outcome, or starts a new one.
• Hash-map mutex poisoning is recovered. A panic from user-provided Hash or Eq code still propagates to that caller.

Compatibility

The minimum supported Rust version is 1.85. The crate uses the Rust 2024 edition and Tokio 1.x.

License

BSD-3-Clause.