fix: check if cctx has already been created before casting inbound vo…

…te (#3636)

changelog.md

@@ -1,5 +1,9 @@
 # CHANGELOG
 
+## Unreleased
+
+* [3636](https://github.com/zeta-chain/node/pull/3636) - add a check to stop posting inbound votes if the cctx has already been created
+
 ## v28.1.0
 
 This is a zetaclient only release.

zetaclient/chains/base/observer.go

@@ -12,6 +12,8 @@ import (
 	lru "github.com/hashicorp/golang-lru"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/zeta-chain/node/pkg/chains"
 	crosschaintypes "github.com/zeta-chain/node/x/crosschain/types"
@@ -406,6 +408,25 @@ func (ob *Observer) PostVoteInbound(
 		logs.FieldCoinType: coinType.String(),
 	}
 
+	cctxIndex := msg.Digest()
+
+	// The cctx is created after the inbound ballot is finalized
+	// 1. if the cctx already exists, we could try voting if the ballot is present
+	// 2. if the cctx exists but the ballot does not exist, we do not need to vote
+	_, err := ob.ZetacoreClient().GetCctxByHash(ctx, cctxIndex)
+	if err == nil {
+		// The cctx exists we should still vote if the ballot is present
+		_, ballotErr := ob.ZetacoreClient().GetBallotByID(ctx, cctxIndex)
+		if ballotErr != nil {
+			// Verify ballot is not found
+			if st, ok := status.FromError(ballotErr); ok && st.Code() == codes.NotFound {
+				// Query for ballot failed, the ballot does not exist we can return
+				ob.logger.Inbound.Info().Fields(lf).Msg("inbound detected: cctx exists but the ballot does not")
+				return cctxIndex, nil
+			}
+		}
+	}
+
 	// 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")

zetaclient/chains/base/observer_test.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"github.com/stretchr/testify/require"
@@ -21,6 +22,8 @@ import (
 	zctx "github.com/zeta-chain/node/zetaclient/context"
 	"github.com/zeta-chain/node/zetaclient/db"
 	"github.com/zeta-chain/node/zetaclient/testutils/mocks"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 const (
@@ -538,9 +541,10 @@ func TestPostVoteInbound(t *testing.T) {
 		ob := newTestSuite(t, chains.Ethereum)
 
 		ob.zetacore.WithPostVoteInbound("", "sampleBallotIndex")
-
 		// post vote inbound
 		msg := sample.InboundVote(coin.CoinType_Gas, chains.Ethereum.ChainId, chains.ZetaChainMainnet.ChainId)
+		ob.zetacore.MockGetCctxByHash(msg.Digest(), errors.New("not found"))
+
 		ballot, err := ob.PostVoteInbound(context.TODO(), &msg, 100000)
 		require.NoError(t, err)
 		require.Equal(t, "sampleBallotIndex", ballot)
@@ -553,12 +557,29 @@ func TestPostVoteInbound(t *testing.T) {
 		// 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)
+		ob.zetacore.MockGetCctxByHash(msg.Digest(), errors.New("not found"))
 
 		// post vote inbound
 		ballot, err := ob.PostVoteInbound(context.TODO(), &msg, 100000)
 		require.NoError(t, err)
 		require.Empty(t, ballot)
 	})
+
+	t.Run("should not post vote cctx already exists and ballot is not found", func(t *testing.T) {
+		//Arrange
+		// create observer
+		ob := newTestSuite(t, chains.Ethereum)
+		// 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)
+		ob.zetacore.MockGetCctxByHash(msg.Digest(), nil)
+		ob.zetacore.MockGetBallotByID(msg.Digest(), status.Error(codes.NotFound, "not found ballot"))
+		// Act
+		ballot, err := ob.PostVoteInbound(context.TODO(), &msg, 100000)
+		// Assert
+		require.NoError(t, err)
+		require.Equal(t, ballot, msg.Digest())
+	})
 }
 
 func createDatabase(t *testing.T) *db.DB {

zetaclient/chains/evm/observer/inbound_test.go

@@ -446,9 +446,10 @@ func Test_ObserveTSSReceiveInBlock(t *testing.T) {
 
 	// test cases
 	tests := []struct {
-		name          string
-		mockEVMClient func(m *mocks.EVMRPCClient)
-		errMsg        string
+		name               string
+		mockEVMClient      func(m *mocks.EVMRPCClient)
+		mockZetacoreClient func(m *mocks.ZetacoreClient)
+		errMsg             string
 	}{
 		{
 			name: "should observe TSS receive in block",
@@ -458,6 +459,9 @@ func Test_ObserveTSSReceiveInBlock(t *testing.T) {
 				m.On("TransactionReceipt", mock.Anything, mock.Anything).Return(receipt, nil)
 				m.On("BlockByNumberCustom", mock.Anything, mock.Anything).Return(block, nil)
 			},
+			mockZetacoreClient: func(m *mocks.ZetacoreClient) {
+				m.On("GetCctxByHash", mock.Anything, mock.Anything).Return(nil, errors.New("not found"))
+			},
 			errMsg: "",
 		},
 		{
@@ -469,7 +473,8 @@ func Test_ObserveTSSReceiveInBlock(t *testing.T) {
 				m.On("BlockNumber", mock.Anything).Return(uint64(0), errors.New("RPC error"))
 				m.On("BlockByNumberCustom", mock.Anything, mock.Anything).Return(nil, errors.New("RPC error"))
 			},
-			errMsg: "error getting block",
+			mockZetacoreClient: nil,
+			errMsg:             "error getting block",
 		},
 		{
 			name: "should not observe on error getting receipt",
@@ -479,7 +484,8 @@ func Test_ObserveTSSReceiveInBlock(t *testing.T) {
 				m.On("TransactionReceipt", mock.Anything, mock.Anything).Return(nil, errors.New("RPC error"))
 				m.On("BlockByNumberCustom", mock.Anything, mock.Anything).Return(block, nil)
 			},
-			errMsg: "error getting receipt",
+			mockZetacoreClient: nil,
+			errMsg:             "error getting receipt",
 		},
 	}
 
@@ -491,6 +497,9 @@ func Test_ObserveTSSReceiveInBlock(t *testing.T) {
 			if tt.mockEVMClient != nil {
 				tt.mockEVMClient(ob.evmMock)
 			}
+			if tt.mockZetacoreClient != nil {
+				tt.mockZetacoreClient(ob.zetacore)
+			}
 
 			err := ob.ObserveTSSReceiveInBlock(ob.ctx, blockNumber)
 			if tt.errMsg != "" {

zetaclient/chains/interfaces/interfaces.go

@@ -85,6 +85,7 @@ type ZetacoreClient interface {
 	GetPendingNoncesByChain(ctx context.Context, chainID int64) (observertypes.PendingNonces, error)
 
 	GetCctxByNonce(ctx context.Context, chainID int64, nonce uint64) (*crosschaintypes.CrossChainTx, error)
+	GetCctxByHash(ctx context.Context, sendHash string) (*crosschaintypes.CrossChainTx, error)
 	GetOutboundTracker(ctx context.Context, chain chains.Chain, nonce uint64) (*crosschaintypes.OutboundTracker, error)
 	GetAllOutboundTrackerByChain(
 		ctx context.Context,
@@ -98,6 +99,7 @@ type ZetacoreClient interface {
 	GetBTCTSSAddress(ctx context.Context, chainID int64) (string, error)
 	GetZetaHotKeyBalance(ctx context.Context) (sdkmath.Int, error)
 	GetInboundTrackersForChain(ctx context.Context, chainID int64) ([]crosschaintypes.InboundTracker, error)
+	GetBallotByID(ctx context.Context, id string) (*observertypes.QueryBallotByIdentifierResponse, error)
 
 	GetUpgradePlan(ctx context.Context) (*upgradetypes.Plan, error)
 

zetaclient/chains/ton/observer/inbound_test.go

@@ -142,6 +142,7 @@ func TestInbound(t *testing.T) {
 			Once()
 
 		ts.MockGetBlockHeader(depositTX.BlockID)
+		ts.MockGetCctxByHash()
 
 		// ACT
 		// Observe inbounds once
@@ -204,6 +205,7 @@ func TestInbound(t *testing.T) {
 			Once()
 
 		ts.MockGetBlockHeader(depositAndCallTX.BlockID)
+		ts.MockGetCctxByHash()
 
 		// ACT
 		// Observe inbounds once
@@ -350,6 +352,7 @@ func TestInbound(t *testing.T) {
 		for _, tx := range txs {
 			ts.MockGetBlockHeader(tx.BlockID)
 		}
+		ts.MockGetCctxByHash()
 
 		// ACT
 		// Observe inbounds once
@@ -417,6 +420,7 @@ func TestInboundTracker(t *testing.T) {
 	})
 	ts.MockGetTransaction(ts.gateway.AccountID(), txWithdrawal)
 	ts.MockGetBlockHeader(txWithdrawal.BlockID)
+	ts.MockGetCctxByHash()
 
 	// Given inbound trackers from zetacore
 	trackers := []cc.InboundTracker{

zetaclient/chains/ton/observer/observer_test.go

@@ -7,6 +7,7 @@ import (
 
 	"cosmossdk.io/math"
 	eth "github.com/ethereum/go-ethereum/common"
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -175,6 +176,11 @@ func (ts *testSuite) MockGetBlockHeader(id ton.BlockIDExt) *mock.Call {
 		Return(blockInfo, nil)
 }
 
+func (ts *testSuite) MockGetCctxByHash() *mock.Call {
+	return ts.zetacore.
+		On("GetCctxByHash", mock.Anything, mock.Anything).Return(nil, errors.New("not found"))
+}
+
 func (ts *testSuite) OnGetInboundTrackersForChain(trackers []cc.InboundTracker) *mock.Call {
 	return ts.zetacore.
 		On("GetInboundTrackersForChain", mock.Anything, ts.chain.ChainId).

zetaclient/testutils/mocks/zetacore_client_opts.go

@@ -38,7 +38,16 @@ func (_m *ZetacoreClient) WithPostVoteInbound(zetaTxHash string, ballotIndex str
 	_m.On("PostVoteInbound", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Maybe().
 		Return(zetaTxHash, ballotIndex, nil)
+	return _m
+}
+
+func (_m *ZetacoreClient) MockGetCctxByHash(cctxIndex string, err error) *ZetacoreClient {
+	_m.On("GetCctxByHash", mock.Anything, cctxIndex).Return(nil, err)
+	return _m
+}
 
+func (_m *ZetacoreClient) MockGetBallotByID(ballotIndex string, err error) *ZetacoreClient {
+	_m.On("GetBallotByID", mock.Anything, ballotIndex).Return(nil, err)
 	return _m
 }