test(host-listener): harden slow-lane local validation

Author: Eikix

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

@@ -61,40 +61,75 @@ cd test-suite/fhevm
 ./fhevm-cli deploy --build --local
 ```
 
-2. Enable the cap on both listener types
+2. Bootstrap gate (required before any slow-lane test)
+
+The validation is invalid until key bootstrap is healthy.
+
+Check:
+
+```bash
+docker logs --since=20m coprocessor-gw-listener | rg -n 'Received ActivateKey event|ActivateKey event successful'
+docker exec -i coprocessor-and-kms-db psql -U postgres -d coprocessor \
+  -c "SELECT tenant_id, sns_pk FROM tenants;"
+docker logs --since=20m coprocessor-sns-worker | rg -n 'bytes_len|No keys available|Fetched keyset'
+```
+
+Expected:
+
+- `coprocessor-gw-listener` logs include `ActivateKey event successful`.
+- `tenants.sns_pk` is not the bootstrap OID anymore.
+- `coprocessor-sns-worker` shows non-zero `bytes_len` and `Fetched keyset` (no repeated `No keys available` loop).
+
+If gate fails, recover by restarting only gw-listener and re-check:
+
+```bash
+cd test-suite/fhevm
+docker compose -p fhevm \
+  --env-file env/staging/.env.coprocessor.local \
+  -f docker-compose/coprocessor-docker-compose.yml \
+  up -d --no-deps coprocessor-gw-listener
+```
+
+3. Enable the cap on both listener types
 
 Temporarily add `--dependent-ops-max-per-chain=<N>` to both services in
 `test-suite/fhevm/docker-compose/coprocessor-docker-compose.yml`:
 
 - `coprocessor-host-listener` command
 - `coprocessor-host-listener-poller` command
 
-Use a low value for validation (example: `2`) so slow-lane is easy to trigger.
+Use a low value for validation (example: `2`).
 Then restart only these services:
 
 ```bash
 cd test-suite/fhevm
-docker compose \
-  --env-file ../env/staging/.env.coprocessor.local \
+docker compose -p fhevm \
+  --env-file env/staging/.env.coprocessor.local \
   -f docker-compose/coprocessor-docker-compose.yml \
   up -d --force-recreate \
   coprocessor-host-listener \
   coprocessor-host-listener-poller
 ```
 
-3. Generate dependent load
+Important:
+
+- Keep `-p fhevm` when restarting manually. Without it, containers can start on a different compose project/network and fail DNS (`host-node`, `db`).
 
-Run `stress_generator` with a dependent synthetic chain scenario
-(`ADDChain` or `MULChain`), and run in parallel a lighter/independent scenario.
+4. Generate dependent load
+
+Use a real host-contract flow so host-listener ingests real TFHE logs:
 
 ```bash
-cd coprocessor/fhevm-engine/stress-test-generator
-export EVGEN_DB_URL='postgresql://postgres:postgres@127.0.0.1:5432/coprocessor'
-export EVGEN_SCENARIO='data/evgen_scenario.csv'
-cargo run --release --bin stress_generator
+cd test-suite/fhevm
+./fhevm-cli test erc20
 ```
 
-4. Validate acceptance criteria in DB
+If this does not trigger slow-lane in local runs, use forcing mode for validation:
+
+- set `--dependent-ops-max-per-chain=1` temporarily
+- rerun `./fhevm-cli test erc20`
+
+5. Validate acceptance criteria in DB
 
 ```sql
 -- A. heavy dependent chains are marked slow
@@ -118,7 +153,17 @@ Expected:
 - At least one chain with `schedule_priority = 1`.
 - Fast (`0`) chains appear before slow (`1`) chains in acquisition order.
 
-5. Validate off-mode (`N=0`)
+6. Run deterministic integration matrix (recommended)
+
+This validates below/at/above threshold behavior with a realistic cap.
+
+```bash
+cd coprocessor/fhevm-engine
+cargo test -p host-listener --test host_listener_integration_tests \
+  test_slow_lane_threshold_matrix_locally -- --nocapture
+```
+
+7. Validate off-mode (`N=0`)
 
 - Set `--dependent-ops-max-per-chain=0` on both listener types.
 - Restart the same two services.

coprocessor/fhevm-engine/host-listener/tests/host_listener_integration_tests.rs

@@ -343,17 +343,6 @@ async fn ingest_blocks_for_receipts(
     Ok(())
 }
 
