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/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..1830b7f862d --- /dev/null +++ b/crates/smoketests/modules/client-connection-http-cancel/src/lib.rs @@ -0,0 +1,25 @@ +use spacetimedb::{log, ReducerContext}; + +#[spacetimedb::table(accessor = marker, public)] +pub struct Marker { + id: u32, +} + +#[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..ec1fb641cb6 100644 --- a/crates/smoketests/tests/smoketests/client_connection_errors.rs +++ b/crates/smoketests/tests/smoketests/client_connection_errors.rs @@ -1,4 +1,9 @@ -use spacetimedb_smoketests::Smoketest; +use socket2::SockRef; +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 +62,121 @@ 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 stream = send_http_reducer_call_and_hold_connection(&test, "slow"); + wait_for_log( + &test, + "http_cancel_repro: slow reducer started", + Duration::from_secs(30), + ); + abort_tcp_stream(stream); + + 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 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 an HTTP call post-connect, got {st_connection_credentials_count}", + ); +} + +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); + + 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 { + 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}"); + + // 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: keep-alive\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(1)); + } +} + +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(5) { + 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(50)); + } + + 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}")) +}