feat(nexus)!: pass msg id between amplifier and core (axelarnetwork#2152

)

* feat(nexus)!: pass msg id between amplifier and core

proto/axelar/nexus/exported/v1beta1/types.proto

@@ -123,4 +123,7 @@ message WasmMessage {
   uint64 source_tx_index = 7 [ (gogoproto.jsontag) = "source_tx_index" ];
   bytes sender = 8 [ (gogoproto.casttype) =
                          "github.com/cosmos/cosmos-sdk/types.AccAddress" ];
+  string id = 9 [
+    (gogoproto.customname) = "ID"
+  ];
 }

x/nexus/exported/types.go

@@ -349,16 +349,43 @@ func FromGeneralMessage(msg GeneralMessage) WasmMessage {
 		PayloadHash:        msg.PayloadHash,
 		SourceTxID:         msg.SourceTxID,
 		SourceTxIndex:      msg.SourceTxIndex,
+		ID:                 msg.ID,
 	}
 }
 
 var _ sdk.Msg = &WasmMessage{}
 
 // ValidateBasic implements sdk.Msg
 func (m WasmMessage) ValidateBasic() error {
-	// The ValidateBasic function allows cheap stateless checks to fail msg handling early during CheckTx.
-	// This message can only be sent by the cosmwasm router as part of the amplifier integration into core,
-	// so there is no point in validating the message here.
+
+	if err := utils.ValidateString(m.ID); err != nil {
+		return sdkerrors.Wrap(err, "invalid wasm message id")
+	}
+
+	if err := m.SourceChain.Validate(); err != nil {
+		return sdkerrors.Wrap(err, "invalid wasm message source chain name")
+	}
+
+	if err := utils.ValidateString(m.SourceAddress); err != nil {
+		return sdkerrors.Wrap(err, "invalid wasm message source address")
+	}
+
+	if err := m.DestinationChain.Validate(); err != nil {
+		return sdkerrors.Wrap(err, "invalid wasm message destination chain name")
+	}
+
+	if err := utils.ValidateString(m.DestinationAddress); err != nil {
+		return sdkerrors.Wrap(err, "invalid wasm message destination address")
+	}
+
+	if len(m.PayloadHash) != 32 {
+		return fmt.Errorf("invalid wasm message payload hash")
+	}
+
+	if len(m.SourceTxID) != 32 {
+		return fmt.Errorf("invalid wasm message source tx id")
+	}
+
 	return nil
 }
 

x/nexus/keeper/msg_dispatcher.go

