Fix #127: Capture async receive timestamp before deserialization

The async standalone server was calling get_monotonic_time_ns() after
transport.receive().await returned, which includes both Tokio executor
scheduling delay and deserialization time in the one-way latency
measurement. The blocking path already used receive_blocking_timed()
which timestamps between raw I/O and deserialization.

Add receive_timed() to the async IpcTransport trait with a default
implementation, then override it in TCP and UDS transports to capture
the monotonic timestamp immediately after read_exact() completes but
before Message::from_bytes() deserialization runs. Update the standalone
server to use receive_timed() for one-way latency measurement.

- Add async receive_timed() default method to IpcTransport trait
- Implement read_message_timed() in TcpSocketTransport
- Implement read_message_timed() in UnixDomainSocketTransport
- Override receive_timed() in both TCP and UDS async transports
- Update standalone_server async one-way path to use receive_timed()
- Remove unused get_monotonic_time_ns import from standalone_server
- All tests passing, clippy clean

AI-assisted-by: Claude Opus 4.6
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: mcurrier2

src/ipc/mod.rs

@@ -730,6 +730,28 @@ pub trait IpcTransport: Send + Sync {
     /// - Async implementation allows cancellation
     async fn receive(&mut self) -> Result<Message>;
 
+    /// Receive a message and capture a monotonic timestamp immediately
+    /// after the raw bytes are read but before deserialization.
+    ///
+    /// This provides more accurate one-way latency measurement by
+    /// excluding deserialization overhead from the receive timestamp,
+    /// mirroring the blocking transport's `receive_blocking_timed()`.
+    ///
+    /// The default implementation captures the timestamp after
+    /// `receive()` returns (including deserialization). Transport
+    /// implementations should override this to place the timestamp
+    /// between raw I/O and deserialization for better accuracy.
+    ///
+    /// # Returns
+    ///
+    /// A tuple of (Message, timestamp_ns) where timestamp_ns is a
+    /// monotonic clock value captured as close to I/O completion as
+    /// possible.
+    async fn receive_timed(&mut self) -> Result<(Message, u64)> {
+        let msg = self.receive().await?;
+        Ok((msg, get_monotonic_time_ns()))
+    }
+
     /// Close the transport
     ///
     /// Cleanly shuts down the transport, releasing all resources

src/ipc/tcp_socket.rs

@@ -1,4 +1,7 @@
-use super::{ConnectionId, IpcError, IpcTransport, Message, TransportConfig, TransportState};
+use super::{
+    get_monotonic_time_ns, ConnectionId, IpcError, IpcTransport, Message, TransportConfig,
+    TransportState,
+};
 use anyhow::{anyhow, Result};
 use async_trait::async_trait;
 use std::collections::HashMap;
@@ -69,6 +72,28 @@ impl TcpSocketTransport {
         Message::from_bytes(&message_data)
     }
 
+    /// Read a message and capture a monotonic timestamp immediately after
+    /// the raw bytes are read but before deserialization. This excludes
+    /// deserialization overhead from one-way latency measurements.
+    async fn read_message_timed(stream: &mut TcpStream) -> Result<(Message, u64)> {
+        let mut len_bytes = [0u8; 4];
+        stream.read_exact(&mut len_bytes).await?;
+        let message_len = u32::from_le_bytes(len_bytes) as usize;
+
+        if message_len > 16 * 1024 * 1024 {
+            return Err(anyhow!("Message too large: {} bytes", message_len));
+        }
+
+        let mut message_data = vec![0u8; message_len];
+        stream.read_exact(&mut message_data).await?;
+
+        // Capture timestamp between raw I/O and deserialization
+        let receive_time_ns = get_monotonic_time_ns();
+
+        let message = Message::from_bytes(&message_data)?;
+        Ok((message, receive_time_ns))
+    }
+
     /// Write a message to the TCP stream.
     ///
     /// Unlike SHM and PMQ, the timestamp is not refreshed
@@ -340,6 +365,36 @@ impl IpcTransport for TcpSocketTransport {
         }
     }
 
