v8: wire more of call-reducer

Author: Centril

crates/core/src/host/v8/mod.rs

@@ -3,20 +3,25 @@
 use super::module_common::{build_common_module_from_raw, ModuleCommon};
 use super::module_host::{CallReducerParams, DynModule, Module, ModuleInfo, ModuleInstance, ModuleRuntime};
 use super::UpdateDatabaseResult;
-use crate::host::wasm_common::module_host_actor::InstanceCommon;
+use crate::host::wasm_common::instrumentation::CallTimes;
+use crate::host::wasm_common::module_host_actor::{
+    EnergyStats, ExecuteResult, ExecutionTimings, InstanceCommon, ReducerOp,
+};
 use crate::host::ArgsTuple;
 use crate::{host::Scheduler, module_host_context::ModuleCreationContext, replica_context::ReplicaContext};
 use anyhow::anyhow;
+use core::time::Duration;
 use de::deserialize_js;
 use error::{catch_exception, exception_already_thrown, ExcResult, Throwable};
 use from_value::cast;
 use key_cache::get_or_create_key_cache;
 use ser::serialize_to_js;
+use spacetimedb_client_api_messages::energy::{EnergyQuanta, ReducerBudget};
 use spacetimedb_datastore::locking_tx_datastore::MutTxId;
 use spacetimedb_datastore::traits::Program;
 use spacetimedb_lib::{ConnectionId, Identity, RawModuleDef};
 use std::sync::{Arc, LazyLock};
-use v8::{Function, HandleScope, Local, Value};
+use v8::{Context, ContextOptions, ContextScope, Function, HandleScope, Isolate, Local, Value};
 
 mod de;
 mod error;
@@ -110,7 +115,7 @@ impl Module for JsModule {
     }
 
     fn info(&self) -> Arc<ModuleInfo> {
-        todo!()
+        self.common.info().clone()
     }
 
     fn create_instance(&self) -> Self::Instance {
@@ -137,8 +142,42 @@ impl ModuleInstance for JsInstance {
         self.common.update_database(replica_ctx, program, old_module_info)
     }
 
-    fn call_reducer(&mut self, _tx: Option<MutTxId>, _params: CallReducerParams) -> super::ReducerCallResult {
-        todo!()
+    fn call_reducer(&mut self, tx: Option<MutTxId>, params: CallReducerParams) -> super::ReducerCallResult {
+        // TODO(centril): snapshots, module->host calls
+        let mut isolate = Isolate::new(<_>::default());
+        let scope = &mut HandleScope::new(&mut isolate);
+        let context = Context::new(scope, ContextOptions::default());
+        let scope = &mut ContextScope::new(scope, context);
+
+        self.common.call_reducer_with_tx(
+            &self.replica_ctx.clone(),
+            tx,
+            params,
+            // TODO(centril): logging.
+            |_ty, _fun, _err| {},
+            |tx, op, _budget| {
+                let call_result = call_call_reducer_from_op(scope, op);
+                // TODO(centril): energy metrering.
+                let energy = EnergyStats {
+                    used: EnergyQuanta::ZERO,
+                    wasmtime_fuel_used: 0,
+                    remaining: ReducerBudget::ZERO,
+                };
+                // TODO(centril): timings.
+                let timings = ExecutionTimings {
+                    total_duration: Duration::ZERO,
+                    wasm_instance_env_call_times: CallTimes::new(),
+                };
+                let exec_result = ExecuteResult {
+                    energy,
+                    timings,
+                    // TODO(centril): memory allocation.
+                    memory_allocation: 0,
+                    call_result,
+                };
+                (tx, exec_result)
+            },
+        )
     }
 }
 
@@ -163,13 +202,25 @@ fn call_free_fun<'scope>(
     fun.call(scope, receiver, args).ok_or_else(exception_already_thrown)
 }
 
