State-viewer uses read-only RocksDB (near#6294)

Fix near#5194

Tested by running 1.21, 1.22, 1.23, 1.24 and 1.25 versions in localnet and attempting to dump their state.
In my localnet tests state viewer was able to open an instance locked by another process. Don't know if it works in the general case. Running multiple state viewers should be totally safe though if the instance is not locked.

Without this change, attempting to dump state can add a column (col49) which makes it impossible to dump state with state-viewer of the corresponding version.
With this change, state-viewer panics earlier and state-viewer of the corresponding version remains able to dump state:

```
Tue 14:17:50 ~/code/nearcore-rocksdb18-2 % RUST_BACKTRACE=all ./target/release/neard --home ~/.near/localnet/node0/ view_state dump_state
thread 'main' panicked at 'Failed to start Epoch Manager: IOErr(Unexpected length of input)', nearcore/src/runtime/mod.rs:198:18
stack backtrace:
   0: rust_begin_unwind
             at /rustc/db9d1b20bba1968c1ec1fc49616d4742c1725b4b/library/std/src/panicking.rs:498:5
   1: core::panicking::panic_fmt
             at /rustc/db9d1b20bba1968c1ec1fc49616d4742c1725b4b/library/core/src/panicking.rs:107:14
   2: core::result::unwrap_failed
             at /rustc/db9d1b20bba1968c1ec1fc49616d4742c1725b4b/library/core/src/result.rs:1613:5
   3: nearcore::runtime::NightshadeRuntime::new
   4: nearcore::runtime::NightshadeRuntime::with_config
   5: state_viewer::commands::load_trie_stop_at_height
   6: state_viewer::commands::dump_state
   7: state_viewer::cli::StateViewerSubCommand::run
   8: neard::cli::NeardCmd::parse_and_run
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.
```

core/store/src/db.rs

@@ -489,6 +489,10 @@ impl Default for RocksDBOptions {
     }
 }
 
