commonware_storage/qmdb/immutable/sync/
mod.rs

1use crate::{
2    Context,
3    index::unordered::Index,
4    journal::{authenticated, contiguous::Mutable},
5    merkle::{
6        Family, Location,
7        full::{self, Merkle},
8    },
9    qmdb::{
10        self, Error,
11        any::ValueEncoding,
12        immutable::{self, CompactDb, Metrics, Operation},
13        operation::Key,
14        sync,
15    },
16    translator::Translator,
17};
18use commonware_codec::{EncodeShared, Read};
19use commonware_cryptography::Hasher;
20use commonware_parallel::Strategy;
21use commonware_utils::range::NonEmptyRange;
22use std::num::NonZeroU64;
23
24impl<F, E, K, V, C, H, T, S> sync::Database for immutable::Immutable<F, E, K, V, C, H, T, S>
25where
26    F: Family,
27    E: Context,
28    K: Key,
29    V: ValueEncoding,
30    C: Mutable<Item = Operation<F, K, V>> + sync::Journal<F, Context = E, Op = Operation<F, K, V>>,
31    C::Item: EncodeShared,
32    C::Config: Clone + Send,
33    H: Hasher,
34    T: Translator,
35    S: Strategy,
36{
37    type Family = F;
38    type Op = Operation<F, K, V>;
39    type Journal = C;
40    type Hasher = H;
41    type Config = immutable::Config<T, C::Config, S>;
42    type Digest = H::Digest;
43    type Context = E;
44
45    /// Returns an [Immutable](immutable::Immutable) initialized from data collected in the sync process.
46    ///
47    /// # Behavior
48    ///
49    /// This method handles different initialization scenarios based on existing data:
50    /// - If the Merkle journal is empty or the last item is before the range start, it creates a
51    ///   fresh Merkle structure from the provided `pinned_nodes`
52    /// - If the Merkle journal has data but is incomplete (has length < range end), missing
53    ///   operations from the log are applied to bring it up to the target state
54    /// - If the Merkle journal has data beyond the range end, it is rewound to match the sync
55    ///   target
56    ///
57    /// # Returns
58    ///
59    /// A [super::Immutable] db populated with the state from the given range.
60    /// The pruning boundary is set to the range start.
61    async fn from_sync_result(
62        context: Self::Context,
63        db_config: Self::Config,
64        log: Self::Journal,
65        pinned_nodes: Option<Vec<Self::Digest>>,
66        range: NonEmptyRange<Location<F>>,
67        apply_batch_size: NonZeroU64,
68    ) -> Result<Self, Error<F>> {
69        let hasher = qmdb::hasher::<H>();
70
71        // Initialize Merkle structure for sync
72        let merkle = Merkle::<F, _, _, S>::init_sync(
73            context.child("merkle"),
74            full::SyncConfig {
75                config: db_config.merkle_config.clone(),
76                range: range.clone(),
77                pinned_nodes,
78            },
79        )
80        .await?;
81
82        let journal = authenticated::Journal::<_, _, _, _, S>::from_components(
83            merkle,
84            log,
85            hasher,
86            apply_batch_size.get(),
87        )
88        .await?;
89
90        let mut snapshot: Index<T, Location<F>> =
91            Index::new(context.child("snapshot"), db_config.translator.clone());
92
93        let (last_commit_loc, inactivity_floor_loc) = {
94            let bounds = journal.journal.bounds();
95            let last_commit_loc = Location::<F>::new(
96                bounds
97                    .end
98                    .checked_sub(1)
99                    .ok_or(Error::HistoricalFloorPruned(Location::new(bounds.end)))?,
100            );
101            let inactivity_floor_loc = crate::qmdb::find_inactivity_floor_at::<F, _>(
102                &journal.journal,
103                Location::new(bounds.end),
104            )
105            .await?;
106
107            // Replay the log from the inactivity floor to build the snapshot. Every retained
108            // location is inserted, mirroring the live apply path, so a repeated key keeps
109            // serving one of its written values across restarts and rewinds.
110            immutable::build_snapshot(
111                inactivity_floor_loc,
112                &journal.journal,
113                &mut snapshot,
114                db_config.init_buffer,
115            )
116            .await?;
117
118            (last_commit_loc, inactivity_floor_loc)
119        };
120        let inactive_peaks = F::inactive_peaks(last_commit_loc + 1, inactivity_floor_loc);
121        let root = journal.root(inactive_peaks)?;
122
123        let metrics = Metrics::new(context);
124        let db = Self {
125            journal,
126            root,
127            snapshot,
128            last_commit_loc,
129            inactivity_floor_loc,
130            metrics,
131        };
132        db.update_metrics();
133
134        db.sync().await
135    }
136
137    async fn persist_sync_result(self) -> Result<Self, Error<F>> {
138        Ok(self)
139    }
140
141    async fn local_pinned_nodes(
142        context: Self::Context,
143        config: &Self::Config,
144        target: &sync::Target<F, Self::Digest>,
145        journal: &Self::Journal,
146    ) -> Result<Option<Vec<Self::Digest>>, Error<F>> {
147        if target.range.start() == Location::new(0)
148            || !sync::journal_covers_range(journal.bounds(), &target.range)
149        {
150            return Ok(None);
151        }
152
153        // The inactivity floor is carried by the last commit operation rather than being
154        // the target range's start.
155        let inactivity_floor =
156            qmdb::find_inactivity_floor_at::<F, _>(journal, target.range.end()).await?;
157
158        sync::local_pinned_nodes::<F, _, H, S>(
159            context,
160            config.merkle_config.clone(),
161            target,
162            inactivity_floor,
163        )
164        .await
165    }
166
167    fn root(&self) -> Self::Digest {
168        self.root()
169    }
170}
171
172impl<F, E, K, V, H, Cfg, S> sync::Database for CompactDb<F, E, K, V, H, Cfg, S>
173where
174    F: Family,
175    E: Context,
176    K: Key,
177    V: ValueEncoding,
178    H: Hasher,
179    S: Strategy,
180    Operation<F, K, V>: EncodeShared,
181    Operation<F, K, V>: Read<Cfg = Cfg>,
182    Cfg: Clone + Send + Sync + 'static,
183{
184    type Family = F;
185    type Op = Operation<F, K, V>;
186    type Journal = sync::journal::Memory<F, E, Operation<F, K, V>>;
187    type Config = immutable::CompactConfig<Cfg, S>;
188    type Digest = H::Digest;
189    type Context = E;
190    type Hasher = H;
191
192    async fn from_sync_result(
193        context: Self::Context,
194        config: Self::Config,
195        log: Self::Journal,
196        pinned_nodes: Option<Vec<Self::Digest>>,
197        range: NonEmptyRange<Location<F>>,
198        _apply_batch_size: NonZeroU64,
199    ) -> Result<Self, Error<F>> {
200        crate::qmdb::compact::from_sync_result(
201            context,
202            config,
203            log,
204            pinned_nodes,
205            range,
206            Self::init_from_sync,
207        )
208        .await
209    }
210
211    async fn persist_sync_result(self) -> Result<Self, Error<F>> {
212        self.sync().await
213    }
214
215    async fn local_pinned_nodes(
216        _context: Self::Context,
217        _config: &Self::Config,
218        _target: &sync::Target<F, Self::Digest>,
219        _journal: &Self::Journal,
220    ) -> Result<Option<Vec<Self::Digest>>, Error<F>> {
221        Ok(None)
222    }
223
224    fn root(&self) -> Self::Digest {
225        self.root()
226    }
227}
228
229#[cfg(test)]
230mod tests;
commonware_storage/qmdb/immutable/sync/mod.rs

commonware_storage/qmdb/immutable/sync/
mod.rs
