[archival deprecation] migrate indexer restorer (#21999)

## Description 

migrates indexer restoration from previous archive to checkpoint bucket

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] gRPC:
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:

Author: phoenix-o

Cargo.lock

@@ -19,9 +19,9 @@ diesel-async = { workspace = true, features = ["bb8", "postgres", "async-connect
 indicatif.workspace = true
 object_store.workspace = true
 prometheus.workspace = true
-sui-archival.workspace = true
 sui-config.workspace = true
 sui-core.workspace = true
+sui-data-ingestion-core.workspace = true
 sui-field-count.workspace = true
 sui-indexer-alt-schema.workspace = true
 sui-indexer-alt-framework.workspace = true

crates/sui-indexer-alt-restorer/Cargo.toml

@@ -1,16 +1,9 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::num::NonZeroUsize;
-
-use prometheus::Registry;
-use tracing::info;
-
-use sui_archival::reader::{ArchiveReader, ArchiveReaderMetrics};
-use sui_config::node::ArchiveReaderConfig;
-use sui_config::object_storage_config::{ObjectStoreConfig, ObjectStoreType};
-
 use crate::Args;
+use sui_data_ingestion_core::end_of_epoch_data;
+use tracing::info;
 
 #[derive(Clone, Debug)]
 pub(crate) struct ArchivalCheckpointInfo {
@@ -21,26 +14,8 @@ impl ArchivalCheckpointInfo {
     /// Reads checkpoint information from archival storage to determine,
     /// specifically the next checkpoint number after `start_epoch` for watermarking.
     pub async fn read_archival_checkpoint_info(args: &Args) -> anyhow::Result<Self> {
-        // Configure GCS object store and initialize archive reader with minimal concurrency.
-        let archive_store_config = ObjectStoreConfig {
-            object_store: Some(ObjectStoreType::GCS),
-            bucket: Some(args.archive_bucket.clone()),
-            object_store_connection_limit: 1, // 1 connection is sufficient
-            no_sign_request: false,
-            ..Default::default()
-        };
-        let archive_reader_config = ArchiveReaderConfig {
-            remote_store_config: archive_store_config,
-            download_concurrency: NonZeroUsize::new(1).unwrap(),
-            ingestion_url: None,
-        };
-        let metrics = ArchiveReaderMetrics::new(&Registry::default());
-        let archive_reader = ArchiveReader::new(archive_reader_config, &metrics)?;
-
-        // Sync and read manifest to get next checkpoint after `start_epoch` for watermarking.
-        archive_reader.sync_manifest_once().await?;
-        let manifest = archive_reader.get_manifest().await?;
-        let next_checkpoint_after_epoch = manifest.next_checkpoint_after_epoch(args.start_epoch);
+        let checkpoints = end_of_epoch_data(args.archive_url.clone(), vec![], 5).await?;
+        let next_checkpoint_after_epoch = checkpoints[args.start_epoch as usize] + 1;
         info!(
             epoch = args.start_epoch,
             checkpoint = next_checkpoint_after_epoch,

crates/sui-indexer-alt-restorer/src/archives.rs

@@ -28,8 +28,8 @@ pub struct Args {
     pub snapshot_bucket: String,
 
     /// Bucket to fetch archive files from.
-    #[clap(long, env = "ARCHIVE_BUCKET", required = true)]
-    pub archive_bucket: String,
+    #[clap(long, env = "ARCHIVE_URL", required = true)]
+    pub archive_url: String,
 
     /// Local directory to temporarily store snapshot files.
     #[clap(long, env = "SNAPSHOT_LOCAL_DIR", required = true)]