Fmt

Signed-off-by: Bob Weinand <bob.weinand@datadoghq.com>

Author: bwoebi

LICENSE-3rdparty.yml

@@ -130,7 +130,7 @@ fn gen_request_enum(enum_name: &Ident, methods: &[MethodInfo]) -> proc_macro2::T
         .collect();
 
     quote! {
-        #[derive(::serde::Serialize, ::serde::Deserialize)]
+        #[derive(::serde::Serialize, ::serde::Deserialize, Debug)]
         pub enum #enum_name {
             #(#variants),*
         }
@@ -341,12 +341,13 @@ fn gen_serve_fn(
                     ::tracing::warn!("IPC serve: failed to receive handles");
                     break;
                 }
-                let recv_counter = handler.recv_counter().fetch_add(1, ::std::sync::atomic::Ordering::Relaxed) + 1;
-                ::tracing::trace!(recv_counter, discriminant, pid = peer.pid, "IPC recv");
+                let recv_counter = handler.recv_counter().load(::std::sync::atomic::Ordering::Relaxed) + 1;
+                ::tracing::trace!(recv_counter, ?req, pid = peer.pid, "IPC recv");
 
                 match req {
                     #(#match_arms)*
                 }
+                handler.recv_counter().fetch_add(1, ::std::sync::atomic::Ordering::Relaxed);
             }
         }
     }
@@ -470,7 +471,7 @@ fn gen_channel(
     }
 }
 
-/// `#[service]` replaces `#[tarpc::service]` + `#[impl_transfer_handles]`.
+/// `#[service]` macro.
 ///
 /// Generates from a `trait` definition:
 /// - `{Trait}Request` enum (Clone, Serialize, Deserialize, TransferHandles)

datadog-ipc-macros/src/lib.rs

@@ -85,17 +85,15 @@ impl ExampleInterface for ExampleServer {
         std::future::ready(())
     }
 
