state-dumper: Don't store state parts in the database (near#12486)

When the state dumper genrates state parts to upload to GCS, it stores
them in the StateParts column before doing so. But it never reads these
parts after storing them. Storing these parts is not a free operation,
because on a forknet run with recent mainnet state, just the storage of
the part took on the order of 50-100 millis. Over several thousand
parts, that's a nontrivial slowdown for no good reason.

nearcore/src/state_sync.rs

@@ -6,7 +6,7 @@ use futures::future::BoxFuture;
 use futures::FutureExt;
 use near_async::time::{Clock, Duration, Instant};
 use near_chain::types::RuntimeAdapter;
-use near_chain::{Chain, ChainGenesis, ChainStoreAccess, DoomslugThresholdMode, Error};
+use near_chain::{Chain, ChainGenesis, DoomslugThresholdMode, Error};
 use near_chain_configs::{ClientConfig, ExternalStorageLocation, MutableValidatorSigner};
 use near_client::sync::external::{
     create_bucket_readwrite, external_storage_location, StateFileType,
@@ -19,9 +19,8 @@ use near_epoch_manager::shard_tracker::ShardTracker;
 use near_epoch_manager::EpochManagerAdapter;
 use near_primitives::hash::CryptoHash;
 use near_primitives::state_part::PartId;
-use near_primitives::state_sync::{StatePartKey, StateSyncDumpProgress};
+use near_primitives::state_sync::StateSyncDumpProgress;
 use near_primitives::types::{AccountId, EpochHeight, EpochId, ShardId, StateRoot};
-use near_store::DBCol;
 use rand::{thread_rng, Rng};
 use std::collections::HashSet;
 use std::sync::atomic::AtomicBool;
@@ -469,15 +468,11 @@ async fn state_sync_dump(
                                     let (part_id, selected_idx) =
                                         select_random_part_id_with_index(&parts_to_dump);
 
-                                    let state_part = obtain_and_store_state_part(
-                                        runtime.as_ref(),
+                                    let state_part = runtime.obtain_state_part(
                                         shard_id,
-                                        sync_hash,
                                         &sync_prev_prev_hash,
                                         &state_root,
-                                        part_id,
-                                        num_parts,
-                                        &chain,
+                                        PartId::new(part_id, num_parts),
                                     );
                                     let state_part = match state_part {
                                         Ok(state_part) => state_part,
@@ -617,31 +612,6 @@ fn update_dumped_size_and_cnt_metrics(
         .set(num_parts as i64);
 }
 
-/// Obtains and then saves the part data.
-fn obtain_and_store_state_part(
-    runtime: &dyn RuntimeAdapter,
-    shard_id: ShardId,
-    sync_hash: CryptoHash,
-    sync_prev_prev_hash: &CryptoHash,
-    state_root: &StateRoot,
-    part_id: u64,
-    num_parts: u64,
-    chain: &Chain,
-) -> Result<Vec<u8>, Error> {
-    let state_part = runtime.obtain_state_part(
-        shard_id,
-        sync_prev_prev_hash,
-        state_root,
-        PartId::new(part_id, num_parts),
-    )?;
-
-    let key = borsh::to_vec(&StatePartKey(sync_hash, shard_id, part_id))?;
-    let mut store_update = chain.chain_store().store().store_update();
-    store_update.set(DBCol::StateParts, &key, &state_part);
-    store_update.commit()?;
-    Ok(state_part)
-}
-
 fn cares_about_shard(
     chain: &Chain,
     shard_id: &ShardId,