feat(bytesbuf)!: make MemoryShared thread-aware (#539)

## Summary

Makes the `bytesbuf` `MemoryShared` abstraction thread-aware, closing a
design gap where our thread-mobile (`Send`) memory providers were
expected to be thread-aware but the abstraction did not require or
express it. A provider now owns the decision of how and when to be
thread-aware, and any adapter around it simply forwards relocation
rather than imposing its own strategy.

## Changes

- `MemoryShared` now has `ThreadAware` as a supertrait, so the
abstraction requires and expresses thread-awareness. This makes it
possible to relocate erased/abstract memory providers correctly across
threads.
- `MemoryShared` gains an object-safe `clone_boxed` method,
blanket-provided for any `Clone` provider, so a type-erased provider can
be duplicated without knowing its concrete type. Every `MemoryShared`
provider is therefore required to be `Clone`.
- Redesigned `CallbackMemory` into a single `CallbackMemory<D>` provider
that pairs thread-aware `data: D` with a bare `reserve_fn: fn(&D, usize)
-> BytesBuf`. Because the reserve function is a bare function pointer it
cannot capture anything, so all reservation state must live in `data`,
which is `ThreadAware` and is relocated together with the provider.
`CallbackMemory::new(data, reserve_fn)` constructs one; `data` is
typically the wrapped inner memory provider, but can be any thread-aware
value (e.g. a tuple pairing the provider with configuration).
- Reworked `OpaqueMemory` to own a `Box<dyn MemoryShared>`, forwarding
both cloning (via `clone_boxed`) and `ThreadAware` relocation to the
wrapped provider. It imposes no thread-awareness strategy of its own,
leaving that decision entirely to the wrapped provider (e.g.
`GlobalPool`, which already manages per-core state internally), and each
clone is an independent provider.
- Gave the test providers `TransparentMemory` and `FixedBlockMemory`
no-op `ThreadAware` impls, since they hold no thread-affine mutable
state.
- Downstream: `http_extensions` drops its now-unnecessary `Unaware`
wrapper around `OpaqueMemory`; `bytesbuf_io` test providers migrate to
the new `CallbackMemory` constructor.

## Breaking changes

- `MemoryShared` gained `ThreadAware` as a supertrait and a new
`clone_boxed` method, so implementors must now also be `ThreadAware`
(and `Clone`, which the blanket impl requires).
- `CallbackMemory`'s constructor changed from taking a capturing
`Fn(usize) -> BytesBuf` closure to taking thread-aware `data` plus a
bare `fn(&D, usize) -> BytesBuf`. State that a closure would previously
have captured must now be passed as `data`.

## Validation

- `just format`, `just readme`, and `just spellcheck` all clean.
- Tests pass for `bytesbuf`, `bytesbuf_io`, and `http_extensions` (unit
+ doctests), including new tests covering
`CallbackMemory`/`OpaqueMemory` relocation forwarding, non-`Debug`
callback data, and the no-op `relocate` on the test providers.
- Clippy `-D warnings` clean; all remaining dependents build.
- Mutation testing of the changed logic (`cargo mutants --in-diff`)
leaves no surviving mutants.

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: 3 people

.spelling

@@ -631,3 +631,4 @@ ASTs
 DAGs
 representable
 rc
+destructure

crates/bytesbuf/Cargo.toml

@@ -46,7 +46,7 @@ new_zealand = { workspace = true }
 nm = { workspace = true }
 num-traits = { workspace = true }
 smallvec = { workspace = true, features = ["const_new", "union"] }
-thread_aware = { workspace = true }
+thread_aware = { workspace = true, features = ["derive"] }
 
 [dev-dependencies]
 alloc_tracker = { workspace = true }

crates/bytesbuf/README.md

@@ -311,13 +311,12 @@ const UDP_CONNECTION_OPTIMAL_MEMORY_CONFIGURATION: MemoryConfiguration = MemoryC
 
 impl HasMemory for UdpConnection {
     fn memory(&self) -> impl MemoryShared {
-        CallbackMemory::new({
-            // Cloning is cheap, as it is a service that shares resources between clones.
-            let io_context = self.io_context.clone();
+        // The I/O memory provider carries the thread-affine state, which is relocated when the
+        // returned provider is moved between threads via a thread-aware runtime mechanism.
+        let io_memory = self.io_context.io_memory();
 
-            move |min_len| {
-                io_context.reserve_io_memory(min_len, UDP_CONNECTION_OPTIMAL_MEMORY_CONFIGURATION)
-            }
+        CallbackMemory::new(io_memory, |io_memory, min_len| {
+            io_memory.reserve_with_config(min_len, &UDP_CONNECTION_OPTIMAL_MEMORY_CONFIGURATION)
         })
     }
 }
@@ -471,7 +470,7 @@ See the `mem::testing` module for details (requires `test-util` Cargo feature).
 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/bytesbuf">source code</a>.
 </sub>
 
- [__cargo_doc2readme_dependencies_info]: ggGmYW0CYXZlMC43LjJhdIQbLiTyV0MU86EbZU15e0PmecoboQ9jo59bnAEbyDXw04U13GlhYvRhcoQb4gvD4zw9iycbirtUutkkzqcbSlRn3SgH6bsbjAbxNhZA3wBhZIGCaGJ5dGVzYnVmZTAuNS42
+ [__cargo_doc2readme_dependencies_info]: ggGmYW0CYXZlMC43LjJhdIQbLiTyV0MU86EbZU15e0PmecoboQ9jo59bnAEbyDXw04U13GlhYvRhcoQbJ1n5emfG9rYb9IJBtfUIpXYbaB5j18j1RxEbDiXhMXybg_xhZIGCaGJ5dGVzYnVmZTAuNS42
  [__link0]: https://docs.rs/bytesbuf/0.5.6/bytesbuf/?search=BytesBuf
  [__link1]: https://docs.rs/bytesbuf/0.5.6/bytesbuf/?search=BytesView
  [__link10]: https://docs.rs/bytesbuf/0.5.6/bytesbuf/?search=BytesView

crates/bytesbuf/examples/bb_has_memory_optimizing.rs

@@ -7,6 +7,7 @@
 
 use bytesbuf::mem::{CallbackMemory, HasMemory, Memory, MemoryShared};
 use bytesbuf::{BytesBuf, BytesView};
+use thread_aware::ThreadAware;
 
 fn main() {
     // In a real application, the I/O context would be provided by the framework.
@@ -66,11 +67,14 @@ const UDP_CONNECTION_OPTIMAL_MEMORY_CONFIGURATION: MemoryConfiguration = MemoryC
 
 impl HasMemory for UdpConnection {
     fn memory(&self) -> impl MemoryShared {
-        CallbackMemory::new({
-            // Cloning is cheap, as it is a service that shares resources between clones.
-            let io_context = self.io_context.clone();
-
-            move |min_len| io_context.reserve_io_memory(min_len, UDP_CONNECTION_OPTIMAL_MEMORY_CONFIGURATION)
+        // The I/O memory provider carries the thread-affine state, which is relocated when the
+        // returned provider is moved between threads via a thread-aware runtime mechanism.
+        let io_memory = self.io_context.io_memory();
+
+        CallbackMemory::new(io_memory, |io_memory, min_len| {
+            // Apply the connection-specific configuration when reserving from the (relocated)
+            // I/O memory provider.
+            io_memory.reserve_with_config(min_len, &UDP_CONNECTION_OPTIMAL_MEMORY_CONFIGURATION)
         })
     }
 }
@@ -88,14 +92,40 @@ impl IoContext {
         Self {}
     }
 
+    /// Returns the thread-affine I/O memory provider that reservations are drawn from.
     #[expect(clippy::unused_self, reason = "for example realism")]
-    pub(crate) fn reserve_io_memory(&self, min_len: usize, _memory_configuration: MemoryConfiguration) -> BytesBuf {
+    pub(crate) fn io_memory(&self) -> IoMemory {
+        IoMemory
+    }
+}
+
+/// The thread-affine I/O memory provider. In a real application this would carry per-thread I/O
+/// resources; here it is a thin wrapper for illustration.
+#[derive(Clone, Debug, ThreadAware)]
+struct IoMemory;
+
+impl IoMemory {
+    #[expect(clippy::unused_self, reason = "for example realism")]
+    fn reserve_with_config(&self, min_len: usize, _memory_configuration: &MemoryConfiguration) -> BytesBuf {
         // This is a wrong way to implement this! Only to make the example compile.
         let memory = bytesbuf::mem::testing::TransparentMemory::new();
         memory.reserve(min_len)
     }
 }
 
+impl Memory for IoMemory {
+    fn reserve(&self, min_bytes: usize) -> BytesBuf {
+        self.reserve_with_config(
+            min_bytes,
+            &MemoryConfiguration {
+                requires_page_alignment: false,
+                zero_memory_on_release: false,
+                requires_registered_memory: false,
+            },
+        )
+    }
+}
+
 #[expect(dead_code, reason = "just an example, fields unused")]
 struct MemoryConfiguration {
     requires_page_alignment: bool,

crates/bytesbuf/examples/bb_optimal_path.rs

@@ -17,6 +17,7 @@ use std::num::NonZero;
 
 use bytesbuf::mem::{BlockSize, CallbackMemory, GlobalPool, HasMemory, MemoryShared};
 use bytesbuf::{BytesBuf, BytesView};
+use thread_aware::ThreadAware;
 
 fn main() {
     // In real-world code, both of these would be provided by the application framework.
@@ -117,11 +118,12 @@ const CONNECTION_OPTIMAL_MEMORY_CONFIGURATION: MemoryConfiguration = MemoryConfi
 
 impl HasMemory for Connection {
     fn memory(&self) -> impl MemoryShared {
-        CallbackMemory::new({
-            // Cloning is cheap, as it is a service that shares resources between clones.
-            let io_context = self.io_context.clone();
+        // The I/O memory provider carries the thread-affine I/O resources, which are relocated when
+        // the returned provider is moved between threads via a thread-aware runtime mechanism.
+        let io_memory = self.io_context.io_memory();
 
-            move |min_len| io_context.reserve_io_memory(min_len, CONNECTION_OPTIMAL_MEMORY_CONFIGURATION)
+        CallbackMemory::new(io_memory, |io_memory, min_len| {
+            io_memory.reserve_with_config(min_len, CONNECTION_OPTIMAL_MEMORY_CONFIGURATION)
         })
     }
 }
@@ -139,8 +141,21 @@ impl IoContext {
         Self {}
     }
 
+    /// Returns the thread-affine I/O memory provider that reservations are drawn from.
     #[expect(clippy::unused_self, reason = "for example realism")]
-    pub(crate) fn reserve_io_memory(&self, min_len: usize, memory_configuration: MemoryConfiguration) -> BytesBuf {
+    pub(crate) fn io_memory(&self) -> IoMemory {
+        IoMemory
+    }
+}
+
+/// The thread-affine I/O memory provider. In a real application this would carry per-thread I/O
+/// resources; here it holds no state and just performs the allocation.
+#[derive(Clone, Debug, ThreadAware)]
+struct IoMemory;
+
+impl IoMemory {
+    #[expect(clippy::unused_self, reason = "for example realism")]
+    fn reserve_with_config(&self, min_len: usize, memory_configuration: MemoryConfiguration) -> BytesBuf {
         let min_len: BlockSize = min_len
             .try_into()
             .expect("this example is limited to max allocation size of BlockSize, just to keep it simple");
@@ -155,7 +170,13 @@ impl IoContext {
     }
 }
 
-#[derive(Debug)]
+impl bytesbuf::mem::Memory for IoMemory {
+    fn reserve(&self, min_bytes: usize) -> BytesBuf {
+        self.reserve_with_config(min_bytes, CONNECTION_OPTIMAL_MEMORY_CONFIGURATION)
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
 #[expect(dead_code, reason = "unused fields just for example realism")]
 struct MemoryConfiguration {
     requires_page_alignment: bool,

crates/bytesbuf/src/lib.rs

@@ -339,29 +339,35 @@
 //!
 //! impl HasMemory for UdpConnection {
 //!     fn memory(&self) -> impl MemoryShared {
-//!         CallbackMemory::new({
-//!             // Cloning is cheap, as it is a service that shares resources between clones.
-//!             let io_context = self.io_context.clone();
+//!         // The I/O memory provider carries the thread-affine state, which is relocated when the
+//!         // returned provider is moved between threads via a thread-aware runtime mechanism.
+//!         let io_memory = self.io_context.io_memory();
 //!
-//!             move |min_len| {
-//!                 io_context.reserve_io_memory(min_len, UDP_CONNECTION_OPTIMAL_MEMORY_CONFIGURATION)
-//!             }
+//!         CallbackMemory::new(io_memory, |io_memory, min_len| {
+//!             io_memory.reserve_with_config(min_len, &UDP_CONNECTION_OPTIMAL_MEMORY_CONFIGURATION)
 //!         })
 //!     }
 //! }
 //!
 //! # use bytesbuf::BytesBuf;
+//! # use bytesbuf::mem::Memory;
+//! # use thread_aware::ThreadAware;
 //! # #[derive(Clone, Debug)]
 //! # struct IoContext;
 //! # impl IoContext {
-//! #     pub fn reserve_io_memory(
-//! #         &self,
-//! #         min_len: usize,
-//! #         _memory_configuration: MemoryConfiguration,
-//! #     ) -> BytesBuf {
-//! #         todo!()
+//! #     pub fn io_memory(&self) -> IoMemory { IoMemory }
+//! # }
+//! # #[derive(Clone, Debug, ThreadAware)]
+//! # struct IoMemory;
+//! # impl IoMemory {
+//! #     fn reserve_with_config(&self, min_len: usize, _configuration: &MemoryConfiguration) -> BytesBuf {
+//! #         bytesbuf::mem::GlobalPool::new().reserve(min_len)
 //! #     }
 //! # }
+//! # impl Memory for IoMemory {
+//! #     fn reserve(&self, min_bytes: usize) -> BytesBuf { self.reserve_with_config(min_bytes, &UDP_CONNECTION_OPTIMAL_MEMORY_CONFIGURATION) }
+//! # }
+//! # #[derive(Clone, Copy)]
 //! # struct MemoryConfiguration { requires_page_alignment: bool, zero_memory_on_release: bool, requires_registered_memory: bool }
 //! ```
 //!