From 62d0fb623e598b752ae4d5f620d6c22f077b967c Mon Sep 17 00:00:00 2001
From: kingchenc <kingchencp@gmail.com>
Date: Fri, 22 May 2026 04:22:46 +0200
Subject: [PATCH] C3: add a read timeout and message-size limits to the Binance
 stream

connect() used connect_async with no WebSocketConfig and next_event
awaited the socket with no deadline, so a stalled server hung the feed
forever and an oversized message could force an unbounded allocation.
connect() now passes a WebSocketConfig capping message/frame size, and
next_event wraps the read in a 300s tokio timeout (well above Binance's
~3-minute ping), surfacing a stall as the new Error::Timeout.
---
 crates/wickra-data/src/error.rs        |  4 ++++
 crates/wickra-data/src/live/binance.rs | 33 ++++++++++++++++++++++----
 2 files changed, 32 insertions(+), 5 deletions(-)

diff --git a/crates/wickra-data/src/error.rs b/crates/wickra-data/src/error.rs
index d6b57267..d977ef06 100644
--- a/crates/wickra-data/src/error.rs
+++ b/crates/wickra-data/src/error.rs
@@ -20,6 +20,10 @@ pub enum Error {
     #[error("malformed payload: {0}")]
     Malformed(String),
 
+    /// A live-feed read exceeded its deadline.
+    #[error("read timed out")]
+    Timeout,
+
     #[cfg(feature = "live-binance")]
     #[error("websocket error: {0}")]
     WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
diff --git a/crates/wickra-data/src/live/binance.rs b/crates/wickra-data/src/live/binance.rs
index 10c92d58..9d6ab469 100644
--- a/crates/wickra-data/src/live/binance.rs
+++ b/crates/wickra-data/src/live/binance.rs
@@ -18,10 +18,13 @@
 //! # Ok(()) }
 //! ```
 
+use std::time::Duration;
+
 use futures_util::SinkExt;
 use futures_util::StreamExt;
 use serde::Deserialize;
 use tokio::net::TcpStream;
+use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
 use tokio_tungstenite::tungstenite::Message;
 use tokio_tungstenite::MaybeTlsStream;
 use tokio_tungstenite::WebSocketStream;
@@ -29,6 +32,18 @@ use tokio_tungstenite::WebSocketStream;
 use crate::error::{Error, Result};
 use wickra_core::Candle;
 
+/// Maximum time to wait for the next WebSocket frame before treating the
+/// connection as stalled. Binance pings roughly every 3 minutes, so a healthy
+/// but quiet stream stays comfortably inside this window.
+const READ_TIMEOUT: Duration = Duration::from_secs(300);
+
+/// Upper bound on an inbound WebSocket message. Kline frames are tiny; this
+/// only caps a pathological or hostile server from forcing an unbounded alloc.
+const MAX_MESSAGE_SIZE: usize = 8 << 20;
+
+/// Upper bound on a single inbound WebSocket frame.
+const MAX_FRAME_SIZE: usize = 2 << 20;
+
 /// Supported Binance kline intervals. The `as_str` value matches Binance's
 /// wire-format strings (`"1m"`, `"5m"`, `"1h"`, etc.).
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -161,7 +176,14 @@ impl BinanceKlineStream {
             streams.join("/")
         );
         let url = url::Url::parse(&url).map_err(|e| Error::Malformed(e.to_string()))?;
-        let (socket, _) = tokio_tungstenite::connect_async(url.as_str()).await?;
+        let ws_config = WebSocketConfig {
+            max_message_size: Some(MAX_MESSAGE_SIZE),
+            max_frame_size: Some(MAX_FRAME_SIZE),
+            ..WebSocketConfig::default()
+        };
+        let (socket, _) =
+            tokio_tungstenite::connect_async_with_config(url.as_str(), Some(ws_config), false)
+                .await?;
         Ok(Self {
             socket,
             interval,
@@ -182,13 +204,14 @@ impl BinanceKlineStream {
             return Ok(None);
         }
         loop {
-            let msg = match self.socket.next().await {
-                Some(Ok(m)) => m,
-                Some(Err(e)) => return Err(Error::from(e)),
-                None => {
+            let msg = match tokio::time::timeout(READ_TIMEOUT, self.socket.next()).await {
+                Ok(Some(Ok(m))) => m,
+                Ok(Some(Err(e))) => return Err(Error::from(e)),
+                Ok(None) => {
                     self.closed = true;
                     return Ok(None);
                 }
+                Err(_elapsed) => return Err(Error::Timeout),
             };
             match msg {
                 Message::Text(text) => {