meta: do not try processing non-meta-enabled containers (#3173)

Author: roman-khimov

cmd/neofs-node/meta.go

@@ -63,6 +63,47 @@ type metaContainerListener struct {
 	prevRes    map[cid.ID]struct{}
 }
 
+func (c *metaContainerListener) IsMineWithMeta(id cid.ID) (bool, error) {
+	curEpoch, err := c.network.Epoch()
+	if err != nil {
+		return false, fmt.Errorf("read current NeoFS epoch: %w", err)
+	}
+	networkMap, err := c.network.GetNetMapByEpoch(curEpoch)
+	if err != nil {
+		return false, fmt.Errorf("read network map at the current epoch #%d: %w", curEpoch, err)
+	}
+	return c.isMineWithMeta(id, networkMap)
+}
+
+func (c *metaContainerListener) 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)
+	}
+
+	const metaOnChainAttr = "__NEOFS__METAINFO_CONSISTENCY"
+	switch cnr.Value.Attribute(metaOnChainAttr) {
+	case "optimistic", "strict":
+	default:
+		return false, nil
+	}
+
+	nodeSets, err := networkMap.ContainerNodes(cnr.Value.PlacementPolicy(), id)
+	if err != nil {
+		return false, fmt.Errorf("apply container storage policy to %s container: %w", id, err)
+	}
+
+	for _, nodeSet := range nodeSets {
+		for _, node := range nodeSet {
+			if bytes.Equal(node.PublicKey(), c.key) {
+				return true, nil
+			}
+		}
+	}
+
+	return false, nil
+}
+
 func (c *metaContainerListener) List() (map[cid.ID]struct{}, error) {
 	actualContainers, err := c.cnrClient.List(nil)
 	if err != nil {
@@ -99,33 +140,14 @@ func (c *metaContainerListener) List() (map[cid.ID]struct{}, error) {
 	var wg errgroup.Group
 	for _, cID := range actualContainers {
 		wg.Go(func() error {
-			cnr, err := c.containers.Get(cID)
-			if err != nil {
-				return fmt.Errorf("read %s container: %w", cID, err)
-			}
-
-			const metaOnChainAttr = "__NEOFS__METAINFO_CONSISTENCY"
-			switch cnr.Value.Attribute(metaOnChainAttr) {
-			case "optimistic", "strict":
-			default:
-				return nil
+			ok, err := c.isMineWithMeta(cID, networkMap)
+			if err != nil || !ok {
+				return err
 			}
 
-			nodeSets, err := networkMap.ContainerNodes(cnr.Value.PlacementPolicy(), cID)
-			if err != nil {
-				return fmt.Errorf("apply container storage policy to %s container: %w", cID, err)
-			}
-
-			for _, nodeSet := range nodeSets {
-				for _, node := range nodeSet {
-					if bytes.Equal(node.PublicKey(), c.key) {
-						locM.Lock()
-						res[cID] = struct{}{}
-						locM.Unlock()
-						return nil
-					}
-				}
-			}
+			locM.Lock()
+			res[cID] = struct{}{}
+			locM.Unlock()
 
 			return nil
 		})

pkg/services/meta/meta.go

@@ -31,6 +31,9 @@ type ContainerLister interface {
 	// List returns node's containers that support chain-based meta data and
 	// any error that does not allow listing.
 	List() (map[cid.ID]struct{}, error)
+	// IsMineWithMeta checks if the given CID has meta enabled and current
+	// node belongs to it.
+	IsMineWithMeta(cid.ID) (bool, error)
 }
 
 // wsClient is for test purposes only.

pkg/services/meta/notifications.go

@@ -119,6 +119,15 @@ func (m *Meta) listenNotifications(ctx context.Context) error {
 				continue
 			}
 
+			ok, err = m.cLister.IsMineWithMeta(ev.cID)
+			if err != nil {
+				l.Error("can't get container data", zap.Error(err))
+				continue
+			}
+			if !ok {
+				continue
+			}
+
 			m.m.Lock()
 
 			st, err := storageForContainer(m.rootPath, ev.cID)

pkg/services/meta/notifications_test.go

@@ -39,6 +39,10 @@ type testContainerLister struct {
 	resErr error
 }
 
+func (t *testContainerLister) IsMineWithMeta(id cid.ID) (bool, error) {
+	return true, nil
+}
+
 func (t *testContainerLister) List() (map[cid.ID]struct{}, error) {
 	return t.res, t.resErr
 }