-    fn shm_sum(
+    async fn shm_sum(
         &self,
         _peer: datadog_ipc::PeerCredentials,
         handle: ShmHandle,
         len: usize,
-    ) -> impl std::future::Future<Output = u64> + Send + '_ {
-        async move {
-            match handle.map() {
-                Ok(mapped) => mapped.as_slice()[..len].iter().map(|&b| b as u64).sum(),
-                Err(_) => u64::MAX,
-            }
+    ) -> u64 {
+        match handle.map() {
+            Ok(mapped) => mapped.as_slice()[..len].iter().map(|&b| b as u64).sum(),
+            Err(_) => u64::MAX,
         }
     }
 

datadog-ipc/src/example_interface.rs

@@ -17,7 +17,7 @@ pub mod client;
 pub mod codec;
 
 pub use client::IpcClientConn;
-pub use platform::{recv_raw_async, send_raw_async};
 pub use platform::{
     max_message_size, PeerCredentials, SeqpacketConn, SeqpacketListener, HANDLE_SUFFIX_SIZE,
 };
+pub use platform::{recv_raw_async, send_raw_async};

datadog-ipc/src/lib.rs

@@ -34,7 +34,7 @@ impl SeqpacketListener {
         let addr = UnixAddr::new_abstract(name).map_err(io::Error::from)?;
         Self::do_bind(addr)
     }
-    
+
     fn do_bind(addr: UnixAddr) -> io::Result<Self> {
         let fd = create_seqpacket_socket()?;
         bind(fd.as_raw_fd(), &addr).map_err(io::Error::from)?;

datadog-ipc/src/platform/unix/sockets/linux.rs

@@ -321,7 +321,11 @@ impl SeqpacketConn {
                 std::mem::size_of::<libc::c_int>() as libc::socklen_t,
             )
         };
-        if ret < 0 { Err(io::Error::last_os_error()) } else { Ok(()) }
+        if ret < 0 {
+            Err(io::Error::last_os_error())
+        } else {
+            Ok(())
+        }
     }
 
     pub fn set_sndbuf_size(&self, size: usize) -> io::Result<()> {
@@ -355,13 +359,13 @@ pub type AsyncConn = AsyncFd<OwnedFd>;
 pub async fn recv_raw_async(fd: &AsyncConn) -> io::Result<(Vec<u8>, Vec<OwnedFd>)> {
     loop {
         let mut guard = fd.readable().await?;
-        // SAFETY: recvmsg writes exactly the first n bytes; we truncate to n before returning,
-        // so no uninitialized bytes are ever exposed to the caller.
         let mut buf = Vec::with_capacity(max_message_size());
-        unsafe { buf.set_len(max_message_size()) };
-        match guard.try_io(|inner| recvmsg_raw(inner.as_raw_fd(), &mut buf, MsgFlags::empty())) {
+        // SAFETY: all bit patterns are valid for u8; recvmsg writes exactly n bytes into
+        // the spare capacity before set_len(n) is called below.
+        let slice = unsafe { std::slice::from_raw_parts_mut(buf.as_mut_ptr(), max_message_size()) };
+        match guard.try_io(|inner| recvmsg_raw(inner.as_raw_fd(), slice, MsgFlags::empty())) {
             Ok(Ok((n, fds))) => {
-                buf.truncate(n);
+                unsafe { buf.set_len(n) };
                 return Ok((buf, fds));
             }
             Ok(Err(e)) => return Err(e),

datadog-ipc/src/platform/unix/sockets/mod.rs

@@ -718,12 +718,16 @@ impl SeqpacketConn {
 pub async fn recv_raw_async(conn: &AsyncConn) -> io::Result<(Vec<u8>, Vec<OwnedHandle>)> {
     let h = conn.handle.as_raw_handle() as SysHANDLE;
     tokio::task::block_in_place(|| {
-        // SAFETY: ReadFile writes exactly the first n bytes; we truncate to n before returning,
-        // so no uninitialized bytes are ever exposed to the caller.
-        let mut buf = Vec::with_capacity(max_message_size() + HANDLE_SUFFIX_SIZE);
-        unsafe { buf.set_len(max_message_size() + HANDLE_SUFFIX_SIZE) };
-        let (n, handles) = pipe_read(h, &mut buf, true)?;
-        buf.truncate(n);
+        let size = max_message_size() + HANDLE_SUFFIX_SIZE;
+        let mut buf = Vec::with_capacity(size);
+        // SAFETY: all bit patterns are valid for u8; pipe_read writes exactly n bytes into
+        // the spare capacity before set_len(n) is called below.
+        let (n, handles) = pipe_read(
+            h,
+            unsafe { std::slice::from_raw_parts_mut(buf.as_mut_ptr(), size) },
+            true,
+        )?;
+        unsafe { buf.set_len(n) };
         Ok((buf, handles))
     })
 }

datadog-ipc/src/platform/windows/sockets.rs

@@ -43,14 +43,15 @@ fn test_shm_handle_transfer() {
 
 /// Verifies that IPC messages larger than 4 KB are handled without panicking.
 ///
-/// Before the fix, Tokio's `NamedPipeServer` registered the pipe handle with mio/IOCP, which
+/// Using Tokio's `NamedPipeServer`, it registered the pipe handle with mio/IOCP, which
 /// posted overlapped `ReadFile` calls into a fixed 4 KB internal buffer.  Messages larger than
 /// 4 KB caused `ReadFile` to return `ERROR_MORE_DATA` synchronously; Windows still queued an
 /// IOCP completion, but mio had already transitioned `io.read` to `State::Err`.  When the
 /// completion fired, mio's `read_done` hit `_ => unreachable!()` (named_pipe.rs:871).
 ///
-/// The fix routes serve-loop I/O through `block_in_place` + direct `ReadFile` into the
-/// caller-supplied large buffer, bypassing mio's 4 KB limit entirely.
+/// Which is why we route serve-loop I/O through `block_in_place` + direct `ReadFile` into the
+/// caller-supplied large buffer, bypassing mio's 4 KB limit entirely. This serves as regression
+/// test.
 #[test]
 fn test_large_message() {
     let (conn_server, conn_client) = SeqpacketConn::socketpair().unwrap();

datadog-ipc/tests/windows_shm.rs

@@ -5,23 +5,23 @@ use super::{
     DynamicInstrumentationConfigState, InstanceId, QueueId, SerializedTracerHeaderTags,
     SessionConfig, SidecarAction,
 };
-use libdd_telemetry::metrics::MetricContext;
 use crate::service::sender::SidecarSender;
 use crate::service::sidecar_interface::SidecarInterfaceChannel;
 use datadog_ipc::platform::{FileBackedHandle, ShmHandle};
 use datadog_ipc::SeqpacketConn;
 use datadog_live_debugger::debugger_defs::DebuggerPayload;
 use datadog_live_debugger::sender::DebuggerType;
 use libdd_common::tag::Tag;
+use libdd_common::MutexExt;
 use libdd_dogstatsd_client::DogStatsDActionOwned;
+use libdd_telemetry::metrics::MetricContext;
 use serde::Serialize;
 use std::sync::Mutex;
 use std::{
     io,
     time::{Duration, Instant},
 };
 use tracing::warn;
-use libdd_common::MutexExt;
 
 /// `SidecarTransport` wraps a [`SidecarSender`] with transparent reconnection support.
 ///
@@ -42,11 +42,15 @@ impl SidecarTransport {
         Self::do_reconnect(&mut self.inner, factory, false);
     }
 
-    pub fn do_reconnect<F>(transport: &mut Mutex<SidecarSender>, factory: F, force_reconnect: bool) -> bool
+    pub fn do_reconnect<F>(
+        transport: &mut Mutex<SidecarSender>,
+        factory: F,
+        force_reconnect: bool,
+    ) -> bool
     where
         F: FnOnce() -> Option<Box<SidecarTransport>>,
     {
-        let mut transport = match transport.lock() {
+        let transport = match transport.get_mut() {
             Ok(t) => t,
             Err(_) => return false,
         };
@@ -120,9 +124,7 @@ impl SidecarTransport {
                 Err(e) => e,
             }
         };
-        if e.kind() == io::ErrorKind::BrokenPipe
-            || e.kind() == io::ErrorKind::ConnectionReset
-        {
+        if e.kind() == io::ErrorKind::BrokenPipe || e.kind() == io::ErrorKind::ConnectionReset {
             if let Some(ref reconnect) = self.reconnect_fn {
                 if Self::do_reconnect(&mut self.inner, reconnect, true) {
                     return f(&mut self.inner.lock_or_panic());
@@ -398,10 +400,7 @@ pub fn send_dogstatsd_actions(
 }
 
 /// Sets x-datadog-test-session-token on all requests for the given session.
-pub fn set_test_session_token(
-    transport: &mut SidecarTransport,
-    token: String,
-) -> io::Result<()> {
+pub fn set_test_session_token(transport: &mut SidecarTransport, token: String) -> io::Result<()> {
     lock_sender(transport)?.set_test_session_token(token);
     Ok(())
 }

datadog-sidecar/src/service/blocking.rs

@@ -69,7 +69,6 @@ fn cancel_if_instance(slot: &mut Option<SidecarInterfaceRequest>, instance_id: &
     }
 }
 
-
 fn cancel_if_queue(
     slot: &mut Option<SidecarInterfaceRequest>,
     instance_id: &InstanceId,
@@ -107,7 +106,11 @@ fn coalesce(outbox: &mut SidecarOutbox, incoming: SidecarInterfaceRequest) {
     } = incoming
     {
         cancel_if_queue(&mut outbox.set_request_config, instance_id, queue_id);
-        cancel_if_queue(&mut outbox.set_universal_service_tags, instance_id, queue_id);
+        cancel_if_queue(
+            &mut outbox.set_universal_service_tags,
+            instance_id,
+            queue_id,
+        );
     }
 
     match incoming {
@@ -280,7 +283,8 @@ impl SidecarSender {
         if self.metric_registrations.contains_key(&metric.name) {
             return;
         }
-        self.metric_registrations.insert(metric.name.clone(), metric.clone());
+        self.metric_registrations
+            .insert(metric.name.clone(), metric.clone());
         let req = SidecarInterfaceRequest::RegisterTelemetryMetric { metric };
         self.channel.send_request_blocking(&req).ok();
     }