diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 574ec089c..4f006d4ae 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -93,6 +93,12 @@ when the database runs in a container. For local Oracle runs, add `-- --test-threads=2` to avoid overwhelming the listener. Check affected matrix results for SQL execution changes; SQLite alone cannot establish portability. +ODBC databases can also run on sqlx 0.9, through the [sqlx-odbc](https://github.com/lovasoa/sqlx-odbc) driver. This is the unstable migration path for [#1056](https://github.com/sqlpage/SQLPage/issues/1056), and releases do not include it. To use it, build with the `unstable-sqlx-next` feature and set `SQLPAGE_UNSTABLE_SQLX_NEXT=true`. With this driver, migrations, `sqlpage_files` and CSV imports stop with an error that names the missing feature. + +```bash +SQLPAGE_UNSTABLE_SQLX_NEXT=true DATABASE_URL='Driver=DuckDB;Database=/tmp/sqlpage-test.duckdb' cargo test --features unstable-sqlx-next +``` + ## Submitting a change Follow nearby code and tests, add regression coverage, and describe the resulting diff --git a/Cargo.lock b/Cargo.lock index 565cc9428..d9924b169 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -292,6 +292,12 @@ dependencies = [ "alloc-no-stdlib", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "android-activity" version = "0.6.1" @@ -1551,6 +1557,9 @@ name = "either" version = "1.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be" +dependencies = [ + "serde", +] [[package]] name = "elliptic-curve" @@ -1984,6 +1993,8 @@ version = "0.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" dependencies = [ + "allocator-api2", + "equivalent", "foldhash", ] @@ -4634,7 +4645,9 @@ dependencies = [ "serde_json", "sha2 0.11.1", "sqlparser", + "sqlx", "sqlx-core-oldapi", + "sqlx-odbc", "tempfile", "tokio", "tokio-stream", @@ -4667,6 +4680,50 @@ dependencies = [ "syn 3.0.10", ] +[[package]] +name = "sqlx" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "378620ccc25c62c89d8be1c819e76a88d59bdcc3304733330788948e619bfd71" +dependencies = [ + "sqlx-core", + "sqlx-macros", +] + +[[package]] +name = "sqlx-core" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05b44e85bf579a8eeb4ceaa77a3a523baf2bf0e9bac7e40f405d537b5d2d5ccb" +dependencies = [ + "base64 0.22.1", + "bigdecimal", + "bytes", + "cfg-if", + "crc", + "crossbeam-queue", + "either", + "event-listener", + "futures-core", + "futures-intrusive", + "futures-io", + "futures-util", + "hashbrown 0.16.1", + "hashlink", + "indexmap 2.14.2", + "log", + "memchr", + "percent-encoding", + "serde", + "sha2 0.10.9", + "smallvec", + "thiserror 2.0.21", + "tokio", + "tokio-stream", + "tracing", + "url", +] + [[package]] name = "sqlx-core-oldapi" version = "0.6.59" @@ -4730,6 +4787,58 @@ dependencies = [ "whoami", ] +[[package]] +name = "sqlx-macros" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bd2b84f2bc39a5705ef27ec785a11c934a41bbd4a24941e257927cddc26b60bf" +dependencies = [ + "proc-macro2", + "quote", + "sqlx-core", + "sqlx-macros-core", + "syn 2.0.119", +] + +[[package]] +name = "sqlx-macros-core" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb8d96de5fdc85a5c4ec813432b523ec637e80ba98f046555f75f7908ddac7c3" +dependencies = [ + "cfg-if", + "dotenvy", + "either", + "heck", + "hex", + "proc-macro2", + "quote", + "serde", + "serde_json", + "sha2 0.10.9", + "sqlx-core", + "syn 2.0.119", + "tokio", + "url", +] + +[[package]] +name = "sqlx-odbc" +version = "0.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23aa8c356e6e3f22dfe20a14a461e0e4ada41b0fc4f14dda65a377f7c163ce9d" +dependencies = [ + "bigdecimal", + "chrono", + "futures-core", + "futures-util", + "log", + "odbc-api", + "sqlx-core", + "thiserror 2.0.21", + "url", +] + [[package]] name = "sqlx-rt-oldapi" version = "0.6.59" diff --git a/Cargo.toml b/Cargo.toml index d5c3b4fa8..7e18d6ce7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -112,6 +112,8 @@ tokio-util = "0.7.12" openidconnect = { version = "4.0.0", default-features = false, features = ["accept-rfc3339-timestamps"] } encoding_rs = "0.8.35" odbc-sys = { version = "0", optional = true } +sqlx-next = { package = "sqlx", version = "0.9", default-features = false, features = ["runtime-tokio"], optional = true } +sqlx-odbc = { version = "0.0.3", default-features = false, features = ["bigdecimal", "chrono"], optional = true } regex = "1" lettre = { version = "0.11", default-features = false, features = [ "aws-lc-rs", @@ -137,6 +139,7 @@ opentelemetry-semantic-conventions = { version = "0.33", features = ["semconv_ex [features] default = [] +unstable-sqlx-next = ["dep:sqlx-next", "dep:sqlx-odbc"] odbc-static = ["odbc-sys", "odbc-sys/vendored-unix-odbc"] lambda-web = [ "dep:actix-http", diff --git a/src/app_config.rs b/src/app_config.rs index 6e9166457..267a88008 100644 --- a/src/app_config.rs +++ b/src/app_config.rs @@ -215,6 +215,10 @@ pub struct AppConfig { /// A separate field for the database password. If set, this will override any password specified in the `database_url`. #[serde(default)] pub database_password: Option, + /// Connects to SQL Server and ODBC databases through sqlx 0.9 instead of sqlx-oldapi. + #[cfg(feature = "unstable-sqlx-next")] + #[serde(default)] + pub unstable_sqlx_next: bool, pub max_database_pool_connections: Option, #[serde( default, @@ -813,16 +817,23 @@ pub fn test_database_url() -> String { std::env::var("DATABASE_URL").unwrap_or_else(|_| "sqlite::memory:".to_string()) } +#[must_use] +pub fn test_unstable_sqlx_next() -> bool { + std::env::var("SQLPAGE_UNSTABLE_SQLX_NEXT").is_ok_and(|value| value == "true") +} + #[cfg(test)] pub mod tests { use super::AppConfig; pub use super::test_database_url; + use super::test_unstable_sqlx_next; #[must_use] pub fn test_config() -> AppConfig { let mut config = serde_json::from_str::( &serde_json::json!({ "database_url": test_database_url(), + "unstable_sqlx_next": test_unstable_sqlx_next(), "listen_on": "localhost:8080" }) .to_string(), diff --git a/src/filesystem.rs b/src/filesystem.rs index 524241c8d..49d6b017f 100644 --- a/src/filesystem.rs +++ b/src/filesystem.rs @@ -1,10 +1,11 @@ use crate::webserver::ErrorWithStatus; use crate::webserver::database::SupportedDatabase; +use crate::webserver::database::driver::unsupported_by_sqlx_next; use crate::webserver::{Database, StatusCodeResultExt, make_placeholder}; use crate::{AppState, TEMPLATES_DIR}; use anyhow::Context; use chrono::{DateTime, Utc}; -use sqlx::any::{AnyStatement, AnyTypeInfo}; +use sqlx::any::{AnyPool, AnyStatement, AnyTypeInfo}; use sqlx::executor::Executor; use sqlx::postgres::types::PgTimeTz; use sqlx::{postgres::Postgres, statement::Statement, types::Type}; @@ -85,9 +86,7 @@ impl FileSystem { (Ok(modified), _) => Ok(modified), (Err(e), Some(db_fs)) if is_path_missing_error(&e) => { // no local file, try the database - db_fs - .file_modified_since_in_db(app_state, path, since) - .await + db_fs.file_modified_since_in_db(path, since).await } (Err(e), _) => { let status = io_error_status(&local_path, &e).await; @@ -143,7 +142,7 @@ impl FileSystem { (Ok(f), _) => Ok(f), (Err(e), Some(db_fs)) if is_path_missing_error(&e) => { // no local file, try the database - db_fs.read_file(app_state, path.as_ref()).await + db_fs.read_file(path.as_ref()).await } (Err(e), _) => { let status = io_error_status(&local_path, &e).await; @@ -201,7 +200,7 @@ impl FileSystem { path.display() ); if let Some(db_fs) = &self.db_fs_queries { - return db_fs.file_exists(app_state, path).await; + return db_fs.file_exists(path).await; } } Ok(local_exists) @@ -313,6 +312,7 @@ async fn file_modified_since_local(path: &Path, since: DateTime) -> tokio:: } pub struct DbFsQueries { + pool: AnyPool, was_modified: AnyStatement<'static>, read_file: AnyStatement<'static>, exists: AnyStatement<'static>, @@ -339,17 +339,22 @@ impl DbFsQueries { async fn init(db: &Database) -> anyhow::Result { log::debug!("Initializing database filesystem queries"); - Self::check_table_available(db).await?; + let pool = db + .connection + .as_old_api() + .ok_or_else(|| unsupported_by_sqlx_next("database-backed files"))? + .clone(); + Self::check_table_available(&pool).await?; Ok(Self { + pool, was_modified: Self::make_was_modified_query(db).await?, read_file: Self::make_read_file_query(db).await?, exists: Self::make_exists_query(db).await?, }) } - async fn check_table_available(db: &Database) -> anyhow::Result<()> { - db.connection - .execute("SELECT 1 FROM sqlpage_files WHERE 1 = 0") + async fn check_table_available(pool: &AnyPool) -> anyhow::Result<()> { + pool.execute("SELECT 1 FROM sqlpage_files WHERE 1 = 0") .await .map(|_| ()) .context("Unable to access sqlpage_files")?; @@ -391,7 +396,6 @@ impl DbFsQueries { async fn file_modified_since_in_db( &self, - app_state: &AppState, path: &Path, since: DateTime, ) -> anyhow::Result { @@ -409,15 +413,12 @@ impl DbFsQueries { self.was_modified.sql(), (since, path) ); - let was_modified_i32 = query - .fetch_optional(&app_state.db.connection) - .await - .with_context(|| { - format!( - "Unable to check when {} was last modified in the database", - path.display() - ) - })?; + let was_modified_i32 = query.fetch_optional(&self.pool).await.with_context(|| { + format!( + "Unable to check when {} was last modified in the database", + path.display() + ) + })?; log::trace!( "DB File {} was modified result: {was_modified_i32:?}", path.display() @@ -425,12 +426,12 @@ impl DbFsQueries { Ok(was_modified_i32 == Some((1,))) } - async fn read_file(&self, app_state: &AppState, path: &Path) -> anyhow::Result> { + async fn read_file(&self, path: &Path) -> anyhow::Result> { log::debug!("Reading file {} from the database", path.display()); self.read_file .query_as::<(Vec,)>() .bind(path.display().to_string()) - .fetch_optional(&app_state.db.connection) + .fetch_optional(&self.pool) .await .map_err(anyhow::Error::from) .and_then(|modified| { @@ -446,7 +447,7 @@ impl DbFsQueries { .with_context(|| format!("Unable to read {} from the database", path.display())) } - async fn file_exists(&self, app_state: &AppState, path: &Path) -> anyhow::Result { + async fn file_exists(&self, path: &Path) -> anyhow::Result { let query = self .exists .query_as::<(i32,)>() @@ -459,7 +460,7 @@ impl DbFsQueries { self.exists.sql(), (path,) ); - let result = query.fetch_optional(&app_state.db.connection).await; + let result = query.fetch_optional(&self.pool).await; log::debug!("DB File exists result: {result:?}"); result.map(|result| result.is_some()).with_context(|| { format!( @@ -488,7 +489,7 @@ async fn test_sql_file_read_utf8() -> anyhow::Result<()> { let create_table_sql = DbFsQueries::get_create_table_sql(state.db.info.database_type); let db = &state.db; - let conn = &db.connection; + let conn = db.connection.as_old_api().unwrap(); // Other tests share this database, so never drop a table their initialized state may use. if conn .execute("SELECT 1 FROM sqlpage_files WHERE 1 = 0") diff --git a/src/telemetry_metrics.rs b/src/telemetry_metrics.rs index 9b2bf6e50..d77251e27 100644 --- a/src/telemetry_metrics.rs +++ b/src/telemetry_metrics.rs @@ -1,8 +1,8 @@ +use crate::webserver::database::driver::DatabasePool; use opentelemetry::global; use opentelemetry::metrics::{Histogram, ObservableGauge}; use opentelemetry_semantic_conventions::attribute as otel; use opentelemetry_semantic_conventions::metric as otel_metric; -use sqlx::any::AnyPool; pub struct TelemetryMetrics { pub http_request_duration: Histogram, @@ -41,7 +41,7 @@ impl Default for TelemetryMetrics { impl TelemetryMetrics { #[must_use] - pub fn new(pool: &AnyPool, db_system_name: &'static str) -> Self { + pub fn new(pool: &DatabasePool, db_system_name: &'static str) -> Self { let meter = global::meter("sqlpage"); let http_request_duration = meter .f64_histogram(otel_metric::HTTP_SERVER_REQUEST_DURATION) diff --git a/src/webserver/database/connect.rs b/src/webserver/database/connect.rs index 84f3a972a..e6f41d18c 100644 --- a/src/webserver/database/connect.rs +++ b/src/webserver/database/connect.rs @@ -1,11 +1,8 @@ -use std::{mem::take, time::Duration}; +use std::{future::Future, mem::take, sync::Arc, time::Duration}; use super::Database; -use crate::{ - ON_CONNECT_FILE, ON_RESET_FILE, - app_config::AppConfig, - webserver::database::{DbInfo, SupportedDatabase}, -}; +use super::driver::DatabasePool; +use crate::{ON_CONNECT_FILE, ON_RESET_FILE, app_config::AppConfig, webserver::database::DbInfo}; use anyhow::Context; use futures_util::future::BoxFuture; use sqlx::connection::{ConnectOptions, Connection}; @@ -23,6 +20,12 @@ impl Database { let mut connect_options: AnyConnectOptions = database_url .parse() .with_context(|| format!("\"{database_url}\" is not a valid database URL. Please change the \"database_url\" option in the configuration file."))?; + #[cfg(feature = "unstable-sqlx-next")] + if config.unstable_sqlx_next + && let Some(driver) = super::sqlx_next::Driver::serving(connect_options.kind()) + { + return Self::init_sqlx_next(config, driver, connect_options.kind()).await; + } if let Some(password) = &config.database_password { set_database_password(&mut connect_options, password); } @@ -35,24 +38,9 @@ impl Database { ); set_custom_connect_options(&mut connect_options, config); log::debug!("Connecting to database: {database_url}"); - let mut retries = config.database_connection_retries; - - let mut conn: AnyConnection = loop { - match AnyConnection::connect_with(&connect_options).await { - Ok(c) => break c, - Err(e) => { - if retries == 0 { - return Err(anyhow::Error::new(e) - .context(format!("Unable to open connection to {database_url}"))); - } - log::warn!("Failed to connect to the database: {e:#}. Retrying in 5 seconds."); - retries -= 1; - tokio::time::sleep(Duration::from_secs(5)).await; - } - } - }; + let mut conn: AnyConnection = + retry_connecting(config, || AnyConnection::connect_with(&connect_options)).await?; let dbms_name: String = conn.dbms_name().await?; - let database_type = SupportedDatabase::from_dbms_name(&dbms_name); drop(conn); let db_kind = connect_options.kind(); @@ -63,74 +51,133 @@ impl Database { log::debug!("Initialized {dbms_name:?} database pool: {pool:#?}"); Ok(Database { - connection: pool, - info: DbInfo { - dbms_name, - database_type, - kind: db_kind, - }, + connection: DatabasePool::OldApi(pool), + info: DbInfo::new(dbms_name, db_kind), + }) + } + + #[cfg(feature = "unstable-sqlx-next")] + async fn init_sqlx_next( + config: &AppConfig, + driver: super::sqlx_next::Driver, + kind: AnyKind, + ) -> anyhow::Result { + log::warn!("Connecting to the {kind:?} database through the unstable sqlx 0.9 drivers"); + let (pool, dbms_name) = driver.connect(config).await?; + log::debug!("Initialized {dbms_name:?} database pool: {pool:#?}"); + Ok(Database { + connection: DatabasePool::SqlxNext(pool), + info: DbInfo::new(dbms_name, kind), }) } fn create_pool_options(config: &AppConfig, kind: AnyKind) -> PoolOptions { + let ConnectionHooks { + on_connect, + on_reset, + } = ConnectionHooks::read(config); let mut pool_options = PoolOptions::new() - .max_connections(if let Some(max) = config.max_database_pool_connections { - max - } else { - // Different databases have a different number of max concurrent connections allowed by default - match kind { - AnyKind::Postgres | AnyKind::Odbc => 50, // Default to PostgreSQL-like limits for Generic - AnyKind::MySql => 75, - AnyKind::Sqlite => { - if config.database_url.contains(":memory:") { - 128 - } else { - 16 - } - } - AnyKind::Mssql => 100, - } - }) + .max_connections(max_connections(config, kind)) .idle_timeout(config.database_connection_idle_timeout) .max_lifetime(config.database_connection_max_lifetime) .acquire_timeout(Duration::from_secs_f64( config.database_connection_acquire_timeout_seconds, )); - pool_options = add_on_return_to_pool(config, pool_options); - pool_options = add_on_connection_handler(config, pool_options); + pool_options = add_on_return_to_pool(pool_options, on_reset); + pool_options = add_on_connection_handler(config, pool_options, on_connect); pool_options } } -fn add_on_return_to_pool(config: &AppConfig, pool_options: PoolOptions) -> PoolOptions { - let on_disconnect_file = config.configuration_directory.join(ON_RESET_FILE); - let sql = if on_disconnect_file.exists() { - log::info!( - "Creating a custom SQL connection cleanup handler from {}", - on_disconnect_file.display() - ); - match std::fs::read_to_string(&on_disconnect_file) { - Ok(sql) => { - log::trace!("The custom SQL connection cleanup handler is:\n{sql}"); - Some(std::sync::Arc::new(sql)) - } +pub(super) async fn retry_connecting( + config: &AppConfig, + mut connect: impl FnMut() -> F, +) -> anyhow::Result +where + F: Future>, + E: std::error::Error + Send + Sync + 'static, +{ + let mut retries = config.database_connection_retries; + loop { + match connect().await { + Ok(connection) => return Ok(connection), Err(e) => { - log::error!( - "Unable to read the file {}: {}", - on_disconnect_file.display(), - e - ); - None + if retries == 0 { + return Err(anyhow::Error::new(e).context(format!( + "Unable to open connection to {}", + config.database_url + ))); + } + log::warn!("Failed to connect to the database: {e:#}. Retrying in 5 seconds."); + retries -= 1; + tokio::time::sleep(Duration::from_secs(5)).await; } } - } else { + } +} + +pub(super) fn max_connections(config: &AppConfig, kind: AnyKind) -> u32 { + if let Some(max) = config.max_database_pool_connections { + return max; + } + // Different databases have a different number of max concurrent connections allowed by default + match kind { + AnyKind::Postgres | AnyKind::Odbc => 50, // Default to PostgreSQL-like limits for Generic + AnyKind::MySql => 75, + AnyKind::Sqlite => { + if config.database_url.contains(":memory:") { + 128 + } else { + 16 + } + } + AnyKind::Mssql => 100, + } +} + +pub(super) struct ConnectionHooks { + pub(super) on_connect: Option>, + pub(super) on_reset: Option>, +} + +impl ConnectionHooks { + pub(super) fn read(config: &AppConfig) -> Self { + Self { + on_connect: read_hook(config, ON_CONNECT_FILE, "database connection handler"), + on_reset: read_hook(config, ON_RESET_FILE, "connection cleanup handler"), + } + } +} + +fn read_hook(config: &AppConfig, file_name: &str, handler: &str) -> Option> { + let hook_file = config.configuration_directory.join(file_name); + if !hook_file.exists() { log::debug!( - "Not creating a custom SQL connection cleanup handler because {} does not exist", - on_disconnect_file.display() + "Not creating a custom SQL {handler} because {} does not exist", + hook_file.display() ); - None - }; + return None; + } + log::info!( + "Creating a custom SQL {handler} from {}", + hook_file.display() + ); + match std::fs::read_to_string(&hook_file) { + Ok(sql) => { + log::trace!("The custom SQL {handler} is:\n{sql}"); + Some(Arc::new(sql)) + } + Err(e) => { + log::error!("Unable to read the file {}: {}", hook_file.display(), e); + None + } + } +} +fn add_on_return_to_pool( + pool_options: PoolOptions, + sql: Option>, +) -> PoolOptions { pool_options.after_release(move |conn, meta| { let sql = sql.clone(); Box::pin(async move { @@ -146,7 +193,7 @@ fn add_on_return_to_pool(config: &AppConfig, pool_options: PoolOptions) -> fn on_return_to_pool( conn: &mut AnyConnection, meta: sqlx::pool::PoolConnectionMetadata, - sql: std::sync::Arc, + sql: Arc, ) -> BoxFuture<'_, Result> { use sqlx::row::Row; Box::pin(async move { @@ -166,36 +213,13 @@ fn on_return_to_pool( fn add_on_connection_handler( config: &AppConfig, pool_options: PoolOptions, + sql: Option>, ) -> PoolOptions { - let on_connect_file = config.configuration_directory.join(ON_CONNECT_FILE); - let on_connect_file_display = on_connect_file.display().to_string(); - let sql = if on_connect_file.exists() { - log::info!( - "Creating a custom SQL database connection handler from {}", - on_connect_file.display() - ); - match std::fs::read_to_string(&on_connect_file) { - Ok(sql) => { - log::trace!("The custom SQL database connection handler is:\n{sql}"); - Some(std::sync::Arc::new(sql)) - } - Err(e) => { - log::error!( - "Unable to read the file {}: {}", - on_connect_file.display(), - e - ); - None - } - } - } else { - log::debug!( - "Not creating a custom SQL database connection handler because {} does not exist", - on_connect_file.display() - ); - None - }; - + let on_connect_file_display = config + .configuration_directory + .join(ON_CONNECT_FILE) + .display() + .to_string(); pool_options.after_connect(move |conn, _| { let sql = sql.clone(); let on_connect_file_display = on_connect_file_display.clone(); @@ -300,7 +324,7 @@ mod tests { .await .unwrap(); let value: String = sqlx::query::query("SELECT value FROM on_connect_marker") - .fetch_one(&db.connection) + .fetch_one(db.connection.as_old_api().unwrap()) .await .unwrap() .try_get(0) @@ -329,7 +353,7 @@ mod tests { let db = Database::init(&config).await.unwrap(); drop(db.connection.acquire().await.unwrap()); let resets: i64 = sqlx::query::query("SELECT COUNT(*) FROM reset_log") - .fetch_one(&db.connection) + .fetch_one(db.connection.as_old_api().unwrap()) .await .unwrap() .try_get(0) diff --git a/src/webserver/database/driver.rs b/src/webserver/database/driver.rs new file mode 100644 index 000000000..e13937c2c --- /dev/null +++ b/src/webserver/database/driver.rs @@ -0,0 +1,262 @@ +use futures_util::StreamExt; +use futures_util::stream::BoxStream; +use serde_json::Value; +use sqlx::Either; +use sqlx::any::{Any, AnyArguments, AnyPool, AnyQueryResult, AnyRow}; +use sqlx::arguments::Arguments; +use sqlx::column::Column; +use sqlx::executor::{Execute, Executor}; +use sqlx::pool::PoolConnection; +use sqlx::row::Row; +use sqlx::value::ValueRef; + +#[cfg(feature = "unstable-sqlx-next")] +use super::sqlx_next; + +#[derive(Clone, Debug)] +pub enum DatabasePool { + OldApi(AnyPool), + #[cfg(feature = "unstable-sqlx-next")] + SqlxNext(sqlx_next::Pool), +} + +#[cfg_attr( + feature = "unstable-sqlx-next", + allow( + clippy::large_enum_variant, + reason = "boxing the default driver's connection would allocate on every acquire" + ) +)] +pub enum PooledConnection { + OldApi(PoolConnection), + #[cfg(feature = "unstable-sqlx-next")] + SqlxNext(sqlx_next::PooledConnection), +} + +/// One physical result of a statement, with each column already decoded to JSON. +pub(super) enum FetchedItem { + Row(Vec<(String, Value)>), + FinishedQuery, +} + +#[derive(Debug)] +pub enum DriverError { + OldApi(sqlx::error::Error), + #[cfg(feature = "unstable-sqlx-next")] + SqlxNext(sqlx_next::Error), +} + +impl DatabasePool { + pub(super) async fn acquire(&self) -> anyhow::Result { + match self { + Self::OldApi(pool) => Ok(PooledConnection::OldApi(pool.acquire().await?)), + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(pool) => Ok(PooledConnection::SqlxNext(pool.acquire().await?)), + } + } + + #[must_use] + pub fn size(&self) -> u32 { + match self { + Self::OldApi(pool) => pool.size(), + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(pool) => pool.size(), + } + } + + #[must_use] + pub fn num_idle(&self) -> usize { + match self { + Self::OldApi(pool) => pool.num_idle(), + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(pool) => pool.num_idle(), + } + } + + pub(super) async fn close(&self) { + match self { + Self::OldApi(pool) => pool.close().await, + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(pool) => pool.close().await, + } + } + + #[must_use] + pub fn as_old_api(&self) -> Option<&AnyPool> { + match self { + Self::OldApi(pool) => Some(pool), + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(_) => None, + } + } +} + +impl PooledConnection { + /// Statements without parameters use the simple query protocol, so they may hold several statements. + pub(super) fn fetch_many<'c>( + &'c mut self, + sql: &'c str, + params: &'c [Option], + ) -> BoxStream<'c, Result> { + match self { + Self::OldApi(connection) => connection + .fetch_many(OldApiStatement { sql, params }) + .map(|result| { + result + .map(fetched_from_old_api) + .map_err(DriverError::OldApi) + }) + .boxed(), + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(connection) => connection + .fetch_many(sql, params) + .map(|result| result.map_err(DriverError::SqlxNext)) + .boxed(), + } + } + + pub(super) async fn execute(&mut self, sql: &str) -> Result<(), DriverError> { + match self { + Self::OldApi(connection) => connection + .execute(sql) + .await + .map(drop) + .map_err(DriverError::OldApi), + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(connection) => { + connection.execute(sql).await.map_err(DriverError::SqlxNext) + } + } + } + + pub fn as_old_api(&mut self) -> Option<&mut PoolConnection> { + match self { + Self::OldApi(connection) => Some(connection), + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(_) => None, + } + } +} + +pub(crate) fn unsupported_by_sqlx_next(capability: &str) -> anyhow::Error { + anyhow::anyhow!( + "The unstable sqlx 0.9 database drivers do not support {capability} yet. \ + Set unstable_sqlx_next to false to use it." + ) +} + +#[must_use] +pub fn is_pool_timeout(error: &anyhow::Error) -> bool { + #[cfg(feature = "unstable-sqlx-next")] + if let Some(sqlx_next::Error::PoolTimedOut) = error.downcast_ref() { + return true; + } + matches!(error.downcast_ref(), Some(sqlx::error::Error::PoolTimedOut)) +} + +impl DriverError { + /// The byte offset in the SQL text where the database located the error. + pub(super) fn offset(&self) -> Option { + match self { + Self::OldApi(sqlx::error::Error::Database(error)) => error.offset(), + Self::OldApi(_) => None, + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(_) => None, + } + } +} + +impl From for DriverError { + fn from(error: sqlx::error::Error) -> Self { + Self::OldApi(error) + } +} + +impl std::fmt::Display for DriverError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::OldApi(error) => error.fmt(f), + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(error) => error.fmt(f), + } + } +} + +impl std::error::Error for DriverError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::OldApi(error) => error.source(), + #[cfg(feature = "unstable-sqlx-next")] + Self::SqlxNext(error) => error.source(), + } + } +} + +fn fetched_from_old_api(result: Either) -> FetchedItem { + match result { + Either::Left(result) => { + log::debug!("Finished query with result: {result:?}"); + FetchedItem::FinishedQuery + } + Either::Right(row) => { + if log::log_enabled!(log::Level::Trace) { + trace_old_api_row(&row); + } + FetchedItem::Row(super::sql_to_json::decode_columns(&row)) + } + } +} + +fn trace_old_api_row(r: &AnyRow) { + use std::fmt::Write; + let columns = r.columns(); + let mut row_str = String::new(); + for (i, col) in columns.iter().enumerate() { + if let Ok(value) = r.try_get_raw(i) { + write!( + &mut row_str, + "[{:?} ({}): {:?}: {:?}]", + col.name(), + if value.is_null() { "NULL" } else { "NOT NULL" }, + col, + value.type_info() + ) + .unwrap(); + } + } + log::trace!("Received db row: {row_str}"); +} + +struct OldApiStatement<'q> { + sql: &'q str, + params: &'q [Option], +} + +impl<'q> Execute<'q, Any> for OldApiStatement<'q> { + fn sql(&self) -> &'q str { + self.sql + } + + fn statement(&self) -> Option<&>::Statement> { + None + } + + fn take_arguments(&mut self) -> Option<>::Arguments> { + if self.params.is_empty() { + return None; + } + let mut arguments = AnyArguments::default(); + for param in self.params { + match param { + None => arguments.add(None::), + Some(value) => arguments.add(value.as_str()), + } + } + Some(arguments) + } + + fn persistent(&self) -> bool { + // Let sqlx create a prepared statement the first time it is executed, and then reuse it. + true + } +} diff --git a/src/webserver/database/error_highlighting.rs b/src/webserver/database/error_highlighting.rs index b457ec995..eb58901ac 100644 --- a/src/webserver/database/error_highlighting.rs +++ b/src/webserver/database/error_highlighting.rs @@ -3,6 +3,7 @@ use std::{ path::{Path, PathBuf}, }; +use super::driver::DriverError; use super::sql::SourceSpan; #[derive(Debug)] @@ -10,7 +11,7 @@ struct NiceDatabaseError { /// The source file that contains the query. source_file: PathBuf, /// The error that occurred. - db_err: sqlx::error::Error, + db_err: DriverError, /// The query that was executed. query: String, /// The start location of the query in the source file, if the query was extracted from a larger file. @@ -43,27 +44,21 @@ impl std::fmt::Display for NiceDatabaseError { self.source_file.display(), self.db_err )?; - if let sqlx::error::Error::Database(db_err) = &self.db_err { - let Some(mut offset) = db_err.offset() else { - write!(f, "{}", self.query)?; - self.show_position_info(f)?; - return Ok(()); - }; - for line in self.query.lines() { - if offset > line.len() { - offset -= line.len() + 1; - } else { - highlight_line_offset(f, line, offset); - self.show_position_info(f)?; - break; - } - } - Ok(()) - } else { + let Some(mut offset) = self.db_err.offset() else { write!(f, "{}", self.query)?; self.show_position_info(f)?; - Ok(()) + return Ok(()); + }; + for line in self.query.lines() { + if offset > line.len() { + offset -= line.len() + 1; + } else { + highlight_line_offset(f, line, offset); + self.show_position_info(f)?; + break; + } } + Ok(()) } } @@ -104,11 +99,11 @@ pub(super) fn is_positioned_error(error: &anyhow::Error) -> bool { pub(super) fn display_db_error( source_file: &Path, query: &str, - db_err: sqlx::error::Error, + db_err: impl Into, ) -> anyhow::Error { anyhow::Error::new(NiceDatabaseError { source_file: source_file.to_path_buf(), - db_err, + db_err: db_err.into(), query: query.to_string(), query_position: None, }) @@ -120,7 +115,7 @@ pub(super) fn display_stmt_db_error( source_file: &Path, query: &str, query_position: SourceSpan, - db_err: sqlx::error::Error, + db_err: DriverError, ) -> anyhow::Error { anyhow::Error::new(NiceDatabaseError { source_file: source_file.to_path_buf(), diff --git a/src/webserver/database/execute_queries.rs b/src/webserver/database/execute_queries.rs index 475380d44..373148601 100644 --- a/src/webserver/database/execute_queries.rs +++ b/src/webserver/database/execute_queries.rs @@ -19,20 +19,13 @@ use crate::webserver::ErrorWithStatus; use crate::webserver::http_request_info::ExecutionContext; use crate::webserver::single_or_vec::SingleOrVec; +use super::driver::{DriverError, FetchedItem, PooledConnection, unsupported_by_sqlx_next}; use super::{Database, DbItem, ScalarSubqueryBehavior, error_highlighting::display_db_error}; -use sqlx::Either; -use sqlx::any::{ - Any, AnyArguments, AnyConnection, AnyQueryResult, AnyRow, AnyStatement, AnyTypeInfo, -}; -use sqlx::arguments::Arguments; -use sqlx::column::Column; -use sqlx::executor::{Execute, Executor}; -use sqlx::pool::PoolConnection; -use sqlx::row::Row as _; +use sqlx::any::{AnyStatement, AnyTypeInfo}; +use sqlx::executor::Executor; use sqlx::statement::Statement; -use sqlx::value::ValueRef; -pub type DbConn = Option>; +pub type DbConn = Option; /// Database events before computed columns and dynamic components are evaluated. enum QueryEvent { @@ -191,6 +184,8 @@ impl Database { param_types: &[AnyTypeInfo], ) -> anyhow::Result> { self.connection + .as_old_api() + .ok_or_else(|| unsupported_by_sqlx_next("database-backed files"))? .prepare_with(query, param_types) .await .map(|s| s.to_owned()) @@ -209,7 +204,9 @@ pub fn stream_query_results_with_conn<'a>( for res in &sql_file.statements { match res { FileStatement::CsvImport(csv_import) => { - let connection = take_connection(&request.app_state.db, db_connection, request).await?; + let connection = take_connection(&request.app_state.db, db_connection, request).await? + .as_old_api() + .ok_or_else(|| unsupported_by_sqlx_next("CSV import"))?; log::debug!("Executing CSV import: {csv_import:?}"); run_csv_import(connection, csv_import, request).await.with_context(|| format!("Failed to import the CSV file {:?} into the table {:?}", csv_import.uploaded_file, csv_import.table_name))?; }, @@ -235,7 +232,7 @@ pub fn stream_query_results_with_conn<'a>( let mut deferred_query_results = Vec::new(); { let connection = take_connection(&request.app_state.db, db_connection, request).await?; - let mut stream = connection.fetch_many(query); + let mut stream = connection.fetch_many(query.sql, &query.param_values); loop { let Some(result) = next_query_event( &mut stream, &mut query_metrics, &mut returned_rows, @@ -358,10 +355,10 @@ async fn execute_static_simple_select( Ok(Value::Object(map)) } -async fn try_rollback_transaction(db_connection: &mut AnyConnection) { +async fn try_rollback_transaction(db_connection: &mut PooledConnection) { log::debug!("Attempting to rollback transaction"); match db_connection.execute("ROLLBACK").await { - Ok(_) => log::debug!("Rolled back transaction"), + Ok(()) => log::debug!("Rolled back transaction"), Err(e) => { log::debug!("There was probably no transaction in progress when this happened: {e:?}"); } @@ -430,7 +427,7 @@ async fn execute_scalar_query<'a>( .scalar_subquery_behavior(); { let connection = take_connection(&request.app_state.db, db_connection, request).await?; - let mut stream = connection.fetch_many(query); + let mut stream = connection.fetch_many(query.sql, &query.param_values); loop { let Some(result) = next_query_event( &mut stream, @@ -501,7 +498,7 @@ async fn execute_scalar_query<'a>( } async fn record_error_and_rollback( - connection: &mut AnyConnection, + connection: &mut PooledConnection, query_metrics: &DbQueryMetricsContext<'_>, returned_rows: i64, error: anyhow::Error, @@ -546,7 +543,7 @@ async fn take_connection<'a>( db: &'a Database, conn: &'a mut DbConn, request: &ExecutionContext, -) -> anyhow::Result<&'a mut PoolConnection> { +) -> anyhow::Result<&'a mut PooledConnection> { if let Some(c) = conn { return Ok(c); } @@ -567,12 +564,12 @@ async fn take_connection<'a>( Ok(connection) } Err(e) => { - let db_name = db.connection.any_kind(); + let db_name = db.info.kind; let active_count = db.connection.size(); let err_msg = format!( "Unable to connect to {db_name:?}. The connection pool currently has {active_count} active connections." ); - Err(anyhow::Error::new(e).context(err_msg)) + Err(e.context(err_msg)) } } } @@ -580,7 +577,7 @@ async fn take_connection<'a>( /// Sets the current `OTel` trace context on the database connection so it is visible /// in `pg_stat_activity.application_name` (`PostgreSQL`) or as a session variable (`MySQL`). /// This allows correlating `SQLPage` traces with database-side monitoring. -async fn set_trace_context(connection: &mut AnyConnection, db: &Database) { +async fn set_trace_context(connection: &mut PooledConnection, db: &Database) { use opentelemetry::trace::TraceContextExt; use tracing_opentelemetry::OpenTelemetrySpanExt; @@ -614,7 +611,7 @@ async fn set_trace_context(connection: &mut AnyConnection, db: &Database) { /// Poll and decode one physical result while measuring only database fetch time. async fn next_query_event( - stream: &mut (impl Stream>> + Unpin), + stream: &mut (impl Stream> + Unpin), metrics: &mut DbQueryMetricsContext<'_>, returned_rows: &mut i64, source_file: &Path, @@ -636,27 +633,22 @@ fn parse_single_sql_result( source_file: &Path, query: &DatabaseQuery, source_span: SourceSpan, - res: sqlx::error::Result>, + res: Result, ) -> anyhow::Result { match res { - Ok(Either::Right(r)) => { - if log::log_enabled!(log::Level::Trace) { - debug_row(&r); - } + Ok(FetchedItem::Row(columns)) => { + let column_count = columns.len(); let (row, mut inputs) = - super::sql_to_json::row_to_json_with_inputs(&r, query.row_input_json.len())?; + super::sql_to_json::row_to_json_with_inputs(columns, query.row_input_json.len())?; decode_json_values(&mut inputs, &query.row_input_json); Ok(QueryEvent::Row(QueryRow { value: row, inputs: RowInputs::new(inputs), - output_column_count: r.columns().len() - query.row_input_json.len() + output_column_count: column_count - query.row_input_json.len() + query.computed_columns.len(), })) } - Ok(Either::Left(res)) => { - log::debug!("Finished query with result: {res:?}"); - Ok(QueryEvent::FinishedQuery) - } + Ok(FetchedItem::FinishedQuery) => Ok(QueryEvent::FinishedQuery), Err(err) => Err(display_stmt_db_error( source_file, &query.sql, @@ -666,26 +658,6 @@ fn parse_single_sql_result( } } -fn debug_row(r: &AnyRow) { - use std::fmt::Write; - let columns = r.columns(); - let mut row_str = String::new(); - for (i, col) in columns.iter().enumerate() { - if let Ok(value) = r.try_get_raw(i) { - write!( - &mut row_str, - "[{:?} ({}): {:?}: {:?}]", - col.name(), - if value.is_null() { "NULL" } else { "NOT NULL" }, - col, - value.type_info() - ) - .unwrap(); - } - } - log::trace!("Received db row: {row_str}"); -} - fn clone_anyhow_err(source_file: &Path, err: &anyhow::Error) -> anyhow::Error { let mut e = anyhow!( "{} contains a syntax error preventing SQLPage from parsing and preparing its SQL statements.", @@ -704,7 +676,6 @@ async fn bind_query<'a>( ) -> anyhow::Result> { let sql = query.sql.as_str(); log::debug!("Preparing statement: {sql}"); - let mut arguments = AnyArguments::default(); let mut param_values = Vec::with_capacity(query.bindings.len()); let mut inputs = NoInputs; for (param_idx, binding) in query.bindings.iter().enumerate() { @@ -718,20 +689,9 @@ async fn bind_query<'a>( param_idx + 1, argument.as_ref().unwrap_or(&Cow::Borrowed("NULL")) ); - param_values.push(argument.as_deref().map(str::to_owned)); - match argument { - None => arguments.add(None::), - Some(Cow::Owned(s)) => arguments.add(s), - Some(Cow::Borrowed(v)) => arguments.add(v), - } + param_values.push(argument.map(Cow::into_owned)); } - let has_arguments = !query.bindings.is_empty(); - Ok(BoundQuery { - sql, - arguments, - has_arguments, - param_values, - }) + Ok(BoundQuery { sql, param_values }) } async fn prepare_query_event( @@ -809,37 +769,12 @@ fn decode_json_values(values: &mut [Value], json_flags: &[bool]) { } } -/// Rewritten SQL and evaluated arguments in the form consumed by `sqlx`. +/// Rewritten SQL and the evaluated values of its parameters. pub struct BoundQuery<'a> { sql: &'a str, - arguments: AnyArguments<'a>, - has_arguments: bool, param_values: Vec>, } -impl<'q> Execute<'q, Any> for BoundQuery<'q> { - fn sql(&self) -> &'q str { - self.sql - } - - fn statement(&self) -> Option<&>::Statement> { - None - } - - fn take_arguments(&mut self) -> Option<>::Arguments> { - if self.has_arguments { - Some(std::mem::take(&mut self.arguments)) - } else { - None - } - } - - fn persistent(&self) -> bool { - // Let sqlx create a prepared statement the first time it is executed, and then reuse it. - true - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/src/webserver/database/migrations.rs b/src/webserver/database/migrations.rs index d850dc37d..d480e6ce7 100644 --- a/src/webserver/database/migrations.rs +++ b/src/webserver/database/migrations.rs @@ -1,4 +1,5 @@ use super::Database; +use super::driver::unsupported_by_sqlx_next; use super::error_highlighting::display_db_error; use crate::MIGRATIONS_DIR; use anyhow; @@ -34,7 +35,11 @@ pub async fn apply(config: &crate::app_config::AppConfig, db: &Database) -> anyh for m in migrator.iter() { log::info!("\t{}", DisplayMigration(m)); } - migrator.run(&db.connection).await.map_err(|err| { + let pool = db + .connection + .as_old_api() + .ok_or_else(|| unsupported_by_sqlx_next("migrations"))?; + migrator.run(pool).await.map_err(|err| { match err { MigrateError::Execute(n, source) => { let migration = failing_migration(&migrator.migrations, n) diff --git a/src/webserver/database/mod.rs b/src/webserver/database/mod.rs index b8c2d6153..dddc5389e 100644 --- a/src/webserver/database/mod.rs +++ b/src/webserver/database/mod.rs @@ -1,6 +1,7 @@ pub mod blob_to_data_url; mod connect; mod csv_import; +pub mod driver; pub mod execute_queries; pub mod migrations; mod sql; @@ -9,6 +10,8 @@ mod sqlpage_functions; mod error_highlighting; mod sql_to_json; +#[cfg(feature = "unstable-sqlx-next")] +mod sqlx_next; pub use sql::SqlFile; use sqlx::any::AnyKind; @@ -125,7 +128,7 @@ impl From for SupportedDatabase { } pub struct Database { - pub connection: sqlx::any::AnyPool, + pub connection: driver::DatabasePool, pub info: DbInfo, } @@ -138,6 +141,16 @@ pub struct DbInfo { pub kind: AnyKind, } +impl DbInfo { + fn new(dbms_name: String, kind: AnyKind) -> Self { + Self { + database_type: SupportedDatabase::from_dbms_name(&dbms_name), + dbms_name, + kind, + } + } +} + impl Database { pub async fn close(&self) -> anyhow::Result<()> { log::info!("Closing all database connections..."); @@ -155,7 +168,7 @@ pub enum DbItem { impl std::fmt::Display for Database { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{:?}", self.connection.any_kind()) + write!(f, "{:?}", self.info.kind) } } diff --git a/src/webserver/database/sql_to_json.rs b/src/webserver/database/sql_to_json.rs index 359471d2a..dbf81fa24 100644 --- a/src/webserver/database/sql_to_json.rs +++ b/src/webserver/database/sql_to_json.rs @@ -44,46 +44,51 @@ pub(super) fn row_to_json(row: &AnyRow) -> Value { Object(map) } -/// Decodes a row into user-visible columns and a private trailing input suffix. +pub(super) fn decode_columns(row: &AnyRow) -> Vec<(String, Value)> { + row.columns() + .iter() + .map(|column| (canonical_col_name(column), sql_to_json(row, column))) + .collect() +} + +/// Splits decoded columns into user-visible columns and a private trailing input suffix. /// -/// Every SQL value is decoded exactly once. Private values are addressed by -/// ordinal, so their generated SQL aliases cannot collide with user columns. +/// Private values are addressed by ordinal, so their generated SQL aliases +/// cannot collide with user columns. pub(super) fn row_to_json_with_inputs( - row: &AnyRow, + mut columns: Vec<(String, Value)>, input_count: usize, ) -> anyhow::Result<(Value, Vec)> { - let columns = row.columns(); let public_count = columns .len() .checked_sub(input_count) .ok_or_else(|| anyhow::anyhow!("The query returned fewer columns than SQLPage expected"))?; - let mut map = Map::new(); - let mut inputs = Vec::with_capacity(input_count); - for column in &columns[..public_count] { - let key = canonical_col_name(column); - let value = sql_to_json(row, column); - map = add_value_to_map(map, (key, value)); - } - for column in &columns[public_count..] { - inputs.push(sql_to_json(row, column)); - } + let inputs = columns + .split_off(public_count) + .into_iter() + .map(|(_, value)| value) + .collect(); + let map = columns.into_iter().fold(Map::new(), add_value_to_map); Ok((Value::Object(map), inputs)) } fn canonical_col_name(col: &AnyColumn) -> String { - // Some databases fold all unquoted identifiers to uppercase but SQLPage uses lowercase property names - if matches!(col.type_info().0, AnyTypeInfoKind::Odbc(_)) - && col - .name() - .chars() - .all(|c| c.is_ascii_uppercase() || c == '_') - { - col.name().to_ascii_lowercase() + if matches!(col.type_info().0, AnyTypeInfoKind::Odbc(_)) { + odbc_column_name(col.name()) } else { col.name().to_owned() } } +/// Some databases fold all unquoted identifiers to uppercase but `SQLPage` uses lowercase property names. +pub(super) fn odbc_column_name(name: &str) -> String { + if name.chars().all(|c| c.is_ascii_uppercase() || c == '_') { + name.to_ascii_lowercase() + } else { + name.to_owned() + } +} + pub(super) fn sql_to_json(row: &AnyRow, col: &AnyColumn) -> Value { let raw_value_result = row.try_get_raw(col.ordinal()); match raw_value_result { @@ -137,7 +142,7 @@ where } } -fn decimal_to_json(decimal: &BigDecimal) -> Value { +pub(super) fn decimal_to_json(decimal: &BigDecimal) -> Value { // to_plain_string always returns a valid JSON string Value::Number(serde_json::Number::from_string_unchecked( decimal.normalized().to_plain_string(), @@ -263,7 +268,7 @@ mod tests { ) .fetch_one(&mut connection) .await?; - let (public, inputs) = row_to_json_with_inputs(&row, 1)?; + let (public, inputs) = row_to_json_with_inputs(decode_columns(&row), 1)?; expect_json_object_equal( &public, &serde_json::json!({ "__sqlpage_input_0": "public" }), diff --git a/src/webserver/database/sqlx_next/mod.rs b/src/webserver/database/sqlx_next/mod.rs new file mode 100644 index 000000000..7fdffd6f6 --- /dev/null +++ b/src/webserver/database/sqlx_next/mod.rs @@ -0,0 +1,216 @@ +mod odbc; + +use std::time::Duration; + +use futures_util::stream::BoxStream; +use futures_util::{StreamExt, TryStreamExt}; +use serde_json::Value; +use sqlx::any::AnyKind; +use sqlx_next::database::Database; +use sqlx_next::decode::Decode; +use sqlx_next::encode::Encode; +use sqlx_next::error::BoxDynError; +use sqlx_next::pool::{PoolConnection, PoolOptions}; +use sqlx_next::{ + Arguments, AssertSqlSafe, ColumnIndex, Either, Execute, Executor, Row, SqlSafeStr, SqlStr, Type, +}; +use sqlx_odbc::Odbc; + +use super::connect::{ConnectionHooks, max_connections}; +use super::driver::FetchedItem; +use crate::app_config::AppConfig; + +pub(super) use sqlx_next::Error; + +#[derive(Clone, Debug)] +pub enum Pool { + Odbc(sqlx_odbc::OdbcPool), +} + +pub enum PooledConnection { + Odbc(PoolConnection), +} + +#[derive(Clone, Copy)] +pub(super) enum Driver { + Odbc, +} + +impl Driver { + pub(super) fn serving(kind: AnyKind) -> Option { + match kind { + AnyKind::Odbc => Some(Self::Odbc), + AnyKind::Postgres | AnyKind::MySql | AnyKind::Sqlite | AnyKind::Mssql => None, + } + } + + /// Returns the pool and the name the database management system reports for itself. + pub(super) async fn connect(self, config: &AppConfig) -> anyhow::Result<(Pool, String)> { + match self { + Self::Odbc => odbc::connect(config).await, + } + } +} + +impl Pool { + pub(super) async fn acquire(&self) -> Result { + Ok(match self { + Self::Odbc(pool) => PooledConnection::Odbc(pool.acquire().await?), + }) + } + + pub(super) fn size(&self) -> u32 { + match self { + Self::Odbc(pool) => pool.size(), + } + } + + pub(super) fn num_idle(&self) -> usize { + match self { + Self::Odbc(pool) => pool.num_idle(), + } + } + + pub(super) async fn close(&self) { + match self { + Self::Odbc(pool) => pool.close().await, + } + } +} + +impl PooledConnection { + pub(super) fn fetch_many<'c>( + &'c mut self, + sql: &'c str, + params: &'c [Option], + ) -> BoxStream<'c, Result> { + match self { + Self::Odbc(connection) => { + fetch_decoded::(&mut **connection, sql, params, odbc::decode_row) + } + } + } + + pub(super) async fn execute(&mut self, sql: &str) -> Result<(), Error> { + let sql = AssertSqlSafe(sql.to_owned()); + match self { + Self::Odbc(connection) => connection.execute(sql).await.map(drop), + } + } +} + +fn fetch_decoded<'c, DB>( + connection: &'c mut DB::Connection, + sql: &'c str, + params: &'c [Option], + decode_row: fn(&DB::Row) -> Vec<(String, Value)>, +) -> BoxStream<'c, Result> +where + DB: Database, + for<'e> &'e mut DB::Connection: Executor<'e, Database = DB>, + DB::QueryResult: std::fmt::Debug, + for<'q> Option<&'q str>: Encode<'q, DB> + Type, +{ + connection + .fetch_many(BoundStatement { sql, params }) + .map_ok(move |result| match result { + Either::Left(result) => { + log::debug!("Finished query with result: {result:?}"); + FetchedItem::FinishedQuery + } + Either::Right(row) => FetchedItem::Row(decode_row(&row)), + }) + .boxed() +} + +struct BoundStatement<'q> { + sql: &'q str, + params: &'q [Option], +} + +impl<'q, DB> Execute<'q, DB> for BoundStatement<'q> +where + DB: Database, + for<'a> Option<&'a str>: Encode<'a, DB> + Type, +{ + fn sql(self) -> SqlStr { + AssertSqlSafe(self.sql).into_sql_str() + } + + fn statement(&self) -> Option<&DB::Statement> { + None + } + + fn take_arguments(&mut self) -> Result, BoxDynError> { + if self.params.is_empty() { + return Ok(None); + } + let mut arguments = DB::Arguments::default(); + for param in self.params { + arguments.add(param.as_deref())?; + } + Ok(Some(arguments)) + } + + fn persistent(&self) -> bool { + true + } +} + +fn pool_options(config: &AppConfig, kind: AnyKind) -> PoolOptions +where + DB: Database, + for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, + bool: Type + for<'r> Decode<'r, DB>, + usize: ColumnIndex, + DB::QueryResult: std::fmt::Debug, +{ + let ConnectionHooks { + on_connect, + on_reset, + } = ConnectionHooks::read(config); + PoolOptions::new() + .max_connections(max_connections(config, kind)) + .idle_timeout(config.database_connection_idle_timeout) + .max_lifetime(config.database_connection_max_lifetime) + .acquire_timeout(Duration::from_secs_f64( + config.database_connection_acquire_timeout_seconds, + )) + .after_connect(move |connection: &mut DB::Connection, _| { + let sql = on_connect.clone(); + Box::pin(async move { + if let Some(sql) = sql { + let result = connection.execute(AssertSqlSafe(sql)).await?; + log::debug!( + "Finished running connection handler on new connection: {result:?}" + ); + } + Ok(()) + }) + }) + .after_release(move |connection: &mut DB::Connection, metadata| { + let sql = on_reset.clone(); + Box::pin(async move { + let Some(sql) = sql else { + return Ok(true); + }; + log::trace!("Running the custom SQL connection cleanup handler. {metadata:?}"); + match connection.fetch_optional(AssertSqlSafe(sql)).await? { + Some(row) => row.try_get::(0), + None => Ok(true), + } + }) + }) +} + +fn decode_or_default<'r, DB, T>(value: DB::ValueRef<'r>) -> T +where + DB: Database, + T: Decode<'r, DB> + Default, +{ + T::decode(value).unwrap_or_else(|error| { + let type_name = std::any::type_name::(); + log::error!("Failed to decode {type_name} value: {error}"); + T::default() + }) +} diff --git a/src/webserver/database/sqlx_next/odbc.rs b/src/webserver/database/sqlx_next/odbc.rs new file mode 100644 index 000000000..768768761 --- /dev/null +++ b/src/webserver/database/sqlx_next/odbc.rs @@ -0,0 +1,74 @@ +use std::time::Duration; + +use anyhow::Context; +use bigdecimal::BigDecimal; +use chrono::{DateTime, FixedOffset, NaiveDate, NaiveTime}; +use serde_json::Value; +use sqlx::any::AnyKind; +use sqlx_next::{Column, ConnectOptions, Row, TypeInfo, ValueRef}; +use sqlx_odbc::{Odbc, OdbcColumn, OdbcConnectOptions, OdbcRow}; + +use super::{Pool, decode_or_default, pool_options}; +use crate::app_config::AppConfig; +use crate::webserver::database::blob_to_data_url::vec_to_data_uri_value; +use crate::webserver::database::connect::retry_connecting; +use crate::webserver::database::sql_to_json::{decimal_to_json, odbc_column_name}; + +pub(super) async fn connect(config: &AppConfig) -> anyhow::Result<(Pool, String)> { + let mut options: OdbcConnectOptions = config.database_url.parse()?; + options + .batch_size(config.max_pending_rows.clamp(1, 1024)) + .max_column_size(None) + .log_statements(log::LevelFilter::Trace) + .log_slow_statements(log::LevelFilter::Warn, Duration::from_millis(250)); + if config.database_password.is_some() { + log::warn!( + "Setting a password for an ODBC connection is not supported via separate config; include credentials in the DSN or connection string" + ); + } + let dbms_name = retry_connecting(config, || options.connect()) + .await? + .dbms_name()?; + let pool = pool_options::(config, AnyKind::Odbc) + .connect_with(options) + .await + .with_context(|| format!("Unable to open connection pool to {}", config.database_url))?; + Ok((Pool::Odbc(pool), dbms_name)) +} + +pub(super) fn decode_row(row: &OdbcRow) -> Vec<(String, Value)> { + row.columns() + .iter() + .map(|column| (odbc_column_name(column.name()), column_to_json(row, column))) + .collect() +} + +fn column_to_json(row: &OdbcRow, column: &OdbcColumn) -> Value { + let value = match row.try_get_raw(column.ordinal()) { + Ok(value) if value.is_null() => return Value::Null, + Ok(value) => value, + Err(error) => { + log::warn!("Unable to extract value from row: {error:?}"); + return Value::Null; + } + }; + match column.type_info().name() { + "REAL" | "FLOAT" | "DOUBLE" => decode_or_default::(value).into(), + "NUMERIC" | "DECIMAL" => decimal_to_json(&decode_or_default::(value)), + "BIGINT" => decode_or_default::(value).into(), + "INTEGER" => decode_or_default::(value).into(), + "SMALLINT" | "TINYINT" => decode_or_default::(value).into(), + "BIT" => decode_or_default::(value).into(), + "DATE" => decode_or_default::(value) + .to_string() + .into(), + "TIME" => decode_or_default::(value) + .to_string() + .into(), + "TIMESTAMP" => decode_or_default::>(value) + .to_rfc3339() + .into(), + "VARBINARY" | "BINARY" => vec_to_data_uri_value(&decode_or_default::>(value)), + _ => decode_or_default::(value).into(), + } +} diff --git a/src/webserver/error.rs b/src/webserver/error.rs index a1ea8954f..c4ea86704 100644 --- a/src/webserver/error.rs +++ b/src/webserver/error.rs @@ -15,6 +15,7 @@ use std::path::PathBuf; use crate::AppState; use crate::app_config::DevOrProd; use crate::webserver::ErrorWithStatus; +use crate::webserver::database::driver::is_pool_timeout; use actix_web::HttpResponseBuilder; use actix_web::error::UrlencodedError; use actix_web::http::{StatusCode, header}; @@ -183,7 +184,7 @@ pub(super) fn anyhow_err_to_actix_resp(e: &anyhow::Error, state: &AppState) -> H "Basic realm=\"Authentication required\", charset=\"UTF-8\"", )); } - } else if let Some(sqlx::error::Error::PoolTimedOut) = e.downcast_ref() { + } else if is_pool_timeout(e) { use rand::RngExt; resp.status(StatusCode::TOO_MANY_REQUESTS).insert_header(( header::RETRY_AFTER, @@ -211,7 +212,7 @@ pub(super) fn anyhow_err_to_actix_resp(e: &anyhow::Error, state: &AppState) -> H fn anyhow_error_status(e: &anyhow::Error) -> Option { if let Some(&ErrorWithStatus { status }) = e.downcast_ref() { Some(status) - } else if let Some(sqlx::error::Error::PoolTimedOut) = e.downcast_ref() { + } else if is_pool_timeout(e) { Some(StatusCode::TOO_MANY_REQUESTS) } else { None diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 088b99beb..63694862a 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -12,7 +12,7 @@ use actix_web::{ }; use sqlpage::{ AppState, - app_config::{AppConfig, test_database_url}, + app_config::{AppConfig, test_database_url, test_unstable_sqlx_next}, telemetry, webserver::http::create_app, }; @@ -112,6 +112,7 @@ pub(crate) async fn make_app_data() -> Data { pub(crate) fn test_config() -> AppConfig { serde_json::from_value(serde_json::json!({ "database_url": test_database_url(), + "unstable_sqlx_next": test_unstable_sqlx_next(), "max_database_pool_connections": 1, "database_connection_retries": 3, "database_connection_acquire_timeout_seconds": 15, diff --git a/tests/configuration_docs/mod.rs b/tests/configuration_docs/mod.rs index 244b07e6e..54be29dce 100644 --- a/tests/configuration_docs/mod.rs +++ b/tests/configuration_docs/mod.rs @@ -35,10 +35,15 @@ fn app_config_option_names() -> BTreeSet { collector .names .iter() + .filter(|name| !is_unreleased(name)) .map(|name| (*name).to_owned()) .collect() } +fn is_unreleased(option_name: &str) -> bool { + option_name.starts_with("unstable_") +} + /// Serde hands the names it expects, renames included, to `deserialize_struct`. #[derive(Default)] struct OptionNameCollector { diff --git a/tests/core/mod.rs b/tests/core/mod.rs index 592360646..4a094601b 100644 --- a/tests/core/mod.rs +++ b/tests/core/mod.rs @@ -17,22 +17,16 @@ async fn store_file_in_db(state: &AppState, path: &str, contents: &[u8]) { let create_table_sql = sqlpage::filesystem::DbFsQueries::get_create_table_sql(state.db.info.database_type); - if state - .db - .connection + let pool = state.db.connection.as_old_api().unwrap(); + if pool .execute("SELECT 1 FROM sqlpage_files WHERE 1 = 0") .await .is_err() { - state.db.connection.execute(create_table_sql).await.unwrap(); + pool.execute(create_table_sql).await.unwrap(); } let delete_sql = format!("DELETE FROM sqlpage_files WHERE path = '{path}'"); - state - .db - .connection - .execute(delete_sql.as_str()) - .await - .unwrap(); + pool.execute(delete_sql.as_str()).await.unwrap(); let insert_sql = format!( "INSERT INTO sqlpage_files(path, contents) VALUES ({}, {})", make_placeholder(state.db.info.kind, 1), @@ -41,7 +35,7 @@ async fn store_file_in_db(state: &AppState, path: &str, contents: &[u8]) { sqlx::query::query(&insert_sql) .bind(path) .bind(contents) - .execute(&state.db.connection) + .execute(pool) .await .unwrap(); } diff --git a/tests/parameter_binding/mod.rs b/tests/parameter_binding/mod.rs index bee65962c..c0ab530fa 100644 --- a/tests/parameter_binding/mod.rs +++ b/tests/parameter_binding/mod.rs @@ -26,7 +26,14 @@ async fn test_parameterized_pages_leave_a_prepared_statement_in_the_cache() -> a ); } - let connection = data.db.connection.acquire().await.unwrap(); + let connection = data + .db + .connection + .as_old_api() + .unwrap() + .acquire() + .await + .unwrap(); assert!( connection.cached_statements_size() > 0, "{:?} ran a parameterized query three times without caching a prepared statement",