@@ -55,17 +55,25 @@ func (m Messenger) DispatchMsg(ctx sdk.Context, contractAddr sdk.AccAddress, _ s
 }
 
 func (m Messenger) routeMsg(ctx sdk.Context, msg exported.WasmMessage) error {
+	if err := msg.ValidateBasic(); err != nil {
+		return err
+	}
+
 	destinationChain, ok := m.GetChain(ctx, msg.DestinationChain)
 	if !ok {
 		return fmt.Errorf("recipient chain %s is not a registered chain", msg.DestinationChain)
 	}
 
-	id, _, _ := m.GenerateMessageID(ctx)
+	// If message already exists, then this is a no-op to avoid causing an error from reverting the whole message batch being routed in Amplifier
+	if _, ok := m.Nexus.GetMessage(ctx, msg.ID); ok {
+		return nil
+	}
+
 	sourceChain := exported.Chain{Name: msg.SourceChain, SupportsForeignAssets: false, KeyType: tss.None, Module: wasmtypes.ModuleName}
 	sender := exported.CrossChainAddress{Chain: sourceChain, Address: msg.SourceAddress}
 	recipient := exported.CrossChainAddress{Chain: destinationChain, Address: msg.DestinationAddress}
 
-	nexusMsg := exported.NewGeneralMessage(id, sender, recipient, msg.PayloadHash, msg.SourceTxID, msg.SourceTxIndex, nil)
+	nexusMsg := exported.NewGeneralMessage(fmt.Sprintf("%s-%s", msg.SourceChain, msg.ID), sender, recipient, msg.PayloadHash, msg.SourceTxID, msg.SourceTxIndex, nil)
 	if err := m.Nexus.SetNewMessage(ctx, nexusMsg); err != nil {
 		return err
 	}

x/nexus/keeper/msg_dispatcher_test.go

@@ -58,7 +58,7 @@ func TestMessenger_DispatchMsg(t *testing.T) {
 	givenMessenger.
 		When("the msg is encoded correctly", func() {
 			msg = wasmvmtypes.CosmosMsg{
-				Custom: []byte("{}"),
+				Custom: []byte("{\"source_chain\":\"sourcechain\",\"source_address\":\"0xb860\",\"destination_chain\":\"Axelarnet\",\"destination_address\":\"axelarvaloper1zh9wrak6ke4n6fclj5e8yk397czv430ygs5jz7\",\"payload_hash\":[187,155,85,102,194,244,135,104,99,51,62,72,31,70,152,53,1,84,37,159,254,98,38,226,131,177,108,225,138,100,188,241],\"source_tx_id\":[187,155,85,102,194,244,135,104,99,51,62,72,31,70,152,53,1,84,37,159,254,98,38,226,131,177,108,225,138,100,188,241],\"source_tx_index\":100, \"id\": \"0x73657e3da2e404f474218fe2789462585d7f6060741bd312c862378cf67ca981-1\"}"),
 			}
 		}).
 		Branch(
@@ -94,7 +94,7 @@ func TestMessenger_DispatchMsg(t *testing.T) {
 	givenMessenger.
 		When("the msg is encoded correctly and the gateway is set correctly", func() {
 			msg = wasmvmtypes.CosmosMsg{
-				Custom: []byte("{}"),
+				Custom: []byte("{\"source_chain\":\"sourcechain\",\"source_address\":\"0xb860\",\"destination_chain\":\"Axelarnet\",\"destination_address\":\"axelarvaloper1zh9wrak6ke4n6fclj5e8yk397czv430ygs5jz7\",\"payload_hash\":[187,155,85,102,194,244,135,104,99,51,62,72,31,70,152,53,1,84,37,159,254,98,38,226,131,177,108,225,138,100,188,241],\"source_tx_id\":[187,155,85,102,194,244,135,104,99,51,62,72,31,70,152,53,1,84,37,159,254,98,38,226,131,177,108,225,138,100,188,241],\"source_tx_index\":100, \"id\": \"0x73657e3da2e404f474218fe2789462585d7f6060741bd312c862378cf67ca981-1\"}"),
 			}
 
 			nexus.GetParamsFunc = func(_ sdk.Context) types.Params {
@@ -125,6 +125,9 @@ func TestMessenger_DispatchMsg(t *testing.T) {
 					nexus.GenerateMessageIDFunc = func(_ sdk.Context) (string, []byte, uint64) {
 						return "1", []byte("1"), 1
 					}
+					nexus.GetMessageFunc = func(_ sdk.Context, _ string) (exported.GeneralMessage, bool) {
+						return exported.GeneralMessage{}, false
+					}
 					nexus.SetNewMessageFunc = func(_ sdk.Context, _ exported.GeneralMessage) error {
 						return fmt.Errorf("set msg error")
 					}
@@ -163,48 +166,67 @@ func TestMessenger_DispatchMsg(t *testing.T) {
 			nexus.GenerateMessageIDFunc = func(_ sdk.Context) (string, []byte, uint64) {
 				return "1", []byte("1"), 1
 			}
+			nexus.GetMessageFunc = func(_ sdk.Context, _ string) (exported.GeneralMessage, bool) {
+				return exported.GeneralMessage{}, false
+			}
 			nexus.SetNewMessageFunc = func(_ sdk.Context, msg exported.GeneralMessage) error {
-				return nil
+				return msg.ValidateBasic()
 			}
 			nexus.RouteMessageFunc = func(ctx sdk.Context, id string, _ ...exported.RoutingContext) error { return nil }
 		}).
 		When("the msg is encoded correctly", func() {
 			msg = wasmvmtypes.CosmosMsg{
-				Custom: []byte("{\"source_chain\":\"sourcechain\",\"source_address\":\"0xb860\",\"destination_chain\":\"Axelarnet\",\"destination_address\":\"axelarvaloper1zh9wrak6ke4n6fclj5e8yk397czv430ygs5jz7\",\"payload_hash\":[187,155,85,102,194,244,135,104,99,51,62,72,31,70,152,53,1,84,37,159,254,98,38,226,131,177,108,225,138,100,188,241],\"source_tx_id\":[47,228],\"source_tx_index\":100}"),
+				Custom: []byte("{\"source_chain\":\"sourcechain\",\"source_address\":\"0xb860\",\"destination_chain\":\"Axelarnet\",\"destination_address\":\"axelarvaloper1zh9wrak6ke4n6fclj5e8yk397czv430ygs5jz7\",\"payload_hash\":[187,155,85,102,194,244,135,104,99,51,62,72,31,70,152,53,1,84,37,159,254,98,38,226,131,177,108,225,138,100,188,241],\"source_tx_id\":[187,155,85,102,194,244,135,104,99,51,62,72,31,70,152,53,1,84,37,159,254,98,38,226,131,177,108,225,138,100,188,241],\"source_tx_index\":100, \"id\": \"0x73657e3da2e404f474218fe2789462585d7f6060741bd312c862378cf67ca981-1\"}"),
 			}
 		}).
 		Branch(
 			When("succeed to route message", func() {
 				nexus.RouteMessageFunc = func(_ sdk.Context, id string, _ ...exported.RoutingContext) error { return nil }
 			}).
-				Then("should route the message", func(t *testing.T) {
-					_, _, err := messenger.DispatchMsg(ctx, contractAddr, "", msg)
-					assert.NoError(t, err)
-
-					assert.Len(t, nexus.SetNewMessageCalls(), 1)
-					assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.Recipient.Chain, axelarnet.Axelarnet)
-					assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.Status, exported.Approved)
-					assert.Nil(t, nexus.SetNewMessageCalls()[0].Msg.Asset)
-
-					assert.Len(t, nexus.RouteMessageCalls(), 1)
-					assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.ID, nexus.RouteMessageCalls()[0].ID)
-				}),
-
-			When("failed to route message", func() {
-				nexus.RouteMessageFunc = func(_ sdk.Context, id string, _ ...exported.RoutingContext) error { return fmt.Errorf("failed") }
-			}).
-				Then("should set message as processing", func(t *testing.T) {
-					_, _, err := messenger.DispatchMsg(ctx, contractAddr, "", msg)
-					assert.NoError(t, err)
-
-					assert.Len(t, nexus.SetNewMessageCalls(), 1)
-					assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.Recipient.Chain, axelarnet.Axelarnet)
-					assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.Status, exported.Approved)
-					assert.Nil(t, nexus.SetNewMessageCalls()[0].Msg.Asset)
-
-					assert.Len(t, nexus.RouteMessageCalls(), 1)
-					assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.ID, nexus.RouteMessageCalls()[0].ID)
-				}),
+				Branch(
+					Then("should route the message", func(t *testing.T) {
+						_, _, err := messenger.DispatchMsg(ctx, contractAddr, "", msg)
+						assert.NoError(t, err)
+
+						assert.Len(t, nexus.SetNewMessageCalls(), 1)
+						assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.Recipient.Chain, axelarnet.Axelarnet)
+						assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.Status, exported.Approved)
+						assert.Nil(t, nexus.SetNewMessageCalls()[0].Msg.Asset)
+
+						assert.Len(t, nexus.RouteMessageCalls(), 1)
+						assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.ID, nexus.RouteMessageCalls()[0].ID)
+					}),
+					When("message already set", func() {
+
+						nexus.GetMessageFunc = func(_ sdk.Context, _ string) (exported.GeneralMessage, bool) {
+							return exported.GeneralMessage{}, true
+						}
+					}).Then("should be a no op", func(t *testing.T) {
+
+						_, _, err := messenger.DispatchMsg(ctx, contractAddr, "", msg)
+						assert.NoError(t, err)
+
+						assert.Len(t, nexus.SetNewMessageCalls(), 0)
+
+						assert.Len(t, nexus.RouteMessageCalls(), 0)
+					}),
+
+					When("failed to route message", func() {
+						nexus.RouteMessageFunc = func(_ sdk.Context, id string, _ ...exported.RoutingContext) error { return fmt.Errorf("failed") }
+					}).
+						Then("should set message as processing", func(t *testing.T) {
+							_, _, err := messenger.DispatchMsg(ctx, contractAddr, "", msg)
+							assert.NoError(t, err)
+
+							assert.Len(t, nexus.SetNewMessageCalls(), 1)
+							assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.Recipient.Chain, axelarnet.Axelarnet)
+							assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.Status, exported.Approved)
+							assert.Nil(t, nexus.SetNewMessageCalls()[0].Msg.Asset)
+
+							assert.Len(t, nexus.RouteMessageCalls(), 1)
+							assert.Equal(t, nexus.SetNewMessageCalls()[0].Msg.ID, nexus.RouteMessageCalls()[0].ID)
+						}),
+				),
 		).
 		Run(t)
 }

