typed_store/rocks/

options.rs

// Copyright (c) Mysten Labs, Inc.
// Modifications Copyright (c) 2026 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

use std::{
    collections::BTreeMap,
    env,
    sync::atomic::{AtomicBool, Ordering},
};

use iota_macros::nondeterministic;
use rocksdb::{BlockBasedOptions, Cache, ReadOptions};
use sysinfo::{MemoryRefreshKind, RefreshKind, System};
use tap::TapFallible;
use tracing::{debug, info, warn};

// Write buffer size per RocksDB instance can be set via the env var below.
// If the env var is not set, use the default value in MiB.
const ENV_VAR_DB_WRITE_BUFFER_SIZE: &str = "DB_WRITE_BUFFER_SIZE_MB";
const DEFAULT_DB_WRITE_BUFFER_SIZE: usize = 1024;

// Write ahead log size per RocksDB instance can be set via the env var below.
// If the env var is not set, use the default value in MiB.
const ENV_VAR_DB_WAL_SIZE: &str = "DB_WAL_SIZE_MB";
const DEFAULT_DB_WAL_SIZE: usize = 1024;

// Environment variable to control behavior of write throughput optimized
// tables.
const ENV_VAR_L0_NUM_FILES_COMPACTION_TRIGGER: &str = "L0_NUM_FILES_COMPACTION_TRIGGER";
const DEFAULT_L0_NUM_FILES_COMPACTION_TRIGGER: usize = 4;
const DEFAULT_UNIVERSAL_COMPACTION_L0_NUM_FILES_COMPACTION_TRIGGER: usize = 80;
const ENV_VAR_MAX_WRITE_BUFFER_SIZE_MB: &str = "MAX_WRITE_BUFFER_SIZE_MB";
const DEFAULT_MAX_WRITE_BUFFER_SIZE_MB: usize = 256;
const ENV_VAR_MAX_WRITE_BUFFER_NUMBER: &str = "MAX_WRITE_BUFFER_NUMBER";
const DEFAULT_MAX_WRITE_BUFFER_NUMBER: usize = 6;
const ENV_VAR_TARGET_FILE_SIZE_BASE_MB: &str = "TARGET_FILE_SIZE_BASE_MB";
const DEFAULT_TARGET_FILE_SIZE_BASE_MB: usize = 128;

// Set to 1 to disable blob storage for transactions and effects.
const ENV_VAR_DISABLE_BLOB_STORAGE: &str = "DISABLE_BLOB_STORAGE";
const ENV_VAR_DB_PARALLELISM: &str = "DB_PARALLELISM";

static FALLOCATE_DISABLED: AtomicBool = AtomicBool::new(false);

/// Stops RocksDB from preallocating file space in databases opened afterwards
/// with [`default_db_options`] or [`bulk_ingestion_options`].
///
/// For tests that run many nodes in one process: on Linux, each write-ahead
/// log otherwise reserves up to 1.1 × the write buffer size on disk.
pub fn disable_fallocate() {
    FALLOCATE_DISABLED.store(true, Ordering::Relaxed);
}

/// Returns the RocksDB default options, with `allow_fallocate` turned off if
/// [`disable_fallocate`] was called.
pub(crate) fn base_db_options() -> rocksdb::Options {
    let mut options = rocksdb::Options::default();
    if !FALLOCATE_DISABLED.load(Ordering::Relaxed) {
        return options;
    }
    // The crate has no setter for `allow_fallocate`. The parsed copy drops the
    // objects attached to the original, so this must run before any are set.
    options
        .get_options_from_string("allow_fallocate=false")
        .expect("allow_fallocate should be a valid RocksDB option")
}

#[derive(Clone, Debug, Default)]
pub struct ReadWriteOptions {
    /// When set, debug log the hash of the key and value bytes when inserting
    /// to this table.
    pub log_value_hash: bool,
}

impl ReadWriteOptions {
    pub fn readopts(&self) -> ReadOptions {
        ReadOptions::default()
    }

    pub fn set_log_value_hash(mut self, log_value_hash: bool) -> Self {
        self.log_value_hash = log_value_hash;
        self
    }
}

#[derive(Default, Clone)]
pub struct DBOptions {
    pub options: rocksdb::Options,
    pub rw_options: ReadWriteOptions,
}

