fix: make thrift transport fallback pool-aware

.gitignore

@@ -4,3 +4,4 @@ target
 /benchmark/output
 .DS_Store
 .trae
+.codex

Cargo.toml

@@ -141,7 +141,7 @@ tokio-native-tls = "0.3"
 tungstenite = "0.28"
 tokio-tungstenite = "0.28"
 
-shmipc = "0.1"
+shmipc = "0.2.0"
 
 [profile.release]
 opt-level = 3
@@ -154,7 +154,6 @@ panic = 'unwind'
 incremental = false
 overflow-checks = false
 
-# [patch.crates-io]
 # pilota = { git = "https://github.com/cloudwego/pilota.git", branch = "main" }
 # pilota-build = { git = "https://github.com/cloudwego/pilota.git", branch = "main" }
 # pilota-thrift-parser = { git = "https://github.com/cloudwego/pilota.git", branch = "main" }

examples/src/thrift/shmipc/client.rs

@@ -1,11 +1,92 @@
 use std::sync::LazyLock;
 
-use volo_thrift::client::CallOpt;
+use motore::{layer::Layer, service::Service};
+use volo::{context::Context, net::Address};
+use volo_thrift::{
+    ClientError,
+    client::CallOpt,
+    context::ClientContext,
+    transport::{DialPlan, SelectedTransport},
+};
+
+/// A per-request layer that dials shmipc first and falls back to a UDS/TCP address.
+///
+/// It follows the `callee.address()` contract: it sets the primary (shmipc) address on the callee
+/// *before* injecting the [`DialPlan`], so `volo-thrift` can validate that the plan primary matches
+/// the callee address. The pool key is chosen per-attempt from the actually selected address, so a
+/// UDS/TCP fallback transport never pollutes the shmipc key. There is no `.dial_plan(...)` builder;
+/// the plan is always injected through the request context.
+#[derive(Clone)]
+struct ShmipcFallbackLayer {
+    plan: DialPlan,
+}
+
+impl ShmipcFallbackLayer {
+    fn new(shmipc_addr: Address, fallback_addr: Address) -> Self {
+        Self {
+            plan: DialPlan::with_fallback(shmipc_addr, fallback_addr),
+        }
+    }
+}
+
+impl<S> Layer<S> for ShmipcFallbackLayer {
+    type Service = ShmipcFallbackService<S>;
+
+    fn layer(self, inner: S) -> Self::Service {
+        ShmipcFallbackService {
+            inner,
+            plan: self.plan,
+        }
+    }
+}
+
+#[derive(Clone)]
+struct ShmipcFallbackService<S> {
+    inner: S,
+    plan: DialPlan,
+}
+
+impl<S, Req> Service<ClientContext, Req> for ShmipcFallbackService<S>
+where
+    S: Service<ClientContext, Req, Error = ClientError> + Send + Sync + 'static,
+    Req: Send + 'static,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+
+    async fn call(&self, cx: &mut ClientContext, req: Req) -> Result<Self::Response, Self::Error> {
+        // 1. Set the primary address first, then inject the plan (order matters).
+        cx.rpc_info_mut()
+            .callee_mut()
+            .set_address(self.plan.primary().clone());
+        cx.extensions_mut().insert(self.plan.clone());
+
+        let resp = self.inner.call(cx, req).await;
+
+        // The actually selected transport is written back by volo-thrift.
+        if let Some(selected) = cx.extensions().get::<SelectedTransport>() {
+            println!(
+                "selected transport: {} (attempt {})",
+                selected.address(),
+                selected.attempt()
+            );
+        }
+        resp
+    }
+}
 
 static CLIENT: LazyLock<volo_gen::thrift_gen::hello::HelloServiceClient> = LazyLock::new(|| {
-    let uds_path = std::os::unix::net::SocketAddr::from_pathname("/tmp/hello_test.sock").unwrap();
+    let shmipc_path =
+        std::os::unix::net::SocketAddr::from_pathname("/tmp/hello_test.sock").unwrap();
+    let shmipc_addr = Address::from(volo::net::ShmipcAddr(shmipc_path));
+    // Fallback to a plain UDS address when shmipc is unavailable.
+    let fallback_path =
+        std::os::unix::net::SocketAddr::from_pathname("/tmp/hello_fallback.sock").unwrap();
+    let fallback_addr = Address::from(fallback_path);
+
     volo_gen::thrift_gen::hello::HelloServiceClientBuilder::new("hello")
-        .address(volo::net::ShmipcAddr(uds_path))
+        .address(shmipc_addr.clone())
+        .layer_outer_front(ShmipcFallbackLayer::new(shmipc_addr, fallback_addr))
         .build()
 });
 

