Make DatabaseKind the single source of truth for RocksDB database names

tillrohrmann · tillrohrmann · commit aa1749ddee9a · 2026-03-27T11:56:43.000+01:00
Move DatabaseKind from node_ctl_svc.proto to common.proto so it is
available in restate-types. Add DatabaseKind::db_name() as the canonical
mapping from kind to RocksDB database name, and store the kind on DbSpec
so that RocksDb::kind() can be queried directly without reverse-
engineering the kind from the name string.

This eliminates the scattered DB_NAME constants from individual crates
and removes the fragile db_name_to_kind() function that could silently
go out of sync if a database was renamed.
diff --git a/crates/bifrost/src/providers/local_loglet/log_store.rs b/crates/bifrost/src/providers/local_loglet/log_store.rs
@@ -19,15 +19,13 @@ use restate_rocksdb::{
 use restate_types::config::{Configuration, LocalLogletOptions};
 use restate_types::errors::MaybeRetryableError;
 use restate_types::live::LiveLoad;
+use restate_types::protobuf::common::DatabaseKind;
 use restate_types::storage::{StorageDecodeError, StorageEncodeError};
 
 use super::keys::{DATA_KEY_PREFIX_LENGTH, MetadataKey, MetadataKind};
 use super::log_state::{LogState, log_state_full_merge, log_state_partial_merge};
 use super::log_store_writer::LogStoreWriter;
 
-// matches the default directory name
-pub(crate) const DB_NAME: &str = "local-loglet";
-
 pub(crate) const DATA_CF: &str = "logstore_data";
 pub(crate) const METADATA_CF: &str = "logstore_metadata";
 
@@ -74,15 +72,21 @@ impl RocksDbLogStore {
         let opts = options.live_load();
         let data_dir = opts.data_dir();
 
-        let db_spec = DbSpecBuilder::new(DbName::new(DB_NAME), data_dir, RocksConfigurator)
-            .add_cf_pattern(CfExactPattern::new(DATA_CF), RocksConfigurator)
-            .add_cf_pattern(CfExactPattern::new(METADATA_CF), RocksConfigurator)
-            // not very important but it's to reduce the number of merges by flushing.
-            // it's also a small cf so it should be quick.
-            .add_to_flush_on_shutdown(CfExactPattern::new(METADATA_CF))
-            .ensure_column_families(cfs)
-            .build()
-            .expect("valid spec");
+        let kind = DatabaseKind::LocalLoglet;
+        let db_spec = DbSpecBuilder::new(
+            DbName::new(kind.db_name()),
+            kind,
+            data_dir,
+            RocksConfigurator,
+        )
+        .add_cf_pattern(CfExactPattern::new(DATA_CF), RocksConfigurator)
+        .add_cf_pattern(CfExactPattern::new(METADATA_CF), RocksConfigurator)
+        // not very important but it's to reduce the number of merges by flushing.
+        // it's also a small cf so it should be quick.
+        .add_to_flush_on_shutdown(CfExactPattern::new(METADATA_CF))
+        .ensure_column_families(cfs)
+        .build()
+        .expect("valid spec");
         let rocksdb = db_manager.open_db(db_spec).await?;
         Ok(Self { rocksdb })
     }
diff --git a/crates/core/protobuf/node_ctl_svc.proto b/crates/core/protobuf/node_ctl_svc.proto
@@ -113,16 +113,8 @@ message EmbeddedMetadataClusterHealth {
   repeated restate.common.NodeId members = 1;
 }
 
-enum DatabaseKind {
-  DATABASE_KIND_UNSPECIFIED = 0;
-  DATABASE_KIND_PARTITION_STORE = 2;
-  DATABASE_KIND_LOG_SERVER = 3;
-  DATABASE_KIND_METADATA_SERVER = 4;
-  DATABASE_KIND_LOCAL_LOGLET = 5;
-}
-
 message TriggerCompactionRequest {
-  repeated DatabaseKind databases = 1;
+  repeated restate.common.DatabaseKind databases = 1;
 }
 
 message TriggerCompactionResponse {
diff --git a/crates/log-server/src/rocksdb_logstore/builder.rs b/crates/log-server/src/rocksdb_logstore/builder.rs
@@ -20,10 +20,10 @@ use restate_rocksdb::{
 use restate_serde_util::ByteCount;
 use restate_types::config::{Configuration, LogServerOptions};
 use restate_types::health::HealthStatus;
-use restate_types::protobuf::common::LogServerStatus;
+use restate_types::protobuf::common::{DatabaseKind, LogServerStatus};
 
 use super::writer::LogStoreWriterBuilder;
-use super::{DATA_CF, DB_NAME, METADATA_CF};
+use super::{DATA_CF, METADATA_CF};
 use super::{RocksDbLogStore, RocksDbLogStoreError};
 use crate::logstore::LogStoreState;
 use crate::rocksdb_logstore::keys::KeyPrefix;
@@ -36,12 +36,14 @@ pub struct RocksDbLogStoreBuilder {
 
 impl RocksDbLogStoreBuilder {
     pub async fn create() -> Result<Self, RocksDbLogStoreError> {
-        let db_name = DbName::new(DB_NAME);
+        let kind = DatabaseKind::LogServer;
+        let db_name = DbName::new(kind.db_name());
         let db_manager = RocksDbManager::get();
         let cfs = vec![CfName::new(DATA_CF), CfName::new(METADATA_CF)];
 
         let db_spec = DbSpecBuilder::new(
             db_name,
+            kind,
             Configuration::pinned().log_server.data_dir(),
             RocksConfigurator,
         )
diff --git a/crates/log-server/src/rocksdb_logstore/mod.rs b/crates/log-server/src/rocksdb_logstore/mod.rs
@@ -29,7 +29,5 @@ pub use self::builder::RocksDbLogStoreBuilder;
 pub use self::store::RocksDbLogStore;
 pub(crate) use error::*;
 
-// matches the default directory name
-const DB_NAME: &str = "log-server";
 pub const DATA_CF: &str = "data";
 pub const METADATA_CF: &str = "metadata";
diff --git a/crates/metadata-server/src/raft/storage/mod.rs b/crates/metadata-server/src/raft/storage/mod.rs
@@ -15,6 +15,5 @@ mod rocksdb_builder;
 pub use rocksdb::{BuildError, Error, RocksDbStorage};
 
 const DATA_DIR: &str = "replicated-metadata-server";
-const DB_NAME: &str = "replicated-metadata-server";
 const DATA_CF: &str = "data";
 const METADATA_CF: &str = "metadata";
diff --git a/crates/metadata-server/src/raft/storage/rocksdb_builder.rs b/crates/metadata-server/src/raft/storage/rocksdb_builder.rs
@@ -18,19 +18,21 @@ use restate_rocksdb::{
     RocksError,
 };
 use restate_types::config::{Configuration, MetadataServerOptions, data_dir};
+use restate_types::protobuf::common::DatabaseKind;
 
-use crate::raft::storage::{DATA_CF, DATA_DIR, DB_NAME, METADATA_CF};
+use crate::raft::storage::{DATA_CF, DATA_DIR, METADATA_CF};
 
 const DATA_CF_BUDGET_RATIO: f64 = 0.85;
 const_assert!(DATA_CF_BUDGET_RATIO < 1.0);
 
 pub async fn build_rocksdb() -> Result<Arc<RocksDb>, RocksError> {
+    let kind = DatabaseKind::MetadataServer;
     let data_dir = data_dir(DATA_DIR);
-    let db_name = DbName::new(DB_NAME);
+    let db_name = DbName::new(kind.db_name());
     let db_manager = RocksDbManager::get();
     let cfs = vec![CfName::new(DATA_CF), CfName::new(METADATA_CF)];
 
-    let db_spec = DbSpecBuilder::new(db_name, data_dir, RocksConfigurator)
+    let db_spec = DbSpecBuilder::new(db_name, kind, data_dir, RocksConfigurator)
         .add_cf_pattern(CfPrefixPattern::new(DATA_CF), RocksConfigurator)
         .add_cf_pattern(CfPrefixPattern::new(METADATA_CF), RocksConfigurator)
         // not very important but it's to reduce the number of merges by flushing.
diff --git a/crates/node/src/network_server/grpc_svc_handler.rs b/crates/node/src/network_server/grpc_svc_handler.rs
@@ -28,7 +28,7 @@ use restate_types::net::connect_opts::GrpcConnectionOptions;
 use restate_core::network::net_util::{DNSResolution, create_tonic_channel};
 use restate_core::protobuf::node_ctl_svc::node_ctl_svc_server::{NodeCtlSvc, NodeCtlSvcServer};
 use restate_core::protobuf::node_ctl_svc::{
-    ClusterHealthResponse, DatabaseCompactionResult, DatabaseKind, EmbeddedMetadataClusterHealth,
+    ClusterHealthResponse, DatabaseCompactionResult, EmbeddedMetadataClusterHealth,
     GetMetadataRequest, GetMetadataResponse, IdentResponse, ProvisionClusterRequest,
     ProvisionClusterResponse, TriggerCompactionRequest, TriggerCompactionResponse,
 };
@@ -43,28 +43,12 @@ use restate_types::logs::metadata::{NodeSetSize, ProviderConfiguration};
 use restate_types::metadata::VersionedValue;
 use restate_types::nodes_config::Role;
 use restate_types::protobuf::cluster::ClusterConfiguration as ProtoClusterConfiguration;
+use restate_types::protobuf::common::DatabaseKind;
 use restate_types::replication::ReplicationProperty;
 use restate_types::storage::StorageCodec;
 
 use crate::{ClusterConfiguration, provision_cluster_metadata};
 
-/// Maps database names to their corresponding DatabaseKind.
-/// Returns None for unrecognized names so they are skipped rather than
-/// accidentally compacted if new databases are added without updating this mapping.
-fn db_name_to_kind(name: &str) -> Option<DatabaseKind> {
-    if name == "log-server" {
-        Some(DatabaseKind::LogServer)
-    } else if name == "replicated-metadata-server" {
-        Some(DatabaseKind::MetadataServer)
-    } else if name == "local-loglet" {
-        Some(DatabaseKind::LocalLoglet)
-    } else if name == "db" || name.starts_with("db-") {
-        Some(DatabaseKind::PartitionStore)
-    } else {
-        None
-    }
-}
-
 pub struct NodeCtlSvcHandler {
     metadata_writer: MetadataWriter,
 }
@@ -306,12 +290,10 @@ impl NodeCtlSvc for NodeCtlSvcHandler {
         // concurrent I/O from multiple databases.
         for db in all_dbs {
             let db_name = db.name().to_string();
+            let kind = db.kind();
 
-            // Check if this database should be compacted. Skip databases whose names
-            // are not recognized so newly added databases are not accidentally compacted.
-            let should_compact = compact_all || {
-                db_name_to_kind(&db_name).is_some_and(|kind| requested_kinds.contains(&kind))
-            };
+            let should_compact = compact_all
+                || (kind != DatabaseKind::Unspecified && requested_kinds.contains(&kind));
 
             if !should_compact {
                 continue;
@@ -471,27 +453,16 @@ fn write_err_to_status(err: WriteError) -> Status {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use restate_types::protobuf::common::DatabaseKind;
 
     #[test]
-    fn test_db_name_to_kind() {
-        assert_eq!(db_name_to_kind("db"), Some(DatabaseKind::PartitionStore));
-        assert_eq!(db_name_to_kind("db-0"), Some(DatabaseKind::PartitionStore));
-        assert_eq!(
-            db_name_to_kind("db-123"),
-            Some(DatabaseKind::PartitionStore)
-        );
-        assert_eq!(db_name_to_kind("log-server"), Some(DatabaseKind::LogServer));
-        assert_eq!(
-            db_name_to_kind("replicated-metadata-server"),
-            Some(DatabaseKind::MetadataServer)
-        );
+    fn test_database_kind_db_names() {
+        assert_eq!(DatabaseKind::LogServer.db_name(), "log-server");
         assert_eq!(
-            db_name_to_kind("local-loglet"),
-            Some(DatabaseKind::LocalLoglet)
+            DatabaseKind::MetadataServer.db_name(),
+            "replicated-metadata-server"
         );
-        // Unknown names return None so they are safely skipped
-        assert_eq!(db_name_to_kind("unknown"), None);
-        assert_eq!(db_name_to_kind("random-name"), None);
+        assert_eq!(DatabaseKind::LocalLoglet.db_name(), "local-loglet");
+        assert_eq!(DatabaseKind::PartitionStore.db_name(), "db");
     }
 }
diff --git a/crates/partition-store/src/partition_store_manager.rs b/crates/partition-store/src/partition_store_manager.rs
@@ -22,6 +22,7 @@ use restate_types::config::Configuration;
 use restate_types::identifiers::{PartitionId, SnapshotId};
 use restate_types::logs::{Lsn, SequenceNumber};
 use restate_types::partitions::Partition;
+use restate_types::protobuf::common::DatabaseKind;
 
 use crate::SnapshotError;
 use crate::memory::MemoryController;
@@ -164,6 +165,7 @@ impl PartitionStoreManager {
 
         let db_spec = DbSpecBuilder::new(
             db_name.clone(),
+            DatabaseKind::PartitionStore,
             Configuration::pinned().worker.storage.data_dir(&db_name),
             configurator.clone(),
         )
diff --git a/crates/rocksdb/src/db_spec.rs b/crates/rocksdb/src/db_spec.rs
@@ -12,6 +12,7 @@ use std::path::PathBuf;
 
 use derive_builder::Builder;
 
+use restate_types::protobuf::common::DatabaseKind;
 use restate_util_string::ReString;
 
 use crate::BoxedCfMatcher;
@@ -145,6 +146,7 @@ pub enum OpenMode {
 #[builder(pattern = "owned", build_fn(name = "build"))]
 pub struct DbSpec {
     pub(crate) name: DbName,
+    pub(crate) kind: DatabaseKind,
     pub(crate) path: PathBuf,
     /// All column families that should be flushed on shutdown, no flush will be performed if empty
     /// which should be the default for most cases.
@@ -174,6 +176,10 @@ impl DbSpec {
         &self.name
     }
 
+    pub fn kind(&self) -> DatabaseKind {
+        self.kind
+    }
+
     pub fn open_mode(&self) -> OpenMode {
         self.db_configurator.get_db_open_mode()
     }
@@ -182,11 +188,13 @@ impl DbSpec {
 impl DbSpecBuilder {
     pub fn new(
         name: DbName,
+        kind: DatabaseKind,
         path: PathBuf,
         configurator: impl DbConfigurator + Send + Sync + 'static,
     ) -> DbSpecBuilder {
         Self {
             name: Some(name),
+            kind: Some(kind),
             path: Some(path),
             db_configurator: Some(Box::new(configurator)),
             ..Self::default()
diff --git a/crates/rocksdb/src/lib.rs b/crates/rocksdb/src/lib.rs
@@ -35,6 +35,7 @@ use rocksdb::statistics::HistogramData;
 use rocksdb::statistics::Ticker;
 
 use restate_core::ShutdownError;
+use restate_types::protobuf::common::DatabaseKind;
 
 // re-exports
 pub use self::db_manager::RocksDbManager;
@@ -112,6 +113,10 @@ impl RocksDb {
         &self.db.spec().name
     }
 
+    pub fn kind(&self) -> DatabaseKind {
+        self.db.spec().kind
+    }
+
     pub(crate) async fn open(
         manager: &'static RocksDbManager,
         spec: DbSpec,
diff --git a/crates/types/protobuf/restate/common.proto b/crates/types/protobuf/restate/common.proto
@@ -129,3 +129,11 @@ enum MetadataKind {
   PARTITION_TABLE = 3;
   LOGS = 4;
 }
+
+enum DatabaseKind {
+  DATABASE_KIND_UNSPECIFIED = 0;
+  DATABASE_KIND_PARTITION_STORE = 2;
+  DATABASE_KIND_LOG_SERVER = 3;
+  DATABASE_KIND_METADATA_SERVER = 4;
+  DATABASE_KIND_LOCAL_LOGLET = 5;
+}
diff --git a/crates/types/src/partition_table.rs b/crates/types/src/partition_table.rs
@@ -19,6 +19,7 @@ use crate::identifiers::{PartitionId, PartitionKey};
 use crate::logs::LogId;
 use crate::metadata::GlobalMetadata;
 use crate::net::metadata::{MetadataContainer, MetadataKind};
+use crate::protobuf::common::DatabaseKind;
 use crate::replication::ReplicationProperty;
 use crate::{Version, Versioned, flexbuffers_storage_encode_decode};
 
@@ -298,13 +299,14 @@ impl Partition {
     }
 
     pub fn db_name(&self) -> DbName {
+        let base = DatabaseKind::PartitionStore.db_name();
         #[cfg(feature = "multi-db")]
         return self
             .db_name
             .clone()
-            .unwrap_or_else(|| DbName::from(format!("db-{}", self.partition_id)));
+            .unwrap_or_else(|| DbName::from(format!("{base}-{}", self.partition_id)));
         #[cfg(not(feature = "multi-db"))]
-        return self.db_name.clone().unwrap_or_else(|| DbName::from("db"));
+        return self.db_name.clone().unwrap_or_else(|| DbName::from(base));
     }
 
     pub fn cf_name(&self) -> CfName {
diff --git a/crates/types/src/protobuf.rs b/crates/types/src/protobuf.rs
@@ -80,6 +80,23 @@ pub mod common {
             false
         }
     }
+
+    impl DatabaseKind {
+        /// The canonical RocksDB database name for this kind.
+        ///
+        /// For [`DatabaseKind::PartitionStore`], this returns the base prefix `"db"`.
+        /// Individual partition databases may be named `"db-{partition_id}"` in
+        /// multi-db mode.
+        pub fn db_name(&self) -> &'static str {
+            match self {
+                DatabaseKind::LogServer => "log-server",
+                DatabaseKind::MetadataServer => "replicated-metadata-server",
+                DatabaseKind::LocalLoglet => "local-loglet",
+                DatabaseKind::PartitionStore => "db",
+                DatabaseKind::Unspecified => "unspecified",
+            }
+        }
+    }
 }
 
 pub mod cluster {
diff --git a/release-notes/unreleased/restatectl-storage-compact.md b/release-notes/unreleased/restatectl-storage-compact.md
@@ -0,0 +1,35 @@
+# Release Notes: `restatectl storage compact` command
+
+## New Feature
+
+### What Changed
+
+A new `restatectl storage compact` command has been added to trigger manual RocksDB
+compaction on cluster nodes. This allows operators to reclaim disk space, reduce read
+amplification, or force compaction after bulk deletions without restarting nodes.
+
+### Why This Matters
+
+RocksDB compactions run automatically in the background, but there are situations where
+operators may want to trigger them explicitly — for example, after a large number of
+invocations have been cleaned up, or as part of routine maintenance to ensure optimal
+storage performance.
+
+### Usage
+
+```bash
+# Compact all databases on all nodes
+restatectl storage compact
+
+# Compact specific database types
+restatectl storage compact -d partition-store
+restatectl storage compact -d log-server,metadata-server
+
+# Target specific nodes
+restatectl storage compact -n N1,N2
+
+# Combine filters
+restatectl storage compact -d partition-store -n N1
+```
+
+**Available database types:** `partition-store`, `log-server`, `metadata-server`, `local-loglet`
diff --git a/tools/bifrost-benchpress/src/main.rs b/tools/bifrost-benchpress/src/main.rs
diff --git a/tools/logserver-bench/src/mixed_workload.rs b/tools/logserver-bench/src/mixed_workload.rs
diff --git a/tools/logserver-bench/src/write_throughput.rs b/tools/logserver-bench/src/write_throughput.rs
diff --git a/tools/restatectl/src/commands/storage/compact.rs b/tools/restatectl/src/commands/storage/compact.rs
