feat(host-listener): account slow-lane over split dependency closures

Eikix · Eikix · commit 3c912c241636 · 2026-02-16T09:47:48.000+01:00
Keep no_fork parallelism, but aggregate inserted-op pressure over split dependency closures for slow-lane classification.
diff --git a/coprocessor/fhevm-engine/host-listener/README.md b/coprocessor/fhevm-engine/host-listener/README.md
@@ -38,12 +38,14 @@ If you want to disable TFHE operation events propagation, you can provide an emp
 `--dependent-ops-max-per-chain` enables slow-lane assignment (`0` disables).
 
 Current behavior:
-- Counter is **per dependence chain, per ingest pass** (block-scoped in normal flow).
-- Count is **unweighted**: `+1` for each newly inserted TFHE op in the chain.
+- Count is **per ingest pass** (block-scoped in normal flow).
+- Count unit is **unweighted**: `+1` for each newly inserted TFHE op.
+- Slow-lane threshold is evaluated on split-dependency closures (connected dcids),
+  then applied to all chains in the over-cap closure.
 - `is_allowed` is **not** part of the counter (a non-allowed op can still be required producer work).
 - It is **not** dependency depth and **not** cumulative across past blocks.
 
-If a chain exceeds the cap in that ingest pass, host-listener marks it slow by
+If a closure exceeds the cap in that ingest pass, host-listener marks its chains slow by
 setting `dependence_chain.schedule_priority = 1` (monotonic via `GREATEST` on
 upsert). tfhe-worker picks fast first (`0`) and processes slow when fast is empty.
 
