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/06/07 12:00:00 by dnettoRaw
8//      ###########      S: 0.6.0
9// =============================================================================
10
11//! Idempotency stores used by runtime controller command deduplication.
12
13use crate::error::{RuntimeError, RuntimeResult};
14use crate::ids::validate_identifier;
15use std::collections::HashMap;
16use std::fmt;
17use std::fs::{self, OpenOptions};
18use std::io::Write;
19use std::path::{Path, PathBuf};
20use std::sync::atomic::{AtomicU64, Ordering};
21use std::time::{SystemTime, UNIX_EPOCH};
22
23/// Stable on-disk format marker for the file idempotency store.
24pub const IDEMPOTENCY_FORMAT_V1: &str = "# appcore-idempotency-v1";
25const MAX_IDEMPOTENCY_FILE_BYTES: u64 = 64 * 1024 * 1024;
26static IDEMPOTENCY_TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);
27
28/// Status of an idempotency execution.
29#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
30pub enum IdempotencyStatus {
31    /// Command execution has been reserved but not completed.
32    Pending,
33    /// Command execution completed and its serialized response is reusable.
34    Resolved {
35        /// Stable response status.
36        response_status: u16,
37        /// Serialized response body.
38        response_body: String,
39    },
40}
41
42/// A stored idempotency execution record.
43#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
44pub struct IdempotencyRecord {
45    /// Validated idempotency key.
46    pub key: String,
47    /// Digest that binds the key to one logical request.
48    pub request_hash: String,
49    /// Current execution status.
50    pub status: IdempotencyStatus,
51    /// Creation timestamp in Unix milliseconds.
52    pub created_at_ms: u64,
53}
54
55/// Durable or process-local idempotency record boundary.
56pub trait IdempotencyStore: Send + Sync {
57    /// Returns a stored record by key.
58    fn get(&self, key: &str) -> RuntimeResult<Option<IdempotencyRecord>>;
59    /// Inserts or replaces a validated record.
60    fn insert(&mut self, record: IdempotencyRecord) -> RuntimeResult<()>;
61    /// Returns the number of active records.
62    fn len(&self) -> usize;
63
64    /// Removes a record when supported.
65    fn remove(&mut self, _key: &str) -> RuntimeResult<()> {
66        Ok(())
67    }
68
69    /// Reports whether no active records exist.
70    fn is_empty(&self) -> bool {
71        self.len() == 0
72    }
73}
74
75/// Process-local idempotency store.
76#[derive(Default)]
77pub struct InMemoryIdempotencyStore {
78    seen: HashMap<String, IdempotencyRecord>,
79}
80
81impl fmt::Debug for InMemoryIdempotencyStore {
82    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
83        f.debug_struct("InMemoryIdempotencyStore")
84            .field("entry_count", &self.seen.len())
85            .finish()
86    }
87}
88
89impl InMemoryIdempotencyStore {
90    /// Creates an empty process-local store.
91    pub fn new() -> Self {
92        Self::default()
93    }
94}
95
96impl IdempotencyStore for InMemoryIdempotencyStore {
97    fn get(&self, key: &str) -> RuntimeResult<Option<IdempotencyRecord>> {
98        Ok(self.seen.get(key).cloned())
99    }
100
101    fn insert(&mut self, record: IdempotencyRecord) -> RuntimeResult<()> {
102        validate_key(&record.key)?;
103        self.seen.insert(record.key.clone(), record);
104        Ok(())
105    }
106
107    fn len(&self) -> usize {
108        self.seen.len()
109    }
110
111    fn remove(&mut self, key: &str) -> RuntimeResult<()> {
112        self.seen.remove(key);
113        Ok(())
114    }
115}
116
117/// Append-oriented local idempotency store with atomic compaction.
118pub struct FileIdempotencyStore {
119    file_path: PathBuf,
120    ttl_ms: Option<u64>,
121    seen: HashMap<String, IdempotencyRecord>,
122}
123
124impl fmt::Debug for FileIdempotencyStore {
125    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
126        f.debug_struct("FileIdempotencyStore")
127            .field("file_path", &self.file_path)
128            .field("ttl_ms", &self.ttl_ms)
129            .field("entry_count", &self.seen.len())
130            .finish()
131    }
132}
133
134fn map_idempotency_io(operation: &'static str, err: std::io::Error) -> RuntimeError {
135    RuntimeError::IdempotencyStoreIo {
136        operation,
137        message: err.to_string(),
138    }
139}
140
141impl FileIdempotencyStore {
142    /// Opens a store without expiration.
143    pub fn new(path: impl AsRef<Path>) -> RuntimeResult<Self> {
144        Self::new_with_ttl(path, None)
145    }
146
147    /// Opens a store with optional record expiration.
148    pub fn new_with_ttl(path: impl AsRef<Path>, ttl_ms: Option<u64>) -> RuntimeResult<Self> {
149        let file_path = path.as_ref().to_path_buf();
150        ensure_parent_dir(&file_path)?;
151        if !file_path.exists() {
152            rewrite_entries(&file_path, &HashMap::new())?;
153        }
154        let (seen, needs_rewrite) = load_entries(&file_path)?;
155        if needs_rewrite {
156            rewrite_entries(&file_path, &seen)?;
157        }
158        let ttl_ms = match ttl_ms {
159            Some(0) => None,
160            other => other,
161        };
162
163        Ok(Self {
164            file_path,
165            ttl_ms,
166            seen,
167        })
168    }
169
170    /// Returns the backing file path.
171    pub fn file_path(&self) -> &Path {
172        &self.file_path
173    }
174
175    /// Removes expired records and atomically rewrites the backing file.
176    pub fn compact(&mut self, now_ms: u64) -> RuntimeResult<usize> {
177        let before = self.seen.len();
178        self.seen
179            .retain(|_, record| !is_expired(record.created_at_ms, self.ttl_ms, now_ms));
180        let removed = before.saturating_sub(self.seen.len());
181
182        rewrite_entries(&self.file_path, &self.seen)?;
183
184        Ok(removed)
185    }
186}
187
188impl IdempotencyStore for FileIdempotencyStore {
189    fn get(&self, key: &str) -> RuntimeResult<Option<IdempotencyRecord>> {
190        let now_ms = now_ms();
191        if let Some(record) = self.seen.get(key) {
192            if is_expired(record.created_at_ms, self.ttl_ms, now_ms) {
193                Ok(None)
194            } else {
195                Ok(Some(record.clone()))
196            }
197        } else {
198            Ok(None)
199        }
200    }
201
202    fn insert(&mut self, record: IdempotencyRecord) -> RuntimeResult<()> {
203        validate_key(&record.key)?;
204        append_entry(&self.file_path, &record)?;
205        self.seen.insert(record.key.clone(), record);
206        Ok(())
207    }
208
209    fn len(&self) -> usize {
210        let now_ms = now_ms();
211        self.seen
212            .values()
213            .filter(|record| !is_expired(record.created_at_ms, self.ttl_ms, now_ms))
214            .count()
215    }
216
217    fn remove(&mut self, key: &str) -> RuntimeResult<()> {
218        if self.seen.remove(key).is_some() {
219            rewrite_entries(&self.file_path, &self.seen)?;
220        }
221        Ok(())
222    }
223}
224
225fn ensure_parent_dir(file_path: &Path) -> RuntimeResult<()> {
226    if let Some(parent) = file_path.parent() {
227        fs::create_dir_all(parent).map_err(|e| map_idempotency_io("create_store_parent_dir", e))?;
228    }
229    Ok(())
230}
231
232fn load_entries(path: &Path) -> RuntimeResult<(HashMap<String, IdempotencyRecord>, bool)> {
233    reject_symlink(path)?;
234    let metadata =
235        fs::metadata(path).map_err(|error| map_idempotency_io("read_store_metadata", error))?;
236    if metadata.len() > MAX_IDEMPOTENCY_FILE_BYTES {
237        return Err(corrupt_idempotency("store exceeds size limit"));
238    }
239    let text = fs::read_to_string(path).map_err(|error| map_idempotency_io("read_store", error))?;
240    let body = split_idempotency_format(&text)?;
241    let (complete, recovered_tail) = complete_line_prefix(body);
242    let mut seen = HashMap::new();
243
244    for line in complete.lines() {
245        let trimmed = line.trim();
246        if trimmed.is_empty() {
247            continue;
248        }
249        let record = parse_idempotency_record(trimmed)?;
250        validate_key(&record.key)?;
251        if matches!(record.status, IdempotencyStatus::Resolved { .. }) {
252            seen.insert(record.key.clone(), record);
253        }
254    }
255    Ok((seen, recovered_tail))
256}
257
258fn append_entry(path: &Path, record: &IdempotencyRecord) -> RuntimeResult<()> {
259    let mut file = OpenOptions::new()
260        .append(true)
261        .open(path)
262        .map_err(|e| map_idempotency_io("open_store_for_append", e))?;
263    let line = serde_json::to_string(record).map_err(|e| RuntimeError::IdempotencyStoreIo {
264        operation: "serialize_store_entry",
265        message: e.to_string(),
266    })?;
267    writeln!(file, "{}", line).map_err(|e| map_idempotency_io("append_store_entry", e))?;
268    file.sync_data()
269        .map_err(|e| map_idempotency_io("sync_store_entry", e))?;
270    Ok(())
271}
272
273fn rewrite_entries(path: &Path, entries: &HashMap<String, IdempotencyRecord>) -> RuntimeResult<()> {
274    let mut rows: Vec<(&String, &IdempotencyRecord)> = entries.iter().collect();
275    rows.sort_by(|a, b| a.0.cmp(b.0));
276
277    let parent = path.parent().unwrap_or_else(|| Path::new("."));
278    reject_symlink(path)?;
279    let temp_name = format!(
280        ".{}.{}-{}.tmp",
281        path.file_name()
282            .and_then(|n| n.to_str())
283            .unwrap_or("idempotency"),
284        std::process::id(),
285        IDEMPOTENCY_TEMP_COUNTER.fetch_add(1, Ordering::Relaxed)
286    );
287    let temp_path = parent.join(temp_name);
288
289    let result = (|| {
290        let mut file = OpenOptions::new()
291            .create_new(true)
292            .write(true)
293            .open(&temp_path)
294            .map_err(|e| map_idempotency_io("open_temp_store_for_rewrite", e))?;
295        writeln!(file, "{IDEMPOTENCY_FORMAT_V1}")
296            .map_err(|e| map_idempotency_io("write_store_format", e))?;
297        for (_, record) in rows {
298            let line =
299                serde_json::to_string(record).map_err(|e| RuntimeError::IdempotencyStoreIo {
300                    operation: "serialize_store_entry",
301                    message: e.to_string(),
302                })?;
303            writeln!(file, "{}", line).map_err(|e| map_idempotency_io("rewrite_store_entry", e))?;
304        }
305        file.sync_all()
306            .map_err(|e| map_idempotency_io("sync_temp_store", e))?;
307        fs::rename(&temp_path, path).map_err(|e| map_idempotency_io("rename_temp_store", e))?;
308        sync_parent_directory(parent)
309    })();
310    if result.is_err() {
311        let _ = fs::remove_file(temp_path);
312    }
313    result
314}
315
316fn split_idempotency_format(text: &str) -> RuntimeResult<&str> {
317    if let Some(body) = text
318        .strip_prefix(IDEMPOTENCY_FORMAT_V1)
319        .and_then(|rest| rest.strip_prefix('\n'))
320    {
321        return Ok(body);
322    }
323    if text == IDEMPOTENCY_FORMAT_V1 {
324        return Ok("");
325    }
326    if text.starts_with("# appcore-") {
327        return Err(corrupt_idempotency("NO MORE SUPPORTED PLEASE UPDATE"));
328    }
329    Err(corrupt_idempotency("NO MORE SUPPORTED PLEASE UPDATE"))
330}
331
332fn complete_line_prefix(body: &str) -> (&str, bool) {
333    if body.is_empty() || body.ends_with('\n') {
334        return (body, false);
335    }
336    match body.rfind('\n') {
337        Some(last_newline) => (&body[..=last_newline], true),
338        None => ("", true),
339    }
340}
341
342fn parse_idempotency_record(line: &str) -> RuntimeResult<IdempotencyRecord> {
343    if !line.starts_with('{') {
344        return Err(corrupt_idempotency("NO MORE SUPPORTED PLEASE UPDATE"));
345    }
346    serde_json::from_str(line).map_err(|_| corrupt_idempotency("invalid JSON record"))
347}
348
349fn reject_symlink(path: &Path) -> RuntimeResult<()> {
350    match fs::symlink_metadata(path) {
351        Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
352            Err(corrupt_idempotency("store path is not a regular file"))
353        }
354        Ok(_) => Ok(()),
355        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
356        Err(error) => Err(map_idempotency_io("inspect_store_path", error)),
357    }
358}
359
360fn corrupt_idempotency(message: &str) -> RuntimeError {
361    RuntimeError::IdempotencyStoreIo {
362        operation: "validate_store",
363        message: message.to_string(),
364    }
365}
366
367#[cfg(unix)]
368fn sync_parent_directory(path: &Path) -> RuntimeResult<()> {
369    fs::File::open(path)
370        .and_then(|directory| directory.sync_all())
371        .map_err(|error| map_idempotency_io("sync_store_parent", error))
372}
373
374#[cfg(not(unix))]
375fn sync_parent_directory(_path: &Path) -> RuntimeResult<()> {
376    Ok(())
377}
378
379fn validate_key(key: &str) -> RuntimeResult<()> {
380    match validate_identifier("IdempotencyKey", key) {
381        Ok(()) => Ok(()),
382        Err(RuntimeError::InvalidIdentifier {
383            reason: "empty", ..
384        }) => Err(RuntimeError::InvalidIdempotencyKey { reason: "empty" }),
385        Err(_) => Err(RuntimeError::InvalidIdempotencyKey {
386            reason: "invalid_char",
387        }),
388    }
389}
390
391fn is_expired(created_at_ms: u64, ttl_ms: Option<u64>, now_ms: u64) -> bool {
392    if created_at_ms == 0 {
393        return false;
394    }
395    match ttl_ms {
396        Some(ttl) => now_ms.saturating_sub(created_at_ms) > ttl,
397        None => false,
398    }
399}
400
401fn now_ms() -> u64 {
402    SystemTime::now()
403        .duration_since(UNIX_EPOCH)
404        .map(|d| d.as_millis() as u64)
405        .unwrap_or(0)
406}
407
408#[cfg(test)]
409#[path = "idempotency_tests.rs"]
410mod tests;
appcore_core/idempotency.rs

appcore_core/
idempotency.rs
