Upgrade mvr-indexer to latest sui-indexer-alt-framework

Cargo.toml

@@ -33,21 +33,24 @@ futures = "0.3.31"
 tempfile = "3.20.0"
 
 # Main Sui Mono-repo
-sui-indexer-alt-metrics = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-mysten-metrics = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-telemetry-subscribers = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-sui-pg-db = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-move-core-types = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-sui-types = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-sui-package-resolver = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-sui-move-build = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-move-binary-format = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-sui-indexer-alt-framework = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-sui-rpc-api = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-sui-storage = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62" }
-sui-field-count = { git = "https://github.com/MystenLabs/sui.git", rev = "3b96ab72dd5db2fb800837d6067bf45839178b62"}
+sui-indexer-alt-metrics = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+mysten-metrics = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+telemetry-subscribers = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+sui-pg-db = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+move-core-types = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+sui-types = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+sui-package-resolver = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+sui-move-build = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+move-binary-format = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+sui-indexer-alt-framework = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+sui-rpc-api = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+sui-storage = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb" }
+sui-field-count = { git = "https://github.com/MystenLabs/sui.git", rev = "211e22a5b0e08f8840f6a3e74120e1b4b04d5adb"}
 
 # New Rust SDK
-sui-sdk-types = { git = "https://github.com/mystenlabs/sui-rust-sdk", package = "sui-sdk-types", rev="048124e484f14b9bf2a402227c9bc255c7621bc1"}
+sui-sdk-types = { git = "https://github.com/MystenLabs/sui-rust-sdk", package = "sui-sdk-types", rev = "339c2272fd5b8fb4e1fa6662cfa9acdbb0d05704", features = ["hash", "serde"] }
+
+# async-graphql (use same fork as sui to avoid version conflicts)
+async-graphql = { git = "https://github.com/amnn/async-graphql", branch = "v7.0.1-react-18-graphiql-4", features = ["dataloader"] }
 
 insta = "1.43.2"

crates/mvr-api/Cargo.toml

@@ -5,7 +5,7 @@ edition = "2021"
 
 [dependencies]
 sui-package-resolver.workspace = true
-async-graphql = { version = "=7.0.1", features = ["dataloader"] }
+async-graphql.workspace = true
 axum = "0.8.1"
 tokio = { workspace = true, features = ["full"] }
 tower-http = { version = "0.6.2", features = ["cors"] }

crates/mvr-api/src/lib.rs

@@ -34,11 +34,7 @@ pub async fn run_server(
     let registry = Registry::new_custom(Some("mvr_api".into()), None)
         .expect("Failed to create Prometheus registry.");
 
-    let metrics = MetricsService::new(
-        MetricsArgs { metrics_address },
-        registry,
-        cancellation_token.clone(),
-    );
+    let metrics = MetricsService::new(MetricsArgs { metrics_address }, registry);
 
     let mvr_source_header = HeaderName::from_str(MVR_SOURCE_HEADER)?;
 

crates/mvr-indexer/src/handlers/git_info_handler.rs

@@ -16,11 +16,10 @@ use mvr_schema::models::GitInfo;
 use serde::de::DeserializeOwned;
 use std::marker::PhantomData;
 use std::sync::Arc;
-use sui_indexer_alt_framework::pipeline::concurrent::Handler;
 use sui_indexer_alt_framework::pipeline::Processor;
-use sui_pg_db::{Connection, Db};
+use sui_indexer_alt_framework::postgres::{handler::Handler, Connection};
+use sui_indexer_alt_framework::types::full_checkpoint_content::Checkpoint;
 use sui_types::base_types::MoveObjectType;
-use sui_types::full_checkpoint_content::CheckpointData;
 use sui_types::object::Object;
 
 pub struct GitInfoHandler<T> {
@@ -92,12 +91,10 @@ impl MoveObjectProcessor<TestnetGitInfo, GitInfo> for GitInfoHandler<TestnetGitI
 }
 
 #[async_trait]
-impl<T: MoveStructType + DeserializeOwned> Handler for GitInfoHandler<T>
+impl<T: MoveStructType + DeserializeOwned + Send + Sync + 'static> Handler for GitInfoHandler<T>
 where
     Self: MoveObjectProcessor<T, GitInfo>,
 {
-    type Store = Db;
-
     async fn commit<'a>(
         values: &[Self::Value],
         conn: &mut Connection<'a>,
@@ -127,30 +124,32 @@ where
     }
 }
 
