From ba757e2723e31c3f1ecaed9b0dee286f4d36b869 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Wed, 1 Jul 2026 12:10:05 -0500 Subject: [PATCH 1/2] fix: fail fast on client send when payload exceeds MAX_TOTAL_CHUNKS The WebSocket client's send path chunked a large ClientRequest and streamed every chunk without checking MAX_TOTAL_CHUNKS. When a payload exceeds the cap (256 chunks x 256 KiB = 64 MiB), the client streamed the ENTIRE oversized payload to the node, which rejects the first chunk during reassembly (ReassemblyBuffer::receive_chunk returns TotalChunksTooLarge). Every byte after the first chunk was wasted bandwidth (a ~450 MiB payload would upload in full before the node rejects it). Add a reusable, tested ensure_chunkable(len) helper in streaming.rs that mirrors the node's reassembly cap, and call it on both send paths (regular/native and browser/wasm) before any chunk is sent, so an oversized request fails fast with a clear TotalChunksTooLarge error and nothing is streamed. Co-Authored-By: Claude Opus 4.8 (1M context) --- rust/src/client_api/browser.rs | 15 ++++++++- rust/src/client_api/regular.rs | 8 ++++- rust/src/client_api/streaming.rs | 58 ++++++++++++++++++++++++++++++++ 3 files changed, 79 insertions(+), 2 deletions(-) diff --git a/rust/src/client_api/browser.rs b/rust/src/client_api/browser.rs index 4e30d48..9a6e72c 100644 --- a/rust/src/client_api/browser.rs +++ b/rust/src/client_api/browser.rs @@ -170,7 +170,7 @@ impl WebApi { } pub async fn send(&mut self, request: ClientRequest<'static>) -> Result<(), Error> { - use super::streaming::{chunk_request, CHUNK_THRESHOLD}; + use super::streaming::{chunk_request, ensure_chunkable, CHUNK_THRESHOLD}; // Check WebSocket ready state before sending. // Per WebSocket spec, send() silently discards data when socket is CLOSING/CLOSED. @@ -195,6 +195,19 @@ impl WebApi { let send = bincode::serialize(&request)?; if send.len() > CHUNK_THRESHOLD { + // Fail fast if the payload would exceed the node's reassembly cap + // (ReassemblyBuffer::receive_chunk rejects total > MAX_TOTAL_CHUNKS on + // the first chunk). Refuse to send anything rather than streaming the + // whole oversized payload just to have the node reject it. + if let Err(e) = ensure_chunkable(send.len()) { + let err = serde_json::json!({ + "error": format!("{e}"), + "origin": "chunk cap check", + "request": format!("{request:?}"), + }); + (self.error_handler)(Error::ConnectionError(err.clone())); + return Err(Error::ConnectionError(err)); + } let stream_id = self.next_stream_id; self.next_stream_id = self.next_stream_id.wrapping_add(1); let chunks = chunk_request(send, stream_id); diff --git a/rust/src/client_api/regular.rs b/rust/src/client_api/regular.rs index 35ac71e..b2f0cb8 100644 --- a/rust/src/client_api/regular.rs +++ b/rust/src/client_api/regular.rs @@ -268,7 +268,7 @@ async fn process_request( req: Option>, next_stream_id: &mut u32, ) -> Result<(), Error> { - use super::streaming::{chunk_request, CHUNK_THRESHOLD}; + use super::streaming::{chunk_request, ensure_chunkable, CHUNK_THRESHOLD}; let req = req.ok_or(Error::ChannelClosed)?; let msg = bincode::serialize(&req) @@ -276,6 +276,12 @@ async fn process_request( .map_err(Error::OtherError)?; if msg.len() > CHUNK_THRESHOLD { + // Fail fast if the payload would exceed the node's reassembly cap + // (ReassemblyBuffer::receive_chunk rejects total > MAX_TOTAL_CHUNKS on the + // first chunk). Refuse to send anything rather than streaming the whole + // oversized payload just to have the node reject it. The error propagates + // to the caller via the request handler. + ensure_chunkable(msg.len()).map_err(|e| Error::OtherError(e.into()))?; let stream_id = *next_stream_id; *next_stream_id = next_stream_id.wrapping_add(1); let chunks = chunk_request(msg, stream_id); diff --git a/rust/src/client_api/streaming.rs b/rust/src/client_api/streaming.rs index 8db1df5..ae9353f 100644 --- a/rust/src/client_api/streaming.rs +++ b/rust/src/client_api/streaming.rs @@ -56,6 +56,30 @@ pub fn chunk_request(data: Vec, stream_id: u32) -> Vec Result<(), StreamError> { + let total = len.div_ceil(CHUNK_SIZE).max(1); + if total > MAX_TOTAL_CHUNKS as usize { + return Err(StreamError::TotalChunksTooLarge { + // Clamp only guards against a theoretical >u32::MAX-chunk (multi-TiB) + // payload; the comparison above is done in `usize` so it is exact. + total: total.min(u32::MAX as usize) as u32, + max: MAX_TOTAL_CHUNKS, + }); + } + Ok(()) +} + /// Split a serialized response payload into `StreamChunk` host response variants. /// /// Uses `Bytes::slice()` internally for zero-copy: each chunk shares the @@ -609,6 +633,40 @@ mod tests { assert_eq!(chunks3.len(), 3); } + #[test] + fn ensure_chunkable_rejects_oversized() { + // A length that needs more than MAX_TOTAL_CHUNKS chunks must be rejected, + // mirroring the node's ReassemblyBuffer cap. Only the length integer is + // passed — no multi-MiB payload is allocated. + let len = (MAX_TOTAL_CHUNKS as usize + 1) * CHUNK_SIZE; + match ensure_chunkable(len) { + Err(StreamError::TotalChunksTooLarge { total, max }) => { + assert_eq!(total, MAX_TOTAL_CHUNKS + 1); + assert_eq!(max, MAX_TOTAL_CHUNKS); + } + other => panic!("expected TotalChunksTooLarge, got {other:?}"), + } + } + + #[test] + fn ensure_chunkable_accepts_at_limit() { + // Exactly MAX_TOTAL_CHUNKS chunks is the largest allowed payload. + let at_limit = MAX_TOTAL_CHUNKS as usize * CHUNK_SIZE; + assert!(ensure_chunkable(at_limit).is_ok()); + // One byte over the exact multiple needs one more chunk → rejected. + assert!(matches!( + ensure_chunkable(at_limit + 1), + Err(StreamError::TotalChunksTooLarge { .. }) + )); + } + + #[test] + fn ensure_chunkable_accepts_small() { + assert!(ensure_chunkable(0).is_ok()); + assert!(ensure_chunkable(1024).is_ok()); + assert!(ensure_chunkable(CHUNK_THRESHOLD).is_ok()); + } + #[test] fn remove_stream_cleans_up() { let mut buf = ReassemblyBuffer::new(); From a5eb7dc77af9b37db6877f6d5e8c422adf854434 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Wed, 1 Jul 2026 12:34:58 -0500 Subject: [PATCH 2/2] test: pin send-path chunk-limit guard against reintroduction Add a call-site regression test (test_send_oversized_rejected_before_streaming) that drives a real WebApi over a localhost socket with an oversized Put (state one byte over the 64 MiB MAX_TOTAL_CHUNKS * CHUNK_SIZE cap) and asserts recv() returns a "exceeds maximum" (TotalChunksTooLarge) error. Verified that removing the ensure_chunkable(...) guard from process_request makes this test fail (the client streams the payload and recv() instead returns a connection-reset error), so the guard can't be silently deleted without CI catching it. The prior ensure_chunkable unit tests only covered the helper, not the wiring. Also document at the guard call site that returning Err tears down the WebApi connection (request_handler break), which is intentional and acceptable for an unsendable over-cap request: the error is still delivered to recv() first, and the browser/wasm path surfaces it without teardown. We deliberately do not thread response_tx through process_request for per-request error reporting. Co-Authored-By: Claude Opus 4.8 (1M context) --- rust/src/client_api/regular.rs | 90 +++++++++++++++++++++++++++++++++- 1 file changed, 88 insertions(+), 2 deletions(-) diff --git a/rust/src/client_api/regular.rs b/rust/src/client_api/regular.rs index b2f0cb8..1da9143 100644 --- a/rust/src/client_api/regular.rs +++ b/rust/src/client_api/regular.rs @@ -279,8 +279,17 @@ async fn process_request( // Fail fast if the payload would exceed the node's reassembly cap // (ReassemblyBuffer::receive_chunk rejects total > MAX_TOTAL_CHUNKS on the // first chunk). Refuse to send anything rather than streaming the whole - // oversized payload just to have the node reject it. The error propagates - // to the caller via the request handler. + // oversized payload just to have the node reject it. + // + // Returning `Err` here breaks the request_handler loop (`break err`), + // which tears down the WebApi connection. That is intentional and + // acceptable: an over-cap request is unsendable and out-of-spec (>64 MiB, + // already above the 50 MiB MAX_STATE_SIZE), and the error is still + // delivered to the caller via `recv()` before teardown. (The browser/wasm + // path surfaces the same error to the JS caller without tearing down the + // connection.) We deliberately do not thread `response_tx` through here to + // report the error per-request; the extra plumbing isn't worth it for a + // request that cannot be sent regardless. ensure_chunkable(msg.len()).map_err(|e| Error::OtherError(e.into()))?; let stream_id = *next_stream_id; *next_stream_id = next_stream_id.wrapping_add(1); @@ -678,6 +687,83 @@ mod test { Ok(()) } + /// Regression test pinning the send-path chunk-limit guard in + /// `process_request`. An oversized request (serialized length above the + /// 64 MiB `MAX_TOTAL_CHUNKS * CHUNK_SIZE` cap) must be rejected locally with a + /// `TotalChunksTooLarge` error *before* any chunk is streamed. + /// + /// Acceptance criterion: if the `ensure_chunkable(...)` call is removed from + /// `process_request`, the client instead streams the whole payload and the + /// delivered error no longer mentions "exceeds maximum", so this test fails. + /// (Verified by temporarily removing the guard.) + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_send_oversized_rejected_before_streaming( + ) -> Result<(), Box> { + use crate::client_api::streaming::{CHUNK_SIZE, MAX_TOTAL_CHUNKS}; + use crate::client_api::ContractRequest; + use crate::prelude::{ + ContractCode, ContractContainer, ContractWasmAPIVersion, Parameters, RelatedContracts, + WrappedContract, WrappedState, + }; + use std::sync::Arc; + + let (listener, port) = bind_free_port().await; + // The server only completes the WS handshake, then drains anything the + // client sends until the client disconnects or a short idle window + // elapses, then drops. With the guard present the client sends nothing (it + // fails locally); if the guard were removed it would stream ~64 MiB of + // chunks, which this drain loop absorbs so the send path can't deadlock. + let server = tokio::task::spawn(async move { + let (tcp, _) = tokio::time::timeout(Duration::from_secs(5), listener.accept()) + .await + .expect("accept timed out") + .expect("accept failed"); + let mut ws = tokio_tungstenite::accept_async(tcp) + .await + .expect("ws handshake failed"); + let _ = tokio::time::timeout(Duration::from_secs(3), async { + while let Some(Ok(_)) = ws.next().await {} + }) + .await; + }); + + let (ws_conn, _) = + tokio_tungstenite::connect_async(format!("ws://localhost:{port}/")).await?; + let mut client = WebApi::start(ws_conn); + + // A Put whose state is one byte over the 64 MiB (256 * 256 KiB) chunk cap; + // serialization overhead only makes it larger, so it needs more than + // MAX_TOTAL_CHUNKS chunks. This is the single deliberate ~64 MiB alloc. + let oversized_state = MAX_TOTAL_CHUNKS as usize * CHUNK_SIZE + 1; + let code = Arc::new(ContractCode::from(vec![1, 2, 3])); + let contract = ContractContainer::Wasm(ContractWasmAPIVersion::V1(WrappedContract::new( + code, + Parameters::from(Vec::new()), + ))); + let request = ClientRequest::ContractOp(ContractRequest::Put { + contract, + state: WrappedState::new(vec![0u8; oversized_state]), + related_contracts: RelatedContracts::new(), + subscribe: false, + blocking_subscribe: false, + }); + + client.send(request).await?; + let err = client + .recv() + .await + .expect_err("oversized request must be rejected, not streamed"); + let msg = format!("{err}"); + assert!( + msg.contains("exceeds maximum"), + "expected a TotalChunksTooLarge error from the send-path guard, got: {msg}" + ); + + drop(client); + let _ = tokio::time::timeout(Duration::from_secs(5), server).await; + Ok(()) + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_recv() -> Result<(), Box> { let (listener, port) = bind_free_port().await;