ir/audit: Use SearchV2 for storage group listing

New search allows to filter out expired storage groups at the listing
stage.

The code volume is reduced in concert.

Closes #3144. Refs #3058.

Signed-off-by: Leonard Lyubich <[email protected]>

Author: cthulhu-rider

CHANGELOG.md

@@ -8,6 +8,7 @@ Changelog for NeoFS Node
 ### Fixed
 
 ### Changed
+- IR calls `ObjectService.SearchV2` to select SG objects now (#3144)
 
 ### Removed
 

pkg/core/client/client.go

@@ -26,6 +26,7 @@ type Client interface {
 	ObjectGetInit(ctx context.Context, containerID cid.ID, objectID oid.ID, signer user.Signer, prm client.PrmObjectGet) (object.Object, *client.PayloadReader, error)
 	ObjectHead(ctx context.Context, containerID cid.ID, objectID oid.ID, signer user.Signer, prm client.PrmObjectHead) (*object.Object, error)
 	ObjectSearchInit(ctx context.Context, containerID cid.ID, signer user.Signer, prm client.PrmObjectSearch) (*client.ObjectListReader, error)
+	SearchObjects(context.Context, cid.ID, object.SearchFilters, []string, string, neofscrypto.Signer, client.SearchObjectsOptions) ([]client.SearchResultItem, string, error)
 	ObjectRangeInit(ctx context.Context, containerID cid.ID, objectID oid.ID, offset, length uint64, signer user.Signer, prm client.PrmObjectRange) (*client.ObjectRangeReader, error)
 	ObjectHash(ctx context.Context, containerID cid.ID, objectID oid.ID, signer user.Signer, prm client.PrmObjectHash) ([][]byte, error)
 	AnnounceLocalTrust(ctx context.Context, epoch uint64, trusts []reputationSDK.Trust, prm client.PrmAnnounceLocalTrust) error

pkg/core/storagegroup/storagegroup.go

@@ -10,20 +10,6 @@ import (
 	"github.com/nspcc-dev/neofs-sdk-go/storagegroup"
 )
 
-// SearchSGPrm groups the parameters which are formed by Processor to search the storage group objects.
-type SearchSGPrm struct {
-	Context context.Context
-
-	Container cid.ID
-
-	NodeInfo client.NodeInfo
-}
-
-// SearchSGDst groups the target values which Processor expects from SG searching to process.
-type SearchSGDst struct {
-	Objects []oid.ID
-}
-
 // GetSGPrm groups parameter of GetSG operation.
 type GetSGPrm struct {
 	Context context.Context
@@ -37,10 +23,11 @@ type GetSGPrm struct {
 
 // SGSource is a storage group information source interface.
 type SGSource interface {
-	// ListSG must list storage group objects in the container. Formed list must be written to destination.
+	// ListSG must list container's storage group objects not expired at the
+	// specified current epoch.
 	//
 	// Must return any error encountered which did not allow to form the list.
-	ListSG(*SearchSGDst, SearchSGPrm) error
+	ListSG(context.Context, client.NodeInfo, cid.ID, uint64) ([]oid.ID, error)
 
 	// GetSG must return storage group object for the provided CID, OID,
 	// container and netmap state.

pkg/innerring/internal/client/client.go

@@ -8,9 +8,7 @@ import (
 	"io"
 
 	clientcore "github.com/nspcc-dev/neofs-node/pkg/core/client"
-	"github.com/nspcc-dev/neofs-node/pkg/services/object_manager/storagegroup"
 	"github.com/nspcc-dev/neofs-sdk-go/client"
-	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
 	"github.com/nspcc-dev/neofs-sdk-go/object"
 	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
 	"github.com/nspcc-dev/neofs-sdk-go/user"
@@ -33,57 +31,6 @@ func (x *Client) SetPrivateKey(key *ecdsa.PrivateKey) {
 	x.signer = user.NewAutoIDSigner(*key)
 }
 
-// SearchSGPrm groups parameters of SearchSG operation.
-type SearchSGPrm struct {
-	contextPrm
-
-	cnrID cid.ID
-}
-
-// SetContainerID sets the ID of the container to search for storage groups.
-func (x *SearchSGPrm) SetContainerID(id cid.ID) {
-	x.cnrID = id
-}
-
-// SearchSGRes groups the resulting values of SearchSG operation.
-type SearchSGRes struct {
-	cliRes []oid.ID
-}
-
-// IDList returns a list of IDs of storage groups in the container.
-func (x SearchSGRes) IDList() []oid.ID {
-	return x.cliRes
-}
-
-var sgFilter = storagegroup.SearchQuery()
-
-// SearchSG lists objects of storage group type in the container.
-//
-// Returns any error which prevented the operation from completing correctly in error return.
-func (x Client) SearchSG(prm SearchSGPrm) (*SearchSGRes, error) {
-	var cliPrm client.PrmObjectSearch
-	cliPrm.SetFilters(sgFilter)
-
-	rdr, err := x.c.ObjectSearchInit(prm.ctx, prm.cnrID, x.signer, cliPrm)
-	if err != nil {
-		return nil, fmt.Errorf("init object search: %w", err)
-	}
-
-	var list []oid.ID
-
-	err = rdr.Iterate(func(id oid.ID) bool {
-		list = append(list, id)
-		return false
-	})
-	if err != nil {
-		return nil, fmt.Errorf("search objects using NeoFS API: %w", err)
-	}
-
-	return &SearchSGRes{
-		cliRes: list,
-	}, nil
-}
-
 // GetObjectPrm groups parameters of GetObject operation.
 type GetObjectPrm struct {
 	getObjectPrm

pkg/innerring/processors/audit/process.go

@@ -118,12 +118,7 @@ func (ap *Processor) findStorageGroups(cnr cid.ID, shuffled netmapcore.Nodes) []
 
 	ln := len(shuffled)
 
-	var (
-		info clientcore.NodeInfo
-		prm  storagegroup.SearchSGPrm
-	)
-
-	prm.Container = cnr
+	var info clientcore.NodeInfo
 
 	for i := range shuffled { // consider iterating over some part of container
 		log := ap.log.With(
@@ -141,22 +136,15 @@ func (ap *Processor) findStorageGroups(cnr cid.ID, shuffled netmapcore.Nodes) []
 		}
 
 		ctx, cancel := context.WithTimeout(context.Background(), ap.searchTimeout)
-
-		prm.Context = ctx
-		prm.NodeInfo = info
-
-		var dst storagegroup.SearchSGDst
-
-		err = ap.sgSrc.ListSG(&dst, prm)
-
+		ids, err := ap.sgSrc.ListSG(ctx, info, cnr, ap.epochSrc.EpochCounter())
 		cancel()
 
 		if err != nil {
 			log.Warn("error in storage group search", zap.Error(err))
 			continue
 		}
 
-		sg = append(sg, dst.Objects...)
+		sg = append(sg, ids...)
 
 		break // we found storage groups, so break loop
 	}

pkg/innerring/rpc.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"fmt"
+	"slices"
+	"strconv"
 	"sync"
 	"time"
 
@@ -14,7 +16,11 @@ import (
 	"github.com/nspcc-dev/neofs-node/pkg/network/cache"
 	"github.com/nspcc-dev/neofs-node/pkg/services/audit/auditor"
 	"github.com/nspcc-dev/neofs-node/pkg/services/object_manager/placement"
+	storagegroupsvc "github.com/nspcc-dev/neofs-node/pkg/services/object_manager/storagegroup"
+	"github.com/nspcc-dev/neofs-sdk-go/client"
 	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
+	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
+	neofsecdsa "github.com/nspcc-dev/neofs-sdk-go/crypto/ecdsa"
 	"github.com/nspcc-dev/neofs-sdk-go/netmap"
 	"github.com/nspcc-dev/neofs-sdk-go/object"
 	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
@@ -226,23 +232,29 @@ func (c *ClientCache) getWrappedClient(info clientcore.NodeInfo) (neofsapiclient
 	return cInternal, nil
 }
 
-func (c ClientCache) ListSG(dst *storagegroup2.SearchSGDst, prm storagegroup2.SearchSGPrm) error {
-	cli, err := c.getWrappedClient(prm.NodeInfo)
+func (c ClientCache) ListSG(ctx context.Context, node clientcore.NodeInfo, cnr cid.ID, notExpiredAt uint64) ([]oid.ID, error) {
+	cli, err := c.Get(node)
 	if err != nil {
-		return fmt.Errorf("could not get API client from cache")
+		return nil, fmt.Errorf("could not get API client from cache: %w", err)
 	}
 
-	var cliPrm neofsapiclient.SearchSGPrm
-
-	cliPrm.SetContext(prm.Context)
-	cliPrm.SetContainerID(prm.Container)
-
-	res, err := cli.SearchSG(cliPrm)
-	if err != nil {
-		return err
+	fs := storagegroupsvc.SearchQuery()
+	fs.AddFilter(object.AttributeExpirationEpoch, strconv.FormatUint(notExpiredAt, 10), object.MatchNumGE)
+	var opts client.SearchObjectsOptions
+	var cursor string
+	var next []client.SearchResultItem
+	var res []oid.ID
+	for {
+		next, cursor, err = cli.SearchObjects(ctx, cnr, fs, nil, cursor, (*neofsecdsa.Signer)(c.key), opts)
+		if err != nil {
+			return nil, fmt.Errorf("search objects RPC: %w", err)
+		}
+		res = slices.Grow(res, len(res)+len(next))
+		for i := range next {
+			res = append(res, next[i].ID)
+		}
+		if cursor == "" {
+			return res, nil
+		}
 	}
-
-	dst.Objects = res.IDList()
-
-	return nil
 }

pkg/network/cache/clients.go

@@ -394,6 +394,17 @@ func (x *connections) AnnounceIntermediateTrust(ctx context.Context, epoch uint6
 	})
 }
 
+func (x *connections) SearchObjects(ctx context.Context, cnr cid.ID, fs object.SearchFilters, attrs []string, cursor string,
+	signer neofscrypto.Signer, opts client.SearchObjectsOptions) ([]client.SearchResultItem, string, error) {
+	var resItems []client.SearchResultItem
+	var resCursor string
+	return resItems, resCursor, x.forEach(ctx, func(ctx context.Context, c *client.Client) error {
+		var err error
+		resItems, resCursor, err = c.SearchObjects(ctx, cnr, fs, attrs, cursor, signer, opts)
+		return err
+	})
+}
+
 func isTempError(err error) bool {
 	st, ok := status.FromError(err)
 	return ok && st.Code() == codes.Unavailable

pkg/network/cache/multi.go

@@ -322,6 +322,16 @@ func (x *multiClient) ObjectSearchInit(ctx context.Context, containerID cid.ID,
 	return
 }
 
+func (x *multiClient) SearchObjects(ctx context.Context, cnr cid.ID, fs objectSDK.SearchFilters, attrs []string, cursor string,
+	signer neofscrypto.Signer, opts client.SearchObjectsOptions) ([]client.SearchResultItem, string, error) {
+	var res []client.SearchResultItem
+	return res, cursor, x.iterateClients(ctx, func(c clientcore.Client) error {
+		var err error
+		res, cursor, err = c.SearchObjects(ctx, cnr, fs, attrs, cursor, signer, opts)
+		return err
+	})
+}
+
 func (x *multiClient) AnnounceLocalTrust(ctx context.Context, epoch uint64, trusts []reputationSDK.Trust, prm client.PrmAnnounceLocalTrust) error {
 	return x.iterateClients(ctx, func(c clientcore.Client) error {
 		return c.AnnounceLocalTrust(ctx, epoch, trusts, prm)