pick(25376): fix(consistency): update formal snapshot to read new checkpoint format (#25377)

## Description

Formal snapshot restore requires fetching the last checkpoint in the
epoch being restored from, which is done through the indexing
framework's ingestion client. This codepath needed to be updated to read
the new checkpoint format.

## Test plan

Tested locally:

```
$ cargo run --bin sui-indexer-alt-consistent-store -- restore \
    --database-path /tmp/mnt                                  \
    --object-file-concurrency 20                              \
    --pipeline balances                                       \
    --pipeline object_by_owner                                \
    --pipeline object_by_type                                 \
    --remote-store-url https://checkpoints.mainnet.sui.io     \
    --http https://formal-snapshot.mainnet.sui.io

```

---

## 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:
- [ ] Indexing Framework:

---------

Co-authored-by: J <129789810+0x-j@users.noreply.github.com>
Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>
Co-authored-by: Jessie Mongeon <133128541+jessiemongeon1@users.noreply.github.com>

Author: 4 people

crates/sui-indexer-alt-consistent-store/src/restore/formal_snapshot.rs

@@ -15,8 +15,7 @@ use object_store::gcp::GoogleCloudStorageBuilder;
 use object_store::http::HttpBuilder;
 use object_store::local::LocalFileSystem;
 use sui_indexer_alt_framework::ingestion::store_client::StoreIngestionClient;
-use sui_indexer_alt_framework::types::full_checkpoint_content::CheckpointData;
-use sui_storage::blob::Blob;
+use sui_indexer_alt_framework::types::full_checkpoint_content::Checkpoint;
 use tracing::info;
 use url::Url;
 
@@ -167,26 +166,21 @@ impl FormalSnapshot {
             .cloned()
             .with_context(|| format!("Cannot find end-of-epoch checkpoint for epoch {epoch}"))?;
 
-        let CheckpointData {
-            checkpoint_summary, ..
-        } = Blob::from_bytes(
-            &client
-                .checkpoint(checkpoint)
-                .await
-                .context("Failed to fetch end-of-epoch checkpoint")?,
-        )
-        .context("Failed to deserialize end-of-epoch checkpoint")?;
+        let Checkpoint { summary, .. } = client
+            .checkpoint(checkpoint)
+            .await
+            .context("Failed to fetch end-of-epoch checkpoint")?;
 
         ensure!(
-            checkpoint_summary.epoch == epoch,
+            summary.epoch == epoch,
             "End-of-epoch checkpoint {checkpoint} does not belong to epoch {epoch}",
         );
 
         let watermark = Watermark {
             epoch_hi_inclusive: epoch,
-            checkpoint_hi_inclusive: checkpoint_summary.sequence_number,
-            tx_hi: checkpoint_summary.network_total_transactions,
-            timestamp_ms_hi_inclusive: checkpoint_summary.timestamp_ms,
+            checkpoint_hi_inclusive: summary.sequence_number,
+            tx_hi: summary.network_total_transactions,
+            timestamp_ms_hi_inclusive: summary.timestamp_ms,
         };
 
         info!(?watermark, "Anchored formal snapshot");

crates/sui-indexer-alt-framework/src/ingestion/decode.rs

@@ -0,0 +1,38 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use prost::Message;
+use sui_rpc::proto::TryFromProtoError;
+use sui_rpc::proto::sui::rpc::v2 as proto;
+
+use crate::types::full_checkpoint_content::Checkpoint;
+
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    #[error("Failed to decompress checkpoint bytes: {0}")]
+    Decompression(#[from] std::io::Error),
+
+    #[error("Failed to deserialize checkpoint protobuf: {0}")]
+    Deserialization(#[from] prost::DecodeError),
+
+    #[error("Failed to convert checkpoint protobuf to checkpoint data: {0}")]
+    ProtoConversion(#[from] Box<TryFromProtoError>),
+}
+
+impl Error {
+    pub(crate) fn reason(&self) -> &'static str {
+        match self {
+            Self::Decompression(_) => "decompression",
+            Self::Deserialization(_) => "deserialization",
+            Self::ProtoConversion(_) => "proto_conversion",
+        }
+    }
+}
+
+/// Decode the bytes of a checkpoint from the remote store. The bytes are expected to be a
+/// [Checkpoint], represented as a protobuf message, in binary form, zstd-compressed.
+pub(crate) fn checkpoint(bytes: &[u8]) -> Result<Checkpoint, Error> {
+    let decompressed = zstd::decode_all(bytes)?;
+    let proto_checkpoint = proto::Checkpoint::decode(&decompressed[..])?;
+    Ok(Checkpoint::try_from(&proto_checkpoint).map_err(Box::new)?)
+}

crates/sui-indexer-alt-framework/src/ingestion/ingestion_client.rs

@@ -17,11 +17,9 @@ use object_store::azure::MicrosoftAzureBuilder;
 use object_store::gcp::GoogleCloudStorageBuilder;
 use object_store::http::HttpBuilder;
 use object_store::local::LocalFileSystem;
-use prost::Message;
 use sui_futures::future::with_slow_future_monitor;
 use sui_rpc::Client;
 use sui_rpc::client::HeadersInterceptor;
-use sui_rpc::proto::sui::rpc::v2::Checkpoint as ProtoCheckpoint;
 use tracing::debug;
 use tracing::error;
 use tracing::warn;
@@ -30,6 +28,7 @@ use url::Url;
 use crate::ingestion::Error as IngestionError;
 use crate::ingestion::MAX_GRPC_MESSAGE_SIZE_BYTES;
 use crate::ingestion::Result as IngestionResult;
+use crate::ingestion::decode;
 use crate::ingestion::store_client::StoreIngestionClient;
 use crate::metrics::CheckpointLagMetricReporter;
 use crate::metrics::IngestionMetrics;
@@ -341,27 +340,10 @@ impl IngestionClient {
                     FetchData::Raw(bytes) => {
                         self.metrics.total_ingested_bytes.inc_by(bytes.len() as u64);
 
-                        let decompressed = zstd::decode_all(&bytes[..]).map_err(|e| {
+                        decode::checkpoint(&bytes).map_err(|e| {
                             self.metrics.inc_retry(
                                 checkpoint,
-                                "decompression",
-                                IngestionError::DeserializationError(checkpoint, e.into()),
-                            )
-                        })?;
-
-                        let proto_checkpoint =
-                            ProtoCheckpoint::decode(&decompressed[..]).map_err(|e| {
-                                self.metrics.inc_retry(
-                                    checkpoint,
-                                    "deserialization",
-                                    IngestionError::DeserializationError(checkpoint, e.into()),
-                                )
-                            })?;
-
-                        Checkpoint::try_from(&proto_checkpoint).map_err(|e| {
-                            self.metrics.inc_retry(
-                                checkpoint,
-                                "proto_conversion",
+                                e.reason(),
                                 IngestionError::DeserializationError(checkpoint, e.into()),
                             )
                         })?

crates/sui-indexer-alt-framework/src/ingestion/mod.rs

@@ -26,6 +26,7 @@ use crate::metrics::IngestionMetrics;
 use crate::types::full_checkpoint_content::Checkpoint;
 
 mod broadcaster;
+pub(crate) mod decode;
 pub mod error;
 pub mod ingestion_client;
 mod rpc_client;

crates/sui-indexer-alt-framework/src/ingestion/store_client.rs

@@ -12,10 +12,12 @@ use serde::de::DeserializeOwned;
 use tracing::debug;
 use tracing::error;
 
+use crate::ingestion::decode;
 use crate::ingestion::ingestion_client::FetchData;
 use crate::ingestion::ingestion_client::FetchError;
 use crate::ingestion::ingestion_client::FetchResult;
 use crate::ingestion::ingestion_client::IngestionClientTrait;
+use crate::types::full_checkpoint_content::Checkpoint;
 
 pub struct StoreIngestionClient {
     store: Arc<dyn ObjectStore>,
@@ -33,9 +35,13 @@ impl StoreIngestionClient {
         Ok(serde_json::from_slice(&bytes)?)
     }
 
-    /// Fetch the bytes for a checkpoint by its sequence number.
-    /// The response is the serialized representation of a checkpoint, as raw bytes.
-    pub async fn checkpoint(&self, checkpoint: u64) -> object_store::Result<Bytes> {
+    /// Fetch and decode checkpoint data by sequence number.
+    pub async fn checkpoint(&self, checkpoint: u64) -> anyhow::Result<Checkpoint> {
+        let bytes = self.checkpoint_bytes(checkpoint).await?;
+        Ok(decode::checkpoint(&bytes)?)
+    }
+
+    async fn checkpoint_bytes(&self, checkpoint: u64) -> object_store::Result<Bytes> {
         self.bytes(ObjectPath::from(format!("{checkpoint}.binpb.zst")))
             .await
     }
@@ -58,7 +64,7 @@ impl IngestionClientTrait for StoreIngestionClient {
     /// - server errors (5xx),
     /// - issues getting a full response.
     async fn fetch(&self, checkpoint: u64) -> FetchResult {
-        match self.checkpoint(checkpoint).await {
+        match self.checkpoint_bytes(checkpoint).await {
             Ok(bytes) => Ok(FetchData::Raw(bytes)),
             Err(ObjectStoreError::NotFound { .. }) => {
                 debug!(checkpoint, "Checkpoint not found");