fix(coprocessor): add threshold for ignoring dependency_count and make dependent list unique

Author: antoniupop

…75a56cf847422c19db036f51169de1b43ef.json …3c35621140ab5ed044d895f480f2ef90d71.jsoncoprocessor/fhevm-engine/.sqlx/query-dd1aca0ab7da74b59bfccede29d6f75a56cf847422c19db036f51169de1b43ef.json renamed to coprocessor/fhevm-engine/.sqlx/query-193e44bce47bd04357a3e86a707c63c35621140ab5ed044d895f480f2ef90d71.json coprocessor/fhevm-engine/.sqlx/query-dd1aca0ab7da74b59bfccede29d6f75a56cf847422c19db036f51169de1b43ef.json renamed to coprocessor/fhevm-engine/.sqlx/query-193e44bce47bd04357a3e86a707c63c35621140ab5ed044d895f480f2ef90d71.json

@@ -802,7 +802,12 @@ impl Database {
                         WHEN dependence_chain.status = 'processed' THEN EXCLUDED.last_updated_at
                         ELSE LEAST(dependence_chain.last_updated_at, EXCLUDED.last_updated_at)
                     END,
-                    dependents = dependence_chain.dependents || EXCLUDED.dependents
+                    dependents = (
+                        SELECT ARRAY(
+                            SELECT DISTINCT d
+                            FROM unnest(dependence_chain.dependents || EXCLUDED.dependents) AS d
+                        )
+                    )
                 "#,
                 chain.hash.to_vec(),
                 last_updated_at,

coprocessor/fhevm-engine/host-listener/src/database/tfhe_event_propagate.rs

@@ -115,6 +115,7 @@ async fn start_coprocessor(rx: Receiver<bool>, app_port: u16, db_url: &str) {
         dcid_cleanup_interval_sec: 0,
         processed_dcid_ttl_sec: 0,
         dcid_max_no_progress_cycles: 2,
+        dcid_ignore_dependency_count_threshold: 100,
     };
 
     std::thread::spawn(move || {

coprocessor/fhevm-engine/tfhe-worker/benches/utils.rs

@@ -121,6 +121,10 @@ pub struct Args {
     #[arg(long, value_parser = clap::value_parser!(u32), default_value_t = 2)]
     pub dcid_max_no_progress_cycles: u32,
 
+    /// Number of no-progress DCID releases before ignoring dependence counter
+    #[arg(long, value_parser = clap::value_parser!(u32), default_value_t = 100)]
+    pub dcid_ignore_dependency_count_threshold: u32,
+
     /// Log level for the application
     #[arg(
         long,

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

@@ -218,7 +218,10 @@ impl LockMngr {
     }
 
     /// Acquire the earliest dependence-chain entry for processing
-    /// sorted by last_updated_at (FIFO).
+    /// sorted by last_updated_at (FIFO). Here we ignore
+    /// dependency_count as reorgs can lead to incorrect counts and
+    /// set of dependents until we add block hashes to transaction
+    /// hashes to uniquely identify transactions.
     /// Returns the dependence_chain_id if a lock was acquired
     pub async fn acquire_early_lock(
         &mut self,

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

@@ -124,6 +124,7 @@ async fn start_coprocessor(rx: Receiver<bool>, app_port: u16, db_url: &str) {
         dcid_cleanup_interval_sec: 0,
         processed_dcid_ttl_sec: 0,
         dcid_max_no_progress_cycles: 2,
+        dcid_ignore_dependency_count_threshold: 100,
     };
 
     std::thread::spawn(move || {

coprocessor/fhevm-engine/tfhe-worker/src/tests/utils.rs

@@ -343,7 +343,9 @@ async fn query_for_work<'a>(
             // If there is a current lock, we extend it and use its dependence_chain_id
             Some((id, reason)) => (Some(id), reason),
             None => {
-                if *no_progress_cycles < 10 * args.dcid_max_no_progress_cycles {
+                if *no_progress_cycles
+                    < args.dcid_ignore_dependency_count_threshold * args.dcid_max_no_progress_cycles
+                {
                     deps_chain_mngr.acquire_next_lock().await?
                 } else {
                     *no_progress_cycles = 0;