diff --git a/crates/sui-rpc/Cargo.toml b/crates/sui-rpc/Cargo.toml index 68ae029cf3..a8d1f27142 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" @@ -57,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"] } +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/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/src/client/mod.rs b/crates/sui-rpc/src/client/mod.rs index ac06bca280..c9603635f9 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; @@ -51,12 +56,52 @@ 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; + +/// 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. 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 +/// connection fails with `DeadlineExceeded` instead of hanging forever. #[derive(Clone)] pub struct Client { uri: http::Uri, + endpoint: tonic::transport::Endpoint, 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, @@ -78,14 +123,32 @@ 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. 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) + /// 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, + body_idle_timeout: Some(DEFAULT_BODY_IDLE_TIMEOUT), request_layer: None, } } @@ -99,33 +162,136 @@ 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 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, + 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 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 + /// 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 @@ -212,6 +378,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) @@ -224,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), ) } @@ -319,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/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/src/client/watchdog.rs b/crates/sui-rpc/src/client/watchdog.rs new file mode 100644 index 0000000000..b2f87be4e0 --- /dev/null +++ b/crates/sui-rpc/src/client/watchdog.rs @@ -0,0 +1,711 @@ +//! 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. +//! +//! 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; +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, + }; + // 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?; + if idle_timeout.is_none() && deadline.is_none() { + return Ok(response); + } + 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. +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: 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, deadline)); + 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 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: Option, + deadline: Option, +) { + 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; + } +} + +/// 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 -- 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: Option, +) -> Option { + loop { + // 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)) => { + permit.send(Ok(frame)); + ControlFlow::Continue(()) + } + Some(Err(status)) => { + permit.send(Err(status)); + ControlFlow::Break(()) + } + None => ControlFlow::Break(()), + } + }; + + 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, + } + } +} + +#[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, Some(Duration::from_secs(30)), None); + + 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, Some(Duration::from_secs(30)), None); + + 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, Some(Duration::from_secs(30)), None); + + 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, Some(Duration::from_secs(30)), None); + + // 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 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(); + + // 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); + } + + /// 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(); + let mut watched = WatchdogBody::spawn(body, Some(Duration::from_secs(30)), None); + + 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, Some(Duration::from_secs(30)), None); + 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), 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" + ); + } +} 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 + )); +} 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()); +} 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" + ); +} 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")); +}