WIP: decoding in GET service

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

Author: cthulhu-rider

cmd/neofs-node/object.go

@@ -716,8 +716,9 @@ func (c *cfg) IsLocalNodePublicKey(b []byte) bool { return c.IsLocalKey(b) }
 // object holders. Resulting slices must not be changed.
 //
 // GetNodesForObject implements [getsvc.NeoFSNetwork].
-func (c *cfg) GetNodesForObject(addr oid.Address) ([][]netmapsdk.NodeInfo, []uint, error) {
-	return c.cfgObject.containerNodes.getNodesForObject(addr)
+func (c *cfg) GetNodesForObject(addr oid.Address) ([][]netmapsdk.NodeInfo, []uint, []iec.Rule, error) {
+	nodeSets, repRules, err := c.cfgObject.containerNodes.getNodesForObject(addr)
+	return nodeSets, repRules, nil, err
 }
 
 type netmapSourceWithNodes struct {

pkg/local_object_storage/engine/ec.go

@@ -0,0 +1,33 @@
+package engine
+
+import (
+	"fmt"
+
+	iec "github.com/nspcc-dev/neofs-node/internal/ec"
+	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
+	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
+	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
+	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
+	"go.uber.org/zap"
+)
+
+// TODO:: docs.
+func (e *StorageEngine) GetECPart(cnr cid.ID, parent oid.ID, pi iec.PartInfo) (objectSDK.Object, error) {
+	// TODO: keep in sync with https://github.com/nspcc-dev/neofs-node/pull/3466.
+	// TODO: metrics and blockErr like Get
+
+	// TODO: sync placement with PUT. They must sort shard equally
+	shs := e.sortedShards(oid.NewAddress(cnr, parent))
+	for i := range shs {
+		obj, err := shs[i].GetECPart(cnr, parent, pi)
+		if err == nil {
+			return obj, nil
+		}
+		// TODO: debug if 404
+		e.log.Info("failed to get EC part from shard", zap.Stringer("shardID", shs[i].ID()), zap.Stringer("container", cnr), zap.Stringer("parent", parent),
+			zap.Int("ruleIdx", pi.RuleIndex), zap.Int("partIdx", pi.Index), zap.Error(err))
+		// FIXME: some errors like 'akready removed' must abort
+	}
+
+	return objectSDK.Object{}, fmt.Errorf("%w: all shards failed", apistatus.ErrObjectNotFound)
+}

pkg/local_object_storage/engine/ec_test.go

@@ -0,0 +1,83 @@
+package engine
+
+import (
+	"strconv"
+	"testing"
+
+	iec "github.com/nspcc-dev/neofs-node/internal/ec"
+	"github.com/nspcc-dev/neofs-node/internal/testutil"
+	"github.com/nspcc-dev/neofs-sdk-go/checksum"
+	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
+	cidtest "github.com/nspcc-dev/neofs-sdk-go/container/id/test"
+	objectsdk "github.com/nspcc-dev/neofs-sdk-go/object"
+	oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test"
+	usertest "github.com/nspcc-dev/neofs-sdk-go/user/test"
+	"github.com/stretchr/testify/require"
+)
+
+func TestStorageEngine_GetECPartByIdx(t *testing.T) {
+	for _, shardNum := range []int{1, 5} {
+		t.Run("shards="+strconv.Itoa(shardNum), func(t *testing.T) {
+			testGetECPart(t, shardNum)
+		})
+	}
+}
+
+func testGetECPart(t *testing.T, shardNum int) {
+	cnr := cidtest.ID()
+	parentID := oidtest.ID()
+	const ruleIdx = 123
+	const partIdx = 456
+
+	var parent objectsdk.Object
+	parent.SetContainerID(cnr)
+	parent.SetID(parentID)
+	parent.SetOwner(usertest.ID())
+	parent.SetPayloadChecksum(checksum.NewSHA256([32]byte(testutil.RandByteSlice(32))))
+
+	part := parent
+	part.SetID(oidtest.OtherID(parentID))
+	part.SetParent(&parent)
+	part.SetAttributes(
+		objectsdk.NewAttribute("__NEOFS__EC_RULE_IDX", strconv.Itoa(ruleIdx)),
+		objectsdk.NewAttribute("__NEOFS__EC_PART_IDX", strconv.Itoa(partIdx)),
+	)
+
+	s := testNewEngineWithShardNum(t, shardNum)
+
+	checkMissingIdxs := func(t *testing.T, ruleIdx, partIdx int) {
+		_, err := s.GetECPart(cnr, parentID, iec.PartInfo{
+			RuleIndex: ruleIdx,
+			Index:     partIdx,
+		})
+		require.ErrorIs(t, err, apistatus.ErrObjectNotFound)
+	}
+
+	checkMissingIdxs(t, ruleIdx-1, partIdx-1)
+	checkMissingIdxs(t, ruleIdx-1, partIdx)
+	checkMissingIdxs(t, ruleIdx-1, partIdx+1)
+	checkMissingIdxs(t, ruleIdx, partIdx-1)
+	checkMissingIdxs(t, ruleIdx, partIdx)
+	checkMissingIdxs(t, ruleIdx, partIdx+1)
+	checkMissingIdxs(t, ruleIdx+1, partIdx-1)
+	checkMissingIdxs(t, ruleIdx+1, partIdx)
+	checkMissingIdxs(t, ruleIdx+1, partIdx+1)
+
+	require.NoError(t, s.Put(&part, nil))
+
+	checkMissingIdxs(t, ruleIdx-1, partIdx-1)
+	checkMissingIdxs(t, ruleIdx-1, partIdx)
+	checkMissingIdxs(t, ruleIdx-1, partIdx+1)
+	checkMissingIdxs(t, ruleIdx, partIdx-1)
+	checkMissingIdxs(t, ruleIdx, partIdx+1)
+	checkMissingIdxs(t, ruleIdx+1, partIdx-1)
+	checkMissingIdxs(t, ruleIdx+1, partIdx)
+	checkMissingIdxs(t, ruleIdx+1, partIdx+1)
+
+	got, err := s.GetECPart(cnr, parentID, iec.PartInfo{
+		RuleIndex: ruleIdx,
+		Index:     partIdx,
+	})
+	require.NoError(t, err)
+	require.Equal(t, part, got)
+}

pkg/local_object_storage/metabase/ec.go

@@ -0,0 +1,62 @@
+package meta
+
+import (
+	"bytes"
+	"fmt"
+	"slices"
+	"strconv"
+
+	iec "github.com/nspcc-dev/neofs-node/internal/ec"
+	objectcore "github.com/nspcc-dev/neofs-node/pkg/core/object"
+	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
+	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
+	"github.com/nspcc-dev/neofs-sdk-go/object"
+	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
+	"go.etcd.io/bbolt"
+)
+
+// TODO: docs.
+func (db *DB) ResolveECPart(cnr cid.ID, parent oid.ID, pi iec.PartInfo) (oid.ID, error) {
+	// FIXME: check degraded
+	// TODO: check already removed case;. https://github.com/nspcc-dev/neofs-node/issues/3502
+	var res oid.ID
+	err := db.boltDB.View(func(tx *bbolt.Tx) error {
+		var err error
+		res, err = db.resolveECPartTx(tx, cnr, parent, pi)
+		return err
+	})
+	return res, err
+}
+
+func (db *DB) resolveECPartTx(tx *bbolt.Tx, cnr cid.ID, parent oid.ID, pi iec.PartInfo) (oid.ID, error) {
+	metaBkt := tx.Bucket(metaBucketKey(cnr))
+	if metaBkt == nil {
+		return oid.ID{}, fmt.Errorf("%w: missing meta bucket", apistatus.ErrObjectNotFound)
+	}
+	metaBktCursor := metaBkt.Cursor()
+
+	pref := slices.Concat([]byte{metaPrefixAttrIDPlain}, []byte(object.FilterParentID), objectcore.MetaAttributeDelimiter,
+		parent[:], objectcore.MetaAttributeDelimiter,
+	)
+	k, _ := metaBktCursor.Seek(pref)
+	partID, ok := bytes.CutPrefix(k, pref)
+	if !ok {
+		return oid.ID{}, fmt.Errorf("%w: not found by index %s", apistatus.ErrObjectNotFound, object.FilterParentID)
+	}
+	if len(partID) != oid.Size {
+		return oid.ID{}, invalidMetaBucketKeyErr(k, fmt.Errorf("wrong OID len %d", len(partID)))
+	}
+
+	// TODO: make one buffer for all keys
+	k = slices.Concat([]byte{metaPrefixIDAttr}, partID, []byte(iec.AttributeRuleIdx), objectcore.MetaAttributeDelimiter, []byte(strconv.Itoa(pi.RuleIndex)))
+	if metaBkt.Get(k) == nil {
+		return oid.ID{}, fmt.Errorf("%w: not found by index %s", apistatus.ErrObjectNotFound, iec.AttributeRuleIdx)
+	}
+
+	k = slices.Concat([]byte{metaPrefixIDAttr}, partID, []byte(iec.AttributePartIdx), objectcore.MetaAttributeDelimiter, []byte(strconv.Itoa(pi.Index)))
+	if metaBkt.Get(k) == nil {
+		return oid.ID{}, fmt.Errorf("%w: not found by index %s", apistatus.ErrObjectNotFound, iec.AttributePartIdx)
+	}
+
+	return oid.ID(partID), nil
+}

pkg/local_object_storage/shard/ec.go

@@ -0,0 +1,39 @@
+package shard
+
+import (
+	"fmt"
+
+	iec "github.com/nspcc-dev/neofs-node/internal/ec"
+	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
+	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
+	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
+	"go.uber.org/zap"
+)
+
+// TODO: docs.
+// TODO: keep in sync with https://github.com/nspcc-dev/neofs-node/pull/3466.
+func (s *Shard) GetECPart(cnr cid.ID, parent oid.ID, pi iec.PartInfo) (objectSDK.Object, error) {
+	// FIXME: metabase can be disabled
+	partID, err := s.metaBase.ResolveECPart(cnr, parent, pi)
+	if err != nil {
+		return objectSDK.Object{}, fmt.Errorf("resolve part ID in metabase: %w", err)
+	}
+
+	partAddr := oid.NewAddress(cnr, partID)
+	if s.hasWriteCache() {
+		obj, err := s.writeCache.Get(partAddr)
+		if err == nil {
+			return *obj, nil
+		}
+
+		// TODO: debug if 404
+		s.log.Info("failed to get EC part from write-cache, trying BLOB storage...", zap.Stringer("partAddr", partAddr), zap.Error(err))
+	}
+
+	obj, err := s.blobStor.Get(partAddr)
+	if err != nil {
+		return objectSDK.Object{}, fmt.Errorf("get from BLOB storage by ID %s: %w", partID, err)
+	}
+
+	return *obj, nil
+}

pkg/services/object/get/container.go

@@ -17,7 +17,9 @@ func (exec *execCtx) executeOnContainer() {
 
 	exec.log.Debug("trying to execute in container...")
 
-	nodeLists, primaryCounts, err := exec.svc.neoFSNet.GetNodesForObject(exec.address())
+	addr := exec.address()
+
+	nodeLists, primaryCounts, _, err := exec.svc.neoFSNet.GetNodesForObject(addr)
 	if err != nil {
 		exec.status = statusUndefined
 		exec.err = err
@@ -34,7 +36,7 @@ func (exec *execCtx) executeOnContainer() {
 	var j, jLim uint
 	primary := true
 
-	for i := 0; i < len(nodeLists); i++ { // do not use for-range!
+	for i := 0; i < len(primaryCounts); i++ { // do not use for-range!
 		if primary {
 			j, jLim = 0, primaryCounts[i]
 		} else {
@@ -81,7 +83,7 @@ func (exec *execCtx) executeOnContainer() {
 			}
 		}
 
-		if primary && i == len(nodeLists)-1 {
+		if primary && i == len(primaryCounts)-1 {
 			// switch to reserve nodes
 			primary = false
 			i = -1