typed_store/
database.rs

1// Copyright (c) 2026 IOTA Stiftung
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{
5    borrow::Borrow,
6    marker::PhantomData,
7    ops::{Bound, Deref, RangeBounds},
8    path::Path,
9    sync::Arc,
10    time::Duration,
11};
12
13use fastcrypto::hash::{Digest, HashFunction};
14use iota_common::debug_fatal;
15use iota_macros::{fail_point, nondeterministic};
16use prometheus_filtered::{Histogram, HistogramTimer};
17use rocksdb::{DBPinnableSlice, Error, LiveFile, ReadOptions, WriteBatch, checkpoint::Checkpoint};
18use serde::{Serialize, de::DeserializeOwned};
19use tokio::sync::oneshot;
20use tracing::{debug, error, instrument, warn};
21use typed_store_error::TypedStoreError;
22
23use crate::{
24    DbIterator,
25    memstore::{InMemoryBatch, InMemoryDB},
26    metrics::{DBMetrics, RocksDBPerfContext, SamplingInterval},
27    rocks::{
28        RocksDB,
29        errors::{typed_store_err_from_bcs_err, typed_store_err_from_rocks_err},
30        options::ReadWriteOptions,
31        rocks_cf, rocks_util,
32        safe_iter::{SafeIter, SafeRevIter},
33    },
34    traits::{Map, TableSummary},
35    util::{be_fix_int_ser, iterator_bounds_with_range, prefix_iterator_bounds},
36};
37
38#[derive(Clone)]
39pub(crate) enum ColumnFamily {
40    Rocks(String),
41    #[allow(dead_code)]
42    InMemory(String),
43}
44
45impl std::fmt::Debug for ColumnFamily {
46    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47        match self {
48            ColumnFamily::Rocks(name) => write!(f, "RocksDB cf: {name}"),
49            ColumnFamily::InMemory(name) => write!(f, "InMemory cf: {name}"),
50        }
51    }
52}
53
54impl ColumnFamily {
55    pub(crate) fn name(&self) -> &str {
56        match self {
57            ColumnFamily::Rocks(name) => name,
58            ColumnFamily::InMemory(name) => name,
59        }
60    }
61
62    pub(crate) fn rocks_cf<'a>(
63        &self,
64        rocks_db: &'a RocksDB,
65    ) -> Arc<rocksdb::BoundColumnFamily<'a>> {
66        match &self {
67            ColumnFamily::Rocks(name) => rocks_db
68                .underlying
69                .cf_handle(name)
70                .expect("Map-keying column family should have been checked at DB creation"),
71            _ => unreachable!("invariant is checked by the caller"),
72        }
73    }
74}
75
76pub(crate) enum Storage {
77    Rocks(RocksDB),
78    #[allow(dead_code)]
79    InMemory(InMemoryDB),
80}
81
82impl std::fmt::Debug for Storage {
83    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
84        match self {
85            Storage::Rocks(db) => write!(f, "RocksDB Storage {db:?}"),
86            Storage::InMemory(db) => write!(f, "InMemoryDB Storage {db:?}"),
87        }
88    }
89}
90
91pub(crate) enum GetResult<'a> {
92    Rocks(DBPinnableSlice<'a>),
93    InMemory(Vec<u8>),
94}
95
96impl Deref for GetResult<'_> {
97    type Target = [u8];
98    fn deref(&self) -> &[u8] {
99        match self {
100            GetResult::Rocks(d) => d.deref(),
101            GetResult::InMemory(d) => d.deref(),
102        }
103    }
104}
105
106pub enum StorageWriteBatch {
107    Rocks(rocksdb::WriteBatch),
108    InMemory(InMemoryBatch),
109}
110
111#[derive(Debug, Default)]
112pub struct MetricConf {
113    pub db_name: String,
114    pub read_sample_interval: SamplingInterval,
115    pub write_sample_interval: SamplingInterval,
116    pub iter_sample_interval: SamplingInterval,
117}
118
119impl MetricConf {
120    pub fn new(db_name: &str) -> Self {
121        if db_name.is_empty() {
122            error!("A meaningful db name should be used for metrics reporting.")
123        }
124        Self {
125            db_name: db_name.to_string(),
126            read_sample_interval: SamplingInterval::default(),
127            write_sample_interval: SamplingInterval::default(),
128            iter_sample_interval: SamplingInterval::default(),
129        }
130    }
131
132    pub fn with_sampling(self, read_interval: SamplingInterval) -> Self {
133        Self {
134            db_name: self.db_name,
135            read_sample_interval: read_interval,
136            write_sample_interval: SamplingInterval::default(),
137            iter_sample_interval: SamplingInterval::default(),
138        }
139    }
140}
141
142const CF_METRICS_REPORT_PERIOD_SECS: u64 = 30;
143
144#[derive(Debug)]
145pub struct Database {
146    pub(crate) storage: Storage,
147    pub(crate) metric_conf: MetricConf,
148}
149
150impl Drop for Database {
151    fn drop(&mut self) {
152        DBMetrics::get().decrement_num_active_dbs(&self.metric_conf.db_name);
153    }
154}
155
156impl Database {
157    pub(crate) fn new(storage: Storage, metric_conf: MetricConf) -> Self {
158        DBMetrics::get().increment_num_active_dbs(&metric_conf.db_name);
159        Self {
160            storage,
161            metric_conf,
162        }
163    }
164
165    pub(crate) fn get<K: AsRef<[u8]>>(
166        &self,
167        cf: &ColumnFamily,
168        key: K,
169        readopts: &ReadOptions,
170    ) -> Result<Option<GetResult<'_>>, TypedStoreError> {
171        match (&self.storage, cf) {
172            (Storage::Rocks(db), ColumnFamily::Rocks(_)) => Ok(db
173                .underlying
174                .get_pinned_cf_opt(&cf.rocks_cf(db), key, readopts)
175                .map_err(typed_store_err_from_rocks_err)?
176                .map(GetResult::Rocks)),
177            (Storage::InMemory(db), ColumnFamily::InMemory(cf_name)) => {
178                Ok(db.get(cf_name, key).map(GetResult::InMemory))
179            }
180
181            _ => Err(TypedStoreError::RocksDB(
182                "typed store invariant violation".to_string(),
183            )),
184        }
185    }
186
187    pub(crate) fn multi_get<I, K>(
188        &self,
189        cf: &ColumnFamily,
190        keys: I,
191        readopts: &ReadOptions,
192    ) -> Vec<Result<Option<GetResult<'_>>, TypedStoreError>>
193    where
194        I: IntoIterator<Item = K>,
195        K: AsRef<[u8]>,
196    {
197        match (&self.storage, cf) {
198            (Storage::Rocks(db), ColumnFamily::Rocks(_)) => {
199                let keys_vec: Vec<K> = keys.into_iter().collect();
200                let res = db.underlying.batched_multi_get_cf_opt(
201                    &cf.rocks_cf(db),
202                    keys_vec.iter(),
203                    // sorted_input
204                    false,
205                    readopts,
206                );
207                res.into_iter()
208                    .map(|r| {
209                        r.map_err(typed_store_err_from_rocks_err)
210                            .map(|item| item.map(GetResult::Rocks))
211                    })
212                    .collect()
213            }
214            (Storage::InMemory(db), ColumnFamily::InMemory(cf_name)) => db
215                .multi_get(cf_name, keys)
216                .into_iter()
217                .map(|r| Ok(r.map(GetResult::InMemory)))
218                .collect(),
219            _ => unreachable!("typed store invariant violation"),
220        }
221    }
222
223    pub fn cf_handle(&self, name: &str) -> Option<()> {
224        match &self.storage {
225            Storage::Rocks(db) => db.underlying.cf_handle(name).map(|_| ()),
226            Storage::InMemory(db) => db.has_cf(name).then_some(()),
227        }
228    }
229
230    pub fn drop_cf(&self, name: &str) -> Result<(), rocksdb::Error> {
231        match &self.storage {
232            Storage::Rocks(db) => db.underlying.drop_cf(name),
233            Storage::InMemory(db) => {
234                db.drop_cf(name);
235                Ok(())
236            }
237        }
238    }
239
240    pub(crate) fn delete_cf<K: AsRef<[u8]>>(
241        &self,
242        cf: &ColumnFamily,
243        key: K,
244    ) -> Result<(), TypedStoreError> {
245        fail_point!("delete-cf-before");
246        let ret = match (&self.storage, cf) {
247            (Storage::Rocks(db), ColumnFamily::Rocks(_)) => db
248                .underlying
249                .delete_cf(&cf.rocks_cf(db), key)
250                .map_err(typed_store_err_from_rocks_err),
251            (Storage::InMemory(db), ColumnFamily::InMemory(cf_name)) => {
252                db.delete(cf_name, key.as_ref());
253                Ok(())
254            }
255            _ => Err(TypedStoreError::RocksDB(
256                "typed store invariant violation".to_string(),
257            )),
258        };
259        fail_point!("delete-cf-after");
260        #[allow(clippy::let_and_return)]
261        ret
262    }
263
264    pub fn path_for_pruning(&self) -> &Path {
265        match &self.storage {
266            Storage::Rocks(rocks) => rocks.underlying.path(),
267            _ => unimplemented!("method is only supported for rocksdb backend"),
268        }
269    }
270
271    pub(crate) fn put_cf(
272        &self,
273        cf: &ColumnFamily,
274        key: Vec<u8>,
275        value: Vec<u8>,
276    ) -> Result<(), TypedStoreError> {
277        fail_point!("put-cf-before");
278        let ret = match (&self.storage, cf) {
279            (Storage::Rocks(db), ColumnFamily::Rocks(_)) => db
280                .underlying
281                .put_cf(&cf.rocks_cf(db), key, value)
282                .map_err(typed_store_err_from_rocks_err),
283            (Storage::InMemory(db), ColumnFamily::InMemory(cf_name)) => {
284                db.put(cf_name, key, value);
285                Ok(())
286            }
287            _ => Err(TypedStoreError::RocksDB(
288                "typed store invariant violation".to_string(),
289            )),
290        };
291        fail_point!("put-cf-after");
292        #[allow(clippy::let_and_return)]
293        ret
294    }
295
296    pub(crate) fn key_may_exist_cf<K: AsRef<[u8]>>(
297        &self,
298        cf: &ColumnFamily,
299        key: K,
300        readopts: &ReadOptions,
301    ) -> bool {
302        match &self.storage {
303            // [`rocksdb::DBWithThreadMode::key_may_exist_cf`] can have false positives,
304            // but no false negatives. We use it to short-circuit the absent case
305            Storage::Rocks(rocks) => {
306                rocks
307                    .underlying
308                    .key_may_exist_cf_opt(&rocks_cf(rocks, cf.name()), key, readopts)
309            }
310            _ => true,
311        }
312    }
313
314    /// Flush the memtable of a single column family to SST files on disk.
315    pub(crate) fn flush_cf(&self, cf: &ColumnFamily) -> Result<(), TypedStoreError> {
316        match &self.storage {
317            // A flush blocks on RocksDB background threads; under the simulator it
318            // must run off the test thread, like the opens in `rocks/mod.rs`, or
319            // real flush durations leak into the simulated schedule. No-op outside
320            // msim.
321            Storage::Rocks(rocks) => nondeterministic!({
322                let cf_name = cf.name();
323                if let Some(handle) = rocks.underlying.cf_handle(cf_name) {
324                    rocks.underlying.flush_cf(&handle).map_err(|e| {
325                        TypedStoreError::RocksDB(format!(
326                            "Failed to flush column family {cf_name}: {e}"
327                        ))
328                    })?;
329                }
330                Ok(())
331            }),
332            // InMemory databases don't need flushing.
333            Storage::InMemory(_) => Ok(()),
334        }
335    }
336
337    /// Iterate all column families and flush the memtables of every column
338    /// family to SST files on disk.
339    pub fn flush_all(&self) -> Result<(), TypedStoreError> {
340        match &self.storage {
341            // See `flush_cf` for why the flushes run off the test thread under
342            // the simulator.
343            Storage::Rocks(rocks) => nondeterministic!({
344                for cf_name in &rocks.cf_names {
345                    if let Some(cf) = rocks.underlying.cf_handle(cf_name) {
346                        rocks.underlying.flush_cf(&cf).map_err(|e| {
347                            TypedStoreError::RocksDB(format!(
348                                "Failed to flush column family {cf_name}: {e}"
349                            ))
350                        })?;
351                    }
352                }
353                Ok(())
354            }),
355            // InMemory databases don't need flushing.
356            Storage::InMemory(_) => Ok(()),
357        }
358    }
359
360    pub fn write(&self, batch: StorageWriteBatch) -> Result<(), TypedStoreError> {
361        self.write_opt(batch, &rocksdb::WriteOptions::default())
362    }
363
364    pub fn write_opt(
365        &self,
366        batch: StorageWriteBatch,
367        write_options: &rocksdb::WriteOptions,
368    ) -> Result<(), TypedStoreError> {
369        fail_point!("batch-write-before");
370        let ret = match (&self.storage, batch) {
371            (Storage::Rocks(rocks), StorageWriteBatch::Rocks(batch)) => rocks
372                .underlying
373                .write_opt(batch, write_options)
374                .map_err(typed_store_err_from_rocks_err),
375            (Storage::InMemory(db), StorageWriteBatch::InMemory(batch)) => {
376                // InMemory doesn't support write options.
377                db.write(batch);
378                Ok(())
379            }
380            _ => Err(TypedStoreError::RocksDB(
381                "using invalid batch type for the database".to_string(),
382            )),
383        };
384        fail_point!("batch-write-after");
385
386        #[allow(clippy::let_and_return)]
387        ret
388    }
389
390    pub(crate) fn compact_range_cf<K: AsRef<[u8]>>(
391        &self,
392        cf: &ColumnFamily,
393        start: Option<K>,
394        end: Option<K>,
395    ) {
396        if let Storage::Rocks(rocksdb) = &self.storage {
397            rocksdb
398                .underlying
399                .compact_range_cf(&rocks_cf(rocksdb, cf.name()), start, end);
400        }
401    }
402
403    pub fn checkpoint(&self, path: &Path) -> Result<(), TypedStoreError> {
404        // TODO: implement for other storage types
405        if let Storage::Rocks(rocks) = &self.storage {
406            let checkpoint =
407                Checkpoint::new(&rocks.underlying).map_err(typed_store_err_from_rocks_err)?;
408            checkpoint
409                .create_checkpoint(path)
410                .map_err(|e| TypedStoreError::RocksDB(e.to_string()))?;
411        }
412        Ok(())
413    }
414
415    pub fn get_sampling_interval(&self) -> SamplingInterval {
416        self.metric_conf.read_sample_interval.new_from_self()
417    }
418
419    pub fn multiget_sampling_interval(&self) -> SamplingInterval {
420        self.metric_conf.read_sample_interval.new_from_self()
421    }
422
423    pub fn write_sampling_interval(&self) -> SamplingInterval {
424        self.metric_conf.write_sample_interval.new_from_self()
425    }
426
427    pub fn iter_sampling_interval(&self) -> SamplingInterval {
428        self.metric_conf.iter_sample_interval.new_from_self()
429    }
430
431    pub(crate) fn db_name(&self) -> String {
432        let name = &self.metric_conf.db_name;
433        if name.is_empty() {
434            "default".to_string()
435        } else {
436            name.clone()
437        }
438    }
439
440    pub fn live_files(&self) -> Result<Vec<LiveFile>, Error> {
441        match &self.storage {
442            Storage::Rocks(rocks) => rocks.underlying.live_files(),
443            _ => Ok(vec![]),
444        }
445    }
446
447    pub(crate) fn try_catch_up_with_primary(&self) -> Result<(), TypedStoreError> {
448        if let Storage::Rocks(rocks) = &self.storage {
449            rocks
450                .underlying
451                .try_catch_up_with_primary()
452                .map_err(typed_store_err_from_rocks_err)?;
453        }
454        Ok(())
455    }
456}
457
458fn rocks_cf_from_db<'a>(
459    db: &'a Database,
460    cf_name: &str,
461) -> Result<Arc<rocksdb::BoundColumnFamily<'a>>, TypedStoreError> {
462    match &db.storage {
463        Storage::Rocks(rocksdb) => Ok(rocksdb
464            .underlying
465            .cf_handle(cf_name)
466            .expect("Map-keying column family should have been checked at DB creation")),
467        _ => Err(TypedStoreError::RocksDB(
468            "using invalid batch type for the database".to_string(),
469        )),
470    }
471}
472
473/// An interface to a rocksDB database, keyed by a columnfamily
474#[derive(Clone, Debug)]
475pub struct DBMap<K, V> {
476    pub db: Arc<Database>,
477    _phantom: PhantomData<fn(K) -> V>,
478    column_family: ColumnFamily,
479    pub opts: ReadWriteOptions,
480    db_metrics: Arc<DBMetrics>,
481    get_sample_interval: SamplingInterval,
482    multiget_sample_interval: SamplingInterval,
483    write_sample_interval: SamplingInterval,
484    iter_sample_interval: SamplingInterval,
485    _metrics_task_cancel_handle: Arc<oneshot::Sender<()>>,
486}
487
488unsafe impl<K: Send, V: Send> Send for DBMap<K, V> {}
489
490impl<K, V> DBMap<K, V> {
491    pub(crate) fn new(
492        db: Arc<Database>,
493        opts: &ReadWriteOptions,
494        column_family: ColumnFamily,
495        is_deprecated: bool,
496    ) -> Self {
497        let db_cloned = Arc::downgrade(&db);
498        let db_metrics = DBMetrics::get();
499        let db_metrics_cloned = db_metrics.clone();
500        let cf = column_family.name().to_string();
501
502        let (sender, mut recv) = tokio::sync::oneshot::channel();
503        if !is_deprecated && matches!(db.storage, Storage::Rocks(_)) {
504            tokio::task::spawn(async move {
505                let mut interval =
506                    tokio::time::interval(Duration::from_secs(CF_METRICS_REPORT_PERIOD_SECS));
507                loop {
508                    tokio::select! {
509                        _ = interval.tick() => {
510                            if let Some(db) = db_cloned.upgrade() {
511                                let cf = cf.clone();
512                                let db_metrics = db_metrics.clone();
513                                if let Err(e) = tokio::task::spawn_blocking(move || {
514                                    Self::report_rocksdb_metrics(&db, &cf, &db_metrics);
515                                }).await {
516                                    error!("Failed to log metrics with error: {}", e);
517                                }
518                            } else {
519                                break;
520                            }
521                        }
522                        _ = &mut recv => break,
523                    }
524                }
525                debug!("Returning the cf metric logging task for DBMap: {}", &cf);
526            });
527        }
528        DBMap {
529            db: db.clone(),
530            opts: opts.clone(),
531            _phantom: PhantomData,
532            column_family,
533            db_metrics: db_metrics_cloned,
534            _metrics_task_cancel_handle: Arc::new(sender),
535            get_sample_interval: db.get_sampling_interval(),
536            multiget_sample_interval: db.multiget_sampling_interval(),
537            write_sample_interval: db.write_sampling_interval(),
538            iter_sample_interval: db.iter_sampling_interval(),
539        }
540    }
541
542    /// Reopens an open database as a typed map operating under a specific
543    /// column family. if no column family is passed, the default column
544    /// family is used.
545    #[instrument(level = "debug", skip(db), err)]
546    pub fn reopen(
547        db: &Arc<Database>,
548        opt_cf: Option<&str>,
549        rw_options: &ReadWriteOptions,
550        is_deprecated: bool,
551    ) -> Result<Self, TypedStoreError> {
552        let cf_key = opt_cf
553            .unwrap_or(rocksdb::DEFAULT_COLUMN_FAMILY_NAME)
554            .to_owned();
555
556        let column_family = match &db.storage {
557            Storage::Rocks(_) => ColumnFamily::Rocks(cf_key),
558            Storage::InMemory(_) => ColumnFamily::InMemory(cf_key),
559        };
560        Ok(DBMap::new(
561            db.clone(),
562            rw_options,
563            column_family,
564            is_deprecated,
565        ))
566    }
567
568    pub fn cf_name(&self) -> &str {
569        self.column_family.name()
570    }
571
572    pub fn batch(&self) -> DBBatch {
573        let batch = match &self.db.storage {
574            Storage::Rocks(_) => StorageWriteBatch::Rocks(WriteBatch::default()),
575            Storage::InMemory(_) => StorageWriteBatch::InMemory(InMemoryBatch::default()),
576        };
577        DBBatch::new(
578            &self.db,
579            batch,
580            &self.db_metrics,
581            &self.write_sample_interval,
582        )
583    }
584
585    /// Flush the memtable of this table's column family to SST files on disk.
586    pub fn flush(&self) -> Result<(), TypedStoreError> {
587        self.db.flush_cf(&self.column_family)
588    }
589
590    /// Iterate all column families and flush the memtables of every column
591    /// family to SST files on disk.
592    pub fn flush_all(&self) -> Result<(), TypedStoreError> {
593        self.db.flush_all()
594    }
595
596    pub fn compact_range<J: Serialize>(&self, start: &J, end: &J) -> Result<(), TypedStoreError> {
597        let from_buf = be_fix_int_ser(start);
598        let to_buf = be_fix_int_ser(end);
599        self.db
600            .compact_range_cf(&self.column_family, Some(from_buf), Some(to_buf));
601        Ok(())
602    }
603
604    pub fn compact_range_raw(
605        &self,
606        cf_name: &str,
607        start: Vec<u8>,
608        end: Vec<u8>,
609    ) -> Result<(), TypedStoreError> {
610        let cf = match &self.db.storage {
611            Storage::Rocks(_) => ColumnFamily::Rocks(cf_name.to_string()),
612            Storage::InMemory(_) => ColumnFamily::InMemory(cf_name.to_string()),
613        };
614        self.db.compact_range_cf(&cf, Some(start), Some(end));
615        Ok(())
616    }
617
618    /// Returns a vector of raw values corresponding to the keys provided.
619    fn multi_get_pinned<J>(
620        &self,
621        keys: impl IntoIterator<Item = J>,
622    ) -> Result<Vec<Option<GetResult<'_>>>, TypedStoreError>
623    where
624        J: Borrow<K>,
625        K: Serialize,
626    {
627        let _timer = self
628            .db_metrics
629            .op_metrics
630            .rocksdb_multiget_latency_seconds
631            .with_label_values(&[self.cf_name()])
632            .start_timer();
633        let perf_ctx = if self.multiget_sample_interval.sample() {
634            Some(RocksDBPerfContext)
635        } else {
636            None
637        };
638        let keys_bytes = keys.into_iter().map(|k| be_fix_int_ser(k.borrow()));
639        let results: Result<Vec<_>, TypedStoreError> = self
640            .db
641            .multi_get(&self.column_family, keys_bytes, &self.opts.readopts())
642            .into_iter()
643            .collect();
644        let entries = results?;
645        let entry_size = entries
646            .iter()
647            .flatten()
648            .map(|entry| entry.len())
649            .sum::<usize>();
650        self.db_metrics
651            .op_metrics
652            .rocksdb_multiget_bytes
653            .with_label_values(&[self.cf_name()])
654            .observe(entry_size as f64);
655        if perf_ctx.is_some() {
656            self.db_metrics
657                .read_perf_ctx_metrics
658                .report_metrics(self.cf_name());
659        }
660        Ok(entries)
661    }
662
663    pub fn checkpoint_db(&self, path: &Path) -> Result<(), TypedStoreError> {
664        self.db.checkpoint(path)
665    }
666
667    pub fn table_summary(&self) -> eyre::Result<TableSummary>
668    where
669        K: Serialize + DeserializeOwned,
670        V: Serialize + DeserializeOwned,
671    {
672        let mut num_keys = 0;
673        let mut key_bytes_total = 0;
674        let mut value_bytes_total = 0;
675        let mut key_hist = hdrhistogram::Histogram::<u64>::new_with_max(100000, 2).unwrap();
676        let mut value_hist = hdrhistogram::Histogram::<u64>::new_with_max(100000, 2).unwrap();
677        for item in self.safe_iter() {
678            let (key, value) = item?;
679            num_keys += 1;
680            let key_len = be_fix_int_ser(key.borrow()).len();
681            let value_len = bcs::to_bytes(value.borrow())?.len();
682            key_bytes_total += key_len;
683            value_bytes_total += value_len;
684            key_hist.record(key_len as u64)?;
685            value_hist.record(value_len as u64)?;
686        }
687        Ok(TableSummary {
688            num_keys,
689            key_bytes_total,
690            value_bytes_total,
691            key_hist,
692            value_hist,
693        })
694    }
695
696    // Creates metrics and context for tracking an iterator usage and performance.
697    fn create_iter_context(
698        &self,
699    ) -> (
700        Option<HistogramTimer>,
701        Option<Histogram>,
702        Option<Histogram>,
703        Option<RocksDBPerfContext>,
704    ) {
705        let timer = self
706            .db_metrics
707            .op_metrics
708            .rocksdb_iter_latency_seconds
709            .with_label_values(&[self.cf_name()])
710            .start_timer();
711        let bytes_scanned = self
712            .db_metrics
713            .op_metrics
714            .rocksdb_iter_bytes
715            .with_label_values(&[self.cf_name()]);
716        let keys_scanned = self
717            .db_metrics
718            .op_metrics
719            .rocksdb_iter_keys
720            .with_label_values(&[self.cf_name()]);
721        let perf_ctx = if self.iter_sample_interval.sample() {
722            Some(RocksDBPerfContext)
723        } else {
724            None
725        };
726        (
727            Some(timer),
728            Some(bytes_scanned),
729            Some(keys_scanned),
730            perf_ctx,
731        )
732    }
733
734    /// Shared rocksdb `SafeIter` construction (raw iterator + scan metrics)
735    /// used by both the forward and reverse raw iterators.
736    fn rocks_safe_iter<'a>(&self, db: &'a RocksDB, readopts: ReadOptions) -> SafeIter<'a, K, V>
737    where
738        K: DeserializeOwned,
739        V: DeserializeOwned,
740    {
741        let db_iter = db
742            .underlying
743            .raw_iterator_cf_opt(&rocks_cf(db, self.column_family.name()), readopts);
744        let (_timer, bytes_scanned, keys_scanned, _perf_ctx) = self.create_iter_context();
745        SafeIter::new(
746            self.cf_name().to_string(),
747            db_iter,
748            _timer,
749            _perf_ctx,
750            bytes_scanned,
751            keys_scanned,
752            Some(self.db_metrics.clone()),
753        )
754    }
755
756    /// Forward iterator over the raw byte bounds `[lower_bound, upper_bound)`;
757    /// both bounds (already serialized, `upper_bound` exclusive) are applied to
758    /// the read options.
759    fn iter_forward_raw(
760        &self,
761        lower_bound: Option<Vec<u8>>,
762        upper_bound: Option<Vec<u8>>,
763    ) -> DbIterator<'_, (K, V)>
764    where
765        K: DeserializeOwned,
766        V: DeserializeOwned,
767    {
768        match &self.db.storage {
769            Storage::Rocks(db) => {
770                let readopts =
771                    rocks_util::apply_range_bounds(self.opts.readopts(), lower_bound, upper_bound);
772                Box::new(self.rocks_safe_iter(db, readopts))
773            }
774            Storage::InMemory(db) => {
775                db.iterator(self.column_family.name(), lower_bound, upper_bound, false)
776            }
777        }
778    }
779
780    /// Reverse counterpart of [`Self::iter_forward_raw`], yielding the same
781    /// keys in descending order. Only the lower bound is applied to the
782    /// read options; the (exclusive) `upper_bound` is enforced by
783    /// [`SafeRevIter`]'s initial seek, because setting an
784    /// iterate-upper-bound would stop `seek_for_prev` from landing on the
785    /// boundary key.
786    fn iter_reversed_raw(
787        &self,
788        lower_bound: Option<Vec<u8>>,
789        upper_bound: Option<Vec<u8>>,
790    ) -> DbIterator<'_, (K, V)>
791    where
792        K: DeserializeOwned,
793        V: DeserializeOwned,
794    {
795        match &self.db.storage {
796            Storage::Rocks(db) => {
797                let readopts =
798                    rocks_util::apply_range_bounds(self.opts.readopts(), lower_bound, None);
799                Box::new(SafeRevIter::new(
800                    self.rocks_safe_iter(db, readopts),
801                    upper_bound,
802                ))
803            }
804            Storage::InMemory(db) => {
805                db.iterator(self.column_family.name(), lower_bound, upper_bound, true)
806            }
807        }
808    }
809
810    /// Reverse counterpart of [`Map::safe_range_iter`]: yields exactly the keys
811    /// of `safe_range_iter(range)` in descending order.
812    ///
813    /// Both directions derive their bounds from the same
814    /// `iterator_bounds_with_range`, so they are guaranteed to cover the
815    /// identical set of keys regardless of the bound inclusivity.
816    pub fn safe_range_iter_reversed(&self, range: impl RangeBounds<K>) -> DbIterator<'_, (K, V)>
817    where
818        K: Serialize + DeserializeOwned,
819        V: DeserializeOwned,
820    {
821        let (lower_bound, upper_bound) = iterator_bounds_with_range(range);
822        self.iter_reversed_raw(lower_bound, upper_bound)
823    }
824
825    /// Forward iterator over every entry whose key begins with `prefix`.
826    ///
827    /// `prefix` is serialized with `be_fix_int_ser` and must form a prefix of
828    /// the column family's key encoding — typically the leading field(s) of a
829    /// tuple key. This avoids constructing artificial maximum keys to bound a
830    /// composite-key scan.
831    pub fn safe_iter_with_prefix<P>(&self, prefix: &P) -> DbIterator<'_, (K, V)>
832    where
833        P: ?Sized + Serialize,
834        K: DeserializeOwned,
835        V: DeserializeOwned,
836    {
837        let (lower_bound, upper_bound) = prefix_iterator_bounds(prefix);
838        self.iter_forward_raw(lower_bound, upper_bound)
839    }
840
841    /// Forward iterator over entries whose key begins with `prefix`, starting
842    /// at `prefix ++ cursor` rather than at the first key under the prefix.
843    ///
844    /// `cursor` is **not** a full key — it is the remainder of the key that
845    /// follows `prefix` (typically the trailing field(s) of a tuple key). Both
846    /// `prefix` and `cursor` are serialized with `be_fix_int_ser` and
847    /// concatenated, mirroring how a composite key is encoded; the scan then
848    /// covers `[prefix ++ cursor, end-of-prefix)`. This lets a paginated scan
849    /// resume from a cursor without an artificial maximum key for the upper
850    /// bound.
851    pub fn safe_iter_with_prefix_from<P, C>(&self, prefix: &P, cursor: &C) -> DbIterator<'_, (K, V)>
852    where
853        P: ?Sized + Serialize,
854        C: ?Sized + Serialize,
855        K: DeserializeOwned,
856        V: DeserializeOwned,
857    {
858        let (lower_bound, upper_bound) = prefix_iterator_bounds(prefix);
859        let lower_bound = lower_bound.map(|mut lower| {
860            lower.extend_from_slice(&be_fix_int_ser(cursor));
861            lower
862        });
863        self.iter_forward_raw(lower_bound, upper_bound)
864    }
865
866    /// Reverse counterpart of [`Self::safe_iter_with_prefix`]: the matching
867    /// entries in descending key order (e.g. for "latest entry under a
868    /// prefix").
869    pub fn safe_iter_with_prefix_reversed<P>(&self, prefix: &P) -> DbIterator<'_, (K, V)>
870    where
871        P: ?Sized + Serialize,
872        K: DeserializeOwned,
873        V: DeserializeOwned,
874    {
875        let (lower_bound, upper_bound) = prefix_iterator_bounds(prefix);
876        self.iter_reversed_raw(lower_bound, upper_bound)
877    }
878}
879
880/// Provides a mutable struct to form a collection of database write operations,
881/// and execute them.
882///
883/// Batching write and delete operations is faster than performing them one by
884/// one and ensures their atomicity,  ie. they are all written or none is.
885/// This is also true of operations across column families in the same database.
886///
887/// Serializations / Deserialization, and naming of column families is performed
888/// by passing a DBMap<K,V> with each operation.
889///
890/// ```
891/// use core::fmt::Error;
892/// use std::sync::Arc;
893///
894/// use prometheus_filtered::Registry;
895/// use tempfile::tempdir;
896/// use typed_store::{Map, metrics::DBMetrics, rocks::*};
897///
898/// #[tokio::main]
899/// async fn main() -> Result<(), Error> {
900///     let rocks = open_cf_opts(
901///         tempfile::tempdir().unwrap(),
902///         None,
903///         MetricConf::default(),
904///         &[
905///             ("First_CF", rocksdb::Options::default()),
906///             ("Second_CF", rocksdb::Options::default()),
907///         ],
908///     )
909///     .unwrap();
910///
911///     let db_cf_1 = DBMap::reopen(
912///         &rocks,
913///         Some("First_CF"),
914///         &ReadWriteOptions::default(),
915///         false,
916///     )
917///     .expect("Failed to open storage");
918///     let keys_vals_1 = (1..100).map(|i| (i, i.to_string()));
919///
920///     let db_cf_2 = DBMap::reopen(
921///         &rocks,
922///         Some("Second_CF"),
923///         &ReadWriteOptions::default(),
924///         false,
925///     )
926///     .expect("Failed to open storage");
927///     let keys_vals_2 = (1000..1100).map(|i| (i, i.to_string()));
928///
929///     let mut batch = db_cf_1.batch();
930///     batch
931///         .insert_batch(&db_cf_1, keys_vals_1.clone())
932///         .expect("Failed to batch insert")
933///         .insert_batch(&db_cf_2, keys_vals_2.clone())
934///         .expect("Failed to batch insert");
935///
936///     let _ = batch.write().expect("Failed to execute batch");
937///     for (k, v) in keys_vals_1 {
938///         let val = db_cf_1.get(&k).expect("Failed to get inserted key");
939///         assert_eq!(Some(v), val);
940///     }
941///
942///     for (k, v) in keys_vals_2 {
943///         let val = db_cf_2.get(&k).expect("Failed to get inserted key");
944///         assert_eq!(Some(v), val);
945///     }
946///     Ok(())
947/// }
948/// ```
949pub struct DBBatch {
950    database: Arc<Database>,
951    batch: StorageWriteBatch,
952    db_metrics: Arc<DBMetrics>,
953    write_sample_interval: SamplingInterval,
954}
955
956impl DBBatch {
957    /// Create a new batch associated with a DB reference.
958    ///
959    /// Use `open_cf` to get the DB reference or an existing open database.
960    pub fn new(
961        dbref: &Arc<Database>,
962        batch: StorageWriteBatch,
963        db_metrics: &Arc<DBMetrics>,
964        write_sample_interval: &SamplingInterval,
965    ) -> Self {
966        DBBatch {
967            database: dbref.clone(),
968            batch,
969            db_metrics: db_metrics.clone(),
970            write_sample_interval: write_sample_interval.clone(),
971        }
972    }
973
974    /// Consume the batch and write its operations to the database
975    #[instrument(level = "trace", skip_all, err)]
976    pub fn write(self) -> Result<(), TypedStoreError> {
977        self.write_opt(&rocksdb::WriteOptions::default())
978    }
979
980    /// Consume the batch and write its operations to the database with custom
981    /// write options
982    #[instrument(level = "trace", skip_all, err)]
983    pub fn write_opt(self, write_options: &rocksdb::WriteOptions) -> Result<(), TypedStoreError> {
984        let db_name = self.database.db_name();
985        let timer = self
986            .db_metrics
987            .op_metrics
988            .rocksdb_batch_commit_latency_seconds
989            .with_label_values(&[&db_name])
990            .start_timer();
991        let batch_size = self.size_in_bytes();
992
993        let perf_ctx = if self.write_sample_interval.sample() {
994            Some(RocksDBPerfContext)
995        } else {
996            None
997        };
998        self.database.write_opt(self.batch, write_options)?;
999        self.db_metrics
1000            .op_metrics
1001            .rocksdb_batch_commit_bytes
1002            .with_label_values(&[&db_name])
1003            .observe(batch_size as f64);
1004
1005        if perf_ctx.is_some() {
1006            self.db_metrics
1007                .write_perf_ctx_metrics
1008                .report_metrics(&db_name);
1009        }
1010        let elapsed = timer.stop_and_record();
1011        if elapsed > 1.0 {
1012            warn!(?elapsed, ?db_name, "very slow batch write");
1013            self.db_metrics
1014                .op_metrics
1015                .rocksdb_very_slow_batch_writes_count
1016                .with_label_values(&[&db_name])
1017                .inc();
1018            self.db_metrics
1019                .op_metrics
1020                .rocksdb_very_slow_batch_writes_duration_ms
1021                .with_label_values(&[&db_name])
1022                .inc_by((elapsed * 1000.0) as u64);
1023        }
1024        Ok(())
1025    }
1026
1027    pub fn size_in_bytes(&self) -> usize {
1028        match self.batch {
1029            StorageWriteBatch::Rocks(ref b) => b.size_in_bytes(),
1030            StorageWriteBatch::InMemory(_) => 0,
1031        }
1032    }
1033
1034    pub fn delete_batch<J: Borrow<K>, K: Serialize, V>(
1035        &mut self,
1036        db: &DBMap<K, V>,
1037        purged_vals: impl IntoIterator<Item = J>,
1038    ) -> Result<(), TypedStoreError> {
1039        if !Arc::ptr_eq(&db.db, &self.database) {
1040            return Err(TypedStoreError::CrossDBBatch);
1041        }
1042
1043        purged_vals
1044            .into_iter()
1045            .try_for_each::<_, Result<_, TypedStoreError>>(|k| {
1046                let k_buf = be_fix_int_ser(k.borrow());
1047                match (&mut self.batch, &db.column_family) {
1048                    (StorageWriteBatch::Rocks(b), ColumnFamily::Rocks(name)) => {
1049                        b.delete_cf(&rocks_cf_from_db(&self.database, name)?, k_buf)
1050                    }
1051                    (StorageWriteBatch::InMemory(b), ColumnFamily::InMemory(name)) => {
1052                        b.delete_cf(name, k_buf)
1053                    }
1054                    _ => Err(TypedStoreError::RocksDB(
1055                        "typed store invariant violation".to_string(),
1056                    ))?,
1057                }
1058                Ok(())
1059            })?;
1060        Ok(())
1061    }
1062
1063    /// Deletes a range of keys between `from` (inclusive) and `to`
1064    /// (non-inclusive) by writing a range delete tombstone in the db map.
1065    /// The effect of this write is visible immediately, i.e. you won't see
1066    /// old values when you do a lookup or scan.
1067    pub fn schedule_delete_range<K: Serialize, V>(
1068        &mut self,
1069        db: &DBMap<K, V>,
1070        from: &K,
1071        to: &K,
1072    ) -> Result<(), TypedStoreError> {
1073        if !Arc::ptr_eq(&db.db, &self.database) {
1074            return Err(TypedStoreError::CrossDBBatch);
1075        }
1076
1077        let from_buf = be_fix_int_ser(from);
1078        let to_buf = be_fix_int_ser(to);
1079
1080        if let StorageWriteBatch::Rocks(b) = &mut self.batch {
1081            b.delete_range_cf(
1082                &rocks_cf_from_db(&self.database, db.cf_name())?,
1083                from_buf,
1084                to_buf,
1085            );
1086        }
1087        Ok(())
1088    }
1089
1090    /// inserts a range of (key, value) pairs given as an iterator
1091    pub fn insert_batch<J: Borrow<K>, K: Serialize, U: Borrow<V>, V: Serialize>(
1092        &mut self,
1093        db: &DBMap<K, V>,
1094        new_vals: impl IntoIterator<Item = (J, U)>,
1095    ) -> Result<&mut Self, TypedStoreError> {
1096        if !Arc::ptr_eq(&db.db, &self.database) {
1097            return Err(TypedStoreError::CrossDBBatch);
1098        }
1099        let mut total = 0usize;
1100        new_vals
1101            .into_iter()
1102            .try_for_each::<_, Result<_, TypedStoreError>>(|(k, v)| {
1103                let k_buf = be_fix_int_ser(k.borrow());
1104                let v_buf = bcs::to_bytes(v.borrow()).map_err(typed_store_err_from_bcs_err)?;
1105                total += k_buf.len() + v_buf.len();
1106                if db.opts.log_value_hash {
1107                    let key_hash = default_hash(&k_buf);
1108                    let value_hash = default_hash(&v_buf);
1109                    debug!(
1110                        "Insert to DB table: {:?}, key_hash: {:?}, value_hash: {:?}",
1111                        db.cf_name(),
1112                        key_hash,
1113                        value_hash
1114                    );
1115                }
1116                match (&mut self.batch, &db.column_family) {
1117                    (StorageWriteBatch::Rocks(b), ColumnFamily::Rocks(name)) => {
1118                        b.put_cf(&rocks_cf_from_db(&self.database, name)?, k_buf, v_buf)
1119                    }
1120                    (StorageWriteBatch::InMemory(b), ColumnFamily::InMemory(name)) => {
1121                        b.put_cf(name, k_buf, v_buf)
1122                    }
1123                    _ => Err(TypedStoreError::RocksDB(
1124                        "typed store invariant violation".to_string(),
1125                    ))?,
1126                }
1127                Ok(())
1128            })?;
1129        self.db_metrics
1130            .op_metrics
1131            .rocksdb_batch_put_bytes
1132            .with_label_values(&[db.cf_name()])
1133            .observe(total as f64);
1134        Ok(self)
1135    }
1136}
1137
1138impl<'a, K, V> Map<'a, K, V> for DBMap<K, V>
1139where
1140    K: Serialize + DeserializeOwned,
1141    V: Serialize + DeserializeOwned,
1142{
1143    type Error = TypedStoreError;
1144
1145    #[instrument(level = "trace", skip_all, err)]
1146    fn contains_key(&self, key: &K) -> Result<bool, TypedStoreError> {
1147        let key_buf = be_fix_int_ser(key);
1148        let readopts = self.opts.readopts();
1149        Ok(self
1150            .db
1151            .key_may_exist_cf(&self.column_family, &key_buf, &readopts)
1152            && self
1153                .db
1154                .get(&self.column_family, &key_buf, &readopts)?
1155                .is_some())
1156    }
1157
1158    #[instrument(level = "trace", skip_all, err)]
1159    fn multi_contains_keys<J>(
1160        &self,
1161        keys: impl IntoIterator<Item = J>,
1162    ) -> Result<Vec<bool>, Self::Error>
1163    where
1164        J: Borrow<K>,
1165    {
1166        let values = self.multi_get_pinned(keys)?;
1167        Ok(values.into_iter().map(|v| v.is_some()).collect())
1168    }
1169
1170    #[instrument(level = "trace", skip_all, err)]
1171    fn get(&self, key: &K) -> Result<Option<V>, TypedStoreError> {
1172        let _timer = self
1173            .db_metrics
1174            .op_metrics
1175            .rocksdb_get_latency_seconds
1176            .with_label_values(&[self.cf_name()])
1177            .start_timer();
1178        let perf_ctx = if self.get_sample_interval.sample() {
1179            Some(RocksDBPerfContext)
1180        } else {
1181            None
1182        };
1183        let key_buf = be_fix_int_ser(key);
1184        let res = self
1185            .db
1186            .get(&self.column_family, &key_buf, &self.opts.readopts())?;
1187        self.db_metrics
1188            .op_metrics
1189            .rocksdb_get_bytes
1190            .with_label_values(&[self.cf_name()])
1191            .observe(res.as_ref().map_or(0.0, |v| v.len() as f64));
1192        if perf_ctx.is_some() {
1193            self.db_metrics
1194                .read_perf_ctx_metrics
1195                .report_metrics(self.cf_name());
1196        }
1197        match res {
1198            Some(data) => {
1199                let value = bcs::from_bytes(&data).map_err(typed_store_err_from_bcs_err);
1200                if value.is_err() {
1201                    let key_hash = default_hash(&key_buf);
1202                    let value_hash = default_hash(&data);
1203                    debug_fatal!(
1204                        "Failed to deserialize value from DB table {:?}, key_hash: {:?}, value_hash: {:?}, error: {:?}",
1205                        self.cf_name(),
1206                        key_hash,
1207                        value_hash,
1208                        value.as_ref().err().unwrap()
1209                    );
1210                }
1211                Ok(Some(value?))
1212            }
1213            None => Ok(None),
1214        }
1215    }
1216
1217    #[instrument(level = "trace", skip_all, err)]
1218    fn insert(&self, key: &K, value: &V) -> Result<(), TypedStoreError> {
1219        let timer = self
1220            .db_metrics
1221            .op_metrics
1222            .rocksdb_put_latency_seconds
1223            .with_label_values(&[self.cf_name()])
1224            .start_timer();
1225        let perf_ctx = if self.write_sample_interval.sample() {
1226            Some(RocksDBPerfContext)
1227        } else {
1228            None
1229        };
1230        let key_buf = be_fix_int_ser(key);
1231        let value_buf = bcs::to_bytes(value).map_err(typed_store_err_from_bcs_err)?;
1232        self.db_metrics
1233            .op_metrics
1234            .rocksdb_put_bytes
1235            .with_label_values(&[self.cf_name()])
1236            .observe((key_buf.len() + value_buf.len()) as f64);
1237        if perf_ctx.is_some() {
1238            self.db_metrics
1239                .write_perf_ctx_metrics
1240                .report_metrics(self.cf_name());
1241        }
1242        self.db.put_cf(&self.column_family, key_buf, value_buf)?;
1243
1244        let elapsed = timer.stop_and_record();
1245        if elapsed > 1.0 {
1246            warn!(?elapsed, cf = ?self.cf_name(), "very slow insert");
1247            self.db_metrics
1248                .op_metrics
1249                .rocksdb_very_slow_puts_count
1250                .with_label_values(&[self.cf_name()])
1251                .inc();
1252            self.db_metrics
1253                .op_metrics
1254                .rocksdb_very_slow_puts_duration_ms
1255                .with_label_values(&[self.cf_name()])
1256                .inc_by((elapsed * 1000.0) as u64);
1257        }
1258
1259        Ok(())
1260    }
1261
1262    #[instrument(level = "trace", skip_all, err)]
1263    fn remove(&self, key: &K) -> Result<(), TypedStoreError> {
1264        let _timer = self
1265            .db_metrics
1266            .op_metrics
1267            .rocksdb_delete_latency_seconds
1268            .with_label_values(&[self.cf_name()])
1269            .start_timer();
1270        let perf_ctx = if self.write_sample_interval.sample() {
1271            Some(RocksDBPerfContext)
1272        } else {
1273            None
1274        };
1275        let key_buf = be_fix_int_ser(key);
1276        self.db.delete_cf(&self.column_family, key_buf)?;
1277        self.db_metrics
1278            .op_metrics
1279            .rocksdb_deletes
1280            .with_label_values(&[self.cf_name()])
1281            .inc();
1282        if perf_ctx.is_some() {
1283            self.db_metrics
1284                .write_perf_ctx_metrics
1285                .report_metrics(self.cf_name());
1286        }
1287        Ok(())
1288    }
1289
1290    /// Writes a range delete tombstone to delete all entries in the db map.
1291    /// The effect of this write is visible immediately, i.e. you won't see
1292    /// old values when you do a lookup or scan.
1293    #[instrument(level = "trace", skip_all, err)]
1294    fn schedule_delete_all(&self) -> Result<(), TypedStoreError> {
1295        let first_key = self.safe_iter().next().transpose()?.map(|(k, _v)| k);
1296        let last_key = self
1297            .safe_range_iter_reversed(..)
1298            .next()
1299            .transpose()?
1300            .map(|(k, _v)| k);
1301        if let Some((first_key, last_key)) = first_key.zip(last_key) {
1302            let mut batch = self.batch();
1303            batch.schedule_delete_range(self, &first_key, &last_key)?;
1304            batch.write()?;
1305        }
1306        Ok(())
1307    }
1308
1309    fn is_empty(&self) -> bool {
1310        self.safe_iter().next().is_none()
1311    }
1312
1313    fn safe_iter(&'a self) -> DbIterator<'a, (K, V)> {
1314        match &self.db.storage {
1315            Storage::Rocks(db) => {
1316                let db_iter = db.underlying.raw_iterator_cf_opt(
1317                    &rocks_cf(db, self.column_family.name()),
1318                    self.opts.readopts(),
1319                );
1320                let (_timer, bytes_scanned, keys_scanned, _perf_ctx) = self.create_iter_context();
1321                Box::new(SafeIter::new(
1322                    self.cf_name().to_string(),
1323                    db_iter,
1324                    _timer,
1325                    _perf_ctx,
1326                    bytes_scanned,
1327                    keys_scanned,
1328                    Some(self.db_metrics.clone()),
1329                ))
1330            }
1331            Storage::InMemory(db) => db.iterator(self.column_family.name(), None, None, false),
1332        }
1333    }
1334
1335    fn safe_iter_with_bounds(
1336        &'a self,
1337        lower_bound: Option<K>,
1338        upper_bound: Option<K>,
1339    ) -> DbIterator<'a, (K, V)> {
1340        let range = (
1341            lower_bound.map(Bound::Included).unwrap_or(Bound::Unbounded),
1342            upper_bound.map(Bound::Excluded).unwrap_or(Bound::Unbounded),
1343        );
1344        self.safe_range_iter(range)
1345    }
1346
1347    fn safe_range_iter(&'a self, range: impl RangeBounds<K>) -> DbIterator<'a, (K, V)> {
1348        let (lower_bound, upper_bound) = iterator_bounds_with_range(range);
1349        self.iter_forward_raw(lower_bound, upper_bound)
1350    }
1351
1352    /// Returns a vector of values corresponding to the keys provided.
1353    #[instrument(level = "trace", skip_all, err)]
1354    fn multi_get<J>(
1355        &self,
1356        keys: impl IntoIterator<Item = J>,
1357    ) -> Result<Vec<Option<V>>, TypedStoreError>
1358    where
1359        J: Borrow<K>,
1360    {
1361        let results = self.multi_get_pinned(keys)?;
1362        let values_parsed: Result<Vec<_>, TypedStoreError> = results
1363            .into_iter()
1364            .map(|value_byte| match value_byte {
1365                Some(data) => Ok(Some(
1366                    bcs::from_bytes(&data).map_err(typed_store_err_from_bcs_err)?,
1367                )),
1368                None => Ok(None),
1369            })
1370            .collect();
1371
1372        values_parsed
1373    }
1374
1375    /// Convenience method for batch insertion
1376    #[instrument(level = "trace", skip_all, err)]
1377    fn multi_insert<J, U>(
1378        &self,
1379        key_val_pairs: impl IntoIterator<Item = (J, U)>,
1380    ) -> Result<(), Self::Error>
1381    where
1382        J: Borrow<K>,
1383        U: Borrow<V>,
1384    {
1385        let mut batch = self.batch();
1386        batch.insert_batch(self, key_val_pairs)?;
1387        batch.write()
1388    }
1389
1390    /// Convenience method for batch removal
1391    #[instrument(level = "trace", skip_all, err)]
1392    fn multi_remove<J>(&self, keys: impl IntoIterator<Item = J>) -> Result<(), Self::Error>
1393    where
1394        J: Borrow<K>,
1395    {
1396        let mut batch = self.batch();
1397        batch.delete_batch(self, keys)?;
1398        batch.write()
1399    }
1400
1401    /// Try to catch up with primary when running as secondary
1402    #[instrument(level = "trace", skip_all, err)]
1403    fn try_catch_up_with_primary(&self) -> Result<(), Self::Error> {
1404        self.db.try_catch_up_with_primary()
1405    }
1406}
1407
1408fn default_hash(value: &[u8]) -> Digest<32> {
1409    let mut hasher = fastcrypto::hash::Blake2b256::default();
1410    hasher.update(value);
1411    hasher.finalize()
1412}
typed_store/database.rs

typed_store/
database.rs
