From 0629d816d4db35dd6caef9e479ac2d42df2dfa09 Mon Sep 17 00:00:00 2001 From: longbowlu Date: Tue, 5 Dec 2023 12:12:07 -0800 Subject: [PATCH] SuiSyncer --- crates/sui-bridge/src/eth_syncer.rs | 23 +- crates/sui-bridge/src/lib.rs | 11 + crates/sui-bridge/src/sui_client.rs | 31 ++- crates/sui-bridge/src/sui_mock_client.rs | 4 +- crates/sui-bridge/src/sui_syncer.rs | 235 +++++++++++++++++- crates/sui-indexer/src/apis/indexer_api_v2.rs | 2 +- crates/sui-indexer/src/indexer_reader.rs | 2 +- .../sui-indexer/src/store/pg_indexer_store.rs | 2 +- crates/sui-json-rpc/src/indexer_api.rs | 4 +- crates/sui-types/src/event.rs | 2 +- 10 files changed, 268 insertions(+), 48 deletions(-) diff --git a/crates/sui-bridge/src/eth_syncer.rs b/crates/sui-bridge/src/eth_syncer.rs index 38e02e85eb8a0f..8e6ede18a86733 100644 --- a/crates/sui-bridge/src/eth_syncer.rs +++ b/crates/sui-bridge/src/eth_syncer.rs @@ -15,6 +15,8 @@ use std::sync::Arc; use tokio::sync::watch; use tokio::task::JoinHandle; use tokio::time::{self, Duration}; +use tokio_retry::strategy::{jitter, ExponentialBackoff}; +use tokio_retry::Retry; const ETH_EVENTS_CHANNEL_SIZE: usize = 1000; const FINALIZED_BLOCK_QUERY_INTERVAL: Duration = Duration::from_secs(2); @@ -94,6 +96,9 @@ where ) .expect("Failed to get last finalzied block from eth client after retry"); tracing::debug!("Last finalized block: {}", new_value); + + // TODO add a metrics for the last finalized block + if new_value > last_block_number { last_finalized_block_sender .send(new_value) @@ -138,29 +143,13 @@ where .send(events) .await .expect("All Eth event channel receivers are closed"); - tracing::info!( - contract_address=?contract_address, - "Observed {len} new events", - ); + tracing::info!(?contract_address, "Observed {len} new Eth events",); } start_block = new_finalized_block + 1; } } } -use tokio_retry::strategy::{jitter, ExponentialBackoff}; -use tokio_retry::Retry; - -#[macro_export] -macro_rules! retry_with_max_delay { - ($func:expr, $max_delay:expr) => {{ - let retry_strategy = ExponentialBackoff::from_millis(100) - .max_delay($max_delay) - .map(jitter); - Retry::spawn(retry_strategy, || $func).await - }}; -} - #[cfg(test)] mod tests { use std::{collections::HashSet, str::FromStr}; diff --git a/crates/sui-bridge/src/lib.rs b/crates/sui-bridge/src/lib.rs index 38f6323863a063..d0027134e7a975 100644 --- a/crates/sui-bridge/src/lib.rs +++ b/crates/sui-bridge/src/lib.rs @@ -9,9 +9,20 @@ pub mod events; pub mod handler; pub mod server; pub mod sui_client; +pub mod sui_syncer; #[cfg(test)] pub(crate) mod eth_mock_provider; #[cfg(test)] pub(crate) mod sui_mock_client; + +#[macro_export] +macro_rules! retry_with_max_delay { + ($func:expr, $max_delay:expr) => {{ + let retry_strategy = ExponentialBackoff::from_millis(100) + .max_delay($max_delay) + .map(jitter); + Retry::spawn(retry_strategy, || $func).await + }}; +} diff --git a/crates/sui-bridge/src/sui_client.rs b/crates/sui-bridge/src/sui_client.rs index 774b845effcab2..d0a5354a0dfe25 100644 --- a/crates/sui-bridge/src/sui_client.rs +++ b/crates/sui-bridge/src/sui_client.rs @@ -24,7 +24,7 @@ use tap::TapFallible; use crate::error::{BridgeError, BridgeResult}; use crate::events::SuiBridgeEvent; -pub(crate) struct SuiClient

