Skip to content

fix(rpc): hold one heavy slot per HTTP batch #336

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rootulp merged 2 commits into from
Nov 9, 2026
Merged

fix(rpc): hold one heavy slot per HTTP batch #336

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
26b3ec5
fix(rpc): hold one heavy slot per HTTP batch
rootulp Nov 9, 2026
ed6f8a5
test(rpc): rebind scoped evidence after batch guard
rootulp Nov 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 2 additions & 2 deletions conformance/mainnet-rpc/readiness.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,8 @@
"bindings": {
"conformance/mainnet-rpc/READINESS.md": "c90b5bc2bf18f5f5b118f2fb75daa6da16bbf2a12168b78877dbc179ea3e4c85",
"conformance/mainnet-rpc/readiness.py": "b285feb0f3ab949254967dd7e4576312ae540893039b96be81f1d5f62212b407",
"conformance/mainnet-rpc/scope-audit.json": "dbfac97d34912fb4039e340eee58cda189a24193969fec2eb2916bad444efeba",
"conformance/mainnet-rpc/scope_audit.py": "5c5c9ab71394797f5a9d5219c372efec7b6aa64fcb5db692ebb76a96780d8333",
"conformance/mainnet-rpc/scope-audit.json": "3d07ce34368ca98e4eecb787cda4055bd1c5513e8b31855b7b3937b9a6525ccb",
"conformance/mainnet-rpc/scope_audit.py": "b47c98547ccaea569ab0e396bbcb5d9cc12ac2704e169870a529f14b0c5c64b5",
"conformance/mainnet-rpc/module-closure/evidence/manifest.json": "605a216586ca3d6d5328f4b36853344816347d9fbc1b1fe01c36c8d85c6703e6",
"conformance/mainnet-rpc/persistence/closure-evidence/manifest.json": "e8ce2ef7043048eff4a66a920ac65c8b3ed3cd213f2cdb510dd22264e9a3932b",
"conformance/mainnet-rpc/provider-soak/v10.5.0/2026-11-02-f51ce12/manifest.json": "d13b92ee1861a7a566b31f9e128850c1db4f79eb48ff183b20ed10c2112ee7bc",
Expand Down
2 changes: 1 addition & 1 deletion conformance/mainnet-rpc/scope-audit.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -466,7 +466,7 @@
{
"source": {
"path": "crates/tier-4-runtime/rpc/src/server.rs",
"sha256": "bbd2b4f1968359e6a0b84447cfdb838da7d608f93cff8886ce723180500fdbd7"
"sha256": "9e8efaf1335e124af116ee67fa6cf0eb4843497116f1dd99b2a9f58cc96549c4"
},
"function": "websocket_write_is_bounded"
},
Expand Down
2 changes: 1 addition & 1 deletion conformance/mainnet-rpc/scope_audit.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import sys

ROOT = Path(__file__).resolve().parents[2]
REVIEWED_AUDIT_SHA256 = "55fbfcc94324021a4cec1196d79ab4efeed658fbdeb4f1ddbfdf11f11b0c24ba"
REVIEWED_AUDIT_SHA256 = "a6d28da90597141cede0d835f4bdd60be9cb6d9e0913664378cca8d90158d5cd"
AUDIT = Path(__file__).with_name("scope-audit.json")
# The exact package/receipt/verifier set is reviewed code. A data edit cannot
# silently remove a campaign, swap its receipt class or skip its verifier.
Expand Down
1 change: 1 addition & 0 deletions crates/tier-4-runtime/rpc/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ ctypes = { path = "../../tier-1-protocol-types/ctypes" }
futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
hex = "0.4"
hyper = { version = "1", features = ["http1", "http2", "server"] }
http-body = "1"
hyper-util = { version = "0.1", features = ["tokio", "server", "server-auto", "service", "http1", "http2"] }
kvdb = { path = "../../tier-2-state-and-store/kvdb" }
mempool = { path = "../mempool" }
Expand Down
114 changes: 110 additions & 4 deletions crates/tier-4-runtime/rpc/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,20 +7,23 @@ use std::collections::BTreeMap;
use std::fs::File;
use std::io::BufReader;
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;

use axum::body::Body;
use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade};
use axum::extract::{FromRequestParts, State};
use axum::http::{HeaderMap, HeaderValue, Method, Request, Response, StatusCode, Version, header};
use bytes::Bytes;
use futures_util::{SinkExt as _, StreamExt as _};
use rustls_pki_types::pem::PemObject as _;
use rustls_pki_types::{CertificateDer, PrivateKeyDer};
use tokio::sync::mpsc;
use tokio_rustls::TlsAcceptor;

use crate::env::{ERR_HEAVY_LIMIT, Environment, MAX_QUERY_LENGTH, Route, find_route};
use crate::env::{ERR_HEAVY_LIMIT, Environment, HeavyGuard, MAX_QUERY_LENGTH, Route, find_route};
use crate::json::{self, J, Node, Val};
use crate::params::{self, Arg};
use crate::query::Query;
Expand Down Expand Up @@ -335,6 +338,18 @@ async fn call_route(
let Some(guard) = env.try_acquire(route) else {
return Err(CallError::Busy);
};
let result = call_route_admitted(env, route, args, remote).await;
drop(guard);
result
}

