wip

Author: tobyhede

.tool-versions

@@ -309,6 +309,8 @@ echo
 mise --env tcp run postgres:setup
 mise --env tls run postgres:setup
 
+mise run test:integration:showcase
+
 echo
 echo '###############################################'
 echo '# Test: Prometheus'
@@ -377,10 +379,8 @@ echo '###############################################'
 echo '# Test: Showcase'
 echo '###############################################'
 echo
-mise --env tls run proxy:up proxy-tls --extra-args "--detach --wait"
-mise --env tls run test:wait_for_postgres_to_quack --port 6432 --max-retries 20 --tls
-RUST_BACKTRACE=full cargo run -p showcase
-mise --env tls run proxy:down
+
+mise run test:integration:showcase
 
 echo
 echo '###############################################'
@@ -637,6 +637,15 @@ else
 fi
 """
 
+[tasks."test:integration:showcase"]
+description = "Run Showcase integration test"
+run = """
+mise --env tls run proxy:up proxy-tls --extra-args "--detach --wait"
+mise --env tls run test:wait_for_postgres_to_quack --port 6432 --max-retries 20 --tls
+RUST_BACKTRACE=full cargo run -p showcase
+mise --env tls run proxy:down
+"""
+
 [tasks.release]
 description = "Publish release artifacts"
 depends = ["release:docker"]

mise.toml

@@ -24,6 +24,7 @@ use std::{
     sync::{Arc, LazyLock, RwLock},
     time::{Duration, Instant},
 };
+use tokio::sync::oneshot;
 use tracing::{debug, error, warn};
 use uuid::Uuid;
 
@@ -311,6 +312,10 @@ where
     }
 
     pub fn set_schema_changed(&self) {
+        debug!(target: CONTEXT,
+            client_id = self.client_id,
+            msg = "Schema changed"
+        );
         let _ = self.schema_changed.write().map(|mut guard| *guard = true);
     }
 
@@ -510,15 +515,38 @@ where
     }
 
     pub async fn reload_schema(&self) {
-        match self.reload_sender.send(ReloadCommand::DatabaseSchema) {
-            Ok(_) => self.set_schema_changed(),
+        let (responder, receiver) = oneshot::channel();
+
+        // let _ = self
+        //     .reload_sender
+        //     .send(ReloadCommand::DatabaseSchema(responder))
+        //     .inspect_err(|err| {
+        //         // Error means a fatal internal error in send.
+        //         // No recovery really possible
+        //         // Log because may break subsequent statements and lead to confusion
+        //         error!(
+        //             target: CONTEXT,
+        //             msg = "Database schema could not be reloaded",
+        //             error = err.to_string()
+        //         )
+        //     });
+
+        match self
+            .reload_sender
+            .send(ReloadCommand::DatabaseSchema(responder))
+        {
+            Ok(_) => (),
             Err(err) => {
                 error!(
                     msg = "Database schema could not be reloaded",
                     error = err.to_string()
                 );
             }
         }
+
+        debug!(target: CONTEXT, msg = "Waiting for schema reload");
+        let response = receiver.await;
+        debug!(target: CONTEXT, msg = "Database schema reloaded", ?response);
     }
 
     pub fn is_passthrough(&self) -> bool {

packages/cipherstash-proxy/src/postgresql/context/mod.rs

@@ -746,10 +746,6 @@ where
         let schema_changed = eql_mapper::collect_ddl(self.context.get_table_resolver(), statement);
 
         if schema_changed {
-            debug!(target: MAPPER,
-                client_id = self.context.client_id,
-                msg = "schema changed"
-            );
             self.context.set_schema_changed();
         }
     }

packages/cipherstash-proxy/src/postgresql/frontend.rs

@@ -9,7 +9,8 @@ use crate::{
 };
 use cipherstash_client::encryption::Plaintext;
 use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
-use tracing::warn;
+use tokio::sync::oneshot::Sender;
+use tracing::{debug, warn};
 
 mod encrypt_config;
 mod schema;
@@ -22,6 +23,8 @@ pub type ReloadSender = UnboundedSender<ReloadCommand>;
 
 type ReloadReceiver = UnboundedReceiver<ReloadCommand>;
 
+pub type ReloadResponder = Sender<()>;
+
 /// SQL Statement for loading encrypt configuration from database
 const ENCRYPT_CONFIG_QUERY: &str = include_str!("./sql/select_config.sql");
 
@@ -31,10 +34,10 @@ const SCHEMA_QUERY: &str = include_str!("./sql/select_table_schemas.sql");
 /// SQL Statement for loading aggregates as part of database schema
 const AGGREGATE_QUERY: &str = include_str!("./sql/select_aggregates.sql");
 
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug)]
 pub enum ReloadCommand {
-    DatabaseSchema,
-    EncryptSchema,
+    DatabaseSchema(ReloadResponder),
+    EncryptSchema(ReloadResponder),
 }
 
 ///
@@ -48,7 +51,6 @@ pub struct Proxy {
     pub eql_version: Option<String>,
     zerokms: ZeroKms,
     reload_sender: ReloadSender,
-    reload_receiver: ReloadReceiver,
 }
 
 impl Proxy {
@@ -59,22 +61,27 @@ impl Proxy {
         // Ensures error on start if credential or network issue
         zerokms.init_cipher(None).await?;
 
-        let encrypt_config = EncryptConfigManager::init(&config.database).await?;
+        let encrypt_config_manager = EncryptConfigManager::init(&config.database).await?;
 
-        let schema = SchemaManager::init(&config.database).await?;
+        let schema_manager = SchemaManager::init(&config.database).await?;
 
         let eql_version = Proxy::eql_version(&config).await?;
 
         let (reload_sender, reload_receiver) = mpsc::unbounded_channel();
 
+        Proxy::receive(
+            reload_receiver,
+            schema_manager.clone(),
+            encrypt_config_manager.clone(),
+        );
+
         Ok(Proxy {
             config: Arc::new(config),
             zerokms,
-            encrypt_config_manager: encrypt_config,
-            schema_manager: schema,
+            encrypt_config_manager,
+            schema_manager,
             eql_version,
             reload_sender,
-            reload_receiver,
         })
     }
 
@@ -97,13 +104,27 @@ impl Proxy {
         Ok(version)
     }
 
-    pub async fn receive(&mut self) {
-        while let Some(command) = self.reload_receiver.recv().await {
-            match command {
-                ReloadCommand::DatabaseSchema => self.schema_manager.reload().await,
-                ReloadCommand::EncryptSchema => self.encrypt_config_manager.reload().await,
+    pub fn receive(
+        mut reload_receiver: ReloadReceiver,
+        schema_manager: SchemaManager,
+        encrypt_config_manager: EncryptConfigManager,
+    ) {
+        tokio::task::spawn(async move {
+            while let Some(command) = reload_receiver.recv().await {
+                debug!(msg = "ReloadCommand received", ?command);
+                match command {
+                    ReloadCommand::DatabaseSchema(responder) => {
+                        schema_manager.reload().await;
+                        encrypt_config_manager.reload().await;
+                        let _ = responder.send(());
+                    }
+                    ReloadCommand::EncryptSchema(responder) => {
+                        encrypt_config_manager.reload().await;
+                        let _ = responder.send(());
+                    }
+                }
             }
-        }
+        });
     }
 
     ///

packages/cipherstash-proxy/src/proxy/mod.rs

@@ -8,6 +8,7 @@ use crate::{
 };
 
 pub async fn insert_test_data() {
+    println!("Insert test data");
     let medications = [
         Medication::new(
             "550e8400-e29b-41d4-a716-446655440001",
@@ -496,8 +497,6 @@ pub async fn clear() {
     //
     // Deleting rows from the eql_v2_configuration table is not officially supported due to the risk of data loss.
     //
-    // TODO: EQL should support safe removal of config rows - at least in some kind of "test" or non-production
-    // mode.
     let sql = r#"
         DELETE
           FROM public.eql_v2_configuration

packages/showcase/src/data.rs

@@ -66,17 +66,18 @@ use crate::{
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    println!("🩺 Healthcare Database Showcase - EQL v2 Searchable Encryption");
+    println!("============================================================");
+
     trace();
     clear().await;
+
     setup_schema().await;
     insert_test_data().await;
     create_enhanced_jsonb_test_data().await;
 
     let client = connect_with_tls(PROXY).await;
 
-    println!("🩺 Healthcare Database Showcase - EQL v2 Searchable Encryption");
-    println!("============================================================");
-
     // Query 1: Get the Aspirin medication ID
     let aspirin_id_sql = "SELECT id FROM medications WHERE name = 'Aspirin';";
     let rows = client.query(aspirin_id_sql, &[]).await.unwrap();

packages/showcase/src/main.rs

@@ -3,5 +3,6 @@ use crate::common::{reset_schema_to, PROXY};
 const SCHEMA: &str = include_str!("./schema.sql");
 
 pub async fn setup_schema() {
+    println!("Setup schema");
     reset_schema_to(SCHEMA, PROXY).await
 }