fix #711

Author: lovasoa

CHANGELOG.md

@@ -1,5 +1,25 @@
 # CHANGELOG.md
 
+## 0.32.0 (unreleased)
+
+- Rollback any open transactions when returning a connection to the pool.
+  - Previously, if an error occurred in the middle of a transaction, the transaction would be left open, and the connection would be returned to the pool. The next request could get a connection with an open half-completed transaction, which could lead to hard to debug issues.
+  - This allows safely using features that require a transaction, like
+    - ```sql
+      BEGIN;
+      CREATE TEMPORARY TABLE t (x int) ON COMMIT DROP; -- postgres syntax
+      -- do something with t
+      -- previously, if an error occurred, the transaction would be left open, and the connection returned to the pool.
+      -- the next request could get a connection where the table `t` still exists, leading to a new error.
+      COMMIT;
+      ```
+    - This will now automatically rollback the transaction, even if an error occurs in the middle of it.
+- Fix a bug where one additional SQL statement was executed after an error occurred in a SQL file. This could cause surprising unexpected behavior.
+  - ```sql
+    insert into t values ($invalid_value); -- if this statement fails, ...
+    insert into t values (42); -- this next statement should not be executed
+    ```
+
 ## 0.31.0 (2024-11-24)
 
 ### 🚀 **New Features**

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "sqlpage"
-version = "0.31.0"
+version = "0.32.0"
 edition = "2021"
 description = "Build data user interfaces entirely in SQL. A web server that takes .sql files and formats the query result using pre-made configurable professional-looking components."
 keywords = ["web", "sql", "framework"]

Cargo.toml

@@ -3,6 +3,7 @@ use std::{mem::take, time::Duration};
 use super::Database;
 use crate::{app_config::AppConfig, ON_CONNECT_FILE};
 use anyhow::Context;
+use futures_util::future::BoxFuture;
 use sqlx::{
     any::{Any, AnyConnectOptions, AnyKind},
     pool::PoolOptions,
@@ -93,12 +94,32 @@ impl Database {
             )
             .acquire_timeout(Duration::from_secs_f64(
                 config.database_connection_acquire_timeout_seconds,
-            ));
+            ))
+            .after_release(on_return_to_pool);
         pool_options = add_on_connection_handler(config, pool_options);
         pool_options
     }
 }
 
+fn on_return_to_pool(
+    conn: &mut sqlx::AnyConnection,
+    meta: sqlx::pool::PoolConnectionMetadata,
+) -> BoxFuture<'_, Result<bool, sqlx::Error>> {
+    Box::pin(async move {
+        match conn.execute("ROLLBACK").await {
+            Ok(r) => log::info!(
+                "Rolled back a transaction that was left open before returning a connection to the pool. Result: {:?}",
+                r
+            ),
+            Err(e) => log::trace!(
+                "Failed to rollback before returning a connection to the pool. There was probably no transaction left open: {e:?}"
+            ),
+        }
+        log::trace!("Releasing connection: {meta:#?}");
+        Ok(true)
+    })
+}
+
 fn add_on_connection_handler(
     config: &AppConfig,
     pool_options: PoolOptions<Any>,

src/webserver/database/connect.rs

@@ -89,19 +89,38 @@ pub fn stream_query_results_with_conn<'a>(
     .map(|res| res.unwrap_or_else(DbItem::Error))
 }
 
