feat(state-sync): drop old state parts at the end of epoch

chain/chain/src/chain.rs

@@ -1815,6 +1815,11 @@ impl Chain {
         };
         let (apply_chunk_work, block_preprocess_info) = preprocess_res;
 
+        if self.epoch_manager.is_next_block_epoch_start(block.header().prev_hash())? {
+            // This is the end of the epoch. Next epoch we will generate new state parts. We can drop the old ones.
+            self.clear_all_downloaded_parts()?;
+        }
+
         // 2) Start creating snapshot if needed.
         if let Err(err) = self.process_snapshot() {
             tracing::error!(target: "state_snapshot", ?err, "Failed to make a state snapshot");
@@ -3016,6 +3021,16 @@ impl Chain {
         chain_store_update.commit()
     }
 
+    /// Drop all downloaded or generated state parts and headers.
+    pub fn clear_all_downloaded_parts(&mut self) -> Result<(), Error> {
+        tracing::debug!(target: "state_sync", "Clear old state parts");
+        let mut store_update = self.chain_store.store().store_update();
+        store_update.delete_all(DBCol::StateParts);
+        store_update.delete_all(DBCol::StateHeaders);
+        store_update.commit()?;
+        Ok(())
+    }
+
     pub fn catchup_blocks_step(
         &mut self,
         me: &Option<AccountId>,

integration-tests/src/tests/client/state_dump.rs

@@ -15,13 +15,11 @@ use near_primitives::block::Tip;
 use near_primitives::shard_layout::ShardUId;
 use near_primitives::state::FlatStateValue;
 use near_primitives::state_part::PartId;
-use near_primitives::state_sync::StatePartKey;
 use near_primitives::transaction::SignedTransaction;
 use near_primitives::types::BlockHeight;
 use near_primitives::validator_signer::{EmptyValidatorSigner, InMemoryValidatorSigner};
 use near_primitives::views::{QueryRequest, QueryResponseKind};
 use near_store::adapter::{StoreAdapter, StoreUpdateAdapter};
-use near_store::DBCol;
 use near_store::Store;
 use nearcore::state_sync::StateSyncDumper;
 use nearcore::test_utils::TestEnvNightshadeSetupExt;
@@ -296,12 +294,10 @@ fn run_state_sync_with_dumped_parts(
         .await
         .unwrap();
 
-        // Simulate state sync
+        // Simulate state sync by reading the dumped parts from the external storage and applying them to the other node
         tracing::info!("syncing node: simulating state sync..");
         env.clients[1].chain.set_state_header(0, sync_hash, state_sync_header).unwrap();
         let runtime_client_1 = Arc::clone(&env.clients[1].runtime_adapter);
-        let runtime_client_0 = Arc::clone(&env.clients[0].runtime_adapter);
-        let client_0_store = runtime_client_0.store();
         let mut store_update = runtime_client_1.store().store_update();
         assert!(runtime_client_1
             .get_flat_storage_manager()
@@ -311,10 +307,16 @@ fn run_state_sync_with_dumped_parts(
             )
             .unwrap());
         store_update.commit().unwrap();
-
+        let shard_id = 0;
         for part_id in 0..num_parts {
-            let key = borsh::to_vec(&StatePartKey(sync_hash, 0, part_id)).unwrap();
-            let part = client_0_store.get(DBCol::StateParts, &key).unwrap().unwrap();
+            let path = root_dir.path().join(external_storage_location(
+                &config.chain_id,
+                &epoch_id,
+                epoch_height,
+                shard_id,
+                &StateFileType::StatePart { part_id, num_parts },
+            ));
+            let part = std::fs::read(&path).expect("Part file not found. It should exist");
 
             runtime_client_1
                 .apply_state_part(0, &state_root, PartId::new(part_id, num_parts), &part, &epoch_id)

o

@@ -0,0 +1,9 @@
+
+running 0 tests
+
+successes:
+
+successes:
+
+test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 394 filtered out; finished in 0.00s
+