Repository: celestiaorg/celestia-app

Issue:
Transactions might accidentally get resigned during evictions
## Description

Currently, the consensus node we're talking to might evict a transaction from its local mempool(which does not mean global eviction) and our way of working around it is resubmitting transaction bytes with the same signature. https://github.com/celestiaorg/celestia-app/blob/bd124cef269e98e80a9e93cd45160c82fe4a510e/pkg/user/tx_client.go#L639

When I was pressure testing TxClient I noticed a strange pattern where an evicted transaction might get sequence mismatch during eviction. This is bad for two reasons:
- Just because this particular node evicted transaction does not mean it won't get proposed by another node.
- Resigning evicted transaction with new nonce will give it a new hash.

This means that we're breaking replay protection and these blobs will most likely get included twice.

## Proposed solution

Keep polling for a period of one minute(could be a different number) and see if it eventually gets included, otherwise return evicted to the user.

Discussion:
@evan-forbes: wait, don't we need to wait one block for the state to reset in the mempool ? or do we already do that?@ninabarbakadze: During evictions? If a transaction was broadcasted and then got evicted it means mempool state was already updated. @evan-forbes: > If a transaction was broadcasted and then got evicted it means mempool state was already updated.

what if the eviction is not due to recheck, but size constraints. afaiu the cosmos-sdk doesn't have a way to revert the state it must wait until the next height.

> This means that we're breaking replay protection and these blobs will most likely get included twice.

if we sign over two different sequences for the same blob then yeah that makes sense. now that we are reverting that in #5783, just to clarify you are saying there are no cases where we want to sign over a new sequence?

can we test to ensure that the txclient doesn't halt after seeing a mismatch? or do we know why are folks still having this issue then?@ninabarbakadze: > what if the eviction is not due to recheck, but size constraints. afaiu the cosmos-sdk doesn't have a way to revert the state it must wait until the next height.

So you think that sequence mismatch is happening because mempool state is not getting updated by the time we try to resubmit?

> if we sign over two different sequences for the same blob then yeah that makes sense. now that we are reverting that in https://github.com/celestiaorg/celestia-app/pull/5783, just to clarify you are saying there are no cases where we want to sign over a new sequence?

We can sign over transaction in two scenarios:
- Tx is just getting submitted has not been added to node's mempool and we're getting sequence mismatch broadcast errors. We can safely resign here.
- Evicted tx that had a TTL that expired. Now we know that this tx will be shown as rejected in ALL node's mempools.

So currently we still maintain functionality to resign during tx submission if we get sequence mismatch.

> can we test to ensure that the txclient doesn't halt after seeing a mismatch? or do we know why are folks still having this issue then?

We do have existing tests for this and tx client should be able to recover. Which is why I'm confused about it still halting. It completely trusts the node and resigns with expected sequence https://github.com/celestiaorg/celestia-app/blob/3faf0bfdd85e272ff70a281c83b2261f929579f8/pkg/user/tx_client_test.go#L200




