fix: fail fast on client send when payload exceeds MAX_TOTAL_CHUNKS

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);

rust/src/client_api/regular.rs

@@ -268,14 +268,29 @@ async fn process_request(
     req: Option<ClientRequest<'static>>,
     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)
         .map_err(Into::into)
         .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.
+        //
+        // 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);
         let chunks = chunk_request(msg, stream_id);
@@ -672,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<dyn std::error::Error + Send + Sync>> {
+        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<dyn std::error::Error + Send + Sync>> {
         let (listener, port) = bind_free_port().await;

rust/src/client_api/streaming.rs

@@ -56,6 +56,30 @@ pub fn chunk_request(data: Vec<u8>, stream_id: u32) -> Vec<ClientRequest<'static
         .collect()
 }
 
+/// Fail-fast check that a serialized payload of `len` bytes fits within
+/// [`MAX_TOTAL_CHUNKS`] before the client streams any chunk.
+///
+/// This mirrors the node's reassembly cap: [`ReassemblyBuffer::receive_chunk`]
+/// rejects a `total` exceeding [`MAX_TOTAL_CHUNKS`] with
+/// [`StreamError::TotalChunksTooLarge`] on the *first* chunk. Without this guard
+/// the client would stream the entire oversized payload (up to hundreds of MiB)
+/// before the node rejects it, wasting all of that bandwidth.
+///
+/// The chunk-count math matches [`chunk_bytes`] exactly, so the `total` computed
+/// here equals the `total` the node validates on the wire.
+pub fn ensure_chunkable(len: usize) -> 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();