Fix the issues with scheduling procedures (#3816)

# Description of Changes

This reapplies the patch from #3704, and fixes the issues that were
causing it to deadlock.

The reason it was deadlocking was that it allowed for the following
sequence of events:
* `SchedulerActor::handle_queued()` begins mutable tx
* `ModuleHost::disconnect_client()` submits call to `call_reducer(tx:
None)`
* scheduler submits call to `call_reducer(tx: Some)`
* `WasmModuleInstance::disconnect_client` now has to try to take tx
lock, but the scheduler's call_reducer already holds it and is behind it
in the queue

So, I moved most of the logic from `handle_queued` back to being
executed in the module worker thread, but kept the code in
`scheduler.rs` so that it can all be reasoned about locally.

Fixes #3645. Should I uncomment the implementation of
`ExportFunctionForScheduledTable for F: Procedure` now?

# Expected complexity level and risk

2 - there's a chance that this patch hasn't fully fixed the deadlock
issue from #3704, but I'm quite confident.

# Testing

- [x] Manually verified that deadlock no longer occurs - previously,
`while true; do python -m smoketests schedule_reducer -k
test_scheduled_table_subscription; done` would freeze up in only 2 or 3
iterations, but now it can run for 10 minutes without issues.

Author: coolreader18

crates/bindings/src/lib.rs

@@ -731,17 +731,16 @@ pub use spacetimedb_bindings_macro::reducer;
 // TODO(procedure-http): add example with an HTTP request.
 // TODO(procedure-transaction): document obtaining and using a transaction within a procedure.
 ///
-// TODO(scheduled-procedures): Uncomment below docs.
-// /// # Scheduled procedures
+/// # Scheduled procedures
 // TODO(docs): after moving scheduled reducer docs into table secion, link there.
-// ///
-// /// Like [reducer]s, procedures can be made **scheduled**.
-// /// This allows calling procedures at a particular time, or in a loop.
-// /// It also allows reducers to enqueue procedure runs.
-// ///
-// /// Scheduled procedures are called on a best-effort basis and may be slightly delayed in their execution
-// /// when a database is under heavy load.
-// ///
+///
+/// Like [reducer]s, procedures can be made **scheduled**.
+/// This allows calling procedures at a particular time, or in a loop.
+/// It also allows reducers to enqueue procedure runs.
+///
+/// Scheduled procedures are called on a best-effort basis and may be slightly delayed in their execution
+/// when a database is under heavy load.
+///
 /// [clients]: https://spacetimedb.com/docs/#client
 // TODO(procedure-async): update docs and examples with `async`-ness.
 #[doc(inline)]

crates/bindings/src/rt.rs

@@ -444,9 +444,7 @@ pub struct FnKindView {
 /// See <https://willcrichton.net/notes/defeating-coherence-rust/> for details on this technique.
 #[cfg_attr(
     feature = "unstable",
-    // TODO(scheduled-procedures): uncomment this, delete other line
-    // doc = "It will be one of [`FnKindReducer`] or [`FnKindProcedure`] in modules that compile successfully."
-    doc = "It will be [`FnKindReducer`] in modules that compile successfully."
+    doc = "It will be one of [`FnKindReducer`] or [`FnKindProcedure`] in modules that compile successfully."
 )]
 #[cfg_attr(
     not(feature = "unstable"),
@@ -467,16 +465,15 @@ impl<'de, TableRow: SpacetimeType + Serialize + Deserialize<'de>, F: Reducer<'de
 {
 }
 
-// TODO(scheduled-procedures): uncomment this to syntactically allow scheduled procedures.
-// #[cfg(feature = "unstable")]
-// impl<
-//         'de,
-//         TableRow: SpacetimeType + Serialize + Deserialize<'de>,
-//         Ret: SpacetimeType + Serialize + Deserialize<'de>,
-//         F: Procedure<'de, (TableRow,), Ret>,
-//     > ExportFunctionForScheduledTable<'de, TableRow, FnKindProcedure<Ret>> for F
-// {
-// }
+#[cfg(feature = "unstable")]
+impl<
+        'de,
+        TableRow: SpacetimeType + Serialize + Deserialize<'de>,
+        Ret: SpacetimeType + Serialize + Deserialize<'de>,
+        F: Procedure<'de, (TableRow,), Ret>,
+    > ExportFunctionForScheduledTable<'de, TableRow, FnKindProcedure<Ret>> for F
+{
+}
 
 // the macro generates <T as SpacetimeType>::make_type::<DummyTypespace>
 pub struct DummyTypespace;

crates/bindings/tests/ui/views.stderr

@@ -361,29 +361,3 @@ error[E0277]: the trait bound `NotSpacetimeType: SpacetimeType` is not satisfied
               ColId
             and $N others
     = note: required for `Option<NotSpacetimeType>` to implement `SpacetimeType`
-
-error[E0631]: type mismatch in function arguments
-   --> tests/ui/views.rs:142:56
-    |
-142 | #[spacetimedb::table(name = scheduled_table, scheduled(scheduled_table_view))]
-    | -------------------------------------------------------^^^^^^^^^^^^^^^^^^^^---
-    | |                                                      |
-    | |                                                      expected due to this
-    | required by a bound introduced by this call
-...
-154 | fn scheduled_table_view(_: &ViewContext, _args: ScheduledTable) -> Vec<Player> {
-    | ------------------------------------------------------------------------------ found signature defined here
-    |
-    = note: expected function signature `for<'a> fn(&'a ReducerContext, ScheduledTable) -> _`
-               found function signature `fn(&ViewContext, ScheduledTable) -> _`
-    = note: required for `for<'a> fn(&'a ViewContext, ScheduledTable) -> Vec<Player> {scheduled_table_view}` to implement `Reducer<'_, (ScheduledTable,)>`
-    = note: required for `for<'a> fn(&'a ViewContext, ScheduledTable) -> Vec<Player> {scheduled_table_view}` to implement `ExportFunctionForScheduledTable<'_, ScheduledTable, {type error}>`
-note: required by a bound in `scheduled_typecheck`
-   --> src/rt.rs
-    |
-    | pub const fn scheduled_typecheck<'de, Row, FnKind>(_x: impl ExportFunctionForScheduledTable<'de, Row, FnKind>)
-    |                                                             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `scheduled_typecheck`
-help: consider wrapping the function in a closure
-    |
-142 | #[spacetimedb::table(name = scheduled_table, scheduled(|arg0: &ReducerContext, arg1: ScheduledTable| scheduled_table_view(/* &ViewContext */, arg1)))]
-    |                                                        +++++++++++++++++++++++++++++++++++++++++++++                     ++++++++++++++++++++++++++

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 {
@@ -3577,6 +3586,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 {
@@ -3887,6 +3897,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 {
@@ -4050,6 +4061,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 {
@@ -4460,6 +4472,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 {
@@ -5013,6 +5026,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,