Fix incentives query filtering

CHANGELOG.md

@@ -63,6 +63,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Bug Fixes
 * [1700](https://github.com/osmosis-labs/osmosis/pull/1700) Upgrade sdk fork with missing snapshot manager fix.
 * [1716](https://github.com/osmosis-labs/osmosis/pull/1716) Fix secondary over-LP shares bug with uneven swap amounts in `CalcJoinPoolShares`.
+* [1759](https://github.com/osmosis-labs/osmosis/pull/1759) Fix pagination filter in incentives query.
 
 ## [v9.0.0 - Nitrogen](https://github.com/osmosis-labs/osmosis/releases/tag/v9.0.0)
 

x/incentives/keeper/grpc_query.go

@@ -63,19 +63,7 @@ func (q Querier) Gauges(goCtx context.Context, req *types.GaugesRequest) (*types
 	}
 
 	ctx := sdk.UnwrapSDKContext(goCtx)
-	gauges := []types.Gauge{}
-	store := ctx.KVStore(q.Keeper.storeKey)
-	valStore := prefix.NewStore(store, types.KeyPrefixGauges)
-
-	pageRes, err := query.FilteredPaginate(valStore, req.Pagination, func(key []byte, value []byte, accumulate bool) (bool, error) {
-		newGauges, err := q.getGaugeFromIDJsonBytes(ctx, value)
-		if err != nil {
-			panic(err)
-		}
-		gauges = append(gauges, newGauges...)
-
-		return true, nil
-	})
+	pageRes, gauges, err := q.filterByPrefixAndDenom(ctx, types.KeyPrefixGauges, "", req.Pagination)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@@ -90,19 +78,8 @@ func (q Querier) ActiveGauges(goCtx context.Context, req *types.ActiveGaugesRequ
 	}
 
 	ctx := sdk.UnwrapSDKContext(goCtx)
-	gauges := []types.Gauge{}
-	store := ctx.KVStore(q.Keeper.storeKey)
-	valStore := prefix.NewStore(store, types.KeyPrefixActiveGauges)
-
-	pageRes, err := query.FilteredPaginate(valStore, req.Pagination, func(key []byte, value []byte, accumulate bool) (bool, error) {
-		newGauges, err := q.getGaugeFromIDJsonBytes(ctx, value)
-		if err != nil {
-			panic(err)
-		}
-		gauges = append(gauges, newGauges...)
 
-		return true, nil
-	})
+	pageRes, gauges, err := q.filterByPrefixAndDenom(ctx, types.KeyPrefixActiveGauges, "", req.Pagination)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@@ -117,19 +94,7 @@ func (q Querier) ActiveGaugesPerDenom(goCtx context.Context, req *types.ActiveGa
 	}
 
 	ctx := sdk.UnwrapSDKContext(goCtx)
-	gauges := []types.Gauge{}
-	store := ctx.KVStore(q.Keeper.storeKey)
-	valStore := prefix.NewStore(store, types.KeyPrefixActiveGauges)
-
-	pageRes, err := query.FilteredPaginate(valStore, req.Pagination, func(key []byte, value []byte, accumulate bool) (bool, error) {
-		activeGauges := q.Keeper.GetActiveGauges(ctx)
-		for _, gauge := range activeGauges {
-			if gauge.DistributeTo.Denom == req.Denom {
-				gauges = append(gauges, gauge)
-			}
-		}
-		return true, nil
-	})
+	pageRes, gauges, err := q.filterByPrefixAndDenom(ctx, types.KeyPrefixActiveGauges, req.Denom, req.Pagination)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@@ -144,19 +109,8 @@ func (q Querier) UpcomingGauges(goCtx context.Context, req *types.UpcomingGauges
 	}
 
 	ctx := sdk.UnwrapSDKContext(goCtx)
-	gauges := []types.Gauge{}
-	store := ctx.KVStore(q.Keeper.storeKey)
-	valStore := prefix.NewStore(store, types.KeyPrefixUpcomingGauges)
 
-	pageRes, err := query.FilteredPaginate(valStore, req.Pagination, func(key []byte, value []byte, accumulate bool) (bool, error) {
-		newGauges, err := q.getGaugeFromIDJsonBytes(ctx, value)
-		if err != nil {
-			panic(err)
-		}
-		gauges = append(gauges, newGauges...)
-
-		return true, nil
-	})
+	pageRes, gauges, err := q.filterByPrefixAndDenom(ctx, types.KeyPrefixUpcomingGauges, "", req.Pagination)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@@ -174,19 +128,7 @@ func (q Querier) UpcomingGaugesPerDenom(goCtx context.Context, req *types.Upcomi
 		return nil, status.Error(codes.InvalidArgument, "invalid denom")
 	}
 
-	gauges := []types.Gauge{}
-	store := ctx.KVStore(q.Keeper.storeKey)
-	prefixStore := prefix.NewStore(store, types.KeyPrefixUpcomingGauges)
-
-	pageRes, err := query.FilteredPaginate(prefixStore, req.Pagination, func(key []byte, value []byte, accumulate bool) (bool, error) {
-		upcomingGauges := q.Keeper.GetUpcomingGauges(ctx)
-		for _, gauge := range upcomingGauges {
-			if gauge.DistributeTo.Denom == req.Denom {
-				gauges = append(gauges, gauge)
-			}
-		}
-		return true, nil
-	})
+	pageRes, gauges, err := q.filterByPrefixAndDenom(ctx, types.KeyPrefixUpcomingGauges, req.Denom, req.Pagination)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@@ -253,3 +195,33 @@ func (q Querier) getGaugeFromIDJsonBytes(ctx sdk.Context, refValue []byte) ([]ty
 
 	return gauges, nil
 }
+
+// Filter gauges based on given prefix type and denom
+func (q Querier) filterByPrefixAndDenom(ctx sdk.Context, prefixType []byte, denom string, pagination *query.PageRequest) (*query.PageResponse, []types.Gauge, error) {
+	gauges := []types.Gauge{}
+	store := ctx.KVStore(q.Keeper.storeKey)
+	valStore := prefix.NewStore(store, prefixType)
+
+	pageRes, err := query.FilteredPaginate(valStore, pagination, func(key []byte, value []byte, accumulate bool) (bool, error) {
+		// This may return multiple gauges at once if two gauges start at the same time.
+		// For now this is treated as an edge case that is not of importance
+		newGauges, err := q.getGaugeFromIDJsonBytes(ctx, value)
+		if err != nil {
+			return false, err
+		}
+		if accumulate {
+			if denom != "" {
+				for _, gauge := range newGauges {
+					if gauge.DistributeTo.Denom != denom {
+						return false, nil
+					}
+					gauges = append(gauges, gauge)
+				}
+			} else {
+				gauges = append(gauges, newGauges...)
+			}
+		}
+		return true, nil
+	})
+	return pageRes, gauges, err
+}

x/incentives/keeper/grpc_query_test.go

@@ -1,10 +1,12 @@
 package keeper_test
 
 import (
+	// "fmt"
 	"time"
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
 
+	query "github.com/cosmos/cosmos-sdk/types/query"
 	"github.com/osmosis-labs/osmosis/v7/x/incentives/types"
 	lockuptypes "github.com/osmosis-labs/osmosis/v7/x/lockup/types"
 	pooltypes "github.com/osmosis-labs/osmosis/v7/x/pool-incentives/types"
@@ -72,6 +74,16 @@ func (suite *KeeperTestSuite) TestGRPCGauges() {
 		StartTime:         startTime,
 	}
 	suite.Require().Equal(res.Data[0].String(), expectedGauge.String())
+
+	// filtering check
+	for i := 0; i < 10; i++ {
+		suite.SetupNewGauge(false, sdk.Coins{sdk.NewInt64Coin("stake", 3)})
+		suite.Ctx = suite.Ctx.WithBlockTime(startTime.Add(time.Second))
+	}
+
+	filter := query.PageRequest{Limit: 10}
+	res, err = suite.querier.Gauges(sdk.WrapSDKContext(suite.Ctx), &types.GaugesRequest{Pagination: &filter})
+	suite.Require().Len(res.Data, 10)
 }
 
 func (suite *KeeperTestSuite) TestGRPCActiveGauges() {
@@ -107,6 +119,23 @@ func (suite *KeeperTestSuite) TestGRPCActiveGauges() {
 		StartTime:         startTime,
 	}
 	suite.Require().Equal(res.Data[0].String(), expectedGauge.String())
+
+	// filtering check
+	for i := 0; i < 20; i++ {
+		_, gauge, _, _ := suite.SetupNewGauge(false, sdk.Coins{sdk.NewInt64Coin("stake", 3)})
+		suite.Ctx = suite.Ctx.WithBlockTime(startTime.Add(time.Second))
+
+		// set up more 9 active gauges => 10
+		if i < 9 {
+			suite.querier.MoveUpcomingGaugeToActiveGauge(suite.Ctx, *gauge)
+		}
+	}
+
+	res, err = suite.querier.ActiveGauges(sdk.WrapSDKContext(suite.Ctx), &types.ActiveGaugesRequest{Pagination: &query.PageRequest{Limit: 5}})
+	suite.Require().Len(res.Data, 5)
+
+	res, err = suite.querier.ActiveGauges(sdk.WrapSDKContext(suite.Ctx), &types.ActiveGaugesRequest{Pagination: &query.PageRequest{Limit: 15}})
+	suite.Require().Len(res.Data, 10)
 }
 
 func (suite *KeeperTestSuite) TestGRPCActiveGaugesPerDenom() {
@@ -140,6 +169,26 @@ func (suite *KeeperTestSuite) TestGRPCActiveGaugesPerDenom() {
 		StartTime:         startTime,
 	}
 	suite.Require().Equal(res.Data[0].String(), expectedGauge.String())
+
+	// filtering check
+	for i := 0; i < 20; i++ {
+		_, gauge, _, _ := suite.SetupNewGaugeWithDenom(false, sdk.Coins{sdk.NewInt64Coin("stake", 3)}, "pool")
+		suite.Ctx = suite.Ctx.WithBlockTime(startTime.Add(time.Second))
+
+		// set up 10 active gauges with "pool" denom
+		if i < 10 {
+			suite.querier.MoveUpcomingGaugeToActiveGauge(suite.Ctx, *gauge)
+		}
+	}
+
+	res, err = suite.querier.ActiveGaugesPerDenom(sdk.WrapSDKContext(suite.Ctx), &types.ActiveGaugesPerDenomRequest{Denom: "lptoken", Pagination: &query.PageRequest{Limit: 5}})
+	suite.Require().Len(res.Data, 1)
+
+	res, err = suite.querier.ActiveGaugesPerDenom(sdk.WrapSDKContext(suite.Ctx), &types.ActiveGaugesPerDenomRequest{Denom: "pool", Pagination: &query.PageRequest{Limit: 5}})
+	suite.Require().Len(res.Data, 5)
+
+	res, err = suite.querier.ActiveGaugesPerDenom(sdk.WrapSDKContext(suite.Ctx), &types.ActiveGaugesPerDenomRequest{Denom: "pool", Pagination: &query.PageRequest{Limit: 15}})
+	suite.Require().Len(res.Data, 10)
 }
 
 func (suite *KeeperTestSuite) TestGRPCUpcomingGauges() {
@@ -172,6 +221,23 @@ func (suite *KeeperTestSuite) TestGRPCUpcomingGauges() {
 		StartTime:         startTime,
 	}
 	suite.Require().Equal(res.Data[0].String(), expectedGauge.String())
+
+	// filtering check
+	for i := 0; i < 20; i++ {
+		_, gauge, _, _ := suite.SetupNewGauge(false, sdk.Coins{sdk.NewInt64Coin("stake", 3)})
+		suite.Ctx = suite.Ctx.WithBlockTime(startTime.Add(time.Second))
+
+		// set up more 9 active gauges => Upcoming = 1 + (20 -9) = 12
+		if i < 9 {
+			suite.querier.MoveUpcomingGaugeToActiveGauge(suite.Ctx, *gauge)
+		}
+	}
+
+	res, err = suite.querier.UpcomingGauges(sdk.WrapSDKContext(suite.Ctx), &types.UpcomingGaugesRequest{Pagination: &query.PageRequest{Limit: 5}})
+	suite.Require().Len(res.Data, 5)
+
+	res, err = suite.querier.UpcomingGauges(sdk.WrapSDKContext(suite.Ctx), &types.UpcomingGaugesRequest{Pagination: &query.PageRequest{Limit: 15}})
+	suite.Require().Len(res.Data, 12)
 }
 
 func (suite *KeeperTestSuite) TestGRPCUpcomingGaugesPerDenom() {
@@ -210,6 +276,26 @@ func (suite *KeeperTestSuite) TestGRPCUpcomingGaugesPerDenom() {
 	res, err = suite.querier.UpcomingGaugesPerDenom(sdk.WrapSDKContext(suite.Ctx), &upcomingGaugeRequest)
 	suite.Require().NoError(err)
 	suite.Require().Len(res.UpcomingGauges, 0)
+
+	// filtering check
+	for i := 0; i < 20; i++ {
+		_, gauge, _, _ := suite.SetupNewGaugeWithDenom(false, sdk.Coins{sdk.NewInt64Coin("stake", 3)}, "pool")
+		suite.Ctx = suite.Ctx.WithBlockTime(startTime.Add(time.Second))
+
+		// set up 10 active gauges with "pool" denom => 10 upcoming
+		if i < 10 {
+			suite.querier.MoveUpcomingGaugeToActiveGauge(suite.Ctx, *gauge)
+		}
+	}
+
+	res, err = suite.querier.UpcomingGaugesPerDenom(sdk.WrapSDKContext(suite.Ctx), &types.UpcomingGaugesPerDenomRequest{Denom: "lptoken", Pagination: &query.PageRequest{Limit: 5}})
+	suite.Require().Len(res.UpcomingGauges, 0)
+
+	res, err = suite.querier.UpcomingGaugesPerDenom(sdk.WrapSDKContext(suite.Ctx), &types.UpcomingGaugesPerDenomRequest{Denom: "pool", Pagination: &query.PageRequest{Limit: 5}})
+	suite.Require().Len(res.UpcomingGauges, 5)
+
+	res, err = suite.querier.UpcomingGaugesPerDenom(sdk.WrapSDKContext(suite.Ctx), &types.UpcomingGaugesPerDenomRequest{Denom: "pool", Pagination: &query.PageRequest{Limit: 15}})
+	suite.Require().Len(res.UpcomingGauges, 10)
 }
 
 func (suite *KeeperTestSuite) TestGRPCRewardsEst() {

x/incentives/keeper/suite_test.go

@@ -129,6 +129,33 @@ func (suite *KeeperTestSuite) SetupNewGauge(isPerpetual bool, coins sdk.Coins) (
 	return suite.setupNewGaugeWithDuration(isPerpetual, coins, defaultLockDuration)
 }
 
+func (suite *KeeperTestSuite) setupNewGaugeWithDenom(isPerpetual bool, coins sdk.Coins, duration time.Duration, denom string) (
+	uint64, *types.Gauge, sdk.Coins, time.Time,
+) {
+	addr := sdk.AccAddress([]byte("Gauge_Creation_Addr_"))
+	startTime2 := time.Now()
+	distrTo := lockuptypes.QueryCondition{
+		LockQueryType: lockuptypes.ByDuration,
+		Denom:         denom,
+		Duration:      duration,
+	}
+
+	// mints coins so supply exists on chain
+	mintCoins := sdk.Coins{sdk.NewInt64Coin(distrTo.Denom, 200)}
+	suite.FundAcc(addr, mintCoins)
+
+	numEpochsPaidOver := uint64(2)
+	if isPerpetual {
+		numEpochsPaidOver = uint64(1)
+	}
+	gaugeID, gauge := suite.CreateGauge(isPerpetual, addr, coins, distrTo, startTime2, numEpochsPaidOver)
+	return gaugeID, gauge, coins, startTime2
+}
+
+func (suite *KeeperTestSuite) SetupNewGaugeWithDenom(isPerpetual bool, coins sdk.Coins, denom string) (uint64, *types.Gauge, sdk.Coins, time.Time) {
+	return suite.setupNewGaugeWithDenom(isPerpetual, coins, defaultLockDuration, denom)
+}
+
 func (suite *KeeperTestSuite) SetupManyLocks(numLocks int, liquidBalance sdk.Coins, coinsPerLock sdk.Coins,
 	lockDuration time.Duration,
 ) []sdk.AccAddress {