check for success before snapshot restore (MystenLabs#15386)

## Description 

It's happened a few times where users try to restore a snapshot that
hasn't been uploaded yet or does not exist, this will cleanly exit in
the case that the success marker hasn't been uploaded.

## Test Plan 

tested locally:
```
$ target/debug/sui-tool download-db-snapshot --epoch "439" --genesis "/opt/sui/config/genesis.blob" --formal --network mainnet --path /opt/sui/db/authorities_db --num-parallel-downloads 50 --snapshot-bucket-type=gcs

thread 'main' panicked at /Users/johnmartin/src/github.com/johnjmartin/sui/crates/sui-tool/src/commands.rs:714:21:
Aborting snapshot restore: missing success marker at mysten-mainnet-formal/epoch_439/_SUCCESS
```
---
If your changes are not user-facing and not a breaking change, you can
skip the following section. Otherwise, please indicate what changed, and
then add to the Release Notes section as highlighted during the release
process.

### Type of Change (Check all that apply)

- [ ] protocol change
- [ ] user-visible impact
- [ ] breaking change for a client SDKs
- [ ] breaking change for FNs (FN binary must upgrade)
- [ ] breaking change for validators or node operators (must upgrade
binaries)
- [ ] breaking change for on-chain data layout
- [ ] necessitate either a data wipe or data migration

### Release notes

crates/sui-storage/src/object_store/util.rs

@@ -34,6 +34,10 @@ pub async fn get<S: ObjectStoreGetExt>(store: &S, src: &Path) -> Result<Bytes> {
     Ok(bytes)
 }
 
+pub async fn exists<S: ObjectStoreGetExt>(store: &S, src: &Path) -> bool {
+    store.get_bytes(src).await.is_ok()
+}
+
 pub async fn put<S: ObjectStorePutExt>(store: &S, src: &Path, bytes: Bytes) -> Result<()> {
     retry(backoff::ExponentialBackoff::default(), || async {
         if !bytes.is_empty() {

crates/sui-tool/src/commands.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{
+    check_completed_snapshot,
     db_tool::{execute_db_tool_command, print_db_all_tables, DbToolCommand},
     download_db_snapshot, download_formal_snapshot, dump_checkpoints_from_archive, get_object,
     get_transaction_block, make_clients, pkg_dump, restore_from_db_checkpoint,
@@ -709,6 +710,12 @@ impl ToolCommand {
                     }
                 };
 
+                if let Err(e) = check_completed_snapshot(&snapshot_store_config, epoch).await {
+                    panic!(
+                        "Aborting snapshot restore: {}, snapshot may not be uploaded yet",
+                        e
+                    );
+                }
                 if formal {
                     let verify = verify.unwrap_or(true);
                     download_formal_snapshot(

crates/sui-tool/src/lib.rs

@@ -21,6 +21,7 @@ use sui_core::authority_client::{AuthorityAPI, NetworkAuthorityClient};
 use sui_network::default_mysten_network_config;
 use sui_protocol_config::Chain;
 use sui_sdk::SuiClientBuilder;
+use sui_storage::object_store::http::HttpDownloaderBuilder;
 use sui_types::accumulator::Accumulator;
 use sui_types::crypto::AuthorityPublicKeyBytes;
 use sui_types::messages_grpc::LayoutGenerationOption;
@@ -48,7 +49,7 @@ use sui_core::epoch::committee_store::CommitteeStore;
 use sui_core::storage::RocksDbStore;
 use sui_snapshot::reader::StateSnapshotReaderV1;
 use sui_snapshot::setup_db_state;
-use sui_storage::object_store::util::{copy_file, get_path};
+use sui_storage::object_store::util::{copy_file, exists, get_path};
 use sui_storage::object_store::{ObjectStoreConfig, ObjectStoreType};
 use sui_storage::verify_checkpoint_range;
 use sui_types::messages_checkpoint::{CheckpointCommitment, ECMHLiveObjectSetDigest};
@@ -794,6 +795,30 @@ fn start_summary_sync(
     })
 }
 
+pub async fn check_completed_snapshot(
+    snapshot_store_config: &ObjectStoreConfig,
+    epoch: EpochId,
+) -> Result<(), anyhow::Error> {
+    let success_marker = format!("epoch_{}/_SUCCESS", epoch);
+    let remote_object_store = if snapshot_store_config.no_sign_request {
+        snapshot_store_config.make_http()?
+    } else {
+        snapshot_store_config.make().map(Arc::new)?
+    };
+    if exists(&remote_object_store, &get_path(success_marker.as_str())).await {
+        Ok(())
+    } else {
+        Err(anyhow!(
+            "missing success marker at {}/{}",
+            snapshot_store_config
+                .bucket
+                .as_ref()
+                .unwrap_or(&"unknown_bucket".to_string()),
+            success_marker
+        ))
+    }
+}
+
 pub async fn download_formal_snapshot(
     path: &Path,
     epoch: EpochId,