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 {

internal/ec/ec.go

@@ -75,6 +75,14 @@ func Decode(rule Rule, dataLen uint64, parts [][]byte) ([]byte, error) {
 		return nil, fmt.Errorf("restore Reed-Solomon: %w", err)
 	}
 
+	return ConcatDataParts(rule, dataLen, parts), nil
+}
+
+// ConcatDataParts returns a new slice of dataLen bytes originating given EC
+// parts according to rule.
+//
+// Panics if there are less than [Rule.DataPartNum] parts.
+func ConcatDataParts(rule Rule, dataLen uint64, parts [][]byte) []byte {
 	// TODO: last part may be shorter, do not overallocate buffer.
-	return slices.Concat(parts[:rule.DataPartNum]...)[:dataLen], nil
+	return slices.Concat(parts[:rule.DataPartNum]...)[:dataLen]
 }

internal/ec/ec_test.go

@@ -1,6 +1,8 @@
 package ec_test
 
 import (
+	"bytes"
+	"fmt"
 	"testing"
 
 	"github.com/klauspost/reedsolomon"
@@ -74,3 +76,44 @@ func TestEncode(t *testing.T) {
 		})
 	}
 }
+
+func testConcatDataParts(t *testing.T, ln uint64) {
+	data := testutil.RandByteSlice(ln)
+
+	for _, rule := range []iec.Rule{
+		{DataPartNum: 3, ParityPartNum: 1},
+		{DataPartNum: 12, ParityPartNum: 4},
+	} {
+		t.Run(fmt.Sprintf("rule=%s", rule.String()), func(t *testing.T) {
+			var parts [][]byte
+			if ln > 0 {
+				rs, err := reedsolomon.New(int(rule.DataPartNum), int(rule.ParityPartNum))
+				require.NoError(t, err)
+
+				parts, err = rs.Split(data)
+				require.NoError(t, err)
+			} else {
+				parts = make([][]byte, rule.DataPartNum+rule.ParityPartNum)
+			}
+
+			got := iec.ConcatDataParts(rule, ln, parts)
+			require.True(t, bytes.Equal(data, got))
+		})
+	}
+}
+
+func TestConcatDataParts(t *testing.T) {
+	for _, ln := range []uint64{
+		0,
+		1,
+		100,
+		1 << 10,
+		1<<10 + 1,
+		10 << 10,
+		10<<10 + 1,
+	} {
+		t.Run(fmt.Sprintf("len=%d", ln), func(t *testing.T) {
+			testConcatDataParts(t, ln)
+		})
+	}
+}

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. Make closer to StorageEngine.Get
+	}
+
+	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/db_test.go

@@ -2,11 +2,13 @@ package meta_test
 
 import (
 	"crypto/rand"
+	"crypto/sha256"
 	"os"
 	"path"
 	"strconv"
 	"testing"
 
+	"github.com/nspcc-dev/neofs-node/internal/testutil"
 	meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase"
 	"github.com/nspcc-dev/neofs-sdk-go/checksum"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
@@ -130,3 +132,12 @@ func setExpiration(o *object.Object, epoch uint64) {
 
 	o.SetAttributes(append(o.Attributes(), attr)...)
 }
+
+func newBlankObject(cnr cid.ID, id oid.ID) object.Object {
+	var obj object.Object
+	obj.SetContainerID(cnr)
+	obj.SetID(id)
+	obj.SetOwner(usertest.ID())
+	obj.SetPayloadChecksum(checksum.NewSHA256([sha256.Size]byte(testutil.RandByteSlice(sha256.Size))))
+	return obj
+}

pkg/local_object_storage/metabase/ec.go

@@ -0,0 +1,100 @@
+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"
+)
+
+// ResolveECPart resolves object that carries EC part produced within cnr for
+// parent object and indexed by pi, checks its availability and returns its ID.
+//
+// If object is missing, ResolveECPart returns [apistatus.ErrObjectNotFound].
+//
+// If object has expired, ResolveECPart returns [ErrObjectIsExpired].
+//
+// If object exists but tombstoned (via [DB.Inhume] or stored tombstone object),
+// ResolveECPart returns [apistatus.ErrObjectAlreadyRemoved].
+//
+// If object is marked as garbage (via [DB.MarkGarbage]), ResolveECPart returns
+// [apistatus.ErrObjectNotFound].
+//
+// If object is locked (via [DB.Lock] or stored locker object), ResolveECPart
+// ignores expiration, tombstone and garbage marks.
+func (db *DB) ResolveECPart(cnr cid.ID, parent oid.ID, pi iec.PartInfo) (oid.ID, error) {
+	db.modeMtx.RLock()
+	defer db.modeMtx.RUnlock()
+	if db.mode.NoMetabase() {
+		return oid.ID{}, ErrDegradedMode
+	}
+
+	// TODO: check already removed and expires cases.
+	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{}, apistatus.ErrObjectNotFound
+	}
+	metaBktCursor := metaBkt.Cursor()
+
+	id, err := db.resolveECPartInMetaBucket(metaBkt, metaBktCursor, parent, pi)
+	if err != nil {
+		return oid.ID{}, err
+	}
+
+	switch objectStatus(tx, metaBktCursor, oid.NewAddress(cnr, id), db.epochState.CurrentEpoch()) {
+	case statusGCMarked:
+		return oid.ID{}, apistatus.ErrObjectNotFound
+	case statusTombstoned:
+		return oid.ID{}, apistatus.ErrObjectAlreadyRemoved
+	case statusExpired:
+		return oid.ID{}, ErrObjectIsExpired
+	}
+
+	return id, nil
+}
+
+func (db *DB) resolveECPartInMetaBucket(bkt *bbolt.Bucket, crs *bbolt.Cursor, parent oid.ID, pi iec.PartInfo) (oid.ID, error) {
+	pref := slices.Concat([]byte{metaPrefixAttrIDPlain}, []byte(object.FilterParentID), objectcore.MetaAttributeDelimiter,
+		parent[:], objectcore.MetaAttributeDelimiter,
+	)
+
+	k, _ := crs.Seek(pref)
+	partID, ok := bytes.CutPrefix(k, pref)
+	if !ok {
+		return oid.ID{}, apistatus.ErrObjectNotFound
+	}
+	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 bkt.Get(k) == nil {
+		return oid.ID{}, apistatus.ErrObjectNotFound
+	}
+
+	k = slices.Concat([]byte{metaPrefixIDAttr}, partID, []byte(iec.AttributePartIdx), objectcore.MetaAttributeDelimiter, []byte(strconv.Itoa(pi.Index)))
+	if bkt.Get(k) == nil {
+		return oid.ID{}, apistatus.ErrObjectNotFound
+	}
+
+	return oid.ID(partID), nil
+}