Merge pull request #23 from 0glabs/mempool

add bridge to sync tx replacement events between app mempool and cometbft mempool during checkTx

Author: Solovyov1796

baseapp/abci.go

@@ -376,19 +376,35 @@ func (app *BaseApp) CheckTx(req abci.RequestCheckTx) abci.ResponseCheckTx {
 		return sdkerrors.ResponseCheckTxWithEvents(err, gInfo.GasWanted, gInfo.GasUsed, anteEvents, app.trace)
 	}
 
-	return abci.ResponseCheckTx{
-		GasWanted:     int64(gInfo.GasWanted), // TODO: Should type accept unsigned ints?
-		GasUsed:       int64(gInfo.GasUsed),   // TODO: Should type accept unsigned ints?
-		Log:           result.Log,
-		Data:          result.Data,
-		Events:        sdk.MarkEventsToIndex(result.Events, app.indexEvents),
-		Priority:      priority,
-		SignerAddress: txInfo.SignerAddress,
-		Nonce:         txInfo.Nonce,
-		GasLimit:      txInfo.GasLimit,
-		GasPrice:      txInfo.GasPrice,
-		Type:          txInfo.Type,
+	resp := abci.ResponseCheckTx{
+		GasWanted: int64(gInfo.GasWanted), // TODO: Should type accept unsigned ints?
+		GasUsed:   int64(gInfo.GasUsed),   // TODO: Should type accept unsigned ints?
+		Log:       result.Log,
+		Data:      result.Data,
+		Events:    sdk.MarkEventsToIndex(result.Events, app.indexEvents),
+		Priority:  priority,
+	}
+
+	txHash := genTxHash(req.Tx)
+
+	app.mempoolSyncLock.Lock()
+	if _, exists := app.mempoolSyncCache[txHash]; exists {
+		replacedTx := app.mempoolSyncCache[txHash]
+		resp.ReplacedTx = make([]byte, len(replacedTx))
+		copy(resp.ReplacedTx, replacedTx)
+		delete(app.mempoolSyncCache, txHash)
+	}
+	app.mempoolSyncLock.Unlock()
+
+	if txInfo != nil {
+		resp.SignerAddress = txInfo.SignerAddress
+		resp.Nonce = txInfo.Nonce
+		resp.GasLimit = txInfo.GasLimit
+		resp.GasPrice = txInfo.GasPrice
+		resp.Type = txInfo.TxType
 	}
+
+	return resp
 }
 
 // DeliverTx implements the ABCI interface and executes a tx in DeliverTx mode.

baseapp/baseapp.go

@@ -1,10 +1,14 @@
 package baseapp
 
 import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"sort"
 	"strings"
+	"sync"
 
 	dbm "github.com/cometbft/cometbft-db"
 	abci "github.com/cometbft/cometbft/abci/types"
@@ -145,6 +149,9 @@ type BaseApp struct { //nolint: maligned
 	abciListeners []ABCIListener
 
 	chainID string
+
+	mempoolSyncLock  sync.RWMutex
+	mempoolSyncCache map[string][]byte
 }
 
 // NewBaseApp returns a reference to an initialized BaseApp. It accepts a
