From 665f69115cb3fc4cb26071c0c0d4d2375d48b1b0 Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Sun, 6 Sep 2026 16:07:45 +0200 Subject: [PATCH 1/5] fix(cubestore): Report a websocket peer that goes away as debug, not error A client that closes without a close handshake ends the connection the same way a graceful one does: the driver terminates connections by design on a heartbeat timeout, a write error and on dispose, and a killed process or a closed browser tab looks the same. Every one of those was logged as an error, so a rolling restart of the API produced one error line per connection and pushed the error rate of a healthy node over its alerting threshold. Classify the read-stream error instead: a vanished peer is reported at debug, a frame arriving after the peer's own close frame at warn, and everything else -- other protocol violations, TLS, capacity and IO errors that are not a vanished peer -- stays at error. IO errors are split by kind rather than lowered as a group, so a genuine failure of the connection is still an error. --- rust/cubestore/cubestore/src/http/mod.rs | 113 ++++++++++++++++++++++- 1 file changed, 112 insertions(+), 1 deletion(-) diff --git a/rust/cubestore/cubestore/src/http/mod.rs b/rust/cubestore/cubestore/src/http/mod.rs index 2f2893d0f8a1d..f6c62e26cce71 100644 --- a/rust/cubestore/cubestore/src/http/mod.rs +++ b/rust/cubestore/cubestore/src/http/mod.rs @@ -30,10 +30,12 @@ use http_auth_basic::Credentials; use log::error; use log::info; use log::trace; +use log::Level; use serde::Deserialize; use std::collections::{BTreeMap, HashMap}; use std::convert::TryFrom; use std::error::Error as StdError; +use std::io; use std::net::SocketAddr; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, SystemTime}; @@ -43,6 +45,7 @@ use tokio::io::{AsyncWriteExt, BufReader}; use tokio::sync::mpsc::Sender; use tokio::sync::{mpsc, Mutex}; use tokio_tungstenite::tungstenite; +use tokio_tungstenite::tungstenite::error::ProtocolError; use tokio_util::sync::CancellationToken; use warp::filters::ws::{Message, Ws}; use warp::http::StatusCode; @@ -149,6 +152,71 @@ fn message_too_large_reason( } } +/// Level at which an error from the websocket read stream is reported. +/// +/// A peer that disappears without a close handshake ends the connection the +/// same way a graceful client does: nothing on this side failed and there is +/// nothing for an operator to act on, while a fleet of clients going away at +/// once — a rolling restart of the API, say — produces one line per connection. +/// Reporting those as errors buries the transport failures that do need +/// attention, so they are separated here. +/// +/// `warp` boxes the underlying `tungstenite` error, so it has to be recovered +/// through `source()`; an error that is not one is left at `Error`, since it is +/// not known to be benign. +fn websocket_error_level(e: &warp::Error) -> Level { + match e + .source() + .and_then(|s| s.downcast_ref::()) + { + Some(e) => tungstenite_error_level(e), + None => Level::Error, + } +} + +fn tungstenite_error_level(e: &tungstenite::Error) -> Level { + match e { + // The socket reached end of file or was reset before a close frame + // arrived. Clients drop connections this way by design — `ws`'s + // `terminate()`, a killed process, a closed browser tab — and the + // server learns nothing else about them. + tungstenite::Error::Protocol(ProtocolError::ResetWithoutClosingHandshake) => Level::Debug, + tungstenite::Error::Io(io) => { + if is_peer_gone(io.kind()) { + Level::Debug + } else { + Level::Error + } + } + // A finished close handshake. The stream reports this as its end rather + // than as an error, so it is not expected here, but it is a normal + // close either way. + tungstenite::Error::ConnectionClosed | tungstenite::Error::AlreadyClosed => Level::Debug, + // The peer sent a frame after its own close frame. Harmless for this + // connection — whatever was already in flight raced the close — but a + // client that does it often is not closing correctly. + tungstenite::Error::Protocol(ProtocolError::ReceivedAfterClosing) => Level::Warn, + // Everything else is either a real transport failure (TLS, capacity, + // an IO error that is not a vanished peer) or a protocol violation the + // client should never commit, such as an invalid opcode, a masking + // violation or an oversized control frame. + _ => Level::Error, + } +} + +/// Whether an IO error means the peer is simply gone, as opposed to the +/// connection failing while the peer is still there. +fn is_peer_gone(kind: io::ErrorKind) -> bool { + matches!( + kind, + io::ErrorKind::ConnectionReset + | io::ErrorKind::ConnectionAborted + | io::ErrorKind::BrokenPipe + | io::ErrorKind::NotConnected + | io::ErrorKind::UnexpectedEof + ) +} + pub struct HttpServer { bind_address: String, sql_service: Arc, @@ -499,7 +567,10 @@ impl HttpServer { error!("Websocket close send error: {:?}", e) } } - None => error!("Websocket error: {:?}", e), + None => log::log!( + websocket_error_level(&e), + "Websocket error: {:?}", e + ), } break; } @@ -1456,6 +1527,46 @@ mod tests { use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; use url::Url; + #[test] + fn websocket_error_levels() { + // A peer gone without a close handshake, in either of the two shapes it + // reaches the read stream in. + assert_eq!( + tungstenite_error_level(&tungstenite::Error::Protocol( + ProtocolError::ResetWithoutClosingHandshake + )), + Level::Debug + ); + assert_eq!( + tungstenite_error_level(&tungstenite::Error::Io(io::Error::from( + io::ErrorKind::ConnectionReset + ))), + Level::Debug + ); + + // A close frame raced by an in-flight frame. + assert_eq!( + tungstenite_error_level(&tungstenite::Error::Protocol( + ProtocolError::ReceivedAfterClosing + )), + Level::Warn + ); + + // A protocol violation and an IO failure that is not a vanished peer. + assert_eq!( + tungstenite_error_level(&tungstenite::Error::Protocol(ProtocolError::InvalidOpcode( + 7 + ))), + Level::Error + ); + assert_eq!( + tungstenite_error_level(&tungstenite::Error::Io(io::Error::from( + io::ErrorKind::PermissionDenied + ))), + Level::Error + ); + } + /// Minimal SqlService that always replies with a fixed DataFrame, used to /// drive process_command in unit tests. struct StubService(Arc); From 48be7195cd845f2fb650130c737c6f060e66e93f Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Tue, 8 Sep 2026 20:54:00 +0200 Subject: [PATCH 2/5] fix(cubestore): Classify websocket send errors the same as read errors A connection breaks in both directions. A peer that goes away with a query in flight fails the send instead of the read, and the send towards a peer that already closed reports it as `ConnectionClosed`, `AlreadyClosed` or `Io(BrokenPipe)` -- so one error line per connection was still logged for a client that merely left. Route every write towards the peer through the same classifier, which never depended on the direction. Also drop the comments that restated the arms they sit on. --- rust/cubestore/cubestore/src/http/mod.rs | 78 +++++++++++------------- 1 file changed, 36 insertions(+), 42 deletions(-) diff --git a/rust/cubestore/cubestore/src/http/mod.rs b/rust/cubestore/cubestore/src/http/mod.rs index f6c62e26cce71..e459dc7dcad8b 100644 --- a/rust/cubestore/cubestore/src/http/mod.rs +++ b/rust/cubestore/cubestore/src/http/mod.rs @@ -152,18 +152,11 @@ fn message_too_large_reason( } } -/// Level at which an error from the websocket read stream is reported. -/// -/// A peer that disappears without a close handshake ends the connection the -/// same way a graceful client does: nothing on this side failed and there is -/// nothing for an operator to act on, while a fleet of clients going away at -/// once — a rolling restart of the API, say — produces one line per connection. -/// Reporting those as errors buries the transport failures that do need -/// attention, so they are separated here. -/// -/// `warp` boxes the underlying `tungstenite` error, so it has to be recovered -/// through `source()`; an error that is not one is left at `Error`, since it is -/// not known to be benign. +/// Level at which a websocket transport error is reported, in either +/// direction. A peer that vanished is not actionable, and a fleet of them going +/// away at once would otherwise bury the transport failures that are. `warp` +/// boxes the `tungstenite` error, so it comes back through `source()`; anything +/// else is not known to be benign and stays at `Error`. fn websocket_error_level(e: &warp::Error) -> Level { match e .source() @@ -176,30 +169,15 @@ fn websocket_error_level(e: &warp::Error) -> Level { fn tungstenite_error_level(e: &tungstenite::Error) -> Level { match e { - // The socket reached end of file or was reset before a close frame - // arrived. Clients drop connections this way by design — `ws`'s - // `terminate()`, a killed process, a closed browser tab — and the - // server learns nothing else about them. + // How a client normally goes away: `terminate()`, a killed process, a + // closed browser tab. Nothing else about it is ever known. tungstenite::Error::Protocol(ProtocolError::ResetWithoutClosingHandshake) => Level::Debug, - tungstenite::Error::Io(io) => { - if is_peer_gone(io.kind()) { - Level::Debug - } else { - Level::Error - } - } - // A finished close handshake. The stream reports this as its end rather - // than as an error, so it is not expected here, but it is a normal - // close either way. + // Reading from, or writing to, a peer that has already closed. tungstenite::Error::ConnectionClosed | tungstenite::Error::AlreadyClosed => Level::Debug, - // The peer sent a frame after its own close frame. Harmless for this - // connection — whatever was already in flight raced the close — but a - // client that does it often is not closing correctly. + tungstenite::Error::Io(io) if is_peer_gone(io.kind()) => Level::Debug, + // A frame that raced the peer's own close frame: harmless once, a + // client that closes incorrectly if it keeps happening. tungstenite::Error::Protocol(ProtocolError::ReceivedAfterClosing) => Level::Warn, - // Everything else is either a real transport failure (TLS, capacity, - // an IO error that is not a vanished peer) or a protocol violation the - // client should never commit, such as an invalid opcode, a masking - // violation or an oversized control frame. _ => Level::Error, } } @@ -527,7 +505,7 @@ impl HttpServer { )); match tokio::time::timeout(EVICTED_CLOSE_TIMEOUT, close).await { Ok(Ok(())) => {} - Ok(Err(e)) => error!("Websocket close send error: {:?}", e), + Ok(Err(e)) => log::log!(websocket_error_level(&e), "Websocket close send error: {:?}", e), Err(_) => log::warn!( "Timed out sending the close frame of an evicted websocket connection" ), @@ -538,7 +516,7 @@ impl HttpServer { trace!("Sending web socket response (process_id: {})", process_id); let send_res = web_socket.send(Message::binary(res.bytes())).await; if let Err(e) = send_res { - error!("Websocket message send error: {:?}", e) + log::log!(websocket_error_level(&e), "Websocket message send error: {:?}", e) } if res.should_close_connection() { log::warn!("Websocket connection closed"); @@ -564,7 +542,7 @@ impl HttpServer { Message::close_with(MESSAGE_TOO_BIG_CLOSE_CODE, reason) ).await; if let Err(e) = send_res { - error!("Websocket close send error: {:?}", e) + log::log!(websocket_error_level(&e), "Websocket close send error: {:?}", e) } } None => log::log!( @@ -602,7 +580,7 @@ impl HttpServer { Message::binary(HttpMessage { message_id, connection_id, command: HttpCommand::Error { error } }.bytes()) ).await; if let Err(e) = send_res { - error!("Websocket message send error: {:?}", e) + log::log!(websocket_error_level(&e), "Websocket message send error: {:?}", e) } continue; } @@ -615,7 +593,7 @@ impl HttpServer { Message::binary(HttpMessage { message_id, connection_id, command: HttpCommand::Error { error: e.to_string() } }.bytes()) ).await; if let Err(e) = send_res { - error!("Websocket message send error: {:?}", e) + log::log!(websocket_error_level(&e), "Websocket message send error: {:?}", e) } break; }, @@ -628,7 +606,7 @@ impl HttpServer { Message::binary(HttpMessage { message_id, connection_id, command: HttpCommand::Error { error: e.to_string() } }.bytes()) ).await; if let Err(e) = send_res { - error!("Websocket message send error: {:?}", e) + log::log!(websocket_error_level(&e), "Websocket message send error: {:?}", e) } break; } @@ -637,7 +615,7 @@ impl HttpServer { } else if msg.is_ping() { let send_res = web_socket.send(Message::pong(Vec::new())).await; if let Err(e) = send_res { - error!("Websocket ping send error: {:?}", e) + log::log!(websocket_error_level(&e), "Websocket ping send error: {:?}", e) } } else if msg.is_close() { break; @@ -1529,8 +1507,8 @@ mod tests { #[test] fn websocket_error_levels() { - // A peer gone without a close handshake, in either of the two shapes it - // reaches the read stream in. + // A peer gone without a close handshake, in either of the two shapes + // the read stream reports it in. assert_eq!( tungstenite_error_level(&tungstenite::Error::Protocol( ProtocolError::ResetWithoutClosingHandshake @@ -1544,6 +1522,22 @@ mod tests { Level::Debug ); + // A write towards a peer that has already left. + assert_eq!( + tungstenite_error_level(&tungstenite::Error::ConnectionClosed), + Level::Debug + ); + assert_eq!( + tungstenite_error_level(&tungstenite::Error::AlreadyClosed), + Level::Debug + ); + assert_eq!( + tungstenite_error_level(&tungstenite::Error::Io(io::Error::from( + io::ErrorKind::BrokenPipe + ))), + Level::Debug + ); + // A close frame raced by an in-flight frame. assert_eq!( tungstenite_error_level(&tungstenite::Error::Protocol( From f1a0ef4c57efd685479c62d43eb288da79d91d43 Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Tue, 8 Sep 2026 21:30:58 +0200 Subject: [PATCH 3/5] fix(cubestore): Import from a generated file in table_partition_split_threshold The test imported a CSV over HTTP from a third-party host, which refuses automated downloads, so the table was never created and the test failed wherever egress is not allowed. The location only ever contributed its size: the threshold under test comes from `estimate_rows`, which divides the file size by a fixed average row length without reading the file. Generate the CSV locally instead, with rows padded so the estimate still lands above the cap being asserted. Both assertions are unchanged, and the test no longer spends a download and ten retries getting there. --- rust/cubestore/cubestore/src/sql/mod.rs | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/rust/cubestore/cubestore/src/sql/mod.rs b/rust/cubestore/cubestore/src/sql/mod.rs index 19a0dbe0533d9..21fe0f6252adc 100644 --- a/rust/cubestore/cubestore/src/sql/mod.rs +++ b/rust/cubestore/cubestore/src/sql/mod.rs @@ -4878,13 +4878,24 @@ mod tests { config.max_partition_split_threshold = 200; config }).start_test_worker(async move |_| { - let url = "https://data.wprdc.org/dataset/0b584c84-7e35-4f4d-a5a2-b01697470c0f/resource/e95dd941-8e47-4460-9bd8-1e51c194370b/download/bikepghpublic.csv"; + // The threshold is derived from the size of the location, + // not from its contents: see ImportServiceImpl::estimate_rows. + // Rows are padded so the estimate lands above + // max_partition_split_threshold per select worker, which is + // what makes the cap observable. + let path = env::temp_dir().join(format!("{}.csv", test_name)); + let padding = "x".repeat(256); + let mut csv = "Response ID,Start Date,End Date\n".to_string(); + for id in 0..813 { + csv += &format!("{},2020-01-01T00:00:00.000Z,{}\n", id, padding); + } + tokio::fs::write(&path, csv).await?; service .exec_query("CREATE SCHEMA IF NOT EXISTS foo") .await?.collect().await?; - let create_table_sql = format!("CREATE TABLE foo.bikes (`Response ID` int, `Start Date` text, `End Date` text) LOCATION '{}'", url); + let create_table_sql = format!("CREATE TABLE foo.bikes (`Response ID` int, `Start Date` text, `End Date` text) LOCATION '{}'", path.to_string_lossy()); service.exec_query(&create_table_sql).await?.collect().await?; From e799245a5bf4611b270f6b5a5037afa3a5773904 Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Wed, 9 Sep 2026 11:45:52 +0200 Subject: [PATCH 4/5] fix(cubestore): Serve create_table_with_url's data ourselves The test imported a CSV over HTTP from a third-party host, which refuses the download, so the table never became ready and the test ran into the harness timeout. This is the only test covering the HTTP import path, so the location has to stay a URL rather than become a local file. Serve the CSV from a listener the test owns, as the TODO the test carried asked for. The body is held back for a moment because the test asserts that a concurrent query cannot see a table whose import is still running, and a local download would otherwise finish before the query is planned. --- .../cubestore-sql-tests/src/files.rs | 77 +++++++++++++++++++ .../cubestore-sql-tests/src/tests.rs | 13 +++- 2 files changed, 87 insertions(+), 3 deletions(-) diff --git a/rust/cubestore/cubestore-sql-tests/src/files.rs b/rust/cubestore/cubestore-sql-tests/src/files.rs index e42cc6a7ca2dd..cb36708b4369c 100644 --- a/rust/cubestore/cubestore-sql-tests/src/files.rs +++ b/rust/cubestore/cubestore-sql-tests/src/files.rs @@ -2,9 +2,14 @@ use cubestore::CubeError; use flate2::read::GzDecoder; use std::io::Cursor; use std::io::Write; +use std::net::SocketAddr; use std::path::Path; +use std::time::Duration; use tar::Archive; use tempfile::NamedTempFile; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::task::JoinHandle; pub fn write_tmp_file(text: &str) -> Result { let mut file = NamedTempFile::new()?; @@ -55,3 +60,75 @@ pub fn recursive_copy_directory(from: &Path, to: &Path) -> Result<(), CubeError> Ok(()) } + +/// Serves one in-memory file over HTTP, for as long as this value is alive. +pub struct TestFileServer { + addr: SocketAddr, + task: JoinHandle<()>, +} + +impl TestFileServer { + pub fn url(&self, name: &str) -> String { + format!("http://{}/{}", self.addr, name) + } +} + +impl Drop for TestFileServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +/// `download_delay` holds back the body, so a caller can observe the state a +/// location is in while its download is still running. Head requests are +/// answered immediately. +pub async fn serve_file( + body: String, + download_delay: Duration, +) -> Result { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let addr = listener.local_addr()?; + let task = tokio::spawn(async move { + while let Ok((socket, _)) = listener.accept().await { + let body = body.clone(); + tokio::spawn(async move { + if let Err(e) = serve_one_request(socket, &body, download_delay).await { + log::error!("Test file server: {}", e); + } + }); + } + }); + Ok(TestFileServer { addr, task }) +} + +async fn serve_one_request( + mut socket: TcpStream, + body: &str, + download_delay: Duration, +) -> Result<(), CubeError> { + let mut request = Vec::new(); + let mut buf = [0u8; 1024]; + while !request.windows(4).any(|w| w == b"\r\n\r\n") { + let read = socket.read(&mut buf).await?; + if read == 0 { + return Ok(()); + } + request.extend_from_slice(&buf[..read]); + } + + let mut response = format!( + "HTTP/1.1 200 OK\r\n\ + Content-Type: text/csv\r\n\ + Content-Length: {}\r\n\ + Connection: close\r\n\r\n", + body.len() + ); + if !request.starts_with(b"HEAD ") { + tokio::time::sleep(download_delay).await; + response += body; + } + + socket.write_all(response.as_bytes()).await?; + socket.shutdown().await?; + Ok(()) +} diff --git a/rust/cubestore/cubestore-sql-tests/src/tests.rs b/rust/cubestore/cubestore-sql-tests/src/tests.rs index 205b1420fc3ca..edfa97d66dc09 100644 --- a/rust/cubestore/cubestore-sql-tests/src/tests.rs +++ b/rust/cubestore/cubestore-sql-tests/src/tests.rs @@ -1,4 +1,4 @@ -use crate::files::write_tmp_file; +use crate::files::{serve_file, write_tmp_file}; use crate::rows::{rows, NULL}; use crate::SqlClient; use async_compression::tokio::write::GzipEncoder; @@ -2469,8 +2469,15 @@ async fn create_table_with_csv_no_header_and_quotes( } async fn create_table_with_url(service: Box) -> Result<(), CubeError> { - // TODO serve this data ourselves - let url = "https://data.wprdc.org/dataset/0b584c84-7e35-4f4d-a5a2-b01697470c0f/resource/e95dd941-8e47-4460-9bd8-1e51c194370b/download/bikepghpublic.csv"; + let mut csv = "Response ID,Start Date,End Date\n".to_string(); + for id in 0..813 { + csv += &format!("{},2020-01-01T00:00:00.000Z,2020-01-02T00:00:00.000Z\n", id); + } + // The body is held back because the query below has to run against a table + // whose import has not finished: only ready tables are visible to the + // planner. + let server = serve_file(csv, Duration::from_millis(500)).await?; + let url = server.url("bikepghpublic.csv"); service .exec_query("CREATE SCHEMA IF NOT EXISTS foo") From 880d8367fc8d3ab3d91035150a06186d893676a0 Mon Sep 17 00:00:00 2001 From: Aleksandr Romanenko Date: Wed, 9 Sep 2026 11:45:53 +0200 Subject: [PATCH 5/5] refactor(cubestore): Drop the websocket error level unit test The test restated the match arms of the classification it covered, so it could only ever fail by being edited alongside them. --- rust/cubestore/cubestore/src/http/mod.rs | 56 ------------------------ 1 file changed, 56 deletions(-) diff --git a/rust/cubestore/cubestore/src/http/mod.rs b/rust/cubestore/cubestore/src/http/mod.rs index e459dc7dcad8b..5c6ec31df3a72 100644 --- a/rust/cubestore/cubestore/src/http/mod.rs +++ b/rust/cubestore/cubestore/src/http/mod.rs @@ -1505,62 +1505,6 @@ mod tests { use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; use url::Url; - #[test] - fn websocket_error_levels() { - // A peer gone without a close handshake, in either of the two shapes - // the read stream reports it in. - assert_eq!( - tungstenite_error_level(&tungstenite::Error::Protocol( - ProtocolError::ResetWithoutClosingHandshake - )), - Level::Debug - ); - assert_eq!( - tungstenite_error_level(&tungstenite::Error::Io(io::Error::from( - io::ErrorKind::ConnectionReset - ))), - Level::Debug - ); - - // A write towards a peer that has already left. - assert_eq!( - tungstenite_error_level(&tungstenite::Error::ConnectionClosed), - Level::Debug - ); - assert_eq!( - tungstenite_error_level(&tungstenite::Error::AlreadyClosed), - Level::Debug - ); - assert_eq!( - tungstenite_error_level(&tungstenite::Error::Io(io::Error::from( - io::ErrorKind::BrokenPipe - ))), - Level::Debug - ); - - // A close frame raced by an in-flight frame. - assert_eq!( - tungstenite_error_level(&tungstenite::Error::Protocol( - ProtocolError::ReceivedAfterClosing - )), - Level::Warn - ); - - // A protocol violation and an IO failure that is not a vanished peer. - assert_eq!( - tungstenite_error_level(&tungstenite::Error::Protocol(ProtocolError::InvalidOpcode( - 7 - ))), - Level::Error - ); - assert_eq!( - tungstenite_error_level(&tungstenite::Error::Io(io::Error::from( - io::ErrorKind::PermissionDenied - ))), - Level::Error - ); - } - /// Minimal SqlService that always replies with a fixed DataFrame, used to /// drive process_command in unit tests. struct StubService(Arc);