commonware_storage/qmdb/immutable/sync/
mod.rs

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

commonware_storage/qmdb/immutable/sync/
mod.rs
