Reapply "Procedures: fix scheduling (#3704)" (#3774)

This reverts commit d26f3a1.

Author: coolreader18

crates/codegen/src/rust.rs

@@ -423,6 +423,7 @@ impl {func_name} for super::RemoteReducers {{
         {callback_id}(self.imp.on_reducer(
             {reducer_name:?},
             Box::new(move |ctx: &super::ReducerEventContext| {{
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {{
                     event: __sdk::ReducerEvent {{
                         reducer: super::Reducer::{enum_variant_name} {{

crates/codegen/tests/snapshots/codegen__codegen_rust.snap

@@ -72,6 +72,7 @@ impl add_player for super::RemoteReducers {
         AddPlayerCallbackId(self.imp.on_reducer(
             "add_player",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::AddPlayer {
@@ -181,6 +182,7 @@ impl add_private for super::RemoteReducers {
         AddPrivateCallbackId(self.imp.on_reducer(
             "add_private",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::AddPrivate {
@@ -294,6 +296,7 @@ age: u8,
         AddCallbackId(self.imp.on_reducer(
             "add",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::Add {
@@ -398,6 +401,7 @@ impl assert_caller_identity_is_module_identity for super::RemoteReducers {
         AssertCallerIdentityIsModuleIdentityCallbackId(self.imp.on_reducer(
             "assert_caller_identity_is_module_identity",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::AssertCallerIdentityIsModuleIdentity {
@@ -527,6 +531,7 @@ impl client_connected for super::RemoteReducers {
         ClientConnectedCallbackId(self.imp.on_reducer(
             "client_connected",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::ClientConnected {
@@ -636,6 +641,7 @@ impl delete_player for super::RemoteReducers {
         DeletePlayerCallbackId(self.imp.on_reducer(
             "delete_player",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::DeletePlayer {
@@ -745,6 +751,7 @@ impl delete_players_by_name for super::RemoteReducers {
         DeletePlayersByNameCallbackId(self.imp.on_reducer(
             "delete_players_by_name",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::DeletePlayersByName {
@@ -1066,6 +1073,7 @@ impl list_over_age for super::RemoteReducers {
         ListOverAgeCallbackId(self.imp.on_reducer(
             "list_over_age",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::ListOverAge {
@@ -1170,6 +1178,7 @@ impl log_module_identity for super::RemoteReducers {
         LogModuleIdentityCallbackId(self.imp.on_reducer(
             "log_module_identity",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::LogModuleIdentity {
@@ -3567,6 +3576,7 @@ impl query_private for super::RemoteReducers {
         QueryPrivateCallbackId(self.imp.on_reducer(
             "query_private",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::QueryPrivate {
@@ -3852,6 +3862,7 @@ impl repeating_test for super::RemoteReducers {
         RepeatingTestCallbackId(self.imp.on_reducer(
             "repeating_test",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::RepeatingTest {
@@ -4015,6 +4026,7 @@ impl say_hello for super::RemoteReducers {
         SayHelloCallbackId(self.imp.on_reducer(
             "say_hello",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::SayHello {
@@ -4325,6 +4337,7 @@ impl test_btree_index_args for super::RemoteReducers {
         TestBtreeIndexArgsCallbackId(self.imp.on_reducer(
             "test_btree_index_args",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::TestBtreeIndexArgs {
@@ -4878,6 +4891,7 @@ arg_4: NamespaceTestF,
         TestCallbackId(self.imp.on_reducer(
             "test",
             Box::new(move |ctx: &super::ReducerEventContext| {
+                #[allow(irrefutable_let_patterns)]
                 let super::ReducerEventContext {
                     event: __sdk::ReducerEvent {
                         reducer: super::Reducer::Test {

crates/core/src/host/mod.rs

@@ -8,6 +8,7 @@ use spacetimedb_lib::bsatn;
 use spacetimedb_lib::de::{serde::SeedWrapper, DeserializeSeed};
 use spacetimedb_lib::ProductValue;
 use spacetimedb_schema::def::deserialize::{ArgsSeed, FunctionDef};
+use spacetimedb_schema::def::ModuleDef;
 
 mod disk_storage;
 mod host_controller;
@@ -41,6 +42,14 @@ pub enum FunctionArgs {
 }
 
 impl FunctionArgs {
+    fn into_tuple_for_def<Def: FunctionDef>(
+        self,
+        module: &ModuleDef,
+        def: &Def,
+    ) -> Result<ArgsTuple, InvalidFunctionArguments> {
+        self.into_tuple(module.arg_seed_for(def))
+    }
+
     fn into_tuple<Def: FunctionDef>(self, seed: ArgsSeed<'_, Def>) -> Result<ArgsTuple, InvalidFunctionArguments> {
         self._into_tuple(seed).map_err(|err| InvalidFunctionArguments {
             err,

crates/core/src/host/module_host.rs

@@ -11,7 +11,6 @@ use crate::error::DBError;
 use crate::estimation::estimate_rows_scanned;
 use crate::hash::Hash;
 use crate::host::host_controller::CallProcedureReturn;
-use crate::host::scheduler::{handle_queued_call_reducer_params, QueueItem};
 use crate::host::v8::JsInstance;
 use crate::host::wasmtime::ModuleInstance;
 use crate::host::{InvalidFunctionArguments, InvalidViewArguments};
@@ -41,7 +40,7 @@ use spacetimedb_client_api_messages::websocket::{ByteListLen, Compression, OneOf
 use spacetimedb_data_structures::error_stream::ErrorStream;
 use spacetimedb_data_structures::map::{HashCollectionExt as _, IntMap};
 use spacetimedb_datastore::error::DatastoreError;
-use spacetimedb_datastore::execution_context::{ExecutionContext, ReducerContext, Workload, WorkloadType};
+use spacetimedb_datastore::execution_context::{Workload, WorkloadType};
 use spacetimedb_datastore::locking_tx_datastore::{MutTxId, ViewCallInfo};
 use spacetimedb_datastore::traits::{IsolationLevel, Program, TxData};
 use spacetimedb_durability::DurableOffset;
@@ -56,7 +55,6 @@ use spacetimedb_primitives::{ArgId, ProcedureId, TableId, ViewFnPtr, ViewId};
 use spacetimedb_query::compile_subscription;
 use spacetimedb_sats::{AlgebraicTypeRef, ProductValue};
 use spacetimedb_schema::auto_migrate::{AutoMigrateError, MigrationPolicy};
-use spacetimedb_schema::def::deserialize::ArgsSeed;
 use spacetimedb_schema::def::{ModuleDef, ProcedureDef, ReducerDef, TableDef, ViewDef};
 use spacetimedb_schema::schema::{Schema, TableSchema};
 use spacetimedb_vm::relation::RelValue;
@@ -608,59 +606,6 @@ pub fn call_identity_connected(
     }
 }
 
-// Only for logging purposes.
-const SCHEDULED_REDUCER: &str = "scheduled_reducer";
-
-pub(crate) fn call_scheduled_reducer(
-    module: &ModuleInfo,
-    queue_item: QueueItem,
-    call_reducer: impl FnOnce(Option<MutTxId>, CallReducerParams) -> (ReducerCallResult, bool),
-) -> (Result<(ReducerCallResult, Timestamp), ReducerCallError>, bool) {
-    extract_trapped(call_scheduled_reducer_inner(module, queue_item, call_reducer))
-}
-
-fn call_scheduled_reducer_inner(
-    module: &ModuleInfo,
-    item: QueueItem,
-    call_reducer: impl FnOnce(Option<MutTxId>, CallReducerParams) -> (ReducerCallResult, bool),
-) -> Result<((ReducerCallResult, Timestamp), bool), ReducerCallError> {
-    let db = &module.relational_db();
-    let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Internal);
-
-    match handle_queued_call_reducer_params(&tx, module, db, item) {
-        Ok(Some(params)) => {
-            // Is necessary to patch the context with the actual calling reducer
-            let reducer_def = module
-                .module_def
-                .get_reducer_by_id(params.reducer_id)
-                .ok_or(ReducerCallError::ScheduleReducerNotFound)?;
-            let reducer = &*reducer_def.name;
-
-            tx.ctx = ExecutionContext::with_workload(
-                tx.ctx.database_identity(),
-                Workload::Reducer(ReducerContext {
-                    name: reducer.into(),
-                    caller_identity: params.caller_identity,
-                    caller_connection_id: params.caller_connection_id,
-                    timestamp: Timestamp::now(),
-                    arg_bsatn: params.args.get_bsatn().clone(),
-                }),
-            );
-
-            let timestamp = params.timestamp;
-            let (res, trapped) = call_reducer(Some(tx), params);
-            Ok(((res, timestamp), trapped))
-        }
-        Ok(None) => Err(ReducerCallError::ScheduleReducerNotFound),
-        Err(err) => Err(ReducerCallError::Args(InvalidReducerArguments(
-            InvalidFunctionArguments {
-                err,
-                function_name: SCHEDULED_REDUCER.into(),
-            },
-        ))),
-    }
-}
-
 pub struct CallReducerParams {
     pub timestamp: Timestamp,
     pub caller_identity: Identity,
@@ -671,9 +616,10 @@ pub struct CallReducerParams {
     pub reducer_id: ReducerId,
     pub args: ArgsTuple,
 }
+
 impl CallReducerParams {
-    /// Returns a set of parameters for a call that came from within
-    /// and without a client/caller/request_id.
+    /// Returns a set of parameters for an internal call
+    /// without a client/caller/request_id.
     pub fn from_system(
         timestamp: Timestamp,
         caller_identity: Identity,
@@ -718,6 +664,26 @@ pub struct CallProcedureParams {
     pub args: ArgsTuple,
 }
 
+impl CallProcedureParams {
+    /// Returns a set of parameters for an internal call
+    /// without a client/caller/request_id.
+    pub fn from_system(
+        timestamp: Timestamp,
+        caller_identity: Identity,
+        procedure_id: ProcedureId,
+        args: ArgsTuple,
+    ) -> Self {
+        Self {
+            timestamp,
+            caller_identity,
+            caller_connection_id: ConnectionId::ZERO,
+            timer: None,
+            procedure_id,
+            args,
+        }
+    }
+}
+
 /// Holds a [`Module`] and a set of [`Instance`]s from it,
 /// and allocates the [`Instance`]s to be used for function calls.
 ///
@@ -1442,8 +1408,9 @@ impl ModuleHost {
         reducer_def: &ReducerDef,
         args: FunctionArgs,
     ) -> Result<CallReducerParams, InvalidReducerArguments> {
-        let reducer_seed = ArgsSeed(module.module_def.typespace().with_type(reducer_def));
-        let args = args.into_tuple(reducer_seed).map_err(InvalidReducerArguments)?;
+        let args = args
+            .into_tuple_for_def(&module.module_def, reducer_def)
+            .map_err(InvalidReducerArguments)?;
         let caller_connection_id = caller_connection_id.unwrap_or(ConnectionId::ZERO);
         Ok(CallReducerParams {
             timestamp: Timestamp::now(),
@@ -1457,6 +1424,21 @@ impl ModuleHost {
         })
     }
 
+    pub async fn call_reducer_with_params(
+        &self,
+        reducer_name: &str,
+        tx: Option<MutTxId>,
+        params: CallReducerParams,
+    ) -> Result<ReducerCallResult, NoSuchModule> {
+        self.call(
+            reducer_name,
+            (tx, params),
+            |(tx, p), inst| inst.call_reducer(tx, p),
+            |(tx, p), inst| inst.call_reducer(tx, p),
+        )
+        .await
+    }
+
     async fn call_reducer_inner(
         &self,
         caller_identity: Identity,
@@ -1468,8 +1450,9 @@ impl ModuleHost {
         reducer_def: &ReducerDef,
         args: FunctionArgs,
     ) -> Result<ReducerCallResult, ReducerCallError> {
-        let reducer_seed = ArgsSeed(self.info.module_def.typespace().with_type(reducer_def));
-        let args = args.into_tuple(reducer_seed).map_err(InvalidReducerArguments)?;
+        let args = args
+            .into_tuple_for_def(&self.info.module_def, reducer_def)
+            .map_err(InvalidReducerArguments)?;
         let caller_connection_id = caller_connection_id.unwrap_or(ConnectionId::ZERO);
         let call_reducer_params = CallReducerParams {
             timestamp: Timestamp::now(),
@@ -1483,12 +1466,7 @@ impl ModuleHost {
         };
 
         Ok(self
-            .call(
-                &reducer_def.name,
-                call_reducer_params,
-                |p, inst| inst.call_reducer(None, p),
-                |p, inst| inst.call_reducer(None, p),
-            )
+            .call_reducer_with_params(&reducer_def.name, None, call_reducer_params)
             .await?)
     }
 
@@ -1593,9 +1571,10 @@ impl ModuleHost {
         procedure_def: &ProcedureDef,
         args: FunctionArgs,
     ) -> Result<CallProcedureReturn, ProcedureCallError> {
-        let procedure_seed = ArgsSeed(self.info.module_def.typespace().with_type(procedure_def));
+        let args = args
+            .into_tuple_for_def(&self.info.module_def, procedure_def)
+            .map_err(InvalidProcedureArguments)?;
         let caller_connection_id = caller_connection_id.unwrap_or(ConnectionId::ZERO);
-        let args = args.into_tuple(procedure_seed).map_err(InvalidProcedureArguments)?;
 
         let params = CallProcedureParams {
             timestamp: Timestamp::now(),
@@ -1605,31 +1584,26 @@ impl ModuleHost {
             procedure_id,
             args,
         };
-        self.call_async_with_instance(&procedure_def.name, async move |inst| match inst {
+        self.call_procedure_with_params(&procedure_def.name, params)
+            .await
+            .map_err(Into::into)
+    }
+
+    // This is not reused in `call_procedure_inner`
+    // due to concerns re. `Timestamp::now`.
+    pub async fn call_procedure_with_params(
+        &self,
+        name: &str,
+        params: CallProcedureParams,
+    ) -> Result<CallProcedureReturn, NoSuchModule> {
+        self.call_async_with_instance(name, async move |inst| match inst {
             Instance::Wasm(mut inst) => (inst.call_procedure(params).await, Instance::Wasm(inst)),
             Instance::Js(inst) => {
                 let (r, s) = inst.call_procedure(params).await;
                 (r, Instance::Js(s))
             }
         })
         .await
-        .map_err(Into::into)
-    }
-
-    // Scheduled reducers require a different function here to call their reducer
-    // because their reducer arguments are stored in the database and need to be fetched
-    // within the same transaction as the reducer call.
-    pub(crate) async fn call_scheduled_reducer(
-        &self,
-        item: QueueItem,
-    ) -> Result<(ReducerCallResult, Timestamp), ReducerCallError> {
-        self.call(
-            SCHEDULED_REDUCER,
-            item,
-            |item, inst| inst.call_scheduled_reducer(item),
-            |item, inst| inst.call_scheduled_reducer(item),
-        )
-        .await?
     }
 
     /// Materializes the views return by the `view_collector`, if not already materialized,
@@ -1722,9 +1696,9 @@ impl ModuleHost {
         let view_def = module_def.view(view_name).ok_or(ViewCallError::NoSuchView)?;
         let fn_ptr = view_def.fn_ptr;
         let row_type = view_def.product_type_ref;
-        let typespace = module_def.typespace().with_type(view_def);
-        let view_seed = ArgsSeed(typespace);
-        let args = args.into_tuple(view_seed).map_err(InvalidViewArguments)?;
+        let args = args
+            .into_tuple_for_def(module_def, view_def)
+            .map_err(InvalidViewArguments)?;
 
         match self
             .call_view_inner(tx, view_name, view_id, table_id, fn_ptr, caller, sender, args, row_type)