feat: use opaque memory in HTTP APIs

crates/bytesbuf/src/mem/opaque_memory.rs

@@ -3,6 +3,7 @@
 
 #[cfg(not(test))]
 use alloc::boxed::Box;
+use core::any::{Any, TypeId};
 
 use thread_aware::ThreadAware;
 
@@ -24,9 +25,21 @@ pub struct OpaqueMemory {
 
 impl OpaqueMemory {
     /// Creates a new instance of the adapter.
+    ///
+    /// # Panics
+    ///
+    /// Panics only if runtime type identification reports [`OpaqueMemory`] but the downcast of
+    /// the same value to [`OpaqueMemory`] fails, which would indicate a standard library defect.
     #[must_use]
-    pub fn new(inner: impl MemoryShared) -> Self {
-        Self { inner: Box::new(inner) }
+    pub fn new<M: MemoryShared>(inner: M) -> Self {
+        if TypeId::of::<M>() == TypeId::of::<Self>() {
+            let inner: Box<dyn Any> = Box::new(inner);
+            *inner
+                .downcast::<Self>()
+                .expect("the concrete type was verified as OpaqueMemory above")
+        } else {
+            Self { inner: Box::new(inner) }
+        }
     }
 
     /// Reserves at least `min_bytes` bytes of memory capacity.
@@ -91,6 +104,16 @@ mod tests {
         assert!(builder.capacity() >= 1024);
     }
 
+    #[test]
+    fn accepts_existing_opaque_memory() {
+        let memory = OpaqueMemory::new(GlobalPool::new());
+        let memory = OpaqueMemory::new(memory);
+
+        let builder = memory.reserve(1024);
+
+        assert!(builder.capacity() >= 1024);
+    }
+
     #[test]
     fn memory_trait() {
         let provider = GlobalPool::new();

crates/fetch/Cargo.toml

@@ -19,9 +19,9 @@ repository = "https://github.com/microsoft/oxidizer/tree/main/crates/fetch"
 
 [package.metadata.cargo_check_external_types]
 allowed_external_types = [
-    "bytesbuf::mem::global::GlobalPool",
     "bytesbuf::mem::has_memory::HasMemory",
     "bytesbuf::mem::memory::Memory",
+    "bytesbuf::mem::opaque_memory::OpaqueMemory",
     "data_privacy::redaction_engine::RedactionEngine",
     "fetch_options::*",
     "fetch_tls::*",

crates/fetch/examples/http_client_app.rs

@@ -6,7 +6,7 @@
 
 use std::sync::Arc;
 
-use bytesbuf::mem::GlobalPool;
+use bytesbuf::mem::{GlobalPool, OpaqueMemory};
 use fetch::HttpClient;
 use fetch::tls::TlsOptions;
 use ohno::ErrorExt;
@@ -18,7 +18,7 @@ use tick::Clock;
 #[fundle::bundle]
 struct App {
     clock: Clock,
-    global_pool: GlobalPool,
+    global_pool: OpaqueMemory,
     client: HttpClient,
 }
 
@@ -42,7 +42,7 @@ async fn main() -> Result<(), ohno::AppError> {
     // Initialize and set up the App instance; fundle ensures all fields are properly constructed.
     let app = App::builder()
         .clock(|_| Clock::new_tokio())
-        .global_pool(|_| GlobalPool::new())
+        .global_pool(|_| OpaqueMemory::new(GlobalPool::new()))
         .client({
             let meter_provider = meter_provider.clone();
             move |x| {

crates/fetch/examples/http_client_custom.rs

@@ -4,7 +4,7 @@
 //! Plugs a custom `EchoHandler` into [`fetch::custom::create_builder`] as the transport
 //! handler. Every request's body is returned verbatim in the response.
 
-use bytesbuf::mem::GlobalPool;
+use bytesbuf::mem::{GlobalPool, OpaqueMemory};
 use fetch::custom::{CustomDeps, Isolation, create_builder};
 use fetch::{HttpRequest, HttpResponse, HttpResponseBuilder};
 use http::StatusCode;
@@ -16,7 +16,7 @@ use tick::Clock;
 async fn main() -> Result<(), ohno::AppError> {
     let deps = CustomDeps {
         clock: Clock::new_tokio(),
-        global_pool: GlobalPool::new(),
+        memory: OpaqueMemory::new(GlobalPool::new()),
         extras: (),
     };
 

crates/fetch/src/custom.rs

@@ -17,7 +17,7 @@ use std::borrow::Cow;
 use std::fmt::Debug;
 use std::sync::Arc;
 
-use bytesbuf::mem::GlobalPool;
+use bytesbuf::mem::OpaqueMemory;
 use http_extensions::{HttpBodyBuilder, RequestHandler};
 use opentelemetry::metrics::Meter;
 use thread_aware::{PerCore, ThreadAware, unaware};
@@ -53,7 +53,7 @@ where
     /// Clock for timing operations and timeouts.
     pub clock: Clock,
     /// Memory pool for usage-neutral memory allocations.
-    pub global_pool: GlobalPool,
+    pub memory: OpaqueMemory,
     /// Extra dependencies forwarded verbatim to [`CustomContext::extras`].
     pub extras: Extras,
 }
@@ -210,12 +210,12 @@ impl HttpClient {
             runtime_name: runtime.into(),
             name: transport.into(),
             clock: deps.clock.clone(),
-            global_pool: deps.global_pool.clone(),
+            memory: deps.memory.clone(),
             isolation,
             inner: thread_aware::Arc::new_with((deps, unaware(factory)), |(deps, factory)| {
                 Arc::new(move |options, meter, pool_index| {
                     let context = CustomContext {
-                        body_builder: create_body_builder(&deps.global_pool, &deps.clock, &options),
+                        body_builder: create_body_builder(&deps.memory, &deps.clock, &options),
                         clock: deps.clock.clone(),
                         pool_index,
                         extras: deps.extras.clone(),
@@ -242,7 +242,7 @@ pub(crate) struct Transport {
     name: Cow<'static, str>,
     inner: thread_aware::Arc<TransportFn, PerCore>,
     clock: Clock,
-    global_pool: GlobalPool,
+    memory: OpaqueMemory,
     isolation: Isolation,
 }
 
@@ -268,7 +268,7 @@ impl Transport {
     }
 
     pub(crate) fn create_body_builder(&self, options: &ClientOptions) -> HttpBodyBuilder {
-        create_body_builder(&self.global_pool, &self.clock, options)
+        create_body_builder(&self.memory, &self.clock, options)
     }
 }
 
@@ -278,7 +278,7 @@ impl Debug for Transport {
     }
 }
 
-pub(crate) fn create_body_builder(pool: &GlobalPool, clock: &Clock, options: &ClientOptions) -> HttpBodyBuilder {
+pub(crate) fn create_body_builder(pool: &OpaqueMemory, clock: &Clock, options: &ClientOptions) -> HttpBodyBuilder {
     HttpBodyBuilder::new(pool.clone(), clock).with_options(options.response_body_options)
 }
 
@@ -288,9 +288,11 @@ mod tests {
     use std::sync::Arc;
     use std::sync::atomic::{AtomicUsize, Ordering};
 
+    use bytesbuf::BytesBuf;
+    use bytesbuf::mem::{GlobalPool, Memory, OpaqueMemory};
     use http::StatusCode;
     use http_extensions::FakeHandler;
-    use thread_aware::unaware;
+    use thread_aware::{ThreadAware, unaware};
 
     use super::{CustomContext, CustomDeps, Isolation, create_builder};
     use crate::HttpResponseBuilder;
@@ -301,7 +303,7 @@ mod tests {
     fn custom_deps() -> CustomDeps {
         CustomDeps {
             clock: FakeDeps::default().clock,
-            global_pool: bytesbuf::mem::GlobalPool::new(),
+            memory: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()),
             extras: (),
         }
     }
@@ -311,6 +313,36 @@ mod tests {
         FakeHandler::from_fn(|_req| HttpResponseBuilder::new_fake().status(StatusCode::OK).build())
     }
 
+    #[derive(Clone, Debug, ThreadAware)]
+    struct CustomMemory {
+        inner: GlobalPool,
+    }
+
+    impl Memory for CustomMemory {
+        fn reserve(&self, min_bytes: usize) -> BytesBuf {
+            self.inner.reserve(min_bytes)
+        }
+    }
+
+    #[cfg_attr(miri, ignore)]
+    #[tokio::test]
+    async fn custom_deps_accept_custom_opaque_memory() {
+        let deps = CustomDeps {
+            clock: FakeDeps::default().clock,
+            memory: OpaqueMemory::new(CustomMemory { inner: GlobalPool::new() }),
+            extras: (),
+        };
+
+        let client = create_builder("test-runtime", "test", ok_factory, Isolation::Shared, deps)
+            .insecure_allow_http()
+            .minimal_pipeline()
+            .build();
+
+        let response = client.post("http://example.com").text("custom pool").fetch().await.unwrap();
+
+        assert_eq!(response.status(), StatusCode::OK);
+    }
+
     #[cfg_attr(miri, ignore)]
     #[tokio::test]
     async fn create_builder_serves_requests_through_custom_pipeline() {
@@ -348,7 +380,7 @@ mod tests {
         let counter = Arc::new(AtomicUsize::new(0));
         let deps = CustomDeps {
             clock: FakeDeps::default().clock,
-            global_pool: bytesbuf::mem::GlobalPool::new(),
+            memory: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()),
             extras: unaware(Arc::clone(&counter)),
         };
 

crates/fetch/src/fake.rs

@@ -84,7 +84,7 @@ impl HttpClient {
             Isolation::Shared,
             CustomDeps {
                 clock: deps.clock,
-                global_pool: bytesbuf::mem::GlobalPool::new(),
+                memory: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()),
                 extras: handler,
             },
         )

crates/fetch/src/tokio.rs

@@ -33,7 +33,7 @@ pub struct TokioDeps {
     /// Clock for timing operations and timeouts.
     pub clock: Clock,
     /// Memory pool for usage-neutral memory allocations.
-    pub global_pool: bytesbuf::mem::GlobalPool,
+    pub global_pool: bytesbuf::mem::OpaqueMemory,
 }
 
 impl Default for TokioDeps {
@@ -47,7 +47,7 @@ impl TokioDeps {
     #[must_use]
     pub fn with_clock(clock: &Clock) -> Self {
         Self {
-            global_pool: bytesbuf::mem::GlobalPool::new(),
+            global_pool: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()),
             clock: clock.clone(),
         }
     }
@@ -64,7 +64,7 @@ impl HttpClient {
     pub fn builder_tokio(deps: impl Into<TokioDeps>) -> HttpClientBuilder {
         let deps = deps.into();
         let clock = deps.clock.clone();
-        let global_pool = deps.global_pool.clone();
+        let memory = deps.global_pool.clone();
 
         // Re-layer on top of the in-crate `builder_custom_internal` path: the
         // full `TokioDeps` rides through `CustomDeps::extras` so that the
@@ -77,7 +77,7 @@ impl HttpClient {
             Isolation::Shared,
             CustomDeps {
                 clock,
-                global_pool,
+                memory,
                 extras: deps,
             },
         )

crates/fetch/tests/telemetry_scope.rs

@@ -94,7 +94,7 @@ async fn custom_transport_scope_attribute() {
 
     let deps = CustomDeps {
         clock: Clock::new_frozen(),
-        global_pool: bytesbuf::mem::GlobalPool::new(),
+        memory: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()),
         extras: (),
     };
 
@@ -128,7 +128,7 @@ async fn custom_transport_instrument_inherits_scope() {
 
     let deps = CustomDeps {
         clock: Clock::new_frozen(),
-        global_pool: bytesbuf::mem::GlobalPool::new(),
+        memory: bytesbuf::mem::OpaqueMemory::new(bytesbuf::mem::GlobalPool::new()),
         extras: (),
     };
 

crates/http_extensions/Cargo.toml

@@ -25,10 +25,10 @@ ignored = ["uuid"]
 allowed_external_types = [
     "futures_core::stream::Stream",
     "bytes::bytes::Bytes",
-    "bytesbuf::mem::global::GlobalPool",
     "bytesbuf::mem::has_memory::HasMemory",
     "bytesbuf::mem::memory::Memory",
     "bytesbuf::mem::memory_shared::MemoryShared",
+    "bytesbuf::mem::opaque_memory::OpaqueMemory",
     "bytesbuf::view::BytesView",
     "http::*",
     "http_body::Body",

crates/http_extensions/benches/http_request_builder.rs

@@ -153,7 +153,7 @@ fn entry(c: &mut Criterion) {
     // Use TransparentMemory instead of GlobalPool so that every reserve() call from the
     // serde_json writer results in a real heap allocation. This makes alloc_tracker report
     // the true number of memory reservations, which GlobalPool would otherwise absorb.
-    let transparent_body_builder = HttpBodyBuilder::with_custom_memory(TransparentMemory::new(), &tick::Clock::new_frozen());
+    let transparent_body_builder = HttpBodyBuilder::new(TransparentMemory::new(), &tick::Clock::new_frozen());
     let operation = session.operation("json_body_large_transparent");
     group.bench_function("json_body_large_transparent", |b| {
         b.iter_custom(|iters| {