scripts/clippy-and-test.sh

@@ -31,6 +31,10 @@ run_clippy() {
 	echo_command cargo clippy -p volo-thrift --no-default-features -- --deny warnings
 	echo_command cargo clippy -p volo-thrift --no-default-features --features multiplex -- --deny warnings
 	echo_command cargo clippy -p volo-thrift --no-default-features --features unsafe-codec -- --deny warnings
+	if [ "${RUN_SHMIPC}" = "yes" ]; then
+		echo_command cargo clippy -p volo-thrift --no-default-features --features shmipc -- --deny warnings
+		echo_command cargo clippy -p volo-thrift --no-default-features --features shmipc,multiplex -- --deny warnings
+	fi
 	echo_command cargo clippy -p volo-grpc --no-default-features -- --deny warnings
 	echo_command cargo clippy -p volo-grpc --no-default-features --features rustls -- --deny warnings
 	echo_command cargo clippy -p volo-grpc --no-default-features --features native-tls -- --deny warnings
@@ -53,6 +57,7 @@ run_clippy() {
 run_test() {
 	echo_command cargo test -p volo-thrift
 	echo_command cargo test -p volo-thrift --features shmipc
+	echo_command cargo test -p volo-thrift --features shmipc,multiplex
 	echo_command cargo test -p volo-grpc --features rustls
 	echo_command cargo test -p volo-http --features client,server,http1,query,form,json,tls,cookie,multipart,ws
 	echo_command cargo test -p volo-http --features client,server,http2,query,form,json,tls,cookie,multipart,ws

volo-thrift/src/client/mod.rs

@@ -17,8 +17,6 @@ use motore::{
 };
 use pilota::thrift::TMessageType;
 use tokio::time::Duration;
-#[cfg(feature = "shmipc")]
-use volo::net::shmipc_fallback::ShmipcMakeTransportWithFallback;
 use volo::{
     FastStr,
     client::WithOptService,
@@ -497,42 +495,6 @@ impl<IL, OL, C, Req, Resp, MkT, MkC, LB> ClientBuilder<IL, OL, C, Req, Resp, MkT
         }
     }
 
-    #[cfg(feature = "shmipc")]
-    /// Set the address for the client with shmipc fallback.
-    ///
-    /// Must be called after `.address(shmipc_addr)`.
-    pub fn shmipc_fallback_address<A: Into<Address>>(
-        self,
-        fallback_addr: A,
-    ) -> ClientBuilder<IL, OL, C, Req, Resp, ShmipcMakeTransportWithFallback, MkC, LB> {
-        let shmipc_addr = self
-            .address
-            .expect("Must call .address() before .with_fallback_address()");
-
-        ClientBuilder {
-            config: self.config,
-            pool: self.pool,
-            caller_name: self.caller_name,
-            callee_name: self.callee_name,
-            address: Some(shmipc_addr),
-            inner_layer: self.inner_layer,
-            outer_layer: self.outer_layer,
-            mk_client: self.mk_client,
-            _marker: PhantomData,
-            make_transport: ShmipcMakeTransportWithFallback::new(
-                DefaultMakeTransport::default(),
-                DefaultMakeTransport::default(),
-                fallback_addr.into(),
-            ),
-            make_codec: self.make_codec,
-            mk_lb: self.mk_lb,
-            disable_timeout_layer: self.disable_timeout_layer,
-            enable_biz_error: self.enable_biz_error,
-            #[cfg(feature = "multiplex")]
-            multiplex: self.multiplex,
-        }
-    }
-
     #[doc(hidden)]
     pub fn get_callee_name(&self) -> &FastStr {
         &self.callee_name