From 851bde5edb408377c3d4a17871ec950cdc90b8e2 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Tue, 14 Jul 2026 21:02:09 -0500 Subject: [PATCH 1/9] sui-rpc: floor hyper at 1.10 to exclude broken h2 versions h2 versions before 0.4.14 have several stream-cancellation flow-control accounting bugs (hyperium/h2#893, #896, #897, #898, and #913, fixed across 0.4.14 and 0.4.15) that can permanently wedge a multiplexed gRPC connection: leaked connection send-window capacity, RST_STREAM frames that are never transmitted, and permanently missed `poll_capacity` wakeups. Add an explicit hyper 1.10 floor to sui-rpc (hyper 1.10 raised its h2 floor to 0.4.14 for the same reason) so downstream consumers cannot resolve a broken h2. The lockfile is not tracked in this repo, so the manifest floor is the durable guard. --- crates/sui-rpc/Cargo.toml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/crates/sui-rpc/Cargo.toml b/crates/sui-rpc/Cargo.toml index 68ae029cf3..ba670d6a80 100644 --- a/crates/sui-rpc/Cargo.toml +++ b/crates/sui-rpc/Cargo.toml @@ -35,6 +35,11 @@ sui-crypto = { version = "0.3.0", path = "../sui-crypto", default-features = fal # dependencies for the protobuf and gRPC definitions bytes = "1.10" +# Not used directly; this floors hyper at 1.10, whose h2 requirement +# excludes the h2 versions with stream-cancellation flow-control +# accounting bugs (hyperium/h2#893, #896, #897, #898, and #913) that can +# permanently wedge a multiplexed gRPC connection. +hyper = { version = "1.10", default-features = false } tonic = { version = "0.14.2", default-features = false, features = ["channel", "codegen", "tls-ring", "tls-webpki-roots", "zstd"] } tonic-prost = "0.14.2" prost = "0.14.1" From 87e73564ce34b10b27ba1300147286b3aa2d2d69 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Tue, 14 Jul 2026 21:04:21 -0500 Subject: [PATCH 2/9] sui-rpc: set explicit HTTP/2 flow-control windows on the client Every RPC made through a Client (and all of its clones) is multiplexed over a single HTTP/2 connection, so the connection-level receive window is shared by all in-flight responses. A streaming response that the application holds without polling pins up to a full stream window of that shared budget, and with hyper's client defaults (2 MiB stream / 5 MiB connection) about three stalled streams exhaust the connection window. From then on every RPC on the channel hangs indefinitely -- response HEADERS are not flow-controlled and still arrive, but response DATA can never be delivered -- while TCP and HTTP/2 keepalives stay healthy, so nothing times out. Client::new now sets the per-stream window to 2 MiB explicitly and raises the connection window to 64 MiB, so ~32 concurrently stalled streams are needed to starve the connection instead of ~3. Add with_initial_stream_window_size and with_initial_connection_window_size overrides, and document that Client::from_endpoint bypasses these defaults and that http2_adaptive_window is not a substitute (adaptive mode starts the connection window at the 64 KiB spec default, making starvation worse). --- crates/sui-rpc/src/client/mod.rs | 71 +++++++++++++++++++++++++++++++- 1 file changed, 69 insertions(+), 2 deletions(-) diff --git a/crates/sui-rpc/src/client/mod.rs b/crates/sui-rpc/src/client/mod.rs index ac06bca280..c110937dfd 100644 --- a/crates/sui-rpc/src/client/mod.rs +++ b/crates/sui-rpc/src/client/mod.rs @@ -51,9 +51,23 @@ const DEFAULT_TCP_KEEPALIVE_RETRIES: u32 = 3; const DEFAULT_HTTP2_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(5); const DEFAULT_HTTP2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(20); +// All RPCs made through a `Client` (and all of its clones) are multiplexed +// over a single HTTP/2 connection, so the connection-level receive window is +// shared by every in-flight response. A streaming response that the +// application holds without polling pins up to a full stream window of that +// shared budget; once the connection window is exhausted, every RPC on the +// channel hangs indefinitely while TCP and HTTP/2 keepalives stay healthy. +// hyper's client defaults (2 MiB stream / 5 MiB connection) let ~3 stalled +// streams starve the connection. Keep the stream window at hyper's default +// but raise the connection window so ~32 concurrently stalled streams are +// needed instead. +const DEFAULT_HTTP2_STREAM_WINDOW_SIZE: u32 = 2 * 1024 * 1024; +const DEFAULT_HTTP2_CONNECTION_WINDOW_SIZE: u32 = 64 * 1024 * 1024; + #[derive(Clone)] pub struct Client { uri: http::Uri, + endpoint: tonic::transport::Endpoint, channel: tonic::transport::Channel, headers: HeadersInterceptor, max_decoding_message_size: Option, @@ -78,11 +92,26 @@ impl Client { /// URL for the public-good, Sui Foundation provided archive for testnet. pub const TESTNET_ARCHIVE: &str = "https://archive.testnet.sui.io"; + /// Build a client from a fully custom [`tonic::transport::Endpoint`]. + /// + /// This bypasses every transport default that [`Client::new`] applies, + /// including the HTTP/2 flow-control windows that protect the shared + /// connection from starvation by stalled streaming responses. Prefer + /// [`Client::new`] plus the `with_*` configuration methods unless an + /// endpoint setting is needed that the client does not expose. + /// + /// In particular, do not rely on + /// [`http2_adaptive_window`](tonic::transport::Endpoint::http2_adaptive_window) + /// as a substitute for large static windows: with adaptive windowing, + /// hyper starts the connection window at the 64 KiB HTTP/2 spec default + /// until bandwidth-delay probing ramps up, so a single stalled stream can + /// starve the whole connection. pub fn from_endpoint(endpoint: &tonic::transport::Endpoint) -> Self { let uri = endpoint.uri().clone(); let channel = endpoint.connect_lazy(); Self { uri, + endpoint: endpoint.clone(), channel, headers: Default::default(), max_decoding_message_size: None, @@ -108,17 +137,20 @@ impl Client { .map_err(tonic::Status::from_error)?; } - let channel = endpoint + let endpoint = endpoint .connect_timeout(DEFAULT_CONNECT_TIMEOUT) .tcp_keepalive(Some(DEFAULT_TCP_KEEPALIVE_IDLE)) .tcp_keepalive_interval(Some(DEFAULT_TCP_KEEPALIVE_INTERVAL)) .tcp_keepalive_retries(Some(DEFAULT_TCP_KEEPALIVE_RETRIES)) .http2_keep_alive_interval(DEFAULT_HTTP2_KEEP_ALIVE_INTERVAL) .keep_alive_timeout(DEFAULT_HTTP2_KEEP_ALIVE_TIMEOUT) - .connect_lazy(); + .initial_stream_window_size(DEFAULT_HTTP2_STREAM_WINDOW_SIZE) + .initial_connection_window_size(DEFAULT_HTTP2_CONNECTION_WINDOW_SIZE); + let channel = endpoint.connect_lazy(); Ok(Self { uri, + endpoint, channel, headers: Default::default(), max_decoding_message_size: None, @@ -126,6 +158,41 @@ impl Client { }) } + /// Set the HTTP/2 per-stream receive window, in bytes. + /// + /// This bounds how much unread response data a single RPC can buffer + /// before the server must stop sending on that stream. It also bounds how + /// much of the shared connection window (see + /// [`with_initial_connection_window_size`](Self::with_initial_connection_window_size)) + /// one stalled stream can pin. Defaults to 2 MiB. + /// + /// This rebuilds the underlying channel, so it must be called before the + /// client is used or cloned; earlier clones keep the previous + /// configuration. + pub fn with_initial_stream_window_size(mut self, size: u32) -> Self { + self.endpoint = self.endpoint.initial_stream_window_size(size); + self.channel = self.endpoint.connect_lazy(); + self + } + + /// Set the HTTP/2 connection-level receive window, in bytes. + /// + /// This window is shared by every RPC multiplexed over the client's + /// single HTTP/2 connection, including all clones of the client. Response + /// data that the application has not yet read counts against it, so it + /// determines how many concurrently stalled streaming responses it takes + /// to starve the connection and hang every other RPC on it. Defaults to + /// 64 MiB (~32 stalled streams at the default 2 MiB stream window). + /// + /// This rebuilds the underlying channel, so it must be called before the + /// client is used or cloned; earlier clones keep the previous + /// configuration. + pub fn with_initial_connection_window_size(mut self, size: u32) -> Self { + self.endpoint = self.endpoint.initial_connection_window_size(size); + self.channel = self.endpoint.connect_lazy(); + self + } + pub fn with_headers(mut self, headers: HeadersInterceptor) -> Self { self.headers = headers; self From b9ab22f7c9642b34bf1e1a633538f0e3bac6083d Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Tue, 14 Jul 2026 21:11:20 -0500 Subject: [PATCH 3/9] sui-rpc: add an idle-body watchdog to the client transport Raising the flow-control windows makes the shared connection hard to starve, but a response stream that the application holds without polling still pins its window forever, and an RPC whose response can no longer make progress still hangs forever. Nothing in the default stack bounds either: HTTP/2 keepalive PINGs are answered by the peer's transport layer even when every stream is stalled, so the connection looks healthy throughout. Add a watchdog layer, on by default, applied inside Client::channel() beneath any user request layers. It moves each response body into a spawned task bridged back to the caller through a bounded mpsc(1) channel. The task's outermost await is an idle timer (default 30 seconds, comfortably above the fullnode's ~5 second subscription watermark cadence), so it stays responsive even when the caller has parked the response and nothing is polling it. A poll-driven timer would not work here: a parked stream is never polled, which is exactly when the timer is needed. On expiry the task drops the inner body, which sends RST_STREAM (a control frame, deliverable even on a starved connection) and releases all of the flow-control window the stream had pinned; the caller observes DeadlineExceeded on its next poll. Dropping the response aborts the task via an abort-on-drop guard, and a panic in the task surfaces as an Internal status rather than a clean end-of-stream. The timeout is configurable with Client::with_body_idle_timeout, disableable with Client::without_body_idle_timeout, and overridable per call (including re-enabling or disabling) by inserting a BodyIdleTimeout extension into the request. --- crates/sui-rpc/Cargo.toml | 2 +- crates/sui-rpc/src/client/mod.rs | 52 ++- crates/sui-rpc/src/client/watchdog.rs | 462 ++++++++++++++++++++++++++ 3 files changed, 514 insertions(+), 2 deletions(-) create mode 100644 crates/sui-rpc/src/client/watchdog.rs diff --git a/crates/sui-rpc/Cargo.toml b/crates/sui-rpc/Cargo.toml index ba670d6a80..9080a78f07 100644 --- a/crates/sui-rpc/Cargo.toml +++ b/crates/sui-rpc/Cargo.toml @@ -62,7 +62,7 @@ proptest = { version = "1.8.0", default-features = false, features = ["std"] } test-strategy = { version = "0.4" } sui-sdk-types = { version = "0.3.0", path = "../sui-sdk-types", default-features = false, features = ["proptest", "serde", "hash"] } serde_json = { version = "1.0.145" } -tokio = { version = "1.40", features = ["rt"] } +tokio = { version = "1.40", features = ["rt", "macros", "test-util"] } [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = ['cfg(doc_cfg)'] } diff --git a/crates/sui-rpc/src/client/mod.rs b/crates/sui-rpc/src/client/mod.rs index c110937dfd..da7debc926 100644 --- a/crates/sui-rpc/src/client/mod.rs +++ b/crates/sui-rpc/src/client/mod.rs @@ -15,6 +15,11 @@ pub use response_ext::ResponseExt; mod interceptors; pub use interceptors::HeadersInterceptor; +mod watchdog; +pub use watchdog::BodyIdleTimeout; +use watchdog::DEFAULT_BODY_IDLE_TIMEOUT; +use watchdog::WatchdogLayer; + mod staking_rewards; pub use staking_rewards::DelegatedStake; @@ -71,6 +76,7 @@ pub struct Client { channel: tonic::transport::Channel, headers: HeadersInterceptor, max_decoding_message_size: Option, + body_idle_timeout: Option, /// Layer to apply to all RPC requests request_layer: Option, @@ -98,7 +104,9 @@ impl Client { /// including the HTTP/2 flow-control windows that protect the shared /// connection from starvation by stalled streaming responses. Prefer /// [`Client::new`] plus the `with_*` configuration methods unless an - /// endpoint setting is needed that the client does not expose. + /// endpoint setting is needed that the client does not expose. The + /// idle-body watchdog (see [`Client::with_body_idle_timeout`]) is part of + /// the client rather than the endpoint and stays enabled. /// /// In particular, do not rely on /// [`http2_adaptive_window`](tonic::transport::Endpoint::http2_adaptive_window) @@ -115,6 +123,7 @@ impl Client { channel, headers: Default::default(), max_decoding_message_size: None, + body_idle_timeout: Some(DEFAULT_BODY_IDLE_TIMEOUT), request_layer: None, } } @@ -154,10 +163,46 @@ impl Client { channel, headers: Default::default(), max_decoding_message_size: None, + body_idle_timeout: Some(DEFAULT_BODY_IDLE_TIMEOUT), request_layer: None, }) } + /// Set the idle timeout for the client's response-body watchdog. + /// Defaults to 30 seconds. + /// + /// The watchdog bounds the time between response-body progress events: if + /// a whole idle period passes without a frame of the response being + /// delivered to the caller -- because the connection is starved or dead, + /// or because the caller has parked a streaming response without polling + /// it -- the watchdog resets the stream, releasing the HTTP/2 + /// flow-control window it had pinned, and the call observes a + /// [`DeadlineExceeded`](tonic::Code::DeadlineExceeded) status on its next + /// poll. This is what turns "an RPC on a starved connection hangs + /// forever" into a bounded failure, and what keeps an abandoned stream + /// from starving the shared connection in the first place. + /// + /// Streams that are legitimately quiet for longer than the timeout (the + /// fullnode's checkpoint subscription is not: it emits watermarks every + /// few seconds) should raise or disable the watchdog for that call with a + /// [`BodyIdleTimeout`] request extension. + pub fn with_body_idle_timeout(mut self, timeout: Duration) -> Self { + self.body_idle_timeout = Some(timeout); + self + } + + /// Disable the client's response-body watchdog (see + /// [`with_body_idle_timeout`](Self::with_body_idle_timeout)). + /// + /// Without it, an RPC whose response can no longer make progress hangs + /// indefinitely; only disable the watchdog when every call is bounded by + /// the caller. It can be re-enabled for individual requests with a + /// [`BodyIdleTimeout`] request extension. + pub fn without_body_idle_timeout(mut self) -> Self { + self.body_idle_timeout = None; + self + } + /// Set the HTTP/2 per-stream receive window, in bytes. /// /// This bounds how much unread response data a single RPC can buffer @@ -279,6 +324,11 @@ impl Client { .service(self.channel.clone()), ); + // Guard every response body with the idle-body watchdog, beneath any + // user layers so their view of the response goes through the + // watchdog's bridge. + let base = BoxService::new(WatchdogLayer::new(self.body_idle_timeout).layer(base)); + // Apply the user's outbound request layer if present. let layered = if let Some(layer) = &self.request_layer { layer.layer(base) diff --git a/crates/sui-rpc/src/client/watchdog.rs b/crates/sui-rpc/src/client/watchdog.rs new file mode 100644 index 0000000000..24fee59547 --- /dev/null +++ b/crates/sui-rpc/src/client/watchdog.rs @@ -0,0 +1,462 @@ +//! Client-side watchdog for response bodies. +//! +//! Every RPC made through a [`Client`](super::Client) is multiplexed over a +//! single HTTP/2 connection whose connection-level receive window is shared +//! by all in-flight responses. Response data counts against that window until +//! the application polls it out of the stream, so a streaming response that +//! is held without being polled pins up to a full stream window of the shared +//! budget. Enough stalled streams starve the connection, and then every RPC +//! on the channel hangs indefinitely while TCP and HTTP/2 keepalives stay +//! healthy. This converts into a permanent hang whenever the task that should +//! be polling a stream is itself blocked on another call on the same starved +//! channel: every party waits for connection window while holding connection +//! window. +//! +//! The watchdog breaks both the starvation and the deadlock. Each response +//! body is moved into a spawned task and bridged back to the caller through a +//! bounded channel. The task's outermost await is an idle timer, so it stays +//! responsive even when the caller has parked the response and nothing is +//! polling it. If a whole idle period passes without a frame being delivered +//! to the caller, the task drops the inner body, which resets the stream +//! (RST_STREAM is a control frame and is deliverable even when the connection +//! window is exhausted) and releases every byte of window the stream had +//! pinned. The caller observes a `DeadlineExceeded` status on its next poll. +//! +//! A timer raced against the caller's own polls would not be enough: a parked +//! stream is never polled, so a poll-driven timer freezes exactly when it is +//! needed. The spawned task gives the timer its own polling root. + +use std::future::Future; +use std::ops::ControlFlow; +use std::pin::Pin; +use std::task::Context; +use std::task::Poll; +use std::time::Duration; + +use bytes::Bytes; +use http_body::Body as _; +use http_body::Frame; +use tonic::Status; +use tonic::body::Body; +use tower::Layer; +use tower::Service; + +/// Default maximum time between response-body progress events before the +/// watchdog resets the stream. Comfortably above the fullnode's subscription +/// watermark cadence (~5s), so a healthy subscription always makes progress +/// well within it. +pub(super) const DEFAULT_BODY_IDLE_TIMEOUT: Duration = Duration::from_secs(30); + +/// Per-request override for the client's idle-body watchdog. +/// +/// Insert it into a request's extensions to raise, lower, or disable the +/// watchdog timeout for that call only: +/// +/// ``` +/// use std::time::Duration; +/// use sui_rpc::client::BodyIdleTimeout; +/// use sui_rpc::proto::sui::rpc::v2::SubscribeCheckpointsRequest; +/// +/// let mut request = tonic::Request::new(SubscribeCheckpointsRequest::default()); +/// request +/// .extensions_mut() +/// .insert(BodyIdleTimeout::new(Duration::from_secs(120))); +/// ``` +#[derive(Debug, Clone, Copy)] +pub struct BodyIdleTimeout(Option); + +impl BodyIdleTimeout { + /// Override the watchdog's idle timeout for this request. + pub fn new(timeout: Duration) -> Self { + Self(Some(timeout)) + } + + /// Disable the watchdog for this request. The response body can then + /// stall indefinitely; only use this when the caller enforces its own + /// bound on the call. + pub fn disabled() -> Self { + Self(None) + } +} + +/// [`Layer`] that applies [`Watchdog`] to the client's transport. +#[derive(Clone)] +pub(super) struct WatchdogLayer { + idle_timeout: Option, +} + +impl WatchdogLayer { + pub(super) fn new(idle_timeout: Option) -> Self { + Self { idle_timeout } + } +} + +impl Layer for WatchdogLayer { + type Service = Watchdog; + + fn layer(&self, inner: S) -> Self::Service { + Watchdog { + inner, + idle_timeout: self.idle_timeout, + } + } +} + +/// Service that moves each response body into a watchdog task (see the +/// module documentation). +pub(super) struct Watchdog { + inner: S, + idle_timeout: Option, +} + +impl Service> for Watchdog +where + S: Service, Response = http::Response>, + S::Future: Send + 'static, + S::Error: Send + 'static, +{ + type Response = http::Response; + type Error = S::Error; + type Future = futures::future::BoxFuture<'static, Result>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, request: http::Request) -> Self::Future { + let idle_timeout = match request.extensions().get::() { + Some(BodyIdleTimeout(override_timeout)) => *override_timeout, + None => self.idle_timeout, + }; + + let response = self.inner.call(request); + Box::pin(async move { + let response = response.await?; + let Some(idle_timeout) = idle_timeout else { + return Ok(response); + }; + Ok(response.map(|body| Body::new(WatchdogBody::spawn(body, idle_timeout)))) + }) + } +} + +/// The caller-facing half of the bridge: receives frames from the watchdog +/// task over a bounded channel and aborts the task if dropped, so dropping a +/// response still resets the underlying stream promptly. +struct WatchdogBody { + rx: tokio::sync::mpsc::Receiver, Status>>, + /// `Some` until the task's completion has been observed. Consulted when + /// the channel closes without a terminal status so that a panic in the + /// watchdog task surfaces as an error instead of a clean end-of-stream. + task: Option>, +} + +impl WatchdogBody { + fn spawn(body: Body, idle_timeout: Duration) -> Self { + // Capacity 1: the bridge only needs to decouple the two polling + // roots, and the tightest bound keeps the transport's own + // flow-control backpressure intact. + let (tx, rx) = tokio::sync::mpsc::channel(1); + let task = tokio::spawn(drive(body, tx, idle_timeout)); + Self { + rx, + task: Some(task), + } + } +} + +impl Drop for WatchdogBody { + fn drop(&mut self) { + if let Some(task) = &self.task { + task.abort(); + } + } +} + +impl http_body::Body for WatchdogBody { + type Data = Bytes; + type Error = Status; + + fn poll_frame( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Status>>> { + let this = &mut *self; + match this.rx.poll_recv(cx) { + Poll::Ready(Some(item)) => Poll::Ready(Some(item)), + Poll::Ready(None) => { + let Some(task) = &mut this.task else { + return Poll::Ready(None); + }; + match Pin::new(task).poll(cx) { + Poll::Ready(result) => { + this.task = None; + match result { + Err(e) if e.is_panic() => Poll::Ready(Some(Err(Status::internal( + "client idle-body watchdog task panicked", + )))), + // A clean exit, or cancellation via our own + // abort-on-drop (not observable here since we + // are being polled, not dropped). + Ok(()) | Err(_) => Poll::Ready(None), + } + } + Poll::Pending => Poll::Pending, + } + } + Poll::Pending => Poll::Pending, + } + } +} + +/// The watchdog task: pump frames from the transport to the bridge channel, +/// bounding each unit of progress with the idle timer. On expiry, drop the +/// body first (resetting the stream and releasing its pinned flow-control +/// window in real time), then deliver the terminal status at the consumer's +/// pace. +async fn drive( + body: Body, + tx: tokio::sync::mpsc::Sender, Status>>, + idle_timeout: Duration, +) { + let cut = pump(body, tx.clone(), idle_timeout).await; + if let Some(status) = cut { + let _ = tx.send(Err(status)).await; + } +} + +/// Returns `Some(status)` if the stream was cut by the watchdog, and `None` +/// on natural end-of-stream, transport error, or consumer hang-up. The inner +/// body is dropped on return in all cases. +async fn pump( + mut body: Body, + tx: tokio::sync::mpsc::Sender, Status>>, + idle_timeout: Duration, +) -> Option { + loop { + // One unit of progress: receive a frame from the transport and hand + // it to the consumer. The timer covers both awaits, so it fires both + // when the transport delivers nothing (a starved or dead connection) + // and when the consumer has parked the response without polling it. + let step = async { + match std::future::poll_fn(|cx| Pin::new(&mut body).poll_frame(cx)).await { + Some(Ok(frame)) => { + if tx.send(Ok(frame)).await.is_err() { + // Consumer dropped the response. + return ControlFlow::Break(()); + } + ControlFlow::Continue(()) + } + Some(Err(status)) => { + let _ = tx.send(Err(status)).await; + ControlFlow::Break(()) + } + None => ControlFlow::Break(()), + } + }; + + match tokio::time::timeout(idle_timeout, step).await { + Ok(ControlFlow::Continue(())) => {} + Ok(ControlFlow::Break(())) => return None, + Err(_) => { + return Some(Status::deadline_exceeded(format!( + "no response-body progress within {idle_timeout:?}; \ + stream reset by the client's idle-body watchdog" + ))); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::AtomicBool; + use std::sync::atomic::Ordering; + + /// Test body driven by an unbounded channel, holding an optional beacon + /// whose `Drop` proves when the watchdog task has dropped the body. + struct TestBody { + rx: tokio::sync::mpsc::UnboundedReceiver, Status>>, + _beacon: Option, + } + + struct DropBeacon(Arc); + impl Drop for DropBeacon { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + impl http_body::Body for TestBody { + type Data = Bytes; + type Error = Status; + + fn poll_frame( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Status>>> { + self.rx.poll_recv(cx) + } + } + + #[allow(clippy::type_complexity)] + fn test_body() -> ( + tokio::sync::mpsc::UnboundedSender, Status>>, + Arc, + Body, + ) { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let dropped = Arc::new(AtomicBool::new(false)); + let body = Body::new(TestBody { + rx, + _beacon: Some(DropBeacon(dropped.clone())), + }); + (tx, dropped, body) + } + + async fn next_frame(body: &mut WatchdogBody) -> Option, Status>> { + std::future::poll_fn(|cx| Pin::new(&mut *body).poll_frame(cx)).await + } + + fn data_frame(bytes: &'static [u8]) -> Frame { + Frame::data(Bytes::from_static(bytes)) + } + + /// `start_paused = true` makes `tokio::time` virtual: sleeps advance only + /// when the runtime is otherwise idle, so these tests run instantly and + /// deterministically. + #[tokio::test(start_paused = true)] + async fn frames_and_trailers_pass_through() { + let (tx, _dropped, body) = test_body(); + let mut watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + + tx.send(Ok(data_frame(b"hello"))).unwrap(); + let frame = next_frame(&mut watched).await.unwrap().unwrap(); + assert_eq!(frame.into_data().unwrap(), Bytes::from_static(b"hello")); + + let mut trailers = http::HeaderMap::new(); + trailers.insert("grpc-status", "0".parse().unwrap()); + tx.send(Ok(Frame::trailers(trailers.clone()))).unwrap(); + let frame = next_frame(&mut watched).await.unwrap().unwrap(); + assert_eq!(frame.into_trailers().unwrap(), trailers); + + drop(tx); + assert!(next_frame(&mut watched).await.is_none()); + } + + #[tokio::test(start_paused = true)] + async fn slow_but_live_stream_is_not_cut() { + let (tx, _dropped, body) = test_body(); + let mut watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + + for _ in 0..5 { + tokio::time::sleep(Duration::from_secs(20)).await; + tx.send(Ok(data_frame(b"tick"))).unwrap(); + let frame = next_frame(&mut watched).await.unwrap().unwrap(); + assert_eq!(frame.into_data().unwrap(), Bytes::from_static(b"tick")); + } + + drop(tx); + assert!(next_frame(&mut watched).await.is_none()); + } + + #[tokio::test(start_paused = true)] + async fn idle_transport_is_cut_and_body_dropped() { + let (_tx, dropped, body) = test_body(); + let mut watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + + let status = next_frame(&mut watched).await.unwrap().unwrap_err(); + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); + assert!(dropped.load(Ordering::SeqCst), "inner body was not dropped"); + assert!(next_frame(&mut watched).await.is_none()); + } + + /// The core starvation defense: when the consumer parks the response + /// without polling it, the watchdog must still fire and drop the inner + /// body (releasing its pinned flow-control window) in wall-clock time, + /// not on the consumer's next poll. + #[tokio::test(start_paused = true)] + async fn parked_consumer_is_cut_without_being_polled() { + let (tx, dropped, body) = test_body(); + let mut watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + + // The consumer takes one frame, then parks the response entirely. + tx.send(Ok(data_frame(b"first"))).unwrap(); + assert!(next_frame(&mut watched).await.unwrap().is_ok()); + + // The transport keeps delivering: one frame sits in the bridge + // channel and the pump blocks handing over the next one. + tx.send(Ok(data_frame(b"buffered"))).unwrap(); + tx.send(Ok(data_frame(b"blocked"))).unwrap(); + + // Without any consumer poll, the idle timer must drop the body. + tokio::time::sleep(Duration::from_secs(60)).await; + assert!( + dropped.load(Ordering::SeqCst), + "inner body was not dropped while the consumer was parked" + ); + + // When the consumer returns it drains the buffered frame, then + // observes the watchdog cut. + let frame = next_frame(&mut watched).await.unwrap().unwrap(); + assert_eq!(frame.into_data().unwrap(), Bytes::from_static(b"buffered")); + let status = next_frame(&mut watched).await.unwrap().unwrap_err(); + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); + } + + #[tokio::test(start_paused = true)] + async fn transport_error_passes_through() { + let (tx, _dropped, body) = test_body(); + let mut watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + + tx.send(Err(Status::unavailable("connection reset"))) + .unwrap(); + let status = next_frame(&mut watched).await.unwrap().unwrap_err(); + assert_eq!(status.code(), tonic::Code::Unavailable); + assert!(next_frame(&mut watched).await.is_none()); + } + + #[tokio::test(start_paused = true)] + async fn dropping_response_aborts_task_and_drops_body() { + let (tx, dropped, body) = test_body(); + let watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + tx.send(Ok(data_frame(b"unread"))).unwrap(); + + drop(watched); + for _ in 0..10 { + tokio::task::yield_now().await; + if dropped.load(Ordering::SeqCst) { + break; + } + } + assert!( + dropped.load(Ordering::SeqCst), + "inner body was not dropped after the response was dropped" + ); + } + + /// A panic in the watchdog task must surface as an error instead of + /// silently closing the bridge, which the consumer could not distinguish + /// from a clean end-of-stream. + #[tokio::test(start_paused = true)] + async fn task_panic_surfaces_as_internal() { + struct PanicBody; + impl http_body::Body for PanicBody { + type Data = Bytes; + type Error = Status; + + fn poll_frame( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll, Status>>> { + panic!("boom from inner body"); + } + } + + let mut watched = WatchdogBody::spawn(Body::new(PanicBody), Duration::from_secs(30)); + let status = next_frame(&mut watched).await.unwrap().unwrap_err(); + assert_eq!(status.code(), tonic::Code::Internal); + } +} From 2e757129c1f3b369e9e859163ff016bff41a86cc Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Tue, 14 Jul 2026 21:15:00 -0500 Subject: [PATCH 4/9] sui-rpc: enforce the grpc-timeout deadline across the response body Previously no per-call deadline was enforced end to end. tonic's channel machinery parses the grpc-timeout header that tonic::Request::set_timeout attaches, but only enforces it until response headers arrive. On a starved connection that is precisely the part that still works -- headers are not flow-controlled -- so the deadline never bounded the actual hang: unary response DATA that can never be delivered, or a streaming body that outlives its budget. The watchdog task now also enforces the request's grpc-timeout deadline across the whole response body, measured from request time so the budget communicated to the server is the same one enforced locally. On expiry the body is dropped (resetting the stream and releasing its pinned flow-control window) and the call observes DeadlineExceeded. This makes Request::set_timeout a total per-call deadline: the server participates through the standard header once it honors it, and the client enforces it locally regardless. Also document the deadline story on Client. --- crates/sui-rpc/src/client/mod.rs | 18 ++ crates/sui-rpc/src/client/watchdog.rs | 242 +++++++++++++++++++++++--- 2 files changed, 231 insertions(+), 29 deletions(-) diff --git a/crates/sui-rpc/src/client/mod.rs b/crates/sui-rpc/src/client/mod.rs index da7debc926..b77e4dc660 100644 --- a/crates/sui-rpc/src/client/mod.rs +++ b/crates/sui-rpc/src/client/mod.rs @@ -69,6 +69,24 @@ const DEFAULT_HTTP2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(20); const DEFAULT_HTTP2_STREAM_WINDOW_SIZE: u32 = 2 * 1024 * 1024; const DEFAULT_HTTP2_CONNECTION_WINDOW_SIZE: u32 = 64 * 1024 * 1024; +/// A gRPC client for the Sui fullnode RPC interface. +/// +/// All RPCs made through a client and its clones are multiplexed over a +/// single HTTP/2 connection. +/// +/// # Timeouts and deadlines +/// +/// No default bounds the total duration of a call; a per-call deadline can be +/// set with [`tonic::Request::set_timeout`]. This attaches the standard +/// `grpc-timeout` header, so a server that supports it enforces the deadline +/// too, and the client enforces it locally end to end: tonic bounds the wait +/// for response headers, and the client's watchdog (see +/// [`with_body_idle_timeout`](Client::with_body_idle_timeout)) bounds the +/// response body against the same deadline. +/// +/// Independent of any deadline, the watchdog resets RPCs whose response body +/// makes no progress for 30 seconds (configurable), so a call on a stalled +/// connection fails with `DeadlineExceeded` instead of hanging forever. #[derive(Clone)] pub struct Client { uri: http::Uri, diff --git a/crates/sui-rpc/src/client/watchdog.rs b/crates/sui-rpc/src/client/watchdog.rs index 24fee59547..d65e3d4752 100644 --- a/crates/sui-rpc/src/client/watchdog.rs +++ b/crates/sui-rpc/src/client/watchdog.rs @@ -25,6 +25,15 @@ //! A timer raced against the caller's own polls would not be enough: a parked //! stream is never polled, so a poll-driven timer freezes exactly when it is //! needed. The spawned task gives the timer its own polling root. +//! +//! The watchdog task also enforces the request's `grpc-timeout` deadline (set +//! with [`tonic::Request::set_timeout`]) across the response body. tonic's +//! own channel machinery only enforces that header until response headers +//! arrive, which on a starved connection is precisely the part that still +//! works -- headers are not flow-controlled -- so without this the deadline +//! never bounds the actual hang. With it, `set_timeout` is a total per-call +//! deadline: the same value is enforced locally end to end and communicated +//! to the server through the standard header. use std::future::Future; use std::ops::ControlFlow; @@ -128,18 +137,45 @@ where Some(BodyIdleTimeout(override_timeout)) => *override_timeout, None => self.idle_timeout, }; + // The deadline starts at request time, so the budget the header + // communicates to the server is the same one enforced locally. + let deadline = parse_grpc_timeout(request.headers()) + .map(|timeout| tokio::time::Instant::now() + timeout); let response = self.inner.call(request); Box::pin(async move { let response = response.await?; - let Some(idle_timeout) = idle_timeout else { + if idle_timeout.is_none() && deadline.is_none() { return Ok(response); - }; - Ok(response.map(|body| Body::new(WatchdogBody::spawn(body, idle_timeout)))) + } + Ok(response.map(|body| Body::new(WatchdogBody::spawn(body, idle_timeout, deadline)))) }) } } +/// Parse the standard `grpc-timeout` request header (1-8 ASCII digits +/// followed by a unit, per the gRPC-over-HTTP/2 spec). Malformed values are +/// ignored, matching tonic's own leniency, since failing the call over an +/// unenforceable header would be worse than not enforcing it. +fn parse_grpc_timeout(headers: &http::HeaderMap) -> Option { + let value = headers.get("grpc-timeout")?.to_str().ok()?; + let (magnitude, unit) = value.split_at(value.len().checked_sub(1)?); + if magnitude.is_empty() || magnitude.len() > 8 || !magnitude.bytes().all(|b| b.is_ascii_digit()) + { + return None; + } + let magnitude: u64 = magnitude.parse().ok()?; + match unit { + "H" => Some(Duration::from_secs(magnitude * 60 * 60)), + "M" => Some(Duration::from_secs(magnitude * 60)), + "S" => Some(Duration::from_secs(magnitude)), + "m" => Some(Duration::from_millis(magnitude)), + "u" => Some(Duration::from_micros(magnitude)), + "n" => Some(Duration::from_nanos(magnitude)), + _ => None, + } +} + /// The caller-facing half of the bridge: receives frames from the watchdog /// task over a bounded channel and aborts the task if dropped, so dropping a /// response still resets the underlying stream promptly. @@ -152,12 +188,16 @@ struct WatchdogBody { } impl WatchdogBody { - fn spawn(body: Body, idle_timeout: Duration) -> Self { + fn spawn( + body: Body, + idle_timeout: Option, + deadline: Option, + ) -> Self { // Capacity 1: the bridge only needs to decouple the two polling // roots, and the tightest bound keeps the transport's own // flow-control backpressure intact. let (tx, rx) = tokio::sync::mpsc::channel(1); - let task = tokio::spawn(drive(body, tx, idle_timeout)); + let task = tokio::spawn(drive(body, tx, idle_timeout, deadline)); Self { rx, task: Some(task), @@ -210,16 +250,28 @@ impl http_body::Body for WatchdogBody { } /// The watchdog task: pump frames from the transport to the bridge channel, -/// bounding each unit of progress with the idle timer. On expiry, drop the -/// body first (resetting the stream and releasing its pinned flow-control -/// window in real time), then deliver the terminal status at the consumer's -/// pace. +/// bounding each unit of progress with the idle timer and the whole body +/// with the request deadline. On expiry, drop the body first (resetting the +/// stream and releasing its pinned flow-control window in real time), then +/// deliver the terminal status at the consumer's pace. async fn drive( body: Body, tx: tokio::sync::mpsc::Sender, Status>>, - idle_timeout: Duration, + idle_timeout: Option, + deadline: Option, ) { - let cut = pump(body, tx.clone(), idle_timeout).await; + let pump = pump(body, tx.clone(), idle_timeout); + let cut = match deadline { + Some(deadline) => tokio::time::timeout_at(deadline, pump) + .await + .unwrap_or_else(|_| { + Some(Status::deadline_exceeded( + "grpc-timeout deadline exceeded before the response body completed; \ + stream reset by the client", + )) + }), + None => pump.await, + }; if let Some(status) = cut { let _ = tx.send(Err(status)).await; } @@ -227,11 +279,12 @@ async fn drive( /// Returns `Some(status)` if the stream was cut by the watchdog, and `None` /// on natural end-of-stream, transport error, or consumer hang-up. The inner -/// body is dropped on return in all cases. +/// body is dropped on return in all cases -- including cancellation by the +/// deadline in [`drive`], which drops this future and the body it owns. async fn pump( mut body: Body, tx: tokio::sync::mpsc::Sender, Status>>, - idle_timeout: Duration, + idle_timeout: Option, ) -> Option { loop { // One unit of progress: receive a frame from the transport and hand @@ -255,15 +308,21 @@ async fn pump( } }; - match tokio::time::timeout(idle_timeout, step).await { - Ok(ControlFlow::Continue(())) => {} - Ok(ControlFlow::Break(())) => return None, - Err(_) => { - return Some(Status::deadline_exceeded(format!( - "no response-body progress within {idle_timeout:?}; \ - stream reset by the client's idle-body watchdog" - ))); - } + let flow = match idle_timeout { + Some(idle_timeout) => match tokio::time::timeout(idle_timeout, step).await { + Ok(flow) => flow, + Err(_) => { + return Some(Status::deadline_exceeded(format!( + "no response-body progress within {idle_timeout:?}; \ + stream reset by the client's idle-body watchdog" + ))); + } + }, + None => step.await, + }; + match flow { + ControlFlow::Continue(()) => {} + ControlFlow::Break(()) => return None, } } } @@ -330,7 +389,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn frames_and_trailers_pass_through() { let (tx, _dropped, body) = test_body(); - let mut watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + let mut watched = WatchdogBody::spawn(body, Some(Duration::from_secs(30)), None); tx.send(Ok(data_frame(b"hello"))).unwrap(); let frame = next_frame(&mut watched).await.unwrap().unwrap(); @@ -349,7 +408,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn slow_but_live_stream_is_not_cut() { let (tx, _dropped, body) = test_body(); - let mut watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + let mut watched = WatchdogBody::spawn(body, Some(Duration::from_secs(30)), None); for _ in 0..5 { tokio::time::sleep(Duration::from_secs(20)).await; @@ -365,7 +424,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn idle_transport_is_cut_and_body_dropped() { let (_tx, dropped, body) = test_body(); - let mut watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + let mut watched = WatchdogBody::spawn(body, Some(Duration::from_secs(30)), None); let status = next_frame(&mut watched).await.unwrap().unwrap_err(); assert_eq!(status.code(), tonic::Code::DeadlineExceeded); @@ -380,7 +439,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn parked_consumer_is_cut_without_being_polled() { let (tx, dropped, body) = test_body(); - let mut watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + let mut watched = WatchdogBody::spawn(body, Some(Duration::from_secs(30)), None); // The consumer takes one frame, then parks the response entirely. tx.send(Ok(data_frame(b"first"))).unwrap(); @@ -409,7 +468,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn transport_error_passes_through() { let (tx, _dropped, body) = test_body(); - let mut watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + let mut watched = WatchdogBody::spawn(body, Some(Duration::from_secs(30)), None); tx.send(Err(Status::unavailable("connection reset"))) .unwrap(); @@ -421,7 +480,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn dropping_response_aborts_task_and_drops_body() { let (tx, dropped, body) = test_body(); - let watched = WatchdogBody::spawn(body, Duration::from_secs(30)); + let watched = WatchdogBody::spawn(body, Some(Duration::from_secs(30)), None); tx.send(Ok(data_frame(b"unread"))).unwrap(); drop(watched); @@ -455,8 +514,133 @@ mod tests { } } - let mut watched = WatchdogBody::spawn(Body::new(PanicBody), Duration::from_secs(30)); + let mut watched = + WatchdogBody::spawn(Body::new(PanicBody), Some(Duration::from_secs(30)), None); let status = next_frame(&mut watched).await.unwrap().unwrap_err(); assert_eq!(status.code(), tonic::Code::Internal); } + + /// The request deadline bounds the whole body even when every individual + /// frame arrives well within the idle timeout. + #[tokio::test(start_paused = true)] + async fn deadline_cuts_a_live_stream() { + let (tx, dropped, body) = test_body(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + let mut watched = + WatchdogBody::spawn(body, Some(Duration::from_secs(3600)), Some(deadline)); + + for _ in 0..4 { + tokio::time::sleep(Duration::from_secs(5)).await; + tx.send(Ok(data_frame(b"tick"))).unwrap(); + assert!(next_frame(&mut watched).await.unwrap().is_ok()); + } + + // Past the deadline the body must be dropped and the consumer must + // observe DeadlineExceeded, even though frames kept flowing. + tokio::time::sleep(Duration::from_secs(15)).await; + assert!( + dropped.load(Ordering::SeqCst), + "inner body was not dropped at the deadline" + ); + let status = next_frame(&mut watched).await.unwrap().unwrap_err(); + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); + } + + /// The deadline is enforced even when the idle watchdog is disabled. + #[tokio::test(start_paused = true)] + async fn deadline_applies_without_idle_timeout() { + let (_tx, dropped, body) = test_body(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + let mut watched = WatchdogBody::spawn(body, None, Some(deadline)); + + let status = next_frame(&mut watched).await.unwrap().unwrap_err(); + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); + assert!(dropped.load(Ordering::SeqCst), "inner body was not dropped"); + } + + /// End-to-end through the service: the `grpc-timeout` header (what + /// `tonic::Request::set_timeout` produces) bounds the response body, and + /// a `BodyIdleTimeout::disabled()` extension does not interfere with it. + #[tokio::test(start_paused = true)] + async fn service_honors_grpc_timeout_header() { + let (_tx, dropped, body) = test_body(); + let mut body = Some(body); + let mut service = WatchdogLayer::new(None).layer(tower::service_fn( + move |_request: http::Request| { + let body = body.take().unwrap(); + async move { Ok::<_, Status>(http::Response::new(body)) } + }, + )); + + let mut request = http::Request::new(Body::empty()); + request + .headers_mut() + .insert("grpc-timeout", "30S".parse().unwrap()); + request.extensions_mut().insert(BodyIdleTimeout::disabled()); + + let response = service.call(request).await.unwrap(); + let mut body = response.into_body(); + let item = std::future::poll_fn(|cx| Pin::new(&mut body).poll_frame(cx)).await; + let status = item.unwrap().unwrap_err(); + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); + assert!(dropped.load(Ordering::SeqCst), "inner body was not dropped"); + } + + /// With neither an idle timeout nor a deadline, the response body must be + /// passed through without spawning a bridge. + #[tokio::test(start_paused = true)] + async fn service_passes_body_through_when_unbounded() { + let (tx, _dropped, body) = test_body(); + let mut body = Some(body); + let mut service = WatchdogLayer::new(None).layer(tower::service_fn( + move |_request: http::Request| { + let body = body.take().unwrap(); + async move { Ok::<_, Status>(http::Response::new(body)) } + }, + )); + + let response = service + .call(http::Request::new(Body::empty())) + .await + .unwrap(); + let mut body = response.into_body(); + + // Nothing cuts the stream no matter how long it idles. + tokio::time::sleep(Duration::from_secs(3600)).await; + tx.send(Ok(data_frame(b"still here"))).unwrap(); + let item = std::future::poll_fn(|cx| Pin::new(&mut body).poll_frame(cx)).await; + assert!(item.unwrap().is_ok()); + } + + #[test] + fn grpc_timeout_parsing() { + fn parse(value: &str) -> Option { + let mut headers = http::HeaderMap::new(); + headers.insert("grpc-timeout", value.parse().unwrap()); + parse_grpc_timeout(&headers) + } + + assert_eq!(parse("2H"), Some(Duration::from_secs(2 * 60 * 60))); + assert_eq!(parse("3M"), Some(Duration::from_secs(180))); + assert_eq!(parse("30S"), Some(Duration::from_secs(30))); + assert_eq!(parse("500m"), Some(Duration::from_millis(500))); + assert_eq!(parse("250u"), Some(Duration::from_micros(250))); + assert_eq!(parse("82n"), Some(Duration::from_nanos(82))); + assert_eq!(parse("99999999S"), Some(Duration::from_secs(99_999_999))); + + // Malformed values are ignored rather than failing the call. + assert_eq!(parse(""), None); + assert_eq!(parse("S"), None); + assert_eq!(parse("30"), None); + assert_eq!(parse("30f"), None); + assert_eq!(parse("-30S"), None); + assert_eq!(parse("+30S"), None); + assert_eq!(parse("3.5S"), None); + assert_eq!(parse("123456789S"), None); + assert_eq!( + parse_grpc_timeout(&http::HeaderMap::new()), + None, + "absent header" + ); + } } From eed9841f2b428faf613a7223d0ed877dcfb3ea16 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Tue, 14 Jul 2026 21:20:36 -0500 Subject: [PATCH 5/9] sui-rpc: add flow-control starvation integration tests Port the standalone reproduction of the indefinite gRPC hang as permanent regression tests. A mock fullnode built from this crate's own generated server stubs pushes large checkpoint-subscription frames as fast as the transport accepts them, and the tests assert each layer of the defense independently: - a control test with the pre-fix configuration (hyper's default 2 MiB stream / 5 MiB connection windows via Client::from_endpoint, watchdog disabled) reproduces the hang with four parked subscription streams and recovers when they are dropped, proving the harness creates real starvation pressure; - the Client::new window defaults absorb eight parked streams while unary calls and an actively polled subscription keep working, with the watchdog disabled to show the windows alone carry the load; - the idle-body watchdog reaps parked streams on the small-window configuration, un-sticking the connection without any help from the caller, and the parked calls observe DeadlineExceeded when finally polled; - tonic::Request::set_timeout cuts a streaming call at its deadline while items are flowing. --- crates/sui-rpc/Cargo.toml | 5 +- crates/sui-rpc/tests/starvation.rs | 282 +++++++++++++++++++++++++++++ 2 files changed, 286 insertions(+), 1 deletion(-) create mode 100644 crates/sui-rpc/tests/starvation.rs diff --git a/crates/sui-rpc/Cargo.toml b/crates/sui-rpc/Cargo.toml index 9080a78f07..a8d1f27142 100644 --- a/crates/sui-rpc/Cargo.toml +++ b/crates/sui-rpc/Cargo.toml @@ -62,7 +62,10 @@ proptest = { version = "1.8.0", default-features = false, features = ["std"] } test-strategy = { version = "0.4" } sui-sdk-types = { version = "0.3.0", path = "../sui-sdk-types", default-features = false, features = ["proptest", "serde", "hash"] } serde_json = { version = "1.0.145" } -tokio = { version = "1.40", features = ["rt", "macros", "test-util"] } +tokio = { version = "1.40", features = ["rt", "rt-multi-thread", "macros", "net", "test-util"] } +# The starvation integration tests host a mock gRPC server built from this +# crate's own generated server stubs. +tonic = { version = "0.14.2", default-features = false, features = ["server", "router"] } [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = ['cfg(doc_cfg)'] } diff --git a/crates/sui-rpc/tests/starvation.rs b/crates/sui-rpc/tests/starvation.rs new file mode 100644 index 0000000000..1fbbb2a602 --- /dev/null +++ b/crates/sui-rpc/tests/starvation.rs @@ -0,0 +1,282 @@ +//! End-to-end regression tests for HTTP/2 connection-level flow-control +//! starvation. +//! +//! Every RPC made through a `Client` is multiplexed over one HTTP/2 +//! connection. Response data counts against the connection-level receive +//! window until the application polls it out of the stream, so streaming +//! responses that are held without being polled pin the shared window. With +//! hyper's default windows (2 MiB stream / 5 MiB connection), a handful of +//! parked streams exhaust the connection window and every RPC on the channel +//! hangs indefinitely while the TCP connection and HTTP/2 keepalives stay +//! healthy. +//! +//! These tests run a mock fullnode built from this crate's own generated +//! server stubs, whose checkpoint subscription pushes large frames as fast as +//! the transport accepts them, and assert: +//! +//! - the failure mode exists (a control test with the old, small windows and +//! the watchdog disabled), so the other tests are known to create real +//! starvation pressure; +//! - the `Client::new` window defaults absorb parked streams that would +//! starve the old configuration; +//! - the idle-body watchdog reaps parked streams, returning their pinned +//! window and un-sticking the connection without any help from the caller; +//! - a per-call `grpc-timeout` deadline (`tonic::Request::set_timeout`) +//! bounds a streaming call end to end even while items are flowing. + +use std::time::Duration; + +use futures::StreamExt; +use proto::ledger_service_server::LedgerService; +use proto::ledger_service_server::LedgerServiceServer; +use proto::subscription_service_server::SubscriptionService; +use proto::subscription_service_server::SubscriptionServiceServer; +use sui_rpc::Client; +use sui_rpc::proto::sui::rpc::v2 as proto; + +/// Generous bound for calls that must complete; failure mode is a hang, so +/// hitting this means the regression is back. +const VICTIM_TIMEOUT: Duration = Duration::from_secs(5); + +/// Bound for calls that must hang in the control test. Locally the victim +/// completes in single-digit milliseconds when the connection is healthy, so +/// this is orders of magnitude past success while keeping the test fast. +const HANG_TIMEOUT: Duration = Duration::from_secs(3); + +#[derive(Clone)] +struct MockServer; + +#[tonic::async_trait] +impl LedgerService for MockServer { + async fn get_service_info( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> { + let mut info = proto::GetServiceInfoResponse::default(); + info.chain_id = Some("mock".to_owned()); + Ok(tonic::Response::new(info)) + } +} + +#[tonic::async_trait] +impl SubscriptionService for MockServer { + async fn subscribe_checkpoints( + &self, + _request: tonic::Request, + ) -> Result< + tonic::Response>, + tonic::Status, + > { + // Emit large frames as fast as the transport will take them, + // mimicking a checkpoint subscription on a busy network. Compression + // is not enabled server-side, so the payload's compressibility is + // irrelevant: each response occupies its full size in the client's + // receive windows. + let stream = futures::stream::iter(0u64..).map(|seq| { + let mut checkpoint = proto::Checkpoint::default(); + checkpoint.sequence_number = Some(seq); + checkpoint.digest = Some("x".repeat(512 * 1024)); + let mut resp = proto::SubscribeCheckpointsResponse::default(); + resp.cursor = Some(seq); + resp.checkpoint = Some(checkpoint); + Ok(resp) + }); + Ok(tonic::Response::new(Box::pin(stream))) + } +} + +async fn spawn_mock_server() -> std::net::SocketAddr { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind mock server listener"); + let addr = listener.local_addr().expect("mock server local addr"); + tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(LedgerServiceServer::new(MockServer)) + .add_service(SubscriptionServiceServer::new(MockServer)) + .serve_with_incoming(tonic::transport::server::TcpIncoming::from(listener)) + .await + .expect("mock server exited with an error"); + }); + addr +} + +/// A client with the pre-fix transport configuration: hyper's default +/// flow-control windows (2 MiB stream / 5 MiB connection) and no watchdog. +fn legacy_client(addr: std::net::SocketAddr) -> Client { + let endpoint = tonic::transport::Endpoint::new(format!("http://{addr}")).expect("endpoint"); + Client::from_endpoint(&endpoint).without_body_idle_timeout() +} + +/// Open `count` checkpoint subscriptions and hold them without polling. The +/// server starts pushing as soon as each response stream opens, so the unread +/// data pins the client's receive windows whether or not the streams are ever +/// polled. +async fn park_subscriptions( + client: &mut Client, + count: usize, +) -> Vec> { + let mut parked = Vec::with_capacity(count); + for _ in 0..count { + let stream = client + .subscription_client() + .subscribe_checkpoints(proto::SubscribeCheckpointsRequest::default()) + .await + .expect("subscribe") + .into_inner(); + parked.push(stream); + } + // Give the server time to fill every parked stream's window. + tokio::time::sleep(Duration::from_secs(1)).await; + parked +} + +async fn victim_call(client: &mut Client, bound: Duration) -> Result<(), String> { + match tokio::time::timeout( + bound, + client + .ledger_client() + .get_service_info(proto::GetServiceInfoRequest::default()), + ) + .await + { + Ok(Ok(_)) => Ok(()), + Ok(Err(status)) => Err(format!("victim call failed with status: {status}")), + Err(_) => Err(format!("victim call hung for more than {bound:?}")), + } +} + +/// Control: with the pre-fix configuration the starvation reproduces, and +/// dropping the parked streams recovers the connection. This proves the mock +/// server creates real starvation pressure, so the passing tests below are +/// known to be exercising the defenses rather than a toothless setup. +#[tokio::test(flavor = "multi_thread")] +async fn control_small_windows_starve_without_watchdog() { + let addr = spawn_mock_server().await; + let mut client = legacy_client(addr); + + victim_call(&mut client, VICTIM_TIMEOUT) + .await + .expect("baseline call on a fresh connection"); + + let parked = park_subscriptions(&mut client, 4).await; + let starved = victim_call(&mut client, HANG_TIMEOUT).await; + assert!( + starved.is_err(), + "victim call should hang with 4 parked streams on 5 MiB connection window" + ); + + // The workaround, mechanized: dropping the stalled calls releases their + // pinned window and un-sticks the connection. + drop(parked); + victim_call(&mut client, VICTIM_TIMEOUT) + .await + .expect("victim call after dropping parked streams"); +} + +/// With the `Client::new` window defaults (64 MiB connection window), the +/// parked streams that starve the legacy configuration are absorbed: unary +/// calls and fresh, actively polled streams keep working. The watchdog is +/// disabled to prove the windows alone carry the load. +#[tokio::test(flavor = "multi_thread")] +async fn default_windows_absorb_parked_streams() { + let addr = spawn_mock_server().await; + let mut client = Client::new(format!("http://{addr}")) + .expect("client") + .without_body_idle_timeout(); + + let _parked = park_subscriptions(&mut client, 8).await; + + victim_call(&mut client, VICTIM_TIMEOUT) + .await + .expect("victim call with 8 parked streams"); + + // A fresh, actively polled subscription must also make progress. + let mut fresh = client + .subscription_client() + .subscribe_checkpoints(proto::SubscribeCheckpointsRequest::default()) + .await + .expect("subscribe") + .into_inner(); + for _ in 0..3 { + let item = tokio::time::timeout(VICTIM_TIMEOUT, fresh.next()) + .await + .expect("fresh subscription made no progress") + .expect("fresh subscription ended unexpectedly") + .expect("fresh subscription item"); + assert!(item.checkpoint.is_some()); + } +} + +/// The idle-body watchdog reaps parked streams on the legacy window +/// configuration: their pinned window is returned without the caller doing +/// anything, the connection un-sticks, and the parked calls observe +/// `DeadlineExceeded` when finally polled. +#[tokio::test(flavor = "multi_thread")] +async fn watchdog_reaps_parked_streams_and_recovers_the_connection() { + let addr = spawn_mock_server().await; + let endpoint = tonic::transport::Endpoint::new(format!("http://{addr}")).expect("endpoint"); + let mut client = + Client::from_endpoint(&endpoint).with_body_idle_timeout(Duration::from_millis(500)); + + let mut parked = park_subscriptions(&mut client, 4).await; + + // Wait past the idle timeout: the watchdog must reset every parked + // stream, releasing the pinned connection window. + tokio::time::sleep(Duration::from_secs(2)).await; + + victim_call(&mut client, VICTIM_TIMEOUT) + .await + .expect("victim call after the watchdog reaped the parked streams"); + + // A parked stream, when finally polled, drains whatever the bridge had + // buffered and then surfaces the watchdog's cut. + let mut stream = parked.pop().expect("parked stream"); + let status = loop { + match tokio::time::timeout(VICTIM_TIMEOUT, stream.next()) + .await + .expect("parked stream made no progress after the watchdog cut") + { + Some(Ok(_buffered)) => continue, + Some(Err(status)) => break status, + None => panic!("parked stream ended cleanly instead of with a status"), + } + }; + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); +} + +/// `tonic::Request::set_timeout` bounds a streaming call end to end: the +/// stream is cut at the deadline even though items are flowing steadily and +/// the idle watchdog never fires. +#[tokio::test(flavor = "multi_thread")] +async fn set_timeout_bounds_a_streaming_call() { + let addr = spawn_mock_server().await; + let mut client = Client::new(format!("http://{addr}")).expect("client"); + + let mut request = tonic::Request::new(proto::SubscribeCheckpointsRequest::default()); + request.set_timeout(Duration::from_secs(1)); + + let mut stream = client + .subscription_client() + .subscribe_checkpoints(request) + .await + .expect("subscribe") + .into_inner(); + + let mut items = 0u64; + let status = loop { + match tokio::time::timeout(VICTIM_TIMEOUT, stream.next()) + .await + .expect("stream made no progress before the deadline") + { + Some(Ok(_)) => items += 1, + Some(Err(status)) => break status, + None => panic!("stream ended cleanly instead of at the deadline"), + } + }; + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); + assert!( + items > 0, + "expected the stream to deliver items until the deadline" + ); +} From 165551494a89e9281d95511b55698d2fb2684475 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Tue, 14 Jul 2026 21:34:29 -0500 Subject: [PATCH 6/9] sui-rpc: reserve bridge capacity before pulling watchdog frames Previously the watchdog task pulled a frame from the transport and then awaited sending it into the bridge channel, so a parked consumer left one frame buffered in the channel plus one held in limbo by the blocked send. Pulling a frame out of the body also releases its HTTP/2 receive-window capacity, inviting the server to send more data than the consumer is actually accepting. Reserve a permit with Sender::reserve before polling the body, so a frame is only taken from the transport once it is immediately deliverable. A parked consumer now buffers at most one frame in the bridge, everything else stays in the transport where it remains subject to flow control, and a dropped consumer is detected before pulling another frame rather than after. The idle-timer semantics are unchanged: the timer still covers both awaits, firing whether the consumer is parked (reserve never resolves) or the transport is starved (no frame arrives). Add a test pinning the no-double-buffering property. --- crates/sui-rpc/src/client/watchdog.rs | 85 +++++++++++++++++++++++---- 1 file changed, 75 insertions(+), 10 deletions(-) diff --git a/crates/sui-rpc/src/client/watchdog.rs b/crates/sui-rpc/src/client/watchdog.rs index d65e3d4752..b2f87be4e0 100644 --- a/crates/sui-rpc/src/client/watchdog.rs +++ b/crates/sui-rpc/src/client/watchdog.rs @@ -287,21 +287,28 @@ async fn pump( idle_timeout: Option, ) -> Option { loop { - // One unit of progress: receive a frame from the transport and hand - // it to the consumer. The timer covers both awaits, so it fires both - // when the transport delivers nothing (a starved or dead connection) - // and when the consumer has parked the response without polling it. + // One unit of progress: wait for bridge capacity, then receive a + // frame from the transport and hand it over through the permit. + // Reserving before polling the body means a frame is only taken from + // the transport once it is immediately deliverable: nothing sits in + // limbo between the transport and the bridge, and the frame's + // receive-window capacity is not released (inviting the server to + // send more) until the consumer can actually accept it. The timer + // covers both awaits, so it fires both when the consumer has parked + // the response without polling it (reserve never resolves) and when + // the transport delivers nothing (a starved or dead connection). let step = async { + let Ok(permit) = tx.reserve().await else { + // Consumer dropped the response. + return ControlFlow::Break(()); + }; match std::future::poll_fn(|cx| Pin::new(&mut body).poll_frame(cx)).await { Some(Ok(frame)) => { - if tx.send(Ok(frame)).await.is_err() { - // Consumer dropped the response. - return ControlFlow::Break(()); - } + permit.send(Ok(frame)); ControlFlow::Continue(()) } Some(Err(status)) => { - let _ = tx.send(Err(status)).await; + permit.send(Err(status)); ControlFlow::Break(()) } None => ControlFlow::Break(()), @@ -446,7 +453,8 @@ mod tests { assert!(next_frame(&mut watched).await.unwrap().is_ok()); // The transport keeps delivering: one frame sits in the bridge - // channel and the pump blocks handing over the next one. + // channel and the pump waits for bridge capacity before taking the + // next one. tx.send(Ok(data_frame(b"buffered"))).unwrap(); tx.send(Ok(data_frame(b"blocked"))).unwrap(); @@ -465,6 +473,63 @@ mod tests { assert_eq!(status.code(), tonic::Code::DeadlineExceeded); } + /// Reserve-before-pull: the pump takes a frame from the transport only + /// once the bridge can accept it, so a parked consumer leaves at most one + /// frame buffered in the bridge and everything else in the transport, + /// where it remains subject to HTTP/2 flow control. + #[tokio::test(start_paused = true)] + async fn parked_consumer_does_not_double_buffer() { + use std::sync::atomic::AtomicU64; + + struct CountingBody { + rx: tokio::sync::mpsc::UnboundedReceiver, Status>>, + pulled: Arc, + } + impl http_body::Body for CountingBody { + type Data = Bytes; + type Error = Status; + + fn poll_frame( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Status>>> { + let poll = self.rx.poll_recv(cx); + if matches!(poll, Poll::Ready(Some(_))) { + self.pulled.fetch_add(1, Ordering::SeqCst); + } + poll + } + } + + let pulled = Arc::new(AtomicU64::new(0)); + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let body = Body::new(CountingBody { + rx, + pulled: pulled.clone(), + }); + let mut watched = WatchdogBody::spawn(body, Some(Duration::from_secs(30)), None); + + // The transport has plenty of frames available. + for _ in 0..5 { + tx.send(Ok(data_frame(b"tick"))).unwrap(); + } + + // The consumer takes one frame and parks. The pump may refill the + // one-slot bridge, but must not pull frames it cannot deliver. + assert!(next_frame(&mut watched).await.unwrap().is_ok()); + tokio::time::sleep(Duration::from_secs(60)).await; + assert_eq!( + pulled.load(Ordering::SeqCst), + 2, + "the pump pulled frames from the transport that it could not deliver" + ); + + // The consumer still sees the buffered frame and then the cut. + assert!(next_frame(&mut watched).await.unwrap().is_ok()); + let status = next_frame(&mut watched).await.unwrap().unwrap_err(); + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); + } + #[tokio::test(start_paused = true)] async fn transport_error_passes_through() { let (tx, _dropped, body) = test_body(); From e1c46d1e459dd9a0810b35ae0a1f81da4844caf7 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Wed, 15 Jul 2026 13:29:31 -0500 Subject: [PATCH 7/9] sui-rpc: expose a client-wide response-headers timeout Previously the client had no way to bound every RPC by default: a deadline had to be set per request with `tonic::Request::set_timeout`, and a call against a server that never responded could wait on response headers indefinitely (headers are not flow-controlled, so neither the flow-control windows nor the idle-body watchdog bound that phase). With this commit, `Client::with_response_headers_timeout` exposes `tonic::transport::Endpoint::timeout`, which tonic enforces locally from request dispatch until response headers arrive. Because a server does not send response headers for a unary call until the handler completes, this effectively bounds the total duration of unary calls, while the timer is dropped once headers arrive, so a client-wide value does not cut off long-lived streaming responses. The timeout is not communicated to the server, and tonic combines it with a per-call `grpc-timeout` deadline by taking the shorter of the two, so a per-call deadline can tighten the local bound but never extend it. Also normalize tonic's `TimeoutExpired` error, which tonic surfaces as `Cancelled`, to `DeadlineExceeded` in the client's error mapping. The gRPC code for an expired deadline is `DeadlineExceeded`, and the watchdog already uses it for body-phase expiry, so both phases of a deadline now report the same code. Add integration tests covering a stalled unary call failing at the timeout with `DeadlineExceeded`, the shorter-of-the-two interaction with per-call deadlines in both directions, a subscription stream outliving the timeout, and a healthy call passing through unaffected. --- crates/sui-rpc/src/client/mod.rs | 80 ++++++++++-- crates/sui-rpc/tests/timeouts.rs | 212 +++++++++++++++++++++++++++++++ 2 files changed, 282 insertions(+), 10 deletions(-) create mode 100644 crates/sui-rpc/tests/timeouts.rs diff --git a/crates/sui-rpc/src/client/mod.rs b/crates/sui-rpc/src/client/mod.rs index b77e4dc660..c9603635f9 100644 --- a/crates/sui-rpc/src/client/mod.rs +++ b/crates/sui-rpc/src/client/mod.rs @@ -76,13 +76,20 @@ const DEFAULT_HTTP2_CONNECTION_WINDOW_SIZE: u32 = 64 * 1024 * 1024; /// /// # Timeouts and deadlines /// -/// No default bounds the total duration of a call; a per-call deadline can be -/// set with [`tonic::Request::set_timeout`]. This attaches the standard -/// `grpc-timeout` header, so a server that supports it enforces the deadline -/// too, and the client enforces it locally end to end: tonic bounds the wait -/// for response headers, and the client's watchdog (see -/// [`with_body_idle_timeout`](Client::with_body_idle_timeout)) bounds the -/// response body against the same deadline. +/// No default bounds the total duration of a call. Two opt-in bounds are +/// available: +/// +/// - A per-call deadline set with [`tonic::Request::set_timeout`]. This +/// attaches the standard `grpc-timeout` header, so a server that supports +/// it enforces the deadline too, and the client enforces it locally end to +/// end: tonic bounds the wait for response headers, and the client's +/// watchdog (see [`with_body_idle_timeout`](Client::with_body_idle_timeout)) +/// bounds the response body against the same deadline. +/// - A client-wide response-headers timeout set with +/// [`with_response_headers_timeout`](Client::with_response_headers_timeout). +/// This is enforced locally only and its timer stops once response headers +/// arrive, so it bounds every unary call (whose headers are not sent until +/// the handler completes) without cutting off long-lived streams. /// /// Independent of any deadline, the watchdog resets RPCs whose response body /// makes no progress for 30 seconds (configurable), so a call on a stalled @@ -155,13 +162,13 @@ impl Client { let uri = uri .try_into() .map_err(Into::into) - .map_err(tonic::Status::from_error)?; + .map_err(status_from_error)?; let mut endpoint = tonic::transport::Endpoint::from(uri.clone()); if uri.scheme() == Some(&http::uri::Scheme::HTTPS) { endpoint = endpoint .tls_config(ClientTlsConfig::new().with_enabled_roots()) .map_err(Into::into) - .map_err(tonic::Status::from_error)?; + .map_err(status_from_error)?; } let endpoint = endpoint @@ -221,6 +228,35 @@ impl Client { self } + /// Set a timeout for the response-headers phase of every RPC made + /// through this client. Disabled by default. + /// + /// The timer covers a request from dispatch on the connection until + /// response headers arrive and is dropped once they do, so a client-wide + /// value does not cut off long-lived streaming responses. Because a + /// server does not send response headers for a unary call until the + /// handler completes, this effectively bounds the total duration of + /// unary calls; the body that follows is bounded by the idle-body + /// watchdog (see [`with_body_idle_timeout`](Self::with_body_idle_timeout)) + /// and, when set, the per-call deadline. Connection establishment is + /// bounded separately by the connect timeout. + /// + /// This timeout is enforced locally only; it is not communicated to the + /// server. When a per-call deadline ([`tonic::Request::set_timeout`]) is + /// also set, the shorter of the two bounds the headers phase locally, so + /// a per-call deadline can tighten this bound but never extend it -- + /// size the timeout for the slowest expected RPC. Expiry surfaces as + /// [`DeadlineExceeded`](tonic::Code::DeadlineExceeded). + /// + /// This rebuilds the underlying channel, so it must be called before the + /// client is used or cloned; earlier clones keep the previous + /// configuration. + pub fn with_response_headers_timeout(mut self, timeout: Duration) -> Self { + self.endpoint = self.endpoint.timeout(timeout); + self.channel = self.endpoint.connect_lazy(); + self + } + /// Set the HTTP/2 per-stream receive window, in bytes. /// /// This bounds how much unread response data a single RPC can buffer @@ -359,7 +395,7 @@ impl Client { // lifetime-inference issues with async_trait and Box. BoxService::new( ServiceBuilder::new() - .map_err(tonic::Status::from_error) + .map_err(status_from_error) .service(layered), ) } @@ -454,3 +490,27 @@ impl Client { }) } } + +/// Map a transport error to a [`tonic::Status`]. +/// +/// tonic surfaces an expired headers-phase timeout (the client's +/// response-headers timeout, or a `grpc-timeout` deadline expiring before +/// response headers arrive) as `Cancelled`. The gRPC code for an expired +/// deadline is `DeadlineExceeded`, and the watchdog already uses it for +/// body-phase expiry, so normalize before delegating to tonic's own mapping. +fn status_from_error(error: BoxError) -> tonic::Status { + let mut source: Option<&(dyn std::error::Error + 'static)> = Some(error.as_ref()); + while let Some(err) = source { + // An embedded `Status` takes precedence, as in tonic's own mapping. + if err.is::() { + break; + } + if err.is::() { + return tonic::Status::deadline_exceeded( + "timeout expired before response headers were received", + ); + } + source = err.source(); + } + tonic::Status::from_error(error) +} diff --git a/crates/sui-rpc/tests/timeouts.rs b/crates/sui-rpc/tests/timeouts.rs new file mode 100644 index 0000000000..6b59ebc02b --- /dev/null +++ b/crates/sui-rpc/tests/timeouts.rs @@ -0,0 +1,212 @@ +//! Integration tests for the client's response-headers timeout +//! (`Client::with_response_headers_timeout`). +//! +//! The timeout feeds `tonic::transport::Endpoint::timeout`, which tonic +//! enforces locally against the wait for response headers only. These tests +//! run a mock fullnode built from this crate's own generated server stubs and +//! assert: +//! +//! - a unary call whose handler never responds (so response headers never +//! arrive) fails at the timeout with `DeadlineExceeded` instead of hanging; +//! - a per-call `grpc-timeout` deadline combines with the timeout by taking +//! the shorter of the two, so it can tighten the local bound but never +//! extend it; +//! - the timer stops once response headers arrive, so a client-wide timeout +//! does not cut off long-lived streaming responses; +//! - a healthy unary call is unaffected. + +use std::time::Duration; + +use futures::StreamExt; +use proto::ledger_service_server::LedgerService; +use proto::ledger_service_server::LedgerServiceServer; +use proto::subscription_service_server::SubscriptionService; +use proto::subscription_service_server::SubscriptionServiceServer; +use sui_rpc::Client; +use sui_rpc::proto::sui::rpc::v2 as proto; + +/// Bound for calls that must complete or fail promptly; generous so slow CI +/// machines do not flake, but far below [`UNARY_STALL`], so hitting it means +/// a timeout that should have fired did not. +const FAILURE_BOUND: Duration = Duration::from_secs(5); + +/// Response-headers timeout used by the tests that expect it to fire. +const HEADERS_TIMEOUT: Duration = Duration::from_millis(400); + +/// How long the mock unary handler stalls before responding. A tonic server +/// does not send response headers for a unary call until the handler +/// returns, so this stalls the headers phase well past every bound above. +const UNARY_STALL: Duration = Duration::from_secs(60); + +/// Interval between items on the mock subscription stream. +const ITEM_INTERVAL: Duration = Duration::from_millis(50); + +#[derive(Clone)] +struct MockServer { + unary_delay: Duration, +} + +#[tonic::async_trait] +impl LedgerService for MockServer { + async fn get_service_info( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> { + tokio::time::sleep(self.unary_delay).await; + let mut info = proto::GetServiceInfoResponse::default(); + info.chain_id = Some("mock".to_owned()); + Ok(tonic::Response::new(info)) + } +} + +#[tonic::async_trait] +impl SubscriptionService for MockServer { + async fn subscribe_checkpoints( + &self, + _request: tonic::Request, + ) -> Result< + tonic::Response>, + tonic::Status, + > { + // Response headers go out as soon as this handler returns; the items + // then trickle out on an interval so the stream stays alive well past + // the response-headers timeout. + let stream = futures::stream::iter(0u64..).then(|seq| async move { + tokio::time::sleep(ITEM_INTERVAL).await; + let mut resp = proto::SubscribeCheckpointsResponse::default(); + resp.cursor = Some(seq); + Ok(resp) + }); + Ok(tonic::Response::new(Box::pin(stream))) + } +} + +async fn spawn_mock_server(unary_delay: Duration) -> std::net::SocketAddr { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind mock server listener"); + let addr = listener.local_addr().expect("mock server local addr"); + tokio::spawn(async move { + let server = MockServer { unary_delay }; + tonic::transport::Server::builder() + .add_service(LedgerServiceServer::new(server.clone())) + .add_service(SubscriptionServiceServer::new(server)) + .serve_with_incoming(tonic::transport::server::TcpIncoming::from(listener)) + .await + .expect("mock server exited with an error"); + }); + addr +} + +/// Issue a unary call and expect it to fail before [`FAILURE_BOUND`], +/// returning the status it failed with. +async fn expect_prompt_failure( + client: &mut Client, + request: tonic::Request, +) -> tonic::Status { + tokio::time::timeout( + FAILURE_BOUND, + client.ledger_client().get_service_info(request), + ) + .await + .expect("call ran past the failure bound instead of timing out") + .expect_err("call against a stalled handler must fail") +} + +/// The response-headers timeout bounds a unary call whose handler never +/// responds, and expiry surfaces as `DeadlineExceeded` (normalized from the +/// `Cancelled` code tonic uses for its timeout). +#[tokio::test(flavor = "multi_thread")] +async fn response_headers_timeout_bounds_a_stalled_unary_call() { + let addr = spawn_mock_server(UNARY_STALL).await; + let mut client = Client::new(format!("http://{addr}")) + .expect("client") + .with_response_headers_timeout(HEADERS_TIMEOUT); + + let request = tonic::Request::new(proto::GetServiceInfoRequest::default()); + let status = expect_prompt_failure(&mut client, request).await; + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); +} + +/// tonic enforces the shorter of the response-headers timeout and a per-call +/// `grpc-timeout` deadline, so a longer per-call deadline cannot extend the +/// locally enforced bound. +#[tokio::test(flavor = "multi_thread")] +async fn per_call_deadline_cannot_extend_the_timeout() { + let addr = spawn_mock_server(UNARY_STALL).await; + let mut client = Client::new(format!("http://{addr}")) + .expect("client") + .with_response_headers_timeout(HEADERS_TIMEOUT); + + let mut request = tonic::Request::new(proto::GetServiceInfoRequest::default()); + request.set_timeout(Duration::from_secs(60)); + + let status = expect_prompt_failure(&mut client, request).await; + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); +} + +/// A per-call deadline shorter than the response-headers timeout tightens +/// the locally enforced bound below it. +#[tokio::test(flavor = "multi_thread")] +async fn per_call_deadline_tightens_below_the_timeout() { + let addr = spawn_mock_server(UNARY_STALL).await; + let mut client = Client::new(format!("http://{addr}")) + .expect("client") + .with_response_headers_timeout(Duration::from_secs(60)); + + let mut request = tonic::Request::new(proto::GetServiceInfoRequest::default()); + request.set_timeout(HEADERS_TIMEOUT); + + let status = expect_prompt_failure(&mut client, request).await; + assert_eq!(status.code(), tonic::Code::DeadlineExceeded); +} + +/// The timer stops once response headers arrive: a streaming call keeps +/// delivering items long past the response-headers timeout, so a client-wide +/// timeout does not cut off long-lived subscriptions. +#[tokio::test(flavor = "multi_thread")] +async fn streams_outlive_the_response_headers_timeout() { + let addr = spawn_mock_server(UNARY_STALL).await; + let mut client = Client::new(format!("http://{addr}")) + .expect("client") + .with_response_headers_timeout(HEADERS_TIMEOUT); + + let mut stream = client + .subscription_client() + .subscribe_checkpoints(proto::SubscribeCheckpointsRequest::default()) + .await + .expect("subscribe") + .into_inner(); + + // 20 items at 50ms apart keep the stream alive for about a second, well + // past the 400ms response-headers timeout. + for expected in 0..20u64 { + let item = tokio::time::timeout(FAILURE_BOUND, stream.next()) + .await + .expect("stream made no progress") + .expect("stream ended unexpectedly") + .expect("stream item"); + assert_eq!(item.cursor, Some(expected)); + } +} + +/// A healthy unary call completes normally with the timeout configured. +#[tokio::test(flavor = "multi_thread")] +async fn healthy_calls_are_unaffected() { + let addr = spawn_mock_server(Duration::ZERO).await; + let mut client = Client::new(format!("http://{addr}")) + .expect("client") + .with_response_headers_timeout(HEADERS_TIMEOUT); + + let info = tokio::time::timeout( + FAILURE_BOUND, + client + .ledger_client() + .get_service_info(proto::GetServiceInfoRequest::default()), + ) + .await + .expect("healthy call hung") + .expect("healthy call failed") + .into_inner(); + assert_eq!(info.chain_id.as_deref(), Some("mock")); +} From 579b472a5b25840a85aa7f575cdfe84364b99fc9 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Wed, 15 Jul 2026 13:53:27 -0500 Subject: [PATCH 8/9] sui-rpc: follow empty pagination pages instead of truncating Previously the list_owned_objects, list_dynamic_fields, list_balances, and list_package_versions streams treated a page with zero items as end-of-stream even when the response carried a next_page_token. The list-pagination contract allows such pages (for example, a filtered scan that hit a server-side budget before finding a match), so the streams would silently drop every remaining page -- data loss with no error. The list_delegated_stake helper already handled this correctly by looping on the token; the four stream helpers were the outlier. With this commit the streams keep following the token until a page yields an item or pagination ends. Also add integration tests against a scripted mock server covering an empty middle page and a mid-pagination error; the empty-page test fails against the previous behavior. --- crates/sui-rpc/src/client/lists.rs | 96 +++++++++++++++-------- crates/sui-rpc/tests/pagination.rs | 121 +++++++++++++++++++++++++++++ 2 files changed, 185 insertions(+), 32 deletions(-) create mode 100644 crates/sui-rpc/tests/pagination.rs diff --git a/crates/sui-rpc/src/client/lists.rs b/crates/sui-rpc/src/client/lists.rs index cd062bba2b..1aeafaac21 100644 --- a/crates/sui-rpc/src/client/lists.rs +++ b/crates/sui-rpc/src/client/lists.rs @@ -37,12 +37,17 @@ impl Client { request, // request (page_token will be updated as we paginate) client, // client for making requests ), - move |(mut iter, has_next_page, mut request, mut client)| async move { + move |(mut iter, mut has_next_page, mut request, mut client)| async move { if let Some(item) = iter.next() { return Some((Ok(item), (iter, has_next_page, request, client))); } - if has_next_page { + // A page may be empty while still carrying a next_page_token + // (for example, a filtered scan that hit a server-side + // budget), so keep following the token until a page yields an + // item or pagination ends. Each response's token reflects + // server-side scan progress, so this terminates. + while has_next_page { let new_request = tonic::Request::from_parts( request.metadata().clone(), request.extensions().clone(), @@ -54,21 +59,24 @@ impl Client { let response = response.into_inner(); let mut iter = response.objects.into_iter(); - let has_next_page = response.next_page_token.is_some(); + has_next_page = response.next_page_token.is_some(); request.get_mut().page_token = response.next_page_token; - iter.next() - .map(|item| (Ok(item), (iter, has_next_page, request, client))) + if let Some(item) = iter.next() { + return Some((Ok(item), (iter, has_next_page, request, client))); + } } Err(e) => { // Return error and terminate stream request.get_mut().page_token = None; - Some((Err(e), (Vec::new().into_iter(), false, request, client))) + return Some(( + Err(e), + (Vec::new().into_iter(), false, request, client), + )); } } - } else { - None } + None }, ) } @@ -98,12 +106,17 @@ impl Client { request, // request (page_token will be updated as we paginate) client, // client for making requests ), - move |(mut iter, has_next_page, mut request, mut client)| async move { + move |(mut iter, mut has_next_page, mut request, mut client)| async move { if let Some(item) = iter.next() { return Some((Ok(item), (iter, has_next_page, request, client))); } - if has_next_page { + // A page may be empty while still carrying a next_page_token + // (for example, a filtered scan that hit a server-side + // budget), so keep following the token until a page yields an + // item or pagination ends. Each response's token reflects + // server-side scan progress, so this terminates. + while has_next_page { let new_request = tonic::Request::from_parts( request.metadata().clone(), request.extensions().clone(), @@ -115,21 +128,24 @@ impl Client { let response = response.into_inner(); let mut iter = response.dynamic_fields.into_iter(); - let has_next_page = response.next_page_token.is_some(); + has_next_page = response.next_page_token.is_some(); request.get_mut().page_token = response.next_page_token; - iter.next() - .map(|item| (Ok(item), (iter, has_next_page, request, client))) + if let Some(item) = iter.next() { + return Some((Ok(item), (iter, has_next_page, request, client))); + } } Err(e) => { // Return error and terminate stream request.get_mut().page_token = None; - Some((Err(e), (Vec::new().into_iter(), false, request, client))) + return Some(( + Err(e), + (Vec::new().into_iter(), false, request, client), + )); } } - } else { - None } + None }, ) } @@ -159,12 +175,17 @@ impl Client { request, // request (page_token will be updated as we paginate) client, // client for making requests ), - move |(mut iter, has_next_page, mut request, mut client)| async move { + move |(mut iter, mut has_next_page, mut request, mut client)| async move { if let Some(item) = iter.next() { return Some((Ok(item), (iter, has_next_page, request, client))); } - if has_next_page { + // A page may be empty while still carrying a next_page_token + // (for example, a filtered scan that hit a server-side + // budget), so keep following the token until a page yields an + // item or pagination ends. Each response's token reflects + // server-side scan progress, so this terminates. + while has_next_page { let new_request = tonic::Request::from_parts( request.metadata().clone(), request.extensions().clone(), @@ -176,21 +197,24 @@ impl Client { let response = response.into_inner(); let mut iter = response.balances.into_iter(); - let has_next_page = response.next_page_token.is_some(); + has_next_page = response.next_page_token.is_some(); request.get_mut().page_token = response.next_page_token; - iter.next() - .map(|item| (Ok(item), (iter, has_next_page, request, client))) + if let Some(item) = iter.next() { + return Some((Ok(item), (iter, has_next_page, request, client))); + } } Err(e) => { // Return error and terminate stream request.get_mut().page_token = None; - Some((Err(e), (Vec::new().into_iter(), false, request, client))) + return Some(( + Err(e), + (Vec::new().into_iter(), false, request, client), + )); } } - } else { - None } + None }, ) } @@ -220,12 +244,17 @@ impl Client { request, // request (page_token will be updated as we paginate) client, // client for making requests ), - move |(mut iter, has_next_page, mut request, mut client)| async move { + move |(mut iter, mut has_next_page, mut request, mut client)| async move { if let Some(item) = iter.next() { return Some((Ok(item), (iter, has_next_page, request, client))); } - if has_next_page { + // A page may be empty while still carrying a next_page_token + // (for example, a filtered scan that hit a server-side + // budget), so keep following the token until a page yields an + // item or pagination ends. Each response's token reflects + // server-side scan progress, so this terminates. + while has_next_page { let new_request = tonic::Request::from_parts( request.metadata().clone(), request.extensions().clone(), @@ -241,21 +270,24 @@ impl Client { let response = response.into_inner(); let mut iter = response.versions.into_iter(); - let has_next_page = response.next_page_token.is_some(); + has_next_page = response.next_page_token.is_some(); request.get_mut().page_token = response.next_page_token; - iter.next() - .map(|item| (Ok(item), (iter, has_next_page, request, client))) + if let Some(item) = iter.next() { + return Some((Ok(item), (iter, has_next_page, request, client))); + } } Err(e) => { // Return error and terminate stream request.get_mut().page_token = None; - Some((Err(e), (Vec::new().into_iter(), false, request, client))) + return Some(( + Err(e), + (Vec::new().into_iter(), false, request, client), + )); } } - } else { - None } + None }, ) } diff --git a/crates/sui-rpc/tests/pagination.rs b/crates/sui-rpc/tests/pagination.rs new file mode 100644 index 0000000000..453ac0c747 --- /dev/null +++ b/crates/sui-rpc/tests/pagination.rs @@ -0,0 +1,121 @@ +//! Regression tests for the automatic pagination in the `list_*` stream +//! helpers. +//! +//! The list-pagination contract allows a server to return an empty page that +//! still carries a `next_page_token` (for example, a filtered scan that hit a +//! server-side budget before finding a match). The pagination streams must +//! keep following the token instead of treating an empty page as +//! end-of-stream, which would silently truncate the results. + +use bytes::Bytes; +use futures::StreamExt; +use proto::state_service_server::StateService; +use proto::state_service_server::StateServiceServer; +use sui_rpc::Client; +use sui_rpc::proto::sui::rpc::v2 as proto; + +/// A scripted `StateService`: each `ListOwnedObjects` page is selected by the +/// request's `page_token` (the page index, or absent for the first page). +#[derive(Clone)] +struct ScriptedServer { + /// One entry per page: the objects to return and whether a next page + /// exists. + pages: Vec<(Vec, bool)>, +} + +impl ScriptedServer { + fn page_index(token: Option<&Bytes>) -> usize { + token + .map(|t| { + String::from_utf8(t.to_vec()) + .expect("test tokens are utf-8") + .parse() + .expect("test tokens are page indexes") + }) + .unwrap_or(0) + } +} + +#[tonic::async_trait] +impl StateService for ScriptedServer { + async fn list_owned_objects( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> { + let index = Self::page_index(request.get_ref().page_token.as_ref()); + let Some((objects, has_next)) = self.pages.get(index) else { + return Err(tonic::Status::invalid_argument(format!( + "requested page {index} beyond the scripted {} pages", + self.pages.len() + ))); + }; + let mut response = proto::ListOwnedObjectsResponse::default(); + response.objects = objects.clone(); + response.next_page_token = has_next.then(|| Bytes::from(format!("{}", index + 1))); + Ok(tonic::Response::new(response)) + } +} + +fn object(id: &str) -> proto::Object { + let mut object = proto::Object::default(); + object.object_id = Some(id.to_owned()); + object +} + +async fn spawn_scripted_server(pages: Vec<(Vec, bool)>) -> Client { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind mock server listener"); + let addr = listener.local_addr().expect("mock server local addr"); + tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(StateServiceServer::new(ScriptedServer { pages })) + .serve_with_incoming(tonic::transport::server::TcpIncoming::from(listener)) + .await + .expect("mock server exited with an error"); + }); + Client::new(format!("http://{addr}")).expect("client") +} + +/// Empty pages that still carry a `next_page_token` must be skipped, not +/// treated as end-of-stream. +#[tokio::test(flavor = "multi_thread")] +async fn empty_page_with_token_does_not_truncate_the_stream() { + let client = spawn_scripted_server(vec![ + (vec![object("0x1"), object("0x2")], true), + (Vec::new(), true), + (Vec::new(), true), + (vec![object("0x3")], false), + ]) + .await; + + let stream = client.list_owned_objects(proto::ListOwnedObjectsRequest::default()); + let items: Vec<_> = stream.collect().await; + + let ids: Vec<_> = items + .into_iter() + .map(|item| item.expect("no page should fail").object_id().to_owned()) + .collect(); + assert_eq!(ids, ["0x1", "0x2", "0x3"]); +} + +/// A page-fetch error is yielded as the final stream item and terminates +/// pagination. +#[tokio::test(flavor = "multi_thread")] +async fn page_error_terminates_the_stream() { + // Two scripted pages; the third fetch (page index 2) falls off the end of + // the script and the mock returns InvalidArgument. + let client = spawn_scripted_server(vec![ + (vec![object("0x1")], true), + (vec![object("0x2")], true), + ]) + .await; + + let mut stream = Box::pin(client.list_owned_objects(proto::ListOwnedObjectsRequest::default())); + + assert_eq!(stream.next().await.unwrap().unwrap().object_id(), "0x1"); + assert_eq!(stream.next().await.unwrap().unwrap().object_id(), "0x2"); + let status = stream.next().await.unwrap().unwrap_err(); + assert_eq!(status.code(), tonic::Code::InvalidArgument); + assert!(stream.next().await.is_none()); +} From 057dc06180cb8deb088aa7d8cd793f01c0ada7bd Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Wed, 15 Jul 2026 13:59:24 -0500 Subject: [PATCH 9/9] sui-rpc: poll the checkpoint subscription while executing execute_transaction_and_wait_for_checkpoint subscribed to the checkpoint stream and then held it unpolled while awaiting ExecuteTransaction and GetTransaction on the same channel -- the hold-and-wait shape at the root of the connection-starvation hangs. Checkpoints keep arriving whether or not anyone reads them, so the parked subscription pinned up to a stream window of the shared connection receive window for the whole execution phase, and once the execution phase outlasted the client's idle-body watchdog the subscription was reset, turning a successful execution into a spurious CheckpointStreamError. Restructure the method to scan the subscription concurrently with the execution phase: the digest is computed from the request up front (so an invalid request now also fails before any network work), and a select loop drives the execute-then-GetTransaction future and the digest scan together. Semantics are preserved: the subscription is opened before execution so the checkpoint cannot be missed, the GetTransaction duplicate-submission shortcut still takes precedence, and the wait timeout still covers only the confirmation phase. A scan result that arrives mid-execution (for example, a duplicate of an already executed transaction landing in a checkpoint) is remembered and used once execution completes. Add regression tests against a mock fullnode: a slow execution with a short watchdog (fails as CheckpointStreamError against the previous implementation), the duplicate-submission shortcut, and fail-fast validation of a request with no transaction. --- .../src/client/transaction_execution.rs | 156 ++++++++----- crates/sui-rpc/tests/execute_and_wait.rs | 216 ++++++++++++++++++ 2 files changed, 311 insertions(+), 61 deletions(-) create mode 100644 crates/sui-rpc/tests/execute_and_wait.rs diff --git a/crates/sui-rpc/src/client/transaction_execution.rs b/crates/sui-rpc/src/client/transaction_execution.rs index 431c22914f..fda4cc65e4 100644 --- a/crates/sui-rpc/src/client/transaction_execution.rs +++ b/crates/sui-rpc/src/client/transaction_execution.rs @@ -85,21 +85,6 @@ impl Client { request: impl tonic::IntoRequest, timeout: Duration, ) -> Result, ExecuteAndWaitError> { - // Subscribe to checkpoint stream before execution to avoid missing the transaction. - // Uses minimal read mask for efficiency since we only nee digest confirmation. - // Once server-side filtering is available, we should filter by transaction digest to - // further reduce bandwidth. - let mut checkpoint_stream = match self - .subscription_client() - .subscribe_checkpoints(SubscribeCheckpointsRequest::default().with_read_mask( - FieldMask::from_str("transactions.digest,sequence_number,summary.timestamp"), - )) - .await - { - Ok(stream) => stream.into_inner(), - Err(e) => return Err(ExecuteAndWaitError::RpcError(e)), - }; - // Calculate digest from the input transaction to avoid relying on response read mask let request = request.into_request(); let transaction = match request.get_ref().transaction_opt() { @@ -112,44 +97,33 @@ impl Client { Err(e) => return Err(ExecuteAndWaitError::ProtoConversionError(e)), }; - let mut response = match self.execution_client().execute_transaction(request).await { - Ok(resp) => resp, - Err(e) => return Err(ExecuteAndWaitError::RpcError(e)), - }; - - // First query the fullnode directly to see if it already has the txn. This is to handle - // the case where an already executed transaction is sent multiple times - if let Ok(resp) = self - .ledger_client() - .get_transaction( - GetTransactionRequest::default() - .with_digest(&executed_txn_digest) - .with_read_mask(FieldMask::from_str("digest,checkpoint,timestamp")), - ) + // Subscribe to checkpoint stream before execution to avoid missing the transaction. + // Uses minimal read mask for efficiency since we only nee digest confirmation. + // Once server-side filtering is available, we should filter by transaction digest to + // further reduce bandwidth. + let mut checkpoint_stream = match self + .subscription_client() + .subscribe_checkpoints(SubscribeCheckpointsRequest::default().with_read_mask( + FieldMask::from_str("transactions.digest,sequence_number,summary.timestamp"), + )) .await - && resp.get_ref().transaction().checkpoint_opt().is_some() { - let checkpoint = resp.get_ref().transaction().checkpoint(); - let timestamp = resp.get_ref().transaction().timestamp; - response - .get_mut() - .transaction_mut() - .set_checkpoint(checkpoint); - response.get_mut().transaction_mut().timestamp = timestamp; - return Ok(response); - } + Ok(stream) => stream.into_inner(), + Err(e) => return Err(ExecuteAndWaitError::RpcError(e)), + }; - // Wait for the transaction to appear in a checkpoint, at which point indexes will have been - // updated. - let timeout_future = tokio::time::sleep(timeout); - let checkpoint_future = async { + // Scan the subscription for the transaction's digest. Every RPC on + // this client shares one HTTP/2 connection, so this future must be + // polled concurrently with the execution phase below: a subscription + // parked while another call is awaited pins its flow-control window + // (checkpoints keep arriving whether or not anyone reads them) and, + // past the idle timeout, gets reset by the client's body watchdog. + let scan = async { while let Some(response) = checkpoint_stream.try_next().await? { let checkpoint = response.checkpoint(); for tx in checkpoint.transactions() { - let digest = tx.digest(); - - if digest == executed_txn_digest { + if tx.digest() == executed_txn_digest { return Ok((checkpoint.sequence_number(), checkpoint.summary().timestamp)); } } @@ -158,25 +132,85 @@ impl Client { "checkpoint stream ended unexpectedly", )) }; + tokio::pin!(scan); + + // Execute, then query the fullnode directly to see if it already has + // the txn in a checkpoint. This is to handle the case where an + // already executed transaction is sent multiple times. + let exec_and_check = async { + let response = self.execution_client().execute_transaction(request).await?; - tokio::select! { - result = checkpoint_future => { - match result { - Ok((checkpoint, timestamp)) => { - response - .get_mut() - .transaction_mut() - .set_checkpoint(checkpoint); - response.get_mut().transaction_mut().timestamp = timestamp; - Ok(response) + let already_checkpointed = match self + .ledger_client() + .get_transaction( + GetTransactionRequest::default() + .with_digest(&executed_txn_digest) + .with_read_mask(FieldMask::from_str("digest,checkpoint,timestamp")), + ) + .await + { + Ok(resp) if resp.get_ref().transaction().checkpoint_opt().is_some() => Some(( + resp.get_ref().transaction().checkpoint(), + resp.get_ref().transaction().timestamp, + )), + _ => None, + }; + + Ok::<_, tonic::Status>((response, already_checkpointed)) + }; + tokio::pin!(exec_and_check); + + // Drive execution and the scan together. The scan can complete first + // (for example, when a duplicate of an already executed transaction + // lands in a checkpoint mid-execution), so remember its outcome; the + // guard keeps a completed scan from being polled again. + let mut scan_result = None; + let (mut response, already_checkpointed) = loop { + tokio::select! { + exec = &mut exec_and_check => { + match exec { + Ok(ok) => break ok, + Err(e) => return Err(ExecuteAndWaitError::RpcError(e)), } - Err(e) => Err(ExecuteAndWaitError::CheckpointStreamError { response, error: e }) } - }, - _ = timeout_future => { - Err(ExecuteAndWaitError::CheckpointTimeout ( response)) + result = &mut scan, if scan_result.is_none() => { + scan_result = Some(result); + } } - } + }; + + // Wait for the transaction to appear in a checkpoint, at which point + // indexes will have been updated. The direct lookup takes precedence: + // when it already places the transaction in a checkpoint there is + // nothing to wait for, even if the scan failed in the meantime. + let (checkpoint, timestamp) = if let Some(found) = already_checkpointed { + found + } else { + let result = match scan_result { + Some(result) => result, + None => { + tokio::select! { + result = &mut scan => result, + _ = tokio::time::sleep(timeout) => { + return Err(ExecuteAndWaitError::CheckpointTimeout(response)); + } + } + } + }; + match result { + Ok(found) => found, + Err(e) => { + return Err(ExecuteAndWaitError::CheckpointStreamError { response, error: e }); + } + } + }; + + response + .get_mut() + .transaction_mut() + .set_checkpoint(checkpoint); + response.get_mut().transaction_mut().timestamp = timestamp; + Ok(response) } /// Retrieves the current reference gas price from the latest epoch information. diff --git a/crates/sui-rpc/tests/execute_and_wait.rs b/crates/sui-rpc/tests/execute_and_wait.rs new file mode 100644 index 0000000000..8e416a462b --- /dev/null +++ b/crates/sui-rpc/tests/execute_and_wait.rs @@ -0,0 +1,216 @@ +//! Regression tests for `execute_transaction_and_wait_for_checkpoint`. +//! +//! The method subscribes to the checkpoint stream before executing so it +//! cannot miss the transaction's checkpoint. Every RPC on a `Client` shares +//! one HTTP/2 connection, so the subscription must be polled *concurrently* +//! with the execution phase: a subscription parked while `ExecuteTransaction` +//! is awaited pins its flow-control window (checkpoints keep arriving whether +//! or not anyone reads them) and, past the idle timeout, gets reset by the +//! client's body watchdog — turning a successful execution into a spurious +//! `CheckpointStreamError`. + +use std::time::Duration; + +use proto::ledger_service_server::LedgerService; +use proto::ledger_service_server::LedgerServiceServer; +use proto::subscription_service_server::SubscriptionService; +use proto::subscription_service_server::SubscriptionServiceServer; +use proto::transaction_execution_service_server::TransactionExecutionService; +use proto::transaction_execution_service_server::TransactionExecutionServiceServer; +use sui_rpc::Client; +use sui_rpc::proto::sui::rpc::v2 as proto; + +/// Cadence of the mock checkpoint subscription. +const CHECKPOINT_TICK: Duration = Duration::from_millis(100); + +#[derive(Clone)] +struct MockServer { + /// Digest of the transaction under test; emitted in the checkpoint at + /// `digest_at_seq`, with filler digests before it. + digest: String, + digest_at_seq: u64, + /// How long `ExecuteTransaction` takes. + exec_delay: Duration, + /// When set, `GetTransaction` reports the transaction as already + /// included in this checkpoint (the duplicate-submission shortcut). + shortcut_checkpoint: Option, +} + +#[tonic::async_trait] +impl TransactionExecutionService for MockServer { + async fn execute_transaction( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> { + tokio::time::sleep(self.exec_delay).await; + Ok(tonic::Response::new( + proto::ExecuteTransactionResponse::default(), + )) + } +} + +#[tonic::async_trait] +impl LedgerService for MockServer { + async fn get_transaction( + &self, + _request: tonic::Request, + ) -> Result, tonic::Status> { + let mut response = proto::GetTransactionResponse::default(); + if let Some(checkpoint) = self.shortcut_checkpoint { + let mut transaction = proto::ExecutedTransaction::default(); + transaction.digest = Some(self.digest.clone()); + transaction.checkpoint = Some(checkpoint); + response.transaction = Some(transaction); + } + Ok(tonic::Response::new(response)) + } +} + +#[tonic::async_trait] +impl SubscriptionService for MockServer { + async fn subscribe_checkpoints( + &self, + _request: tonic::Request, + ) -> Result< + tonic::Response>, + tonic::Status, + > { + let digest = self.digest.clone(); + let digest_at_seq = self.digest_at_seq; + let stream = futures::stream::unfold(0u64, move |seq| { + let digest = digest.clone(); + async move { + tokio::time::sleep(CHECKPOINT_TICK).await; + let mut transaction = proto::ExecutedTransaction::default(); + transaction.digest = Some(if seq == digest_at_seq { + digest + } else { + "filler".to_owned() + }); + let mut checkpoint = proto::Checkpoint::default(); + checkpoint.sequence_number = Some(seq); + checkpoint.transactions = vec![transaction]; + let mut resp = proto::SubscribeCheckpointsResponse::default(); + resp.cursor = Some(seq); + resp.checkpoint = Some(checkpoint); + Some((Ok(resp), seq + 1)) + } + }); + Ok(tonic::Response::new(Box::pin(stream))) + } +} + +async fn spawn_mock_server(mock: MockServer) -> std::net::SocketAddr { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind mock server listener"); + let addr = listener.local_addr().expect("mock server local addr"); + tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(TransactionExecutionServiceServer::new(mock.clone())) + .add_service(LedgerServiceServer::new(mock.clone())) + .add_service(SubscriptionServiceServer::new(mock)) + .serve_with_incoming(tonic::transport::server::TcpIncoming::from(listener)) + .await + .expect("mock server exited with an error"); + }); + addr +} + +/// A minimal valid transaction, plus the digest the client will derive from +/// it. +fn test_transaction() -> (proto::Transaction, String) { + let transaction = sui_sdk_types::Transaction { + kind: sui_sdk_types::TransactionKind::ProgrammableTransaction( + sui_sdk_types::ProgrammableTransaction { + inputs: Vec::new(), + commands: Vec::new(), + }, + ), + sender: sui_sdk_types::Address::ZERO, + gas_payment: sui_sdk_types::GasPayment { + objects: Vec::new(), + owner: sui_sdk_types::Address::ZERO, + price: 1, + budget: 1, + }, + expiration: sui_sdk_types::TransactionExpiration::None, + }; + let digest = transaction.digest().to_string(); + (proto::Transaction::from(transaction), digest) +} + +/// The subscription must survive a slow execution: with an idle timeout well +/// below the execution latency, the stream is only alive at scan time if it +/// was polled throughout. Before the concurrent-scan fix, the watchdog reset +/// the parked subscription during execution and this returned +/// `CheckpointStreamError` despite a successful execution. +#[tokio::test(flavor = "multi_thread")] +async fn subscription_survives_slow_execution() { + let (transaction, digest) = test_transaction(); + let digest_at_seq = 50; + let addr = spawn_mock_server(MockServer { + digest, + digest_at_seq, + exec_delay: Duration::from_secs(4), + shortcut_checkpoint: None, + }) + .await; + + let mut client = Client::new(format!("http://{addr}")) + .expect("client") + .with_body_idle_timeout(Duration::from_millis(1500)); + + let request = proto::ExecuteTransactionRequest::default().with_transaction(transaction); + let response = client + .execute_transaction_and_wait_for_checkpoint(request, Duration::from_secs(30)) + .await + .expect("execution and checkpoint confirmation"); + assert_eq!(response.get_ref().transaction().checkpoint(), digest_at_seq); +} + +/// The duplicate-submission shortcut still short-circuits the checkpoint +/// wait: when `GetTransaction` already places the transaction in a +/// checkpoint, that checkpoint is returned without waiting for the +/// subscription scan (whose digest checkpoint here would only arrive far +/// past the wait timeout). +#[tokio::test(flavor = "multi_thread")] +async fn already_checkpointed_transaction_short_circuits() { + let (transaction, digest) = test_transaction(); + let addr = spawn_mock_server(MockServer { + digest, + digest_at_seq: 10_000, + exec_delay: Duration::ZERO, + shortcut_checkpoint: Some(7), + }) + .await; + + let mut client = Client::new(format!("http://{addr}")).expect("client"); + + let request = proto::ExecuteTransactionRequest::default().with_transaction(transaction); + let response = client + .execute_transaction_and_wait_for_checkpoint(request, Duration::from_secs(5)) + .await + .expect("execution with an already checkpointed transaction"); + assert_eq!(response.get_ref().transaction().checkpoint(), 7); +} + +/// A request without a transaction fails fast, before any network work. +#[tokio::test(flavor = "multi_thread")] +async fn missing_transaction_fails_without_rpc() { + // No server at this address: the validation error must surface before + // any RPC is attempted. + let mut client = Client::new("http://127.0.0.1:1").expect("client"); + + let err = client + .execute_transaction_and_wait_for_checkpoint( + proto::ExecuteTransactionRequest::default(), + Duration::from_secs(1), + ) + .await + .expect_err("request without a transaction"); + assert!(matches!( + err, + sui_rpc::client::ExecuteAndWaitError::MissingTransaction + )); +}