+fn col_name(col: DBCol) -> String {
+    format!("col{}", col as usize)
+}
+
 impl RocksDBOptions {
     /// Once the disk space is below the `free_disk_space_warn_threshold`, RocksDB will emit an warning message every [`interval`](RocksDBOptions::check_free_space_interval) write.
     pub fn free_disk_space_warn_threshold(mut self, warn_treshold: bytesize::ByteSize) -> Self {
@@ -525,11 +529,14 @@ impl RocksDBOptions {
 
     /// Opens a read only database.
     pub fn read_only<P: AsRef<std::path::Path>>(self, path: P) -> Result<RocksDB, DBError> {
-        let options = self.rocksdb_options.unwrap_or_default();
-        let cf_names: Vec<_> = self.cf_names.unwrap_or_else(|| vec!["col0".to_string()]);
-        let db = DB::open_cf_for_read_only(&options, path, cf_names.iter(), false)?;
-        let cfs =
-            cf_names.iter().map(|n| db.cf_handle(n).unwrap() as *const ColumnFamily).collect();
+        use strum::IntoEnumIterator;
+        let options = self.rocksdb_options.unwrap_or_else(rocksdb_options);
+        let cf_with_opts = DBCol::iter().map(|col| (col_name(col), rocksdb_column_options(col)));
+        let db = DB::open_cf_with_opts_for_read_only(&options, path, cf_with_opts, false)?;
+        let cfs = DBCol::iter()
+            .map(|col| db.cf_handle(&col_name(col)).unwrap() as *const ColumnFamily)
+            .collect();
+
         Ok(RocksDB {
             db,
             cfs,
@@ -544,17 +551,11 @@ impl RocksDBOptions {
     pub fn read_write<P: AsRef<std::path::Path>>(self, path: P) -> Result<RocksDB, DBError> {
         use strum::IntoEnumIterator;
         let options = self.rocksdb_options.unwrap_or_else(rocksdb_options);
-        let cf_names = self
-            .cf_names
-            .unwrap_or_else(|| DBCol::iter().map(|col| format!("col{}", col as usize)).collect());
+        let cf_names =
+            self.cf_names.unwrap_or_else(|| DBCol::iter().map(|col| col_name(col)).collect());
         let cf_descriptors = self.cf_descriptors.unwrap_or_else(|| {
             DBCol::iter()
-                .map(|col| {
-                    ColumnFamilyDescriptor::new(
-                        format!("col{}", col as usize),
-                        rocksdb_column_options(col),
-                    )
-                })
+                .map(|col| ColumnFamilyDescriptor::new(col_name(col), rocksdb_column_options(col)))
                 .collect()
         });
         let db = DB::open_cf_descriptors(&options, path, cf_descriptors)?;
@@ -882,7 +883,7 @@ impl RocksDB {
         })
     }
 
-    fn new_read_only<P: AsRef<std::path::Path>>(path: P) -> Result<Self, DBError> {
+    pub fn new_read_only<P: AsRef<std::path::Path>>(path: P) -> Result<Self, DBError> {
         RocksDBOptions::default().read_only(path)
     }
 

core/store/src/db/v6_to_v7.rs

@@ -4,7 +4,7 @@ use byteorder::{LittleEndian, ReadBytesExt};
 use rocksdb::{ColumnFamilyDescriptor, MergeOperands, Options};
 use strum::IntoEnumIterator;
 
-use crate::db::{rocksdb_column_options, DBError, RocksDB, RocksDBOptions};
+use crate::db::{col_name, rocksdb_column_options, DBError, RocksDB, RocksDBOptions};
 use crate::DBCol;
 
 fn refcount_merge_v6(
@@ -59,14 +59,11 @@ fn rocksdb_column_options_v6(col: DBCol) -> Options {
 impl RocksDB {
     pub(crate) fn new_v6<P: AsRef<std::path::Path>>(path: P) -> Result<Self, DBError> {
         RocksDBOptions::default()
-            .cf_names(DBCol::iter().map(|col| format!("col{}", col as usize)).collect())
+            .cf_names(DBCol::iter().map(|col| col_name(col)).collect())
             .cf_descriptors(
                 DBCol::iter()
                     .map(|col| {
-                        ColumnFamilyDescriptor::new(
-                            format!("col{}", col as usize),
-                            rocksdb_column_options_v6(col),
-                        )
+                        ColumnFamilyDescriptor::new(col_name(col), rocksdb_column_options_v6(col))
                     })
                     .collect(),
             )

core/store/src/lib.rs

@@ -300,6 +300,13 @@ pub fn create_store(path: &Path) -> Store {
     Store::new(db)
 }
 
+/// Creates a store which is unable to modify an existing RocksDB instance.
+/// Panics if a write operation is attempted.
+pub fn open_read_only_store(path: &Path) -> Store {
+    let db = Arc::new(RocksDB::new_read_only(path).expect("Failed to open the database"));
+    Store::new(db)
+}
+
 /// Reads an object from Trie.
 /// # Errors
 /// see StorageError

tools/state-viewer/src/cli.rs

@@ -9,7 +9,7 @@ use near_primitives::hash::CryptoHash;
 use near_primitives::sharding::ChunkHash;
 use near_primitives::types::{BlockHeight, ShardId};
 use near_primitives::version::{DB_VERSION, PROTOCOL_VERSION};
-use near_store::{create_store, Store};
+use near_store::{open_read_only_store, Store};
 use nearcore::{get_default_home, get_store_path, load_config, NearConfig};
 use once_cell::sync::Lazy;
 use std::path::{Path, PathBuf};
@@ -114,7 +114,7 @@ pub enum StateViewerSubCommand {
 impl StateViewerSubCommand {
     pub fn run(self, home_dir: &Path, genesis_validation: GenesisValidationMode) {
         let near_config = load_config(home_dir, genesis_validation);
-        let store = create_store(&get_store_path(home_dir));
+        let store = open_read_only_store(&get_store_path(home_dir));
         match self {
             StateViewerSubCommand::Peers => peers(store),
             StateViewerSubCommand::State => state(home_dir, near_config, store),