1use anyhow::anyhow;
5use async_trait::async_trait;
6use core::panic;
7use fastcrypto::traits::ToFromBytes;
8use std::collections::HashMap;
9use std::str::from_utf8;
10use std::sync::Arc;
11use std::time::Duration;
12use sui_json_rpc_types::BcsEvent;
13use sui_json_rpc_types::{EventFilter, Page, SuiEvent};
14use sui_json_rpc_types::{
15 EventPage, SuiExecutionStatus, SuiObjectDataOptions, SuiTransactionBlockResponseOptions,
16};
17use sui_rpc::field::{FieldMask, FieldMaskUtil};
18use sui_rpc::proto::sui::rpc::v2::{
19 Checkpoint, ExecuteTransactionRequest, ExecutedTransaction, GetCheckpointRequest,
20 GetObjectRequest, GetServiceInfoRequest, GetTransactionRequest, Object,
21 Transaction as ProtoTransaction, UserSignature as ProtoUserSignature,
22};
23use sui_sdk::{SuiClient as SuiSdkClient, SuiClientBuilder};
24use sui_sdk_types::Address;
25use sui_types::BRIDGE_PACKAGE_ID;
26use sui_types::SUI_BRIDGE_OBJECT_ID;
27use sui_types::TypeTag;
28use sui_types::base_types::ObjectRef;
29use sui_types::base_types::SequenceNumber;
30use sui_types::bridge::{
31 BridgeSummary, BridgeWrapper, MoveTypeBridgeMessageKey, MoveTypeBridgeRecord,
32};
33use sui_types::bridge::{BridgeTrait, BridgeTreasurySummary};
34use sui_types::bridge::{MoveTypeBridgeMessage, MoveTypeParsedTokenTransferMessage};
35use sui_types::bridge::{MoveTypeCommitteeMember, MoveTypeTokenTransferPayload};
36use sui_types::collection_types::LinkedTableNode;
37use sui_types::gas_coin::GasCoin;
38use sui_types::object::Owner;
39use sui_types::parse_sui_type_tag;
40use sui_types::transaction::ObjectArg;
41use sui_types::transaction::SharedObjectMutability;
42use sui_types::transaction::Transaction;
43use sui_types::{Identifier, base_types::ObjectID, digests::TransactionDigest, event::EventID};
44use tokio::sync::OnceCell;
45use tracing::{error, warn};
46
47use crate::crypto::BridgeAuthorityPublicKey;
48use crate::error::{BridgeError, BridgeResult};
49use crate::events::SuiBridgeEvent;
50use crate::metrics::BridgeMetrics;
51use crate::retry_with_max_elapsed_time;
52use crate::types::BridgeActionStatus;
53use crate::types::ParsedTokenTransferMessage;
54use crate::types::SuiEvents;
55use crate::types::{BridgeAction, BridgeAuthority, BridgeCommittee};
56
57pub struct SuiClient<P> {
58 inner: P,
59 bridge_metrics: Arc<BridgeMetrics>,
60}
61
62pub type SuiBridgeClient = SuiClient<SuiClientInternal>;
63
64pub struct SuiClientInternal {
65 jsonrpc_client: SuiSdkClient,
66 grpc_client: sui_rpc::Client,
67}
68
69#[derive(Clone, Debug)]
70pub struct ExecuteTransactionResult {
71 pub status: SuiExecutionStatus,
72 pub events: Vec<SuiEvent>,
73}
74
75impl SuiBridgeClient {
76 pub async fn new(rpc_url: &str, bridge_metrics: Arc<BridgeMetrics>) -> anyhow::Result<Self> {
77 let jsonrpc_client = SuiClientBuilder::default()
78 .build(rpc_url)
79 .await
80 .map_err(|e| {
81 anyhow!("Can't establish connection with Sui Rpc {rpc_url}. Error: {e}")
82 })?;
83 let grpc_client = sui_rpc::Client::new(rpc_url)?;
84 let inner = SuiClientInternal {
85 jsonrpc_client,
86 grpc_client,
87 };
88 let self_ = Self {
89 inner,
90 bridge_metrics,
91 };
92 self_.describe().await.map_err(|e| anyhow::anyhow!("{e}"))?;
93 Ok(self_)
94 }
95
96 pub fn jsonrpc_client(&self) -> &SuiSdkClient {
97 &self.inner.jsonrpc_client
98 }
99
100 pub fn grpc_client(&self) -> &sui_rpc::Client {
101 &self.inner.grpc_client
102 }
103}
104
105impl<P> SuiClient<P>
106where
107 P: SuiClientInner,
108{
109 pub fn new_for_testing(inner: P) -> Self {
110 Self {
111 inner,
112 bridge_metrics: Arc::new(BridgeMetrics::new_for_testing()),
113 }
114 }
115
116 async fn describe(&self) -> Result<(), BridgeError> {
118 let chain_id = self.inner.get_chain_identifier().await?;
119 let block_number = self.inner.get_latest_checkpoint_sequence_number().await?;
120 tracing::info!(
121 "SuiClient is connected to chain {chain_id}, current block number: {block_number}"
122 );
123 Ok(())
124 }
125
126 pub async fn get_mutable_bridge_object_arg_must_succeed(&self) -> ObjectArg {
131 static ARG: OnceCell<ObjectArg> = OnceCell::const_new();
132 *ARG.get_or_init(|| async move {
133 let Ok(Ok(bridge_object_arg)) = retry_with_max_elapsed_time!(
134 self.inner.get_mutable_bridge_object_arg(),
135 Duration::from_secs(30)
136 ) else {
137 panic!("Failed to get bridge object arg after retries");
138 };
139 bridge_object_arg
140 })
141 .await
142 }
143
144 pub async fn query_events_by_module(
146 &self,
147 package: ObjectID,
148 module: Identifier,
149 cursor: Option<EventID>,
151 ) -> BridgeResult<Page<SuiEvent, EventID>> {
152 let filter = EventFilter::MoveEventModule {
153 package,
154 module: module.clone(),
155 };
156 let events = self.inner.query_events(filter.clone(), cursor).await?;
157
158 assert!(
160 events
161 .data
162 .iter()
163 .all(|event| event.type_.address.as_ref() == package.as_ref()
164 && event.type_.module == module)
165 );
166 Ok(events)
167 }
168
169 pub async fn get_bridge_action_by_tx_digest_and_event_idx_maybe(
173 &self,
174 tx_digest: &TransactionDigest,
175 event_idx: u16,
176 ) -> BridgeResult<BridgeAction> {
177 let events = self.inner.get_events_by_tx_digest(*tx_digest).await?;
178 let event = events
179 .events
180 .get(event_idx as usize)
181 .ok_or(BridgeError::NoBridgeEventsInTxPosition)?;
182 if event.type_.address.as_ref() != BRIDGE_PACKAGE_ID.as_ref() {
183 return Err(BridgeError::BridgeEventInUnrecognizedSuiPackage);
184 }
185 let bridge_event = SuiBridgeEvent::try_from_sui_event(event)?
186 .ok_or(BridgeError::NoBridgeEventsInTxPosition)?;
187
188 bridge_event
189 .try_into_bridge_action()
190 .ok_or(BridgeError::BridgeEventNotActionable)
191 }
192
193 pub async fn get_bridge_summary(&self) -> BridgeResult<BridgeSummary> {
194 self.inner.get_bridge_summary().await
195 }
196
197 pub async fn is_bridge_paused(&self) -> BridgeResult<bool> {
198 self.get_bridge_summary()
199 .await
200 .map(|summary| summary.is_frozen)
201 }
202
203 pub async fn get_treasury_summary(&self) -> BridgeResult<BridgeTreasurySummary> {
204 Ok(self.get_bridge_summary().await?.treasury)
205 }
206
207 pub async fn get_token_id_map(&self) -> BridgeResult<HashMap<u8, TypeTag>> {
208 self.get_bridge_summary()
209 .await?
210 .treasury
211 .id_token_type_map
212 .into_iter()
213 .map(|(id, name)| {
214 parse_sui_type_tag(&format!("0x{name}"))
215 .map(|name| (id, name))
216 .map_err(|e| {
217 BridgeError::InternalError(format!(
218 "Failed to retrieve token id mapping: {e}, type name: {name}"
219 ))
220 })
221 })
222 .collect()
223 }
224
225 pub async fn get_notional_values(&self) -> BridgeResult<HashMap<u8, u64>> {
226 let bridge_summary = self.get_bridge_summary().await?;
227 bridge_summary
228 .treasury
229 .id_token_type_map
230 .iter()
231 .map(|(id, type_name)| {
232 bridge_summary
233 .treasury
234 .supported_tokens
235 .iter()
236 .find_map(|(tn, metadata)| {
237 if type_name == tn {
238 Some((*id, metadata.notional_value))
239 } else {
240 None
241 }
242 })
243 .ok_or(BridgeError::InternalError(
244 "Error encountered when retrieving token notional values.".into(),
245 ))
246 })
247 .collect()
248 }
249
250 pub async fn get_bridge_committee(&self) -> BridgeResult<BridgeCommittee> {
251 let bridge_summary = self.inner.get_bridge_summary().await?;
252 let move_type_bridge_committee = bridge_summary.committee;
253
254 let mut authorities = vec![];
255 for (_, member) in move_type_bridge_committee.members {
257 let MoveTypeCommitteeMember {
258 sui_address,
259 bridge_pubkey_bytes,
260 voting_power,
261 http_rest_url,
262 blocklisted,
263 } = member;
264 let pubkey = BridgeAuthorityPublicKey::from_bytes(&bridge_pubkey_bytes)?;
265 let base_url = from_utf8(&http_rest_url).unwrap_or_else(|_e| {
266 warn!(
267 "Bridge authority address: {}, pubkey: {:?} has invalid http url: {:?}",
268 sui_address, bridge_pubkey_bytes, http_rest_url
269 );
270 ""
271 });
272 authorities.push(BridgeAuthority {
273 sui_address,
274 pubkey,
275 voting_power,
276 base_url: base_url.into(),
277 is_blocklisted: blocklisted,
278 });
279 }
280 BridgeCommittee::new(authorities)
281 }
282
283 pub async fn get_chain_identifier(&self) -> BridgeResult<String> {
284 self.inner.get_chain_identifier().await
285 }
286
287 pub async fn get_reference_gas_price_until_success(&self) -> u64 {
288 loop {
289 let Ok(Ok(rgp)) = retry_with_max_elapsed_time!(
290 self.inner.get_reference_gas_price(),
291 Duration::from_secs(30)
292 ) else {
293 self.bridge_metrics
294 .sui_rpc_errors
295 .with_label_values(&["get_reference_gas_price"])
296 .inc();
297 error!("Failed to get reference gas price");
298 continue;
299 };
300 return rgp;
301 }
302 }
303
304 pub async fn get_latest_checkpoint_sequence_number(&self) -> BridgeResult<u64> {
305 self.inner.get_latest_checkpoint_sequence_number().await
306 }
307
308 pub async fn execute_transaction_block_with_effects(
309 &self,
310 tx: sui_types::transaction::Transaction,
311 ) -> BridgeResult<ExecuteTransactionResult> {
312 self.inner.execute_transaction_block_with_effects(tx).await
313 }
314
315 pub async fn get_token_transfer_action_onchain_status_until_success(
317 &self,
318 source_chain_id: u8,
319 seq_number: u64,
320 ) -> BridgeActionStatus {
321 loop {
322 let bridge_object_arg = self.get_mutable_bridge_object_arg_must_succeed().await;
323 let Ok(Ok(status)) = retry_with_max_elapsed_time!(
324 self.inner.get_token_transfer_action_onchain_status(
325 bridge_object_arg,
326 source_chain_id,
327 seq_number
328 ),
329 Duration::from_secs(30)
330 ) else {
331 self.bridge_metrics
332 .sui_rpc_errors
333 .with_label_values(&["get_token_transfer_action_onchain_status"])
334 .inc();
335 error!(
336 source_chain_id,
337 seq_number, "Failed to get token transfer action onchain status"
338 );
339 continue;
340 };
341 return status;
342 }
343 }
344
345 pub async fn get_token_transfer_action_onchain_signatures_until_success(
346 &self,
347 source_chain_id: u8,
348 seq_number: u64,
349 ) -> Option<Vec<Vec<u8>>> {
350 loop {
351 let bridge_object_arg = self.get_mutable_bridge_object_arg_must_succeed().await;
352 let Ok(Ok(sigs)) = retry_with_max_elapsed_time!(
353 self.inner.get_token_transfer_action_onchain_signatures(
354 bridge_object_arg,
355 source_chain_id,
356 seq_number
357 ),
358 Duration::from_secs(30)
359 ) else {
360 self.bridge_metrics
361 .sui_rpc_errors
362 .with_label_values(&["get_token_transfer_action_onchain_signatures"])
363 .inc();
364 error!(
365 source_chain_id,
366 seq_number, "Failed to get token transfer action onchain signatures"
367 );
368 continue;
369 };
370 return sigs;
371 }
372 }
373
374 pub async fn get_parsed_token_transfer_message(
375 &self,
376 source_chain_id: u8,
377 seq_number: u64,
378 ) -> BridgeResult<Option<ParsedTokenTransferMessage>> {
379 let bridge_object_arg = self.get_mutable_bridge_object_arg_must_succeed().await;
380 let message = self
381 .inner
382 .get_parsed_token_transfer_message(bridge_object_arg, source_chain_id, seq_number)
383 .await?;
384 Ok(match message {
385 Some(payload) => Some(ParsedTokenTransferMessage::try_from(payload)?),
386 None => None,
387 })
388 }
389
390 pub async fn get_bridge_record(
391 &self,
392 source_chain_id: u8,
393 seq_number: u64,
394 ) -> Result<Option<MoveTypeBridgeRecord>, BridgeError> {
395 self.inner
396 .get_bridge_record(source_chain_id, seq_number)
397 .await
398 }
399
400 pub async fn get_gas_data_panic_if_not_gas(
401 &self,
402 gas_object_id: ObjectID,
403 ) -> (GasCoin, ObjectRef, Owner) {
404 self.inner
405 .get_gas_data_panic_if_not_gas(gas_object_id)
406 .await
407 }
408}
409
410#[async_trait]
412pub trait SuiClientInner: Send + Sync {
413 async fn query_events(
414 &self,
415 query: EventFilter,
416 cursor: Option<EventID>,
417 ) -> Result<EventPage, BridgeError>;
418
419 async fn get_events_by_tx_digest(
420 &self,
421 tx_digest: TransactionDigest,
422 ) -> Result<SuiEvents, BridgeError>;
423
424 async fn get_chain_identifier(&self) -> Result<String, BridgeError>;
425
426 async fn get_reference_gas_price(&self) -> Result<u64, BridgeError>;
427
428 async fn get_latest_checkpoint_sequence_number(&self) -> Result<u64, BridgeError>;
429
430 async fn get_mutable_bridge_object_arg(&self) -> Result<ObjectArg, BridgeError>;
431
432 async fn get_bridge_summary(&self) -> Result<BridgeSummary, BridgeError>;
433
434 async fn execute_transaction_block_with_effects(
435 &self,
436 tx: Transaction,
437 ) -> Result<ExecuteTransactionResult, BridgeError>;
438
439 async fn get_token_transfer_action_onchain_status(
440 &self,
441 bridge_object_arg: ObjectArg,
442 source_chain_id: u8,
443 seq_number: u64,
444 ) -> Result<BridgeActionStatus, BridgeError>;
445
446 async fn get_token_transfer_action_onchain_signatures(
447 &self,
448 bridge_object_arg: ObjectArg,
449 source_chain_id: u8,
450 seq_number: u64,
451 ) -> Result<Option<Vec<Vec<u8>>>, BridgeError>;
452
453 async fn get_parsed_token_transfer_message(
454 &self,
455 bridge_object_arg: ObjectArg,
456 source_chain_id: u8,
457 seq_number: u64,
458 ) -> Result<Option<MoveTypeParsedTokenTransferMessage>, BridgeError>;
459
460 async fn get_bridge_record(
461 &self,
462 source_chain_id: u8,
463 seq_number: u64,
464 ) -> Result<Option<MoveTypeBridgeRecord>, BridgeError>;
465
466 async fn get_gas_data_panic_if_not_gas(
467 &self,
468 gas_object_id: ObjectID,
469 ) -> (GasCoin, ObjectRef, Owner);
470}
471
472#[async_trait]
473impl SuiClientInner for SuiSdkClient {
474 async fn query_events(
475 &self,
476 query: EventFilter,
477 cursor: Option<EventID>,
478 ) -> Result<EventPage, BridgeError> {
479 self.event_api()
480 .query_events(query, cursor, None, false)
481 .await
482 .map_err(Into::into)
483 }
484
485 async fn get_events_by_tx_digest(
486 &self,
487 _tx_digest: TransactionDigest,
488 ) -> Result<SuiEvents, BridgeError> {
489 unimplemented!("use gRPC implementation")
490 }
491
492 async fn get_chain_identifier(&self) -> Result<String, BridgeError> {
493 unimplemented!("use gRPC implementation")
494 }
495
496 async fn get_reference_gas_price(&self) -> Result<u64, BridgeError> {
497 unimplemented!("use gRPC implementation")
498 }
499
500 async fn get_latest_checkpoint_sequence_number(&self) -> Result<u64, BridgeError> {
501 unimplemented!("use gRPC implementation")
502 }
503
504 async fn get_mutable_bridge_object_arg(&self) -> Result<ObjectArg, BridgeError> {
505 unimplemented!("use gRPC implementation")
506 }
507
508 async fn get_bridge_summary(&self) -> Result<BridgeSummary, BridgeError> {
509 unimplemented!("use gRPC implementation")
510 }
511
512 async fn get_token_transfer_action_onchain_status(
513 &self,
514 _bridge_object_arg: ObjectArg,
515 _source_chain_id: u8,
516 _seq_number: u64,
517 ) -> Result<BridgeActionStatus, BridgeError> {
518 unimplemented!("use gRPC implementation")
519 }
520
521 async fn get_token_transfer_action_onchain_signatures(
522 &self,
523 _bridge_object_arg: ObjectArg,
524 _source_chain_id: u8,
525 _seq_number: u64,
526 ) -> Result<Option<Vec<Vec<u8>>>, BridgeError> {
527 unimplemented!("use gRPC implementation")
528 }
529
530 async fn execute_transaction_block_with_effects(
531 &self,
532 tx: Transaction,
533 ) -> Result<ExecuteTransactionResult, BridgeError> {
534 use sui_json_rpc_types::SuiTransactionBlockEffectsAPI;
535 match self.quorum_driver_api().execute_transaction_block(
536 tx,
537 SuiTransactionBlockResponseOptions::new().with_effects().with_events(),
538 Some(sui_types::quorum_driver_types::ExecuteTransactionRequestType::WaitForEffectsCert),
539 ).await {
540 Ok(response) => {
541 let effects = response.effects.expect("We requested effects but got None.");
542 let events = response.events.expect("We requested events but got None.");
543 Ok(ExecuteTransactionResult {
544 status: effects.status().clone(),
545 events: events.data,
546 })
547 }
548 Err(e) => Err(BridgeError::SuiTxFailureGeneric(e.to_string())),
549 }
550 }
551
552 async fn get_parsed_token_transfer_message(
553 &self,
554 _bridge_object_arg: ObjectArg,
555 _source_chain_id: u8,
556 _seq_number: u64,
557 ) -> Result<Option<MoveTypeParsedTokenTransferMessage>, BridgeError> {
558 unimplemented!("use gRPC implementation")
559 }
560
561 async fn get_bridge_record(
562 &self,
563 _source_chain_id: u8,
564 _seq_number: u64,
565 ) -> Result<Option<MoveTypeBridgeRecord>, BridgeError> {
566 unimplemented!("use gRPC implementation")
567 }
568
569 async fn get_gas_data_panic_if_not_gas(
570 &self,
571 gas_object_id: ObjectID,
572 ) -> (GasCoin, ObjectRef, Owner) {
573 loop {
574 match self
575 .read_api()
576 .get_object_with_options(
577 gas_object_id,
578 SuiObjectDataOptions::default().with_owner().with_content(),
579 )
580 .await
581 .map(|resp| resp.data)
582 {
583 Ok(Some(gas_obj)) => {
584 let owner = gas_obj.owner.clone().expect("Owner is requested");
585 let gas_coin = GasCoin::try_from(&gas_obj)
586 .unwrap_or_else(|err| panic!("{} is not a gas coin: {err}", gas_object_id));
587 return (gas_coin, gas_obj.object_ref(), owner);
588 }
589 other => {
590 warn!("Can't get gas object: {:?}: {:?}", gas_object_id, other);
591 tokio::time::sleep(Duration::from_secs(5)).await;
592 }
593 }
594 }
595 }
596}
597
598#[async_trait]
599impl SuiClientInner for sui_rpc::Client {
600 async fn query_events(
601 &self,
602 _query: EventFilter,
603 _cursor: Option<EventID>,
604 ) -> Result<EventPage, BridgeError> {
605 unimplemented!("query_events not supported in gRPC");
608 }
609
610 async fn get_events_by_tx_digest(
611 &self,
612 tx_digest: TransactionDigest,
613 ) -> Result<SuiEvents, BridgeError> {
614 let mut client = self.clone();
615 let resp = client
616 .ledger_client()
617 .get_transaction(
618 GetTransactionRequest::new(&(tx_digest.into())).with_read_mask(
619 FieldMask::from_paths([
620 ExecutedTransaction::path_builder().digest(),
621 ExecutedTransaction::path_builder().events().finish(),
622 ExecutedTransaction::path_builder().checkpoint(),
623 ExecutedTransaction::path_builder().timestamp(),
624 ]),
625 ),
626 )
627 .await?
628 .into_inner();
629 let resp = resp.transaction();
630
631 Ok(SuiEvents {
632 transaction_digest: tx_digest,
633 checkpoint: resp.checkpoint_opt(),
634 timestamp_ms: resp
635 .timestamp_opt()
636 .map(|timestamp| sui_rpc::proto::proto_to_timestamp_ms(*timestamp))
637 .transpose()?,
638 events: resp
639 .events()
640 .events()
641 .iter()
642 .enumerate()
643 .map(|(idx, event)| {
644 Ok(SuiEvent {
645 id: EventID {
646 tx_digest,
647 event_seq: idx as u64,
648 },
649 package_id: event.package_id().parse()?,
650 transaction_module: Identifier::new(event.module())?,
651 sender: event.sender().parse()?,
652 type_: event.event_type().parse()?,
653 parsed_json: Default::default(),
654 bcs: BcsEvent::Base64 {
655 bcs: event.contents().value().into(),
656 },
657 timestamp_ms: None,
658 })
659 })
660 .collect::<Result<_, BridgeError>>()?,
661 })
662 }
663
664 async fn get_chain_identifier(&self) -> Result<String, BridgeError> {
665 let chain_id = self
666 .clone()
667 .ledger_client()
668 .get_service_info(GetServiceInfoRequest::default())
669 .await?
670 .into_inner()
671 .chain_id()
672 .parse::<sui_types::digests::CheckpointDigest>()?;
673
674 Ok(sui_types::digests::ChainIdentifier::from(chain_id).to_string())
675 }
676
677 async fn get_reference_gas_price(&self) -> Result<u64, BridgeError> {
678 let mut client = self.clone();
679 sui_rpc::Client::get_reference_gas_price(&mut client)
680 .await
681 .map_err(Into::into)
682 }
683
684 async fn get_latest_checkpoint_sequence_number(&self) -> Result<u64, BridgeError> {
685 let mut client = self.clone();
686 let resp =
687 client
688 .ledger_client()
689 .get_checkpoint(GetCheckpointRequest::latest().with_read_mask(
690 FieldMask::from_paths([Checkpoint::path_builder().sequence_number()]),
691 ))
692 .await?
693 .into_inner();
694 Ok(resp.checkpoint().sequence_number())
695 }
696
697 async fn get_mutable_bridge_object_arg(&self) -> Result<ObjectArg, BridgeError> {
698 let owner = self
699 .clone()
700 .ledger_client()
701 .get_object(
702 GetObjectRequest::new(&(SUI_BRIDGE_OBJECT_ID.into())).with_read_mask(
703 FieldMask::from_paths([Object::path_builder().owner().finish()]),
704 ),
705 )
706 .await?
707 .into_inner()
708 .object()
709 .owner()
710 .to_owned();
711 Ok(ObjectArg::SharedObject {
712 id: SUI_BRIDGE_OBJECT_ID,
713 initial_shared_version: SequenceNumber::from_u64(owner.version()),
714 mutability: SharedObjectMutability::Mutable,
715 })
716 }
717
718 async fn get_bridge_summary(&self) -> Result<BridgeSummary, BridgeError> {
719 static BRIDGE_VERSION_ID: tokio::sync::OnceCell<Address> =
720 tokio::sync::OnceCell::const_new();
721
722 let bridge_version_id = BRIDGE_VERSION_ID
723 .get_or_try_init::<BridgeError, _, _>(|| async {
724 let bridge_wrapper_bcs = self
725 .clone()
726 .ledger_client()
727 .get_object(
728 GetObjectRequest::new(&(SUI_BRIDGE_OBJECT_ID.into())).with_read_mask(
729 FieldMask::from_paths([Object::path_builder().contents().finish()]),
730 ),
731 )
732 .await?
733 .into_inner()
734 .object()
735 .contents()
736 .to_owned();
737
738 let bridge_wrapper: BridgeWrapper = bcs::from_bytes(bridge_wrapper_bcs.value())?;
739
740 Ok(bridge_wrapper.version.id.id.bytes.into())
741 })
742 .await?;
743
744 let bridge_inner_id = bridge_version_id
745 .derive_dynamic_child_id(&sui_sdk_types::TypeTag::U64, &bcs::to_bytes(&1u64).unwrap());
746
747 let field_bcs = self
748 .clone()
749 .ledger_client()
750 .get_object(GetObjectRequest::new(&bridge_inner_id).with_read_mask(
751 FieldMask::from_paths([Object::path_builder().contents().finish()]),
752 ))
753 .await?
754 .into_inner()
755 .object()
756 .contents()
757 .to_owned();
758
759 let field: sui_types::dynamic_field::Field<u64, sui_types::bridge::BridgeInnerV1> =
760 bcs::from_bytes(field_bcs.value())?;
761 let summary = field.value.try_into_bridge_summary()?;
762 Ok(summary)
763 }
764
765 async fn get_token_transfer_action_onchain_status(
766 &self,
767 _bridge_object_arg: ObjectArg,
768 source_chain_id: u8,
769 seq_number: u64,
770 ) -> Result<BridgeActionStatus, BridgeError> {
771 let record = self.get_bridge_record(source_chain_id, seq_number).await?;
772 let Some(record) = record else {
773 return Ok(BridgeActionStatus::NotFound);
774 };
775
776 if record.claimed {
777 Ok(BridgeActionStatus::Claimed)
778 } else if record.verified_signatures.is_some() {
779 Ok(BridgeActionStatus::Approved)
780 } else {
781 Ok(BridgeActionStatus::Pending)
782 }
783 }
784
785 async fn get_token_transfer_action_onchain_signatures(
786 &self,
787 _bridge_object_arg: ObjectArg,
788 source_chain_id: u8,
789 seq_number: u64,
790 ) -> Result<Option<Vec<Vec<u8>>>, BridgeError> {
791 let record = self.get_bridge_record(source_chain_id, seq_number).await?;
792 Ok(record.and_then(|record| record.verified_signatures))
793 }
794
795 async fn execute_transaction_block_with_effects(
796 &self,
797 tx: Transaction,
798 ) -> Result<ExecuteTransactionResult, BridgeError> {
799 use move_core_types::language_storage::StructTag;
800 use sui_rpc::proto::sui::rpc::v2::ExecutedTransaction as ProtoExecutedTransaction;
801 use sui_sdk_types::SignedTransaction;
802
803 let signed_tx: SignedTransaction = tx.try_into().map_err(|e| {
804 BridgeError::SuiTxFailureGeneric(format!("Failed to convert transaction: {:?}", e))
805 })?;
806
807 let proto_tx: ProtoTransaction = signed_tx.transaction.into();
808 let proto_sigs: Vec<ProtoUserSignature> =
809 signed_tx.signatures.into_iter().map(Into::into).collect();
810
811 let request = ExecuteTransactionRequest::default()
812 .with_transaction(proto_tx)
813 .with_signatures(proto_sigs)
814 .with_read_mask(FieldMask::from_paths([
815 ProtoExecutedTransaction::path_builder()
816 .effects()
817 .status()
818 .finish(),
819 ProtoExecutedTransaction::path_builder()
820 .events()
821 .events()
822 .finish(),
823 ]));
824
825 let response = self
826 .clone()
827 .execution_client()
828 .execute_transaction(request)
829 .await
830 .map_err(|e| BridgeError::SuiTxFailureGeneric(format!("gRPC execute failed: {:?}", e)))?
831 .into_inner();
832
833 let executed_tx = response.transaction();
834
835 let effects = executed_tx.effects();
836 let status = effects.status();
837
838 let sui_status = if status.success() {
839 SuiExecutionStatus::Success
840 } else {
841 let error = status.error();
842 let description = error.description().to_string();
843
844 let failure_msg = if !description.is_empty() {
845 description
846 } else {
847 format!("{:?}", error.kind())
848 };
849
850 SuiExecutionStatus::Failure { error: failure_msg }
851 };
852
853 let sui_events: Vec<SuiEvent> = executed_tx
854 .events()
855 .events()
856 .iter()
857 .filter_map(|event| {
858 let package_id: ObjectID = event.package_id().parse().ok()?;
859 let module = event.module().to_string();
860 let sender: sui_types::base_types::SuiAddress = event.sender().parse().ok()?;
861
862 let event_type_tag: sui_types::TypeTag =
863 parse_sui_type_tag(event.event_type()).ok()?;
864 let struct_tag: StructTag = match event_type_tag {
865 sui_types::TypeTag::Struct(s) => *s,
866 _ => return None,
867 };
868 let contents = event.contents();
869 let bcs_bytes = contents.value().to_vec();
870
871 Some(SuiEvent {
872 id: EventID {
873 tx_digest: TransactionDigest::default(),
874 event_seq: 0,
875 },
876 package_id,
877 transaction_module: Identifier::new(module).ok()?,
878 sender,
879 type_: struct_tag,
880 parsed_json: serde_json::Value::Null,
881 bcs: BcsEvent::new(bcs_bytes),
882 timestamp_ms: None,
883 })
884 })
885 .collect();
886
887 Ok(ExecuteTransactionResult {
888 status: sui_status,
889 events: sui_events,
890 })
891 }
892
893 async fn get_parsed_token_transfer_message(
894 &self,
895 _bridge_object_arg: ObjectArg,
896 source_chain_id: u8,
897 seq_number: u64,
898 ) -> Result<Option<MoveTypeParsedTokenTransferMessage>, BridgeError> {
899 let record = self.get_bridge_record(source_chain_id, seq_number).await?;
900
901 let Some(record) = record else {
902 return Ok(None);
903 };
904 let MoveTypeBridgeMessage {
905 message_type: _,
906 message_version,
907 seq_num,
908 source_chain,
909 payload,
910 } = record.message;
911
912 let mut parsed_payload: MoveTypeTokenTransferPayload = bcs::from_bytes(&payload)?;
913
914 parsed_payload.amount = u64::from_be_bytes(parsed_payload.amount.to_le_bytes());
916
917 Ok(Some(MoveTypeParsedTokenTransferMessage {
918 message_version,
919 seq_num,
920 source_chain,
921 payload,
922 parsed_payload,
923 }))
924 }
925
926 async fn get_bridge_record(
927 &self,
928 source_chain_id: u8,
929 seq_number: u64,
930 ) -> Result<Option<MoveTypeBridgeRecord>, BridgeError> {
931 static BRIDGE_RECORDS_ID: tokio::sync::OnceCell<Address> =
932 tokio::sync::OnceCell::const_new();
933
934 let records_id = BRIDGE_RECORDS_ID
935 .get_or_try_init(|| async {
936 self.get_bridge_summary()
937 .await
938 .map(|summary| summary.bridge_records_id.into())
939 })
940 .await?;
941
942 let record_id = {
943 let key = MoveTypeBridgeMessageKey {
944 source_chain: source_chain_id,
945 message_type: crate::types::BridgeActionType::TokenTransfer as u8,
946 bridge_seq_num: seq_number,
947 };
948 let key_bytes = bcs::to_bytes(&key)?;
949 let key_type = sui_sdk_types::StructTag::new(
950 Address::from(BRIDGE_PACKAGE_ID),
951 sui_sdk_types::Identifier::from_static("message"),
952 sui_sdk_types::Identifier::from_static("BridgeMessageKey"),
953 vec![],
954 );
955
956 records_id.derive_dynamic_child_id(&(key_type.into()), &key_bytes)
957 };
958
959 let response =
960 match self
961 .clone()
962 .ledger_client()
963 .get_object(GetObjectRequest::new(&record_id).with_read_mask(
964 FieldMask::from_paths([Object::path_builder().contents().finish()]),
965 ))
966 .await
967 {
968 Ok(response) => response,
969 Err(status) => {
970 if status.code() == tonic::Code::NotFound {
971 return Ok(None);
972 } else {
973 return Err(status.into());
974 }
975 }
976 };
977
978 let field_bcs = response.into_inner().object().contents().to_owned();
979
980 let field: sui_types::dynamic_field::Field<
981 MoveTypeBridgeMessageKey,
982 LinkedTableNode<MoveTypeBridgeMessageKey, MoveTypeBridgeRecord>,
983 > = bcs::from_bytes(field_bcs.value())?;
984
985 Ok(Some(field.value.value))
986 }
987
988 async fn get_gas_data_panic_if_not_gas(
989 &self,
990 gas_object_id: ObjectID,
991 ) -> (GasCoin, ObjectRef, Owner) {
992 loop {
993 let result = async {
994 let resp = self
995 .clone()
996 .ledger_client()
997 .get_object(
998 GetObjectRequest::new(&(gas_object_id.into())).with_read_mask(
999 FieldMask::from_paths([Object::path_builder().bcs().finish()]),
1000 ),
1001 )
1002 .await?
1003 .into_inner();
1004
1005 let obj = resp.object();
1006 let object: sui_types::object::Object = obj.bcs().deserialize().map_err(|e| {
1007 BridgeError::Generic(format!("Failed to deserialize object from BCS: {e}"))
1008 })?;
1009
1010 let object_ref = object.compute_object_reference();
1011 let owner = object.owner().clone();
1012 let gas_coin = GasCoin::try_from(&object).map_err(|e| {
1013 BridgeError::Generic(format!("Failed to convert object to gas coin: {e}"))
1014 })?;
1015
1016 Ok::<_, BridgeError>((gas_coin, object_ref, owner))
1017 }
1018 .await;
1019
1020 match result {
1021 Ok(data) => return data,
1022 Err(e) => {
1023 warn!("Can't get gas object: {:?}: {:?}", gas_object_id, e);
1024 tokio::time::sleep(Duration::from_secs(5)).await;
1025 }
1026 }
1027 }
1028 }
1029}
1030
1031#[async_trait]
1032impl SuiClientInner for SuiClientInternal {
1033 async fn query_events(
1034 &self,
1035 query: EventFilter,
1036 cursor: Option<EventID>,
1037 ) -> Result<EventPage, BridgeError> {
1038 self.jsonrpc_client.query_events(query, cursor).await
1039 }
1040
1041 async fn get_events_by_tx_digest(
1042 &self,
1043 tx_digest: TransactionDigest,
1044 ) -> Result<SuiEvents, BridgeError> {
1045 self.grpc_client.get_events_by_tx_digest(tx_digest).await
1046 }
1047
1048 async fn get_chain_identifier(&self) -> Result<String, BridgeError> {
1049 self.grpc_client.get_chain_identifier().await
1050 }
1051
1052 async fn get_reference_gas_price(&self) -> Result<u64, BridgeError> {
1053 self.grpc_client.get_reference_gas_price().await
1054 }
1055
1056 async fn get_latest_checkpoint_sequence_number(&self) -> Result<u64, BridgeError> {
1057 self.grpc_client
1058 .get_latest_checkpoint_sequence_number()
1059 .await
1060 }
1061
1062 async fn get_mutable_bridge_object_arg(&self) -> Result<ObjectArg, BridgeError> {
1063 self.grpc_client.get_mutable_bridge_object_arg().await
1064 }
1065
1066 async fn get_bridge_summary(&self) -> Result<BridgeSummary, BridgeError> {
1067 self.grpc_client.get_bridge_summary().await
1068 }
1069
1070 async fn get_token_transfer_action_onchain_status(
1071 &self,
1072 bridge_object_arg: ObjectArg,
1073 source_chain_id: u8,
1074 seq_number: u64,
1075 ) -> Result<BridgeActionStatus, BridgeError> {
1076 self.grpc_client
1077 .get_token_transfer_action_onchain_status(
1078 bridge_object_arg,
1079 source_chain_id,
1080 seq_number,
1081 )
1082 .await
1083 }
1084
1085 async fn get_token_transfer_action_onchain_signatures(
1086 &self,
1087 bridge_object_arg: ObjectArg,
1088 source_chain_id: u8,
1089 seq_number: u64,
1090 ) -> Result<Option<Vec<Vec<u8>>>, BridgeError> {
1091 self.grpc_client
1092 .get_token_transfer_action_onchain_signatures(
1093 bridge_object_arg,
1094 source_chain_id,
1095 seq_number,
1096 )
1097 .await
1098 }
1099
1100 async fn execute_transaction_block_with_effects(
1101 &self,
1102 tx: Transaction,
1103 ) -> Result<ExecuteTransactionResult, BridgeError> {
1104 self.grpc_client
1105 .execute_transaction_block_with_effects(tx)
1106 .await
1107 }
1108
1109 async fn get_parsed_token_transfer_message(
1110 &self,
1111 bridge_object_arg: ObjectArg,
1112 source_chain_id: u8,
1113 seq_number: u64,
1114 ) -> Result<Option<MoveTypeParsedTokenTransferMessage>, BridgeError> {
1115 self.grpc_client
1116 .get_parsed_token_transfer_message(bridge_object_arg, source_chain_id, seq_number)
1117 .await
1118 }
1119
1120 async fn get_bridge_record(
1121 &self,
1122 source_chain_id: u8,
1123 seq_number: u64,
1124 ) -> Result<Option<MoveTypeBridgeRecord>, BridgeError> {
1125 self.grpc_client
1126 .get_bridge_record(source_chain_id, seq_number)
1127 .await
1128 }
1129
1130 async fn get_gas_data_panic_if_not_gas(
1131 &self,
1132 gas_object_id: ObjectID,
1133 ) -> (GasCoin, ObjectRef, Owner) {
1134 self.jsonrpc_client
1135 .get_gas_data_panic_if_not_gas(gas_object_id)
1136 .await
1137 }
1138}
1139
1140#[cfg(test)]
1141mod tests {
1142 use crate::crypto::BridgeAuthorityKeyPair;
1143 use crate::e2e_tests::test_utils::TestClusterWrapperBuilder;
1144 use crate::types::SuiToEthTokenTransfer;
1145 use crate::{
1146 events::{EmittedSuiToEthTokenBridgeV1, MoveTokenDepositedEvent},
1147 sui_mock_client::SuiMockClient,
1148 test_utils::{
1149 approve_action_with_validator_secrets, bridge_token, get_test_eth_to_sui_bridge_action,
1150 get_test_sui_to_eth_bridge_action,
1151 },
1152 };
1153 use ethers::types::Address as EthAddress;
1154 use move_core_types::account_address::AccountAddress;
1155 use serde::{Deserialize, Serialize};
1156 use std::str::FromStr;
1157 use sui_json_rpc_types::BcsEvent;
1158 use sui_types::base_types::SuiAddress;
1159 use sui_types::bridge::{BridgeChainId, TOKEN_ID_SUI, TOKEN_ID_USDC};
1160 use sui_types::crypto::get_key_pair;
1161
1162 use super::*;
1163 use crate::events::{SuiToEthTokenBridgeV1, init_all_struct_tags};
1164
1165 #[tokio::test]
1166 async fn get_bridge_action_by_tx_digest_and_event_idx_maybe() {
1167 telemetry_subscribers::init_for_testing();
1171 let mock_client = SuiMockClient::default();
1172 let sui_client = SuiClient::new_for_testing(mock_client.clone());
1173 let tx_digest = TransactionDigest::random();
1174
1175 init_all_struct_tags();
1177
1178 let sanitized_event_1 = EmittedSuiToEthTokenBridgeV1 {
1179 nonce: 1,
1180 sui_chain_id: BridgeChainId::SuiTestnet,
1181 sui_address: SuiAddress::random_for_testing_only(),
1182 eth_chain_id: BridgeChainId::EthSepolia,
1183 eth_address: EthAddress::random(),
1184 token_id: TOKEN_ID_SUI,
1185 amount_sui_adjusted: 100,
1186 };
1187 let emitted_event_1 = MoveTokenDepositedEvent {
1188 seq_num: sanitized_event_1.nonce,
1189 source_chain: sanitized_event_1.sui_chain_id as u8,
1190 sender_address: sanitized_event_1.sui_address.to_vec(),
1191 target_chain: sanitized_event_1.eth_chain_id as u8,
1192 target_address: sanitized_event_1.eth_address.as_bytes().to_vec(),
1193 token_type: sanitized_event_1.token_id,
1194 amount_sui_adjusted: sanitized_event_1.amount_sui_adjusted,
1195 };
1196
1197 let mut sui_event_1 = SuiEvent::random_for_testing();
1198 sui_event_1.type_ = SuiToEthTokenBridgeV1.get().unwrap().clone();
1199 sui_event_1.bcs = BcsEvent::new(bcs::to_bytes(&emitted_event_1).unwrap());
1200
1201 #[derive(Serialize, Deserialize)]
1202 struct RandomStruct {}
1203
1204 let event_2: RandomStruct = RandomStruct {};
1205 let mut sui_event_2 = SuiEvent::random_for_testing();
1207 sui_event_2.type_ = SuiToEthTokenBridgeV1.get().unwrap().clone();
1208 sui_event_2.type_.module = Identifier::from_str("unrecognized_module").unwrap();
1209 sui_event_2.bcs = BcsEvent::new(bcs::to_bytes(&event_2).unwrap());
1210
1211 let mut sui_event_3 = sui_event_1.clone();
1213 sui_event_3.type_.address = AccountAddress::random();
1214
1215 mock_client.add_events_by_tx_digest(
1216 tx_digest,
1217 vec![
1218 sui_event_1.clone(),
1219 sui_event_2.clone(),
1220 sui_event_1.clone(),
1221 sui_event_3.clone(),
1222 ],
1223 );
1224 let expected_action = BridgeAction::SuiToEthTokenTransfer(SuiToEthTokenTransfer {
1225 nonce: sanitized_event_1.nonce,
1226 sui_chain_id: sanitized_event_1.sui_chain_id,
1227 eth_chain_id: sanitized_event_1.eth_chain_id,
1228 sui_address: sanitized_event_1.sui_address,
1229 eth_address: sanitized_event_1.eth_address,
1230 token_id: sanitized_event_1.token_id,
1231 amount_adjusted: sanitized_event_1.amount_sui_adjusted,
1232 });
1233 assert_eq!(
1234 sui_client
1235 .get_bridge_action_by_tx_digest_and_event_idx_maybe(&tx_digest, 0)
1236 .await
1237 .unwrap(),
1238 expected_action,
1239 );
1240 assert_eq!(
1241 sui_client
1242 .get_bridge_action_by_tx_digest_and_event_idx_maybe(&tx_digest, 2)
1243 .await
1244 .unwrap(),
1245 expected_action,
1246 );
1247 assert!(matches!(
1248 sui_client
1249 .get_bridge_action_by_tx_digest_and_event_idx_maybe(&tx_digest, 1)
1250 .await
1251 .unwrap_err(),
1252 BridgeError::NoBridgeEventsInTxPosition
1253 ),);
1254 assert!(matches!(
1255 sui_client
1256 .get_bridge_action_by_tx_digest_and_event_idx_maybe(&tx_digest, 3)
1257 .await
1258 .unwrap_err(),
1259 BridgeError::BridgeEventInUnrecognizedSuiPackage
1260 ),);
1261 assert!(matches!(
1262 sui_client
1263 .get_bridge_action_by_tx_digest_and_event_idx_maybe(&tx_digest, 4)
1264 .await
1265 .unwrap_err(),
1266 BridgeError::NoBridgeEventsInTxPosition
1267 ),);
1268
1269 sui_event_2.type_ = SuiToEthTokenBridgeV1.get().unwrap().clone();
1271 mock_client.add_events_by_tx_digest(tx_digest, vec![sui_event_2]);
1272 sui_client
1273 .get_bridge_action_by_tx_digest_and_event_idx_maybe(&tx_digest, 2)
1274 .await
1275 .unwrap_err();
1276 }
1277
1278 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
1282 async fn test_get_action_onchain_status_for_sui_to_eth_transfer() {
1283 telemetry_subscribers::init_for_testing();
1284 let mut bridge_keys = vec![];
1285 for _ in 0..=3 {
1286 let (_, kp): (_, BridgeAuthorityKeyPair) = get_key_pair();
1287 bridge_keys.push(kp);
1288 }
1289 let mut test_cluster = TestClusterWrapperBuilder::new()
1290 .with_bridge_authority_keys(bridge_keys)
1291 .with_deploy_tokens(true)
1292 .build()
1293 .await;
1294
1295 let bridge_metrics = Arc::new(BridgeMetrics::new_for_testing());
1296 let sui_client =
1297 SuiClient::new(&test_cluster.inner.fullnode_handle.rpc_url, bridge_metrics)
1298 .await
1299 .unwrap();
1300 let bridge_authority_keys = test_cluster.authority_keys_clone();
1301
1302 test_cluster
1304 .trigger_reconfiguration_if_not_yet_and_assert_bridge_committee_initialized()
1305 .await;
1306 let context = &mut test_cluster.inner.wallet;
1307 let sender = context.active_address().unwrap();
1308 let usdc_amount = 5000000;
1309 let bridge_object_arg = sui_client
1310 .get_mutable_bridge_object_arg_must_succeed()
1311 .await;
1312 let id_token_map = sui_client.get_token_id_map().await.unwrap();
1313
1314 let action = get_test_eth_to_sui_bridge_action(None, Some(usdc_amount), Some(sender), None);
1316 let usdc_object_ref = approve_action_with_validator_secrets(
1317 context,
1318 bridge_object_arg,
1319 action.clone(),
1320 &bridge_authority_keys,
1321 Some(sender),
1322 &id_token_map,
1323 )
1324 .await
1325 .unwrap();
1326
1327 let status = sui_client
1328 .inner
1329 .get_token_transfer_action_onchain_status(
1330 bridge_object_arg,
1331 action.chain_id() as u8,
1332 action.seq_number(),
1333 )
1334 .await
1335 .unwrap();
1336 assert_eq!(status, BridgeActionStatus::Claimed);
1337
1338 let eth_recv_address = EthAddress::random();
1341 let bridge_event = bridge_token(
1342 context,
1343 eth_recv_address,
1344 usdc_object_ref,
1345 id_token_map.get(&TOKEN_ID_USDC).unwrap().clone(),
1346 bridge_object_arg,
1347 )
1348 .await;
1349 assert_eq!(bridge_event.nonce, 0);
1350 assert_eq!(bridge_event.sui_chain_id, BridgeChainId::SuiCustom);
1351 assert_eq!(bridge_event.eth_chain_id, BridgeChainId::EthCustom);
1352 assert_eq!(bridge_event.eth_address, eth_recv_address);
1353 assert_eq!(bridge_event.sui_address, sender);
1354 assert_eq!(bridge_event.token_id, TOKEN_ID_USDC);
1355 assert_eq!(bridge_event.amount_sui_adjusted, usdc_amount);
1356
1357 let action = get_test_sui_to_eth_bridge_action(
1358 None,
1359 None,
1360 Some(bridge_event.nonce),
1361 Some(bridge_event.amount_sui_adjusted),
1362 Some(bridge_event.sui_address),
1363 Some(bridge_event.eth_address),
1364 Some(TOKEN_ID_USDC),
1365 );
1366 let status = sui_client
1367 .inner
1368 .get_token_transfer_action_onchain_status(
1369 bridge_object_arg,
1370 action.chain_id() as u8,
1371 action.seq_number(),
1372 )
1373 .await
1374 .unwrap();
1375 assert_eq!(status, BridgeActionStatus::Pending);
1377
1378 approve_action_with_validator_secrets(
1380 context,
1381 bridge_object_arg,
1382 action.clone(),
1383 &bridge_authority_keys,
1384 None,
1385 &id_token_map,
1386 )
1387 .await;
1388
1389 let status = sui_client
1390 .inner
1391 .get_token_transfer_action_onchain_status(
1392 bridge_object_arg,
1393 action.chain_id() as u8,
1394 action.seq_number(),
1395 )
1396 .await
1397 .unwrap();
1398 assert_eq!(status, BridgeActionStatus::Approved);
1399
1400 let action =
1402 get_test_sui_to_eth_bridge_action(None, None, Some(100), None, None, None, None);
1403 let status = sui_client
1404 .inner
1405 .get_token_transfer_action_onchain_status(
1406 bridge_object_arg,
1407 action.chain_id() as u8,
1408 action.seq_number(),
1409 )
1410 .await
1411 .unwrap();
1412 assert_eq!(status, BridgeActionStatus::NotFound);
1413 }
1414}