fix deadlock

Author: Shubham8287

crates/client-api/src/routes/database.rs

@@ -263,22 +263,13 @@ pub async fn prepare<S: ControlStateDelegate + NodeDelegate>(
 
     let (module, Database { owner_identity, .. }) = find_module_and_database(&worker_ctx, name_or_identity).await?;
 
-    let connection_id = generate_random_connection_id();
-
-    module
-        .call_identity_connected(auth.into(), connection_id)
-        .await
-        .map_err(client_connected_error_to_response)?;
-
+    // 2PC prepare is a server-to-server call; no client lifecycle management needed.
+    // call_identity_connected/disconnected submit jobs to the module's executor, which
+    // will be blocked holding the 2PC write lock after prepare_reducer returns — deadlock.
     let result = module
-        .prepare_reducer(caller_identity, Some(connection_id), &reducer, args)
+        .prepare_reducer(caller_identity, None, &reducer, args)
         .await;
 
-    module
-        .call_identity_disconnected(caller_identity, connection_id)
-        .await
-        .map_err(client_disconnected_error_to_response)?;
-
     match result {
         Ok((prepare_id, rcr, return_value)) => {
             let (status, body) =

crates/core/src/host/mod.rs

@@ -1,6 +1,7 @@
 use anyhow::Context;
 use bytes::Bytes;
 use bytestring::ByteString;
+use core::future::Future;
 use derive_more::Display;
 use enum_map::Enum;
 use once_cell::sync::OnceCell;
@@ -10,6 +11,31 @@ use spacetimedb_lib::ProductValue;
 use spacetimedb_schema::def::deserialize::{ArgsSeed, FunctionDef};
 use spacetimedb_schema::def::ModuleDef;
 
+/// Block on `fut` from a synchronous context that may be inside a Tokio runtime.
+///
+/// `Handle::block_on` and `block_in_place` both panic when the calling thread is
+/// a custom (`std::thread::spawn`) thread that has entered the runtime via
+/// `Handle::enter()` — which is exactly the pattern used by `SingleCoreExecutor`.
+///
+/// The fix (same as the non-2PC `call_reducer_on_db` path): spawn a **scoped**
+/// OS thread.  The scoped thread starts with no Tokio context, so `Handle::block_on`
+/// works normally and drives the future using the **original** runtime's I/O reactor
+/// and connection pools.
+///
+/// Use this for every place in the 2PC / cross-DB call paths that needs to
+/// synchronously drive a future from blocking (WASM executor) context.
+pub(crate) fn block_on_scoped<F>(handle: &tokio::runtime::Handle, fut: F) -> F::Output
+where
+    F: Future + Send,
+    F::Output: Send,
+{
+    std::thread::scope(|s| {
+        s.spawn(|| handle.block_on(fut))
+            .join()
+            .expect("block_on_scoped: thread panicked")
+    })
+}
+
 mod disk_storage;
 mod host_controller;
 mod module_common;

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

@@ -13,6 +13,7 @@ use crate::host::module_host::{
     ViewCallResult, ViewCommand, ViewCommandResult, ViewOutcome,
 };
 use crate::host::scheduler::{CallScheduledFunctionResult, ScheduledFunctionParams};
+use crate::host::block_on_scoped;
 use crate::host::{
     ArgsTuple, ModuleHost, ProcedureCallError, ProcedureCallResult, ReducerCallError, ReducerCallResult, ReducerId,
     ReducerOutcome, Scheduler, UpdateDatabaseResult,
@@ -698,7 +699,7 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
         if let Some(prepare_offset) = marker_tx_data.tx_offset() {
             if let Some(mut durable) = stdb.durable_tx_offset() {
                 let handle = tokio::runtime::Handle::current();
-                let _ = handle.block_on(durable.wait_for(prepare_offset));
+                let _ = block_on_scoped(&handle, durable.wait_for(prepare_offset));
             }
         }
 
@@ -725,11 +726,13 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
             // Without this, A could delete its coordinator log entry while B's commit
             // is still in-memory — a B crash at that point would leave the tx uncommitted
             // with no way to recover (A has already forgotten it committed).
-            let handle = tokio::runtime::Handle::current();
             if let Some(mut durable) = stdb.durable_tx_offset() {
-                if let Ok(offset) = handle.block_on(commit_result.tx_offset) {
-                    let _ = handle.block_on(durable.wait_for(offset));
-                }
+                let handle = tokio::runtime::Handle::current();
+                block_on_scoped(&handle, async move {
+                    if let Ok(offset) = commit_result.tx_offset.await {
+                        let _ = durable.wait_for(offset).await;
+                    }
+                });
             }
 
             // Notify coordinator that B has committed so it can delete its coordinator log entry.
@@ -738,7 +741,7 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
             let router = replica_ctx.call_reducer_router.clone();
             let client_http = replica_ctx.call_reducer_client.clone();
             let auth_token = replica_ctx.call_reducer_auth_token.clone();
-            handle.spawn(send_ack_commit_to_coordinator(
+            tokio::runtime::Handle::current().spawn(send_ack_commit_to_coordinator(
                 client_http,
                 router,
                 auth_token,
@@ -787,19 +790,13 @@ impl<T: WasmInstance> WasmModuleInstance<T> {
         let auth_token = replica_ctx.call_reducer_auth_token.clone();
         let prepare_id_owned = prepare_id.to_owned();
         loop {
-            let decision = std::thread::scope(|s| {
-                s.spawn(|| {
-                    handle.block_on(Self::query_coordinator_status(
-                        &client,
-                        &router,
-                        auth_token.clone(),
-                        coordinator_identity,
-                        &prepare_id_owned,
-                    ))
-                })
-                .join()
-                .expect("coordinator poll thread panicked")
-            });
+            let decision = block_on_scoped(&handle, Self::query_coordinator_status(
+                &client,
+                &router,
+                auth_token.clone(),
+                coordinator_identity,
+                &prepare_id_owned,
+            ));
             match decision {
                 Some(commit) => return commit,
                 None => std::thread::sleep(Duration::from_secs(5)),
@@ -1157,9 +1154,7 @@ impl InstanceCommon {
 
             let replica_ctx = inst.replica_ctx().clone();
             let handle = tokio::runtime::Handle::current();
-            std::thread::scope(|s| {
-                s.spawn(|| {
-                    handle.block_on(async {
+            block_on_scoped(&handle, async {
                         // Wait for A's coordinator log (committed atomically with the tx) to be
                         // durable before sending COMMIT to B.  This guarantees that if A crashes
                         // after sending COMMIT, recovery can retransmit from the durable log.
@@ -1218,10 +1213,6 @@ impl InstanceCommon {
                                 }
                             }
                         }
-                    });
-                })
-                .join()
-                .expect("2PC coordination thread panicked");
             });
         }
 

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

@@ -1996,19 +1996,11 @@ impl WasmInstanceEnv {
             let args_buf = mem.deref_slice(args_ptr, args_len)?;
             let args = bytes::Bytes::copy_from_slice(args_buf);
 
-            // Reducers run inside a tokio LocalSet (single-threaded), so block_in_place
-            // is unavailable and futures::executor::block_on can't drive tokio I/O.
-            // Spawn a new OS thread and call Handle::block_on from there, which is
-            // designed to be called from synchronous (non-async) contexts.
             let handle = tokio::runtime::Handle::current();
             let fut = env
                 .instance_env
                 .call_reducer_on_db(database_identity, &reducer_name, args);
-            let result = std::thread::scope(|s| {
-                s.spawn(|| handle.block_on(fut))
-                    .join()
-                    .expect("call_reducer_on_db: worker thread panicked")
-            });
+            let result = super::super::block_on_scoped(&handle, fut);
 
             match result {
                 Ok((status, body)) => {
@@ -2067,11 +2059,7 @@ impl WasmInstanceEnv {
             let fut = env
                 .instance_env
                 .call_reducer_on_db_2pc(database_identity, &reducer_name, args);
-            let result = std::thread::scope(|s| {
-                s.spawn(|| handle.block_on(fut))
-                    .join()
-                    .expect("call_reducer_on_db_2pc: worker thread panicked")
-            });
+            let result = super::super::block_on_scoped(&handle, fut);
 
             match result {
                 Ok((status, body, prepare_id)) => {

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -29,6 +29,7 @@ use crate::{
 use crate::{
     locking_tx_datastore::ViewCallInfo,
     system_tables::{
+        ST_2PC_COORDINATOR_LOG_ID, ST_2PC_COORDINATOR_LOG_IDX, ST_2PC_STATE_ID, ST_2PC_STATE_IDX,
         ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ACCESSOR_IDX, ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_IDX,
         ST_EVENT_TABLE_ID, ST_EVENT_TABLE_IDX, ST_INDEX_ACCESSOR_ID, ST_INDEX_ACCESSOR_IDX, ST_TABLE_ACCESSOR_ID,
         ST_TABLE_ACCESSOR_IDX, ST_VIEW_COLUMN_ID, ST_VIEW_COLUMN_IDX, ST_VIEW_ID, ST_VIEW_IDX, ST_VIEW_PARAM_ID,
@@ -478,6 +479,9 @@ impl CommittedState {
         self.create_table(ST_INDEX_ACCESSOR_ID, schemas[ST_INDEX_ACCESSOR_IDX].clone());
         self.create_table(ST_COLUMN_ACCESSOR_ID, schemas[ST_COLUMN_ACCESSOR_IDX].clone());
 
+        self.create_table(ST_2PC_STATE_ID, schemas[ST_2PC_STATE_IDX].clone());
+        self.create_table(ST_2PC_COORDINATOR_LOG_ID, schemas[ST_2PC_COORDINATOR_LOG_IDX].clone());
+
         // Insert the sequences into `st_sequences`
         let (st_sequences, blob_store, pool) =
             self.get_table_and_blob_store_or_create(ST_SEQUENCE_ID, &schemas[ST_SEQUENCE_IDX]);

crates/datastore/src/system_tables.rs

@@ -286,6 +286,8 @@ pub(crate) const ST_EVENT_TABLE_IDX: usize = 16;
 pub(crate) const ST_TABLE_ACCESSOR_IDX: usize = 17;
 pub(crate) const ST_INDEX_ACCESSOR_IDX: usize = 18;
 pub(crate) const ST_COLUMN_ACCESSOR_IDX: usize = 19;
+pub(crate) const ST_2PC_STATE_IDX: usize = 20;
+pub(crate) const ST_2PC_COORDINATOR_LOG_IDX: usize = 21;
 
 macro_rules! st_fields_enum {
     ($(#[$attr:meta])* enum $ty_name:ident { $($name:expr, $var:ident = $discr:expr,)* }) => {