/// Write options for a record that must be on disk before the write returns,
/// rather than in the operating system's cache until some later sync.
pub fn synced_write_options() -> rocksdb::WriteOptions {
    let mut opts = rocksdb::WriteOptions::default();
    opts.set_sync(true);
    opts
}

/// The smallest SST file buffer limit, so a small memory budget does not
/// split the rows into many tiny files.
const MIN_SST_BUFFER_SIZE: usize = 64 << 20;

/// RocksDB options tuned for one-shot bulk ingestion into a freshly created
/// store, e.g. initial index building, inserting the genesis snapshot, or
/// restoring a formal snapshot.
pub struct BulkIngestionOptions {
    pub db_options: rocksdb::Options,
    pub column_family_options: DBOptions,
    /// The memtable memory one column family may take.
    cf_memory_budget: usize,
}

impl BulkIngestionOptions {
    /// How much memory each of `writers` concurrent
    /// [`SstFileBuffer`](crate::rocks::SstFileBuffer)s may fill before writing
    /// a file: one column family's memtable budget, shared between them.
    pub fn sst_buffer_limit(&self, writers: usize) -> usize {
        (self.cf_memory_budget / writers.max(1)).max(MIN_SST_BUFFER_SIZE)
    }

    /// Per-table config applying the bulk column-family options to every
    /// listed table.
    pub fn table_config(
        &self,
        table_names: impl IntoIterator<Item = String>,
    ) -> DBMapTableConfigMap {
        DBMapTableConfigMap::new(
            table_names
                .into_iter()
                .map(|name| (name, self.column_family_options.clone()))
                .collect(),
        )
    }
}

/// Memtable budget for a bulk-ingestion store, as `(db_write_buffer_size,
/// cf_memory_budget)`: 80% of system RAM across the whole database and 25%
/// for a single column family.
fn write_buffer_budget(total_memory_bytes: u64) -> (usize, usize) {
    // A failed memory probe reports 0, and RocksDB reads a whole-database cap
    // of 0 as unlimited; a measured limit is used as is, however small.
    const FALLBACK_TOTAL_MEMORY_BYTES: u64 = 4 << 30;

    let total_memory_bytes = if total_memory_bytes == 0 {
        FALLBACK_TOTAL_MEMORY_BYTES as f64
    } else {
        total_memory_bytes as f64
    };
    (
        (total_memory_bytes * 0.8) as usize,
        (total_memory_bytes * 0.25) as usize,
    )
}

pub fn bulk_ingestion_options() -> BulkIngestionOptions {
    let total_memory_bytes = available_memory_bytes();
    let num_cpus = num_cpus::get();
    let (db_write_buffer_size, cf_memory_budget) = write_buffer_budget(total_memory_bytes);

    let mut db_options = base_db_options();

    // `unordered_write` gives a large speedup for bulk writes; relaxed write
    // ordering is acceptable because the store is rebuilt from scratch on
    // failure.
    db_options.set_unordered_write(true);

    // Allow CPU-intensive flushing to use all cores.
    db_options.set_max_background_jobs(num_cpus as i32);

    // Upper bound on memtable memory: 80% of RAM, sized so flushing keeps up
    // with the writers.
    db_options.set_db_write_buffer_size(db_write_buffer_size);

    // Per-column-family options: Create options with compactions disabled and large
    // write buffers. Each CF can use up to 25% of system RAM, but total is
    // still limited by set_db_write_buffer_size configured above.
    let mut column_family_options = default_db_options();
    column_family_options
        .options
        .set_disable_auto_compactions(true);

    // Disable the write backpressure that kicks in as L0 files build up: `-1`
    // turns off the compaction and slowdown triggers, and the stop trigger is
    // pushed out of reach. These are per-column-family options, so they must be
    // set here rather than on `db_options` to reach the index CFs. Compaction
    // itself is disabled above, so no L0 files are ever compacted away.
    column_family_options
        .options
        .set_level_zero_file_num_compaction_trigger(-1);
    column_family_options
        .options
        .set_level_zero_slowdown_writes_trigger(-1);
    column_family_options
        .options
        .set_level_zero_stop_writes_trigger(i32::MAX);

    const MIN_BUFFER_SIZE: usize = 64 * 1024 * 1024; // 64 MiB
    // More CPUs means more parallel flushing capacity, so target more buffers,
    // but never shrink a buffer below the minimum.
    let target_buffer_count = num_cpus.max(2);
    let buffer_size = (cf_memory_budget / target_buffer_count).max(MIN_BUFFER_SIZE);
    let buffer_count = (cf_memory_budget / buffer_size).clamp(2, target_buffer_count) as i32;
    column_family_options
        .options
        .set_write_buffer_size(buffer_size);
    column_family_options
        .options
        .set_max_write_buffer_number(buffer_count);

    debug!(
        total_memory_bytes,
        num_cpus,
        db_write_buffer_size,
        buffer_size,
        buffer_count,
        "configured bulk ingestion options"
    );

    BulkIngestionOptions {
        db_options,
        column_family_options,
        cf_memory_budget,
    }
}

