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 record expiration.
151    pub fn new_with_ttl(path: impl AsRef<Path>, ttl_ms: Option<u64>) -> RuntimeResult<Self> {
152        let file_path = path.as_ref().to_path_buf();
153        ensure_parent_dir(&file_path)?;
154        if !file_path.exists() {
155            rewrite_entries(&file_path, &HashMap::new(), None, |_, _| true)?;
156        }
157        let loaded = load_entries(&file_path)?;
158        let (seen, file_state) = if loaded.needs_rewrite {
159            let rewritten = rewrite_entries(&file_path, &loaded.entries, None, |_, _| true)?;
160            (rewritten.entries, rewritten.file_state)
161        } else {
162            (loaded.entries, loaded.file_state)
163        };
164        let ttl_ms = match ttl_ms {
165            Some(0) => None,
166            other => other,
167        };
168
169        Ok(Self {
170            file_path,
171            ttl_ms,
172            seen,
173            file_state,
174        })
175    }
176
177    /// Returns the backing file path.
178    pub fn file_path(&self) -> &Path {
179        &self.file_path
180    }
181
182    /// Removes expired records and atomically rewrites the backing file.
183    pub fn compact(&mut self, now_ms: u64) -> RuntimeResult<usize> {
184        let before = self.seen.len();
185        let ttl_ms = self.ttl_ms;
186        let rewritten = rewrite_entries(&self.file_path, &self.seen, None, |_, location| {
187            !is_expired(location.created_at_ms, ttl_ms, now_ms)
188        })?;
189        let removed = before.saturating_sub(rewritten.entries.len());
190        self.seen = rewritten.entries;
191        self.file_state = rewritten.file_state;
192
193        Ok(removed)
194    }
195}
196
197impl IdempotencyStore for FileIdempotencyStore {
198    fn get(&self, key: &str) -> RuntimeResult<Option<IdempotencyRecord>> {
199        let now_ms = now_ms();
200        if let Some(location) = self.seen.get(key) {
201            if is_expired(location.created_at_ms, self.ttl_ms, now_ms) {
202                Ok(None)
203            } else {
204                read_entry(&self.file_path, key, location).map(Some)
205            }
206        } else {
207            Ok(None)
208        }
209    }
210
211    fn insert(&mut self, record: IdempotencyRecord) -> RuntimeResult<()> {
212        validate_key(&record.key)?;
213        ensure_active_capacity(&self.seen, &record.key)?;
214        let record_bytes = encoded_record_bytes(&record)?;
215        if self.requires_rewrite(record_bytes) {
216            return self.replace_and_rewrite(record);
217        }
218        let appended = append_entry(&self.file_path, &record, self.file_state.bytes)?;
219        self.file_state.bytes = self.file_state.bytes.saturating_add(appended.written_bytes);
220        self.file_state.records = self.file_state.records.saturating_add(1);
221        self.seen.insert(record.key, appended.location);
222        Ok(())
223    }
224
225    fn len(&self) -> usize {
226        let now_ms = now_ms();
227        self.seen
228            .values()
229            .filter(|location| !is_expired(location.created_at_ms, self.ttl_ms, now_ms))
230            .count()
231    }
232
233    fn remove(&mut self, key: &str) -> RuntimeResult<()> {
234        if self.seen.contains_key(key) {
235            let rewritten = rewrite_entries(&self.file_path, &self.seen, None, |candidate, _| {
236                candidate != key
237            })?;
238            self.seen = rewritten.entries;
239            self.file_state = rewritten.file_state;
240        }
241        Ok(())
242    }
243}
244
245impl FileIdempotencyStore {
246    fn requires_rewrite(&self, record_bytes: u64) -> bool {
247        self.file_state.records >= MAX_PERSISTED_IDEMPOTENCY_RECORDS
248            || self
249                .file_state
250                .bytes
251                .checked_add(record_bytes.saturating_add(1))
252                .is_none_or(|bytes| bytes > MAX_IDEMPOTENCY_FILE_BYTES)
253    }
254
255    fn replace_and_rewrite(&mut self, record: IdempotencyRecord) -> RuntimeResult<()> {
256        let rewritten = rewrite_entries(&self.file_path, &self.seen, Some(&record), |_, _| true)?;
257        self.seen = rewritten.entries;
258        self.file_state = rewritten.file_state;
259        Ok(())
260    }
261}
262
263fn ensure_parent_dir(file_path: &Path) -> RuntimeResult<()> {
264    if let Some(parent) = file_path.parent() {
265        fs::create_dir_all(parent).map_err(|e| map_idempotency_io("create_store_parent_dir", e))?;
266    }
267    Ok(())
268}
269
270fn validate_key(key: &str) -> RuntimeResult<()> {
271    match validate_identifier("IdempotencyKey", key) {
272        Ok(()) => Ok(()),
273        Err(RuntimeError::InvalidIdentifier {
274            reason: "empty", ..
275        }) => Err(RuntimeError::InvalidIdempotencyKey { reason: "empty" }),
276        Err(_) => Err(RuntimeError::InvalidIdempotencyKey {
277            reason: "invalid_char",
278        }),
279    }
280}
281
282fn ensure_active_capacity<T>(entries: &HashMap<String, T>, key: &str) -> RuntimeResult<()> {
283    if !entries.contains_key(key) && entries.len() >= MAX_ACTIVE_IDEMPOTENCY_RECORDS {
284        return Err(RuntimeError::IdempotencyStoreIo {
285            operation: "validate_store",
286            message: "active record limit exceeded".to_string(),
287        });
288    }
289    Ok(())
290}
291
292fn is_expired(created_at_ms: u64, ttl_ms: Option<u64>, now_ms: u64) -> bool {
293    if created_at_ms == 0 {
294        return false;
295    }
296    match ttl_ms {
297        Some(ttl) => now_ms.saturating_sub(created_at_ms) > ttl,
298        None => false,
299    }
300}
301
302fn now_ms() -> u64 {
303    SystemTime::now()
304        .duration_since(UNIX_EPOCH)
305        .map(|d| d.as_millis() as u64)
306        .unwrap_or(0)
307}
308
309#[cfg(test)]
310#[path = "idempotency_tests.rs"]
311mod tests;
appcore_core/idempotency.rs

appcore_core/
idempotency.rs
