diff --git a/src/filesystem.rs b/src/filesystem.rs index 524241c8..b2759d3f 100644 --- a/src/filesystem.rs +++ b/src/filesystem.rs @@ -4,7 +4,7 @@ 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 +85,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 +141,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 +199,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 +311,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 +338,18 @@ 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().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 +391,6 @@ impl DbFsQueries { async fn file_modified_since_in_db( &self, - app_state: &AppState, path: &Path, since: DateTime, ) -> anyhow::Result { @@ -409,15 +408,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 +421,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 +442,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 +455,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 +484,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(); // 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 9b2bf6e5..d77251e2 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 3c0e4203..65d7af24 100644 --- a/src/webserver/database/connect.rs +++ b/src/webserver/database/connect.rs @@ -1,6 +1,7 @@ use std::{future::Future, mem::take, sync::Arc, time::Duration}; use super::Database; +use super::driver::DatabasePool; use crate::{ ON_CONNECT_FILE, ON_RESET_FILE, app_config::AppConfig, @@ -49,7 +50,7 @@ impl Database { log::debug!("Initialized {dbms_name:?} database pool: {pool:#?}"); Ok(Database { - connection: pool, + connection: DatabasePool::OldApi(pool), info: DbInfo { dbms_name, database_type, @@ -311,7 +312,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()) .await .unwrap() .try_get(0) @@ -340,7 +341,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()) .await .unwrap() .try_get(0) diff --git a/src/webserver/database/driver.rs b/src/webserver/database/driver.rs new file mode 100644 index 00000000..81972475 --- /dev/null +++ b/src/webserver/database/driver.rs @@ -0,0 +1,208 @@ +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; + +#[derive(Clone, Debug)] +pub enum DatabasePool { + OldApi(AnyPool), +} + +pub enum PooledConnection { + OldApi(PoolConnection), +} + +/// 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), +} + +impl DatabasePool { + pub(super) async fn acquire(&self) -> anyhow::Result { + match self { + Self::OldApi(pool) => Ok(PooledConnection::OldApi(pool.acquire().await?)), + } + } + + #[must_use] + pub fn size(&self) -> u32 { + match self { + Self::OldApi(pool) => pool.size(), + } + } + + #[must_use] + pub fn num_idle(&self) -> usize { + match self { + Self::OldApi(pool) => pool.num_idle(), + } + } + + pub(super) async fn close(&self) { + match self { + Self::OldApi(pool) => pool.close().await, + } + } + + #[must_use] + pub fn as_old_api(&self) -> &AnyPool { + match self { + Self::OldApi(pool) => pool, + } + } +} + +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(), + } + } + + 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), + } + } + + pub fn as_old_api(&mut self) -> &mut PoolConnection { + match self { + Self::OldApi(connection) => connection, + } + } +} + +#[must_use] +pub fn is_pool_timeout(error: &anyhow::Error) -> bool { + 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, + } + } +} + +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), + } + } +} + +impl std::error::Error for DriverError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::OldApi(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 b457ec99..eb58901a 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 475380d4..ee893f5e 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}; 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,7 @@ impl Database { param_types: &[AnyTypeInfo], ) -> anyhow::Result> { self.connection + .as_old_api() .prepare_with(query, param_types) .await .map(|s| s.to_owned()) @@ -209,7 +203,8 @@ 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(); 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 +230,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 +353,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 +425,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 +496,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 +541,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 +562,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 +575,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 +609,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 +631,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 +656,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 +674,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 +687,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 +767,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 d850dc37..a76416b5 100644 --- a/src/webserver/database/migrations.rs +++ b/src/webserver/database/migrations.rs @@ -34,29 +34,32 @@ 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| { - match err { - MigrateError::Execute(n, source) => { - let migration = failing_migration(&migrator.migrations, n) - .expect("sqlx reports the version of a migration it just ran"); - let source_file = migrations_dir.join(format!( - "{:04}_{}{}", - n, - migration.description.replace(' ', "_"), - migration.migration_type.suffix() - )); - display_db_error(&source_file, &migration.sql, source).context(format!( - "Failed to apply {} migration {}", - db, - DisplayMigration(migration) - )) + migrator + .run(db.connection.as_old_api()) + .await + .map_err(|err| { + match err { + MigrateError::Execute(n, source) => { + let migration = failing_migration(&migrator.migrations, n) + .expect("sqlx reports the version of a migration it just ran"); + let source_file = migrations_dir.join(format!( + "{:04}_{}{}", + n, + migration.description.replace(' ', "_"), + migration.migration_type.suffix() + )); + display_db_error(&source_file, &migration.sql, source).context(format!( + "Failed to apply {} migration {}", + db, + DisplayMigration(migration) + )) + } + source => anyhow::Error::new(source), } - source => anyhow::Error::new(source), - } - .context(format!( - "Failed to apply database migrations from {MIGRATIONS_DIR:?}" - )) - })?; + .context(format!( + "Failed to apply database migrations from {MIGRATIONS_DIR:?}" + )) + })?; Ok(()) } diff --git a/src/webserver/database/mod.rs b/src/webserver/database/mod.rs index b8c2d615..9625f3cb 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; @@ -125,7 +126,7 @@ impl From for SupportedDatabase { } pub struct Database { - pub connection: sqlx::any::AnyPool, + pub connection: driver::DatabasePool, pub info: DbInfo, } @@ -155,7 +156,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 359471d2..7fec8d07 100644 --- a/src/webserver/database/sql_to_json.rs +++ b/src/webserver/database/sql_to_json.rs @@ -44,29 +44,31 @@ 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)) } @@ -263,7 +265,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/error.rs b/src/webserver/error.rs index a1ea8954..c4ea8670 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/core/mod.rs b/tests/core/mod.rs index 59236064..4a39f9e2 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(); + 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 bee65962..2d029bd4 100644 --- a/tests/parameter_binding/mod.rs +++ b/tests/parameter_binding/mod.rs @@ -26,7 +26,7 @@ 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().acquire().await.unwrap(); assert!( connection.cached_statements_size() > 0, "{:?} ran a parameterized query three times without caching a prepared statement",