feat: disable write stall on fullnodes perpetual db (#6817)

# Description of change

Port
MystenLabs/sui@42dbb62

From Sui's commit description:
For fullnodes that do not prune the `perpetual` DB, especially
`transactions` and `effects` cfs, they can run into write stalls
occasionally that makes the fullnode non operational. Since fullnodes
have to accept all writes from checkpoints, throttling writes do not
seem to make much sense.
Write stalling on validators is left enabled.

## Links to any relevant issues

Part of #3990 

## Type of change
- Enhancement 

---------

Co-authored-by: muXxer <[email protected]>

Author: jkrvivian

crates/iota-config/src/node.rs

@@ -250,6 +250,12 @@ pub struct NodeConfig {
     #[serde(default)]
     pub verifier_signing_config: VerifierSigningConfig,
 
+    /// If a value is set, it determines if writes to DB can stall, which can
+    /// halt the whole process. By default, write stall is enabled on
+    /// validators but not on fullnodes.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub enable_db_write_stall: Option<bool>,
+
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub iota_names_config: Option<IotaNamesConfig>,
 }

crates/iota-core/src/authority/authority_store_tables.rs

@@ -13,11 +13,10 @@ use typed_store::{
     DBMapUtils,
     metrics::SamplingInterval,
     rocks::{
-        DBBatch, DBMap, DBOptions, MetricConf, ReadWriteOptions, default_db_options,
+        DBBatch, DBMap, DBMapTableConfigMap, DBOptions, MetricConf, default_db_options,
         read_size_from_env,
         util::{empty_compaction_filter, reference_count_merge_operator},
     },
-    rocksdb::Options,
     traits::{Map, TableSummary, TypedStoreDebug},
 };
 
@@ -37,6 +36,22 @@ const ENV_VAR_EFFECTS_BLOCK_CACHE_SIZE: &str = "EFFECTS_BLOCK_CACHE_MB";
 const ENV_VAR_EVENTS_BLOCK_CACHE_SIZE: &str = "EVENTS_BLOCK_CACHE_MB";
 const ENV_VAR_INDIRECT_OBJECTS_BLOCK_CACHE_SIZE: &str = "INDIRECT_OBJECTS_BLOCK_CACHE_MB";
 
+/// Options to apply to every column family of the `perpetual` DB.
+#[derive(Default)]
+pub struct AuthorityPerpetualTablesOptions {
+    /// Whether to enable write stalling on all column families.
+    pub enable_write_stall: bool,
+}
+
+impl AuthorityPerpetualTablesOptions {
+    fn apply_to(&self, mut db_options: DBOptions) -> DBOptions {
+        if !self.enable_write_stall {
+            db_options = db_options.disable_write_throttling();
+        }
+        db_options
+    }
+}
+
 /// AuthorityPerpetualTables contains data that must be preserved from one epoch
 /// to the next.
 #[derive(DBMapUtils)]
@@ -56,21 +71,17 @@ pub struct AuthorityPerpetualTables {
     /// executed transactions whose effects have not yet been written out,
     /// and which must be retried. But, they cannot be retried unless their
     /// input objects are still accessible!
-    #[default_options_override_fn = "objects_table_default_config"]
     pub(crate) objects: DBMap<ObjectKey, StoreObjectWrapper>,
 
-    #[default_options_override_fn = "indirect_move_objects_table_default_config"]
     pub(crate) indirect_move_objects: DBMap<ObjectContentDigest, StoreMoveObjectWrapper>,
 
     /// Object references of currently active objects that can be mutated.
-    #[default_options_override_fn = "live_owned_object_markers_table_default_config"]
     pub(crate) live_owned_object_markers: DBMap<ObjectRef, ()>,
 
     /// This is a map between the transaction digest and the corresponding
     /// transaction that's known to be executable. This means that it may
     /// have been executed locally, or it may have been synced through
     /// state-sync but hasn't been executed yet.
-    #[default_options_override_fn = "transactions_table_default_config"]
     pub(crate) transactions: DBMap<TransactionDigest, TrustedTransaction>,
 
     /// A map between the transaction digest of a certificate to the effects of
@@ -85,7 +96,6 @@ pub struct AuthorityPerpetualTables {
     ///
     /// It's also possible for the effects to be reverted if the transaction
     /// didn't make it into the epoch.
-    #[default_options_override_fn = "effects_table_default_config"]
     pub(crate) effects: DBMap<TransactionEffectsDigest, TransactionEffects>,
 
     /// Transactions that have been executed locally on this node. We need this
@@ -99,7 +109,6 @@ pub struct AuthorityPerpetualTables {
     // We could potentially remove this if we decided not to provide events in the execution path.
     // TODO: Figure out what to do with this table in the long run.
     // Also we need a pruning policy for this table. We can prune this table along with tx/effects.
-    #[default_options_override_fn = "events_table_default_config"]
     pub(crate) events: DBMap<(TransactionEventsDigest, usize), Event>,
 
     /// Epoch and checkpoint of transactions finalized by checkpoint
@@ -156,13 +165,45 @@ impl AuthorityPerpetualTables {
         parent_path.join("perpetual")
     }
 
-    pub fn open(parent_path: &Path, db_options: Option<Options>) -> Self {
+    pub fn open(
+        parent_path: &Path,
+        db_options_override: Option<AuthorityPerpetualTablesOptions>,
+    ) -> Self {
+        let db_options_override = db_options_override.unwrap_or_default();
+        let db_options =
+            db_options_override.apply_to(default_db_options().optimize_db_for_write_throughput(4));
+        let table_options = DBMapTableConfigMap::new(BTreeMap::from([
+            (
+                "objects".to_string(),
+                objects_table_config(db_options.clone()),
+            ),
+            (
+                "indirect_move_objects".to_string(),
+                indirect_move_objects_table_config(db_options.clone()),
+            ),
+            (
+                "live_owned_object_markers".to_string(),
+                live_owned_object_markers_table_config(db_options.clone()),
+            ),
+            (
+                "transactions".to_string(),
+                transactions_table_config(db_options.clone()),
+            ),
+            (
+                "effects".to_string(),
+                effects_table_config(db_options.clone()),
+            ),
+            (
+                "events".to_string(),
+                events_table_config(db_options.clone()),
+            ),
+        ]));
         Self::open_tables_read_write(
             Self::path(parent_path),
             MetricConf::new("perpetual")
                 .with_sampling(SamplingInterval::new(Duration::from_secs(60), 0)),
-            db_options,
-            None,
+            Some(db_options.options),
+            Some(table_options),
         )
     }
 
@@ -605,57 +646,58 @@ impl Iterator for LiveSetIter<'_> {
 }
 
 // These functions are used to initialize the DB tables
-fn live_owned_object_markers_table_default_config() -> DBOptions {
+fn live_owned_object_markers_table_config(db_options: DBOptions) -> DBOptions {
     DBOptions {
-        options: default_db_options()
+        options: db_options
+            .clone()
             .optimize_for_write_throughput()
             .optimize_for_read(read_size_from_env(ENV_VAR_LOCKS_BLOCK_CACHE_SIZE).unwrap_or(1024))
             .options,
-        rw_options: ReadWriteOptions::default().set_ignore_range_deletions(false),
+        rw_options: db_options.rw_options.set_ignore_range_deletions(false),
     }
 }
 
-fn objects_table_default_config() -> DBOptions {
-    default_db_options()
+fn objects_table_config(db_options: DBOptions) -> DBOptions {
+    db_options
         .optimize_for_write_throughput()
         .optimize_for_read(read_size_from_env(ENV_VAR_OBJECTS_BLOCK_CACHE_SIZE).unwrap_or(5 * 1024))
 }
 
-fn transactions_table_default_config() -> DBOptions {
-    default_db_options()
+fn transactions_table_config(db_options: DBOptions) -> DBOptions {
+    db_options
         .optimize_for_write_throughput()
         .optimize_for_point_lookup(
             read_size_from_env(ENV_VAR_TRANSACTIONS_BLOCK_CACHE_SIZE).unwrap_or(512),
         )
 }
 
-fn effects_table_default_config() -> DBOptions {
-    default_db_options()
+fn effects_table_config(db_options: DBOptions) -> DBOptions {
+    db_options
         .optimize_for_write_throughput()
         .optimize_for_point_lookup(
             read_size_from_env(ENV_VAR_EFFECTS_BLOCK_CACHE_SIZE).unwrap_or(1024),
         )
 }
 
-fn events_table_default_config() -> DBOptions {
-    default_db_options()
+fn events_table_config(db_options: DBOptions) -> DBOptions {
+    db_options
         .optimize_for_write_throughput()
         .optimize_for_read(read_size_from_env(ENV_VAR_EVENTS_BLOCK_CACHE_SIZE).unwrap_or(1024))
 }
 
-fn indirect_move_objects_table_default_config() -> DBOptions {
-    let mut options = default_db_options()
+fn indirect_move_objects_table_config(mut db_options: DBOptions) -> DBOptions {
+    db_options = db_options
         .optimize_for_write_throughput()
         .optimize_for_point_lookup(
             read_size_from_env(ENV_VAR_INDIRECT_OBJECTS_BLOCK_CACHE_SIZE).unwrap_or(512),
         );
-    options.options.set_merge_operator(
+    db_options.options.set_merge_operator(
         "refcount operator",
         reference_count_merge_operator,
         reference_count_merge_operator,
     );
-    options
+    db_options
         .options
         .set_compaction_filter("empty filter", empty_compaction_filter);
-    options
+    db_options
 }

crates/iota-node/src/lib.rs

@@ -35,7 +35,7 @@ use iota_core::{
     authority::{
         AuthorityState, AuthorityStore, CHAIN_IDENTIFIER, RandomnessRoundReceiver,
         authority_per_epoch_store::AuthorityPerEpochStore,
-        authority_store_tables::AuthorityPerpetualTables,
+        authority_store_tables::{AuthorityPerpetualTables, AuthorityPerpetualTablesOptions},
         epoch_start_configuration::{EpochFlag, EpochStartConfigTrait, EpochStartConfiguration},
     },
     authority_aggregator::{AuthAggMetrics, AuthorityAggregator},
@@ -479,10 +479,12 @@ impl IotaNode {
             None,
         ));
 
-        let perpetual_options = default_db_options().optimize_db_for_write_throughput(4);
+        // By default, only enable write stall on validators for perpetual db.
+        let enable_write_stall = config.enable_db_write_stall.unwrap_or(is_validator);
+        let perpetual_tables_options = AuthorityPerpetualTablesOptions { enable_write_stall };
         let perpetual_tables = Arc::new(AuthorityPerpetualTables::open(
             &config.db_path().join("store"),
-            Some(perpetual_options.options),
+            Some(perpetual_tables_options),
         ));
         let is_genesis = perpetual_tables
             .database_is_empty()

crates/iota-swarm-config/src/node_config_builder.rs

@@ -225,6 +225,7 @@ impl ValidatorConfigBuilder {
             execution_cache: ExecutionCacheConfig::default(),
             enable_validator_tx_finalizer: true,
             verifier_signing_config: VerifierSigningConfig::default(),
+            enable_db_write_stall: None,
             iota_names_config: None,
         }
     }
@@ -519,6 +520,7 @@ impl FullnodeConfigBuilder {
             // This is a validator specific feature.
             enable_validator_tx_finalizer: false,
             verifier_signing_config: VerifierSigningConfig::default(),
+            enable_db_write_stall: None,
             iota_names_config: None,
         }
     }

crates/typed-store-derive/src/lib.rs

@@ -323,6 +323,14 @@ pub fn derive_dbmap_utils_general(input: TokenStream) -> TokenStream {
                 remove_deprecated_tables: bool,
             ) -> Self {
                 let path = &path;
+                let default_cf_opt = if let Some(opt) = global_db_options_override.as_ref() {
+                    typed_store::rocks::DBOptions {
+                        options: opt.clone(),
+                        rw_options: typed_store::rocks::default_db_options().rw_options,
+                    }
+                } else {
+                    typed_store::rocks::default_db_options()
+                };
                 let (db, rwopt_cfs) = {
                     let opt_cfs = match tables_db_options_override {
                         None => [
@@ -332,7 +340,7 @@ pub fn derive_dbmap_utils_general(input: TokenStream) -> TokenStream {
                         ],
                         Some(o) => [
                             #(
-                                (stringify!(#cf_names).to_owned(), o.to_map().get(stringify!(#cf_names)).unwrap().clone()),
+                                (stringify!(#cf_names).to_owned(), o.to_map().get(stringify!(#cf_names)).unwrap_or(&default_cf_opt).clone()),
                             )*
                         ]
                     };