Repository: celestiaorg/celestia-app

Issue:
Using multiple endpoints increments the sequence each time
### Summary

As was pointed out in this [PR](https://github.com/celestiaorg/celestia-app/pull/5776#discussion_r2353727083), when we use multiple endpoints, we incorrectly increment the sequence for each submission/endpoint. 

Hints from issue comments:

Solution patch:
diff --git a/Makefile b/Makefile
index 1bf7d632..0b6d12a8 100644
--- a/Makefile
+++ b/Makefile
@@ -315,7 +315,7 @@ test-race:
 # TODO: Remove the -skip flag once the following tests no longer contain data races.
 # https://github.com/celestiaorg/celestia-app/issues/1369
 	@echo "--> Running tests in race mode"
-	@go test -timeout 15m ./... -v -race -skip "TestPrepareProposalConsistency|TestIntegrationTestSuite|TestSquareSizeIntegrationTest|TestStandardSDKIntegrationTestSuite|TestTxsimCommandFlags|TestTxsimCommandEnvVar|TestTxsimDefaultKeypath|TestMintIntegrationTestSuite|TestUpgrade|TestMaliciousTestNode|TestBigBlobSuite|TestQGBIntegrationSuite|TestSignerTestSuite|TestPriorityTestSuite|TestTimeInPrepareProposalContext|TestCLITestSuite|TestLegacyUpgrade|TestSignerTwins|TestConcurrentTxSubmission|TestTxClientTestSuite|Test_testnode|TestEvictions|TestEstimateGasUsed|TestEstimateGasPrice|TestWithEstimatorService|TestTxsOverMaxTxSizeGetRejected|TestStart_Success|TestReadBlockchainHeaders|TestPrepareProposalCappingNumberOfMessages|TestGasEstimatorE2E|TestGasEstimatorE2EWithNetworkMinGasPrice|TestRejections|TestClaimRewardsAfterFullUndelegation"
+	@go test -timeout 15m ./... -v -race -skip "TestPrepareProposalConsistency|TestIntegrationTestSuite|TestSquareSizeIntegrationTest|TestStandardSDKIntegrationTestSuite|TestTxsimCommandFlags|TestTxsimCommandEnvVar|TestTxsimDefaultKeypath|TestMintIntegrationTestSuite|TestUpgrade|TestMaliciousTestNode|TestBigBlobSuite|TestQGBIntegrationSuite|TestSignerTestSuite|TestPriorityTestSuite|TestTimeInPrepareProposalContext|TestCLITestSuite|TestLegacyUpgrade|TestSignerTwins|TestConcurrentTxSubmission|TestTxClientTestSuite|Test_testnode|TestEvictions|TestEstimateGasUsed|TestEstimateGasPrice|TestWithEstimatorService|TestTxsOverMaxTxSizeGetRejected|TestStart_Success|TestReadBlockchainHeaders|TestPrepareProposalCappingNumberOfMessages|TestGasEstimatorE2E|TestGasEstimatorE2EWithNetworkMinGasPrice|TestRejections|TestClaimRewardsAfterFullUndelegation|TestParallelTxSubmission"
 .PHONY: test-race
 
 ## test-bench: Run benchmark unit tests.
diff --git a/pkg/user/account.go b/pkg/user/account.go
index aaa0e45a..056aaa9e 100644
--- a/pkg/user/account.go
+++ b/pkg/user/account.go
@@ -7,6 +7,7 @@ import (
 	cryptotypes "github.com/cosmos/cosmos-sdk/crypto/types"
 	"github.com/cosmos/cosmos-sdk/types"
 	authtypes "github.com/cosmos/cosmos-sdk/x/auth/types"
+	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
 	"google.golang.org/grpc"
 )
 
@@ -81,3 +82,21 @@ func QueryAccount(ctx context.Context, conn *grpc.ClientConn, registry codectype
 	accNum, seqNum = acc.GetAccountNumber(), acc.GetSequence()
 	return accNum, seqNum, nil
 }
+
+// QueryAccountBalance fetches the balance of a specific denomination for an account.
+func QueryAccountBalance(ctx context.Context, conn *grpc.ClientConn, _ codectypes.InterfaceRegistry, address types.AccAddress, denom string) (types.Coin, error) {
+	balanceQuery := banktypes.NewQueryClient(conn)
+	balanceResp, err := balanceQuery.Balance(ctx, &banktypes.QueryBalanceRequest{
+		Address: address.String(),
+		Denom:   denom,
+	})
+	if err != nil {
+		return types.Coin{}, err
+	}
+
+	if balanceResp.Balance == nil {
+		return types.NewInt64Coin(denom, 0), nil
+	}
+
+	return *balanceResp.Balance, nil
+}
diff --git a/pkg/user/parallel_tx_submission.go b/pkg/user/parallel_tx_submission.go
new file mode 100644
index 00000000..5a9886dc
--- /dev/null
+++ b/pkg/user/parallel_tx_submission.go
@@ -0,0 +1,505 @@
+package user
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+
+	"cosmossdk.io/x/feegrant"
+	"github.com/celestiaorg/celestia-app/v6/pkg/appconsts"
+	"github.com/celestiaorg/go-square/v3/share"
+	"github.com/cosmos/cosmos-sdk/crypto/hd"
+	"github.com/cosmos/cosmos-sdk/crypto/keyring"
+	sdktypes "github.com/cosmos/cosmos-sdk/types"
+	bank "github.com/cosmos/cosmos-sdk/x/bank/types"
+)
+
+// SubmissionJob represents a transaction submission task for parallel processing
+type SubmissionJob struct {
+	Blobs    []*share.Blob
+	Options  []TxOption
+	Ctx      context.Context
+	ResultsC chan SubmissionResult
+}
+
+// SubmissionResult contains the result of a parallel transaction submission
+type SubmissionResult struct {
+	Signer     string
+	TxResponse *TxResponse
+	Error      error
+}
+
+// txQueue manages parallel transaction submission
+type txQueue struct {
+	client      *TxClient
+	jobQueue    chan *SubmissionJob
+	workers     []*txWorker
+	started     atomic.Bool
+	ctx         context.Context
+	cancel      context.CancelFunc
+	initialized atomic.Bool // whether workers have been initialized
+	wg          sync.WaitGroup
+}
+
+// txWorker represents a worker that processes transactions using a specific account
+type txWorker struct {
+	id          int
+	accountName string
+	address     string
+	client      *TxClient
+	jobQueue    chan *SubmissionJob
+}
+
+const (
+	defaultParallelQueueSize = 100
+)
+
+func newTxQueue(client *TxClient, numWorkers int) *txQueue {
+	pool := &txQueue{
+		client:   client,
+		jobQueue: make(chan *SubmissionJob, defaultParallelQueueSize),
+		workers:  make([]*txWorker, numWorkers),
+	}
+
+	// Create workers: first worker always uses existing signer account
+	for i := range numWorkers {
+		var accountName, address string
+
+		if i == 0 {
+			// First worker uses the current default account
+			accountName = client.DefaultAccountName()
+			address = client.DefaultAddress().String()
+		} else {
+			// Additional workers use generated account names
+			accountName = fmt.Sprintf("parallel-worker-%d", i)
+
+			// Get worker address from keyring if account exists
+			if record, err := client.signer.keys.Key(accountName); err == nil {
+				if addr, err := record.GetAddress(); err == nil {
+					address = addr.String()
+				}
+			}
+		}
+
+		worker := &txWorker{
+			id:          i,
+			accountName: accountName,
+			address:     address,
+			client:      client,
+			jobQueue:    pool.jobQueue,
+		}
+		pool.workers[i] = worker
+	}
+
+	return pool
+}
+
+// start initiates all workers in the pool
+func (p *txQueue) start(ctx context.Context) error {
+	if p.started.Load() {
+		return nil
+	}
+
+	// Initialize worker accounts if needed
+	if !p.initialized.Load() {
+		if err := p.initializeWorkerAccounts(ctx); err != nil {
+			return fmt.Errorf("failed to initialize worker accounts: %w", err)
+		}
+		p.initialized.Store(true)
+	}
+
+	// Recreate job queue channel if it was closed during previous stop
+	p.jobQueue = make(chan *SubmissionJob, defaultParallelQueueSize)
+	// Update workers to use new job queue BEFORE starting goroutines
+	for _, worker := range p.workers {
+		worker.jobQueue = p.jobQueue
+	}
+
+	// Create a new context for this pool instance
+	p.ctx, p.cancel = context.WithCancel(ctx)
+
+	// Set started flag before starting workers to prevent race
+	p.started.Store(true)
+
+	// Start workers after everything is set up
+	for _, worker := range p.workers {
+		p.wg.Add(1)
+		go func(w *txWorker) {
+			defer p.wg.Done()
+			w.start(p.ctx)
+		}(worker)
+	}
+
+	return nil
+}
+
+// stop shuts down all workers in the pool
+func (p *txQueue) stop() {
+	if !p.started.Load() {
+		return
+	}
+
+	if p.cancel != nil {
+		p.cancel()
+	}
+
+	// Close the job queue to signal workers to stop accepting new jobs
+	close(p.jobQueue)
+
+	// Wait for all workers to finish before marking as stopped
+	p.wg.Wait()
+
+	p.started.Store(false)
+}
+
+// submitJob submits a job to the parallel worker pool
+func (p *txQueue) submitJob(job *SubmissionJob) {
+	if !p.started.Load() || p.ctx == nil {
+		job.ResultsC <- SubmissionResult{Error: errors.New("tx queue not started")}
+		return
+	}
+
+	select {
+	case p.jobQueue <- job:
+	case <-p.ctx.Done():
+		job.ResultsC <- SubmissionResult{Error: errors.New("tx queue full or has stopped")}
+	}
+}
+
+// isStarted returns whether the tx queue is started
+func (p *txQueue) isStarted() bool {
+	return p.started.Load()
+}
+
+// start begins the worker's job processing loop
+func (w *txWorker) start(ctx context.Context) {
+	for {
+		select {
+		case job, ok := <-w.jobQueue:
+			if !ok {
+				// Channel closed, exit worker
+				return
+			}
+			w.processJob(job, ctx)
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// processJob handles the actual transaction submission for a job
+func (w *txWorker) processJob(job *SubmissionJob, workerCtx context.Context) {
+	jobCtx := job.Ctx
+	if jobCtx == nil {
+		jobCtx = context.Background()
+	}
+
+	select {
+	case <-jobCtx.Done():
+		job.ResultsC <- SubmissionResult{Signer: w.address, Error: jobCtx.Err()}
+		return
+	case <-workerCtx.Done():
+		job.ResultsC <- SubmissionResult{Signer: w.address, Error: workerCtx.Err()}
+		return
+	default:
+	}
+
+	options := job.Options
+
+	// Only add fee granter option for workers that aren't the primary account (worker 0)
+	if w.id != 0 {
+		// Add fee granter option so master account pays for worker transaction fees
+		options = append([]TxOption{SetFeeGranter(w.client.DefaultAddress())}, options...)
+	}
+
+	// Use the worker's dedicated account to submit the transaction
+	txResponse, err := w.client.SubmitPayForBlobWithAccount(jobCtx, w.accountName, job.Blobs, options...)
+
+	result := SubmissionResult{
+		Signer:     w.address,
+		TxResponse: txResponse,
+		Error:      err,
+	}
+
+	// Send result back through the job-specific results channel
+	job.ResultsC <- result
+}
+
+// initializeWorkerAccounts creates and initializes all worker accounts for parallel submission.
+// It creates the accounts in the keyring if they don't exist, funds them with a small balance,
+// and sets up fee grants so the main account pays for transaction fees.
+func (p *txQueue) initializeWorkerAccounts(ctx context.Context) error {
+	// No work required if we've already initialized all workers.
+	if p.initialized.Load() {
+		return nil
+	}
+
+	// If we only have one worker (index 0), skip all initialization as it uses the existing signer account
+	if len(p.workers) == 1 {
+		p.initialized.Store(true)
+		return nil
+	}
+
+	// Get the list of worker accounts that need to be initialized
+	// Skip the first worker (index 0) as it always uses the existing signer account
+	var workersToInit []*txWorker
+	var workersToLoad []*txWorker
+	for i, worker := range p.workers {
+		if i == 0 {
+			// Skip first worker - it uses existing account
+			continue
+		}
+
+		// Check if account exists in signer
+		if _, exists := p.client.signer.accounts[worker.accountName]; !exists {
+			// Check if account exists in keyring but not loaded in signer
+			if _, err := p.client.signer.keys.Key(worker.accountName); err == nil {
+				// Account exists in keyring but not loaded - add to load list
+				workersToLoad = append(workersToLoad, worker)
+			} else {
+				// Account doesn't exist anywhere - needs full initialization
+				workersToInit = append(workersToInit, worker)
+			}
+		}
+	}
+
+	// Load existing accounts from keyring into signer
+	if len(workersToLoad) > 0 {
+		for _, worker := range workersToLoad {
+			if err := p.client.loadWorkerAccount(worker); err != nil {
+				return fmt.Errorf("failed to load existing worker account %s: %w", worker.accountName, err)
+			}
+		}
+	}
+
+	if len(workersToInit) == 0 {
+		p.initialized.Store(true)
+		return nil // All accounts already exist
+	}
+
+	// Create accounts in keyring if they don't exist
+	for _, worker := range workersToInit {
+		if err := p.client.createWorkerAccount(worker.accountName); err != nil {
+			return fmt.Errorf("failed to create worker account %s: %w", worker.accountName, err)
+		}
+	}
+
+	// Fund accounts and set up fee grants in a single transaction
+	if err := p.client.fundAndGrantWorkerAccounts(ctx, workersToInit); err != nil {
+		return fmt.Errorf("failed to fund and grant worker accounts: %w", err)
+	}
+
+	return nil
+}
+
+// createFeeGrantMessages creates fee grant messages for workers that don't already have grants
+func (client *TxClient) createFeeGrantMessages(ctx context.Context, workers []*txWorker) ([]sdktypes.Msg, uint64, error) {
+	msgs := make([]sdktypes.Msg, 0, len(workers))
+	totalGasLimit := uint64(0)
+	masterAddress := client.defaultAddress
+
+	for _, worker := range workers {
+		// Get worker address
+		record, err := client.signer.keys.Key(worker.accountName)
+		if err != nil {
+			return nil, 0, fmt.Errorf("failed to get worker account %s from keyring: %w", worker.accountName, err)
+		}
+
+		workerAddress, err := record.GetAddress()
+		if err != nil {
+			return nil, 0, fmt.Errorf("failed to get address for worker account %s: %w", worker.accountName, err)
+		}
+
+		// Check if fee grant already exists
+		hasGrant, err := client.hasFeeGrant(ctx, masterAddress, workerAddress)
+		if err != nil {
+			return nil, 0, fmt.Errorf("failed to check fee grant for worker %s: %w", worker.accountName, err)
+		}
+
+		if !hasGrant {
+			// Create feegrant message so master account pays for worker fees
+			feegrantMsg, err := feegrant.NewMsgGrantAllowance(
+				&feegrant.BasicAllowance{}, // Unlimited allowance
+				masterAddress,
+				workerAddress,
+			)
+			if err != nil {
+				return nil, 0, fmt.Errorf("failed to create feegrant message for worker %s: %w", worker.accountName, err)
+			}
+			msgs = append(msgs, feegrantMsg)
+			totalGasLimit += FeegrantGasLimit
+		}
+	}
+
+	return msgs, totalGasLimit, nil
+}
+
+// loadWorkerAccount loads an existing account from keyring into the signer
+func (client *TxClient) loadWorkerAccount(worker *txWorker) error {
+	// Get account from keyring
+	record, err := client.signer.keys.Key(worker.accountName)
+	if err != nil {
+		return fmt.Errorf("failed to get worker account %s from keyring: %w", worker.accountName, err)
+	}
+
+	workerAddress, err := record.GetAddress()
+	if err != nil {
+		return fmt.Errorf("failed to get address for worker account %s: %w", worker.accountName, err)
+	}
+
+	// Query account info from chain
+	accNum, seqNum, err := QueryAccount(context.Background(), client.conns[0], client.registry, workerAddress)
+	if err != nil {
+		return fmt.Errorf("failed to query worker account %s on chain: %w", worker.accountName, err)
+	}
+
+	// Add account to signer
+	account := NewAccount(worker.accountName, accNum, seqNum)
+	if err := client.signer.AddAccount(account); err != nil {
+		return fmt.Errorf("failed to add worker account %s to signer: %w", worker.accountName, err)
+	}
+
+	// Update worker address
+	worker.address = workerAddress.String()
+
+	return nil
+}
+
+// createWorkerAccount creates a new account in the keyring
+func (client *TxClient) createWorkerAccount(accountName string) error {
+	// Check if account already exists in keyring
+	if _, err := client.signer.keys.Key(accountName); err == nil {
+		return nil // Account already exists
+	}
+
+	// Create new account
+	path := hd.CreateHDPath(sdktypes.CoinType, 0, 0).String()
+	_, _, err := client.signer.keys.NewMnemonic(accountName, keyring.English, path, keyring.DefaultBIP39Passphrase, hd.Secp256k1)
+	if err != nil {
+		return fmt.Errorf("failed to create account %s in keyring: %w", accountName, err)
+	}
+
+	return nil
+}
+
+// accountNeedsFunding checks if an account needs funding by querying its balance
+func (client *TxClient) accountNeedsFunding(ctx context.Context, address sdktypes.AccAddress) (bool, error) {
+	// Query account balance
+	balance, err := QueryAccountBalance(ctx, client.conns[0], client.registry, address, appconsts.BondDenom)
+	if err != nil {
+		// If account doesn't exist, it needs funding
+		return true, nil
+	}
+
+	// Check if balance is less than the default worker balance
+	// Note: we check for >= DefaultWorkerBalance to avoid re-funding accounts that already have sufficient balance
+	return balance.Amount.Int64() < DefaultWorkerBalance, nil
+}
+
+// hasFeeGrant checks if a fee grant exists between granter and grantee
+func (client *TxClient) hasFeeGrant(ctx context.Context, granter, grantee sdktypes.AccAddress) (bool, error) {
+	feegrantQuery := feegrant.NewQueryClient(client.conns[0])
+	_, err := feegrantQuery.Allowance(ctx, &feegrant.QueryAllowanceRequest{
+		Granter: granter.String(),
+		Grantee: grantee.String(),
+	})
+	if err != nil {
+		// If error contains "not found" or similar, grant doesn't exist
+		return false, nil
+	}
+	return true, nil
+}
+
+// fundAndGrantWorkerAccounts sends funds to worker accounts and sets up fee grants
+func (client *TxClient) fundAndGrantWorkerAccounts(ctx context.Context, workers []*txWorker) error {
+	if len(workers) == 0 {
+		return nil
+	}
+
+	msgs := make([]sdktypes.Msg, 0, len(workers)*2) // Each worker needs up to 2 msgs: send + feegrant
+	totalGasLimit := uint64(0)
+
+	masterAddress := client.defaultAddress
+
+	// Create send messages for funding accounts that need funding
+	for _, worker := range workers {
+		// Get worker address
+		record, err := client.signer.keys.Key(worker.accountName)
+		if err != nil {
+			return fmt.Errorf("failed to get worker account %s from keyring: %w", worker.accountName, err)
+		}
+
+		workerAddress, err := record.GetAddress()
+		if err != nil {
+			return fmt.Errorf("failed to get address for worker account %s: %w", worker.accountName, err)
+		}
+
+		// Check if account already has sufficient balance
+		needsFunding, err := client.accountNeedsFunding(ctx, workerAddress)
+		if err != nil {
+			// If we can't check balance, assume it needs funding to avoid blocking
+			needsFunding = true
+		}
+
+		if needsFunding {
+			// Create send message to fund the account
+			sendMsg := bank.NewMsgSend(
+				masterAddress,
+				workerAddress,
+				sdktypes.NewCoins(sdktypes.NewInt64Coin(appconsts.BondDenom, DefaultWorkerBalance)),
+			)
+			msgs = append(msgs, sendMsg)
+			totalGasLimit += SendGasLimit
+		}
+	}
+
+	// Add fee grant messages
+	feeGrantMsgs, feeGrantGasLimit, err := client.createFeeGrantMessages(ctx, workers)
+	if err != nil {
+		return err
+	}
+	msgs = append(msgs, feeGrantMsgs...)
+	totalGasLimit += feeGrantGasLimit
+
+	// Submit the initialization transaction only if there are messages to send
+	if len(msgs) > 0 {
+		opts := []TxOption{SetGasLimit(totalGasLimit)}
+		_, err = client.SubmitTx(ctx, msgs, opts...)
+		if err != nil {
+			return fmt.Errorf("failed to submit initialization transaction: %w", err)
+		}
+	}
+
+	// Add the worker accounts to the signer
+	for _, worker := range workers {
+		record, err := client.signer.keys.Key(worker.accountName)
+		if err != nil {
+			return fmt.Errorf("failed to get worker account %s from keyring: %w", worker.accountName, err)
+		}
+
+		workerAddress, err := record.GetAddress()
+		if err != nil {
+			return fmt.Errorf("failed to get address for worker account %s: %w", worker.accountName, err)
+		}
+
+		// Query the account info from chain
+		accNum, seqNum, err := QueryAccount(ctx, client.conns[0], client.registry, workerAddress)
+		if err != nil {
+			return fmt.Errorf("failed to query worker account %s on chain: %w", worker.accountName, err)
+		}
+
+		// Add account to signer
+		account := NewAccount(worker.accountName, accNum, seqNum)
+		if err := client.signer.AddAccount(account); err != nil {
+			return fmt.Errorf("failed to add worker account %s to signer: %w", worker.accountName, err)
+		}
+
+		// Update worker address now that account is created
+		worker.address = workerAddress.String()
+	}
+
+	client.txQueue.initialized.Store(true)
+	return nil
+}
diff --git a/pkg/user/tx_client.go b/pkg/user/tx_client.go
index 7d636c5c..a168a3ed 100644
--- a/pkg/user/tx_client.go
+++ b/pkg/user/tx_client.go
@@ -42,6 +42,10 @@ import (
 const (
 	DefaultPollTime          = 3 * time.Second
 	txTrackerPruningInterval = 10 * time.Minute
+	// Gas limits for initialization transactions
+	SendGasLimit         = 100000
+	FeegrantGasLimit     = 800000
+	DefaultWorkerBalance = 1
 	// evictionPollTimeOut is the timeout for checking if an evicted transaction
 	// gets committed after experiencing a broadcast error during resubmission
 	evictionPollTimeOut = 1 * time.Minute
@@ -121,6 +125,12 @@ func WithDefaultAccount(name string) Option {
 		}
 		c.defaultAccount = name
 		c.defaultAddress = addr
+
+		// Update worker 0's account if tx queue already exists
+		if c.txQueue != nil && len(c.txQueue.workers) > 0 {
+			c.txQueue.workers[0].accountName = name
+			c.txQueue.workers[0].address = addr.String()
+		}
 	}
 }
 
@@ -142,6 +152,31 @@ func WithAdditionalCoreEndpoints(conns []*grpc.ClientConn) Option {
 	}
 }
 
+// WithTxWorkers enables parallel transaction submission with the specified number of worker accounts.
+// Worker accounts are automatically generated with hardcoded names unless numWorkers is 1, in which case
+// the existing default account is used.
+// Workers are initialized automatically when SetupTxClient is called.
+func WithTxWorkers(numWorkers int) Option {
+	if numWorkers <= 0 {
+		return func(*TxClient) {}
+	}
+
+	return func(c *TxClient) {
+		pool := newTxQueue(c, numWorkers)
+		c.txQueue = pool
+	}
+}
+
+// WithParallelQueueSize sets the buffer size for the parallel submission job queue.
+// Default is 100 if not specified.
+func WithParallelQueueSize(size int) Option {
+	return func(c *TxClient) {
+		if c.txQueue != nil {
+			c.txQueue.jobQueue = make(chan *SubmissionJob, size)
+		}
+	}
+}
+
 // TxClient is an abstraction for building, signing, and broadcasting Celestia transactions
 // It supports multiple accounts. If none is specified, it will
 // try to use the default account.
@@ -162,6 +197,8 @@ type TxClient struct {
 	// that was submitted to the chain
 	txTracker           map[string]txInfo
 	gasEstimationClient gasestimation.GasEstimatorClient
+	// txQueue manages parallel transaction submission when enabled
+	txQueue *txQueue
 }
 
 // NewTxClient returns a new TxClient
@@ -185,6 +222,7 @@ func NewTxClient(
 	if err != nil {
 		return nil, err
 	}
+
 	txClient := &TxClient{
 		signer:              signer,
 		registry:            registry,
@@ -201,9 +239,10 @@ func NewTxClient(
 		opt(txClient)
 	}
 
-	// Sanity check to ensure we don't have more than 3 connections
-	if len(txClient.conns) > 3 {
-		txClient.conns = txClient.conns[:3]
+	// Always create a tx queue with at least 1 worker (the default account)
+	// unless already configured by WithTxWorkers option
+	if txClient.txQueue == nil {
+		txClient.txQueue = newTxQueue(txClient, 1)
 	}
 
 	return txClient, nil
@@ -253,23 +292,58 @@ func SetupTxClient(
 		return nil, fmt.Errorf("failed to create signer: %w", err)
 	}
 
-	return NewTxClient(encCfg.Codec, signer, conn, encCfg.InterfaceRegistry, options...)
+	txClient, err := NewTxClient(encCfg.Codec, signer, conn, encCfg.InterfaceRegistry, options...)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := txClient.txQueue.start(ctx); err != nil {
+		return nil, fmt.Errorf("failed to start tx queue: %w", err)
+	}
+
+	return txClient, nil
 }
 
 // SubmitPayForBlob forms a transaction from the provided blobs, signs it, and submits it to the chain.
 // TxOptions may be provided to set the fee and gas limit.
+// This method uses the tx queue infrastructure and blocks until the transaction is confirmed.
 func (client *TxClient) SubmitPayForBlob(ctx context.Context, blobs []*share.Blob, opts ...TxOption) (*TxResponse, error) {
-	resp, err := client.BroadcastPayForBlob(ctx, blobs, opts...)
-	if err != nil {
-		return nil, err
+	if client.txQueue == nil {
+		return nil, errors.New("tx queue not configured")
 	}
 
-	span := trace.SpanFromContext(ctx)
-	span.AddEvent("txclient: broadcasted PFB", trace.WithAttributes(
-		attribute.Int("num_blobs", len(blobs)),
-	))
+	if !client.txQueue.started.Load() {
+		return nil, errors.New("tx queue not started")
+	}
 
-	return client.ConfirmTx(ctx, resp.TxHash)
+	resultsC := make(chan SubmissionResult, 1)
+	defer close(resultsC)
+
+	job := &SubmissionJob{
+		Blobs:    blobs,
+		Options:  opts,
+		Ctx:      ctx,
+		ResultsC: resultsC,
+	}
+
+	client.SubmitJob(job)
+
+	// Block waiting for the result
+	result := <-resultsC
+	if result.Error != nil {
+		return result.TxResponse, result.Error
+	}
+
+	return result.TxResponse, nil
+}
+
+// SubmitJob submits a job to the tx queue for parallel processing
+func (client *TxClient) SubmitJob(job *SubmissionJob) {
+	if client.txQueue != nil {
+		client.txQueue.submitJob(job)
+	} else {
+		job.ResultsC <- SubmissionResult{Error: errors.New("tx queue not configured")}
+	}
 }
 
 // SubmitPayForBlobWithAccount forms a transaction from the provided blobs, signs it with the provided account, and submits it to the chain.
@@ -601,7 +675,7 @@ func (client *TxClient) submitToMultipleConnections(ctx context.Context, txBytes
 	close(errCh)
 
 	// Return first successful response, if any
-	if resp, ok := <-respCh; ok {
+	if resp, ok := <-respCh; ok && resp != nil {
 		client.trackTransaction(signer, resp.TxHash, txBytes)
 
 		if err := client.signer.IncrementSequence(signer); err != nil {
@@ -936,7 +1010,8 @@ func (client *TxClient) getAccountNameFromMsgs(msgs []sdktypes.Msg) (string, err
 	return record.Name, nil
 }
 
-// trackTransaction tracks a transaction in the tx client's local tx tracker.
+// trackTransaction tracks a transaction without acquiring the mutex.
+// This should only be called when the caller already holds the mutex.
 func (client *TxClient) trackTransaction(signer, txHash string, txBytes []byte) {
 	sequence := client.signer.Account(signer).Sequence()
 	client.txTracker[txHash] = txInfo{
@@ -960,6 +1035,56 @@ func (client *TxClient) Signer() *Signer {
 	return client.signer
 }
 
+// StartTxQueueForTest starts the tx queue for testing purposes.
+// This function is only intended for use in tests.
+func (client *TxClient) StartTxQueueForTest(ctx context.Context) error {
+	if client.txQueue == nil {
+		return nil
+	}
+	return client.txQueue.start(ctx)
+}
+
+// StopTxQueueForTest stops the tx queue for testing purposes.
+// This function is only intended for use in tests.
+func (client *TxClient) StopTxQueueForTest() {
+	if client.txQueue != nil {
+		client.txQueue.stop()
+	}
+}
+
+// IsTxQueueStartedForTest returns whether the tx queue is started, for testing purposes.
+// This function is only intended for use in tests.
+func (client *TxClient) IsTxQueueStartedForTest() bool {
+	if client.txQueue == nil {
+		return false
+	}
+	return client.txQueue.isStarted()
+}
+
+// TxQueueWorkerCount returns the number of workers in the tx queue
+func (client *TxClient) TxQueueWorkerCount() int {
+	if client.txQueue == nil {
+		return 0
+	}
+	return len(client.txQueue.workers)
+}
+
+// TxQueueWorkerAddress returns the address for the worker at the given index
+func (client *TxClient) TxQueueWorkerAddress(index int) string {
+	if client.txQueue == nil || index < 0 || index >= len(client.txQueue.workers) {
+		return ""
+	}
+	return client.txQueue.workers[index].address
+}
+
+// TxQueueWorkerAccountName returns the account name for the worker at the given index
+func (client *TxClient) TxQueueWorkerAccountName(index int) string {
+	if client.txQueue == nil || index < 0 || index >= len(client.txQueue.workers) {
+		return ""
+	}
+	return client.txQueue.workers[index].accountName
+}
+
 // QueryMinimumGasPrice queries both the nodes local and network wide
 // minimum gas prices, returning the maximum of the two.
 func QueryMinimumGasPrice(ctx context.Context, grpcConn *grpc.ClientConn) (float64, error) {



Test patch:
diff --git a/pkg/user/e2e_test.go b/pkg/user/e2e_test.go
index b471ae12..bdc102fe 100644
--- a/pkg/user/e2e_test.go
+++ b/pkg/user/e2e_test.go
@@ -1,78 +1,136 @@
 package user_test
 
 import (
-	"context"
-	"errors"
 	"fmt"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/celestiaorg/celestia-app/v6/app"
+	"github.com/celestiaorg/celestia-app/v6/app/encoding"
 	"github.com/celestiaorg/celestia-app/v6/pkg/appconsts"
 	"github.com/celestiaorg/celestia-app/v6/pkg/user"
 	"github.com/celestiaorg/celestia-app/v6/test/util/blobfactory"
 	"github.com/celestiaorg/celestia-app/v6/test/util/random"
 	"github.com/celestiaorg/celestia-app/v6/test/util/testnode"
 	"github.com/celestiaorg/go-square/v3/share"
-	"github.com/cometbft/cometbft/config"
 	"github.com/stretchr/testify/require"
 )
 
-func TestConcurrentTxSubmission(t *testing.T) {
+func TestParallelTxSubmission(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping in short mode")
 	}
 
-	// Iterate over all mempool versions
-	mempools := []string{config.MempoolTypeFlood, config.MempoolTypePriority, config.MempoolTypeCAT}
-	for _, mempool := range mempools {
-		t.Run(fmt.Sprintf("mempool %s", mempool), func(t *testing.T) {
-			// Setup network
-			tmConfig := testnode.DefaultTendermintConfig()
-			tmConfig.Mempool.Type = mempool
-			tmConfig.Consensus.TimeoutCommit = 5 * time.Second
-			ctx, _, _ := testnode.NewNetwork(t, testnode.DefaultConfig().WithTendermintConfig(tmConfig))
-			_, err := ctx.WaitForHeight(1)
-			require.NoError(t, err)
-
-			// Setup signer
-			txClient, err := testnode.NewTxClientFromContext(ctx)
-			require.NoError(t, err)
-
-			// Pregenerate all the blobs
-			numTxs := 100
-			blobs := blobfactory.ManyRandBlobs(random.New(), blobfactory.Repeat(2048, numTxs)...)
-
-			// Prepare transactions
-			var (
-				wg    sync.WaitGroup
-				errCh = make(chan error, 1)
-			)
-
-			subCtx, cancel := context.WithCancel(ctx.GoContext())
-			defer cancel()
-			time.AfterFunc(time.Minute, cancel)
-			for i := range numTxs {
-				wg.Add(1)
-				go func(b *share.Blob) {
-					defer wg.Done()
-					_, err := txClient.SubmitPayForBlob(subCtx, []*share.Blob{b}, user.SetGasLimitAndGasPrice(500_000, appconsts.DefaultMinGasPrice))
-					if err != nil && !errors.Is(err, context.Canceled) {
-						// only catch the first error
-						select {
-						case errCh <- err:
-							cancel()
-						default:
-						}
-					}
-				}(blobs[i])
+	// Setup network
+	tmConfig := testnode.DefaultTendermintConfig()
+	tmConfig.Consensus.TimeoutCommit = 3 * time.Second
+	ctx, _, _ := testnode.NewNetwork(t, testnode.DefaultConfig().WithTendermintConfig(tmConfig))
+	_, err := ctx.WaitForHeight(1)
+	require.NoError(t, err)
+
+	// Setup signer with parallel workers (accounts will be auto-created and started)
+	numWorkers := 5
+	encCfg := encoding.MakeConfig(app.ModuleEncodingRegisters...)
+	txWorkersOpt := user.WithTxWorkers(numWorkers)
+	txClient, err := user.SetupTxClient(ctx.GoContext(), ctx.Keyring, ctx.GRPCClient, encCfg, txWorkersOpt)
+	require.NoError(t, err)
+
+	// Pool should already be started by SetupTxClient
+	require.True(t, txClient.IsTxQueueStartedForTest())
+
+	// Generate test blobs
+	numJobs := 10
+	blobs := blobfactory.ManyRandBlobs(random.New(), blobfactory.Repeat(1024, numJobs)...)
+
+	// Submit jobs in parallel using goroutines
+	var wg sync.WaitGroup
+	errCh := make(chan error, numJobs)
+	for i := range numJobs {
+		wg.Add(1)
+		go func(idx int) { //nolint:contextcheck
+			defer wg.Done()
+			resp, err := txClient.SubmitPayForBlob(ctx.GoContext(), []*share.Blob{blobs[idx]}, user.SetGasLimitAndGasPrice(500_000, appconsts.DefaultMinGasPrice))
+			if err != nil {
+				errCh <- fmt.Errorf("transaction %d failed: %w", idx, err)
+				return
 			}
-			wg.Wait()
-			select {
-			case err := <-errCh:
-				require.NoError(t, err)
-			default:
+			if resp == nil || resp.TxHash == "" {
+				errCh <- fmt.Errorf("transaction %d returned nil or empty response", idx)
 			}
-		})
+		}(i)
+	}
+
+	// Wait for all to complete
+	wg.Wait()
+	close(errCh)
+
+	// Check for any errors
+	for err := range errCh {
+		require.NoError(t, err)
 	}
+
+	t.Logf("Successfully submitted %d parallel transactions", numJobs)
+
+	// Part 2: Test shutdown behavior and verify pool can be stopped properly
+	t.Log("Testing parallel pool shutdown...")
+
+	// Stop the parallel pool
+	txClient.StopTxQueueForTest()
+	require.False(t, txClient.IsTxQueueStartedForTest())
+
+	// Verify that new submissions fail when tx queue is stopped
+	_, err = txClient.SubmitPayForBlob(ctx.GoContext(), []*share.Blob{blobs[0]})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "tx queue not started")
+
+	t.Log("Successfully verified parallel pool shutdown behavior")
+
+	// Part 3: Test restart scenario - create new client and verify existing accounts are reused
+	t.Log("Testing restart scenario with existing worker accounts...")
+
+	// Create a new TxClient using the same keyring (simulating restart)
+	// Use the same default account as the first client to avoid using a worker account as default
+	originalDefaultAccount := txClient.DefaultAccountName()
+	txClient2, err := user.SetupTxClient(ctx.GoContext(), ctx.Keyring, ctx.GRPCClient, encCfg, user.WithTxWorkers(numWorkers*3), user.WithDefaultAccount(originalDefaultAccount))
+	require.NoError(t, err)
+
+	// Tx queue should already be started by SetupTxClient
+	require.True(t, txClient2.IsTxQueueStartedForTest())
+
+	// Submit jobs in parallel using goroutines
+	var wg2 sync.WaitGroup
+	errCh2 := make(chan error, numJobs)
+	for i := range numJobs {
+		wg2.Add(1)
+		go func(idx int) {
+			defer wg2.Done()
+			resp, err := txClient2.SubmitPayForBlob(ctx.GoContext(), []*share.Blob{blobs[idx]}, user.SetGasLimitAndGasPrice(500_000, appconsts.DefaultMinGasPrice))
+			if err != nil {
+				errCh2 <- fmt.Errorf("transaction %d failed: %w", idx, err)
+				return
+			}
+			if resp == nil || resp.TxHash == "" {
+				errCh2 <- fmt.Errorf("transaction %d returned nil or empty response", idx)
+			}
+		}(i)
+	}
+
+	// Wait for all to complete
+	wg2.Wait()
+	close(errCh2)
+
+	// Check for any errors
+	for err := range errCh2 {
+		require.NoError(t, err)
+	}
+
+	// Check that worker accounts exist in keyring before initialization
+	for i := 1; i < numWorkers*3; i++ {
+		accountName := fmt.Sprintf("parallel-worker-%d", i)
+		_, err := ctx.Keyring.Key(accountName)
+		require.NoError(t, err, "worker account %s should exist in keyring", accountName)
+	}
+
+	t.Log("Successfully verified restart scenario - existing worker accounts were reused and no new accounts were created")
 }
diff --git a/pkg/user/mock_tx_server_test.go b/pkg/user/mock_tx_server_test.go
index 89910044..cf270adb 100644
--- a/pkg/user/mock_tx_server_test.go
+++ b/pkg/user/mock_tx_server_test.go
@@ -20,6 +20,8 @@ import (
 // BroadcastHandler is a function type for handling broadcast requests
 type BroadcastHandler func(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error)
 
+type TxStatusHandler func(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error)
+
 // mockTxServer implements both gRPC ServiceServer and TxServer interfaces for mocking broadcast and tx status responses
 type mockTxServer struct {
 	sdktx.UnimplementedServiceServer
@@ -27,10 +29,23 @@ type mockTxServer struct {
 	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
-	broadcastHandler    BroadcastHandler                  // Custom broadcast handler
+
+	// Optional custom handlers - if set, these override default behavior
+	broadcastHandler BroadcastHandler
+	txStatusHandler  TxStatusHandler
 }
 
 func (m *mockTxServer) TxStatus(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error) {
+	return m.txStatusHandler(ctx, req)
+}
+
+func (m *mockTxServer) BroadcastTx(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
+	return m.broadcastHandler(ctx, req)
+}
+
+// defaultTxStatusHandler implements the original default behavior for TxStatus
+func (m *mockTxServer) defaultTxStatusHandler(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error) {
+	// Use predefined response sequences
 	if responses, exists := m.txStatusResponses[req.TxId]; exists {
 		callCount := m.txStatusCallCounts[req.TxId]
 		m.txStatusCallCounts[req.TxId]++
@@ -52,13 +67,10 @@ func (m *mockTxServer) TxStatus(ctx context.Context, req *tx.TxStatusRequest) (*
 	}, nil
 }
 
-func (m *mockTxServer) BroadcastTx(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
-	// Use custom handler if provided
-	if m.broadcastHandler != nil {
-		return m.broadcastHandler(ctx, req)
-	}
-
-	// Default behavior: Same hash for all broadcast calls
+// nolint:unused
+// defaultBroadcastHandler implements the original default behavior for BroadcastTx
+func (m *mockTxServer) defaultBroadcastHandler(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
+	// Default behavior: same hash for all broadcast calls
 	txHash := "test-tx-hash-123"
 
 	// Increment broadcast call count
@@ -88,15 +100,89 @@ func (m *mockTxServer) BroadcastTx(ctx context.Context, req *sdktx.BroadcastTxRe
 	}, nil
 }
 
+// nolint:unused
+// setupTxClientWithMockGRPCServer 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) {
+	// Use default handlers for backward compatibility
+	return setupTxClientWithMockGRPCServerAndHandlers(t, responseSequences, nil, nil, opts...)
+}
+
+// nolint:unused
+// setupTxClientWithMockGRPCServerAndHandlers creates a TxClient with optional custom handlers
+func setupTxClientWithMockGRPCServerAndHandlers(t *testing.T, responseSequences map[string][]*tx.TxStatusResponse, broadcastHandler func(context.Context, *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error), txStatusHandler func(context.Context, *tx.TxStatusRequest) (*tx.TxStatusResponse, error), opts ...user.Option) (*user.TxClient, *grpc.ClientConn) {
+	// Create mock server with provided response sequences and handlers
+	mockServer := &mockTxServer{
+		txStatusResponses:   responseSequences,
+		txStatusCallCounts:  make(map[string]int),
+		broadcastCallCounts: make(map[string]int),
+	}
+
+	// Set handlers: use provided custom handlers or default to original behavior
+	if broadcastHandler != nil {
+		mockServer.broadcastHandler = broadcastHandler
+	} else {
+		mockServer.broadcastHandler = mockServer.defaultBroadcastHandler
+	}
+
+	if txStatusHandler != nil {
+		mockServer.txStatusHandler = txStatusHandler
+	} else {
+		mockServer.txStatusHandler = mockServer.defaultTxStatusHandler
+	}
+
+	// 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
+	conn, err := grpc.NewClient(
+		"passthrough:///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
+}
+
 // createMockServer creates a single mock gRPC server with the given configuration
 func createMockServer(t *testing.T, txStatusResponses map[string][]*tx.TxStatusResponse, broadcastHandler BroadcastHandler) *grpc.ClientConn {
 	mockServer := &mockTxServer{
 		txStatusResponses:   txStatusResponses,
 		txStatusCallCounts:  make(map[string]int),
 		broadcastCallCounts: make(map[string]int),
-		broadcastHandler:    broadcastHandler,
 	}
 
+	// Set handlers: use provided custom handler or default behavior
+	if broadcastHandler != nil {
+		mockServer.broadcastHandler = broadcastHandler
+	} else {
+		mockServer.broadcastHandler = mockServer.defaultBroadcastHandler
+	}
+	mockServer.txStatusHandler = mockServer.defaultTxStatusHandler
+
 	// Set up in-memory gRPC server
 	lis := bufconn.Listen(1024 * 1024)
 	s := grpc.NewServer()
@@ -110,8 +196,8 @@ func createMockServer(t *testing.T, txStatusResponses map[string][]*tx.TxStatusR
 	}()
 
 	// Create client connection
-	//nolint:staticcheck
-	conn, err := grpc.DialContext(context.Background(), "bufnet",
+	conn, err := grpc.NewClient(
+		"passthrough:///bufnet",
 		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
 			return lis.Dial()
 		}),
diff --git a/pkg/user/parallel_tx_submission_test.go b/pkg/user/parallel_tx_submission_test.go
new file mode 100644
index 00000000..3265512f
--- /dev/null
+++ b/pkg/user/parallel_tx_submission_test.go
@@ -0,0 +1,520 @@
+package user_test
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/celestiaorg/celestia-app/v6/app"
+	"github.com/celestiaorg/celestia-app/v6/app/encoding"
+	"github.com/celestiaorg/celestia-app/v6/app/grpc/tx"
+	"github.com/celestiaorg/celestia-app/v6/pkg/user"
+	"github.com/celestiaorg/go-square/v3/share"
+	abci "github.com/cometbft/cometbft/abci/types"
+	"github.com/cometbft/cometbft/rpc/core"
+	"github.com/cosmos/cosmos-sdk/crypto/hd"
+	"github.com/cosmos/cosmos-sdk/crypto/keyring"
+	sdktypes "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/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+)
+
+func newMockTxClientWithCustomHandlers(t *testing.T, broadcastHandler func(context.Context, *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error), txStatusHandler func(context.Context, *tx.TxStatusRequest) (*tx.TxStatusResponse, error), workerAccounts []string) (*user.TxClient, *grpc.ClientConn) {
+	t.Helper()
+
+	encCfg := encoding.MakeConfig(app.ModuleEncodingRegisters...)
+	kr := keyring.NewInMemory(encCfg.Codec)
+
+	accountNames := make([]string, 0, len(workerAccounts)+1)
+	accountNames = append(accountNames, "master")
+	accountNames = append(accountNames, workerAccounts...)
+
+	for i, name := range accountNames {
+		path := hd.CreateHDPath(sdktypes.CoinType, 0, uint32(i)).String()
+		_, _, err := kr.NewMnemonic(name, keyring.English, path, keyring.DefaultBIP39Passphrase, hd.Secp256k1)
+		require.NoError(t, err)
+	}
+
+	accounts := make([]*user.Account, len(accountNames))
+	for i, name := range accountNames {
+		accounts[i] = user.NewAccount(name, uint64(i+1), 0)
+	}
+
+	signer, err := user.NewSigner(kr, encCfg.TxConfig, "test-chain", accounts...)
+	require.NoError(t, err)
+
+	options := []user.Option{
+		user.WithPollTime(10 * time.Millisecond),
+	}
+	if len(workerAccounts) > 0 {
+		// For mock tests, use WithTxWorkers since we're providing mock accounts
+		option := user.WithTxWorkers(len(workerAccounts))
+		options = append(options, option)
+	}
+
+	return setupTxClientWithMockGRPCServerAndSigner(t, make(map[string][]*tx.TxStatusResponse), broadcastHandler, txStatusHandler, encCfg, signer, options...)
+}
+
+func setupTxClientWithMockGRPCServerAndSigner(t *testing.T, responseSequences map[string][]*tx.TxStatusResponse, broadcastHandler func(context.Context, *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error), txStatusHandler func(context.Context, *tx.TxStatusRequest) (*tx.TxStatusResponse, error), encCfg encoding.Config, signer *user.Signer, opts ...user.Option) (*user.TxClient, *grpc.ClientConn) {
+	t.Helper()
+
+	mockServer := &mockTxServer{
+		txStatusResponses:   responseSequences,
+		txStatusCallCounts:  make(map[string]int),
+		broadcastCallCounts: make(map[string]int),
+	}
+
+	// Set handlers: use provided custom handlers or default to original behavior
+	if broadcastHandler != nil {
+		mockServer.broadcastHandler = broadcastHandler
+	} else {
+		mockServer.broadcastHandler = mockServer.defaultBroadcastHandler
+	}
+
+	if txStatusHandler != nil {
+		mockServer.txStatusHandler = txStatusHandler
+	} else {
+		mockServer.txStatusHandler = mockServer.defaultTxStatusHandler
+	}
+
+	lis := bufconn.Listen(1024 * 1024)
+	s := grpc.NewServer()
+	sdktx.RegisterServiceServer(s, mockServer)
+	tx.RegisterTxServer(s, mockServer)
+
+	go func() {
+		if err := s.Serve(lis); err != nil {
+			t.Logf("Server exited with error: %v", err)
+		}
+	}()
+
+	conn, err := grpc.NewClient(
+		"passthrough:///bufnet",
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return lis.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+	require.NoError(t, err)
+
+	mockTxClient, err := user.NewTxClient(
+		encCfg.Codec,
+		signer,
+		conn,
+		encCfg.InterfaceRegistry,
+		opts...,
+	)
+	require.NoError(t, err)
+
+	// Start the tx queue (mirroring SetupTxClient behavior)
+	err = mockTxClient.StartTxQueueForTest(context.Background())
+	require.NoError(t, err)
+
+	return mockTxClient, conn
+}
+
+func randomBlob(t *testing.T) *share.Blob {
+	t.Helper()
+
+	blob, err := share.NewBlob(share.RandomBlobNamespace(), []byte("hello world"), share.ShareVersionZero, nil)
+	require.NoError(t, err)
+	return blob
+}
+
+func hashTxBytes(txBytes []byte) string {
+	sum := sha256.Sum256(txBytes)
+	return strings.ToUpper(hex.EncodeToString(sum[:]))
+}
+
+func TestParallelSubmitPayForBlobSuccess(t *testing.T) {
+	t.Parallel()
+
+	statusStore := make(map[string]*tx.TxStatusResponse)
+	var mu sync.Mutex
+
+	broadcastHandler := func(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
+		hash := hashTxBytes(req.TxBytes)
+
+		mu.Lock()
+		statusStore[hash] = &tx.TxStatusResponse{
+			Height:        33,
+			ExecutionCode: abci.CodeTypeOK,
+			Status:        core.TxStatusCommitted,
+		}
+		mu.Unlock()
+
+		return &sdktx.BroadcastTxResponse{
+			TxResponse: &sdktypes.TxResponse{
+				Code:   abci.CodeTypeOK,
+				TxHash: hash,
+			},
+		}, nil
+	}
+	txStatusHandler := func(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error) {
+		mu.Lock()
+		defer mu.Unlock()
+
+		if resp, ok := statusStore[strings.ToUpper(req.TxId)]; ok {
+			return resp, nil
+		}
+
+		return &tx.TxStatusResponse{Status: core.TxStatusPending}, nil
+	}
+
+	workerAccounts := []string{"parallel-worker-1", "parallel-worker-2"}
+	client, conn := newMockTxClientWithCustomHandlers(t, broadcastHandler, txStatusHandler, workerAccounts)
+	defer conn.Close()
+	require.True(t, client.TxQueueWorkerCount() > 0)
+
+	blob := randomBlob(t)
+
+	jobCount := 3
+	results := make([]user.SubmissionResult, 0, jobCount)
+	var wg sync.WaitGroup
+	var resultsMu sync.Mutex
+
+	// Submit all jobs in parallel
+	for range jobCount {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			resp, err := client.SubmitPayForBlob(context.Background(), []*share.Blob{blob})
+			resultsMu.Lock()
+			defer resultsMu.Unlock()
+			if err != nil {
+				results = append(results, user.SubmissionResult{Error: err})
+			} else {
+				results = append(results, user.SubmissionResult{TxResponse: resp})
+			}
+		}()
+	}
+
+	// Wait for all to complete
+	wg.Wait()
+
+	// Verify results
+	for _, result := range results {
+		require.NoError(t, result.Error)
+		require.NotNil(t, result.TxResponse)
+		require.Equal(t, abci.CodeTypeOK, result.TxResponse.Code)
+	}
+
+	var totalSequence uint64
+	for i := 0; i < client.TxQueueWorkerCount(); i++ {
+		accountName := client.TxQueueWorkerAccountName(i)
+		totalSequence += client.Signer().Account(accountName).Sequence()
+	}
+	require.Equal(t, uint64(len(results)), totalSequence)
+}
+
+func TestParallelSubmitPayForBlobBroadcastError(t *testing.T) {
+	t.Parallel()
+
+	broadcastHandler := func(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
+		hash := hashTxBytes(req.TxBytes)
+		return &sdktx.BroadcastTxResponse{
+			TxResponse: &sdktypes.TxResponse{
+				Code:   5,
+				TxHash: hash,
+				RawLog: "insufficient funds",
+			},
+		}, nil
+	}
+	txStatusHandler := func(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error) {
+		return &tx.TxStatusResponse{Status: core.TxStatusCommitted}, nil
+	}
+
+	workerAccounts := []string{"parallel-worker-1"}
+	client, conn := newMockTxClientWithCustomHandlers(t, broadcastHandler, txStatusHandler, workerAccounts)
+	defer conn.Close()
+	require.True(t, client.TxQueueWorkerCount() > 0)
+
+	blob := randomBlob(t)
+
+	_, err := client.SubmitPayForBlob(context.Background(), []*share.Blob{blob})
+	require.Error(t, err)
+
+	var broadcastErr *user.BroadcastTxError
+	require.ErrorAs(t, err, &broadcastErr)
+	require.Equal(t, uint32(5), broadcastErr.Code)
+
+	for i := 0; i < client.TxQueueWorkerCount(); i++ {
+		accountName := client.TxQueueWorkerAccountName(i)
+		require.Equal(t, uint64(0), client.Signer().Account(accountName).Sequence())
+	}
+}
+
+func TestParallelSubmissionSignerAddress(t *testing.T) {
+	t.Parallel()
+
+	statusStore := make(map[string]*tx.TxStatusResponse)
+	var mu sync.Mutex
+
+	broadcastHandler := func(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
+		hash := hashTxBytes(req.TxBytes)
+
+		mu.Lock()
+		statusStore[hash] = &tx.TxStatusResponse{
+			Height:        33,
+			ExecutionCode: abci.CodeTypeOK,
+			Status:        core.TxStatusCommitted,
+		}
+		mu.Unlock()
+
+		return &sdktx.BroadcastTxResponse{
+			TxResponse: &sdktypes.TxResponse{
+				Code:   abci.CodeTypeOK,
+				TxHash: hash,
+			},
+		}, nil
+	}
+	txStatusHandler := func(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error) {
+		mu.Lock()
+		defer mu.Unlock()
+
+		if resp, ok := statusStore[strings.ToUpper(req.TxId)]; ok {
+			return resp, nil
+		}
+
+		return &tx.TxStatusResponse{Status: core.TxStatusPending}, nil
+	}
+
+	workerAccounts := []string{"parallel-worker-1", "parallel-worker-2"}
+	client, conn := newMockTxClientWithCustomHandlers(t, broadcastHandler, txStatusHandler, workerAccounts)
+	defer conn.Close()
+	require.True(t, client.TxQueueWorkerCount() > 0)
+
+	blob := randomBlob(t)
+
+	// Submit multiple jobs to test different workers
+	jobCount := 4
+	results := make([]user.SubmissionResult, 0, jobCount)
+	var wg sync.WaitGroup
+	var resultsMu sync.Mutex
+
+	// Submit all jobs in parallel
+	for range jobCount {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			resp, err := client.SubmitPayForBlob(context.Background(), []*share.Blob{blob})
+
+			resultsMu.Lock()
+			defer resultsMu.Unlock()
+			if err != nil {
+				results = append(results, user.SubmissionResult{Error: err})
+			} else {
+				// Get the signer from the worker - for this test we need to track which worker was used
+				// Since we can't easily get this from the response, we'll just verify the responses are valid
+				results = append(results, user.SubmissionResult{TxResponse: resp})
+			}
+		}()
+	}
+
+	// Wait for all to complete
+	wg.Wait()
+
+	// Verify all results
+	for _, result := range results {
+		require.NoError(t, result.Error)
+		require.NotNil(t, result.TxResponse)
+		require.Equal(t, abci.CodeTypeOK, result.TxResponse.Code)
+	}
+
+	// Verify that multiple workers were used by checking sequence numbers
+	var totalSequence uint64
+	for i := 0; i < client.TxQueueWorkerCount(); i++ {
+		accountName := client.TxQueueWorkerAccountName(i)
+		totalSequence += client.Signer().Account(accountName).Sequence()
+	}
+	require.Equal(t, uint64(len(results)), totalSequence)
+}
+
+func TestParallelPoolRestart(t *testing.T) {
+	t.Parallel()
+
+	statusStore := make(map[string]*tx.TxStatusResponse)
+	var mu sync.Mutex
+
+	broadcastHandler := func(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
+		hash := hashTxBytes(req.TxBytes)
+
+		mu.Lock()
+		statusStore[hash] = &tx.TxStatusResponse{
+			Height:        12,
+			ExecutionCode: abci.CodeTypeOK,
+			Status:        core.TxStatusCommitted,
+		}
+		mu.Unlock()
+
+		return &sdktx.BroadcastTxResponse{
+			TxResponse: &sdktypes.TxResponse{
+				Code:   abci.CodeTypeOK,
+				TxHash: hash,
+			},
+		}, nil
+	}
+
+	txStatusHandler := func(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		if resp, ok := statusStore[strings.ToUpper(req.TxId)]; ok {
+			return resp, nil
+		}
+		return &tx.TxStatusResponse{Status: core.TxStatusPending}, nil
+	}
+
+	workerAccounts := []string{"parallel-worker-1"}
+	client, conn := newMockTxClientWithCustomHandlers(t, broadcastHandler, txStatusHandler, workerAccounts)
+	defer conn.Close()
+
+	require.True(t, client.TxQueueWorkerCount() > 0)
+
+	ctx := context.Background()
+	require.True(t, client.IsTxQueueStartedForTest())
+
+	blob := randomBlob(t)
+
+	submitAndAssert := func(t *testing.T) {
+		t.Helper()
+
+		resp, err := client.SubmitPayForBlob(ctx, []*share.Blob{blob})
+		require.NoError(t, err)
+		require.NotNil(t, resp)
+		require.Equal(t, abci.CodeTypeOK, resp.Code)
+	}
+
+	submitAndAssert(t)
+
+	client.StopTxQueueForTest()
+	require.False(t, client.IsTxQueueStartedForTest())
+
+	require.NoError(t, client.StartTxQueueForTest(ctx))
+	require.True(t, client.IsTxQueueStartedForTest())
+
+	submitAndAssert(t)
+}
+
+func TestParallelSubmitPayForBlobContextCancellation(t *testing.T) {
+	t.Parallel()
+
+	startedCh := make(chan struct{})
+	broadcastHandler := func(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
+		close(startedCh)
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+
+	txStatusHandler := func(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error) {
+		return &tx.TxStatusResponse{Status: core.TxStatusPending}, nil
+	}
+
+	workerAccounts := []string{"parallel-worker-1"}
+	client, conn := newMockTxClientWithCustomHandlers(t, broadcastHandler, txStatusHandler, workerAccounts)
+	defer conn.Close()
+
+	require.True(t, client.TxQueueWorkerCount() > 0)
+
+	blob := randomBlob(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	errCh := make(chan error, 1)
+	go func() {
+		_, err := client.SubmitPayForBlob(ctx, []*share.Blob{blob})
+		errCh <- err
+	}()
+
+	select {
+	case <-startedCh:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("broadcast handler was not invoked")
+	}
+
+	cancel()
+
+	select {
+	case err := <-errCh:
+		require.Error(t, err)
+		st, ok := status.FromError(err)
+		require.True(t, ok, "expected gRPC status error")
+		require.Equal(t, codes.Canceled, st.Code())
+	case <-time.After(5 * time.Second):
+		t.Fatalf("timeout waiting for cancellation result")
+	}
+}
+
+func TestSingleWorkerNoFeeGranter(t *testing.T) {
+	t.Parallel()
+
+	statusStore := make(map[string]*tx.TxStatusResponse)
+	var mu sync.Mutex
+	var capturedTx sdktypes.Tx
+
+	encCfg := encoding.MakeConfig(app.ModuleEncodingRegisters...)
+
+	broadcastHandler := func(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
+		hash := hashTxBytes(req.TxBytes)
+
+		// Decode and capture the transaction
+		decodedTx, err := encCfg.TxConfig.TxDecoder()(req.TxBytes)
+		if err != nil {
+			return nil, err
+		}
+		capturedTx = decodedTx
+
+		mu.Lock()
+		statusStore[hash] = &tx.TxStatusResponse{
+			Height:        100,
+			ExecutionCode: abci.CodeTypeOK,
+			Status:        core.TxStatusCommitted,
+		}
+		mu.Unlock()
+
+		return &sdktx.BroadcastTxResponse{
+			TxResponse: &sdktypes.TxResponse{
+				Code:   abci.CodeTypeOK,
+				TxHash: hash,
+			},
+		}, nil
+	}
+	txStatusHandler := func(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error) {
+		mu.Lock()
+		defer mu.Unlock()
+
+		if resp, ok := statusStore[strings.ToUpper(req.TxId)]; ok {
+			return resp, nil
+		}
+
+		return &tx.TxStatusResponse{Status: core.TxStatusPending}, nil
+	}
+
+	// Create client with default single worker (no WithTxWorkers option)
+	client, conn := newMockTxClientWithCustomHandlers(t, broadcastHandler, txStatusHandler, nil)
+	defer conn.Close()
+	require.Equal(t, 1, client.TxQueueWorkerCount())
+
+	blob := randomBlob(t)
+
+	// Submit via the new SubmitPayForBlob which uses the parallel pool
+	resp, err := client.SubmitPayForBlob(context.Background(), []*share.Blob{blob})
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	require.Equal(t, abci.CodeTypeOK, resp.Code)
+
+	// Verify that no fee granter was set in the transaction
+	feeTx, ok := capturedTx.(sdktypes.FeeTx)
+	require.True(t, ok, "Transaction should implement FeeTx interface")
+	require.Nil(t, feeTx.FeeGranter(), "Single worker should not use fee granter")
+}
diff --git a/pkg/user/tx_client_test.go b/pkg/user/tx_client_test.go
index 8ed16a27..7582b4f9 100644
--- a/pkg/user/tx_client_test.go
+++ b/pkg/user/tx_client_test.go
@@ -406,7 +406,7 @@ func TestEvictions(t *testing.T) {
 			},
 		}
 
-		mockTxClient, conns := setupTxClientWithMockServers(t, []BroadcastHandler{nil}, responseSequences, user.WithPollTime(1*time.Second))
+		mockTxClient, conns := setupTxClientWithMockServers(t, []BroadcastHandler{nil}, responseSequences, user.WithPollTime(100*time.Millisecond))
 		defer conns[0].Close()
 
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
@@ -579,7 +579,7 @@ func setupTxClient(
 	testnodeConfig := testnode.DefaultConfig().
 		WithTendermintConfig(defaultTmConfig).
 		WithFundedAccounts(accounts...).
-		WithDelayedPrecommitTimeout(400 * time.Millisecond)
+		WithDelayedPrecommitTimeout(300 * time.Millisecond)
 	testnodeConfig.Genesis.ConsensusParams.Block.MaxBytes = blocksize
 
 	ctx, _, _ := testnode.NewNetwork(t, testnodeConfig)
diff --git a/pkg/user/worker_account_test.go b/pkg/user/worker_account_test.go
new file mode 100644
index 00000000..f7a3ae47
--- /dev/null
+++ b/pkg/user/worker_account_test.go
@@ -0,0 +1,156 @@
+package user_test
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/celestiaorg/celestia-app/v6/app"
+	"github.com/celestiaorg/celestia-app/v6/app/encoding"
+	"github.com/celestiaorg/celestia-app/v6/pkg/user"
+	"github.com/cosmos/cosmos-sdk/crypto/hd"
+	"github.com/cosmos/cosmos-sdk/crypto/keyring"
+	sdktypes "github.com/cosmos/cosmos-sdk/types"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/test/bufconn"
+)
+
+// testSetup holds common test infrastructure
+type testSetup struct {
+	encCfg encoding.Config
+	kr     keyring.Keyring
+	signer *user.Signer
+	conn   *grpc.ClientConn
+}
+
+// newTestSetup creates common test infrastructure with the specified accounts
+func newTestSetup(t *testing.T, accountNames []string) *testSetup {
+	t.Helper()
+
+	encCfg := encoding.MakeConfig(app.ModuleEncodingRegisters...)
+	kr := keyring.NewInMemory(encCfg.Codec)
+
+	accounts := make([]*user.Account, len(accountNames))
+	for i, name := range accountNames {
+		path := hd.CreateHDPath(sdktypes.CoinType, 0, uint32(i)).String()
+		_, _, err := kr.NewMnemonic(name, keyring.English, path, keyring.DefaultBIP39Passphrase, hd.Secp256k1)
+		require.NoError(t, err)
+		accounts[i] = user.NewAccount(name, uint64(i+1), 0)
+	}
+
+	signer, err := user.NewSigner(kr, encCfg.TxConfig, "test-chain", accounts...)
+	require.NoError(t, err)
+
+	// Create mock GRPC connection
+	listener := bufconn.Listen(1024 * 1024)
+	conn, err := grpc.NewClient("", grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+		return listener.Dial()
+	}), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+
+	return &testSetup{
+		encCfg: encCfg,
+		kr:     kr,
+		signer: signer,
+		conn:   conn,
+	}
+}
+
+// Close cleans up test resources
+func (ts *testSetup) Close() {
+	ts.conn.Close()
+}
+
+// newTxClientWithWorkers creates a TxClient with parallel workers using the specified default account
+func (ts *testSetup) newTxClientWithWorkers(t *testing.T, numWorkers int, defaultAccount string) *user.TxClient {
+	t.Helper()
+
+	txWorkersOpt := user.WithTxWorkers(numWorkers)
+	client, err := user.NewTxClient(ts.encCfg.Codec, ts.signer, ts.conn, ts.encCfg.InterfaceRegistry,
+		user.WithDefaultAccount(defaultAccount), txWorkersOpt)
+	require.NoError(t, err)
+
+	return client
+}
+
+func TestWorkerZeroAlwaysUsesMainAccount(t *testing.T) {
+	t.Parallel()
+
+	customAccountName := "my-custom-main-account"
+	setup := newTestSetup(t, []string{customAccountName})
+	defer setup.Close()
+
+	// Test with different numbers of workers - worker 0 should always use main account
+	testCases := []struct {
+		name       string
+		numWorkers int
+	}{
+		{"Single Worker", 1},
+		{"Two Workers", 2},
+		{"Five Workers", 5},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			client := setup.newTxClientWithWorkers(t, tc.numWorkers, customAccountName)
+
+			// Verify client uses our custom account as default
+			require.Equal(t, customAccountName, client.DefaultAccountName())
+
+			// Verify tx queue exists
+			require.True(t, client.TxQueueWorkerCount() > 0)
+
+			// Get all workers
+			require.Equal(t, tc.numWorkers, client.TxQueueWorkerCount())
+
+			// CRITICAL TEST: Worker 0 must ALWAYS use the main account passed to TxClient
+			worker0Name := client.TxQueueWorkerAccountName(0)
+			require.Equal(t, customAccountName, worker0Name,
+				"Worker 0 must always use the main account passed to TxClient")
+			require.Equal(t, client.DefaultAccountName(), worker0Name,
+				"Worker 0 must use the same account as client.DefaultAccountName()")
+
+			// For multiple workers, verify additional workers use generated names
+			if tc.numWorkers > 1 {
+				for i := 1; i < tc.numWorkers; i++ {
+					expectedName := "parallel-worker-" + string(rune('0'+i))
+					require.Equal(t, expectedName, client.TxQueueWorkerAccountName(i),
+						"Additional workers should use generated names")
+					require.NotEqual(t, customAccountName, client.TxQueueWorkerAccountName(i),
+						"Additional workers should NOT use the main account")
+				}
+			}
+		})
+	}
+}
+
+func TestWorkerZeroWithDifferentDefaultAccounts(t *testing.T) {
+	t.Parallel()
+
+	accountNames := []string{"account-alpha", "account-beta", "account-gamma"}
+	setup := newTestSetup(t, accountNames)
+	defer setup.Close()
+
+	// Test that worker 0 uses whichever account is set as default
+	for _, defaultAccount := range accountNames {
+		t.Run("DefaultAccount_"+defaultAccount, func(t *testing.T) {
+			client := setup.newTxClientWithWorkers(t, 3, defaultAccount)
+
+			// Verify the client uses the specified default account
+			require.Equal(t, defaultAccount, client.DefaultAccountName())
+
+			// Get workers
+			require.Equal(t, 3, client.TxQueueWorkerCount())
+
+			// CRITICAL: Worker 0 must use whatever account is configured as default
+			require.Equal(t, defaultAccount, client.TxQueueWorkerAccountName(0),
+				"Worker 0 must always use the configured default account (%s)", defaultAccount)
+
+			// Other workers use generated names
+			require.Equal(t, "parallel-worker-1", client.TxQueueWorkerAccountName(1))
+			require.Equal(t, "parallel-worker-2", client.TxQueueWorkerAccountName(2))
+		})
+	}
+}
diff --git a/test/util/grpctest/mocks.go b/test/util/grpctest/mocks.go
index 204e56c7..2d93a91b 100644
--- a/test/util/grpctest/mocks.go
+++ b/test/util/grpctest/mocks.go
@@ -8,6 +8,7 @@ import (
 	"net"
 	"testing"
 
+	"github.com/celestiaorg/celestia-app/v6/app/grpc/tx"
 	"github.com/celestiaorg/celestia-app/v6/test/util/testnode"
 	sdktx "github.com/cosmos/cosmos-sdk/types/tx"
 	"github.com/stretchr/testify/require"
@@ -18,8 +19,10 @@ import (
 // MockTxService allows controlling the behavior of BroadcastTx calls.
 type MockTxService struct {
 	sdktx.UnimplementedServiceServer // Embed the unimplemented server
+	tx.UnimplementedTxServer         // Embed the unimplemented server
 
 	BroadcastHandler func(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error)
+	TxStatusHandler  func(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error)
 }
 
 func (m *MockTxService) BroadcastTx(ctx context.Context, req *sdktx.BroadcastTxRequest) (*sdktx.BroadcastTxResponse, error) {
@@ -29,6 +32,13 @@ func (m *MockTxService) BroadcastTx(ctx context.Context, req *sdktx.BroadcastTxR
 	return nil, fmt.Errorf("MockTxService.BroadcastHandler not set")
 }
 
+func (m *MockTxService) TxStatus(ctx context.Context, req *tx.TxStatusRequest) (*tx.TxStatusResponse, error) {
+	if m.TxStatusHandler != nil {
+		return m.TxStatusHandler(ctx, req)
+	}
+	return nil, fmt.Errorf("MockTxService.TxStatusHandler not set")
+}
+
 func (m *MockTxService) Simulate(context.Context, *sdktx.SimulateRequest) (*sdktx.SimulateResponse, error) {
 	return nil, errors.New("Simulate not implemented in mock")
 }
@@ -76,6 +86,7 @@ func StartMockServer(t *testing.T, service *MockTxService) *grpc.ClientConn {
 
 	s := grpc.NewServer()
 	sdktx.RegisterServiceServer(s, service)
+	tx.RegisterTxServer(s, service)
 
 	go func() {
 		if err := s.Serve(lis); err != nil && !errors.Is(err, grpc.ErrServerStopped) {