siderite_backends/mysql/
native.rs

1//! Opt-in native MySQL adapter (feature `mysql-native`).
2//!
3//! Completed ORM statements retain clean prepared-statement sessions.
4//! Interrupted exchanges and unfinished transactions retire their socket.
5//! Raw SQL/scripts and schema connections are retired after use because
6//! their session state and locks cannot be safely inferred. Writes are never
7//! retried after a connection error or an ambiguous commit outcome.
8//!
9//! This is an experimental adapter. Adoption still requires live recovery,
10//! TLS, migration and matched end-to-end performance evidence. The existing
11//! SQLx adapter remains the compatibility default.
12
13mod codec;
14#[cfg(test)]
15mod contract;
16mod lease;
17#[cfg(test)]
18mod options_tests;
19
20use super::io::MySqlIo;
21use super::metadata::TableCache;
22use super::returning::{Prepared, prepare, synthesize, synthesized_result};
23use super::returning::{table_of, write_returning};
24use crate::shared::affected_result;
25use crate::sql::{CompiledQuery, MySql, compile, compile_write};
26use async_trait::async_trait;
27use lease::Lease;
28use mysql_async::{Opts, OptsBuilder, Pool, PoolConstraints, PoolOpts};
29use siderite_orm::{Backend, BackendCapabilities, BackendError, ExecResult};
30use siderite_orm::{Executor, IsolationLevel, OrmError, QueryError, QueryPlan};
31use siderite_orm::{QueryResult, Transaction, Value, WritePlan};
32use std::fmt::{self, Debug, Formatter};
33use std::sync::Arc;
34use std::time::Duration;
35use tokio::sync::{MappedMutexGuard, Mutex, MutexGuard, Semaphore};
36
37type Reply<T> = Result<T, OrmError>;
38type Args = Vec<Value>;
39type Scope = Box<dyn Transaction>;
40type ReadResult = Reply<QueryResult>;
41
42/// Native pool admission and checkout limits.
43#[derive(Debug, Clone)]
44pub struct NativeMySqlOptions {
45    /// Maximum simultaneously held database connections.
46    pub max_connections: usize,
47    /// Maximum admitted callers waiting for a connection.
48    pub max_waiters: usize,
49    /// Deadline for obtaining a configured connection.
50    pub acquire_timeout: Duration,
51}
52
53impl Default for NativeMySqlOptions {
54    fn default() -> Self {
55        Self {
56            max_connections: 10,
57            max_waiters: 100,
58            acquire_timeout: Duration::from_secs(10),
59        }
60    }
61}
62
63/// MySQL ORM adapter using retained clean mysql_async sessions.
64#[derive(Clone)]
65pub struct NativeMySqlBackend {
66    pool: Pool,
67    admission: Arc<Semaphore>,
68    acquire_timeout: Duration,
69    max_connections: usize,
70    tables: Arc<TableCache>,
71}
72
73impl NativeMySqlBackend {
74    /// Connect with the default bounded admission settings.
75    ///
76    /// Args:
77    ///     url: MySQL connection URL, including the database name.
78    ///
79    /// Returns:
80    ///     A connected native backend, or a configuration/connection error.
81    pub async fn connect(url: &str) -> Result<Self, BackendError> {
82        Self::connect_with(url, NativeMySqlOptions::default()).await
83    }
84
85    /// Connect with explicit admission and checkout limits.
86    ///
87    /// Args:
88    ///     url: Native-driver MySQL connection URL.
89    ///     options: Connection count, waiting count and checkout deadline.
90    ///
91    /// Returns:
92    ///     A connected backend, or an invalid configuration/connection error.
93    pub async fn connect_with(
94        url: &str,
95        options: NativeMySqlOptions,
96    ) -> Result<Self, BackendError> {
97        let parsed = Opts::from_url(url).map_err(|_| {
98            let message = "invalid native MySQL URL";
99            connection_error(message)
100        })?;
101        Self::connect_options(parsed, options).await
102    }
103
104    /// Connect with native options, including verified TLS and custom roots.
105    ///
106    /// Args:
107    ///     connection: Driver options; TLS roots and identity policy retained.
108    ///     options: Connection count, admission and checkout deadline.
109    ///
110    /// Returns:
111    ///     Warm backend, or configuration/session/connection failure.
112    ///
113    /// Native pool bounds, session setup, found-rows and statement cache
114    /// override corresponding driver settings to preserve adapter contracts.
115    /// Custom initialization must leave no transaction or lock open.
116    pub async fn connect_options(
117        connection: Opts,
118        options: NativeMySqlOptions,
119    ) -> Result<Self, BackendError> {
120        let capacity = options
121            .max_connections
122            .checked_add(options.max_waiters)
123            .filter(|capacity| *capacity <= Semaphore::MAX_PERMITS)
124            .ok_or_else(|| connection_error("invalid admission capacity"))?;
125        // With zero idle TTL, the minimum is the retention threshold.
126        // Retain every clean connection up to the configured pool bound.
127        let count = options.max_connections;
128        let bounds = PoolConstraints::new(count, count)
129            .filter(|_| options.max_connections > 0)
130            .ok_or_else(|| connection_error("invalid connection count"))?;
131        if options.acquire_timeout.is_zero() {
132            return Err(connection_error("checkout deadline must be positive"));
133        }
134        let mut setup = super::SESSION_SETUP.to_vec();
135        setup.push("SET autocommit = 1");
136        let acquire_timeout = options.acquire_timeout;
137        let connect_options = OptsBuilder::from_opts(connection)
138            .client_found_rows(true)
139            .stmt_cache_size(128)
140            .setup(setup)
141            .pool_opts(
142                PoolOpts::default()
143                    .with_constraints(bounds)
144                    .with_reset_connection(false),
145            );
146        let backend = Self {
147            pool: Pool::new(connect_options),
148            admission: Arc::new(Semaphore::new(capacity)),
149            acquire_timeout,
150            max_connections: count,
151            tables: Arc::default(),
152        };
153        backend.warm().await?;
154        Ok(backend)
155    }
156
157    /// Disconnect the pool after all held connections have been released.
158    ///
159    /// Returns:
160    ///     Success after disconnect, or the native driver error.
161    pub async fn close(&self) -> Result<(), BackendError> {
162        self.pool
163            .clone()
164            .disconnect()
165            .await
166            .map_err(|error| connection_error(error.to_string()))
167    }
168
169    async fn checkout(&self) -> Result<Lease, OrmError> {
170        let permit = Arc::clone(&self.admission)
171            .try_acquire_owned()
172            .map_err(|_| connection_error("native MySQL admission is full"))?;
173        let wait = self.pool.get_conn();
174        let conn = tokio::time::timeout(self.acquire_timeout, wait)
175            .await
176            .map_err(|_| connection_error("native MySQL checkout timed out"))?
177            .map_err(lease::error)?;
178        Ok(Lease::new(conn, permit))
179    }
180
181    /// Initialize all configured pool slots within one checkout deadline.
182    ///
183    /// Returns:
184    ///     Success after every slot is configured, or a checkout/setup error.
185    pub async fn warm(&self) -> Result<(), BackendError> {
186        let work = async {
187            let mut leases = Vec::with_capacity(self.max_connections);
188            for _ in 0..self.max_connections {
189                leases.push(self.checkout().await?);
190            }
191            Ok::<_, OrmError>(leases)
192        };
193        let leases = tokio::time::timeout(self.acquire_timeout, work)
194            .await
195            .map_err(|_| connection_error("native pool warming timed out"))?
196            .map_err(|error| connection_error(error.to_string()))?;
197        drop(leases);
198        Ok(())
199    }
200
201    fn autocommit(&self) -> bool {
202        true
203    }
204}
205
206trait LeaseAccess {
207    fn lease(&mut self) -> &mut Lease;
208}
209
210impl Debug for NativeMySqlBackend {
211    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
212        // mysql_async's Pool Debug includes its URL-derived password.
213        formatter
214            .debug_struct("NativeMySqlBackend")
215            .field("max_connections", &self.max_connections)
216            .field("acquire_timeout", &self.acquire_timeout)
217            .finish_non_exhaustive()
218    }
219}
220
221impl LeaseAccess for Lease {
222    fn lease(&mut self) -> &mut Lease {
223        self
224    }
225}
226
227impl LeaseAccess for MappedMutexGuard<'_, Lease> {
228    fn lease(&mut self) -> &mut Lease {
229        self
230    }
231}
232
233/// Compile before checkout, and share executor behavior for pools/scopes.
234macro_rules! executor {
235    ($target:ty) => {
236        #[async_trait]
237        impl Executor for $target {
238            fn capabilities(&self) -> BackendCapabilities {
239                BackendCapabilities::mysql()
240            }
241
242            async fn fetch(&self, plan: &QueryPlan) -> Reply<QueryResult> {
243                let query = compile(plan, &MySql)?;
244                let mut access = self.checkout().await?;
245                access.lease().fetch(&query.sql, query.params).await
246            }
247
248            async fn execute(&self, plan: &WritePlan) -> Reply<ExecResult> {
249                let prepared = Write::prepare(plan)?;
250                let mut access = self.checkout().await?;
251                write(
252                    access.lease(),
253                    &self.tables,
254                    plan,
255                    prepared,
256                    self.autocommit(),
257                )
258                .await
259            }
260
261            async fn fetch_raw(&self, sql: &str, params: Args) -> ReadResult {
262                self.tables.clear();
263                let mut access = self.checkout().await?;
264                let conn = access.lease();
265                conn.retire = true;
266                conn.fetch(sql, params).await
267            }
268
269            async fn execute_raw(&self, sql: &str, args: Args) -> Reply<u64> {
270                self.tables.clear();
271                let mut access = self.checkout().await?;
272                let conn = access.lease();
273                conn.retire = true;
274                conn.run(sql, args).await.map(|done| done.rows_affected())
275            }
276
277            async fn execute_script(&self, sql: &str) -> Reply<()> {
278                self.tables.clear();
279                let mut access = self.checkout().await?;
280                let conn = access.lease();
281                conn.retire = true;
282                conn.control(sql).await
283            }
284        }
285    };
286}
287
288executor!(NativeMySqlBackend);
289executor!(NativeTransaction);
290
291enum Write {
292    Plain(CompiledQuery),
293    Returning(Prepared),
294}
295
296impl Write {
297    fn prepare(plan: &WritePlan) -> Result<Self, OrmError> {
298        if plan.returning().is_empty() {
299            compile_write(plan, &MySql).map(Self::Plain)
300        } else {
301            prepare(plan).map(Self::Returning)
302        }
303    }
304}
305
306async fn write(
307    conn: &mut Lease,
308    tables: &TableCache,
309    plan: &WritePlan,
310    prepared: Write,
311    autocommit: bool,
312) -> Reply<ExecResult> {
313    let prepared = match prepared {
314        Write::Returning(prepared) => prepared,
315        Write::Plain(query) => {
316            let done = conn.run(&query.sql, query.params).await?;
317            return Ok(affected_result(done.rows_affected()));
318        }
319    };
320    let info = tables.table(conn, table_of(plan)).await?;
321    if autocommit {
322        if let WritePlan::Insert(insert) = plan
323            && let Some(row) = synthesize(&info, insert)
324        {
325            let write = prepared.write;
326            let done = conn.run(&write.sql, write.params).await?;
327            return synthesized_result(&info, row, &done);
328        }
329        begin(conn, None).await?;
330    }
331    let result = write_returning(conn, &info, plan, prepared).await?;
332    if autocommit {
333        conn.control("COMMIT").await?;
334        conn.in_transaction = false;
335    }
336    Ok(result)
337}
338
339#[async_trait]
340impl Backend for NativeMySqlBackend {
341    fn read_parameter_count(&self, plan: &QueryPlan) -> Result<Option<usize>, OrmError> {
342        Ok(Some(compile(plan, &MySql)?.params.len()))
343    }
344
345    async fn begin(&self, isolation: Option<IsolationLevel>) -> Reply<Scope> {
346        if let Some(level) = isolation {
347            self.capabilities()
348                .require(siderite_orm::Feature::Isolation(level))?;
349        }
350        let mut conn = self.checkout().await?;
351        begin(&mut conn, isolation).await?;
352        Ok(Box::new(NativeTransaction {
353            slot: Mutex::new(Some(conn)),
354            tables: Arc::clone(&self.tables),
355            transactional: true,
356        }))
357    }
358
359    async fn begin_schema(&self, transactional: bool) -> Reply<Scope> {
360        if transactional {
361            return self.begin(None).await;
362        }
363        let mut conn = self.checkout().await?;
364        conn.retire = true;
365        Ok(Box::new(NativeTransaction {
366            slot: Mutex::new(Some(conn)),
367            tables: Arc::clone(&self.tables),
368            transactional: false,
369        }))
370    }
371}
372
373async fn begin(conn: &mut Lease, level: Option<IsolationLevel>) -> Reply<()> {
374    conn.in_transaction = true;
375    if let Some(level) = level {
376        let sql = format!(
377            "SET TRANSACTION ISOLATION LEVEL {}",
378            super::isolation_sql(level)
379        );
380        conn.control(&sql).await?;
381    }
382    conn.control("START TRANSACTION").await
383}
384
385struct NativeTransaction {
386    slot: Mutex<Option<Lease>>,
387    tables: Arc<TableCache>,
388    transactional: bool,
389}
390
391impl NativeTransaction {
392    async fn checkout(&self) -> Result<MappedMutexGuard<'_, Lease>, OrmError> {
393        MutexGuard::try_map(self.slot.lock().await, Option::as_mut)
394            .map_err(|_| QueryError::TransactionClosed.into())
395    }
396
397    fn autocommit(&self) -> bool {
398        false
399    }
400
401    async fn finish(&self, commit: bool) -> Reply<()> {
402        let mut conn = self
403            .slot
404            .lock()
405            .await
406            .take()
407            .ok_or(QueryError::TransactionClosed)?;
408        if self.transactional {
409            let sql = if commit { "COMMIT" } else { "ROLLBACK" };
410            conn.control(sql).await?;
411            conn.in_transaction = false;
412        }
413        Ok(())
414    }
415}
416
417#[async_trait]
418impl Transaction for NativeTransaction {
419    async fn commit(&self) -> Reply<()> {
420        self.finish(true).await
421    }
422
423    async fn rollback(&self) -> Reply<()> {
424        self.finish(false).await
425    }
426}
427
428fn connection_error(message: impl Into<String>) -> BackendError {
429    BackendError::Connection(message.into())
430}
siderite_backends/mysql/native.rs

siderite_backends/mysql/
native.rs
