runlight_sqlx/
mysql.rs

1//! MySQL 8.4 and MariaDB 11.4, on the app's pool. Each statement runs on the pool on its own; a
2//! transaction or a lock holds one connection until it ends.
3
4use std::sync::Arc;
5
6use runlight::BoxFuture;
7use runlight::store::{Db, DbError, Dialect, Hold, Param, Row, fill_placeholders, mysql_literal};
8use sqlx::pool::PoolConnection;
9use sqlx::{AssertSqlSafe, MySql, MySqlPool};
10use tokio::sync::Mutex;
11
12use crate::rows::{db_error, mysql_row};
13
14/// One lock per database, so installs sharing a server do not wait on each other. Lock names are 64
15/// characters at most.
16const LOCK: &str = "CONCAT('runlight:', LEFT(SHA2(COALESCE(DATABASE(), ''), 256), 48))";
17
18/// The store's driver over a MySQL pool. SQL written for SQLite and Postgres is sent as MySQL reads it:
19/// a "quoted" name in backticks, a backslash inside 'text' doubled, and each value written into the
20/// statement, escaped, as mysql2 fills them in.
21#[derive(Clone)]
22pub struct MysqlDb {
23    pool: MySqlPool,
24}
25
26impl MysqlDb {
27    /// The store's driver over a pool.
28    pub fn new(pool: MySqlPool) -> MysqlDb {
29        MysqlDb { pool }
30    }
31}
32
33/// The session's sql_mode as it was, less NO_BACKSLASH_ESCAPES, wherever it sits in the list.
34const NO_BACKSLASH_ESCAPES_OFF: &str = "SET SESSION sql_mode = TRIM(BOTH ',' FROM REPLACE(CONCAT(',', @@SESSION.sql_mode, ','), ',NO_BACKSLASH_ESCAPES,', ','))";
35
36/// A statement with its values written in. Values are escaped with backslashes, which a server set to
37/// NO_BACKSLASH_ESCAPES reads as text, so a quote could end the value early: the statement goes after
38/// one that drops that mode from the connection, an app's own pool's too, in the same round trip.
39fn text(sql: &str, params: &[Param]) -> Result<AssertSqlSafe<String>, DbError> {
40    Ok(AssertSqlSafe(format!("{NO_BACKSLASH_ESCAPES_OFF}; {}", fill_placeholders(sql, params, true, mysql_literal)?)))
41}
42
43async fn exec(conn: &mut PoolConnection<MySql>, sql: &str) -> Result<Vec<Row>, DbError> {
44    let rows = sqlx::raw_sql(AssertSqlSafe(sql.to_string())).fetch_all(&mut **conn).await.map_err(db_error)?;
45    rows.iter().map(mysql_row).collect()
46}
47
48impl Db for MysqlDb {
49    fn dialect(&self) -> Dialect {
50        Dialect::Mysql
51    }
52
53    fn all<'a>(&'a self, sql: &'a str, params: Vec<Param>) -> BoxFuture<'a, Result<Vec<Row>, DbError>> {
54        Box::pin(async move {
55            let rows = sqlx::raw_sql(text(sql, &params)?).fetch_all(&self.pool).await.map_err(db_error)?;
56            rows.iter().map(mysql_row).collect()
57        })
58    }
59
60    fn run<'a>(&'a self, sql: &'a str, params: Vec<Param>) -> BoxFuture<'a, Result<u64, DbError>> {
61        Box::pin(async move {
62            Ok(sqlx::raw_sql(text(sql, &params)?).execute(&self.pool).await.map_err(db_error)?.rows_affected())
63        })
64    }
65
66    fn hold(&self, kind: Hold) -> BoxFuture<'_, Result<Arc<dyn Db>, DbError>> {
67        Box::pin(async move {
68            // In its slot from the start, so a request cancelled part way closes the connection rather than
69            // give it back to the pool in a transaction or with the lock taken.
70            let mut slot = Slot { conn: Some(self.pool.acquire().await.map_err(db_error)?), nested: false };
71            let conn = slot.conn.as_mut().ok_or_else(ended)?;
72            match kind {
73                Hold::Transaction => begin(conn).await?,
74                Hold::Exclusive => loop {
75                    let rows = exec(conn, &format!("SELECT GET_LOCK({LOCK}, 5) AS ok")).await?;
76                    let row = rows.into_iter().next().unwrap_or_default();
77                    if row.is_null("ok") {
78                        return Err(DbError("Runlight: MySQL refused the lock for creating tables".into()));
79                    }
80                    if row.num("ok") == 1.0 {
81                        break;
82                    }
83                    // Not got within 5 seconds: another process is creating the tables. Ask again.
84                },
85            }
86            Ok(Arc::new(Held { conn: Arc::new(Mutex::new(slot)), kind, role: Role::Owner }) as Arc<dyn Db>)
87        })
88    }
89}
90
91/// As Postgres does by default: each statement sees what was committed before it began, and InnoDB
92/// takes no gap locks, so two writers to neighbouring rows do not deadlock.
93async fn begin(conn: &mut PoolConnection<MySql>) -> Result<(), DbError> {
94    exec(conn, "SET TRANSACTION ISOLATION LEVEL READ COMMITTED").await?;
95    exec(conn, "START TRANSACTION").await?;
96    Ok(())
97}
98
99/// The held connection. Let go of with it still in (a request cancelled before `finish`), it closes the
100/// connection, so no transaction or lock goes back to the pool.
101struct Slot {
102    conn: Option<PoolConnection<MySql>>,
103    /// A transaction begun inside the lock that has not ended.
104    nested: bool,
105}
106
107impl Slot {
108    fn close(&mut self) {
109        if let Some(conn) = self.conn.take() {
110            drop(conn.detach());
111        }
112    }
113}
114
115impl Drop for Slot {
116    fn drop(&mut self) {
117        self.close();
118    }
119}
120
121/// One connection held for a transaction or a lock.
122struct Held {
123    conn: Arc<Mutex<Slot>>,
124    kind: Hold,
125    role: Role,
126}
127
128impl Drop for Held {
129    fn drop(&mut self) {
130        // Its owner gone without `finish`: closed now rather than when the last joined handle goes.
131        if self.role == Role::Owner
132            && let Ok(mut slot) = self.conn.try_lock()
133        {
134            slot.close();
135        }
136    }
137}
138
139/// What finishing a held connection does.
140#[derive(Clone, Copy, PartialEq, Eq)]
141enum Role {
142    /// Ends the transaction or lets the lock go, and gives the connection back.
143    Owner,
144    /// A transaction begun inside a lock: ends the transaction, and keeps the connection.
145    Nested,
146    /// Joined what was already held: ends nothing.
147    Joined,
148}
149
150fn ended() -> DbError {
151    DbError("Runlight: the held connection has been let go".into())
152}
153
154impl Db for Held {
155    fn dialect(&self) -> Dialect {
156        Dialect::Mysql
157    }
158
159    fn all<'a>(&'a self, sql: &'a str, params: Vec<Param>) -> BoxFuture<'a, Result<Vec<Row>, DbError>> {
160        Box::pin(async move {
161            let mut slot = self.conn.lock().await;
162            let conn = slot.conn.as_mut().ok_or_else(ended)?;
163            let rows = sqlx::raw_sql(text(sql, &params)?).fetch_all(&mut **conn).await.map_err(db_error)?;
164            rows.iter().map(mysql_row).collect()
165        })
166    }
167
168    fn run<'a>(&'a self, sql: &'a str, params: Vec<Param>) -> BoxFuture<'a, Result<u64, DbError>> {
169        Box::pin(async move {
170            let mut slot = self.conn.lock().await;
171            let conn = slot.conn.as_mut().ok_or_else(ended)?;
172            Ok(sqlx::raw_sql(text(sql, &params)?).execute(&mut **conn).await.map_err(db_error)?.rows_affected())
173        })
174    }
175
176    fn hold(&self, kind: Hold) -> BoxFuture<'_, Result<Arc<dyn Db>, DbError>> {
177        Box::pin(async move {
178            if kind == Hold::Transaction && self.kind == Hold::Exclusive {
179                let mut slot = self.conn.lock().await;
180                let slot = &mut *slot;
181                let conn = slot.conn.as_mut().ok_or_else(ended)?;
182                // Marked before it is sent: a transaction cancelled part way may still have begun.
183                slot.nested = true;
184                begin(conn).await?;
185                return Ok(Arc::new(Held { conn: self.conn.clone(), kind, role: Role::Nested }) as Arc<dyn Db>);
186            }
187            Ok(Arc::new(Held { conn: self.conn.clone(), kind: self.kind, role: Role::Joined }) as Arc<dyn Db>)
188        })
189    }
190
191    fn finish(&self, ok: bool) -> BoxFuture<'_, Result<(), DbError>> {
192        Box::pin(async move {
193            let mut slot = self.conn.lock().await;
194            match self.role {
195                Role::Joined => Ok(()),
196                Role::Nested => {
197                    let slot = &mut *slot;
198                    let Some(conn) = slot.conn.as_mut() else { return Ok(()) };
199                    exec(conn, if ok { "COMMIT" } else { "ROLLBACK" }).await?;
200                    slot.nested = false;
201                    Ok(())
202                }
203                // A transaction inside the lock that never ended: closing the connection rolls it back and
204                // lets the lock go.
205                Role::Owner if slot.nested => {
206                    slot.close();
207                    Ok(())
208                }
209                Role::Owner => {
210                    let Some(mut conn) = slot.conn.take() else { return Ok(()) };
211                    let result = match self.kind {
212                        Hold::Transaction => exec(&mut conn, if ok { "COMMIT" } else { "ROLLBACK" }).await.map(|_| ()),
213                        // A connection whose release failed may still have the lock, so it is closed below,
214                        // and a lost connection ends its session, and the lock with it.
215                        Hold::Exclusive => exec(&mut conn, &format!("DO RELEASE_LOCK({LOCK})")).await.map(|_| ()),
216                    };
217                    if result.is_err() {
218                        // A connection still in its transaction or holding the lock must never go back to
219                        // the pool.
220                        drop(conn.detach());
221                    }
222                    if self.kind == Hold::Exclusive { Ok(()) } else { result }
223                }
224            }
225        })
226    }
227}
runlight_sqlx/mysql.rs

runlight_sqlx/
mysql.rs
