feat(kms-connector): event listening via eth_getLogs (#2166)

* feat(kms-connector): eth_getLogs event listening

* chore(kms-connector): event listener error handling

* chore(kms-connector): code cleaning

* fix(kms-connector): error metric handling

* fix(kms-connector): fix flaky tests

* chore(kms-connector): batch db insertion

* chore(kms-connector): rename metric

* chore(kms-connector): code cleaning

* fix(kms-connector): fix ob1 error

* chore(kms-connector): docs

* chore(kms-connector): docs

Co-authored-by: David <dvdplm@gmail.com>

* chore(kms-connector): u8 to usize

---------

Co-authored-by: David <dvdplm@gmail.com>

Author: eudelins-zama

docs/metrics/metrics.md

@@ -228,11 +228,11 @@ Metrics for zkproof-worker are to be added in future releases, if/when needed. C
  - **Alarm**: If the counter is a flat line over a period of time, only for `event_type` `public_decryption_request` and `user_decryption_request`.
    - **Recommendation**: 0 for more than 1 minute, i.e. `increase(counter{event_type="..."}[1m]) == 0`.
 
-#### Metric Name: `kms_connector_gw_listener_event_received_errors`
+#### Metric Name: `kms_connector_gw_listener_event_listening_errors`
  - **Type**: Counter
  - **Labels**:
-   - `event_type`: see [description](#metric-name-kms_connector_gw_listener_event_received_counter)
- - **Description**: Counts the number of errors encountered by the GW listener while receiving events.
+   - `contract`: can be used to filter by contract (decryption, kmsgeneration).
+ - **Description**: Counts the number of errors encountered by the GW listener while listening for events.
  - **Alarm**: If the counter increases over a period of time.
    - **Recommendation**: more than 60 failures in 1 minute, i.e. `sum(increase(counter[1m])) > 60`.
 

kms-connector/.sqlx/query-08628a859406ce5a66dda3b9729847a3846a59691b8dde5f518c2cb40b687742.json

@@ -42,14 +42,22 @@ database_url = "postgres://postgres:postgres@localhost/kms-connector"
 # ENV: KMS_CONNECTOR_TASK_LIMIT
 # task_limit = 1000
 
-# The polling interval for decryption requests (optional, defaults to 1s)
+# The polling interval for decryption requests (optional, defaults to 500ms)
 # ENV: KMS_CONNECTOR_DECRYPTION_POLLING (format: https://docs.rs/humantime/latest/humantime/)
-# decryption_polling = "1s"
+# decryption_polling = "500ms"
 
 # The polling interval for key management requests (optional, defaults to 30s)
 # ENV: KMS_CONNECTOR_KMS_GENERATION_POLLING (format: https://docs.rs/humantime/latest/humantime/)
 # key_management_polling = "30s"
 
+# Maximum number of blocks per eth_getLogs RPC call (optional, defaults to 100)
+# ENV: KMS_CONNECTOR_GET_LOGS_BATCH_SIZE
+# get_logs_batch_size = 100
+
+# Maximum consecutive polling errors before the listener stops (optional, defaults to 20)
+# ENV: KMS_CONNECTOR_MAX_CONSECUTIVE_POLLING_ERRORS
+# max_consecutive_polling_errors = 20
+
 # Block number to start processing decryption events from (optional, defaults to latest block if not set)
 # ENV: KMS_CONNECTOR_DECRYPTION_FROM_BLOCK_NUMBER
 # decryption_from_block_number = 1234

kms-connector/.sqlx/query-7e6031fa69c6a6884bb9b3a7b2e19db0fef204f71ad8dde9bd8bc808b0cb25fd.json

@@ -62,6 +62,14 @@ pub struct Config {
     #[serde(with = "humantime_serde", default = "default_key_management_polling")]
     pub key_management_polling: Duration,
 
+    /// The maximum number of blocks to fetch per `eth_getLogs` request.
+    #[serde(default = "default_get_logs_batch_size")]
+    pub get_logs_batch_size: u64,
+
+    /// Maximum number of consecutive polling errors before stopping the loop.
+    #[serde(default = "default_max_consecutive_polling_errors")]
+    pub max_consecutive_polling_errors: usize,
+
     /// Optional block number to start processing decryption events from.
     pub decryption_from_block_number: Option<u64>,
     /// Optional block number to start processing KMS operation events from.
@@ -75,13 +83,21 @@ fn default_service_name() -> String {
 }
 
 fn default_decryption_polling() -> Duration {
-    Duration::from_secs(1)
+    Duration::from_millis(500)
 }
 
 fn default_key_management_polling() -> Duration {
     Duration::from_secs(30)
 }
 
+fn default_get_logs_batch_size() -> u64 {
+    100
+}
+
+fn default_max_consecutive_polling_errors() -> usize {
+    20
+}
+
 // Default implementation for testing purpose
 impl Default for Config {
     fn default() -> Self {
@@ -101,6 +117,8 @@ impl Default for Config {
             healthcheck_timeout: default_healthcheck_timeout(),
             decryption_polling: default_decryption_polling(),
             key_management_polling: default_key_management_polling(),
+            get_logs_batch_size: default_get_logs_batch_size(),
+            max_consecutive_polling_errors: default_max_consecutive_polling_errors(),
             decryption_from_block_number: None,
             kms_operation_from_block_number: None,
         }
@@ -125,6 +143,8 @@ mod tests {
             env::remove_var("KMS_CONNECTOR_DECRYPTION_CONTRACT__ADDRESS");
             env::remove_var("KMS_CONNECTOR_KMS_GENERATION_CONTRACT__ADDRESS");
             env::remove_var("KMS_CONNECTOR_SERVICE_NAME");
+            env::remove_var("KMS_CONNECTOR_GET_LOGS_BATCH_SIZE");
+            env::remove_var("KMS_CONNECTOR_MAX_CONSECUTIVE_POLLING_ERRORS");
         }
     }
 
@@ -207,15 +227,27 @@ mod tests {
         // Set an environment variable to override the file
         let gateway_chain_id = 77737;
         let service_name = "kms-connector-override";
+        let get_logs_batch_size: u64 = 500;
+        let max_consecutive_polling_errors = 5;
         let mut expected_config = example_config.clone();
         expected_config.gateway_chain_id = gateway_chain_id;
         expected_config.service_name = service_name.to_string();
+        expected_config.get_logs_batch_size = get_logs_batch_size;
+        expected_config.max_consecutive_polling_errors = max_consecutive_polling_errors;
         unsafe {
             env::set_var(
                 "KMS_CONNECTOR_GATEWAY_CHAIN_ID",
                 gateway_chain_id.to_string(),
             );
             env::set_var("KMS_CONNECTOR_SERVICE_NAME", service_name);
+            env::set_var(
+                "KMS_CONNECTOR_GET_LOGS_BATCH_SIZE",
+                get_logs_batch_size.to_string(),
+            );
+            env::set_var(
+                "KMS_CONNECTOR_MAX_CONSECUTIVE_POLLING_ERRORS",
+                max_consecutive_polling_errors.to_string(),
+            );
         }
 
         // Load config from both sources