fix(host-listener): bound slow-priority reset load in off mode

Author: Eikix

coprocessor/fhevm-engine/host-listener/README.md

@@ -46,8 +46,8 @@ slow-lane work when fast lane is empty, which isolates heavy dependent traffic
 without reordering within a chain.
 
 When set to `0`, host-listener disables slow-lane decisions, skips dependent-op
-throttling accounting, and resets existing `schedule_priority` values to `0` at
-startup so scheduling is effectively FIFO by `last_updated_at`.
+throttling accounting, and progressively resets pending slow chains back to
+`schedule_priority = 0` in bounded batches during ingest.
 
 ## Events in FHEVM
 

coprocessor/fhevm-engine/host-listener/src/bin/poller.rs

@@ -122,7 +122,7 @@ struct Args {
     #[arg(
         long,
         default_value_t = 0,
-        help = "Max weighted dependent ops per chain before slow-lane (0 disables and keeps all chains fast)"
+        help = "Max weighted dependent ops per chain before slow-lane (0 disables; pending slow chains are reset in bounded batches)"
     )]
     pub dependent_ops_max_per_chain: u32,
 }

coprocessor/fhevm-engine/host-listener/src/cmd/mod.rs

@@ -128,7 +128,7 @@ pub struct Args {
     #[arg(
         long,
         default_value_t = 0,
-        help = "Max weighted dependent ops per chain before slow-lane (0 disables and keeps all chains fast)"
+        help = "Max weighted dependent ops per chain before slow-lane (0 disables; pending slow chains are reset in bounded batches)"
     )]
     pub dependent_ops_max_per_chain: u32,
 
@@ -1075,16 +1075,6 @@ pub async fn main(args: Args) -> anyhow::Result<()> {
         ));
     }
 
-    if args.dependent_ops_max_per_chain == 0 {
-        let reset = db.reset_schedule_priorities().await?;
-        if reset > 0 {
-            info!(
-                count = reset,
-                "Slow-lane disabled: reset priorities to fast"
-            );
-        }
-    }
-
     let health_check = HealthCheck {
         blockchain_timeout_tick: log_iter.tick_timeout.clone(),
         blockchain_tick: log_iter.tick_block.clone(),

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

@@ -15,6 +15,8 @@ use crate::database::tfhe_event_propagate::{
     tfhe_result_handle, ChainHash, Database, LogTfhe,
 };
 
+const SLOW_PRIORITY_RESET_BATCH_SIZE: i64 = 5_000;
+
 pub struct BlockLogs<T> {
     pub logs: Vec<T>,
     pub summary: BlockSummary,
@@ -157,6 +159,20 @@ pub async fn ingest_block_logs(
     .await;
 
     let slow_lane_enabled = options.dependent_ops_max_per_chain > 0;
+    if !slow_lane_enabled {
+        let reset = db
+            .reset_schedule_priorities_batch(
+                &mut tx,
+                SLOW_PRIORITY_RESET_BATCH_SIZE,
+            )
+            .await?;
+        if reset > 0 {
+            info!(
+                count = reset,
+                "Slow-lane disabled: reset a batch of priorities to fast"
+            );
+        }
+    }
     let mut dependent_ops_by_chain: HashMap<ChainHash, DependentOpsStats> =
         HashMap::new();
     for tfhe_log in tfhe_event_log {

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

@@ -175,15 +175,32 @@ impl Database {
         }
     }
 
-    pub async fn reset_schedule_priorities(&self) -> Result<u64, SqlxError> {
+    pub async fn reset_schedule_priorities_batch(
+        &self,
+        tx: &mut Transaction<'_>,
+        batch_size: i64,
+    ) -> Result<u64, SqlxError> {
         let rows = sqlx::query(
             r#"
-            UPDATE dependence_chain
+            WITH batch AS (
+                SELECT dependence_chain_id
+                FROM dependence_chain
+                WHERE schedule_priority = 1
+                  AND status = 'updated'
+                  AND worker_id IS NULL
+                  AND dependency_count = 0
+                ORDER BY last_updated_at ASC
+                LIMIT $1
+                FOR UPDATE SKIP LOCKED
+            )
+            UPDATE dependence_chain dc
             SET schedule_priority = 0
-            WHERE schedule_priority <> 0
+            FROM batch
+            WHERE dc.dependence_chain_id = batch.dependence_chain_id
             "#,
         )
-        .execute(&self.pool().await)
+        .bind(batch_size)
+        .execute(tx.deref_mut())
         .await?;
         Ok(rows.rows_affected())
     }

coprocessor/fhevm-engine/host-listener/src/poller/mod.rs

@@ -158,16 +158,6 @@ pub async fn run_poller(config: PollerConfig) -> Result<()> {
         ));
     }
 
-    if config.dependent_ops_max_per_chain == 0 {
-        let reset = db.reset_schedule_priorities().await?;
-        if reset > 0 {
-            info!(
-                count = reset,
-                "Slow-lane disabled: reset priorities to fast"
-            );
-        }
-    }
-
     let initial_anchor =
         db.poller_get_last_caught_up_block(chain_id as i64).await?;
     db.tick.update();