implements sync state rpc

crates/rooch-executor/src/actor/messages.rs

@@ -223,3 +223,12 @@ pub struct SaveStateChangeSetMessage {
 impl Message for SaveStateChangeSetMessage {
     type Result = Result<()>;
 }
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct GetStateChangeSetsMessage {
+    pub tx_orders: Vec<u64>,
+}
+
+impl Message for GetStateChangeSetsMessage {
+    type Result = Result<Vec<Option<StateChangeSetExt>>>;
+}

crates/rooch-executor/src/actor/reader_executor.rs

@@ -3,8 +3,8 @@
 
 use super::messages::{
     AnnotatedStatesMessage, ExecuteViewFunctionMessage, GetAnnotatedEventsByEventHandleMessage,
-    GetAnnotatedEventsByEventIDsMessage, GetEventsByEventHandleMessage, RefreshStateMessage,
-    StatesMessage,
+    GetAnnotatedEventsByEventIDsMessage, GetEventsByEventHandleMessage, GetStateChangeSetsMessage,
+    RefreshStateMessage, StatesMessage,
 };
 use crate::actor::messages::{
     GetEventsByEventIDsMessage, GetTxExecutionInfosByHashMessage, ListAnnotatedStatesMessage,
@@ -24,7 +24,7 @@ use moveos_types::function_return_value::AnnotatedFunctionReturnValue;
 use moveos_types::moveos_std::event::EventHandle;
 use moveos_types::moveos_std::event::{AnnotatedEvent, Event};
 use moveos_types::moveos_std::object::ObjectMeta;
-use moveos_types::state::{AnnotatedState, ObjectState};
+use moveos_types::state::{AnnotatedState, ObjectState, StateChangeSetExt};
 use moveos_types::state_resolver::RootObjectResolver;
 use moveos_types::state_resolver::{AnnotatedStateKV, AnnotatedStateReader, StateKV, StateReader};
 use moveos_types::transaction::TransactionExecutionInfo;
@@ -323,3 +323,17 @@ impl Handler<EventData> for ReaderExecutorActor {
         Ok(())
     }
 }
+
+#[async_trait]
+impl Handler<GetStateChangeSetsMessage> for ReaderExecutorActor {
+    async fn handle(
+        &mut self,
+        msg: GetStateChangeSetsMessage,
+        _ctx: &mut ActorContext,
+    ) -> Result<Vec<Option<StateChangeSetExt>>> {
+        let GetStateChangeSetsMessage { tx_orders } = msg;
+        self.rooch_store
+            .state_store
+            .multi_get_state_change_set(tx_orders)
+    }
+}

crates/rooch-executor/src/proxy/mod.rs

@@ -3,9 +3,9 @@
 
 use crate::actor::messages::{
     ConvertL2TransactionData, DryRunTransactionResult, GetAnnotatedEventsByEventIDsMessage,
-    GetEventsByEventHandleMessage, GetEventsByEventIDsMessage, GetTxExecutionInfosByHashMessage,
-    ListAnnotatedStatesMessage, ListStatesMessage, RefreshStateMessage, SaveStateChangeSetMessage,
-    ValidateL1BlockMessage, ValidateL1TxMessage,
+    GetEventsByEventHandleMessage, GetEventsByEventIDsMessage, GetStateChangeSetsMessage,
+    GetTxExecutionInfosByHashMessage, ListAnnotatedStatesMessage, ListStatesMessage,
+    RefreshStateMessage, SaveStateChangeSetMessage, ValidateL1BlockMessage, ValidateL1TxMessage,
 };
 use crate::actor::reader_executor::ReaderExecutorActor;
 use crate::actor::{
@@ -251,6 +251,15 @@ impl ExecutorProxy {
             .map_err(|e| anyhow!(format!("Save state change set error: {:?}", e)))
     }
 
+    pub async fn get_state_change_sets(
+        &self,
+        tx_orders: Vec<u64>,
+    ) -> Result<Vec<Option<StateChangeSetExt>>> {
+        self.reader_actor
+            .send(GetStateChangeSetsMessage { tx_orders })
+            .await?
+    }
+
     pub async fn chain_id(&self) -> Result<ChainID> {
         self.get_states(AccessPath::object(ChainID::chain_id_object_id()))
             .await?

crates/rooch-indexer/src/actor/indexer.rs

@@ -237,15 +237,12 @@ impl Handler<IndexerRevertMessage> for IndexerActor {
     async fn handle(&mut self, msg: IndexerRevertMessage, _ctx: &mut ActorContext) -> Result<()> {
         let IndexerRevertMessage {
             revert_tx_order,
-            // revert_ledger_tx,
-            // revert_execution_info,
             revert_state_change_set,
             root,
             object_mapping,
         } = msg;
 
         self.root = root;
-        // let tx_order = ledger_transaction.sequence_info.tx_order;
 
         // 1. revert indexer transaction
         self.indexer_store
@@ -259,14 +256,6 @@ impl Handler<IndexerRevertMessage> for IndexerActor {
         let mut state_index_generator = IndexerObjectStatesIndexGenerator::default();
         let mut indexer_object_state_change_set = IndexerObjectStateChangeSet::default();
 
-        // // set genesis tx_order and state_index_generator for new indexer revert
-        // let tx_order: u64 = 0;
-        // let last_state_index = self
-        //     .query_last_state_index_by_tx_order(tx_order, state_type.clone())
-        //     .await?;
-        // let mut state_index_generator = last_state_index.map_or(0, |x| x + 1);
-
-        // let object_mapping = HashMap::<ObjectID, ObjectMeta>::new();
         for (_feild_key, object_change) in revert_state_change_set.state_change_set.changes {
             let _ = handle_revert_object_change(
                 &mut state_index_generator,
@@ -279,7 +268,6 @@ impl Handler<IndexerRevertMessage> for IndexerActor {
         self.indexer_store
             .apply_object_states(indexer_object_state_change_set)?;
 
-        // self.indexer_store.revert_states(object_state_change_set)?;
         Ok(())
     }
 }

crates/rooch-indexer/src/actor/messages.rs

@@ -153,8 +153,6 @@ impl Message for IndexerApplyObjectStatesMessage {
 #[derive(Debug, Serialize, Deserialize)]
 pub struct IndexerRevertMessage {
     pub revert_tx_order: u64,
-    // pub revert_ledger_tx: LedgerTransaction,
-    // pub revert_execution_info: TransactionExecutionInfo,
     pub revert_state_change_set: StateChangeSetExt,
     pub root: ObjectMeta,
     pub object_mapping: HashMap<ObjectID, ObjectMeta>,

crates/rooch-indexer/src/proxy/mod.rs

@@ -214,17 +214,13 @@ impl IndexerProxy {
     pub async fn revert_indexer(
         &self,
         revert_tx_order: u64,
-        // revert_ledger_tx: LedgerTransaction,
-        // revert_execution_info: TransactionExecutionInfo,
         revert_state_change_set: StateChangeSetExt,
         root: ObjectMeta,
         object_mapping: HashMap<ObjectID, ObjectMeta>,
     ) -> Result<()> {
         self.actor
             .notify(IndexerRevertMessage {
                 revert_tx_order,
-                // revert_ledger_tx,
-                // revert_execution_info,
                 revert_state_change_set,
                 root,
                 object_mapping,

crates/rooch-indexer/src/tests/test_indexer.rs

@@ -62,7 +62,7 @@ async fn test_transaction_store() -> Result<()> {
     let transactions = vec![indexer_transaction];
     indexer_store.persist_transactions(transactions)?;
 
-    let filter = TransactionFilter::Sender(random_moveos_tx.ctx.sender.into());
+    let filter = TransactionFilter::Sender(random_moveos_tx.ctx.sender);
     let query_transactions =
         indexer_reader.query_transactions_with_filter(filter, None, 1, true)?;
     assert_eq!(query_transactions.len(), 1);
@@ -97,7 +97,7 @@ async fn test_event_store() -> Result<()> {
     let events = vec![indexer_event];
     indexer_store.persist_events(events)?;
 
-    let filter = EventFilter::Sender(random_moveos_tx.ctx.sender.into());
+    let filter = EventFilter::Sender(random_moveos_tx.ctx.sender);
     let query_events = indexer_reader.query_events_with_filter(filter, None, 1, true)?;
     assert_eq!(query_events.len(), 1);
     Ok(())
@@ -250,7 +250,7 @@ async fn test_escape_transaction() -> Result<()> {
     let transactions = vec![indexer_transaction];
     indexer_store.persist_transactions(transactions)?;
 
-    let filter = TransactionFilter::Sender(random_moveos_tx.ctx.sender.into());
+    let filter = TransactionFilter::Sender(random_moveos_tx.ctx.sender);
     let query_transactions =
         indexer_reader.query_transactions_with_filter(filter, None, 1, true)?;
     assert_eq!(query_transactions.len(), 1);

crates/rooch-open-rpc-spec/schemas/openrpc.json

@@ -746,6 +746,44 @@
           "$ref": "#/components/schemas/primitive_types::H256"
         }
       }
+    },
+    {
+      "name": "rooch_syncStates",
+      "description": "Sync state change sets",
+      "params": [
+        {
+          "name": "filter",
+          "required": true,
+          "schema": {
+            "$ref": "#/components/schemas/SyncStateFilterView"
+          }
+        },
+        {
+          "name": "cursor",
+          "schema": {
+            "$ref": "#/components/schemas/u64"
+          }
+        },
+        {
+          "name": "limit",
+          "schema": {
+            "$ref": "#/components/schemas/u64"
+          }
+        },
+        {
+          "name": "query_option",
+          "schema": {
+            "$ref": "#/components/schemas/QueryOptions"
+          }
+        }
+      ],
+      "result": {
+        "name": "StateChangeSetPageView",
+        "required": true,
+        "schema": {
+          "$ref": "#/components/schemas/PageView_for_StateChangeSetWithTxOrderView_and_u64"
+        }
+      }
     }
   ],
   "components": {
@@ -2479,6 +2517,35 @@
           }
         }
       },
+      "PageView_for_StateChangeSetWithTxOrderView_and_u64": {
+        "description": "`next_cursor` points to the last item in the page; Reading with `next_cursor` will start from the next item after `next_cursor` if `next_cursor` is `Some`, otherwise it will start from the first item.",
+        "type": "object",
+        "required": [
+          "data",
+          "has_next_page"
+        ],
+        "properties": {
+          "data": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/StateChangeSetWithTxOrderView"
+            }
+          },
+          "has_next_page": {
+            "type": "boolean"
+          },
+          "next_cursor": {
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/u64"
+              },
+              {
+                "type": "null"
+              }
+            ]
+          }
+        }
+      },
       "PageView_for_StateKVView_and_String": {
         "description": "`next_cursor` points to the last item in the page; Reading with `next_cursor` will start from the next item after `next_cursor` if `next_cursor` is `Some`, otherwise it will start from the first item.",
         "type": "object",
@@ -2712,6 +2779,21 @@
           }
         }
       },
+      "StateChangeSetWithTxOrderView": {
+        "type": "object",
+        "required": [
+          "state_change_set",
+          "tx_order"
+        ],
+        "properties": {
+          "state_change_set": {
+            "$ref": "#/components/schemas/StateChangeSetView"
+          },
+          "tx_order": {
+            "$ref": "#/components/schemas/u64"
+          }
+        }
+      },
       "StateKVView": {
         "type": "object",
         "required": [
@@ -2742,6 +2824,30 @@
           }
         }
       },
+      "SyncStateFilterView": {
+        "oneOf": [
+          {
+            "description": "Sync by object id.",
+            "type": "object",
+            "required": [
+              "object_i_d"
+            ],
+            "properties": {
+              "object_i_d": {
+                "$ref": "#/components/schemas/ObjectID"
+              }
+            },
+            "additionalProperties": false
+          },
+          {
+            "description": "Sync all.",
+            "type": "string",
+            "enum": [
+              "all"
+            ]
+          }
+        ]
+      },
       "TransactionExecutionInfoView": {
         "type": "object",
         "required": [

crates/rooch-rpc-api/src/api/rooch_api.rs

@@ -11,8 +11,9 @@ use crate::jsonrpc_types::{
     DryRunTransactionResponseView, EventOptions, EventPageView, ExecuteTransactionResponseView,
     FieldKeyView, FunctionCallView, H256View, IndexerEventPageView, IndexerObjectStatePageView,
     IndexerStateIDView, ModuleABIView, ObjectIDVecView, ObjectIDView, ObjectStateFilterView,
-    ObjectStateView, QueryOptions, RoochAddressView, StateOptions, StatePageView, StrView,
-    StructTagView, TransactionWithInfoPageView, TxOptions,
+    ObjectStateView, QueryOptions, RoochAddressView, StateChangeSetPageView, StateOptions,
+    StatePageView, StrView, StructTagView, SyncStateFilterView, TransactionWithInfoPageView,
+    TxOptions,
 };
 use crate::RpcResult;
 use jsonrpsee::proc_macros::rpc;
@@ -198,4 +199,15 @@ pub trait RoochAPI {
         repair_type: RepairIndexerTypeView,
         repair_params: RepairIndexerParamsView,
     ) -> RpcResult<()>;
+
+    /// Sync state change sets
+    #[method(name = "syncStates")]
+    async fn sync_states(
+        &self,
+        filter: SyncStateFilterView,
+        // exclusive cursor if `Some`, otherwise start from the beginning
+        cursor: Option<StrView<u64>>,
+        limit: Option<StrView<u64>>,
+        query_option: Option<QueryOptions>,
+    ) -> RpcResult<StateChangeSetPageView>;
 }

crates/rooch-rpc-api/src/jsonrpc_types/rooch_types.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use super::event_view::IndexerEventIDView;
-use super::{HumanReadableDisplay, IndexerStateIDView};
+use super::{HumanReadableDisplay, IndexerStateIDView, StateChangeSetWithTxOrderView};
 use crate::jsonrpc_types::account_view::BalanceInfoView;
 use crate::jsonrpc_types::btc::ord::InscriptionStateView;
 use crate::jsonrpc_types::btc::utxo::UTXOStateView;
@@ -29,6 +29,7 @@ pub type IndexerObjectStatePageView = PageView<IndexerObjectStateView, IndexerSt
 
 pub type UTXOPageView = PageView<UTXOStateView, IndexerStateIDView>;
 pub type InscriptionPageView = PageView<InscriptionStateView, IndexerStateIDView>;
+pub type StateChangeSetPageView = PageView<StateChangeSetWithTxOrderView, StrView<u64>>;
 
 /// `next_cursor` points to the last item in the page;
 /// Reading with `next_cursor` will start from the next item after `next_cursor` if