feat(s2n-quic-tests): implement a blocklist for test events (#2872)

Author: boquan-fang

quic/s2n-quic-tests/README.md

@@ -126,11 +126,31 @@ Implement the `s2n_quic_core::packet::interceptor::Interceptor` trait and config
 
 ```rust
 let client = Client::builder()
-    .with_io(handle.builder().build().unwrap())?  
-    .with_packet_interceptor(interceptor)?  
+    .with_io(handle.builder().build().unwrap())?
+    .with_event(tracing_events(true, model.clone()))?
+    .with_packet_interceptor(interceptor)?
     .start()?;
 ```
 
+### Blocklist Event Subscriber
+
+The test suite includes a `BlocklistSubscriber` utility that can be used to detect and fail tests when specific unwanted events occur. This subscriber works by panicking when it encounters events that have been added to the blocklist, such as certain types of packet losses.
+
+To use the blocklist subscriber in tests:
+
+```rust
+// Use with a client or server
+let client = Client::builder()
+    .with_io(handle.builder().build().unwrap())?
+    .with_tls(certificates::CERT_PEM)?
+    .with_event(tracing_events(true, model.clone()))?
+    .start()?;
+```
+
+When the `with_blocklist` parameter is set to `true`, the `tracing_events` function returns both the standard tracing subscriber and the blocklist subscriber. The standard subscriber logs events as usual, while the blocklist subscriber will cause a test to fail immediately if a blocklisted event is encountered. If certain unwanted events are expected during the test, switching the `with_blocklist` parameter to false to disable the blocklist feature.
+
+This is particularly useful for tests that need to verify that certain error conditions don't occur, as the test will fail early with a clear error message rather than continuing with potentially incorrect behavior.
+
 ### Common Setup
 
 `src/lib.rs` provides common utilities for setting up test clients and servers with various configurations.

quic/s2n-quic-tests/src/lib.rs

@@ -4,8 +4,8 @@
 use s2n_quic::{
     client::Connect,
     provider::{
-        event,
-        io::testing::{primary, spawn, Handle, Result},
+        event::{self, events},
+        io::testing::{primary, spawn, Handle, Model, Result},
     },
     stream::PeerStream,
     Client, Server,
@@ -21,7 +21,124 @@ mod tests;
 
 pub static SERVER_CERTS: (&str, &str) = (certificates::CERT_PEM, certificates::KEY_PEM);
 
-pub fn tracing_events() -> event::tracing::Subscriber {
+/// A subscriber that panics when a blocklisted event is encountered
+pub struct BlocklistSubscriber {
+    blocklist_enabled: bool,
+    network_env: Model,
+}
+
+impl BlocklistSubscriber {
+    pub fn new(blocklist_enabled: bool, network_env: Model) -> Self {
+        Self {
+            blocklist_enabled,
+            network_env,
+        }
+    }
+
+    pub fn max_udp_payload(&self) -> u16 {
+        self.network_env.max_udp_payload()
+    }
+}
+
+impl event::Subscriber for BlocklistSubscriber {
+    type ConnectionContext = ();
+
+    fn create_connection_context(
+        &mut self,
+        _meta: &events::ConnectionMeta,
+        _info: &events::ConnectionInfo,
+    ) -> Self::ConnectionContext {
+    }
+
+    fn on_datagram_dropped(
+        &mut self,
+        _context: &mut Self::ConnectionContext,
+        _meta: &events::ConnectionMeta,
+        event: &events::DatagramDropped,
+    ) {
+        if self.blocklist_enabled {
+            panic!(
+                "Blacklisted datagram dropped event encountered: {:?}",
+                event
+            );
+        }
+    }
+
+    fn on_packet_dropped(
+        &mut self,
+        _context: &mut Self::ConnectionContext,
+        _meta: &events::ConnectionMeta,
+        event: &events::PacketDropped,
+    ) {
+        if matches!(
+            event,
+            events::PacketDropped {
+                reason: events::PacketDropReason::DecryptionFailed { .. }
+                    | events::PacketDropReason::UnprotectFailed { .. }
+                    | events::PacketDropReason::VersionMismatch { .. }
+                    | events::PacketDropReason::UndersizedInitialPacket { .. }
+                    | events::PacketDropReason::InitialConnectionIdInvalidSpace { .. },
+                ..
+            }
+        ) && self.blocklist_enabled
+        {
+            panic!("Blocklisted packet dropped event encountered: {:?}", event);
+        }
+    }
+
+    fn on_packet_lost(
+        &mut self,
+        _context: &mut Self::ConnectionContext,
+        _meta: &events::ConnectionMeta,
+        event: &events::PacketLost,
+    ) {
+        // packet which is smaller than the MTU should not be lost
+        if event.bytes_lost < self.max_udp_payload() && self.blocklist_enabled {
+            panic!("Bytes lost is {} and max udp payload is {}\nBlocklisted packet lost event encountered: {:?}", event.bytes_lost, self.max_udp_payload(), event);
+        }
+    }
+
+    fn on_platform_tx_error(
+        &mut self,
+        _meta: &events::EndpointMeta,
+        event: &events::PlatformTxError,
+    ) {
+        if self.blocklist_enabled {
+            panic!(
+                "Blocklisted platform tx error event encountered: {:?}",
+                event
+            );
+        }
+    }
+
+    fn on_platform_rx_error(
+        &mut self,
+        _meta: &events::EndpointMeta,
+        event: &events::PlatformRxError,
+    ) {
+        if self.blocklist_enabled {
+            panic!(
+                "Blocklisted platform rx error event encountered: {:?}",
+                event
+            );
+        }
+    }
+
+    fn on_endpoint_datagram_dropped(
+        &mut self,
+        _meta: &events::EndpointMeta,
+        event: &events::EndpointDatagramDropped,
+    ) {
+        if self.blocklist_enabled {
+            panic!(
+                "Blocklisted endpoint datagram dropped event encountered: {:?}",
+                event
+            );
+        }
+    }
+}
+
+pub fn tracing_events(with_blocklist: bool, network_env: Model) -> impl event::Subscriber {
     use std::sync::Once;
 
     static TRACING: Once = Once::new();
@@ -59,7 +176,10 @@ pub fn tracing_events() -> event::tracing::Subscriber {
             .init();
     });
 
-    event::tracing::Subscriber::default()
+    (
+        event::tracing::Subscriber::default(),
+        BlocklistSubscriber::new(with_blocklist, network_env),
+    )
 }
 
 pub fn start_server(mut server: Server) -> Result<SocketAddr> {
@@ -96,22 +216,27 @@ pub fn start_server(mut server: Server) -> Result<SocketAddr> {
     Ok(server_addr)
 }
 
-pub fn server(handle: &Handle) -> Result<SocketAddr> {
-    let server = build_server(handle)?;
+pub fn server(handle: &Handle, network_env: Model) -> Result<SocketAddr> {
+    let server = build_server(handle, network_env)?;
     start_server(server)
 }
 
-pub fn build_server(handle: &Handle) -> Result<Server> {
+pub fn build_server(handle: &Handle, network_env: Model) -> Result<Server> {
     Ok(Server::builder()
         .with_io(handle.builder().build().unwrap())?
         .with_tls(SERVER_CERTS)?
-        .with_event(tracing_events())?
+        .with_event(tracing_events(true, network_env))?
         .with_random(Random::with_seed(123))?
         .start()?)
 }
 
-pub fn client(handle: &Handle, server_addr: SocketAddr) -> Result {
-    let client = build_client(handle)?;
+pub fn client(
+    handle: &Handle,
+    server_addr: SocketAddr,
+    network_env: Model,
+    with_blocklist: bool,
+) -> Result {
+    let client = build_client(handle, network_env, with_blocklist)?;
     start_client(client, server_addr, Data::new(10_000))
 }
 
@@ -146,21 +271,15 @@ pub fn start_client(client: Client, server_addr: SocketAddr, data: Data) -> Resu
     Ok(())
 }
 
-pub fn build_client(handle: &Handle) -> Result<Client> {
+pub fn build_client(handle: &Handle, network_env: Model, with_blocklist: bool) -> Result<Client> {
     Ok(Client::builder()
         .with_io(handle.builder().build().unwrap())?
         .with_tls(certificates::CERT_PEM)?
-        .with_event(tracing_events())?
+        .with_event(tracing_events(with_blocklist, network_env))?
         .with_random(Random::with_seed(123))?
         .start()?)
 }
 
-pub fn client_server(handle: &Handle) -> Result<SocketAddr> {
-    let addr = server(handle)?;
-    client(handle, addr)?;
-    Ok(addr)
-}
-
 pub struct Random {
     inner: rand_chacha::ChaCha8Rng,
 }

quic/s2n-quic-tests/src/tests/blackhole.rs

@@ -5,18 +5,22 @@ use super::*;
 
 fn blackhole(model: Model, blackhole_duration: Duration) {
     test(model.clone(), |handle| {
+        let model_for_spawn = model.clone();
         spawn(async move {
             // switch back and forth between blackhole and not
             loop {
                 delay(blackhole_duration).await;
                 // drop all packets
-                model.set_drop_rate(1.0);
+                model_for_spawn.set_drop_rate(1.0);
 
                 delay(blackhole_duration).await;
-                model.set_drop_rate(0.0);
+                model_for_spawn.set_drop_rate(0.0);
             }
         });
-        client_server(handle)
+
+        let addr = server(handle, model.clone())?;
+        client(handle, addr, model.clone(), true)?;
+        Ok(addr)
     })
     .unwrap();
 }

quic/s2n-quic-tests/src/tests/buffer_limit.rs

@@ -30,7 +30,7 @@ fn buffer_limit_test() {
     let client_hello_subscriber = recorder::TlsClientHello::new();
     let client_hello_event = client_hello_subscriber.events();
 
-    test(model, |handle| {
+    test(model.clone(), |handle| {
         let server = tls::Server::builder()
             .with_application_protocols(["h3"].iter())
             .unwrap()
@@ -43,7 +43,7 @@ fn buffer_limit_test() {
             .with_io(handle.builder().build()?)?
             .with_tls(server)?
             .with_event((
-                tracing_events(),
+                tracing_events(true, model.clone()),
                 (client_hello_subscriber, connection_closed_subscriber),
             ))?
             .with_random(Random::with_seed(456))?
@@ -66,7 +66,7 @@ fn buffer_limit_test() {
         let client = Client::builder()
             .with_io(handle.builder().build()?)?
             .with_tls(client)?
-            .with_event(tracing_events())?
+            .with_event(tracing_events(true, model.clone()))?
             .with_random(Random::with_seed(456))?
             .start()?;
 

quic/s2n-quic-tests/src/tests/chain.rs

@@ -78,11 +78,11 @@ where
     let model = Model::default();
     model.set_delay(Duration::from_millis(50));
 
-    test(model, |handle| {
+    test(model.clone(), |handle| {
         let server = Server::builder()
             .with_io(handle.builder().build()?)?
             .with_tls(build_server_mtls_provider(certificates::MTLS_CA_CERT)?)?
-            .with_event((Chain, tracing_events()))?
+            .with_event((Chain, tracing_events(true, model.clone())))?
             .start()?;
         let (send, server_chain) = tokio::sync::mpsc::channel(1);
         let server_chain = Arc::new(tokio::sync::Mutex::new(server_chain));
@@ -92,7 +92,7 @@ where
         let client = Client::builder()
             .with_io(handle.builder().build().unwrap())?
             .with_tls(build_client_mtls_provider(certificates::MTLS_CA_CERT)?)?
-            .with_event((Chain, tracing_events()))?
+            .with_event((Chain, tracing_events(true, model.clone())))?
             .start()?;
 
         // show it working for several connections

quic/s2n-quic-tests/src/tests/client_handshake_confirm.rs

@@ -137,12 +137,12 @@ where
     let model = Model::default();
     model.set_delay(Duration::from_millis(50));
 
-    test(model, |handle| {
+    test(model.clone(), |handle| {
         let server_tls = build_server_mtls_provider(server_cert)?;
         let server = Server::builder()
             .with_io(handle.builder().build()?)?
             .with_tls(server_tls)?
-            .with_event(tracing_events())?
+            .with_event(tracing_events(true, model.clone()))?
             .with_random(Random::with_seed(456))?
             .start()?;
 
@@ -152,7 +152,7 @@ where
         let client = Client::builder()
             .with_io(handle.builder().build().unwrap())?
             .with_tls(client_tls)?
-            .with_event((ClientConfirm, tracing_events()))?
+            .with_event((ClientConfirm, tracing_events(true, model.clone())))?
             .with_random(Random::with_seed(456))?
             .start()?;
 

quic/s2n-quic-tests/src/tests/connection_limits.rs

@@ -30,16 +30,20 @@ fn connection_limits() {
     }
 
     let model = Model::default();
-    test(model, |handle| {
+    test(model.clone(), |handle| {
         let server = Server::builder()
             .with_io(handle.builder().build()?)?
+            .with_event(tracing_events(true, model.clone()))?
             .with_tls(SERVER_CERTS)?
+            .with_event(tracing_events(true, model.clone()))?
             .with_limits(LimitsProvider)?
             .start()?;
 
         let client = Client::builder()
             .with_io(handle.builder().build().unwrap())?
+            .with_event(tracing_events(true, model.clone()))?
             .with_tls(certificates::CERT_PEM)?
+            .with_event(tracing_events(true, model.clone()))?
             .start()?;
         let addr = start_server(server)?;
         start_client(client, addr, Data::new(1000))?;