/// Available memory in bytes, honoring cgroup limits in containerized
/// environments and falling back to total system memory.
fn available_memory_bytes() -> u64 {
    // Under the simulator sysinfo must run off the test thread, where the
    // intercepted clock and rng would make its probes a source of
    // non-determinism. No-op outside msim.
    nondeterministic!({
        // `RefreshKind::nothing().with_memory(..)` avoids collecting other,
        // slower stats.
        let mut sys = System::new_with_specifics(
            RefreshKind::nothing().with_memory(MemoryRefreshKind::everything()),
        );
        sys.refresh_memory();

        if let Some(cgroup_limits) = sys.cgroup_limits() {
            // `total_memory` is 0 when there is no cgroup limit.
            if cgroup_limits.total_memory > 0 {
                debug!(
                    limit = cgroup_limits.total_memory,
                    "using cgroup memory limit"
                );
                return cgroup_limits.total_memory;
            }
        }

        let total = sys.total_memory();
        debug!(total, "using total system memory");
        total
    })
}

#[derive(Clone)]
pub struct DBMapTableConfigMap(BTreeMap<String, DBOptions>);
impl DBMapTableConfigMap {
    pub fn new(map: BTreeMap<String, DBOptions>) -> Self {
        Self(map)
    }

    pub fn to_map(&self) -> BTreeMap<String, DBOptions> {
        self.0.clone()
    }
}

impl DBOptions {
    // Optimize lookup perf for tables where no scans are performed.
    // If non-trivial number of values can be > 512B in size, it is beneficial to
    // also specify optimize_for_large_values_no_scan().
    pub fn optimize_for_point_lookup(mut self, block_cache_size_mb: usize) -> DBOptions {
        // NOTE: this overwrites the block options.
        self.options
            .optimize_for_point_lookup(block_cache_size_mb as u64);
        self
    }

    // Optimize write and lookup perf for tables which are rarely scanned, and have
    // large values. https://rocksdb.org/blog/2021/05/26/integrated-blob-db.html
    pub fn optimize_for_large_values_no_scan(mut self, min_blob_size: u64) -> DBOptions {
        if env::var(ENV_VAR_DISABLE_BLOB_STORAGE).is_ok() {
            info!("Large value blob storage optimization is disabled via env var.");
            return self;
        }

        // Blob settings.
        self.options.set_enable_blob_files(true);
        self.options
            .set_blob_compression_type(rocksdb::DBCompressionType::Lz4);
        self.options.set_enable_blob_gc(true);
        // Since each blob can have non-trivial size overhead, and compression does not
        // work across blobs, set a min blob size in bytes to so small
        // transactions and effects are kept in sst files.
        self.options.set_min_blob_size(min_blob_size);

        // Increase write buffer size to 256MiB.
        let write_buffer_size = read_size_from_env(ENV_VAR_MAX_WRITE_BUFFER_SIZE_MB)
            .unwrap_or(DEFAULT_MAX_WRITE_BUFFER_SIZE_MB)
            * 1024
            * 1024;
        self.options.set_write_buffer_size(write_buffer_size);
        // Since large blobs are not in sst files, reduce the target file size and base
        // level target size.
        let target_file_size_base = 64 << 20;
        self.options
            .set_target_file_size_base(target_file_size_base);
        // Level 1 default to 64MiB * 4 ~ 256MiB.
        let max_level_zero_file_num = read_size_from_env(ENV_VAR_L0_NUM_FILES_COMPACTION_TRIGGER)
            .unwrap_or(DEFAULT_L0_NUM_FILES_COMPACTION_TRIGGER);
        self.options
            .set_max_bytes_for_level_base(target_file_size_base * max_level_zero_file_num as u64);

        self
    }

