fix: Reset counter store on startup when env var exists (#700)

* fix: Reset counter store on startup when env var exists

* chore: PR suggestions

Author: zeljkoX

src/bootstrap/config_processor.rs

@@ -751,6 +751,10 @@ where
         app_state.network_repository.drop_all_entries().await?;
         app_state.plugin_repository.drop_all_entries().await?;
         app_state.api_key_repository.drop_all_entries().await?;
+        app_state
+            .transaction_counter_store
+            .drop_all_entries()
+            .await?;
     }
 
     info!("Processing config file");
@@ -1796,6 +1800,48 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_reset_storage_on_start_clears_transaction_counter() -> Result<()> {
+        let config = create_minimal_test_config();
+        let server_config = Arc::new(create_test_server_config_with_settings(
+            RepositoryStorageType::InMemory,
+            true,
+        ));
+
+        let app_state = ThinData(create_test_app_state());
+
+        // Seed transaction counter with a value that simulates an inflated nonce
+        app_state
+            .transaction_counter_store
+            .set("test-relayer-1", "0xABC", 999)
+            .await
+            .unwrap();
+        assert_eq!(
+            app_state
+                .transaction_counter_store
+                .get("test-relayer-1", "0xABC")
+                .await
+                .unwrap(),
+            Some(999)
+        );
+
+        // Process config with reset_storage_on_start = true
+        process_config_file(config, server_config, &app_state).await?;
+
+        // Transaction counter should have been cleared
+        assert_eq!(
+            app_state
+                .transaction_counter_store
+                .get("test-relayer-1", "0xABC")
+                .await
+                .unwrap(),
+            None,
+            "Transaction counter should be cleared when RESET_STORAGE_ON_START is true"
+        );
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_should_process_config_file_redis_storage_empty_repositories() -> Result<()> {
         let config = create_minimal_test_config();

src/repositories/transaction_counter/mod.rs

@@ -59,6 +59,10 @@ pub trait TransactionCounterTrait {
 
     async fn set(&self, relayer_id: &str, address: &str, value: u64)
         -> Result<(), RepositoryError>;
+
+    /// Remove all stored counter entries from the underlying backend.
+    /// Intended for startup reset flows when `RESET_STORAGE_ON_START` is enabled.
+    async fn drop_all_entries(&self) -> Result<(), RepositoryError>;
 }
 
 /// Enum wrapper for different transaction counter repository implementations
@@ -137,6 +141,15 @@ impl TransactionCounterTrait for TransactionCounterRepositoryStorage {
             }
         }
     }
+
+    async fn drop_all_entries(&self) -> Result<(), RepositoryError> {
+        match self {
+            TransactionCounterRepositoryStorage::InMemory(counter) => {
+                counter.drop_all_entries().await
+            }
+            TransactionCounterRepositoryStorage::Redis(counter) => counter.drop_all_entries().await,
+        }
+    }
 }
 
 #[cfg(test)]
@@ -175,4 +188,17 @@ mod tests {
         let new_value = repo.decrement("test_relayer", "0x1234").await.unwrap();
         assert_eq!(new_value, 100);
     }
+
+    #[tokio::test]
+    async fn test_enum_wrapper_drop_all_entries() {
+        let repo = TransactionCounterRepositoryStorage::new_in_memory();
+
+        repo.set("relayer_1", "0x1234", 100).await.unwrap();
+        repo.set("relayer_2", "0x5678", 200).await.unwrap();
+
+        repo.drop_all_entries().await.unwrap();
+
+        assert_eq!(repo.get("relayer_1", "0x1234").await.unwrap(), None);
+        assert_eq!(repo.get("relayer_2", "0x5678").await.unwrap(), None);
+    }
 }

src/repositories/transaction_counter/transaction_counter_in_memory.rs

@@ -66,6 +66,11 @@ impl TransactionCounterTrait for InMemoryTransactionCounter {
             .insert((relayer_id.to_string(), address.to_string()), value);
         Ok(())
     }
