zerodds_durability_store/
memory.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3
4//! In-memory [`DurabilityStore`] — the cold store for `TRANSIENT` (no disk)
5//! and the reference/test vehicle for the trait semantics.
6
7use alloc::collections::BTreeMap;
8use alloc::string::{String, ToString};
9use alloc::vec::Vec;
10use std::sync::Mutex;
11use std::time::SystemTime;
12
13use zerodds_qos::policies::history::HistoryKind;
14
15use crate::contract::Contract;
16use crate::error::{Result, StoreError};
17use crate::model::{Cursor, DurabilitySample, Page, Selector, StoreStats};
18use crate::store::{DEFAULT_PAGE, DurabilityStore};
19
20/// `(topic, instance_key)` map key. Ranging over a topic prefix yields all of
21/// its instances in `O(log n + k)`.
22type Key = (String, [u8; 16]);
23
24#[derive(Default)]
25struct Slot {
26    /// Samples kept sorted by `sequence`.
27    samples: Vec<DurabilitySample>,
28    /// `unregister` wall-clock time, if any.
29    unregistered_at: Option<SystemTime>,
30}
31
32#[derive(Default)]
33struct Inner {
34    default_contract: Contract,
35    contracts: BTreeMap<String, Contract>,
36    by_key: BTreeMap<Key, Slot>,
37}
38
39impl Inner {
40    fn contract_for(&self, topic: &str) -> Contract {
41        self.contracts
42            .get(topic)
43            .copied()
44            .unwrap_or(self.default_contract)
45    }
46
47    /// Inclusive `(topic, ..)` range bounds for the BTreeMap.
48    fn topic_range(topic: &str) -> (Key, Key) {
49        (
50            (topic.to_string(), [0u8; 16]),
51            (topic.to_string(), [0xffu8; 16]),
52        )
53    }
54
55    fn topic_sample_count(&self, topic: &str) -> usize {
56        let (lo, hi) = Self::topic_range(topic);
57        self.by_key
58            .range(lo..=hi)
59            .map(|(_, s)| s.samples.len())
60            .sum()
61    }
62
63    fn topic_instance_count(&self, topic: &str) -> usize {
64        let (lo, hi) = Self::topic_range(topic);
65        self.by_key.range(lo..=hi).count()
66    }
67}
68
69/// In-memory, contract-bounded durability store. Thread-safe via a single
70/// mutex. Used directly as the `TRANSIENT` cold store, and as the hot tier
71/// inside [`crate::TieredStore`].
72#[derive(Default)]
73pub struct InMemoryStore {
74    inner: Mutex<Inner>,
75}
76
77impl InMemoryStore {
78    /// New empty store with the given default contract (used for topics
79    /// without an explicit [`set_contract`](DurabilityStore::set_contract)).
80    #[must_use]
81    pub fn with_default_contract(default_contract: Contract) -> Self {
82        Self {
83            inner: Mutex::new(Inner {
84                default_contract,
85                ..Inner::default()
86            }),
87        }
88    }
89
90    fn lock(&self) -> Result<std::sync::MutexGuard<'_, Inner>> {
91        self.inner
92            .lock()
93            .map_err(|_| StoreError::Poisoned("in-memory store"))
94    }
95}
96
97/// Inserts `sample` into `slot`, enforcing the per-instance history policy.
98/// `KEEP_LAST` evicts the lowest sequence(s); `KEEP_ALL` rejects past the cap.
99/// Returns the signed change in the slot's sample count.
100fn push_into_slot(slot: &mut Slot, sample: DurabilitySample, contract: &Contract) -> Result<isize> {
101    match contract.history_kind {
102        HistoryKind::KeepAll => {
103            // A re-send of an existing sequence is an idempotent replace — it
104            // does not grow the slot, so the cap must not reject it.
105            let is_resend = slot
106                .samples
107                .binary_search_by(|s| s.sequence.cmp(&sample.sequence))
108                .is_ok();
109            if !is_resend
110                && contract.per_instance_bounded()
111                && slot.samples.len() >= contract.max_samples_per_instance as usize
112            {
113                return Err(StoreError::OutOfResources("max_samples_per_instance"));
114            }
115            let before = slot.samples.len();
116            insert_sorted(&mut slot.samples, sample);
117            Ok(slot.samples.len() as isize - before as isize)
118        }
119        HistoryKind::KeepLast => {
120            let before = slot.samples.len();
121            insert_sorted(&mut slot.samples, sample);
122            let depth = contract.effective_depth();
123            while slot.samples.len() > depth {
124                slot.samples.remove(0); // lowest sequence (sorted)
125            }
126            Ok(slot.samples.len() as isize - before as isize)
127        }
128    }
129}
130
131/// Inserts keeping `samples` sorted by `sequence` (duplicates by sequence
132/// replace in place — a re-sent sample is idempotent).
133fn insert_sorted(samples: &mut Vec<DurabilitySample>, sample: DurabilitySample) {
134    match samples.binary_search_by(|s| s.sequence.cmp(&sample.sequence)) {
135        Ok(pos) => samples[pos] = sample,
136        Err(pos) => samples.insert(pos, sample),
137    }
138}
139
140impl DurabilityStore for InMemoryStore {
141    fn set_contract(&self, topic: &str, contract: Contract) -> Result<()> {
142        self.lock()?.contracts.insert(topic.to_string(), contract);
143        Ok(())
144    }
145
146    fn store(&self, sample: DurabilitySample) -> Result<()> {
147        let mut g = self.lock()?;
148        let contract = g.contract_for(&sample.topic);
149        let topic = sample.topic.clone();
150        let key: Key = (topic.clone(), sample.instance_key);
151
152        // A re-send of an existing (topic, instance, sequence) is an idempotent
153        // replace — it grows nothing, so no cap may reject it.
154        let is_resend = g
155            .by_key
156            .get(&key)
157            .map(|slot| {
158                slot.samples
159                    .binary_search_by(|s| s.sequence.cmp(&sample.sequence))
160                    .is_ok()
161            })
162            .unwrap_or(false);
163
164        // Global per-topic caps (only meaningful under KEEP_ALL; KEEP_LAST
165        // self-bounds via depth but max_samples still applies as a ceiling).
166        if !is_resend
167            && contract.samples_bounded()
168            && g.topic_sample_count(&topic) >= contract.max_samples as usize
169            && matches!(contract.history_kind, HistoryKind::KeepAll)
170        {
171            return Err(StoreError::OutOfResources("max_samples"));
172        }
173        let new_instance = !g.by_key.contains_key(&key);
174        if new_instance
175            && contract.instances_bounded()
176            && g.topic_instance_count(&topic) >= contract.max_instances as usize
177        {
178            return Err(StoreError::OutOfResources("max_instances"));
179        }
180        let slot = g.by_key.entry(key).or_default();
181        push_into_slot(slot, sample, &contract)?;
182        Ok(())
183    }
184
185    fn query(&self, topic: &str, selector: &Selector) -> Result<Page> {
186        let g = self.lock()?;
187        let (lo, hi) = Inner::topic_range(topic);
188        let mut matched: Vec<DurabilitySample> = g
189            .by_key
190            .range(lo..=hi)
191            .flat_map(|(_, s)| s.samples.iter())
192            .filter(|s| selector.matches(s))
193            .cloned()
194            .collect();
195        matched.sort_by_key(|s| (s.instance_key, s.sequence));
196        let limit = selector.limit.unwrap_or(DEFAULT_PAGE);
197        let exhausted = matched.len() <= limit;
198        matched.truncate(limit);
199        let next: Option<Cursor> = if exhausted {
200            None
201        } else {
202            matched.last().map(|s| (s.instance_key, s.sequence))
203        };
204        Ok(Page {
205            samples: matched,
206            next,
207        })
208    }
209
210    fn unregister(&self, topic: &str, instance_key: &[u8; 16], now: SystemTime) -> Result<()> {
211        let mut g = self.lock()?;
212        if let Some(slot) = g.by_key.get_mut(&(topic.to_string(), *instance_key)) {
213            slot.unregistered_at = Some(now);
214        }
215        Ok(())
216    }
217
218    fn cleanup(&self, now: SystemTime) -> Result<usize> {
219        let mut g = self.lock()?;
220        let due: Vec<Key> = g
221            .by_key
222            .iter()
223            .filter_map(|(k, slot)| {
224                let ts = slot.unregistered_at?;
225                let delay = g.contract_for(&k.0).cleanup_delay;
226                let deadline = ts.checked_add(delay)?;
227                if now >= deadline {
228                    Some(k.clone())
229                } else {
230                    None
231                }
232            })
233            .collect();
234        let removed = due.len();
235        for k in due {
236            g.by_key.remove(&k);
237        }
238        Ok(removed)
239    }
240
241    fn stats(&self, topic: &str) -> Result<StoreStats> {
242        let g = self.lock()?;
243        let (lo, hi) = Inner::topic_range(topic);
244        let mut stats = StoreStats::default();
245        for (_, slot) in g.by_key.range(lo..=hi) {
246            stats.instances += 1;
247            stats.samples += slot.samples.len();
248            stats.bytes += slot
249                .samples
250                .iter()
251                .map(|s| s.payload.len() as u64)
252                .sum::<u64>();
253        }
254        Ok(stats)
255    }
256}
zerodds_durability_store/memory.rs

zerodds_durability_store/
memory.rs
