Try fix

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

Author: cthulhu-rider

pkg/services/object/get/exec.go

@@ -35,7 +35,8 @@ type execCtx struct {
 
 	ctx context.Context
 
-	prm RangePrm
+	prm          RangePrm
+	prmRangeHash *RangeHashPrm
 
 	statusError
 
@@ -81,6 +82,12 @@ 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
@@ -411,8 +418,15 @@ 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

@@ -69,19 +69,29 @@ 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)
+	return s.getRange(ctx, prm, nodeLists, repRules, ecRules, nil)
 }
 
-func (s *Service) getRange(ctx context.Context, prm RangePrm, nodeLists [][]netmapsdk.NodeInfo, repRules []uint, ecRules []iec.Rule) error {
+func (s *Service) getRange(ctx context.Context, prm RangePrm, nodeLists [][]netmapsdk.NodeInfo, repRules []uint, ecRules []iec.Rule,
+	hashPrm *RangeHashPrm) 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
+		err := s.get(ctx, prm.commonPrm, withPreSortedContainerNodes(nodeLists[:len(repRules)], repRules), withPayloadRange(prm.rng), withHash(hashPrm)).err
 		if len(ecRules) == 0 || !errors.Is(err, apistatus.ErrObjectNotFound) {
 			return err
 		}
 	}
 
+	ecNodeLists := nodeLists[len(repRules):]
+	if hashPrm != nil && prm.rangeForwarder != nil && !localNodeInSets(s.neoFSNet, nodeLists) {
+		hashes, err := s.proxyHashRequest(ctx, ecNodeLists, prm.rangeForwarder)
+		if err == nil {
+			hashPrm.forwardedRangeHashResponse = hashes
+		}
+		return err
+	}
+
 	return s.copyECObjectRange(ctx, prm.objWriter, prm.addr.Container(), prm.addr.Object(), prm.common.SessionToken(),
-		ecRules, nodeLists[len(repRules):], prm.rng.GetOffset(), prm.rng.GetLength())
+		ecRules, ecNodeLists, prm.rng.GetOffset(), prm.rng.GetLength())
 }
 
 func (s *Service) GetRangeHash(ctx context.Context, prm RangeHashPrm) (*RangeHashRes, error) {
@@ -90,10 +100,6 @@ func (s *Service) GetRangeHash(ctx context.Context, prm RangeHashPrm) (*RangeHas
 		return nil, fmt.Errorf("get nodes for object: %w", err)
 	}
 
-	if 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 {
@@ -113,10 +119,17 @@ func (s *Service) GetRangeHash(ctx context.Context, prm RangeHashPrm) (*RangeHas
 			hash: util.NewSaltingWriter(h, prm.salt),
 		})
 
-		if err := s.getRange(ctx, rngPrm, nodeLists, repRules, ecRules); err != nil {
+		if err := s.getRange(ctx, rngPrm, nodeLists, repRules, ecRules, &prm); 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))
 	}
 
@@ -125,7 +138,7 @@ 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) {
+func (s *Service) proxyHashRequest(ctx context.Context, sortedNodeLists [][]netmapsdk.NodeInfo, proxyFn RangeRequestForwarder) ([][]byte, error) {
 	for i := range sortedNodeLists {
 		for j := range sortedNodeLists[i] {
 			conn, node, err := s.conns.(*clientCacheWrapper)._connect(sortedNodeLists[i][j])
@@ -138,7 +151,7 @@ func (s *Service) proxyHashRequest(ctx context.Context, sortedNodeLists [][]netm
 
 			hashes, err := proxyFn(ctx, node, conn)
 			if err == nil {
-				return &RangeHashRes{hashes: hashes}, nil
+				return hashes, nil
 			}
 
 			if errors.Is(err, apistatus.ErrObjectAlreadyRemoved) || errors.Is(err, apistatus.ErrObjectAccessDenied) ||

pkg/services/object/get/prm.go

@@ -33,6 +33,8 @@ 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

@@ -214,6 +214,11 @@ 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 {