Add distributed deadlock detection via call edge tracking

Before making a cross-database reducer call, register an edge A -> B with
a CallEdgeTracker. If adding the edge would create a cycle (distributed
deadlock), the tracker returns an error. The caller retries with exponential
backoff (5 attempts), then fails with a deadlock error.

New trait CallEdgeTracker in core::host::call_edge_tracker with:
  - register_edge(call_id, caller, callee) -> Result<()>
  - unregister_edge(call_id) -> Result<()>
  - unregister_all_edges() -> Result<()> (crash cleanup on startup)

NoopCallEdgeTracker for standalone (always allows calls).
Cloud implementation will call control DB reducers for cycle detection.

Also added register/unregister_reducer_call_edge methods to
ControlStateWriteAccess trait (no-op in standalone).

Author: cloutiertyler

crates/client-api/src/lib.rs

@@ -344,6 +344,23 @@ pub trait ControlStateWriteAccess: Send + Sync {
         owner_identity: &Identity,
         domain_names: &[DomainName],
     ) -> anyhow::Result<SetDomainsResult>;
+
+    // Cross-database call edge tracking (distributed deadlock detection)
+
+    /// Register a cross-database call edge for deadlock detection.
+    /// Returns `Err` if the edge would create a cycle (distributed deadlock).
+    async fn register_reducer_call_edge(
+        &self,
+        call_id: &str,
+        caller: &Identity,
+        callee: &Identity,
+    ) -> anyhow::Result<()>;
+
+    /// Unregister a cross-database call edge after the call completes.
+    async fn unregister_reducer_call_edge(&self, call_id: &str) -> anyhow::Result<()>;
+
+    /// Unregister all edges for this node (crash cleanup on startup).
+    async fn unregister_all_reducer_call_edges_for_node(&self) -> anyhow::Result<()>;
 }
 
 #[async_trait]
@@ -454,6 +471,23 @@ impl<T: ControlStateWriteAccess + ?Sized> ControlStateWriteAccess for Arc<T> {
             .replace_dns_records(database_identity, owner_identity, domain_names)
             .await
     }
+
+    async fn register_reducer_call_edge(
+        &self,
+        call_id: &str,
+        caller: &Identity,
+        callee: &Identity,
+    ) -> anyhow::Result<()> {
+        (**self).register_reducer_call_edge(call_id, caller, callee).await
+    }
+
+    async fn unregister_reducer_call_edge(&self, call_id: &str) -> anyhow::Result<()> {
+        (**self).unregister_reducer_call_edge(call_id).await
+    }
+
+    async fn unregister_all_reducer_call_edges_for_node(&self) -> anyhow::Result<()> {
+        (**self).unregister_all_reducer_call_edges_for_node().await
+    }
 }
 
 #[async_trait]

crates/core/src/host/call_edge_tracker.rs

@@ -0,0 +1,59 @@
+/// Trait for tracking cross-database call edges for distributed deadlock detection.
+///
+/// Before making a cross-database reducer call, the caller registers an edge
+/// A -> B (caller -> callee). If this would create a cycle in the call graph,
+/// the registration fails with an error, indicating a potential distributed deadlock.
+///
+/// Implementations differ between deployment modes:
+///
+/// - **Standalone** -- [`NoopCallEdgeTracker`] always returns `Ok(())`.
+///   Single-node deployments cannot have distributed deadlocks.
+///
+/// - **Cluster** -- Calls a reducer on the control database that inserts the edge
+///   and runs cycle detection. Returns `Err` if a cycle is found.
+use spacetimedb_lib::Identity;
+use std::future::Future;
+use std::pin::Pin;
+
+pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
+
+pub trait CallEdgeTracker: Send + Sync + 'static {
+    /// Register a call edge: `caller` is about to call `callee`.
+    ///
+    /// Returns `Ok(())` if the edge was registered (no cycle).
+    /// Returns `Err` if registering this edge would create a cycle.
+    fn register_edge<'a>(
+        &'a self,
+        call_id: &'a str,
+        caller: Identity,
+        callee: Identity,
+    ) -> BoxFuture<'a, anyhow::Result<()>>;
+
+    /// Unregister a call edge after the call completes (success or failure).
+    fn unregister_edge<'a>(&'a self, call_id: &'a str) -> BoxFuture<'a, anyhow::Result<()>>;
+
+    /// Unregister all edges for this node (crash cleanup on startup).
+    fn unregister_all_edges<'a>(&'a self) -> BoxFuture<'a, anyhow::Result<()>>;
+}
+
+/// No-op implementation for standalone (single-node) deployments.
+pub struct NoopCallEdgeTracker;
+
+impl CallEdgeTracker for NoopCallEdgeTracker {
+    fn register_edge<'a>(
+        &'a self,
+        _call_id: &'a str,
+        _caller: Identity,
+        _callee: Identity,
+    ) -> BoxFuture<'a, anyhow::Result<()>> {
+        Box::pin(async { Ok(()) })
+    }
+
+    fn unregister_edge<'a>(&'a self, _call_id: &'a str) -> BoxFuture<'a, anyhow::Result<()>> {
+        Box::pin(async { Ok(()) })
+    }
+
+    fn unregister_all_edges<'a>(&'a self) -> BoxFuture<'a, anyhow::Result<()>> {
+        Box::pin(async { Ok(()) })
+    }
+}

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