-impl<T: MoveStructType + DeserializeOwned> Processor for GitInfoHandler<T>
+#[async_trait]
+impl<T: MoveStructType + DeserializeOwned + Send + Sync + 'static> Processor for GitInfoHandler<T>
 where
     Self: MoveObjectProcessor<T, GitInfo>,
 {
     const NAME: &'static str = Self::PROC_NAME;
     type Value = GitInfo;
 
-    fn process(&self, checkpoint: &Arc<CheckpointData>) -> anyhow::Result<Vec<Self::Value>> {
+    async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Self::Value>> {
         checkpoint
             .transactions
             .iter()
             .try_fold(vec![], |result, tx| {
-                tx.output_objects.iter().try_fold(result, |mut result, o| {
-                    if matches!(o.type_(), Some(t) if t == &self.type_) {
-                        if let Some(move_obj) = o.data.try_as_move() {
-                            result.push(Self::process_move_object(
-                                self.chain_id.clone(),
-                                bcs::from_bytes(move_obj.contents())?,
-                                o,
-                            )?)
+                tx.output_objects(&checkpoint.object_set)
+                    .try_fold(result, |mut result, o| {
+                        if matches!(o.type_(), Some(t) if t == &self.type_) {
+                            if let Some(move_obj) = o.data.try_as_move() {
+                                result.push(Self::process_move_object(
+                                    self.chain_id.clone(),
+                                    bcs::from_bytes(move_obj.contents())?,
+                                    o,
+                                )?)
+                            }
                         }
-                    }
-                    Ok(result)
-                })
+                        Ok(result)
+                    })
             })
     }
 }

crates/mvr-indexer/src/handlers/mod.rs

@@ -15,11 +15,10 @@ use move_core_types::language_storage::StructTag as MoveStructTag;
 use mvr_schema::models::Package;
 use std::str::FromStr;
 use std::sync::Arc;
-use sui_indexer_alt_framework::pipeline::concurrent::Handler;
 use sui_indexer_alt_framework::pipeline::Processor;
-use sui_pg_db::{Connection, Db};
+use sui_indexer_alt_framework::postgres::{handler::Handler, Connection};
+use sui_indexer_alt_framework::types::full_checkpoint_content::Checkpoint;
 use sui_sdk_types::StructTag;
-use sui_types::full_checkpoint_content::CheckpointData;
 use tracing::info;
 
 // Convert rust sdk struct tag to move struct tag.
@@ -80,21 +79,20 @@ pub struct NoOpsHandler;
 
 #[async_trait]
 impl Handler for NoOpsHandler {
-    type Store = Db;
-
     async fn commit<'a>(_: &[Self::Value], _: &mut Connection<'a>) -> anyhow::Result<usize> {
         Ok(0)
     }
 }
 
+#[async_trait]
 impl Processor for NoOpsHandler {
     const NAME: &'static str = "No Ops Handler";
     type Value = Package;
 
-    fn process(&self, checkpoint: &Arc<CheckpointData>) -> anyhow::Result<Vec<Self::Value>> {
+    async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Self::Value>> {
         info!(
             "Processed checkpoint: {}",
-            checkpoint.checkpoint_summary.sequence_number
+            checkpoint.summary.sequence_number
         );
         Ok(vec![])
     }

crates/mvr-indexer/src/handlers/name_record_handler.rs

@@ -12,13 +12,12 @@ use mvr_types::name::Name;
 use mvr_types::name_service::Domain;
 use std::str::FromStr;
 use std::sync::Arc;
-use sui_indexer_alt_framework::pipeline::concurrent::Handler;
 use sui_indexer_alt_framework::pipeline::Processor;
-use sui_pg_db::{Connection, Db};
+use sui_indexer_alt_framework::postgres::{handler::Handler, Connection};
+use sui_indexer_alt_framework::types::full_checkpoint_content::Checkpoint;
 use sui_sdk_types::StructTag;
 use sui_types::base_types::MoveObjectType;
 use sui_types::dynamic_field::Field;
-use sui_types::full_checkpoint_content::CheckpointData;
 
 pub struct NameRecordHandler {
     type_: MoveObjectType,
@@ -53,8 +52,6 @@ impl NameRecordHandler {
 }
 #[async_trait]
 impl Handler for NameRecordHandler {
-    type Store = Db;
-
     async fn commit<'a>(
         values: &[Self::Value],
         conn: &mut Connection<'a>,
@@ -83,44 +80,46 @@ impl Handler for NameRecordHandler {
     }
 }
 
+#[async_trait]
 impl Processor for NameRecordHandler {
     const NAME: &'static str = "NameRecord";
     type Value = NameRecord;
 
-    fn process(&self, checkpoint: &Arc<CheckpointData>) -> anyhow::Result<Vec<Self::Value>> {
+    async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Self::Value>> {
         checkpoint
             .transactions
             .iter()
             .try_fold(vec![], |result, tx| {
-                tx.output_objects.iter().try_fold(result, |mut result, o| {
-                    // TODO: do we need to check if the parent of the DF table entry is the MVR registry?
-                    if matches!(o.type_(), Some(t) if t == &self.type_) {
-                        if let Some(move_obj) = o.data.try_as_move() {
-                            let data: Field<Name, AppRecord> =
-                                bcs::from_bytes(move_obj.contents())?;
-                            let Name { org, app } = data.name;
-                            let name = Name::new(Domain { labels: org.labels }, app);
-                            let AppRecord {
-                                app_info,
-                                networks,
-                                metadata,
-                                ..
-                            } = data.value;
+                tx.output_objects(&checkpoint.object_set)
+                    .try_fold(result, |mut result, o| {
+                        // TODO: do we need to check if the parent of the DF table entry is the MVR registry?
+                        if matches!(o.type_(), Some(t) if t == &self.type_) {
+                            if let Some(move_obj) = o.data.try_as_move() {
+                                let data: Field<Name, AppRecord> =
+                                    bcs::from_bytes(move_obj.contents())?;
+                                let Name { org, app } = data.name;
+                                let name = Name::new(Domain { labels: org.labels }, app);
+                                let AppRecord {
+                                    app_info,
+                                    networks,
+                                    metadata,
+                                    ..
+                                } = data.value;
 
-                            result.push(NameRecord {
-                                name: name.to_string(),
-                                object_version: o.version().value() as i64,
-                                mainnet_id: app_info
-                                    .and_then(|info| Some(info.package_info_id?.to_string())),
-                                testnet_id: networks
-                                    .get(&TESTNET_CHAIN_ID.to_string())
-                                    .and_then(|info| Some(info.package_info_id?.to_string())),
-                                metadata: serde_json::to_value(into_hash_map(metadata))?,
-                            })
+                                result.push(NameRecord {
+                                    name: name.to_string(),
+                                    object_version: o.version().value() as i64,
+                                    mainnet_id: app_info
+                                        .and_then(|info| Some(info.package_info_id?.to_string())),
+                                    testnet_id: networks
+                                        .get(&TESTNET_CHAIN_ID.to_string())
+                                        .and_then(|info| Some(info.package_info_id?.to_string())),
+                                    metadata: serde_json::to_value(into_hash_map(metadata))?,
+                                })
+                            }
                         }
-                    }
-                    Ok(result)
-                })
+                        Ok(result)
+                    })
             })
     }
 }

crates/mvr-indexer/src/handlers/package_handler.rs

@@ -1,18 +1,19 @@
 use crate::{MAINNET_CHAIN_ID, TESTNET_CHAIN_ID};
+use async_trait::async_trait;
 use chrono::DateTime;
 use diesel_async::RunQueryDsl;
 use itertools::Itertools;
 use move_binary_format::CompiledModule;
 use mvr_schema::models::{Package, PackageDependency};
 use std::collections::HashSet;
 use std::sync::Arc;
-use sui_indexer_alt_framework::pipeline::concurrent::Handler;
 use sui_indexer_alt_framework::pipeline::Processor;
-use sui_pg_db::{Connection, Db};
+use sui_indexer_alt_framework::postgres::{handler::Handler, Connection};
+use sui_indexer_alt_framework::types::full_checkpoint_content::Checkpoint;
 use sui_types::base_types::ObjectID;
-use sui_types::full_checkpoint_content::CheckpointData;
 use sui_types::move_package::MovePackage;
 use sui_types::object::Data;
+use sui_types::transaction::TransactionDataAPI;
 
 pub struct PackageHandler<const MAINNET: bool>;
 
@@ -24,10 +25,8 @@ impl<const MAINNET: bool> PackageHandler<MAINNET> {
     };
 }
 
-#[async_trait::async_trait]
+#[async_trait]
 impl<const MAINNET: bool> Handler for PackageHandler<MAINNET> {
-    type Store = Db;
-
     async fn commit<'a>(
         values: &[Self::Value],
         conn: &mut Connection<'a>,
@@ -53,6 +52,7 @@ impl<const MAINNET: bool> Handler for PackageHandler<MAINNET> {
     }
 }
 
+#[async_trait]
 impl<const MAINNET: bool> Processor for PackageHandler<MAINNET> {
     const NAME: &'static str = if MAINNET {
         "Package - Mainnet"
@@ -61,15 +61,14 @@ impl<const MAINNET: bool> Processor for PackageHandler<MAINNET> {
     };
     type Value = Package;
 
-    fn process(&self, checkpoint: &Arc<CheckpointData>) -> anyhow::Result<Vec<Self::Value>> {
-        let timestamp =
-            DateTime::from_timestamp_millis(checkpoint.checkpoint_summary.timestamp_ms as i64)
-                .unwrap()
-                .naive_utc();
+    async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Self::Value>> {
+        let timestamp = DateTime::from_timestamp_millis(checkpoint.summary.timestamp_ms as i64)
+            .unwrap()
+            .naive_utc();
 
         let mut results = vec![];
         for tx in &checkpoint.transactions {
-            for o in &tx.output_objects {
+            for o in tx.output_objects(&checkpoint.object_set) {
                 if let Data::Package(p) = &o.data {
                     let package_id = p.id().to_hex_uncompressed();
                     let deps = package_dependencies(Self::CHAIN_ID.to_string(), p)?;
@@ -80,7 +79,7 @@ impl<const MAINNET: bool> Processor for PackageHandler<MAINNET> {
                         move_package: bcs::to_bytes(p)?,
                         chain_id: Self::CHAIN_ID.to_string(),
                         tx_hash: tx.transaction.digest().base58_encode(),
-                        sender: tx.transaction.sender_address().to_string(),
+                        sender: tx.transaction.sender().to_string(),
                         timestamp,
                         deps,
                     })

crates/mvr-indexer/src/handlers/package_info_handler.rs

@@ -12,12 +12,11 @@ use mvr_schema::models::PackageInfo;
 use serde::de::DeserializeOwned;
 use std::marker::PhantomData;
 use std::sync::Arc;
-use sui_indexer_alt_framework::pipeline::concurrent::Handler;
 use sui_indexer_alt_framework::pipeline::Processor;
-use sui_pg_db::{Connection, Db};
+use sui_indexer_alt_framework::postgres::{handler::Handler, Connection};
+use sui_indexer_alt_framework::types::full_checkpoint_content::Checkpoint;
 use sui_sdk_types::Address;
 use sui_types::collection_types::VecMap;
-use sui_types::full_checkpoint_content::CheckpointData;
 use sui_types::object::Object;
 
 const DEFAULT_NAME_KEY: &str = "default";
@@ -114,12 +113,10 @@ impl MoveObjectProcessor<TestnetPackageInfo, PackageInfo>
 }
 
 #[async_trait]
-impl<T: MoveStructType + DeserializeOwned> Handler for PackageInfoHandler<T>
+impl<T: MoveStructType + DeserializeOwned + Send + Sync + 'static> Handler for PackageInfoHandler<T>
 where
     Self: MoveObjectProcessor<T, PackageInfo>,
 {
-    type Store = Db;
-
     async fn commit<'a>(
         values: &[Self::Value],
         conn: &mut Connection<'a>,
@@ -149,16 +146,18 @@ where
     }
 }
 
-impl<T: MoveStructType + DeserializeOwned> Processor for PackageInfoHandler<T>
+#[async_trait]
+impl<T: MoveStructType + DeserializeOwned + Send + Sync + 'static> Processor
+    for PackageInfoHandler<T>
 where
     Self: MoveObjectProcessor<T, PackageInfo>,
 {
     const NAME: &'static str = Self::PROC_NAME;
     type Value = PackageInfo;
 
-    fn process(&self, checkpoint: &Arc<CheckpointData>) -> anyhow::Result<Vec<Self::Value>> {
+    async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Self::Value>> {
         checkpoint.transactions.iter().try_fold(vec![], |result, tx| {
-            tx.output_objects.iter().try_fold(result, |mut result, obj| {
+            tx.output_objects(&checkpoint.object_set).try_fold(result, |mut result, obj| {
                 if matches!(obj.type_(), Some(t) if matches!(t.other(), Some(s) if s == &convert_struct_tag(T::struct_type())) ) {
                     if let Some(move_obj) = obj.data.try_as_move() {
                         let move_obj: T = bcs::from_bytes(move_obj.contents())?;