feat: split mutex for each account

go.mod

@@ -32,6 +32,7 @@ require (
 	github.com/gorilla/mux v1.8.1
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0
 	github.com/hashicorp/go-metrics v0.5.3
+	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/holiman/uint256 v1.3.0
 	github.com/initia-labs/OPinit v0.4.1
 	github.com/initia-labs/initia v0.4.1
@@ -163,7 +164,6 @@ require (
 	github.com/hashicorp/go-safetemp v1.0.0 // indirect
 	github.com/hashicorp/go-version v1.7.0 // indirect
 	github.com/hashicorp/golang-lru v1.0.2 // indirect
-	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/hashicorp/yamux v0.1.1 // indirect
 	github.com/hdevalence/ed25519consensus v0.1.0 // indirect

jsonrpc/backend/backend.go

@@ -5,6 +5,7 @@
 	"sync"
 
 	bigcache "github.com/allegro/bigcache/v3"
+	lrucache "github.com/hashicorp/golang-lru/v2"
 
 	"cosmossdk.io/log"
 	"github.com/cosmos/cosmos-sdk/client"
@@ -19,7 +20,7 @@
 	logger log.Logger
 
 	queuedTxs *bigcache.BigCache
-	sendTxMut sync.Mutex
+	accMuts   *lrucache.Cache[string, *sync.Mutex]
 
 	ctx       context.Context
 	svrCtx    *server.Context
@@ -54,13 +55,22 @@
 		return nil, err
 	}
 
+	// support concurrent 100 accounts mutex
+	accMuts, err := lrucache.New[string, *sync.Mutex](100)
+	if err != nil {
+		return nil, err
+	}
+
 	return &JSONRPCBackend{
-		app:       app,
-		logger:    logger,
+		app:    app,
+		logger: logger,
+
 		queuedTxs: queuedTxs,
+		accMuts:   accMuts,
+
 		ctx:       ctx,
 		svrCtx:    svrCtx,
 		clientCtx: clientCtx,
 		cfg:       cfg,
 	}, nil
 }

jsonrpc/backend/tx.go

@@ -4,6 +4,7 @@
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 
 	"cosmossdk.io/collections"
 	sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
@@ -36,9 +37,6 @@
 }
 
 func (b *JSONRPCBackend) SendTx(tx *coretypes.Transaction) error {
-	b.sendTxMut.Lock()
-	defer b.sendTxMut.Unlock()
-
 	queryCtx, err := b.getQueryCtx()
 	if err != nil {
 		return err
@@ -70,28 +68,39 @@
 	accSeq := uint64(0)
 	sender := sdk.AccAddress(sig.PubKey.Address().Bytes())
 
+	// hold mutex for each sender
+	senderHex := hexutil.Encode(sender.Bytes())
+	accMut, ok := b.accMuts.Get(senderHex)
+	if !ok {
+		accMut = &sync.Mutex{}
+		_ = b.accMuts.Add(senderHex, accMut)
+	}
+
+	accMut.Lock()
+	defer accMut.Unlock()
+
 	checkCtx := b.app.GetContextForCheckTx(nil)
 	if acc := b.app.AccountKeeper.GetAccount(checkCtx, sender); acc != nil {
 		accSeq = acc.GetSequence()
 	}
 
-	b.logger.Debug("enqueue tx", "sender", sender, "txSeq", txSeq, "accSeq", accSeq)
-	cacheKey := fmt.Sprintf("%X-%d", sender.Bytes(), txSeq)
+	b.logger.Debug("enqueue tx", "sender", senderHex, "txSeq", txSeq, "accSeq", accSeq)
+	cacheKey := fmt.Sprintf("%s-%d", senderHex, txSeq)
 	if err := b.queuedTxs.Set(cacheKey, txBytes); err != nil {
 		b.logger.Error("failed to enqueue tx", "key", cacheKey, "err", err)
 		return NewInternalError("failed to enqueue tx")
 	}
 
 	// check if there are queued txs which can be sent
 	for {
-		cacheKey := fmt.Sprintf("%X-%d", sender.Bytes(), accSeq)
+		cacheKey := fmt.Sprintf("%s-%d", senderHex, accSeq)
 		if txBytes, err := b.queuedTxs.Get(cacheKey); err == nil {
 			if err := b.queuedTxs.Delete(cacheKey); err != nil {
 				b.logger.Error("failed to delete queued tx", "key", cacheKey, "err", err)
 				return NewInternalError("failed to delete queued tx")
 			}
 
-			b.logger.Debug("broadcast queued tx", "sender", sender, "txSeq", accSeq)
+			b.logger.Debug("broadcast queued tx", "sender", senderHex, "txSeq", accSeq)
 			res, err := b.clientCtx.BroadcastTxSync(txBytes)
 			if err != nil {
 				return err