/// Calls a route whose heavy-request slot is already held by the caller.
async fn call_route_admitted(
env: &Arc<Environment>,
route: &'static Route,
args: Vec<Arg>,
remote: &str,
) -> Result<J, CallError> {
if route.name == "broadcast_tx_commit" {
return broadcast_tx_commit(
env,
Expand All @@ -347,7 +362,6 @@ async fn call_route(
let env2 = Arc::clone(env);
let name = route.name;
tokio::task::spawn_blocking(move || {
let _guard = guard;
if name == "abci_query" {
Comment on lines 364 to 365

@chatgpt-codex-connector chatgpt-codex-connector Bot Nov 9, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Keep the guard alive inside detached blocking work

If an HTTP client disconnects while this spawn_blocking task is running, Hyper drops the request future and therefore the guard held by jsonrpc_handler/call_route, but Tokio's already-started blocking task continues executing. Repeating heavy requests followed by immediate disconnects can consequently run an unbounded number of heavy operations despite max_concurrent_heavy_requests; the previous implementation moved the guard into this closure specifically so cancellation could not release the slot before the blocking work finished.

Useful? React with 👍 / 👎.

let arg = |i: usize| args.get(i).cloned().unwrap_or(Arg::Null);
env2.abci_query_response(arg(0).str(), arg(1).bytes(), arg(2).int(), arg(3).bool())
Expand Down Expand Up @@ -487,6 +501,32 @@ struct Ctx {
remote: SocketAddr,
}

/// Keeps a heavy-request slot until Hyper finishes consuming the response.
struct GuardedBody {
inner: Body,
_guard: HeavyGuard,
}

impl http_body::Body for GuardedBody {
type Data = Bytes;
type Error = axum::Error;

fn poll_frame(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
http_body::Body::poll_frame(Pin::new(&mut self.inner), cx)
}

fn is_end_stream(&self) -> bool {
http_body::Body::is_end_stream(&self.inner)
}

fn size_hint(&self) -> http_body::SizeHint {
http_body::Body::size_hint(&self.inner)
}
}

fn json_response(status: StatusCode, body: String, cache: bool) -> Response<Body> {
let mut r = Response::new(Body::from(body));
*r.status_mut() = status;
Expand All @@ -503,6 +543,23 @@ fn json_response(status: StatusCode, body: String, cache: bool) -> Response<Body
r
}

fn guarded_json_response(
status: StatusCode,
body: String,
cache: bool,
guard: Option<HeavyGuard>,
) -> Response<Body> {
let Some(guard) = guard else {
return json_response(status, body, cache);
};
let mut response = json_response(status, String::new(), cache);
*response.body_mut() = Body::new(GuardedBody {
inner: Body::from(body),
_guard: guard,
});
response
}

fn not_found() -> Response<Body> {
let mut r = Response::new(Body::from("404 page not found\n"));
*r.status_mut() = StatusCode::NOT_FOUND;
Expand Down Expand Up @@ -697,6 +754,7 @@ async fn jsonrpc_handler(ctx: &Ctx, body: &[u8], host: &str) -> Response<Body> {
};
let mut responses = Vec::new();
let mut cache = true;
let mut heavy_guard = None;
for req in requests {
let Some(id) = req.id.clone() else { continue };
let route = find_route(&req.method, ctx.env.config.unsafe_routes);
Expand Down Expand Up @@ -734,7 +792,20 @@ async fn jsonrpc_handler(ctx: &Ctx, body: &[u8], host: &str) -> Response<Body> {
if cache && !route.cacheable_with(&args) {
cache = false;
}
match call_route(&ctx.env, route, args, &ctx.remote.to_string()).await {
let result = if route.heavy {
if heavy_guard.is_none() {
let Some(guard) = ctx.env.try_acquire(route) else {
responses.push(RpcResponse::internal(Some(id), ERR_HEAVY_LIMIT));
cache = false;
continue;
};
heavy_guard = Some(guard);
}
call_route_admitted(&ctx.env, route, args, &ctx.remote.to_string()).await
} else {
call_route(&ctx.env, route, args, &ctx.remote.to_string()).await
};
match result {
Ok(j) => responses.push(RpcResponse::ok(Some(id), &j)),
Err(CallError::Busy) => {
responses.push(RpcResponse::internal(Some(id), ERR_HEAVY_LIMIT));
Expand All @@ -755,7 +826,12 @@ async fn jsonrpc_handler(ctx: &Ctx, body: &[u8], host: &str) -> Response<Body> {
if responses.is_empty() {
return Response::new(Body::empty());
}
json_response(StatusCode::OK, encode_responses(&responses), cache)
guarded_json_response(
StatusCode::OK,
encode_responses(&responses),
cache,
heavy_guard,
)
}

/// Applies the CORS policy (`rs/cors` subset).
Expand Down Expand Up @@ -1711,6 +1787,36 @@ pub(crate) mod broadcast_tests {
);
}

#[tokio::test]
async fn heavy_batch_holds_one_slot_until_response_body_is_consumed() {
let mut env = environment(Ok(abci::ResponseCheckTx::default()), 1_000);
Arc::get_mut(&mut env)
.expect("unshared test environment")
.config
.max_concurrent_heavy_requests = 1;
let ctx = Ctx {
env: Arc::clone(&env),
remote: "127.0.0.1:1234".parse().expect("remote address"),
};
let request = br#"[
{"jsonrpc":"2.0","id":1,"method":"genesis"},
{"jsonrpc":"2.0","id":2,"method":"genesis"}
]"#;

let response = jsonrpc_handler(&ctx, request, "localhost").await;
let heavy = find_route("genesis", false).expect("heavy route");
assert!(env.try_acquire(heavy).is_none());

let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.expect("response body");
let body = String::from_utf8(body.to_vec()).expect("UTF-8 response");
assert!(body.contains("\"id\":1"));
assert!(body.contains("\"id\":2"));
assert!(!body.contains(ERR_HEAVY_LIMIT));
assert!(env.try_acquire(heavy).is_some());
}

async fn wait_for_subscription(env: &Environment, client: &str) {
tokio::time::timeout(Duration::from_secs(1), async {
while env.event_bus.num_client_subscriptions(client) == 0 {
Expand Down
Loading