@@ -165,6 +172,7 @@ func NewBaseApp(
 		msgServiceRouter: NewMsgServiceRouter(),
 		txDecoder:        txDecoder,
 		fauxMerkleMode:   false,
+		mempoolSyncCache: make(map[string][]byte),
 	}
 
 	for _, option := range options {
@@ -727,6 +735,13 @@ func (app *BaseApp) runTx(mode runTxMode, txBytes []byte) (gInfo sdk.GasInfo, re
 	}
 
 	if mode == runTxModeCheck {
+		if app.txInfoExtracter != nil {
+			txInfo, err = app.txInfoExtracter(ctx, tx)
+			if err != nil {
+				return gInfo, nil, anteEvents, priority, nil, err
+			}
+		}
+
 		err = app.mempool.Insert(ctx, tx)
 		if err != nil {
 			return gInfo, nil, anteEvents, priority, nil, err
@@ -779,13 +794,6 @@ func (app *BaseApp) runTx(mode runTxMode, txBytes []byte) (gInfo sdk.GasInfo, re
 		}
 	}
 
-	if mode == runTxModeCheck && app.txInfoExtracter != nil {
-		txInfo, err = app.txInfoExtracter(ctx, tx)
-		if err != nil {
-			return gInfo, nil, anteEvents, priority, nil, err
-		}
-	}
-
 	return gInfo, result, anteEvents, priority, txInfo, err
 }
 
@@ -923,3 +931,31 @@ func (app *BaseApp) ProcessProposalVerifyTx(txBz []byte) (sdk.Tx, error) {
 func (app *BaseApp) Close() error {
 	return nil
 }
+
+func (app *BaseApp) RegisterMempoolTxReplacedEvent(ctx context.Context, oldTx, newTx sdk.Tx) error {
+	newRawTx, err := app.txEncoder(newTx)
+	if err != nil {
+		return err
+	}
+
+	oldRawTx, err := app.txEncoder(oldTx)
+	if err != nil {
+		return err
+	}
+
+	newRawTxHash := genTxHash(newRawTx)
+
+	app.mempoolSyncLock.Lock()
+	app.mempoolSyncCache[newRawTxHash] = oldRawTx
+	app.mempoolSyncLock.Unlock()
+
+	oldRawTxHash := genTxHash(oldRawTx)
+	app.Logger().Debug("Mempool tx replaced event registered", "new_tx_hash", newRawTxHash, "old_tx_hash", oldRawTxHash)
+
+	return nil
+}
+
+func genTxHash(tx []byte) string {
+	hash := sha256.Sum256(tx)
+	return hex.EncodeToString(hash[:])
+}

types/mempool/mempool.go

@@ -25,6 +25,16 @@ type Mempool interface {
 	Remove(sdk.Tx) error
 }
 
+type MempoolExt interface {
+	Mempool
+
+	// SelectBy use callback to iterate over the mempool, it's thread-safe to use.
+	SelectBy(context.Context, [][]byte, func(sdk.Tx) bool)
+
+	// GetSenderUncommittedTxsCount returns the number of uncommitted transactions for a given sender.
+	GetSenderUncommittedTxnCount(ctx context.Context, sender string) int
+}
+
 // Iterator defines an app-side mempool iterator interface that is as minimal as possible.  The order of iteration
 // is determined by the app-side mempool implementation.
 type Iterator interface {
@@ -39,3 +49,23 @@ var (
 	ErrTxNotFound           = errors.New("tx not found in mempool")
 	ErrMempoolTxMaxCapacity = errors.New("pool reached max tx capacity")
 )
+
+func SelectBy(ctx context.Context, mempool Mempool, txs [][]byte, callback func(sdk.Tx) bool) {
+	if ext, ok := mempool.(MempoolExt); ok {
+		ext.SelectBy(ctx, txs, callback)
+		return
+	}
+
+	// fallback to old behavior, without holding the lock while iteration.
+	iter := mempool.Select(ctx, txs)
+	for iter != nil && callback(iter.Tx()) {
+		iter = iter.Next()
+	}
+}
+
+func GetSenderUncommittedTxnCount(ctx context.Context, mempool Mempool, sender string) int {
+	if ext, ok := mempool.(MempoolExt); ok {
+		return ext.GetSenderUncommittedTxnCount(ctx, sender)
+	}
+	return 0
+}

types/tx_msg.go

@@ -77,7 +77,7 @@ type (
 		Nonce         uint64
 		GasLimit      uint64
 		GasPrice      uint64
-		Type          int32
+		TxType        int32
 	}
 
 	TxInfoExtracter func(ctx Context, tx Tx) (*TxInfo, error)

x/auth/ante/validator_tx_fee.go

@@ -45,22 +45,42 @@ func checkTxFeeWithValidatorMinGasPrices(ctx sdk.Context, tx sdk.Tx) (sdk.Coins,
 	return feeCoins, priority, nil
 }
 
+const (
+	chainBaseDenom = "neuron"
+	cosmosGasDenom = "ua0gi"
+	standardDenom  = "a0gi"
+
+	cosmosGasDenomConversionMultiplier = 1e12
+	standardDenomConversionMultiplier  = 1e18
+)
+
 // getTxPriority returns a naive tx priority based on the amount of the smallest denomination of the gas price
 // provided in a transaction.
 // NOTE: This implementation should be used with a great consideration as it opens potential attack vectors
 // where txs with multiple coins could not be prioritize as expected.
-func getTxPriority(fee sdk.Coins, gas int64) int64 {
+func getTxPriority(fee sdk.Coins, gasLimit int64) int64 {
 	var priority int64
+
+	totalFeeInBaseDenom := sdk.NewInt(0)
+
 	for _, c := range fee {
-		p := int64(math.MaxInt64)
-		gasPrice := c.Amount.QuoRaw(gas)
-		if gasPrice.IsInt64() {
-			p = gasPrice.Int64()
-		}
-		if priority == 0 || p < priority {
-			priority = p
+		if c.Denom == chainBaseDenom {
+			totalFeeInBaseDenom = totalFeeInBaseDenom.Add(c.Amount)
+		} else if c.Denom == cosmosGasDenom {
+			totalFeeInBaseDenom = totalFeeInBaseDenom.Add(c.Amount.MulRaw(cosmosGasDenomConversionMultiplier))
+		} else if c.Denom == standardDenom {
+			totalFeeInBaseDenom = totalFeeInBaseDenom.Add(c.Amount.MulRaw(standardDenomConversionMultiplier))
+		} else {
+			// ignore other denominations
 		}
 	}
 
+	gasPrice := totalFeeInBaseDenom.QuoRaw(gasLimit)
+	if gasPrice.IsInt64() {
+		priority = gasPrice.Int64()
+	} else {
+		priority = int64(math.MaxInt64)
+	}
+
 	return priority
 }