feat: nonce handling with signer

app/errors/insufficient_gas_price_test.go

@@ -47,9 +47,7 @@ func TestInsufficientMinGasPriceIntegration(t *testing.T) {
 	msg, err := blob.NewMsgPayForBlobs(signer.Address().String(), appconsts.LatestVersion, b)
 	require.NoError(t, err)
 
-	tx, err := signer.CreateTx([]sdk.Msg{msg}, user.SetGasLimit(gasLimit), user.SetFeeAmount(fee))
-	require.NoError(t, err)
-	sdkTx, err := enc.TxConfig.TxDecoder()(tx)
+	sdkTx, err := signer.CreateTx([]sdk.Msg{msg}, user.SetGasLimit(gasLimit), user.SetFeeAmount(fee))
 	require.NoError(t, err)
 
 	decorator := ante.NewDeductFeeDecorator(testApp.AccountKeeper, testApp.BankKeeper, testApp.FeeGrantKeeper, nil)

app/errors/nonce_mismatch.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"strings"
 
 	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
 )
@@ -13,16 +14,31 @@ func IsNonceMismatch(err error) bool {
 	return errors.Is(err, sdkerrors.ErrWrongSequence)
 }
 
+// IsNonceMismatch checks if the error code matches the sequence mismatch.
+func IsNonceMismatchCode(code uint32) bool {
+	return code == sdkerrors.ErrWrongSequence.ABCICode()
+}
+
 // ParseNonceMismatch extracts the expected sequence number from the
 // ErrWrongSequence error.
 func ParseNonceMismatch(err error) (uint64, error) {
 	if !IsNonceMismatch(err) {
 		return 0, errors.New("error is not a sequence mismatch")
 	}
 
-	numbers := regexpInt.FindAllString(err.Error(), -1)
+	return ParseExpectedSequence(err.Error())
+}
+
+// ParseExpectedSequence extracts the expected sequence number from the
+// ErrWrongSequence error.
+func ParseExpectedSequence(str string) (uint64, error) {
+	if !strings.HasPrefix(str, "account sequence mismatch") {
+		return 0, fmt.Errorf("unexpected wrong sequence error: %s", str)
+	}
+
+	numbers := regexpInt.FindAllString(str, -1)
 	if len(numbers) != 2 {
-		return 0, fmt.Errorf("unexpected wrong sequence error: %w", err)
+		return 0, fmt.Errorf("expected two numbers in string, got %d", len(numbers))
 	}
 
 	// the first number is the expected sequence number

app/errors/nonce_mismatch_test.go

@@ -41,9 +41,7 @@ func TestNonceMismatchIntegration(t *testing.T) {
 	msg, err := blob.NewMsgPayForBlobs(signer.Address().String(), appconsts.LatestVersion, b)
 	require.NoError(t, err)
 
-	tx, err := signer.CreateTx([]sdk.Msg{msg})
-	require.NoError(t, err)
-	sdkTx, err := enc.TxConfig.TxDecoder()(tx)
+	sdkTx, err := signer.CreateTx([]sdk.Msg{msg})
 	require.NoError(t, err)
 
 	decorator := ante.NewSigVerificationDecorator(testApp.AccountKeeper, encCfg.TxConfig.SignModeHandler())

app/test/big_blob_test.go

@@ -77,11 +77,9 @@ func (s *BigBlobSuite) TestErrBlobsTooLarge() {
 
 	for _, tc := range testCases {
 		s.Run(tc.name, func() {
-			blobTx, err := signer.CreatePayForBlob([]*blob.Blob{tc.blob}, user.SetGasLimit(1e9), user.SetFee(2000000))
-			require.NoError(t, err)
 			subCtx, cancel := context.WithTimeout(s.cctx.GoContext(), 30*time.Second)
 			defer cancel()
-			res, err := signer.BroadcastTx(subCtx, blobTx)
+			res, err := signer.SubmitPayForBlob(subCtx, []*blob.Blob{tc.blob}, user.SetGasLimit(1e9), user.SetFee(2000000))
 			require.NoError(t, err)
 			require.NotNil(t, res)
 			require.Equal(t, tc.want, res.Code, res.Logs)

app/test/priority_test.go

@@ -76,16 +76,15 @@ func (s *PriorityTestSuite) TestPriorityByGasPrice() {
 		blobSize := uint32(100)
 		gasLimit := blobtypes.DefaultEstimateGas([]uint32{blobSize})
 		gasPrice := s.rand.Float64()
-		btx, err := signer.CreatePayForBlob(
+		resp, err := signer.SubmitPayForBlob(
+			s.cctx.GoContext(),
 			blobfactory.ManyBlobs(
 				s.rand,
 				[]namespace.Namespace{namespace.RandomBlobNamespace()},
 				[]int{100}),
 			user.SetGasLimitAndFee(gasLimit, gasPrice),
 		)
 		require.NoError(t, err)
-		resp, err := signer.BroadcastTx(s.cctx.GoContext(), btx)
-		require.NoError(t, err)
 		require.Equal(t, abci.CodeTypeOK, resp.Code, resp.RawLog)
 		hashes = append(hashes, resp.TxHash)
 	}

go.work.sum

@@ -927,6 +927,7 @@ github.com/moby/term v0.0.0-20221205130635-1aeaba878587/go.mod h1:8FzsFHVUBGZdbD
 github.com/moricho/tparallel v0.3.0/go.mod h1:leENX2cUv7Sv2qDgdi0D0fCftN8fRC67Bcn8pqzeYNI=
 github.com/moul/http2curl v1.0.0/go.mod h1:8UbvGypXm98wA/IqH45anm5Y2Z6ep6O31QGOAZ3H0fQ=
 github.com/mrunalp/fileutils v0.5.0/go.mod h1:M1WthSahJixYnrXQl/DFQuteStB1weuxD2QJNHXfbSQ=
+github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f h1:y5//uYreIhSUg3J1GEMiLbxo1LJaP8RfCpH6pymGZus=
 github.com/nakabonne/nestif v0.3.1/go.mod h1:9EtoZochLn5iUprVDmDjqGKPofoUEBL8U4Ngq6aY7OE=
 github.com/nats-io/jwt/v2 v2.0.3/go.mod h1:VRP+deawSXyhNjXmxPCHskrR6Mq50BqpEI5SEcNiGlY=
 github.com/nats-io/nats-server/v2 v2.5.0/go.mod h1:Kj86UtrXAL6LwYRA6H4RqzkHhK0Vcv2ZnKD5WbQ1t3g=

pkg/user/e2e_test.go

@@ -0,0 +1,92 @@
+package user_test
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/celestiaorg/celestia-app/pkg/appconsts"
+	"github.com/celestiaorg/celestia-app/pkg/user"
+	"github.com/celestiaorg/celestia-app/test/util/blobfactory"
+	"github.com/celestiaorg/celestia-app/test/util/testnode"
+	"github.com/celestiaorg/go-square/blob"
+	"github.com/stretchr/testify/require"
+	tmrand "github.com/tendermint/tendermint/libs/rand"
+)
+
+func TestConcurrentTxSubmission(t *testing.T) {
+	// Setup network
+	tmConfig := testnode.DefaultTendermintConfig()
+	tmConfig.Consensus.TimeoutCommit = 10 * time.Second
+	ctx, _, _ := testnode.NewNetwork(t, testnode.DefaultConfig().WithTendermintConfig(tmConfig))
+	_, err := ctx.WaitForHeight(1)
+	require.NoError(t, err)
+
+	// Setup signer
+	signer, err := testnode.NewSingleSignerFromContext(ctx)
+	require.NoError(t, err)
+
+	// Pregenerate all the blobs
+	numTxs := 10
+	blobs := blobfactory.ManyRandBlobs(tmrand.NewRand(), blobfactory.Repeat(2048, numTxs)...)
+
+	// Prepare transactions
+	var (
+		wg    sync.WaitGroup
+		errCh = make(chan error)
+	)
+
+	subCtx, cancel := context.WithCancel(ctx.GoContext())
+	defer cancel()
+	time.AfterFunc(time.Minute, cancel)
+	for i := 0; i < numTxs; i++ {
+		wg.Add(1)
+		go func(b *blob.Blob) {
+			defer wg.Done()
+			_, err := signer.SubmitPayForBlob(subCtx, []*blob.Blob{b}, user.SetGasLimitAndFee(500_000, appconsts.DefaultGlobalMinGasPrice))
+			if err != nil && !errors.Is(err, context.Canceled) {
+				// only catch the first error
+				select {
+				case errCh <- err:
+					cancel()
+				default:
+				}
+			}
+		}(blobs[i])
+	}
+	wg.Wait()
+
+	select {
+	case err := <-errCh:
+		require.NoError(t, err)
+	default:
+	}
+}
+
+func TestSignerTwins(t *testing.T) {
+	// Setup network
+	tmConfig := testnode.DefaultTendermintConfig()
+	tmConfig.Consensus.TimeoutCommit = 10 * time.Second
+	ctx, _, _ := testnode.NewNetwork(t, testnode.DefaultConfig().WithTendermintConfig(tmConfig))
+	_, err := ctx.WaitForHeight(1)
+	require.NoError(t, err)
+
+	signer1, err := testnode.NewSingleSignerFromContext(ctx)
+	require.NoError(t, err)
+	signer2, err := testnode.NewSingleSignerFromContext(ctx)
+	require.NoError(t, err)
+
+	blobs := blobfactory.ManyRandBlobs(tmrand.NewRand(), blobfactory.Repeat(2048, 8)...)
+
+	_, err = signer1.SubmitPayForBlob(ctx.GoContext(), blobs[:1], user.SetGasLimitAndFee(500_000, appconsts.DefaultGlobalMinGasPrice))
+	require.NoError(t, err)
+
+	_, err = signer2.SubmitPayForBlob(ctx.GoContext(), blobs[1:3], user.SetGasLimitAndFee(500_000, appconsts.DefaultGlobalMinGasPrice))
+	require.NoError(t, err)
+
+	signer1.ForceSetSequence(4)
+	_, err = signer1.SubmitPayForBlob(ctx.GoContext(), blobs[3:5], user.SetGasLimitAndFee(500_000, appconsts.DefaultGlobalMinGasPrice))
+	require.NoError(t, err)
+}