Add per-request deadlines to sui-rpc unary calls

Cargo.toml

@@ -33,7 +33,7 @@ serde = "1.0.219"
 serde_derive = "1.0.219"
 
 axum = "0.8.4"
-tower = "0.5.3"
+tower = { version = "0.5.3", features = ["timeout"] }
 tower-http = "0.6"
 reqwest = { version = "0.12", default-features = false, features = ["http2", "json", "rustls-tls"] }
 tokio = { version = "1.46.1", features = ["full"] }

crates/hashi/Cargo.toml

@@ -88,3 +88,4 @@ ed25519-dalek.workspace = true
 test-strategy.workspace = true
 proptest.workspace = true
 tracing-test = "0.2"
+http-body = "1"

crates/hashi/src/cli/commands/balance.rs

@@ -22,7 +22,7 @@ pub async fn run(config: &CliConfig, address: &str, output_format: OutputFormat)
     let btc_type = format!("{}::btc::BTC", config.package_id());
     let btc_struct_tag: StructTag = btc_type.parse().context("Failed to parse hBTC coin type")?;
 
-    let mut client = sui_rpc::Client::new(&config.sui_rpc_url)?;
+    let mut client = crate::sui_rpc_client::new_sui_rpc_client(&config.sui_rpc_url)?;
 
     let response = client
         .state_client()

crates/hashi/src/cli/commands/deposit.rs

@@ -202,7 +202,7 @@ async fn request(
         TxMode::Execute => print_info("Submitting deposit request on Sui..."),
     }
 
