Improve GET/HEAD replica localization

CHANGELOG.md

@@ -30,6 +30,7 @@ Changelog for NeoFS Node
 - Policer now uses an in-memory queue for redundant object removal, reducing metabase writes (#4216)
 - Session token storage now indexes tokens by expiration epoch for efficient cleanup (#4243)
 - Session key encryption now uses HPKE (#4247)
+- SN object GET/HEAD server now traverses nodes strictly by weight, and falls back to local storage (#4232)
 
 ### Removed
 - fstree-decompress utility (#4223)

internal/netmap/netmap.go

@@ -11,11 +11,13 @@ import (
 // least one element from nodeSets.
 func NodeSetsContainPublicKeyFunc(nodeSets [][]netmap.NodeInfo, checkPubKeyFn func([]byte) bool) bool {
 	return slices.ContainsFunc(nodeSets, func(nodeSet []netmap.NodeInfo) bool {
-		return nodeSetContainsPublicKeyFunc(nodeSet, checkPubKeyFn)
+		return NodeSetContainsPublicKeyFunc(nodeSet, checkPubKeyFn)
 	})
 }
 
-func nodeSetContainsPublicKeyFunc(nodeSet []netmap.NodeInfo, checkPubKeyFn func([]byte) bool) bool {
+// NodeSetContainsPublicKeyFunc checks whether checkPubKeyFn is true for at
+// least one element from nodeSet.
+func NodeSetContainsPublicKeyFunc(nodeSet []netmap.NodeInfo, checkPubKeyFn func([]byte) bool) bool {
 	return slices.ContainsFunc(nodeSet, func(node netmap.NodeInfo) bool {
 		return checkPubKeyFn(node.PublicKey())
 	})

pkg/services/object/get/container.go

@@ -2,16 +2,13 @@ package getsvc
 
 import (
 	"context"
+	"errors"
 
+	inetmap "github.com/nspcc-dev/neofs-node/internal/netmap"
 	"go.uber.org/zap"
 )
 
 func (exec *execCtx) executeOnContainer() {
-	if exec.isLocal() {
-		exec.log.Debug("return result directly")
-		return
-	}
-
 	exec.log.Debug("trying to execute in container...")
 
 	addr := exec.address()
@@ -25,10 +22,29 @@ func (exec *execCtx) executeOnContainer() {
 			exec.status = statusUndefined
 			exec.err = err
 			exec.log.Debug("failed to list storage nodes for the object", zap.Error(err))
+			exec.logOtherError()
 			return
 		}
 	}
 
+	var localDone bool
+	execLocal := func() bool {
+		if localDone {
+			return false
+		}
+		localDone = true
+		return exec.executeLocal()
+	}
+
+	// try local node first if it is a primary holder
+	for i, n := range primaryCounts {
+		if inetmap.NodeSetContainsPublicKeyFunc(nodeLists[i][:n], exec.svc.neoFSNet.IsLocalNodePublicKey) {
+			if execLocal() {
+				return
+			}
+		}
+	}
+
 	ctx, cancel := context.WithCancel(exec.context())
 	defer cancel()
 
@@ -50,14 +66,24 @@ func (exec *execCtx) executeOnContainer() {
 				exec.log.Debug("interrupt placement iteration by context",
 					zap.Error(ctx.Err()),
 				)
-
+				if exec.err == nil {
+					exec.err = ctx.Err()
+				}
+				exec.logOtherError()
 				return
 			default:
 			}
 
 			bKey := nodeLists[i][j].PublicKey()
 			strKey := string(bKey)
-			if _, ok := mProcessedNodes[strKey]; ok || exec.svc.neoFSNet.IsLocalNodePublicKey(bKey) {
+			if _, ok := mProcessedNodes[strKey]; ok {
+				continue
+			}
+
+			if exec.svc.neoFSNet.IsLocalNodePublicKey(bKey) {
+				if execLocal() {
+					return
+				}
 				continue
 			}
 
@@ -68,7 +94,10 @@ func (exec *execCtx) executeOnContainer() {
 			//  we reach the best result - split info with linking object ID.
 			if exec.processNode(nodeLists[i][j]) {
 				exec.log.Debug("completing the operation")
-				return
+				exec.analyzeContainerStatus()
+				if exec.responseStarted || !errors.Is(exec.err, errNoLinkNoLastPart) {
+					return
+				}
 			}
 		}
 
@@ -78,4 +107,10 @@ func (exec *execCtx) executeOnContainer() {
 			i = -1
 		}
 	}
+
+	// Local node is not in the container now. But it can still store the object.
+	// One scenario is when the container nodes have changed, the local node is no
+	// longer in it, and the object hasn't migrated yet. To cover such cases, lookup
+	// local storage as a last-ditch effort to find the object.
+	execLocal()
 }

pkg/services/object/get/get.go

@@ -205,38 +205,62 @@ func (s *Service) get(ctx context.Context, prm commonPrm, opts ...execOption) st
 func (exec *execCtx) execute() {
 	exec.log.Debug("serving request...")
 
-	// perform local operation
-	exec.executeLocal()
+	if exec.isLocal() {
+		exec.executeLocal()
+		exec.log.Debug("return result directly")
+		return
+	}
 
-	exec.analyzeStatus(true)
+	exec.executeOnContainer()
 }
 
-func (exec *execCtx) analyzeStatus(execCnr bool) {
+func (exec *execCtx) analyzeLocalStatus() bool {
 	// analyze local result
 	switch exec.status {
 	case statusOK:
-		exec.log.Debug("operation finished successfully")
+		exec.logSuccess()
 	case statusVIRTUAL:
-		exec.log.Debug("requested object is virtual")
-		if exec.collectOnly {
-			return
-		}
-		exec.assemble()
-		if errors.Is(exec.err, errNoLinkNoLastPart) && execCnr {
-			exec.executeOnContainer()
-			exec.analyzeStatus(false)
-		}
+		exec.handleVirtualObject()
+		return !errors.Is(exec.err, errNoLinkNoLastPart)
 	case statusAPIResponse:
-		exec.log.Debug("received api response locally, return directly", zap.Error(exec.err))
-		return
+		exec.logAPIError()
 	default:
-		exec.log.Debug("operation finished with error",
-			zap.Error(exec.err),
-		)
+		exec.logOtherError()
+		return exec.responseStarted
+	}
 
-		if execCnr && !exec.responseStarted {
-			exec.executeOnContainer()
-			exec.analyzeStatus(false)
-		}
+	return true
+}
+
+func (exec *execCtx) analyzeContainerStatus() {
+	switch exec.status {
+	case statusOK:
+		exec.logSuccess()
+	case statusVIRTUAL:
+		exec.handleVirtualObject()
+	case statusAPIResponse:
+		exec.logAPIError()
+	default:
+		exec.logOtherError()
+	}
+}
+
+func (exec *execCtx) handleVirtualObject() {
+	exec.log.Debug("requested object is virtual")
+	if exec.collectOnly {
+		return
 	}
+	exec.assemble()
+}
+
+func (exec *execCtx) logSuccess() {
+	exec.log.Debug("operation finished successfully")
+}
+
+func (exec *execCtx) logAPIError() {
+	exec.log.Debug("received api response locally, return directly", zap.Error(exec.err))
+}
+
+func (exec *execCtx) logOtherError() {
+	exec.log.Debug("operation finished with error", zap.Error(exec.err))
 }

pkg/services/object/get/get_test.go

@@ -92,7 +92,8 @@ func TestGetExtendedRangeOfECPart(t *testing.T) {
 }
 
 type testNeoFS struct {
-	vectors map[oid.Address][][]netmap.NodeInfo
+	vectors        map[oid.Address][][]netmap.NodeInfo
+	localPublicKey []byte
 }
 
 type testClientCache struct {
@@ -116,7 +117,9 @@ func newTestStorage() *testStorage {
 	}
 }
 
-func (g *testNeoFS) IsLocalNodePublicKey([]byte) bool { return false }
+func (g *testNeoFS) IsLocalNodePublicKey(pubKey []byte) bool {
+	return bytes.Equal(g.localPublicKey, pubKey)
+}
 
 func (g *testNeoFS) GetNodesForObject(addr oid.Address) ([][]netmap.NodeInfo, []uint, []iec.Rule, error) {
 	nodeLists, ok := g.vectors[addr]
@@ -1404,7 +1407,7 @@ func TestDoubleWriteHeaderOnPayloadReadFailure(t *testing.T) {
 		err:       readErr,
 	}
 
-	anyNodeLists, nodeStrs := testNodeMatrix(t, []int{1})
+	anyNodeLists, nodeStrs := testNodeMatrix(t, []int{2})
 
 	clientCache := &testClientCache{
 		clients: make(map[string]*testClient),
@@ -1422,6 +1425,7 @@ func TestDoubleWriteHeaderOnPayloadReadFailure(t *testing.T) {
 		vectors: map[oid.Address][][]netmap.NodeInfo{
 			addr: anyNodeLists,
 		},
+		localPublicKey: anyNodeLists[0][0].PublicKey(),
 	}
 
 	writer := &trackingWriter{}
@@ -1452,7 +1456,7 @@ func TestDoubleWriteHeaderOnChunkWriteFailure(t *testing.T) {
 	storage := newTestStorage()
 	storage.addPhy(addr, obj)
 
-	anyNodeLists, nodeStrs := testNodeMatrix(t, []int{1})
+	anyNodeLists, nodeStrs := testNodeMatrix(t, []int{2})
 
 	clientCache := &testClientCache{
 		clients: make(map[string]*testClient),
@@ -1470,6 +1474,7 @@ func TestDoubleWriteHeaderOnChunkWriteFailure(t *testing.T) {
 		vectors: map[oid.Address][][]netmap.NodeInfo{
 			addr: anyNodeLists,
 		},
+		localPublicKey: anyNodeLists[0][0].PublicKey(),
 	}
 
 	writeErr := errors.New("simulated chunk write error")

pkg/services/object/get/local.go

@@ -13,7 +13,7 @@ import (
 	"go.uber.org/zap"
 )
 
-func (exec *execCtx) executeLocal() {
+func (exec *execCtx) executeLocal() bool {
 	var err error
 
 	exec.collectedHeader, exec.collectedReader, err = exec.svc.localStorage.get(exec)
@@ -40,7 +40,7 @@ func (exec *execCtx) executeLocal() {
 			exec.status = statusNotFound
 			exec.err = err
 
-			return
+			break
 		}
 
 		exec.status = statusAPIResponse
@@ -50,6 +50,8 @@ func (exec *execCtx) executeLocal() {
 		mergeSplitInfo(exec.splitInfo(), errSplitInfo.SplitInfo())
 		exec.err = object.NewSplitInfoError(exec.infoSplit)
 	}
+
+	return exec.analyzeLocalStatus()
 }
 
 func (s *Service) copyLocalObjectHeader(ctx context.Context, dst internal.HeaderWriter, cnr cid.ID, id oid.ID, raw bool) error {