Merge branch 'master' into kim/cli/graceful-disconnect-subscribe

kim · web-flow · commit 8d63f34e66f1 · 2025-07-09T10:01:45.000+02:00
diff --git a/crates/bindings-macro/src/sats.rs b/crates/bindings-macro/src/sats.rs
@@ -347,7 +347,7 @@ pub(crate) fn derive_deserialize(ty: &SatsType<'_>) -> TokenStream {
     de_generics.params.insert(0, de_lt_param.into());
     let (de_impl_generics, _, de_where_clause) = de_generics.split_for_impl();
 
-    let (iter_n, iter_n2, iter_n3) = (0usize.., 0usize.., 0usize..);
+    let (iter_n, iter_n2, iter_n3, iter_n4) = (0usize.., 0usize.., 0usize.., 0usize..);
 
     match &ty.data {
         SatsTypeData::Product(fields) => {
@@ -443,8 +443,8 @@ pub(crate) fn derive_deserialize(ty: &SatsType<'_>) -> TokenStream {
                     impl #de_impl_generics #spacetimedb_lib::de::FieldNameVisitor<'de> for __ProductVisitor #ty_generics #de_where_clause {
                         type Output = __ProductFieldIdent;
 
-                        fn field_names(&self, names: &mut dyn #spacetimedb_lib::de::ValidNames) {
-                            names.extend::<&[&str]>(&[#(#field_strings),*])
+                        fn field_names(&self) -> impl '_ + Iterator<Item = Option<&str>> {
+                            [#(#field_strings),*].into_iter().map(Some)
                         }
 
                         fn visit<__E: #spacetimedb_lib::de::Error>(self, name: &str) -> Result<Self::Output, __E> {
@@ -453,6 +453,13 @@ pub(crate) fn derive_deserialize(ty: &SatsType<'_>) -> TokenStream {
                                 _ => Err(#spacetimedb_lib::de::Error::unknown_field_name(name, &self)),
                             }
                         }
+
+                        fn visit_seq<__E: #spacetimedb_lib::de::Error>(self, index: usize, name: &str) -> Result<Self::Output, __E> {
+                            match index {
+                                #(#iter_n4 => Ok(__ProductFieldIdent::#field_names),)*
+                                _ => Err(#spacetimedb_lib::de::Error::unknown_field_name(name, &self)),
+                            }
+                        }
                     }
 
                     #[allow(non_camel_case_types)]
@@ -516,11 +523,11 @@ pub(crate) fn derive_deserialize(ty: &SatsType<'_>) -> TokenStream {
                         #(#variant_idents,)*
                     }
 
-                    impl #de_impl_generics #spacetimedb_lib::de::VariantVisitor for __SumVisitor #ty_generics #de_where_clause {
+                    impl #de_impl_generics #spacetimedb_lib::de::VariantVisitor<'de> for __SumVisitor #ty_generics #de_where_clause {
                         type Output = __Variant;
 
-                        fn variant_names(&self, names: &mut dyn #spacetimedb_lib::de::ValidNames) {
-                            names.extend::<&[&str]>(&[#(#variant_names,)*])
+                        fn variant_names(&self) -> impl '_ + Iterator<Item = &str> {
+                            [#(#variant_names,)*].into_iter()
                         }
 
                         fn visit_tag<E: #spacetimedb_lib::de::Error>(self, __tag: u8) -> Result<Self::Output, E> {
diff --git a/crates/core/src/db/datastore/system_tables.rs b/crates/core/src/db/datastore/system_tables.rs
@@ -816,12 +816,15 @@ pub struct ModuleKind(u8);
 impl ModuleKind {
     /// The [`ModuleKind`] of WASM-based modules.
     pub const WASM: ModuleKind = ModuleKind(0);
+    /// The [`ModuleKind`] of JS modules.
+    pub const JS: ModuleKind = ModuleKind(1);
 }
 
 impl From<crate::messages::control_db::HostType> for ModuleKind {
     fn from(host_type: crate::messages::control_db::HostType) -> Self {
         match host_type {
             crate::messages::control_db::HostType::Wasm => Self::WASM,
+            crate::messages::control_db::HostType::Js => Self::JS,
         }
     }
 }
diff --git a/crates/core/src/host/host_controller.rs b/crates/core/src/host/host_controller.rs
@@ -9,6 +9,8 @@ use crate::db::db_metrics::DB_METRICS;
 use crate::db::relational_db::{self, DiskSizeFn, RelationalDB, Txdata};
 use crate::db::{self, spawn_tx_metrics_recorder};
 use crate::energy::{EnergyMonitor, EnergyQuanta, NullEnergyMonitor};
+use crate::host::module_host::ModuleRuntime as _;
+use crate::host::v8::V8Runtime;
 use crate::messages::control_db::{Database, HostType};
 use crate::module_host_context::ModuleCreationContext;
 use crate::replica_context::ReplicaContext;
@@ -106,12 +108,14 @@ pub struct HostController {
 
 struct HostRuntimes {
     wasmtime: WasmtimeRuntime,
+    v8: V8Runtime,
 }
 
 impl HostRuntimes {
     fn new(data_dir: Option<&ServerDataDir>) -> Arc<Self> {
         let wasmtime = WasmtimeRuntime::new(data_dir);
-        Arc::new(Self { wasmtime })
+        let v8 = V8Runtime::default();
+        Arc::new(Self { wasmtime, v8 })
     }
 }
 
@@ -671,19 +675,25 @@ async fn make_module_host(
     //       threads, but those aren't for computation. Also, wasmtime uses rayon
     //       to run compilation in parallel, so it'll need to run stuff in rayon anyway.
     asyncify(move || {
+        let mcc = ModuleCreationContext {
+            replica_ctx,
+            scheduler,
+            program: &program,
+            energy_monitor,
+        };
+
+        let start = Instant::now();
         let module_host = match host_type {
             HostType::Wasm => {
-                let mcc = ModuleCreationContext {
-                    replica_ctx,
-                    scheduler,
-                    program: &program,
-                    energy_monitor,
-                };
-                let start = Instant::now();
                 let actor = runtimes.wasmtime.make_actor(mcc)?;
                 trace!("wasmtime::make_actor blocked for {:?}", start.elapsed());
                 ModuleHost::new(actor, unregister, core)
             }
+            HostType::Js => {
+                let actor = runtimes.v8.make_actor(mcc)?;
+                trace!("v8::make_actor blocked for {:?}", start.elapsed());
+                ModuleHost::new(actor, unregister, core)
+            }
         };
         Ok((program, module_host))
     })
diff --git a/crates/core/src/host/mod.rs b/crates/core/src/host/mod.rs
@@ -16,8 +16,10 @@ mod host_controller;
 pub mod module_host;
 pub mod scheduler;
 pub mod wasmtime;
+
 // Visible for integration testing.
 pub mod instance_env;
+pub mod v8; // only pub for testing
 mod wasm_common;
 
 pub use disk_storage::DiskStorage;
diff --git a/crates/core/src/host/module_host.rs b/crates/core/src/host/module_host.rs
@@ -12,6 +12,7 @@ use crate::execution_context::{ExecutionContext, ReducerContext, Workload, Workl
 use crate::hash::Hash;
 use crate::identity::Identity;
 use crate::messages::control_db::Database;
+use crate::module_host_context::ModuleCreationContext;
 use crate::replica_context::ReplicaContext;
 use crate::sql::ast::SchemaViewer;
 use crate::sql::parser::RowLevelExpr;
@@ -306,6 +307,12 @@ impl ReducersMap {
     }
 }
 
+/// A runtime that can create modules.
+pub trait ModuleRuntime {
+    /// Creates a module based on the context `mcc`.
+    fn make_actor(&self, mcc: ModuleCreationContext<'_>) -> anyhow::Result<impl Module>;
+}
+
 pub trait DynModule: Send + Sync + 'static {
     fn replica_ctx(&self) -> &Arc<ReplicaContext>;
     fn scheduler(&self) -> &Scheduler;
diff --git a/crates/core/src/host/v8/mod.rs b/crates/core/src/host/v8/mod.rs
@@ -0,0 +1,96 @@
+use crate::{
+    db::datastore::locking_tx_datastore::MutTxId,
+    host::{
+        module_host::{DynModule, Module, ModuleInfo, ModuleInstance, ModuleRuntime},
+        Scheduler,
+    },
+    module_host_context::ModuleCreationContext,
+    replica_context::ReplicaContext,
+};
+use anyhow::anyhow;
+use std::sync::{Arc, LazyLock};
+
+use super::module_host::CallReducerParams;
+
+/// The V8 runtime, for modules written in e.g., JS or TypeScript.
+#[derive(Default)]
+pub struct V8Runtime {
+    _priv: (),
+}
+
+impl ModuleRuntime for V8Runtime {
+    fn make_actor(&self, mcc: ModuleCreationContext<'_>) -> anyhow::Result<impl Module> {
+        V8_RUNTIME_GLOBAL.make_actor(mcc)
+    }
+}
+
+static V8_RUNTIME_GLOBAL: LazyLock<V8RuntimeInner> = LazyLock::new(V8RuntimeInner::new);
+
+/// The actual V8 runtime, with initialization of V8.
+struct V8RuntimeInner {
+    _priv: (),
+}
+
+impl V8RuntimeInner {
+    #[allow(clippy::new_without_default)]
+    const fn new() -> Self {
+        // TODO: actually setup V8.
+
+        Self { _priv: () }
+    }
+
+    fn make_actor(&self, _: ModuleCreationContext<'_>) -> anyhow::Result<impl Module> {
+        Err::<JsModule, _>(anyhow!("v8_todo"))
+    }
+}
+
+#[derive(Clone)]
+struct JsModule;
+
+impl DynModule for JsModule {
+    fn replica_ctx(&self) -> &Arc<ReplicaContext> {
+        todo!()
+    }
+
+    fn scheduler(&self) -> &Scheduler {
+        todo!()
+    }
+}
+
+impl Module for JsModule {
+    type Instance = JsInstance;
+
+    type InitialInstances<'a> = std::iter::Empty<JsInstance>;
+
+    fn initial_instances(&mut self) -> Self::InitialInstances<'_> {
+        std::iter::empty()
+    }
+
+    fn info(&self) -> Arc<ModuleInfo> {
+        todo!()
+    }
+
+    fn create_instance(&self) -> Self::Instance {
+        todo!()
+    }
+}
+
+struct JsInstance;
+
+impl ModuleInstance for JsInstance {
+    fn trapped(&self) -> bool {
+        todo!()
+    }
+
+    fn update_database(
+        &mut self,
+        _program: crate::db::datastore::traits::Program,
+        _old_module_info: Arc<ModuleInfo>,
+    ) -> anyhow::Result<super::UpdateDatabaseResult> {
+        todo!()
+    }
+
+    fn call_reducer(&mut self, _tx: Option<MutTxId>, _params: CallReducerParams) -> super::ReducerCallResult {
+        todo!()
+    }
+}
diff --git a/crates/core/src/host/wasm_common/module_host_actor.rs b/crates/core/src/host/wasm_common/module_host_actor.rs
@@ -122,8 +122,6 @@ pub enum DescribeError {
     Decode(#[from] DecodeError),
     #[error(transparent)]
     RuntimeError(anyhow::Error),
-    #[error("unimplemented RawModuleDef version")]
-    UnimplementedRawModuleDefVersion,
 }
 
 impl<T: WasmModule> WasmModuleHostActor<T> {
diff --git a/crates/core/src/host/wasmtime/mod.rs b/crates/core/src/host/wasmtime/mod.rs
@@ -7,6 +7,7 @@ use wasmtime::{Engine, Linker, Module, StoreContext, StoreContextMut};
 
 use crate::energy::{EnergyQuanta, ReducerBudget};
 use crate::error::NodesError;
+use crate::host::module_host::ModuleRuntime;
 use crate::module_host_context::ModuleCreationContext;
 
 mod wasm_instance_env;
@@ -80,11 +81,10 @@ impl WasmtimeRuntime {
         config.cache_config_load(tmpfile.path())?;
         Ok(())
     }
+}
 
-    pub fn make_actor(
-        &self,
-        mcc: ModuleCreationContext,
-    ) -> Result<impl super::module_host::Module, ModuleCreationError> {
+impl ModuleRuntime for WasmtimeRuntime {
+    fn make_actor(&self, mcc: ModuleCreationContext) -> anyhow::Result<impl super::module_host::Module> {
         let module = Module::new(&self.engine, &mcc.program.bytes).map_err(ModuleCreationError::WasmCompileError)?;
 
         let func_imports = module
diff --git a/crates/core/src/messages/control_db.rs b/crates/core/src/messages/control_db.rs
@@ -74,4 +74,5 @@ pub struct NodeStatus {
 #[repr(i32)]
 pub enum HostType {
     Wasm = 0,
+    Js,
 }
diff --git a/crates/core/src/subscription/module_subscription_actor.rs b/crates/core/src/subscription/module_subscription_actor.rs
@@ -2301,7 +2301,7 @@ mod tests {
 
         // We should have evaluated queries for `x = 3` and `x = 4`
         assert_eq!(metrics.delta_queries_evaluated, 2);
-        assert_eq!(metrics.delta_queries_matched, 2);
+        assert_eq!(metrics.delta_queries_matched, 1);
 
         // UPDATE v SET x = 0 WHERE id = 3
         let metrics = commit_tx(
@@ -2319,19 +2319,17 @@ mod tests {
 
         // Insert new row into `u` that joins with `x = 5`
         // UPDATE v SET x = 6 WHERE id = 5
+        // Should result in a no-op
         let metrics = commit_tx(
             &db,
             &subs,
             [(v_id, product![5u64, 5u64, 5u64])],
             [(v_id, product![5u64, 6u64, 6u64]), (u_id, product![5u64, 6u64, 7u64])],
         )?;
 
-        // Results in a no-op
-        assert_tx_update_for_table(&mut rx, u_id, &schema, [], []).await;
-
         // We should only have evaluated the query for `x = 5`
         assert_eq!(metrics.delta_queries_evaluated, 1);
-        assert_eq!(metrics.delta_queries_matched, 1);
+        assert_eq!(metrics.delta_queries_matched, 0);
 
         Ok(())
     }
diff --git a/crates/execution/src/lib.rs b/crates/execution/src/lib.rs
@@ -115,12 +115,25 @@ pub trait DeltaStore {
     }
 }
 
-#[derive(Clone, PartialEq, Eq)]
+#[derive(Clone)]
 pub enum Row<'a> {
     Ptr(RowRef<'a>),
     Ref(&'a ProductValue),
 }
 
+impl PartialEq for Row<'_> {
+    fn eq(&self, other: &Self) -> bool {
+        match (self, other) {
+            (Self::Ptr(x), Self::Ptr(y)) => x == y,
+            (Self::Ref(x), Self::Ref(y)) => x == y,
+            (Self::Ptr(x), Self::Ref(y)) => x == *y,
+            (Self::Ref(x), Self::Ptr(y)) => y == *x,
+        }
+    }
+}
+
+impl Eq for Row<'_> {}
+
 impl Hash for Row<'_> {
     fn hash<H: Hasher>(&self, state: &mut H) {
         match self {
diff --git a/crates/sats/src/algebraic_value/de.rs b/crates/sats/src/algebraic_value/de.rs
@@ -173,12 +173,12 @@ impl SumAccess {
     }
 }
 
-impl de::SumAccess<'_> for SumAccess {
+impl<'de> de::SumAccess<'de> for SumAccess {
     type Error = ValueDeserializeError;
 
     type Variant = ValueDeserializer;
 
-    fn variant<V: de::VariantVisitor>(self, visitor: V) -> Result<(V::Output, Self::Variant), Self::Error> {
+    fn variant<V: de::VariantVisitor<'de>>(self, visitor: V) -> Result<(V::Output, Self::Variant), Self::Error> {
         let tag = visitor.visit_tag(self.sum.tag)?;
         let val = *self.sum.value;
         Ok((tag, ValueDeserializer { val }))
@@ -313,7 +313,7 @@ impl<'de> de::SumAccess<'de> for &'de SumAccess {
 
     type Variant = &'de ValueDeserializer;
 
-    fn variant<V: de::VariantVisitor>(self, visitor: V) -> Result<(V::Output, Self::Variant), Self::Error> {
+    fn variant<V: de::VariantVisitor<'de>>(self, visitor: V) -> Result<(V::Output, Self::Variant), Self::Error> {
         let tag = visitor.visit_tag(self.sum.tag)?;
         Ok((tag, ValueDeserializer::from_ref(&self.sum.value)))
     }
diff --git a/crates/sats/src/bsatn/de.rs b/crates/sats/src/bsatn/de.rs
@@ -146,7 +146,7 @@ impl<'de, R: BufReader<'de>> SumAccess<'de> for Deserializer<'_, R> {
     type Error = DecodeError;
     type Variant = Self;
 
-    fn variant<V: de::VariantVisitor>(self, visitor: V) -> Result<(V::Output, Self::Variant), Self::Error> {
+    fn variant<V: de::VariantVisitor<'de>>(self, visitor: V) -> Result<(V::Output, Self::Variant), Self::Error> {
         let tag = self.reader.get_u8()?;
         visitor.visit_tag(tag).map(|variant| (variant, self))
     }
diff --git a/crates/sats/src/de.rs b/crates/sats/src/de.rs
diff --git a/crates/sats/src/de/impls.rs b/crates/sats/src/de/impls.rs
diff --git a/crates/sats/src/de/serde.rs b/crates/sats/src/de/serde.rs
diff --git a/crates/sats/src/layout.rs b/crates/sats/src/layout.rs
diff --git a/crates/standalone/src/subcommands/extract_schema.rs b/crates/standalone/src/subcommands/extract_schema.rs

Original file line number	Diff line number	Diff line change
`@@ -816,12 +816,15 @@ pub struct ModuleKind(u8);`
`816`	`816`	`impl ModuleKind {`
`817`	`817`	/// The [`ModuleKind`] of WASM-based modules.
`818`	`818`	`pub const WASM: ModuleKind = ModuleKind(0);`
	`819`	+ /// The [`ModuleKind`] of JS modules.
	`820`	`+ pub const JS: ModuleKind = ModuleKind(1);`
`819`	`821`	`}`
`820`	`822`
`821`	`823`	`impl From<crate::messages::control_db::HostType> for ModuleKind {`
`822`	`824`	`fn from(host_type: crate::messages::control_db::HostType) -> Self {`
`823`	`825`	`match host_type {`
`824`	`826`	`crate::messages::control_db::HostType::Wasm => Self::WASM,`
	`827`	`+ crate::messages::control_db::HostType::Js => Self::JS,`
`825`	`828`	`}`
`826`	`829`	`}`
`827`	`830`	`}`
Original file line number	Diff line number	Diff line change
`@@ -122,8 +122,6 @@ pub enum DescribeError {`
`122`	`122`	`Decode(#[from] DecodeError),`
`123`	`123`	`#[error(transparent)]`
`124`	`124`	`RuntimeError(anyhow::Error),`
`125`		`- #[error("unimplemented RawModuleDef version")]`
`126`		`- UnimplementedRawModuleDefVersion,`
`127`	`125`	`}`
`128`	`126`
`129`	`127`	`impl<T: WasmModule> WasmModuleHostActor<T> {`
Original file line number	Diff line number	Diff line change
`@@ -74,4 +74,5 @@ pub struct NodeStatus {`
`74`	`74`	`#[repr(i32)]`
`75`	`75`	`pub enum HostType {`
`76`	`76`	`Wasm = 0,`
	`77`	`+ Js,`
`77`	`78`	`}`