chain: gc partial chunks on archival nodes (near#6439)

This is commit 6be2e0e upstream.

Start garbage collecting ColPartialChunks and ColInvalidChunks on
archival nodes.  The former is quite sizeable column and its data can
be recovered from ColChunks.  The latter is only needed when operating
at head.

Note that this is likely insufficient for the garbage collection to
happen in reasonable time (since with current default options we’re
garbage collecting only two heights at a time).  It’s best to clean
out the two columns.

Issue: near#6242

chain/chain/src/chain.rs

@@ -851,6 +851,30 @@ impl Chain {
         Ok(())
     }
 
+    /// Garbage collect data which archival node doesn’t need to keep.
+    ///
+    /// Normally, archival nodes keep all the data from the genesis block and
+    /// don’t run garbage collection.  On the other hand, for better performance
+    /// the storage contains some data duplication, i.e. values in some of the
+    /// columns can be recomputed from data in different columns.  To save on
+    /// storage, archival nodes do garbage collect that data.
+    ///
+    /// `gc_height_limit` limits how many heights will the function process.
+    pub fn clear_archive_data(&mut self, gc_height_limit: BlockHeightDelta) -> Result<(), Error> {
+        let head = self.store.head()?;
+        let gc_stop_height = self.runtime_adapter.get_gc_stop_height(&head.last_block_hash);
+        if gc_stop_height > head.height {
+            return Err(ErrorKind::GCError(
+                "gc_stop_height cannot be larger than head.height".into(),
+            )
+            .into());
+        }
+
+        let mut chain_store_update = self.store.store_update();
+        chain_store_update.clear_redundant_chunk_data(gc_stop_height, gc_height_limit)?;
+        chain_store_update.commit()
+    }
+
     pub fn clear_forks_data(
         &mut self,
         tries: ShardTries,

chain/chain/src/store.rs

@@ -27,9 +27,9 @@ use near_primitives::transaction::{
 use near_primitives::trie_key::{trie_key_parsers, TrieKey};
 use near_primitives::types::chunk_extra::ChunkExtra;
 use near_primitives::types::{
-    AccountId, BlockExtra, BlockHeight, EpochId, GCCount, NumBlocks, ShardId, StateChanges,
-    StateChangesExt, StateChangesForSplitStates, StateChangesKinds, StateChangesKindsExt,
-    StateChangesRequest,
+    AccountId, BlockExtra, BlockHeight, BlockHeightDelta, EpochId, GCCount, NumBlocks, ShardId,
+    StateChanges, StateChangesExt, StateChangesForSplitStates, StateChangesKinds,
+    StateChangesKindsExt, StateChangesRequest,
 };
 use near_primitives::utils::{get_block_shard_id, index_to_bytes, to_timestamp};
 use near_primitives::views::LightClientBlockView;
@@ -2021,6 +2021,47 @@ impl<'a> ChainStoreUpdate<'a> {
         Ok(())
     }
 
+    /// Clears chunk data which can be computed from other data in the storage.
+    ///
+    /// We are storing PartialEncodedChunk objects in the ColPartialChunks in
+    /// the storage.  However, those objects can be computed from data in
+    /// ColChunks and as such are redundant.  For performance reasons we want to
+    /// keep that data when operating at head of the chain but the data can be
+    /// safely removed from archival storage.
+    ///
+    /// `gc_stop_height` indicates height starting from which no data should be
+    /// garbage collected.  Roughly speaking this represents start of the ‘hot’
+    /// data that we want to keep.
+    ///
+    /// `gt_height_limit` indicates limit of how many non-empty heights to
+    /// process.  This limit means that the method may stop garbage collection
+    /// before reaching `gc_stop_height`.
+    pub fn clear_redundant_chunk_data(
+        &mut self,
+        gc_stop_height: BlockHeight,
+        gc_height_limit: BlockHeightDelta,
+    ) -> Result<(), Error> {
+        let mut height = self.chunk_tail()?;
+        let mut remaining = gc_height_limit;
+        while height < gc_stop_height && remaining > 0 {
+            let chunk_hashes = self.chain_store.get_all_chunk_hashes_by_height(height)?;
+            height += 1;
+            if !chunk_hashes.is_empty() {
+                remaining -= 1;
+                for chunk_hash in chunk_hashes {
+                    let chunk_header_hash = chunk_hash.into();
+                    self.gc_col(ColPartialChunks, &chunk_header_hash);
+                    // Data in ColInvalidChunks isn’t technically redundant (it
+                    // cannot be calculated from other data) but it is data we
+                    // don’t need for anything so it can be deleted as well.
+                    self.gc_col(ColInvalidChunks, &chunk_header_hash);
+                }
+            }
+        }
+        self.update_chunk_tail(height);
+        Ok(())
+    }
+
     fn get_shard_uids_to_gc(
         &mut self,
         runtime_adapter: &dyn RuntimeAdapter,

chain/client/src/client.rs

@@ -1043,17 +1043,20 @@ impl Client {
                 self.chain.get_block_header(last_final_block).map_or(0, |header| header.height())
             };
             self.chain.blocks_with_missing_chunks.prune_blocks_below_height(last_finalized_height);
-            if !self.config.archive {
-                let timer = metrics::GC_TIME.start_timer();
-                if let Err(err) = self
-                    .chain
-                    .clear_data(self.runtime_adapter.get_tries(), self.config.gc_blocks_limit)
-                {
-                    error!(target: "client", "Can't clear old data, {:?}", err);
-                    debug_assert!(false);
-                };
-                timer.observe_duration();
-            }
+
+            let timer = metrics::GC_TIME.start_timer();
+            let gc_blocks_limit = self.config.gc_blocks_limit;
+            let result = if self.config.archive {
+                self.chain.clear_archive_data(gc_blocks_limit)
+            } else {
+                let tries = self.runtime_adapter.get_tries();
+                self.chain.clear_data(tries, gc_blocks_limit)
+            };
+            if let Err(err) = result {
+                error!(target: "client", "Can't clear old data, {:?}", err);
+                debug_assert!(false);
+            };
+            timer.observe_duration();
 
             if self.runtime_adapter.is_next_block_epoch_start(block.hash()).unwrap_or(false) {
                 let next_epoch_protocol_version = unwrap_or_return!(self