openzeppelin_relayer/repositories/transaction_counter/
transaction_counter_redis.rs

1//! Redis implementation of the transaction counter.
2//!
3//! This module provides a Redis-based implementation of the `TransactionCounterTrait`,
4//! allowing transaction counters to be stored and retrieved from a Redis database.
5//! The implementation includes comprehensive error handling, logging, and atomic operations
6//! to ensure consistency when incrementing and decrementing counters.
7
8use super::TransactionCounterTrait;
9use crate::models::RepositoryError;
10use crate::repositories::redis_base::RedisRepository;
11use crate::utils::RedisConnections;
12use async_trait::async_trait;
13use redis::AsyncCommands;
14use std::fmt;
15use std::sync::Arc;
16use tracing::debug;
17
18const COUNTER_PREFIX: &str = "transaction_counter";
19
20#[derive(Clone)]
21pub struct RedisTransactionCounter {
22    pub connections: Arc<RedisConnections>,
23    pub key_prefix: String,
24}
25
26impl RedisRepository for RedisTransactionCounter {}
27
28impl fmt::Debug for RedisTransactionCounter {
29    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
30        f.debug_struct("RedisTransactionCounter")
31            .field("key_prefix", &self.key_prefix)
32            .finish()
33    }
34}
35
36impl RedisTransactionCounter {
37    pub fn new(
38        connections: Arc<RedisConnections>,
39        key_prefix: String,
40    ) -> Result<Self, RepositoryError> {
41        if key_prefix.is_empty() {
42            return Err(RepositoryError::InvalidData(
43                "Redis key prefix cannot be empty".to_string(),
44            ));
45        }
46
47        Ok(Self {
48            connections,
49            key_prefix,
50        })
51    }
52
53    /// Generate key for transaction counter: {prefix}:transaction_counter:{relayer_id}:{address}
54    fn counter_key(&self, relayer_id: &str, address: &str) -> String {
55        format!(
56            "{}:{}:{}:{}",
57            self.key_prefix, COUNTER_PREFIX, relayer_id, address
58        )
59    }
60}
61
62#[async_trait]
63impl TransactionCounterTrait for RedisTransactionCounter {
64    async fn get(&self, relayer_id: &str, address: &str) -> Result<Option<u64>, RepositoryError> {
65        if relayer_id.is_empty() {
66            return Err(RepositoryError::InvalidData(
67                "Relayer ID cannot be empty".to_string(),
68            ));
69        }
70
71        if address.is_empty() {
72            return Err(RepositoryError::InvalidData(
73                "Address cannot be empty".to_string(),
74            ));
75        }
76
77        let key = self.counter_key(relayer_id, address);
78        debug!(relayer_id = %relayer_id, address = %address, "getting counter for relayer and address");
79
80        let mut conn = self
81            .get_connection(self.connections.reader(), "get")
82            .await?;
83
84        let value: Option<u64> = conn
85            .get(&key)
86            .await
87            .map_err(|e| self.map_redis_error(e, "get_counter"))?;
88
89        debug!(value = ?value, "retrieved counter value");
90        Ok(value)
91    }
92
93    async fn get_and_increment(
94        &self,
95        relayer_id: &str,
96        address: &str,
97    ) -> Result<u64, RepositoryError> {
98        if relayer_id.is_empty() {
99            return Err(RepositoryError::InvalidData(
100                "Relayer ID cannot be empty".to_string(),
101            ));
102        }
103
104        if address.is_empty() {
105            return Err(RepositoryError::InvalidData(
106                "Address cannot be empty".to_string(),
107            ));
108        }
109
110        let key = self.counter_key(relayer_id, address);
111        debug!(relayer_id = %relayer_id, address = %address, "getting and incrementing counter for relayer and address");
112
113        let mut conn = self
114            .get_connection(self.connections.primary(), "get_and_increment")
115            .await?;
116
117        // Use Redis INCR for atomic increment
118        let new_value: u64 = conn
119            .incr(&key, 1)
120            .await
121            .map_err(|e| self.map_redis_error(e, "get_and_increment"))?;
122
123        let current = new_value.saturating_sub(1);
124
125        debug!(from = %current, to = %(current + 1), "counter incremented");
126        Ok(current)
127    }
128
129    async fn decrement(&self, relayer_id: &str, address: &str) -> Result<u64, RepositoryError> {
130        if relayer_id.is_empty() {
131            return Err(RepositoryError::InvalidData(
132                "Relayer ID cannot be empty".to_string(),
133            ));
134        }
135
136        if address.is_empty() {
137            return Err(RepositoryError::InvalidData(
138                "Address cannot be empty".to_string(),
139            ));
140        }
141
142        let key = self.counter_key(relayer_id, address);
143        debug!(relayer_id = %relayer_id, address = %address, "decrementing counter for relayer and address");
144
145        let mut conn = self
146            .get_connection(self.connections.primary(), "decrement")
147            .await?;
148
149        // Check if counter exists first
150        let exists: bool = conn
151            .exists(&key)
152            .await
153            .map_err(|e| self.map_redis_error(e, "check_counter_exists"))?;
154
155        if !exists {
156            return Err(RepositoryError::NotFound(format!(
157                "Counter not found for relayer {relayer_id} and address {address}"
158            )));
159        }
160
161        // Use Redis DECR and correct if it goes below 0
162        let new_value: i64 = conn
163            .decr(&key, 1)
164            .await
165            .map_err(|e| self.map_redis_error(e, "decrement_counter"))?;
166
167        let new_value = if new_value < 0 {
168            // Correct negative values back to 0
169            let _: () = conn
170                .set(&key, 0)
171                .await
172                .map_err(|e| self.map_redis_error(e, "correct_negative_counter"))?;
173            0u64
174        } else {
175            new_value as u64
176        };
177
178        debug!(new_value = %new_value, "counter decremented");
179        Ok(new_value)
180    }
181
182    async fn set(
183        &self,
184        relayer_id: &str,
185        address: &str,
186        value: u64,
187    ) -> Result<(), RepositoryError> {
188        if relayer_id.is_empty() {
189            return Err(RepositoryError::InvalidData(
190                "Relayer ID cannot be empty".to_string(),
191            ));
192        }
193
194        if address.is_empty() {
195            return Err(RepositoryError::InvalidData(
196                "Address cannot be empty".to_string(),
197            ));
198        }
199
200        let key = self.counter_key(relayer_id, address);
201        debug!(relayer_id = %relayer_id, address = %address, value = %value, "setting counter for relayer and address");
202
203        let mut conn = self
204            .get_connection(self.connections.primary(), "set")
205            .await?;
206
207        let _: () = conn
208            .set(&key, value)
209            .await
210            .map_err(|e| self.map_redis_error(e, "set_counter"))?;
211
212        debug!(value = %value, "counter set");
213        Ok(())
214    }
215}
216
217#[cfg(test)]
218mod tests {
219    use super::*;
220    use std::sync::Arc;
221    use tokio;
222    use uuid::Uuid;
223
224    async fn setup_test_repo() -> RedisTransactionCounter {
225        let redis_url =
226            std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
227        let cfg = deadpool_redis::Config::from_url(&redis_url);
228        let pool = Arc::new(
229            cfg.builder()
230                .expect("Failed to create pool builder")
231                .max_size(16)
232                .runtime(deadpool_redis::Runtime::Tokio1)
233                .build()
234                .expect("Failed to build Redis pool"),
235        );
236        let connections = Arc::new(RedisConnections::new_single_pool(pool));
237
238        RedisTransactionCounter::new(connections, "test_counter".to_string())
239            .expect("Failed to create Redis transaction counter")
240    }
241
242    #[tokio::test]
243    #[ignore = "Requires active Redis instance"]
244    async fn test_get_nonexistent_counter() {
245        let repo = setup_test_repo().await;
246        let random_id = Uuid::new_v4().to_string();
247        let result = repo.get(&random_id, "0x1234").await.unwrap();
248        assert_eq!(result, None);
249    }
250
251    #[tokio::test]
252    #[ignore = "Requires active Redis instance"]
253    async fn test_set_and_get_counter() {
254        let repo = setup_test_repo().await;
255        let relayer_id = uuid::Uuid::new_v4().to_string();
256        let address = uuid::Uuid::new_v4().to_string();
257
258        repo.set(&relayer_id, &address, 100).await.unwrap();
259        let result = repo.get(&relayer_id, &address).await.unwrap();
260        assert_eq!(result, Some(100));
261    }
262
263    #[tokio::test]
264    #[ignore = "Requires active Redis instance"]
265    async fn test_get_and_increment() {
266        let repo = setup_test_repo().await;
267        let relayer_id = uuid::Uuid::new_v4().to_string();
268        let address = uuid::Uuid::new_v4().to_string();
269
270        // First increment should return 0 and set to 1
271        let result = repo.get_and_increment(&relayer_id, &address).await.unwrap();
272        assert_eq!(result, 0);
273
274        let current = repo.get(&relayer_id, &address).await.unwrap();
275        assert_eq!(current, Some(1));
276
277        // Second increment should return 1 and set to 2
278        let result = repo.get_and_increment(&relayer_id, &address).await.unwrap();
279        assert_eq!(result, 1);
280
281        let current = repo.get(&relayer_id, &address).await.unwrap();
282        assert_eq!(current, Some(2));
283    }
284
285    #[tokio::test]
286    #[ignore = "Requires active Redis instance"]
287    async fn test_decrement() {
288        let repo = setup_test_repo().await;
289        let relayer_id = uuid::Uuid::new_v4().to_string();
290        let address = uuid::Uuid::new_v4().to_string();
291
292        // Set initial value
293        repo.set(&relayer_id, &address, 5).await.unwrap();
294
295        // Decrement should return 4
296        let result = repo.decrement(&relayer_id, &address).await.unwrap();
297        assert_eq!(result, 4);
298
299        let current = repo.get(&relayer_id, &address).await.unwrap();
300        assert_eq!(current, Some(4));
301    }
302
303    #[tokio::test]
304    #[ignore = "Requires active Redis instance"]
305    async fn test_decrement_not_found() {
306        let repo = setup_test_repo().await;
307        let result = repo.decrement("nonexistent", "0x1234").await;
308        assert!(matches!(result, Err(RepositoryError::NotFound(_))));
309    }
310
311    #[tokio::test]
312    #[ignore = "Requires active Redis instance"]
313    async fn test_empty_validation() {
314        let repo = setup_test_repo().await;
315
316        // Test empty relayer_id
317        let result = repo.get("", "0x1234").await;
318        assert!(matches!(result, Err(RepositoryError::InvalidData(_))));
319
320        // Test empty address
321        let result = repo.get("relayer", "").await;
322        assert!(matches!(result, Err(RepositoryError::InvalidData(_))));
323    }
324
325    #[tokio::test]
326    #[ignore = "Requires active Redis instance"]
327    async fn test_multiple_relayers() {
328        let repo = setup_test_repo().await;
329        let relayer_1 = uuid::Uuid::new_v4().to_string();
330        let relayer_2 = uuid::Uuid::new_v4().to_string();
331        let address_1 = uuid::Uuid::new_v4().to_string();
332        let address_2 = uuid::Uuid::new_v4().to_string();
333
334        // Set different values for different relayer/address combinations
335        repo.set(&relayer_1, &address_1, 100).await.unwrap();
336        repo.set(&relayer_1, &address_2, 200).await.unwrap();
337        repo.set(&relayer_2, &address_1, 300).await.unwrap();
338
339        // Verify independent counters
340        assert_eq!(repo.get(&relayer_1, &address_1).await.unwrap(), Some(100));
341        assert_eq!(repo.get(&relayer_1, &address_2).await.unwrap(), Some(200));
342        assert_eq!(repo.get(&relayer_2, &address_1).await.unwrap(), Some(300));
343
344        // Verify independent increments
345        assert_eq!(
346            repo.get_and_increment(&relayer_1, &address_1)
347                .await
348                .unwrap(),
349            100
350        );
351        assert_eq!(
352            repo.get_and_increment(&relayer_1, &address_1)
353                .await
354                .unwrap(),
355            101
356        );
357        assert_eq!(
358            repo.get_and_increment(&relayer_1, &address_2)
359                .await
360                .unwrap(),
361            200
362        );
363        assert_eq!(
364            repo.get_and_increment(&relayer_1, &address_2)
365                .await
366                .unwrap(),
367            201
368        );
369        assert_eq!(repo.get(&relayer_2, &address_1).await.unwrap(), Some(300));
370    }
371
372    #[tokio::test]
373    #[ignore = "Requires active Redis instance"]
374    async fn test_concurrent_get_and_increment() {
375        let repo = setup_test_repo().await;
376        let relayer_id = uuid::Uuid::new_v4().to_string();
377        let address = uuid::Uuid::new_v4().to_string();
378
379        // Set initial value
380        repo.set(&relayer_id, &address, 100).await.unwrap();
381
382        // Create multiple concurrent tasks that increment the counter
383        let handles: Vec<_> = (0..10)
384            .map(|_| {
385                let repo = repo.clone();
386                let relayer_id = relayer_id.clone();
387                let address = address.clone();
388                tokio::spawn(
389                    async move { repo.get_and_increment(&relayer_id, &address).await.unwrap() },
390                )
391            })
392            .collect();
393
394        // Wait for all tasks to complete and collect results
395        let mut results = Vec::new();
396        for handle in handles {
397            results.push(handle.await.unwrap());
398        }
399
400        // Sort results to check they are sequential
401        results.sort();
402
403        // Verify we get exactly the values 100-109 (no duplicates, no gaps)
404        let expected: Vec<u64> = (100..110).collect();
405        assert_eq!(results, expected);
406
407        // Verify final value is 110
408        let final_value = repo.get(&relayer_id, &address).await.unwrap();
409        assert_eq!(final_value, Some(110));
410    }
411}
openzeppelin_relayer/repositories/transaction_counter/transaction_counter_redis.rs

openzeppelin_relayer/repositories/transaction_counter/
transaction_counter_redis.rs