x/nexus/keeper/wasm_message_route.go

@@ -26,7 +26,12 @@ func NewMessageRoute(nexus types.Nexus, account types.AccountKeeper, wasm types.
 			return fmt.Errorf("gateway is not set")
 		}
 
-		bz, err := json.Marshal(request{RouteMessagesFromNexus: []exported.WasmMessage{exported.FromGeneralMessage(msg)}})
+		wasmMsg := exported.FromGeneralMessage(msg)
+		if err := wasmMsg.ValidateBasic(); err != nil {
+			return err
+		}
+
+		bz, err := json.Marshal(request{RouteMessagesFromNexus: []exported.WasmMessage{wasmMsg}})
 		if err != nil {
 			return nil
 		}

x/nexus/types/expected_keepers.go

@@ -43,6 +43,7 @@ type Nexus interface {
 	RateLimitTransfer(ctx sdk.Context, chain exported.ChainName, asset sdk.Coin, direction exported.TransferDirection) error
 	GenerateMessageID(ctx sdk.Context) (string, []byte, uint64)
 	SetNewMessage(ctx sdk.Context, msg exported.GeneralMessage) error
+	GetMessage(ctx sdk.Context, id string) (exported.GeneralMessage, bool)
 	SetMessageExecuted(ctx sdk.Context, id string) error
 	RouteMessage(ctx sdk.Context, id string, routingCtx ...exported.RoutingContext) error
 	DequeueRouteMessage(ctx sdk.Context) (exported.GeneralMessage, bool)