From 7731764f63c415c40c3430d1a40fcda1ae6c44f8 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Wed, 8 Jul 2026 12:01:26 +0530 Subject: [PATCH 1/2] http disconnect cancel smoketest --- crates/smoketests/modules/Cargo.lock | 8 ++ crates/smoketests/modules/Cargo.toml | 1 + .../client-connection-http-cancel/Cargo.toml | 12 ++ .../client-connection-http-cancel/src/lib.rs | 30 +++++ .../smoketests/client_connection_errors.rs | 119 +++++++++++++++++- 5 files changed, 169 insertions(+), 1 deletion(-) create mode 100644 crates/smoketests/modules/client-connection-http-cancel/Cargo.toml create mode 100644 crates/smoketests/modules/client-connection-http-cancel/src/lib.rs diff --git a/crates/smoketests/modules/Cargo.lock b/crates/smoketests/modules/Cargo.lock index 2ce4b735807..e7adacae94a 100644 --- a/crates/smoketests/modules/Cargo.lock +++ b/crates/smoketests/modules/Cargo.lock @@ -609,6 +609,14 @@ dependencies = [ "spacetimedb", ] +[[package]] +name = "smoketest-module-client-connection-http-cancel" +version = "0.1.0" +dependencies = [ + "log", + "spacetimedb", +] + [[package]] name = "smoketest-module-client-connection-reject" version = "0.1.0" diff --git a/crates/smoketests/modules/Cargo.toml b/crates/smoketests/modules/Cargo.toml index c23de0030f6..c5a3faffebb 100644 --- a/crates/smoketests/modules/Cargo.toml +++ b/crates/smoketests/modules/Cargo.toml @@ -82,6 +82,7 @@ members = [ "delete-database", "client-connection-reject", "client-connection-disconnect-panic", + "client-connection-http-cancel", "sql-connect-hook", # Log filtering tests diff --git a/crates/smoketests/modules/client-connection-http-cancel/Cargo.toml b/crates/smoketests/modules/client-connection-http-cancel/Cargo.toml new file mode 100644 index 00000000000..46ecd40b509 --- /dev/null +++ b/crates/smoketests/modules/client-connection-http-cancel/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "smoketest-module-client-connection-http-cancel" +version = "0.1.0" +edition = "2021" +publish = false + +[lib] +crate-type = ["cdylib"] + +[dependencies] +spacetimedb.workspace = true +log.workspace = true diff --git a/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs b/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs new file mode 100644 index 00000000000..be0548aa6be --- /dev/null +++ b/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs @@ -0,0 +1,30 @@ +use spacetimedb::{log, ReducerContext, Table}; + +#[spacetimedb::table(accessor = marker, public)] +pub struct Marker { + id: u32, +} + +#[spacetimedb::reducer(init)] +pub fn init(ctx: &ReducerContext) { + ctx.db.marker().insert(Marker { id: 0 }); +} + +#[spacetimedb::reducer(client_connected)] +pub fn connected(ctx: &ReducerContext) { + log::info!("http_cancel_repro: connected {}", ctx.sender()); +} + +#[spacetimedb::reducer(client_disconnected)] +pub fn disconnected(ctx: &ReducerContext) { + log::info!("http_cancel_repro: disconnected {}", ctx.sender()); +} + +#[spacetimedb::reducer] +pub fn slow(_ctx: &ReducerContext) { + log::info!("http_cancel_repro: slow reducer started"); + for i in 0..300_000_000u64 { + core::hint::black_box(i); + } + log::info!("http_cancel_repro: slow reducer finished"); +} diff --git a/crates/smoketests/tests/smoketests/client_connection_errors.rs b/crates/smoketests/tests/smoketests/client_connection_errors.rs index c937d750ce4..baeb5172225 100644 --- a/crates/smoketests/tests/smoketests/client_connection_errors.rs +++ b/crates/smoketests/tests/smoketests/client_connection_errors.rs @@ -1,4 +1,8 @@ -use spacetimedb_smoketests::Smoketest; +use spacetimedb_smoketests::{require_local_server, Smoketest}; +use std::io::Write; +use std::net::TcpStream; +use std::thread; +use std::time::{Duration, Instant}; /// Test that client_connected returning an error rejects the connection #[test] @@ -57,3 +61,116 @@ fn test_client_disconnected_error_still_deletes_st_client() { "Expected at most 1 st_client row (the SQL connection itself), got {row_count}: {sql_out}", ); } + +#[test] +fn test_http_reducer_call_cancel_still_deletes_st_client() { + require_local_server!(); + + let test = Smoketest::builder() + .precompiled_module("client-connection-http-cancel") + .build(); + + let streams = send_http_reducer_calls_and_hold_connections(&test, "slow", 64); + wait_for_log( + &test, + "http_cancel_repro: slow reducer started", + Duration::from_secs(30), + ); + drop(streams); + + wait_for_log( + &test, + "http_cancel_repro: slow reducer finished", + Duration::from_secs(30), + ); + + let (st_client_count, st_connection_credentials_count) = wait_for_client_rows_to_clear(&test); + assert!( + st_client_count <= 1, + "Expected at most 1 st_client row (the SQL connection itself) after dropping 64 HTTP calls post-connect, got {st_client_count}", + ); + assert!( + st_connection_credentials_count <= 1, + "Expected at most 1 st_connection_credentials row (the SQL connection itself) after dropping 64 HTTP calls post-connect, got {st_connection_credentials_count}", + ); +} + +fn send_http_reducer_calls_and_hold_connections(test: &Smoketest, reducer: &str, count: usize) -> Vec { + let host = test.server_host(); + let request = http_reducer_call_request(test, reducer); + + (0..count) + .map(|_| { + let mut stream = TcpStream::connect(host).expect("Failed to connect to local server"); + stream.set_nodelay(true).expect("Failed to set TCP_NODELAY"); + stream + .write_all(request.as_bytes()) + .expect("Failed to write HTTP request"); + stream.flush().expect("Failed to flush HTTP request"); + stream + }) + .collect() +} + +fn http_reducer_call_request(test: &Smoketest, reducer: &str) -> String { + let identity = test.database_identity.as_ref().expect("No database published"); + let host = test.server_host(); + let token = test.read_token().expect("Failed to read auth token"); + let path = format!("/v1/database/{identity}/call/{reducer}"); + + format!( + "POST {path} HTTP/1.1\r\n\ + Host: {host}\r\n\ + Authorization: Bearer {token}\r\n\ + Content-Type: application/json\r\n\ + Content-Length: 2\r\n\ + Connection: close\r\n\ + \r\n\ + []" + ) +} + +fn wait_for_log(test: &Smoketest, needle: &str, timeout: Duration) -> Vec { + let start = Instant::now(); + loop { + let logs = test.logs(200).unwrap(); + if logs.iter().any(|l| l.contains(needle)) { + return logs; + } + assert!( + start.elapsed() < timeout, + "Timed out waiting for log containing {needle:?}: {:?}", + logs + ); + thread::sleep(Duration::from_millis(100)); + } +} + +fn wait_for_client_rows_to_clear(test: &Smoketest) -> (usize, usize) { + let start = Instant::now(); + let mut last = (usize::MAX, usize::MAX); + + while start.elapsed() < Duration::from_secs(20) { + last = ( + sql_count(test, "st_client"), + sql_count(test, "st_connection_credentials"), + ); + if last.0 <= 1 && last.1 <= 1 { + return last; + } + thread::sleep(Duration::from_millis(250)); + } + + last +} + +fn sql_count(test: &Smoketest, table: &str) -> usize { + let output = test + .sql(&format!("SELECT COUNT(*) AS count FROM {table}")) + .unwrap_or_else(|err| panic!("Failed to count rows in {table}: {err:#}")); + + output + .lines() + .find_map(|line| line.trim().parse::().ok()) + .unwrap_or_else(|| panic!("Failed to parse COUNT(*) output for {table}: {output}")) +} From bd27fd81c90f6017f29e89825a9a7b9df779f943 Mon Sep 17 00:00:00 2001 From: Shubham Mishra Date: Wed, 8 Jul 2026 17:51:21 +0530 Subject: [PATCH 2/2] make smoketest simpler for single client to hide false positives --- Cargo.lock | 1 + crates/smoketests/Cargo.toml | 1 + .../client-connection-http-cancel/src/lib.rs | 7 +-- .../smoketests/client_connection_errors.rs | 46 +++++++++++-------- 4 files changed, 29 insertions(+), 26 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 40ddf5ea619..dd6f6c6ca43 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8850,6 +8850,7 @@ dependencies = [ "regex", "reqwest", "serde_json", + "socket2 0.5.10", "spacetimedb-core", "spacetimedb-guard", "tempfile", diff --git a/crates/smoketests/Cargo.toml b/crates/smoketests/Cargo.toml index 473be886c55..fa2461b13c5 100644 --- a/crates/smoketests/Cargo.toml +++ b/crates/smoketests/Cargo.toml @@ -21,6 +21,7 @@ spacetimedb-core.workspace = true cargo_metadata.workspace = true assert_cmd = "2" predicates = "3" +socket2.workspace = true tokio.workspace = true tokio-postgres.workspace = true xmltree.workspace = true diff --git a/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs b/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs index be0548aa6be..1830b7f862d 100644 --- a/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs +++ b/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs @@ -1,15 +1,10 @@ -use spacetimedb::{log, ReducerContext, Table}; +use spacetimedb::{log, ReducerContext}; #[spacetimedb::table(accessor = marker, public)] pub struct Marker { id: u32, } -#[spacetimedb::reducer(init)] -pub fn init(ctx: &ReducerContext) { - ctx.db.marker().insert(Marker { id: 0 }); -} - #[spacetimedb::reducer(client_connected)] pub fn connected(ctx: &ReducerContext) { log::info!("http_cancel_repro: connected {}", ctx.sender()); diff --git a/crates/smoketests/tests/smoketests/client_connection_errors.rs b/crates/smoketests/tests/smoketests/client_connection_errors.rs index baeb5172225..ec1fb641cb6 100644 --- a/crates/smoketests/tests/smoketests/client_connection_errors.rs +++ b/crates/smoketests/tests/smoketests/client_connection_errors.rs @@ -1,3 +1,4 @@ +use socket2::SockRef; use spacetimedb_smoketests::{require_local_server, Smoketest}; use std::io::Write; use std::net::TcpStream; @@ -70,13 +71,13 @@ fn test_http_reducer_call_cancel_still_deletes_st_client() { .precompiled_module("client-connection-http-cancel") .build(); - let streams = send_http_reducer_calls_and_hold_connections(&test, "slow", 64); + let stream = send_http_reducer_call_and_hold_connection(&test, "slow"); wait_for_log( &test, "http_cancel_repro: slow reducer started", Duration::from_secs(30), ); - drop(streams); + abort_tcp_stream(stream); wait_for_log( &test, @@ -87,29 +88,32 @@ fn test_http_reducer_call_cancel_still_deletes_st_client() { let (st_client_count, st_connection_credentials_count) = wait_for_client_rows_to_clear(&test); assert!( st_client_count <= 1, - "Expected at most 1 st_client row (the SQL connection itself) after dropping 64 HTTP calls post-connect, got {st_client_count}", + "Expected at most 1 st_client row (the SQL connection itself) after dropping an HTTP call post-connect, got {st_client_count}", ); assert!( st_connection_credentials_count <= 1, - "Expected at most 1 st_connection_credentials row (the SQL connection itself) after dropping 64 HTTP calls post-connect, got {st_connection_credentials_count}", + "Expected at most 1 st_connection_credentials row (the SQL connection itself) after dropping an HTTP call post-connect, got {st_connection_credentials_count}", ); } -fn send_http_reducer_calls_and_hold_connections(test: &Smoketest, reducer: &str, count: usize) -> Vec { +fn send_http_reducer_call_and_hold_connection(test: &Smoketest, reducer: &str) -> TcpStream { let host = test.server_host(); let request = http_reducer_call_request(test, reducer); - (0..count) - .map(|_| { - let mut stream = TcpStream::connect(host).expect("Failed to connect to local server"); - stream.set_nodelay(true).expect("Failed to set TCP_NODELAY"); - stream - .write_all(request.as_bytes()) - .expect("Failed to write HTTP request"); - stream.flush().expect("Failed to flush HTTP request"); - stream - }) - .collect() + let mut stream = TcpStream::connect(host).expect("Failed to connect to local server"); + stream.set_nodelay(true).expect("Failed to set TCP_NODELAY"); + stream + .write_all(request.as_bytes()) + .expect("Failed to write HTTP request"); + stream.flush().expect("Failed to flush HTTP request"); + stream +} + +fn abort_tcp_stream(stream: TcpStream) { + SockRef::from(&stream) + .set_linger(Some(Duration::ZERO)) + .expect("Failed to set SO_LINGER=0"); + drop(stream); } fn http_reducer_call_request(test: &Smoketest, reducer: &str) -> String { @@ -118,13 +122,15 @@ fn http_reducer_call_request(test: &Smoketest, reducer: &str) -> String { let token = test.read_token().expect("Failed to read auth token"); let path = format!("/v1/database/{identity}/call/{reducer}"); + // Keep the HTTP/1.1 connection alive so dropping the TCP stream is an + // unexpected client disconnect while the reducer call is still active. format!( "POST {path} HTTP/1.1\r\n\ Host: {host}\r\n\ Authorization: Bearer {token}\r\n\ Content-Type: application/json\r\n\ Content-Length: 2\r\n\ - Connection: close\r\n\ + Connection: keep-alive\r\n\ \r\n\ []" ) @@ -142,7 +148,7 @@ fn wait_for_log(test: &Smoketest, needle: &str, timeout: Duration) -> Vec (usize, usize) { let start = Instant::now(); let mut last = (usize::MAX, usize::MAX); - while start.elapsed() < Duration::from_secs(20) { + while start.elapsed() < Duration::from_secs(5) { last = ( sql_count(test, "st_client"), sql_count(test, "st_connection_credentials"), @@ -158,7 +164,7 @@ fn wait_for_client_rows_to_clear(test: &Smoketest) -> (usize, usize) { if last.0 <= 1 && last.1 <= 1 { return last; } - thread::sleep(Duration::from_millis(250)); + thread::sleep(Duration::from_millis(50)); } last