sov-db: add offline NOMT hashtable-bucket migration utility

CHANGELOG.md

@@ -1,3 +1,6 @@
+# 2026-06-18
+- #2988 Adds `sov_db::migration` (`rebuild_kernel`/`rebuild_user`) and the `migrate_nomt_buckets` example: an offline, root-verified rebuild of a namespace's NOMT with a different `hashtable_buckets`, reading the bucket-count-independent flat state instead of replaying DA. Gives operators an actionable answer to the `kernel_state` "Occupancy rate for NOMT hashtable is too high. Please update buckets size and resync database" warning when the kernel keyspace outgrows the fixed page table. Additive; no behavior change and no new dependencies.
+
 # 2026-06-12
 - #2974 Adds in-process aggregation of JSON-RPC call statistics, covering every call on both HTTP and WebSocket. New `monitoring.rpc_aggregation` config section (`flush_interval_secs`, `slow_call_threshold_ms`, `max_slow_call_points_per_window`; defaults apply when omitted). Two new measurements: `sov_rollup_rpc_aggregated` (one point per `(method, transport)` per flush window with call/error/batch/cancellation/notification counters and cumulative latency buckets) and `sov_rollup_rpc_slow_calls` (individual points for calls over the slow threshold, capped per window, with `status` and `cancelled` tags). Calls whose future is dropped mid-flight (client disconnect/timeout) are recorded as `cancelled`. Recording is lock-free on the hot path (bounded channel + atomics).
   * **Breaking Change** `sov_stf_runner::rpc_module_to_router` now takes an `RpcAggregationConfig` parameter and additionally returns the `Arc<RpcStatsAggregator>`; callers must drive `RpcStatsAggregator::run_flush_loop` (the SDK's `start_http_server` does this and registers the task) for aggregated metrics to be emitted.

crates/full-node/sov-db/examples/migrate_nomt_buckets.rs

@@ -0,0 +1,81 @@
+//! Offline tool to rebuild a namespace's NOMT page-table with a new
+//! `hashtable_buckets`, without replaying DA. Run with the node **stopped**.
+//!
+//! ```text
+//! cargo run -p sov-db --example migrate_nomt_buckets -- \
+//!     --data-dir /path/to/rollup/data \
+//!     --namespace kernel \
+//!     --source-buckets 256000 \
+//!     --new-buckets 4000000 \
+//!     --out /path/to/rollup/data/kernel_nomt_db.new
+//! ```
+//!
+//! The rebuilt root is verified equal to the source root before anything is
+//! persisted. On success, stop the node, swap `--out` in for the live
+//! `<namespace>_nomt_db`, set the matching `*_hashtable_buckets` in the config to
+//! `--new-buckets`, and restart.
+//!
+//! NOTE: the hasher is fixed to `sha2::Sha256` here, which matches the default
+//! `CryptoSpec::Hasher`. If your rollup uses a different state hasher, adjust it.
+
+use std::path::PathBuf;
+
+use sha2::Sha256;
+use sov_db::migration::{rebuild_kernel, rebuild_user};
+
+fn arg(args: &[String], flag: &str) -> Option<String> {
+    args.iter()
+        .position(|a| a == flag)
+        .and_then(|i| args.get(i + 1).cloned())
+}
+
+fn main() -> anyhow::Result<()> {
+    let args: Vec<String> = std::env::args().collect();
+
+    let data_dir = PathBuf::from(
+        arg(&args, "--data-dir").ok_or_else(|| anyhow::anyhow!("missing --data-dir"))?,
+    );
+    let namespace = arg(&args, "--namespace")
+        .ok_or_else(|| anyhow::anyhow!("missing --namespace (kernel|user)"))?;
+    let new_buckets: u32 = arg(&args, "--new-buckets")
+        .ok_or_else(|| anyhow::anyhow!("missing --new-buckets"))?
+        .parse()?;
+    let out = PathBuf::from(arg(&args, "--out").ok_or_else(|| anyhow::anyhow!("missing --out"))?);
+    // The bucket count the existing DB was created with. Defaults to the sov-db
+    // kernel default (256_000); pass explicitly for the user namespace.
+    let source_buckets: u32 = arg(&args, "--source-buckets")
+        .map(|s| s.parse())
+        .transpose()?
+        .unwrap_or(256_000);
+    let commit_concurrency: usize = arg(&args, "--commit-concurrency")
+        .map(|s| s.parse())
+        .transpose()?
+        .unwrap_or(2);
+
+    let report = match namespace.as_str() {
+        "kernel" => rebuild_kernel::<Sha256>(
+            &data_dir,
+            source_buckets,
+            new_buckets,
+            &out,
+            commit_concurrency,
+        )?,
+        "user" => rebuild_user::<Sha256>(
+            &data_dir,
+            source_buckets,
+            new_buckets,
+            &out,
+            commit_concurrency,
+        )?,
+        other => anyhow::bail!("unknown --namespace `{other}` (expected kernel|user)"),
+    };
+
+    println!(
+        "migrated {} entries into {} buckets; verified root {} — wrote {}",
+        report.entries,
+        report.new_buckets,
+        report.root,
+        out.display()
+    );
+    Ok(())
+}

crates/full-node/sov-db/src/lib.rs

@@ -33,6 +33,7 @@ pub mod namespaces;
 /// Configuration for `sov-db`
 pub mod config;
 pub(crate) mod metrics;
+pub mod migration;
 /// Implements state pruning functionality for removing old versions of keys.
 pub mod pruner;
 /// Implements a wrapper around [NOMT](https://github.com/thrumdev/nomt) meant for storing rollup state.

crates/full-node/sov-db/src/migration.rs

@@ -0,0 +1,184 @@
+//! Offline migration of a namespace's NOMT page-table (`hashtable_buckets`) to a
+//! different bucket count, **without replaying DA**.
+//!
+//! `hashtable_buckets` is fixed at DB creation and cannot be changed in place
+//! (`nomt` has no in-place resize), yet the kernel keyspace grows over a chain's
+//! lifetime, so a long-running node can outgrow its page table. The only documented
+//! remedy ("resync database") means replaying DA, which is impossible for a node
+//! without archival DA.
+//!
+//! This rebuilds the trie purely from the **flat** state store, which is independent
+//! of the bucket count: it enumerates `(key, value)` from the namespace's flat column
+//! family, re-derives each authenticated leaf, and commits them into a fresh NOMT
+//! opened with the new bucket count. The rebuilt root is asserted equal to the source
+//! root **before anything is persisted**, so a faulty reconstruction aborts instead of
+//! corrupting state.
+//!
+//! Run it with the node **stopped** (so the flat store and the source NOMT are
+//! consistent), then swap the rebuilt directory in and set `kernel_hashtable_buckets`
+//! / `user_hashtable_buckets` in the config to the new value before restarting.
+
+use std::path::Path;
+
+use nomt::hasher::BinaryHasher;
+use nomt::trie::KeyPath;
+use nomt::{KeyReadWrite, Nomt, Options, SessionParams};
+use sov_rollup_interface::reexports::digest::{self, Digest};
+
+use crate::namespaces::{KernelNamespace, Namespace, UserNamespace};
+
+/// Directory suffix of the flat state DB (mirrors `flat_db.rs` `DB_PATH_SUFFIX`).
+const FLAT_STATE_DB_SUFFIX: &str = "state-db";
+/// NOMT directory for the kernel namespace (mirrors `config.rs`).
+const KERNEL_NOMT_DIR: &str = "kernel_nomt_db";
+/// NOMT directory for the user namespace (mirrors `config.rs`).
+const USER_NOMT_DIR: &str = "user_nomt_db";
+
+/// Outcome of a successful migration run.
+#[derive(Debug)]
+pub struct MigrationReport {
+    /// Number of flat entries migrated.
+    pub entries: u64,
+    /// Bucket count of the rebuilt page table.
+    pub new_buckets: u32,
+    /// The (matching) state root, as `nomt` prints it.
+    pub root: String,
+}
+
+/// Re-derive the authenticated leaf value committed into the trie for a flat value.
+///
+/// Mirrors `sov_state`'s `NodeLeaf::make_leaf` + `combine_val_hash_and_size`: the
+/// 32-byte value hash followed by the value length as a little-endian `u32`.
+/// Correctness is not assumed — it is enforced by the root-equality gate in
+/// [`rebuild`].
+fn leaf_commitment<H: Digest<OutputSize = digest::typenum::U32>>(value: &[u8]) -> Vec<u8> {
+    let mut out = H::digest(value).to_vec();
+    out.extend_from_slice(&(value.len() as u32).to_le_bytes());
+    out
+}
+
+fn make_options(path: &Path, buckets: u32, commit_concurrency: usize) -> Options {
+    let mut o = Options::new();
+    o.metrics(true);
+    o.rollback(true);
+    o.max_rollback_log_len(1);
+    o.commit_concurrency(commit_concurrency);
+    o.hashtable_buckets(buckets);
+    o.path(path.to_path_buf());
+    o
+}
+
+/// Rebuild the **kernel** namespace NOMT with `new_buckets` into `out_dir`.
+///
+/// `data_dir` is the rollup storage path (the directory that contains
+/// `kernel_nomt_db` and `state-db`). `source_buckets` is the bucket count the
+/// existing DB was created with (the default is `256_000` if it was never set).
+pub fn rebuild_kernel<H>(
+    data_dir: &Path,
+    source_buckets: u32,
+    new_buckets: u32,
+    out_dir: &Path,
+    commit_concurrency: usize,
+) -> anyhow::Result<MigrationReport>
+where
+    H: Digest<OutputSize = digest::typenum::U32> + Send + Sync + 'static,
+{
+    rebuild::<H>(
+        data_dir,
+        KERNEL_NOMT_DIR,
+        KernelNamespace::STATE_VALUES_TABLE_NAME,
+        source_buckets,
+        new_buckets,
+        out_dir,
+        commit_concurrency,
+    )
+}
+
+/// Rebuild the **user** namespace NOMT with `new_buckets` into `out_dir`.
+pub fn rebuild_user<H>(
+    data_dir: &Path,
+    source_buckets: u32,
+    new_buckets: u32,
+    out_dir: &Path,
+    commit_concurrency: usize,
+) -> anyhow::Result<MigrationReport>
+where
+    H: Digest<OutputSize = digest::typenum::U32> + Send + Sync + 'static,
+{
+    rebuild::<H>(
+        data_dir,
+        USER_NOMT_DIR,
+        UserNamespace::STATE_VALUES_TABLE_NAME,
+        source_buckets,
+        new_buckets,
+        out_dir,
+        commit_concurrency,
+    )
+}
+
+#[allow(clippy::too_many_arguments)]
+fn rebuild<H>(
+    data_dir: &Path,
+    nomt_dir: &str,
+    cf_name: &str,
+    source_buckets: u32,
+    new_buckets: u32,
+    out_dir: &Path,
+    commit_concurrency: usize,
+) -> anyhow::Result<MigrationReport>
+where
+    H: Digest<OutputSize = digest::typenum::U32> + Send + Sync + 'static,
+{
+    // 1. Source root (the value the rebuild must reproduce).
+    let source = Nomt::<BinaryHasher<H>>::open(make_options(
+        &data_dir.join(nomt_dir),
+        source_buckets,
+        commit_concurrency,
+    ))?;
+    let source_root = format!("{:?}", source.root());
+
+    // 2. Enumerate the namespace's flat values (independent of bucket count). The flat
+    //    key/value bytes are the canonical `SlotKey`/`SlotValue` encodings.
+    let statedb = data_dir.join(FLAT_STATE_DB_SUFFIX);
+    let cfs = rockbound::rocksdb::DB::list_cf(&rockbound::rocksdb::Options::default(), &statedb)?;
+    let db = rockbound::rocksdb::DB::open_cf_for_read_only(
+        &rockbound::rocksdb::Options::default(),
+        &statedb,
+        &cfs,
+        false,
+    )?;
+    let cf = db
+        .cf_handle(cf_name)
+        .ok_or_else(|| anyhow::anyhow!("flat column family `{cf_name}` not found"))?;
+
+    let mut writes: Vec<(KeyPath, KeyReadWrite)> = Vec::new();
+    for item in db.iterator_cf(&cf, rockbound::rocksdb::IteratorMode::Start) {
+        let (k, v) = item?;
+        let key_path: KeyPath = H::digest(&k).into();
+        writes.push((
+            key_path,
+            KeyReadWrite::Write(Some(leaf_commitment::<H>(&v))),
+        ));
+    }
+    let entries = writes.len() as u64;
+    writes.sort_by(|a, b| a.0.cmp(&b.0));
+    writes.dedup_by(|a, b| a.0 == b.0);
+
+    // 3. Commit into a fresh NOMT with the new bucket count and verify the root.
+    let out =
+        Nomt::<BinaryHasher<H>>::open(make_options(out_dir, new_buckets, commit_concurrency))?;
+    let session = out.begin_session(SessionParams::default());
+    let finished = session.finish(writes)?;
+    let new_root = format!("{:?}", finished.root());
+    anyhow::ensure!(
+        new_root == source_root,
+        "root mismatch — refusing to persist. source={source_root} rebuilt={new_root}"
+    );
+    finished.commit(&out)?;
+
+    Ok(MigrationReport {
+        entries,
+        new_buckets,
+        root: new_root,
+    })
+}