diff --git a/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md b/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md new file mode 100644 index 000000000..7d4c22a89 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md @@ -0,0 +1,71 @@ +# October 23, MySQL statement closes and late timers + +- **Found by.** Server #714. Twice on October 23, the hosted MariaDB 10.11 + `codec-deadline-health` cell recorded a native timer more than the 1 s + maximum late. In one, `TimerFired` came 1.098 s after `fire_at`. MySQL 8.0 + and 8.4 cells fail the same way, less often. +- **Reproduction.** A disposable 4-vCPU VM ran the unchanged cell on main + `f67f2816e`, with temporary timing in a local-only build: + - MariaDB failed 2 of the first 8 cells, with timers 1.228 s and 1.213 s + late; + - PostgreSQL 17 and SQLite passed all 8 cells, at most 171 ms late. +- **Attribution.** Every late timer waited behind the global transition lock: + - In the worst MariaDB cell, the tick's activity-deadline sweep waited + 543 ms behind a workflow completion that held the lock 690 ms. The sweep + then held it 407 ms itself, and the timer fired 1.228 s late. + - Ordinary transitions held the lock 300 ms to 1 s. With every statement + logged by MariaDB, those transactions spent a median of 5 ms in the + database (p90 9 ms) across their 30 to 48 statements. + - Native was not CPU-starved. Wake-up probes on an async task and an OS + thread rarely saw 50 ms delays, and giving native 2 CPUs left the holds + unchanged. + - Timestamps on each statement inside the transactions showed gaps of + 40 to 100 ms spread across ordinary statements, many near 58 ms. That is + Nagle's algorithm waiting for a delayed acknowledgement. +- **Cause.** `COM_STMT_CLOSE` has no reply. The published `sqlx-mysql` 0.9.0 + flushes it in its own write: when its 100-entry statement cache evicts a + statement, and after executing a statement it does not cache. Before the + next command it also flushes anything still buffered. The next command is + then a second small write while the close is still unacknowledged, so the + kernel holds it about 40 ms. SQLx sets no `TCP_NODELAY`; PHP's mysqlnd does. + PostgreSQL's driver sends its closes together with the next request. +- **Change.** In the vendored driver, `MySqlStream::queue_packet` buffers each + statement close with sequence number 0. On the path before the next command, + `wait_until_ready` no longer flushes a buffer that holds only replyless + closes, so they leave in the same write as that command. `Connection::flush` + and `clear_cached_statements` still send queued closes before returning; + Security's review of the first revision found that they did not. Packets, + order and server behavior are unchanged. The patch record is + `rust/vendor/sqlx-mysql/DURABLE-PATCH.md`. +- **Tests.** A proxy records the TCP read chunks it receives from the client. + TCP is a byte stream, so a chunk boundary shows bytes that arrived + separately. + - `qualification_statement_close_leaves_with_the_next_command` uses a + one-entry statement cache and one uncached statement, so both close paths + are exercised. With the fix it passes on MySQL 8.0.46 and MariaDB 10.11 in + 0.01 s. With the published driver it fails on both: every close arrives + as a chunk of its own, and the same 14 statements take 0.6 s. + - `qualification_explicit_flush_sends_queued_statement_closes`: a queued + close stays buffered until `flush`, which sends it with no later command + and leaves `should_flush` false. + - `qualification_clearing_the_statement_cache_closes_them_now`: + `clear_cached_statements` sends all three closes before it returns. On + MySQL 8.0 with the performance schema on, the server's statements for the + connection drop from three to none. + - Both lifecycle tests fail with the first revision (`f31c4c361`) on MySQL + 8.0 and MariaDB, and pass with this one. The existing cancelled-boundary + qualification still passes. +- **Hosted-cell A/B on the VM.** Interleaved cells with the spec and limits + unchanged. Both builds were instrumented: one without the driver change, one + with its first revision. That revision already buffered the closes, but + `wait_until_ready` still flushed them before draining. + + | | Cells passed | Timer lateness | Holds of 300 ms or more | + | --- | --- | --- | --- | + | MariaDB without the fix | 3 of 6 | 190 to 1045 ms | 54 | + | MariaDB with the fix | 6 of 6 | 49 to 289 ms | 0 | + | MySQL 8.0 and 8.4 without the fix | 5 of 8 | 138 to 1099 ms | 77 | + | MySQL 8.0 and 8.4 with the fix | 8 of 8 | 116 to 299 ms | 3 | + + Clean release builds of main and of this branch are compared across all five + families in the PR's evidence comment. diff --git a/rust/src/runtime/mysql_pool_tests.rs b/rust/src/runtime/mysql_pool_tests.rs index a4780a930..20b524bf7 100644 --- a/rust/src/runtime/mysql_pool_tests.rs +++ b/rust/src/runtime/mysql_pool_tests.rs @@ -221,3 +221,246 @@ async fn qualification_cancelled_pool_boundaries_preserve_fresh_reads_and_commit .await .expect("bounded real transaction-boundary qualification"); } + +const COM_STMT_CLOSE: u8 = 0x19; + +/// Forward clients to `upstream`, recording each TCP read chunk received from +/// a client. TCP is a byte stream: a chunk boundary shows bytes that arrived +/// separately, not a write call. +async fn record_client_chunks( + listener: TcpListener, + upstream: String, + writes: Arc>>>, +) { + loop { + let (client, _) = listener.accept().await.unwrap(); + let server = TcpStream::connect(&upstream).await.unwrap(); + let writes = writes.clone(); + tokio::spawn(async move { + let (mut client_read, mut client_write) = client.into_split(); + let (mut server_read, mut server_write) = server.into_split(); + let upstream = async { + let mut bytes = vec![0; 65536]; + loop { + let count = client_read.read(&mut bytes).await?; + if count == 0 { + break; + } + writes.lock().unwrap().push(bytes[..count].to_vec()); + server_write.write_all(&bytes[..count]).await?; + } + server_write.shutdown().await + }; + let downstream = async { + tokio::io::copy(&mut server_read, &mut client_write).await?; + client_write.shutdown().await + }; + let _: (std::io::Result<()>, std::io::Result<()>) = tokio::join!(upstream, downstream); + }); + } +} + +/// Each client command packet (sequence 0): its command byte, and whether a +/// received chunk ended with it. +fn recorded_commands(writes: &[Vec]) -> Vec<(u8, bool)> { + let mut stream = Vec::new(); + let mut write_ends = std::collections::BTreeSet::new(); + for write in writes { + stream.extend_from_slice(write); + write_ends.insert(stream.len()); + } + let mut commands = Vec::new(); + let mut at = 0; + while at + 4 <= stream.len() { + let length = usize::from(stream[at]) + | usize::from(stream[at + 1]) << 8 + | usize::from(stream[at + 2]) << 16; + let end = at + 4 + length; + if end > stream.len() { + break; + } + if stream[at + 3] == 0 && length > 0 { + commands.push((stream[at + 4], write_ends.contains(&end))); + } + at = end; + } + commands +} + +/// A statement close has no reply. Sent on its own, it leaves +/// unacknowledged data in flight, and Nagle's algorithm then holds the next +/// command until the server's delayed acknowledgement, about 40 ms per close. +/// The proxy then receives the close as a chunk of its own. Under the +/// transition lock those stalls added up to a late timer (#714). +#[tokio::test] +#[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"] +async fn qualification_statement_close_leaves_with_the_next_command() { + tokio::time::timeout(Duration::from_secs(15), async { + let url = std::env::var("DW_TEST_MYSQL_URL").unwrap(); + let direct = MySqlConnectOptions::from_str(&url).unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let writes = Arc::new(std::sync::Mutex::new(Vec::new())); + let proxy = tokio::spawn(record_client_chunks( + listener, + format!("{}:{}", direct.get_host(), direct.get_port()), + writes.clone(), + )); + // Plain text so the proxy can read packets; one cached statement, so + // each new statement evicts and closes the previous one. + let options = direct + .host("127.0.0.1") + .port(port) + .ssl_mode(sqlx::mysql::MySqlSslMode::Disabled) + .statement_cache_capacity(1); + let mut connection = MySqlConnection::connect_with(&options).await.unwrap(); + for round in 0..4_i64 { + for (sql, step) in [("SELECT ? + 1", 1), ("SELECT ? + 2", 2), ("SELECT ? + 3", 3)] { + let value: i64 = sqlx::query_scalar(sql).bind(round).fetch_one(&mut connection).await.unwrap(); + assert_eq!(value, round + step); + } + } + // An uncached statement is closed right after it executes. + let value: i64 = sqlx::query_scalar("SELECT ? + 4") + .bind(1_i64) + .persistent(false) + .fetch_one(&mut connection) + .await + .unwrap(); + assert_eq!(value, 5); + let value: i64 = sqlx::query_scalar("SELECT ? + 5").bind(1_i64).fetch_one(&mut connection).await.unwrap(); + assert_eq!(value, 6); + connection.close().await.unwrap(); + proxy.abort(); + let commands = recorded_commands(&writes.lock().unwrap()); + let closes = commands.iter().filter(|(command, _)| *command == COM_STMT_CLOSE).count(); + assert!(closes >= 12, "eviction and uncached closes were exercised: {closes}"); + assert!( + commands.iter().all(|(command, ended)| *command != COM_STMT_CLOSE || !ended), + "a statement close arrived as a chunk of its own: {commands:?}" + ); + }) + .await + .expect("bounded statement-close chunk qualification"); +} + +/// A plaintext connection through a recording proxy. +async fn proxied_connection( + capacity: usize, +) -> (MySqlConnection, Arc>>>, tokio::task::JoinHandle<()>, MySqlConnectOptions) { + let url = std::env::var("DW_TEST_MYSQL_URL").unwrap(); + let direct = MySqlConnectOptions::from_str(&url).unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let writes = Arc::new(std::sync::Mutex::new(Vec::new())); + let proxy = tokio::spawn(record_client_chunks( + listener, + format!("{}:{}", direct.get_host(), direct.get_port()), + writes.clone(), + )); + let options = direct + .clone() + .host("127.0.0.1") + .port(port) + .ssl_mode(sqlx::mysql::MySqlSslMode::Disabled) + .statement_cache_capacity(capacity); + (MySqlConnection::connect_with(&options).await.unwrap(), writes, proxy, direct) +} + +fn closes(writes: &std::sync::Mutex>>) -> usize { + recorded_commands(&writes.lock().unwrap()) + .iter() + .filter(|(command, _)| *command == COM_STMT_CLOSE) + .count() +} + +/// Wait, bounded, until the proxy has received `count` statement closes. +async fn closes_arrive(writes: &std::sync::Mutex>>, count: usize) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while closes(writes) < count { + assert!( + tokio::time::Instant::now() < deadline, + "{count} statement closes did not leave the client: {:?}", + recorded_commands(&writes.lock().unwrap()) + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +/// `Connection::flush` sends queued closes by itself: no later command is +/// needed, and the connection reports nothing left to flush. +#[tokio::test] +#[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"] +async fn qualification_explicit_flush_sends_queued_statement_closes() { + use sqlx::Connection as _; + tokio::time::timeout(Duration::from_secs(15), async { + let (mut connection, writes, proxy, _) = proxied_connection(1).await; + // An uncached statement queues its close after it executes. + let value: i64 = sqlx::query_scalar("SELECT ? + 4") + .bind(1_i64) + .persistent(false) + .fetch_one(&mut connection) + .await + .unwrap(); + assert_eq!(value, 5); + assert!(connection.should_flush(), "the close is queued"); + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(closes(&writes), 0, "the queued close waits for a command or flush"); + connection.flush().await.unwrap(); + assert!(!connection.should_flush(), "an explicit flush leaves nothing buffered"); + closes_arrive(&writes, 1).await; + connection.close().await.unwrap(); + proxy.abort(); + }) + .await + .expect("bounded explicit-flush qualification"); +} + +/// `clear_cached_statements` closes every cached statement on the server +/// before it returns, without a later command. +#[tokio::test] +#[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"] +async fn qualification_clearing_the_statement_cache_closes_them_now() { + use sqlx::Connection as _; + tokio::time::timeout(Duration::from_secs(15), async { + let (mut connection, writes, proxy, direct) = proxied_connection(100).await; + // A separate session can see this connection's server-side + // statements where the performance schema is on (MySQL 8 default). + let thread: u64 = sqlx::query_scalar("SELECT CONNECTION_ID()").fetch_one(&mut connection).await.unwrap(); + connection.clear_cached_statements().await.unwrap(); + let before = closes(&writes); + for sql in ["SELECT ? + 1", "SELECT ? + 2", "SELECT ? + 3"] { + let _: i64 = sqlx::query_scalar(sql).bind(1_i64).fetch_one(&mut connection).await.unwrap(); + } + assert_eq!(connection.cached_statements_size(), 3); + let mut observer = MySqlConnection::connect_with(&direct).await.unwrap(); + let schema: i64 = sqlx::query_scalar("SELECT @@performance_schema").fetch_one(&mut observer).await.unwrap(); + let server_statements = async |observer: &mut MySqlConnection| -> i64 { + sqlx::query_scalar("SELECT COUNT(*) FROM performance_schema.prepared_statements_instances p JOIN performance_schema.threads t ON t.THREAD_ID=p.OWNER_THREAD_ID WHERE t.PROCESSLIST_ID=?") + .bind(thread) + .fetch_one(observer) + .await + .unwrap() + }; + if schema == 1 { + let held = server_statements(&mut observer).await; + assert_eq!(held, 3, "the server holds this connection's three statements"); + } + connection.clear_cached_statements().await.unwrap(); + assert_eq!(connection.cached_statements_size(), 0); + assert!(!connection.should_flush(), "clearing leaves nothing buffered"); + closes_arrive(&writes, before + 3).await; + if schema == 1 { + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while server_statements(&mut observer).await != 0 { + assert!(tokio::time::Instant::now() < deadline, "the server still holds the statements"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + observer.close().await.unwrap(); + connection.close().await.unwrap(); + proxy.abort(); + }) + .await + .expect("bounded statement-cache clear qualification"); +} diff --git a/rust/vendor/sqlx-mysql/DURABLE-PATCH.md b/rust/vendor/sqlx-mysql/DURABLE-PATCH.md index f08664b46..f4792828c 100644 --- a/rust/vendor/sqlx-mysql/DURABLE-PATCH.md +++ b/rust/vendor/sqlx-mysql/DURABLE-PATCH.md @@ -1,7 +1,7 @@ -# SQLx MySQL hard-close patch +# SQLx MySQL driver patches -This directory contains the official `sqlx-mysql` 0.9.0 crate with one scoped -driver correction. Its published archive has SHA-256 +This directory contains the official `sqlx-mysql` 0.9.0 crate with two scoped +driver corrections. Its published archive has SHA-256 `90b8020fe17c5f2c245bfa2505d7ef59c5604839527c740266ad2214acebea27` and VCS commit `003b698e99e024f3621b8043a2426fde5b741171`. @@ -19,10 +19,49 @@ The original and flush-before-gate TLS controls fail with the published driver; both pass with this correction. Qualification and original evidence are in [Server #429](https://github.com/durable-workflow/server/pull/429#issuecomment-6135933370). -All upstream source is retained. The only source-code change is in -`src/connection/mod.rs`. The packaged license-reference files are replaced with +## Statement closes leave with the next command + +`COM_STMT_CLOSE` has no server reply. The published driver flushes it on its +own write: when its statement cache evicts a statement, and right after +executing a statement it does not cache. The next command is then a second +small write while the close is still unacknowledged, so Nagle's algorithm +holds it until the server's delayed acknowledgement, about 40 ms. SQLx does not +set `TCP_NODELAY`; PHP's mysqlnd and libmysqlclient do. + +Inside the native transition lock, a transaction issues 30 to 48 statements, +so these stalls added up to holds of 0.3 to 1 s under load. A due timer +waiting behind them fired more than the 1 s maximum late (Server #714). + +`MySqlStream::queue_packet` now buffers each close with the reset sequence +number that `send_packet` would use. On the path before the next command +(`wait_until_ready`), a buffer that holds only such closes is not flushed, so +they leave in the same write as that command. The server receives the same +packets in the same order. A connection dropped before its next command leaves +the close unsent; the server releases that connection's prepared statements +when it closes. + +The public lifecycle is unchanged. `Connection::flush` sends every buffered +byte, queued closes included, and leaves `should_flush` false. +`clear_cached_statements` sends its closes before it returns. Neither needs a +later command. + +Three tests in `rust/src/runtime/mysql_pool_tests.rs` record the TCP read +chunks a proxy receives from the client: +- `qualification_statement_close_leaves_with_the_next_command` fails with the + published driver, where each close arrives as a chunk of its own, and passes + with this correction; +- `qualification_explicit_flush_sends_queued_statement_closes` and + `qualification_clearing_the_statement_cache_closes_them_now` prove the two + lifecycle calls without a later command. Where the performance schema is on, + the second also sees the server's statements for the connection drop from + three to none. + +All upstream source is retained. The source-code changes are in +`src/connection/mod.rs`, `src/connection/stream.rs` and +`src/connection/executor.rs`. The packaged license-reference files are replaced with the full MIT and Apache license texts from the same upstream commit. This file records the patch; no upstream release is represented as containing it. Remove this override when a reviewed published driver provides equivalent -hard-close behavior and passes the same database and TLS qualification. +hard-close behavior and statement-close batching (or `TCP_NODELAY`), and passes +the same database and TLS qualification. diff --git a/rust/vendor/sqlx-mysql/src/connection/executor.rs b/rust/vendor/sqlx-mysql/src/connection/executor.rs index ee59d03d0..6c12d29fc 100755 --- a/rust/vendor/sqlx-mysql/src/connection/executor.rs +++ b/rust/vendor/sqlx-mysql/src/connection/executor.rs @@ -92,8 +92,7 @@ impl MySqlConnection { { self.inner .stream - .send_packet(StmtClose { statement: id }) - .await?; + .queue_packet(StmtClose { statement: id })?; } Ok((id, metadata)) @@ -168,7 +167,7 @@ impl MySqlConnection { }) .await?; - self.inner.stream.send_packet(StmtClose { statement: id }).await?; + self.inner.stream.queue_packet(StmtClose { statement: id })?; MySqlValueFormat::Binary } @@ -337,8 +336,7 @@ impl<'c> Executor<'c> for &'c mut MySqlConnection { self.inner .stream - .send_packet(StmtClose { statement: id }) - .await?; + .queue_packet(StmtClose { statement: id })?; metadata }; @@ -367,8 +365,7 @@ impl<'c> Executor<'c> for &'c mut MySqlConnection { self.inner .stream - .send_packet(StmtClose { statement: id }) - .await?; + .queue_packet(StmtClose { statement: id })?; let columns = (*metadata.columns).clone(); diff --git a/rust/vendor/sqlx-mysql/src/connection/mod.rs b/rust/vendor/sqlx-mysql/src/connection/mod.rs index 110ae4f1e..2513cf3cd 100755 --- a/rust/vendor/sqlx-mysql/src/connection/mod.rs +++ b/rust/vendor/sqlx-mysql/src/connection/mod.rs @@ -94,7 +94,11 @@ impl Connection for MySqlConnection { #[doc(hidden)] fn flush(&mut self) -> impl Future> + Send + '_ { - self.inner.stream.wait_until_ready() + async move { + // `Connection::flush` sends everything, queued closes included. + self.inner.stream.flush_queued().await?; + self.inner.stream.wait_until_ready().await + } } fn cached_statements_size(&self) -> usize { @@ -105,13 +109,13 @@ impl Connection for MySqlConnection { while let Some((statement_id, _)) = self.inner.cache_statement.remove_lru() { self.inner .stream - .send_packet(StmtClose { + .queue_packet(StmtClose { statement: statement_id, - }) - .await?; + })?; } - Ok(()) + // The server closes them now, not with some later command. + self.inner.stream.flush_queued().await } #[doc(hidden)] diff --git a/rust/vendor/sqlx-mysql/src/connection/stream.rs b/rust/vendor/sqlx-mysql/src/connection/stream.rs index e6aa8b48c..c16a21eb0 100755 --- a/rust/vendor/sqlx-mysql/src/connection/stream.rs +++ b/rust/vendor/sqlx-mysql/src/connection/stream.rs @@ -19,6 +19,9 @@ pub struct MySqlStream> { pub(crate) sequence_id: u8, pub(crate) waiting: VecDeque, pub(crate) is_tls: bool, + // The write buffer holds only commands without a reply, queued to leave + // with the next command. + pub(crate) queued_unanswered: bool, } #[derive(Debug, PartialEq, Eq)] @@ -56,13 +59,27 @@ impl MySqlStream { sequence_id: 0, socket: BufferedSocket::new(socket), is_tls: false, + queued_unanswered: false, } } - pub(crate) async fn wait_until_ready(&mut self) -> Result<(), Error> { + /// Send every buffered byte now, including queued commands without a + /// reply. Explicit flushes and statement cache clears use this; only the + /// path before the next command leaves queued closes to share its write. + pub(crate) async fn flush_queued(&mut self) -> Result<(), Error> { if !self.socket.write_buffer().is_empty() { self.socket.flush().await?; } + self.queued_unanswered = false; + Ok(()) + } + + pub(crate) async fn wait_until_ready(&mut self) -> Result<(), Error> { + // Queued commands without a reply wait for the next command; nothing + // pending needs them to be sent first. + if !self.socket.write_buffer().is_empty() && !self.queued_unanswered { + self.socket.flush().await?; + } while !self.waiting.is_empty() { while self.waiting.front() == Some(&Waiting::Row) { @@ -108,10 +125,26 @@ impl MySqlStream { Ok(()) } + /// Buffer a command that has no server reply (`COM_STMT_CLOSE`) so it + /// leaves in the same write as the next command. Flushing it alone leaves + /// unacknowledged data in flight, and Nagle's algorithm then holds the next + /// command until the server's delayed acknowledgement (about 40 ms). + pub(crate) fn queue_packet<'en, T>(&mut self, payload: T) -> Result<(), Error> + where + T: ProtocolEncode<'en, Capabilities>, + { + let unanswered = self.socket.write_buffer().is_empty() || self.queued_unanswered; + self.sequence_id = 0; + self.write_packet(payload)?; + self.queued_unanswered = unanswered; + Ok(()) + } + pub(crate) fn write_packet<'en, T>(&mut self, payload: T) -> Result<(), Error> where T: ProtocolEncode<'en, Capabilities>, { + self.queued_unanswered = false; self.socket .write_with(Packet(payload), (self.capabilities, &mut self.sequence_id)) } @@ -213,6 +246,7 @@ impl MySqlStream { sequence_id: self.sequence_id, waiting: self.waiting, is_tls: self.is_tls, + queued_unanswered: self.queued_unanswered, } } } diff --git a/rust/vendor/sqlx-mysql/src/connection/tls.rs b/rust/vendor/sqlx-mysql/src/connection/tls.rs index 9034fbd63..9ac71f676 100755 --- a/rust/vendor/sqlx-mysql/src/connection/tls.rs +++ b/rust/vendor/sqlx-mysql/src/connection/tls.rs @@ -97,6 +97,7 @@ impl WithSocket for MapStream { sequence_id: self.sequence_id, waiting: self.waiting, is_tls: true, + queued_unanswered: false, } } }