+    async fn receive_timed(&mut self) -> Result<(Message, u64)> {
+        if self.state != TransportState::Connected {
+            return Err(anyhow!("Transport not connected"));
+        }
+
+        if self.stream.is_none() {
+            if let Some(listener) = self.listener.as_ref() {
+                let (stream, client_addr) = listener.accept().await?;
+                debug!(
+                    "TCP Socket server accepted connection from: {}",
+                    client_addr
+                );
+                let std_stream = stream.into_std()?;
+                let socket = socket2::Socket::from(std_stream.try_clone()?);
+                socket.set_nodelay(true)?;
+                socket.set_recv_buffer_size(self.buffer_size)?;
+                socket.set_send_buffer_size(self.buffer_size)?;
+                self.stream = Some(TcpStream::from_std(std_stream)?);
+            }
+        }
+
+        if let Some(ref mut stream) = self.stream {
+            let (message, ts) = Self::read_message_timed(stream).await?;
+            debug!("Received message {} via TCP Socket", message.id);
+            Ok((message, ts))
+        } else {
+            Err(anyhow!("No active stream available"))
+        }
+    }
+
     async fn close(&mut self) -> Result<()> {
         debug!("Closing TCP Socket transport");
 

src/ipc/unix_domain_socket.rs

@@ -1,4 +1,7 @@
-use super::{ConnectionId, IpcError, IpcTransport, Message, TransportConfig, TransportState};
+use super::{
+    get_monotonic_time_ns, ConnectionId, IpcError, IpcTransport, Message, TransportConfig,
+    TransportState,
+};
 use anyhow::{anyhow, Result};
 use async_trait::async_trait;
 use std::collections::HashMap;
@@ -66,6 +69,28 @@ impl UnixDomainSocketTransport {
         Message::from_bytes(&message_data)
     }
 
+    /// Read a message and capture a monotonic timestamp immediately after
+    /// the raw bytes are read but before deserialization. This excludes
+    /// deserialization overhead from one-way latency measurements.
+    async fn read_message_timed(stream: &mut UnixStream) -> Result<(Message, u64)> {
+        let mut len_bytes = [0u8; 4];
+        stream.read_exact(&mut len_bytes).await?;
+        let message_len = u32::from_le_bytes(len_bytes) as usize;
+
+        if message_len > 16 * 1024 * 1024 {
+            return Err(anyhow!("Message too large: {} bytes", message_len));
+        }
+
+        let mut message_data = vec![0u8; message_len];
+        stream.read_exact(&mut message_data).await?;
+
+        // Capture timestamp between raw I/O and deserialization
+        let receive_time_ns = get_monotonic_time_ns();
+
+        let message = Message::from_bytes(&message_data)?;
+        Ok((message, receive_time_ns))
+    }
+
     /// Write a message to the Unix stream.
     ///
     /// Unlike SHM and PMQ, the timestamp is not refreshed
@@ -320,6 +345,27 @@ impl IpcTransport for UnixDomainSocketTransport {
         }
     }
 
+    async fn receive_timed(&mut self) -> Result<(Message, u64)> {
+        if self.state != TransportState::Connected {
+            return Err(anyhow!("Transport not connected"));
+        }
+
+        if self.stream.is_none() {
+            if let Some(listener) = self.listener.as_ref() {
+                let (stream, _) = listener.accept().await?;
+                self.stream = Some(stream);
+            }
+        }
+
+        if let Some(ref mut stream) = self.stream {
+            let (message, ts) = Self::read_message_timed(stream).await?;
+            debug!("Received message {} via Unix Domain Socket", message.id);
+            Ok((message, ts))
+        } else {
+            Err(anyhow!("No active stream available"))
+        }
+    }
+
     async fn close(&mut self) -> Result<()> {
         debug!("Closing Unix Domain Socket transport");
 

src/standalone_server.rs

@@ -19,8 +19,8 @@ use tracing_subscriber::filter::LevelFilter;
 use crate::benchmark::BenchmarkConfig;
 use crate::cli::{Args, IpcMechanism};
 use crate::ipc::{
-    get_monotonic_time_ns, BlockingTransport, BlockingTransportFactory, Message, MessageType,
-    TransportConfig, TransportFactory,
+    BlockingTransport, BlockingTransportFactory, Message, MessageType, TransportConfig,
+    TransportFactory,
 };
 use crate::logging::ColorizedFormatter;
 use crate::metrics::{LatencyType, MetricsCollector};
@@ -747,15 +747,14 @@ pub async fn run_standalone_server_async_single(
             info!("Shutdown signal received, exiting");
             break;
         }
-        match transport.receive().await {
-            Ok(msg) => {
+        match transport.receive_timed().await {
+            Ok((msg, receive_time_ns)) => {
                 if msg.message_type == MessageType::Shutdown {
                     debug!("Server received shutdown message, exiting");
                     break;
                 }
 
                 if msg.message_type == MessageType::OneWay && msg.id != u64::MAX {
-                    let receive_time_ns = get_monotonic_time_ns();
                     let latency_ns = receive_time_ns.saturating_sub(msg.timestamp);
                     let latency = std::time::Duration::from_nanos(latency_ns);
                     one_way_metrics.record_message(config.message_size, Some(latency))?;