sequel_mcp/sql/
cancel.rs

1//! Server-side statement cancellation (D2).
2//!
3//! `MAX_EXECUTION_TIME`/`max_statement_time` only cover SELECTs and are
4//! not guaranteed immediate, so this module provides active cancellation:
5//! when a deadline expires, `KILL QUERY <id>` is issued from a separate
6//! same-user control connection against the executing connection's id
7//! (captured via `CONNECTION_ID()` before the statement runs). The
8//! executing connection is only reused after its transaction state is
9//! verified clean; anything inconclusive discards it. A cancelled
10//! mutation is never retried automatically.
11
12use mysql_async::Pool;
13use mysql_async::prelude::Queryable;
14use thiserror::Error;
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum CancelOutcome {
18    /// Statement finished before the deadline (no cancellation).
19    Completed,
20    /// Deadline fired; KILL QUERY succeeded; the executing future
21    /// resolved; the connection verified clean and stays in the pool.
22    Interrupted,
23    /// Deadline fired and the outcome could not be established — the
24    /// executing connection was discarded. The caller must surface an
25    /// uncertain result and must not retry.
26    Uncertain,
27}
28
29#[derive(Debug, Error)]
30pub enum CancellationError {
31    #[error("statement deadline exceeded; interrupted via KILL QUERY ({0}ms)")]
32    Interrupted(u64),
33    #[error(
34        "statement deadline exceeded; cancellation inconclusive — connection discarded ({0}ms)"
35    )]
36    Uncertain(u64),
37}
38
39/// Issue `KILL QUERY <connection_id>` on a separate connection from the
40/// same pool (same user may kill its own threads).
41pub async fn kill_query(pool: &Pool, connection_id: u64) -> Result<(), String> {
42    let mut control = pool
43        .get_conn()
44        .await
45        .map_err(|e| format!("control connection failed: {e}"))?;
46    // The identifier originates from the server (CONNECTION_ID), never
47    // user input, and is a u64 — formatting cannot inject. KILL is not
48    // universally preparable, so it is sent as literal SQL.
49    let sql = format!("KILL QUERY {connection_id}");
50    control
51        .query_drop(sql.as_str())
52        .await
53        .map_err(|e| format!("KILL QUERY failed: {e}"))
54}
55
56/// Verify a post-cancellation connection is clean: no open transaction,
57/// autocommit restored, and it answers a health query. `true` = reusable.
58pub async fn verify_connection_clean(conn: &mut mysql_async::Conn) -> Result<bool, String> {
59    let row: Option<(i8, i8)> = conn
60        .exec_first::<(i8, i8), _, _>("SELECT @@autocommit, @@in_transaction", ())
61        .await
62        .map_err(|e| format!("verify query failed: {e}"))?;
63    match row {
64        // in_transaction must be 0; autocommit is informational (we set
65        // session state per operation anyway).
66        Some((_, in_txn)) => Ok(in_txn == 0),
67        None => Ok(false),
68    }
69}
sequel_mcp/sql/cancel.rs

sequel_mcp/sql/
cancel.rs
