feat(serf-compio)!: API parity with serf-reactor (+ 3 real defects fixed)

serf-compio/Cargo.toml

@@ -11,8 +11,12 @@ description = "compio-based async driver for the Sans-I/O serf machine"
 default = ["tcp", "tag-regex"]
 # Plain-TCP reliable coordinator.
 tcp = ["serf-proto/tcp", "serf-driver/tcp"]
-# TLS-over-TCP reliable coordinator (implies tcp).
-tls = ["tcp", "serf-proto/tls", "compio/rustls"]
+# TLS-over-TCP reliable coordinator (implies tcp). The record layer comes from
+# serf-proto; a crypto-backend feature (tls-rustls-*) supplies the rustls
+# provider via memberlist-proto (serf-proto exposes no rustls provider itself).
+tls = ["tcp", "serf-proto/tls", "serf-driver/tls", "compio/rustls"]
+tls-rustls-ring = ["tls", "memberlist-proto/tls-rustls-ring"]
+tls-rustls-aws-lc-rs = ["tls", "memberlist-proto/tls-rustls-aws-lc-rs"]
 # QUIC coordinator.
 quic = ["serf-proto/quic", "serf-driver/quic"]
 quic-rustls-ring = ["quic", "serf-proto/quic-rustls-ring", "serf-driver/quic-rustls-ring"]
@@ -28,6 +32,10 @@ chacha20-poly1305 = [
 ]
 # Regex-backed tag-filter matching.
 tag-regex = ["serf-proto/tag-regex", "serf-driver/tag-regex"]
+# Test-only fault-injection surface (forwards serf-proto's `test` feature: a
+# `MessageDropper` installed via `VoidDelegate::with_message_dropper`). Compiled
+# out of every non-test build.
+test = ["serf-proto/test"]
 # Emit `tracing` spans around the public driver operations.
 tracing = ["dep:tracing", "serf-driver/tracing"]
 # Optional config layering: `serde` adds Serialize/Deserialize; `clap` adds CLI flags.
@@ -112,7 +120,7 @@ required-features = ["tcp"]
 # delegate, snapshot persistence, user event, query round-trip, graceful leave).
 [[test]]
 name = "tls"
-required-features = ["tls"]
+required-features = ["tls-rustls-ring"]
 
 # The real-node QUIC test suite: loopback nodes over a quinn config bundle
 # driving the QUIC pump's whole command surface (join/converge, user event, query
@@ -123,5 +131,19 @@ name = "quic"
 required-features = ["quic-rustls-ring"]
 
 [package.metadata.docs.rs]
-all-features = true
+# A single coherent crypto-provider set (ring, not aws-lc-rs) so the doc build
+# documents every gated item without pulling conflicting rustls providers.
+features = [
+  "tls-rustls-ring",
+  "quic-rustls-ring",
+  "coordinates",
+  "aes-gcm",
+  "chacha20-poly1305",
+  "tag-regex",
+  "tracing",
+  "serde",
+  "clap",
+  "dns",
+  "getifs",
+]
 rustdoc-args = ["--cfg", "docsrs"]

serf-compio/src/delegate/mod.rs

@@ -144,6 +144,16 @@ pub trait Delegate:
   type Id;
   /// Address type — always `SocketAddr` in compio.
   type Address;
+
+  /// Test-only inbound message-drop hook. The driver installs the returned
+  /// [`MessageDropper`](serf_proto::MessageDropper) on the machine so a test can
+  /// drop selected inbound membership messages; `None` (the default) drops
+  /// nothing. Gated behind the `test` feature — no production use.
+  #[cfg(feature = "test")]
+  #[cfg_attr(docsrs, doc(cfg(feature = "test")))]
+  fn message_dropper(&self) -> Option<std::sync::Arc<dyn serf_proto::MessageDropper>> {
+    None
+  }
 }
 
 /// Observer the driver notifies after it rotates the LIVE wire keyring, so an

serf-compio/src/delegate/tests.rs

@@ -45,3 +45,34 @@ fn a_sync_predicate_satisfies_the_merge_delegate() {
     "a permit-all predicate admits the exchange"
   );
 }