{ +pub struct SuiClient

{ inner: P, } @@ -84,10 +84,7 @@ where let mut is_first_page = true; let mut all_events: Vec = vec![]; loop { - let events = self - .inner - .query_events(filter.clone(), cursor.clone()) - .await?; + let events = self.inner.query_events(filter.clone(), cursor).await?; if events.data.is_empty() { return Ok(Page { data: all_events, @@ -97,7 +94,7 @@ where } // unwrap safe: we just checked data is not empty - let new_cursor = events.data.last().unwrap().id.clone(); + let new_cursor = events.data.last().unwrap().id; // Now check if we need to query more events for the sake of // paginating in transaction granularity @@ -349,7 +346,7 @@ mod tests { next_cursor: Some(event_2.id.clone()), has_next_page: true, }; - mock_client.add_event_response(package, module.clone(), event_1.id.clone(), events_page_2); + mock_client.add_event_response(package, module.clone(), event_1.id, events_page_2); // page 3 (event 3, event 4, different tx_digest) let mut event_3 = SuiEvent::random_for_testing(); event_3.id.tx_digest = event_2.id.tx_digest; @@ -358,10 +355,10 @@ mod tests { assert_ne!(event_3.id.tx_digest, event_4.id.tx_digest); let events_page_3 = EventPage { data: vec![event_3.clone(), event_4.clone()], - next_cursor: Some(event_4.id.clone()), + next_cursor: Some(event_4.id), has_next_page: true, }; - mock_client.add_event_response(package, module.clone(), event_2.id.clone(), events_page_3); + mock_client.add_event_response(package, module.clone(), event_2.id, events_page_3); let page: Page = sui_client .query_events_by_module(package, module.clone(), cursor) .await @@ -390,12 +387,12 @@ mod tests { // second page assert_eq!( mock_client.pop_front_past_event_query_params().unwrap(), - (package, module.clone(), event_1.id.clone()) + (package, module.clone(), event_1.id) ); // third page assert_eq!( mock_client.pop_front_past_event_query_params().unwrap(), - (package, module.clone(), event_2.id.clone()) + (package, module.clone(), event_2.id) ); // no more assert_eq!(mock_client.pop_front_past_event_query_params(), None); @@ -403,10 +400,10 @@ mod tests { // Case 4, modify page 3 in case 3 to return event_4 only let events_page_3 = EventPage { data: vec![event_4.clone()], - next_cursor: Some(event_4.id.clone()), + next_cursor: Some(event_4.id), has_next_page: true, }; - mock_client.add_event_response(package, module.clone(), event_2.id.clone(), events_page_3); + mock_client.add_event_response(package, module.clone(), event_2.id, events_page_3); let page: Page = sui_client .query_events_by_module(package, module.clone(), cursor) .await @@ -434,12 +431,12 @@ mod tests { // second page assert_eq!( mock_client.pop_front_past_event_query_params().unwrap(), - (package, module.clone(), event_1.id.clone()) + (package, module.clone(), event_1.id) ); // third page assert_eq!( mock_client.pop_front_past_event_query_params().unwrap(), - (package, module.clone(), event_2.id.clone()) + (package, module.clone(), event_2.id) ); // no more assert_eq!(mock_client.pop_front_past_event_query_params(), None); @@ -447,10 +444,10 @@ mod tests { // Case 5, modify page 2 in case 3 to mark has_next_page as false let events_page_2 = EventPage { data: vec![event_2.clone()], - next_cursor: Some(event_2.id.clone()), + next_cursor: Some(event_2.id), has_next_page: false, }; - mock_client.add_event_response(package, module.clone(), event_1.id.clone(), events_page_2); + mock_client.add_event_response(package, module.clone(), event_1.id, events_page_2); let page: Page = sui_client .query_events_by_module(package, module.clone(), cursor) .await diff --git a/crates/sui-bridge/src/sui_mock_client.rs b/crates/sui-bridge/src/sui_mock_client.rs index a3b6b726af1970..48edd2705dff4f 100644 --- a/crates/sui-bridge/src/sui_mock_client.rs +++ b/crates/sui-bridge/src/sui_mock_client.rs @@ -78,10 +78,10 @@ impl SuiClientInner for SuiMockClient { self.past_event_query_params.lock().unwrap().push_back(( package, module.clone(), - cursor.clone(), + cursor, )); Ok(events - .get(&(package, module.clone(), cursor.clone())) + .get(&(package, module.clone(), cursor)) .cloned() .unwrap_or_else(|| { panic!( diff --git a/crates/sui-bridge/src/sui_syncer.rs b/crates/sui-bridge/src/sui_syncer.rs index 2ab3a9f636a764..fbd6d288b508f8 100644 --- a/crates/sui-bridge/src/sui_syncer.rs +++ b/crates/sui-bridge/src/sui_syncer.rs @@ -1,14 +1,237 @@ // Copyright (c) Mysten Labs, Inc. // SPDX-License-Identifier: Apache-2.0 -//! The Suisyncer module is responsible for synchronizing Events emitted on Sui blockchain from -//! concerned bridge packages. Each package is associated with a cursor, and the syncer will -//! only query from that cursor onwards. It's likely that SuiSyncer only tracks one package. +//! The SuiSyncer module is responsible for synchronizing Events emitted on Sui blockchain from +//! concerned bridge packages. +use crate::{ + error::BridgeResult, + retry_with_max_delay, + sui_client::{SuiClient, SuiClientInner}, +}; +use mysten_metrics::spawn_logged_monitored_task; +use std::{collections::HashMap, sync::Arc}; +use sui_json_rpc_types::SuiEvent; +use sui_types::{ + base_types::ObjectID, digests::TransactionDigest, Identifier, SUI_SYSTEM_PACKAGE_ID, +}; +use tokio::{ + task::JoinHandle, + time::{self, Duration}, +}; +use tokio_retry::strategy::{jitter, ExponentialBackoff}; +use tokio_retry::Retry; +// TODO: use the right package id +const PACKAGE_ID: ObjectID = SUI_SYSTEM_PACKAGE_ID; +const SUI_EVENTS_CHANNEL_SIZE: usize = 1000; +/// Map from contract address to their start block. +pub type SuiTargetModules = HashMap; -pub struct EthSyncer

{ - eth_client: Arc, - contract_addresses: EthTargetAddresses, +pub struct SuiSyncer { + sui_client: Arc>, + // The last transaction that the syncer has fully processed. + // Syncer will resume post this transaction (i.e. exclusive), when it starts. + cursors: SuiTargetModules, +} + +impl SuiSyncer +where + C: SuiClientInner + 'static, +{ + pub fn new(sui_client: Arc>, cursors: SuiTargetModules) -> Self { + Self { + sui_client, + cursors, + } + } + + pub async fn run( + self, + query_interval: Duration, + ) -> BridgeResult<( + Vec>, + mysten_metrics::metered_channel::Receiver>, + )> { + let (events_tx, events_rx) = mysten_metrics::metered_channel::channel( + SUI_EVENTS_CHANNEL_SIZE, + &mysten_metrics::get_metrics() + .unwrap() + .channels + .with_label_values(&["sui_events_queue"]), + ); + + let mut task_handles = vec![]; + for (module, cursor) in self.cursors { + let events_rx_clone = events_tx.clone(); + let sui_client_clone = self.sui_client.clone(); + task_handles.push(spawn_logged_monitored_task!( + Self::run_event_listening_task( + module, + cursor, + events_rx_clone, + sui_client_clone, + query_interval + ) + )); + } + Ok((task_handles, events_rx)) + } + + async fn run_event_listening_task( + // The module where interested events are defined. + // Moudle is always of bridge package 0x9. + module: Identifier, + mut next_cursor: TransactionDigest, + events_sender: mysten_metrics::metered_channel::Sender>, + sui_client: Arc>, + query_interval: Duration, + ) { + tracing::info!( + ?module, + ?next_cursor, + "Starting sui events listening task from tx_digest {next_cursor}" + ); + let mut interval = time::interval(query_interval); + interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + let events = retry_with_max_delay!( + sui_client.query_events_by_module(PACKAGE_ID, module.clone(), next_cursor), + Duration::from_secs(600) + ) + .expect("Failed to query events from sui client after retry"); + + let len = events.data.len(); + if len != 0 { + events_sender + .send(events.data) + .await + .expect("All Sui event channel receivers are closed"); + // Unwrap: `query_events_by_module` always returns Some `next_cursor` + // If the events list is empty, `next_cursor` will be the same as `start_tx_digest` + next_cursor = events.next_cursor.unwrap(); + tracing::info!(?module, ?next_cursor, "Observed {len} new Sui events"); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use crate::{sui_client::SuiClient, sui_mock_client::SuiMockClient}; + use prometheus::Registry; + use sui_json_rpc_types::EventPage; + use sui_types::{digests::TransactionDigest, event::EventID, Identifier}; + use tokio::time::timeout; + + #[tokio::test] + async fn test_sui_syncer_basic() -> anyhow::Result<()> { + telemetry_subscribers::init_for_testing(); + let registry = Registry::new(); + mysten_metrics::init_metrics(®istry); + + let mock = SuiMockClient::default(); + let client = Arc::new(SuiClient::new_for_testing(mock.clone())); + let module_foo = Identifier::new("Foo").unwrap(); + let module_bar = Identifier::new("Bar").unwrap(); + let empty_events = EventPage::empty(); + let cursor = TransactionDigest::random(); + add_event_response(&mock, module_foo.clone(), cursor, empty_events.clone()); + add_event_response(&mock, module_bar.clone(), cursor, empty_events.clone()); + + let target_modules = HashMap::from_iter(vec![ + (module_foo.clone(), cursor), + (module_bar.clone(), cursor), + ]); + let interval = Duration::from_millis(200); + let (_handles, mut events_rx) = SuiSyncer::new(client, target_modules) + .run(interval) + .await + .unwrap(); + + // Initially there are no events + assert_no_more_events(interval, &mut events_rx).await; + + // Module Foo has new events + let event_1: SuiEvent = SuiEvent::random_for_testing(); + let module_foo_events_1: sui_json_rpc_types::Page = EventPage { + data: vec![event_1.clone(), event_1.clone()], + next_cursor: None, + has_next_page: false, + }; + add_event_response( + &mock, + module_foo.clone(), + event_1.id.tx_digest, + empty_events.clone(), + ); + add_event_response( + &mock, + module_foo.clone(), + cursor, + module_foo_events_1.clone(), + ); + + let received_events = events_rx.recv().await.unwrap(); + assert_eq!(received_events.len(), 2); + assert_eq!(received_events[0].id, event_1.id); + assert_eq!(received_events[1].id, event_1.id); + // No more + assert_no_more_events(interval, &mut events_rx).await; + + // Module Bar has new events + let event_2: SuiEvent = SuiEvent::random_for_testing(); + let module_bar_events_1 = EventPage { + data: vec![event_2.clone()], + next_cursor: None, + has_next_page: false, + }; + add_event_response( + &mock, + module_bar.clone(), + event_2.id.tx_digest, + empty_events.clone(), + ); + + add_event_response(&mock, module_bar.clone(), cursor, module_bar_events_1); + + let received_events = events_rx.recv().await.unwrap(); + assert_eq!(received_events.len(), 1); + assert_eq!(received_events[0].id, event_2.id); + // No more + assert_no_more_events(interval, &mut events_rx).await; + + Ok(()) + } + + async fn assert_no_more_events( + interval: Duration, + events_rx: &mut mysten_metrics::metered_channel::Receiver>, + ) { + match timeout(interval * 2, events_rx.recv()).await { + Err(_e) => (), + other => panic!("Should have timed out, but got: {:?}", other), + }; + } + + fn add_event_response( + mock: &SuiMockClient, + module: Identifier, + cursor: TransactionDigest, + events: EventPage, + ) { + mock.add_event_response( + PACKAGE_ID, + module.clone(), + EventID { + tx_digest: cursor, + event_seq: u16::MAX as u64, + }, + events.clone(), + ); + } } diff --git a/crates/sui-indexer/src/apis/indexer_api_v2.rs b/crates/sui-indexer/src/apis/indexer_api_v2.rs index 01138e8a87302d..cacc65b1e17166 100644 --- a/crates/sui-indexer/src/apis/indexer_api_v2.rs +++ b/crates/sui-indexer/src/apis/indexer_api_v2.rs @@ -190,7 +190,7 @@ impl IndexerApiServer for IndexerApiV2 { let has_next_page = results.len() > limit; results.truncate(limit); - let next_cursor = results.last().map(|o| o.id.clone()); + let next_cursor = results.last().map(|o| o.id); Ok(Page { data: results, next_cursor, diff --git a/crates/sui-indexer/src/indexer_reader.rs b/crates/sui-indexer/src/indexer_reader.rs index 173d311a92a8a1..9e753f1c1ef755 100644 --- a/crates/sui-indexer/src/indexer_reader.rs +++ b/crates/sui-indexer/src/indexer_reader.rs @@ -1052,7 +1052,7 @@ impl IndexerReader { limit: usize, descending_order: bool, ) -> IndexerResult> { - let (tx_seq, event_seq) = if let Some(cursor) = cursor.clone() { + let (tx_seq, event_seq) = if let Some(cursor) = cursor { let EventID { tx_digest, event_seq, diff --git a/crates/sui-indexer/src/store/pg_indexer_store.rs b/crates/sui-indexer/src/store/pg_indexer_store.rs index 36ff4d81928405..ae8f1ddc155525 100644 --- a/crates/sui-indexer/src/store/pg_indexer_store.rs +++ b/crates/sui-indexer/src/store/pg_indexer_store.rs @@ -463,7 +463,7 @@ impl PgIndexerStore { page_limit -= 1; let has_next_page = sui_event_vec.len() > page_limit; sui_event_vec.truncate(page_limit); - let next_cursor = sui_event_vec.last().map(|e| e.id.clone()); + let next_cursor = sui_event_vec.last().map(|e| e.id); Ok(EventPage { data: sui_event_vec, next_cursor, diff --git a/crates/sui-json-rpc/src/indexer_api.rs b/crates/sui-json-rpc/src/indexer_api.rs index 655ef7a85ea5ed..9a600025b03d66 100644 --- a/crates/sui-json-rpc/src/indexer_api.rs +++ b/crates/sui-json-rpc/src/indexer_api.rs @@ -257,7 +257,7 @@ impl IndexerApiServer for IndexerApi { .query_events( &self.transaction_kv_store, query, - cursor.clone(), + cursor, limit + 1, descending, ) @@ -265,7 +265,7 @@ impl IndexerApiServer for IndexerApi { .map_err(Error::from)?; let has_next_page = data.len() > limit; data.truncate(limit); - let next_cursor = data.last().map_or(cursor, |e| Some(e.id.clone())); + let next_cursor = data.last().map_or(cursor, |e| Some(e.id)); self.metrics .query_events_result_size .report(data.len() as u64); diff --git a/crates/sui-types/src/event.rs b/crates/sui-types/src/event.rs index 35af980a9ba036..b505442d55bc50 100755 --- a/crates/sui-types/src/event.rs +++ b/crates/sui-types/src/event.rs @@ -41,7 +41,7 @@ pub struct EventEnvelope { /// Unique ID of a Sui Event, the ID is a combination of tx seq number and event seq number, /// the ID is local to this particular fullnode and will be different from other fullnode. #[serde_as] -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, Hash)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, Hash)] #[serde(rename_all = "camelCase")] pub struct EventID { pub tx_digest: TransactionDigest,