diff --git a/go.mod b/go.mod index 689fa40a3f..c3738407b7 100644 --- a/go.mod +++ b/go.mod @@ -22,7 +22,7 @@ require ( github.com/nspcc-dev/neo-go v0.121.0 github.com/nspcc-dev/neofs-api-go/v2 v2.14.1-0.20240827150555-5ce597aa14ea github.com/nspcc-dev/neofs-contract v0.26.1 - github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.20.0.20260703200507-4f450009c764 + github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.20.0.20260722162447-2a7efa169828 github.com/nspcc-dev/tzhash v1.8.4 github.com/panjf2000/ants/v2 v2.11.5 github.com/prometheus/client_golang v1.23.2 diff --git a/go.sum b/go.sum index 4aa6bd36ca..26adca4f00 100644 --- a/go.sum +++ b/go.sum @@ -191,8 +191,8 @@ github.com/nspcc-dev/neofs-api-go/v2 v2.14.1-0.20240827150555-5ce597aa14ea h1:mK github.com/nspcc-dev/neofs-api-go/v2 v2.14.1-0.20240827150555-5ce597aa14ea/go.mod h1:YzhD4EZmC9Z/PNyd7ysC7WXgIgURc9uCG1UWDeV027Y= github.com/nspcc-dev/neofs-contract v0.26.1 h1:7Ii7Q4L3au408LOsIWKiSgfnT1g8G9jo3W7381d41T8= github.com/nspcc-dev/neofs-contract v0.26.1/go.mod h1:pevVF9OWdEN5bweKxOu6ryZv9muCEtS1ppzYM4RfBIo= -github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.20.0.20260703200507-4f450009c764 h1:7zHfAeRQsWx68owaPuyBSIAs4GIUBL5q6Q335PPGjvo= -github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.20.0.20260703200507-4f450009c764/go.mod h1:MqOYvyQgnXgEYLVzxQcZ12gp/wrJtDqMzMfjaBsisBM= +github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.20.0.20260722162447-2a7efa169828 h1:86GhprLAnQEHAUEU6vtW+FQlETjSzLnDff1/UsS9gxE= +github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.20.0.20260722162447-2a7efa169828/go.mod h1:py6ZfE9p0Bz/ZaisbY/beuCPFh+p/WbiLDyqKLeso9g= github.com/nspcc-dev/rfc6979 v0.2.4 h1:NBgsdCjhLpEPJZqmC9rciMZDcSY297po2smeaRjw57k= github.com/nspcc-dev/rfc6979 v0.2.4/go.mod h1:86ylDw6Kss+P6v4QAJqo1Sp3mC0/Zr9G97xSjQ9TuFg= github.com/nspcc-dev/tzhash v1.8.4 h1:lvuPGWsqEo9dVEvo/kdNLKv/Cy0yxRs9z5hJp8VcBuo= diff --git a/internal/ec/ec.go b/internal/ec/ec.go index 7246065c51..b785ad0dd2 100644 --- a/internal/ec/ec.go +++ b/internal/ec/ec.go @@ -1,6 +1,8 @@ package ec import ( + "crypto/sha256" + "encoding/hex" "fmt" "slices" "strconv" @@ -11,9 +13,10 @@ import ( // Erasure coding attributes. const ( - AttributePrefix = "__NEOFS__EC_" - AttributeRuleIdx = AttributePrefix + "RULE_IDX" - AttributePartIdx = AttributePrefix + "PART_IDX" + AttributePrefix = "__NEOFS__EC_" + AttributeRuleIdx = AttributePrefix + "RULE_IDX" + AttributePartIdx = AttributePrefix + "PART_IDX" + AttributePartsHashes = AttributePrefix + "PART_HASHES" ) // Rule represents erasure coding rule for object payload's encoding and placement. @@ -35,27 +38,37 @@ func (x Rule) String() string { // [Rule.DataPartNum], last data part is aligned with zeros. // // If data is empty, all parts are nil. -func Encode(rule Rule, data []byte) ([][]byte, error) { +func Encode(rule Rule, data []byte) ([][]byte, []string, error) { if len(data) == 0 { - return make([][]byte, rule.DataPartNum+rule.ParityPartNum), nil + return make([][]byte, rule.DataPartNum+rule.ParityPartNum), nil, nil } // TODO: Explore reedsolomon.Option for performance improvement. https://github.com/nspcc-dev/neofs-node/issues/3501 enc, err := newCoderForRule(rule) if err != nil { - return nil, err + return nil, nil, err } parts, err := enc.Split(data) if err != nil { - return nil, fmt.Errorf("split data: %w", err) + return nil, nil, fmt.Errorf("split data: %w", err) } if err := enc.Encode(parts); err != nil { - return nil, fmt.Errorf("calculate Reed-Solomon parity: %w", err) + return nil, nil, fmt.Errorf("calculate Reed-Solomon parity: %w", err) + } + + var ( + sums = make([]string, len(parts)) + sum [sha256.Size]byte + ) + + for i, p := range parts { + sum = sha256.Sum256(p) + sums[i] = hex.EncodeToString(sum[:]) } - return parts, nil + return parts, sums, nil } // Decode decodes source data of known len from EC parts obtained by applying diff --git a/pkg/core/object/ec.go b/pkg/core/object/ec.go index 03f655683f..7ae0828061 100644 --- a/pkg/core/object/ec.go +++ b/pkg/core/object/ec.go @@ -13,7 +13,7 @@ import ( func checkEC(hdr object.Object, rules []netmap.ECRule, blank bool, isParent bool) (bool, error) { attrs := hdr.Attributes() var ecAttr string - if len(attrs) > 0 { + if !isParent && len(attrs) > 0 { if first := attrs[0].Key(); strings.HasPrefix(first, iec.AttributePrefix) { for i := 1; i < len(attrs); i++ { if !strings.HasPrefix(attrs[i].Key(), iec.AttributePrefix) { diff --git a/pkg/network/cache/clients.go b/pkg/network/cache/clients.go index 14eef79314..1dcfc0b920 100644 --- a/pkg/network/cache/clients.go +++ b/pkg/network/cache/clients.go @@ -239,7 +239,7 @@ func (x *Clients) initConnection(ctx context.Context, pub []byte, uri string) (* if err != nil { // should never happen return nil, fmt.Errorf("init gRPC client conn: %w", err) } - res, err := client.NewGRPC(ctx, pub, grpcConn, x.signBufPool, x.streamMsgTimeout, nil) + res, err := client.NewGRPC(ctx, pub, grpcConn, x.signBufPool, x.streamMsgTimeout) if err != nil { _ = grpcConn.Close() return res, fmt.Errorf("init NeoFS API client from gRPC client conn: %w", err) diff --git a/pkg/services/object/put/distributed.go b/pkg/services/object/put/distributed.go index 4c9776b605..19c2c1a435 100644 --- a/pkg/services/object/put/distributed.go +++ b/pkg/services/object/put/distributed.go @@ -6,8 +6,10 @@ import ( "encoding/binary" "errors" "fmt" + "io" "math" "slices" + "strings" "sync" "sync/atomic" @@ -73,9 +75,18 @@ type distributedTarget struct { localOnly bool + // EC handling rules and caches: + // // When object from request is an EC part, ecPart.RuleIndex is >= 0. // Otherwise, ecPart.RuleIndex is negative. - ecPart iec.PartInfo + // When request means node-side EC encoding, ecRules is non-empty, + // original object payload is stored in objectPayload and payload + // objects are stored in encodedECParts. + ecPart iec.PartInfo + ecRules []iec.Rule + payloadAlreadyRead bool + encodedECParts [][][]byte // according to ecRules indexing + objectPayload []byte initialPolicy *netmap.InitialPlacementPolicy @@ -109,6 +120,63 @@ func newMaxReplicasError(maxReplicas, done uint, lastRule int, lastRuleErr error return newCompletionError(err, done > 0) } +// modifyECParentObject modifies parent header for EC split by attaching EC +// part hashes to parent object's attributes. It must be called only with +// non-empty t.ecRules. +func (t *distributedTarget) modifyECParentObject(hdr *object.Object, reader io.Reader) error { + if len(t.ecRules) == 0 { + panic("incorrect usage of EC parts modifier with undefined EC rules") + } + + var ( + b = getPayload() + payloadLen = int(hdr.PayloadSize()) + ) + if cap(b) < payloadLen { + putPayload(b) + b = make([]byte, 0, payloadLen) + } + buff := bytes.NewBuffer(b) + + n, err := io.Copy(buff, reader) + if err != nil { + return fmt.Errorf("reading EC object payload from slicer: %w", err) + } + if int(n) != payloadLen { + return fmt.Errorf("read %d bytes from slicer, but claimed payload len is %d ", n, payloadLen) + } + t.payloadAlreadyRead = true + t.objectPayload = buff.Bytes() + + var hashes string + for i, rule := range t.ecRules { + payloadParts, sums, err := iec.Encode(rule, t.objectPayload) + if err != nil { + return fmt.Errorf("split object payload into EC parts for rule #%d (%s): %w", i, rule, err) + } + + t.encodedECParts = append(t.encodedECParts, payloadParts) + hashes = strings.Join(sums, ",") + } + + var ( + foundAttr bool + attrs = hdr.Attributes() + ) + for _, attr := range attrs { + if attr.Key() == iec.AttributePartsHashes { + foundAttr = true + attr.SetValue(hashes) + } + } + if !foundAttr { + attrs = append(attrs, object.NewAttribute(iec.AttributePartsHashes, hashes)) + } + hdr.SetAttributes(attrs...) + + return nil +} + func (t *distributedTarget) WriteHeader(hdr *object.Object) error { payloadLen := hdr.PayloadSize() if payloadLen > math.MaxInt { @@ -140,7 +208,14 @@ func (t *distributedTarget) WriteHeader(hdr *object.Object) error { } func (t *distributedTarget) Write(p []byte) (n int, err error) { - t.encodedObject.b = append(t.encodedObject.b, p...) + if t.payloadAlreadyRead { + if t.objectPayload != nil { + t.encodedObject.b = append(t.encodedObject.b, t.objectPayload...) + t.objectPayload = nil + } + } else { + t.encodedObject.b = append(t.encodedObject.b, p...) + } return len(p), nil } @@ -148,6 +223,7 @@ func (t *distributedTarget) Write(p []byte) (n int, err error) { func (t *distributedTarget) Close() (oid.ID, error) { defer func() { putPayload(t.encodedObject.b) + t.payloadAlreadyRead = false t.encodedObject.b = nil t.resetMetaCollection() }() @@ -375,11 +451,7 @@ nextRule: continue } - payloadParts, err := iec.Encode(ecRules[ecRuleIdx], obj.Payload()) - if err != nil { - return fmt.Errorf("split object payload into EC parts for rule #%d (%s): %w", ecRuleIdx, ecRules[ecRuleIdx], err) - } - + payloadParts := t.encodedECParts[ruleIdx] fin, err := handleECRule(ruleIdx, ecRuleIdx, payloadParts, ecRules[ecRuleIdx]) if err != nil { return err @@ -470,28 +542,12 @@ func (t *distributedTarget) replicateRemainingPrimaryNodes(obj object.Object, no } func (t *distributedTarget) replicateRemainingECRules(obj object.Object, ecRules []iec.Rule, nodeLists [][]netmap.NodeInfo, applied []bool) { - encodedByRule := make(map[iec.Rule][][]byte, len(ecRules)) - for ruleIdx := range ecRules { if applied[ruleIdx] { continue } - payloadParts, ok := encodedByRule[ecRules[ruleIdx]] - if !ok { - var err error - payloadParts, err = iec.Encode(ecRules[ruleIdx], obj.Payload()) - if err != nil { - t.placementIterator.log.Info("failed to encode object for post-placement EC replication", - zap.Stringer("object", obj.Address()), - zap.Int("rule_idx", ruleIdx), - zap.Stringer("rule", ecRules[ruleIdx]), - zap.Error(err)) - continue - } - encodedByRule[ecRules[ruleIdx]] = payloadParts - } - + payloadParts := t.encodedECParts[ruleIdx] totalParts := len(payloadParts) for partIdx := range payloadParts { partObj, err := formObjectForECPart(t.sessionSigner, obj, ruleIdx, partIdx, payloadParts) diff --git a/pkg/services/object/put/prm.go b/pkg/services/object/put/prm.go index 9d107f9518..3e9a842e81 100644 --- a/pkg/services/object/put/prm.go +++ b/pkg/services/object/put/prm.go @@ -20,6 +20,7 @@ type PutInitPrm struct { containerNodes ContainerNodes ecPart iec.PartInfo + ecRules []iec.Rule localNodeInContainer bool localSignerRFC6979 neofscrypto.Signer localNodeSigner neofscrypto.Signer diff --git a/pkg/services/object/put/slice.go b/pkg/services/object/put/slice.go index da750bcdd3..f807671d6c 100644 --- a/pkg/services/object/put/slice.go +++ b/pkg/services/object/put/slice.go @@ -4,11 +4,13 @@ import ( "context" "errors" "fmt" + "io" "math" "github.com/nspcc-dev/neofs-node/pkg/services/object/internal" "github.com/nspcc-dev/neofs-sdk-go/client" apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status" + "github.com/nspcc-dev/neofs-sdk-go/container" "github.com/nspcc-dev/neofs-sdk-go/object" oid "github.com/nspcc-dev/neofs-sdk-go/object/id" "github.com/nspcc-dev/neofs-sdk-go/object/slicer" @@ -17,15 +19,18 @@ import ( "github.com/nspcc-dev/neofs-sdk-go/user" ) +type splitObjectModifier func(*object.Object, io.Reader) error + type slicingTarget struct { ctx context.Context + cnr container.Container signer user.Signer sessionToken *session.Object sessionTokenV2 *sessionv2.Token currentEpoch uint64 maxObjSize uint64 - - nextTarget internal.Target + splitModifier splitObjectModifier + nextTarget internal.Target payloadWriter *slicer.PayloadWriter } @@ -41,6 +46,7 @@ func newSlicingTarget( sessionTokenV2 *sessionv2.Token, curEpoch uint64, initNextTarget internal.Target, + splitModifier splitObjectModifier, ) internal.Target { return &slicingTarget{ ctx: ctx, @@ -49,6 +55,7 @@ func newSlicingTarget( sessionTokenV2: sessionTokenV2, currentEpoch: curEpoch, maxObjSize: maxObjSize, + splitModifier: splitModifier, nextTarget: initNextTarget, } } @@ -62,6 +69,9 @@ func (x *slicingTarget) WriteHeader(hdr *object.Object) error { } else if x.sessionToken != nil { opts.SetSession(*x.sessionToken) } + if x.splitModifier != nil { + opts.SetSplitChainModifier(x.splitModifier) + } if payloadSize := hdr.PayloadSize(); payloadSize != 0 && payloadSize != math.MaxUint64 { // https://github.com/nspcc-dev/neofs-api/blob/d95228c40283cf6e188073a87a802af7e5dc0a7d/object/types.proto#L93-L95 diff --git a/pkg/services/object/put/streamer.go b/pkg/services/object/put/streamer.go index 2e21c25922..adc91b3a84 100644 --- a/pkg/services/object/put/streamer.go +++ b/pkg/services/object/put/streamer.go @@ -84,7 +84,7 @@ func (p *Streamer) initTarget(prm *PutInitPrm) error { p.target = &validatingTarget{ l: p.log, ctx: p.ctx, - nextTarget: p.newCommonTarget(prm), + nextTarget: p.newDistrubutedWriter(prm), fmt: p.fmtValidator, quotaLimiter: p.quotaLimiter, cachedCnr: prm.cnr, @@ -151,6 +151,13 @@ func (p *Streamer) initTarget(prm *PutInitPrm) error { sessionSigner := user.NewAutoIDSigner(*sessionKey) prm.sessionSigner = sessionSigner + var ( + dTarget = p.newDistrubutedWriter(prm) + splitModifier splitObjectModifier + ) + if len(prm.ecRules) > 0 && prm.ecPart.RuleIndex < 0 { + splitModifier = dTarget.modifyECParentObject + } p.target = &validatingTarget{ l: p.log, ctx: p.ctx, @@ -166,7 +173,8 @@ func (p *Streamer) initTarget(prm *PutInitPrm) error { sToken, sTokenV2, p.networkState.CurrentEpoch(), - p.newCommonTarget(prm), + dTarget, + splitModifier, ), } @@ -206,7 +214,8 @@ func (p *Streamer) preparePrm(prm *PutInitPrm) error { return fmt.Errorf("select storage nodes for the container: %w", err) } cnrNodes := prm.containerNodes.Unsorted() - ecRulesN := len(prm.containerNodes.ECRules()) + ecRules := prm.containerNodes.ECRules() + ecRulesN := len(ecRules) if typ := prm.hdr.Type(); ecRulesN > 0 && typ != object.TypeTombstone && typ != object.TypeLock && typ != object.TypeLink { ecPart, err := iec.GetPartInfo(*prm.hdr) if err != nil { @@ -230,6 +239,7 @@ func (p *Streamer) preparePrm(prm *PutInitPrm) error { } prm.ecPart = ecPart + prm.ecRules = ecRules } else { prm.localNodeInContainer = localNodeInSets(p.neoFSNet, cnrNodes) prm.ecPart.RuleIndex = -1 @@ -244,7 +254,7 @@ func (p *Streamer) preparePrm(prm *PutInitPrm) error { return nil } -func (p *Streamer) newCommonTarget(prm *PutInitPrm) internal.Target { +func (p *Streamer) newDistrubutedWriter(prm *PutInitPrm) *distributedTarget { var relay func(nodeDesc) error if p.relay != nil { relay = func(node nodeDesc) error { @@ -276,6 +286,7 @@ func (p *Streamer) newCommonTarget(prm *PutInitPrm) internal.Target { fmt: p.fmtValidator, containerNodes: prm.containerNodes, ecPart: prm.ecPart, + ecRules: prm.ecRules, localNodeInContainer: prm.localNodeInContainer, localNodeSigner: prm.localNodeSigner, sessionSigner: prm.sessionSigner,