Merge pull request #2780 from fermyon/sqlite-statements

Run sqlite statements

Author: rylev

Cargo.lock

@@ -68,16 +68,6 @@ error: the `wasm32-wasi` target is not installed
     std::fs::create_dir_all("target/test-programs").unwrap();
 
     build_wasm_test_program("core-wasi-test.wasm", "crates/core/tests/core-wasi-test");
-    // build_wasm_test_program("redis-rust.wasm", "crates/trigger-redis/tests/rust");
-    // build_wasm_test_program(
-    //     "spin-http-benchmark.wasm",
-    //     "crates/trigger-http/benches/spin-http-benchmark",
-    // );
-    // build_wasm_test_program(
-    //     "wagi-benchmark.wasm",
-    //     "crates/trigger-http/benches/wagi-benchmark",
-    // );
-    // build_wasm_test_program("timer_app_example.wasm", "examples/spin-timer/app-example");
 
     cargo_build(TIMER_TRIGGER_INTEGRATION_TEST);
 }

build.rs

@@ -66,15 +66,12 @@ impl v2::HostConnection for InstanceState {
         if !self.allowed_databases.contains(&database) {
             return Err(v2::Error::AccessDenied);
         }
-        (self.get_connection_creator)(&database)
+        let conn = (self.get_connection_creator)(&database)
             .ok_or(v2::Error::NoSuchDatabase)?
-            .create_connection()
-            .await
-            .and_then(|conn| {
-                self.connections
-                    .push(conn)
-                    .map_err(|()| v2::Error::Io("too many connections opened".to_string()))
-            })
+            .create_connection()?;
+        self.connections
+            .push(conn)
+            .map_err(|()| v2::Error::Io("too many connections opened".to_string()))
             .map(Resource::new_own)
     }
 

crates/factor-sqlite/src/host.rs

@@ -81,10 +81,7 @@ impl Factor for SqliteFactor {
             get_connection_creator(label).is_some()
         })?;
 
-        Ok(AppState {
-            allowed_databases,
-            get_connection_creator,
-        })
+        Ok(AppState::new(allowed_databases, get_connection_creator))
     }
 
     fn prepare<T: spin_factors::RuntimeFactors>(
@@ -149,28 +146,51 @@ pub trait DefaultLabelResolver: Send + Sync {
     fn default(&self, label: &str) -> Option<Arc<dyn ConnectionCreator>>;
 }
 
+#[derive(Clone)]
 pub struct AppState {
     /// A map from component id to a set of allowed database labels.
     allowed_databases: HashMap<String, Arc<HashSet<String>>>,
     /// A function for mapping from database name to a connection creator.
     get_connection_creator: host::ConnectionCreatorGetter,
 }
 
+impl AppState {
+    /// Create a new `AppState`
+    pub fn new(
+        allowed_databases: HashMap<String, Arc<HashSet<String>>>,
+        get_connection_creator: host::ConnectionCreatorGetter,
+    ) -> Self {
+        Self {
+            allowed_databases,
+            get_connection_creator,
+        }
+    }
+
+    /// Get a connection for a given database label.
+    ///
+    /// Returns `None` if there is no connection creator for the given label.
+    pub async fn get_connection(
+        &self,
+        label: &str,
+    ) -> Option<Result<Box<dyn Connection>, v2::Error>> {
+        let connection = (self.get_connection_creator)(label)?.create_connection();
+        Some(connection)
+    }
+}
+
 /// A creator of a connections for a particular SQLite database.
-#[async_trait]
 pub trait ConnectionCreator: Send + Sync {
     /// Get a *new* [`Connection`]
     ///
     /// The connection should be a new connection, not a reused one.
-    async fn create_connection(&self) -> Result<Box<dyn Connection + 'static>, v2::Error>;
+    fn create_connection(&self) -> Result<Box<dyn Connection + 'static>, v2::Error>;
 }
 
-#[async_trait::async_trait]
 impl<F> ConnectionCreator for F
 where
     F: Fn() -> anyhow::Result<Box<dyn Connection + 'static>> + Send + Sync + 'static,
 {
-    async fn create_connection(&self) -> Result<Box<dyn Connection + 'static>, v2::Error> {
+    fn create_connection(&self) -> Result<Box<dyn Connection + 'static>, v2::Error> {
         (self)().map_err(|_| v2::Error::InvalidConnection)
     }
 }

crates/factor-sqlite/src/lib.rs

@@ -143,9 +143,8 @@ impl spin_factor_sqlite::DefaultLabelResolver for DefaultLabelResolver {
 /// A connection creator that always returns an error.
 struct InvalidConnectionCreator;
 
-#[async_trait::async_trait]
 impl spin_factor_sqlite::ConnectionCreator for InvalidConnectionCreator {
-    async fn create_connection(
+    fn create_connection(
         &self,
     ) -> Result<Box<dyn spin_factor_sqlite::Connection + 'static>, spin_world::v2::sqlite::Error>
     {

crates/factor-sqlite/tests/factor_test.rs

@@ -32,7 +32,12 @@ impl<T: RuntimeFactors, U: Send + 'static> FactorsExecutor<T, U> {
             hooks: Default::default(),
         })
     }
+}
 
