Preserve SQLite streaming in native drivers

Author: lovasoa

Cargo.lock

@@ -58,16 +58,17 @@ actix-web-httpauth = "0.8.0"
 rand = "0.10.0"
 actix-multipart = "0.7.2"
 base64 = "0.22"
+bytes = "1"
 hmac = "0.13"
 sha2 = "0.11"
-rustls-acme = "0.15"
+rustls-acme = { version = "0.15", default-features = false, features = ["ring", "tls12", "webpki-roots"] }
 dotenvy = "0.15.7"
 csv-async = { version = "1.2.6", features = ["tokio"] }
-rustls = { version = "0.23" } # keep in sync with actix-web, awc, and rustls-acme
+rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] } # keep in sync with actix-web, awc, and rustls-acme
 rustls-native-certs = "0.8.1"
 awc = { version = "3", features = ["rustls-0_23-webpki-roots"] }
 clap = { version = "4.5.17", features = ["derive"] }
-tokio-util = "0.7.12"
+tokio-util = { version = "0.7.12", features = ["compat"] }
 openidconnect = { version = "4.0.0", default-features = false, features = ["accept-rfc3339-timestamps"] }
 encoding_rs = "0.8.35"
 regex = "1"
@@ -96,7 +97,7 @@ tokio = { version = "1", features = ["rt", "time", "test-util"] }
 
 [build-dependencies]
 awc = { version = "3", features = ["rustls-0_23-webpki-roots"] }
-rustls = "0.23"
+rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] }
 actix-rt = "2.8"
 libflate = "2"
 futures-util = "0.3.21"

Cargo.toml