+
+    async fn drop_all_entries(&self) -> Result<(), RepositoryError> {
+        self.store.clear();
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -104,6 +109,23 @@ mod tests {
         assert_eq!(store.get(relayer_id, address).await.unwrap(), Some(100));
     }
 
+    #[tokio::test]
+    async fn test_drop_all_entries() {
+        let store = InMemoryTransactionCounter::new();
+
+        store.set("relayer_1", "0x1234", 100).await.unwrap();
+        store.set("relayer_1", "0x5678", 200).await.unwrap();
+        store.set("relayer_2", "0x1234", 300).await.unwrap();
+
+        assert_eq!(store.get("relayer_1", "0x1234").await.unwrap(), Some(100));
+
+        store.drop_all_entries().await.unwrap();
+
+        assert_eq!(store.get("relayer_1", "0x1234").await.unwrap(), None);
+        assert_eq!(store.get("relayer_1", "0x5678").await.unwrap(), None);
+        assert_eq!(store.get("relayer_2", "0x1234").await.unwrap(), None);
+    }
+
     #[tokio::test]
     async fn test_multiple_relayers() {
         let store = InMemoryTransactionCounter::new();

src/repositories/transaction_counter/transaction_counter_redis.rs

@@ -212,6 +212,54 @@ impl TransactionCounterTrait for RedisTransactionCounter {
         debug!(value = %value, "counter set");
         Ok(())
     }
+
+    async fn drop_all_entries(&self) -> Result<(), RepositoryError> {
+        let mut conn = self
+            .get_connection(self.connections.primary(), "drop_all_entries")
+            .await?;
+
+        let pattern = format!("{}:{}:*", self.key_prefix, COUNTER_PREFIX);
+        debug!(pattern = %pattern, "dropping all transaction counter entries");
+
+        // Phase 1: Collect all matching keys without mutating the keyspace.
+        // Deleting during SCAN can cause hash table rehashing, which may skip keys.
+        let mut cursor: u64 = 0;
+        let mut all_keys: Vec<String> = Vec::new();
+
+        loop {
+            let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
+                .cursor_arg(cursor)
+                .arg("MATCH")
+                .arg(&pattern)
+                .arg("COUNT")
+                .arg(100)
+                .query_async(&mut conn)
+                .await
+                .map_err(|e| self.map_redis_error(e, "drop_all_entries_scan"))?;
+
+            all_keys.extend(keys);
+
+            cursor = next_cursor;
+            if cursor == 0 {
+                break;
+            }
+        }
+
+        // Phase 2: Batch delete all collected keys.
+        if !all_keys.is_empty() {
+            let mut pipe = redis::pipe();
+            pipe.atomic();
+            for key in &all_keys {
+                pipe.del(key);
+            }
+            pipe.exec_async(&mut conn)
+                .await
+                .map_err(|e| self.map_redis_error(e, "drop_all_entries_delete"))?;
+        }
+
+        debug!(total_deleted = %all_keys.len(), "dropped all transaction counter entries");
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -222,6 +270,10 @@ mod tests {
     use uuid::Uuid;
 
     async fn setup_test_repo() -> RedisTransactionCounter {
+        setup_test_repo_with_prefix("test_counter").await
+    }
+
+    async fn setup_test_repo_with_prefix(prefix: &str) -> RedisTransactionCounter {
         let redis_url =
             std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
         let cfg = deadpool_redis::Config::from_url(&redis_url);
@@ -235,7 +287,7 @@ mod tests {
         );
         let connections = Arc::new(RedisConnections::new_single_pool(pool));
 
-        RedisTransactionCounter::new(connections, "test_counter".to_string())
+        RedisTransactionCounter::new(connections, prefix.to_string())
             .expect("Failed to create Redis transaction counter")
     }
 
@@ -369,6 +421,35 @@ mod tests {
         assert_eq!(repo.get(&relayer_2, &address_1).await.unwrap(), Some(300));
     }
 
+    #[tokio::test]
+    #[ignore = "Requires active Redis instance"]
+    async fn test_drop_all_entries() {
+        let prefix = format!("test_drop_{}", uuid::Uuid::new_v4());
+        let repo = setup_test_repo_with_prefix(&prefix).await;
+        let relayer_1 = uuid::Uuid::new_v4().to_string();
+        let relayer_2 = uuid::Uuid::new_v4().to_string();
+        let address_1 = uuid::Uuid::new_v4().to_string();
+        let address_2 = uuid::Uuid::new_v4().to_string();
+
+        // Set up multiple counters
+        repo.set(&relayer_1, &address_1, 100).await.unwrap();
+        repo.set(&relayer_1, &address_2, 200).await.unwrap();
+        repo.set(&relayer_2, &address_1, 300).await.unwrap();
+
+        // Verify they exist
+        assert_eq!(repo.get(&relayer_1, &address_1).await.unwrap(), Some(100));
+        assert_eq!(repo.get(&relayer_1, &address_2).await.unwrap(), Some(200));
+        assert_eq!(repo.get(&relayer_2, &address_1).await.unwrap(), Some(300));
+
+        // Drop all
+        repo.drop_all_entries().await.unwrap();
+
+        // Verify all are gone
+        assert_eq!(repo.get(&relayer_1, &address_1).await.unwrap(), None);
+        assert_eq!(repo.get(&relayer_1, &address_2).await.unwrap(), None);
+        assert_eq!(repo.get(&relayer_2, &address_1).await.unwrap(), None);
+    }
+
     #[tokio::test]
     #[ignore = "Requires active Redis instance"]
     async fn test_concurrent_get_and_increment() {