fix(socket): fix subscriber reconnects with QUIC

msg-socket/src/pub/mod.rs

@@ -378,7 +378,7 @@ mod tests {
         // Try to connect and subscribe before the publisher is up
         sub_socket.connect("0.0.0.0:6662").await.unwrap();
         sub_socket.subscribe("HELLO".to_string()).await.unwrap();
-        tokio::time::sleep(Duration::from_millis(500)).await;
+        tokio::time::sleep(Duration::from_millis(1000)).await;
 
         pub_socket.bind("0.0.0.0:6662").await.unwrap();
         tokio::time::sleep(Duration::from_millis(2000)).await;

msg-socket/src/sub/driver.rs

@@ -112,15 +112,25 @@ where
     /// De-activates a publisher by setting it to [`PublisherState::Inactive`]. This will initialize
     /// the backoff stream.
     fn reset_publisher(&mut self, addr: SocketAddr) {
+        tracing::debug!("Resetting publisher at {addr:?}");
         self.publishers.insert(
             addr,
             PublisherState::Inactive {
                 addr,
-                backoff: ExponentialBackoff::new(Duration::from_millis(10), 16),
+                backoff: ExponentialBackoff::new(Duration::from_millis(50), 16),
             },
         );
     }
 
+    /// Returns true if we're already connected to the given publisher address.
+    fn is_connected(&self, addr: &SocketAddr) -> bool {
+        if let Some(PublisherState::Active { .. }) = self.publishers.get(addr) {
+            return true;
+        }
+
+        false
+    }
+
     /// Subscribes to a topic on all publishers.
     fn subscribe(&mut self, topic: String) {
         let mut inactive = Vec::new();
@@ -273,6 +283,12 @@ where
     }
 
     fn on_connection(&mut self, addr: SocketAddr, io: T::Io) {
+        if self.is_connected(&addr) {
+            // We're already connected to this publisher
+            warn!(%addr, "Already connected to publisher");
+            return;
+        }
+
         // This should spawn a new task tied to this connection, and
         debug!("Connection to {} established, spawning session", addr);