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};
13use crate::topic_meta::TopicMeta;
14
15/// Default page size for the [`DurabilityStore::replay_for_topic`] convenience
16/// drain when a selector does not set its own `limit`.
17pub const DEFAULT_PAGE: usize = 1024;
18
19/// A durable, contract-bounded sample store. Implementations are the cold
20/// adapters (sqlite / file / lakehouse / in-memory). All methods are
21/// `&self` + `Send + Sync`: the daemon shares one store across threads.
22///
23/// Retention is bounded ONLY by the topic's `Contract` (ADR 0009 inv. 1):
24/// implementations enforce it on [`store`](DurabilityStore::store) (drop-oldest
25/// under `KEEP_LAST`, [`StoreError::OutOfResources`](crate::StoreError) under
26/// `KEEP_ALL` caps).
27pub trait DurabilityStore: Send + Sync {
28    /// Registers (or replaces) the retention [`Contract`] for `topic`. The
29    /// daemon calls this when it starts serving a topic. Topics without an
30    /// explicit contract use the store's default ([`Contract::default`]).
31    ///
32    /// Contract-only: it does not persist a type name or keyedness, so a
33    /// topic that only ever saw `set_contract` has no [`TopicMeta`]
34    /// ([`topic_meta`](Self::topic_meta) returns `None` for it) — use
35    /// [`set_topic_meta`](Self::set_topic_meta) when serving a topic needs to
36    /// survive a restart without a writer/reader being re-discovered first.
37    /// When both were ever called for the same topic, `set_topic_meta`'s
38    /// contract wins (it is the more recent, more complete registration).
39    ///
40    /// # Errors
41    /// Backend/I/O failure.
42    fn set_contract(&self, topic: &str, contract: Contract) -> Result<()>;
43
44    /// Persists what the durability service (or any other caller) needs to
45    /// resume serving `topic` after a restart, with no writer or reader
46    /// re-discovered first: its type name, keyedness, and the retention
47    /// contract in force (see [`TopicMeta`]). Called once, when a topic is
48    /// first served. Also becomes the contract
49    /// [`store`](Self::store)/[`cleanup`](Self::cleanup) enforce for `topic`
50    /// (superseding a plain [`set_contract`](Self::set_contract) for the same
51    /// topic — see its doc).
52    ///
53    /// # Errors
54    /// Backend/I/O failure.
55    fn set_topic_meta(&self, topic: &str, meta: TopicMeta) -> Result<()>;
56
57    /// The persisted [`TopicMeta`] for `topic`, if
58    /// [`set_topic_meta`](Self::set_topic_meta) was ever called for it.
59    /// `None` for a topic that only ever saw
60    /// [`set_contract`](Self::set_contract) (contract-only, no served
61    /// type/keyedness known) or was never registered at all.
62    ///
63    /// # Errors
64    /// Backend/I/O failure.
65    fn topic_meta(&self, topic: &str) -> Result<Option<TopicMeta>>;
66
67    /// Persists `sample` durably, enforcing the topic contract. Returns
68    /// [`StoreError::OutOfResources`](crate::StoreError) when a `KEEP_ALL` cap
69    /// is hit.
70    ///
71    /// # Errors
72    /// Contract cap exceeded, or a backend/I/O failure.
73    fn store(&self, sample: DurabilitySample) -> Result<()>;
74
75    /// Returns one ordered [`Page`] of samples for `topic` matching `selector`.
76    /// Ordering is `(instance_key, sequence)`. Drive pagination via
77    /// [`Page::next`] + [`Selector::after_cursor`].
78    ///
79    /// # Errors
80    /// Backend/I/O failure.
81    fn query(&self, topic: &str, selector: &Selector) -> Result<Page>;
82
83    /// Marks an instance as unregistered at `now`; its samples become eligible
84    /// for purge after `now + cleanup_delay` (applied by [`cleanup`]).
85    ///
86    /// [`cleanup`]: DurabilityStore::cleanup
87    ///
88    /// # Errors
89    /// Backend/I/O failure.
90    fn unregister(&self, topic: &str, instance_key: &[u8; 16], now: SystemTime) -> Result<()>;
91
92    /// Purges instances whose `unregister` time + cleanup-delay is `<= now`.
93    /// Returns the number of purged instances.
94    ///
95    /// # Errors
96    /// Backend/I/O failure.
97    fn cleanup(&self, now: SystemTime) -> Result<usize>;
98
99    /// Aggregate counters for `topic`.
100    ///
101    /// # Errors
102    /// Backend/I/O failure.
103    fn stats(&self, topic: &str) -> Result<StoreStats>;
104
105    /// Convenience: drains the full contracted history for `topic` by paging
106    /// through [`query`](DurabilityStore::query). Prefer paged `query` for
107    /// large contracts; this materializes everything into a `Vec`.
108    ///
109    /// # Errors
110    /// Backend/I/O failure from the underlying `query`.
111    fn replay_for_topic(&self, topic: &str) -> Result<Vec<DurabilitySample>> {
112        let mut out = Vec::new();
113        let mut sel = Selector {
114            limit: Some(DEFAULT_PAGE),
115            ..Selector::default()
116        };
117        loop {
118            let page = self.query(topic, &sel)?;
119            let got = page.samples.len();
120            out.extend(page.samples);
121            match page.next {
122                Some(cursor) if got > 0 => sel = sel.after_cursor(cursor),
123                _ => break,
124            }
125        }
126        Ok(out)
127    }
128}
zerodds_durability_store/store.rs

zerodds_durability_store/
store.rs