+/// Transforms a stream of database items to stop processing after encountering the first error.
+/// The error item itself is still emitted before stopping.
 pub fn stop_at_first_error(
     results_stream: impl Stream<Item = DbItem>,
 ) -> impl Stream<Item = DbItem> {
-    let mut has_error = false;
-    results_stream.take_while(move |item| {
-        // We stop the stream AFTER the first error, so that the error is still returned to the client, but the rest of the queries are not executed.
-        let should_continue = !has_error;
-        if let DbItem::Error(err) = item {
-            log::error!("{err:?}");
-            has_error = true;
+    // We need a oneshot channel rather than a simple boolean flag because
+    // take_while would poll the stream one extra time after the error,
+    // while take_until stops immediately when the future completes
+    let (error_tx, error_rx) = tokio::sync::oneshot::channel();
+    let mut error_tx = Some(error_tx);
+
+    results_stream
+        .inspect(move |item| {
+            if let DbItem::Error(err) = item {
+                log::error!("{err:?}");
+                if let Some(tx) = error_tx.take() {
+                    let _ = tx.send(());
+                }
+            }
+        })
+        .take_until(error_rx)
+}
+
+pub(crate) async fn rollback_transaction(db_connection: &mut DbConn) {
+    if let Some(conn) = db_connection.as_mut() {
+        match conn.execute("ROLLBACK").await {
+            Ok(r) => log::debug!("Rolled back transaction with result: {:?}", r),
+            Err(e) => log::debug!("Failed to rollback transaction: {e:?}"),
         }
-        futures_util::future::ready(should_continue)
-    })
+    } else {
+        log::debug!("No connection to rollback a transaction");
+    }
 }
 
 /// Executes the sqlpage pseudo-functions contained in a static simple select

src/webserver/database/execute_queries.rs

@@ -1,6 +1,6 @@
 use crate::render::{AnyRenderBodyContext, HeaderContext, PageContext};
 use crate::webserver::content_security_policy::ContentSecurityPolicy;
-use crate::webserver::database::execute_queries::stop_at_first_error;
+use crate::webserver::database::execute_queries::{rollback_transaction, stop_at_first_error};
 use crate::webserver::database::{execute_queries::stream_query_results_with_conn, DbItem};
 use crate::webserver::http_request_info::extract_request_info;
 use crate::webserver::ErrorWithStatus;
@@ -173,32 +173,35 @@ async fn render_sql(
         let database_entries_stream =
             stream_query_results_with_conn(&sql_file, &mut req_param, &mut conn);
         let database_entries_stream = stop_at_first_error(database_entries_stream);
-        let response_with_writer = build_response_header_and_stream(
-            Arc::clone(&app_state),
-            database_entries_stream,
-            request_context,
-        )
-        .await;
-        match response_with_writer {
-            Ok(ResponseWithWriter::RenderStream {
-                http_response,
-                renderer,
+        {
+            let response_with_writer = build_response_header_and_stream(
+                Arc::clone(&app_state),
                 database_entries_stream,
-            }) => {
-                resp_send
-                    .send(http_response)
-                    .unwrap_or_else(|e| log::error!("could not send headers {e:?}"));
-                stream_response(database_entries_stream, renderer).await;
-            }
-            Ok(ResponseWithWriter::FinishedResponse { http_response }) => {
-                resp_send
-                    .send(http_response)
-                    .unwrap_or_else(|e| log::error!("could not send headers {e:?}"));
-            }
-            Err(err) => {
-                send_anyhow_error(&err, resp_send, app_state.config.environment);
+                request_context,
+            )
+            .await;
+            match response_with_writer {
+                Ok(ResponseWithWriter::RenderStream {
+                    http_response,
+                    renderer,
+                    database_entries_stream,
+                }) => {
+                    resp_send
+                        .send(http_response)
+                        .unwrap_or_else(|e| log::error!("could not send headers {e:?}"));
+                    stream_response(database_entries_stream, renderer).await;
+                }
+                Ok(ResponseWithWriter::FinishedResponse { http_response }) => {
+                    resp_send
+                        .send(http_response)
+                        .unwrap_or_else(|e| log::error!("could not send headers {e:?}"));
+                }
+                Err(err) => {
+                    send_anyhow_error(&err, resp_send, app_state.config.environment);
+                }
             }
         }
+        rollback_transaction(&mut conn).await;
     });
     resp_recv.await.map_err(ErrorInternalServerError)
 }

src/webserver/http.rs

@@ -0,0 +1,6 @@
+BEGIN;
+CREATE TEMPORARY TABLE t(f INTEGER NOT NULL);
+INSERT INTO t(f) VALUES ($x);
+COMMIT;
+
+select 'text' as component, f as contents from t;

tests/failed_transaction.sql

@@ -469,6 +469,37 @@ async fn test_csv_upload() -> actix_web::Result<()> {
     Ok(())
 }
 
+#[actix_web::test]
+async fn test_transaction_error() -> actix_web::Result<()> {
+    // First, request the page without any parameter. It should fail because
+    // of the not null constraint.
+    // But then, when we request again with a parameter, we should not see any side
+    // effect coming from the first transaction, and it should succeed
+    let data = make_app_data().await;
+    let req = get_request_to_with_data("/tests/failed_transaction.sql", data.clone())
+        .await?
+        .to_srv_request();
+    let resp = main_handler(req).await?;
+    let body = test::read_body(resp).await;
+    let body_str = String::from_utf8(body.to_vec()).unwrap();
+    assert!(
+        body_str.contains("constraint failed"),
+        "{body_str}\nexpected to contain: constraint failed"
+    );
+    // Now query again, with ?x=1447
+    let req = get_request_to_with_data("/tests/failed_transaction.sql?x=1447", data)
+        .await?
+        .to_srv_request();
+    let resp = main_handler(req).await?;
+    let body = test::read_body(resp).await;
+    let body_str = String::from_utf8(body.to_vec()).unwrap();
+    assert!(
+        body_str.contains("1447"),
+        "{body_str}\nexpected to contain: 1447"
+    );
+    Ok(())
+}
+
 #[actix_web::test]
 /// `/sqlpage/migrations/0001_init.sql` should return a 403 Forbidden
 async fn privileged_paths_are_not_accessible() {
@@ -561,8 +592,10 @@ async fn test_official_website_basic_auth_example() {
     );
 }
 
-async fn get_request_to(path: &str) -> actix_web::Result<TestRequest> {
-    let data = make_app_data().await;
+async fn get_request_to_with_data(
+    path: &str,
+    data: actix_web::web::Data<AppState>,
+) -> actix_web::Result<TestRequest> {
     Ok(test::TestRequest::get()
         .uri(path)
         .insert_header(ContentType::plaintext())
@@ -571,6 +604,11 @@ async fn get_request_to(path: &str) -> actix_web::Result<TestRequest> {
         .app_data(data))
 }
 
+async fn get_request_to(path: &str) -> actix_web::Result<TestRequest> {
+    let data = make_app_data().await;
+    get_request_to_with_data(path, data).await
+}
+
 async fn make_app_data_from_config(config: AppConfig) -> actix_web::web::Data<AppState> {
     let state = AppState::init(&config).await.unwrap();
     actix_web::web::Data::new(state)
@@ -614,6 +652,7 @@ pub fn test_config() -> AppConfig {
     serde_json::from_str::<AppConfig>(&format!(
         r#"{{
         "database_url": "{}",
+        "max_database_pool_connections": 1,
         "database_connection_retries": 3,
         "database_connection_acquire_timeout_seconds": 15,
         "allow_exec": true,