appcore_peer_rpc/
replay.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: replay.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/24 11:51:10 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/24 11:51:10 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Bounded replay-store policy shared by authenticated Runtime ingress.
12
13use crate::{PeerNonceStore, PeerRpcError};
14use std::collections::{BTreeMap, VecDeque};
15use std::sync::Mutex;
16
17/// Capacity, TTL, and cleanup policy for a replay store.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub struct ReplayStoreConfig {
20    /// Maximum number of live replay identities.
21    pub max_entries: usize,
22    /// Maximum time a caller-provided expiry may retain an identity.
23    pub ttl_ms: u64,
24    /// Maximum interval between full expired-entry cleanup passes.
25    pub cleanup_interval_ms: u64,
26}
27
28impl ReplayStoreConfig {
29    /// Validates and creates a replay-store policy.
30    pub fn new(
31        max_entries: usize,
32        ttl_ms: u64,
33        cleanup_interval_ms: u64,
34    ) -> Result<Self, PeerRpcError> {
35        if max_entries == 0 || ttl_ms == 0 || cleanup_interval_ms == 0 {
36            return Err(PeerRpcError::InvalidEnvelope(
37                "replay_store_policy_invalid".to_string(),
38            ));
39        }
40        Ok(Self {
41            max_entries,
42            ttl_ms,
43            cleanup_interval_ms,
44        })
45    }
46}
47
48impl Default for ReplayStoreConfig {
49    fn default() -> Self {
50        Self {
51            max_entries: super::MAX_NONCE_CACHE_ENTRIES,
52            ttl_ms: 60_000,
53            cleanup_interval_ms: 1_000,
54        }
55    }
56}
57
58/// Point-in-time replay-store metrics.
59#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
60pub struct ReplayStoreMetrics {
61    /// Current number of live entries.
62    pub entries: usize,
63    /// Identities accepted since creation.
64    pub accepted: u64,
65    /// Live replay attempts rejected since creation.
66    pub replays: u64,
67    /// Expired identities removed since creation.
68    pub expired: u64,
69    /// Requests rejected because all bounded entries were live.
70    pub capacity_rejections: u64,
71    /// Full cleanup passes executed since creation.
72    pub cleanups: u64,
73}
74
75/// Replay protection with explicit cleanup and observable bounds.
76pub trait ReplayStore: PeerNonceStore {
77    /// Removes entries whose effective TTL has elapsed.
78    fn cleanup(&self, now_ms: u64) -> Result<usize, PeerRpcError>;
79    /// Returns bounded non-sensitive metrics.
80    fn metrics(&self) -> ReplayStoreMetrics;
81}
82
83#[derive(Debug, Default)]
84struct ReplayState {
85    entries: BTreeMap<String, u64>,
86    lru: VecDeque<String>,
87    last_cleanup_ms: u64,
88    metrics: ReplayStoreMetrics,
89}
90
91/// Process-local TTL replay store with LRU ordering and fail-closed capacity.
92///
93/// Live entries are never evicted merely to make space because doing so would
94/// reopen their replay window. LRU ordering is used to remove expired entries
95/// deterministically; a full live set rejects new requests.
96#[derive(Debug)]
97pub struct BoundedReplayStore {
98    config: ReplayStoreConfig,
99    state: Mutex<ReplayState>,
100}
101
102impl BoundedReplayStore {
103    /// Creates an empty bounded replay store.
104    pub fn new(config: ReplayStoreConfig) -> Self {
105        Self {
106            config,
107            state: Mutex::new(ReplayState::default()),
108        }
109    }
110
111    fn cleanup_locked(&self, state: &mut ReplayState, now_ms: u64) -> usize {
112        let before = state.entries.len();
113        state.entries.retain(|_, expiry| *expiry > now_ms);
114        state
115            .lru
116            .retain(|identity| state.entries.contains_key(identity));
117        let removed = before.saturating_sub(state.entries.len());
118        state.last_cleanup_ms = now_ms;
119        state.metrics.expired = state.metrics.expired.saturating_add(removed as u64);
120        state.metrics.cleanups = state.metrics.cleanups.saturating_add(1);
121        state.metrics.entries = state.entries.len();
122        removed
123    }
124
125    fn cleanup_if_due(&self, state: &mut ReplayState, now_ms: u64) {
126        if now_ms.saturating_sub(state.last_cleanup_ms) >= self.config.cleanup_interval_ms {
127            self.cleanup_locked(state, now_ms);
128        }
129    }
130
131    fn touch_lru(state: &mut ReplayState, identity: &str) {
132        if let Some(position) = state.lru.iter().position(|entry| entry == identity) {
133            state.lru.remove(position);
134        }
135        state.lru.push_back(identity.to_string());
136    }
137}
138
139impl PeerNonceStore for BoundedReplayStore {
140    fn check_and_record(
141        &self,
142        nonce: &str,
143        expires_at_ms: u64,
144        now_ms: u64,
145    ) -> Result<(), PeerRpcError> {
146        if nonce.is_empty() || expires_at_ms <= now_ms {
147            return Err(PeerRpcError::InvalidEnvelope(
148                "replay_identity_invalid".to_string(),
149            ));
150        }
151        let mut state = self
152            .state
153            .lock()
154            .map_err(|_| PeerRpcError::InvalidEnvelope("replay_store_poisoned".to_string()))?;
155        self.cleanup_if_due(&mut state, now_ms);
156        if state.entries.contains_key(nonce) {
157            Self::touch_lru(&mut state, nonce);
158            state.metrics.replays = state.metrics.replays.saturating_add(1);
159            return Err(PeerRpcError::NonceReplay);
160        }
161        if state.entries.len() >= self.config.max_entries {
162            self.cleanup_locked(&mut state, now_ms);
163            if state.entries.len() >= self.config.max_entries {
164                state.metrics.capacity_rejections =
165                    state.metrics.capacity_rejections.saturating_add(1);
166                return Err(PeerRpcError::NonceCacheFull);
167            }
168        }
169        let ttl_expiry = now_ms.saturating_add(self.config.ttl_ms);
170        let effective_expiry = expires_at_ms.min(ttl_expiry);
171        state.entries.insert(nonce.to_string(), effective_expiry);
172        state.lru.push_back(nonce.to_string());
173        state.metrics.accepted = state.metrics.accepted.saturating_add(1);
174        state.metrics.entries = state.entries.len();
175        Ok(())
176    }
177}
178
179impl ReplayStore for BoundedReplayStore {
180    fn cleanup(&self, now_ms: u64) -> Result<usize, PeerRpcError> {
181        let mut state = self
182            .state
183            .lock()
184            .map_err(|_| PeerRpcError::InvalidEnvelope("replay_store_poisoned".to_string()))?;
185        Ok(self.cleanup_locked(&mut state, now_ms))
186    }
187
188    fn metrics(&self) -> ReplayStoreMetrics {
189        self.state
190            .lock()
191            .map(|state| state.metrics)
192            .unwrap_or_default()
193    }
194}
195
196#[cfg(test)]
197mod tests {
198    use super::*;
199
200    fn store(max_entries: usize) -> BoundedReplayStore {
201        BoundedReplayStore::new(ReplayStoreConfig::new(max_entries, 100, 10).unwrap())
202    }
203
204    #[test]
205    fn rejects_live_replay_and_reports_metrics() {
206        let store = store(2);
207        assert!(store.check_and_record("nonce-a", 100, 1).is_ok());
208        assert_eq!(
209            store.check_and_record("nonce-a", 100, 2),
210            Err(PeerRpcError::NonceReplay)
211        );
212        assert_eq!(
213            store.metrics(),
214            ReplayStoreMetrics {
215                entries: 1,
216                accepted: 1,
217                replays: 1,
218                ..ReplayStoreMetrics::default()
219            }
220        );
221    }
222
223    #[test]
224    fn expires_entries_and_applies_ttl_cap() {
225        let store = store(1);
226        assert!(store.check_and_record("nonce-a", 10_000, 1).is_ok());
227        assert_eq!(
228            store.check_and_record("nonce-b", 10_000, 2),
229            Err(PeerRpcError::NonceCacheFull)
230        );
231        assert!(store.check_and_record("nonce-b", 10_000, 102).is_ok());
232        assert_eq!(store.metrics().entries, 1);
233        assert_eq!(store.metrics().expired, 1);
234    }
235
236    #[test]
237    fn full_live_store_fails_closed_without_lru_eviction() {
238        let store = store(1);
239        store.check_and_record("nonce-a", 100, 1).unwrap();
240
241        assert_eq!(
242            store.check_and_record("nonce-b", 100, 2),
243            Err(PeerRpcError::NonceCacheFull)
244        );
245        assert_eq!(
246            store.check_and_record("nonce-a", 100, 3),
247            Err(PeerRpcError::NonceReplay)
248        );
249        assert_eq!(store.metrics().capacity_rejections, 1);
250    }
251
252    #[test]
253    fn replay_touch_updates_lru_age_without_evicting_live_entries() {
254        let store = store(2);
255        store.check_and_record("nonce-a", 100, 1).unwrap();
256        store.check_and_record("nonce-b", 100, 2).unwrap();
257        assert_eq!(
258            store.check_and_record("nonce-a", 100, 3),
259            Err(PeerRpcError::NonceReplay)
260        );
261
262        let state = store.state.lock().unwrap();
263        assert_eq!(
264            state.lru.iter().map(String::as_str).collect::<Vec<_>>(),
265            vec!["nonce-b", "nonce-a"]
266        );
267    }
268}
appcore_peer_rpc/replay.rs

appcore_peer_rpc/
replay.rs