@@ -12,7 +12,7 @@ use std::time::Duration;
 
 #[actix_rt::main]
 async fn main() {
-    rustls::crypto::aws_lc_rs::default_provider()
+    rustls::crypto::ring::default_provider()
         .install_default()
         .unwrap();
 

build.rs

@@ -1,7 +1,7 @@
 use crate::webserver::ErrorWithStatus;
 use crate::webserver::database::SupportedDatabase;
-use crate::webserver::{Database, StatusCodeResultExt, make_placeholder};
 use crate::webserver::database::{DbParam, driver::DbValue};
+use crate::webserver::{Database, StatusCodeResultExt, make_placeholder};
 use crate::{AppState, TEMPLATES_DIR};
 use anyhow::Context;
 use chrono::{DateTime, Utc};
@@ -267,9 +267,9 @@ impl DbFsQueries {
         log::debug!("Initializing database filesystem queries");
         Self::check_table_available(db).await?;
         Ok(Self {
-            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?,
+            was_modified: Self::make_was_modified_query(db),
+            read_file: Self::make_read_file_query(db),
+            exists: Self::make_exists_query(db),
         })
     }
 
@@ -284,31 +284,31 @@ impl DbFsQueries {
         Ok(())
     }
 
-    async fn make_was_modified_query(db: &Database) -> anyhow::Result<String> {
+    fn make_was_modified_query(db: &Database) -> String {
         let was_modified_query = format!(
             "SELECT 1 from sqlpage_files WHERE last_modified >= {} AND path = {}",
             make_placeholder(db.info.kind, 1),
             make_placeholder(db.info.kind, 2)
         );
         log::debug!("Preparing the database filesystem was_modified_query: {was_modified_query}");
-        Ok(was_modified_query)
+        was_modified_query
     }
 
-    async fn make_read_file_query(db: &Database) -> anyhow::Result<String> {
+    fn make_read_file_query(db: &Database) -> String {
         let read_file_query = format!(
             "SELECT contents from sqlpage_files WHERE path = {}",
             make_placeholder(db.info.kind, 1),
         );
         log::debug!("Preparing the database filesystem read_file_query: {read_file_query}");
-        Ok(read_file_query)
+        read_file_query
     }
 
-    async fn make_exists_query(db: &Database) -> anyhow::Result<String> {
+    fn make_exists_query(db: &Database) -> String {
         let exists_query = format!(
             "SELECT 1 from sqlpage_files WHERE path = {}",
             make_placeholder(db.info.kind, 1),
         );
-        Ok(exists_query)
+        exists_query
     }
 
     async fn file_modified_since_in_db(
@@ -351,26 +351,26 @@ impl DbFsQueries {
         log::debug!("Reading file {} from the database", path.display());
         let mut conn = app_state.db.connection.acquire().await?;
         conn.fetch_optional(
-                &self.read_file,
-                &[DbParam::Text(path.display().to_string())],
-            )
-            .await
-            .map_err(anyhow::Error::from)
-            .and_then(|row| {
-                if let Some(row) = row {
-                    match row.values.first() {
-                        Some(DbValue::Bytes(bytes)) => Ok(bytes.clone()),
-                        Some(DbValue::Text(text)) => Ok(text.as_bytes().to_vec()),
-                        _ => Ok(Vec::new()),
-                    }
-                } else {
-                    Err(ErrorWithStatus {
-                        status: actix_web::http::StatusCode::NOT_FOUND,
-                    }
-                    .into())
+            &self.read_file,
+            &[DbParam::Text(path.display().to_string())],
+        )
+        .await
+        .map_err(anyhow::Error::from)
+        .and_then(|row| {
+            if let Some(row) = row {
+                match row.values.first() {
+                    Some(DbValue::Bytes(bytes)) => Ok(bytes.clone()),
+                    Some(DbValue::Text(text)) => Ok(text.as_bytes().to_vec()),
+                    _ => Ok(Vec::new()),
+                }
+            } else {
+                Err(ErrorWithStatus {
+                    status: actix_web::http::StatusCode::NOT_FOUND,
                 }
-            })
-            .with_context(|| format!("Unable to read {} from the database", path.display()))
+                .into())
+            }
+        })
+        .with_context(|| format!("Unable to read {} from the database", path.display()))
     }
 
     async fn file_exists(&self, app_state: &AppState, path: &Path) -> anyhow::Result<bool> {
@@ -412,9 +412,9 @@ 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 mut conn = db.connection.acquire().await?;
-    conn.execute_command("DROP TABLE IF EXISTS sqlpage_files", &[]).await?;
+    conn.execute_command("DROP TABLE IF EXISTS sqlpage_files", &[])
+        .await?;
     log::debug!("Creating table sqlpage_files: {create_table_sql}");
     conn.execute_command(create_table_sql, &[]).await?;
 
@@ -425,13 +425,13 @@ async fn test_sql_file_read_utf8() -> anyhow::Result<()> {
         make_placeholder(dbms, 2)
     );
     conn.execute_command(
-            &insert_sql,
-            &[
-                DbParam::Text("unit test file.txt".into()),
-                DbParam::Bytes("Héllö world! 😀".as_bytes().to_vec()),
-            ],
-        )
-        .await?;
+        &insert_sql,
+        &[
+            DbParam::Text("unit test file.txt".into()),
+            DbParam::Bytes("Héllö world! 😀".as_bytes().to_vec()),
+        ],
+    )
+    .await?;
 
     let fs = FileSystem::init("/", db).await;
     let actual = fs

src/filesystem.rs

@@ -37,7 +37,7 @@
 //! When processing a request, `SQLPage`:
 //!
 //! 1. Parses the SQL using sqlparser-rs. Once a SQL file is parsed, it is cached for later reuse.
-//! 2. Executes queries through sqlx.
+//! 2. Executes queries through native database drivers.
 //! 3. Finds the requested component's handlebars template in the database or in the filesystem.
 //! 4. Maps results to the component template, using handlebars-rs.
 //! 5. Streams rendered HTML to the client.

src/lib.rs

@@ -1,8 +1,8 @@
+use crate::webserver::database::DbPool;
 use opentelemetry::global;
 use opentelemetry::metrics::{Histogram, ObservableGauge};
 use opentelemetry_semantic_conventions::attribute as otel;
 use opentelemetry_semantic_conventions::metric as otel_metric;
-use crate::webserver::database::DbPool;
 
 pub struct TelemetryMetrics {
     pub http_request_duration: Histogram<f64>,

src/telemetry_metrics.rs

@@ -23,8 +23,8 @@ impl Database {
         }
         log::debug!("Connecting to a {db_kind:?} database on {database_url}");
         let on_connect_sql = read_connection_handler(config, ON_CONNECT_FILE);
-        let _on_reset_sql = read_connection_handler(config, ON_RESET_FILE);
-        if _on_reset_sql.is_some() {
+        let on_reset_sql = read_connection_handler(config, ON_RESET_FILE);
+        if on_reset_sql.is_some() {
             log::warn!(
                 "{ON_RESET_FILE} is currently ignored by the native driver pool because connections are not reused yet"
             );
@@ -111,7 +111,10 @@ fn read_connection_handler(config: &AppConfig, file_name: &str) -> Option<String
         );
         return None;
     }
-    log::info!("Creating a custom SQL connection handler from {}", file.display());
+    log::info!(
+        "Creating a custom SQL connection handler from {}",
+        file.display()
+    );
     match std::fs::read_to_string(&file) {
         Ok(sql) => {
             log::trace!("The custom SQL connection handler is:\n{sql}");

src/webserver/database/connect.rs

@@ -324,7 +324,9 @@ async fn test_end_to_end() {
             uploaded_file: "my_file.csv".into(),
         }
     );
-    let db = crate::webserver::Database::init(&test_config()).await.unwrap();
+    let db = crate::webserver::Database::init(&test_config())
+        .await
+        .unwrap();
     let mut conn = db.connection.acquire().await.unwrap();
     conn.execute_command("CREATE TABLE my_table (col1 TEXT, col2 TEXT)", &[])
         .await
@@ -339,8 +341,14 @@ async fn test_end_to_end() {
         .into_iter()
         .filter_map(|item| match item {
             super::driver::DbStatementResult::Row(row) => Some((
-                match &row.values[0] { super::driver::DbValue::Text(s) => s.clone(), other => format!("{other:?}") },
-                match &row.values[1] { super::driver::DbValue::Text(s) => s.clone(), other => format!("{other:?}") },
+                match &row.values[0] {
+                    super::driver::DbValue::Text(s) => s.clone(),
+                    other => format!("{other:?}"),
+                },
+                match &row.values[1] {
+                    super::driver::DbValue::Text(s) => s.clone(),
+                    other => format!("{other:?}"),
+                },
             )),
             super::driver::DbStatementResult::Finished => None,
         })