diff --git a/CHANGELOG.md b/CHANGELOG.md index ef1252f1f9..338f5360a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,7 @@ Changelog for NeoFS Node - Store in metabase associated object ID in bytes instead of Base58 (#3971) - Optimized local RANGE request execution (#3967) - GET now supports payload ranges (#3991) +- Optimized EC GET request execution (#3996) ### Removed - `policer.max_workers` configuration (#3920) diff --git a/cmd/neofs-node/object.go b/cmd/neofs-node/object.go index 8e70c94131..3e6aaef050 100644 --- a/cmd/neofs-node/object.go +++ b/cmd/neofs-node/object.go @@ -345,7 +345,7 @@ func initObjectService(c *cfg) { putSvc: sPut, keys: keyStorage, } - server := objectService.New(objSvc, mNumber, c.cfgObject.pool.search, fsChain, storage, c.metaService, c.key.PrivateKey, c.metricsCollector, aclChecker, aclSvc, coreConstructor) + server := objectService.New(objSvc, mNumber, c.cfgObject.pool.search, fsChain, storage, c.metaService, c.key.PrivateKey, c.metricsCollector, aclChecker, aclSvc, coreConstructor, c.log) os.server = server svcDesc := protoobject.ObjectService_ServiceDesc diff --git a/internal/object/wire.go b/internal/object/wire.go index cae2ce1ff2..a022432ace 100644 --- a/internal/object/wire.go +++ b/internal/object/wire.go @@ -196,7 +196,23 @@ func GetParentNonPayloadFieldBounds(buf []byte) (iprotobuf.FieldBounds, iprotobu return idf, sigf, hdrf, nil } - splitf, err := iprotobuf.GetLENFieldBounds(buf[rootHdrf.ValueFrom:rootHdrf.To], protoobject.FieldHeaderSplit) + return getParentNonPayloadFieldBounds(buf, rootHdrf.ValueFrom, rootHdrf.To) +} + +// GetParentNonPayloadFieldBoundsHeader is GetParentNonPayloadFieldBounds +// analogue accepting header. +func GetParentNonPayloadFieldBoundsHeader(buf []byte) (iprotobuf.FieldBounds, iprotobuf.FieldBounds, iprotobuf.FieldBounds, error) { + if len(buf) == 0 { + return iprotobuf.FieldBounds{}, iprotobuf.FieldBounds{}, iprotobuf.FieldBounds{}, errEmptyData + } + + return getParentNonPayloadFieldBounds(buf, 0, len(buf)) +} + +func getParentNonPayloadFieldBounds(buf []byte, hdrFrom, hdrTo int) (iprotobuf.FieldBounds, iprotobuf.FieldBounds, iprotobuf.FieldBounds, error) { + var idf, sigf, hdrf iprotobuf.FieldBounds + + splitf, err := iprotobuf.GetLENFieldBounds(buf[hdrFrom:hdrTo], protoobject.FieldHeaderSplit) if err != nil { return idf, sigf, hdrf, err } @@ -205,8 +221,9 @@ func GetParentNonPayloadFieldBounds(buf []byte) (iprotobuf.FieldBounds, iprotobu return idf, sigf, hdrf, nil } - buf = buf[:rootHdrf.ValueFrom+splitf.To] - off := rootHdrf.ValueFrom + splitf.ValueFrom + buf = buf[:hdrFrom+splitf.To] + off := hdrFrom + splitf.ValueFrom + var prevNum protowire.Number loop: for { @@ -270,7 +287,7 @@ func GetPayloadLengthAndFieldOffset(buf []byte) (uint64, int, error) { var pldLen uint64 if !hf.IsMissing() { - pldLen, err = iprotobuf.GetUint64Field(buf[hf.ValueFrom:hf.To], protoobject.FieldHeaderPayloadLength) + pldLen, err = GetPayloadLengthHeader(buf[hf.ValueFrom:hf.To]) if err != nil { return 0, 0, fmt.Errorf("seek payload length field in header: %w", err) } @@ -289,3 +306,19 @@ func GetPayloadLengthAndFieldOffset(buf []byte) (uint64, int, error) { return pldLen, off + tagLn, nil } + +// GetPayloadLengthHeader reads payload length header. If field is missing, no +// error is returned. +// +// Message should have ascending field order, otherwise error returns. +func GetPayloadLengthHeader(buf []byte) (uint64, error) { + return iprotobuf.GetUint64Field(buf, protoobject.FieldHeaderPayloadLength) +} + +// GetTypeHeader reads object type header. If field is missing, no error is +// returned. +// +// Message should have ascending field order, otherwise error returns. +func GetTypeHeader(buf []byte) (object.Type, error) { + return iprotobuf.GetEnumField[object.Type](buf, protoobject.FieldHeaderObjectType) +} diff --git a/internal/protobuf/api.go b/internal/protobuf/api.go index 94d99e0971..36ffaf8dc9 100644 --- a/internal/protobuf/api.go +++ b/internal/protobuf/api.go @@ -9,6 +9,7 @@ import ( "github.com/nspcc-dev/neo-go/pkg/crypto/hash" "github.com/nspcc-dev/neo-go/pkg/encoding/address" "github.com/nspcc-dev/neofs-sdk-go/checksum" + cid "github.com/nspcc-dev/neofs-sdk-go/container/id" neofscrypto "github.com/nspcc-dev/neofs-sdk-go/crypto" "github.com/nspcc-dev/neofs-sdk-go/object" oid "github.com/nspcc-dev/neofs-sdk-go/object/id" @@ -21,7 +22,10 @@ import ( // Fixed message lengths. const ( - ObjectIDLength = 1 + 1 + oid.Size + ObjectIDLength = 1 + 1 + oid.Size + ContainerIDLength = 1 + 1 + cid.Size + ObjectAddressLength = 1 + 1 + ObjectIDLength + + 1 + 1 + ContainerIDLength ) // Message length limits. diff --git a/internal/protobuf/api_test.go b/internal/protobuf/api_test.go new file mode 100644 index 0000000000..0d5939bce1 --- /dev/null +++ b/internal/protobuf/api_test.go @@ -0,0 +1,26 @@ +package protobuf_test + +import ( + "testing" + + iprotobuf "github.com/nspcc-dev/neofs-node/internal/protobuf" + cidtest "github.com/nspcc-dev/neofs-sdk-go/container/id/test" + oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test" + "github.com/stretchr/testify/require" +) + +func TestMessageLengths(t *testing.T) { + for _, tc := range []struct { + name string + msg interface{ MarshaledSize() int } + cnst int + }{ + {name: "object ID", msg: oidtest.ID().ProtoMessage(), cnst: iprotobuf.ObjectIDLength}, + {name: "container ID", msg: cidtest.ID().ProtoMessage(), cnst: iprotobuf.ContainerIDLength}, + {name: "object address", msg: oidtest.Address().ProtoMessage(), cnst: iprotobuf.ObjectAddressLength}, + } { + t.Run(tc.name, func(t *testing.T) { + require.EqualValues(t, tc.msg.MarshaledSize(), tc.cnst) + }) + } +} diff --git a/internal/protobuf/seekers.go b/internal/protobuf/seekers.go index 8333ecb805..fddbe4b069 100644 --- a/internal/protobuf/seekers.go +++ b/internal/protobuf/seekers.go @@ -96,3 +96,23 @@ func GetUint64Field(buf []byte, num protowire.Number) (uint64, error) { u, _, err := ParseUint64Field(buf[off+tagLn:], num, typ) return u, err } + +// GetEnumField seeks enum field in buf by number and parses it. If field is +// missing, no error is returned. +// +// Message should have ascending field order, otherwise error returns. +// +// If there is an error, its text contains num. +func GetEnumField[T ~int32](buf []byte, num protowire.Number) (T, error) { + off, tagLn, typ, err := SeekFieldByNumber(buf, num) + if err != nil { + return 0, err + } + + if off < 0 { + return 0, nil + } + + u, _, err := ParseEnumField[T](buf[off+tagLn:], num, typ) + return u, err +} diff --git a/internal/protobuf/tags.go b/internal/protobuf/tags.go index 5705c7b8b5..036ec0b055 100644 --- a/internal/protobuf/tags.go +++ b/internal/protobuf/tags.go @@ -8,6 +8,9 @@ const ( TagBytes4 = 34 TagBytes5 = 42 TagBytes6 = 50 + TagBytes7 = 58 + TagBytes8 = 66 + TagBytes9 = 74 ) // One-byte tags for VARINT fields. diff --git a/internal/protobuf/tags_test.go b/internal/protobuf/tags_test.go index 09f0c25db6..cec20fbe12 100644 --- a/internal/protobuf/tags_test.go +++ b/internal/protobuf/tags_test.go @@ -20,6 +20,9 @@ func TestTags(t *testing.T) { {tag: iprotobuf.TagBytes4, num: 4}, {tag: iprotobuf.TagBytes5, num: 5}, {tag: iprotobuf.TagBytes6, num: 6}, + {tag: iprotobuf.TagBytes7, num: 7}, + {tag: iprotobuf.TagBytes8, num: 8}, + {tag: iprotobuf.TagBytes9, num: 9}, } { require.EqualValues(t, protowire.EncodeTag(protowire.Number(tc.num), protowire.BytesType), tc.tag) diff --git a/pkg/core/client/client.go b/pkg/core/client/client.go index eb44a377c0..01785963a5 100644 --- a/pkg/core/client/client.go +++ b/pkg/core/client/client.go @@ -2,6 +2,7 @@ package client import ( "context" + "errors" "io" "github.com/nspcc-dev/neofs-sdk-go/client" @@ -29,6 +30,14 @@ type Client interface { AnnounceIntermediateTrust(ctx context.Context, epoch uint64, trust reputationSDK.PeerToPeerTrust, prm client.PrmAnnounceIntermediateTrust) error } +// ErrSkipConnection is returned to skip connection. +var ErrSkipConnection = errors.New("connection skipped") + +// ErrAllConnectionsSkipped allows to check whether +// [MultiAddressClient.ForAnyGRPCConn] error is returns because all connections +// are unavailable or skipped. +var ErrAllConnectionsSkipped = errors.New("all connections skipped") + // MultiAddressClient is an interface of the // Client that supports multihost work. type MultiAddressClient interface { @@ -36,5 +45,9 @@ type MultiAddressClient interface { // ForAnyGRPCConn executes op over gRPC connections to given multi-address // endpoint-by-endpoint until success. - ForAnyGRPCConn(context.Context, func(context.Context, *grpc.ClientConn) error) error + // + // If next endpoint is unavailable or f returns [ErrSkipConnection] for it, + // ForAnyGRPCConn continues. If this happens on all endpoints, ForAnyGRPCConn + // returns [ErrAllConnectionsSkipped]. + ForAnyGRPCConn(ctx context.Context, f func(context.Context, *grpc.ClientConn) error) error } diff --git a/pkg/network/cache/clients.go b/pkg/network/cache/clients.go index f17afdcac0..06e961cd28 100644 --- a/pkg/network/cache/clients.go +++ b/pkg/network/cache/clients.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/hex" + "errors" "fmt" "io" "iter" @@ -279,20 +280,26 @@ func (x *connections) all(f func(ma string, c *client.Client) bool) { } func (x *connections) forAny(ctx context.Context, f func(context.Context, *client.Client) error) error { - var firstErr error + var firstUnavailableErr error for ma, c := range x.all { err := f(ctx, c) if err == nil { return nil } - if !isTempError(err) { + if errors.Is(err, clientcore.ErrSkipConnection) { + continue + } + if !isUnavailableError(err) { return newEndpointError(ma, err) } - if firstErr == nil { - firstErr = newEndpointError(ma, err) + if firstUnavailableErr == nil { + firstUnavailableErr = newEndpointError(ma, err) } } - return newMultiEndpointError(x.nodeID, firstErr) + if firstUnavailableErr == nil { + return clientcore.ErrAllConnectionsSkipped + } + return newMultiEndpointError(x.nodeID, firstUnavailableErr) } func (x *connections) ForAnyGRPCConn(ctx context.Context, f func(context.Context, *grpc.ClientConn) error) error { @@ -318,7 +325,7 @@ func (x *connections) ReplicateObject(ctx context.Context, id oid.ID, src io.Rea if err == nil { return sig, nil } - if !isTempError(err) { + if !isUnavailableError(err) { return nil, newEndpointError(ma, err) } if _, errSeek := src.Seek(0, io.SeekStart); errSeek != nil { @@ -401,7 +408,7 @@ func (x *connections) SearchObjects(ctx context.Context, cnr cid.ID, fs object.S }) } -func isTempError(err error) bool { +func isUnavailableError(err error) bool { st, ok := status.FromError(err) return ok && st.Code() == codes.Unavailable } @@ -410,6 +417,32 @@ func newEndpointError(addr string, err error) error { return fmt.Errorf("%s: %w", addr, err) } +type multiEndpointError struct { + nodeID string + first error +} + +// Error implements built-in [error]. +func (x multiEndpointError) Error() string { + return fmt.Sprintf("all %s endpoints failed, first error: %s", x.nodeID, x.first) +} + +// Unwrap implements interface for [errors] package. +func (x multiEndpointError) Unwrap() error { + return x.first +} + +// Is implements interface for [errors.Is]. +func (x multiEndpointError) Is(target error) bool { + if errors.Is(target, clientcore.ErrAllConnectionsSkipped) { + return true + } + return errors.Is(x.first, target) +} + func newMultiEndpointError(nodeID string, first error) error { - return fmt.Errorf("all %s endpoints failed, first error: %w", nodeID, first) + return multiEndpointError{ + nodeID: nodeID, + first: first, + } } diff --git a/pkg/network/cache/clients_internal_test.go b/pkg/network/cache/clients_internal_test.go new file mode 100644 index 0000000000..f881ab1520 --- /dev/null +++ b/pkg/network/cache/clients_internal_test.go @@ -0,0 +1,18 @@ +package cache + +import ( + "errors" + "testing" + + coreclient "github.com/nspcc-dev/neofs-node/pkg/core/client" + "github.com/stretchr/testify/require" +) + +func TestMultiEndpointError(t *testing.T) { + firstErr := errors.New("test err") + + err := newMultiEndpointError("NODE_X", firstErr) + require.EqualError(t, err, "all NODE_X endpoints failed, first error: test err") + require.ErrorIs(t, err, firstErr) + require.ErrorIs(t, err, coreclient.ErrAllConnectionsSkipped) +} diff --git a/pkg/services/object/get.go b/pkg/services/object/get.go index eb6a174f7f..1aa1b10801 100644 --- a/pkg/services/object/get.go +++ b/pkg/services/object/get.go @@ -4,27 +4,82 @@ import ( "bytes" "context" "crypto/sha256" + "encoding/binary" "errors" "fmt" "io" + "strconv" + "sync" + iec "github.com/nspcc-dev/neofs-node/internal/ec" + iobject "github.com/nspcc-dev/neofs-node/internal/object" iprotobuf "github.com/nspcc-dev/neofs-node/internal/protobuf" "github.com/nspcc-dev/neofs-node/internal/protobuf/protoscan" + coreclient "github.com/nspcc-dev/neofs-node/pkg/core/client" + "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/engine" aclsvc "github.com/nspcc-dev/neofs-node/pkg/services/object/acl/v2" + getsvc "github.com/nspcc-dev/neofs-node/pkg/services/object/get" apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status" + cid "github.com/nspcc-dev/neofs-sdk-go/container/id" + neofsecdsa "github.com/nspcc-dev/neofs-sdk-go/crypto/ecdsa" + "github.com/nspcc-dev/neofs-sdk-go/object" + oid "github.com/nspcc-dev/neofs-sdk-go/object/id" protoobject "github.com/nspcc-dev/neofs-sdk-go/proto/object" protorefs "github.com/nspcc-dev/neofs-sdk-go/proto/refs" + protosession "github.com/nspcc-dev/neofs-sdk-go/proto/session" protostatus "github.com/nspcc-dev/neofs-sdk-go/proto/status" + "go.uber.org/zap" "google.golang.org/grpc" "google.golang.org/grpc/mem" "google.golang.org/protobuf/encoding/protowire" ) +var ( + getStreamDesc = &grpc.StreamDesc{ + StreamName: "Get", + ServerStreams: true, + ClientStreams: false, + } + getRangeStreamDesc = &grpc.StreamDesc{ + StreamName: "GetRange", + ServerStreams: true, + ClientStreams: false, + } +) + type getStreamProgress struct { headWas bool readPayload int } +func callServerStream(ctx context.Context, conn *grpc.ClientConn, method string, streamDesc *grpc.StreamDesc, request any) (grpc.ClientStream, error) { + stream, err := conn.NewStream(ctx, streamDesc, method, + grpc.StaticMethod(), + grpc.ForceCodecV2(iprotobuf.BufferedCodec{}), + ) + if err != nil { + return nil, fmt.Errorf("stream opening failed: %w", err) + } + + if err = stream.SendMsg(request); err != nil { + return nil, fmt.Errorf("send request: %w", err) + } + + if err = stream.CloseSend(); err != nil { + return nil, fmt.Errorf("close send: %w", err) + } + + return stream, nil +} + +func callGet(ctx context.Context, conn *grpc.ClientConn, request any) (grpc.ClientStream, error) { + return callServerStream(ctx, conn, protoobject.ObjectService_Get_FullMethodName, getStreamDesc, request) +} + +func callRange(ctx context.Context, conn *grpc.ClientConn, request any) (grpc.ClientStream, error) { + return callServerStream(ctx, conn, protoobject.ObjectService_GetRange_FullMethodName, getRangeStreamDesc, request) +} + // returns: // - nil on completed object transmission // - [object.SplitInfoError]/nil on split info response and unset/set raw flag in request @@ -32,18 +87,9 @@ type getStreamProgress struct { // - nil on other API statuses // - any other transport/protocol error otherwise func (x *getProxyContext) continueWithConn(ctx context.Context, conn *grpc.ClientConn) error { - stream, err := conn.NewStream(ctx, &protoobject.ObjectService_ServiceDesc.Streams[0], protoobject.ObjectService_Get_FullMethodName, - grpc.StaticMethod(), - grpc.ForceCodecV2(iprotobuf.BufferedCodec{}), - ) + stream, err := callGet(ctx, conn, x.req) if err != nil { - return fmt.Errorf("stream opening failed: %w", err) - } - if err = stream.SendMsg(x.req); err != nil { - return fmt.Errorf("send request: %w", err) - } - if err = stream.CloseSend(); err != nil { - return fmt.Errorf("close send: %w", err) + return err } var prog getStreamProgress @@ -84,7 +130,7 @@ func (x *getProxyContext) validateEOF(prog getStreamProgress) error { return nil } -func (x *getProxyContext) handleGetResponse(streamProg *getStreamProgress, respBuf mem.BufferSlice) (bool, bool, error) { +func handleResponseCodeAndBody(respBuf mem.BufferSlice) (uint32, iprotobuf.BuffersSlice, error) { var code uint32 var body iprotobuf.BuffersSlice @@ -107,6 +153,12 @@ func (x *getProxyContext) handleGetResponse(streamProg *getStreamProgress, respB } err := protoscan.ScanMessage(iprotobuf.NewBuffersSlice(respBuf), protoscan.ResponseScheme, opts) + + return code, body, err +} + +func (x *getProxyContext) handleGetResponse(streamProg *getStreamProgress, respBuf mem.BufferSlice) (bool, bool, error) { + code, body, err := handleResponseCodeAndBody(respBuf) if err != nil { return false, false, err } @@ -129,14 +181,14 @@ func (x *getProxyContext) handleGetResponse(streamProg *getStreamProgress, respB return false, sent, nil } -func (x *getProxyContext) handleResponseBody(streamProg *getStreamProgress, respBuf mem.BufferSlice, buffers iprotobuf.BuffersSlice) (bool, error) { +func handleGetResponseBodyOneof(headWas *bool, buffers iprotobuf.BuffersSlice) (protowire.Number, iprotobuf.BuffersSlice, error) { var oneofNum protowire.Number var oneofFld iprotobuf.BuffersSlice var opts protoscan.ScanMessageOptions opts.InterceptBytes = func(num protowire.Number, buffers iprotobuf.BuffersSlice) error { if num == protoobject.FieldGetResponseBodyChunk { - if !streamProg.headWas { + if !*headWas { return errors.New("incorrect message sequence") } oneofNum, oneofFld = num, buffers @@ -148,20 +200,50 @@ func (x *getProxyContext) handleResponseBody(streamProg *getStreamProgress, resp default: return protoscan.ErrContinue case protoobject.FieldGetResponseBodyInit: - if streamProg.headWas { + if *headWas { return errors.New("incorrect message sequence") } - streamProg.headWas = true + *headWas = true oneofNum, oneofFld = num, buffers return nil case protoobject.FieldGetResponseBodySplitInfo: - streamProg.headWas = true + *headWas = true oneofNum, oneofFld = num, buffers return nil } } err := protoscan.ScanMessage(buffers, protoscan.ObjectGetResponseBodyScheme, opts) + + return oneofNum, oneofFld, err +} + +func handleRangeResponseBodyOneof(buffers iprotobuf.BuffersSlice) (protowire.Number, iprotobuf.BuffersSlice, error) { + var oneofNum protowire.Number + var oneofFld iprotobuf.BuffersSlice + + var opts protoscan.ScanMessageOptions + opts.InterceptBytes = func(num protowire.Number, buffers iprotobuf.BuffersSlice) error { + if num == protoobject.FieldRangeResponseBodyChunk { + oneofNum, oneofFld = num, buffers + } + return nil + } + opts.InterceptNested = func(num protowire.Number, buffers iprotobuf.BuffersSlice) error { + if num != protoobject.FieldRangeResponseBodySplitInfo { + return protoscan.ErrContinue + } + oneofNum, oneofFld = num, buffers + return nil + } + + err := protoscan.ScanMessage(buffers, protoscan.ObjectGetRangeResponseBodyScheme, opts) + + return oneofNum, oneofFld, err +} + +func (x *getProxyContext) handleResponseBody(streamProg *getStreamProgress, respBuf mem.BufferSlice, buffers iprotobuf.BuffersSlice) (bool, error) { + oneofNum, oneofFld, err := handleGetResponseBodyOneof(&streamProg.headWas, buffers) if err != nil { return false, err } @@ -287,3 +369,1118 @@ func (x *getProxyContext) handleChunkResponse(streamProg *getStreamProgress, res func (x *getProxyContext) handleSplitInfo(respBuf mem.BufferSlice, buffers iprotobuf.BuffersSlice) (bool, error) { return handleSplitInfo(x.req.GetBody().GetRaw(), x.respStream.base, respBuf, buffers) } + +type preparedRangeRequest struct { + offset uint64 + length uint64 + buffer mem.Buffer +} + +type getECTransport struct { + server *Server + request *protoobject.GetRequest + signResponses bool + responseStream grpc.ServerStream + + getPartRequest mem.Buffer + getPartRequestInfo iec.PartInfo + + getPartRangeRequestsMtx sync.RWMutex + getPartRangeRequests map[iec.PartInfo]*preparedRangeRequest + + rangeViaGet bool +} + +// CopyLocalECPartParentHeaderAndPayload implements [getsvc.GetECRequestTransport]. +func (x *getECTransport) CopyLocalECPartParentHeaderAndPayload(_ context.Context, storage *engine.StorageEngine, partInfo iec.PartInfo) (bool, uint64, uint64, uint64, error) { + // TODO: handle request fields once and reuse + addr := x.request.GetBody().GetAddress() + cnr, err := cid.DecodeBytes(addr.GetContainerId().GetValue()) + if err != nil { + return false, 0, 0, 0, fmt.Errorf("invalid container ID in request: %w", err) + } + id, err := oid.DecodeBytes(addr.GetObjectId().GetValue()) + if err != nil { + return false, 0, 0, 0, fmt.Errorf("invalid object ID in request: %w", err) + } + + logError := func(msg string, err error) { + x.server.log.Warn(msg, zap.Stringer("container", cnr), zap.Stringer("parent", id), + zap.Int("ruleIdx", partInfo.RuleIndex), zap.Int("partIdx", partInfo.Index), zap.Error(err)) + } + + hdrMemBuf, buf := getBufferForHeadResponse() + + prefixLen, stream, err := storage.ReadECPart(cnr, id, partInfo, buf) + if err != nil { + var splitErr *object.SplitInfoError + if errors.Is(err, apistatus.ErrObjectAlreadyRemoved) || errors.As(err, &splitErr) { + return false, 0, 0, 0, err + } + logError("local storage failure (read EC part)", err) + return false, 0, 0, 0, nil + } + + defer stream.Close() + + _, _, partHdrf, err := iobject.GetNonPayloadFieldBounds(buf[:prefixLen]) + if err != nil { + return false, 0, 0, 0, fmt.Errorf("parse first %d bytes of object protobuf: %w", prefixLen, err) + } + + partHdrBuf := buf[partHdrf.ValueFrom:partHdrf.To] + + typ, err := iobject.GetTypeHeader(partHdrBuf) + if err != nil { + logError("invalid local object header (get type)", err) + return false, 0, 0, 0, nil + } + if typ == object.TypeLink { + return false, 0, 0, 0, getsvc.ErrLinker + } + + partPldLen, err := iobject.GetPayloadLengthHeader(partHdrBuf) + if err != nil { + logError("invalid local object header (get payload length)", err) + return false, 0, 0, 0, nil + } + + parentIDf, parentSigf, parentHdrf, err := iobject.GetParentNonPayloadFieldBoundsHeader(partHdrBuf) + if err != nil { + logError("invalid local object header (get parent fields)", err) + return false, 0, 0, 0, nil + } + + parentPldLen, err := iobject.GetPayloadLengthHeader(partHdrBuf[parentHdrf.ValueFrom:parentHdrf.To]) + if err != nil { + logError("invalid local object header (get payload length from parent header)", err) + return false, 0, 0, 0, nil + } + + var n int + + if !parentIDf.IsMissing() { + // ID has same tag in header and split header + n = copy(buf, partHdrBuf[parentIDf.From:parentIDf.To]) + } + + if !parentSigf.IsMissing() { + partHdrBuf[parentSigf.From] = iprotobuf.TagBytes2 + n += copy(buf[n:], partHdrBuf[parentSigf.From:parentSigf.To]) + } + + if !partHdrf.IsMissing() { + partHdrBuf[parentHdrf.From] = iprotobuf.TagBytes3 + n += copy(buf[n:], partHdrBuf[parentHdrf.From:parentHdrf.To]) + } + + err = x.server.copyGetStream(x.responseStream, hdrMemBuf, buf, prefixLen, n, stream, partHdrf.To, x.signResponses) + if err != nil { + var e copyReadError + if !errors.As(err, &e) { + return false, 0, 0, 0, err + } + logError("local storage stream failure (read EC part)", err) + return true, parentPldLen, partPldLen, uint64(e.written), nil + } + + return true, parentPldLen, partPldLen, partPldLen, nil +} + +// CopyLocalECPartRange implements [getsvc.GetECRequestTransport]. +func (x *getECTransport) CopyLocalECPartRange(ctx context.Context, storage *engine.StorageEngine, partInfo iec.PartInfo, off, ln uint64, controlCh <-chan struct{}) (uint64, error) { + // TODO: handle request fields once and reuse + addr := x.request.GetBody().GetAddress() + cnr, err := cid.DecodeBytes(addr.GetContainerId().GetValue()) + if err != nil { + return 0, fmt.Errorf("invalid container ID in request: %w", err) + } + id, err := oid.DecodeBytes(addr.GetObjectId().GetValue()) + if err != nil { + return 0, fmt.Errorf("invalid object ID in request: %w", err) + } + + logError := func(msg string, err error) { + x.server.log.Warn(msg, zap.Stringer("container", cnr), zap.Stringer("parent", id), + zap.Int("ruleIdx", partInfo.RuleIndex), zap.Int("partIdx", partInfo.Index), zap.Uint64("off", off), zap.Uint64("ln", ln), zap.Error(err)) + } + + hdrMemBuf, buf := getBufferForHeadResponse() + stream, err := storage.ReadECPartRange(cnr, id, partInfo, off, ln, buf) + hdrMemBuf.Free() + if err != nil { + if errors.Is(err, apistatus.ErrObjectAlreadyRemoved) { + return 0, err + } + logError("local storage failure (read EC part range)", err) + return 0, nil + } + + if stream == nil { + return 0, nil + } + + defer stream.Close() + + if controlCh != nil { + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-controlCh: + } + } + + err = x.server.copyRangeStream(x.responseStream, stream, x.signResponses, shiftPayloadChunkInGetResponseBuffer) + if err != nil { + var e copyReadError + if !errors.As(err, &e) { + return 0, err + } + logError("local storage stream failure (read EC part range)", err) + return uint64(e.written), nil + } + + return ln, nil +} + +func (x *getECTransport) initGetPartRequest(partInfo iec.PartInfo) error { + if x.getPartRequestInfo == partInfo && x.getPartRequest != nil { + return nil + } + + reqObj := x.request.GetBody().GetAddress() + cnr := reqObj.GetContainerId().GetValue() + parent := reqObj.GetObjectId().GetValue() + + reqMetaHdr := x.request.GetMetaHeader() + sessionToken := reqMetaHdr.GetSessionTokenV2() + sessionTokenV1 := reqMetaHdr.GetSessionToken() + + var err error + x.getPartRequest, err = x.server.makeGetECPartRequest(cnr, parent, partInfo, sessionToken, sessionTokenV1) + if err != nil { + return fmt.Errorf("make GET request: %w", err) + } + + x.getPartRequestInfo = partInfo + + return nil +} + +// CopyRemoteECPartParentHeaderAndPayload implements [getsvc.GetECRequestTransport]. +func (x *getECTransport) CopyRemoteECPartParentHeaderAndPayload(ctx context.Context, conn coreclient.MultiAddressClient, partInfo iec.PartInfo) (bool, uint64, uint64, uint64, error) { + var copiedHdr bool + var parentPldLen uint64 + var partPldLen uint64 + var copiedPartPld uint64 + + err := conn.ForAnyGRPCConn(ctx, func(ctx context.Context, conn *grpc.ClientConn) error { + if !copiedHdr { + if err := x.initGetPartRequest(partInfo); err != nil { + return err + } + + var err error + copiedHdr, parentPldLen, partPldLen, copiedPartPld, err = x.copyRemotePart(ctx, conn) + if err != nil { + return err + } + + if copiedHdr && copiedPartPld == partPldLen { + return nil + } + + return coreclient.ErrSkipConnection + } + + copiedFromNode, err := x.copyRemotePartRange(ctx, conn, partInfo, copiedPartPld, parentPldLen-copiedPartPld, nil) + if err != nil { + return err + } + + copiedPartPld += copiedFromNode + if copiedPartPld > partPldLen { + return fmt.Errorf("part payload overflow: full %d bytes, copied %d", partPldLen, copiedPartPld) + } + + if copiedPartPld == partPldLen { + return nil + } + + return coreclient.ErrSkipConnection + }) + if err != nil && !errors.Is(err, coreclient.ErrAllConnectionsSkipped) { + return false, 0, 0, 0, err + } + + return copiedHdr, parentPldLen, partPldLen, copiedPartPld, nil +} + +func (x *getECTransport) copyRemotePart(ctx context.Context, conn *grpc.ClientConn) (bool, uint64, uint64, uint64, error) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + stream, err := callGet(ctx, conn, x.getPartRequest) + if err != nil { + if errors.Is(err, ctx.Err()) { + return false, 0, 0, 0, err + } + // TODO: if error is due to incorrect request, error should be returned. How to catch this? + x.server.log.Warn("GET object API failure (call)", zap.String("node", conn.Target()), zap.Error(err)) + return false, 0, 0, 0, nil + } + + var copiedHdr bool + var parentPldLen uint64 + var partPldLen uint64 + var copiedPartPldLen uint64 + + var headWas bool + for { + var respBuf mem.BufferSlice + if err = stream.RecvMsg(&respBuf); err != nil { + if errors.Is(err, ctx.Err()) { + return false, 0, 0, 0, err + } + if !errors.Is(err, io.EOF) { + x.server.log.Warn("GET object API failure (receive message)", zap.String("node", conn.Target()), zap.Error(err)) + } + break + } + + code, body, err := handleResponseCodeAndBody(respBuf) + if err != nil { + respBuf.Free() + return false, 0, 0, 0, err + } + + if code == protostatus.ObjectNotFound { + respBuf.Free() + if headWas { + return false, 0, 0, 0, errors.New("received object not found status after header") + } + return false, 0, 0, 0, nil + } + + if code != protostatus.OK { + if err = x.responseStream.SendMsg(respBuf); err != nil { + return false, 0, 0, 0, fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) + } + return false, 0, 0, 0, getsvc.ErrResponded + } + + num, fld, err := handleGetResponseBodyOneof(&headWas, body) + if err != nil { + respBuf.Free() + return false, 0, 0, 0, err + } + + switch num { + default: + respBuf.Free() + return false, 0, 0, 0, errors.New("none of the supported oneof fields are specified") + case protoobject.FieldGetResponseBodyInit: + var parentID, parentSig, parentHdr iprotobuf.BuffersSlice + parentID, parentSig, parentHdr, parentPldLen, partPldLen, err = handleGetECPartResponseInit(fld) + if err != nil { + respBuf.Free() + return false, 0, 0, 0, err + } + + err = x.server.writeInitGetResponseBuffers(x.responseStream, parentID, parentSig, parentHdr, x.signResponses) + respBuf.Free() + if err != nil { + return false, 0, 0, 0, err + } + + copiedHdr = true + case protoobject.FieldGetResponseBodyChunk: + copiedPartPldLen += uint64(fld.Len()) + if copiedPartPldLen > partPldLen { + respBuf.Free() + return false, 0, 0, 0, fmt.Errorf("part payload overflow: full %d bytes, copied %d", partPldLen, copiedPartPldLen) + } + if copiedPartPldLen > parentPldLen { + respBuf.Free() + return false, 0, 0, 0, fmt.Errorf("parent payload overflow: full %d bytes, copied %d", parentPldLen, copiedPartPldLen) + } + + if err = x.responseStream.SendMsg(respBuf); err != nil { + return false, 0, 0, 0, fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) + } + case protoobject.FieldGetResponseBodySplitInfo: + err := _handleSplitInfo(fld, true) + respBuf.Free() + return false, 0, 0, 0, err + } + } + + return copiedHdr, parentPldLen, partPldLen, copiedPartPldLen, nil +} + +func (x *getECTransport) copyRemotePartRangeViaGet(ctx context.Context, conn *grpc.ClientConn, partInfo iec.PartInfo, off, ln uint64, controlCh <-chan struct{}) (uint64, error) { + if err := x.initGetPartRequest(partInfo); err != nil { + return 0, err + } + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + stream, err := callGet(ctx, conn, x.getPartRequest) + if err != nil { + if errors.Is(err, ctx.Err()) { + return 0, err + } + // TODO: if error is due to incorrect request, error should be returned. How to catch this? + x.server.log.Warn("GET object API failure (call)", zap.String("node", conn.Target()), zap.Error(err)) + return 0, nil + } + + if controlCh != nil { + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-controlCh: + } + } + + var copied uint64 + + var streamOff uint64 + var headWas bool + for { + var respBuf mem.BufferSlice + if err = stream.RecvMsg(&respBuf); err != nil { + if errors.Is(err, ctx.Err()) { + return 0, err + } + fin := errors.Is(err, io.EOF) + if fin && copied < ln { + return 0, fmt.Errorf("received less bytes than requested: expected %d, got %d", ln, copied) + } + if !fin { + x.server.log.Warn("GET object API failure (receive message)", zap.String("node", conn.Target()), zap.Error(err)) + } + break + } + + code, body, err := handleResponseCodeAndBody(respBuf) + if err != nil { + respBuf.Free() + return 0, err + } + + if code == protostatus.ObjectNotFound { + respBuf.Free() + if headWas { + return 0, errors.New("received object not found status after header") + } + return 0, nil + } + + if code != protostatus.OK { + if err = x.responseStream.SendMsg(respBuf); err != nil { + return 0, fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) + } + return 0, getsvc.ErrResponded + } + + num, fld, err := handleGetResponseBodyOneof(&headWas, body) + if err != nil { + respBuf.Free() + return 0, err + } + + switch num { + default: + respBuf.Free() + return 0, errors.New("none of the supported oneof fields are specified") + case protoobject.FieldGetResponseBodyInit: + _, _, _, _, full, err := handleGetECPartResponseInit(fld) + respBuf.Free() + if err != nil { + return 0, err + } + + if ln != 0 { + if off >= full || full-off < ln { + return 0, apistatus.ErrObjectOutOfRange + } + } + case protoobject.FieldGetResponseBodyChunk: + var chunkFrom, chunkTo int + chunkLen := fld.Len() + + if off > streamOff { + diff := off - streamOff + if uint64(chunkLen) <= diff { // already copied + streamOff += uint64(chunkLen) + break + } + chunkFrom = int(diff) + } + + left := ln - copied + + if uint64(chunkLen-chunkFrom) > left { + chunkTo = chunkFrom + int(left) + } else { + chunkTo = chunkLen + } + + if chunkFrom > 0 || chunkTo < chunkLen { + if chunkFrom > 0 { + _, ok := fld.MoveNext(chunkFrom) + if !ok { + respBuf.Free() + return 0, fmt.Errorf("%w while moving to offset=%d in chunk with length=%d", io.ErrUnexpectedEOF, chunkFrom, chunkLen) + } + } + if chunkTo < chunkLen { + var ok bool + fld, ok = fld.MoveNext(chunkTo - chunkFrom) + if !ok { + respBuf.Free() + return 0, fmt.Errorf("%w while moving to offset=%d in chunk with length=%d", io.ErrUnexpectedEOF, chunkTo-chunkFrom, chunkLen-chunkFrom) + } + } + resp, err := x.server.makeGetChunkResponse(fld, x.signResponses) + respBuf.Free() + if err != nil { + return 0, fmt.Errorf("make RANGE response: %w", err) + } + respBuf = mem.BufferSlice{resp} + } + + if err = x.responseStream.SendMsg(respBuf); err != nil { + return 0, fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) + } + + copied += uint64(chunkTo - chunkFrom) + if copied == ln { + return copied, nil + } + case protoobject.FieldGetResponseBodySplitInfo: + err := _handleSplitInfo(fld, true) + respBuf.Free() + return 0, err + } + } + + return copied, nil +} + +func handleGetECPartResponseInit(buffers iprotobuf.BuffersSlice) (iprotobuf.BuffersSlice, iprotobuf.BuffersSlice, iprotobuf.BuffersSlice, uint64, uint64, error) { + var parentID, parentSig, parentHdr iprotobuf.BuffersSlice + var parentPldLen uint64 + var partPldLen uint64 + + var opts protoscan.ScanMessageOptions + opts.InterceptNested = func(num protowire.Number, buffers iprotobuf.BuffersSlice) error { + if num != protoobject.FieldGetResponseBodyInitHeader { + return protoscan.ErrContinue + } + + var opts protoscan.ScanMessageOptions + opts.InterceptUint64 = func(num protowire.Number, u uint64) error { + if num == protoobject.FieldHeaderPayloadLength { + partPldLen = u + } + return nil + } + opts.InterceptEnum = func(num protowire.Number, e int32) error { + if num == protoobject.FieldHeaderObjectType && e == int32(protoobject.ObjectType_LINK) { + return getsvc.ErrLinker + } + return nil + } + // TODO: consider returning parent header from the server (by specific flag for example). + // Now only payload len is used from the part header. It can be calculated from rule and parent len. + opts.InterceptNested = func(num protowire.Number, buffers iprotobuf.BuffersSlice) error { + if num != protoobject.FieldHeaderSplit { + return protoscan.ErrContinue + } + + var opts protoscan.ScanMessageOptions + opts.InterceptNested = func(num protowire.Number, buffers iprotobuf.BuffersSlice) error { + switch num { //nolint:exhaustive + case protoobject.FieldHeaderSplitParentHeader: + var opts protoscan.ScanMessageOptions + opts.InterceptUint64 = func(num protowire.Number, u uint64) error { + if num == protoobject.FieldHeaderPayloadLength { + parentPldLen = u + } + return nil + } + if err := protoscan.ScanMessage(buffers, protoscan.ObjectHeaderScheme, opts); err != nil { + return err + } + parentHdr = buffers + return nil + case protoobject.FieldHeaderSplitParent: + parentID = buffers + case protoobject.FieldHeaderSplitParentSignature: + parentSig = buffers + } + return protoscan.ErrContinue + } + + return protoscan.ScanMessage(buffers, protoscan.ObjectSplitHeaderScheme, opts) + } + + return protoscan.ScanMessage(buffers, protoscan.ObjectHeaderScheme, opts) + } + + err := protoscan.ScanMessage(buffers, protoscan.ObjectGetResponseInitScheme, opts) + if err != nil { + return iprotobuf.BuffersSlice{}, iprotobuf.BuffersSlice{}, iprotobuf.BuffersSlice{}, 0, 0, err + } + + return parentID, parentSig, parentHdr, parentPldLen, partPldLen, nil +} + +func (x *getECTransport) CopyRemoteECPartRange(ctx context.Context, conn coreclient.MultiAddressClient, partInfo iec.PartInfo, off uint64, ln uint64, controlCh <-chan struct{}) (uint64, error) { + var copiedPld uint64 + + err := conn.ForAnyGRPCConn(ctx, func(ctx context.Context, conn *grpc.ClientConn) error { + copiedFromNode, err := x.copyRemotePartRange(ctx, conn, partInfo, off+copiedPld, ln-copiedPld, controlCh) + if err != nil { + return err + } + + copiedPld += copiedFromNode + if copiedPld > ln { + return fmt.Errorf("received %d bytes while %d requested", copiedPld, ln) + } + + if copiedPld == ln { + return nil + } + + return coreclient.ErrSkipConnection + }) + if err != nil && !errors.Is(err, coreclient.ErrAllConnectionsSkipped) { + return 0, err + } + + return copiedPld, nil +} + +func (x *getECTransport) copyRemotePartRange(ctx context.Context, conn *grpc.ClientConn, partInfo iec.PartInfo, off uint64, ln uint64, controlCh <-chan struct{}) (uint64, error) { + if x.rangeViaGet { + return x.copyRemotePartRangeViaGet(ctx, conn, partInfo, off, ln, controlCh) + } + + request, err := x.makeGetECPartRangeRequest(partInfo, off, ln) + if err != nil { + return 0, fmt.Errorf("make request: %w", err) + } + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + stream, err := callRange(ctx, conn, request) + if err != nil { + if errors.Is(err, ctx.Err()) { + return 0, err + } + x.server.log.Warn("RANGE object API failure (call)", zap.String("node", conn.Target()), zap.Error(err)) + return 0, nil + } + + if controlCh != nil { + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-controlCh: + } + } + + var copiedPld uint64 + + for first := true; ; first = false { + var respBuf mem.BufferSlice + if err = stream.RecvMsg(&respBuf); err != nil { + if errors.Is(err, ctx.Err()) { + return 0, err + } + fin := errors.Is(err, io.EOF) + if fin && copiedPld < ln { + return 0, fmt.Errorf("received less bytes than requested: expected %d, got %d", ln, copiedPld) + } + if !fin { + x.server.log.Warn("RANGE object API failure (receive message)", zap.String("node", conn.Target()), zap.Error(err)) + } + break + } + + code, body, err := handleResponseCodeAndBody(respBuf) + if err != nil { + respBuf.Free() + return 0, err + } + + if code == protostatus.ObjectNotFound { + respBuf.Free() + if !first { + return 0, errors.New("received object not found status in non-first message") + } + return 0, nil + } + + // TODO: track https://github.com/nspcc-dev/neofs-node/issues/3547 + if code == protostatus.ObjectAccessDenied { + respBuf.Free() + if !first { + return 0, errors.New("received access denied status in non-first message") + } + x.rangeViaGet = true + return x.copyRemotePartRangeViaGet(ctx, conn, partInfo, off, ln, controlCh) + } + + if code != protostatus.OK { + if err = x.responseStream.SendMsg(respBuf); err != nil { + return 0, fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) + } + return 0, getsvc.ErrResponded + } + + num, fld, err := handleRangeResponseBodyOneof(body) + if err != nil { + respBuf.Free() + return 0, err + } + + switch num { + default: + respBuf.Free() + return 0, errors.New("none of the supported oneof fields are specified") + case protoobject.FieldRangeResponseBodyChunk: + copiedPld += uint64(fld.Len()) + if copiedPld > ln { + respBuf.Free() + return 0, fmt.Errorf("received more bytes than requested: expected %d, got %d", ln, copiedPld) + } + + // In fact, the only difference is in the 'body.chunk' field. With https://github.com/nspcc-dev/neofs-api/pull/389 it's gonna be GET only. + getRespBuf, err := x.server.makeGetChunkResponse(fld, x.signResponses) + respBuf.Free() + if err != nil { + return 0, fmt.Errorf("make GET chunk response: %w", err) + } + + if err = x.responseStream.SendMsg(getRespBuf); err != nil { + return 0, fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) + } + case protoobject.FieldGetResponseBodySplitInfo: + respBuf.Free() + if !first { + return 0, errors.New("received split info in non-first message") + } + return 0, errors.New("unexpected split info status response") + } + } + + return copiedPld, nil +} + +func (s *Server) makeGetECPartRequest(cnr, parent []byte, partInfo iec.PartInfo, sessionToken *protosession.SessionTokenV2, sessionTokenV1 *protosession.SessionToken) (mem.Buffer, error) { + ruleIdxStr := strconv.Itoa(partInfo.RuleIndex) + partIdxStr := strconv.Itoa(partInfo.Index) + + ruleIdxHdrLen := calculateXHeaderLength(iec.AttributeRuleIdx, ruleIdxStr) + partIdxHdrLen := calculateXHeaderLength(iec.AttributePartIdx, partIdxStr) + + sessionTokenLen := sessionToken.MarshaledSize() + sessionTokenV1Len := sessionTokenV1.MarshaledSize() + + metaHdrLen := calculateGetECPartRequestMetaHeaderLength(ruleIdxHdrLen, partIdxHdrLen, sessionTokenLen, sessionTokenV1Len) + + reqLen := 1 + 1 + getByAddressRequestBodyLen + // first 1 for iprotobuf.TagBytes1 + 1 + protowire.SizeBytes(metaHdrLen) + // 1 for iprotobuf.TagBytes2 + 1 + 2 + requestVerificationHeaderECDSAWIthSHA512Len // 1 for iprotobuf.TagBytes3 + + // TODO: try with sync.Pool + buf := make([]byte, reqLen) + + n, err := s.writeGetECPartRequest(buf, cnr, parent, metaHdrLen, ruleIdxHdrLen, ruleIdxStr, partIdxHdrLen, partIdxStr, + sessionTokenLen, sessionToken, sessionTokenV1Len, sessionTokenV1) + if err != nil { + return nil, err + } + if n != len(buf) { + return nil, fmt.Errorf("got wrong request length: expected %d, got %d", n, len(buf)) + } + + return mem.SliceBuffer(buf), nil +} + +func (x *getECTransport) makeGetECPartRangeRequest(partInfo iec.PartInfo, off, ln uint64) (mem.Buffer, error) { + x.getPartRangeRequestsMtx.RLock() + req := x.getPartRangeRequests[partInfo] + x.getPartRangeRequestsMtx.RUnlock() + + if req == nil { + x.getPartRangeRequestsMtx.Lock() + + req = x.getPartRangeRequests[partInfo] + if req == nil { + if x.getPartRangeRequests == nil { + x.getPartRangeRequests = make(map[iec.PartInfo]*preparedRangeRequest, 1) + } + req = new(preparedRangeRequest) + x.getPartRangeRequests[partInfo] = req + } + + x.getPartRangeRequestsMtx.Unlock() + } + + if req.offset == off && req.length == ln && req.buffer != nil { + return req.buffer, nil + } + + reqObj := x.request.GetBody().GetAddress() + cnr := reqObj.GetContainerId().GetValue() + parent := reqObj.GetObjectId().GetValue() + + reqMetaHdr := x.request.GetMetaHeader() + sessionToken := reqMetaHdr.GetSessionTokenV2() + sessionTokenV1 := reqMetaHdr.GetSessionToken() + + reqBuf, err := x.server.makeGetECPartRangeRequest(cnr, parent, partInfo, off, ln, sessionToken, sessionTokenV1) + if err != nil { + // stream is closed by context cancellation + return nil, err + } + + req.buffer = reqBuf + req.offset = off + req.length = ln + + return reqBuf, nil +} + +func (s *Server) makeGetECPartRangeRequest(cnr, parent []byte, partInfo iec.PartInfo, off, ln uint64, sessionToken *protosession.SessionTokenV2, sessionTokenV1 *protosession.SessionToken) (mem.Buffer, error) { + ruleIdxStr := strconv.Itoa(partInfo.RuleIndex) + partIdxStr := strconv.Itoa(partInfo.Index) + + ruleIdxHdrLen := calculateXHeaderLength(iec.AttributeRuleIdx, ruleIdxStr) + partIdxHdrLen := calculateXHeaderLength(iec.AttributePartIdx, partIdxStr) + + sessionTokenLen := sessionToken.MarshaledSize() + sessionTokenV1Len := sessionTokenV1.MarshaledSize() + + metaHdrLen := calculateGetECPartRequestMetaHeaderLength(ruleIdxHdrLen, partIdxHdrLen, sessionTokenLen, sessionTokenV1Len) + + var rngLen int + if off != 0 { + rngLen = 1 + protowire.SizeVarint(off) // 1 for iprotobuf.TagVarint1 + } + if ln != 0 { + rngLen += 1 + protowire.SizeVarint(ln) // 1 for iprotobuf.TagVarint2 + } + + bodyLen := getByAddressRequestBodyLen + + if rngLen != 0 { + bodyLen += 1 + protowire.SizeBytes(rngLen) // 1 for iprotobuf.TagBytes2 + } + + reqLen := 1 + protowire.SizeBytes(bodyLen) + // 1 for iprotobuf.TagBytes1 + 1 + protowire.SizeBytes(metaHdrLen) + // 1 for iprotobuf.TagBytes2 + 1 + 2 + requestVerificationHeaderECDSAWIthSHA512Len // 1 for iprotobuf.TagBytes3 + + // TODO: try with sync.Pool + buf := make([]byte, reqLen) + + n, err := s.writeGetECPartRangeRequest(buf, bodyLen, cnr, parent, rngLen, off, ln, + metaHdrLen, ruleIdxHdrLen, ruleIdxStr, partIdxHdrLen, partIdxStr, + sessionTokenLen, sessionToken, sessionTokenV1Len, sessionTokenV1) + if err != nil { + return nil, err + } + if n != len(buf) { + return nil, fmt.Errorf("got wrong request length: expected %d, got %d", n, len(buf)) + } + + return mem.SliceBuffer(buf), nil +} + +func (s *Server) writeGetECPartRequest(buf []byte, cnr []byte, parent []byte, metaHdrLen int, ruleIdxHdrLen int, ruleIdxHdr string, partIdxHdrLen int, partIdxHdr string, + sessionTokenLen int, sessionToken *protosession.SessionTokenV2, sessionTokenV1Len int, sessionTokenV1 *protosession.SessionToken) (int, error) { + // TODO: can be calculated once and reused + originSig, err := neofsecdsa.Signer(s.signer).Sign(nil) + if err != nil { + return 0, fmt.Errorf("sign empty data: %w", err) + } + + // body + buf[0] = iprotobuf.TagBytes1 + buf[1] = getByAddressRequestBodyLen + buf[2] = iprotobuf.TagBytes1 // address + buf[3] = iprotobuf.ObjectAddressLength + buf[4] = iprotobuf.TagBytes1 // CID + buf[5] = iprotobuf.ContainerIDLength + buf[6] = iprotobuf.TagBytes1 // value + buf[7] = cid.Size + copy(buf[8:], cnr) + buf[40] = iprotobuf.TagBytes2 // OID + buf[41] = iprotobuf.ObjectIDLength + buf[42] = iprotobuf.TagBytes1 // value + buf[43] = oid.Size + copy(buf[44:], parent) + + bodySig, err := signECDSAWithSHA512(s.signer, buf[2:76]) + if err != nil { + return 0, fmt.Errorf("sign body: %w", err) + } + + // meta header + buf[76] = iprotobuf.TagBytes2 + off := 77 + binary.PutUvarint(buf[77:], uint64(metaHdrLen)) + + from := off + + off += copy(buf[off:], currentVersionResponseMetaHeader) + off += writeRequestMetaXHeader(buf[off:], ruleIdxHdrLen, iec.AttributeRuleIdx, ruleIdxHdr) + off += writeRequestMetaXHeader(buf[off:], partIdxHdrLen, iec.AttributePartIdx, partIdxHdr) + + if sessionTokenV1 != nil { + off += writeStablyMarshalledField(buf[off:], iprotobuf.TagBytes5, sessionTokenV1Len, sessionTokenV1) + } + + if sessionToken != nil { + off += writeStablyMarshalledField(buf[off:], iprotobuf.TagBytes9, sessionTokenLen, sessionToken) + } + + metaHdrSig, err := signECDSAWithSHA512(s.signer, buf[from:off]) + if err != nil { + return 0, fmt.Errorf("sign meta header: %w", err) + } + + // verification header + off += writeRequestVerificationHeader(buf[off:], s.pubKeyBytes, bodySig, metaHdrSig, originSig) + + return off, nil +} + +func (s *Server) writeGetECPartRangeRequest(buf []byte, bodyLen int, cnr []byte, parent []byte, rngLen int, off uint64, ln uint64, + metaHdrLen int, ruleIdxHdrLen int, ruleIdxHdr string, partIdxHdrLen int, partIdxHdr string, sessionTokenLen int, sessionToken *protosession.SessionTokenV2, + sessionTokenV1Len int, sessionTokenV1 *protosession.SessionToken) (int, error) { + // TODO: can be calculated once and reused + originSig, err := neofsecdsa.Signer(s.signer).Sign(nil) + if err != nil { + return 0, fmt.Errorf("sign empty data: %w", err) + } + + // body + buf[0] = iprotobuf.TagBytes1 + n := 1 + binary.PutUvarint(buf[1:], uint64(bodyLen)) + from := n + buf[n] = iprotobuf.TagBytes1 // address + n++ + buf[n] = iprotobuf.ObjectAddressLength + n++ + buf[n] = iprotobuf.TagBytes1 // CID + n++ + buf[n] = iprotobuf.ContainerIDLength + n++ + buf[n] = iprotobuf.TagBytes1 // value + n++ + buf[n] = cid.Size + n++ + n += copy(buf[n:], cnr) + buf[n] = iprotobuf.TagBytes2 // OID + n++ + buf[n] = iprotobuf.ObjectIDLength + n++ + buf[n] = iprotobuf.TagBytes1 // value + n++ + buf[n] = oid.Size + n++ + n += copy(buf[n:], parent) + if rngLen != 0 { + buf[n] = iprotobuf.TagBytes2 // range + n++ + n += binary.PutUvarint(buf[n:], uint64(rngLen)) + if off != 0 { + buf[n] = iprotobuf.TagVarint1 + n++ + n += binary.PutUvarint(buf[n:], off) + } + if ln != 0 { + buf[n] = iprotobuf.TagVarint2 + n++ + n += binary.PutUvarint(buf[n:], ln) + } + } + + bodySig, err := signECDSAWithSHA512(s.signer, buf[from:n]) + if err != nil { + return 0, fmt.Errorf("sign body: %w", err) + } + + // meta header + buf[n] = iprotobuf.TagBytes2 + n++ + n += binary.PutUvarint(buf[n:], uint64(metaHdrLen)) + + from = n + + n += copy(buf[n:], currentVersionResponseMetaHeader) + n += writeRequestMetaXHeader(buf[n:], ruleIdxHdrLen, iec.AttributeRuleIdx, ruleIdxHdr) + n += writeRequestMetaXHeader(buf[n:], partIdxHdrLen, iec.AttributePartIdx, partIdxHdr) + + if sessionTokenV1 != nil { + n += writeStablyMarshalledField(buf[n:], iprotobuf.TagBytes5, sessionTokenV1Len, sessionTokenV1) + } + + if sessionToken != nil { + n += writeStablyMarshalledField(buf[n:], iprotobuf.TagBytes9, sessionTokenLen, sessionToken) + } + + metaHdrSig, err := signECDSAWithSHA512(s.signer, buf[from:n]) + if err != nil { + return 0, fmt.Errorf("sign meta header: %w", err) + } + + // verification header + n += writeRequestVerificationHeader(buf[n:], s.pubKeyBytes, bodySig, metaHdrSig, originSig) + + return n, nil +} + +func (s *Server) writeInitGetResponseBuffers(respStream grpc.ServerStream, id, sig, hdr iprotobuf.BuffersSlice, signResponse bool) error { + idLen := id.Len() + sigLen := sig.Len() + hdrLen := hdr.Len() + + initFldLen := calculateInitGetResponseFieldLength(idLen, sigLen, hdrLen) + + bodyLen := 1 + protowire.SizeBytes(initFldLen) // 1 for iprotobuf.TagBytes1 + + respLen := 1 + protowire.SizeBytes(bodyLen) // 1 for iprotobuf.TagBytes1 + + if signResponse { + respLen += 1 + protowire.SizeBytes(requestVerificationHeaderECDSAWIthSHA512Len) // 1 for iprotobuf.TagBytes3 + } + + var respBuf mem.Buffer + var buf mem.SliceBuffer + if respLen <= headResponseBufferLen { + hb, _ := getBufferForHeadResponse() + respBuf = hb + buf = hb.SliceBuffer + } else { + buf = make(mem.SliceBuffer, respLen) + respBuf = buf + } + + // body + buf[0] = iprotobuf.TagBytes1 + off := 1 + binary.PutUvarint(buf[1:], uint64(bodyLen)) + bodyFrom := off + // init + buf[off] = iprotobuf.TagBytes1 + off++ + off += binary.PutUvarint(buf[off:], uint64(initFldLen)) + // id + buf[off] = iprotobuf.TagBytes1 + off++ + off += binary.PutUvarint(buf[off:], uint64(idLen)) + off += id.CopyTo(buf[off:]) + // signature + buf[off] = iprotobuf.TagBytes2 + off++ + off += binary.PutUvarint(buf[off:], uint64(sigLen)) + off += sig.CopyTo(buf[off:]) + // header + buf[off] = iprotobuf.TagBytes3 + off++ + off += binary.PutUvarint(buf[off:], uint64(hdrLen)) + off += hdr.CopyTo(buf[off:]) + + if signResponse { + n, err := s.signResponse(buf[off:], buf[bodyFrom:off], nil) + if err != nil { + respBuf.Free() + return fmt.Errorf("sign response: %w", err) + } + off += n + } + + if respLen <= headResponseBufferLen { + respBuf.(*iprotobuf.MemBuffer).SetBounds(0, off) + } + + return respStream.SendMsg(respBuf) +} + +func (s *Server) makeGetChunkResponse(chunk iprotobuf.BuffersSlice, sign bool) (mem.Buffer, error) { + chunkLen := chunk.Len() + + bodyLen := 1 + protowire.SizeBytes(chunkLen) // 1 for iprotobuf.TagBytes1 + + respLen := 1 + protowire.SizeBytes(bodyLen) // 1 for iprotobuf.TagBytes1 + + if sign { + respLen += 1 + protowire.SizeBytes(requestVerificationHeaderECDSAWIthSHA512Len) // 1 for iprotobuf.TagBytes3 + } + + var respBuf mem.Buffer + var buf mem.SliceBuffer + if respLen <= maxGetResponseChunkLen { + hb, _ := getBufferForChunkGetResponse() + respBuf = hb + buf = hb.SliceBuffer + } else { + buf = make(mem.SliceBuffer, respLen) + respBuf = buf + } + + // body + buf[0] = iprotobuf.TagBytes1 + off := 1 + binary.PutUvarint(buf[1:], uint64(bodyLen)) + bodyFrom := off + // chunk + buf[off] = iprotobuf.TagBytes2 + off++ + off += binary.PutUvarint(buf[off:], uint64(chunkLen)) + off += chunk.CopyTo(buf[off:]) + + if sign { + n, err := s.signResponse(buf[off:], buf[bodyFrom:off], nil) + if err != nil { + return nil, fmt.Errorf("sign response: %w", err) + } + off += n + } + + if respLen <= maxGetResponseChunkLen { + respBuf.(*iprotobuf.MemBuffer).SetBounds(0, off) + } + + return respBuf, nil +} + +func calculateGetECPartRequestMetaHeaderLength(ruleIdxHdrLen, partIdxHdrLen, sessionTokenLen, sessionTokenV1Len int) int { + metaHdrLen := len(currentVersionResponseMetaHeader) + + 1 + protowire.SizeBytes(ruleIdxHdrLen) + // 1 for iprotobuf.TagBytes4 + 1 + protowire.SizeBytes(partIdxHdrLen) // 1 for iprotobuf.TagBytes4 + + if sessionTokenLen > 0 { + metaHdrLen += 1 + protowire.SizeBytes(sessionTokenLen) // 1 for iprotobuf.TagBytes9 + } + + if sessionTokenV1Len > 0 { + metaHdrLen += 1 + protowire.SizeBytes(sessionTokenV1Len) // 1 for iprotobuf.TagBytes5 + } + + return metaHdrLen +} + +func calculateInitGetResponseFieldLength(idLen, sigLen, hdrLen int) int { + return 1 + protowire.SizeBytes(idLen) + // 1 for iprotobuf.TagBytes1 + 1 + protowire.SizeBytes(sigLen) + // 1 for iprotobuf.TagBytes2 + 1 + protowire.SizeBytes(hdrLen) // 1 for iprotobuf.TagBytes3 +} diff --git a/pkg/services/object/get/ec.go b/pkg/services/object/get/ec.go index 02e6d4b96f..904368d356 100644 --- a/pkg/services/object/get/ec.go +++ b/pkg/services/object/get/ec.go @@ -15,6 +15,7 @@ import ( iec "github.com/nspcc-dev/neofs-node/internal/ec" islices "github.com/nspcc-dev/neofs-node/internal/slices" + "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/engine" "github.com/nspcc-dev/neofs-node/pkg/services/object/internal" apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status" "github.com/nspcc-dev/neofs-sdk-go/container" @@ -192,7 +193,29 @@ func (s *Service) getECObjectHeaderByRule(ctx context.Context, localNodeKey ecds // Returns [apistatus.ErrObjectAlreadyRemoved] if the object was marked for // removal. Returns [apistatus.ErrObjectNotFound] otherwise. func (s *Service) copyECObject(ctx context.Context, cnr cid.ID, parent oid.ID, sTok *session.Object, - rules []iec.Rule, sortedNodeLists [][]netmap.NodeInfo, dst ObjectWriter) error { + rules []iec.Rule, sortedNodeLists [][]netmap.NodeInfo, dst ObjectWriter, transport GetECRequestTransport) error { + var partial partialObjectCopy + if transport != nil { + err := s.streamECObject(ctx, transport, rules[0], 0, sortedNodeLists[0]) + var split *object.SplitInfoError + switch { + case err == nil: // default behaves the same, but this case detects success faster + return nil + case errors.Is(err, ErrLinker): + // TODO: reuse linker from response + case errors.As(err, &partial): + // TODO: optimize following code by taking into account partial success + case errors.As(err, &split): + info := split.SplitInfo() + if info == nil { + return errors.New("no info in size-split error") + } + return s.copySplitECObjectByInfo(ctx, dst, cnr, parent, sTok, rules, sortedNodeLists, 0, *info) + default: + return err + } + } + // TODO: sort EC rules by complexity and try simpler ones first. Note that rule idxs passed as arguments must be kept. for i := range rules { obj, err := s.restoreFromECPartsByRule(ctx, cnr, parent, sTok, rules[i], i, sortedNodeLists[i]) @@ -200,6 +223,16 @@ func (s *Service) copyECObject(ctx context.Context, cnr cid.ID, parent oid.ID, s if obj.Type() == object.TypeLink && obj.GetID() != parent { return s.copySplitECObjectByLinker(ctx, dst, cnr, parent, sTok, rules, sortedNodeLists, i, obj) } + if transport != nil && partial.copiedHeader { + pld := obj.Payload() + if partial.copiedPayloadLength >= uint64(len(pld)) { + return fmt.Errorf("unexpected full object payload len %d bytes while %d have already been copied", len(pld), partial.copiedPayloadLength) + } + if err := dst.WriteChunk(pld[partial.copiedPayloadLength:]); err != nil { + return fmt.Errorf("write payload: %w", err) + } + return nil + } if err := copyObject(dst, obj); err != nil { return fmt.Errorf("copy object: %w", err) } @@ -290,7 +323,7 @@ func (s *Service) copySplitECObjectByLinker(ctx context.Context, dst ObjectWrite func (s *Service) copySizeSplitECObjectByParts(ctx context.Context, dst ObjectWriter, cnr cid.ID, parent oid.ID, sTok *session.Object, rules []iec.Rule, sortedNodeLists [][]netmap.NodeInfo, fromRule int, parentHdr *object.Object, sizeSplitParts []object.MeasuredObject) error { if err := dst.WriteHeader(parentHdr); err != nil { - return fmt.Errorf("%w: write parent header: %w", errStreamFailure, err) + return fmt.Errorf("%w: write parent header: %w", ErrResponseStreamFailure, err) } nextPart: @@ -305,7 +338,7 @@ nextPart: return fmt.Errorf("get size-split part #%d=%s: unexpected linker", partIdx, partID) } if err := dst.WriteChunk(obj.Payload()); err != nil { - return fmt.Errorf("failed to write size-split part #%d=%s: %w: %w", partIdx, partID, errStreamFailure, err) + return fmt.Errorf("failed to write size-split part #%d=%s: %w: %w", partIdx, partID, ErrResponseStreamFailure, err) } continue nextPart } @@ -694,7 +727,7 @@ func (s *Service) copyECObjectRange(ctx context.Context, dst ChunkWriter, cnr ci fullPldLen, off, ln, nil) } if err == nil || errors.Is(err, apistatus.ErrObjectAlreadyRemoved) || errors.Is(err, apistatus.ErrObjectOutOfRange) || - errors.Is(err, apistatus.ErrObjectAccessDenied) || errors.Is(err, errStreamFailure) || errors.Is(err, ctx.Err()) { + errors.Is(err, apistatus.ErrObjectAccessDenied) || errors.Is(err, ErrResponseStreamFailure) || errors.Is(err, ctx.Err()) { return err } @@ -886,7 +919,7 @@ nextPart: } if errors.Is(err, apistatus.ErrObjectAlreadyRemoved) || errors.Is(err, apistatus.ErrObjectOutOfRange) || - errors.Is(err, errStreamFailure) || errors.Is(err, partCtx.Err()) { + errors.Is(err, ErrResponseStreamFailure) || errors.Is(err, partCtx.Err()) { return err } @@ -1013,7 +1046,7 @@ func (s *Service) copyECObjectRangeByParts(ctx context.Context, dst ChunkWriter, failedPartIdx, failedPartWritten, written, err := s.copyECPartsRanges(stageCtx, dst, localNodeKey, cnr, parent, sTok, rule, ruleIdx, sortedNodes, fullPartLen, firstPartIdx, firstPartOff, lastPartIdx, lastPartTo, firstPartStream) if err == nil || errors.Is(err, apistatus.ErrObjectAlreadyRemoved) || errors.Is(err, apistatus.ErrObjectOutOfRange) || - errors.Is(err, apistatus.ErrObjectAccessDenied) || errors.Is(err, errStreamFailure) || errors.Is(err, ctx.Err()) { + errors.Is(err, apistatus.ErrObjectAccessDenied) || errors.Is(err, ErrResponseStreamFailure) || errors.Is(err, ctx.Err()) { return written, err } @@ -1062,7 +1095,7 @@ func (s *Service) copyECObjectRangeByParts(ctx context.Context, dst ChunkWriter, } if err := dst.WriteChunk(parts[partIdx][from:to]); err != nil { - return written, fmt.Errorf("%w: write [%d:%d] range of EC part #%d: %w", errStreamFailure, from, to, partIdx, err) + return written, fmt.Errorf("%w: write [%d:%d] range of EC part #%d: %w", ErrResponseStreamFailure, from, to, partIdx, err) } written += to - from @@ -1138,7 +1171,7 @@ func (s *Service) copyECPartsRanges(ctx context.Context, dst ChunkWriter, localN rc.Close() if err != nil { if n == partLen { // should never happen - err = fmt.Errorf("%w: received needed range with error for EC part #%d: %w", errStreamFailure, curIdx, err) + err = fmt.Errorf("%w: received needed range with error for EC part #%d: %w", ErrResponseStreamFailure, curIdx, err) } return curIdx, n, written, err } @@ -1351,6 +1384,241 @@ func (s *Service) getECPartRangeFromNode(ctx context.Context, cnr cid.ID, parent return rc, nil } +func (s *Service) streamECObject(ctx context.Context, transport GetECRequestTransport, rule iec.Rule, ruleIdx int, sortedNodes []netmap.NodeInfo) error { + copiedHdr, fullPldLen, partPldLen, copiedPldLen, err := s.streamFirstECPart(ctx, transport, rule, ruleIdx, sortedNodes) + if err != nil { + if errors.Is(err, ErrResponded) { + return nil + } + return fmt.Errorf("part#0: %w", err) + } + + if !copiedHdr { + return partialObjectCopy{} + } + + if rule.DataPartNum == 1 && copiedPldLen == partPldLen { + return nil + } + + partial := partialObjectCopy{ + copiedHeader: true, + copiedPayloadLength: copiedPldLen, + } + + if copiedPldLen < partPldLen || rule.DataPartNum == 1 { + return partial + } + + if rule.DataPartNum == 2 { + partInfo := iec.PartInfo{ + RuleIndex: ruleIdx, + Index: 1, + } + copiedPartPld, err := s.streamECPartRangePrefix(ctx, transport, rule, partInfo, sortedNodes, fullPldLen-partPldLen, nil) + if err != nil { + if errors.Is(err, ErrResponded) { + return nil + } + return fmt.Errorf("part#1: %w", err) + } + + partial.copiedPayloadLength += copiedPartPld + if partial.copiedPayloadLength > fullPldLen { + return fmt.Errorf("payload overflow: full %d bytes, copied %d", fullPldLen, partial.copiedPayloadLength) + } + + if partial.copiedPayloadLength == fullPldLen { + return nil + } + + return partial + } + + type task struct { + partIdx int + controlCh <-chan struct{} + } + taskCh := make(chan task, 1) + + var resErr atomic.Value + ctx, cancel := context.WithCancel(ctx) + + work := func() { + defer cancel() + for { + select { + case <-ctx.Done(): + return + case t := <-taskCh: + var ln uint64 + var nextControlCh chan struct{} + if t.partIdx < int(rule.DataPartNum)-1 { + ln = partPldLen + + nextControlCh = make(chan struct{}, 1) + taskCh <- task{ + partIdx: t.partIdx + 1, + controlCh: nextControlCh, + } + } else { + // last part can be suffixed with zeros which should not be transmitted + ln = fullPldLen - partPldLen*uint64(rule.DataPartNum-1) + } + + partInfo := iec.PartInfo{ + RuleIndex: ruleIdx, + Index: t.partIdx, + } + copiedPartPld, err := s.streamECPartRangePrefix(ctx, transport, rule, partInfo, sortedNodes, ln, t.controlCh) + if err != nil { + resErr.CompareAndSwap(nil, fmt.Errorf("part#%d: %w", t.partIdx, err)) + return + } + + if copiedPartPld > 0 { + // data is copied sequentially, so no concurrency here + partial.copiedPayloadLength += copiedPartPld + if partial.copiedPayloadLength > fullPldLen { + resErr.CompareAndSwap(nil, fmt.Errorf("payload overflow: full %d bytes, copied %d", fullPldLen, partial.copiedPayloadLength)) + return + } + } + + if copiedPartPld < ln || nextControlCh == nil { + return + } + + close(nextControlCh) + } + } + } + + taskCh <- task{ + partIdx: 1, + controlCh: nil, // do not wait + } + + go work() + work() + + <-ctx.Done() + + if val := resErr.Load(); val != nil { + return val.(error) + } + + if partial.copiedPayloadLength == fullPldLen { + return nil + } + + return partial +} + +func (s *Service) streamFirstECPart(ctx context.Context, transport GetECRequestTransport, rule iec.Rule, ruleIdx int, sortedNodes []netmap.NodeInfo) (bool, uint64, uint64, uint64, error) { + var err error + var copiedHdr bool + var parentPldLen uint64 + var partPldLen uint64 + var copiedPartPld uint64 + + partInfo := iec.PartInfo{ + RuleIndex: ruleIdx, + Index: 0, + } + + for nodeIdx := range iec.NodeSequenceForPart(0, int(rule.DataPartNum+rule.ParityPartNum), len(sortedNodes)) { + local := s.neoFSNet.IsLocalNodePublicKey(sortedNodes[nodeIdx].PublicKey()) + + if !copiedHdr { + if local { + copiedHdr, parentPldLen, partPldLen, copiedPartPld, err = transport.CopyLocalECPartParentHeaderAndPayload(ctx, s.localObjects.(*engine.StorageEngine), partInfo) + } else { + conn, connErr := s.conns.(*clientCacheWrapper).connect(ctx, sortedNodes[nodeIdx]) + if connErr != nil { + s.logSNConnFailure(sortedNodes[nodeIdx], connErr) + continue + } + + copiedHdr, parentPldLen, partPldLen, copiedPartPld, err = transport.CopyRemoteECPartParentHeaderAndPayload(ctx, conn, partInfo) + } + if err != nil { + return false, 0, 0, 0, err + } + + // TODO: verify partPldLen against parentPldLen + + if copiedHdr && copiedPartPld == partPldLen { + break + } + + continue + } + + var copiedFromNode uint64 + if local { + copiedFromNode, err = transport.CopyLocalECPartRange(ctx, s.localObjects.(*engine.StorageEngine), partInfo, copiedPartPld, parentPldLen-copiedPartPld, nil) + } else { + conn, connErr := s.conns.(*clientCacheWrapper).connect(ctx, sortedNodes[nodeIdx]) + if connErr != nil { + s.logSNConnFailure(sortedNodes[nodeIdx], connErr) + continue + } + + copiedFromNode, err = transport.CopyRemoteECPartRange(ctx, conn, partInfo, copiedPartPld, parentPldLen-copiedPartPld, nil) + } + if err != nil { + return false, 0, 0, 0, err + } + + copiedPartPld += copiedFromNode + if copiedPartPld > partPldLen { + return false, 0, 0, 0, fmt.Errorf("part payload overflow: full %d bytes, copied %d", partPldLen, copiedPartPld) + } + + if copiedPartPld == partPldLen { + break + } + } + + return copiedHdr, parentPldLen, partPldLen, copiedPartPld, nil +} + +func (s *Service) streamECPartRangePrefix(ctx context.Context, transport GetECRequestTransport, rule iec.Rule, partInfo iec.PartInfo, sortedNodes []netmap.NodeInfo, ln uint64, controlCh <-chan struct{}) (uint64, error) { + var err error + var copiedLen uint64 + + for nodeIdx := range iec.NodeSequenceForPart(partInfo.Index, int(rule.DataPartNum+rule.ParityPartNum), len(sortedNodes)) { + var copiedLenNode uint64 + if s.neoFSNet.IsLocalNodePublicKey(sortedNodes[nodeIdx].PublicKey()) { + copiedLenNode, err = transport.CopyLocalECPartRange(ctx, s.localObjects.(*engine.StorageEngine), partInfo, copiedLen, ln-copiedLen, controlCh) + } else { + conn, connErr := s.conns.(*clientCacheWrapper).connect(ctx, sortedNodes[nodeIdx]) + if connErr != nil { + s.logSNConnFailure(sortedNodes[nodeIdx], connErr) + continue + } + + copiedLenNode, err = transport.CopyRemoteECPartRange(ctx, conn, partInfo, copiedLen, ln-copiedLen, controlCh) + } + + if err != nil { + return 0, err + } + + copiedLen += copiedLenNode + if copiedLen > ln { + return 0, fmt.Errorf("part payload range overflow: requested %d bytes, copied %d", ln, copiedLen) + } + + if copiedLen == ln { + break + } + } + + return copiedLen, nil +} + // returns [iec.PartInfo.RuleIndex] = -1 if request is not for particular EC part. func checkECPartInfoRequest(xHdrs []string, cnr container.Container) (iec.PartInfo, error) { var res iec.PartInfo diff --git a/pkg/services/object/get/exec.go b/pkg/services/object/get/exec.go index 83f3278d8d..6ae18569c8 100644 --- a/pkg/services/object/get/exec.go +++ b/pkg/services/object/get/exec.go @@ -22,8 +22,6 @@ const ( streamChunkSize = 256 * 1024 // 256 KiB ) -var errStreamFailure = errors.New("stream failure") - type statusError struct { status int err error @@ -512,7 +510,7 @@ func copyPayloadStreamBuffer(w ChunkWriter, r io.Reader, buf []byte) (uint64, er n, err := r.Read(buf) if n > 0 { if writeErr := w.WriteChunk(buf[:n]); writeErr != nil { - return done, fmt.Errorf("%w: %w", errStreamFailure, writeErr) + return done, fmt.Errorf("%w: %w", ErrResponseStreamFailure, writeErr) } done += uint64(n) } diff --git a/pkg/services/object/get/get.go b/pkg/services/object/get/get.go index e675e1b0e1..0ce9f89612 100644 --- a/pkg/services/object/get/get.go +++ b/pkg/services/object/get/get.go @@ -112,7 +112,7 @@ func (s *Service) Get(ctx context.Context, prm Prm) error { } return s.copyECObject(ctx, prm.addr.Container(), prm.addr.Object(), prm.common.SessionToken(), - ecRules, ecNodeLists, prm.objWriter) + ecRules, ecNodeLists, prm.objWriter, prm.ecTransport) } // GetRange serves a request to get an object by address, and returns Streamer instance. diff --git a/pkg/services/object/get/prm.go b/pkg/services/object/get/prm.go index 7354125ad4..6b7ce7e444 100644 --- a/pkg/services/object/get/prm.go +++ b/pkg/services/object/get/prm.go @@ -32,6 +32,8 @@ type Prm struct { submitLocalGetStreamFn SubmitStreamFunc forwardRequestFn ForwardGetRequestFunc + + ecTransport GetECRequestTransport } // RangePrm groups parameters of GetRange service call. @@ -256,3 +258,8 @@ func (p *RangePrm) WithBuffer(buffer []byte, submitStreamFn SubmitDataStreamFunc func (p *RangePrm) SetRequestForwarder(f ForwardRangeRequestFunc) { p.forwardRequestFn = f } + +// WithECTransport specifies transport layer to for EC handling. +func (p *Prm) WithECTransport(transport GetECRequestTransport) { + p.ecTransport = transport +} diff --git a/pkg/services/object/get/service.go b/pkg/services/object/get/service.go index aa496d6f0c..61dd6126c6 100644 --- a/pkg/services/object/get/service.go +++ b/pkg/services/object/get/service.go @@ -3,6 +3,8 @@ package getsvc import ( "context" "crypto/ecdsa" + "encoding/hex" + "errors" "io" iec "github.com/nspcc-dev/neofs-node/internal/ec" @@ -50,6 +52,66 @@ type NeoFSNetwork interface { IsLocalNodePublicKey([]byte) bool } +// ErrResponded is returned when server successfully finished processing of some +// request. +var ErrResponded = errors.New(" responded") + +// ErrResponseStreamFailure is returned when response stream no longer works. +var ErrResponseStreamFailure = errors.New("stream failure") + +// ErrLinker is returned when object of LINK type is received. +var ErrLinker = errors.New("linker") + +// 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 + // succeeded, CopyRemoteECPartParentHeaderAndPayload sends parent header and + // part payload to the client, and returns: + // - flag whether parent header was copied or not; + // - payload length of original requested object in bytes; + // - payload length of part object in bytes; + // - number of part's payload bytes copied. + // + // If response stream fails, [ErrResponseStreamFailure] is returned. + // + // If the node responds with split object info, + // CopyRemoteECPartParentHeaderAndPayload converts it to + // [*object.SplitInfoError] and returns. + // + // If the node responds with any failure status other than 'not found', the + // response is copied to the client and [ErrResponded] is returned. + // + // If the node responds with object of LINK type, [ErrLinker] is returned. + // + // Otherwise, no error is returned. Copying can be incomplete in this case. + // + // CopyRemoteECPartParentHeaderAndPayload is never called concurrently. + CopyRemoteECPartParentHeaderAndPayload(ctx context.Context, conn client.MultiAddressClient, partInfo iec.PartInfo) (bool, uint64, uint64, uint64, error) + // 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 + // node using conn to it. If succeeded, CopyRemoteECPartRange sends with payload + // range to the client, and returns number of bytes copied. + // + // If response stream fails, [ErrResponseStreamFailure] is returned. + // + // If the node responds with any failure status other than 'not found', the + // response is copied to the client and [ErrResponded] is returned. + // + // Otherwise, no error is returned. Copying can be incomplete in this case. + // + // If controlCh is passed, CopyRemoteECPartRange blocks copying data until signal + // from it. + // + // CopyRemoteECPartRange can be called concurrently for different partInfo, but + // never for the same one. + CopyRemoteECPartRange(ctx context.Context, conn client.MultiAddressClient, partInfo iec.PartInfo, off, ln uint64, controlCh <-chan struct{}) (uint64, error) + // CopyLocalECPartRange works like CopyRemoteECPartRange but locally. + CopyLocalECPartRange(ctx context.Context, storage *engine.StorageEngine, partInfo iec.PartInfo, off, ln uint64, ch <-chan struct{}) (uint64, error) +} + // Service utility serving requests of Object.Get service. type Service struct { *cfg @@ -185,3 +247,10 @@ func WithNNSResolver(resolver sessionv2.NNSResolver) Option { c.nnsResolver = resolver } } + +func (s *Service) logSNConnFailure(node netmapsdk.NodeInfo, err error) { + s.log.Warn("remote SN connection failure", + zap.String("publicKey", hex.EncodeToString(node.PublicKey())), + zap.Error(err), + ) +} diff --git a/pkg/services/object/get/util.go b/pkg/services/object/get/util.go index 398fcd5060..7257b62a34 100644 --- a/pkg/services/object/get/util.go +++ b/pkg/services/object/get/util.go @@ -568,3 +568,18 @@ func localNodeInSet(n NeoFSNetwork, nodes []netmap.NodeInfo) bool { func zapEndpoints(info netmap.NodeInfo) zap.Field { return zap.Strings("address group", slices.Collect(info.NetworkEndpoints())) } + +// partialObjectCopy contains information about incomplete copying of some +// object. +type partialObjectCopy struct { + // Whether header was copied or not. + copiedHeader bool + // Number of payload bytes copied. + copiedPayloadLength uint64 +} + +// Error implements [error]. +func (x partialObjectCopy) Error() string { + return fmt.Sprintf("incomplete object copy (copied header: %t, copied payload: %d bytes)", + x.copiedHeader, x.copiedPayloadLength) +} diff --git a/pkg/services/object/get_test.go b/pkg/services/object/get_test.go index 3aabc8a9ed..ebb13febaf 100644 --- a/pkg/services/object/get_test.go +++ b/pkg/services/object/get_test.go @@ -28,6 +28,7 @@ import ( usertest "github.com/nspcc-dev/neofs-sdk-go/user/test" "github.com/nspcc-dev/neofs-sdk-go/version" "github.com/stretchr/testify/require" + "go.uber.org/zap" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/test/bufconn" @@ -55,7 +56,7 @@ func TestServer_Get_Local(t *testing.T) { ) handlers := &getOnlyHandler{svc: handler} - srv := New(handlers, 0, nil, fsChain, nil, nil, signer.ECDSAPrivateKey, mtrc, aclChecker, reqInfoExt, nil) + srv := New(handlers, 0, nil, fsChain, nil, nil, signer.ECDSAPrivateKey, mtrc, aclChecker, reqInfoExt, nil, zap.NewNop()) for _, pldLen := range []uint64{ 0, 1, @@ -179,7 +180,7 @@ func TestServer_Get_Remote(t *testing.T) { ) handlers := getOnlyHandler{svc: handler} - srv := New(handlers, 0, nil, fsChain, nil, nil, signer.ECDSAPrivateKey, &mtrc, aclChecker, reqInfoExt, nil) + srv := New(handlers, 0, nil, fsChain, nil, nil, signer.ECDSAPrivateKey, &mtrc, aclChecker, reqInfoExt, nil, zap.NewNop()) t.Run("object", func(t *testing.T) { const payloadLen = 100 << 10 diff --git a/pkg/services/object/head_test.go b/pkg/services/object/head_test.go index 615def197d..0926b7e7c1 100644 --- a/pkg/services/object/head_test.go +++ b/pkg/services/object/head_test.go @@ -25,6 +25,7 @@ import ( usertest "github.com/nspcc-dev/neofs-sdk-go/user/test" "github.com/nspcc-dev/neofs-sdk-go/version" "github.com/stretchr/testify/require" + "go.uber.org/zap" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/test/bufconn" @@ -56,7 +57,7 @@ func TestServer_Head_Local(t *testing.T) { ) handlers := headOnlyHandler{svc: handler} - srv := New(handlers, 0, nil, fsChain, nil, nil, signer.ECDSAPrivateKey, mtrc, aclChecker, reqInfoExt, nil) + srv := New(handlers, 0, nil, fsChain, nil, nil, signer.ECDSAPrivateKey, mtrc, aclChecker, reqInfoExt, nil, zap.NewNop()) assertWithVersion := func(t *testing.T, ver version.Version) *protoobject.HeadResponse { req := newLocalHeadRequest(t, ver, obj.Address(), signer) @@ -136,7 +137,7 @@ func TestServer_Head_Remote(t *testing.T) { ) handlers := headOnlyHandler{svc: handler} - srv := New(handlers, 0, nil, fsChain, nil, nil, signer.ECDSAPrivateKey, mtrc, aclChecker, reqInfoExt, nil) + srv := New(handlers, 0, nil, fsChain, nil, nil, signer.ECDSAPrivateKey, mtrc, aclChecker, reqInfoExt, nil, zap.NewNop()) t.Run("EC part", func(t *testing.T) { nodes := make([]netmap.NodeInfo, 3) @@ -188,7 +189,7 @@ func TestServer_Head_Remote(t *testing.T) { ) handlers := headOnlyHandler{svc: handler} - srv := New(handlers, 0, nil, fsChain, nil, nil, signer.ECDSAPrivateKey, mtrc, aclChecker, reqInfoExt, nil) + srv := New(handlers, 0, nil, fsChain, nil, nil, signer.ECDSAPrivateKey, mtrc, aclChecker, reqInfoExt, nil, zap.NewNop()) t.Run("header", func(t *testing.T) { const payloadLen = 100 << 10 diff --git a/pkg/services/object/proto.go b/pkg/services/object/proto.go index 405baf36d5..2c5cb7bd2a 100644 --- a/pkg/services/object/proto.go +++ b/pkg/services/object/proto.go @@ -1,10 +1,13 @@ package object import ( + "crypto/ecdsa" "encoding/binary" "fmt" "io" + "github.com/nspcc-dev/neo-go/pkg/crypto/keys" + "github.com/nspcc-dev/neo-go/pkg/smartcontract" iobject "github.com/nspcc-dev/neofs-node/internal/object" iprotobuf "github.com/nspcc-dev/neofs-node/internal/protobuf" neofscrypto "github.com/nspcc-dev/neofs-sdk-go/crypto" @@ -27,6 +30,16 @@ const ( getResponseChunkBufferLen = maxChunkOffsetInGetResponse + maxGetResponseChunkLen + maxResponseVerificationHeaderLen ) +// Fixed message lengths. +const ( + getByAddressRequestBodyLen = 1 + 1 + iprotobuf.ObjectAddressLength + compressedECDSAPublicKeyLen = smartcontract.PublicKeyLen + ecdsaWithSHA256SignatureValueLen = 1 + keys.SignatureLen + ecdsaWithSHA512SignatureLen = 1 + 1 + compressedECDSAPublicKeyLen + + 1 + 1 + ecdsaWithSHA256SignatureValueLen // scheme is 0 + requestVerificationHeaderECDSAWIthSHA512Len = (1 + 1 + ecdsaWithSHA512SignatureLen) * 3 +) + var currentVersionResponseMetaHeader []byte func init() { @@ -268,3 +281,75 @@ func getBufferForChunkGetResponse() (*iprotobuf.MemBuffer, []byte) { chunkBuf := item.SliceBuffer[maxChunkOffsetInGetResponse:] return item, chunkBuf[:len(chunkBuf)-maxResponseVerificationHeaderLen] } + +func calculateXHeaderLength(key, val string) int { + return 1 + protowire.SizeBytes(len(key)) + // 1 for iprotobuf.TagBytes1 + 1 + protowire.SizeBytes(len(val)) // 1 for iprotobuf.TagBytes2 +} + +func writeRequestMetaXHeader(buf []byte, ln int, key, val string) int { + buf[0] = iprotobuf.TagBytes4 + off := 1 + binary.PutUvarint(buf[1:], uint64(ln)) + return off + writeXHeader(buf[off:], key, val) +} + +func writeXHeader(buf []byte, key, val string) int { + off := writeBytesField(buf, iprotobuf.TagBytes1, key) + return off + writeBytesField(buf[off:], iprotobuf.TagBytes2, val) +} + +func writeBytesField[T string | []byte](buf []byte, tag byte, val T) int { + buf[0] = tag + off := 1 + binary.PutUvarint(buf[1:], uint64(len(val))) + return off + copy(buf[off:], val) +} + +func writeStablyMarshalledField(buf []byte, tag byte, ln int, fld interface { + MarshalStable([]byte) +}) int { + buf[0] = tag + off := 1 + binary.PutUvarint(buf[1:], uint64(ln)) + fld.MarshalStable(buf[off:]) + return off + ln +} + +func signECDSAWithSHA512(privKey ecdsa.PrivateKey, data []byte) ([]byte, error) { + sig, err := neofsecdsa.Signer(privKey).Sign(data) + if err != nil { + return nil, err + } + + if len(sig) != ecdsaWithSHA256SignatureValueLen { + return nil, fmt.Errorf("wrong signature len: expected %d, got %d", ecdsaWithSHA256SignatureValueLen, len(sig)) + } + + return sig, nil +} + +func writeRequestVerificationHeader(buf []byte, pubKey, bodySIg, metaSig, originSig []byte) int { + buf[0] = iprotobuf.TagBytes3 + off := 1 + binary.PutUvarint(buf[1:], requestVerificationHeaderECDSAWIthSHA512Len) + off += writeRequestVerificationSignature(buf[off:], iprotobuf.TagBytes1, pubKey, bodySIg) + off += writeRequestVerificationSignature(buf[off:], iprotobuf.TagBytes2, pubKey, metaSig) + off += writeRequestVerificationSignature(buf[off:], iprotobuf.TagBytes3, pubKey, originSig) + return off +} + +func writeRequestVerificationSignature(buf []byte, tag byte, pubKey, sig []byte) int { + buf[0] = tag + buf[1] = ecdsaWithSHA512SignatureLen + return 2 + writeECDSAWithSHA512Signature(buf[2:], pubKey, sig) +} + +func writeECDSAWithSHA512Signature(buf []byte, pubKey, sig []byte) int { + // key + buf[0] = iprotobuf.TagBytes1 // key + buf[1] = compressedECDSAPublicKeyLen + off := 2 + copy(buf[2:], pubKey) + // value + buf[off] = iprotobuf.TagBytes2 + off++ + buf[off] = ecdsaWithSHA256SignatureValueLen + off++ + return off + copy(buf[off:], sig) // scheme is 0 +} diff --git a/pkg/services/object/server.go b/pkg/services/object/server.go index f1d14ec77e..4cc7820f38 100644 --- a/pkg/services/object/server.go +++ b/pkg/services/object/server.go @@ -48,6 +48,7 @@ import ( "github.com/nspcc-dev/neofs-sdk-go/user" "github.com/nspcc-dev/neofs-sdk-go/version" "github.com/panjf2000/ants/v2" + "go.uber.org/zap" "google.golang.org/grpc" grpccodes "google.golang.org/grpc/codes" "google.golang.org/grpc/mem" @@ -207,23 +208,30 @@ type Server struct { reqInfoProc ACLInfoExtractor nodeClients ClientConstructor searchWorkers *ants.Pool + log *zap.Logger } // New provides protoobject.ObjectServiceServer for the given parameters. -func New(hs Handlers, magicNumber uint32, sp *ants.Pool, fsChain FSChain, st Storage, metaSvc *metasvc.Meta, signer ecdsa.PrivateKey, m MetricCollector, ac aclsvc.ACLChecker, rp ACLInfoExtractor, cs ClientConstructor) *Server { +func New(hs Handlers, magicNumber uint32, sp *ants.Pool, fsChain FSChain, st Storage, metaSvc *metasvc.Meta, signer ecdsa.PrivateKey, m MetricCollector, ac aclsvc.ACLChecker, rp ACLInfoExtractor, cs ClientConstructor, log *zap.Logger) *Server { + pubKeyBytes := (*keys.PublicKey)(&signer.PublicKey).Bytes() + if len(pubKeyBytes) != compressedECDSAPublicKeyLen { + panic(fmt.Sprintf("wrong public key len: expected %d, got %d", compressedECDSAPublicKeyLen, len(pubKeyBytes))) + } + return &Server{ handlers: hs, fsChain: fsChain, storage: st, meta: metaSvc, signer: signer, - pubKeyBytes: (*keys.PublicKey)(&signer.PublicKey).Bytes(), + pubKeyBytes: pubKeyBytes, mNumber: magicNumber, metrics: m, aclChecker: ac, reqInfoProc: rp, nodeClients: cs, searchWorkers: sp, + log: log, } } @@ -976,6 +984,13 @@ func (s *Server) Get(req *protoobject.GetRequest, gStream protoobject.ObjectServ return s.sendStatusGetResponse(gStream, err, needSignResp) } + p.WithECTransport(&getECTransport{ + server: s, + request: req, + signResponses: needSignResp, + responseStream: gStream, + }) + // TODO: consider optimization // We could acquire ~256K buffer (like for chunks) if storage would try to read it full. // Then small objects would fit into a single buffer, and for large ones it'd be possible to @@ -1001,6 +1016,9 @@ func (s *Server) Get(req *protoobject.GetRequest, gStream protoobject.ObjectServ err = s.handlers.Get(gStream.Context(), p) if err != nil { + if errors.Is(err, getsvc.ErrResponseStreamFailure) { + return err + } return s.sendStatusGetResponse(gStream, err, needSignResp) } @@ -1023,7 +1041,7 @@ func (s *Server) Get(req *protoobject.GetRequest, gStream protoobject.ObjectServ pldFldOff := max(idf.To, sigf.To, hdrf.To) - err = s.copyGetStream(gStream, hdrRespBuf, hdrBuf, hdrLen, stream, pldFldOff, needSignResp) // defer + err = s.copyGetStream(gStream, hdrRespBuf, hdrBuf, hdrLen, pldFldOff, stream, pldFldOff, needSignResp) // defer if err != nil { return s.sendStatusGetResponse(gStream, err, needSignResp) } @@ -1031,12 +1049,12 @@ func (s *Server) Get(req *protoobject.GetRequest, gStream protoobject.ObjectServ return nil } -func (s *Server) copyGetStream(gStream protoobject.ObjectService_GetServer, hdrRespBuf *iprotobuf.MemBuffer, hdrBuf []byte, - hdrLen int, stream io.Reader, pldFldOff int, needSignResp bool) error { +func (s *Server) copyGetStream(gStream grpc.ServerStream, hdrRespBuf *iprotobuf.MemBuffer, hdrBuf []byte, + prefixLen, hdrTo int, stream io.Reader, pldFldOff int, needSignResp bool) error { var chunkRespBuf *iprotobuf.MemBuffer var chunkBuf []byte - prereadPldLen := hdrLen - pldFldOff + prereadPldLen := prefixLen - pldFldOff if prereadPldLen > 0 { chunkRespBuf, chunkBuf = getBufferForChunkGetResponse() copy(chunkBuf, hdrBuf[pldFldOff:][:prereadPldLen]) @@ -1046,7 +1064,7 @@ func (s *Server) copyGetStream(gStream protoobject.ObjectService_GetServer, hdrR // `header.payload_length` field. } - bodyf := shiftHeaderInGetResponseBuffer(hdrRespBuf.SliceBuffer, hdrBuf[:pldFldOff]) + bodyf := shiftHeaderInGetResponseBuffer(hdrRespBuf.SliceBuffer, hdrBuf[:hdrTo]) if needSignResp { n, err := s.signResponse(hdrRespBuf.SliceBuffer[bodyf.To:], hdrRespBuf.SliceBuffer[bodyf.ValueFrom:bodyf.To], nil) @@ -1069,7 +1087,7 @@ func (s *Server) copyGetStream(gStream protoobject.ObjectService_GetServer, hdrR if chunkRespBuf != nil { chunkRespBuf.Free() } - return err + return fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) } if chunkRespBuf == nil { @@ -1088,7 +1106,10 @@ func (s *Server) copyGetStream(gStream protoobject.ObjectService_GetServer, hdrR streamDone := errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) if err != nil && !streamDone { chunkRespBuf.Free() - return fmt.Errorf("read payload stream: %w", err) + return copyReadError{ + error: fmt.Errorf("read payload stream: %w", err), + written: sent, + } } n += prereadPldLen @@ -1121,7 +1142,7 @@ func (s *Server) copyGetStream(gStream protoobject.ObjectService_GetServer, hdrR chunkRespBuf.SetBounds(bodyf.From, bodyf.To) if err = gStream.SendMsg(chunkRespBuf); err != nil { - return err + return fmt.Errorf("%w: %w", getsvc.ErrResponseStreamFailure, err) } sent += n if streamDone { @@ -1355,7 +1376,7 @@ func (s *Server) GetRange(req *protoobject.GetRangeRequest, gStream protoobject. return nil } - err = s.copyRangeStream(gStream, stream, needSignResponse) + err = s.copyRangeStream(gStream, stream, needSignResponse, shiftPayloadChunkInRangeResponseBuffer) if err != nil { return s.sendStatusRangeResponse(gStream, err, req) } @@ -1363,7 +1384,8 @@ func (s *Server) GetRange(req *protoobject.GetRangeRequest, gStream protoobject. return nil } -func (s *Server) copyRangeStream(gStream protoobject.ObjectService_GetRangeServer, stream io.Reader, needSignResp bool) error { +func (s *Server) copyRangeStream(gStream grpc.ServerStream, stream io.Reader, needSignResp bool, shiftFn func([]byte, int, int) iprotobuf.FieldBounds) error { + var sent int for { // chunk response buffers for GET completely suitable for RANGE respBuf, buf := getBufferForChunkGetResponse() @@ -1372,7 +1394,10 @@ func (s *Server) copyRangeStream(gStream protoobject.ObjectService_GetRangeServe streamDone := errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) if err != nil && !streamDone { respBuf.Free() - return fmt.Errorf("read payload stream: %w", err) + return copyReadError{ + error: fmt.Errorf("read payload stream: %w", err), + written: sent, + } } if n == 0 { @@ -1380,7 +1405,7 @@ func (s *Server) copyRangeStream(gStream protoobject.ObjectService_GetRangeServe return nil } - bodyf := shiftPayloadChunkInRangeResponseBuffer(respBuf.SliceBuffer, maxChunkOffsetInGetResponse, n) + bodyf := shiftFn(respBuf.SliceBuffer, maxChunkOffsetInGetResponse, n) if needSignResp { n, err := s.signResponse(respBuf.SliceBuffer[bodyf.To:], respBuf.SliceBuffer[bodyf.ValueFrom:bodyf.To], nil) @@ -1392,9 +1417,15 @@ func (s *Server) copyRangeStream(gStream protoobject.ObjectService_GetRangeServe } respBuf.SetBounds(bodyf.From, bodyf.To) - if err = gStream.SendMsg(respBuf); err != nil || streamDone { + if err = gStream.SendMsg(respBuf); err != nil { return err } + + sent += n + + if streamDone { + return nil + } } } diff --git a/pkg/services/object/server_test.go b/pkg/services/object/server_test.go index 9c66ae890d..0d011d449a 100644 --- a/pkg/services/object/server_test.go +++ b/pkg/services/object/server_test.go @@ -54,6 +54,7 @@ import ( "github.com/nspcc-dev/neofs-sdk-go/user" "github.com/panjf2000/ants/v2" "github.com/stretchr/testify/require" + "go.uber.org/zap" "google.golang.org/grpc" ) @@ -341,7 +342,7 @@ func TestServer_Replicate(t *testing.T) { var noCallReqProc noCallTestReqInfoExtractor var noCallCs noCallClients sp := newSearchPool(t) - noCallSrv := New(noCallObjSvc, 0, sp, &noCallFSChain, noCallStorage, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs) + noCallSrv := New(noCallObjSvc, 0, sp, &noCallFSChain, noCallStorage, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs, zap.NewNop()) clientSigner := neofscryptotest.Signer() clientPubKey := neofscrypto.PublicKeyBytes(clientSigner.Public()) serverPubKey := neofscrypto.PublicKeyBytes(neofscryptotest.Signer().Public()) @@ -505,7 +506,7 @@ func TestServer_Replicate(t *testing.T) { t.Run("apply storage policy failure", func(t *testing.T) { fsChain := newTestFSChain(t, serverPubKey, clientPubKey, cnr) - srv := New(noCallObjSvc, 0, sp, fsChain, noCallStorage, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs) + srv := New(noCallObjSvc, 0, sp, fsChain, noCallStorage, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs, zap.NewNop()) fsChain.cnrErr = errors.New("any error") @@ -517,7 +518,7 @@ func TestServer_Replicate(t *testing.T) { t.Run("client or server mismatches object's storage policy", func(t *testing.T) { fsChain := newTestFSChain(t, serverPubKey, clientPubKey, cnr) - srv := New(noCallObjSvc, 0, sp, fsChain, noCallStorage, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs) + srv := New(noCallObjSvc, 0, sp, fsChain, noCallStorage, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs, zap.NewNop()) fsChain.serverOutsideCnr = true fsChain.clientOutsideCnr = true @@ -538,7 +539,7 @@ func TestServer_Replicate(t *testing.T) { t.Run("local storage failure", func(t *testing.T) { fsChain := newTestFSChain(t, serverPubKey, clientPubKey, cnr) s := newTestStorage(t, req.Object) - srv := New(noCallObjSvc, 0, sp, fsChain, s, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs) + srv := New(noCallObjSvc, 0, sp, fsChain, s, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs, zap.NewNop()) s.storeErr = errors.New("any error") @@ -554,7 +555,7 @@ func TestServer_Replicate(t *testing.T) { reqForSignature, o := anyValidRequest(t, clientSigner, cnr, objID) fsChain := newTestFSChain(t, serverPubKey, clientPubKey, cnr) s := newTestStorage(t, reqForSignature.Object) - srv := New(noCallObjSvc, mNumber, sp, fsChain, s, nil, signer.ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs) + srv := New(noCallObjSvc, mNumber, sp, fsChain, s, nil, signer.ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs, zap.NewNop()) t.Run("signature not requested", func(t *testing.T) { resp, err := srv.Replicate(context.Background(), reqForSignature) @@ -594,7 +595,7 @@ func TestServer_Replicate(t *testing.T) { t.Run("OK", func(t *testing.T) { fsChain := newTestFSChain(t, serverPubKey, clientPubKey, cnr) s := newTestStorage(t, req.Object) - srv := New(noCallObjSvc, 0, sp, fsChain, s, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs) + srv := New(noCallObjSvc, 0, sp, fsChain, s, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, noCallACLChecker, noCallReqProc, noCallCs, zap.NewNop()) resp, err := srv.Replicate(context.Background(), req) require.NoError(t, err) @@ -660,7 +661,7 @@ func BenchmarkServer_Replicate(b *testing.B) { var fsChain nopFSChain sp := newSearchPool(b) - srv := New(nil, 0, sp, fsChain, nopStorage{}, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, nopACLChecker{}, nopReqInfoExtractor{}, noCallClients{}) + srv := New(nil, 0, sp, fsChain, nopStorage{}, nil, neofscryptotest.Signer().ECDSAPrivateKey, nopMetrics{}, nopACLChecker{}, nopReqInfoExtractor{}, noCallClients{}, zap.NewNop()) for _, tc := range []struct { name string diff --git a/pkg/services/object/util.go b/pkg/services/object/util.go index dbf50dd33a..f57d531b4c 100644 --- a/pkg/services/object/util.go +++ b/pkg/services/object/util.go @@ -17,10 +17,19 @@ import ( ) func handleSplitInfo(raw bool, respStream grpc.ServerStream, respBuf mem.BufferSlice, buffers iprotobuf.BuffersSlice) (bool, error) { + err := _handleSplitInfo(buffers, !raw) + if err != nil { + return false, err + } + + return true, respStream.SendMsg(respBuf) +} + +func _handleSplitInfo(buffers iprotobuf.BuffersSlice, compose bool) error { var si object.SplitInfo var opts protoscan.ScanMessageOptions - if !raw { + if compose { opts.InterceptBytes = func(num protowire.Number, buffers iprotobuf.BuffersSlice) error { if num == protoobject.FieldSplitInfoSplitID { id := object.NewSplitIDFromV2(buffers.ReadOnlyData()) @@ -56,14 +65,14 @@ func handleSplitInfo(raw bool, respStream grpc.ServerStream, respBuf mem.BufferS err := protoscan.ScanMessage(buffers, protoscan.ObjectSplitInfoScheme, opts) if err != nil { - return false, fmt.Errorf("handle split info field: %w", err) + return fmt.Errorf("handle split info field: %w", err) } - if !raw { - return false, object.NewSplitInfoError(&si) + if compose { + return object.NewSplitInfoError(&si) } - return true, respStream.SendMsg(respBuf) + return nil } func (s *Server) sendChunkResponse(respStream grpc.ServerStream, respBuf mem.BufferSlice, chunkBuffers iprotobuf.BuffersSlice, @@ -123,3 +132,12 @@ func (s *Server) sendChunkResponse(respStream grpc.ServerStream, respBuf mem.Buf return remoteSent, nil } + +type copyReadError struct { + error + written int +} + +func (x copyReadError) Unwrap() error { + return x.error +}