fix(market): narrow owner queries to the filtered deployment

x/market/keeper/grpc_query.go

@@ -148,6 +148,35 @@
 	}, nil
 }
 
+// ownerOrderPrefix extends the owner prefix with the dseq, gseq and oseq filters, in key order,
+// so a query for one deployment reads that deployment instead of everything the owner has in
+// state. A resume key outside the narrowed prefix keeps the wider one, or the range is invalid.
+func ownerOrderPrefix(owner string, dseq uint64, gseq, oseq uint32, resume *keys.OrderPrimaryKey) keys.OrderPrimaryKey {
+	if dseq == 0 || (resume != nil && resume.K2() != dseq) {
+		return collections.QuadPrefix[string, uint64, uint32, uint32](owner)
+	}
+
+	if gseq == 0 || (resume != nil && resume.K3() != gseq) {
+		return collections.QuadSuperPrefix[string, uint64, uint32, uint32](owner, dseq)
+	}
+
+	if oseq == 0 || (resume != nil && resume.K4() != oseq) {
+		return collections.QuadSuperPrefix3[string, uint64, uint32, uint32](owner, dseq, gseq)
+	}
+
+	return collections.Join4(owner, dseq, gseq, oseq)
+}
+
+// orderPartOf returns the order part of a bid or lease resume key, or nil when there is none.
+func orderPartOf(pk *collections.Pair[keys.OrderPrimaryKey, keys.ProviderPartKey]) *keys.OrderPrimaryKey {
+	if pk == nil {
+		return nil
+	}
+
+	order := pk.K1()
+	return &order
+}
+
 // ordersOwnerPath iterates the primary map with an owner prefix.
 func (k Querier) ordersOwnerPath(
 	ctx sdk.Context,
@@ -163,7 +192,7 @@
 	}
 
 	// Build range on primary map
