WIP

Cargo.toml

@@ -31,5 +31,9 @@ smol_str = "0.2"
 smallvec = "1"
 rand = "0.8"
 
+# memberlist-types = { version = "0.2", path = "../memberlist/types", default-features = false }
+# memberlist-core = { version = "0.2", path = "../memberlist/core", default-features = false }
+# memberlist = { version = "0.2", path = "../memberlist/memberlist", default-features = false }
+
 ruserf-core = { path = "core", version = "0.1.0", default-features = false }
 ruserf-types = { path = "types", version = "0.1.0", default-features = false }

core/src/lib.rs

@@ -98,8 +98,8 @@ pub mod tests {
     use std::sync::Once;
     static TRACE: Once = Once::new();
     TRACE.call_once(|| {
-      let filter =
-        std::env::var("RUSERF_TESTING_LOG").unwrap_or_else(|_| "ruserf_core=trace".to_owned());
+      let filter = std::env::var("RUSERF_TESTING_LOG")
+        .unwrap_or_else(|_| "ruserf_core=info,memberlist_core=debug".to_owned());
       memberlist_core::tracing::subscriber::set_global_default(
         tracing_subscriber::fmt::fmt()
           .without_time()
@@ -120,7 +120,7 @@ pub mod tests {
     B: FnOnce(F) -> F::Output,
     F: std::future::Future<Output = ()>,
   {
-    // initialize_tests_tracing();
+    initialize_tests_tracing();
     block_on(fut);
   }
 }

core/src/serf/api.rs

@@ -589,7 +589,7 @@ where
       // channel signals that we are cleaned up outside of Serf.
       *s = SerfState::Shutdown;
     }
-
+    tracing::error!("debug: shutdown memberlist");
     self.inner.memberlist.shutdown().await?;
     self.inner.shutdown_tx.close();
 
@@ -598,6 +598,7 @@ where
       snap.wait().await;
     }
 
+    tracing::error!("debug: shutdown serf handles");
     loop {
       if let Ok(mut handles) = self.inner.handles.try_borrow_mut() {
         let mut futs = core::mem::take(&mut *handles);

core/src/serf/base/tests/serf.rs

@@ -725,10 +725,6 @@ pub async fn serf_name_resolution<T>(
       panic!("timed out");
     }
   }
-
-  for s in serfs.iter() {
-    s.shutdown().await.unwrap();
-  }
 }
 
 /// Unit test for serf local member

core/src/serf/base/tests/serf/event.rs

@@ -867,10 +867,6 @@ where
   }
   assert_eq!(acks.len(), 2, "missing acks {acks:?}");
   assert_eq!(responses.len(), 1, "missing responses {responses:?}");
-
-  for s in serfs {
-    let _ = s.shutdown().await;
-  }
 }
 
 /// Unit test for serf query filter
@@ -992,10 +988,6 @@ pub async fn serf_query_filter<T>(
 
   assert_eq!(acks.len(), 1, "missing acks {acks:?}");
   assert_eq!(responses.len(), 1, "missing responses {responses:?}");
-
-  for s in serfs {
-    let _ = s.shutdown().await;
-  }
 }
 
 /// Unit test for serf query deduplicate

core/src/serf/base/tests/serf/reconnect.rs

@@ -143,10 +143,6 @@ pub async fn serf_reconnect_same_ip<T, R, F>(
     .collect(),
   )
   .await;
-
-  for s in serfs.iter() {
-    s.shutdown().await.unwrap();
-  }
 }
 
 #[derive(Clone)]

core/src/serf/base/tests/serf/remove.rs

@@ -86,10 +86,6 @@ pub async fn serf_remove_failed_node<T>(
       panic!("Failed to remove failed node");
     }
   }
-
-  for s in serfs.iter() {
-    s.shutdown().await.unwrap();
-  }
 }
 
 /// Unit test for remove failed node prune
@@ -157,10 +153,6 @@ pub async fn serf_remove_failed_node_prune<T>(
   serfs.swap(1, 2);
 
   wait_until_num_nodes(2, &serfs[..2]).await;
-
-  for s in serfs.iter() {
-    s.shutdown().await.unwrap();
-  }
 }
 
 /// Unit test for remove failed node ourself