fix(coprocessor): fix handling of dependence cycles

Author: antoniupop

coprocessor/fhevm-engine/scheduler/src/dfg.rs

@@ -11,6 +11,7 @@ use daggy::petgraph::{
 };
 use daggy::{petgraph::graph::node_index, Dag, NodeIndex};
 use fhevm_engine_common::types::{Handle, SupportedFheCiphertexts, SupportedFheOperations};
+use tracing::error;
 
 #[derive(Debug)]
 pub struct DFGOp {
@@ -145,14 +146,59 @@ impl DFTxGraph {
                 }
             }
         }
+
+        // We build a replica of the graph and map it to the
+        // underlying DiGraph so we can identify cycles.
+        let mut digraph = self.graph.map(|idx, _| idx, |_, _| ()).graph().clone();
         // Add transaction dependence edges
-        for (producer, consumer) in dependence_pairs {
-            // Error only occurs in case of cyclic dependence which
-            // shoud not be possible between transactions. In that
-            // case, the whole cycle should be put in an error state.
-            self.graph
-                .add_edge(*producer, consumer, ())
-                .map_err(|_| SchedulerError::CyclicDependence)?;
+        for (producer, consumer) in dependence_pairs.iter() {
+            digraph.add_edge(**producer, *consumer, ());
+        }
+        let mut tarjan = daggy::petgraph::algo::TarjanScc::new();
+        let mut sccs = Vec::new();
+        tarjan.run(&self.graph.graph(), |scc| {
+            if scc.len() > 1 {
+                // All non-singleton SCCs in a directed graph are
+                // dependence cycles
+                sccs.push(scc.to_vec());
+            }
+        });
+        if !sccs.is_empty() {
+            for scc in sccs {
+                error!(target: "scheduler", { cycle_size = ?scc.len() },
+		       "Dependence cycle detected");
+                for idx in scc {
+                    // Mark the node as uncomputable so we don't go
+                    // and mark as completed operations that are in
+                    // error.
+                    self.set_uncomputable(idx, &Dag::new())?;
+                    let tx = self
+                        .graph
+                        .node_weight(idx)
+                        .ok_or(SchedulerError::DataflowGraphError)?;
+                    error!(target: "scheduler", { transaction_id = ?hex::encode(tx.transaction_id.clone()) },
+		       "Transaction is part of a dependence cycle");
+                    for (_, op) in tx.graph.graph.node_references() {
+                        self.results.push(DFGTxResult {
+                            transaction_id: tx.transaction_id.clone(),
+                            handle: op.result_handle.to_vec(),
+                            compressed_ct: Err(SchedulerError::CyclicDependence.into()),
+                        });
+                    }
+                }
+            }
+            return Err(SchedulerError::CyclicDependence.into());
+        } else {
+            // If no dependence cycles were found, then we can
+            // complete the graph and proceed to execution
+            for (producer, consumer) in dependence_pairs.iter() {
+                // The error case here should not happen as we've
+                // already covered it by testing for SCCs in the graph
+                // first
+                self.graph
+                    .add_edge(**producer, *consumer, ())
+                    .map_err(|_| SchedulerError::CyclicDependence)?;
+            }
         }
         Ok(())
     }

coprocessor/fhevm-engine/tfhe-worker/src/tfhe_worker.rs

@@ -443,7 +443,12 @@ async fn build_transaction_graph_and_execute<'a>(
     loop_ctx: &opentelemetry::Context,
 ) -> Result<DFTxGraph, Box<dyn std::error::Error + Send + Sync>> {
     let mut tx_graph = DFTxGraph::default();
-    tx_graph.build(tenant_txs)?;
+    if tx_graph.build(tenant_txs).is_err() {
+        // If we had an error while building the graph, we don't
+        // execute anything and return to allow any set results
+        // (essentially errors) to be set in DB.
+        return Ok(tx_graph);
+    }
     let cts_to_query = tx_graph.needed_map.keys().cloned().collect::<Vec<_>>();
     let ciphertext_map =
         query_ciphertexts(&cts_to_query, *tenant_id, trx, tracer, loop_ctx).await?;