-	prefix := collections.QuadPrefix[string, uint64, uint32, uint32](owner)
+	prefix := ownerOrderPrefix(owner, req.Filters.DSeq, req.Filters.GSeq, req.Filters.OSeq, resumePK)
 	r := new(collections.Range[keys.OrderPrimaryKey]).Prefix(prefix)
 	if resumePK != nil {
 		if req.Pagination.Reverse {
@@ -249,7 +278,7 @@
 		var iter indexes.MultiIterator[int32, keys.OrderPrimaryKey]
 		var err error
 
 		if idx == 0 && resumePK != nil {
 			r := collections.NewPrefixedPairRange[int32, keys.OrderPrimaryKey](int32(state)).StartInclusive(*resumePK)
 			if req.Pagination.Reverse {
 				r = collections.NewPrefixedPairRange[int32, keys.OrderPrimaryKey](int32(state)).EndInclusive(*resumePK).Descending()
@@ -425,7 +454,7 @@
 		stateSet[types.Bid_State(s)] = true
 	}
 
-	orderPrefix := collections.QuadPrefix[string, uint64, uint32, uint32](owner)
+	orderPrefix := ownerOrderPrefix(owner, req.Filters.DSeq, req.Filters.GSeq, req.Filters.OSeq, orderPartOf(resumePK))
 	bidPrefix := collections.PairPrefix[keys.OrderPrimaryKey, keys.ProviderPartKey](orderPrefix)
 	r := new(collections.Range[keys.BidPrimaryKey]).Prefix(bidPrefix)
 	if resumePK != nil {
@@ -804,7 +833,7 @@
 		stateSet[v1.Lease_State(s)] = true
 	}
 
-	orderPrefix := collections.QuadPrefix[string, uint64, uint32, uint32](owner)
+	orderPrefix := ownerOrderPrefix(owner, req.Filters.DSeq, req.Filters.GSeq, req.Filters.OSeq, orderPartOf(resumePK))
 	leasePrefix := collections.PairPrefix[keys.OrderPrimaryKey, keys.ProviderPartKey](orderPrefix)
 	r := new(collections.Range[keys.LeasePrimaryKey]).Prefix(leasePrefix)
 	if resumePK != nil {

x/market/keeper/grpc_query_test.go

@@ -9,10 +9,12 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
+	storetypes "cosmossdk.io/store/types"
 	"github.com/cosmos/cosmos-sdk/baseapp"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	sdkquery "github.com/cosmos/cosmos-sdk/types/query"
 
+	dv1 "pkg.akt.dev/go/node/deployment/v1"
 	mv1 "pkg.akt.dev/go/node/market/v1"
 	mvbeta "pkg.akt.dev/go/node/market/v1beta5"
 	"pkg.akt.dev/go/testutil"
@@ -1240,3 +1242,161 @@ func TestGRPCQueryLeases(t *testing.T) {
 		})
 	}
 }
+
+func TestGRPCQueryOwnerFiltersReadOnlyMatchingEntries(t *testing.T) {
+	suite := setupTest(t)
+	prepareBankMocks(suite)
+	querier := suite.keeper.NewQuerier()
+
+	owner := testutil.AccAddress(t).String()
+	target := dv1.MakeGroupID(dv1.DeploymentID{Owner: owner, DSeq: 50}, 1)
+	leaseID := createGroupLease(t, suite.TestSuite, target)
+
+	type idFilter struct {
+		dseq uint64
+		gseq uint32
+		oseq uint32
+	}
+
+	// gasByQuery runs the orders, bids and leases queries for the owner with the given ID filter,
+	// checks that each returns only the target, and returns the gas each one consumed.
+	gasByQuery := func(filter idFilter) [3]uint64 {
+		var gas [3]uint64
+
+		gas[0] = gasUsed(suite.ctx, func(ctx sdk.Context) {
+			res, err := querier.Orders(ctx, &mvbeta.QueryOrdersRequest{
+				Filters: mvbeta.OrderFilters{Owner: owner, DSeq: filter.dseq, GSeq: filter.gseq, OSeq: filter.oseq},
+			})
+			require.NoError(t, err)
+			require.Len(t, res.Orders, 1)
+			require.Equal(t, leaseID.OrderID(), res.Orders[0].ID)
+		})
+
+		gas[1] = gasUsed(suite.ctx, func(ctx sdk.Context) {
+			res, err := querier.Bids(ctx, &mvbeta.QueryBidsRequest{
+				Filters: mvbeta.BidFilters{Owner: owner, DSeq: filter.dseq, GSeq: filter.gseq, OSeq: filter.oseq},
+			})
+			require.NoError(t, err)
+			require.Len(t, res.Bids, 1)
+			require.Equal(t, leaseID.BidID(), res.Bids[0].Bid.ID)
+		})
+
+		gas[2] = gasUsed(suite.ctx, func(ctx sdk.Context) {
+			res, err := querier.Leases(ctx, &mvbeta.QueryLeasesRequest{
+				Filters: mv1.LeaseFilters{Owner: owner, DSeq: filter.dseq, GSeq: filter.gseq, OSeq: filter.oseq},
+			})
+			require.NoError(t, err)
+			require.Len(t, res.Leases, 1)
+			require.Equal(t, leaseID, res.Leases[0].Lease.ID)
+		})
+
+		return gas
+	}
+
+	byDeployment := idFilter{dseq: target.DSeq}
+	byGroup := idFilter{dseq: target.DSeq, gseq: target.GSeq}
+	byOrder := idFilter{dseq: target.DSeq, gseq: target.GSeq, oseq: leaseID.OSeq}
+
+	deploymentGas := gasByQuery(byDeployment)
+	groupGas := gasByQuery(byGroup)
+	orderGas := gasByQuery(byOrder)
+
+	// other deployments of the owner, sorting both before and after the target
+	for dseq := uint64(1); dseq <= 100; dseq++ {
+		if dseq != target.DSeq {
+			createGroupLease(t, suite.TestSuite, dv1.MakeGroupID(dv1.DeploymentID{Owner: owner, DSeq: dseq}, 1))
+		}
+	}
+
+	require.Equal(t, deploymentGas, gasByQuery(byDeployment), "dseq filter read other deployments")
+	require.Equal(t, groupGas, gasByQuery(byGroup), "gseq filter read other deployments")
+	require.Equal(t, orderGas, gasByQuery(byOrder), "oseq filter read other deployments")
+
+	// a second group in the target deployment, which only the gseq and oseq filters exclude
+	sibling, _ := createGroupBid(t, suite.TestSuite, dv1.MakeGroupID(target.DeploymentID(), 2))
+	require.NoError(t, suite.keeper.CreateLease(suite.ctx, sibling))
+
+	require.Equal(t, groupGas, gasByQuery(byGroup), "gseq filter read other groups")
+	require.Equal(t, orderGas, gasByQuery(byOrder), "oseq filter read other groups")
+}
+
+func TestGRPCQueryOwnerResumeKeyOutsideDSeqFilter(t *testing.T) {
+	suite := setupTest(t)
+	prepareBankMocks(suite)
+
+	owner := testutil.AccAddress(t).String()
+	for dseq := uint64(1); dseq <= 3; dseq++ {
+		createGroupLease(t, suite.TestSuite, dv1.MakeGroupID(dv1.DeploymentID{Owner: owner, DSeq: dseq}, 1))
+	}
+
+	// the first page ends on dseq 1, so the next key resumes at dseq 2, past a dseq 1 filter
+	firstPage := &sdkquery.PageRequest{Limit: 1}
+
+	orders, err := suite.queryClient.Orders(suite.ctx, &mvbeta.QueryOrdersRequest{
+		Filters:    mvbeta.OrderFilters{Owner: owner},
+		Pagination: firstPage,
+	})
+	require.NoError(t, err)
+	require.NotEmpty(t, orders.Pagination.NextKey)
+
+	resumedOrders, err := suite.queryClient.Orders(suite.ctx, &mvbeta.QueryOrdersRequest{
+		Filters:    mvbeta.OrderFilters{Owner: owner, DSeq: 1},
+		Pagination: &sdkquery.PageRequest{Key: orders.Pagination.NextKey},
+	})
+	require.NoError(t, err)
+	require.Empty(t, resumedOrders.Orders)
+
+	bids, err := suite.queryClient.Bids(suite.ctx, &mvbeta.QueryBidsRequest{
+		Filters:    mvbeta.BidFilters{Owner: owner},
+		Pagination: &sdkquery.PageRequest{Limit: 1},
+	})
+	require.NoError(t, err)
+	require.NotEmpty(t, bids.Pagination.NextKey)
+
+	resumedBids, err := suite.queryClient.Bids(suite.ctx, &mvbeta.QueryBidsRequest{
+		Filters:    mvbeta.BidFilters{Owner: owner, DSeq: 1},
+		Pagination: &sdkquery.PageRequest{Key: bids.Pagination.NextKey},
+	})
+	require.NoError(t, err)
+	require.Empty(t, resumedBids.Bids)
+
+	leases, err := suite.queryClient.Leases(suite.ctx, &mvbeta.QueryLeasesRequest{
+		Filters:    mv1.LeaseFilters{Owner: owner},
+		Pagination: &sdkquery.PageRequest{Limit: 1},
+	})
+	require.NoError(t, err)
+	require.NotEmpty(t, leases.Pagination.NextKey)
+
+	resumedLeases, err := suite.queryClient.Leases(suite.ctx, &mvbeta.QueryLeasesRequest{
+		Filters:    mv1.LeaseFilters{Owner: owner, DSeq: 1},
+		Pagination: &sdkquery.PageRequest{Key: leases.Pagination.NextKey},
+	})
+	require.NoError(t, err)
+	require.Empty(t, resumedLeases.Leases)
+}
+
+func prepareBankMocks(suite *grpcTestSuite) {
+	suite.PrepareMocks(func(ts *state.TestSuite) {
+		bkeeper := ts.BankKeeper()
+
+		bkeeper.
+			On("SendCoinsFromAccountToModule", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+			Return(nil)
+		bkeeper.
+			On("SendCoinsFromModuleToAccount", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+			Return(nil)
+		bkeeper.
+			On("SendCoinsFromModuleToModule", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+			Return(nil)
+
+		bkeeper.On("BurnCoins", mock.Anything, mock.Anything, mock.Anything).
+			Return(nil)
+	})
+}
+
+// gasUsed returns the store gas a query consumes, which grows with every key it reads.
+func gasUsed(ctx sdk.Context, query func(sdk.Context)) uint64 {
+	meter := storetypes.NewInfiniteGasMeter()
+	query(ctx.WithGasMeter(meter))
+	return meter.GasConsumed()
+}

x/market/keeper/keeper_test.go

@@ -9,6 +9,7 @@ import (
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
 
+	dv1 "pkg.akt.dev/go/node/deployment/v1"
 	dtypes "pkg.akt.dev/go/node/deployment/v1beta4"
 	mv1 "pkg.akt.dev/go/node/market/v1"
 	mvbeta "pkg.akt.dev/go/node/market/v1beta5"
@@ -410,9 +411,15 @@ func Test_OnGroupClosed(t *testing.T) {
 }
 
 func createLease(t testing.TB, suite *state.TestSuite) mv1.LeaseID {
+	t.Helper()
+	return createGroupLease(t, suite, dv1.MakeGroupID(testutil.DeploymentID(t), 0))
+}
+
+// createGroupLease creates an order, bid and lease for the given group.
+func createGroupLease(t testing.TB, suite *state.TestSuite, gid dv1.GroupID) mv1.LeaseID {
 	t.Helper()
 	ctx := suite.Context()
-	bid, order := createBid(t, suite)
+	bid, order := createGroupBid(t, suite, gid)
 	keeper := suite.MarketKeeper()
 
 	err := keeper.CreateLease(ctx, bid)
@@ -460,9 +467,15 @@ func createLease(t testing.TB, suite *state.TestSuite) mv1.LeaseID {
 }
 
 func createBid(t testing.TB, suite *state.TestSuite) (mvbeta.Bid, mvbeta.Order) {
+	t.Helper()
+	return createGroupBid(t, suite, dv1.MakeGroupID(testutil.DeploymentID(t), 0))
+}
+
+// createGroupBid creates an order and a bid on it for the given group.
+func createGroupBid(t testing.TB, suite *state.TestSuite, gid dv1.GroupID) (mvbeta.Bid, mvbeta.Order) {
 	t.Helper()
 	ctx := suite.Context()
-	order, gspec := createOrder(t, suite.Context(), suite.MarketKeeper())
+	order, gspec := createGroupOrder(t, suite.Context(), suite.MarketKeeper(), gid)
 	provider := testutil.AccAddress(t)
 	price := testutil.ACTDecCoinRandom(t)
 	roffer := mvbeta.ResourceOfferFromRU(gspec.Resources)
@@ -498,7 +511,13 @@ func createBid(t testing.TB, suite *state.TestSuite) (mvbeta.Bid, mvbeta.Order)
 
 func createOrder(t testing.TB, ctx sdk.Context, keeper keeper.IKeeper) (mvbeta.Order, dtypes.GroupSpec) {
 	t.Helper()
-	group := testutil.DeploymentGroup(t, testutil.DeploymentID(t), 0)
+	return createGroupOrder(t, ctx, keeper, dv1.MakeGroupID(testutil.DeploymentID(t), 0))
+}
+
+// createGroupOrder creates an order for the given group.
+func createGroupOrder(t testing.TB, ctx sdk.Context, keeper keeper.IKeeper, gid dv1.GroupID) (mvbeta.Order, dtypes.GroupSpec) {
+	t.Helper()
+	group := testutil.DeploymentGroup(t, gid.DeploymentID(), gid.GSeq)
 
 	order, err := keeper.CreateOrder(ctx, group.ID, group.GroupSpec, nil)
 	require.NoError(t, err)