fix: avoid endless tx rescan when the inbound vote message is invalid

changelog.md

@@ -13,6 +13,7 @@
 ### Fixes
 
 * [3206](https://github.com/zeta-chain/node/pull/3206) - skip Solana unsupported transaction version to not block inbound observation
+* [3184](https://github.com/zeta-chain/node/pull/3184) - zetaclient should not retry if inbound vote message validation fails
 
 ## v23.0.0
 

testutil/sample/crosschain.go

@@ -292,7 +292,7 @@ func ZetaAccounting(t *testing.T, index string) types.ZetaAccounting {
 
 func InboundVote(coinType coin.CoinType, from, to int64) types.MsgVoteInbound {
 	return types.MsgVoteInbound{
-		Creator:            "",
+		Creator:            Bech32AccAddress().String(),
 		Sender:             EthAddress().String(),
 		SenderChainId:      Chain(from).ChainId,
 		Receiver:           EthAddress().String(),

x/crosschain/types/message_vote_inbound.go

@@ -22,7 +22,7 @@ const MaxMessageLength = 10240
 // InboundVoteOption is a function that sets some option on the inbound vote message
 type InboundVoteOption func(*MsgVoteInbound)
 
-// WithMemoRevertOptions sets the revert options for inbound vote message
+// WithRevertOptions sets the revert options for inbound vote message
 func WithRevertOptions(revertOptions RevertOptions) InboundVoteOption {
 	return func(msg *MsgVoteInbound) {
 		msg.RevertOptions = revertOptions

zetaclient/chains/base/observer.go

@@ -466,7 +466,7 @@ func (ob *Observer) ReadLastTxScannedFromDB() (string, error) {
 	return lastTx.Hash, nil
 }
 
-// PostVoteInbound posts a vote for the given vote message
+// PostVoteInbound posts a vote for the given vote message and returns the ballot.
 func (ob *Observer) PostVoteInbound(
 	ctx context.Context,
 	msg *crosschaintypes.MsgVoteInbound,
@@ -477,19 +477,26 @@ func (ob *Observer) PostVoteInbound(
 	var (
 		txHash   = msg.InboundHash
 		coinType = msg.CoinType
-		chainID  = ob.Chain().ChainId
 	)
 
-	zetaHash, ballot, err := ob.ZetacoreClient().PostVoteInbound(ctx, gasLimit, retryGasLimit, msg)
-
+	// prepare logger fields
 	lf := map[string]any{
-		"inbound.chain_id":         chainID,
-		"inbound.coin_type":        coinType.String(),
-		"inbound.external_tx_hash": txHash,
-		"inbound.ballot_index":     ballot,
-		"inbound.zeta_tx_hash":     zetaHash,
+		logs.FieldMethod:   "PostVoteInbound",
+		logs.FieldTx:       txHash,
+		logs.FieldCoinType: coinType.String(),
+	}
+
+	// make sure the message is valid to avoid unnecessary retries
+	if err := msg.ValidateBasic(); err != nil {
+		ob.logger.Inbound.Warn().Err(err).Fields(lf).Msg("invalid inbound vote message")
+		return "", nil
 	}
 
+	// post vote to zetacore
+	zetaHash, ballot, err := ob.ZetacoreClient().PostVoteInbound(ctx, gasLimit, retryGasLimit, msg)
+	lf[logs.FieldZetaTx] = zetaHash
+	lf[logs.FieldBallot] = ballot
+
 	switch {
 	case err != nil:
 		ob.logger.Inbound.Error().Err(err).Fields(lf).Msg("inbound detected: error posting vote")

zetaclient/chains/base/observer_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/zeta-chain/node/pkg/chains"
 	"github.com/zeta-chain/node/pkg/coin"
 	"github.com/zeta-chain/node/testutil/sample"
+	crosschaintypes "github.com/zeta-chain/node/x/crosschain/types"
 	observertypes "github.com/zeta-chain/node/x/observer/types"
 	"github.com/zeta-chain/node/zetaclient/chains/base"
 	"github.com/zeta-chain/node/zetaclient/chains/interfaces"
@@ -633,6 +634,24 @@ func TestPostVoteInbound(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, "sampleBallotIndex", ballot)
 	})
+
+	t.Run("should not post vote if message basic validation fails", func(t *testing.T) {
+		// create observer
+		ob := createObserver(t, chains.Ethereum, defaultAlertLatency)
+
+		// create mock zetacore client
+		zetacoreClient := mocks.NewZetacoreClient(t)
+		ob = ob.WithZetacoreClient(zetacoreClient)
+
+		// create sample message with long Message
+		msg := sample.InboundVote(coin.CoinType_Gas, chains.Ethereum.ChainId, chains.ZetaChainMainnet.ChainId)
+		msg.Message = strings.Repeat("1", crosschaintypes.MaxMessageLength+1)
+
+		// post vote inbound
+		ballot, err := ob.PostVoteInbound(context.TODO(), &msg, 100000)
+		require.NoError(t, err)
+		require.Empty(t, ballot)
+	})
 }
 
 func TestAlertOnRPCLatency(t *testing.T) {

zetaclient/chains/bitcoin/observer/event.go

@@ -19,20 +19,7 @@
 	"github.com/zeta-chain/node/zetaclient/compliance"
 	"github.com/zeta-chain/node/zetaclient/config"
 	"github.com/zeta-chain/node/zetaclient/logs"
-)
-
-// InboundProcessability is an enum representing the processability of an inbound
-type InboundProcessability int
-
-const (
-	// InboundProcessabilityGood represents a processable inbound
-	InboundProcessabilityGood InboundProcessability = iota
-
-	// InboundProcessabilityDonation represents a donation inbound
-	InboundProcessabilityDonation
-
-	// InboundProcessabilityComplianceViolation represents a compliance violation
-	InboundProcessabilityComplianceViolation
+	clienttypes "github.com/zeta-chain/node/zetaclient/types"
 )
 
 // BTCInboundEvent represents an incoming transaction event
@@ -62,11 +49,11 @@
 	TxHash string
 }
 
-// Processability returns the processability of the inbound event
-func (event *BTCInboundEvent) Processability() InboundProcessability {
+// Category returns the category of the inbound event
+func (event *BTCInboundEvent) Category() clienttypes.InboundCategory {
 	// compliance check on sender and receiver addresses
 	if config.ContainRestrictedAddress(event.FromAddress, event.ToAddress) {
-		return InboundProcessabilityComplianceViolation
+		return clienttypes.InboundCategoryRestricted
 	}
 
 	// compliance check on receiver, revert/abort addresses in standard memo
@@ -76,16 +63,16 @@
 			event.MemoStd.RevertOptions.RevertAddress,
 			event.MemoStd.RevertOptions.AbortAddress,
 		) {
-			return InboundProcessabilityComplianceViolation
+			return clienttypes.InboundCategoryRestricted
 		}
 	}
 
 	// donation check
 	if bytes.Equal(event.MemoBytes, []byte(constant.DonationMessage)) {
-		return InboundProcessabilityDonation
+		return clienttypes.InboundCategoryDonation
 	}
 
-	return InboundProcessabilityGood
+	return clienttypes.InboundCategoryGood
 }
 
 // DecodeMemoBytes decodes the contained memo bytes as either standard or legacy memo
@@ -164,25 +151,22 @@
 	return nil
 }
 
-// CheckEventProcessability checks if the inbound event is processable
-func (ob *Observer) CheckEventProcessability(event BTCInboundEvent) bool {
-	// check if the event is processable
-	switch result := event.Processability(); result {
-	case InboundProcessabilityGood:
+// IsEventProcessable checks if the inbound event is processable
+func (ob *Observer) IsEventProcessable(event BTCInboundEvent) bool {
+	logFields := map[string]any{logs.FieldTx: event.TxHash}
+
+	switch category := event.Category(); category {
+	case clienttypes.InboundCategoryGood:
 		return true
-	case InboundProcessabilityDonation:
-		logFields := map[string]any{
-			logs.FieldChain: ob.Chain().ChainId,
-			logs.FieldTx:    event.TxHash,
-		}
+	case clienttypes.InboundCategoryDonation:
 		ob.Logger().Inbound.Info().Fields(logFields).Msgf("thank you rich folk for your donation!")
 		return false
-	case InboundProcessabilityComplianceViolation:
+	case clienttypes.InboundCategoryRestricted:
 		compliance.PrintComplianceLog(ob.logger.Inbound, ob.logger.Compliance,
 			false, ob.Chain().ChainId, event.TxHash, event.FromAddress, event.ToAddress, "BTC")
 		return false
 	default:
-		ob.Logger().Inbound.Error().Msgf("unreachable code got InboundProcessability: %v", result)
+		ob.Logger().Inbound.Error().Fields(logFields).Msgf("unreachable code got InboundCategory: %v", category)
 		return false
 	}
 }

zetaclient/chains/bitcoin/observer/event_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/zeta-chain/node/zetaclient/keys"
 	"github.com/zeta-chain/node/zetaclient/testutils"
 	"github.com/zeta-chain/node/zetaclient/testutils/mocks"
+	clienttypes "github.com/zeta-chain/node/zetaclient/types"
 )
 
 // createTestBtcEvent creates a test BTC inbound event
@@ -41,7 +42,7 @@ func createTestBtcEvent(
 	}
 }
 
-func Test_CheckProcessability(t *testing.T) {
+func Test_Category(t *testing.T) {
 	// setup compliance config
 	cfg := config.Config{
 		ComplianceConfig: sample.ComplianceConfig(),
@@ -52,26 +53,26 @@ func Test_CheckProcessability(t *testing.T) {
 	tests := []struct {
 		name     string
 		event    *observer.BTCInboundEvent
-		expected observer.InboundProcessability
+		expected clienttypes.InboundCategory
 	}{
 		{
-			name: "should return InboundProcessabilityGood for a processable inbound event",
+			name: "should return InboundCategoryGood for a processable inbound event",
 			event: &observer.BTCInboundEvent{
 				FromAddress: "tb1quhassyrlj43qar0mn0k5sufyp6mazmh2q85lr6ex8ehqfhxpzsksllwrsu",
 				ToAddress:   testutils.TSSAddressBTCAthens3,
 			},
-			expected: observer.InboundProcessabilityGood,
+			expected: clienttypes.InboundCategoryGood,
 		},
 		{
-			name: "should return InboundProcessabilityComplianceViolation for a restricted sender address",
+			name: "should return InboundCategoryRestricted for a restricted sender address",
 			event: &observer.BTCInboundEvent{
 				FromAddress: sample.RestrictedBtcAddressTest,
 				ToAddress:   testutils.TSSAddressBTCAthens3,
 			},
-			expected: observer.InboundProcessabilityComplianceViolation,
+			expected: clienttypes.InboundCategoryRestricted,
 		},
 		{
-			name: "should return InboundProcessabilityComplianceViolation for a restricted receiver address in standard memo",
+			name: "should return InboundCategoryRestricted for a restricted receiver address in standard memo",
 			event: &observer.BTCInboundEvent{
 				FromAddress: "tb1quhassyrlj43qar0mn0k5sufyp6mazmh2q85lr6ex8ehqfhxpzsksllwrsu",
 				ToAddress:   testutils.TSSAddressBTCAthens3,
@@ -81,10 +82,10 @@ func Test_CheckProcessability(t *testing.T) {
 					},
 				},
 			},
-			expected: observer.InboundProcessabilityComplianceViolation,
+			expected: clienttypes.InboundCategoryRestricted,
 		},
 		{
-			name: "should return InboundProcessabilityComplianceViolation for a restricted revert address in standard memo",
+			name: "should return InboundCategoryRestricted for a restricted revert address in standard memo",
 			event: &observer.BTCInboundEvent{
 				FromAddress: "tb1quhassyrlj43qar0mn0k5sufyp6mazmh2q85lr6ex8ehqfhxpzsksllwrsu",
 				ToAddress:   testutils.TSSAddressBTCAthens3,
@@ -96,22 +97,22 @@ func Test_CheckProcessability(t *testing.T) {
 					},
 				},
 			},
-			expected: observer.InboundProcessabilityComplianceViolation,
+			expected: clienttypes.InboundCategoryRestricted,
 		},
 		{
-			name: "should return InboundProcessabilityDonation for a donation inbound event",
+			name: "should return InboundCategoryDonation for a donation inbound event",
 			event: &observer.BTCInboundEvent{
 				FromAddress: "tb1quhassyrlj43qar0mn0k5sufyp6mazmh2q85lr6ex8ehqfhxpzsksllwrsu",
 				ToAddress:   testutils.TSSAddressBTCAthens3,
 				MemoBytes:   []byte(constant.DonationMessage),
 			},
-			expected: observer.InboundProcessabilityDonation,
+			expected: clienttypes.InboundCategoryDonation,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result := tt.event.Processability()
+			result := tt.event.Category()
 			require.Equal(t, tt.expected, result)
 		})
 	}
@@ -301,7 +302,7 @@ func Test_ValidateStandardMemo(t *testing.T) {
 	}
 }
 
-func Test_CheckEventProcessability(t *testing.T) {
+func Test_IsEventProcessable(t *testing.T) {
 	// can use any bitcoin chain for testing
 	chain := chains.BitcoinMainnet
 	params := mocks.MockChainParams(chain.ChainId, 10)
@@ -344,7 +345,7 @@ func Test_CheckEventProcessability(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result := ob.CheckEventProcessability(tt.event)
+			result := ob.IsEventProcessable(tt.event)
 			require.Equal(t, tt.result, result)
 		})
 	}

zetaclient/chains/bitcoin/observer/inbound.go

@@ -282,21 +282,7 @@
 		return msg.Digest(), nil
 	}
 
-	zetaHash, ballot, err := ob.ZetacoreClient().PostVoteInbound(
-		ctx,
-		zetacore.PostVoteInboundGasLimit,
-		zetacore.PostVoteInboundExecutionGasLimit,
-		msg,
-	)
-	if err != nil {
-		ob.logger.Inbound.Error().Err(err).Msg("error posting to zetacore")
-		return "", err
-	} else if zetaHash != "" {
-		ob.logger.Inbound.Info().Msgf("BTC deposit detected and reported: PostVoteInbound zeta tx hash: %s inbound %s ballot %s fee %v",
-			zetaHash, txHash, ballot, event.DepositorFee)
-	}
-
-	return msg.Digest(), nil
+	return ob.PostVoteInbound(ctx, msg, zetacore.PostVoteInboundExecutionGasLimit)
 }
 
 // FilterAndParseIncomingTx given txs list returned by the "getblock 2" RPC command, return the txs that are relevant to us
@@ -332,12 +318,15 @@
 }
 
 // GetInboundVoteFromBtcEvent converts a BTCInboundEvent to a MsgVoteInbound to enable voting on the inbound on zetacore
+//
+// Returns:
+//   - a valid MsgVoteInbound message, or
+//   - nil if no valid message can be created for whatever reasons:
+//     invalid data, not processable, invalid amount, etc.
 func (ob *Observer) GetInboundVoteFromBtcEvent(event *BTCInboundEvent) *crosschaintypes.MsgVoteInbound {
 	// prepare logger fields
 	lf := map[string]any{
-		logs.FieldModule: logs.ModNameInbound,
 		logs.FieldMethod: "GetInboundVoteFromBtcEvent",
-		logs.FieldChain:  ob.Chain().ChainId,
 		logs.FieldTx:     event.TxHash,
 	}
 
@@ -349,7 +338,7 @@
 	}
 
 	// check if the event is processable
-	if !ob.CheckEventProcessability(*event) {
+	if !ob.IsEventProcessable(*event) {
 		return nil
 	}