fix(turso): unbound params executed as NULL — mysql CLI 'commands out of sync'

crates/litewire-turso/src/lib.rs

@@ -61,6 +61,16 @@
 //! Anything else the engine cannot do surfaces as the engine's own error
 //! text mapped into [`BackendError::Sqlite`], which the wire frontends'
 //! error classifiers already understand (SQLite-style message shapes).
+//!
+//! # Bind-count parity with the rusqlite backend
+//!
+//! The engine executes statements with unbound parameters as `NULL`
+//! instead of erroring. This backend rejects a parameter-count mismatch
+//! with the same "Wrong number of parameters passed to query" error the
+//! rusqlite backend produces. Without it, the `mysql` >= 8.1 CLI's
+//! `select $$` startup probe (which must fail) returns a result set the
+//! CLI never consumes, and every later statement dies client-side with
+//! CR 2014 "Commands out of sync".
 
 pub mod cdc;
 
@@ -347,6 +357,147 @@ fn sql_uses_pragma_tvf(sql: &str) -> bool {
     sql.to_ascii_lowercase().contains("pragma_")
 }
 
+/// Number of parameter slots a SQL statement declares, following SQLite's
+/// tokenizer rules: `?` takes the next free index, `?NNN` takes index `NNN`,
+/// and `:name` / `@name` / `$name` each take the next free index the first
+/// time the name appears. `$name` additionally swallows SQLite's TCL-heritage
+/// suffixes (`$name::seg` repetitions and one trailing `(...)`) as part of
+/// the variable name, matching tokenize.c. The result is the highest index
+/// assigned — the same value `sqlite3_bind_parameter_count()` reports.
+///
+/// Quoted strings (`'…'`), quoted identifiers (`"…"`, `` `…` ``, `[…]`),
+/// line comments (`--`) and block comments (`/* … */`) are skipped.
+///
+/// TODO(turso >0.7): delete this scanner and delegate to the statement's
+/// parameter count once the turso crate exposes it publicly.
+fn expected_param_count(sql: &str) -> usize {
+    let bytes = sql.as_bytes();
+    let mut i = 0;
+    let mut max_index = 0usize;
+    // Distinct named parameters seen so far (each gets one index).
+    let mut named: Vec<&str> = Vec::new();
+    while i < bytes.len() {
+        match bytes[i] {
+            // String literal or quoted identifier; doubled quote escapes.
+            q @ (b'\'' | b'"' | b'`') => {
+                i += 1;
+                while i < bytes.len() {
+                    if bytes[i] == q {
+                        if i + 1 < bytes.len() && bytes[i + 1] == q {
+                            i += 2;
+                            continue;
+                        }
+                        break;
+                    }
+                    i += 1;
+                }
+                i += 1;
+            }
+            // Bracket-quoted identifier (accepted by SQLite for MS compat).
+            b'[' => {
+                while i < bytes.len() && bytes[i] != b']' {
+                    i += 1;
+                }
+                i += 1;
+            }
+            // -- line comment
+            b'-' if bytes.get(i + 1) == Some(&b'-') => {
+                while i < bytes.len() && bytes[i] != b'\n' {
+                    i += 1;
+                }
+            }
+            // /* block comment */
+            b'/' if bytes.get(i + 1) == Some(&b'*') => {
+                i += 2;
+                while i + 1 < bytes.len() && !(bytes[i] == b'*' && bytes[i + 1] == b'/') {
+                    i += 1;
+                }
+                i = (i + 2).min(bytes.len());
+            }
+            // `?` or `?NNN`
+            b'?' => {
+                i += 1;
+                let start = i;
+                while i < bytes.len() && bytes[i].is_ascii_digit() {
+                    i += 1;
+                }
+                if i > start {
+                    let n: usize = sql[start..i].parse().unwrap_or(0);
+                    max_index = max_index.max(n);
+                } else {
+                    max_index += 1;
+                }
+            }
+            // `:name`, `@name`, `$name`. SQLite also accepts a bare `$`
+            // variable — `SELECT $$` parses as a single parameter (the
+            // mysql 8.4 CLI exploits exactly that in its startup probe).
+            b':' | b'@' | b'$' => {
+                let start = i;
+                i += 1;
+                while i < bytes.len()
+                    && (bytes[i].is_ascii_alphanumeric() || bytes[i] == b'_' || bytes[i] == b'$')
+                {
+                    i += 1;
+                }
+                if bytes[start] == b'$' {
+                    // TCL-heritage suffixes are part of the variable name
+                    // (sqlite tokenize.c): `$name::seg` repetitions, then
+                    // optionally one non-nested `(...)`.
+                    while i + 1 < bytes.len() && bytes[i] == b':' && bytes[i + 1] == b':' {
+                        i += 2;
+                        while i < bytes.len()
+                            && (bytes[i].is_ascii_alphanumeric()
+                                || bytes[i] == b'_'
+                                || bytes[i] == b'$')
+                        {
+                            i += 1;
+                        }
+                    }
+                    if i < bytes.len() && bytes[i] == b'(' {
+                        while i < bytes.len() && bytes[i] != b')' {
+                            i += 1;
+                        }
+                        i = (i + 1).min(bytes.len());
+                    }
+                }
+                if i > start + 1 || bytes[start] == b'$' {
+                    let name = &sql[start..i];
+                    // O(n) scan is fine: real statements carry < 10 distinct
+                    // named parameters.
+                    if !named.contains(&name) {
+                        named.push(name);
+                        max_index += 1;
+                    }
+                }
+            }
+            _ => i += 1,
+        }
+    }
+    max_index
+}
+
+/// Reject a bind-count mismatch with the same error shape the rusqlite
+/// backend produces (`rusqlite::Error::InvalidParameterCount`).
+///
+/// The Turso engine (0.7.0) silently leaves unbound parameters NULL and
+/// executes anyway. SQLite's C API technically does the same, but every
+/// SQLite *binding* litewire sits on (rusqlite here, and what wire clients
+/// expect from a MySQL-shaped server) treats a count mismatch as an error.
+/// The failure mode of not checking is severe: Oracle's `mysql` >= 8.1 CLI
+/// probes dollar-quoting support at startup with `select $$` and expects an
+/// error reply. If the server instead returns a result set, the CLI never
+/// reads it, its client-side state machine wedges, and every subsequent
+/// statement fails with CR 2014 "Commands out of sync".
+fn check_param_count(sql: &str, got: usize) -> Result<(), BackendError> {
+    let needed = expected_param_count(sql);
+    if got != needed {
+        return Err(BackendError::Sqlite(format!(
+            "Wrong number of parameters passed to query. Got {got}, needed {needed}"
+        )));
+    }
+    Ok(())
+}
+
 #[async_trait::async_trait]
 impl BackendConn for TursoConn {
     async fn query(&self, sql: &str, params: &[Value]) -> Result<ResultSet, BackendError> {
@@ -355,6 +506,10 @@ impl BackendConn for TursoConn {
         // `prepare_cached` interns the parsed statement in the engine's
         // per-connection cache, mirroring the rusqlite backend.
         let mut stmt = conn.prepare_cached(sql).await.map_err(map_turso_err)?;
+        // Engine parse errors surface first (prepare above); then enforce
+        // the bind count like rusqlite does — the engine itself would
+        // silently run with unbound parameters as NULL.
+        check_param_count(sql, params.len())?;
 
         let columns = to_columns(&stmt.columns());
         let col_count = columns.len();
@@ -396,6 +551,8 @@ impl BackendConn for TursoConn {
         reject_unsupported(sql)?;
         let conn = self.conn.lock().await;
         let mut stmt = conn.prepare_cached(sql).await.map_err(map_turso_err)?;
+        // Same bind-count parity check as `query` (see `check_param_count`).
+        check_param_count(sql, params.len())?;
         let affected = stmt
             .execute(to_params(params))
             .await
@@ -713,6 +870,81 @@ mod tests {
         assert_eq!(cols[2].decltype.as_deref(), Some("BLOB"));
     }
 
+    // -- Bind-count parity (mysql CLI `select $$` probe regression) -------
+
+    #[test]
+    fn param_count_scanner() {
+        assert_eq!(expected_param_count("SELECT 1"), 0);
+        assert_eq!(expected_param_count("SELECT ?"), 1);
+        assert_eq!(expected_param_count("SELECT ?, ?"), 2);
+        assert_eq!(expected_param_count("SELECT ?2"), 2);
+        assert_eq!(expected_param_count("SELECT ?3, ?"), 4);
+        assert_eq!(expected_param_count("SELECT :a, :b, :a"), 2);
+        assert_eq!(expected_param_count("SELECT @x, $y"), 2);
+        // The mysql >= 8.1 CLI probe: one TCL-style `$` parameter.
+        assert_eq!(expected_param_count("SELECT $$"), 1);
+        // Placeholders inside literals/identifiers/comments don't count.
+        assert_eq!(expected_param_count("SELECT '?', \"?\", `?` -- ?"), 0);
+        assert_eq!(expected_param_count("SELECT 'it''s ?' /* :x */, [?]"), 0);
+        assert_eq!(expected_param_count("SELECT 'a@b.c', '$5'"), 0);
+        // TCL-heritage `$` suffixes are part of the variable name, matching
+        // sqlite3_bind_parameter_count() (review differential-test finding).
+        assert_eq!(expected_param_count("SELECT $foo::type"), 1);
+        assert_eq!(expected_param_count("SELECT $foo::type(1,2)"), 1);
+        assert_eq!(expected_param_count("SELECT $foo(1)"), 1);
+        assert_eq!(expected_param_count("SELECT $a::b::c, $a::b::c"), 1);
+    }
+
+    #[tokio::test]
+    async fn unbound_parameter_rejected_like_rusqlite() {
+        // Regression: Oracle's mysql 8.4 CLI sends `select $$` at startup
+        // (dollar-quoting detection) and requires an error reply. Turso
+        // executes unbound parameters as NULL, so this returned a result
+        // set — which the CLI never reads, wedging its state machine into
+        // CR 2014 "Commands out of sync" for every following statement.
+        let backend = Turso::memory().await.unwrap();
+
+        let err = backend.query("SELECT $$", &[]).await.unwrap_err();
+        assert!(
+            err.to_string()
+                .contains("Wrong number of parameters passed to query. Got 0, needed 1"),
+            "expected rusqlite-parity bind-count error, got: {err}"
+        );
+
+        let err = backend.query("SELECT ?", &[]).await.unwrap_err();
+        assert!(err.to_string().contains("Wrong number of parameters"));
+    }
+
+    #[tokio::test]
+    async fn execute_bind_count_mismatch_rejected() {
+        let backend = Turso::memory().await.unwrap();
+        backend
+            .execute("CREATE TABLE t (v TEXT)", &[])
+            .await
+            .unwrap();
+
+        // Too few.
+        let err = backend
+            .execute("INSERT INTO t VALUES (?1)", &[])
+            .await
+            .unwrap_err();
+        assert!(err.to_string().contains("Got 0, needed 1"), "got: {err}");
+
+        // Too many.
+        let err = backend
+            .query("SELECT ?1", &[Value::Integer(1), Value::Integer(2)])
+            .await
+            .unwrap_err();
+        assert!(err.to_string().contains("Got 2, needed 1"), "got: {err}");
+
+        // Exact count still works.
+        let rs = backend
+            .query("SELECT ?1", &[Value::Integer(7)])
+            .await
+            .unwrap();
+        assert_eq!(rs.rows[0][0], Value::Integer(7));
+    }
+
     #[tokio::test]
     async fn vacuum_rejected_with_clear_error() {
         let backend = Turso::memory().await.unwrap();

crates/litewire/tests/mysql_turso_e2e.rs

@@ -0,0 +1,102 @@
+//! End-to-end regression test: MySQL frontend backed by the experimental
+//! Turso engine backend (`--features turso`).
+//!
+//! Guards the wire property that broke Oracle's `mysql` >= 8.1 CLI: the
+//! CLI probes dollar-quoting support at startup with `select $$` and
+//! requires an ERR reply. When the backend executed the statement with the
+//! unbound `$$` parameter as NULL, the server answered with a result set
+//! the CLI never reads — wedging the client state machine so every later
+//! statement failed with CR 2014 "Commands out of sync".
+
+#![cfg(feature = "turso")]
+
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use mysql_async::prelude::*;
+use mysql_async::{Conn, Opts, OptsBuilder};
+use tokio::net::TcpListener;
+
+async fn free_port() -> u16 {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    listener.local_addr().unwrap().port()
+}
+
+/// Start litewire's MySQL frontend on `port`, backed by an in-memory Turso
+/// engine database.
+async fn start_litewire_turso(port: u16) -> tokio::task::JoinHandle<()> {
+    let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
+    let backend = litewire::litewire_turso::Turso::memory().await.unwrap();
+    let config = litewire::litewire_mysql::MysqlFrontendConfig { listen: addr };
+    let frontend = litewire::litewire_mysql::MysqlFrontend::new(config, Arc::new(backend));
+
+    tokio::spawn(async move {
+        frontend.serve().await.unwrap();
+    })
+}
+
+async fn connect(port: u16) -> Conn {
+    let opts: Opts = OptsBuilder::default()
+        .ip_or_hostname("127.0.0.1")
+        .tcp_port(port)
+        .user(Some("root"))
+        .pass(Some(""))
+        .db_name(Some("test"))
+        .into();
+
+    for i in 0..20 {
+        match Conn::new(opts.clone()).await {
+            Ok(conn) => return conn,
+            Err(_) if i < 19 => tokio::time::sleep(std::time::Duration::from_millis(50)).await,
+            Err(e) => panic!("failed to connect after retries: {e}"),
+        }
+    }
+    unreachable!()
+}
+
+/// `select $$` (the mysql >= 8.1 CLI dollar-quoting probe) must produce a
+/// server ERR packet — never a result set — and the connection must remain
+/// usable afterwards.
+#[tokio::test]
+async fn dollar_probe_errors_and_connection_stays_in_sync() {
+    let port = free_port().await;
+    let _server = start_litewire_turso(port).await;
+    let mut conn = connect(port).await;
+
+    let probe = conn.query_iter("select $$").await;
+    match probe {
+        Err(mysql_async::Error::Server(e)) => {
+            assert!(
+                e.message.contains("Wrong number of parameters"),
+                "expected bind-count error, got: {e}"
+            );
+        }
+        Err(other) => panic!("expected a server ERR packet, got transport error: {other}"),
+        Ok(_) => panic!("`select $$` must not return a result set"),
+    }
+
+    // The stream must still be in sync: same connection, next statement.
+    let rows: Vec<(i64,)> = conn.query("SELECT 1").await.unwrap();
+    assert_eq!(rows, vec![(1,)]);
+
+    drop(conn);
+}
+
+/// Sanity: normal queries work over the wire against the Turso backend.
+#[tokio::test]
+async fn basic_crud_over_wire() {
+    let port = free_port().await;
+    let _server = start_litewire_turso(port).await;
+    let mut conn = connect(port).await;
+
+    conn.query_drop("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL)")
+        .await
+        .unwrap();
+    conn.exec_drop("INSERT INTO users (name) VALUES (?)", ("Alice",))
+        .await
+        .unwrap();
+    let rows: Vec<(i64, String)> = conn.query("SELECT id, name FROM users").await.unwrap();
+    assert_eq!(rows, vec![(1, "Alice".to_string())]);
+
+    drop(conn);
+}