-async fn emit_dependent_burst(
-    setup: &Setup,
-    input_handle: Option<FixedBytes<32>>,
-    depth: usize,
-) -> Result<
-    (Vec<alloy::rpc::types::TransactionReceipt>, FixedBytes<32>),
-    anyhow::Error,
-> {
-    emit_dependent_burst_seeded(setup, input_handle, depth, 1_u64).await
-}
-
 async fn ingest_dependent_burst_seeded(
     db: &mut Database,
     setup: &Setup,
@@ -480,7 +469,7 @@ async fn test_bad_chain_id() {
 
 #[tokio::test]
 #[serial(db)]
-async fn test_slow_lane_marks_heavy_chain_locally() -> Result<(), anyhow::Error>
+async fn test_slow_lane_threshold_matrix_locally() -> Result<(), anyhow::Error>
 {
     let setup = setup_with_block_time(None, 3.0).await?;
     let mut db = Database::new(
@@ -490,27 +479,111 @@ async fn test_slow_lane_marks_heavy_chain_locally() -> Result<(), anyhow::Error>
     )
     .await?;
 
-    let (receipts, _) = emit_dependent_burst(&setup, None, 4).await?;
-    ingest_blocks_for_receipts(
-        &mut db,
-        &setup,
-        &receipts,
-        IngestOptions {
-            dependence_by_connexity: false,
-            dependence_cross_block: true,
-            dependent_ops_max_per_chain: 1,
-        },
+    let cases = [
+        ("below_cap", 63_usize, 64_u32, 0_i16, 11_u64),
+        ("at_cap", 64_usize, 64_u32, 0_i16, 12_u64),
+        ("above_cap", 65_usize, 64_u32, 1_i16, 13_u64),
+    ];
+
+    let mut seen_chains = HashSet::new();
+    for (name, depth, cap, expected_priority, seed) in cases {
+        let last_handle = ingest_dependent_burst_seeded(
+            &mut db, &setup, None, depth, seed, cap,
+        )
+        .await?;
+        let dep_chain_id =
+            dep_chain_id_for_output_handle(&setup, last_handle).await?;
+        assert!(
+            seen_chains.insert(dep_chain_id.clone()),
+            "matrix case {name} reused an existing dependence chain"
+        );
+
+        let schedule_priority = sqlx::query_scalar::<_, i16>(
+            "SELECT schedule_priority FROM dependence_chain WHERE dependence_chain_id = $1",
+        )
+        .bind(&dep_chain_id)
+        .fetch_one(&setup.db_pool)
+        .await?;
+        assert_eq!(
+            schedule_priority, expected_priority,
+            "case={name} depth={depth} cap={cap}"
+        );
+    }
+
+    Ok(())
+}
+
+#[tokio::test]
+#[serial(db)]
+async fn test_slow_lane_cross_block_sustained_below_cap_stays_fast_locally(
+) -> Result<(), anyhow::Error> {
+    let setup = setup_with_block_time(None, 3.0).await?;
+    let mut db = Database::new(
+        &setup.args.database_url,
+        &setup.args.coprocessor_api_key.unwrap(),
+        setup.args.dependence_cache_size,
     )
     .await?;
 
-    let slow_chain_count = sqlx::query_scalar::<_, i64>(
-        "SELECT COUNT(*) FROM dependence_chain WHERE schedule_priority = 1",
+    let cap = 64_u32;
+    let burst_depth = 8_usize;
+    let rounds = 4_u64;
+
+    let mut current_handle: Option<FixedBytes<32>> = None;
+    let mut seen_block_numbers = HashSet::new();
+
+    for round in 0..rounds {
+        let seed = 101_u64 + round;
+        let (receipts, last_output_handle) = emit_dependent_burst_seeded(
+            &setup,
+            current_handle,
+            burst_depth,
+            seed,
+        )
+        .await?;
+
+        for receipt in &receipts {
+            let block_number =
+                receipt.block_number.expect("receipt has block number");
+            seen_block_numbers.insert(block_number);
+        }
+
+        ingest_blocks_for_receipts(
+            &mut db,
+            &setup,
+            &receipts,
+            IngestOptions {
+                dependence_by_connexity: false,
+                dependence_cross_block: true,
+                dependent_ops_max_per_chain: cap,
+            },
+        )
+        .await?;
+
+        current_handle = Some(last_output_handle);
+        tokio::time::sleep(tokio::time::Duration::from_secs(4)).await;
+    }
+
+    assert!(
+        seen_block_numbers.len() > 1,
+        "test must span multiple blocks"
+    );
+
+    let dep_chain_id = dep_chain_id_for_output_handle(
+        &setup,
+        current_handle.expect("final output handle exists"),
+    )
+    .await?;
+    let schedule_priority = sqlx::query_scalar::<_, i16>(
+        "SELECT schedule_priority FROM dependence_chain WHERE dependence_chain_id = $1",
     )
+    .bind(&dep_chain_id)
     .fetch_one(&setup.db_pool)
     .await?;
-    assert!(
-        slow_chain_count > 0,
-        "heavy dependent chain should be assigned to slow lane"
+
+    assert_eq!(
+        schedule_priority, 0,
+        "current behavior: below-cap batches do not accumulate into slow lane across blocks"
     );
     Ok(())
 }