reintroduce clear_resharding_data (#10211)

Tested on updated #10193

`is_next_block_epoch_start` reads `BlockInfo` of the first block of the
epoch, which is not correct in the case of garbage collection.
The previous version of the test was passing, because that `BlockInfo`
was still in the cache of the epoch manager (as we call
`is_next_block_epoch_start` on every block)

Implemented safer version of `is_next_block_epoch_start` --
`is_last_block_in_finished_epoch` (didn't think too much about the name,
open to suggestions).
`is_last_block_in_finished_epoch` works by relying on the fact that if
we processed block, and it was the last block of an epoch according to
`is_next_block_epoch_start`, then we wrote an `EpochInfo` with the hash
of that block, and `EpochInfo` is not garbage collectible.

chain/chain/src/chain.rs

@@ -1007,12 +1007,11 @@ impl Chain {
                         *block_hash,
                         GCMode::Canonical(tries.clone()),
                     )?;
-                    // TODO(resharding): Call clear_resharding_data once we figure out what's wrong
-                    // chain_store_update.clear_resharding_data(
-                    //     self.runtime_adapter.as_ref(),
-                    //     self.epoch_manager.as_ref(),
-                    //     *block_hash,
-                    // )?;
+                    chain_store_update.clear_resharding_data(
+                        self.runtime_adapter.as_ref(),
+                        self.epoch_manager.as_ref(),
+                        *block_hash,
+                    )?;
                     gc_blocks_remaining -= 1;
                 } else {
                     return Err(Error::GCError(

chain/chain/src/store.rs

@@ -2340,7 +2340,7 @@ impl<'a> ChainStoreUpdate<'a> {
     ) -> Result<(), Error> {
         // Need to check if this is the last block of the epoch where resharding is completed
         // which means shard layout changed in the previous epoch
-        if !epoch_manager.is_next_block_epoch_start(&block_hash)? {
+        if !epoch_manager.is_last_block_in_finished_epoch(&block_hash)? {
             return Ok(());
         }
 

chain/chain/src/test_utils/kv_runtime.rs

@@ -547,6 +547,10 @@ impl EpochManagerAdapter for MockEpochManager {
             != self.get_epoch_and_valset(prev_prev_hash)?.0)
     }
 
+    fn is_last_block_in_finished_epoch(&self, hash: &CryptoHash) -> Result<bool, EpochError> {
+        self.is_next_block_epoch_start(hash)
+    }
+
     fn get_epoch_id_from_prev_block(
         &self,
         parent_hash: &CryptoHash,

chain/epoch-manager/src/adapter.rs

@@ -76,6 +76,12 @@ pub trait EpochManagerAdapter: Send + Sync {
     /// Returns true, if given hash is last block in it's epoch.
     fn is_next_block_epoch_start(&self, parent_hash: &CryptoHash) -> Result<bool, EpochError>;
 
+    /// Returns true, if given hash is in an epoch that already finished.
+    /// `is_next_block_epoch_start` works even if we didn't fully process the provided block.
+    /// This function works even if we garbage collected `BlockInfo` of the first block of the epoch.
+    /// Thus, this function is better suited for use in garbage collection.
+    fn is_last_block_in_finished_epoch(&self, hash: &CryptoHash) -> Result<bool, EpochError>;
+
     /// Get epoch id given hash of previous block.
     fn get_epoch_id_from_prev_block(&self, parent_hash: &CryptoHash)
         -> Result<EpochId, EpochError>;
@@ -475,6 +481,11 @@ impl EpochManagerAdapter for EpochManagerHandle {
         epoch_manager.is_next_block_epoch_start(parent_hash)
     }
 
+    fn is_last_block_in_finished_epoch(&self, hash: &CryptoHash) -> Result<bool, EpochError> {
+        let epoch_manager = self.read();
+        epoch_manager.is_last_block_in_finished_epoch(hash)
+    }
+
     fn get_epoch_id_from_prev_block(
         &self,
         parent_hash: &CryptoHash,

chain/epoch-manager/src/lib.rs

@@ -1086,6 +1086,20 @@ impl EpochManager {
         self.is_next_block_in_next_epoch(&block_info)
     }
 
+    /// Relies on the fact that last block hash of an epoch is an EpochId of next next epoch.
+    /// If this block is the last one in some epoch, and we fully processed it, there will be `EpochInfo` record with `hash` key.
+    fn is_last_block_in_finished_epoch(&self, hash: &CryptoHash) -> Result<bool, EpochError> {
+        match self.get_epoch_info(&EpochId(*hash)) {
+            Ok(_) => Ok(true),
+            Err(EpochError::IOErr(msg)) => Err(EpochError::IOErr(msg)),
+            Err(EpochError::MissingBlock(_)) => Ok(false),
+            Err(err) => {
+                warn!(target: "epoch_manager", ?err, "Unexpected error in is_last_block_in_finished_epoch");
+                Ok(false)
+            }
+        }
+    }
+
     pub fn get_epoch_id_from_prev_block(
         &self,
         parent_hash: &CryptoHash,

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

@@ -1059,13 +1059,11 @@ fn test_shard_layout_upgrade_gc_impl(resharding_type: ReshardingType, rng_seed:
 }
 
 #[test]
-#[ignore]
 fn test_shard_layout_upgrade_gc() {
     test_shard_layout_upgrade_gc_impl(ReshardingType::V1, 44);
 }
 
 #[test]
-#[ignore]
 fn test_shard_layout_upgrade_gc_v2() {
     // TODO(resharding) remove those checks once rolled out
     if checked_feature!("stable", SimpleNightshadeV2, PROTOCOL_VERSION) {