tokio-boring: record when the stream was first read

tokio-boring/Cargo.toml

@@ -32,7 +32,7 @@ fips-precompiled = ["fips"]
 [dependencies]
 boring = { workspace = true, default-features = false }
 boring-sys = { workspace = true, default-features = false }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["time"] }
 
 [dev-dependencies]
 futures = { workspace = true }

tokio-boring/src/bridge.rs

@@ -4,10 +4,12 @@ use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll, Waker};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+use tokio::time::Instant;
 
 pub(crate) struct AsyncStreamBridge<S> {
     pub(crate) stream: S,
     waker: Option<Waker>,
+    pub(crate) first_read: Option<Instant>,
 }
 
 impl<S> AsyncStreamBridge<S> {
@@ -18,6 +20,7 @@ impl<S> AsyncStreamBridge<S> {
         Self {
             stream,
             waker: None,
+            first_read: None,
         }
     }
 
@@ -45,14 +48,20 @@ where
     S: AsyncRead + Unpin,
 {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        self.with_context(|ctx, stream| {
+        let n = self.with_context(|ctx, stream| {
             let mut buf = ReadBuf::new(buf);
 
             match stream.poll_read(ctx, &mut buf)? {
                 Poll::Ready(()) => Ok(buf.filled().len()),
                 Poll::Pending => Err(io::Error::from(io::ErrorKind::WouldBlock)),
             }
-        })
+        })?;
+
+        if n > 0 && self.first_read.is_none() {
+            self.first_read = Some(Instant::now());
+        }
+
+        Ok(n)
     }
 }
 

tokio-boring/src/lib.rs

@@ -24,6 +24,7 @@ use std::io::{self, Write};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+use tokio::time::Instant;
 
 mod async_callbacks;
 mod bridge;
@@ -157,6 +158,19 @@ impl<S> SslStream<S> {
         &mut self.0.get_mut().stream
     }
 
+    /// Returns when a read from the underlying stream first returned data, or
+    /// `None` if nothing has been read yet.
+    ///
+    /// For a server this is the first read of the peer's ClientHello, and for
+    /// a client the first read of the ServerHello. This is useful for measuring
+    /// handshake latency without counting time spent waiting for the peer. The
+    /// timestamp is taken when the read completes, so socket buffering and task
+    /// scheduling can put it some time after the bytes actually arrived.
+    #[must_use]
+    pub fn first_read_at(&self) -> Option<Instant> {
+        self.0.get_ref().first_read
+    }
+
     fn run_in_context<F, R>(&mut self, ctx: &mut Context<'_>, f: F) -> R
     where
         F: FnOnce(&mut ssl::SslStream<AsyncStreamBridge<S>>) -> R,

tokio-boring/tests/client_server.rs

@@ -57,3 +57,31 @@ async fn handshake_error() {
 
     future::join(server, client).await;
 }
+
+#[tokio::test]
+async fn first_read_at_excludes_wait_before_client_hello() {
+    const DELAY: std::time::Duration = std::time::Duration::from_millis(200);
+
+    let (stream, addr) = create_server(|_| ());
+
+    let server = async {
+        let started_at = tokio::time::Instant::now();
+        let stream = stream.await.unwrap();
+        let first_read = stream.first_read_at().expect("ClientHello was read");
+
+        assert!(first_read.duration_since(started_at) >= DELAY);
+    };
+
+    let client = async {
+        let config = common::create_connector(|b| b.set_ca_file("tests/cert.pem"))
+            .configure()
+            .unwrap();
+        let tcp = TcpStream::connect(&addr).await.unwrap();
+        tokio::time::sleep(DELAY).await;
+        tokio_boring::connect(config, "localhost", tcp)
+            .await
+            .unwrap();
+    };
+
+    future::join(server, client).await;
+}