feat(capablities): sleep & spawn capabilities

Author: Aaalibaba42

Cargo.lock

@@ -18,6 +18,7 @@ anyhow = { version = "1.0" }
 arrayref = "0.3.7"
 priority-queue = "2.1.1"
 libdd-common = { path = "../libdd-common" }
+libdd-capabilities-impl = { path = "../libdd-capabilities-impl" }
 datadog-sidecar-macros = { path = "../datadog-sidecar-macros" }
 
 libdd-telemetry = { path = "../libdd-telemetry", features = ["tracing"] }

datadog-sidecar/Cargo.toml

@@ -16,7 +16,7 @@ use datadog_ipc::platform::NamedShmHandle;
 use futures::future::Shared;
 use futures::FutureExt;
 use http::uri::PathAndQuery;
-use libdd_common::DefaultHttpClient;
+use libdd_capabilities_impl::NativeCapabilities;
 use libdd_common::{Endpoint, MutexExt};
 use libdd_data_pipeline::agent_info::schema::AgentInfoStruct;
 use libdd_data_pipeline::agent_info::{fetch_info_with_state, FetchInfoStatus};
@@ -103,7 +103,7 @@ impl AgentInfoFetcher {
             fetch_endpoint.url = http::Uri::from_parts(parts).unwrap();
             loop {
                 let fetched =
-                    fetch_info_with_state::<DefaultHttpClient>(&fetch_endpoint, state.as_deref())
+                    fetch_info_with_state::<NativeCapabilities>(&fetch_endpoint, state.as_deref())
                         .await;
                 let mut complete_fut = None;
                 {

datadog-sidecar/src/service/agent_info.rs

@@ -5,8 +5,8 @@ use super::TraceSendData;
 use crate::agent_remote_config::AgentRemoteConfigWriter;
 use datadog_ipc::platform::NamedShmHandle;
 use futures::future::join_all;
-use libdd_common::capabilities::HttpClientTrait;
-use libdd_common::{DefaultHttpClient, Endpoint, MutexExt};
+use libdd_capabilities_impl::{HttpClientCapability, NativeCapabilities};
+use libdd_common::{Endpoint, MutexExt};
 use libdd_trace_utils::trace_utils;
 use libdd_trace_utils::trace_utils::SendData;
 use libdd_trace_utils::trace_utils::SendDataResult;
@@ -95,7 +95,7 @@ pub(crate) struct TraceFlusher {
     pub(crate) min_force_drop_size_bytes: AtomicU32, // put a limit on memory usage
     remote_config: Mutex<AgentRemoteConfigs>,
     pub metrics: Mutex<TraceFlusherMetrics>,
-    client: DefaultHttpClient,
+    capabilities: NativeCapabilities,
 }
 impl Default for TraceFlusher {
     fn default() -> Self {
@@ -106,7 +106,7 @@ impl Default for TraceFlusher {
             min_force_drop_size_bytes: AtomicU32::new(trace_utils::MAX_PAYLOAD_SIZE as u32),
             remote_config: Mutex::new(Default::default()),
             metrics: Mutex::new(Default::default()),
-            client: DefaultHttpClient::new_client(),
+            capabilities: NativeCapabilities::new_client(),
         }
     }
 }
@@ -248,7 +248,7 @@ impl TraceFlusher {
 
     async fn send_and_handle_trace(&self, send_data: SendData) {
         let endpoint = send_data.get_target().clone();
-        let response = send_data.send(&self.client).await;
+        let response = send_data.send(&self.capabilities).await;
         self.metrics.lock_or_panic().update(&response);
         match response.last_result {
             Ok(response) => {

datadog-sidecar/src/service/tracing/trace_flusher.rs

@@ -20,3 +20,4 @@ bytes = "1"
 http = "1"
 libdd-capabilities = { path = "../libdd-capabilities", version = "0.1.0" }
 libdd-common = { path = "../libdd-common", version = "3.0.2" }
+tokio = { version = "1", features = ["time", "rt"] }

libdd-capabilities-impl/Cargo.toml

@@ -8,11 +8,13 @@ Native implementations of `libdd-capabilities` traits.
 
 ## Capabilities
 
-- **`DefaultHttpClient`**: HTTP client backed by hyper and the `libdd-common` connector infrastructure (supports Unix sockets, HTTPS with rustls, Windows named pipes).
+- **`NativeHttpClient`**: HTTP client backed by hyper and the `libdd-common` connector infrastructure (supports Unix sockets, HTTPS with rustls, Windows named pipes).
+- **`NativeSleepCapability`**: Sleep backed by `tokio::time::sleep`.
+- **`NativeSpawnCapability`**: Task spawning backed by `tokio::runtime::Handle::spawn`.
 
 ## Types
 
-- **`NativeCapabilities`**: Bundle type alias that implements all capability traits using native backends. Currently delegates to `DefaultHttpClient`; as more capability traits are added (spawn, sleep, etc.), this type will implement all of them.
+- **`NativeCapabilities`**: Bundle struct that implements all capability traits using native backends. Delegates to `NativeHttpClient`, `NativeSleepCapability`, and `NativeSpawnCapability`.
 
 ## Usage
 

libdd-capabilities-impl/README.md

@@ -1,7 +1,7 @@
 // Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
-//! Re-exports `DefaultHttpClient` from `libdd-common`, where it lives alongside
+//! Re-exports `NativeHttpClient` from `libdd-common`, where it lives alongside
 //! the hyper infrastructure it wraps.
 
-pub use libdd_common::DefaultHttpClient;
+pub use libdd_common::NativeHttpClient;

libdd-capabilities-impl/src/http.rs

@@ -8,34 +8,58 @@
 //! etc.). Leaf crates (FFI, benchmarks) pin this type as the generic parameter.
 
 mod http;
+pub mod sleep;
+pub mod spawn;
 
 use core::future::Future;
+use std::time::Duration;
 
-pub use http::DefaultHttpClient;
+pub use http::NativeHttpClient;
 use libdd_capabilities::http::HttpError;
-pub use libdd_capabilities::HttpClientTrait;
+pub use libdd_capabilities::HttpClientCapability;
 use libdd_capabilities::MaybeSend;
+pub use libdd_capabilities::SleepCapability;
+pub use libdd_capabilities::SpawnCapability;
+pub use sleep::NativeSleepCapability;
+pub use spawn::{NativeJoinHandle, NativeSpawnCapability};
 
 /// Bundle struct for native platform capabilities.
 ///
-/// Delegates to [`DefaultHttpClient`] for HTTP. As more capability traits are
-/// added (spawn, sleep, etc.), additional fields and impls are added here
-/// without changing the type identity — consumers see the same
-/// `NativeCapabilities` throughout.
+/// Delegates to [`NativeHttpClient`] for HTTP, [`NativeSleepCapability`] for
+/// sleep, and [`NativeSpawnCapability`] for task spawning.
 ///
 /// Individual capability traits keep minimal per-function bounds (e.g.
-/// functions that only need HTTP require just `H: HttpClientTrait`, not the
+/// functions that only need HTTP require just `H: HttpClientCapability`, not the
 /// full bundle) so that native callers like the sidecar can use
-/// `DefaultHttpClient` directly without pulling in this bundle.
+/// `NativeHttpClient` directly without pulling in this bundle.
 #[derive(Clone, Debug)]
 pub struct NativeCapabilities {
-    http: DefaultHttpClient,
+    http: NativeHttpClient,
+    sleep: NativeSleepCapability,
+    spawn: NativeSpawnCapability,
 }
 
-impl HttpClientTrait for NativeCapabilities {
+impl NativeCapabilities {
+    /// Create a bundle with an explicit tokio runtime handle for spawning.
+    ///
+    /// Prefer `new_client()` (via `HttpClientCapability`) when already inside
+    /// a tokio context. This constructor exists for test code that owns a
+    /// `SharedRuntime` and needs to pass its handle explicitly.
+    pub fn new(handle: tokio::runtime::Handle) -> Self {
+        Self {
+            http: NativeHttpClient::new_client(),
+            sleep: NativeSleepCapability,
+            spawn: NativeSpawnCapability::new(handle),
+        }
+    }
+}
+
+impl HttpClientCapability for NativeCapabilities {
     fn new_client() -> Self {
         Self {
-            http: DefaultHttpClient::new_client(),
+            http: NativeHttpClient::new_client(),
+            sleep: NativeSleepCapability,
+            spawn: NativeSpawnCapability::from_current(),
         }
     }
 
@@ -46,3 +70,21 @@ impl HttpClientTrait for NativeCapabilities {
         self.http.request(req)
     }
 }
+
+impl SleepCapability for NativeCapabilities {
+    fn sleep(&self, duration: Duration) -> impl Future<Output = ()> + MaybeSend {
+        self.sleep.sleep(duration)
+    }
+}
+
+impl SpawnCapability for NativeCapabilities {
+    type JoinHandle<T: MaybeSend + 'static> = NativeJoinHandle<T>;
+
+    fn spawn<F, T>(&self, future: F) -> NativeJoinHandle<T>
+    where
+        F: Future<Output = T> + MaybeSend + 'static,
+        T: MaybeSend + 'static,
+    {
+        self.spawn.spawn(future)
+    }
+}

libdd-capabilities-impl/src/lib.rs

@@ -0,0 +1,19 @@
+// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+//! Native sleep implementation backed by `tokio::time::sleep`.
+
+use core::future::Future;
+use std::time::Duration;
+
+use libdd_capabilities::maybe_send::MaybeSend;
+use libdd_capabilities::sleep::SleepCapability;
+
+#[derive(Clone, Debug)]
+pub struct NativeSleepCapability;
+
+impl SleepCapability for NativeSleepCapability {
+    fn sleep(&self, duration: Duration) -> impl Future<Output = ()> + MaybeSend {
+        tokio::time::sleep(duration)
+    }
+}

libdd-capabilities-impl/src/sleep.rs

@@ -0,0 +1,61 @@
+// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+//! Native spawn implementation backed by `tokio::runtime::Handle::spawn`.
+
+use core::future::Future;
+use core::pin::Pin;
+use core::task::{Context, Poll};
+
+use libdd_capabilities::maybe_send::MaybeSend;
+use libdd_capabilities::spawn::SpawnCapability;
+use tokio::task::JoinHandle;
+
+#[derive(Clone, Debug)]
+pub struct NativeSpawnCapability {
+    handle: tokio::runtime::Handle,
+}
+
+impl NativeSpawnCapability {
+    pub fn new(handle: tokio::runtime::Handle) -> Self {
+        Self { handle }
+    }
+
+    pub fn from_current() -> Self {
+        Self {
+            handle: tokio::runtime::Handle::current(),
+        }
+    }
+}
+
+impl SpawnCapability for NativeSpawnCapability {
+    type JoinHandle<T: MaybeSend + 'static> = NativeJoinHandle<T>;
+
+    fn spawn<F, T>(&self, future: F) -> NativeJoinHandle<T>
+    where
+        F: Future<Output = T> + MaybeSend + 'static,
+        T: MaybeSend + 'static,
+    {
+        NativeJoinHandle(self.handle.spawn(future))
+    }
+}
+
+/// Newtype wrapping `tokio::task::JoinHandle<T>` that normalises the output to
+/// `T` instead of `Result<T, JoinError>`.
+///
+/// A `JoinError` means the spawned task panicked or was aborted. Workers use
+/// `CancellationToken` for graceful shutdown, so `JoinError` indicates a bug.
+pub struct NativeJoinHandle<T>(JoinHandle<T>);
+
+impl<T> Future for NativeJoinHandle<T> {
+    type Output = T;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
+        // JoinHandle<T>: Unpin, so Pin::new is safe.
+        match Pin::new(&mut self.get_mut().0).poll(cx) {
+            Poll::Ready(Ok(val)) => Poll::Ready(val),
+            Poll::Ready(Err(e)) => panic!("spawned task failed: {e}"),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}