fix: reject multiple statements in single-statement APIs

libsql/src/connection.rs

@@ -26,6 +26,11 @@ pub enum Op {
 pub(crate) trait Conn {
     async fn execute(&self, sql: &str, params: Params) -> Result<u64>;
 
+    async fn query(&self, sql: &str, params: Params) -> Result<Rows> {
+        let stmt = self.prepare(sql).await?;
+        stmt.query(params).await
+    }
+
     async fn execute_batch(&self, sql: &str) -> Result<BatchRows>;
 
     async fn execute_transactional_batch(&self, sql: &str) -> Result<BatchRows>;
@@ -191,9 +196,7 @@ impl Connection {
     /// For more info on how to pass params check [`IntoParams`]'s docs and on how to
     /// extract values out of the rows check the [`Rows`] docs.
     pub async fn query(&self, sql: &str, params: impl IntoParams) -> Result<Rows> {
-        let stmt = self.prepare(sql).await?;
-
-        stmt.query(params).await
+        self.conn.query(sql, params.into_params()?).await
     }
 
     /// Prepares a cached statement.

libsql/src/local/connection.rs

@@ -132,7 +132,8 @@ impl Connection {
         P: TryInto<Params>,
         P::Error: Into<crate::BoxError>,
     {
-        let stmt = Statement::prepare(self.clone(), self.raw, sql.into().as_str())?;
+        let sql = sql.into();
+        let stmt = Statement::prepare(self.clone(), self.raw, sql.as_str())?;
         let params = params
             .try_into()
             .map_err(|e| Error::ToSqlConversionFailure(e.into()))?;
@@ -331,13 +332,39 @@ impl Connection {
         P: TryInto<Params>,
         P::Error: Into<crate::BoxError>,
     {
-        let stmt = Statement::prepare(self.clone(), self.raw, sql.into().as_str())?;
+        let sql = sql.into();
+        let stmt = self.prepare_single_statement(&sql)?;
         let params = params
             .try_into()
             .map_err(|e| Error::ToSqlConversionFailure(e.into()))?;
         stmt.execute(&params)
     }
 
+    /// Prepare a statement for the single-statement execute API.
+    ///
+    /// SQLite prepares the first statement in an input string and returns a
+    /// pointer to any remaining SQL.  `execute` is intentionally a
+    /// single-statement operation; callers that need more than one statement
+    /// should use `execute_batch` instead.  Check the tail before stepping the
+    /// first statement so a rejected input has no partial side effects.
+    pub(crate) fn prepare_single_statement(&self, sql: &str) -> Result<Statement> {
+        let stmt = Statement::prepare(self.clone(), self.raw, sql)?;
+        let tail = stmt.tail();
+        if tail != 0 && tail < sql.len() {
+            // A tail containing only whitespace or comments produces a null
+            // statement, and is valid after the first statement.  Preparing
+            // the tail lets SQLite perform that distinction for us.
+            let trailing = Statement::prepare(self.clone(), self.raw, &sql[tail..])?;
+            if !trailing.inner.raw_stmt.is_null() {
+                return Err(Error::Misuse(
+                    "multiple SQL statements are not supported; use execute_batch instead"
+                        .to_string(),
+                ));
+            }
+        }
+        Ok(stmt)
+    }
+
     /// Execute the SQL statement synchronously.
     ///
     /// This method never blocks the thread until, but instead returns a

libsql/src/local/impls.rs

@@ -23,6 +23,12 @@ impl Conn for LibsqlConnection {
         self.conn.execute(sql, params)
     }
 
+    async fn query(&self, sql: &str, params: Params) -> Result<Rows> {
+        let stmt = self.conn.prepare_single_statement(sql)?;
+        let rows = stmt.query(&params)?;
+        Ok(Rows::new(LibsqlRows(rows)))
+    }
+
     async fn execute_batch(&self, sql: &str) -> Result<BatchRows> {
         self.conn.execute_batch(sql)
     }

libsql/src/parser.rs

@@ -192,6 +192,20 @@ impl StmtKind {
 }
 
 impl Statement {
+    /// Reject input containing more than one SQL statement.
+    ///
+    /// The single-statement APIs use this check before dispatching to either
+    /// the local replica or the remote writer.  `execute_batch` remains the
+    /// explicit API for multi-statement input.
+    pub fn ensure_single(stmts: &[Self]) -> Result<()> {
+        if stmts.len() > 1 {
+            return Err(Error::Misuse(
+                "multiple SQL statements are not supported; use execute_batch instead".into(),
+            ));
+        }
+        Ok(())
+    }
+
     pub fn empty() -> Self {
         Self {
             stmt: String::new(),

libsql/src/replication/connection.rs

@@ -311,6 +311,7 @@ impl RemoteConnection {
 impl Conn for RemoteConnection {
     async fn execute(&self, sql: &str, params: Params) -> Result<u64> {
         let stmts = parser::Statement::parse(sql).collect::<Result<Vec<_>>>()?;
+        parser::Statement::ensure_single(&stmts)?;
 
         if self.should_execute_local(&stmts[..])? {
             // TODO(lucio): See if we can arc the params here to cheaply clone
@@ -602,6 +603,7 @@ pub struct RemoteStatement {
 impl RemoteStatement {
     pub async fn prepare(conn: RemoteConnection, sql: &str) -> Result<Self> {
         let stmts = parser::Statement::parse(sql).collect::<Result<Vec<_>>>()?;
+        parser::Statement::ensure_single(&stmts)?;
 
         if conn.should_execute_local(&stmts[..])? {
             tracing::trace!("Preparing {sql} locally");

libsql/tests/integration_tests.rs

@@ -629,6 +629,82 @@ async fn transaction() {
     assert!(rows.next().await.unwrap().is_none());
 }
 
+#[tokio::test]
+async fn single_statement_apis_reject_multiple_statements() {
+    let db = Database::open(":memory:").unwrap();
+    let conn = db.connect().unwrap();
+    conn.execute("CREATE TABLE values_table (value INTEGER)", ())
+        .await
+        .unwrap();
+
+    let result = conn
+        .execute(
+            "INSERT INTO values_table VALUES (1); UPDATE values_table SET value = 2;",
+            (),
+        )
+        .await;
+    assert!(
+        matches!(result, Err(libsql::Error::Misuse(message)) if message.contains("multiple SQL statements"))
+    );
+
+    let mut rows = conn
+        .query("SELECT value FROM values_table", ())
+        .await
+        .unwrap();
+    assert!(rows.next().await.unwrap().is_none());
+}
+
+#[tokio::test]
+async fn query_rejects_multiple_statements_without_side_effects() {
+    let db = Database::open(":memory:").unwrap();
+    let conn = db.connect().unwrap();
+    conn.execute("CREATE TABLE values_table (value INTEGER)", ())
+        .await
+        .unwrap();
+
+    let result = conn
+        .query(
+            "INSERT INTO values_table VALUES (1); UPDATE values_table SET value = 2;",
+            (),
+        )
+        .await;
+    assert!(
+        matches!(result, Err(libsql::Error::Misuse(message)) if message.contains("multiple SQL statements"))
+    );
+
+    let mut rows = conn
+        .query("SELECT value FROM values_table", ())
+        .await
+        .unwrap();
+    assert!(rows.next().await.unwrap().is_none());
+}
+
+#[tokio::test]
+async fn single_statement_apis_allow_trailing_comments() {
+    let db = Database::open(":memory:").unwrap();
+    let conn = db.connect().unwrap();
+    conn.execute("CREATE TABLE values_table (value INTEGER)", ())
+        .await
+        .unwrap();
+
+    conn.execute(
+        "INSERT INTO values_table VALUES (1); -- trailing comment",
+        (),
+    )
+    .await
+    .unwrap();
+
+    let mut rows = conn
+        .query("SELECT value FROM values_table", ())
+        .await
+        .unwrap();
+    assert_eq!(
+        rows.next().await.unwrap().unwrap().get::<i64>(0).unwrap(),
+        1
+    );
+    assert!(rows.next().await.unwrap().is_none());
+}
+
 #[tokio::test]
 async fn custom_params() {
     let conn = setup().await;

libsql/tests/replication.rs

@@ -53,6 +53,36 @@ async fn inject_frames() {
         10
     );
 
+    // Single-statement APIs must reject a second statement before dispatching
+    // to the embedded replica, so the first statement cannot partially apply.
+    let result = conn
+        .execute("INSERT INTO test VALUES (99); UPDATE test SET c = 42", ())
+        .await;
+    assert!(
+        matches!(result, Err(libsql::Error::Misuse(message)) if message.contains("multiple SQL statements"))
+    );
+
+    let result = conn
+        .query("INSERT INTO test VALUES (100); SELECT c FROM test", ())
+        .await;
+    assert!(
+        matches!(result, Err(libsql::Error::Misuse(message)) if message.contains("multiple SQL statements"))
+    );
+
+    let mut rows = conn.query("select count(*) from test", ()).await.unwrap();
+    assert_eq!(
+        *rows
+            .next()
+            .await
+            .unwrap()
+            .unwrap()
+            .get_value(0)
+            .unwrap()
+            .as_integer()
+            .unwrap(),
+        10
+    );
+
     // inject the same frames again, this should be idempotent
     let mut frames: Vec<FrameMut> = DB
         .chunks(LIBSQL_PAGE_SIZE)