appcore_core/
idempotency.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: idempotency.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/05/31 13:38:42 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/24 16:07:49 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Idempotency stores used by runtime controller command deduplication.
12
13use crate::error::{RuntimeError, RuntimeResult};
14use crate::idempotency_file::{
15    append_entry, encoded_record_bytes, load_entries, read_entry, rewrite_entries, RecordLocation,
16    StoreFileState, MAX_ACTIVE_IDEMPOTENCY_RECORDS, MAX_IDEMPOTENCY_FILE_BYTES,
17    MAX_PERSISTED_IDEMPOTENCY_RECORDS,
18};
19use crate::ids::validate_identifier;
20use std::collections::HashMap;
21use std::fmt;
22use std::fs;
23use std::path::{Path, PathBuf};
24use std::time::{SystemTime, UNIX_EPOCH};
25
26/// Stable on-disk format marker for the file idempotency store.
27pub const IDEMPOTENCY_FORMAT_V1: &str = "# appcore-idempotency-v1";
28
29/// Status of an idempotency execution.
30#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
31pub enum IdempotencyStatus {
32    /// Command execution has been reserved but not completed.
33    Pending,
34    /// Command execution completed and its serialized response is reusable.
35    Resolved {
36        /// Stable response status.
37        response_status: u16,
38        /// Serialized response body.
39        response_body: String,
40    },
41}
42
43/// A stored idempotency execution record.
44#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
45pub struct IdempotencyRecord {
46    /// Validated idempotency key.
47    pub key: String,
48    /// Digest that binds the key to one logical request.
49    pub request_hash: String,
50    /// Current execution status.
51    pub status: IdempotencyStatus,
52    /// Creation timestamp in Unix milliseconds.
53    pub created_at_ms: u64,
54}
55
56/// Durable or process-local idempotency record boundary.
57pub trait IdempotencyStore: Send + Sync {
58    /// Returns a stored record by key.
59    fn get(&self, key: &str) -> RuntimeResult<Option<IdempotencyRecord>>;
60    /// Inserts or replaces a validated record.
61    fn insert(&mut self, record: IdempotencyRecord) -> RuntimeResult<()>;
62    /// Returns the number of active records.
63    fn len(&self) -> usize;
64
65    /// Removes a record when supported.
66    fn remove(&mut self, _key: &str) -> RuntimeResult<()> {
67        Ok(())
68    }
69
70    /// Reports whether no active records exist.
71    fn is_empty(&self) -> bool {
72        self.len() == 0
73    }
74}
75
76/// Process-local idempotency store.
77#[derive(Default)]
78pub struct InMemoryIdempotencyStore {
79    seen: HashMap<String, IdempotencyRecord>,
80}
81
82impl fmt::Debug for InMemoryIdempotencyStore {
83    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
84        f.debug_struct("InMemoryIdempotencyStore")
85            .field("entry_count", &self.seen.len())
86            .finish()
87    }
88}
89
90impl InMemoryIdempotencyStore {
91    /// Creates an empty process-local store.
92    pub fn new() -> Self {
93        Self::default()
94    }
95}
96
97impl IdempotencyStore for InMemoryIdempotencyStore {
98    fn get(&self, key: &str) -> RuntimeResult<Option<IdempotencyRecord>> {
99        Ok(self.seen.get(key).cloned())
100    }
101
102    fn insert(&mut self, record: IdempotencyRecord) -> RuntimeResult<()> {
103        validate_key(&record.key)?;
104        ensure_active_capacity(&self.seen, &record.key)?;
105        self.seen.insert(record.key.clone(), record);
106        Ok(())
107    }
108
109    fn len(&self) -> usize {
110        self.seen.len()
111    }
112
113    fn remove(&mut self, key: &str) -> RuntimeResult<()> {
114        self.seen.remove(key);
115        Ok(())
116    }
117}
118
119/// Append-oriented local idempotency store with atomic compaction.
120pub struct FileIdempotencyStore {
121    file_path: PathBuf,
122    ttl_ms: Option<u64>,
123    seen: HashMap<String, RecordLocation>,
124    file_state: StoreFileState,
125}
126
127impl fmt::Debug for FileIdempotencyStore {
128    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
129        f.debug_struct("FileIdempotencyStore")
130            .field("file_path", &self.file_path)
131            .field("ttl_ms", &self.ttl_ms)
132            .field("entry_count", &self.seen.len())
133            .finish()
134    }
135}
136
137fn map_idempotency_io(operation: &'static str, err: std::io::Error) -> RuntimeError {
138    RuntimeError::IdempotencyStoreIo {
139        operation,
140        message: err.to_string(),
141    }
142}
143
144impl FileIdempotencyStore {
145    /// Opens a store without expiration.
146    pub fn new(path: impl AsRef<Path>) -> RuntimeResult<Self> {
147        Self::new_with_ttl(path, None)
148    }
149
150    /// Opens a store with optional expiration of resolved records.
151    /// Pending reservations never expire automatically; reconcile their effects.
152    pub fn new_with_ttl(path: impl AsRef<Path>, ttl_ms: Option<u64>) -> RuntimeResult<Self> {
153        let file_path = path.as_ref().to_path_buf();
154        ensure_parent_dir(&file_path)?;
155        if !file_path.exists() {
156            rewrite_entries(&file_path, &HashMap::new(), None, |_, _| true)?;
157        }
158        let loaded = load_entries(&file_path)?;
159        let (seen, file_state) = if loaded.needs_rewrite {
160            let rewritten = rewrite_entries(&file_path, &loaded.entries, None, |_, _| true)?;
161            (rewritten.entries, rewritten.file_state)
162        } else {
163            (loaded.entries, loaded.file_state)
164        };
165        let ttl_ms = match ttl_ms {
166            Some(0) => None,
167            other => other,
168        };
169
170        Ok(Self {
171            file_path,
172            ttl_ms,
173            seen,
174            file_state,
175        })
176    }
177
178    /// Returns the backing file path.
179    pub fn file_path(&self) -> &Path {
180        &self.file_path
181    }
182
183    /// Removes expired records and atomically rewrites the backing file.
184    pub fn compact(&mut self, now_ms: u64) -> RuntimeResult<usize> {
185        let before = self.seen.len();
186        let ttl_ms = self.ttl_ms;
187        let rewritten = rewrite_entries(&self.file_path, &self.seen, None, |_, location| {
188            location.pending || !is_expired(location.created_at_ms, ttl_ms, now_ms)
189        })?;
190        let removed = before.saturating_sub(rewritten.entries.len());
191        self.seen = rewritten.entries;
192        self.file_state = rewritten.file_state;
193
194        Ok(removed)
195    }
196}
197
198impl IdempotencyStore for FileIdempotencyStore {
199    fn get(&self, key: &str) -> RuntimeResult<Option<IdempotencyRecord>> {
200        let now_ms = now_ms();
201        if let Some(location) = self.seen.get(key) {
202            if !location.pending && is_expired(location.created_at_ms, self.ttl_ms, now_ms) {
203                Ok(None)
204            } else {
205                read_entry(&self.file_path, key, location).map(Some)
206            }
207        } else {
208            Ok(None)
209        }
210    }
211
212    fn insert(&mut self, record: IdempotencyRecord) -> RuntimeResult<()> {
213        validate_key(&record.key)?;
214        ensure_active_capacity(&self.seen, &record.key)?;
215        let record_bytes = encoded_record_bytes(&record)?;
216        if self.requires_rewrite(record_bytes) {
217            return self.replace_and_rewrite(record);
218        }
219        let appended = append_entry(&self.file_path, &record, self.file_state.bytes)?;
220        self.file_state.bytes = self.file_state.bytes.saturating_add(appended.written_bytes);
221        self.file_state.records = self.file_state.records.saturating_add(1);
222        self.seen.insert(record.key, appended.location);
223        Ok(())
224    }
225
226    fn len(&self) -> usize {
227        let now_ms = now_ms();
228        self.seen
229            .values()
230            .filter(|location| {
231                location.pending || !is_expired(location.created_at_ms, self.ttl_ms, now_ms)
232            })
233            .count()
234    }
235
236    fn remove(&mut self, key: &str) -> RuntimeResult<()> {
237        if self.seen.contains_key(key) {
238            let rewritten = rewrite_entries(&self.file_path, &self.seen, None, |candidate, _| {
239                candidate != key
240            })?;
241            self.seen = rewritten.entries;
242            self.file_state = rewritten.file_state;
243        }
244        Ok(())
245    }
246}
247
248impl FileIdempotencyStore {
249    fn requires_rewrite(&self, record_bytes: u64) -> bool {
250        self.file_state.records >= MAX_PERSISTED_IDEMPOTENCY_RECORDS
251            || self
252                .file_state
253                .bytes
254                .checked_add(record_bytes.saturating_add(1))
255                .is_none_or(|bytes| bytes > MAX_IDEMPOTENCY_FILE_BYTES)
256    }
257
258    fn replace_and_rewrite(&mut self, record: IdempotencyRecord) -> RuntimeResult<()> {
259        let rewritten = rewrite_entries(&self.file_path, &self.seen, Some(&record), |_, _| true)?;
260        self.seen = rewritten.entries;
261        self.file_state = rewritten.file_state;
262        Ok(())
263    }
264}
265
266fn ensure_parent_dir(file_path: &Path) -> RuntimeResult<()> {
267    if let Some(parent) = file_path.parent() {
268        fs::create_dir_all(parent).map_err(|e| map_idempotency_io("create_store_parent_dir", e))?;
269    }
270    Ok(())
271}
272
273fn validate_key(key: &str) -> RuntimeResult<()> {
274    match validate_identifier("IdempotencyKey", key) {
275        Ok(()) => Ok(()),
276        Err(RuntimeError::InvalidIdentifier {
277            reason: "empty", ..
278        }) => Err(RuntimeError::InvalidIdempotencyKey { reason: "empty" }),
279        Err(_) => Err(RuntimeError::InvalidIdempotencyKey {
280            reason: "invalid_char",
281        }),
282    }
283}
284
285fn ensure_active_capacity<T>(entries: &HashMap<String, T>, key: &str) -> RuntimeResult<()> {
286    if !entries.contains_key(key) && entries.len() >= MAX_ACTIVE_IDEMPOTENCY_RECORDS {
287        return Err(RuntimeError::IdempotencyStoreIo {
288            operation: "validate_store",
289            message: "active record limit exceeded".to_string(),
290        });
291    }
292    Ok(())
293}
294
295fn is_expired(created_at_ms: u64, ttl_ms: Option<u64>, now_ms: u64) -> bool {
296    if created_at_ms == 0 {
297        return false;
298    }
299    match ttl_ms {
300        Some(ttl) => now_ms.saturating_sub(created_at_ms) > ttl,
301        None => false,
302    }
303}
304
305fn now_ms() -> u64 {
306    SystemTime::now()
307        .duration_since(UNIX_EPOCH)
308        .map(|d| d.as_millis() as u64)
309        .unwrap_or(0)
310}
311
312#[cfg(test)]
313#[path = "idempotency_tests.rs"]
314mod tests;
appcore_core/idempotency.rs

appcore_core/
idempotency.rs
