feat(coprocessor): drift detection in gw-listener (#2096)

* feat(coprocessor): add opt-in drift detection to gw-listener

Compare local ciphertext digests against on-chain consensus from the
CiphertextCommits contract. Enabled via --ciphertext-commits-address.

Adds early-warning logging when peer submissions diverge and structured
warn logs when local digest mismatches consensus. Four new Prometheus
counters provide observability. Detection-only — no recovery action.

* fix(coprocessor): propagate consensus errors and defer unresolved drift checks

- P1: Propagate DB errors from handle_consensus instead of swallowing
  them, so the block is retried via the outer backoff loop.
- P2: Queue unresolved consensus events (local digest not yet computed)
  in a bounded pending queue (10k cap) and retry each block tick, so
  late-arriving local digests are still checked.
- Refactor comparison logic into try_resolve_consensus shared by both
  the initial check and the retry path.

* refactor(coprocessor): trim drift detection to m1

* chore(coprocessor): drop m1 wording

* feat(coprocessor): track drift variants and lagging coprocessors

* refactor(coprocessor): fetch expected senders from gateway config

* feat(coprocessor): rebuild drift tracking after restart

* fix(coprocessor): silence drift replay on restart

* fix(coprocessor): keep consensus checks until local digests exist

* chore(coprocessor): apply rustfmt cleanup

* Fix drift detector rebuild and sender refresh

* Fix catch-up sender seeding and batch finalization

* Wire drift detection into gw-listener chart

* Bump coprocessor chart version

* Add ciphertext drift e2e workflow

* Remove stray plan file

* Keep e2e drift test env intact

* Tighten drift listener and e2e injector

* Use DB trigger for ciphertext drift e2e

* Fix drift handle completion and e2e exit path

* test(e2e): assert ciphertext drift via listener logs

* feat(gw-listener): add local attribution fields to drift logs

* feat(gw-listener): expose drift tuning histograms

* refactor(gw-listener): simplify drift detector and listener internals

- Avoid double variant_summaries() calls by binding result once
- Add has_multiple_variants() for cheap filter without string allocation
- Lazy-clone current_expected_senders only on new handle insertion
- Remove dead missing_submission_reported field
- Embed EventContext in ConsensusState instead of duplicating fields
- Use fetch_optional instead of fetch_all+assert for single-row query
- Hoist filter_addresses out of the polling loop

* test(drift-detector): add boundary and edge-case tests

Cover gaps identified in test audit:
- Grace period boundary: handle not evicted within grace, evicted at boundary
- No-consensus timeout boundary: handle not evicted within window, evicted at boundary
- Consensus before any submission: creates handle, evicts after grace
- Equivocation: same sender different digests warns but keeps first submission
- Duplicate submission: same sender same digests silently ignored
- Local digest not ready: consensus defers check, completes after DB update
- Local digest match: no false-positive drift when digests agree
- Local check not ready eviction: consensus with pending check times out

* refactor(tests): extract make_consensus_state helper, remove dead code

Simplify drift_detector tests by extracting a shared constructor,
removing an unused pool variable, and adding metric assertions.

* Fix review findings: metric pollution, DB waste, completion bug, e2e robustness

- Guard CONSENSUS_LATENCY_BLOCKS_HISTOGRAM behind alerts_enabled to prevent
  rebuild replay from polluting latency metrics
- Early-return from try_check_local_consensus when !alerts_enabled to skip
  unnecessary DB queries during rebuild; handles are re-checked via
  refresh_pending_consensus_checks once alerts resume
- Change finish_if_complete threshold from == to >= so unexpected senders
  don't prevent handle completion (previously hung until timeout eviction)
- Capture injector exit code in e2e script instead of letting set -e abort
  on wait; report injector failure explicitly
- Add test: unexpected_sender_does_not_block_completion

* docs(gw-listener): clarify consensus invariant

* refactor(gw-listener): simplify drift detector helpers and listener dispatch

Inline trivial free functions (has_multiple_variants, seen_sender_strings,
missing_sender_strings, address_strings, digest_pair_from_db_digests) at
call sites and delete them. Remove redundant .or_else() in sender_seed_block
computation. Convert if-else address dispatch to match. Add submit() test
helper to compress repetitive observe_submission boilerplate.

* rename submit() to submit_digest_event_and_drift_check() in tests

* refactor(gw-listener): simplify DriftDetector internals

- HandleState::new associated function replaces free function
- Inline DeferredMetrics fields directly into DriftDetector
- has_multiple_variants() avoids allocation in hot-path variant check
- classify_handle + HandleDisposition enum replaces continue chains in evict_stale
- end_of_batch/end_of_rebuild facade methods encapsulate multi-step protocols
- Fix dead unwrap_or in finish_if_complete (consensus already checked is_some)

* docs(gw-listener): document rebuild_drift_detector and facade methods

* fix(gw-listener): clarify drift counter description

Address review feedback: the old wording "peer submissions or consensus
mismatch" was ambiguous. Reword to make the two detection paths explicit.

* address review feedback: metrics docs, default timeouts, clap requires

- Clarify drift counter description in metrics.rs
- Document all 5 drift metrics in docs/metrics/metrics.md
- Bump drift timeout defaults from 50/10 to 3000/3000 blocks (~5 min at
  100ms block time) to accommodate coprocessors stuck for minutes
- Replace manual bail with clap requires on ciphertext_commits_address

* refactor(gw-listener): rename alerts_enabled to replaying (inverted)

The field name now reflects what the detector is *doing* (replaying
historical logs) rather than a side-effect toggle. Inverted semantics:
`replaying: true` suppresses warns and DB queries, `false` is normal
operation.

Also removes redundant `if self.alerts_enabled` guard in
try_check_local_consensus — already short-circuited by the early return
at the top of the function.

* address review feedback: type alias, flush_metrics, tx-senders signature

- Add `type CiphertextDigest = FixedBytes<32>` to avoid literal 32
- Document implicit upper bound on open_handles
- Remove redundant early-return guard in flush_metrics
- fetch_expected_coprocessor_tx_senders now requires gateway_config_address
  parameter; caller decides whether to call based on config

* refactor(gw-listener): rename HandleDisposition to HandleOutcome

Use domain-meaningful names: HandleOutcome describes what happened to
a handle, not an internal mechanism. Variants renamed:
- KeepWaiting → Pending
- NoConsensusTimeout → GatewayNeverReachedConsensus
- ConsensusUncheckedTimeout → LocalDigestNeverAppeared
- MissingSubmissionPostGrace → NotAllCoprocessorsSubmitted

* address self-review: fix double-counted metrics, remove dead code, improve clarity

- Fix verify_proof metrics being incremented both inside the function
  and at the call site (bug: double-counting)
- Remove unnecessary .clone() on owned event field
- Replace unreachable Pending return with unreachable!()
- Replace cross-function .unwrap() with let-else in evict_stale
- Collapse two identical "local digests not ready" debug branches
- Use < instead of != in classify_handle (invariant is always <)
- Rename fetch_expected_coprocessor_tx_senders to fetch_expected_senders
- Remove redundant replaying guard in finalize_completed_without_consensus

* fix(gw-listener): prevent drift detection errors from killing the listener

All drift detection code paths now log-and-continue instead of
propagating errors to run_get_logs. A transient DB or RPC failure
during drift checking should not interrupt processing of
VerifyProofRequest, ActivateKey, or ActivateCrs events.

* refactor(gw-listener): replace block-based drift timeouts with wall-clock Duration

The gateway chain only produces blocks when there are transactions, making
block count an unreliable proxy for elapsed time. Switch drift detection
from block-number arithmetic to std::time::Instant/Duration:

- ConfigSettings and CLI args use Duration (humantime format, e.g. "5m")
- EventContext carries observed_at: Instant
- HandleState/ConsensusState carry first_seen_at/received_at: Instant
- classify_handle compares wall-clock elapsed time against Duration thresholds
- Block numbers retained for logging/metrics (still useful diagnostics)
- Removes get_block_number RPC call from rebuild_drift_detector (no longer needed)

* docs(gw-listener): update metrics docs for wall-clock timeouts, clarify Default is test-only

* refactor(gw-listener): address review nits on drift detector

- Prefix DriftDetector fields with `drift_` for consistency with ConfigSettings
- Move classify_handle from free function to &self method on DriftDetector

Author: Eikix

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

@@ -238,6 +238,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,31 @@ struct Conf {
         help = "Skip VerifyProofRequest events during replay"
     )]
     pub replay_skip_verify_proof: bool,
+
+    #[arg(
+        long,
+        requires = "gateway_config_address",
+        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>,
+
+    /// How long to wait for the gateway to emit a consensus event after the
+    /// first submission is seen. Wall-clock duration — the default of 5 minutes
+    /// accommodates coprocessors that may be stuck for a few minutes.
+    #[arg(long, default_value = "5m", value_parser = parse_duration, requires = "ciphertext_commits_address")]
+    drift_no_consensus_timeout: Duration,
+
+    /// After consensus, how many additional blocks to wait for remaining
+    /// coprocessors to submit their ciphertext material. Wall-clock duration.
+    #[arg(long, default_value = "5m", value_parser = parse_duration, requires = "ciphertext_commits_address")]
+    drift_post_consensus_grace: Duration,
 }
 
 fn install_signal_handlers(cancel_token: CancellationToken) -> anyhow::Result<()> {
@@ -171,6 +196,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: conf.drift_no_consensus_timeout,
+        drift_post_consensus_grace: conf.drift_post_consensus_grace,
     };
 
     let gw_listener = GatewayListener::new(