+// Calls the `__call_reducer__` function on the global proxy object using `op`.
+fn call_call_reducer_from_op(scope: &mut HandleScope<'_>, op: ReducerOp<'_>) -> anyhow::Result<Result<(), Box<str>>> {
+    call_call_reducer(
+        scope,
+        op.id.into(),
+        op.caller_identity,
+        op.caller_connection_id,
+        op.timestamp.to_micros_since_unix_epoch(),
+        op.args,
+    )
+}
+
 // Calls the `__call_reducer__` function on the global proxy object.
 fn call_call_reducer(
     scope: &mut HandleScope<'_>,
     reducer_id: u32,
     sender: &Identity,
     conn_id: &ConnectionId,
-    timestamp: u64,
+    timestamp: i64,
     reducer_args: &ArgsTuple,
 ) -> anyhow::Result<Result<(), Box<str>>> {
     // Get a cached version of the `__call_reducer__` property.

crates/core/src/host/wasm_common/module_host_actor.rs

@@ -1,4 +1,3 @@
-use bytes::Bytes;
 use prometheus::{Histogram, IntCounter, IntGauge};
 use spacetimedb_lib::db::raw_def::v9::Lifecycle;
 use spacetimedb_schema::auto_migrate::ponder_migrate;
@@ -16,7 +15,7 @@ use crate::host::module_host::{
     CallReducerParams, DatabaseUpdate, DynModule, EventStatus, Module, ModuleEvent, ModuleFunctionCall, ModuleInfo,
     ModuleInstance,
 };
-use crate::host::{ReducerCallResult, ReducerId, ReducerOutcome, Scheduler, UpdateDatabaseResult};
+use crate::host::{ArgsTuple, ReducerCallResult, ReducerId, ReducerOutcome, Scheduler, UpdateDatabaseResult};
 use crate::identity::Identity;
 use crate::messages::control_db::HostType;
 use crate::module_host_context::ModuleCreationContext;
@@ -333,7 +332,7 @@ impl InstanceCommon {
     /// The method also performs various measurements and records energy usage,
     /// as well as broadcasting a [`ModuleEvent`] containing information about
     /// the outcome of the call.
-    fn call_reducer_with_tx(
+    pub(crate) fn call_reducer_with_tx(
         &mut self,
         replica_ctx: &ReplicaContext,
         tx: Option<MutTxId>,
@@ -379,7 +378,7 @@ impl InstanceCommon {
             caller_identity: &caller_identity,
             caller_connection_id: &caller_connection_id,
             timestamp,
-            arg_bytes: args.get_bsatn().clone(),
+            args: &args,
         };
 
         let workload = Workload::Reducer(ReducerContext::from(op.clone()));
@@ -626,8 +625,8 @@ pub struct ReducerOp<'a> {
     pub caller_identity: &'a Identity,
     pub caller_connection_id: &'a ConnectionId,
     pub timestamp: Timestamp,
-    /// The BSATN-serialized arguments passed to the reducer.
-    pub arg_bytes: Bytes,
+    /// The arguments passed to the reducer.
+    pub args: &'a ArgsTuple,
 }
 
 impl From<ReducerOp<'_>> for execution_context::ReducerContext {
@@ -638,15 +637,15 @@ impl From<ReducerOp<'_>> for execution_context::ReducerContext {
             caller_identity,
             caller_connection_id,
             timestamp,
-            arg_bytes,
+            args,
         }: ReducerOp<'_>,
     ) -> Self {
         Self {
             name: name.to_owned(),
             caller_identity: *caller_identity,
             caller_connection_id: *caller_connection_id,
             timestamp,
-            arg_bsatn: arg_bytes.clone(),
+            arg_bsatn: args.get_bsatn().clone(),
         }
     }
 }

crates/core/src/host/wasmtime/wasmtime_module.rs

@@ -204,7 +204,8 @@ impl module_host_actor::WasmInstance for WasmtimeInstance {
         let [conn_id_0, conn_id_1] = bytemuck::must_cast(op.caller_connection_id.as_le_byte_array());
 
         // Prepare arguments to the reducer + the error sink & start timings.
-        let (args_source, errors_sink) = store.data_mut().start_reducer(op.name, op.arg_bytes, op.timestamp);
+        let args_bytes = op.args.get_bsatn().clone();
+        let (args_source, errors_sink) = store.data_mut().start_reducer(op.name, args_bytes, op.timestamp);
 
         let call_result = self.call_reducer.call(
             &mut *store,