zerodds_durability_store/
store.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3
4//! The [`DurabilityStore`] trait — the storage abstraction every cold adapter
5//! implements (ADR 0009, layer ②b).
6
7use alloc::vec::Vec;
8use std::time::SystemTime;
9
10use crate::contract::Contract;
11use crate::error::Result;
12use crate::model::{DurabilitySample, Page, Selector, StoreStats};
13
14/// Default page size for the [`DurabilityStore::replay_for_topic`] convenience
15/// drain when a selector does not set its own `limit`.
16pub const DEFAULT_PAGE: usize = 1024;
17
18/// A durable, contract-bounded sample store. Implementations are the cold
19/// adapters (sqlite / file / lakehouse / in-memory). All methods are
20/// `&self` + `Send + Sync`: the daemon shares one store across threads.
21///
22/// Retention is bounded ONLY by the topic's `Contract` (ADR 0009 inv. 1):
23/// implementations enforce it on [`store`](DurabilityStore::store) (drop-oldest
24/// under `KEEP_LAST`, [`StoreError::OutOfResources`](crate::StoreError) under
25/// `KEEP_ALL` caps).
26pub trait DurabilityStore: Send + Sync {
27    /// Registers (or replaces) the retention [`Contract`] for `topic`. The
28    /// daemon calls this when it starts serving a topic. Topics without an
29    /// explicit contract use the store's default ([`Contract::default`]).
30    ///
31    /// # Errors
32    /// Backend/I/O failure.
33    fn set_contract(&self, topic: &str, contract: Contract) -> Result<()>;
34
35    /// Persists `sample` durably, enforcing the topic contract. Returns
36    /// [`StoreError::OutOfResources`](crate::StoreError) when a `KEEP_ALL` cap
37    /// is hit.
38    ///
39    /// # Errors
40    /// Contract cap exceeded, or a backend/I/O failure.
41    fn store(&self, sample: DurabilitySample) -> Result<()>;
42
43    /// Returns one ordered [`Page`] of samples for `topic` matching `selector`.
44    /// Ordering is `(instance_key, sequence)`. Drive pagination via
45    /// [`Page::next`] + [`Selector::after_cursor`].
46    ///
47    /// # Errors
48    /// Backend/I/O failure.
49    fn query(&self, topic: &str, selector: &Selector) -> Result<Page>;
50
51    /// Marks an instance as unregistered at `now`; its samples become eligible
52    /// for purge after `now + cleanup_delay` (applied by [`cleanup`]).
53    ///
54    /// [`cleanup`]: DurabilityStore::cleanup
55    ///
56    /// # Errors
57    /// Backend/I/O failure.
58    fn unregister(&self, topic: &str, instance_key: &[u8; 16], now: SystemTime) -> Result<()>;
59
60    /// Purges instances whose `unregister` time + cleanup-delay is `<= now`.
61    /// Returns the number of purged instances.
62    ///
63    /// # Errors
64    /// Backend/I/O failure.
65    fn cleanup(&self, now: SystemTime) -> Result<usize>;
66
67    /// Aggregate counters for `topic`.
68    ///
69    /// # Errors
70    /// Backend/I/O failure.
71    fn stats(&self, topic: &str) -> Result<StoreStats>;
72
73    /// Convenience: drains the full contracted history for `topic` by paging
74    /// through [`query`](DurabilityStore::query). Prefer paged `query` for
75    /// large contracts; this materializes everything into a `Vec`.
76    ///
77    /// # Errors
78    /// Backend/I/O failure from the underlying `query`.
79    fn replay_for_topic(&self, topic: &str) -> Result<Vec<DurabilitySample>> {
80        let mut out = Vec::new();
81        let mut sel = Selector {
82            limit: Some(DEFAULT_PAGE),
83            ..Selector::default()
84        };
85        loop {
86            let page = self.query(topic, &sel)?;
87            let got = page.samples.len();
88            out.extend(page.samples);
89            match page.next {
90                Some(cursor) if got > 0 => sel = sel.after_cursor(cursor),
91                _ => break,
92            }
93        }
94        Ok(out)
95    }
96}
zerodds_durability_store/store.rs

zerodds_durability_store/
store.rs