+impl<T: RuntimeFactors, U: Send + 'static> FactorsExecutor<T, U>
+where
+    T::AppState: Sync,
+{
     /// Adds the given [`ExecutorHooks`] to this executor.
     ///
     /// Hooks are run in the order they are added.
@@ -53,7 +58,7 @@ impl<T: RuntimeFactors, U: Send + 'static> FactorsExecutor<T, U> {
             .context("failed to configure app")?;
 
         for hooks in &mut self.hooks {
-            hooks.configure_app(&configured_app)?;
+            hooks.configure_app(&configured_app).await?;
         }
 
         let mut component_instance_pres = HashMap::new();
@@ -75,9 +80,14 @@ impl<T: RuntimeFactors, U: Send + 'static> FactorsExecutor<T, U> {
     }
 }
 
-pub trait ExecutorHooks<T: RuntimeFactors, U>: Send + Sync {
+#[async_trait]
+pub trait ExecutorHooks<T, U>: Send + Sync
+where
+    T: RuntimeFactors,
+    T::AppState: Sync,
+{
     /// Configure app hooks run immediately after [`RuntimeFactors::configure_app`].
-    fn configure_app(&mut self, configured_app: &ConfiguredApp<T>) -> anyhow::Result<()> {
+    async fn configure_app(&mut self, configured_app: &ConfiguredApp<T>) -> anyhow::Result<()> {
         let _ = configured_app;
         Ok(())
     }
@@ -131,7 +141,12 @@ impl<T: RuntimeFactors, U: Send + 'static> FactorsExecutorApp<T, U> {
             .with_context(|| format!("no such component {component_id:?}"))?;
         Ok(instance_pre.component())
     }
+}
 
+impl<T: RuntimeFactors, U: Send + 'static> FactorsExecutorApp<T, U>
+where
+    T::AppState: Sync,
+{
     /// Returns an instance builder for the given component ID.
     pub fn prepare(&self, component_id: &str) -> anyhow::Result<FactorsInstanceBuilder<T, U>> {
         let app_component = self

crates/factors-executor/src/lib.rs

@@ -42,8 +42,12 @@ spin-factors-executor = { path = "../factors-executor" }
 spin-runtime-config = { path = "../runtime-config" }
 spin-telemetry = { path = "../telemetry" }
 terminal = { path = "../terminal" }
-tokio = { version = "1.23", features = ["fs"] }
+tokio = { version = "1.23", features = ["fs", "rt"] }
 tracing = { workspace = true }
 
+[dev-dependencies]
+spin-world = { path = "../world" }
+tempfile = "3.12"
+
 [lints]
 workspace = true

crates/trigger/Cargo.toml

@@ -1,4 +1,5 @@
 mod launch_metadata;
+mod sqlite_statements;
 mod summary;
 
 use std::future::Future;
@@ -13,6 +14,7 @@ use spin_common::{arg_parser::parse_kv, sloth};
 use spin_core::async_trait;
 use spin_factors_executor::{ComponentLoader, FactorsExecutor};
 use spin_runtime_config::{ResolvedRuntimeConfig, UserProvidedPath};
+use sqlite_statements::SqlStatementExecutorHook;
 use summary::KeyValueDefaultStoreSummaryHook;
 
 use crate::factors::{TriggerFactors, TriggerFactorsRuntimeConfig};
@@ -198,6 +200,7 @@ impl<T: Trigger> FactorsTriggerCommand<T> {
                     state_dir: self.state_dir.as_deref(),
                     local_app_dir: local_app_dir.as_deref(),
                     initial_key_values: self.key_values,
+                    sqlite_statements: self.sqlite_statements,
                     allow_transient_write: self.allow_transient_write,
                     follow_components,
                     log_dir: self.log,
@@ -277,6 +280,8 @@ pub struct TriggerAppOptions<'a> {
     local_app_dir: Option<&'a str>,
     /// Initial key/value pairs to set in the app's default store.
     initial_key_values: Vec<(String, String)>,
+    /// SQLite statements to run.
+    sqlite_statements: Vec<String>,
     /// Whether to allow transient writes to mounted files
     allow_transient_write: bool,
     /// Which components should have their logs followed.
@@ -351,8 +356,6 @@ impl<T: Trigger> TriggerAppBuilder<T> {
         )
         .context("failed to create factors")?;
 
-        // TODO(factors): handle: self.sqlite_statements
-
         // TODO: port the rest of the component loader logic
         struct SimpleComponentLoader;
 
@@ -421,6 +424,7 @@ impl<T: Trigger> TriggerAppBuilder<T> {
         // TODO:
         // builder.hooks(SummariseRuntimeConfigHook::new(&self.runtime_config_file));
         executor.add_hooks(KeyValueDefaultStoreSummaryHook);
+        executor.add_hooks(SqlStatementExecutorHook::new(options.sqlite_statements));
         // builder.hooks(SqlitePersistenceMessageHook);
 
         let configured_app = {