test(s2n-quic-dc): shared caches tests (#2876)

Author: mehnazyunus

dc/s2n-quic-dc/src/path/secret/map.rs

@@ -298,6 +298,12 @@ impl Map {
         self.store.test_stop_cleaner();
     }
 
+    #[doc(hidden)]
+    #[cfg(test)]
+    pub fn reset_all_senders(&self) {
+        self.store.reset_all_senders();
+    }
+
     #[doc(hidden)]
     #[cfg(any(test, feature = "testing"))]
     pub fn test_insert(&self, peer: SocketAddr) {

dc/s2n-quic-dc/src/path/secret/map/entry.rs

@@ -300,6 +300,11 @@ impl Entry {
     pub fn application_data(&self) -> &Option<ApplicationData> {
         &self.application_data
     }
+
+    #[cfg(test)]
+    pub fn reset_sender_counter(&self) {
+        self.sender.reset_counter();
+    }
 }
 
 impl receiver::Error {

dc/s2n-quic-dc/src/path/secret/map/state.rs

@@ -1029,6 +1029,14 @@ where
             Ok(None)
         }
     }
+
+    #[cfg(test)]
+    fn reset_all_senders(&self) {
+        let peer_map = self.peers.0.read();
+        for entry in peer_map.iter() {
+            entry.reset_sender_counter();
+        }
+    }
 }
 
 impl<C, S> Drop for State<C, S>

dc/s2n-quic-dc/src/path/secret/map/store.rs

@@ -145,4 +145,7 @@ pub trait Store: 'static + Send + Sync {
         &self,
         session: &dyn s2n_quic_core::crypto::tls::TlsSession,
     ) -> Result<Option<ApplicationData>, ApplicationDataError>;
+
+    #[cfg(test)]
+    fn reset_all_senders(&self);
 }

dc/s2n-quic-dc/src/path/secret/sender.rs

@@ -73,6 +73,11 @@ impl State {
         // Update the key to the new minimum to start at.
         self.current_id.fetch_max(*min_key_id, Ordering::Relaxed);
     }
+
+    #[cfg(test)]
+    pub fn reset_counter(&self) {
+        self.current_id.store(0, Ordering::Relaxed);
+    }
 }
 
 #[test]

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

@@ -781,6 +781,7 @@ where
                 .on_decrypt_success(recv_buffer.into())
                 .is_err()
             {
+                // we just close the stream
                 return Ok(ControlFlow::Continue(())).into();
             };
 
@@ -813,16 +814,12 @@ where
                 recv_buffer,
             );
 
-            let sender = uds::sender::Sender::new()?;
-            let dest_path = self.dest_path.clone();
+            let sender = uds::sender::Sender::new(&self.dest_path)?;
             let tcp_stream = socket.into_std()?;
 
             // FIXME make this a manual Future impl instead of Box
-            let send_future = Box::pin(async move {
-                sender
-                    .send_msg(&buffer, &dest_path, tcp_stream.as_fd())
-                    .await
-            });
+            let send_future =
+                Box::pin(async move { sender.send_msg(&buffer, tcp_stream.as_fd()).await });
 
             let event_data = SocketEventData {
                 credential_id: credentials.id.to_vec(),