sn/object: Initial support for erasure coding policies by HASH server

Closes #3656. Refs #3423, #3654, #526.

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

Author: cthulhu-rider

pkg/services/object/get/exec.go

@@ -35,8 +35,7 @@ type execCtx struct {
 
 	ctx context.Context
 
-	prm          RangePrm
-	prmRangeHash *RangeHashPrm
+	prm RangePrm
 
 	statusError
 
@@ -82,12 +81,6 @@ func withPayloadRange(r *objectSDK.Range) execOption {
 	}
 }
 
-func withHash(p *RangeHashPrm) execOption {
-	return func(ctx *execCtx) {
-		ctx.prmRangeHash = p
-	}
-}
-
 func withLogger(l *zap.Logger) execOption {
 	return func(ctx *execCtx) {
 		ctx.log = l
@@ -418,15 +411,8 @@ func (exec execCtx) isForwardingEnabled() bool {
 	return exec.prm.forwarder != nil
 }
 
-// isRangeHashForwardingEnabled returns true if common execution
-// parameters has GETRANGEHASH request forwarding closure set.
-func (exec execCtx) isRangeHashForwardingEnabled() bool {
-	return exec.prm.rangeForwarder != nil
-}
-
 // disableForwarding removes request forwarding closure from common
 // parameters, so it won't be inherited in new execution contexts.
 func (exec *execCtx) disableForwarding() {
 	exec.prm.SetRequestForwarder(nil)
-	exec.prm.SetRangeHashRequestForwarder(nil)
 }

pkg/services/object/get/get.go

@@ -5,8 +5,11 @@ import (
 	"errors"
 	"fmt"
 
+	iec "github.com/nspcc-dev/neofs-node/internal/ec"
+	"github.com/nspcc-dev/neofs-node/pkg/network"
 	"github.com/nspcc-dev/neofs-node/pkg/util"
 	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
+	netmapsdk "github.com/nspcc-dev/neofs-sdk-go/netmap"
 	"github.com/nspcc-dev/neofs-sdk-go/object"
 	"go.uber.org/zap"
 )
@@ -66,6 +69,10 @@ func (s *Service) GetRange(ctx context.Context, prm RangePrm) error {
 		return s.copyLocalECPartRange(prm.objWriter, prm.addr.Container(), prm.addr.Object(), pi, prm.rng.GetOffset(), prm.rng.GetLength())
 	}
 
+	return s.getRange(ctx, prm, nodeLists, repRules, ecRules)
+}
+
+func (s *Service) getRange(ctx context.Context, prm RangePrm, nodeLists [][]netmapsdk.NodeInfo, repRules []uint, ecRules []iec.Rule) error {
 	if len(repRules) > 0 { // REP format does not require encoding
 		err := s.get(ctx, prm.commonPrm, withPreSortedContainerNodes(nodeLists[:len(repRules)], repRules), withPayloadRange(prm.rng)).err
 		if len(ecRules) == 0 || !errors.Is(err, apistatus.ErrObjectNotFound) {
@@ -78,6 +85,15 @@ func (s *Service) GetRange(ctx context.Context, prm RangePrm) error {
 }
 
 func (s *Service) GetRangeHash(ctx context.Context, prm RangeHashPrm) (*RangeHashRes, error) {
+	nodeLists, repRules, ecRules, err := s.neoFSNet.GetNodesForObject(prm.addr)
+	if err != nil {
+		return nil, fmt.Errorf("get nodes for object: %w", err)
+	}
+
+	if prm.common.SessionToken() == nil && prm.rangeForwarder != nil && !localNodeInSets(s.neoFSNet, nodeLists) {
+		return s.proxyHashRequest(ctx, nodeLists, prm.rangeForwarder)
+	}
+
 	hashes := make([][]byte, 0, len(prm.rngs))
 
 	for _, rng := range prm.rngs {
@@ -97,17 +113,10 @@ func (s *Service) GetRangeHash(ctx context.Context, prm RangeHashPrm) (*RangeHas
 			hash: util.NewSaltingWriter(h, prm.salt),
 		})
 
-		if err := s.get(ctx, rngPrm.commonPrm, withHash(&prm), withPayloadRange(rngPrm.rng)).err; err != nil {
+		if err := s.getRange(ctx, rngPrm, nodeLists, repRules, ecRules); err != nil {
 			return nil, err
 		}
 
-		if prm.forwardedRangeHashResponse != nil {
-			// forwarder request case; no need to collect the other
-			// parts, the whole response has already been received
-			hashes = prm.forwardedRangeHashResponse
-			break
-		}
-
 		hashes = append(hashes, h.Sum(nil))
 	}
 
@@ -116,6 +125,34 @@ func (s *Service) GetRangeHash(ctx context.Context, prm RangeHashPrm) (*RangeHas
 	}, nil
 }
 
+func (s *Service) proxyHashRequest(ctx context.Context, sortedNodeLists [][]netmapsdk.NodeInfo, proxyFn RangeRequestForwarder) (*RangeHashRes, error) {
+	for i := range sortedNodeLists {
+		for j := range sortedNodeLists[i] {
+			conn, node, err := s.conns.(*clientCacheWrapper)._connect(sortedNodeLists[i][j])
+			if err != nil {
+				// TODO: implement address list stringer for lazy encoding
+				s.log.Debug("get conn to remote node",
+					zap.String("addresses", network.StringifyGroup(node.AddressGroup())), zap.Error(err))
+				continue
+			}
+
+			hashes, err := proxyFn(ctx, node, conn)
+			if err == nil {
+				return &RangeHashRes{hashes: hashes}, nil
+			}
+
+			if errors.Is(err, apistatus.ErrObjectAlreadyRemoved) || errors.Is(err, apistatus.ErrObjectAccessDenied) ||
+				errors.Is(err, apistatus.ErrObjectOutOfRange) || errors.Is(err, ctx.Err()) {
+				return nil, err
+			}
+
+			s.log.Debug("HASH proxy failed", zap.Error(err))
+		}
+	}
+
+	return nil, apistatus.ErrObjectNotFound
+}
+
 // Head reads object header from container.
 //
 // Returns ErrNotFound if the header was not received for the call.

pkg/services/object/get/prm.go

@@ -33,8 +33,6 @@ type RangeHashPrm struct {
 	rngs []object.Range
 
 	salt []byte
-
-	forwardedRangeHashResponse [][]byte
 }
 
 type RequestForwarder func(context.Context, coreclient.NodeInfo, coreclient.MultiAddressClient) (*object.Object, error)

pkg/services/object/get/util.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"slices"
 
 	coreclient "github.com/nspcc-dev/neofs-node/pkg/core/client"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/engine"
@@ -213,11 +214,6 @@ func (c *clientWrapper) getObject(exec *execCtx, info coreclient.NodeInfo) (*obj
 		return hdr, nil, nil
 	}
 
-	if rngH := exec.prmRangeHash; rngH != nil && exec.isRangeHashForwardingEnabled() {
-		exec.prmRangeHash.forwardedRangeHashResponse, err = exec.prm.rangeForwarder(exec.ctx, info, c.client)
-		return nil, nil, err
-	}
-
 	// we don't specify payload writer because we accumulate
 	// the object locally (even huge).
 	if rng := exec.ctxRange(); rng != nil {
@@ -405,11 +401,16 @@ func (c *clientCacheWrapper) InitGetObjectRangeStream(ctx context.Context, node
 }
 
 func (c *clientCacheWrapper) connect(node netmap.NodeInfo) (coreclient.MultiAddressClient, error) {
+	conn, _, err := c._connect(node)
+	return conn, err
+}
+
+func (c *clientCacheWrapper) _connect(node netmap.NodeInfo) (coreclient.MultiAddressClient, coreclient.NodeInfo, error) {
 	// TODO: code is copied from pkg/services/object/get/container.go:63. Worth sharing?
 	// TODO: we may waste resources doing this per request. Make once on network map change instead.
 	var ag network.AddressGroup
 	if err := ag.FromIterator(network.NodeEndpointsIterator(node)); err != nil {
-		return nil, fmt.Errorf("decode SN network addresses: %w", err)
+		return nil, coreclient.NodeInfo{}, fmt.Errorf("decode SN network addresses: %w", err)
 	}
 
 	var ni coreclient.NodeInfo
@@ -418,10 +419,10 @@ func (c *clientCacheWrapper) connect(node netmap.NodeInfo) (coreclient.MultiAddr
 
 	conn, err := c.cache.Get(ni)
 	if err != nil {
-		return nil, fmt.Errorf("get conn: %w", err)
+		return nil, coreclient.NodeInfo{}, fmt.Errorf("get conn: %w", err)
 	}
 
-	return conn, nil
+	return conn, ni, nil
 }
 
 // TODO: share.
@@ -436,3 +437,15 @@ func convertContextStatus(err error) error {
 		return context.DeadlineExceeded
 	}
 }
+
+func localNodeInSets(n NeoFSNetwork, nodeSets [][]netmap.NodeInfo) bool {
+	return slices.ContainsFunc(nodeSets, func(nodeSet []netmap.NodeInfo) bool {
+		return localNodeInSet(n, nodeSet)
+	})
+}
+
+func localNodeInSet(n NeoFSNetwork, nodes []netmap.NodeInfo) bool {
+	return slices.ContainsFunc(nodes, func(node netmap.NodeInfo) bool {
+		return n.IsLocalNodePublicKey(node.PublicKey())
+	})
+}