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.
This commit is contained in:
@@ -20,6 +20,10 @@ pub enum Error {
|
|||||||
#[error("malformed payload: {0}")]
|
#[error("malformed payload: {0}")]
|
||||||
Malformed(String),
|
Malformed(String),
|
||||||
|
|
||||||
|
/// A live-feed read exceeded its deadline.
|
||||||
|
#[error("read timed out")]
|
||||||
|
Timeout,
|
||||||
|
|
||||||
#[cfg(feature = "live-binance")]
|
#[cfg(feature = "live-binance")]
|
||||||
#[error("websocket error: {0}")]
|
#[error("websocket error: {0}")]
|
||||||
WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
|
WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
|
||||||
|
|||||||
@@ -18,10 +18,13 @@
|
|||||||
//! # Ok(()) }
|
//! # Ok(()) }
|
||||||
//! ```
|
//! ```
|
||||||
|
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use futures_util::SinkExt;
|
use futures_util::SinkExt;
|
||||||
use futures_util::StreamExt;
|
use futures_util::StreamExt;
|
||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
use tokio::net::TcpStream;
|
use tokio::net::TcpStream;
|
||||||
|
use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
|
||||||
use tokio_tungstenite::tungstenite::Message;
|
use tokio_tungstenite::tungstenite::Message;
|
||||||
use tokio_tungstenite::MaybeTlsStream;
|
use tokio_tungstenite::MaybeTlsStream;
|
||||||
use tokio_tungstenite::WebSocketStream;
|
use tokio_tungstenite::WebSocketStream;
|
||||||
@@ -29,6 +32,18 @@ use tokio_tungstenite::WebSocketStream;
|
|||||||
use crate::error::{Error, Result};
|
use crate::error::{Error, Result};
|
||||||
use wickra_core::Candle;
|
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
|
/// Supported Binance kline intervals. The `as_str` value matches Binance's
|
||||||
/// wire-format strings (`"1m"`, `"5m"`, `"1h"`, etc.).
|
/// wire-format strings (`"1m"`, `"5m"`, `"1h"`, etc.).
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
@@ -161,7 +176,14 @@ impl BinanceKlineStream {
|
|||||||
streams.join("/")
|
streams.join("/")
|
||||||
);
|
);
|
||||||
let url = url::Url::parse(&url).map_err(|e| Error::Malformed(e.to_string()))?;
|
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 {
|
Ok(Self {
|
||||||
socket,
|
socket,
|
||||||
interval,
|
interval,
|
||||||
@@ -182,13 +204,14 @@ impl BinanceKlineStream {
|
|||||||
return Ok(None);
|
return Ok(None);
|
||||||
}
|
}
|
||||||
loop {
|
loop {
|
||||||
let msg = match self.socket.next().await {
|
let msg = match tokio::time::timeout(READ_TIMEOUT, self.socket.next()).await {
|
||||||
Some(Ok(m)) => m,
|
Ok(Some(Ok(m))) => m,
|
||||||
Some(Err(e)) => return Err(Error::from(e)),
|
Ok(Some(Err(e))) => return Err(Error::from(e)),
|
||||||
None => {
|
Ok(None) => {
|
||||||
self.closed = true;
|
self.closed = true;
|
||||||
return Ok(None);
|
return Ok(None);
|
||||||
}
|
}
|
||||||
|
Err(_elapsed) => return Err(Error::Timeout),
|
||||||
};
|
};
|
||||||
match msg {
|
match msg {
|
||||||
Message::Text(text) => {
|
Message::Text(text) => {
|
||||||
|
|||||||
Reference in New Issue
Block a user