Rust #325: send MySQL statement closes with the next command (#714)

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.

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<std::sync::Mutex<Vec<Vec<u8>>>>,
+) {
+    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<u8>]) -> 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<std::sync::Mutex<Vec<Vec<u8>>>>, 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<Vec<Vec<u8>>>) -> 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<Vec<Vec<u8>>>, 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");
+}

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.