@ -1,90 +1,31 @@
use axum ::{
extract ::{ Query , State } ,
routing , Router ,
} ;
use crate ::settings ::matching_list ::MatchingList ;
use axum ::{ extract ::State , routing , Json , Router } ;
use derive_new ::new ;
use hyperlane_core ::{ ChainCommunicationError , QueueOperation , H256 } ;
use serde ::Deserialize ;
use std ::str ::FromStr ;
use tokio ::sync ::broadcast ::Sender ;
const MESSAGE_RETRY_API_BASE : & str = "/message_retry" ;
#[ derive(Clone, Debug, PartialEq, Eq) ]
pub enum MessageRetryRequest {
MessageId ( H256 ) ,
DestinationDomain ( u32 ) ,
}
impl PartialEq < QueueOperation > for & MessageRetryRequest {
fn eq ( & self , other : & QueueOperation ) -> bool {
match self {
MessageRetryRequest ::MessageId ( message_id ) = > message_id = = & other . id ( ) ,
MessageRetryRequest ::DestinationDomain ( destination_domain ) = > {
destination_domain = = & other . destination_domain ( ) . id ( )
}
}
}
}
#[ derive(new, Clone) ]
pub struct MessageRetryApi {
tx : Sender < MessageRetryRequest > ,
}
#[ derive(Deserialize) ]
struct RawMessageRetryRequest {
message_id : Option < String > ,
destination_domain : Option < u32 > ,
}
impl TryFrom < RawMessageRetryRequest > for Vec < MessageRetryRequest > {
type Error = ChainCommunicationError ;
fn try_from ( request : RawMessageRetryRequest ) -> Result < Self , Self ::Error > {
let mut retry_requests = Vec ::new ( ) ;
if let Some ( message_id ) = request . message_id {
retry_requests . push ( MessageRetryRequest ::MessageId ( H256 ::from_str ( & message_id ) ? ) ) ;
}
if let Some ( destination_domain ) = request . destination_domain {
retry_requests . push ( MessageRetryRequest ::DestinationDomain ( destination_domain ) ) ;
}
Ok ( retry_requests )
}
tx : Sender < MatchingList > ,
}
async fn retry_message (
State ( tx ) : State < Sender < MessageRetryReque st > > ,
Query ( request ) : Query < RawMessageRetryReque st> ,
State ( tx ) : State < Sender < MatchingList > > ,
Json ( retry_req_payload ) : Json < MatchingList > ,
) -> String {
let retry_requests : Vec < MessageRetryRequest > = match request . try_into ( ) {
Ok ( retry _requests ) = > retry_requests ,
match tx . send ( retry_req_payload ) {
Ok ( _ ) = > "Moved message(s) to the front of the queue" . to_string ( ) ,
// Technically it's bad practice to print the error message to the user, but
// this endpoint is for debugging purposes only.
Err ( err ) = > {
return format! ( "Failed to parse retry request: {}" , err ) ;
}
} ;
if retry_requests . is_empty ( ) {
return "No retry requests found. Please provide either a message_id or destination_domain." . to_string ( ) ;
Err ( err ) = > format! ( "Failed to send retry request to the queue: {}" , err ) ,
}
if let Err ( err ) = retry_requests
. into_iter ( )
. map ( | req | tx . send ( req ) )
. collect ::< Result < Vec < _ > , _ > > ( )
{
return format! ( "Failed to send retry request to the queue: {}" , err ) ;
}
"Moved message(s) to the front of the queue" . to_string ( )
}
impl MessageRetryApi {
pub fn router ( & self ) -> Router {
Router ::new ( )
. route ( "/" , routing ::ge t( retry_message ) )
. route ( "/" , routing ::post ( retry_message ) )
. with_state ( self . tx . clone ( ) )
}
@ -95,18 +36,20 @@ impl MessageRetryApi {
#[ cfg(test) ]
mod tests {
use crate ::server ::ENDPOINT_MESSAGES_QUEUE_SIZE ;
use crate ::{ msg ::op_queue ::test ::MockPendingOperation , server ::ENDPOINT_MESSAGES_QUEUE_SIZE } ;
use super ::* ;
use axum ::http ::StatusCode ;
use ethers ::utils ::hex ::ToHex ;
use hyperlane_core ::{ HyperlaneMessage , QueueOperation } ;
use serde_json ::json ;
use std ::net ::SocketAddr ;
use tokio ::sync ::broadcast ::{ Receiver , Sender } ;
fn setup_test_server ( ) -> ( SocketAddr , Receiver < MessageRetryReque st > ) {
let broadcast_tx = Sender ::< MessageRetryReque st > ::new ( ENDPOINT_MESSAGES_QUEUE_SIZE ) ;
fn setup_test_server ( ) -> ( SocketAddr , Receiver < MatchingLi st > ) {
let broadcast_tx = Sender ::< MatchingLi st > ::new ( ENDPOINT_MESSAGES_QUEUE_SIZE ) ;
let message_retry_api = MessageRetryApi ::new ( broadcast_tx . clone ( ) ) ;
let ( path , retry_router ) = message_retry_api . get_route ( ) ;
let app = Router ::new ( ) . nest ( path , retry_router ) ;
// Running the app in the background using a test server
@ -122,49 +65,186 @@ mod tests {
async fn test_message_id_retry ( ) {
let ( addr , mut rx ) = setup_test_server ( ) ;
// Create a random message ID
let message_id = H256 ::random ( ) ;
let client = reqwest ::Client ::new ( ) ;
// Create a random message with a random message ID
let message = HyperlaneMessage ::default ( ) ;
let pending_operation = MockPendingOperation ::with_message_data ( message . clone ( ) ) ;
let matching_list_body = json ! ( [
{
"messageid" : message . id ( )
}
] ) ;
// Send a GET request to the server
let response = reqwest ::get ( format! (
"http://{}{}?message_id={}" ,
addr ,
MESSAGE_RETRY_API_BASE ,
message_id . encode_hex ::< String > ( )
) )
. await
. unwrap ( ) ;
// Send a POST request to the server
let response = client
. post ( format! ( "http://{}{}" , addr , MESSAGE_RETRY_API_BASE ) )
. json ( & matching_list_body ) // Set the request body
. send ( )
. await
. unwrap ( ) ;
// Check that the response status code is OK
assert_eq! ( response . status ( ) , StatusCode ::OK ) ;
assert_eq! (
rx . try_recv ( ) . unwrap ( ) ,
MessageRetryRequest ::MessageId ( message_id )
) ;
let list = rx . try_recv ( ) . unwrap ( ) ;
// Check that the list received by the server matches the pending operation
assert! ( list . op_matches ( & ( Box ::new ( pending_operation ) as QueueOperation ) ) ) ;
}
#[ tokio::test ]
async fn test_destination_domain_retry ( ) {
let ( addr , mut rx ) = setup_test_server ( ) ;
// Create a random destination domain
let destination_domain = 42 ;
let client = reqwest ::Client ::new ( ) ;
let mut message = HyperlaneMessage ::default ( ) ;
// Use a random destination domain
message . destination = 42 ;
let pending_operation = MockPendingOperation ::with_message_data ( message . clone ( ) ) ;
let matching_list_body = json ! ( [
{
"destinationdomain" : message . destination
}
] ) ;
// Send a POST request to the server
let response = client
. post ( format! ( "http://{}{}" , addr , MESSAGE_RETRY_API_BASE ) )
. json ( & matching_list_body ) // Set the request body
. send ( )
. await
. unwrap ( ) ;
// Check that the response status code is OK
assert_eq! ( response . status ( ) , StatusCode ::OK ) ;
let list = rx . try_recv ( ) . unwrap ( ) ;
// Check that the list received by the server matches the pending operation
assert! ( list . op_matches ( & ( Box ::new ( pending_operation ) as QueueOperation ) ) ) ;
}
#[ tokio::test ]
async fn test_origin_domain_retry ( ) {
let ( addr , mut rx ) = setup_test_server ( ) ;
let client = reqwest ::Client ::new ( ) ;
let mut message = HyperlaneMessage ::default ( ) ;
// Use a random origin domain
message . origin = 42 ;
let pending_operation = MockPendingOperation ::with_message_data ( message . clone ( ) ) ;
let matching_list_body = json ! ( [
{
"origindomain" : message . origin
}
] ) ;
// Send a POST request to the server
let response = client
. post ( format! ( "http://{}{}" , addr , MESSAGE_RETRY_API_BASE ) )
. json ( & matching_list_body ) // Set the request body
. send ( )
. await
. unwrap ( ) ;
// Check that the response status code is OK
assert_eq! ( response . status ( ) , StatusCode ::OK ) ;
let list = rx . try_recv ( ) . unwrap ( ) ;
// Check that the list received by the server matches the pending operation
assert! ( list . op_matches ( & ( Box ::new ( pending_operation ) as QueueOperation ) ) ) ;
}
#[ tokio::test ]
async fn test_sender_address_retry ( ) {
let ( addr , mut rx ) = setup_test_server ( ) ;
let client = reqwest ::Client ::new ( ) ;
let message = HyperlaneMessage ::default ( ) ;
let pending_operation = MockPendingOperation ::with_message_data ( message . clone ( ) ) ;
let matching_list_body = json ! ( [
{
"senderaddress" : message . sender
}
] ) ;
// Send a POST request to the server
let response = client
. post ( format! ( "http://{}{}" , addr , MESSAGE_RETRY_API_BASE ) )
. json ( & matching_list_body ) // Set the request body
. send ( )
. await
. unwrap ( ) ;
// Check that the response status code is OK
assert_eq! ( response . status ( ) , StatusCode ::OK ) ;
let list = rx . try_recv ( ) . unwrap ( ) ;
// Check that the list received by the server matches the pending operation
assert! ( list . op_matches ( & ( Box ::new ( pending_operation ) as QueueOperation ) ) ) ;
}
#[ tokio::test ]
async fn test_recipient_address_retry ( ) {
let ( addr , mut rx ) = setup_test_server ( ) ;
let client = reqwest ::Client ::new ( ) ;
let message = HyperlaneMessage ::default ( ) ;
let pending_operation = MockPendingOperation ::with_message_data ( message . clone ( ) ) ;
let matching_list_body = json ! ( [
{
"recipientaddress" : message . recipient
}
] ) ;
// Send a POST request to the server
let response = client
. post ( format! ( "http://{}{}" , addr , MESSAGE_RETRY_API_BASE ) )
. json ( & matching_list_body ) // Set the request body
. send ( )
. await
. unwrap ( ) ;
// Check that the response status code is OK
assert_eq! ( response . status ( ) , StatusCode ::OK ) ;
let list = rx . try_recv ( ) . unwrap ( ) ;
// Check that the list received by the server matches the pending operation
assert! ( list . op_matches ( & ( Box ::new ( pending_operation ) as QueueOperation ) ) ) ;
}
#[ tokio::test ]
async fn test_multiple_retry ( ) {
let ( addr , mut rx ) = setup_test_server ( ) ;
let client = reqwest ::Client ::new ( ) ;
let mut message = HyperlaneMessage ::default ( ) ;
// Use a random origin domain
message . origin = 42 ;
let pending_operation = MockPendingOperation ::with_message_data ( message . clone ( ) ) ;
let matching_list_body = json ! ( [
{
"origindomain" : message . origin
} ,
{
"destinationdomain" : message . destination
} ,
{
"messageid" : message . id ( )
}
] ) ;
// Send a GET request to the server
let response = reqwest ::get ( format! (
"http://{}{}?destination_domain={}" ,
addr , MESSAGE_RETRY_API_BASE , destination_domain
) )
. await
. unwrap ( ) ;
// Send a POS T request to the server
let response = client
. post ( format! ( "http://{}{}" , addr , MESSAGE_RETRY_API_BASE ) )
. json ( & matching_list_body ) // Set the request body
. send ( )
. await
. unwrap ( ) ;
// Check that the response status code is OK
assert_eq! ( response . status ( ) , StatusCode ::OK ) ;
assert_eq! (
rx . try_recv ( ) . unwrap ( ) ,
MessageRetryRequest ::DestinationDomain ( destination_domain )
) ;
let list = rx . try_recv ( ) . unwrap ( ) ;
// Check that the list received by the server matches the pending operation
assert! ( list . op_matches ( & ( Box ::new ( pending_operation ) as QueueOperation ) ) ) ;
}
}
Mantas-M
3 weeks ago
committed by
GitHub