Skip to content

EC GET without part idxs #4125

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
cthulhu-rider wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

EC GET without part idxs #4125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7096a4c
sn/object: Allow ranged GET without part index
cthulhu-rider Aug 10, 2026
673cede
sn/object: Send requests without part index in EC GET
cthulhu-rider Aug 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 33 additions & 39 deletions pkg/services/object/get.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,11 +333,11 @@ type getECTransport struct {
signResponses bool
responseStream grpc.ServerStream

getPartRequest mem.Buffer
getPartRequestInfo iec.PartInfo
getPartRequest mem.Buffer
getPartRequestRuleIdx int

getPartRangeRequestsMtx sync.RWMutex
getPartRangeRequests map[iec.PartInfo]*preparedRangeRequest
getPartRangeRequests map[int]*preparedRangeRequest
}

// CopyLocalECPartParentHeaderAndPayload implements [getsvc.GetECRequestTransport].
Expand Down Expand Up @@ -479,24 +479,24 @@ func (x *getECTransport) CopyLocalECPartRange(ctx context.Context, storage *engi
return ln, nil
}

func (x *getECTransport) initGetPartRequest(remoteServerAPIVersion *protorefs.Version, partInfo iec.PartInfo) error {
if x.getPartRequestInfo == partInfo && x.getPartRequest != nil {
func (x *getECTransport) initGetPartRequest(remoteServerAPIVersion *protorefs.Version, ruleIdx int) error {
if x.getPartRequestRuleIdx == ruleIdx && x.getPartRequest != nil {
return nil
}

var err error
x.getPartRequest, err = x.server.makeGetECPartRequest(remoteServerAPIVersion, x.requestContainer, x.requestObject, partInfo)
x.getPartRequest, err = x.server.makeGetECPartRequest(remoteServerAPIVersion, x.requestContainer, x.requestObject, ruleIdx)
if err != nil {
return fmt.Errorf("make GET request: %w", err)
}

x.getPartRequestInfo = partInfo
x.getPartRequestRuleIdx = ruleIdx

return nil
}

// CopyRemoteECPartParentHeaderAndPayload implements [getsvc.GetECRequestTransport].
func (x *getECTransport) CopyRemoteECPartParentHeaderAndPayload(ctx context.Context, conn clientcore.MultiAddressClient, partInfo iec.PartInfo) (bool, uint64, uint64, uint64, error) {
func (x *getECTransport) CopyRemoteECPartParentHeaderAndPayload(ctx context.Context, conn clientcore.MultiAddressClient, ruleIdx int) (bool, uint64, uint64, uint64, error) {
var copiedHdr bool
var parentPldLen uint64
var partPldLen uint64
Expand All @@ -506,7 +506,8 @@ func (x *getECTransport) CopyRemoteECPartParentHeaderAndPayload(ctx context.Cont

err := conn.ForAnyGRPCConn(ctx, func(ctx context.Context, conn *grpc.ClientConn) error {
if !copiedHdr {
if err := x.initGetPartRequest(connAPIVersion, partInfo); err != nil {
// FIXME: another part can be received. Same for ranges.
if err := x.initGetPartRequest(connAPIVersion, ruleIdx); err != nil {
return err
}

Expand All @@ -523,7 +524,7 @@ func (x *getECTransport) CopyRemoteECPartParentHeaderAndPayload(ctx context.Cont
return clientcore.ErrSkipConnection
}

copiedFromNode, err := x.copyRemotePartRange(ctx, conn, connAPIVersion, partInfo, copiedPartPld, partPldLen-copiedPartPld, nil)
copiedFromNode, err := x.copyRemotePartRange(ctx, conn, connAPIVersion, ruleIdx, copiedPartPld, partPldLen-copiedPartPld, nil)
if err != nil {
return err
}
Expand Down Expand Up @@ -650,8 +651,8 @@ func (x *getECTransport) copyRemotePart(ctx context.Context, conn *grpc.ClientCo
return copiedHdr, parentPldLen, partPldLen, copiedPartPldLen, nil
}

func (x *getECTransport) copyRemotePartRange(ctx context.Context, conn *grpc.ClientConn, connAPIVersion *protorefs.Version, partInfo iec.PartInfo, off, ln uint64, controlCh <-chan bool) (uint64, error) {
request, err := x.makeGetECPartRangeRequest(connAPIVersion, partInfo, off, ln)
func (x *getECTransport) copyRemotePartRange(ctx context.Context, conn *grpc.ClientConn, connAPIVersion *protorefs.Version, ruleIdx int, off, ln uint64, controlCh <-chan bool) (uint64, error) {
request, err := x.makeGetECPartRangeRequest(connAPIVersion, ruleIdx, off, ln)
if err != nil {
return 0, fmt.Errorf("make request: %w", err)
}
Expand Down Expand Up @@ -831,7 +832,7 @@ func handleGetECPartResponseInit(buffers iprotobuf.BuffersSlice) (iprotobuf.Buff
return parentID, parentSig, parentHdr, parentPldLen, partPldLen, nil
}

func (x *getECTransport) CopyRemoteECPartRange(ctx context.Context, conn clientcore.MultiAddressClient, partInfo iec.PartInfo, off uint64, ln uint64, full bool, controlCh <-chan bool) (uint64, error) {
func (x *getECTransport) CopyRemoteECPartRange(ctx context.Context, conn clientcore.MultiAddressClient, ruleIdx int, off uint64, ln uint64, full bool, controlCh <-chan bool) (uint64, error) {
var copiedPld uint64

connAPIVersion := conn.APIVersion()
Expand All @@ -842,7 +843,7 @@ func (x *getECTransport) CopyRemoteECPartRange(ctx context.Context, conn clientc
reqLen = ln - copiedPld
}

copiedFromNode, err := x.copyRemotePartRange(ctx, conn, connAPIVersion, partInfo, off+copiedPld, reqLen, controlCh)
copiedFromNode, err := x.copyRemotePartRange(ctx, conn, connAPIVersion, ruleIdx, off+copiedPld, reqLen, controlCh)
if err != nil {
return err
}
Expand All @@ -865,14 +866,12 @@ func (x *getECTransport) CopyRemoteECPartRange(ctx context.Context, conn clientc
return copiedPld, nil
}

func (s *Server) makeGetECPartRequest(remoteServerAPIVersion *protorefs.Version, cnr cid.ID, parent oid.ID, partInfo iec.PartInfo) (mem.Buffer, error) {
ruleIdxStr := strconv.Itoa(partInfo.RuleIndex)
partIdxStr := strconv.Itoa(partInfo.Index)
func (s *Server) makeGetECPartRequest(remoteServerAPIVersion *protorefs.Version, cnr cid.ID, parent oid.ID, ruleIdx int) (mem.Buffer, error) {
ruleIdxStr := strconv.Itoa(ruleIdx)

ruleIdxHdrLen := calculateXHeaderLength(iec.AttributeRuleIdx, ruleIdxStr)
partIdxHdrLen := calculateXHeaderLength(iec.AttributePartIdx, partIdxStr)

metaHdrLen := calculateGetECPartRequestMetaHeaderLength(ruleIdxHdrLen, partIdxHdrLen)
metaHdrLen := calculateGetECPartRequestMetaHeaderLength(ruleIdxHdrLen)

verifHdrSigCount := getRequestVerificationSignaturesCount(remoteServerAPIVersion)

Expand All @@ -883,7 +882,7 @@ func (s *Server) makeGetECPartRequest(remoteServerAPIVersion *protorefs.Version,
// TODO: try with sync.Pool
buf := make([]byte, reqLen)

n, err := s.writeGetECPartRequest(buf, cnr, parent, metaHdrLen, ruleIdxHdrLen, ruleIdxStr, partIdxHdrLen, partIdxStr, verifHdrSigCount)
n, err := s.writeGetECPartRequest(buf, cnr, parent, metaHdrLen, ruleIdxHdrLen, ruleIdxStr, verifHdrSigCount)
if err != nil {
return nil, err
}
Expand All @@ -894,21 +893,21 @@ func (s *Server) makeGetECPartRequest(remoteServerAPIVersion *protorefs.Version,
return mem.SliceBuffer(buf), nil
}

func (x *getECTransport) makeGetECPartRangeRequest(remoteServerAPIVersion *protorefs.Version, partInfo iec.PartInfo, off, ln uint64) (mem.Buffer, error) {
func (x *getECTransport) makeGetECPartRangeRequest(remoteServerAPIVersion *protorefs.Version, ruleIdx int, off, ln uint64) (mem.Buffer, error) {
x.getPartRangeRequestsMtx.RLock()
req := x.getPartRangeRequests[partInfo]
req := x.getPartRangeRequests[ruleIdx]
x.getPartRangeRequestsMtx.RUnlock()

if req == nil {
x.getPartRangeRequestsMtx.Lock()

req = x.getPartRangeRequests[partInfo]
req = x.getPartRangeRequests[ruleIdx]
if req == nil {
if x.getPartRangeRequests == nil {
x.getPartRangeRequests = make(map[iec.PartInfo]*preparedRangeRequest, 1)
x.getPartRangeRequests = make(map[int]*preparedRangeRequest, 1)
}
req = new(preparedRangeRequest)
x.getPartRangeRequests[partInfo] = req
x.getPartRangeRequests[ruleIdx] = req
}

x.getPartRangeRequestsMtx.Unlock()
Expand All @@ -918,7 +917,7 @@ func (x *getECTransport) makeGetECPartRangeRequest(remoteServerAPIVersion *proto
return req.buffer, nil
}

reqBuf, err := x.server.makeGetECPartRangeRequest(remoteServerAPIVersion, x.requestContainer, x.requestObject, partInfo, off, ln)
reqBuf, err := x.server.makeGetECPartRangeRequest(remoteServerAPIVersion, x.requestContainer, x.requestObject, ruleIdx, off, ln)
if err != nil {
// stream is closed by context cancellation
return nil, err
Expand All @@ -931,14 +930,12 @@ func (x *getECTransport) makeGetECPartRangeRequest(remoteServerAPIVersion *proto
return reqBuf, nil
}

func (s *Server) makeGetECPartRangeRequest(remoteServerAPIVersion *protorefs.Version, cnr cid.ID, parent oid.ID, partInfo iec.PartInfo, off, ln uint64) (mem.Buffer, error) {
ruleIdxStr := strconv.Itoa(partInfo.RuleIndex)
partIdxStr := strconv.Itoa(partInfo.Index)
func (s *Server) makeGetECPartRangeRequest(remoteServerAPIVersion *protorefs.Version, cnr cid.ID, parent oid.ID, ruleIdx int, off, ln uint64) (mem.Buffer, error) {
ruleIdxStr := strconv.Itoa(ruleIdx)

ruleIdxHdrLen := calculateXHeaderLength(iec.AttributeRuleIdx, ruleIdxStr)
partIdxHdrLen := calculateXHeaderLength(iec.AttributePartIdx, partIdxStr)

metaHdrLen := calculateGetECPartRequestMetaHeaderLength(ruleIdxHdrLen, partIdxHdrLen)
metaHdrLen := calculateGetECPartRequestMetaHeaderLength(ruleIdxHdrLen)

var rngLen int
if off != 0 {
Expand Down Expand Up @@ -967,7 +964,7 @@ func (s *Server) makeGetECPartRangeRequest(remoteServerAPIVersion *protorefs.Ver
buf := make([]byte, reqLen)

n, err := s.writeGetECPartRangeRequest(buf, bodyLen, cnr, parent, rngLen, off, ln,
metaHdrLen, ruleIdxHdrLen, ruleIdxStr, partIdxHdrLen, partIdxStr, verifHdrSigCount)
metaHdrLen, ruleIdxHdrLen, ruleIdxStr, verifHdrSigCount)
if err != nil {
return nil, err
}
Expand All @@ -978,7 +975,7 @@ func (s *Server) makeGetECPartRangeRequest(remoteServerAPIVersion *protorefs.Ver
return mem.SliceBuffer(buf), nil
}

func (s *Server) writeGetECPartRequest(buf []byte, cnr cid.ID, parent oid.ID, metaHdrLen int, ruleIdxHdrLen int, ruleIdxHdr string, partIdxHdrLen int, partIdxHdr string, verifHdrSigCount int) (int, error) {
func (s *Server) writeGetECPartRequest(buf []byte, cnr cid.ID, parent oid.ID, metaHdrLen int, ruleIdxHdrLen int, ruleIdxHdr string, verifHdrSigCount int) (int, error) {
var originSig []byte
var err error
if verifHdrSigCount == 3 {
Expand Down Expand Up @@ -1017,7 +1014,6 @@ func (s *Server) writeGetECPartRequest(buf []byte, cnr cid.ID, parent oid.ID, me

off += copy(buf[off:], currentVersionResponseMetaHeader)
off += writeRequestMetaXHeader(buf[off:], ruleIdxHdrLen, iec.AttributeRuleIdx, ruleIdxHdr)
off += writeRequestMetaXHeader(buf[off:], partIdxHdrLen, iec.AttributePartIdx, partIdxHdr)

metaHdrSig, err := signECDSAWithSHA512(s.signer, buf[from:off])
if err != nil {
Expand All @@ -1031,7 +1027,7 @@ func (s *Server) writeGetECPartRequest(buf []byte, cnr cid.ID, parent oid.ID, me
}

func (s *Server) writeGetECPartRangeRequest(buf []byte, bodyLen int, cnr cid.ID, parent oid.ID, rngLen int, off uint64, ln uint64,
metaHdrLen int, ruleIdxHdrLen int, ruleIdxHdr string, partIdxHdrLen int, partIdxHdr string, verifHdrSigCount int) (int, error) {
metaHdrLen int, ruleIdxHdrLen int, ruleIdxHdr string, verifHdrSigCount int) (int, error) {
var originSig []byte
var err error
if verifHdrSigCount == 3 {
Expand Down Expand Up @@ -1101,7 +1097,6 @@ func (s *Server) writeGetECPartRangeRequest(buf []byte, bodyLen int, cnr cid.ID,

n += copy(buf[n:], currentVersionResponseMetaHeader)
n += writeRequestMetaXHeader(buf[n:], ruleIdxHdrLen, iec.AttributeRuleIdx, ruleIdxHdr)
n += writeRequestMetaXHeader(buf[n:], partIdxHdrLen, iec.AttributePartIdx, partIdxHdr)

metaHdrSig, err := signECDSAWithSHA512(s.signer, buf[from:n])
if err != nil {
Expand Down Expand Up @@ -1180,10 +1175,9 @@ func (s *Server) writeInitGetResponseBuffers(respStream grpc.ServerStream, id, s
return respStream.SendMsg(respBuf)
}

func calculateGetECPartRequestMetaHeaderLength(ruleIdxHdrLen, partIdxHdrLen int) int {
func calculateGetECPartRequestMetaHeaderLength(ruleIdxHdrLen int) int {
return len(currentVersionResponseMetaHeader) +
1 + protowire.SizeBytes(ruleIdxHdrLen) + // 1 for iprotobuf.TagBytes4
1 + protowire.SizeBytes(partIdxHdrLen) // 1 for iprotobuf.TagBytes4
1 + protowire.SizeBytes(ruleIdxHdrLen) // 1 for iprotobuf.TagBytes4
}

func calculateInitGetResponseFieldLength(idLen, sigLen, hdrLen int) int {
Expand Down
8 changes: 3 additions & 5 deletions pkg/services/object/get/ec.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1533,7 +1533,7 @@ func (s *Service) streamFirstECPart(ctx context.Context, transport GetECRequestT
continue
}

copiedHdr, parentPldLen, partPldLen, copiedPartPld, err = transport.CopyRemoteECPartParentHeaderAndPayload(ctx, conn, partInfo)
copiedHdr, parentPldLen, partPldLen, copiedPartPld, err = transport.CopyRemoteECPartParentHeaderAndPayload(ctx, conn, ruleIdx)
}
if err != nil {
return false, 0, 0, 0, err
Expand Down Expand Up @@ -1562,7 +1562,7 @@ func (s *Service) streamFirstECPart(ctx context.Context, transport GetECRequestT
continue
}

copiedFromNode, err = transport.CopyRemoteECPartRange(ctx, conn, partInfo, copiedPartPld, partPldLen-copiedPartPld, copiedPartPld == 0, nil)
copiedFromNode, err = transport.CopyRemoteECPartRange(ctx, conn, ruleIdx, copiedPartPld, partPldLen-copiedPartPld, copiedPartPld == 0, nil)
}
if err != nil {
return false, 0, 0, 0, err
Expand Down Expand Up @@ -1600,7 +1600,7 @@ func (s *Service) streamECPartRangePrefix(ctx context.Context, transport GetECRe
continue
}

copiedLenNode, err = transport.CopyRemoteECPartRange(ctx, conn, partInfo, copiedLen, ln-copiedLen, full && copiedLen == 0, controlCh)
copiedLenNode, err = transport.CopyRemoteECPartRange(ctx, conn, partInfo.RuleIndex, copiedLen, ln-copiedLen, full && copiedLen == 0, controlCh)
}

if err != nil {
Expand Down Expand Up @@ -1693,8 +1693,6 @@ func checkECPartInfoGetRequest(neofs NeoFSNetwork, prm Prm) (iec.PartInfo, error

res.RuleIndex = -1
return res, nil
} else if partIdxStr == "" && neofs == nil {
return res, fmt.Errorf("request must have %s header for EC objects", iec.AttributePartIdx)
}

var (
Expand Down
9 changes: 2 additions & 7 deletions pkg/services/object/get/get.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,8 @@ import (

// Get serves a request to get an object by address, and returns Streamer instance.
func (s *Service) Get(ctx context.Context, prm Prm) error {
var neofsNet NeoFSNetwork
// range requests do not support fetching additional info about EC parts
if !prm.payloadRange.IsSet() {
neofsNet = s.neoFSNet
}

pi, err := checkECPartInfoGetRequest(neofsNet, prm)
// TODO: upd API
pi, err := checkECPartInfoGetRequest(s.neoFSNet, prm)

@cthulhu-rider cthulhu-rider Aug 31, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is wrong for any case except the best one. For example, if SN#1 stores part#2, payload will be streamed 0-2-...

so, the only thing we can optimize for now is header+payload GET which is used to request part#0 only. For optimal case, optimization won't be visible

if err != nil {
// TODO: track https://github.com/nspcc-dev/neofs-api/issues/269.
return fmt.Errorf("invalid request: %w", err)
Expand Down
12 changes: 6 additions & 6 deletions pkg/services/object/get/service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ var ErrUnavailableNode = errors.New("unavailable node")
// GetECRequestTransport is used to serve GET requests for EC objects.
type GetECRequestTransport interface {
// CopyRemoteECPartParentHeaderAndPayload requests originally requested object's
// EC part identified by partInfo from remote storage node using conn to it. If
// EC part identified by policy rule index from remote storage node using conn to it. If
// succeeded, CopyRemoteECPartParentHeaderAndPayload sends parent header and
// part payload to the client, and returns:
// - flag whether parent header was copied or not;
Expand All @@ -93,11 +93,12 @@ type GetECRequestTransport interface {
// Otherwise, no error is returned. Copying can be incomplete in this case.
//
// CopyRemoteECPartParentHeaderAndPayload is never called concurrently.
CopyRemoteECPartParentHeaderAndPayload(ctx context.Context, conn clientcore.MultiAddressClient, partInfo iec.PartInfo) (bool, uint64, uint64, uint64, error)
CopyRemoteECPartParentHeaderAndPayload(ctx context.Context, conn clientcore.MultiAddressClient, ruleIdx int) (bool, uint64, uint64, uint64, error)
// TODO: same for local case?
// CopyLocalECPartParentHeaderAndPayload works like CopyRemoteECPartParentHeaderAndPayload but locally.
CopyLocalECPartParentHeaderAndPayload(ctx context.Context, storage *engine.StorageEngine, partInfo iec.PartInfo) (bool, uint64, uint64, uint64, error)
// CopyRemoteECPartRange requests specified payload range of originally
// requested object's EC part identified by partInfo pair from remote storage
// requested object's EC part identified by policy rule index from remote storage
// node using conn to it. If succeeded, CopyRemoteECPartRange sends with payload
// range to the client, and returns number of bytes copied.
//
Expand All @@ -115,9 +116,8 @@ type GetECRequestTransport interface {
// from it. On true, CopyRemoteECPartRange returns [ErrAborted] instantly
// without copying. Otherwise, i.e. if on false or close, copying starts.
//
// CopyRemoteECPartRange can be called concurrently for different partInfo, but
// never for the same one.
CopyRemoteECPartRange(ctx context.Context, conn clientcore.MultiAddressClient, partInfo iec.PartInfo, off, ln uint64, full bool, controlCh <-chan bool) (uint64, error)
// CopyRemoteECPartRange is never called concurrently for the same ruleIdx.
CopyRemoteECPartRange(ctx context.Context, conn clientcore.MultiAddressClient, ruleIdx int, off, ln uint64, full bool, controlCh <-chan bool) (uint64, error)
// CopyLocalECPartRange works like CopyRemoteECPartRange but locally.
CopyLocalECPartRange(ctx context.Context, storage *engine.StorageEngine, partInfo iec.PartInfo, off, ln uint64, ch <-chan bool) (uint64, error)
}
Expand Down
Loading