feat(coprocessor): drift detection in gw-listener (M1)

.github/workflows/test-suite-e2e-tests.yml

@@ -233,6 +233,11 @@ jobs:
         run: |
           ./fhevm-cli test hcu-block-cap
 
+      - name: Ciphertext drift test
+        working-directory: test-suite/fhevm
+        run: |
+          ./fhevm-cli test ciphertext-drift
+
       - name: Host listener poller test
         working-directory: test-suite/fhevm
         run: |

charts/coprocessor/Chart.yaml

@@ -1,6 +1,6 @@
 name: coprocessor
 description: A helm chart to distribute and deploy Zama fhevm Co-Processor services
-version: 0.8.4
+version: 0.8.5
 apiVersion: v2
 keywords:
   - fhevm

charts/coprocessor/values.yaml

@@ -417,12 +417,31 @@ gwListener:
   replicas: 1
 
   env:
-    # =========================================================================
-    # NEW ENVIRONMENT VARIABLES
-    # =========================================================================
-    # The address of the KMSGeneration contract
+    - name: DATABASE_URL
+      valueFrom:
+        secretKeyRef:
+          name: coprocessor-db-url
+          key: coprocessor-db-url
+    - name: INPUT_VERIFICATION_ADDRESS
+      valueFrom:
+        configMapKeyRef:
+          name: gateway-sc-addresses
+          key: input_verification.address
     - name: KMS_GENERATION_ADDRESS
-      value: ""
+      valueFrom:
+        configMapKeyRef:
+          name: gateway-sc-addresses
+          key: kms_generation.address
+    - name: CIPHERTEXT_COMMITS_ADDRESS
+      valueFrom:
+        configMapKeyRef:
+          name: gateway-sc-addresses
+          key: ciphertext_commits.address
+    - name: GATEWAY_CONFIG_ADDRESS
+      valueFrom:
+        configMapKeyRef:
+          name: gateway-sc-addresses
+          key: gateway_config.address
 
   # Command line arguments for the gateway listener
   args:
@@ -432,6 +451,8 @@ gwListener:
     - --gw-url=ws://gateway-rpc-node:8548
     - --input-verification-address=$(INPUT_VERIFICATION_ADDRESS)
     - --kms-generation-address=$(KMS_GENERATION_ADDRESS)
+    - --ciphertext-commits-address=$(CIPHERTEXT_COMMITS_ADDRESS)
+    - --gateway-config-address=$(GATEWAY_CONFIG_ADDRESS)
     - --error-sleep-initial-secs=1
     - --error-sleep-max-secs=10
     - --health-check-port=8080

coprocessor/fhevm-engine/db-migration/migrations/20260311154000_gw_listener_earliest_open_ct_block.sql

@@ -0,0 +1,3 @@
+ALTER TABLE gw_listener_last_block
+ADD COLUMN IF NOT EXISTS earliest_open_ct_commits_block BIGINT
+CHECK (earliest_open_ct_commits_block >= 0);

coprocessor/fhevm-engine/gw-listener/src/bin/gw_listener.rs

@@ -89,6 +89,32 @@ struct Conf {
         help = "Skip VerifyProofRequest events during replay"
     )]
     pub replay_skip_verify_proof: bool,
+
+    #[arg(long, help = "CiphertextCommits contract address for drift detection")]
+    ciphertext_commits_address: Option<Address>,
+
+    #[arg(
+        long,
+        requires = "ciphertext_commits_address",
+        help = "GatewayConfig contract address used to fetch coprocessor tx-senders"
+    )]
+    gateway_config_address: Option<Address>,
+
+    #[arg(
+        long,
+        default_value_t = 50,
+        requires = "ciphertext_commits_address",
+        help = "Drop a handle and alert if no consensus arrives after this many blocks"
+    )]
+    drift_no_consensus_timeout_blocks: u64,
+
+    #[arg(
+        long,
+        default_value_t = 10,
+        requires = "ciphertext_commits_address",
+        help = "After consensus, wait this many blocks for missing coprocessor submissions"
+    )]
+    drift_post_consensus_grace_blocks: u64,
 }
 
 fn install_signal_handlers(cancel_token: CancellationToken) -> anyhow::Result<()> {
@@ -156,6 +182,11 @@ async fn main() -> anyhow::Result<()> {
     else {
         anyhow::bail!("--host-chain-id or CHAIN_ID env var is missing.")
     };
+    if conf.ciphertext_commits_address.is_some() && conf.gateway_config_address.is_none() {
+        anyhow::bail!(
+            "--gateway-config-address is required when --ciphertext-commits-address is set."
+        );
+    }
     let config = ConfigSettings {
         host_chain_id,
         database_url: conf.database_url.clone().unwrap_or_default(),
@@ -171,6 +202,10 @@ async fn main() -> anyhow::Result<()> {
         replay_from_block: conf.replay_from_block,
         replay_skip_verify_proof: conf.replay_skip_verify_proof,
         log_last_processed_every_number_of_updates: conf.log_last_processed_every_number_of_updates,
+        ciphertext_commits_address: conf.ciphertext_commits_address,
+        gateway_config_address: conf.gateway_config_address,
+        drift_no_consensus_timeout_blocks: conf.drift_no_consensus_timeout_blocks,
+        drift_post_consensus_grace_blocks: conf.drift_post_consensus_grace_blocks,
     };
 
     let gw_listener = GatewayListener::new(