fix: backport(v29) check if cctx has already been created before casting inbound vote (#3664)

* check if cctx exists before casting inbound vote

* update comments

* return cctxindex instead of empty string

* added a status check

* status check for ballot check

* fix unit tests

* add changelog

* update changelog

Author: kingpinXD

changelog.md

@@ -1,5 +1,11 @@
 # CHANGELOG
 
+## v29.0.1
+
+### Fixes
+
+* [3664](https://github.com/zeta-chain/node/pull/3664/) - add a check to stop posting inbound votes if the cctx has already been created
+
 ## v29.0.0
 
 ### Breaking Changes

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"
@@ -401,6 +403,25 @@ func (ob *Observer) PostVoteInbound(
 		logs.FieldConfirmationMode: msg.ConfirmationMode.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

@@ -1,12 +1,14 @@
 package base_test
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"os"
 	"strings"
 	"testing"
 
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"github.com/stretchr/testify/require"
@@ -21,6 +23,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 (
@@ -518,9 +522,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(errors.New("not found"))
+
 		ballot, err := ob.PostVoteInbound(context.TODO(), &msg, 100000)
 		require.NoError(t, err)
 		require.Equal(t, "sampleBallotIndex", ballot)
@@ -533,12 +538,64 @@ 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(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)
+
+		ob.zetacore.WithPostVoteInbound("", "sampleBallotIndex")
+		msg := sample.InboundVote(coin.CoinType_Gas, chains.Ethereum.ChainId, chains.ZetaChainMainnet.ChainId)
+
+		ob.zetacore.MockGetCctxByHash(nil)
+		ob.zetacore.MockGetBallotByID(msg.Digest(), status.Error(codes.NotFound, "not found ballot"))
+
+		var logBuffer bytes.Buffer
+		consoleWriter := zerolog.ConsoleWriter{Out: &logBuffer}
+		logger := zerolog.New(consoleWriter)
+		ob.Observer.Logger().Inbound = logger
+
+		// Act
+		ballot, err := ob.PostVoteInbound(context.TODO(), &msg, 100000)
+		// Assert
+		require.NoError(t, err)
+		require.Equal(t, ballot, msg.Digest())
+
+		logOutput := logBuffer.String()
+		require.Contains(t, logOutput, "inbound detected: cctx exists but the ballot does not")
+	})
+
+	t.Run("should post vote cctx already exists but ballot is found", func(t *testing.T) {
+		//Arrange
+		// create observer
+		ob := newTestSuite(t, chains.Ethereum)
+
+		msg := sample.InboundVote(coin.CoinType_Gas, chains.Ethereum.ChainId, chains.ZetaChainMainnet.ChainId)
+		ob.zetacore.WithPostVoteInbound(sample.ZetaIndex(t), msg.Digest())
+		ob.zetacore.MockGetCctxByHash(nil)
+		ob.zetacore.MockGetBallotByID(msg.Digest(), nil)
+
+		var logBuffer bytes.Buffer
+		consoleWriter := zerolog.ConsoleWriter{Out: &logBuffer}
+		logger := zerolog.New(consoleWriter)
+		ob.Observer.Logger().Inbound = logger
+
+		// Act
+		ballot, err := ob.PostVoteInbound(context.TODO(), &msg, 100000)
+		// Assert
+		require.NoError(t, err)
+		require.Equal(t, ballot, msg.Digest())
+
+		logOutput := logBuffer.String()
+		require.Contains(t, logOutput, "inbound detected: vote posted")
+	})
 }
 
 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

@@ -91,6 +91,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,
@@ -104,6 +105,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/sui/observer/observer_test.go

@@ -8,6 +8,7 @@ import (
 
 	"cosmossdk.io/math"
 	"github.com/block-vision/sui-go-sdk/models"
+	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -112,6 +113,7 @@ func TestObserver(t *testing.T) {
 
 		// Given inbound votes catches so we can assert them later
 		ts.CatchInboundVotes()
+		ts.zetaMock.MockGetCctxByHash(errors.New("not found"))
 
 		// ACT
 		err := ts.ObserveInbound(ts.ctx)
@@ -187,6 +189,7 @@ func TestObserver(t *testing.T) {
 
 		// Given votes catcher
 		ts.CatchInboundVotes()
+		ts.zetaMock.MockGetCctxByHash(errors.New("not found"))
 
 		// ACT
 		err := ts.ProcessInboundTrackers(ts.ctx)

zetaclient/chains/ton/observer/inbound_test.go

@@ -141,6 +141,7 @@ func TestInbound(t *testing.T) {
 			Once()
 
 		ts.MockGetBlockHeader(depositTX.BlockID)
+		ts.MockGetCctxByHash()
 
 		// ACT
 		// Observe inbounds once
@@ -205,6 +206,7 @@ func TestInbound(t *testing.T) {
 			Once()
 
 		ts.MockGetBlockHeader(depositAndCallTX.BlockID)
+		ts.MockGetCctxByHash()
 
 		// ACT
 		// Observe inbounds once
@@ -347,6 +349,7 @@ func TestInbound(t *testing.T) {
 		for _, tx := range txs {
 			ts.MockGetBlockHeader(tx.BlockID)
 		}
+		ts.MockGetCctxByHash()
 
 		// ACT
 		// Observe inbounds once
@@ -414,6 +417,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.go

@@ -49,7 +49,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(err error) *ZetacoreClient {
+	_m.On("GetCctxByHash", mock.Anything, mock.Anything).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
 }