    // Optimize tables with a mix of lookup and scan workloads.
    pub fn optimize_for_read(mut self, block_cache_size_mb: usize) -> DBOptions {
        self.options
            .set_block_based_table_factory(&get_block_options(block_cache_size_mb, 16 << 10));
        self
    }

    // Optimize DB receiving significant insertions.
    pub fn optimize_db_for_write_throughput(mut self, db_max_write_buffer_gb: u64) -> DBOptions {
        self.options
            .set_db_write_buffer_size(db_max_write_buffer_gb as usize * 1024 * 1024 * 1024);
        self.options
            .set_max_total_wal_size(db_max_write_buffer_gb * 1024 * 1024 * 1024);
        self
    }

    // Optimize tables receiving significant insertions.
    pub fn optimize_for_write_throughput(mut self) -> DBOptions {
        // Increase write buffer size to 256MiB.
        let write_buffer_size = read_size_from_env(ENV_VAR_MAX_WRITE_BUFFER_SIZE_MB)
            .unwrap_or(DEFAULT_MAX_WRITE_BUFFER_SIZE_MB)
            * 1024
            * 1024;
        self.options.set_write_buffer_size(write_buffer_size);
        // Increase write buffers to keep to 6 before slowing down writes.
        let max_write_buffer_number = read_size_from_env(ENV_VAR_MAX_WRITE_BUFFER_NUMBER)
            .unwrap_or(DEFAULT_MAX_WRITE_BUFFER_NUMBER);
        self.options
            .set_max_write_buffer_number(max_write_buffer_number.try_into().unwrap());
        // Keep 1 write buffer so recent writes can be read from memory.
        self.options
            .set_max_write_buffer_size_to_maintain((write_buffer_size).try_into().unwrap());

        // Increase compaction trigger for level 0 to 6.
        let max_level_zero_file_num = read_size_from_env(ENV_VAR_L0_NUM_FILES_COMPACTION_TRIGGER)
            .unwrap_or(DEFAULT_L0_NUM_FILES_COMPACTION_TRIGGER);
        self.options.set_level_zero_file_num_compaction_trigger(
            max_level_zero_file_num.try_into().unwrap(),
        );
        self.options.set_level_zero_slowdown_writes_trigger(
            (max_level_zero_file_num * 12).try_into().unwrap(),
        );
        self.options
            .set_level_zero_stop_writes_trigger((max_level_zero_file_num * 16).try_into().unwrap());

        // Increase sst file size to 128MiB.
        self.options
            .set_target_file_size_base(target_file_size_base());

        // Increase level 1 target size to 256MiB * 6 ~ 1.5GiB.
        self.options
            .set_max_bytes_for_level_base((write_buffer_size * max_level_zero_file_num) as u64);

        self
    }

