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::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;
26// appcore-norm: allow(global-state) reason: atomic sequence prevents process-local temporary path collisions
27static IDEMPOTENCY_TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);
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        self.seen.insert(record.key.clone(), record);
105        Ok(())
106    }
107
108    fn len(&self) -> usize {
109        self.seen.len()
110    }
111
112    fn remove(&mut self, key: &str) -> RuntimeResult<()> {
113        self.seen.remove(key);
114        Ok(())
115    }
116}
117
118/// Append-oriented local idempotency store with atomic compaction.
119pub struct FileIdempotencyStore {
120    file_path: PathBuf,
121    ttl_ms: Option<u64>,
122    seen: HashMap<String, IdempotencyRecord>,
123}
124
125impl fmt::Debug for FileIdempotencyStore {
126    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
127        f.debug_struct("FileIdempotencyStore")
128            .field("file_path", &self.file_path)
129            .field("ttl_ms", &self.ttl_ms)
130            .field("entry_count", &self.seen.len())
131            .finish()
132    }
133}
134
135fn map_idempotency_io(operation: &'static str, err: std::io::Error) -> RuntimeError {
136    RuntimeError::IdempotencyStoreIo {
137        operation,
138        message: err.to_string(),
139    }
140}
141
142impl FileIdempotencyStore {
143    /// Opens a store without expiration.
144    pub fn new(path: impl AsRef<Path>) -> RuntimeResult<Self> {
145        Self::new_with_ttl(path, None)
146    }
147
148    /// Opens a store with optional record expiration.
149    pub fn new_with_ttl(path: impl AsRef<Path>, ttl_ms: Option<u64>) -> RuntimeResult<Self> {
150        let file_path = path.as_ref().to_path_buf();
151        ensure_parent_dir(&file_path)?;
152        if !file_path.exists() {
153            rewrite_entries(&file_path, &HashMap::new())?;
154        }
155        let (seen, needs_rewrite) = load_entries(&file_path)?;
156        if needs_rewrite {
157            rewrite_entries(&file_path, &seen)?;
158        }
159        let ttl_ms = match ttl_ms {
160            Some(0) => None,
161            other => other,
162        };
163
164        Ok(Self {
165            file_path,
166            ttl_ms,
167            seen,
168        })
169    }
170
171    /// Returns the backing file path.
172    pub fn file_path(&self) -> &Path {
173        &self.file_path
174    }
175
176    /// Removes expired records and atomically rewrites the backing file.
177    pub fn compact(&mut self, now_ms: u64) -> RuntimeResult<usize> {
178        let before = self.seen.len();
179        self.seen
180            .retain(|_, record| !is_expired(record.created_at_ms, self.ttl_ms, now_ms));
181        let removed = before.saturating_sub(self.seen.len());
182
183        rewrite_entries(&self.file_path, &self.seen)?;
184
185        Ok(removed)
186    }
187}
188
189impl IdempotencyStore for FileIdempotencyStore {
190    fn get(&self, key: &str) -> RuntimeResult<Option<IdempotencyRecord>> {
191        let now_ms = now_ms();
192        if let Some(record) = self.seen.get(key) {
193            if is_expired(record.created_at_ms, self.ttl_ms, now_ms) {
194                Ok(None)
195            } else {
196                Ok(Some(record.clone()))
197            }
198        } else {
199            Ok(None)
200        }
201    }
202
203    fn insert(&mut self, record: IdempotencyRecord) -> RuntimeResult<()> {
204        validate_key(&record.key)?;
205        append_entry(&self.file_path, &record)?;
206        self.seen.insert(record.key.clone(), record);
207        Ok(())
208    }
209
210    fn len(&self) -> usize {
211        let now_ms = now_ms();
212        self.seen
213            .values()
214            .filter(|record| !is_expired(record.created_at_ms, self.ttl_ms, now_ms))
215            .count()
216    }
217
218    fn remove(&mut self, key: &str) -> RuntimeResult<()> {
219        if self.seen.remove(key).is_some() {
220            rewrite_entries(&self.file_path, &self.seen)?;
221        }
222        Ok(())
223    }
224}
225
226fn ensure_parent_dir(file_path: &Path) -> RuntimeResult<()> {
227    if let Some(parent) = file_path.parent() {
228        fs::create_dir_all(parent).map_err(|e| map_idempotency_io("create_store_parent_dir", e))?;
229    }
230    Ok(())
231}
232
233fn load_entries(path: &Path) -> RuntimeResult<(HashMap<String, IdempotencyRecord>, bool)> {
234    reject_symlink(path)?;
235    let metadata =
236        fs::metadata(path).map_err(|error| map_idempotency_io("read_store_metadata", error))?;
237    if metadata.len() > MAX_IDEMPOTENCY_FILE_BYTES {
238        return Err(corrupt_idempotency("store exceeds size limit"));
239    }
240    let text = fs::read_to_string(path).map_err(|error| map_idempotency_io("read_store", error))?;
241    let body = split_idempotency_format(&text)?;
242    let (complete, recovered_tail) = complete_line_prefix(body);
243    let mut seen = HashMap::new();
244
245    for line in complete.lines() {
246        let trimmed = line.trim();
247        if trimmed.is_empty() {
248            continue;
249        }
250        let record = parse_idempotency_record(trimmed)?;
251        validate_key(&record.key)?;
252        if matches!(record.status, IdempotencyStatus::Resolved { .. }) {
253            seen.insert(record.key.clone(), record);
254        }
255    }
256    Ok((seen, recovered_tail))
257}
258
259fn append_entry(path: &Path, record: &IdempotencyRecord) -> RuntimeResult<()> {
260    let mut file = OpenOptions::new()
261        .append(true)
262        .open(path)
263        .map_err(|e| map_idempotency_io("open_store_for_append", e))?;
264    let line = serde_json::to_string(record).map_err(|e| RuntimeError::IdempotencyStoreIo {
265        operation: "serialize_store_entry",
266        message: e.to_string(),
267    })?;
268    writeln!(file, "{}", line).map_err(|e| map_idempotency_io("append_store_entry", e))?;
269    file.sync_data()
270        .map_err(|e| map_idempotency_io("sync_store_entry", e))?;
271    Ok(())
272}
273
274fn rewrite_entries(path: &Path, entries: &HashMap<String, IdempotencyRecord>) -> RuntimeResult<()> {
275    let mut rows: Vec<(&String, &IdempotencyRecord)> = entries.iter().collect();
276    rows.sort_by(|a, b| a.0.cmp(b.0));
277
278    let parent = path.parent().unwrap_or_else(|| Path::new("."));
279    reject_symlink(path)?;
280    let temp_name = format!(
281        ".{}.{}-{}.tmp",
282        path.file_name()
283            .and_then(|n| n.to_str())
284            .unwrap_or("idempotency"),
285        std::process::id(),
286        IDEMPOTENCY_TEMP_COUNTER.fetch_add(1, Ordering::Relaxed)
287    );
288    let temp_path = parent.join(temp_name);
289
290    let result = (|| {
291        let mut file = OpenOptions::new()
292            .create_new(true)
293            .write(true)
294            .open(&temp_path)
295            .map_err(|e| map_idempotency_io("open_temp_store_for_rewrite", e))?;
296        writeln!(file, "{IDEMPOTENCY_FORMAT_V1}")
297            .map_err(|e| map_idempotency_io("write_store_format", e))?;
298        for (_, record) in rows {
299            let line =
300                serde_json::to_string(record).map_err(|e| RuntimeError::IdempotencyStoreIo {
301                    operation: "serialize_store_entry",
302                    message: e.to_string(),
303                })?;
304            writeln!(file, "{}", line).map_err(|e| map_idempotency_io("rewrite_store_entry", e))?;
305        }
306        file.sync_all()
307            .map_err(|e| map_idempotency_io("sync_temp_store", e))?;
308        fs::rename(&temp_path, path).map_err(|e| map_idempotency_io("rename_temp_store", e))?;
309        sync_parent_directory(parent)
310    })();
311    if result.is_err() {
312        let _ = fs::remove_file(temp_path);
313    }
314    result
315}
316
317fn split_idempotency_format(text: &str) -> RuntimeResult<&str> {
318    if let Some(body) = text
319        .strip_prefix(IDEMPOTENCY_FORMAT_V1)
320        .and_then(|rest| rest.strip_prefix('\n'))
321    {
322        return Ok(body);
323    }
324    if text == IDEMPOTENCY_FORMAT_V1 {
325        return Ok("");
326    }
327    if text.starts_with("# appcore-") {
328        return Err(corrupt_idempotency("NO MORE SUPPORTED PLEASE UPDATE"));
329    }
330    Err(corrupt_idempotency("NO MORE SUPPORTED PLEASE UPDATE"))
331}
332
333fn complete_line_prefix(body: &str) -> (&str, bool) {
334    if body.is_empty() || body.ends_with('\n') {
335        return (body, false);
336    }
337    match body.rfind('\n') {
338        Some(last_newline) => (&body[..=last_newline], true),
339        None => ("", true),
340    }
341}
342
343fn parse_idempotency_record(line: &str) -> RuntimeResult<IdempotencyRecord> {
344    if !line.starts_with('{') {
345        return Err(corrupt_idempotency("NO MORE SUPPORTED PLEASE UPDATE"));
346    }
347    serde_json::from_str(line).map_err(|_| corrupt_idempotency("invalid JSON record"))
348}
349
350fn reject_symlink(path: &Path) -> RuntimeResult<()> {
351    match fs::symlink_metadata(path) {
352        Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
353            Err(corrupt_idempotency("store path is not a regular file"))
354        }
355        Ok(_) => Ok(()),
356        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
357        Err(error) => Err(map_idempotency_io("inspect_store_path", error)),
358    }
359}
360
361fn corrupt_idempotency(message: &str) -> RuntimeError {
362    RuntimeError::IdempotencyStoreIo {
363        operation: "validate_store",
364        message: message.to_string(),
365    }
366}
367
368#[cfg(unix)]
369fn sync_parent_directory(path: &Path) -> RuntimeResult<()> {
370    fs::File::open(path)
371        .and_then(|directory| directory.sync_all())
372        .map_err(|error| map_idempotency_io("sync_store_parent", error))
373}
374
375#[cfg(not(unix))]
376fn sync_parent_directory(_path: &Path) -> RuntimeResult<()> {
377    Ok(())
378}
379
380fn validate_key(key: &str) -> RuntimeResult<()> {
381    match validate_identifier("IdempotencyKey", key) {
382        Ok(()) => Ok(()),
383        Err(RuntimeError::InvalidIdentifier {
384            reason: "empty", ..
385        }) => Err(RuntimeError::InvalidIdempotencyKey { reason: "empty" }),
386        Err(_) => Err(RuntimeError::InvalidIdempotencyKey {
387            reason: "invalid_char",
388        }),
389    }
390}
391
392fn is_expired(created_at_ms: u64, ttl_ms: Option<u64>, now_ms: u64) -> bool {
393    if created_at_ms == 0 {
394        return false;
395    }
396    match ttl_ms {
397        Some(ttl) => now_ms.saturating_sub(created_at_ms) > ttl,
398        None => false,
399    }
400}
401
402fn now_ms() -> u64 {
403    SystemTime::now()
404        .duration_since(UNIX_EPOCH)
405        .map(|d| d.as_millis() as u64)
406        .unwrap_or(0)
407}
408
409#[cfg(test)]
410#[path = "idempotency_tests.rs"]
411mod tests;
appcore_core/idempotency.rs

appcore_core/
idempotency.rs
