commonware_storage/qmdb/sync/
database.rs

1use crate::{
2    Context,
3    merkle::{Family, Location, full},
4    qmdb::sync::{Journal, Target},
5    translator::Translator,
6};
7use commonware_cryptography::{Digest, Hasher};
8use commonware_parallel::Strategy;
9use commonware_utils::range::NonEmptyRange;
10use std::{future::Future, num::NonZeroU64};
11
12/// Database configuration that can produce the configuration for its sync journal.
13pub trait Config {
14    type JournalConfig;
15    fn journal_config(&self) -> Self::JournalConfig;
16}
17
18impl<T: Translator, J: Clone, S: Strategy> Config for crate::qmdb::any::Config<T, J, S> {
19    type JournalConfig = J;
20
21    fn journal_config(&self) -> Self::JournalConfig {
22        self.journal_config.clone()
23    }
24}
25
26impl<T: Translator, C: Clone, S: Strategy> Config for crate::qmdb::immutable::Config<T, C, S> {
27    type JournalConfig = C;
28
29    fn journal_config(&self) -> Self::JournalConfig {
30        self.log.clone()
31    }
32}
33
34impl<J: Clone, S: Strategy> Config for crate::qmdb::keyless::Config<J, S> {
35    type JournalConfig = J;
36
37    fn journal_config(&self) -> Self::JournalConfig {
38        self.log.clone()
39    }
40}
41
42impl<C: Clone + Send + Sync + 'static, S: Strategy> Config for crate::qmdb::compact::Config<C, S> {
43    type JournalConfig = ();
44
45    fn journal_config(&self) -> Self::JournalConfig {}
46}
47
48pub trait Database: Sized + Send {
49    type Family: Family;
50    type Op: Send + Sync;
51    type Journal: Journal<Self::Family, Context = Self::Context, Op = Self::Op>;
52    type Config: Config<JournalConfig = <Self::Journal as Journal<Self::Family>>::Config>;
53    type Digest: Digest;
54    type Context: commonware_runtime::Storage
55        + commonware_runtime::Clock
56        + commonware_runtime::Metrics;
57    type Hasher: commonware_cryptography::Hasher<Digest = Self::Digest>;
58
59    /// Build a database from the journal and pinned nodes populated by the sync engine.
60    fn from_sync_result(
61        context: Self::Context,
62        config: Self::Config,
63        journal: Self::Journal,
64        pinned_nodes: Option<Vec<Self::Digest>>,
65        range: NonEmptyRange<Location<Self::Family>>,
66        apply_batch_size: NonZeroU64,
67    ) -> impl Future<Output = Result<Self, crate::qmdb::Error<Self::Family>>> + Send;
68
69    /// Persist any state that must remain provisional until the engine verifies the rebuilt root.
70    ///
71    /// The engine calls this only after [`Self::root`] matches the requested target. Implementations
72    /// that persist everything in [`Self::from_sync_result`] must explicitly return `Ok(self)`.
73    fn persist_sync_result(
74        self,
75    ) -> impl Future<Output = Result<Self, crate::qmdb::Error<Self::Family>>> + Send;
76
77    /// Return locally available pinned nodes for the target, if persisted local state can
78    /// authenticate them.
79    ///
80    /// Returning `Some` lets a completed sync journal reuse pinned nodes from an on-disk
81    /// database instead of fetching them from peers. Returning `None` always falls back to
82    /// fetching from peers.
83    fn local_pinned_nodes(
84        context: Self::Context,
85        config: &Self::Config,
86        target: &crate::qmdb::sync::Target<Self::Family, Self::Digest>,
87        journal: &Self::Journal,
88    ) -> impl Future<Output = Result<Option<Vec<Self::Digest>>, crate::qmdb::Error<Self::Family>>> + Send;
89
90    /// Get the root digest of the database for verification
91    fn root(&self) -> Self::Digest;
92}
93
94/// Whether a completed sync journal's `bounds` cover `range`: retained data reaches back to
95/// `range.start()` and ends exactly at `range.end()`.
96pub(crate) fn journal_covers_range<F: Family>(
97    bounds: std::ops::Range<u64>,
98    range: &NonEmptyRange<Location<F>>,
99) -> bool {
100    Location::new(bounds.start) <= range.start() && Location::new(bounds.end) == range.end()
101}
102
103/// Shared body for [`Database::local_pinned_nodes`] implementations backed by a persisted
104/// [`full::Merkle`]. Reopens it from `config` under `context` and returns the pinned nodes at
105/// `target.range.start()` if the persisted bounds cover the target and the root, computed with
106/// `inactivity_floor`, matches `target.root`. Returns `Ok(None)` when the persisted state
107/// cannot authenticate the target, including when an interrupted reset left required local
108/// nodes unavailable.
109pub(crate) async fn local_pinned_nodes<F, E, H, S>(
110    context: E,
111    config: full::Config<S>,
112    target: &Target<F, H::Digest>,
113    inactivity_floor: Location<F>,
114) -> Result<Option<Vec<H::Digest>>, crate::qmdb::Error<F>>
115where
116    F: Family,
117    E: Context,
118    H: Hasher,
119    S: Strategy,
120{
121    let hasher = crate::qmdb::hasher::<H>();
122
123    // A crash can persist a node-journal reset before its replacement metadata.
124    // Missing local pins then use a peer-authenticated boundary. Other errors still propagate.
125    let merkle = match full::Merkle::<F, _, _, S>::init(context, &hasher, config).await {
126        Ok(merkle) => merkle,
127        Err(crate::merkle::Error::MissingNode(_)) => return Ok(None),
128        Err(err) => return Err(err.into()),
129    };
130    let bounds = merkle.bounds();
131    if bounds.start > target.range.start() || bounds.end != target.range.end() {
132        return Ok(None);
133    }
134
135    let inactive_peaks = F::inactive_peaks(target.range.end(), inactivity_floor);
136    if merkle.root(&hasher, inactive_peaks)? != target.root {
137        return Ok(None);
138    }
139
140    merkle
141        .pinned_nodes_at(target.range.start())
142        .await
143        .map(Some)
144        .map_err(Into::into)
145}
146
147#[cfg(test)]
148mod tests {
149    use super::{journal_covers_range, local_pinned_nodes};
150    use crate::{
151        journal::contiguous::fixed,
152        merkle::{Location, Position, full, mmr::Family as MmrFamily},
153        qmdb::sync::Target,
154    };
155    use commonware_cryptography::{Sha256, sha256::Digest};
156    use commonware_parallel::Sequential;
157    use commonware_runtime::{
158        BufferPooler, Runner as _, Supervisor as _, buffer::paged::CacheRef, deterministic,
159    };
160    use commonware_utils::{NZU16, NZU64, NZUsize, non_empty_range, range::NonEmptyRange};
161
162    fn merkle_config(pooler: &impl BufferPooler) -> full::Config<Sequential> {
163        full::Config {
164            journal_partition: "local-pins-journal".into(),
165            metadata_partition: "local-pins-metadata".into(),
166            items_per_blob: NZU64!(7),
167            write_buffer: NZUsize!(1024),
168            replay_buffer: NZUsize!(1024),
169            strategy: Sequential,
170            page_cache: CacheRef::from_pooler(pooler, NZU16!(111), NZUsize!(5)),
171        }
172    }
173
174    #[test]
175    fn test_journal_covers_range() {
176        let range: NonEmptyRange<Location<MmrFamily>> =
177            non_empty_range!(Location::new(10), Location::new(20));
178
179        // Bounds reaching at least back to the start and ending exactly at the end cover.
180        assert!(journal_covers_range(10..20, &range));
181        assert!(journal_covers_range(5..20, &range));
182
183        // Bounds starting after the range start do not cover.
184        assert!(!journal_covers_range(11..20, &range));
185
186        // Bounds ending anywhere but exactly at the range end do not cover.
187        assert!(!journal_covers_range(10..19, &range));
188        assert!(!journal_covers_range(10..21, &range));
189        assert!(!journal_covers_range(0..0, &range));
190    }
191
192    #[test]
193    fn local_pinned_nodes_treats_interrupted_reset_as_unavailable() {
194        deterministic::Runner::default().start(|context| async move {
195            let hasher = crate::qmdb::hasher::<Sha256>();
196            let config = merkle_config(&context);
197            let mut merkle = full::Merkle::<MmrFamily, _, Digest, Sequential>::init(
198                context.child("init"),
199                &hasher,
200                config.clone(),
201            )
202            .await
203            .unwrap();
204            let mut batch = merkle.new_batch();
205            for i in 0u64..50 {
206                batch = batch.add(&hasher, &i.to_be_bytes());
207            }
208            let batch = merkle.with_mem(|mem| batch.merkleize(mem, &hasher));
209            merkle = merkle.apply_batch(&batch).unwrap();
210            let merkle = merkle.sync().await.unwrap();
211            let merkle = merkle.prune(Location::new(30)).await.unwrap();
212            let merkle = merkle.sync().await.unwrap();
213            drop(merkle);
214
215            // Model a crash after an incompatible reset durably cleared the node journal but
216            // before it replaced the prior target's pinned metadata.
217            let restart = Location::new(7);
218            let journal_config = fixed::Config {
219                partition: config.journal_partition.clone(),
220                items_per_blob: config.items_per_blob,
221                page_cache: config.page_cache.clone(),
222                write_buffer: config.write_buffer,
223                replay_buffer: config.replay_buffer,
224            };
225            let journal = fixed::Journal::<_, Digest>::init(
226                context.child("interrupted_reset"),
227                journal_config,
228            )
229            .await
230            .unwrap();
231            let reset_pos = Position::<MmrFamily>::try_from(restart).unwrap();
232            let journal = journal.clear_to_size(*reset_pos).await.unwrap();
233            drop(journal);
234
235            let target = Target {
236                root: Digest::from([0; 32]),
237                range: non_empty_range!(restart, Location::new(20)),
238            };
239            let pinned = local_pinned_nodes::<MmrFamily, _, Sha256, Sequential>(
240                context.child("local_pins"),
241                config,
242                &target,
243                restart,
244            )
245            .await
246            .unwrap();
247            assert!(pinned.is_none());
248        });
249    }
250}
commonware_storage/qmdb/sync/database.rs

commonware_storage/qmdb/sync/
database.rs
