From 81d4ff9cc76ac1ce4626176d85936f1474537964 Mon Sep 17 00:00:00 2001 From: Ngo Quoc Dat Date: Thu, 22 Oct 2026 17:02:56 +0700 Subject: [PATCH 1/2] fix(linux-drivers): stop an editor script before its next statement on MySQL, MariaDB and SQLite Sent whole, MySQL starts the next statement once the last result is written, so a KILL that missed let it run while the run read Cancelled. --- linux/crates/drivers/mssql/src/cancel.rs | 4 +- .../drivers/mssql/src/in_band_request.rs | 41 +- linux/crates/drivers/mysql/src/cancel.rs | 44 ++- linux/crates/drivers/mysql/src/editor_run.rs | 55 +-- .../drivers/mysql/src/editor_session.rs | 132 +++++-- linux/crates/drivers/mysql/src/lib.rs | 1 + .../crates/drivers/mysql/src/script_split.rs | 362 ++++++++++++++++++ .../mysql/tests/integration/editor_session.rs | 177 +++++++-- linux/crates/drivers/sqlite/src/connection.rs | 2 +- .../drivers/sqlite/src/editor_session.rs | 37 +- .../drivers/sqlite/src/result_stream.rs | 13 +- .../sqlite/tests/integration/sessions.rs | 38 +- .../sqlite/tests/integration/support.rs | 13 +- linux/docs/adding-drivers.md | 1 + linux/docs/drivers/mysql.md | 10 +- linux/docs/drivers/sql-server.md | 2 + linux/docs/drivers/sqlite.md | 2 +- 17 files changed, 801 insertions(+), 133 deletions(-) create mode 100644 linux/crates/drivers/mysql/src/script_split.rs diff --git a/linux/crates/drivers/mssql/src/cancel.rs b/linux/crates/drivers/mssql/src/cancel.rs index ddb820918b..0de760706f 100644 --- a/linux/crates/drivers/mssql/src/cancel.rs +++ b/linux/crates/drivers/mssql/src/cancel.rs @@ -52,8 +52,8 @@ impl EngineStop for MssqlStop { Ok(()) } - /// The work returns `Cancelled` only once the server has confirmed - /// the attention, or once a stop arrived before anything was sent. + /// The work returns `Cancelled` once the server has confirmed the + /// attention, and for a cancel that found nothing left to stop. fn acknowledged(&self, error: &DriverError) -> bool { matches!(error, DriverError::Cancelled) } diff --git a/linux/crates/drivers/mssql/src/in_band_request.rs b/linux/crates/drivers/mssql/src/in_band_request.rs index 63d5ea2186..0f965c81e8 100644 --- a/linux/crates/drivers/mssql/src/in_band_request.rs +++ b/linux/crates/drivers/mssql/src/in_band_request.rs @@ -252,7 +252,7 @@ async fn stopped( match halt { Halt::Unsent | Halt::Finished => { let clean = !roll_back || roll_back_open(client, until).await; - (DriverError::Cancelled, clean) + (stop_error(watch.request(), None), clean) } Halt::Boundary | Halt::MidToken => { let answer = attention(client, until).await; @@ -260,17 +260,25 @@ async fn stopped( Answer::Unanswered => StopProgress::Failed, Answer::Acknowledged | Answer::Garbled => StopProgress::Delivered, }); - match answer { - Answer::Acknowledged => { - let clean = halt == Halt::Boundary && (!roll_back || roll_back_open(client, until).await); - (DriverError::Cancelled, clean) - } - Answer::Garbled | Answer::Unanswered => (unconfirmed_stop(watch.request()), false), - } + let clean = match answer { + Answer::Acknowledged => halt == Halt::Boundary && (!roll_back || roll_back_open(client, until).await), + Answer::Garbled | Answer::Unanswered => false, + }; + (stop_error(watch.request(), Some(answer)), clean) } } } +/// Only an acknowledged attention says the server stopped the request. +/// One that was never sent, or ended on its own while the stop waited +/// for a boundary, ran as far as it was going to. +fn stop_error(request: &StopRequest, attention: Option) -> DriverError { + match attention { + Some(Answer::Acknowledged) => DriverError::Cancelled, + Some(Answer::Garbled | Answer::Unanswered) | None => unconfirmed_stop(request), + } +} + /// Send the attention and read up to its acknowledgement, by `until`. pub(crate) async fn attention(client: &mut MssqlClient, until: Instant) -> Answer { match tokio::time::timeout_at(until, client.cancel_query()).await { @@ -405,6 +413,23 @@ mod tests { assert!(matches!(outcome, Err(Interrupt::Stopped(Halt::Finished)))); } + #[test] + fn only_an_acknowledged_attention_reads_as_the_server_stopping_it() { + let deadline = StopRequest::new(); + deadline.request(StopReason::DeadlineElapsed); + let unconfirmed = DriverError::Timeout { + phase: tablepro_core::TimeoutPhase::Statement, + server_cancelled: false, + }; + + assert_eq!( + stop_error(&deadline, Some(Answer::Acknowledged)), + DriverError::Cancelled + ); + assert_eq!(stop_error(&deadline, None), unconfirmed); + assert_eq!(stop_error(&deadline, Some(Answer::Unanswered)), unconfirmed); + } + #[tokio::test] async fn a_broken_or_garbled_response_while_stopping_ends_the_connection() { let garbled = tiberius::error::Error::Conversion("nothing to see".into()); diff --git a/linux/crates/drivers/mysql/src/cancel.rs b/linux/crates/drivers/mysql/src/cancel.rs index 17c23d387f..0eb68d8a87 100644 --- a/linux/crates/drivers/mysql/src/cancel.rs +++ b/linux/crates/drivers/mysql/src/cancel.rs @@ -37,7 +37,16 @@ impl ControlLink { /// it rather than leaving a half-read answer for the next one. pub(crate) async fn kill_query(&self, id: u64) -> Result<(), DriverError> { let mut held = self.connection.lock().await; - let mut physical = match held.take() { + // The server may have closed the kept one since the last stop, at + // its wait_timeout or a restart. A ping says so without sending + // the KILL, which, once sent, must not be sent again elsewhere. + let mut kept = held.take(); + if let Some(physical) = kept.as_mut() + && physical.connection.ping().await.is_err() + { + kept = None; + } + let mut physical = match kept { Some(physical) => physical, None => { self.endpoint @@ -45,23 +54,9 @@ impl ControlLink { .await? } }; - let flavour = physical.flavour; - let killed = physical - .connection - .execute(AssertSqlSafe(format!("KILL QUERY {id}"))) - .await - .map_err(|error| map_error(&error, flavour)); - match killed { - Ok(_) => { - *held = Some(physical); - Ok(()) - } - Err(error) if server_number(&error) == Some(NO_SUCH_THREAD) => { - *held = Some(physical); - Ok(()) - } - Err(error) => Err(error), - } + kill(&mut physical, id).await?; + *held = Some(physical); + Ok(()) } pub(crate) async fn close(&self) { @@ -73,6 +68,19 @@ impl ControlLink { } } +async fn kill(physical: &mut MySqlPhysical, id: u64) -> Result<(), DriverError> { + let flavour = physical.flavour; + let killed = physical + .connection + .execute(AssertSqlSafe(format!("KILL QUERY {id}"))) + .await + .map_err(|error| map_error(&error, flavour)); + match killed { + Err(error) if server_number(&error) != Some(NO_SUCH_THREAD) => Err(error), + _ => Ok(()), + } +} + /// How far a call's KILL QUERY got. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum StopProgress { diff --git a/linux/crates/drivers/mysql/src/editor_run.rs b/linux/crates/drivers/mysql/src/editor_run.rs index dab336105b..f80bb3c71e 100644 --- a/linux/crates/drivers/mysql/src/editor_run.rs +++ b/linux/crates/drivers/mysql/src/editor_run.rs @@ -12,8 +12,6 @@ use futures_util::TryStreamExt; use sqlx::mysql::{MySqlConnection, MySqlRow}; use sqlx::{AssertSqlSafe, Either, Executor as _, Row as _}; use tablepro_core::column::ResultColumn; -use tablepro_core::sql_syntax::SqlGrammar; -use tablepro_core::sql_syntax::script::{LexicalSettings, ScriptPlan, ScriptTokenKind}; use tablepro_core::{DriverError, ResultEvent, ResultSink, RowLimit, SinkFlow, Truncation, Value}; use tokio_util::sync::CancellationToken; @@ -23,6 +21,7 @@ use crate::errors::map_error; use crate::flavour::MySqlFlavour; use crate::mysql_timestamp_mode::MySqlTimestampMode; use crate::mysql_wire_format::MySqlWireFormat; +use crate::script_split; /// How many rows travel in one `Rows` event. const ROWS_PER_EVENT: usize = 256; @@ -31,34 +30,23 @@ const ROWS_PER_EVENT: usize = 256; pub(crate) struct RunOutcome { /// The sink stopped reading, so the run was cut short. pub consumer_gone: bool, + /// A stop came between two statements, so the rest was not sent. + pub halted: bool, } -/// The texts to send. A script with mysql-client DELIMITER lines is -/// split where the client would split it, since the server does not -/// know the directive; anything else goes whole, and the server runs it -/// as one multi-statement batch. -pub(crate) fn batches(sql: &str) -> Vec<&str> { - let plan = ScriptPlan::build(sql, SqlGrammar::MySql, LexicalSettings::default_for(SqlGrammar::MySql)); - let directives = plan - .tokens(0..sql.len()) - .any(|token| token.kind == ScriptTokenKind::Directive); - if !directives || !plan.diagnostics().is_empty() { - return vec![sql]; - } - plan.statements().iter().map(|statement| statement.text(sql)).collect() -} - -/// Run the script. When the sink stops reading, `consumer_gone` is -/// cancelled so the caller can stop the server. The batch already sent -/// is still read to its end, unshown, since only that frees the -/// connection, and no later batch is sent. +/// Run the script one text per round trip (`script_split`), sending the +/// next only while `halt` says no stop has come. When the sink stops +/// reading, `stopped` is cancelled so the caller can stop the server. +/// The text already sent is still read to its end, unshown, since only +/// that frees the connection. pub(crate) async fn run_script( connection: &mut MySqlConnection, flavour: MySqlFlavour, sql: &str, limit: RowLimit, sink: &mut dyn ResultSink, - consumer_gone: &CancellationToken, + stopped: &CancellationToken, + halt: impl Fn() -> bool, ) -> Result { let mut emitter = Emitter { sink, @@ -67,10 +55,15 @@ pub(crate) async fn run_script( sets: 0, gone: false, }; - for text in batches(sql) { + let mut halted = false; + for text in script_split::texts(sql) { if emitter.gone { break; } + if stopped.is_cancelled() || halt() { + halted = true; + break; + } let results = connection.fetch_many(sqlx::raw_sql(AssertSqlSafe(text.to_owned()))); let mut results = std::pin::pin!(results); while let Some(item) = results.try_next().await.map_err(|error| map_error(&error, flavour))? { @@ -79,12 +72,13 @@ pub(crate) async fn run_script( Either::Left(done) => emitter.done(done.rows_affected()).await, } if emitter.gone { - consumer_gone.cancel(); + stopped.cancel(); } } } Ok(RunOutcome { consumer_gone: emitter.gone, + halted, }) } @@ -195,17 +189,4 @@ mod tests { assert_eq!(truncation(101, 100), (Truncation::ServerStopped, None)); assert_eq!(truncation(150_000, 100), (Truncation::ReadPastLimit, Some(150_000))); } - - #[test] - fn delimiter_scripts_split_where_the_client_would() { - let script = "DELIMITER //\nCREATE PROCEDURE p() BEGIN SELECT 1; SELECT 2; END//\nDELIMITER ;\nCALL p();"; - - let split = batches(script); - - assert_eq!(split.len(), 2, "{split:?}"); - assert!(split[0].starts_with("CREATE PROCEDURE"), "{split:?}"); - assert!(split[0].ends_with("END"), "{split:?}"); - assert!(split[1].starts_with("CALL p()"), "{split:?}"); - assert_eq!(batches("SELECT 1; SELECT 2"), ["SELECT 1; SELECT 2"]); - } } diff --git a/linux/crates/drivers/mysql/src/editor_session.rs b/linux/crates/drivers/mysql/src/editor_session.rs index 10bdd7cdd2..7907f61351 100644 --- a/linux/crates/drivers/mysql/src/editor_session.rs +++ b/linux/crates/drivers/mysql/src/editor_session.rs @@ -157,7 +157,7 @@ impl Shared { /// Run the script, stopping it on a cancel, at the deadline, when /// the session closes or when the sink stops reading. A stop is KILL /// QUERY on a connection of the session's own; the statement then - /// has the cancel grace to end. + /// has the cancel grace to end, and nothing after it is sent. async fn supervise( &self, held: &mut Held, @@ -166,52 +166,88 @@ impl Shared { sink: &mut dyn ResultSink, ) -> Result<(RunEnd, bool), DriverError> { let limit = options.row_limit; - if let Err(error) = set_select_limit(held, self.flavour, limit).await { - return finished(Err(error)).map(|end| (end, true)); - } let deadline = options.deadline.map(|limit| Instant::now() + limit); - let consumer_gone = CancellationToken::new(); - let running = run_script( - &mut held.connection, - self.flavour, - sql.as_str(), - limit, - sink, - &consumer_gone, - ); + let stopped = CancellationToken::new(); + // Read synchronously before each statement goes out, since the + // select below sees a stop only once the run waits. + let halt = || self.stop_due(options, deadline).is_some(); + let flavour = self.flavour; + let running = async { + set_select_limit(held, flavour, limit).await?; + run_script(&mut held.connection, flavour, sql.as_str(), limit, sink, &stopped, halt).await + }; let mut running = std::pin::pin!(running); let stop = tokio::select! { - outcome = &mut running => return finished(outcome).map(|end| (end, true)), + outcome = &mut running => { + return match outcome { + // Stopped between statements: nothing runs to KILL. + Ok(RunOutcome { halted: true, .. }) => { + let stop = self.stop_due(options, deadline).unwrap_or(Stop::Cancelled); + Ok((stopped_end(stop, false), true)) + } + outcome => finished(outcome).map(|end| (end, true)), + }; + } () = options.cancel.cancelled() => Stop::Cancelled, () = self.closing.cancelled() => Stop::Cancelled, - () = consumer_gone.cancelled() => Stop::Cancelled, + () = stopped.cancelled() => Stop::Cancelled, () = elapsed(deadline) => Stop::DeadlineElapsed, }; + stopped.cancel(); let until = Instant::now() + self.liveness.cancel_grace; - // Nothing more is read until the KILL has landed. A batch the - // server is still sending then waits on the socket, so it cannot - // move on to its next statement before the KILL stops it. - let delivered = tokio::time::timeout_at(until, self.control.kill_query(self.connection_id)) + // Nothing more is read until the KILL has its answer, so it lands + // while the statement it names is still the connection's last. + let answered = tokio::time::timeout_at(until, self.control.kill_query(self.connection_id)) .await - .ok() - .map(|sent| sent.is_ok()); + .is_ok(); let settled = tokio::time::timeout_at(until, &mut running).await.ok(); - let server_cancelled = delivered == Some(true); - let end = match stop { - Stop::Cancelled => RunEnd::Cancelled, - Stop::DeadlineElapsed => RunEnd::TimedOut { server_cancelled }, - }; + let confirmed = matches!(&settled, Some(Err(error)) if confirms_stop(error)); + let end = stopped_end(stop, confirmed); match settled { - // Whatever the statement did once it was stopped, the run - // ends as stopped: its result may be cut short by the KILL. - Some(Ok(_)) => Ok((end, true)), Some(Err(DriverError::ConnectionLost { during })) => Err(DriverError::ConnectionLost { during }), - Some(Err(_)) => Ok((end, true)), + // An unanswered KILL may still land, on whatever runs next here. + _ if !answered => Ok((end, false)), + Some(_) => Ok((end, true)), // The statement would not stop, so the connection still has // its answer coming and cannot take another run. None => Ok((end, false)), } } + + fn stop_due(&self, options: &RunOptions, deadline: Option) -> Option { + if options.cancel.is_cancelled() || self.closing.is_cancelled() { + Some(Stop::Cancelled) + } else if deadline.is_some_and(|at| Instant::now() >= at) { + Some(Stop::DeadlineElapsed) + } else { + None + } + } +} + +/// A stopped run reads as stopped whatever its statement did after the +/// stop, since nothing after it ran. +fn stopped_end(stop: Stop, confirmed: bool) -> RunEnd { + match stop { + Stop::Cancelled => RunEnd::Cancelled, + Stop::DeadlineElapsed => RunEnd::TimedOut { + server_cancelled: confirmed, + }, + } +} + +/// The statement ended with the server's word that it was stopped: 1317 +/// from the KILL, or the server's own statement timeout. An OK, or a +/// KILL that only returned OK, says nothing about it. +fn confirms_stop(error: &DriverError) -> bool { + matches!( + error, + DriverError::Cancelled + | DriverError::Timeout { + phase: TimeoutPhase::Statement, + server_cancelled: true, + } + ) } async fn set_select_limit(held: &mut Held, flavour: MySqlFlavour, limit: RowLimit) -> Result<(), DriverError> { @@ -228,8 +264,10 @@ async fn set_select_limit(held: &mut Held, flavour: MySqlFlavour, limit: RowLimi fn finished(outcome: Result) -> Result { match outcome { - Ok(RunOutcome { consumer_gone: true }) => Ok(RunEnd::Cancelled), - Ok(RunOutcome { consumer_gone: false }) => Ok(RunEnd::Completed), + Ok(RunOutcome { + consumer_gone: true, .. + }) => Ok(RunEnd::Cancelled), + Ok(RunOutcome { .. }) => Ok(RunEnd::Completed), // A connection that is gone ends the session; a statement the // server refused ends the run. Err(error @ DriverError::ConnectionLost { .. }) => Err(error), @@ -316,3 +354,33 @@ impl EditorSession for MysqlEditorSession { shared.control.close().await; } } + +#[cfg(test)] +mod tests { + use tablepro_core::BusyReason; + + use super::*; + + #[test] + fn a_timeout_reads_as_stopped_by_the_server_only_when_the_statement_said_so() { + let server_timeout = DriverError::Timeout { + phase: TimeoutPhase::Statement, + server_cancelled: true, + }; + + assert!(confirms_stop(&DriverError::Cancelled)); + assert!(confirms_stop(&server_timeout)); + assert!(!confirms_stop(&DriverError::Busy(BusyReason::NoFreeConnection))); + assert_eq!( + stopped_end(Stop::DeadlineElapsed, false), + RunEnd::TimedOut { + server_cancelled: false + } + ); + assert_eq!( + stopped_end(Stop::DeadlineElapsed, true), + RunEnd::TimedOut { server_cancelled: true } + ); + assert_eq!(stopped_end(Stop::Cancelled, false), RunEnd::Cancelled); + } +} diff --git a/linux/crates/drivers/mysql/src/lib.rs b/linux/crates/drivers/mysql/src/lib.rs index 710f1c8b26..d6c152a150 100644 --- a/linux/crates/drivers/mysql/src/lib.rs +++ b/linux/crates/drivers/mysql/src/lib.rs @@ -26,6 +26,7 @@ mod mysql_temporal; mod mysql_timestamp_mode; mod mysql_wire_format; mod physical; +mod script_split; mod server_zone; mod session_statements; mod standalone; diff --git a/linux/crates/drivers/mysql/src/script_split.rs b/linux/crates/drivers/mysql/src/script_split.rs new file mode 100644 index 0000000000..cd8d2d2ec1 --- /dev/null +++ b/linux/crates/drivers/mysql/src/script_split.rs @@ -0,0 +1,362 @@ +//! An editor script goes one statement per round trip. Sent whole, the +//! server starts each next statement once the last one's result is +//! written, read or not, so a KILL that misses lets the rest run. + +use std::ops::Range; + +use tablepro_core::sql_syntax::SqlGrammar; +use tablepro_core::sql_syntax::script::{LexicalSettings, ScriptPlan, ScriptToken, ScriptTokenKind}; + +/// Split where the mysql client splits: at each `;`, or at the delimiter +/// a DELIMITER line sets. +pub(crate) fn texts(sql: &str) -> Vec<&str> { + let plan = ScriptPlan::build(sql, SqlGrammar::MySql, LexicalSettings::default_for(SqlGrammar::MySql)); + // A string or comment that never closes: the server reports it. + if !plan.diagnostics().is_empty() { + return vec![sql]; + } + let tokens: Vec = plan.tokens(0..sql.len()).collect(); + let units = units(sql, &tokens); + let directives = tokens.iter().any(|token| token.kind == ScriptTokenKind::Directive); + // One statement goes as written: the client's view of where it ends + // can differ from the server's, as under NO_BACKSLASH_ESCAPES. + if !directives && (units.len() <= 1 || whole_by_server(sql, &tokens, &units)) { + return vec![sql]; + } + units.into_iter().filter_map(|unit| sql.get(unit)).collect() +} + +/// The text between each two terminators, less the whitespace and plain +/// comments at its edges. The server runs what a `/*!` comment holds, so +/// such a comment is code, even standing alone. +fn units(sql: &str, tokens: &[ScriptToken]) -> Vec> { + let mut units = Vec::new(); + let mut code: Option> = None; + for token in tokens { + if matches!(token.kind, ScriptTokenKind::Semicolon | ScriptTokenKind::Directive) { + units.extend(code.take()); + } else if !token.kind.is_trivia() || executable(sql, token) { + code = Some(code.map_or(token.range.clone(), |code| code.start..token.range.end)); + } + } + units.extend(code); + units +} + +fn executable(sql: &str, token: &ScriptToken) -> bool { + token.kind == ScriptTokenKind::BlockComment + && sql + .get(token.range.clone()) + .is_some_and(|text| text.starts_with("/*!") || text.starts_with("/*M!")) +} + +/// The client's split holds only for statements it reads as the server +/// does. A statement it does not know may have a body, which only the +/// server's grammar can end, and the session's sql_mode decides how some +/// text reads: a backslash in a string, `[name]` under MariaDB's MSSQL +/// mode, `$$` quotes on MySQL 8.4. A `/*!` comment may split on its own. +fn whole_by_server(sql: &str, tokens: &[ScriptToken], units: &[Range]) -> bool { + let read_by_the_server = |token: &ScriptToken| { + let text = sql.get(token.range.clone()).unwrap_or_default(); + match token.kind { + ScriptTokenKind::BlockComment => executable(sql, token), + ScriptTokenKind::StringLiteral => text.contains('\\'), + ScriptTokenKind::Word => text.contains('$'), + ScriptTokenKind::Punct => !PLAIN_PUNCTUATION.contains(text), + _ => false, + } + }; + tokens.iter().any(read_by_the_server) + || units.iter().any(|unit| { + let first = tokens.partition_point(|token| token.range.end <= unit.start); + let last = tokens.partition_point(|token| token.range.start < unit.end); + !plain(sql, tokens.get(first..last).unwrap_or_default()) + }) +} + +const PLAIN_PUNCTUATION: &str = ",.=<>!+-*/%&|^~:@?"; + +/// The words that start a statement with no body of its own. +const PLAIN_STARTS: [&str; 54] = [ + "ANALYZE", + "BACKUP", + "BINLOG", + "CACHE", + "CALL", + "CHANGE", + "CHECK", + "CHECKSUM", + "CLONE", + "COMMIT", + "DEALLOCATE", + "DELETE", + "DESC", + "DESCRIBE", + "DO", + "DROP", + "EXECUTE", + "EXPLAIN", + "FLUSH", + "GRANT", + "HANDLER", + "HELP", + "IMPORT", + "INSERT", + "INSTALL", + "KILL", + "LOAD", + "LOCK", + "OPTIMIZE", + "PREPARE", + "PURGE", + "RELEASE", + "RENAME", + "REPAIR", + "REPLACE", + "RESET", + "RESTART", + "REVOKE", + "ROLLBACK", + "SAVEPOINT", + "SELECT", + "SET", + "SHOW", + "SHUTDOWN", + "START", + "TABLE", + "TRUNCATE", + "UNINSTALL", + "UNLOCK", + "UPDATE", + "USE", + "VALUES", + "WITH", + "XA", +]; + +/// Objects whose definition never holds a `;`. +const PLAIN_OBJECTS: [&str; 18] = [ + "TABLE", + "TEMPORARY", + "VIEW", + "INDEX", + "UNIQUE", + "FULLTEXT", + "SPATIAL", + "DATABASE", + "SCHEMA", + "USER", + "ROLE", + "SERVER", + "SEQUENCE", + "TABLESPACE", + "LOGFILE", + "RESOURCE", + "UNDO", + "INSTANCE", +]; + +fn plain(sql: &str, tokens: &[ScriptToken]) -> bool { + let text = |token: &ScriptToken| sql.get(token.range.clone()).unwrap_or_default(); + let mut rest = tokens; + let Some(first) = next_significant(&mut rest) else { + return true; + }; + if first.kind != ScriptTokenKind::Word { + return false; + } + let word = text(first); + if is(word, &["CREATE", "ALTER"]) { + return !creates_program(sql, rest); + } + // BEGIN and BEGIN WORK start a transaction; anything else after it + // is a block: BEGIN NOT ATOMIC, or sql_mode=ORACLE. + if is(word, &["BEGIN"]) { + return next_significant(&mut rest) + .is_none_or(|token| token.kind == ScriptTokenKind::Word && is(text(token), &["WORK"])); + } + is(word, &PLAIN_STARTS) +} + +/// Whether the header after CREATE or ALTER may name a stored program, +/// read past `OR REPLACE`, `AGGREGATE`, `DEFINER = user`, `SQL SECURITY` +/// and `ALGORITHM`. Anything but a plain object counts, since sending a +/// plain statement whole costs less than cutting a body apart. +fn creates_program(sql: &str, tokens: &[ScriptToken]) -> bool { + let text = |token: &ScriptToken| sql.get(token.range.clone()).unwrap_or_default(); + let mut rest = tokens; + loop { + let Some(token) = next_significant(&mut rest) else { + return false; + }; + if token.kind != ScriptTokenKind::Word { + return true; + } + let word = text(token); + match word.to_ascii_uppercase().as_str() { + "OR" | "REPLACE" | "AGGREGATE" => {} + "SQL" | "ALGORITHM" => { + next_significant(&mut rest); + next_significant(&mut rest); + } + "DEFINER" => { + next_significant(&mut rest); + skip_user(sql, &mut rest); + } + _ => return !is(word, &PLAIN_OBJECTS), + } + } +} + +/// Past `CURRENT_USER`, `CURRENT_USER()`, `name`, or `name@host` with the +/// host written as one run of tokens (`127.0.0.1`). +fn skip_user(sql: &str, tokens: &mut &[ScriptToken]) { + let is_punct = |token: Option<&ScriptToken>, punct: &str| { + token.is_some_and(|token| sql.get(token.range.clone()) == Some(punct)) + }; + let Some(name) = next_significant(tokens) else { + return; + }; + let peek = |tokens: &[ScriptToken]| tokens.iter().find(|token| !token.kind.is_trivia()).cloned(); + if sql + .get(name.range.clone()) + .is_some_and(|text| text.eq_ignore_ascii_case("CURRENT_USER")) + { + if peek(tokens).is_some_and(|token| token.kind == ScriptTokenKind::OpenParen) { + next_significant(tokens); + next_significant(tokens); + } + return; + } + if is_punct(peek(tokens).as_ref(), "@") { + next_significant(tokens); + next_significant(tokens); + let joined = tokens + .iter() + .position(|token| token.kind.is_trivia()) + .unwrap_or(tokens.len()); + *tokens = tokens.get(joined..).unwrap_or_default(); + } +} + +fn next_significant<'a>(tokens: &mut &'a [ScriptToken]) -> Option<&'a ScriptToken> { + let at = tokens.iter().position(|token| !token.kind.is_trivia())?; + let token = tokens.get(at)?; + *tokens = tokens.get(at + 1..).unwrap_or_default(); + Some(token) +} + +fn is(text: &str, words: &[&str]) -> bool { + words.iter().any(|word| text.eq_ignore_ascii_case(word)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_plain_script_goes_one_statement_at_a_time() { + assert_eq!(texts("SELECT 1; SELECT 2"), ["SELECT 1", "SELECT 2"]); + assert_eq!( + texts("BEGIN;\nINSERT INTO t VALUES (';');\n-- done\nCOMMIT;"), + ["BEGIN", "INSERT INTO t VALUES (';')", "COMMIT"] + ); + assert_eq!(texts("BEGIN WORK; SELECT 1"), ["BEGIN WORK", "SELECT 1"]); + assert_eq!(texts("XA BEGIN 'x'; SELECT 1"), ["XA BEGIN 'x'", "SELECT 1"]); + assert_eq!( + texts("SET @a := -1; SELECT @@session.sql_mode, j->>'$.x' FROM t WHERE a <> 2 OR b IS NULL"), + [ + "SET @a := -1", + "SELECT @@session.sql_mode, j->>'$.x' FROM t WHERE a <> 2 OR b IS NULL" + ] + ); + assert_eq!( + texts("CREATE TABLE t (id int); DROP PROCEDURE p"), + ["CREATE TABLE t (id int)", "DROP PROCEDURE p"] + ); + assert_eq!( + texts( + "CREATE TABLE x (event int); CREATE DEFINER = admin@localhost SQL SECURITY INVOKER VIEW v AS SELECT 1" + ), + [ + "CREATE TABLE x (event int)", + "CREATE DEFINER = admin@localhost SQL SECURITY INVOKER VIEW v AS SELECT 1" + ] + ); + } + + #[test] + fn one_statement_goes_as_written() { + for script in [ + "-- note\n/* plain */ ;;", + "SELECT 1;\n-- trailing", + "SELECT 'a\\'; SELECT 'b -- ';", + "/*M!100100 SET @a = 1 */", + ] { + assert_eq!(texts(script), [script], "{script}"); + } + } + + #[test] + fn delimiter_scripts_split_where_the_client_would() { + let script = "DELIMITER //\nCREATE PROCEDURE p() BEGIN SELECT 1; SELECT 2; END//\nDELIMITER ;\nCALL p();"; + + assert_eq!( + texts(script), + ["CREATE PROCEDURE p() BEGIN SELECT 1; SELECT 2; END", "CALL p()"] + ); + } + + /// mysqldump writes a trigger as a run of `/*!` comments, which the + /// server runs. + #[test] + fn executable_comments_are_sent() { + let dump = "DELIMITER ;;\n/*!50003 CREATE*/ /*!50003 TRIGGER t1 BEFORE INSERT ON t FOR EACH ROW \ + BEGIN SET NEW.a = 1; END */;;\nDELIMITER ;\n/*!40101 SET NAMES utf8mb4 */;"; + + assert_eq!( + texts(dump), + [ + "/*!50003 CREATE*/ /*!50003 TRIGGER t1 BEFORE INSERT ON t FOR EACH ROW BEGIN SET NEW.a = 1; END */", + "/*!40101 SET NAMES utf8mb4 */" + ] + ); + } + + #[test] + fn a_body_only_the_server_can_end_goes_whole() { + let whole = [ + "CREATE TABLE tt (a int); CREATE TRIGGER b BEFORE INSERT ON tt FOR EACH ROW BEGIN SET NEW.a = 1; END", + "CREATE DEFINER = admin@localhost PROCEDURE p() IF 1 THEN SELECT 1; END IF", + "ALTER EVENT e DO BEGIN DELETE FROM t; END", + "CREATE OR REPLACE DEFINER = 'a'@'%' AGGREGATE FUNCTION f(x int) RETURNS int BEGIN RETURN 1; END", + "CREATE DEFINER = CURRENT_USER() TRIGGER g BEFORE INSERT ON t FOR EACH ROW BEGIN SET NEW.id = 1; END", + "CREATE DEFINER=root@127.0.0.1 PROCEDURE p() BEGIN SELECT 1; END", + "CREATE DEFINER = CURRENT_USER( ) PROCEDURE p() BEGIN SELECT 1; SELECT 2; END", + "CREATE DEFINER = 'a' @ 'b' EVENT e ON SCHEDULE EVERY 1 DAY DO BEGIN DELETE FROM t; END", + "CREATE DEFINER = user@localhost PROCEDURE p() BEGIN SELECT 1; END", + "/*!40101 SET NAMES utf8mb4 */; SELECT 1", + "BEGIN NOT ATOMIC SELECT 1; END", + "IF 1 THEN SELECT 1; END IF", + "WHILE @a < 3 DO SET @a = @a + 1; END WHILE", + "l1: LOOP LEAVE l1; END LOOP", + "`l1`: BEGIN NOT ATOMIC SELECT 1; END", + "SELECT 1 /*! ; DELETE FROM t */; SELECT 2", + "SELECT 'a\\'; DELETE FROM t; SELECT \\''; SELECT 1", + "SELECT 1 AS [a;b]; SELECT 2", + "SELECT $$a;b$$; SELECT 1", + "SELECT {d '2020-01-01'}; SELECT 1", + "BEGIN \"p\"(); END", + "(SELECT 1); DELETE FROM t", + ]; + + for script in whole { + assert_eq!(texts(script), [script], "{script}"); + } + } + + #[test] + fn unterminated_text_goes_whole() { + assert_eq!(texts("SELECT 1; SELECT 'open"), ["SELECT 1; SELECT 'open"]); + } +} diff --git a/linux/crates/drivers/mysql/tests/integration/editor_session.rs b/linux/crates/drivers/mysql/tests/integration/editor_session.rs index ecb850891d..236d599ca2 100644 --- a/linux/crates/drivers/mysql/tests/integration/editor_session.rs +++ b/linux/crates/drivers/mysql/tests/integration/editor_session.rs @@ -2,23 +2,26 @@ //! server: MariaDB's @@in_transaction, MySQL's performance_schema. And //! what a run does when its reader goes away. +use std::time::Duration; + use tablepro_core::{ - CallOptions, EditorSession, EditorSessionState, RowLimit, RunEnd, RunOptions, TransactionState, UserSql, + CallOptions, EditorSession, EditorSessionState, RowLimit, RunEnd, RunOptions, RunSummary, TransactionState, UserSql, }; use tablepro_test_fixtures::MysqlFlavour; use tokio_util::sync::CancellationToken; use crate::support::{ - CALL, EventSink, Events, FLAVOURS, connect, connect_root, count, run_ok, session, setup, start, within, + CALL, EventSink, Events, FLAVOURS, connect, connect_root, count, rows, run_ok, run_with, session, session_scalar, + setup, start, text, within, }; -/// A hundred thousand rows of a kilobyte each: far more than the socket -/// buffers hold, so the server is still sending the SELECT when the -/// reader leaves. The DELETE behind it must never run. -const WIDE_THEN_DELETE: &str = "WITH d AS (SELECT 0 AS n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 \ +/// A thousand rows, which the server has long finished sending when the +/// reader leaves at the first of them, then a DELETE that must never +/// reach it. +const ROWS_THEN_DELETE: &str = "WITH d AS (SELECT 0 AS n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 \ UNION ALL SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7 UNION ALL SELECT 8 \ UNION ALL SELECT 9) \ - SELECT REPEAT('x', 1000) AS filler FROM d a, d b, d c, d e, d f; \ + SELECT a.n FROM d a, d b, d c; \ DELETE FROM t"; async fn state_after(session: &dyn EditorSession, sql: &str) -> TransactionState { @@ -89,9 +92,21 @@ async fn mysql_user_without_ps_select_unknown() { connection.close().await.expect("a close"); } +async fn until_gone(editor: &dyn EditorSession, options: RunOptions) -> RunSummary { + let events = Events::default(); + editor + .run( + UserSql::new(ROWS_THEN_DELETE), + options, + Box::new(EventSink::new(&events).gone_after(1)), + ) + .await + .expect("the run") +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[ignore = "requires docker"] -async fn consumer_gone_kills_the_rest_of_the_batch() { +async fn consumer_gone_sends_nothing_after_the_running_statement() { for flavour in FLAVOURS { let fixture = start(flavour).await; let connection = connect(&fixture, false).await; @@ -101,23 +116,16 @@ async fn consumer_gone_kills_the_rest_of_the_batch() { ) .await; let editor = session(connection.as_ref()).await; - let events = Events::default(); - let options = - RunOptions::new(CancellationToken::new()).with_row_limit(RowLimit::new(1_000_000).expect("a limit")); - - let summary = editor - .run( - UserSql::new(WIDE_THEN_DELETE), - options, - Box::new(EventSink::new(&events).gone_after(1)), - ) - .await - .expect("the run"); - assert_eq!(summary.end, RunEnd::Cancelled, "{flavour:?}"); + let gone = until_gone(editor.as_ref(), RunOptions::new(CancellationToken::new())).await; + assert_eq!(editor.state(), EditorSessionState::Ready, "{flavour:?}"); + // A killed SLEEP returns 1 with no error, so its statement ends + // as if nothing had stopped it. + let stopped = stopped_after(editor.as_ref(), "SELECT SLEEP(30); DELETE FROM t").await; + + assert_eq!(gone.end, RunEnd::Cancelled, "{flavour:?}"); + assert_eq!(stopped, RunEnd::Cancelled, "{flavour:?}"); assert_eq!(editor.state(), EditorSessionState::Ready, "{flavour:?}"); - // The next run reads whatever the batch left on the wire first, - // so by now a DELETE that was still coming has run. run_ok(editor.as_ref(), "SELECT 1").await; assert_eq!( count(connection.as_ref(), "SELECT COUNT(*) FROM t").await, @@ -128,3 +136,126 @@ async fn consumer_gone_kills_the_rest_of_the_batch() { connection.close().await.expect("a close"); } } + +/// Stop `sql` half a second in, as the Stop button does. +async fn stopped_after(editor: &dyn EditorSession, sql: &str) -> RunEnd { + let cancel = CancellationToken::new(); + let cancelling = { + let cancel = cancel.clone(); + async move { + tokio::time::sleep(Duration::from_millis(500)).await; + cancel.cancel(); + } + }; + let ((summary, _), ()) = tokio::join!(run_with(editor, sql, RunOptions::new(cancel)), cancelling); + summary.end +} + +/// The login is locked once the session is open, so the KILL's own +/// connection is refused: only the statements left unsent keep the row. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires docker"] +async fn a_stop_whose_kill_is_refused_sends_nothing_after_the_running_statement() { + for flavour in FLAVOURS { + let fixture = start(flavour).await; + let root = connect_root(&fixture).await; + let admin = session(root.as_ref()).await; + let connection = connect(&fixture, false).await; + setup( + connection.as_ref(), + &["CREATE TABLE t (id int PRIMARY KEY)", "INSERT INTO t VALUES (1)"], + ) + .await; + let gone = session(connection.as_ref()).await; + let late = session(connection.as_ref()).await; + run_ok(admin.as_ref(), "ALTER USER 'tablepro'@'%' ACCOUNT LOCK").await; + + let summary = until_gone(gone.as_ref(), RunOptions::new(CancellationToken::new())).await; + let timed_out = run_with( + late.as_ref(), + "SELECT 1 FROM DUAL WHERE SLEEP(3) = 0; DELETE FROM t", + RunOptions::new(CancellationToken::new()).with_deadline(Some(Duration::from_millis(200))), + ) + .await + .0; + + assert_eq!(summary.end, RunEnd::Cancelled, "{flavour:?}"); + assert_eq!( + timed_out.end, + RunEnd::TimedOut { + server_cancelled: false + }, + "{flavour:?}" + ); + assert_eq!(count(root.as_ref(), "SELECT COUNT(*) FROM t").await, 1, "{flavour:?}"); + run_ok(admin.as_ref(), "ALTER USER 'tablepro'@'%' ACCOUNT UNLOCK").await; + for editor in [gone, late, admin] { + editor.close().await; + } + connection.close().await.expect("a close"); + root.close().await.expect("a close"); + } +} + +/// The server closes the KILL's own connection between two stops, as its +/// wait_timeout does to an idle one: the second stop still lands. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires docker"] +async fn a_stop_lands_after_the_server_closed_the_kill_connection() { + for flavour in FLAVOURS { + let fixture = start(flavour).await; + let root = connect_root(&fixture).await; + let admin = session(root.as_ref()).await; + let connection = connect(&fixture, false).await; + let editor = session(connection.as_ref()).await; + let editor_id = session_scalar(editor.as_ref(), "SELECT CONNECTION_ID()").await; + let sleeping = "SELECT 1 FROM DUAL WHERE SLEEP(30) = 0"; + assert_eq!(stopped_after(editor.as_ref(), sleeping).await, RunEnd::Cancelled); + let others = + format!("SELECT ID FROM information_schema.PROCESSLIST WHERE USER = 'tablepro' AND ID <> {editor_id}"); + for row in rows(root.as_ref(), &others).await { + let id = row.first().map(text).expect("an id"); + run_ok(admin.as_ref(), &format!("KILL CONNECTION {id}")).await; + } + + let end = stopped_after(editor.as_ref(), sleeping).await; + + assert_eq!(end, RunEnd::Cancelled, "{flavour:?}"); + assert_eq!(editor.state(), EditorSessionState::Ready, "{flavour:?}"); + assert_eq!(session_scalar(editor.as_ref(), "SELECT 7").await, "7", "{flavour:?}"); + for editor in [editor, admin] { + editor.close().await; + } + connection.close().await.expect("a close"); + root.close().await.expect("a close"); + } +} + +/// A run's first statement is the row limit's SET, and a server that has +/// stopped answering must not keep the run past its deadline there. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires docker"] +async fn a_deadline_reaches_a_run_still_setting_its_row_limit() { + let fixture = start(MysqlFlavour::Mysql).await; + let connection = connect(&fixture, false).await; + let editor = session(connection.as_ref()).await; + run_ok(editor.as_ref(), "SELECT 1").await; + fixture.container().pause().await.expect("a paused server"); + let options = RunOptions::new(CancellationToken::new()) + .with_row_limit(RowLimit::new(5).expect("a limit")) + .with_deadline(Some(Duration::from_millis(300))); + + let returned = tokio::time::timeout(Duration::from_secs(10), run_with(editor.as_ref(), "SELECT 1", options)).await; + fixture.container().unpause().await.expect("a running server"); + + let (summary, _) = returned.expect("the run returned while the server was paused"); + assert_eq!( + summary.end, + RunEnd::TimedOut { + server_cancelled: false + } + ); + assert_eq!(editor.state(), EditorSessionState::Lost); + editor.close().await; + connection.close().await.expect("a close"); +} diff --git a/linux/crates/drivers/sqlite/src/connection.rs b/linux/crates/drivers/sqlite/src/connection.rs index 88e7ab08f1..dacd00a17f 100644 --- a/linux/crates/drivers/sqlite/src/connection.rs +++ b/linux/crates/drivers/sqlite/src/connection.rs @@ -392,7 +392,7 @@ impl Connection for SqliteConnection { run_bound(held.connection(), statement, sink.as_mut()).await } ReadStatement::User(sql) => { - run_script(held.connection(), sql.as_str(), None, sink.as_mut()).await + run_script(held.connection(), sql.as_str(), None, sink.as_mut(), || false).await } }, Err(error) => Err(error), diff --git a/linux/crates/drivers/sqlite/src/editor_session.rs b/linux/crates/drivers/sqlite/src/editor_session.rs index 5fb3cb2c12..8d40cde552 100644 --- a/linux/crates/drivers/sqlite/src/editor_session.rs +++ b/linux/crates/drivers/sqlite/src/editor_session.rs @@ -170,16 +170,27 @@ impl Shared { .deadline .or(self.statement_timeout) .map(|limit| Instant::now() + limit); - let running = run_script(connection, sql.as_str(), Some(options.row_limit), sink); + // Read before each statement, since the select below sees a stop + // only once the run waits. + let halt = || stop_due(options, flag, deadline).is_some(); + let running = run_script(connection, sql.as_str(), Some(options.row_limit), sink, halt); tokio::pin!(running); let stop = tokio::select! { - outcome = &mut running => return finished(outcome), + outcome = &mut running => { + return match outcome { + Ok(outcome) if outcome.halted => { + Ok(halted_end(stop_due(options, flag, deadline).unwrap_or(Stop::Cancelled))) + } + outcome => finished(outcome), + }; + } () = options.cancel.cancelled() => Stop::Cancelled, () = deadline::elapsed(deadline) => Stop::DeadlineElapsed, }; flag.raise(); let settled = tokio::time::timeout(self.cancel_grace, &mut running).await; Ok(match (stop, settled) { + (_, Ok(Ok(outcome))) if outcome.halted => halted_end(stop), (_, Ok(Ok(outcome))) => completed(outcome), (Stop::Cancelled, Ok(Err(error))) if flag.stopped(&error) => RunEnd::Cancelled, (Stop::DeadlineElapsed, Ok(Err(error))) if flag.stopped(&error) => { @@ -216,6 +227,28 @@ fn finished(outcome: Result) -> Result) -> Option { + if deadline.is_some_and(|at| Instant::now() >= at) && !options.cancel.is_cancelled() { + Some(Stop::DeadlineElapsed) + } else if options.cancel.is_cancelled() || flag.is_raised() { + Some(Stop::Cancelled) + } else { + None + } +} + +/// The statement before the stop ended on its own, so SQLite stopped +/// nothing. +fn halted_end(stop: Stop) -> RunEnd { + match stop { + Stop::Cancelled => RunEnd::Cancelled, + Stop::DeadlineElapsed => RunEnd::TimedOut { + server_cancelled: false, + }, + } +} + fn completed(outcome: StreamOutcome) -> RunEnd { match outcome.consumer_gone { true => RunEnd::Cancelled, diff --git a/linux/crates/drivers/sqlite/src/result_stream.rs b/linux/crates/drivers/sqlite/src/result_stream.rs index 68fedd0aa5..05e0db441f 100644 --- a/linux/crates/drivers/sqlite/src/result_stream.rs +++ b/linux/crates/drivers/sqlite/src/result_stream.rs @@ -35,6 +35,8 @@ pub(crate) struct StreamOutcome { pub rows: u64, /// The sink stopped reading, so the rest of the run was skipped. pub consumer_gone: bool, + /// A stop came between two statements, so the rest was not run. + pub halted: bool, result_sets: usize, } @@ -75,16 +77,23 @@ pub(crate) async fn count(connection: &mut SqlxConnection, statement: &Statement } } -/// Run SQL the user wrote, one statement after another, until one fails -/// or the sink stops reading. +/// Run SQL the user wrote, one statement after another, until one fails, +/// the sink stops reading or `halt` says a stop came. The interrupt +/// reaches a statement only every thousand instructions, so a short one +/// would otherwise run after the stop. pub(crate) async fn run_script( connection: &mut SqlxConnection, sql: &str, limit: Option, sink: &mut dyn ResultSink, + halt: impl Fn() -> bool, ) -> Result { let mut outcome = StreamOutcome::default(); for statement in split_statements(sql) { + if halt() { + outcome.halted = true; + break; + } run_statement(connection, statement, &[], limit, sink, &mut outcome).await?; if outcome.consumer_gone { break; diff --git a/linux/crates/drivers/sqlite/tests/integration/sessions.rs b/linux/crates/drivers/sqlite/tests/integration/sessions.rs index 9348a2ec9b..97467a25eb 100644 --- a/linux/crates/drivers/sqlite/tests/integration/sessions.rs +++ b/linux/crates/drivers/sqlite/tests/integration/sessions.rs @@ -10,7 +10,7 @@ use tablepro_core::{ use tokio::time::Instant; use tokio_util::sync::CancellationToken; -use crate::support::{TestFile, apply, batch, call, connect, query, run, run_with, statement}; +use crate::support::{Collected, TestFile, apply, batch, call, connect, query, run, run_with, statement}; const UNBOUNDED: &str = "WITH RECURSIVE r(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM r) SELECT count(*) FROM r"; @@ -122,6 +122,42 @@ async fn recursive_cte_cancelled() { connection.close().await.unwrap(); } +/// The interrupt reaches a statement only every thousand instructions, +/// and the DELETE needs fewer, so only not starting it keeps the row. +#[tokio::test] +async fn a_stop_during_one_statement_runs_nothing_after_it() { + let file = TestFile::new().await; + let connection = file.open().await; + let session = connection + .open_editor_session(EditorSessionOptions::default(), CancellationToken::new()) + .await + .unwrap(); + let (created, _) = run( + session.as_ref(), + "CREATE TABLE t (id INTEGER PRIMARY KEY); INSERT INTO t VALUES (1)", + ) + .await; + assert_eq!(created.end, RunEnd::Completed); + let cancel = CancellationToken::new(); + let collected = Collected::default(); + + let summary = session + .run( + UserSql::new("SELECT 1; DELETE FROM t"), + RunOptions::new(cancel.clone()), + collected.cancelling(cancel), + ) + .await + .unwrap(); + + assert_eq!(summary.end, RunEnd::Cancelled); + assert_eq!(session.state(), EditorSessionState::Ready); + let (_, counted) = run(session.as_ref(), "SELECT count(*) FROM t").await; + assert_eq!(counted.rows(), vec![vec![Value::Int(1)]]); + session.close().await; + connection.close().await.unwrap(); +} + #[tokio::test] async fn a_run_past_its_deadline_times_out_on_the_server() { let file = TestFile::new().await; diff --git a/linux/crates/drivers/sqlite/tests/integration/support.rs b/linux/crates/drivers/sqlite/tests/integration/support.rs index ccc33401d3..66b5d9ddee 100644 --- a/linux/crates/drivers/sqlite/tests/integration/support.rs +++ b/linux/crates/drivers/sqlite/tests/integration/support.rs @@ -106,7 +106,13 @@ pub struct Collected(Arc>>); impl Collected { pub fn sink(&self) -> Box { - Box::new(CollectingSink(self.clone())) + Box::new(CollectingSink(self.clone(), None)) + } + + /// Cancels `cancel` when the first result set starts, as a Stop + /// pressed while the first rows come in. + pub fn cancelling(&self, cancel: CancellationToken) -> Box { + Box::new(CollectingSink(self.clone(), Some(cancel))) } pub fn events(&self) -> Vec { @@ -125,11 +131,14 @@ impl Collected { } } -struct CollectingSink(Collected); +struct CollectingSink(Collected, Option); #[async_trait] impl ResultSink for CollectingSink { async fn accept(&mut self, event: ResultEvent) -> SinkFlow { + if let (ResultEvent::ResultSetStarted { .. }, Some(cancel)) = (&event, &self.1) { + cancel.cancel(); + } self.0.0.lock().unwrap_or_else(PoisonError::into_inner).push(event); SinkFlow::Continue } diff --git a/linux/docs/adding-drivers.md b/linux/docs/adding-drivers.md index 9dcc43b481..4234ccf5e6 100644 --- a/linux/docs/adding-drivers.md +++ b/linux/docs/adding-drivers.md @@ -278,6 +278,7 @@ An editor tab holds one session for as long as it is open, because the state a u - `run(sql, options, sink)` takes `UserSql` as typed and `RunOptions`: `row_limit`, `deadline` and `cancel`. A caller stops a run through `RunOptions::cancel`. - A dropped caller is a cancel. The run goes on a task of its own, or an actor's, and `run` arms `options.cancel.clone().drop_guard()` until the outcome arrives, so the stop still goes out and the session settles to `Ready` or `Lost`. `MysqlEditorSession::run` is the shape to copy. - The app's editor splits a script with the engine's grammar and sends one statement per run, or one `GO` batch on SQL Server. A driver still runs whatever text it is given, a whole script included. +- A stopped run sends nothing more, whether or not the stop reached the server: the statement running, or the batch the server runs as one (a `GO` batch on SQL Server, text MySQL has to send whole), is its last. `TimedOut { server_cancelled: true }` means the server said it stopped the statement, never only that the stop went out. - A statement the server refused ends the run with `RunEnd::Failed` in the `RunSummary`; an `Err` from `run` means the session itself failed. - `RunSummary` carries `elapsed`, the `RunEnd` (`Completed`, `Failed`, `Cancelled`, `TimedOut { server_cancelled }`) and the `TransactionState` after the run. `Unknown` is an answer of its own: the tab's close guard warns instead of guessing. - `close` stops a running statement, then closes the connection within `close_timeout`. diff --git a/linux/docs/drivers/mysql.md b/linux/docs/drivers/mysql.md index f1b2c3d5f8..6f27e22385 100644 --- a/linux/docs/drivers/mysql.md +++ b/linux/docs/drivers/mysql.md @@ -73,7 +73,7 @@ A save or a schema change runs in the server's own zone, `@@GLOBAL.time_zone`, a ## Stops and timeouts -A stop is `KILL QUERY ` on the control connection, which reaches the server over the same route as the connection it stops. The id comes from `CONNECTION_ID()` at login. It names the connection, not the statement: `StopDelivery::OutOfBand(StopScope::Connection)`. A pool connection goes back to the pool only when the server confirmed the stop (the statement ended with 1317, or with the server's own timeout after the `KILL` landed) and the `KILL` itself finished; a call stopped between statements also gives up its connection, since a `KILL` may be on its way to it. Error 1094 (no such thread) counts as delivered: there was nothing left to stop. +A stop is `KILL QUERY ` on the control connection, which reaches the server over the same route as the connection it stops. The id comes from `CONNECTION_ID()` at login. It names the connection, not the statement: `StopDelivery::OutOfBand(StopScope::Connection)`. A pool connection goes back to the pool only when the server confirmed the stop (the statement ended with 1317, or with the server's own timeout after the `KILL` landed) and the `KILL` itself finished; a call stopped between statements also gives up its connection, since a `KILL` may be on its way to it. Error 1094 (no such thread) counts as delivered: there was nothing left to stop. A control connection the server closed since the last stop, at its `wait_timeout` or a restart, is replaced and the `KILL` sent again. | Server answer | Reported as | |---|---| @@ -118,9 +118,11 @@ Whether a write can be undone depends on the table's storage engine, not on the ## Editor sessions -- The editor sends one statement per run. Text with several statements runs as the server's own multi-statement batch over the text protocol, except that mysql-client `DELIMITER` lines split it where the client would, because the server does not know the directive. +- The editor sends one statement per run. Text with several statements is split where the mysql client splits it, at each `;` or at the delimiter a `DELIMITER` line sets, and goes one statement per round trip. Sent whole, the server starts each next statement as soon as the last one's result is written, read or not, so a stop could not keep it from running. A `/*!` comment is code there, so it is sent. +- Text without `DELIMITER` lines goes as written, as the server's own batch, when it holds one statement, or when only the server can tell where its statements end: a statement that does not open with a word known to start one with no body, such as a stored program or a compound statement, a `/*!` comment, or text the session's `sql_mode` reads its own way (a backslash in a string, `[name]`, `$$` quotes). - The row limit is `SET SESSION sql_select_limit = `, sent when the limit changes. The server stops a plain `SELECT` there, and the set ends with `Truncation::ServerStopped` and no total. A `SELECT` with a `LIMIT` of its own, or one inside a routine, is read to its end: the rows past the limit are counted, never decoded, and the set ends with `Truncation::ReadPastLimit` and its total. A set that ends exactly one row past the limit reads as `ServerStopped` either way. -- A stop sends `KILL QUERY` on the session's own control connection and gives the statement the cancel grace to end. A statement that has not ended by then ends the session, because its answer is still on the way. -- When the grid stops reading, the run sends the same `KILL QUERY` and ends as cancelled. The batch already sent is read to its end, unshown, and no later batch is sent. +- A stop sends `KILL QUERY` on the session's own control connection and gives the statement the cancel grace to end, and nothing after it is sent, whatever the `KILL` did. A statement that has not ended by then ends the session, because its answer is still on the way, and so does a `KILL` that has not answered, because it may still land on whatever runs next. The row limit's `SET` is watched the same way. +- A run stopped at its deadline reads as `TimedOut { server_cancelled: true }` only when the statement ended with 1317 or the server's own timeout. A bare `SELECT SLEEP(n)` the `KILL` stopped returns 1 with no error, so it reads as `false`. +- When the grid stops reading, the run sends the same `KILL QUERY` and ends as cancelled. The statement already sent is read to its end, unshown, and nothing after it is sent. - sqlx describes a result column by its type alone, with no fractional digits and no BIT length. A run shows a DATETIME(6) holding `12:00:00.5` as `12:00:00.5` and a BIT(1) in the whole byte it arrived in, `00000001`. A table tab and its export take both from the catalogue and show `12:00:00.500000` and `1`. - The transaction state after each run comes from `@@in_transaction` on MariaDB and from `performance_schema.events_transactions_current` on MySQL. On MySQL before 8.0.16, with `performance_schema` off, or for a login that cannot read it, the state is `Unknown`, and the tab's close guard warns instead of guessing. diff --git a/linux/docs/drivers/sql-server.md b/linux/docs/drivers/sql-server.md index aaa04ca6b8..38804753ef 100644 --- a/linux/docs/drivers/sql-server.md +++ b/linux/docs/drivers/sql-server.md @@ -70,6 +70,8 @@ tiberius borrows its client for the whole of a request, so nothing outside the t 2. sends the attention and waits up to half of the grace for the server's acknowledgement; 3. on a workspace connection, rolls back any transaction the request left open (`IF @@TRANCOUNT > 0 ROLLBACK TRANSACTION`), because an attention does not end one. An editor session keeps the user's transaction. +A request that ends on its own within that quarter gets no attention, and reads as `Cancelled`, or `Timeout { server_cancelled: false }` when the deadline asked for it, since the server stopped nothing. + The connection is reused only when the attention was acknowledged at a token boundary. One stopped in the middle of a token, or whose attention got no acknowledgement or something else back, is closed. During a save the attention is never sent while `COMMIT TRANSACTION` is in flight. tiberius 0.13 stops reading at the first end of message, so an attention sent while SQL Server is still waiting on the request, as in a `WAITFOR`, is never acknowledged, and the connection is closed. A pooled connection is replaced; an editor session ends and the tab opens a new one on its next run. The fix is tiberius-rs/tiberius#459. Until a release has it, `.config/nextest.toml` holds the three tests that need it out of the default run: diff --git a/linux/docs/drivers/sqlite.md b/linux/docs/drivers/sqlite.md index c9c8422281..7f6fc6620b 100644 --- a/linux/docs/drivers/sqlite.md +++ b/linux/docs/drivers/sqlite.md @@ -111,5 +111,5 @@ SQLite's pragmas do not report a foreign key's constraint name, so the definitio - A script is split where SQLite itself would end a statement, so the semicolons inside a trigger body do not end it. Text the splitter cannot finish, such as an unterminated string, goes to SQLite whole, so the error is SQLite's own. The splitter is tested against `sqlite3_complete` (`tests/script_plan_matches_sqlite3_complete.rs`). - Each statement is prepared first, so a `SELECT` that finds nothing still has its columns. A column with no declared type gets its type from the first row, sent as `ResultEvent::ColumnsResolved`. - The row limit stops the read one row past the limit, and the set ends with `Truncation::ReadPastLimit`. SQLite computes a row only when it is stepped, so the rest of the result is never produced. -- A stop raises the run's flag and gives the statement the cancel grace to notice it; past that its read is dropped. +- A stop raises the run's flag and gives the statement the cancel grace to notice it; past that its read is dropped. The flag reaches a statement only every 1,000 instructions, so the run also reads it before each statement and starts none after a stop. A run stopped between statements reads as `Cancelled`, or `TimedOut { server_cancelled: false }` at its deadline. - The transaction state after a run comes from `sqlite3_get_autocommit`. From 125f6c0898103864b0448195e8d02757e9ae5489 Mon Sep 17 00:00:00 2001 From: Ngo Quoc Dat Date: Thu, 22 Oct 2026 17:03:46 +0700 Subject: [PATCH 2/2] docs(linux): changelog for #3780 --- linux/changelog.d/3780.fixed.1.md | 1 + linux/changelog.d/3780.fixed.2.md | 1 + linux/changelog.d/3780.fixed.md | 1 + 3 files changed, 3 insertions(+) create mode 100644 linux/changelog.d/3780.fixed.1.md create mode 100644 linux/changelog.d/3780.fixed.2.md create mode 100644 linux/changelog.d/3780.fixed.md diff --git a/linux/changelog.d/3780.fixed.1.md b/linux/changelog.d/3780.fixed.1.md new file mode 100644 index 0000000000..4d8e936bcd --- /dev/null +++ b/linux/changelog.d/3780.fixed.1.md @@ -0,0 +1 @@ +Stopping a MySQL or MariaDB query could fail after the tab sat idle for hours, or while the run was starting. diff --git a/linux/changelog.d/3780.fixed.2.md b/linux/changelog.d/3780.fixed.2.md new file mode 100644 index 0000000000..4a0edf09cd --- /dev/null +++ b/linux/changelog.d/3780.fixed.2.md @@ -0,0 +1 @@ +A MySQL, MariaDB or SQL Server query that finished just as its time limit passed said the server had stopped it. diff --git a/linux/changelog.d/3780.fixed.md b/linux/changelog.d/3780.fixed.md new file mode 100644 index 0000000000..0db2bf18ff --- /dev/null +++ b/linux/changelog.d/3780.fixed.md @@ -0,0 +1 @@ +Stopping a MySQL or MariaDB query could still run the statements after it in the same block.