diff --git a/crates/litewire-turso/src/lib.rs b/crates/litewire-turso/src/lib.rs index 72c429a..ba657f5 100644 --- a/crates/litewire-turso/src/lib.rs +++ b/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 { @@ -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(); diff --git a/crates/litewire/tests/mysql_turso_e2e.rs b/crates/litewire/tests/mysql_turso_e2e.rs new file mode 100644 index 0000000..adc0b1c --- /dev/null +++ b/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); +}