fix(linux-drivers): stop an editor script before its next statement on MySQL, MariaDB and SQLite

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.

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.

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.

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)
     }

linux/crates/drivers/mssql/src/in_band_request.rs

@@ -252,25 +252,33 @@ 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;
             stop.record(match answer {
                 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<Answer>) -> 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());

linux/crates/drivers/mysql/src/cancel.rs

@@ -37,31 +37,26 @@ 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
                     .connect_physical(SessionKind::Control, &no_limits(), &CancellationToken::new())
                     .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 {

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<RunOutcome, DriverError> {
     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"]);
-    }
 }