node/meta: fill full object index in container storage

Make underlying KV database store the same index structure as in the new
metabase structure for search V2. Exclude repetitive fields from the database in
the old MPT form. Receive object headers via internal object get (head) service.
Closes #3139.

Signed-off-by: Pavel Karpy <[email protected]>

Author: carpawell

cmd/neofs-node/config.go

@@ -441,7 +441,7 @@ type cfgGRPC struct {
 }
 
 type cfgMeta struct {
-	cLister meta.ContainerLister
+	network meta.NeoFSNetwork
 }
 
 type cfgMorph struct {

cmd/neofs-node/meta.go

@@ -11,8 +11,11 @@ import (
 	"github.com/nspcc-dev/neofs-node/pkg/core/netmap"
 	cntClient "github.com/nspcc-dev/neofs-node/pkg/morph/client/container"
 	"github.com/nspcc-dev/neofs-node/pkg/services/meta"
+	getsvc "github.com/nspcc-dev/neofs-node/pkg/services/object/get"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
 	netmapsdk "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"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 )
@@ -22,22 +25,23 @@ func initMeta(c *cfg) {
 		initMorphComponents(c)
 	}
 
-	c.cfgMeta.cLister = &metaContainerListener{
+	c.cfgMeta.network = &neofsNetwork{
 		key:        c.binPublicKey,
 		cnrClient:  c.basics.cCli,
 		containers: c.cfgObject.cnrSource,
 		network:    c.basics.netMapSource,
+		header:     c.cfgObject.getSvc,
 	}
 
 	var err error
 	p := meta.Parameters{
-		Logger:          c.log.With(zap.String("service", "meta data")),
-		ContainerLister: c.cfgMeta.cLister,
-		Timeout:         c.applicationConfiguration.fsChain.dialTimeout,
-		NeoEnpoints:     c.applicationConfiguration.fsChain.endpoints,
-		ContainerHash:   c.basics.containerSH,
-		NetmapHash:      c.basics.netmapSH,
-		RootPath:        c.applicationConfiguration.metadata.path,
+		Logger:        c.log.With(zap.String("service", "meta data")),
+		Network:       c.cfgMeta.network,
+		Timeout:       c.applicationConfiguration.fsChain.dialTimeout,
+		NeoEnpoints:   c.applicationConfiguration.fsChain.endpoints,
+		ContainerHash: c.basics.containerSH,
+		NetmapHash:    c.basics.netmapSH,
+		RootPath:      c.applicationConfiguration.metadata.path,
 	}
 	c.shared.metaService, err = meta.New(p)
 	fatalOnErr(err)
@@ -50,20 +54,36 @@ func initMeta(c *cfg) {
 	}))
 }
 
-type metaContainerListener struct {
+type neofsNetwork struct {
 	key []byte
 
 	cnrClient  *cntClient.Client
 	containers container.Source
 	network    netmap.Source
+	header     *getsvc.Service
 
 	m          sync.RWMutex
 	prevCnrs   []cid.ID
 	prevNetMap *netmapsdk.NetMap
 	prevRes    map[cid.ID]struct{}
 }
 
-func (c *metaContainerListener) IsMineWithMeta(id cid.ID) (bool, error) {
+func (c *neofsNetwork) Head(ctx context.Context, cID cid.ID, oID oid.ID) (object.Object, error) {
+	var hw headerWriter
+	var addr oid.Address
+	addr.SetContainer(cID)
+	addr.SetObject(oID)
+
+	var hPrm getsvc.HeadPrm
+	hPrm.SetHeaderWriter(&hw)
+	hPrm.WithAddress(addr)
+
+	err := c.header.Head(ctx, hPrm)
+
+	return *hw.h, err
+}
+
+func (c *neofsNetwork) IsMineWithMeta(id cid.ID) (bool, error) {
 	curEpoch, err := c.network.Epoch()
 	if err != nil {
 		return false, fmt.Errorf("read current NeoFS epoch: %w", err)
@@ -75,7 +95,7 @@ func (c *metaContainerListener) IsMineWithMeta(id cid.ID) (bool, error) {
 	return c.isMineWithMeta(id, networkMap)
 }
 
-func (c *metaContainerListener) isMineWithMeta(id cid.ID, networkMap *netmapsdk.NetMap) (bool, error) {
+func (c *neofsNetwork) isMineWithMeta(id cid.ID, networkMap *netmapsdk.NetMap) (bool, error) {
 	cnr, err := c.containers.Get(id)
 	if err != nil {
 		return false, fmt.Errorf("read %s container: %w", id, err)
@@ -104,7 +124,7 @@ func (c *metaContainerListener) isMineWithMeta(id cid.ID, networkMap *netmapsdk.
 	return false, nil
 }
 
-func (c *metaContainerListener) List() (map[cid.ID]struct{}, error) {
+func (c *neofsNetwork) List() (map[cid.ID]struct{}, error) {
 	actualContainers, err := c.cnrClient.List(nil)
 	if err != nil {
 		return nil, fmt.Errorf("read containers: %w", err)

pkg/services/meta/blocks.go

@@ -8,9 +8,10 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/core/mpt"
 	"github.com/nspcc-dev/neo-go/pkg/neorpc"
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 )
 
-func (m *Meta) handleBlock(b *block.Header) error {
+func (m *Meta) handleBlock(ctx context.Context, b *block.Header) error {
 	h := b.Hash()
 	ind := b.Index
 	l := m.l.With(zap.Stringer("block hash", h), zap.Uint32("index", ind))
@@ -35,6 +36,11 @@ func (m *Meta) handleBlock(b *block.Header) error {
 	m.m.RLock()
 	defer m.m.RUnlock()
 
+	var wg errgroup.Group
+	wg.SetLimit(1024)
+	ctx, cancel := context.WithTimeout(ctx, m.timeout)
+	defer cancel()
+
 	for _, n := range res.Application {
 		ev, err := parseObjNotification(n)
 		if err != nil {
@@ -48,13 +54,21 @@ func (m *Meta) handleBlock(b *block.Header) error {
 			continue
 		}
 
-		err = m.handleObjectNotification(s, ev)
-		if err != nil {
-			l.Error("handling object notification", zap.Error(err))
-			continue
-		}
+		wg.Go(func() error {
+			err := m.handleObjectNotification(ctx, s, ev)
+			if err != nil {
+				return fmt.Errorf("handling %s/%s object notification: %w", ev.cID, ev.oID, err)
+			}
+
+			l.Debug("handled object notification successfully", zap.Stringer("cID", ev.cID), zap.Stringer("oID", ev.oID))
 
-		l.Debug("handled object notification successfully", zap.Stringer("cID", ev.cID), zap.Stringer("oID", ev.oID))
+			return nil
+		})
+	}
+
+	err = wg.Wait()
+	if err != nil {
+		l.Error("failed to handle block's notifications", zap.Error(err))
 	}
 
 	for _, st := range m.storages {
@@ -91,7 +105,7 @@ func (m *Meta) blockFetcher(ctx context.Context, buff <-chan *block.Header) {
 		case <-ctx.Done():
 			return
 		case b := <-buff:
-			err := m.handleBlock(b)
+			err := m.handleBlock(ctx, b)
 			if err != nil {
 				m.l.Error("block handling failed", zap.Error(err))
 				continue