    // Optimize tables receiving significant insertions, without any deletions.
    // TODO: merge this function with optimize_for_write_throughput(), and use a
    // flag to indicate if deletion is received.
    pub fn optimize_for_write_throughput_no_deletion(mut self) -> DBOptions {
        // Increase write buffer size to 256MiB.
        let write_buffer_size = read_size_from_env(ENV_VAR_MAX_WRITE_BUFFER_SIZE_MB)
            .unwrap_or(DEFAULT_MAX_WRITE_BUFFER_SIZE_MB)
            * 1024
            * 1024;
        self.options.set_write_buffer_size(write_buffer_size);
        // Increase write buffers to keep to 6 before slowing down writes.
        let max_write_buffer_number = read_size_from_env(ENV_VAR_MAX_WRITE_BUFFER_NUMBER)
            .unwrap_or(DEFAULT_MAX_WRITE_BUFFER_NUMBER);
        self.options
            .set_max_write_buffer_number(max_write_buffer_number.try_into().unwrap());
        // Keep 1 write buffer so recent writes can be read from memory.
        self.options
            .set_max_write_buffer_size_to_maintain((write_buffer_size).try_into().unwrap());

        // Switch to universal compactions.
        self.options
            .set_compaction_style(rocksdb::DBCompactionStyle::Universal);
        let mut compaction_options = rocksdb::UniversalCompactOptions::default();
        compaction_options.set_max_size_amplification_percent(10000);
        compaction_options.set_stop_style(rocksdb::UniversalCompactionStopStyle::Similar);
        self.options
            .set_universal_compaction_options(&compaction_options);

        let max_level_zero_file_num = read_size_from_env(ENV_VAR_L0_NUM_FILES_COMPACTION_TRIGGER)
            .unwrap_or(DEFAULT_UNIVERSAL_COMPACTION_L0_NUM_FILES_COMPACTION_TRIGGER);
        self.options.set_level_zero_file_num_compaction_trigger(
            max_level_zero_file_num.try_into().unwrap(),
        );
        self.options.set_level_zero_slowdown_writes_trigger(
            (max_level_zero_file_num * 12).try_into().unwrap(),
        );
        self.options
            .set_level_zero_stop_writes_trigger((max_level_zero_file_num * 16).try_into().unwrap());

        // Increase sst file size to 128MiB.
        self.options
            .set_target_file_size_base(target_file_size_base());

        // This should be a no-op for universal compaction but increasing it to be safe.
        self.options
            .set_max_bytes_for_level_base((write_buffer_size * max_level_zero_file_num) as u64);

        self
    }

    // Overrides the block options with different block cache size and block size.
    pub fn set_block_options(
        mut self,
        block_cache_size_mb: usize,
        block_size_bytes: usize,
    ) -> DBOptions {
        self.options
            .set_block_based_table_factory(&get_block_options(
                block_cache_size_mb,
                block_size_bytes,
            ));
        self
    }

    // Disables write stalling and stopping based on pending compaction bytes.
    pub fn disable_write_throttling(mut self) -> DBOptions {
        self.options.set_soft_pending_compaction_bytes_limit(0);
        self.options.set_hard_pending_compaction_bytes_limit(0);
        self.options.set_level_zero_slowdown_writes_trigger(512);
        self.options.set_level_zero_stop_writes_trigger(1024);
        self
    }
}

/// Creates a default RocksDB option, to be used when RocksDB option is
/// unspecified.
pub fn default_db_options() -> DBOptions {
    let mut opt = base_db_options();

    // One common issue when running tests on Mac is that the default ulimit is too
    // low, leading to I/O errors such as "Too many open files". Raising fdlimit
    // to bypass it.
    if let Some(limit) = fdlimit::raise_fd_limit() {
        // on windows raise_fd_limit return None
        opt.set_max_open_files((limit / 8) as i32);
    }

    // The table cache is locked for updates and this determines the number
    // of shards, ie 2^10. Increase in case of lock contentions.
    opt.set_table_cache_num_shard_bits(10);

    // LSM compression settings
    opt.set_compression_type(rocksdb::DBCompressionType::Lz4);
    opt.set_bottommost_compression_type(rocksdb::DBCompressionType::Zstd);
    opt.set_bottommost_zstd_max_train_bytes(1024 * 1024, true);

    // IOTA uses multiple RocksDB in a node, so total sizes of write buffers and WAL
    // can be higher than the limits below.
    //
    // RocksDB also exposes the option to configure total write buffer size across
    // multiple instances via `write_buffer_manager`. But the write buffer flush
    // policy (flushing the buffer receiving the next write) may not work well.
    // So sticking to per-db write buffer size limit for now.
    //
    // The environment variables are only meant to be emergency overrides. They may
    // go away in future. It is preferable to update the default value, or
    // override the option in code.
    opt.set_db_write_buffer_size(
        read_size_from_env(ENV_VAR_DB_WRITE_BUFFER_SIZE).unwrap_or(DEFAULT_DB_WRITE_BUFFER_SIZE)
            * 1024
            * 1024,
    );
    opt.set_max_total_wal_size(
        read_size_from_env(ENV_VAR_DB_WAL_SIZE).unwrap_or(DEFAULT_DB_WAL_SIZE) as u64 * 1024 * 1024,
    );

    // Num threads for compactions and memtable flushes.
    opt.increase_parallelism(read_size_from_env(ENV_VAR_DB_PARALLELISM).unwrap_or(8) as i32);

    opt.set_enable_pipelined_write(true);

    // Increase block size to 16KiB.
    // https://github.com/EighteenZi/rocksdb_wiki/blob/master/Memory-usage-in-RocksDB.md#indexes-and-filter-blocks
    opt.set_block_based_table_factory(&get_block_options(128, 16 << 10));

    // Set memtable bloomfilter.
    opt.set_memtable_prefix_bloom_ratio(0.02);

    DBOptions {
        options: opt,
        rw_options: ReadWriteOptions::default(),
    }
}

