fix(kms-connector): listen from latest block if catchup fail (#1737)

* fix(kms-connector): listen from latest block if catchup fail

* chore(kms-connector): add tests

* ci(kms-connector): upgrade cargo-audit (#1671)

* chore(kms-connector): update codeowner

Author: eudelins-zama

.github/CODEOWNERS

@@ -11,7 +11,7 @@
 /host-contracts/ @zama-ai/fhevm-gateway
 /protocol-contracts/ @zama-ai/fhevm-gateway
 /library-solidity/ @zama-ai/fhevm-gateway
-/kms-connector/ @zama-ai/fhevm-gateway
+/kms-connector/ @zama-ai/fhevm-gateway @dartdart26
 
 # Coprocessor Team ownership
 /coprocessor/ @zama-ai/fhevm-coprocessor

.github/workflows/kms-connector-dependency-analysis.yml

@@ -30,6 +30,7 @@ jobs:
             rust-files:
               - .github/workflows/kms-connector-dependency-analysis.yml
               - kms-connector/**
+
   dependencies-check:
     name: kms-connector-dependency-analysis/dependencies-check (bpr)
     needs: check-changes
@@ -50,13 +51,13 @@ jobs:
           toolchain: stable
 
       - name: Install cargo-binstall
-        uses: cargo-bins/cargo-binstall@84ca29d5c1719e79e23b6af147555a8f4dac79d6 # v1.10.14
+        uses: cargo-bins/cargo-binstall@80aaafe04903087c333980fa2686259ddd34b2d9 # v1.16.6
 
       - name: Install cargo tools
         run: |
           cargo binstall --no-confirm --force \
-            cargo-audit@0.21.2 \
-            cargo-deny@0.16.2
+            cargo-audit@0.22.0 \
+            cargo-deny@0.18.9
 
       - name: Check that Cargo.lock is the source of truth
         run: |
@@ -72,4 +73,3 @@ jobs:
         run: |
           cd kms-connector
           cargo-audit audit
-

kms-connector/Cargo.lock

@@ -167,9 +167,10 @@ where
         event_poller.poller = event_poller.poller.with_poll_interval(poll_interval);
         info!("✓ Subscribed to {event_type} events");
 
-        self.catchup_past_events::<E>(&mut last_block_polled, event_type)
+        let _ = self
+            .catchup_past_events::<E>(&mut last_block_polled, event_type)
             .await
-            .map_err(|e| anyhow!("Failed to catch up past {event_type} events: {e}"))?;
+            .inspect_err(|e| warn!("Failed to catch up past {event_type} events: {e}"));
 
         select! {
             _ = self.process_events(event_type, event_poller, &mut last_block_polled) => (),
@@ -407,7 +408,7 @@ mod tests {
     #[timeout(Duration::from_secs(90))]
     #[tokio::test]
     async fn test_reset_filter_stops_listener() {
-        let (_test_instance, asserter, gw_listener) = test_setup().await;
+        let (_test_instance, asserter, gw_listener) = test_setup(None).await;
 
         asserter.push_failure(ErrorPayload {
             code: -32000,
@@ -418,11 +419,31 @@ mod tests {
         gw_listener.subscribe(EventType::KeygenRequest).await;
     }
 
+    #[rstest::rstest]
+    #[timeout(Duration::from_secs(90))]
+    #[tokio::test]
+    async fn test_failed_catchup_does_not_stop_listener() {
+        let (mut test_instance, asserter, gw_listener) = test_setup(Some(0)).await;
+
+        asserter.push_failure(ErrorPayload {
+            code: -32002,
+            message: "request timed out".into(),
+            data: None,
+        });
+
+        let event_type = EventType::KeygenRequest;
+        tokio::spawn(gw_listener.subscribe(event_type));
+        test_instance.wait_for_log("Failed to catch up").await;
+        test_instance
+            .wait_for_log(&format!("Waiting for next {event_type}"))
+            .await;
+    }
+
     #[rstest::rstest]
     #[timeout(Duration::from_secs(90))]
     #[tokio::test]
     async fn test_listener_ended_by_end_of_any_task() {
-        let (mut test_instance, _asserter, gw_listener) = test_setup().await;
+        let (mut test_instance, _asserter, gw_listener) = test_setup(None).await;
 
         // Will stop because some subcription tasks will not be able to init their event filter
         gw_listener.start().await;
@@ -438,7 +459,9 @@ mod tests {
         RootProvider,
     >;
 
-    async fn test_setup() -> (TestInstance, Asserter, GatewayListener<MockProvider>) {
+    async fn test_setup(
+        from_block_number: Option<u64>,
+    ) -> (TestInstance, Asserter, GatewayListener<MockProvider>) {
         let test_instance = TestInstanceBuilder::db_setup().await.unwrap();
 
         // Create a mocked `alloy::Provider`
@@ -452,6 +475,7 @@ mod tests {
         let config = Config {
             decryption_polling: Duration::from_millis(500),
             key_management_polling: Duration::from_millis(500),
+            from_block_number,
             ..Default::default()
         };
         let listener = GatewayListener::new(