feat(s2n-quic-dc): Expose acceptor timing to application (#3006)

Author: Mark-Simulacrum

dc/s2n-quic-dc/src/stream/application.rs

@@ -27,38 +27,54 @@ pub struct Builder<Sub: event::Subscriber> {
     pub app_queue_time: Option<Timestamp>,
 }
 
+/// Carries timestamps of events before the stream is returned to the application in accept().
+#[non_exhaustive]
+pub struct AcceptInfo {
+    /// How long the stream spent inside the dcQUIC acceptor before being enqueued for the
+    /// application.
+    pub dc_quic_accept_time: Duration,
+    /// How long the stream spent enqueued for the application.
+    pub app_queue_sojourn_time: Duration,
+}
+
 impl<Sub> Builder<Sub>
 where
     Sub: event::Subscriber,
 {
     /// Builds the stream and emits an event indicating that the stream was built
     #[inline]
-    pub(crate) fn accept(self) -> io::Result<(Stream<Sub>, Duration)> {
-        let sojourn_time = {
-            let remote_address = self.shared.remote_addr();
-            let remote_address = &remote_address;
-            let creds = self.shared.credentials();
-            let credential_id = &*creds.id;
-            let stream_id = creds.key_id.as_u64();
-            let now = self.shared.common.clock.get_time();
-            let total_sojourn_time = now.saturating_duration_since(self.kernel_start_time);
-            let queue_sojourn_time =
-                now.saturating_duration_since(self.app_queue_time.expect("set by accept_stream"));
-
-            self.shared
-                .endpoint_publisher(now)
-                .on_acceptor_stream_dequeued(event::builder::AcceptorStreamDequeued {
-                    remote_address,
-                    credential_id,
-                    stream_id,
-                    sojourn_time: total_sojourn_time,
-                    queue_sojourn_time,
-                });
-
-            queue_sojourn_time
-        };
-
-        self.build().map(|stream| (stream, sojourn_time))
+    pub(crate) fn accept(self) -> io::Result<(Stream<Sub>, AcceptInfo)> {
+        let kernel_start_time = self.kernel_start_time;
+        let app_queue_time = self.app_queue_time.expect("set by accept_stream");
+        let remote_address = self.shared.remote_addr();
+        let remote_address = &remote_address;
+        let creds = self.shared.credentials();
+        let credential_id = &*creds.id;
+        let stream_id = creds.key_id.as_u64();
+        let now = self.shared.common.clock.get_time();
+        let total_sojourn_time = now.saturating_duration_since(self.kernel_start_time);
+        let queue_sojourn_time = now.saturating_duration_since(app_queue_time);
+
+        self.shared
+            .endpoint_publisher(now)
+            .on_acceptor_stream_dequeued(event::builder::AcceptorStreamDequeued {
+                remote_address,
+                credential_id,
+                stream_id,
+                sojourn_time: total_sojourn_time,
+                queue_sojourn_time,
+            });
+
+        self.build().map(|stream| {
+            (
+                stream,
+                AcceptInfo {
+                    dc_quic_accept_time: app_queue_time
+                        .saturating_duration_since(kernel_start_time),
+                    app_queue_sojourn_time: now.saturating_duration_since(app_queue_time),
+                },
+            )
+        })
     }
 
     #[inline]

dc/s2n-quic-dc/src/stream/server/accept.rs

@@ -4,7 +4,7 @@
 use crate::{
     event,
     stream::{
-        application::{Builder as StreamBuilder, Stream},
+        application::{AcceptInfo, Builder as StreamBuilder, Stream},
         server::stats,
     },
     sync::mpmc as channel,
@@ -37,7 +37,7 @@ where
 pub async fn accept<Sub>(
     streams: &Receiver<Sub>,
     stats: &stats::Sender,
-) -> io::Result<(Stream<Sub>, SocketAddr)>
+) -> io::Result<(Stream<Sub>, AcceptInfo, SocketAddr)>
 where
     Sub: event::Subscriber,
 {
@@ -49,10 +49,10 @@ where
     })?;
 
     // build the stream inside the application context
-    let (stream, sojourn_time) = stream.accept()?;
-    stats.send(sojourn_time);
+    let (stream, info) = stream.accept()?;
+    stats.send(info.app_queue_sojourn_time);
 
     let remote_addr = stream.peer_addr()?;
 
-    Ok((stream, remote_addr))
+    Ok((stream, info, remote_addr))
 }

dc/s2n-quic-dc/src/stream/server/tokio.rs

@@ -100,6 +100,18 @@ impl<H: Handshake + Clone, S: event::Subscriber + Clone> Server<H, S> {
 
     #[inline]
     pub async fn accept(&self) -> io::Result<(crate::stream::application::Stream<S>, SocketAddr)> {
+        let (stream, _info, addr) = accept::accept(&self.streams, &self.stats).await?;
+        Ok((stream, addr))
+    }
+
+    #[inline]
+    pub async fn accept_with_info(
+        &self,
+    ) -> io::Result<(
+        crate::stream::application::Stream<S>,
+        crate::stream::application::AcceptInfo,
+        SocketAddr,
+    )> {
         accept::accept(&self.streams, &self.stats).await
     }
 

dc/s2n-quic-dc/src/stream/testing.rs

@@ -607,7 +607,9 @@ impl Server {
     }
 
     pub async fn accept(&self) -> io::Result<(Stream, SocketAddr)> {
-        stream_server::accept::accept(&self.receiver, &self.stats).await
+        let (stream, _info, addr) =
+            stream_server::accept::accept(&self.receiver, &self.stats).await?;
+        Ok((stream, addr))
     }
 
     pub fn subscriber(&self) -> Arc<testing::Subscriber> {