Add CallEdgeTracker trait for distributed deadlock detection

When database A calls a reducer on database B, register the edge A -> B
with a CallEdgeTracker. If the edge would create a cycle (B is already
waiting for A), return CycleDetected error to abort the call before it
deadlocks.

- New trait CallEdgeTracker with register_edge, unregister_edge, is_cancelled
- NoopCallEdgeTracker for standalone (always succeeds, no cycle detection)
- Edge registration in call_reducer_on_db and call_reducer_on_db_2pc
- New NodesError::CycleDetected variant
- New errno::CYCLE_DETECTED (22) for wasm ABI
- Cloud implementation will provide a real tracker that talks to the
  control DB for cycle detection

Author: cloutiertyler

crates/bindings-sys/src/lib.rs

@@ -1520,7 +1520,7 @@ pub fn call_reducer_on_db(
     // on transport failure. Unlike other ABI functions, a non-zero return value here
     // does NOT indicate a generic errno — it's the HTTP status code. Only HTTP_ERROR
     // specifically signals a transport-level failure.
-    if status == Errno::HTTP_ERROR.code() {
+    if status == Errno::HTTP_ERROR.code() || status == Errno::CYCLE_DETECTED.code() {
         Err(out)
     } else {
         Ok((status, out))
@@ -1551,7 +1551,7 @@ pub fn call_reducer_on_db_2pc(
             &mut out,
         )
     };
-    if status == Errno::HTTP_ERROR.code() {
+    if status == Errno::HTTP_ERROR.code() || status == Errno::CYCLE_DETECTED.code() {
         Err(out)
     } else {
         Ok((status, out))

crates/core/src/error.rs

@@ -277,6 +277,8 @@ pub enum NodesError {
     ScheduleError(#[source] ScheduleError),
     #[error("HTTP request failed: {0}")]
     HttpError(String),
+    #[error("distributed deadlock detected: call would create a cycle")]
+    CycleDetected,
 }
 
 impl From<DBError> for NodesError {

crates/core/src/host/call_edge_tracker.rs

@@ -0,0 +1,61 @@
+//! Distributed deadlock detection via call-edge tracking.
+//!
+//! When database A calls a reducer on database B, the edge A -> B is registered
+//! with the control DB. If inserting the edge creates a cycle in the call graph,
+//! the call that created the cycle is cancelled.
+//!
+//! The [`CallEdgeTracker`] trait provides the interface. Standalone uses a no-op
+//! implementation; cloud implements cycle detection via the control DB.
+
+use spacetimedb_lib::Identity;
+
+/// Error returned when registering a call edge would create a cycle.
+#[derive(Debug)]
+pub struct CycleDetected;
+
+impl std::fmt::Display for CycleDetected {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "distributed deadlock detected: call would create a cycle")
+    }
+}
+
+impl std::error::Error for CycleDetected {}
+
+/// Tracks cross-database call edges for distributed deadlock detection.
+///
+/// Methods are blocking (not async) because they are called from the database
+/// thread, which must not enter an async runtime.
+pub trait CallEdgeTracker: Send + Sync {
+    /// Register that `caller` is about to call a reducer on `target`.
+    ///
+    /// Returns `Err(CycleDetected)` if the edge would create a cycle in the
+    /// call graph (i.e. `target` is already waiting, directly or transitively,
+    /// for `caller`).
+    fn register_edge(&self, caller: Identity, target: Identity) -> Result<(), CycleDetected>;
+
+    /// Remove the edge after the call completes (success or failure).
+    fn unregister_edge(&self, caller: Identity, target: Identity);
+
+    /// Check whether a pending call from `caller` to `target` has been
+    /// asynchronously cancelled (e.g. because a cycle was detected after
+    /// the edge was registered).
+    fn is_cancelled(&self, caller: Identity, target: Identity) -> bool;
+}
+
+/// No-op implementation for standalone (single-node) deployments.
+///
+/// Always succeeds, never detects cycles. Distributed deadlocks are not
+/// possible when the control DB is not involved.
+pub struct NoopCallEdgeTracker;
+
+impl CallEdgeTracker for NoopCallEdgeTracker {
+    fn register_edge(&self, _caller: Identity, _target: Identity) -> Result<(), CycleDetected> {
+        Ok(())
+    }
+
+    fn unregister_edge(&self, _caller: Identity, _target: Identity) {}
+
+    fn is_cancelled(&self, _caller: Identity, _target: Identity) -> bool {
+        false
+    }
+}

crates/core/src/host/host_controller.rs

@@ -729,6 +729,7 @@ async fn make_replica_ctx(
         call_reducer_auth_token,
         prepared_txs: crate::host::prepared_tx::PreparedTransactions::new(),
         on_panic: std::sync::Arc::new(std::sync::OnceLock::new()),
+        call_edge_tracker: std::sync::Arc::new(crate::host::call_edge_tracker::NoopCallEdgeTracker),
     })
 }
 

crates/core/src/host/instance_env.rs

@@ -1016,10 +1016,16 @@ impl InstanceEnv {
         // accepts the request without generating a fresh ephemeral identity per call.
         let auth_token = self.replica_ctx.call_reducer_auth_token.clone();
         let caller_identity = self.replica_ctx.database.database_identity;
+        let edge_tracker = self.replica_ctx.call_edge_tracker.clone();
 
         async move {
             let start = Instant::now();
 
+            // Register call edge for distributed deadlock detection.
+            edge_tracker
+                .register_edge(caller_identity, database_identity)
+                .map_err(|_| NodesError::CycleDetected)?;
+
             let base_url = router
                 .resolve_base_url(database_identity)
                 .await
@@ -1048,6 +1054,9 @@ impl InstanceEnv {
             }
             .await;
 
+            // Unregister the edge now that the call has completed.
+            edge_tracker.unregister_edge(caller_identity, database_identity);
+
             WORKER_METRICS
                 .cross_db_reducer_calls_total
                 .with_label_values(&caller_identity)
@@ -1082,10 +1091,16 @@ impl InstanceEnv {
         let reducer_name = reducer_name.to_owned();
         let auth_token = self.replica_ctx.call_reducer_auth_token.clone();
         let caller_identity = self.replica_ctx.database.database_identity;
+        let edge_tracker = self.replica_ctx.call_edge_tracker.clone();
 
         async move {
             let start = Instant::now();
 
+            // Register call edge for distributed deadlock detection.
+            edge_tracker
+                .register_edge(caller_identity, database_identity)
+                .map_err(|_| NodesError::CycleDetected)?;
+
             let base_url = router
                 .resolve_base_url(database_identity)
                 .await
@@ -1120,6 +1135,9 @@ impl InstanceEnv {
             }
             .await;
 
+            // Unregister the edge now that the call has completed.
+            edge_tracker.unregister_edge(caller_identity, database_identity);
+
             WORKER_METRICS
                 .cross_db_reducer_calls_total
                 .with_label_values(&caller_identity)
@@ -1511,6 +1529,7 @@ mod test {
                 call_reducer_auth_token: None,
                 prepared_txs: crate::host::prepared_tx::PreparedTransactions::new(),
                 on_panic: std::sync::Arc::new(std::sync::OnceLock::new()),
+                call_edge_tracker: std::sync::Arc::new(crate::host::call_edge_tracker::NoopCallEdgeTracker),
             },
             runtime,
         ))

crates/core/src/host/mod.rs

@@ -36,6 +36,7 @@ where
     })
 }
 
+pub mod call_edge_tracker;
 mod disk_storage;
 mod host_controller;
 mod module_common;

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

@@ -2016,6 +2016,15 @@ impl WasmInstanceEnv {
                     bytes_source.0.write_to(mem, out)?;
                     Ok(errno::HTTP_ERROR.get() as u32)
                 }
+                Err(NodesError::CycleDetected) => {
+                    let err_msg = "distributed deadlock detected: call would create a cycle";
+                    let err_bytes = bsatn::to_vec(&err_msg).with_context(|| {
+                        "Failed to BSATN-serialize cycle detection error"
+                    })?;
+                    let bytes_source = WasmInstanceEnv::create_bytes_source(env, err_bytes.into())?;
+                    bytes_source.0.write_to(mem, out)?;
+                    Ok(errno::CYCLE_DETECTED.get() as u32)
+                }
                 Err(e) => Err(WasmError::Db(e)),
             }
         })
