WIP

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

Author: cthulhu-rider

cmd/neofs-node/object.go

@@ -199,7 +199,7 @@ func initObjectService(c *cfg) {
 		),
 	)
 
-	c.policer = policer.New(
+	c.policer = policer.New(neofsecdsa.Signer(c.key.PrivateKey),
 		policer.WithLogger(c.log),
 		policer.WithLocalStorage(ls),
 		policer.WithRemoteHeader(

internal/ec/ec.go

@@ -2,6 +2,7 @@ package ec
 
 import (
 	"fmt"
+	"iter"
 	"slices"
 	"strconv"
 
@@ -40,10 +41,9 @@ func Encode(rule Rule, data []byte) ([][]byte, error) {
 		return make([][]byte, rule.DataPartNum+rule.ParityPartNum), nil
 	}
 
-	// TODO: Explore reedsolomon.Option for performance improvement. https://github.com/nspcc-dev/neofs-node/issues/3501
-	enc, err := reedsolomon.New(int(rule.DataPartNum), int(rule.ParityPartNum))
-	if err != nil { // should never happen with correct rule
-		return nil, fmt.Errorf("init Reed-Solomon encoder: %w", err)
+	enc, err := newCoderForRule(rule)
+	if err != nil {
+		return nil, err
 	}
 
 	parts, err := enc.Split(data)
@@ -61,26 +61,44 @@ func Encode(rule Rule, data []byte) ([][]byte, error) {
 // Decode decodes source data of known len from EC parts obtained by applying
 // specified rule.
 func Decode(rule Rule, dataLen uint64, parts [][]byte) ([]byte, error) {
-	// TODO: Explore reedsolomon.Option for performance improvement. https://github.com/nspcc-dev/neofs-node/issues/3501
-	dec, err := reedsolomon.New(int(rule.DataPartNum), int(rule.ParityPartNum))
-	if err != nil { // should never happen with correct rule
-		return nil, fmt.Errorf("init Reed-Solomon decoder: %w", err)
+	if err := decodeIndex(rule, parts, func(yield func(int) bool) {
+		for i := range int(rule.DataPartNum) {
+			if !yield(i) {
+				return
+			}
+		}
+	}); err != nil {
+		return nil, err
 	}
 
-	required := make([]bool, rule.DataPartNum+rule.ParityPartNum)
-	for i := range rule.DataPartNum {
-		required[i] = true
+	if got := islices.TwoDimSliceElementCount(parts[:rule.DataPartNum]); uint64(got) < dataLen {
+		return nil, fmt.Errorf("sum len of received data parts is less than full len: %d < %d", got, dataLen)
+	}
+
+	return ConcatDataParts(rule, dataLen, parts), nil
+}
+
+// DecodeIndexes decodes specified EC parts obtained by applying specified rule.
+func DecodeIndexes(rule Rule, parts [][]byte, idxs []int) error {
+	return decodeIndex(rule, parts, slices.Values(idxs))
+}
+
+func decodeIndex(rule Rule, parts [][]byte, idxs iter.Seq[int]) error {
+	rs, err := newCoderForRule(rule)
+	if err != nil {
+		return err
 	}
 
-	if err := dec.ReconstructSome(parts, required); err != nil {
-		return nil, fmt.Errorf("restore Reed-Solomon: %w", err)
+	required := make([]bool, rule.DataPartNum+rule.ParityPartNum)
+	for idx := range idxs {
+		required[idx] = true
 	}
 
-	if got := islices.TwoDimSliceElementCount(parts[:rule.DataPartNum]); uint64(got) < dataLen {
-		return nil, fmt.Errorf("sum len of received data parts is less than full len: %d < %d", got, dataLen)
+	if err := rs.ReconstructSome(parts, required); err != nil {
+		return fmt.Errorf("restore Reed-Solomon: %w", err)
 	}
 
-	return ConcatDataParts(rule, dataLen, parts), nil
+	return nil
 }
 
 // ConcatDataParts returns a new slice of dataLen bytes originating given EC
@@ -91,3 +109,13 @@ 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]
 }
+
+func newCoderForRule(rule Rule) (reedsolomon.Encoder, error) {
+	// TODO: Explore reedsolomon.Option for performance improvement. https://github.com/nspcc-dev/neofs-node/issues/3501
+	enc, err := reedsolomon.New(int(rule.DataPartNum), int(rule.ParityPartNum))
+	if err != nil { // should never happen with correct rule
+		return nil, fmt.Errorf("init Reed-Solomon decoder: %w", err)
+	}
+
+	return enc, nil
+}

internal/ec/objects.go

@@ -53,13 +53,25 @@ func GetPartInfo(obj object.Object) (PartInfo, error) {
 
 // FormObjectForECPart forms object for EC part produced from given parent object.
 func FormObjectForECPart(signer neofscrypto.Signer, parent object.Object, part []byte, partInfo PartInfo) (object.Object, error) {
+	return formObjectForECPart(signer, parent, part, partInfo, true)
+}
+
+// FormObjectForECPartWithoutSession is similar to [FormObjectForECPart] but
+// leaves session token unset.
+func FormObjectForECPartWithoutSession(signer neofscrypto.Signer, parent object.Object, part []byte, partInfo PartInfo) (object.Object, error) {
+	return formObjectForECPart(signer, parent, part, partInfo, false)
+}
+
+func formObjectForECPart(signer neofscrypto.Signer, parent object.Object, part []byte, partInfo PartInfo, withSession bool) (object.Object, error) {
 	var obj object.Object
 	obj.SetVersion(parent.Version())
 	obj.SetContainerID(parent.GetContainerID())
 	obj.SetOwner(parent.Owner())
 	obj.SetCreationEpoch(parent.CreationEpoch())
 	obj.SetType(object.TypeRegular)
-	obj.SetSessionToken(parent.SessionToken())
+	if withSession {
+		obj.SetSessionToken(parent.SessionToken())
+	}
 
 	obj.SetParent(&parent)
 	iobject.SetIntAttribute(&obj, AttributeRuleIdx, partInfo.RuleIndex)

pkg/local_object_storage/engine/ec.go

@@ -104,3 +104,76 @@ loop:
 
 	return object.Object{}, nil, apistatus.ErrObjectNotFound
 }
+
+// HeadECPart is similar to [StorageEngine.GetECPart] but returns only the header.
+// TODO: unit tests.
+func (e *StorageEngine) HeadECPart(cnr cid.ID, parent oid.ID, pi iec.PartInfo) (object.Object, error) {
+	if e.metrics != nil {
+		defer elapsed(e.metrics.AddHeadECPartDuration)()
+	}
+
+	e.blockMtx.RLock()
+	defer e.blockMtx.RUnlock()
+	if e.blockErr != nil {
+		return object.Object{}, e.blockErr
+	}
+
+	// TODO: sync placement with PUT. They should sort shards equally, but now PUT sorts by part ID.
+	//  https://github.com/nspcc-dev/neofs-node/issues/3537
+	s := e.sortShardsFn(e, oid.NewAddress(cnr, parent))
+
+	var partID oid.ID
+loop:
+	for i := range s {
+		hdr, err := s[i].shardIface.HeadECPart(cnr, parent, pi)
+		switch {
+		case err == nil:
+			return hdr, nil
+		case errors.Is(err, apistatus.ErrObjectAlreadyRemoved):
+			return object.Object{}, err
+		case errors.Is(err, meta.ErrObjectIsExpired):
+			return object.Object{}, apistatus.ErrObjectNotFound // like Get
+		case errors.As(err, (*ierrors.ObjectID)(&partID)):
+			if partID.IsZero() {
+				panic("zero object ID returned as error")
+			}
+
+			e.log.Info("EC part's object ID resolved in shard but reading failed, continue by ID",
+				zap.Stringer("container", cnr), zap.Stringer("parent", parent),
+				zap.Int("ecRule", pi.RuleIndex), zap.Int("partIdx", pi.Index),
+				zap.Stringer("shardID", s[i].shardIface.ID()), zap.Error(err))
+			// TODO: need report error? Same for other places. https://github.com/nspcc-dev/neofs-node/issues/3538
+
+			s = s[i+1:]
+			break loop
+		case errors.Is(err, apistatus.ErrObjectNotFound):
+		default:
+			e.log.Info("failed to get EC part header from shard, ignore error",
+				zap.Stringer("container", cnr), zap.Stringer("parent", parent),
+				zap.Int("ecRule", pi.RuleIndex), zap.Int("partIdx", pi.Index),
+				zap.Stringer("shardID", s[i].shardIface.ID()), zap.Error(err))
+		}
+	}
+
+	if partID.IsZero() {
+		return object.Object{}, apistatus.ErrObjectNotFound
+	}
+
+	for i := range s {
+		// get an object bypassing the metabase. We can miss deletion or expiration mark. Get behaves like this, so here too.
+		hdr, err := s[i].shardIface.Head(oid.NewAddress(cnr, partID), true)
+		switch {
+		case err == nil:
+			return *hdr, nil
+		case errors.Is(err, apistatus.ErrObjectNotFound):
+		default:
+			e.log.Info("failed to get EC part header from shard bypassing metabase, ignore error",
+				zap.Stringer("container", cnr), zap.Stringer("parent", parent),
+				zap.Int("ecRule", pi.RuleIndex), zap.Int("partIdx", pi.Index),
+				zap.Stringer("partID", partID),
+				zap.Stringer("shardID", s[i].shardIface.ID()), zap.Error(err))
+		}
+	}
+
+	return object.Object{}, apistatus.ErrObjectNotFound
+}

pkg/local_object_storage/engine/engine.go

@@ -15,6 +15,7 @@ import (
 	"github.com/nspcc-dev/neofs-node/pkg/util"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
 	"github.com/nspcc-dev/neofs-sdk-go/object"
+	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
 	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
 	"go.uber.org/zap"
 )
@@ -47,6 +48,8 @@ type shardInterface interface {
 	ID() *shard.ID
 	GetStream(oid.Address, bool) (*object.Object, io.ReadCloser, error)
 	GetECPart(cid.ID, oid.ID, iec.PartInfo) (object.Object, io.ReadCloser, error)
+	Head(oid.Address, bool) (*objectSDK.Object, error)
+	HeadECPart(cid.ID, oid.ID, iec.PartInfo) (object.Object, error)
 }
 
 type shardWrapper struct {

pkg/local_object_storage/engine/metrics.go

@@ -18,6 +18,7 @@ type MetricRegister interface {
 	AddSearchDuration(d time.Duration)
 	AddListObjectsDuration(d time.Duration)
 	AddGetECPartDuration(d time.Duration)
+	AddHeadECPartDuration(d time.Duration)
 
 	SetObjectCounter(shardID, objectType string, v uint64)
 	AddToObjectCounter(shardID, objectType string, delta int)

pkg/local_object_storage/shard/ec.go

@@ -65,3 +65,33 @@ func (s *Shard) GetECPart(cnr cid.ID, parent oid.ID, pi iec.PartInfo) (object.Ob
 
 	return *hdr, rdr, nil
 }
+
+// HeadECPart is similar to [Shard.GetECPart] but returns only the header.
+// TODO: unit tests.
+func (s *Shard) HeadECPart(cnr cid.ID, parent oid.ID, pi iec.PartInfo) (object.Object, error) {
+	partID, err := s.metaBaseIface.ResolveECPart(cnr, parent, pi)
+	if err != nil {
+		return object.Object{}, fmt.Errorf("resolve part ID in metabase: %w", err)
+	}
+
+	partAddr := oid.NewAddress(cnr, partID)
+	if s.hasWriteCache() {
+		hdr, err := s.writeCache.Head(partAddr)
+		if err == nil {
+			return *hdr, nil
+		}
+
+		if errors.Is(err, apistatus.ErrObjectNotFound) {
+			s.log.Debug("EC part object is missing in write-cache, fallback to BLOB storage", zap.Stringer("partAddr", partAddr), zap.Error(err))
+		} else {
+			s.log.Info("failed to get EC part object header from write-cache, fallback to BLOB storage", zap.Stringer("partAddr", partAddr), zap.Error(err))
+		}
+	}
+
+	hdr, err := s.blobStor.Head(partAddr)
+	if err != nil {
+		return object.Object{}, fmt.Errorf("get header from BLOB storage by ID %w: %w", ierrors.ObjectID(partID), err)
+	}
+
+	return *hdr, nil
+}

pkg/metrics/engine.go

@@ -21,6 +21,7 @@ type (
 		searchDuration                prometheus.Histogram
 		listObjectsDuration           prometheus.Histogram
 		getECPartDuration             prometheus.Histogram
+		headECPartDuration            prometheus.Histogram
 
 		containerSize prometheus.GaugeVec
 		payloadSize   prometheus.GaugeVec
@@ -122,6 +123,13 @@ func newEngineMetrics() engineMetrics {
 			Name:      "get_ec_part_time",
 			Help:      "Engine 'get EC part' operations handling time",
 		})
+
+		headECPartDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: storageNodeNameSpace,
+			Subsystem: engineSubsystem,
+			Name:      "head_ec_part_time",
+			Help:      "Engine 'head EC part' operations handling time",
+		})
 	)
 
 	var (
@@ -161,6 +169,7 @@ func newEngineMetrics() engineMetrics {
 		searchDuration:                searchDuration,
 		listObjectsDuration:           listObjectsDuration,
 		getECPartDuration:             getECPartDuration,
+		headECPartDuration:            headECPartDuration,
 		containerSize:                 *containerSize,
 		payloadSize:                   *payloadSize,
 		capacitySize:                  *capacitySize,
@@ -181,6 +190,7 @@ func (m engineMetrics) register() {
 	prometheus.MustRegister(m.searchDuration)
 	prometheus.MustRegister(m.listObjectsDuration)
 	prometheus.MustRegister(m.getECPartDuration)
+	prometheus.MustRegister(m.headECPartDuration)
 	prometheus.MustRegister(m.containerSize)
 	prometheus.MustRegister(m.payloadSize)
 	prometheus.MustRegister(m.capacitySize)
@@ -238,6 +248,10 @@ func (m engineMetrics) AddGetECPartDuration(d time.Duration) {
 	m.getECPartDuration.Observe(d.Seconds())
 }
 
+func (m engineMetrics) AddHeadECPartDuration(d time.Duration) {
+	m.headECPartDuration.Observe(d.Seconds())
+}
+
 func (m engineMetrics) AddToContainerSize(cnrID string, size int64) {
 	m.containerSize.With(
 		prometheus.Labels{

pkg/services/object/get/ec.go

@@ -47,6 +47,20 @@ func (s *Service) copyLocalECPart(dst ObjectWriter, cnr cid.ID, parent oid.ID, p
 	return nil
 }
 
+// similar to copyLocalECPart but returns only the header.
+func (s *Service) copyLocalECPartHeader(dst internal.HeaderWriter, cnr cid.ID, parent oid.ID, pi iec.PartInfo) error {
+	hdr, err := s.localObjects.HeadECPart(cnr, parent, pi)
+	if err != nil {
+		return fmt.Errorf("get object header from local storage: %w", err)
+	}
+
+	if err := dst.WriteHeader(&hdr); err != nil {
+		return fmt.Errorf("write header: %w", err)
+	}
+
+	return nil
+}
+
 func (s *Service) copyECObject(ctx context.Context, cnr cid.ID, parent oid.ID, sTok *session.Object,
 	ecRules []iec.Rule, sortedNodeLists [][]netmap.NodeInfo, dst ObjectWriter) error {
 	obj, err := s.restoreFromECParts(ctx, cnr, parent, sTok, ecRules, sortedNodeLists)
@@ -527,3 +541,19 @@ func checkECAttributesInReceivedObject(hdr object.Object, ruleIdx, partIdx strin
 
 	return fmt.Errorf("not all EC attributes received: requested %d, got %d", expected, found)
 }
+
+func checkPartRequestAgainstPolicy(ecRules []iec.Rule, pi iec.PartInfo) error {
+	if len(ecRules) == 0 {
+		return errors.New("EC part requested in container without EC policy")
+	}
+
+	if pi.RuleIndex >= len(ecRules) {
+		return fmt.Errorf("EC rule index overflows container policy: idx=%d,rules=%d", pi.RuleIndex, len(ecRules))
+	}
+
+	if total := ecRules[pi.RuleIndex].DataPartNum + ecRules[pi.RuleIndex].ParityPartNum; pi.Index >= int(total) {
+		return fmt.Errorf("EC part index overflows container policy: idx=%d,parts=%d", pi.Index, total)
+	}
+
+	return nil
+}

pkg/services/object/get/get.go

@@ -97,11 +97,26 @@ func (s *Service) GetRangeHash(ctx context.Context, prm RangeHashPrm) (*RangeHas
 // Returns ErrNotFound if the header was not received for the call.
 // Returns SplitInfoError if object is virtual and raw flag is set.
 func (s *Service) Head(ctx context.Context, prm HeadPrm) error {
+	pi, err := checkECPartInfoRequest(prm.common.XHeaders())
+	if err != nil {
+		// TODO: track https://github.com/nspcc-dev/neofs-api/issues/269.
+		return fmt.Errorf("invalid request: %w", err)
+	}
+
 	nodeLists, repRules, ecRules, err := s.neoFSNet.GetNodesForObject(prm.addr)
 	if err != nil {
 		return fmt.Errorf("get nodes for object: %w", err)
 	}
 
+	if pi.RuleIndex >= 0 {
+		if err := checkPartRequestAgainstPolicy(ecRules, pi); err != nil {
+			// TODO: track https://github.com/nspcc-dev/neofs-api/issues/269.
+			return fmt.Errorf("invalid request: %w", err)
+		}
+		// TODO: deny if node is not in the container?
+		return s.copyLocalECPartHeader(prm.objWriter, prm.addr.Container(), prm.addr.Object(), pi)
+	}
+
 	if len(repRules) > 0 {
 		err := s.get(ctx, prm.commonPrm, headOnly(), withPreSortedContainerNodes(nodeLists[:len(repRules)], repRules)).err
 		if len(ecRules) == 0 || !errors.Is(err, apistatus.ErrObjectNotFound) {