fn get_block_options(block_cache_size_mb: usize, block_size_bytes: usize) -> BlockBasedOptions {
    // Set options mostly similar to those used in optimize_for_point_lookup(),
    // except non-default binary and hash index, to hopefully reduce lookup
    // latencies without causing any regression for scanning, with slightly more
    // memory usages. https://github.com/facebook/rocksdb/blob/11cb6af6e5009c51794641905ca40ce5beec7fee/options/options.cc#L611-L621
    let mut block_options = BlockBasedOptions::default();
    // Overrides block size.
    block_options.set_block_size(block_size_bytes);
    // Configure a block cache.
    block_options.set_block_cache(&Cache::new_lru_cache(block_cache_size_mb << 20));
    // Set a bloomfilter with 1% false positive rate.
    block_options.set_bloom_filter(10.0, false);
    // From https://github.com/EighteenZi/rocksdb_wiki/blob/master/Block-Cache.md#caching-index-and-filter-blocks
    block_options.set_pin_l0_filter_and_index_blocks_in_cache(true);
    block_options
}

pub fn list_tables(path: std::path::PathBuf) -> eyre::Result<Vec<String>> {
    const DB_DEFAULT_CF_NAME: &str = "default";

    let opts = rocksdb::Options::default();
    rocksdb::DBWithThreadMode::<rocksdb::MultiThreaded>::list_cf(&opts, path)
        .map_err(|e| e.into())
        .map(|q| {
            q.iter()
                .filter_map(|s| {
                    // The `default` table is not used
                    if s != DB_DEFAULT_CF_NAME {
                        Some(s.clone())
                    } else {
                        None
                    }
                })
                .collect()
        })
}

/// The size, in bytes, at which compactions under
/// [`DBOptions::optimize_for_write_throughput`] and
/// [`DBOptions::optimize_for_write_throughput_no_deletion`] cut their output
/// files.
pub fn target_file_size_base() -> u64 {
    read_size_from_env(ENV_VAR_TARGET_FILE_SIZE_BASE_MB).unwrap_or(DEFAULT_TARGET_FILE_SIZE_BASE_MB)
        as u64
        * 1024
        * 1024
}

pub fn read_size_from_env(var_name: &str) -> Option<usize> {
    env::var(var_name)
        .ok()?
        .parse::<usize>()
        .tap_err(|e| {
            warn!(
                "Env var {} does not contain valid usize integer: {}",
                var_name, e
            )
        })
        .ok()
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The memtable budget is a fixed share of the measured memory, with a
    /// safe fallback when the probe fails.
    #[test]
    fn write_buffer_budget_scales_with_available_memory() {
        const TOTAL: u64 = 64 << 30;

        let (db_one, cf_one) = write_buffer_budget(TOTAL);
        assert_eq!(db_one, (TOTAL as f64 * 0.8) as usize);
        assert_eq!(cf_one, (TOTAL as f64 * 0.25) as usize);

        // A failed memory probe must not turn the whole-database cap into
        // RocksDB's unlimited.
        let (db_unknown, cf_unknown) = write_buffer_budget(0);
        assert_eq!(write_buffer_budget(4 << 30), (db_unknown, cf_unknown));
        assert!(db_unknown > 0 && cf_unknown > 0);

        // A measured limit below the fallback is used as is: budgeting above
        // a cgroup limit would get the process killed.
        const SMALL: u64 = 2 << 30;
        let (db_small, cf_small) = write_buffer_budget(SMALL);
        assert_eq!(db_small, (SMALL as f64 * 0.8) as usize);
        assert_eq!(cf_small, (SMALL as f64 * 0.25) as usize);
        assert!(db_small < db_unknown);
    }
}