+
+/// A BOXED machine merge delegate satisfies the trait — the shape the node
+/// constructors' `Option<Box<dyn MergeDelegate<..>>>` slot takes, so this pins
+/// that the re-exported trait and its `Box` blanket impl compose.
+#[cfg(any(feature = "tcp", feature = "quic"))]
+#[test]
+fn boxed_merge_delegate_satisfies_trait() {
+  struct AcceptAll;
+  impl MergeDelegate<SmolStr, SocketAddr> for AcceptAll {
+    fn notify_merge(
+      &self,
+      _peers: memberlist_proto::MaybeOwned<
+        '_,
+        [memberlist_proto::typed::NodeState<SmolStr, SocketAddr>],
+      >,
+    ) -> bool {
+      true
+    }
+  }
+  fn assert_merge<T>(t: &T) -> bool
+  where
+    T: MergeDelegate<SmolStr, SocketAddr>,
+  {
+    t.notify_merge(memberlist_proto::MaybeOwned::Borrowed(&[]))
+  }
+  let boxed: Box<dyn MergeDelegate<SmolStr, SocketAddr>> = Box::new(AcceptAll);
+  assert!(
+    assert_merge(&boxed),
+    "the boxed predicate's verdict is the one the machine acts on"
+  );
+}

serf-compio/src/delegate/void.rs

@@ -21,6 +21,10 @@ use super::KeyringDelegate;
 #[cfg_attr(docsrs, doc(cfg(any(feature = "tcp", feature = "quic"))))]
 pub struct VoidDelegate<I, A> {
   _phantom: PhantomData<fn(I, A)>,
+  /// Test-only inbound message-drop hook returned via
+  /// [`Delegate::message_dropper`]; `None` in every real build.
+  #[cfg(feature = "test")]
+  message_dropper: Option<std::sync::Arc<dyn serf_proto::MessageDropper>>,
 }
 
 #[cfg(any(feature = "tcp", feature = "quic"))]
@@ -30,8 +34,24 @@ impl<I, A> VoidDelegate<I, A> {
   pub const fn new() -> Self {
     Self {
       _phantom: PhantomData,
+      #[cfg(feature = "test")]
+      message_dropper: None,
     }
   }
+
+  /// Attach a test-only [`MessageDropper`](serf_proto::MessageDropper) surfaced
+  /// through [`Delegate::message_dropper`], so a test node drops selected
+  /// inbound membership messages. Test fault injection only.
+  #[cfg(feature = "test")]
+  #[cfg_attr(docsrs, doc(cfg(feature = "test")))]
+  #[must_use]
+  pub fn with_message_dropper(
+    mut self,
+    dropper: std::sync::Arc<dyn serf_proto::MessageDropper>,
+  ) -> Self {
+    self.message_dropper = Some(dropper);
+    self
+  }
 }
 
 #[cfg(any(feature = "tcp", feature = "quic"))]
@@ -77,6 +97,11 @@ where
 {
   type Id = I;
   type Address = A;
+
+  #[cfg(feature = "test")]
+  fn message_dropper(&self) -> Option<std::sync::Arc<dyn serf_proto::MessageDropper>> {
+    self.message_dropper.clone()
+  }
 }
 
 /// A keyring delegate that persists nothing.

serf-compio/src/driver/options/tests.rs

@@ -476,3 +476,21 @@ fn tracing_forwards_to_the_shared_engines() {
     )
   };
 }
+
+/// `Default` and `new()` are the same source of truth for the stream knobs, and
+/// the defaults they produce are themselves admissible — a node built with no
+/// explicit stream configuration passes the same `validate` gate `Transport::new`
+/// applies.
+#[test]
+fn stream_transport_options_default_matches_new() {
+  let d = StreamTransportOptions::default();
+  let n = StreamTransportOptions::new();
+  assert_eq!(d.dial_timeout(), n.dial_timeout());
+  assert_eq!(d.close_timeout(), n.close_timeout());
+  assert_eq!(d.bridge_inbound_cap(), n.bridge_inbound_cap());
+  assert_eq!(d.bridge_recv_buf_len(), n.bridge_recv_buf_len());
+  assert!(
+    d.validate().is_ok(),
+    "the default stream knobs are admissible"
+  );
+}