aster_forge_db/
connection.rs

1//! Database connection helpers and `SQLite` reader/writer pool setup.
2//!
3//! The module wraps `SeaORM` connection creation with retry behavior, optional metrics callbacks, and
4//! SQLite-specific pooling rules. `SQLite` writer connections are constrained so transaction
5//! serialization is explicit, while read-only handles can still be split out when the URL supports
6//! it.
7
8use crate::{DbError, Result, retry};
9use sea_orm::{ConnectOptions, ConnectionTrait, Database, DatabaseConnection, SqlxSqliteConnector};
10use std::sync::Arc;
11
12use aster_forge_metrics::{
13    DbMetricBackend, DbQueryKind, DbQueryMetric, NoopDbMetrics, SharedDbMetricsRecorder,
14};
15
16/// Database connection URL input.
17#[derive(Clone, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
18#[serde(untagged)]
19pub enum DatabaseUrl {
20    /// A complete database URL. Existing percent-encoded URLs use this compatibility mode.
21    Url(String),
22    /// A base URL without userinfo plus raw credentials that Forge injects safely.
23    Credentials {
24        /// Absolute database URL without username or password.
25        base_url: String,
26        /// Raw username, without percent encoding.
27        #[serde(default, skip_serializing)]
28        username: Option<String>,
29        /// Raw password, without percent encoding.
30        #[serde(default, skip_serializing)]
31        password: Option<String>,
32    },
33}
34
35impl DatabaseUrl {
36    /// Creates a separated-credentials database URL input.
37    pub fn credentials(
38        base_url: impl Into<String>,
39        username: Option<String>,
40        password: Option<String>,
41    ) -> Self {
42        Self::Credentials {
43            base_url: base_url.into(),
44            username,
45            password,
46        }
47    }
48
49    /// Returns the complete URL when this input uses compatibility mode.
50    #[must_use]
51    pub fn as_url(&self) -> Option<&str> {
52        match self {
53            Self::Url(url) => Some(url),
54            Self::Credentials { .. } => None,
55        }
56    }
57
58    /// Returns the complete URL mutably when this input uses compatibility mode.
59    pub fn as_url_mut(&mut self) -> Option<&mut String> {
60        match self {
61            Self::Url(url) => Some(url),
62            Self::Credentials { .. } => None,
63        }
64    }
65
66    fn resolve(&self) -> Result<String> {
67        match self {
68            Self::Url(url) => Ok(url.clone()),
69            Self::Credentials {
70                base_url,
71                username,
72                password,
73            } => aster_forge_utils::url::url_with_credentials(
74                base_url,
75                username.as_deref(),
76                password.as_deref(),
77                "database base URL",
78            )
79            .map(std::convert::Into::into)
80            .map_err(|error| {
81                DbError::non_retryable(format!(
82                    "invalid database connection configuration: {error}"
83                ))
84            }),
85        }
86    }
87}
88
89impl std::fmt::Debug for DatabaseUrl {
90    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        match self {
92            Self::Url(_) => formatter.write_str("DatabaseUrl::Url(<redacted>)"),
93            Self::Credentials { .. } => formatter.write_str("DatabaseUrl::Credentials(<redacted>)"),
94        }
95    }
96}
97
98impl From<String> for DatabaseUrl {
99    fn from(url: String) -> Self {
100        Self::Url(url)
101    }
102}
103
104impl From<&str> for DatabaseUrl {
105    fn from(url: &str) -> Self {
106        Self::Url(url.to_string())
107    }
108}
109
110/// Database connection configuration.
111#[derive(Clone, Debug)]
112pub struct DatabaseConfig {
113    /// Database URL understood by SeaORM/sqlx.
114    pub url: DatabaseUrl,
115    /// Maximum pool size for non-SQLite connections and `SQLite` reader pools.
116    pub pool_size: u32,
117    /// Number of retries for connection establishment.
118    pub retry_count: u32,
119}
120
121impl DatabaseConfig {
122    /// Creates a config with conservative default pool and retry settings.
123    pub fn new(url: impl Into<String>) -> Self {
124        Self {
125            url: DatabaseUrl::Url(url.into()),
126            pool_size: 5,
127            retry_count: 3,
128        }
129    }
130
131    /// Creates a config from a base URL and raw credentials.
132    pub fn with_credentials(
133        base_url: impl Into<String>,
134        username: Option<String>,
135        password: Option<String>,
136    ) -> Self {
137        Self {
138            url: DatabaseUrl::credentials(base_url, username, password),
139            pool_size: 5,
140            retry_count: 3,
141        }
142    }
143}
144
145/// Pair of writer and reader database handles.
146#[derive(Clone)]
147pub struct DbHandles {
148    writer: DatabaseConnection,
149    reader: DatabaseConnection,
150    sqlite_read_write_split: bool,
151}
152
153impl DbHandles {
154    /// Creates writer/reader handles backed by the same connection.
155    #[must_use]
156    pub fn single(db: DatabaseConnection) -> Self {
157        Self {
158            writer: db.clone(),
159            reader: db,
160            sqlite_read_write_split: false,
161        }
162    }
163
164    /// Returns the writer connection.
165    #[must_use]
166    pub fn writer(&self) -> &DatabaseConnection {
167        &self.writer
168    }
169
170    /// Returns the reader connection.
171    #[must_use]
172    pub fn reader(&self) -> &DatabaseConnection {
173        &self.reader
174    }
175
176    /// Returns whether `SQLite` writer and reader use separate pools.
177    #[must_use]
178    pub fn sqlite_read_write_split(&self) -> bool {
179        self.sqlite_read_write_split
180    }
181
182    /// Closes the underlying database pools.
183    ///
184    /// Split `SQLite` handles own two independent pools, so both reader and writer must be closed.
185    /// Single-handle configurations clone the same pool into both fields and only close the writer
186    /// once.
187    ///
188    /// Both pools are always asked to close, even when the reader close fails: an early return
189    /// would leak the writer pool (`SQLite` WAL/file handles) with no way to retry, because `close`
190    /// consumes the handles. The reader error, being the first failure, is the one returned.
191    ///
192    /// # Errors
193    ///
194    /// Returns an error when the database operation fails.
195    pub async fn close(self) -> Result<()> {
196        let reader_result = if self.sqlite_read_write_split {
197            Some(self.reader.close().await)
198        } else {
199            None
200        };
201        let writer_result = self.writer.close().await;
202        first_close_error(reader_result, writer_result)
203    }
204}
205
206/// Returns the first close failure after both pools have been asked to close.
207fn first_close_error(
208    reader: Option<std::result::Result<(), sea_orm::DbErr>>,
209    writer: std::result::Result<(), sea_orm::DbErr>,
210) -> Result<()> {
211    if let Some(Err(error)) = reader {
212        return Err(DbError::from(error));
213    }
214    writer.map_err(DbError::from)
215}
216
217/// Connects to the configured database and installs a metrics callback.
218///
219/// # Errors
220///
221/// Returns an error when the database operation fails.
222pub async fn connect_with_metrics(
223    cfg: &DatabaseConfig,
224    metrics: SharedDbMetricsRecorder,
225) -> Result<DatabaseConnection> {
226    let url = cfg.url.resolve()?;
227    let retry_config = retry::RetryConfig {
228        max_retries: cfg.retry_count,
229        ..retry::RetryConfig::connection()
230    };
231    retry::with_retry(&retry_config, || {
232        Box::pin(connect_once(cfg, &url, metrics.clone()))
233    })
234    .await
235}
236
237/// Connects to the configured database without metrics.
238///
239/// # Errors
240///
241/// Returns an error when the database operation fails.
242pub async fn connect(cfg: &DatabaseConfig) -> Result<DatabaseConnection> {
243    connect_with_metrics(cfg, Arc::new(NoopDbMetrics)).await
244}
245
246/// Creates reader/writer handles for an existing writer connection and metrics recorder.
247///
248/// # Errors
249///
250/// Returns an error when the database operation fails.
251pub async fn connect_reader_for_writer_with_metrics(
252    cfg: &DatabaseConfig,
253    writer: DatabaseConnection,
254    metrics: SharedDbMetricsRecorder,
255) -> Result<DbHandles> {
256    let url = normalize_database_url(&cfg.url.resolve()?);
257    if !sqlite_reader_pool_enabled(&url) {
258        return Ok(DbHandles::single(writer));
259    }
260
261    let retry_config = retry::RetryConfig {
262        max_retries: cfg.retry_count,
263        ..retry::RetryConfig::connection()
264    };
265    let reader = retry::with_retry(&retry_config, || {
266        connect_sqlite_reader_once(cfg, &url, metrics.clone())
267    })
268    .await?;
269    Ok(DbHandles {
270        writer,
271        reader,
272        sqlite_read_write_split: true,
273    })
274}
275
276/// Creates reader/writer handles for an existing writer connection without metrics.
277///
278/// # Errors
279///
280/// Returns an error when the database operation fails.
281pub async fn connect_reader_for_writer(
282    cfg: &DatabaseConfig,
283    writer: DatabaseConnection,
284) -> Result<DbHandles> {
285    connect_reader_for_writer_with_metrics(cfg, writer, Arc::new(NoopDbMetrics)).await
286}
287
288async fn connect_once(
289    cfg: &DatabaseConfig,
290    database_url: &str,
291    metrics: SharedDbMetricsRecorder,
292) -> Result<DatabaseConnection> {
293    let url = normalize_database_url(database_url);
294    let is_sqlite = url.starts_with("sqlite:");
295    // SQLite relies on a single pooled connection so concurrent writers are serialized at
296    // connection acquisition; repo-layer "lock" helpers do not emulate row locks there.
297    let max_connections = if is_sqlite { 1 } else { cfg.pool_size };
298
299    let mut opt = ConnectOptions::new(&url);
300    opt.max_connections(max_connections)
301        .min_connections(1)
302        .sqlx_logging(false)
303        .test_before_acquire(true);
304
305    // SeaORM's generic Database::connect() pre-validates URLs with url::Url::parse(),
306    // which rejects Windows-style SQLite paths containing backslashes. Route SQLite
307    // through sqlx's dedicated connector instead so platform-native paths keep working.
308    let db = if is_sqlite {
309        SqlxSqliteConnector::connect(opt)
310            .await
311            .map_err(DbError::database_connection)?
312    } else {
313        Database::connect(opt)
314            .await
315            .map_err(DbError::database_connection)?
316    };
317
318    let backend = db.get_database_backend();
319    tracing::info!(backend = ?backend, "database connected");
320
321    if is_sqlite {
322        tracing::info!(max_connections, "applying SQLite PRAGMA optimizations");
323        // PRAGMA failures classify through DbError::from: transient lock contention
324        // (SQLITE_BUSY while another process initializes the file) stays retryable,
325        // persistent failures (permissions, corrupt file) fail fast.
326        db.execute_unprepared("PRAGMA journal_mode=WAL;")
327            .await
328            .map_err(DbError::from)?;
329        db.execute_unprepared("PRAGMA busy_timeout=15000;")
330            .await
331            .map_err(DbError::from)?;
332        db.execute_unprepared("PRAGMA synchronous=NORMAL;")
333            .await
334            .map_err(DbError::from)?;
335        db.execute_unprepared("PRAGMA foreign_keys=ON;")
336            .await
337            .map_err(DbError::from)?;
338    }
339
340    let mut db = db;
341    install_db_metrics(&mut db, metrics);
342
343    Ok(db)
344}
345
346async fn connect_sqlite_reader_once(
347    cfg: &DatabaseConfig,
348    normalized_writer_url: &str,
349    metrics: SharedDbMetricsRecorder,
350) -> Result<DatabaseConnection> {
351    let reader_url = sqlite_reader_url(normalized_writer_url);
352    let max_connections = cfg.pool_size.max(1);
353    let mut opt = ConnectOptions::new(&reader_url);
354    opt.max_connections(max_connections)
355        .min_connections(1)
356        .sqlx_logging(false)
357        .test_before_acquire(true)
358        .map_sqlx_sqlite_pool_opts(|pool_options| {
359            pool_options.after_connect(|conn, _meta| {
360                Box::pin(async move {
361                    use sea_orm::sqlx::Executor;
362
363                    conn.execute("PRAGMA busy_timeout=15000;").await?;
364                    conn.execute("PRAGMA synchronous=NORMAL;").await?;
365                    conn.execute("PRAGMA foreign_keys=ON;").await?;
366                    conn.execute("PRAGMA query_only=ON;").await?;
367                    Ok(())
368                })
369            })
370        });
371
372    let mut db = SqlxSqliteConnector::connect(opt)
373        .await
374        .map_err(DbError::database_connection)?;
375    install_db_metrics(&mut db, metrics);
376
377    tracing::info!(
378        max_connections,
379        "SQLite reader pool connected with query_only pragma"
380    );
381    Ok(db)
382}
383
384fn normalize_database_url(database_url: &str) -> String {
385    if database_url == "sqlite::memory:" {
386        return database_url.to_string();
387    }
388
389    if database_url.starts_with("sqlite://") && !database_url.contains('?') {
390        return format!("{database_url}?mode=rwc");
391    }
392
393    database_url.to_string()
394}
395
396fn sqlite_reader_pool_enabled(normalized_url: &str) -> bool {
397    normalized_url.starts_with("sqlite:") && !is_sqlite_memory_url(normalized_url)
398}
399
400fn is_sqlite_memory_url(normalized_url: &str) -> bool {
401    normalized_url == "sqlite::memory:"
402        || normalized_url
403            .split_once('?')
404            .is_some_and(|(_, query)| query.split('&').any(|param| param == "mode=memory"))
405}
406
407fn sqlite_reader_url(normalized_writer_url: &str) -> String {
408    let Some((base, query)) = normalized_writer_url.split_once('?') else {
409        return format!("{normalized_writer_url}?mode=ro");
410    };
411
412    let mut saw_mode = false;
413    let query = query
414        .split('&')
415        .filter(|param| !param.is_empty())
416        .map(|param| {
417            if param.starts_with("mode=") {
418                saw_mode = true;
419                "mode=ro"
420            } else {
421                param
422            }
423        })
424        .collect::<Vec<_>>();
425
426    if saw_mode {
427        format!("{base}?{}", query.join("&"))
428    } else if query.is_empty() {
429        format!("{base}?mode=ro")
430    } else {
431        format!("{base}?mode=ro&{}", query.join("&"))
432    }
433}
434
435fn install_db_metrics(db: &mut DatabaseConnection, metrics: SharedDbMetricsRecorder) {
436    if !metrics.enabled() {
437        return;
438    }
439
440    db.set_metric_callback(move |info| {
441        metrics.record_db_query(&db_query_metric_from_sea_orm(info));
442    });
443}
444
445fn db_metric_backend_from_sea_orm(backend: sea_orm::DbBackend) -> DbMetricBackend {
446    match backend {
447        sea_orm::DbBackend::Sqlite => DbMetricBackend::Sqlite,
448        sea_orm::DbBackend::MySql => DbMetricBackend::MySql,
449        sea_orm::DbBackend::Postgres => DbMetricBackend::Postgres,
450        _ => DbMetricBackend::Other,
451    }
452}
453
454fn db_query_kind_from_sql(sql: &str) -> DbQueryKind {
455    match sql
456        .trim_start()
457        .split_ascii_whitespace()
458        .next()
459        .unwrap_or_default()
460        .to_ascii_uppercase()
461        .as_str()
462    {
463        "SELECT" => DbQueryKind::Select,
464        "INSERT" => DbQueryKind::Insert,
465        "UPDATE" => DbQueryKind::Update,
466        "DELETE" => DbQueryKind::Delete,
467        "WITH" => DbQueryKind::With,
468        "BEGIN" | "COMMIT" | "ROLLBACK" | "SAVEPOINT" | "RELEASE" => DbQueryKind::Transaction,
469        "CREATE" | "ALTER" | "DROP" | "TRUNCATE" => DbQueryKind::Ddl,
470        "PRAGMA" => DbQueryKind::Pragma,
471        _ => DbQueryKind::Other,
472    }
473}
474
475fn db_query_metric_from_sea_orm(info: &sea_orm::metric::Info<'_>) -> DbQueryMetric {
476    DbQueryMetric::new(
477        db_metric_backend_from_sea_orm(info.statement.db_backend),
478        db_query_kind_from_sql(&info.statement.sql),
479        info.failed,
480        info.elapsed,
481    )
482}
483
484#[cfg(test)]
485mod tests {
486    use super::{first_close_error, normalize_database_url};
487    use crate::connection::{DatabaseConfig, DatabaseUrl};
488    use aster_forge_metrics::{DbQueryKind, NoopDbMetrics};
489    use aster_forge_test::temp::SqliteTestDatabase;
490    use sea_orm::{ConnectionTrait, DbErr, TransactionTrait};
491    use std::sync::Arc;
492
493    #[test]
494    fn credentialed_database_url_is_resolved_and_debug_is_redacted() {
495        let raw_password = "db#[]{}^+=*@:/?%\u{5bc6}\u{7801}";
496        let input = DatabaseUrl::credentials(
497            "postgres://db.example:5432/app?sslmode=require",
498            Some("app-user".to_string()),
499            Some(raw_password.to_string()),
500        );
501
502        let resolved = input.resolve().unwrap();
503        assert!(resolved.starts_with("postgres://app-user:"));
504        assert!(resolved.ends_with("@db.example:5432/app?sslmode=require"));
505        assert!(!resolved.contains(raw_password));
506
507        let debug = format!("{input:?}");
508        assert_eq!(debug, "DatabaseUrl::Credentials(<redacted>)");
509        assert!(!debug.contains(raw_password));
510    }
511
512    #[test]
513    fn credentialed_database_url_deserializes_without_serializing_secrets() {
514        let raw_username = "app-user@example.com";
515        let raw_password = "db#[]{}^+=*@:/?%secret";
516        let input: DatabaseUrl = serde_json::from_str(&format!(
517            r#"{{"base_url":"postgres://db.example:5432/app","username":"{raw_username}","password":"{raw_password}"}}"#,
518        ))
519        .unwrap();
520
521        assert_eq!(
522            input,
523            DatabaseUrl::credentials(
524                "postgres://db.example:5432/app",
525                Some(raw_username.to_string()),
526                Some(raw_password.to_string()),
527            )
528        );
529
530        let serialized = serde_json::to_string(&input).unwrap();
531        assert!(serialized.contains("postgres://db.example:5432/app"));
532        assert!(!serialized.contains(raw_username));
533        assert!(!serialized.contains(raw_password));
534        assert!(!serialized.contains("db%23%5B%5D"));
535    }
536
537    #[tokio::test]
538    async fn invalid_credentialed_database_url_returns_non_retryable_redacted_error() {
539        let raw_password = "raw#database-secret";
540        let mut config = DatabaseConfig::with_credentials(
541            "postgres://existing@db.example/app",
542            Some("replacement".to_string()),
543            Some(raw_password.to_string()),
544        );
545        config.retry_count = 10;
546
547        let error = super::connect(&config).await.unwrap_err();
548        assert!(matches!(error, crate::DbError::NonRetryable(_)));
549        assert!(
550            error
551                .to_string()
552                .contains("must not already include userinfo")
553        );
554        assert!(!error.to_string().contains(raw_password));
555    }
556
557    #[test]
558    fn first_close_error_prefers_the_reader_failure() {
559        let error = first_close_error(
560            Some(Err(DbErr::Custom("reader blew up".to_string()))),
561            Err(DbErr::Custom("writer blew up".to_string())),
562        )
563        .expect_err("both pools failing must surface an error");
564        assert!(error.to_string().contains("reader blew up"));
565
566        let error = first_close_error(
567            Some(Err(DbErr::Custom("reader blew up".to_string()))),
568            Ok(()),
569        )
570        .expect_err("reader failure must surface even when the writer closes cleanly");
571        assert!(error.to_string().contains("reader blew up"));
572    }
573
574    #[test]
575    fn first_close_error_surfaces_writer_failure_and_clean_runs() {
576        let error = first_close_error(
577            Some(Ok(())),
578            Err(DbErr::Custom("writer blew up".to_string())),
579        )
580        .expect_err("writer failure must surface when the reader closed cleanly");
581        assert!(error.to_string().contains("writer blew up"));
582
583        // Single-handle configurations close only the writer.
584        let error = first_close_error(None, Err(DbErr::Custom("writer blew up".to_string())))
585            .expect_err("writer failure must surface without a split reader");
586        assert!(error.to_string().contains("writer blew up"));
587
588        assert!(first_close_error(Some(Ok(())), Ok(())).is_ok());
589        assert!(first_close_error(None, Ok(())).is_ok());
590    }
591
592    #[test]
593    fn sqlite_urls_without_query_default_to_rwc_mode() {
594        assert_eq!(
595            normalize_database_url("sqlite:///var/lib/asterdrive/app.db"),
596            "sqlite:///var/lib/asterdrive/app.db?mode=rwc"
597        );
598        assert_eq!(
599            normalize_database_url("sqlite://data/asterdrive.db"),
600            "sqlite://data/asterdrive.db?mode=rwc"
601        );
602    }
603
604    #[test]
605    fn sqlite_memory_and_existing_queries_are_preserved() {
606        assert_eq!(normalize_database_url("sqlite::memory:"), "sqlite::memory:");
607        assert_eq!(
608            normalize_database_url("sqlite:///var/lib/asterdrive/app.db?mode=ro"),
609            "sqlite:///var/lib/asterdrive/app.db?mode=ro"
610        );
611        assert_eq!(
612            normalize_database_url("postgres://user:pass@localhost/asterdrive"),
613            "postgres://user:pass@localhost/asterdrive"
614        );
615    }
616
617    #[test]
618    fn sqlite_reader_pool_skips_memory_databases() {
619        assert!(!super::sqlite_reader_pool_enabled("sqlite::memory:"));
620        assert!(!super::sqlite_reader_pool_enabled(
621            "sqlite://memory-test?mode=memory&cache=shared"
622        ));
623        assert!(super::sqlite_reader_pool_enabled(
624            "sqlite:///var/lib/asterdrive/app.db?mode=rwc"
625        ));
626    }
627
628    #[test]
629    fn sqlite_reader_url_forces_read_only_mode() {
630        assert_eq!(
631            super::sqlite_reader_url("sqlite:///var/lib/asterdrive/app.db?mode=rwc"),
632            "sqlite:///var/lib/asterdrive/app.db?mode=ro"
633        );
634        assert_eq!(
635            super::sqlite_reader_url("sqlite:///var/lib/asterdrive/app.db?cache=shared"),
636            "sqlite:///var/lib/asterdrive/app.db?mode=ro&cache=shared"
637        );
638        assert_eq!(
639            super::sqlite_reader_url("sqlite:///var/lib/asterdrive/app.db?"),
640            "sqlite:///var/lib/asterdrive/app.db?mode=ro"
641        );
642        assert_eq!(
643            super::sqlite_reader_url("sqlite:///var/lib/asterdrive/app.db?&cache=shared"),
644            "sqlite:///var/lib/asterdrive/app.db?mode=ro&cache=shared"
645        );
646        assert_eq!(
647            super::sqlite_reader_url("sqlite:///var/lib/asterdrive/app.db?mode=rwc&"),
648            "sqlite:///var/lib/asterdrive/app.db?mode=ro"
649        );
650    }
651
652    #[test]
653    fn db_query_kind_from_sql_matches_low_cardinality_labels() {
654        assert_eq!(
655            super::db_query_kind_from_sql(" SELECT 1"),
656            DbQueryKind::Select
657        );
658        assert_eq!(
659            super::db_query_kind_from_sql("insert into users values (1)"),
660            DbQueryKind::Insert
661        );
662        assert_eq!(
663            super::db_query_kind_from_sql("UPDATE users SET name = ?"),
664            DbQueryKind::Update
665        );
666        assert_eq!(
667            super::db_query_kind_from_sql("delete from users"),
668            DbQueryKind::Delete
669        );
670        assert_eq!(
671            super::db_query_kind_from_sql("WITH recent AS (SELECT 1) SELECT * FROM recent"),
672            DbQueryKind::With
673        );
674        assert_eq!(
675            super::db_query_kind_from_sql("BEGIN"),
676            DbQueryKind::Transaction
677        );
678        assert_eq!(
679            super::db_query_kind_from_sql("CREATE TABLE example (id integer)"),
680            DbQueryKind::Ddl
681        );
682        assert_eq!(
683            super::db_query_kind_from_sql("PRAGMA foreign_keys=ON"),
684            DbQueryKind::Pragma
685        );
686        assert_eq!(super::db_query_kind_from_sql("VACUUM"), DbQueryKind::Other);
687    }
688
689    #[tokio::test]
690    async fn sqlite_connector_accepts_windows_style_urls() {
691        let url = format!(
692            "sqlite://windows\\sqlite-url-{}?mode=memory&cache=shared",
693            uuid::Uuid::new_v4()
694        );
695        let db = super::connect_with_metrics(
696            &DatabaseConfig {
697                url: url.into(),
698                pool_size: 10,
699                retry_count: 3,
700            },
701            Arc::new(NoopDbMetrics),
702        )
703        .await
704        .expect("sqlite connection should succeed for Windows-style URL");
705
706        db.execute_unprepared("SELECT 1;")
707            .await
708            .expect("sqlite query should succeed");
709    }
710
711    #[tokio::test]
712    async fn sqlite_memory_handles_use_single_connection() {
713        let cfg = DatabaseConfig {
714            url: "sqlite::memory:".into(),
715            pool_size: 4,
716            retry_count: 0,
717        };
718        let writer = super::connect_with_metrics(&cfg, Arc::new(NoopDbMetrics))
719            .await
720            .expect("sqlite memory writer should connect");
721        let handles =
722            super::connect_reader_for_writer_with_metrics(&cfg, writer, Arc::new(NoopDbMetrics))
723                .await
724                .expect("sqlite memory handles should connect");
725
726        assert!(!handles.sqlite_read_write_split());
727        assert_eq!(
728            handles.writer().get_database_backend(),
729            handles.reader().get_database_backend()
730        );
731
732        handles
733            .close()
734            .await
735            .expect("single sqlite handles should close");
736    }
737
738    #[tokio::test]
739    async fn sqlite_reader_pool_is_query_only() {
740        let database = SqliteTestDatabase::new("reader-pool");
741        let cfg = DatabaseConfig {
742            url: database.url().into(),
743            pool_size: 4,
744            retry_count: 0,
745        };
746        let writer = super::connect_with_metrics(&cfg, Arc::new(NoopDbMetrics))
747            .await
748            .expect("sqlite writer should connect");
749        let handles =
750            super::connect_reader_for_writer_with_metrics(&cfg, writer, Arc::new(NoopDbMetrics))
751                .await
752                .expect("sqlite handles should connect");
753        assert!(handles.sqlite_read_write_split());
754
755        handles
756            .writer()
757            .execute_unprepared("CREATE TABLE reader_guard (id INTEGER PRIMARY KEY);")
758            .await
759            .expect("writer should create table");
760
761        let write_result = handles
762            .reader()
763            .execute_unprepared("INSERT INTO reader_guard (id) VALUES (1);")
764            .await;
765        assert!(write_result.is_err(), "reader pool must reject writes");
766
767        handles
768            .close()
769            .await
770            .expect("split sqlite handles should close");
771    }
772
773    #[tokio::test]
774    async fn sqlite_reader_pool_reads_while_writer_connection_is_busy() {
775        let database = SqliteTestDatabase::new("reader-writer-split");
776        let cfg = DatabaseConfig {
777            url: database.url().into(),
778            pool_size: 4,
779            retry_count: 0,
780        };
781        let writer = super::connect_with_metrics(&cfg, Arc::new(NoopDbMetrics))
782            .await
783            .expect("sqlite writer should connect");
784        let handles =
785            super::connect_reader_for_writer_with_metrics(&cfg, writer, Arc::new(NoopDbMetrics))
786                .await
787                .expect("sqlite handles should connect");
788        assert!(handles.sqlite_read_write_split());
789
790        handles
791            .writer()
792            .execute_unprepared("CREATE TABLE split_guard (id INTEGER PRIMARY KEY, name TEXT);")
793            .await
794            .expect("writer should create table");
795        handles
796            .writer()
797            .execute_unprepared("INSERT INTO split_guard (id, name) VALUES (1, 'ready');")
798            .await
799            .expect("writer should seed row");
800
801        let txn = handles
802            .writer()
803            .begin()
804            .await
805            .expect("writer transaction should begin");
806        txn.execute_unprepared("UPDATE split_guard SET name = 'held' WHERE id = 1;")
807            .await
808            .expect("writer transaction should hold a write lock");
809
810        let read = tokio::time::timeout(std::time::Duration::from_millis(250), async {
811            handles
812                .reader()
813                .query_one_raw(sea_orm::Statement::from_string(
814                    sea_orm::DbBackend::Sqlite,
815                    "SELECT name FROM split_guard WHERE id = 1",
816                ))
817                .await
818        })
819        .await
820        .expect("reader should not wait on the writer pool queue")
821        .expect("reader query should succeed")
822        .expect("reader query should return row");
823        let name: String = read
824            .try_get_by_index(0)
825            .expect("reader query should decode name");
826        assert_eq!(name, "ready");
827
828        txn.rollback()
829            .await
830            .expect("writer transaction should roll back");
831
832        handles
833            .close()
834            .await
835            .expect("split sqlite handles should close after reader query");
836    }
837}
aster_forge_db/connection.rs

aster_forge_db/
connection.rs