diff --git a/coprocessor/fhevm-engine/host-listener/src/database/dependence_chains.rs b/coprocessor/fhevm-engine/host-listener/src/database/dependence_chains.rs
@@ -415,7 +415,7 @@ async fn grouping_to_chains_connex(
                     size: tx.size,
                     before_size: 0,
                     dependencies: vec![],
-                    inheritance_parents: vec![],
+                    split_dependencies: vec![],
                     dependents: vec![],
                     allowed_handle: tx.allowed_handle.clone(),
                     new_chain,
@@ -523,7 +523,7 @@ fn grouping_to_chains_no_fork(
                         size: 0,
                         before_size: 0,
                         dependencies: vec![],
-                        inheritance_parents: vec![],
+                        split_dependencies: vec![],
                         dependents: vec![],
                         allowed_handle: tx.allowed_handle.clone(), // needed to publish in cache
                         new_chain: false,
@@ -544,15 +544,15 @@ fn grouping_to_chains_no_fork(
             }
             debug!("Creating new chain for tx {:?} with block dependencies {:?}, outer dependencies {:?}, before_size {}",
 		   tx, dependencies_block, dependencies_outer, before_size);
-            let inheritance_parents =
+            let split_dependencies =
                 [dependencies_block.clone(), dependencies_outer.clone()]
                     .concat();
             let new_chain = Chain {
                 hash: tx.tx_hash,
                 size: tx.size,
                 before_size,
                 dependencies: dependencies_block,
-                inheritance_parents,
+                split_dependencies,
                 dependents: vec![],
                 allowed_handle: tx.allowed_handle.clone(),
                 new_chain: true,
@@ -1022,7 +1022,7 @@ mod tests {
         Chain {
             hash: TransactionHash::with_last_byte(last_byte),
             dependencies: vec![],
-            inheritance_parents: vec![],
+            split_dependencies: vec![],
             dependents: vec![],
             size: 1,
             before_size: 0,
diff --git a/coprocessor/fhevm-engine/host-listener/src/database/ingest.rs b/coprocessor/fhevm-engine/host-listener/src/database/ingest.rs
@@ -51,7 +51,7 @@ fn propagate_slow_lane_to_dependents(
     let mut dependents_by_dependency: HashMap<ChainHash, Vec<ChainHash>> =
         HashMap::new();
     for chain in chains {
-        for dependency in &chain.inheritance_parents {
+        for dependency in &chain.split_dependencies {
             dependents_by_dependency
                 .entry(*dependency)
                 .or_default()
@@ -74,6 +74,64 @@ fn propagate_slow_lane_to_dependents(
     }
 }
 
+fn classify_slow_by_split_dependency_closure(
+    chains: &[Chain],
+    dependent_ops_by_chain: &HashMap<ChainHash, u64>,
+    max_per_chain: u64,
+) -> HashSet<ChainHash> {
+    let chain_ids = chains
+        .iter()
+        .map(|chain| chain.hash)
+        .collect::<HashSet<_>>();
+    let mut neighbors: HashMap<ChainHash, HashSet<ChainHash>> =
+        HashMap::with_capacity(chains.len());
+    for chain in chains {
+        neighbors.entry(chain.hash).or_default();
+        for dependency in &chain.split_dependencies {
+            if !chain_ids.contains(dependency) {
+                continue;
+            }
+            neighbors.entry(chain.hash).or_default().insert(*dependency);
+            neighbors.entry(*dependency).or_default().insert(chain.hash);
+        }
+    }
+
+    let mut visited = HashSet::with_capacity(chains.len());
+    let mut slow_dep_chain_ids = HashSet::new();
+    for chain in chains {
+        if visited.contains(&chain.hash) {
+            continue;
+        }
+        let mut component = Vec::new();
+        let mut stack = vec![chain.hash];
+        visited.insert(chain.hash);
+        while let Some(current) = stack.pop() {
+            component.push(current);
+            if let Some(next_neighbors) = neighbors.get(&current) {
+                for next in next_neighbors {
+                    if visited.insert(*next) {
+                        stack.push(*next);
+                    }
+                }
+            }
+        }
+
+        let component_ops =
+            component.iter().fold(0_u64, |sum, dep_chain_id| {
+                sum.saturating_add(
+                    dependent_ops_by_chain
+                        .get(dep_chain_id)
+                        .copied()
+                        .unwrap_or(0),
+                )
+            });
+        if component_ops > max_per_chain {
+            slow_dep_chain_ids.extend(component);
+        }
+    }
+    slow_dep_chain_ids
+}
+
 pub async fn ingest_block_logs(
     chain_id: ChainId,
     db: &mut Database,
@@ -205,20 +263,17 @@ pub async fn ingest_block_logs(
     let mut slow_dep_chain_ids: HashSet<ChainHash> = HashSet::new();
     if slow_lane_enabled {
         let max_per_chain = u64::from(options.dependent_ops_max_per_chain);
-        for chain in &chains {
-            if let Some(chain_dep_ops) = dependent_ops_by_chain.get(&chain.hash)
-            {
-                if *chain_dep_ops > max_per_chain {
-                    slow_dep_chain_ids.insert(chain.hash);
-                }
-            }
-        }
+        slow_dep_chain_ids = classify_slow_by_split_dependency_closure(
+            &chains,
+            &dependent_ops_by_chain,
+            max_per_chain,
+        );
 
         let parent_dep_chain_ids = chains
             .iter()
             .flat_map(|chain| {
                 chain
-                    .inheritance_parents
+                    .split_dependencies
                     .iter()
                     .map(|dependency| dependency.to_vec())
             })
@@ -276,7 +331,7 @@ mod tests {
                 .iter()
                 .map(|dep| FixedBytes::<32>::from([*dep; 32]))
                 .collect(),
-            inheritance_parents: dependencies
+            split_dependencies: dependencies
                 .iter()
                 .map(|dep| FixedBytes::<32>::from([*dep; 32]))
                 .collect(),
@@ -305,4 +360,31 @@ mod tests {
         assert!(slow_dep_chain_ids.contains(&chains[2].hash));
         assert!(!slow_dep_chain_ids.contains(&chains[3].hash));
     }
+
+    #[test]
+    fn classifies_slow_by_split_dependency_closure_sum() {
+        let chains = vec![
+            fixture_chain(1, &[]),
+            fixture_chain(2, &[1]),
+            fixture_chain(3, &[2]),
+            fixture_chain(4, &[]),
+        ];
+        let dependent_ops_by_chain = HashMap::from([
+            (chains[0].hash, 30_u64),
+            (chains[1].hash, 20_u64),
+            (chains[2].hash, 20_u64),
+            (chains[3].hash, 10_u64),
+        ]);
+
+        let slow_dep_chain_ids = classify_slow_by_split_dependency_closure(
+            &chains,
+            &dependent_ops_by_chain,
+            64,
+        );
+
+        assert!(slow_dep_chain_ids.contains(&chains[0].hash));
+        assert!(slow_dep_chain_ids.contains(&chains[1].hash));
+        assert!(slow_dep_chain_ids.contains(&chains[2].hash));
+        assert!(!slow_dep_chain_ids.contains(&chains[3].hash));
+    }
 }
diff --git a/coprocessor/fhevm-engine/host-listener/src/database/tfhe_event_propagate.rs b/coprocessor/fhevm-engine/host-listener/src/database/tfhe_event_propagate.rs
@@ -55,9 +55,9 @@ static SLOW_LANE_MARKED_CHAINS_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(
 pub struct Chain {
     pub hash: ChainHash,
     pub dependencies: Vec<ChainHash>,
-    // Ingest-only metadata for slow-lane inheritance propagation.
+    // Ingest-only metadata for dependency links split by no_fork grouping.
     // Not used by scheduler execution ordering.
-    pub inheritance_parents: Vec<ChainHash>,
+    pub split_dependencies: Vec<ChainHash>,
     pub dependents: Vec<ChainHash>,
     pub allowed_handle: Vec<Handle>,
     pub size: u64,
