(v0.26 <- #1851) Optimize Pricefeed EndBlocker (#1895)

* Optimize Pricefeed EndBlocker (#1851)

* optimize pricefeed endblocker to iterate all markets only once to remove
overhead of opening and closing iterator for each market individually.
In addition, extend tests to cover 100% of abci and price updating
behavior.

* use test cases that can't be confused with mean to ensure median is
always used

* update kvtool to master branch

* update changelog

---------

Co-authored-by: Nick DeLuca <nickdeluca08@gmail.com>

.tool-versions

@@ -1,2 +1,2 @@
-golang 1.21
+golang 1.21.9
 nodejs 18.16.0

CHANGELOG.md

@@ -36,6 +36,9 @@ Ref: https://keepachangelog.com/en/1.0.0/
 
 ## [Unreleased]
 
+### Improvements
+- (pricefeed) [#1851] optimize EndBlocker to iterate all markets only once
+
 ## [v0.26.0]
 
 ### Features
@@ -330,6 +333,7 @@ the [changelog](https://github.com/cosmos/cosmos-sdk/blob/v0.38.4/CHANGELOG.md).
 - [#257](https://github.com/Kava-Labs/kava/pulls/257) Include scripts to run
   large-scale simulations remotely using aws-batch
 
+[#1851]: https://github.com/Kava-Labs/kava/pull/1851
 [#1846]: https://github.com/Kava-Labs/kava/pull/1846
 [#1848]: https://github.com/Kava-Labs/kava/pull/1848
 [#1839]: https://github.com/Kava-Labs/kava/pull/1839

tests/e2e/kvtool

@@ -1 +1 @@
-Subproject commit 79bb25f47d20c10e70f0539c519240fba82b3302
+Subproject commit 567dfb80905be7d83ba2d33cc5dba9daf0af789c

x/pricefeed/abci.go

@@ -1,7 +1,6 @@
 package pricefeed
 
 import (
-	"errors"
 	"time"
 
 	"github.com/cosmos/cosmos-sdk/telemetry"
@@ -14,15 +13,5 @@ import (
 func EndBlocker(ctx sdk.Context, k keeper.Keeper) {
 	defer telemetry.ModuleMeasureSince(types.ModuleName, time.Now(), telemetry.MetricKeyEndBlocker)
 
-	// Update the current price of each asset.
+	k.SetCurrentPricesForAllMarkets(ctx)
-	for _, market := range k.GetMarkets(ctx) {
-		if !market.Active {
-			continue
-		}
-
-		err := k.SetCurrentPrices(ctx, market.MarketID)
-		if err != nil && !errors.Is(err, types.ErrNoValidPrice) {
-			panic(err)
-		}
-	}
 }

x/pricefeed/abci_test.go

@@ -0,0 +1,20 @@
+package pricefeed_test
+
+import (
+	"testing"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/kava-labs/kava/x/pricefeed"
+	"github.com/kava-labs/kava/x/pricefeed/keeper"
+	"github.com/kava-labs/kava/x/pricefeed/testutil"
+)
+
+func TestEndBlocker_UpdatesMultipleMarkets(t *testing.T) {
+	testutil.SetCurrentPrices_PriceCalculations(t, func(ctx sdk.Context, keeper keeper.Keeper) {
+		pricefeed.EndBlocker(ctx, keeper)
+	})
+
+	testutil.SetCurrentPrices_EventEmission(t, func(ctx sdk.Context, keeper keeper.Keeper) {
+		pricefeed.EndBlocker(ctx, keeper)
+	})
+}

x/pricefeed/keeper/keeper.go

@@ -131,6 +131,74 @@ func (k Keeper) SetCurrentPrices(ctx sdk.Context, marketID string) error {
 	return nil
 }
 
+// SetCurrentPricesForAllMarkets updates the price of an asset to the median of all valid oracle inputs
+func (k Keeper) SetCurrentPricesForAllMarkets(ctx sdk.Context) {
+	orderedMarkets := []string{}
+	marketPricesByID := make(map[string]types.CurrentPrices)
+
+	for _, market := range k.GetMarkets(ctx) {
+		if market.Active {
+			orderedMarkets = append(orderedMarkets, market.MarketID)
+			marketPricesByID[market.MarketID] = types.CurrentPrices{}
+		}
+	}
+
+	iterator := sdk.KVStorePrefixIterator(ctx.KVStore(k.key), types.RawPriceFeedPrefix)
+	for ; iterator.Valid(); iterator.Next() {
+		var postedPrice types.PostedPrice
+		k.cdc.MustUnmarshal(iterator.Value(), &postedPrice)
+
+		prices, found := marketPricesByID[postedPrice.MarketID]
+		if !found {
+			continue
+		}
+
+		// filter out expired prices
+		if postedPrice.Expiry.After(ctx.BlockTime()) {
+			marketPricesByID[postedPrice.MarketID] = append(prices, types.NewCurrentPrice(postedPrice.MarketID, postedPrice.Price))
+		}
+	}
+	iterator.Close()
+
+	for _, marketID := range orderedMarkets {
+		// store current price
+		validPrevPrice := true
+		prevPrice, err := k.GetCurrentPrice(ctx, marketID)
+		if err != nil {
+			validPrevPrice = false
+		}
+
+		notExpiredPrices, _ := marketPricesByID[marketID]
+
+		if len(notExpiredPrices) == 0 {
+			// NOTE: The current price stored will continue storing the most recent (expired)
+			// price if this is not set.
+			// This zero's out the current price stored value for that market and ensures
+			// that CDP methods that GetCurrentPrice will return error.
+			k.setCurrentPrice(ctx, marketID, types.CurrentPrice{})
+			continue
+		}
+
+		medianPrice := k.CalculateMedianPrice(notExpiredPrices)
+
+		// check case that market price was not set in genesis
+		//if validPrevPrice && !medianPrice.Equal(prevPrice.Price) {
+		if validPrevPrice && !medianPrice.Equal(prevPrice.Price) {
+			// only emit event if price has changed
+			ctx.EventManager().EmitEvent(
+				sdk.NewEvent(
+					types.EventTypeMarketPriceUpdated,
+					sdk.NewAttribute(types.AttributeMarketID, marketID),
+					sdk.NewAttribute(types.AttributeMarketPrice, medianPrice.String()),
+				),
+			)
+		}
+
+		currentPrice := types.NewCurrentPrice(marketID, medianPrice)
+		k.setCurrentPrice(ctx, marketID, currentPrice)
+	}
+}
+
 func (k Keeper) setCurrentPrice(ctx sdk.Context, marketID string, currentPrice types.CurrentPrice) {
 	store := ctx.KVStore(k.key)
 	store.Set(types.CurrentPriceKey(marketID), k.cdc.MustMarshal(&currentPrice))

x/pricefeed/keeper/keeper_test.go

@@ -1,6 +1,7 @@
 package keeper_test
 
 import (
+	"errors"
 	"testing"
 	"time"
 
@@ -11,6 +12,8 @@ import (
 	tmprototypes "github.com/cometbft/cometbft/proto/tendermint/types"
 
 	"github.com/kava-labs/kava/app"
+	"github.com/kava-labs/kava/x/pricefeed/keeper"
+	"github.com/kava-labs/kava/x/pricefeed/testutil"
 	"github.com/kava-labs/kava/x/pricefeed/types"
 )
 
@@ -236,3 +239,33 @@ func TestKeeper_ExpiredSetCurrentPrices(t *testing.T) {
 	_, err = keeper.GetCurrentPrice(ctx, "tstusd")
 	require.ErrorIs(t, types.ErrNoValidPrice, err, "current prices should be invalid")
 }
+
+func TestKeeper_SetCurrentPricesForAllMarkets_PriceUpdate(t *testing.T) {
+	testutil.SetCurrentPrices_PriceCalculations(t, func(ctx sdk.Context, keeper keeper.Keeper) {
+		keeper.SetCurrentPricesForAllMarkets(ctx)
+	})
+}
+
+func TestKeeper_SetCurrentPricesForAllMarkets_EventEmission(t *testing.T) {
+	testutil.SetCurrentPrices_EventEmission(t, func(ctx sdk.Context, keeper keeper.Keeper) {
+		keeper.SetCurrentPricesForAllMarkets(ctx)
+	})
+}
+
+func TestKeeper_SetCurrentPrices_MatchesAllMarketsBehavior(t *testing.T) {
+	testFunc := func(ctx sdk.Context, k keeper.Keeper) {
+		for _, market := range k.GetMarkets(ctx) {
+			if !market.Active {
+				continue
+			}
+
+			err := k.SetCurrentPrices(ctx, market.MarketID)
+			if err != nil && !errors.Is(err, types.ErrNoValidPrice) {
+				panic(err)
+			}
+		}
+	}
+
+	testutil.SetCurrentPrices_PriceCalculations(t, testFunc)
+	testutil.SetCurrentPrices_EventEmission(t, testFunc)
+}

x/pricefeed/testutil/helpers.go

@@ -0,0 +1,236 @@
+package testutil
+
+import (
+	"testing"
+	"time"
+
+	tmprototypes "github.com/cometbft/cometbft/proto/tendermint/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/kava-labs/kava/app"
+	"github.com/kava-labs/kava/x/pricefeed/keeper"
+	"github.com/kava-labs/kava/x/pricefeed/types"
+)
+
+func SetCurrentPrices_PriceCalculations(t *testing.T, f func(ctx sdk.Context, keeper keeper.Keeper)) {
+	_, addrs := app.GeneratePrivKeyAddressPairs(5)
+	tApp := app.NewTestApp()
+	ctx := tApp.NewContext(true, tmprototypes.Header{}).
+		WithBlockTime(time.Now().UTC())
+	keeper := tApp.GetPriceFeedKeeper()
+
+	params := types.Params{
+		Markets: []types.Market{
+			// valid previous price, expired prices, price change, and active
+			{MarketID: "asset1:usd", BaseAsset: "asset1", QuoteAsset: "usd", Oracles: addrs, Active: true},
+			// same data as asset1, but not active and should be ignored
+			{MarketID: "asset2:usd", BaseAsset: "asset2", QuoteAsset: "usd", Oracles: addrs, Active: false},
+			// same data as asset1 except no valid previous price
+			{MarketID: "asset3:usd", BaseAsset: "asset3", QuoteAsset: "usd", Oracles: addrs, Active: true},
+			// previous price set, but no valid prices
+			{MarketID: "asset4:usd", BaseAsset: "asset4", QuoteAsset: "usd", Oracles: addrs, Active: true},
+			// same as market one except different prices
+			{MarketID: "asset5:usd", BaseAsset: "asset5", QuoteAsset: "usd", Oracles: addrs, Active: true},
+		},
+	}
+	keeper.SetParams(ctx, params)
+
+	// need price equal to block time and after block time
+	blockTime := time.Now()
+	initialPriceExpiry := blockTime.Add(1 * time.Hour)
+
+	_, err := keeper.SetPrice(ctx, addrs[0], "asset1:usd", sdk.MustNewDecFromStr("1"), initialPriceExpiry)
+	require.NoError(t, err)
+	_, err = keeper.SetPrice(ctx, addrs[0], "asset2:usd", sdk.MustNewDecFromStr("1"), initialPriceExpiry)
+	require.NoError(t, err)
+	_, err = keeper.SetPrice(ctx, addrs[0], "asset4:usd", sdk.MustNewDecFromStr("1"), initialPriceExpiry)
+	require.NoError(t, err)
+	_, err = keeper.SetPrice(ctx, addrs[0], "asset5:usd", sdk.MustNewDecFromStr("10"), initialPriceExpiry)
+	require.NoError(t, err)
+
+	ctx = ctx.WithBlockTime(blockTime)
+	f(ctx, keeper)
+
+	// price should be set
+	price, err := keeper.GetCurrentPrice(ctx, "asset1:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.OneDec(), price.Price)
+	// not an active market, so price is not set
+	price, err = keeper.GetCurrentPrice(ctx, "asset2:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	// no price posted
+	price, err = keeper.GetCurrentPrice(ctx, "asset3:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	// price set initially
+	price, err = keeper.GetCurrentPrice(ctx, "asset4:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.OneDec(), price.Price)
+	price, err = keeper.GetCurrentPrice(ctx, "asset5:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.MustNewDecFromStr("10.0"), price.Price)
+
+	_, err = keeper.SetPrice(ctx, addrs[1], "asset1:usd", sdk.MustNewDecFromStr("2"), initialPriceExpiry.Add(1*time.Hour))
+	require.NoError(t, err)
+	_, err = keeper.SetPrice(ctx, addrs[1], "asset2:usd", sdk.MustNewDecFromStr("2"), initialPriceExpiry.Add(1*time.Hour))
+	require.NoError(t, err)
+	_, err = keeper.SetPrice(ctx, addrs[1], "asset5:usd", sdk.MustNewDecFromStr("20"), initialPriceExpiry.Add(1*time.Hour))
+	require.NoError(t, err)
+
+	blockTime = blockTime.Add(30 * time.Minute)
+	ctx = ctx.WithBlockTime(blockTime)
+	f(ctx, keeper)
+
+	// price should be set
+	price, err = keeper.GetCurrentPrice(ctx, "asset1:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.MustNewDecFromStr("1.5"), price.Price)
+	// not an active market, so price is not set
+	price, err = keeper.GetCurrentPrice(ctx, "asset2:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	// no price posted
+	price, err = keeper.GetCurrentPrice(ctx, "asset3:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	// price set initially
+	price, err = keeper.GetCurrentPrice(ctx, "asset4:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.OneDec(), price.Price)
+	price, err = keeper.GetCurrentPrice(ctx, "asset5:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.MustNewDecFromStr("15.0"), price.Price)
+
+	_, err = keeper.SetPrice(ctx, addrs[2], "asset1:usd", sdk.MustNewDecFromStr("30"), initialPriceExpiry.Add(1*time.Hour))
+	require.NoError(t, err)
+	_, err = keeper.SetPrice(ctx, addrs[2], "asset2:usd", sdk.MustNewDecFromStr("30"), initialPriceExpiry.Add(1*time.Hour))
+	require.NoError(t, err)
+	_, err = keeper.SetPrice(ctx, addrs[2], "asset5:usd", sdk.MustNewDecFromStr("30"), initialPriceExpiry.Add(1*time.Hour))
+	require.NoError(t, err)
+
+	blockTime = blockTime.Add(15 * time.Minute)
+	ctx = ctx.WithBlockTime(blockTime)
+	f(ctx, keeper)
+
+	// price should be set
+	price, err = keeper.GetCurrentPrice(ctx, "asset1:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.MustNewDecFromStr("2.0"), price.Price)
+	// not an active market, so price is not set
+	price, err = keeper.GetCurrentPrice(ctx, "asset2:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	// no price posted
+	price, err = keeper.GetCurrentPrice(ctx, "asset3:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	// price set initially
+	price, err = keeper.GetCurrentPrice(ctx, "asset4:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.OneDec(), price.Price)
+	price, err = keeper.GetCurrentPrice(ctx, "asset5:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.MustNewDecFromStr("20.0"), price.Price)
+
+	blockTime = blockTime.Add(15 * time.Minute)
+	ctx = ctx.WithBlockTime(blockTime)
+	f(ctx, keeper)
+
+	// price should be set
+	price, err = keeper.GetCurrentPrice(ctx, "asset1:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.MustNewDecFromStr("16"), price.Price)
+	// not an active market, so price is not set
+	price, err = keeper.GetCurrentPrice(ctx, "asset2:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	// no price posted
+	price, err = keeper.GetCurrentPrice(ctx, "asset3:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	// price set initially, now expired
+	price, err = keeper.GetCurrentPrice(ctx, "asset4:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	price, err = keeper.GetCurrentPrice(ctx, "asset5:usd")
+	require.NoError(t, err)
+	require.Equal(t, sdk.MustNewDecFromStr("25.0"), price.Price)
+
+	blockTime = blockTime.Add(10 * time.Hour)
+	ctx = ctx.WithBlockTime(blockTime)
+	f(ctx, keeper)
+
+	// all prices expired now
+	price, err = keeper.GetCurrentPrice(ctx, "asset1:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	price, err = keeper.GetCurrentPrice(ctx, "asset2:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	price, err = keeper.GetCurrentPrice(ctx, "asset3:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	price, err = keeper.GetCurrentPrice(ctx, "asset4:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+	price, err = keeper.GetCurrentPrice(ctx, "asset5:usd")
+	require.Equal(t, types.ErrNoValidPrice, err)
+}
+
+func SetCurrentPrices_EventEmission(t *testing.T, f func(ctx sdk.Context, keeper keeper.Keeper)) {
+	_, addrs := app.GeneratePrivKeyAddressPairs(5)
+	tApp := app.NewTestApp()
+	ctx := tApp.NewContext(true, tmprototypes.Header{}).
+		WithBlockTime(time.Now().UTC())
+	keeper := tApp.GetPriceFeedKeeper()
+
+	params := types.Params{
+		Markets: []types.Market{
+			{MarketID: "asset1:usd", BaseAsset: "asset1", QuoteAsset: "usd", Oracles: addrs, Active: true},
+		},
+	}
+	keeper.SetParams(ctx, params)
+
+	blockTime := time.Now()
+	initialPriceExpiry := blockTime.Add(1 * time.Hour)
+
+	// post a price
+	_, err := keeper.SetPrice(ctx, addrs[0], "asset1:usd", sdk.MustNewDecFromStr("1"), initialPriceExpiry)
+	require.NoError(t, err)
+
+	// reset context with fresh event manager
+	ctx = ctx.WithBlockTime(blockTime).WithEventManager(sdk.NewEventManager())
+	f(ctx, keeper)
+
+	// no previous price so no event
+	require.Equal(t, 0, len(ctx.EventManager().Events()))
+
+	// post same price from another oracle
+	_, err = keeper.SetPrice(ctx, addrs[1], "asset1:usd", sdk.MustNewDecFromStr("1"), initialPriceExpiry)
+	require.NoError(t, err)
+
+	blockTime = blockTime.Add(10 * time.Second)
+	ctx = ctx.WithBlockTime(blockTime).WithEventManager(sdk.NewEventManager())
+	f(ctx, keeper)
+
+	// no price change so no event
+	require.Equal(t, 0, len(ctx.EventManager().Events()))
+
+	// post price changes
+	_, err = keeper.SetPrice(ctx, addrs[2], "asset1:usd", sdk.MustNewDecFromStr("2"), initialPriceExpiry)
+	require.NoError(t, err)
+	_, err = keeper.SetPrice(ctx, addrs[3], "asset1:usd", sdk.MustNewDecFromStr("10"), initialPriceExpiry)
+	require.NoError(t, err)
+	_, err = keeper.SetPrice(ctx, addrs[4], "asset1:usd", sdk.MustNewDecFromStr("10"), initialPriceExpiry)
+	require.NoError(t, err)
+
+	blockTime = blockTime.Add(10 * time.Second)
+	ctx = ctx.WithBlockTime(blockTime).WithEventManager(sdk.NewEventManager())
+	f(ctx, keeper)
+
+	// price is changes so event should be emitted
+	require.Equal(t, 1, len(ctx.EventManager().Events()))
+
+	event := ctx.EventManager().Events()[0]
+
+	// has correct event type
+	assert.Equal(t, types.EventTypeMarketPriceUpdated, event.Type)
+	// has correct attributes
+	marketID, found := event.GetAttribute(types.AttributeMarketID)
+	require.True(t, found)
+	marketPrice, found := event.GetAttribute(types.AttributeMarketPrice)
+	require.True(t, found)
+	// attributes have correct values
+	assert.Equal(t, "asset1:usd", marketID.Value)
+	assert.Equal(t, sdk.MustNewDecFromStr("2").String(), marketPrice.Value)
+}