feat: Add uniflight crate for duplicate request coalescing

.gitignore

@@ -35,4 +35,5 @@ _manifest
 ARROW
 
 # Agent files
-.claude
+.claude
+CLAUDE.md

.spelling

@@ -1,3 +1,4 @@
+304
 →
 0.X.Y
 100k
@@ -82,6 +83,8 @@ C-SERDE
 C-SMART-PTR
 deallocate
 Debuggability
+Deduplicate
+deduplicating
 deduplication
 deque
 Deque
@@ -279,6 +282,8 @@ unconfigured
 uncontended
 unhandleable
 unicode
+uniflight
+uniflight's
 Uninit
 unordered
 unredacted

CHANGELOG.md

@@ -20,3 +20,4 @@ Please see each crate's change log below:
 - [`thread_aware_macros`](./crates/thread_aware_macros/CHANGELOG.md)
 - [`thread_aware_macros_impl`](./crates/thread_aware_macros_impl/CHANGELOG.md)
 - [`tick`](./crates/tick/CHANGELOG.md)
+- [`uniflight`](./crates/uniflight/CHANGELOG.md)

Cargo.toml

@@ -43,13 +43,18 @@ thread_aware = { path = "crates/thread_aware", default-features = false, version
 thread_aware_macros = { path = "crates/thread_aware_macros", default-features = false, version = "0.6.1" }
 thread_aware_macros_impl = { path = "crates/thread_aware_macros_impl", default-features = false, version = "0.6.1" }
 tick = { path = "crates/tick", default-features = false, version = "0.1.2" }
+uniflight = { path = "crates/uniflight", default-features = false, version = "0.1.0" }
 
 # external dependencies
+ahash = { version = "0.8", default-features = false }
 alloc_tracker = { version = "0.5.9", default-features = false }
+anyhow = { version = "1.0.100", default-features = false }
+async-once-cell = { version = "0.5", default-features = false }
 bytes = { version = "1.11.0", default-features = false }
 chrono = { version = "0.4.40", default-features = false }
 chrono-tz = { version = "0.10.4", default-features = false }
 criterion = { version = "0.8.1", default-features = false }
+dashmap = { version = "6.1", default-features = false }
 derive_more = { version = "2.0.1", default-features = false }
 duct = { version = "1.1.1", default-features = false }
 dynosaur = { version = "0.3.0", default-features = false }
@@ -73,6 +78,7 @@ once_cell = { version = "1.21.3", default-features = false }
 opentelemetry = { version = "0.31.0", default-features = false }
 opentelemetry-stdout = { version = "0.31.0", default-features = false }
 opentelemetry_sdk = { version = "0.31.0", default-features = false }
+parking_lot = { version = "0.12.5", default-features = false }
 pin-project-lite = { version = "0.2.13", default-features = false }
 pretty_assertions = { version = "1.4.1", default-features = false }
 prettyplease = { version = "0.2.37", default-features = false }
@@ -101,6 +107,7 @@ trait-variant = { version = "0.1.2", default-features = false }
 trybuild = { version = "1.0.114", default-features = false }
 typeid = { version = "1.0.3", default-features = false }
 windows-sys = { version = "0.61.2", default-features = false }
+xutex = { version = "0.2.0", default-features = false }
 xxhash-rust = { version = "0.8.15", default-features = false }
 
 [workspace.lints.rust]

README.md

@@ -36,6 +36,7 @@ These are the primary crates built out of this repo:
 - [`seatbelt`](./crates/seatbelt/README.md) - Resilience and recovery mechanisms for fallible operations.
 - [`thread_aware`](./crates/thread_aware/README.md) - Facilities to support thread-isolated state.
 - [`tick`](./crates/tick/README.md) - Provides primitives to interact with and manipulate machine time.
+- [`uniflight`](./crates/uniflight/README.md) - Coalesces duplicate async tasks into a single execution.
 
 ## About this Repo
 

crates/uniflight/CHANGELOG.md

@@ -0,0 +1,8 @@
+# Changelog
+
+## [0.1.0] - 2025-12-10
+
+- 🧩 Miscellaneous
+
+  - Initial commit of uniflight
+

crates/uniflight/Cargo.toml

@@ -0,0 +1,53 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+[package]
+name = "uniflight"
+description = "Coalesces duplicate async tasks into a single execution."
+version = "0.1.0"
+readme = "README.md"
+keywords = ["oxidizer", "coalescing", "stempede", "singleflight", "deduplication"]
+categories = ["concurrency"]
+
+edition.workspace = true
+rust-version.workspace = true
+authors.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+
+[package.metadata.cargo_check_external_types]
+allowed_external_types = [
+    "thread_aware::cell::builtin::PerProcess",
+    "thread_aware::cell::storage::Strategy",
+    "thread_aware::core::ThreadAware",
+]
+
+[dependencies]
+ahash = { workspace = true, default-features = false, features = ["std"] }
+async-once-cell.workspace = true
+dashmap.workspace = true
+futures-util = { workspace = true, default-features = false, features = ["std", "alloc"] }
+thread_aware.workspace = true
+
+[dev-dependencies]
+criterion = { workspace = true, features = ["async_tokio"] }
+futures-util = { workspace = true, features = ["alloc", "std"] }
+mutants.workspace = true
+tick = { workspace = true, features = ["tokio"] }
+tokio = { workspace = true, features = [
+    "macros",
+    "rt",
+    "time",
+    "rt-multi-thread",
+] }
+
+[lints]
+workspace = true
+
+[[bench]]
+name = "performance"
+harness = false
+
+[[example]]
+name = "cache_population"

crates/uniflight/README.md

@@ -0,0 +1,145 @@
+<div align="center">
+ <img src="./logo.png" alt="Uniflight Logo" width="96">
+
+# Uniflight
+
+[![crate.io](https://img.shields.io/crates/v/uniflight.svg)](https://crates.io/crates/uniflight)
+[![docs.rs](https://docs.rs/uniflight/badge.svg)](https://docs.rs/uniflight)
+[![MSRV](https://img.shields.io/crates/msrv/uniflight)](https://crates.io/crates/uniflight)
+[![CI](https://github.com/microsoft/oxidizer/actions/workflows/main.yml/badge.svg?event=push)](https://github.com/microsoft/oxidizer/actions/workflows/main.yml)
+[![Coverage](https://codecov.io/gh/microsoft/oxidizer/graph/badge.svg?token=FCUG0EL5TI)](https://codecov.io/gh/microsoft/oxidizer)
+[![License](https://img.shields.io/badge/license-MIT-blue.svg)](../../LICENSE)
+<a href="../.."><img src="../../logo.svg" alt="This crate was developed as part of the Oxidizer project" width="20"></a>
+
+</div>
+
+Coalesces duplicate async tasks into a single execution.
+
+This crate provides [`Merger`][__link0], a mechanism for deduplicating concurrent async operations.
+When multiple tasks request the same work (identified by a key), only the first task (the
+“leader”) performs the actual work while subsequent tasks (the “followers”) wait and receive
+a clone of the result.
+
+## When to Use
+
+Use `Merger` when you have expensive or rate-limited operations that may be requested
+concurrently with the same parameters:
+
+* **Cache population**: Prevent thundering herd when a cache entry expires
+* **API calls**: Deduplicate concurrent requests to the same endpoint
+* **Database queries**: Coalesce identical queries issued simultaneously
+* **File I/O**: Avoid reading the same file multiple times concurrently
+
+## Example
+
+```rust
+use uniflight::Merger;
+
+let group: Merger<String, String> = Merger::new();
+
+// Multiple concurrent calls with the same key will share a single execution.
+// Note: you can pass &str directly when the key type is String.
+let result = group.execute("user:123", || async {
+    // This expensive operation runs only once, even if called concurrently
+    "expensive_result".to_string()
+}).await.expect("leader should not panic");
+```
+
+## Flexible Key Types
+
+The [`Merger::execute`][__link1] method accepts keys using [`Borrow`][__link2] semantics, allowing you to pass
+borrowed forms of the key type. For example, with `Merger<String, T>`, you can pass `&str`
+directly without allocating:
+
+```rust
+let merger: Merger<String, i32> = Merger::new();
+
+// Pass &str directly - no need to call .to_string()
+let result = merger.execute("my-key", || async { 42 }).await;
+assert_eq!(result, Ok(42));
+```
+
+## Thread-Aware Scoping
+
+`Merger` supports thread-aware scoping via a [`Strategy`][__link3]
+type parameter. This controls how the internal state is partitioned across threads/NUMA nodes:
+
+* [`PerProcess`][__link4] (default): Single global state, maximum deduplication
+* [`PerNuma`][__link5]: Separate state per NUMA node, NUMA-local memory access
+* [`PerCore`][__link6]: Separate state per core, no deduplication (useful for already-partitioned work)
+
+```rust
+use uniflight::Merger;
+use thread_aware::PerNuma;
+
+// NUMA-aware merger - each NUMA node gets its own deduplication scope
+let merger: Merger<String, String, PerNuma> = Merger::new_per_numa();
+```
+
+## Cancellation and Panic Handling
+
+`Merger` handles task cancellation and panics explicitly:
+
+* If the leader task is cancelled or dropped, a follower becomes the new leader
+* If the leader task panics, followers receive [`LeaderPanicked`][__link7] error with the panic message
+* Followers that join before the leader completes receive the value the leader returns
+
+When a panic occurs, followers are notified via the error type rather than silently
+retrying. The panic message is captured and available via [`LeaderPanicked::message`][__link8]:
+
+```rust
+let merger: Merger<String, String> = Merger::new();
+match merger.execute("key", || async { "result".to_string() }).await {
+    Ok(value) => println!("got {value}"),
+    Err(err) => {
+        println!("leader panicked: {}", err.message());
+        // Decide whether to retry
+    }
+}
+```
+
+## Memory Management
+
+Completed entries are automatically removed from the internal map when the last caller
+finishes. This ensures no stale entries accumulate over time.
+
+## Type Requirements
+
+The value type `T` must implement [`Clone`][__link9] because followers receive a clone of the
+leader’s result. The key type `K` must implement [`Hash`][__link10] and [`Eq`][__link11].
+
+## Thread Safety
+
+[`Merger`][__link12] is `Send` and `Sync`, and can be shared across threads. The returned futures
+are `Send` when the closure, future, key, and value types are `Send`.
+
+## Performance
+
+Run benchmarks with `cargo bench -p uniflight`. The suite covers:
+
+* `single_call`: Baseline latency with no contention
+* `high_contention_100`: 100 concurrent tasks on the same key
+* `distributed_10x10`: 10 keys with 10 tasks each
+
+Use `--save-baseline` and `--baseline` flags to track regressions over time.
+
+
+<hr/>
+<sub>
+This crate was developed as part of <a href="../..">The Oxidizer Project</a>. Browse this crate's <a href="https://github.com/microsoft/oxidizer/tree/main/crates/uniflight">source code</a>.
+</sub>
+
+ [__cargo_doc2readme_dependencies_info]: ggGkYW0CYXSEGy4k8ldDFPOhG2VNeXtD5nnKG6EPY6OfW5wBG8g18NOFNdxpYXKEGxgwNFq9VUtfG5xaBNm6U4VGG97W2YkyKkPjG4KVgSbTgdOrYWSCgmx0aHJlYWRfYXdhcmVlMC42LjGCaXVuaWZsaWdodGUwLjEuMA
+ [__link0]: https://docs.rs/uniflight/0.1.0/uniflight/struct.Merger.html
+ [__link1]: https://docs.rs/uniflight/0.1.0/uniflight/?search=Merger::execute
+ [__link10]: https://doc.rust-lang.org/stable/std/?search=hash::Hash
+ [__link11]: https://doc.rust-lang.org/stable/std/cmp/trait.Eq.html
+ [__link12]: https://docs.rs/uniflight/0.1.0/uniflight/struct.Merger.html
+ [__link2]: https://doc.rust-lang.org/stable/std/?search=borrow::Borrow
+ [__link3]: https://docs.rs/thread_aware/0.6.1/thread_aware/?search=storage::Strategy
+ [__link4]: https://docs.rs/thread_aware/0.6.1/thread_aware/?search=PerProcess
+ [__link5]: https://docs.rs/thread_aware/0.6.1/thread_aware/?search=PerNuma
+ [__link6]: https://docs.rs/thread_aware/0.6.1/thread_aware/?search=PerCore
+ [__link7]: https://docs.rs/uniflight/0.1.0/uniflight/struct.LeaderPanicked.html
+ [__link8]: https://docs.rs/uniflight/0.1.0/uniflight/?search=LeaderPanicked::message
+ [__link9]: https://doc.rust-lang.org/stable/std/clone/trait.Clone.html