From f31c4c36130c77c46ca6cc4f7cab5307f1480714 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 22:40:17 +0000 Subject: [PATCH 1/2] Send MySQL statement closes with the next command COM_STMT_CLOSE has no reply. The published sqlx-mysql flushed it on its own write after a cache eviction or an uncached execute, and flushed it again before draining results. The next command then waited behind Nagle's algorithm for the server's delayed acknowledgement, about 40 ms. Across the 30-48 statements of a transition that held the global lock for 0.3-1 s under load, and a due timer behind it fired more than the 1 s bound late (#714). The vendored driver now queues each close with sequence 0 and leaves a buffer of replyless closes unflushed until the next command, so both go out in one write. A proxy-recorded qualification test fails with the published driver and passes with the change on MySQL 8.0 and MariaDB. --- .../2026-10-23-mysql-statement-close.md | 57 +++++++++ rust/src/runtime/mysql_pool_tests.rs | 119 ++++++++++++++++++ rust/vendor/sqlx-mysql/DURABLE-PATCH.md | 38 +++++- .../sqlx-mysql/src/connection/executor.rs | 11 +- rust/vendor/sqlx-mysql/src/connection/mod.rs | 5 +- .../sqlx-mysql/src/connection/stream.rs | 25 +++- rust/vendor/sqlx-mysql/src/connection/tls.rs | 1 + 7 files changed, 239 insertions(+), 17 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md 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..d5377cc5d --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md @@ -0,0 +1,57 @@ +# 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. `wait_until_ready` no longer flushes + a buffer that holds only replyless closes, so they leave in the same write as + the next command. Packets, order and server behavior are unchanged. The + patch record is `rust/vendor/sqlx-mysql/DURABLE-PATCH.md`. +- **Test.** `qualification_statement_close_leaves_with_the_next_command` + records the client's writes through a proxy. It 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 ends its own + write, and the same 14 statements take 0.6 s. + - 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..e6f40c497 100644 --- a/rust/src/runtime/mysql_pool_tests.rs +++ b/rust/src/runtime/mysql_pool_tests.rs @@ -221,3 +221,122 @@ async fn qualification_cancelled_pool_boundaries_preserve_fresh_reads_and_commit .await .expect("bounded real transaction-boundary qualification"); } + +/// Forward one client to `upstream`, recording each client write as it +/// arrives. +async fn record_client_writes( + 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 +/// recorded write 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. Flushed 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. +/// 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() { + const COM_STMT_CLOSE: u8 = 0x19; + 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_writes( + 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 ended its own write: {commands:?}" + ); + }) + .await + .expect("bounded statement-close write qualification"); +} diff --git a/rust/vendor/sqlx-mysql/DURABLE-PATCH.md b/rust/vendor/sqlx-mysql/DURABLE-PATCH.md index f08664b46..117531a0c 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,36 @@ 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, so it leaves in the same write as the next +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. + +`qualification_statement_close_leaves_with_the_next_command` in +`rust/src/runtime/mysql_pool_tests.rs` records every client write through a +proxy. It fails with the published driver, where each close ends its write, and +passes with this correction. + +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..c20bfc026 100755 --- a/rust/vendor/sqlx-mysql/src/connection/mod.rs +++ b/rust/vendor/sqlx-mysql/src/connection/mod.rs @@ -105,10 +105,9 @@ 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(()) diff --git a/rust/vendor/sqlx-mysql/src/connection/stream.rs b/rust/vendor/sqlx-mysql/src/connection/stream.rs index e6aa8b48c..590ce1ecd 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,11 +59,14 @@ 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> { - if !self.socket.write_buffer().is_empty() { + // 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?; } @@ -108,10 +114,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 +235,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, } } } From dc58f9345c05b3185426baee2d72249887605470 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 23:48:10 +0000 Subject: [PATCH 2/2] Keep explicit MySQL flushes and cache clears sending statement closes Security's review of f31c4c361: Connection::flush delegated to wait_until_ready, which now leaves a buffer of queued closes unsent, and clear_cached_statements only queued its closes. Both report success with the bytes still buffered. Explicit flushes and cache clears now send every buffered byte before returning (MySqlStream::flush_queued); only the path before the next command keeps closes to share that command's write. New MySQL and MariaDB controls prove both calls without a later command; they fail on f31c4c361. The close test now describes what it observes: TCP read chunks at a proxy, not write calls. --- .../2026-10-23-mysql-statement-close.md | 36 +++-- rust/src/runtime/mysql_pool_tests.rs | 144 ++++++++++++++++-- rust/vendor/sqlx-mysql/DURABLE-PATCH.md | 31 ++-- rust/vendor/sqlx-mysql/src/connection/mod.rs | 9 +- .../sqlx-mysql/src/connection/stream.rs | 11 ++ 5 files changed, 199 insertions(+), 32 deletions(-) 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 index d5377cc5d..7d4c22a89 100644 --- 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 @@ -30,17 +30,31 @@ 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. `wait_until_ready` no longer flushes - a buffer that holds only replyless closes, so they leave in the same write as - the next command. Packets, order and server behavior are unchanged. The - patch record is `rust/vendor/sqlx-mysql/DURABLE-PATCH.md`. -- **Test.** `qualification_statement_close_leaves_with_the_next_command` - records the client's writes through a proxy. It 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 ends its own - write, and the same 14 statements take 0.6 s. - - The existing cancelled-boundary qualification still passes. + 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 diff --git a/rust/src/runtime/mysql_pool_tests.rs b/rust/src/runtime/mysql_pool_tests.rs index e6f40c497..20b524bf7 100644 --- a/rust/src/runtime/mysql_pool_tests.rs +++ b/rust/src/runtime/mysql_pool_tests.rs @@ -222,9 +222,12 @@ async fn qualification_cancelled_pool_boundaries_preserve_fresh_reads_and_commit .expect("bounded real transaction-boundary qualification"); } -/// Forward one client to `upstream`, recording each client write as it -/// arrives. -async fn record_client_writes( +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>>>, @@ -258,7 +261,7 @@ async fn record_client_writes( } /// Each client command packet (sequence 0): its command byte, and whether a -/// recorded write ended with it. +/// 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(); @@ -284,21 +287,21 @@ fn recorded_commands(writes: &[Vec]) -> Vec<(u8, bool)> { commands } -/// A statement close has no reply. Flushed on its own, it leaves +/// 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. -/// Under the transition lock those stalls added up to a late timer (#714). +/// 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() { - const COM_STMT_CLOSE: u8 = 0x19; 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_writes( + let proxy = tokio::spawn(record_client_chunks( listener, format!("{}:{}", direct.get_host(), direct.get_port()), writes.clone(), @@ -334,9 +337,130 @@ async fn qualification_statement_close_leaves_with_the_next_command() { assert!(closes >= 12, "eviction and uncached closes were exercised: {closes}"); assert!( commands.iter().all(|(command, ended)| *command != COM_STMT_CLOSE || !ended), - "a statement close ended its own write: {commands:?}" + "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-close write qualification"); + .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 117531a0c..f4792828c 100644 --- a/rust/vendor/sqlx-mysql/DURABLE-PATCH.md +++ b/rust/vendor/sqlx-mysql/DURABLE-PATCH.md @@ -33,15 +33,28 @@ 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, so it leaves in the same write as the next -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. - -`qualification_statement_close_leaves_with_the_next_command` in -`rust/src/runtime/mysql_pool_tests.rs` records every client write through a -proxy. It fails with the published driver, where each close ends its write, and -passes with this correction. +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 diff --git a/rust/vendor/sqlx-mysql/src/connection/mod.rs b/rust/vendor/sqlx-mysql/src/connection/mod.rs index c20bfc026..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 { @@ -110,7 +114,8 @@ impl Connection for MySqlConnection { })?; } - 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 590ce1ecd..c16a21eb0 100755 --- a/rust/vendor/sqlx-mysql/src/connection/stream.rs +++ b/rust/vendor/sqlx-mysql/src/connection/stream.rs @@ -63,6 +63,17 @@ impl MySqlStream { } } + /// 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.