Solution patch:
diff --git a/pkg/user/tx_client.go b/pkg/user/tx_client.go
index c8fd658c..236b71c6 100644
--- a/pkg/user/tx_client.go
+++ b/pkg/user/tx_client.go
@@ -40,6 +40,7 @@ import (
 const (
 	DefaultPollTime          = 3 * time.Second
 	txTrackerPruningInterval = 10 * time.Minute
+	evictionPollTime         = 1 * time.Minute
 )
 
 type Option func(client *TxClient)
@@ -407,8 +408,10 @@ func (client *TxClient) BroadcastTx(ctx context.Context, msgs []sdktypes.Msg, op
 	return client.broadcastTxAndIncrementSequence(ctx, client.conns[0], txBytes, account)
 }
 
+// broadcastTxAndIncrementSequence submits a transaction to the chain and increments the sequence of the signer.
+// It will retry the transaction if it encounters a sequence mismatch.
 func (client *TxClient) broadcastTxAndIncrementSequence(ctx context.Context, conn *grpc.ClientConn, txBytes []byte, signer string) (*sdktypes.TxResponse, error) {
-	resp, err := client.broadcastTx(ctx, conn, txBytes, signer)
+	resp, err := client.broadcastTxWithRetry(ctx, conn, txBytes, signer)
 	if err != nil {
 		return nil, err
 	}
@@ -426,9 +429,8 @@ func (client *TxClient) broadcastTxAndIncrementSequence(ctx context.Context, con
 	return resp, nil
 }
 
-// broadcastTx resubmits a transaction that was evicted from the mempool.
-// Unlike the initial broadcast, it doesn't increment the signer's sequence number.
-func (client *TxClient) broadcastTx(ctx context.Context, conn *grpc.ClientConn, txBytes []byte, signer string) (*sdktypes.TxResponse, error) {
+// broadcastTx broadcasts a transaction to the chain and returns the response.
+func (client *TxClient) broadcastTx(ctx context.Context, conn *grpc.ClientConn, txBytes []byte) (*sdktypes.TxResponse, error) {
 	txClient := sdktx.NewServiceClient(conn)
 	resp, err := txClient.BroadcastTx(
 		ctx,
@@ -441,17 +443,6 @@ func (client *TxClient) broadcastTx(ctx context.Context, conn *grpc.ClientConn,
 		return nil, err
 	}
 	if resp.TxResponse.Code != abci.CodeTypeOK {
-		if apperrors.IsNonceMismatchCode(resp.TxResponse.Code) {
-			expectedSequence, err := apperrors.ParseExpectedSequence(resp.TxResponse.RawLog)
-			if err != nil {
-				return nil, fmt.Errorf("error parsing sequence mismatch: %w. RawLog: %s", err, resp.TxResponse.RawLog)
-			}
-			if err := client.signer.SetSequence(signer, expectedSequence); err != nil {
-				return nil, fmt.Errorf("setting sequence: %w", err)
-			}
-
-			return client.retryBroadcastingTx(ctx, txBytes)
-		}
 		broadcastTxErr := &BroadcastTxError{
 			TxHash:   resp.TxResponse.TxHash,
 			Code:     resp.TxResponse.Code,
@@ -463,6 +454,27 @@ func (client *TxClient) broadcastTx(ctx context.Context, conn *grpc.ClientConn,
 	return resp.TxResponse, nil
 }
 
+// broadcastTxWithRetry broadcasts a transaction and retries if it encounters a sequence mismatch.
+func (client *TxClient) broadcastTxWithRetry(ctx context.Context, conn *grpc.ClientConn, txBytes []byte, signer string) (*sdktypes.TxResponse, error) {
+	resp, err := client.broadcastTx(ctx, conn, txBytes)
+	if err != nil {
+		broadcastTxErr, ok := err.(*BroadcastTxError)
+		if !ok || !apperrors.IsNonceMismatchCode(broadcastTxErr.Code) {
+			return nil, err
+		}
+		expectedSequence, err := apperrors.ParseExpectedSequence(broadcastTxErr.ErrorLog)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing sequence mismatch: %w. ErrorLog: %s", err, broadcastTxErr.ErrorLog)
+		}
+		if err := client.signer.SetSequence(signer, expectedSequence); err != nil {
+			return nil, fmt.Errorf("setting sequence: %w", err)
+		}
+		return client.retryBroadcastingTx(ctx, txBytes)
+	}
+
+	return resp, nil
+}
+
 // retryBroadcastingTx creates a new transaction by copying over an existing transaction but creates a new signature with the
 // new sequence number. It then calls `broadcastTx` and attempts to submit the transaction.
 func (s *TxClient) retryBroadcastingTx(ctx context.Context, txBytes []byte) (*sdktypes.TxResponse, error) {
@@ -518,7 +530,7 @@ func (s *TxClient) retryBroadcastingTx(ctx context.Context, txBytes []byte) (*sd
 		}
 	}
 
-	broadcastTxResp, err := s.broadcastTx(ctx, s.conns[0], newTxBytes, signer)
+	broadcastTxResp, err := s.broadcastTxWithRetry(ctx, s.conns[0], newTxBytes, signer)
 	if err != nil {
 		return nil, err
 	}
@@ -594,6 +606,7 @@ func (client *TxClient) ConfirmTx(ctx context.Context, txHash string) (*TxRespon
 
 	pollTicker := time.NewTicker(client.pollTime)
 	defer pollTicker.Stop()
+	var evictionPollTimeStart *time.Time
 
 	for {
 		resp, err := txClient.TxStatus(ctx, &tx.TxStatusRequest{TxId: txHash})
@@ -601,15 +614,15 @@ func (client *TxClient) ConfirmTx(ctx context.Context, txHash string) (*TxRespon
 			return nil, err
 		}
 
+		if evictionPollTimeStart != nil {
+			if time.Since(*evictionPollTimeStart) > evictionPollTime {
+				return nil, fmt.Errorf("eviction poll timeout: transaction %s was evicted ", txHash)
+			}
+		}
+
 		switch resp.Status {
 		case core.TxStatusPending:
 			// Continue polling if the transaction is still pending
-			select {
-			case <-ctx.Done():
-				return nil, ctx.Err()
-			case <-pollTicker.C:
-				continue
-			}
 		case core.TxStatusCommitted:
 			txResponse := &TxResponse{
 				Height: resp.Height,
@@ -628,14 +641,27 @@ func (client *TxClient) ConfirmTx(ctx context.Context, txHash string) (*TxRespon
 			client.deleteFromTxTracker(txHash)
 			return txResponse, nil
 		case core.TxStatusEvicted:
-			_, signer, exists := client.GetTxFromTxTracker(txHash)
+			_, _, exists := client.GetTxFromTxTracker(txHash)
 			if !exists {
 				return nil, fmt.Errorf("tx: %s not found in txTracker; likely failed during broadcast", txHash)
 			}
-			// Resubmit straight away in the event of eviction and keep polling until tx is committed
-			_, err := client.broadcastTx(ctx, client.conns[0], client.txTracker[txHash].txBytes, signer)
+
+			if evictionPollTimeStart != nil {
+				// Eviction timer is running, no need to resubmit again
+				break
+			}
+
+			// If we're not already tracking eviction timeout, try to resubmit
+			_, err := client.broadcastTx(ctx, client.conns[0], client.txTracker[txHash].txBytes)
 			if err != nil {
-				return nil, fmt.Errorf("resubmission for evicted tx with hash %s failed: %w", txHash, err)
+				// Check if the error is a broadcast tx error
+				_, ok := err.(*BroadcastTxError)
+				if !ok {
+					return nil, err
+				}
+				// Start eviction timeout timer on any broadcast error during resubmission
+				now := time.Now()
+				evictionPollTimeStart = &now
 			}
 		case core.TxStatusRejected:
 			sequence, signer, exists := client.GetTxFromTxTracker(txHash)
@@ -656,6 +682,14 @@ func (client *TxClient) ConfirmTx(ctx context.Context, txHash string) (*TxRespon
 			}
 			return nil, fmt.Errorf("transaction with hash %s not found", txHash)
 		}
+
+		// Single ticker wait point for all continuing cases
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-pollTicker.C:
+			continue
+		}
 	}
 }
 



Test patch:
diff --git a/pkg/user/mock_tx_server_test.go b/pkg/user/mock_tx_server_test.go
new file mode 100644
index 00000000..3a8911bb
--- /dev/null
+++ b/pkg/user/mock_tx_server_test.go
@@ -0,0 +1,126 @@
+package user_test
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/celestiaorg/celestia-app/v5/app/grpc/tx"
+	"github.com/celestiaorg/celestia-app/v5/pkg/user"
+	abci "github.com/cometbft/cometbft/abci/types"
+	"github.com/cometbft/cometbft/rpc/core"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdktx "github.com/cosmos/cosmos-sdk/types/tx"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/test/bufconn"
+)
+
+// mockTxServer implements both gRPC ServiceServer and TxServer interfaces for mocking broadcast and tx status responses
+type mockTxServer struct {
+	sdktx.UnimplementedServiceServer
+	tx.UnimplementedTxServer
+	txStatusResponses   map[string][]*tx.TxStatusResponse // txHash with sequence of responses
+	txStatusCallCounts  map[string]int                    // txHash with number of TxStatus calls made
+	broadcastCallCounts map[string]int                    // txHash with number of BroadcastTx calls made
+}
+
+func (m *mockTxServer) TxStatus(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error) {
+	if responses, exists := m.txStatusResponses[req.TxId]; exists {
+		callCount := m.txStatusCallCounts[req.TxId]
+		m.txStatusCallCounts[req.TxId]++
+
+		// Return response based on call count
+		if callCount < len(responses) {
+			resp := responses[callCount]
+			return resp, nil
+		}
+		// If we've exhausted predefined responses, return the last response
+		lastResp := responses[len(responses)-1]
+		return lastResp, nil
+	}
+
+	// Default response
+	return &tx.TxStatusResponse{
+		Status: core.TxStatusPending,
+		Height: 1,
+	}, nil
+}
+
+func (m *mockTxServer) BroadcastTx(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
+	// Same hash for all broadcast calls
+	txHash := "test-tx-hash-123"
+
+	// Increment broadcast call count
+	m.broadcastCallCounts[txHash]++
+
+	// When it's called for the second time return sequence mismatch error
+	if m.broadcastCallCounts[txHash] > 1 {
+		return &sdktx.BroadcastTxResponse{
+			TxResponse: &sdk.TxResponse{
+				TxHash: txHash,
+				Code:   32, // sequence mismatch error code
+				Height: 1,
+				RawLog: "account sequence mismatch",
+			},
+		}, nil
+	}
+
+	// Return successful broadcast response
+	return &sdktx.BroadcastTxResponse{
+		TxResponse: &sdk.TxResponse{
+			TxHash: txHash,
+			Code:   abci.CodeTypeOK,
+			Height: 1,
+			Data:   "",
+			RawLog: "",
+		},
+	}, nil
+}
+
+// setupTxClientWithMockGPRCServer creates a TxClient connected to a mock gRPC server that lets you mock broadcast and tx status responses
+func setupTxClientWithMockGRPCServer(t *testing.T, responseSequences map[string][]*tx.TxStatusResponse, opts ...user.Option) (*user.TxClient, *grpc.ClientConn) {
+	// Create mock server with provided response sequences
+	mockServer := &mockTxServer{
+		txStatusResponses:   responseSequences,
+		txStatusCallCounts:  make(map[string]int),
+		broadcastCallCounts: make(map[string]int),
+	}
+
+	// Set up in-memory gRPC server
+	lis := bufconn.Listen(1024 * 1024)
+	s := grpc.NewServer()
+	sdktx.RegisterServiceServer(s, mockServer) // For BroadcastTx
+	tx.RegisterTxServer(s, mockServer)         // For TxStatus
+
+	go func() {
+		if err := s.Serve(lis); err != nil {
+			t.Logf("Server exited with error: %v", err)
+		}
+	}()
+
+	// Create client connection
+	//nolint:staticcheck
+	conn, err := grpc.DialContext(context.Background(), "bufnet",
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return lis.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	require.NoError(t, err)
+
+	// Create TxClient with mock connection
+	encCfg, txClient, _ := setupTxClientWithDefaultParams(t)
+
+	mockTxClient, err := user.NewTxClient(
+		encCfg.Codec,
+		txClient.Signer(),
+		conn,
+		encCfg.InterfaceRegistry,
+		opts...,
+	)
+	require.NoError(t, err)
+
+	return mockTxClient, conn
+}
diff --git a/pkg/user/tx_client_test.go b/pkg/user/tx_client_test.go
index 779635aa..efe432f2 100644
--- a/pkg/user/tx_client_test.go
+++ b/pkg/user/tx_client_test.go
@@ -353,6 +353,68 @@ func TestEvictions(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, txInfo.Status, core.TxStatusCommitted)
 	}
+
+	t.Run("gets nonce mismatch during eviction and gets confirmed in the meantime", func(t *testing.T) {
+		expectedTxHash := "test-tx-hash-123" // This matches what our mock server returns
+
+		// Happy path: PENDING -> EVICTED -> COMMITTED
+		responseSequences := map[string][]*tx.TxStatusResponse{
+			expectedTxHash: {
+				{Status: core.TxStatusPending},
+				{Status: core.TxStatusEvicted},
+				{Status: core.TxStatusEvicted},
+				{Status: core.TxStatusEvicted},
+				{Status: core.TxStatusCommitted, Height: 102, ExecutionCode: 0},
+			},
+		}
+
+		mockTxClient, conn := setupTxClientWithMockGRPCServer(t, responseSequences, user.WithPollTime(1*time.Second))
+		defer conn.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+
+		blobs := blobfactory.ManyRandBlobs(random.New(), 1, 1)
+		resp, err := mockTxClient.BroadcastPayForBlob(ctx, blobs, user.SetFee(1), user.SetGasLimit(1))
+		require.NoError(t, err)
+		require.Equal(t, resp.Code, abci.CodeTypeOK)
+		require.Equal(t, resp.TxHash, expectedTxHash)
+
+		response, err := mockTxClient.ConfirmTx(ctx, resp.TxHash)
+		require.NoError(t, err)
+		require.Equal(t, response.TxHash, resp.TxHash)
+		require.Equal(t, response.Code, abci.CodeTypeOK)
+	})
+
+	t.Run("gets nonce mismatch during eviction and hits eviction poll timeout", func(t *testing.T) {
+		expectedTxHash := "test-tx-hash-123"
+
+		// Transaction will get evicted and then resubmitted with nonce mismatch error
+		// Define response sequence: PENDING -> EVICTED -> Sequence mismatch -> EVICTED -> Eviction Poll Timeout
+		// Responses will return evicted until the eviction poll timeout is reached
+		responseSequences := map[string][]*tx.TxStatusResponse{
+			expectedTxHash: {
+				{Status: core.TxStatusPending}, // 1st call
+				{Status: core.TxStatusEvicted}, // 2nd call
+				{Status: core.TxStatusEvicted}, // 3rd and following calls
+			},
+		}
+
+		mockTxClient, conn := setupTxClientWithMockGRPCServer(t, responseSequences, user.WithPollTime(5*time.Second))
+		defer conn.Close()
+
+		ctx := context.Background()
+
+		blobs := blobfactory.ManyRandBlobs(random.New(), 1, 1)
+		resp, err := mockTxClient.BroadcastPayForBlob(ctx, blobs, user.SetFee(1), user.SetGasLimit(1))
+		require.NoError(t, err)
+		require.Equal(t, resp.Code, abci.CodeTypeOK)
+		require.Equal(t, resp.TxHash, expectedTxHash)
+
+		_, err = mockTxClient.ConfirmTx(ctx, resp.TxHash)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "eviction poll timeout")
+	})
 }
 
 // TestWithEstimatorService ensures that if the WithEstimatorService