tatara_core/domain/
idempotency.rs

1//! Idempotency layer for Raft commands.
2//!
3//! Prevents duplicate operations during leader transitions or retries.
4//! Based on the exactly-once semantics pattern:
5//!   idempotency_key → dedup store → TTL expiry → cached response.
6//!
7//! Reference: Exactly-once semantics in distributed systems
8//! (foundational to Raft, Kafka, and production consensus systems).
9
10use chrono::{DateTime, Duration, Utc};
11use serde::{Deserialize, Serialize};
12use std::collections::HashMap;
13
14/// Default TTL for idempotency keys (5 minutes).
15const DEFAULT_TTL_SECS: i64 = 300;
16
17/// A deduplication store that tracks recently processed idempotency keys.
18/// Stored in ClusterState and replicated via Raft.
19#[derive(Debug, Clone, Serialize, Deserialize, Default)]
20pub struct IdempotencyStore {
21    /// key → (response, expires_at)
22    entries: HashMap<String, IdempotencyEntry>,
23}
24
25#[derive(Debug, Clone, Serialize, Deserialize)]
26pub struct IdempotencyEntry {
27    /// The cached response for this key.
28    pub response: String, // Serialized ClusterResponse
29    /// When this entry expires.
30    pub expires_at: DateTime<Utc>,
31}
32
33impl IdempotencyStore {
34    pub fn new() -> Self {
35        Self {
36            entries: HashMap::new(),
37        }
38    }
39
40    /// Check if an idempotency key has already been processed.
41    /// Returns the cached response if it has, None if it hasn't.
42    pub fn check(&self, key: &str) -> Option<&str> {
43        let entry = self.entries.get(key)?;
44        if Utc::now() < entry.expires_at {
45            Some(&entry.response)
46        } else {
47            None // Expired
48        }
49    }
50
51    /// Record that an idempotency key has been processed with the given response.
52    pub fn record(&mut self, key: String, response: String) {
53        self.entries.insert(
54            key,
55            IdempotencyEntry {
56                response,
57                expires_at: Utc::now() + Duration::seconds(DEFAULT_TTL_SECS),
58            },
59        );
60    }
61
62    /// Record with a custom TTL.
63    pub fn record_with_ttl(&mut self, key: String, response: String, ttl_secs: i64) {
64        self.entries.insert(
65            key,
66            IdempotencyEntry {
67                response,
68                expires_at: Utc::now() + Duration::seconds(ttl_secs),
69            },
70        );
71    }
72
73    /// Garbage collect expired entries. Call periodically.
74    pub fn gc(&mut self) {
75        let now = Utc::now();
76        self.entries.retain(|_, entry| entry.expires_at > now);
77    }
78
79    /// Number of active (non-expired) entries.
80    pub fn len(&self) -> usize {
81        let now = Utc::now();
82        self.entries.values().filter(|e| e.expires_at > now).count()
83    }
84
85    pub fn is_empty(&self) -> bool {
86        self.len() == 0
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93
94    #[test]
95    fn test_idempotency_check_and_record() {
96        let mut store = IdempotencyStore::new();
97
98        // First time: not found
99        assert!(store.check("key-1").is_none());
100
101        // Record
102        store.record("key-1".to_string(), "response-1".to_string());
103
104        // Second time: found
105        assert_eq!(store.check("key-1"), Some("response-1"));
106    }
107
108    #[test]
109    fn test_idempotency_expiry() {
110        let mut store = IdempotencyStore::new();
111
112        // Record with 0-second TTL (already expired)
113        store.record_with_ttl("key-1".to_string(), "response-1".to_string(), -1);
114
115        // Should not find expired key
116        assert!(store.check("key-1").is_none());
117    }
118
119    #[test]
120    fn test_gc() {
121        let mut store = IdempotencyStore::new();
122
123        // Record one valid and one expired
124        store.record("valid".to_string(), "ok".to_string());
125        store.record_with_ttl("expired".to_string(), "old".to_string(), -1);
126
127        store.gc();
128
129        assert_eq!(store.entries.len(), 1);
130        assert!(store.entries.contains_key("valid"));
131    }
132
133    #[test]
134    fn test_len_excludes_expired() {
135        let mut store = IdempotencyStore::new();
136        store.record("valid".to_string(), "ok".to_string());
137        store.record_with_ttl("expired".to_string(), "old".to_string(), -1);
138
139        assert_eq!(store.len(), 1);
140    }
141}
tatara_core/domain/idempotency.rs

tatara_core/domain/
idempotency.rs