@@ -2081,6 +2090,15 @@ impl WasmInstanceEnv {
                     bytes_source.0.write_to(mem, out)?;
                     Ok(errno::HTTP_ERROR.get() as u32)
                 }
+                Err(NodesError::CycleDetected) => {
+                    let err_msg = "distributed deadlock detected: call would create a cycle";
+                    let err_bytes = bsatn::to_vec(&err_msg).with_context(|| {
+                        "Failed to BSATN-serialize cycle detection error"
+                    })?;
+                    let bytes_source = WasmInstanceEnv::create_bytes_source(env, err_bytes.into())?;
+                    bytes_source.0.write_to(mem, out)?;
+                    Ok(errno::CYCLE_DETECTED.get() as u32)
+                }
                 Err(e) => Err(WasmError::Db(e)),
             }
         })

crates/core/src/replica_context.rs

@@ -3,6 +3,7 @@ use spacetimedb_commitlog::SizeOnDisk;
 use super::database_logger::DatabaseLogger;
 use crate::db::relational_db::RelationalDB;
 use crate::error::DBError;
+use crate::host::call_edge_tracker::CallEdgeTracker;
 use crate::host::prepared_tx::PreparedTransactions;
 use crate::host::reducer_router::ReducerCallRouter;
 use crate::messages::control_db::Database;
@@ -72,6 +73,9 @@ pub struct ReplicaContext {
     /// async task that can't panic on the WASM executor thread (e.g., 2PC persistence
     /// abort in Round 2).  Set once by `launch_module`; empty in tests.
     pub on_panic: Arc<OnceLock<Box<dyn Fn() + Send + Sync + 'static>>>,
+    /// Distributed deadlock detection via call-edge tracking.
+    /// Standalone: no-op. Cloud: registers edges with control DB for cycle detection.
+    pub call_edge_tracker: Arc<dyn CallEdgeTracker>,
 }
 
 impl ReplicaContext {

crates/primitives/src/errno.rs

@@ -35,6 +35,7 @@ macro_rules! errnos {
                 "ABI call can only be made while within a read-only transaction"
             ),
             HTTP_ERROR(21, "The HTTP request failed"),
+            CYCLE_DETECTED(22, "Distributed deadlock detected: call would create a cycle"),
         );
     };
 }