feat(mirror): Clear state parts gcp bucket on reset (near#12502)

The bucket is created along with the state dumper node. If the state
dumper node is present we want to clear the bucket every time we reset
the network to a previous state.
This can be on `hard-reset`, `new-test` or when we run `reset` to
restore from a backup.

The command will be executed on the dumper node to make sure that it has
the edit access to the bucket.

pytest/tests/mocknet/mirror.py

@@ -136,6 +136,7 @@ def hard_reset_cmd(args, traffic_generator, nodes):
     if sys.stdin.readline().strip() != 'yes':
         return
     init_neard_runners(args, traffic_generator, nodes, remove_home_dir=True)
+    _clear_state_parts_if_exists(args, nodes)
 
 
 def restart_cmd(args, traffic_generator, nodes):
@@ -235,6 +236,25 @@ def _apply_config_changes(node, state_sync_location):
         do_update_config(node, f'{key}={json.dumps(change)}')
 
 
+def _clear_state_parts_if_exists(args, nodes):
+    """Looks for the state dumper and clears the GCP bucket where it dumped the parts."""
+    # TODO: Maybe add an argument to set the epoch height from where we want to cleanup.
+    # It still works without it because the dumper node will start dumping the current epoch after reset.
+
+    state_dumper_node = nodes.find(lambda n: n.want_state_dump)
+    if state_dumper_node is None:
+        logger.info('No state dumper node found, skipping state parts cleanup.')
+        return
+
+    logger.info('State dumper node found, cleaning up state parts.')
+    bucket_name = _get_state_parts_bucket_name(args)
+    state_dumper_node.run_cmd(f'gsutil -m rm -r gs://{bucket_name}/chain_id=\*')
+
+
+def _get_state_parts_bucket_name(args):
+    return f'near-state-dumper-mocknet-{args.chain_id}-{args.start_height}-{args.unique_id}'
+
+
 def new_test_cmd(args, traffic_generator, nodes):
     prompt_setup_flags(args, [n.name() for n in nodes if n.want_state_dump])
 
@@ -282,12 +302,7 @@ def new_test_cmd(args, traffic_generator, nodes):
         }
     else:
         if args.gcs_state_sync:
-            location = {
-                "GCS": {
-                    "bucket":
-                        f'near-state-dumper-mocknet-{args.chain_id}-{args.start_height}-{args.unique_id}'
-                }
-            }
+            location = {"GCS": {"bucket": _get_state_parts_bucket_name(args)}}
         else:
             location = None
     logger.info('Applying default config changes')
@@ -297,6 +312,8 @@ def new_test_cmd(args, traffic_generator, nodes):
         logger.info('Configuring nodes for stateless protocol')
         pmap(lambda node: _apply_stateless_config(args, node), nodes)
 
+    _clear_state_parts_if_exists(args, nodes)
+
 
 def status_cmd(args, traffic_generator, nodes):
     targeted = nodes + to_list(traffic_generator)
@@ -345,6 +362,7 @@ def reset_cmd(args, traffic_generator, nodes):
     logger.info(
         'Data dir reset in progress. Run the `status` command to see when this is finished. Until it is finished, neard runners may not respond to HTTP requests.'
     )
+    _clear_state_parts_if_exists(args, nodes)
 
 
 def make_backup_cmd(args, traffic_generator, nodes):