Oracle midblock

app/ante.go

@@ -86,6 +86,7 @@ func NewAnteHandlerAndDepGenerator(options HandlerOptions) (sdk.AnteHandler, sdk
 		sdk.DefaultWrappedAnteDecorator(wasmkeeper.NewLimitSimulationGasDecorator(options.WasmConfig.SimulationGasLimit)), // after setup context to enforce limits early
 		sdk.DefaultWrappedAnteDecorator(ante.NewRejectExtensionOptionsDecorator()),
 		oracle.NewSpammingPreventionDecorator(*options.OracleKeeper),
+		oracle.NewOracleVoteAloneDecorator(),
 		sdk.DefaultWrappedAnteDecorator(ante.NewValidateBasicDecorator()),
 		sdk.DefaultWrappedAnteDecorator(ante.NewTxTimeoutHeightDecorator()),
 		sdk.DefaultWrappedAnteDecorator(ante.NewValidateMemoDecorator(options.AccountKeeper)),

app/antedecorators/gas_test.go

@@ -13,30 +13,18 @@ import (
 	"github.com/tendermint/tendermint/proto/tendermint/types"
 )
 
-type TestTx struct {
-	msgs []sdk.Msg
-}
-
-func (t TestTx) GetMsgs() []sdk.Msg {
-	return t.msgs
-}
-
-func (t TestTx) ValidateBasic() error {
-	return nil
-}
-
 func TestMultiplierGasSetter(t *testing.T) {
-	app := app.Setup(false)
+	testApp := app.Setup(false)
 	contractAddr, err := sdk.AccAddressFromBech32("sei1y3pxq5dp900czh0mkudhjdqjq5m8cpmmps8yjw")
 	require.NoError(t, err)
-	ctx := app.NewContext(false, types.Header{}).WithBlockHeight(2)
+	ctx := testApp.NewContext(false, types.Header{}).WithBlockHeight(2)
 	testMsg := wasmtypes.MsgExecuteContract{
 		Contract: "sei1y3pxq5dp900czh0mkudhjdqjq5m8cpmmps8yjw",
 		Msg:      []byte("{}"),
 	}
-	testTx := TestTx{msgs: []sdk.Msg{&testMsg}}
+	testTx := app.NewTestTx([]sdk.Msg{&testMsg})
 	// discounted mapping
-	app.AccessControlKeeper.SetWasmDependencyMapping(ctx, accesscontrol.WasmDependencyMapping{
+	testApp.AccessControlKeeper.SetWasmDependencyMapping(ctx, accesscontrol.WasmDependencyMapping{
 		Enabled:         true,
 		ContractAddress: contractAddr.String(),
 		AccessOps: []accesscontrol.AccessOperationWithSelector{
@@ -52,12 +40,12 @@ func TestMultiplierGasSetter(t *testing.T) {
 			},
 		},
 	})
-	gasMeterSetter := antedecorators.GetGasMeterSetter(app.AccessControlKeeper)
+	gasMeterSetter := antedecorators.GetGasMeterSetter(testApp.AccessControlKeeper)
 	ctxWithGasMeter := gasMeterSetter(false, ctx, 1000, testTx)
 	ctxWithGasMeter.GasMeter().ConsumeGas(2, "")
 	require.Equal(t, uint64(1), ctxWithGasMeter.GasMeter().GasConsumed())
 	// not discounted mapping
-	app.AccessControlKeeper.SetWasmDependencyMapping(ctx, accesscontrol.WasmDependencyMapping{
+	testApp.AccessControlKeeper.SetWasmDependencyMapping(ctx, accesscontrol.WasmDependencyMapping{
 		Enabled:         true,
 		ContractAddress: contractAddr.String(),
 		AccessOps: []accesscontrol.AccessOperationWithSelector{

app/app.go

@@ -1175,6 +1175,60 @@ func (app *App) ProcessTxs(
 	return txResults, ctx
 }
 
+func (app *App) PartitionOracleVoteTxs(ctx sdk.Context, txs [][]byte) (oracleVoteTxs, otherTxs [][]byte) {
+	for _, tx := range txs {
+		decodedTx, err := app.txDecoder(tx)
+		if err != nil {
+			ctx.Logger().Error(fmt.Sprintf("Error decoding tx for partitioning: %v", err))
+			// if theres an issue decoding, add it to `otherTxs` for normal processing and continue
+			otherTxs = append(otherTxs, tx)
+			continue
+		}
+		oracleVote := false
+		// if theres an oracle vote msg, we want to add to oracleVoteTxs
+	msgLoop:
+		for _, msg := range decodedTx.GetMsgs() {
+			switch msg.(type) {
+			case *oracletypes.MsgAggregateExchangeRateVote:
+				oracleVote = true
+			default:
+				oracleVote = false
+				break msgLoop
+			}
+		}
+		if oracleVote {
+			oracleVoteTxs = append(oracleVoteTxs, tx)
+		} else {
+			otherTxs = append(otherTxs, tx)
+		}
+
+	}
+	return oracleVoteTxs, otherTxs
+}
+
+func (app *App) BuildDependenciesAndRunTxs(ctx sdk.Context, txs [][]byte) ([]*abci.ExecTxResult, sdk.Context) {
+	var txResults []*abci.ExecTxResult
+
+	dependencyDag, err := app.AccessControlKeeper.BuildDependencyDag(ctx, app.txDecoder, app.GetAnteDepGenerator(), txs)
+
+	switch err {
+	case nil:
+		// Start with a fresh state for the MemCache
+		ctx = ctx.WithContextMemCache(sdk.NewContextMemCache())
+		txResults, ctx = app.ProcessTxs(ctx, txs, dependencyDag, app.ProcessBlockConcurrent)
+	case acltypes.ErrGovMsgInBlock:
+		ctx.Logger().Info(fmt.Sprintf("Gov msg found while building DAG, processing synchronously: %s", err))
+		txResults = app.ProcessBlockSynchronous(ctx, txs)
+		metrics.IncrDagBuildErrorCounter(metrics.GovMsgInBlock)
+	default:
+		ctx.Logger().Error(fmt.Sprintf("Error while building DAG, processing synchronously: %s", err))
+		txResults = app.ProcessBlockSynchronous(ctx, txs)
+		metrics.IncrDagBuildErrorCounter(metrics.FailedToBuild)
+	}
+
+	return txResults, ctx
+}
+
 func (app *App) ProcessBlock(ctx sdk.Context, txs [][]byte, req BlockProcessRequest, lastCommit abci.CommitInfo) ([]abci.Event, []*abci.ExecTxResult, abci.ResponseEndBlock, error) {
 	goCtx := app.decorateContextWithDexMemState(ctx.Context())
 	ctx = ctx.WithContext(goCtx)
@@ -1205,27 +1259,20 @@ func (app *App) ProcessBlock(ctx sdk.Context, txs [][]byte, req BlockProcessRequ
 	beginBlockResp := app.BeginBlock(ctx, beginBlockReq)
 	events = append(events, beginBlockResp.Events...)
 
-	midBlockEvents := app.MidBlock(ctx, req.GetHeight())
-	events = append(events, midBlockEvents...)
+	var txResults []*abci.ExecTxResult
 
-	dependencyDag, err := app.AccessControlKeeper.BuildDependencyDag(ctx, app.txDecoder, app.GetAnteDepGenerator(), txs)
+	oracleTxs, txs := app.PartitionOracleVoteTxs(ctx, txs)
 
-	var txResults []*abci.ExecTxResult
+	// run the oracle txs
+	oracleResults, ctx := app.BuildDependenciesAndRunTxs(ctx, oracleTxs)
+	txResults = append(txResults, oracleResults...)
 
-	switch err {
-	case nil:
-		// Start with a fresh state for the MemCache
-		ctx = ctx.WithContextMemCache(sdk.NewContextMemCache())
-		txResults, ctx = app.ProcessTxs(ctx, txs, dependencyDag, app.ProcessBlockConcurrent)
-	case acltypes.ErrGovMsgInBlock:
-		ctx.Logger().Info(fmt.Sprintf("Gov msg found while building DAG, processing synchronously: %s", err))
-		txResults = app.ProcessBlockSynchronous(ctx, txs)
-		metrics.IncrDagBuildErrorCounter(metrics.GovMsgInBlock)
-	default:
-		ctx.Logger().Error(fmt.Sprintf("Error while building DAG, processing synchronously: %s", err))
-		txResults = app.ProcessBlockSynchronous(ctx, txs)
-		metrics.IncrDagBuildErrorCounter(metrics.FailedToBuild)
-	}
+	midBlockEvents := app.MidBlock(ctx, req.GetHeight())
+	events = append(events, midBlockEvents...)
+
+	// run other txs
+	otherResults, ctx := app.BuildDependenciesAndRunTxs(ctx, txs)
+	txResults = append(txResults, otherResults...)
 
 	// Finalize all Bank Module Transfers here so that events are included
 	lazyWriteEvents := app.BankKeeper.WriteDeferredOperations(ctx)

app/app_test.go

@@ -9,8 +9,10 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkacltypes "github.com/cosmos/cosmos-sdk/types/accesscontrol"
 	acltypes "github.com/cosmos/cosmos-sdk/x/accesscontrol/types"
+	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
 	"github.com/k0kubun/pp/v3"
 	"github.com/sei-protocol/sei-chain/app"
+	oracletypes "github.com/sei-protocol/sei-chain/x/oracle/types"
 	"github.com/stretchr/testify/require"
 	abci "github.com/tendermint/tendermint/abci/types"
 )
@@ -155,3 +157,113 @@ func TestProcessTxsClearCacheOnFail(t *testing.T) {
 	require.Equal(t, 0, len(testWrapper.Ctx.ContextMemCache().GetDeferredWithdrawals().GetSortedKeys()))
 	require.Equal(t, 0, len(testWrapper.Ctx.ContextMemCache().GetDeferredSends().GetSortedKeys()))
 }
+
+func TestPartitionOracleTxs(t *testing.T) {
+	tm := time.Now().UTC()
+	valPub := secp256k1.GenPrivKey().PubKey()
+
+	testWrapper := app.NewTestWrapper(t, tm, valPub)
+
+	account := sdk.AccAddress(valPub.Address()).String()
+	validator := sdk.ValAddress(valPub.Address()).String()
+
+	oracleMsg := &oracletypes.MsgAggregateExchangeRateVote{
+		ExchangeRates: "1.2uatom",
+		Feeder:        account,
+		Validator:     validator,
+	}
+
+	otherMsg := &stakingtypes.MsgDelegate{
+		DelegatorAddress: account,
+		ValidatorAddress: validator,
+		Amount:           sdk.NewCoin("usei", sdk.NewInt(1)),
+	}
+
+	txEncoder := app.MakeEncodingConfig().TxConfig.TxEncoder()
+	oracleTxBuilder := app.MakeEncodingConfig().TxConfig.NewTxBuilder()
+	otherTxBuilder := app.MakeEncodingConfig().TxConfig.NewTxBuilder()
+	mixedTxBuilder := app.MakeEncodingConfig().TxConfig.NewTxBuilder()
+
+	err := oracleTxBuilder.SetMsgs(oracleMsg)
+	require.NoError(t, err)
+	oracleTx, err := txEncoder(oracleTxBuilder.GetTx())
+	require.NoError(t, err)
+
+	err = otherTxBuilder.SetMsgs(otherMsg)
+	require.NoError(t, err)
+	otherTx, err := txEncoder(otherTxBuilder.GetTx())
+	require.NoError(t, err)
+
+	// this should be treated as non-oracle vote
+	err = mixedTxBuilder.SetMsgs([]sdk.Msg{oracleMsg, otherMsg}...)
+	require.NoError(t, err)
+	mixedTx, err := txEncoder(mixedTxBuilder.GetTx())
+	require.NoError(t, err)
+
+	txs := [][]byte{
+		oracleTx,
+		otherTx,
+		mixedTx,
+	}
+
+	oracleTxs, otherTxs := testWrapper.App.PartitionOracleVoteTxs(testWrapper.Ctx, txs)
+	require.Equal(t, oracleTxs, [][]byte{oracleTx})
+	require.Equal(t, otherTxs, [][]byte{otherTx, mixedTx})
+}
+
+func TestProcessOracleAndOtherTxsSuccess(t *testing.T) {
+	tm := time.Now().UTC()
+	valPub := secp256k1.GenPrivKey().PubKey()
+
+	testWrapper := app.NewTestWrapper(t, tm, valPub)
+
+	account := sdk.AccAddress(valPub.Address()).String()
+	validator := sdk.ValAddress(valPub.Address()).String()
+
+	oracleMsg := &oracletypes.MsgAggregateExchangeRateVote{
+		ExchangeRates: "1.2uatom",
+		Feeder:        account,
+		Validator:     validator,
+	}
+
+	otherMsg := &stakingtypes.MsgDelegate{
+		DelegatorAddress: account,
+		ValidatorAddress: validator,
+		Amount:           sdk.NewCoin("usei", sdk.NewInt(1)),
+	}
+
+	oracleTxBuilder := app.MakeEncodingConfig().TxConfig.NewTxBuilder()
+	otherTxBuilder := app.MakeEncodingConfig().TxConfig.NewTxBuilder()
+	txEncoder := app.MakeEncodingConfig().TxConfig.TxEncoder()
+
+	err := oracleTxBuilder.SetMsgs(oracleMsg)
+	require.NoError(t, err)
+	oracleTx, err := txEncoder(oracleTxBuilder.GetTx())
+	require.NoError(t, err)
+
+	err = otherTxBuilder.SetMsgs(otherMsg)
+	require.NoError(t, err)
+	otherTx, err := txEncoder(otherTxBuilder.GetTx())
+	require.NoError(t, err)
+
+	txs := [][]byte{
+		oracleTx,
+		otherTx,
+	}
+
+	req := &abci.RequestFinalizeBlock{
+		Height: 1,
+	}
+	_, txResults, _, _ := testWrapper.App.ProcessBlock(
+		testWrapper.Ctx.WithBlockHeight(
+			1,
+		).WithBlockGasMeter(
+			sdk.NewInfiniteGasMeter(),
+		),
+		txs,
+		req,
+		req.DecidedLastCommit,
+	)
+
+	require.Equal(t, 2, len(txResults))
+}

app/test_helpers.go

@@ -25,6 +25,22 @@ import (
 
 const TestContract = "TEST"
 
+type TestTx struct {
+	msgs []sdk.Msg
+}
+
+func NewTestTx(msgs []sdk.Msg) TestTx {
+	return TestTx{msgs: msgs}
+}
+
+func (t TestTx) GetMsgs() []sdk.Msg {
+	return t.msgs
+}
+
+func (t TestTx) ValidateBasic() error {
+	return nil
+}
+
 type TestWrapper struct {
 	suite.Suite
 

x/oracle/abci.go

@@ -13,12 +13,6 @@ import (
 
 func MidBlocker(ctx sdk.Context, k keeper.Keeper) {
 	defer telemetry.ModuleMeasureSince(types.ModuleName, time.Now(), telemetry.MetricKeyMidBlocker)
-	ctx.Logger().Info("Running Oracle MidBlocker")
-	// TODO: this needs to be refactored to perform relevant endblocker logic to finalize oracle prices in a later PR
-}
-
-func EndBlocker(ctx sdk.Context, k keeper.Keeper) {
-	defer telemetry.ModuleMeasureSince(types.ModuleName, time.Now(), telemetry.MetricKeyEndBlocker)
 
 	params := k.GetParams(ctx)
 	if utils.IsPeriodLastBlock(ctx, params.VotePeriod) {
@@ -139,6 +133,12 @@ func EndBlocker(ctx sdk.Context, k keeper.Keeper) {
 		}
 	}
 
+}
+
+func EndBlocker(ctx sdk.Context, k keeper.Keeper) {
+	defer telemetry.ModuleMeasureSince(types.ModuleName, time.Now(), telemetry.MetricKeyEndBlocker)
+
+	params := k.GetParams(ctx)
 	// Do slash who did miss voting over threshold and
 	// reset miss counters of all validators at the last block of slash window
 	if utils.IsPeriodLastBlock(ctx, params.SlashWindow) {