-    let mut client = sui_rpc::Client::new(&config.sui_rpc_url)?;
+    let mut client = crate::sui_rpc_client::new_sui_rpc_client(&config.sui_rpc_url)?;
     let outcome = crate::sui_tx_executor::finalize(
         &mut client,
         signer.as_ref(),
@@ -339,7 +339,7 @@ async fn request_all(
     // 3 dynamic-field ops per deposit × 1000 object-runtime cap = 333/PTB.
     const CHUNK_SIZE: usize = 333;
 
-    let sui_client = sui_rpc::Client::new(&config.sui_rpc_url)?;
+    let sui_client = crate::sui_rpc_client::new_sui_rpc_client(&config.sui_rpc_url)?;
     let mut executor = crate::sui_tx_executor::SuiTxExecutor::new(sui_client, signer, hashi_ids);
 
     let txid_address =

crates/hashi/src/cli/commands/withdraw.rs

@@ -86,7 +86,7 @@ async fn request(
         .context("Withdrawal address does not match the configured Bitcoin network")?;
     let destination_bytes = witness_program_from_address(&btc_addr)?;
 
-    let mut client = sui_rpc::Client::new(&config.sui_rpc_url)?;
+    let mut client = crate::sui_rpc_client::new_sui_rpc_client(&config.sui_rpc_url)?;
 
     // A single request supports all tx modes (execute / dry-run /
     // serialize-unsigned) via the builder + finalize path.
@@ -206,7 +206,7 @@ async fn cancel(config: &CliConfig, tx_opts: &TxOptions, request_id: &str) -> Re
         TxMode::Execute => print_info("Cancelling withdrawal..."),
     }
 
-    let mut client = sui_rpc::Client::new(&config.sui_rpc_url)?;
+    let mut client = crate::sui_rpc_client::new_sui_rpc_client(&config.sui_rpc_url)?;
     let outcome = crate::sui_tx_executor::finalize(
         &mut client,
         signer.as_ref(),

crates/hashi/src/cli/mod.rs

@@ -1261,7 +1261,7 @@ pub async fn run_publish(opts: PublishOpts) -> anyhow::Result<()> {
     }
 
     // Connect to RPC
-    let mut client = sui_rpc::Client::new(&opts.sui_rpc_url)?;
+    let mut client = crate::sui_rpc_client::new_sui_rpc_client(&opts.sui_rpc_url)?;
 
     // Publish
     print_info("Publishing ...");
@@ -1323,7 +1323,7 @@ pub async fn run_launch(opts: LaunchOpts) -> anyhow::Result<()> {
 
     print_info(&format!("Sui RPC: {}", opts.sui_rpc_url));
 
-    let mut client = sui_rpc::Client::new(&opts.sui_rpc_url)?;
+    let mut client = crate::sui_rpc_client::new_sui_rpc_client(&opts.sui_rpc_url)?;
 
     // Pre-flight: read the launch state and who would form the initial
     // committee. No hard failures yet — status mode reports every state.
@@ -1606,7 +1606,7 @@ pub async fn run_register(opts: RegisterOpts) -> anyhow::Result<()> {
     if opts.serialize_unsigned {
         // Build the transaction and print as base64 without executing.
         // No private key is required for this path.
-        let mut client = sui_rpc::Client::new(&sui_rpc_url)?;
+        let mut client = crate::sui_rpc_client::new_sui_rpc_client(&sui_rpc_url)?;
         let hashi_ids = config.hashi_ids();
 
         print_info("Building registration transaction ...");
@@ -1643,7 +1643,7 @@ pub async fn run_register(opts: RegisterOpts) -> anyhow::Result<()> {
         }
     }
 
-    let client = sui_rpc::Client::new(&sui_rpc_url)?;
+    let client = crate::sui_rpc_client::new_sui_rpc_client(&sui_rpc_url)?;
     let signer = config.operator_private_key()?;
     let hashi_ids = config.hashi_ids();
     let mut executor = crate::sui_tx_executor::SuiTxExecutor::new(client, signer, hashi_ids);

crates/hashi/src/lib.rs

@@ -30,6 +30,7 @@ pub mod mpc;
 pub mod onchain;
 pub mod publish;
 pub mod storage;
+pub mod sui_rpc_client;
 pub mod sui_tx_executor;
 pub mod tls;
 pub mod utxo_pool;
@@ -574,7 +575,7 @@ impl Hashi {
         use sui_rpc::proto::sui::rpc::v2::GetServiceInfoRequest;
 
         let sui_rpc_url = self.config.sui_rpc.as_deref().unwrap();
-        let mut client = sui_rpc::Client::new(sui_rpc_url)?;
+        let mut client = sui_rpc_client::new_sui_rpc_client(sui_rpc_url)?;
 
         let service_info = client
             .ledger_client()

crates/hashi/src/onchain/mod.rs

@@ -128,7 +128,7 @@ impl OnchainState {
         grpc_max_decoding_message_size: Option<usize>,
         metrics: Option<Arc<crate::metrics::Metrics>>,
     ) -> Result<(Self, Service)> {
-        let mut client = Client::new(sui_rpc_url)?;
+        let mut client = crate::sui_rpc_client::new_sui_rpc_client(sui_rpc_url)?;
         // The scrape client reads the full on-chain state (the largest
         // responses), so it needs the decode limit too — not just `committees`.
         if let Some(limit) = grpc_max_decoding_message_size {

crates/hashi/src/onchain/watcher.rs

@@ -64,7 +64,7 @@ pub async fn watcher(sui_rpc_url: String, state: OnchainState, metrics: Option<A
         // same channel can reuse a wedged h2 connection — the one whose stream
         // just stalled — and silently hang again; a new client forces a clean
         // connection. Reconnects are rare, so the extra setup cost is fine.
-        let mut client = match Client::new(&sui_rpc_url) {
+        let mut client = match crate::sui_rpc_client::new_sui_rpc_client(&sui_rpc_url) {
             Ok(client) => client,
             Err(e) => {
                 tracing::warn!("error creating Sui RPC client: {e}");

crates/hashi/src/sui_rpc_client.rs

@@ -0,0 +1,172 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::time::Duration;
+
+const SUI_RPC_REQUEST_TIMEOUT: Duration = Duration::from_secs(60);
+
+pub fn new_sui_rpc_client(url: &str) -> Result<sui_rpc::Client, tonic::Status> {
+    new_client_with_deadline(url, SUI_RPC_REQUEST_TIMEOUT)
+}
+
+fn new_client_with_deadline(
+    url: &str,
+    deadline: Duration,
+) -> Result<sui_rpc::Client, tonic::Status> {
+    Ok(sui_rpc::Client::new(url)?.request_layer(tower::timeout::TimeoutLayer::new(deadline)))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::convert::Infallible;
+    use std::task::Context;
+    use std::task::Poll;
+    use sui_rpc::proto::sui::rpc::v2::GetServiceInfoRequest;
+    use sui_rpc::proto::sui::rpc::v2::SubscribeCheckpointsRequest;
+
+    #[derive(Clone)]
+    struct HangingLedgerService;
+
+    impl tonic::server::NamedService for HangingLedgerService {
+        const NAME: &'static str = "sui.rpc.v2.LedgerService";
+    }
+
+    impl tower::Service<http::Request<tonic::body::Body>> for HangingLedgerService {
+        type Response = http::Response<tonic::body::Body>;
+        type Error = Infallible;
+        type Future = futures::future::Pending<Result<Self::Response, Self::Error>>;
+
+        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn call(&mut self, _req: http::Request<tonic::body::Body>) -> Self::Future {
+            futures::future::pending()
+        }
+    }
+
+    struct PendingBody;
+
+    impl http_body::Body for PendingBody {
+        type Data = bytes::Bytes;
+        type Error = Infallible;
+
+        fn poll_frame(
+            self: std::pin::Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+        ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
+            Poll::Pending
+        }
+    }
+
+    #[derive(Clone)]
+    struct SilentSubscriptionService;
+
+    impl tonic::server::NamedService for SilentSubscriptionService {
+        const NAME: &'static str = "sui.rpc.v2.SubscriptionService";
+    }
+
+    impl tower::Service<http::Request<tonic::body::Body>> for SilentSubscriptionService {
+        type Response = http::Response<tonic::body::Body>;
+        type Error = Infallible;
+        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
+
+        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn call(&mut self, _req: http::Request<tonic::body::Body>) -> Self::Future {
+            std::future::ready(Ok(http::Response::builder()
+                .status(http::StatusCode::OK)
+                .header(http::header::CONTENT_TYPE, "application/grpc")
+                .body(tonic::body::Body::new(PendingBody))
+                .unwrap()))
+        }
+    }
+
+    async fn spawn_server<S>(svc: S) -> std::net::SocketAddr
+    where
+        S: tower::Service<
+                http::Request<tonic::body::Body>,
+                Response = http::Response<tonic::body::Body>,
+                Error = Infallible,
+            > + tonic::server::NamedService
+            + Clone
+            + Send
+            + Sync
+            + 'static,
+        S::Future: Send + 'static,
+    {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let incoming = futures::stream::unfold(listener, |listener| async move {
+            let result = listener.accept().await.map(|(stream, _)| stream);
+            Some((result, listener))
+        });
+        tokio::spawn(
+            tonic::transport::Server::builder()
+                .add_service(svc)
+                .serve_with_incoming(incoming),
+        );
+        addr
+    }
+
+    #[tokio::test]
+    async fn deadline_fails_stuck_request_instead_of_hanging() {
+        let addr = spawn_server(HangingLedgerService).await;
+        let url = format!("http://{addr}");
+
+        let mut plain = sui_rpc::Client::new(url.as_str()).unwrap();
+        let hung = tokio::time::timeout(
+            Duration::from_secs(2),
+            plain
+                .ledger_client()
+                .get_service_info(GetServiceInfoRequest::default()),
+        )
+        .await;
+        assert!(
+            hung.is_err(),
+            "un-deadlined request should still be pending"
+        );
+
+        let mut client =
+            new_client_with_deadline(url.as_str(), Duration::from_millis(500)).unwrap();
+        let result = tokio::time::timeout(
+            Duration::from_secs(5),
+            client
+                .ledger_client()
+                .get_service_info(GetServiceInfoRequest::default()),
+        )
+        .await
+        .expect("deadline should fire well before 5s");
+        assert!(result.is_err(), "stuck request must surface an error");
+    }
+
+    #[tokio::test]
+    async fn deadline_spares_established_subscription_stream() {
+        let addr = spawn_server(SilentSubscriptionService).await;
+        let mut client = new_client_with_deadline(
+            format!("http://{addr}").as_str(),
+            Duration::from_millis(500),
+        )
+        .unwrap();
+
+        let response = tokio::time::timeout(
+            Duration::from_secs(5),
+            client
+                .subscription_client()
+                .subscribe_checkpoints(SubscribeCheckpointsRequest::default()),
+        )
+        .await
+        .expect("subscription accept should be fast")
+        .expect("subscription should be accepted");
+
+        let mut messages = response.into_inner();
+        let past_deadline = tokio::time::timeout(Duration::from_secs(2), messages.message()).await;
+        assert!(
+            past_deadline.is_err(),
+            "established stream should still be pending past the deadline, got {past_deadline:?}"
+        );
+    }
+}