@@ -1011,15 +1011,39 @@ impl InstanceEnv {
         let client = self.replica_ctx.call_reducer_client.clone();
         let router = self.replica_ctx.call_reducer_router.clone();
         let reducer_name = reducer_name.to_owned();
-        // Node-level auth token: a single token minted at startup and shared by all replicas
-        // on this node. Passed as a Bearer token so `anon_auth_middleware` on the target node
-        // 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.
+            // Retry with backoff if a cycle is detected.
+            let call_id = uuid::Uuid::new_v4().to_string();
+            const MAX_RETRIES: u32 = 5;
+            const BACKOFF_MS: u64 = 100;
+            for attempt in 0..MAX_RETRIES {
+                match edge_tracker.register_edge(&call_id, caller_identity, database_identity).await {
+                    Ok(()) => break,
+                    Err(e) if attempt < MAX_RETRIES - 1 => {
+                        log::warn!(
+                            "Cycle detected on call {caller_identity} -> {database_identity} \
+                             (attempt {attempt}): {e}; retrying"
+                        );
+                        tokio::time::sleep(std::time::Duration::from_millis(
+                            BACKOFF_MS * 2u64.pow(attempt),
+                        ))
+                        .await;
+                    }
+                    Err(e) => {
+                        return Err(NodesError::HttpError(format!(
+                            "distributed deadlock detected: {caller_identity} -> {database_identity}: {e}"
+                        )));
+                    }
+                }
+            }
+
             let base_url = router
                 .resolve_base_url(database_identity)
                 .await
@@ -1048,6 +1072,9 @@ impl InstanceEnv {
             }
             .await;
 
+            // Unregister the call edge (regardless of success/failure).
+            let _ = edge_tracker.unregister_edge(&call_id).await;
+
             WORKER_METRICS
                 .cross_db_reducer_calls_total
                 .with_label_values(&caller_identity)
@@ -1082,10 +1109,36 @@ 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.
+            let call_id = uuid::Uuid::new_v4().to_string();
+            const MAX_RETRIES: u32 = 5;
+            const BACKOFF_MS: u64 = 100;
+            for attempt in 0..MAX_RETRIES {
+                match edge_tracker.register_edge(&call_id, caller_identity, database_identity).await {
+                    Ok(()) => break,
+                    Err(e) if attempt < MAX_RETRIES - 1 => {
+                        log::warn!(
+                            "Cycle detected on 2PC call {caller_identity} -> {database_identity} \
+                             (attempt {attempt}): {e}; retrying"
+                        );
+                        tokio::time::sleep(std::time::Duration::from_millis(
+                            BACKOFF_MS * 2u64.pow(attempt),
+                        ))
+                        .await;
+                    }
+                    Err(e) => {
+                        return Err(NodesError::HttpError(format!(
+                            "distributed deadlock detected: {caller_identity} -> {database_identity}: {e}"
+                        )));
+                    }
+                }
+            }
+
             let base_url = router
                 .resolve_base_url(database_identity)
                 .await
@@ -1120,6 +1173,9 @@ impl InstanceEnv {
             }
             .await;
 
+            // Unregister the call edge (regardless of success/failure).
+            let _ = edge_tracker.unregister_edge(&call_id).await;
+
             WORKER_METRICS
                 .cross_db_reducer_calls_total
                 .with_label_values(&caller_identity)
@@ -1511,6 +1567,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: Arc::new(crate::host::call_edge_tracker::NoopCallEdgeTracker),
             },
             runtime,
         ))

crates/core/src/host/mod.rs

@@ -45,6 +45,7 @@ pub mod prepared_tx;
 pub mod scheduler;
 pub mod wasmtime;
 
+pub mod call_edge_tracker;
 // Visible for integration testing.
 pub mod instance_env;
 pub mod reducer_router;

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: tracks cross-database call edges.
+    /// Standalone uses [`crate::host::call_edge_tracker::NoopCallEdgeTracker`].
+    pub call_edge_tracker: Arc<dyn CallEdgeTracker>,
 }
 
 impl ReplicaContext {

crates/standalone/src/lib.rs

@@ -484,6 +484,25 @@ impl spacetimedb_client_api::ControlStateWriteAccess for StandaloneEnv {
             .control_db
             .spacetime_replace_domains(database_identity, owner_identity, domain_names)?)
     }
+
+    // Distributed deadlock detection: no-op in standalone (single node).
+
+    async fn register_reducer_call_edge(
+        &self,
+        _call_id: &str,
+        _caller: &Identity,
+        _callee: &Identity,
+    ) -> anyhow::Result<()> {
+        Ok(())
+    }
+
+    async fn unregister_reducer_call_edge(&self, _call_id: &str) -> anyhow::Result<()> {
+        Ok(())
+    }
+
+    async fn unregister_all_reducer_call_edges_for_node(&self) -> anyhow::Result<()> {
+        Ok(())
+    }
 }
 
 impl spacetimedb_client_api::Authorization for StandaloneEnv {