Pipeline wasm module operations

crates/cli/src/subcommands/subscribe.rs

@@ -240,6 +240,8 @@ enum Error {
     },
     #[error("encountered failed transaction: {reason}")]
     TransactionFailure { reason: Box<str> },
+    #[error("encountered error in initial subscribe: {reason}")]
+    SubscribeFailure { reason: Box<str> },
     #[error("error formatting response: {source:#}")]
     Reformat {
         #[source]
@@ -295,6 +297,9 @@ where
                 }
                 break;
             }
+            ws_v1::ServerMessage::SubscriptionError(error) => {
+                return Err(Error::SubscribeFailure { reason: error.error });
+            }
             ws_v1::ServerMessage::TransactionUpdate(ws_v1::TransactionUpdate { status, .. }) => {
                 return Err(match status {
                     ws_v1::UpdateStatus::Failed(msg) => Error::TransactionFailure { reason: msg },
@@ -341,6 +346,9 @@ where
                     details: "received a second initial subscription update",
                 })
             }
+            ws_v1::ServerMessage::SubscriptionError(error) => {
+                return Err(Error::SubscribeFailure { reason: error.error });
+            }
             ws_v1::ServerMessage::TransactionUpdateLight(ws_v1::TransactionUpdateLight { update, .. })
             | ws_v1::ServerMessage::TransactionUpdate(ws_v1::TransactionUpdate {
                 status: ws_v1::UpdateStatus::Committed(update),

crates/core/src/config.rs

@@ -137,6 +137,7 @@ impl fmt::Display for MetadataFile {
 pub struct ConfigFile {
     pub certificate_authority: Option<CertificateAuthority>,
     pub logs: LogConfig,
+    pub wasm: WasmConfig,
     pub v8: V8Config,
 }
 
@@ -148,6 +149,8 @@ struct ConfigFileToml {
     #[serde(default)]
     logs: LogConfig,
     #[serde(default)]
+    wasm: WasmConfigToml,
+    #[serde(default)]
     v8: V8ConfigToml,
     #[serde(default)]
     v8_heap_policy: V8HeapPolicyConfig,
@@ -162,6 +165,9 @@ impl<'de> serde::Deserialize<'de> for ConfigFile {
         Ok(Self {
             certificate_authority: config.certificate_authority,
             logs: config.logs,
+            wasm: WasmConfig {
+                procedure_instance_pool_size: config.wasm.procedure_instance_pool_size,
+            },
             v8: V8Config {
                 procedure_instance_pool_size: config.v8.procedure_instance_pool_size,
                 heap_policy: config.v8_heap_policy,
@@ -206,6 +212,37 @@ pub struct LogConfig {
     pub directives: Vec<String>,
 }
 
+#[derive(Clone, Copy, Debug)]
+pub struct WasmConfig {
+    pub procedure_instance_pool_size: NonZeroUsize,
+}
+
+impl Default for WasmConfig {
+    fn default() -> Self {
+        Self {
+            procedure_instance_pool_size: default_wasm_procedure_instance_pool_size(),
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, serde::Deserialize)]
+#[serde(rename_all = "kebab-case")]
+struct WasmConfigToml {
+    #[serde(
+        default = "default_wasm_procedure_instance_pool_size",
+        deserialize_with = "de_nz_usize"
+    )]
+    pub procedure_instance_pool_size: NonZeroUsize,
+}
+
+impl Default for WasmConfigToml {
+    fn default() -> Self {
+        Self {
+            procedure_instance_pool_size: default_wasm_procedure_instance_pool_size(),
+        }
+    }
+}
+
 #[derive(Clone, Copy, Debug)]
 pub struct V8Config {
     pub procedure_instance_pool_size: NonZeroUsize,
@@ -323,6 +360,10 @@ fn default_v8_procedure_instance_pool_size() -> NonZeroUsize {
     std::thread::available_parallelism().unwrap_or_else(|_| NonZeroUsize::new(1).unwrap())
 }
 
+fn default_wasm_procedure_instance_pool_size() -> NonZeroUsize {
+    std::thread::available_parallelism().unwrap_or_else(|_| NonZeroUsize::new(1).unwrap())
+}
+
 fn de_nz_usize<'de, D>(deserializer: D) -> Result<NonZeroUsize, D::Error>
 where
     D: serde::Deserializer<'de>,
@@ -521,6 +562,10 @@ mod tests {
     fn v8_heap_policy_defaults_when_omitted() {
         let config: ConfigFile = toml::from_str("").unwrap();
 
+        assert_eq!(
+            config.wasm.procedure_instance_pool_size,
+            default_wasm_procedure_instance_pool_size()
+        );
         assert_eq!(
             config.v8.procedure_instance_pool_size,
             default_v8_procedure_instance_pool_size()
@@ -538,6 +583,9 @@ mod tests {
     #[test]
     fn v8_heap_policy_parses_from_toml() {
         let toml = r#"
+            [wasm]
+            procedure-instance-pool-size = 4
+
             [v8]
             procedure-instance-pool-size = 3
 
@@ -551,6 +599,7 @@ mod tests {
 
         let config: ConfigFile = toml::from_str(toml).unwrap();
 
+        assert_eq!(config.wasm.procedure_instance_pool_size.get(), 4);
         assert_eq!(config.v8.procedure_instance_pool_size.get(), 3);
         assert_eq!(config.v8.heap_policy.heap_check_request_interval, None);
         assert_eq!(

crates/core/src/host/host_controller.rs

@@ -3,7 +3,7 @@ use super::scheduler::SchedulerStarter;
 use super::wasmtime::WasmtimeRuntime;
 use super::{Scheduler, UpdateDatabaseResult};
 use crate::client::{ClientActorId, ClientName};
-use crate::config::V8Config;
+use crate::config::{V8Config, WasmConfig};
 use crate::database_logger::DatabaseLogger;
 use crate::db::persistence::PersistenceProvider;
 use crate::db::relational_db::{self, spawn_view_cleanup_loop, DiskSizeFn, RelationalDB, Txdata};
@@ -124,10 +124,22 @@ pub(crate) struct HostRuntimes {
     v8: V8Runtime,
 }
 
+#[derive(Clone, Copy, Debug, Default)]
+pub struct HostRuntimeConfig {
+    pub wasm: WasmConfig,
+    pub v8: V8Config,
+}
+
+impl HostRuntimeConfig {
+    pub fn new(wasm: WasmConfig, v8: V8Config) -> Self {
+        Self { wasm, v8 }
+    }
+}
+
 impl HostRuntimes {
-    fn new(data_dir: Option<&ServerDataDir>, v8_config: V8Config) -> Arc<Self> {
-        let wasmtime = WasmtimeRuntime::new(data_dir);
-        let v8 = V8Runtime::new(v8_config);
+    fn new(data_dir: Option<&ServerDataDir>, config: HostRuntimeConfig) -> Arc<Self> {
+        let wasmtime = WasmtimeRuntime::new(data_dir, config.wasm);
+        let v8 = V8Runtime::new(config.v8);
         Arc::new(Self { wasmtime, v8 })
     }
 }
@@ -211,7 +223,7 @@ impl HostController {
     pub fn new(
         data_dir: Arc<ServerDataDir>,
         default_config: db::Config,
-        v8_config: V8Config,
+        runtime_config: HostRuntimeConfig,
         program_storage: ProgramStorage,
         energy_monitor: Arc<impl EnergyMonitor>,
         persistence: Arc<dyn PersistenceProvider>,
@@ -223,7 +235,7 @@ impl HostController {
             program_storage,
             energy_monitor,
             persistence,
-            runtimes: HostRuntimes::new(Some(&data_dir), v8_config),
+            runtimes: HostRuntimes::new(Some(&data_dir), runtime_config),
             data_dir,
             page_pool: PagePool::new(default_config.page_pool_max_size),
             bsatn_rlb_pool: BsatnRowListBuilderPool::new(),
@@ -1365,7 +1377,7 @@ pub async fn extract_schema(program_bytes: Box<[u8]>, host_type: HostType) -> an
     extract_schema_with_pools(
         PagePool::new(None),
         BsatnRowListBuilderPool::new(),
-        &HostRuntimes::new(None, V8Config::default()),
+        &HostRuntimes::new(None, HostRuntimeConfig::default()),
         program_bytes,
         host_type,
     )

crates/core/src/host/mod.rs

@@ -26,7 +26,7 @@ mod wasm_common;
 pub use disk_storage::DiskStorage;
 pub use host_controller::{
     extract_schema, CallProcedureReturn, CallResult, ExternalDurability, ExternalStorage, HostController,
-    MigratePlanResult, ProcedureCallResult, ProgramStorage, ReducerCallResult, ReducerOutcome,
+    HostRuntimeConfig, MigratePlanResult, ProcedureCallResult, ProgramStorage, ReducerCallResult, ReducerOutcome,
 };
 pub use module_host::{ModuleHost, NoSuchModule, ProcedureCallError, ReducerCallError, UpdateDatabaseResult};
 pub use scheduler::Scheduler;