[WIP]feat: introduce 2-phases aggregation

app/ante/cosmos/context.go

@@ -41,7 +41,7 @@ func (sud SetUpContextDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, simulate
 
 	// If the transaction is an OracleCreatePriceTx, we set the gas meter to infinite
 	// NOTE: this works fine with cosmossdk-noopmempool+tendermint-mempool, since tenderminit used gasMeter.Limit() to check the gaswanted of tx insteand of tx.GetGas(), but if works with application mempool, we need to be caution of the tx.GetGas() since oracle create-price tx doesn't really pay.
-	if anteutils.IsOracleCreatePriceTx(tx) {
+	if _, ok, _ := anteutils.OracleCreatePriceTx(tx); ok {
 		newCtx = ctx.WithGasMeter(evmostypes.NewInfiniteGasMeterWithLimit(0))
 		return next(newCtx, tx, simulate)
 	}

app/ante/cosmos/fees.go

@@ -59,8 +59,14 @@ func (dfd DeductFeeDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, simulate bo
 		return ctx, errorsmod.Wrap(errortypes.ErrTxDecode, "Tx must be a FeeTx")
 	}
 
-	if anteutils.IsOracleCreatePriceTx(tx) {
-		newCtx := ctx.WithPriority(math.MaxInt64)
+	if _, isOracle, isRawData := anteutils.OracleCreatePriceTx(tx); isOracle {
+		var newCtx sdk.Context
+		if isRawData {
+			newCtx = ctx.WithPriority(math.MinInt64)
+		} else {
+			// set lowest priority for rawData tx, we include rawdata tx from imua-mempool to make sure that big raw data piece don't take all the space in a block, and the 'delivered' tx will be removed from consensus-mempoool by 'recheckTx' after commit
+			newCtx = ctx.WithPriority(math.MaxInt64)
+		}
 		return next(newCtx, tx, simulate)
 	}
 

app/ante/cosmos/min_price.go

@@ -32,7 +32,7 @@ func (mpd MinGasPriceDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, simulate
 		return ctx, errorsmod.Wrapf(errortypes.ErrInvalidType, "invalid transaction type %T, expected sdk.FeeTx", tx)
 	}
 
-	if anteutils.IsOracleCreatePriceTx(tx) {
+	if _, ok, _ := anteutils.OracleCreatePriceTx(tx); ok {
 		return next(ctx, tx, simulate)
 	}
 

app/ante/cosmos/oracle_two_phases.go

@@ -0,0 +1,42 @@
+package cosmos
+
+import (
+	"errors"
+	"fmt"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/imua-xyz/imuachain/app/ante/utils"
+)
+
+type OracleTwoPhasesDecorator struct {
+	ok utils.OracleKeeper
+}
+
+func NewOracleTwoPhasesDecorator(oracleKeeper utils.OracleKeeper) OracleTwoPhasesDecorator {
+	return OracleTwoPhasesDecorator{
+		ok: oracleKeeper,
+	}
+}
+
+func (otpd OracleTwoPhasesDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, simulate bool, next sdk.AnteHandler) (sdk.Context, error) {
+	msgs, _, isRawData := utils.OracleCreatePriceTx(tx)
+	if isRawData {
+		pieceWithProof, ok := otpd.ok.GetPieceWithProof(msgs[0])
+		// valid failed when getting pieceWithProof
+		if !ok {
+			return ctx, errors.New("failed to valid and get pieceWithProof with a tx with oracle rawData")
+		}
+		proofPath := otpd.ok.MinimalProofPathByIndex(msgs[0].FeederID, uint32(pieceWithProof.Index))
+		if len(proofPath) != int(pieceWithProof.ProofSize()) {
+			return ctx, fmt.Errorf("rawData proofPath size not match, expected:%d, got:%d", len(proofPath), pieceWithProof.ProofSize())
+		}
+		// the proofPath need to be exactly the same of both value and order
+		for i, index := range proofPath {
+			if pieceWithProof.Proof[i].Index != index {
+				return ctx, fmt.Errorf("rawData proofPath didn't include necessary index on position:%d of path:%d", i, index)
+			}
+		}
+	}
+
+	return next(ctx, tx, simulate)
+}

app/ante/cosmos/sigverify.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"strconv"
 
 	"github.com/cosmos/cosmos-sdk/crypto/keys/ed25519"
 	kmultisig "github.com/cosmos/cosmos-sdk/crypto/keys/multisig"
@@ -21,7 +22,6 @@ import (
 	authsigning "github.com/cosmos/cosmos-sdk/x/auth/signing"
 	"github.com/cosmos/cosmos-sdk/x/auth/types"
 	"github.com/imua-xyz/imuachain/app/ante/utils"
-	oracletypes "github.com/imua-xyz/imuachain/x/oracle/types"
 )
 
 var (
@@ -59,7 +59,7 @@ func NewSetPubKeyDecorator(ak authante.AccountKeeper) SetPubKeyDecorator {
 
 func (spkd SetPubKeyDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, simulate bool, next sdk.AnteHandler) (sdk.Context, error) {
 	// skip publickkey set for oracle create-price message
-	if utils.IsOracleCreatePriceTx(tx) {
+	if _, ok, _ := utils.OracleCreatePriceTx(tx); ok {
 		sigTx, ok := tx.(authsigning.SigVerifiableTx)
 		if !ok {
 			return ctx, sdkerrors.ErrTxDecode.Wrap("invalid transaction type, expected SigVerifiableTx")
@@ -173,7 +173,7 @@ func (sgcd SigGasConsumeDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, simula
 		return ctx, sdkerrors.ErrTxDecode.Wrap("invalid transaction type, expected SigVerifiableTx")
 	}
 
-	if utils.IsOracleCreatePriceTx(tx) {
+	if _, isOracle, _ := utils.OracleCreatePriceTx(tx); isOracle {
 		return next(ctx, tx, simulate)
 	}
 
@@ -258,7 +258,7 @@ func OnlyLegacyAminoSigners(sigData signing.SignatureData) bool {
 }
 
 func (svd SigVerificationDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, simulate bool, next sdk.AnteHandler) (newCtx sdk.Context, err error) {
-	if utils.IsOracleCreatePriceTx(tx) {
+	if _, ok, _ := utils.OracleCreatePriceTx(tx); ok {
 		sigTx, ok := tx.(authsigning.SigVerifiableTx)
 		if !ok {
 			return ctx, sdkerrors.ErrTxDecode.Wrap("invalid transaction type, expected SigVerifiableTx")
@@ -386,9 +386,40 @@ func NewIncrementSequenceDecorator(ak authante.AccountKeeper, oracleKeeper utils
 
 func (isd IncrementSequenceDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, simulate bool, next sdk.AnteHandler) (sdk.Context, error) {
 	// oracle create-price message dont need to increment sequence, check its nonce instead
-	if utils.IsOracleCreatePriceTx(tx) {
-		for _, msg := range tx.GetMsgs() {
-			msg := msg.(*oracletypes.MsgCreatePrice)
+	if msgs, isOracle, isRawData := utils.OracleCreatePriceTx(tx); isOracle {
+		msg := msgs[0]
+		if isRawData {
+			// TODO(leonz): move this to ValidateBasic
+			if len(msg.Prices) != 1 || len(msg.Prices[0].Prices) == 0 {
+				return ctx, errors.New("invalid raw data price format")
+			}
+			pieceIndexTmp, err := strconv.ParseUint(msg.Prices[0].Prices[0].DetID, 10, 32)
+			pieceIndex := uint32(pieceIndexTmp)
+			if err != nil {
+				return ctx, fmt.Errorf("invalid piece index parsed from DetID, got:%s", msg.Prices[0].Prices[0].DetID)
+			}
+			if nextPieceIndex, found := isd.oracleKeeper.NextPieceIndexByFeederID(ctx, msg.FeederID); found {
+				if pieceIndex < nextPieceIndex {
+					return ctx, fmt.Errorf("piece index must be at least %d, got:%d", nextPieceIndex, pieceIndex)
+				}
+				if (ctx.IsCheckTx() && pieceIndex >= nextPieceIndex+uint32(isd.oracleKeeper.GetMaxNonceFromCache())) ||
+					(!ctx.IsCheckTx() && pieceIndex > nextPieceIndex) {
+					return ctx, fmt.Errorf("invalid piece index, nextPieceIndex:%d, got:%d, isCheckTx:%t", nextPieceIndex, pieceIndex, ctx.IsCheckTx())
+				}
+			} else {
+				return ctx, fmt.Errorf("no valid nextPieceIndex for feederID:%d", msg.FeederID)
+			}
+
+			if accAddress, err := sdk.AccAddressFromBech32(msg.Creator); err != nil {
+				return ctx, errors.New("invalid address")
+			} else if _, err := isd.oracleKeeper.CheckAndIncreaseNextPieceIndex(ctx, sdk.ConsAddress(accAddress).String(), msg.FeederID, pieceIndex); err != nil {
+				return ctx, err
+			}
+
+			return next(ctx, tx, simulate)
+		}
+
+		for _, msg := range msgs {
 			if accAddress, err := sdk.AccAddressFromBech32(msg.Creator); err != nil {
 				return ctx, errors.New("invalid address")
 				// #nosec G115  // safe conversion

app/ante/cosmos/txsize_gas.go

@@ -38,9 +38,12 @@ func (cgts ConsumeTxSizeGasDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, sim
 	}
 
 	// Skip gas consumption if tx is an OracleCreatePriceTx
-	if anteutils.IsOracleCreatePriceTx(tx) {
-		if len(ctx.TxBytes()) > anteutils.TxSizeLimit {
-			return ctx, sdkerrors.ErrTxTooLarge.Wrapf("oracle create-price tx has exceeds size limit, limit:%d, got:%d", anteutils.TxSizeLimit, len(ctx.TxBytes()))
+	if _, isOracle, isRawData := anteutils.OracleCreatePriceTx(tx); isOracle {
+		if isRawData && len(ctx.TxBytes()) > anteutils.TxSizeLimitOracleRawData {
+			return ctx, sdkerrors.ErrTxTooLarge.Wrapf("oracle create-price tx exceeds size limit, limit:%d, got:%d", anteutils.TxSizeLimitOracleRawData, len(ctx.TxBytes()))
+		}
+		if len(ctx.TxBytes()) > anteutils.TxSizeLimitOraclePrice {
+			return ctx, sdkerrors.ErrTxTooLarge.Wrapf("oracle create-price tx with rawdata exceeds size limit, limit:%d, got:%d", anteutils.TxSizeLimitOraclePrice, len(ctx.TxBytes()))
 		}
 		return next(ctx, tx, simulate)
 	}

app/ante/evm/fee_market.go

@@ -30,7 +30,7 @@ func NewGasWantedDecorator(
 }
 
 func (gwd GasWantedDecorator) AnteHandle(ctx sdk.Context, tx sdk.Tx, simulate bool, next sdk.AnteHandler) (newCtx sdk.Context, err error) {
-	if anteutils.IsOracleCreatePriceTx(tx) {
+	if _, ok, _ := anteutils.OracleCreatePriceTx(tx); ok {
 		return next(ctx, tx, simulate)
 	}
 	evmParams := gwd.evmKeeper.GetParams(ctx)

app/ante/handler_options.go

@@ -120,6 +120,7 @@ func newCosmosAnteHandler(options HandlerOptions) sdk.AnteHandler {
 		cosmosante.NewIncrementSequenceDecorator(options.AccountKeeper, options.OracleKeeper),
 		ibcante.NewRedundantRelayDecorator(options.IBCKeeper),
 		evmante.NewGasWantedDecorator(options.EvmKeeper, options.FeeMarketKeeper),
+		cosmosante.NewOracleTwoPhasesDecorator(options.OracleKeeper),
 	)
 }
 

app/ante/utils/interfaces.go

@@ -3,6 +3,7 @@ package utils
 import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
+	oracletypes "github.com/imua-xyz/imuachain/x/oracle/types"
 )
 
 // BankKeeper defines the exposed interface for using functionality of the bank keeper
@@ -27,4 +28,9 @@ type StakingKeeper interface {
 // OracleKeeper defines the exposed interface for using functionality of the oracle keeper
 type OracleKeeper interface {
 	CheckAndIncreaseNonce(ctx sdk.Context, validator string, feederID uint64, nonce uint32) (prevNonce uint32, err error)
+	NextPieceIndexByFeederID(ctx sdk.Context, feederID uint64) (uint32, bool)
+	CheckAndIncreaseNextPieceIndex(ctx sdk.Context, validator string, feederID uint64, NextPieceIndex uint32) (nextPieceIndex uint32, err error)
+	GetMaxNonceFromCache() int32
+	GetPieceWithProof(msg *oracletypes.MsgCreatePrice) (*oracletypes.PieceWithProof, bool)
+	MinimalProofPathByIndex(feederID uint64, index uint32) []uint32
 }

app/ante/utils/oracle.go

@@ -6,17 +6,32 @@ import (
 )
 
 // TxSizeLimit limits max size of a create-price tx, this is calculated based on one nativeTokenbalance message of worst case(max size), which will need 576 bytes for balance update
-const TxSizeLimit = 1000
+// 48*1024+5*32+6*4 +... TODO(leonz): ensure the proto cost, now use a fixed 200B
+const (
+	TxSizeLimitOraclePrice   = 49600
+	TxSizeLimitOracleRawData = 200
+)
 
-func IsOracleCreatePriceTx(tx sdk.Tx) bool {
+// TODO(leonz): return additional error info
+func OracleCreatePriceTx(tx sdk.Tx) (msgsO []*oracletypes.MsgCreatePrice, validOracle, validRawData bool) {
 	msgs := tx.GetMsgs()
 	if len(msgs) == 0 {
-		return false
+		return nil, false, false
 	}
+	msgsO = make([]*oracletypes.MsgCreatePrice, 0, len(msgs))
 	for _, msg := range msgs {
-		if _, ok := msg.(*oracletypes.MsgCreatePrice); !ok {
-			return false
+		msgO, ok := msg.(*oracletypes.MsgCreatePrice)
+		if !ok {
+			return nil, false, false
+		}
+		if msgO.IsPhaseTwo() {
+			if len(msgs) > 1 {
+				return nil, false, false
+			}
+			msgsO = append(msgsO, msgO)
+			return msgsO, true, true
 		}
+		msgsO = append(msgsO, msgO)
 	}
-	return true
+	return msgsO, true, false
 }

app/app.go

@@ -105,7 +105,6 @@ import (
 	"github.com/cosmos/cosmos-sdk/store/streaming"
 	storetypes "github.com/cosmos/cosmos-sdk/store/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
-	"github.com/cosmos/cosmos-sdk/types/mempool"
 	"github.com/cosmos/cosmos-sdk/types/module"
 	"github.com/cosmos/cosmos-sdk/version"
 	srvflags "github.com/evmos/evmos/v16/server/flags"
@@ -354,7 +353,7 @@ type ImuachainApp struct {
 	OperatorKeeper   operatorKeeper.Keeper
 	ImSlashKeeper    slashKeeper.Keeper
 	AVSManagerKeeper avsManagerKeeper.Keeper
-	OracleKeeper     oracleKeeper.Keeper
+	OracleKeeper     *oracleKeeper.Keeper
 	ImmintKeeper     immintkeeper.Keeper
 	DistrKeeper      distrkeeper.Keeper
 
@@ -394,10 +393,10 @@ func NewImuachainApp(
 
 	eip712.SetEncodingConfig(encodingConfig)
 
+	var oKeeper *oracleKeeper.Keeper
 	// Setup Mempool and Proposal Handlers
 	baseAppOptions = append(baseAppOptions, func(app *baseapp.BaseApp) {
-		// NOTE: we use a NoOpMempool here, for oracle create-price, it works fine since we have set a infinitgasmeterwithlimit in the ante handler to avoid the out-of-gas error no matter what the amount/gas is set by tx builder, and we set the highest priority for oracle create-price txs to work properly with tendermint mempool to make sure oracle creat-prie tx will be included in the mempool if received. And if we want to use some other application mempool, we need to take care of the gas limit and gas price in the oracle create-price txs.(we don't need to bother this since tendermint mempool use gasMeter.limit() instead of tx.Gas())
-		mempool := mempool.NoOpMempool{}
+		mempool := NewImuaMempool(oKeeper, encodingConfig.TxConfig.TxDecoder())
 		app.SetMempool(mempool)
 		handler := baseapp.NewDefaultProposalHandler(mempool, app)
 		app.SetPrepareProposal(handler.PrepareProposalHandler())
@@ -462,6 +461,7 @@ func NewImuachainApp(
 		keys:              keys,
 		tkeys:             tkeys,
 		memKeys:           memKeys,
+		OracleKeeper:      oKeeper,
 	}
 
 	// init params keeper and subspaces
@@ -544,7 +544,7 @@ func NewImuachainApp(
 
 	// asset and client chain registry.
 	app.AssetsKeeper = assetsKeeper.NewKeeper(
-		keys[assetsTypes.StoreKey], appCodec, &app.OracleKeeper, app.AccountKeeper,
+		keys[assetsTypes.StoreKey], appCodec, app.OracleKeeper, app.AccountKeeper,
 		app.BankKeeper, &app.DelegationKeeper, authAddrString,
 	)
 
@@ -582,7 +582,7 @@ func NewImuachainApp(
 	)
 
 	// x/oracle is not fully integrated (or enabled) but allows for exchange rates to be added.
-	app.OracleKeeper = oracleKeeper.NewKeeper(
+	*app.OracleKeeper = oracleKeeper.NewKeeper(
 		appCodec, keys[oracleTypes.StoreKey], memKeys[oracleTypes.MemStoreKey],
 		app.GetSubspace(oracleTypes.ModuleName), app.StakingKeeper,
 		&app.DelegationKeeper, &app.AssetsKeeper, authAddrString,
@@ -696,7 +696,7 @@ func NewImuachainApp(
 		keys[operatorTypes.StoreKey], appCodec,
 		app.AssetsKeeper,
 		&app.DelegationKeeper, // intentionally a pointer, since not yet initialized.
-		&app.OracleKeeper,
+		app.OracleKeeper,
 		&app.AVSManagerKeeper,
 		&app.StakingKeeper,
 		delegationTypes.VirtualSlashKeeper{},
@@ -904,7 +904,7 @@ func NewImuachainApp(
 		reward.NewAppModule(appCodec, app.RewardKeeper),
 		imslash.NewAppModule(appCodec, app.ImSlashKeeper),
 		avs.NewAppModule(appCodec, app.AVSManagerKeeper),
-		oracle.NewAppModule(appCodec, app.OracleKeeper, app.AccountKeeper, app.BankKeeper),
+		oracle.NewAppModule(appCodec, *app.OracleKeeper, app.AccountKeeper, app.BankKeeper),
 		distr.NewAppModule(appCodec, app.DistrKeeper),
 	)
 
@@ -1106,6 +1106,7 @@ func NewImuachainApp(
 func (app *ImuachainApp) Name() string { return app.BaseApp.Name() }
 
 func (app *ImuachainApp) setAnteHandler(txConfig client.TxConfig, maxGasWanted uint64) {
+	app.OracleKeeper.GetPieceWithProof(nil)
 	options := ante.HandlerOptions{
 		Cdc:                    app.appCodec,
 		AccountKeeper:          app.AccountKeeper,