diff --git a/CHANGELOG.md b/CHANGELOG.md index 62442bf8e6..06f4c8eb44 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,8 @@ Changelog for NeoFS Node ### Added - Online FSTree layout reshaping via the `blobstor.allow_depth_change` configuration option (#4149) +- SN now serves `ReplicateV2` API (#4168) +- Container SN now uses `ReplicateV2` API in PUT handler when available (#4168) ### Fixed - Broken EC object GET with all data parts missing (#4173) @@ -18,6 +20,7 @@ Changelog for NeoFS Node - GetRange object service method (#4167) ### Updated +- `github.com/nspcc-dev/neofs-sdk-go` module to `v1.0.0-rc.22.0.20260911084039-e749b2b839ac` (#4168) ### Updating from v0.56.0 To change an FSTree layout depth, configure the new `blobstor.depth` and set diff --git a/cmd/neofs-node/object.go b/cmd/neofs-node/object.go index 4a86003783..15eeaa60fc 100644 --- a/cmd/neofs-node/object.go +++ b/cmd/neofs-node/object.go @@ -264,7 +264,7 @@ func initObjectService(c *cfg) { }) os := &objectSource{signer: neofsecdsa.SignerRFC6979(c.key.PrivateKey), get: sGet} - sPut := putsvc.NewService(&transport{clients: putConstructor}, c, c.metaService, + sPut := putsvc.NewService(c, c.metaService, initQuotas(c.cCli, c.cfgObject.quotasTTL), c.containerPayments, putsvc.WithKeyStorage(keyStorage), diff --git a/cmd/neofs-node/transport.go b/cmd/neofs-node/transport.go deleted file mode 100644 index 1caca99057..0000000000 --- a/cmd/neofs-node/transport.go +++ /dev/null @@ -1,72 +0,0 @@ -package main - -import ( - "context" - "fmt" - - apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status" - "github.com/nspcc-dev/neofs-sdk-go/netmap" - protoobject "github.com/nspcc-dev/neofs-sdk-go/proto/object" - "google.golang.org/grpc" - "google.golang.org/grpc/encoding" - "google.golang.org/grpc/encoding/proto" - "google.golang.org/grpc/mem" -) - -type transport struct { - clients *coreClientConstructor -} - -// SendReplicationRequestToNode connects to described node and sends prepared -// replication request message to it. -func (x *transport) SendReplicationRequestToNode(ctx context.Context, req []byte, node netmap.NodeInfo) ([]byte, error) { - c, err := x.clients.Get(ctx, node) - if err != nil { - return nil, fmt.Errorf("connect to remote node: %w", err) - } - - var res []byte - return res, c.ForAnyGRPCConn(ctx, func(ctx context.Context, conn *grpc.ClientConn) error { - // this will be changed during NeoFS API Go deprecation. Code most likely be - // placed in SDK - var resp protoobject.ReplicateResponse - err := conn.Invoke(ctx, protoobject.ObjectService_Replicate_FullMethodName, req, &resp, binaryMessageOnly) - if err != nil { - return fmt.Errorf("API transport (op=%s): %w", protoobject.ObjectService_Replicate_FullMethodName, err) - } - res, err = replicationResultFromResponse(&resp) - return err - }) -} - -// [encoding.Codec] making Marshal to accept and forward []byte messages only. -var binaryMessageOnly = grpc.ForceCodecV2(protoCodecBinaryRequestOnly{}) - -type protoCodecBinaryRequestOnly struct{} - -func (protoCodecBinaryRequestOnly) Name() string { - // may be any non-empty, conflicts are unlikely to arise - return "neofs_binary_sender" -} - -func (protoCodecBinaryRequestOnly) Marshal(msg any) (mem.BufferSlice, error) { - bMsg, ok := msg.([]byte) - if ok { - return mem.BufferSlice{mem.SliceBuffer(bMsg)}, nil - } - - return nil, fmt.Errorf("message is not of type %T", bMsg) -} - -func (protoCodecBinaryRequestOnly) Unmarshal(data mem.BufferSlice, msg any) error { - return encoding.GetCodecV2(proto.Name).Unmarshal(data, msg) -} - -func replicationResultFromResponse(m *protoobject.ReplicateResponse) ([]byte, error) { - err := apistatus.ToError(m.GetStatus()) - if err != nil { - return nil, err - } - - return m.GetObjectSignature(), nil -} diff --git a/go.mod b/go.mod index c8a2a45fb2..8a627a5db8 100644 --- a/go.mod +++ b/go.mod @@ -22,7 +22,7 @@ require ( github.com/nspcc-dev/neo-go v0.123.0 github.com/nspcc-dev/neofs-api-go/v2 v2.14.1-0.20240827150555-5ce597aa14ea github.com/nspcc-dev/neofs-contract v0.26.2-0.20260902204128-d68d58f8e0e2 - github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.22.0.20260904102537-e8003800aad2 + github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.22.0.20260911084039-e749b2b839ac github.com/nspcc-dev/tzhash v1.8.4 github.com/panjf2000/ants/v2 v2.12.1 github.com/prometheus/client_golang v1.24.1 diff --git a/go.sum b/go.sum index ab6dbb6708..4741d2d79a 100644 --- a/go.sum +++ b/go.sum @@ -189,8 +189,8 @@ github.com/nspcc-dev/neofs-api-go/v2 v2.14.1-0.20240827150555-5ce597aa14ea h1:mK github.com/nspcc-dev/neofs-api-go/v2 v2.14.1-0.20240827150555-5ce597aa14ea/go.mod h1:YzhD4EZmC9Z/PNyd7ysC7WXgIgURc9uCG1UWDeV027Y= github.com/nspcc-dev/neofs-contract v0.26.2-0.20260902204128-d68d58f8e0e2 h1:EUu96Lb7VjaQ53ppMmTR9FeeqJw8g7zt3sBRDvPSLqc= github.com/nspcc-dev/neofs-contract v0.26.2-0.20260902204128-d68d58f8e0e2/go.mod h1:H8J/j8vmlK3QsPCz0e65v0ie4i59ENRrdwv7wUZvUIo= -github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.22.0.20260904102537-e8003800aad2 h1:mMrlj7HKYirfwtFSdQEYu2QR5MZe81OFFFZWp6Z5+Sg= -github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.22.0.20260904102537-e8003800aad2/go.mod h1:1jTSRnrBKHCX2nhlBDxUj9bHPb+TCRO6kAEI3XCSyOs= +github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.22.0.20260911084039-e749b2b839ac h1:PvW5leyLsQRHsFOojeb0+tCDs1MlXn11WbG4fi5/Ouw= +github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.22.0.20260911084039-e749b2b839ac/go.mod h1:8Nz0EKfdWZVIyq7bEtus0A0j0SCewKX9IfOAeZW5G+A= github.com/nspcc-dev/rfc6979 v0.2.4 h1:NBgsdCjhLpEPJZqmC9rciMZDcSY297po2smeaRjw57k= github.com/nspcc-dev/rfc6979 v0.2.4/go.mod h1:86ylDw6Kss+P6v4QAJqo1Sp3mC0/Zr9G97xSjQ9TuFg= github.com/nspcc-dev/tzhash v1.8.4 h1:lvuPGWsqEo9dVEvo/kdNLKv/Cy0yxRs9z5hJp8VcBuo= diff --git a/internal/object/api.go b/internal/object/api.go new file mode 100644 index 0000000000..b22fba0ea3 --- /dev/null +++ b/internal/object/api.go @@ -0,0 +1,6 @@ +package object + +import "github.com/nspcc-dev/neofs-sdk-go/version" + +// ReplicateV2FirstAPIVersion is an API version when ReplicateV2 was added. +var ReplicateV2FirstAPIVersion = version.New(2, 27) diff --git a/pkg/core/client/client.go b/pkg/core/client/client.go index 362b7959df..3a7b13ade1 100644 --- a/pkg/core/client/client.go +++ b/pkg/core/client/client.go @@ -13,6 +13,7 @@ import ( protorefs "github.com/nspcc-dev/neofs-sdk-go/proto/refs" "github.com/nspcc-dev/neofs-sdk-go/reputation" "github.com/nspcc-dev/neofs-sdk-go/user" + "github.com/nspcc-dev/neofs-sdk-go/version" "google.golang.org/grpc" ) @@ -62,3 +63,11 @@ func IsMutuallyAuthenticated(c any) bool { x, ok := c.(interface{ IsMutuallyAuthenticated() bool }) return ok && x.IsMutuallyAuthenticated() } + +// CompareAPIVersion performs three-way comparison of API server version against +// the given one. +func CompareAPIVersion(c MultiAddressClient, v version.Version) int { + cvMsg := c.APIVersion() + cv := version.New(cvMsg.GetMajor(), cvMsg.GetMinor()) + return cv.Compare(v) +} diff --git a/pkg/services/object/common.go b/pkg/services/object/common.go index 0f1f75e163..5f537e2bb2 100644 --- a/pkg/services/object/common.go +++ b/pkg/services/object/common.go @@ -253,3 +253,23 @@ func (s *Server) _handleRequestMetaHeader(metaHdr *protosession.RequestMetaHeade return reqMD, nil } + +func newBadRequestStatus(msg string) *protostatus.Status { + return newStatus(protostatus.BadRequest, msg) +} + +func newInternalServerErrorStatus(msg string) *protostatus.Status { + return newStatus(protostatus.InternalServerError, msg) +} + +func newContainerNotFoundStatus(msg string) *protostatus.Status { + return newStatus(protostatus.ContainerNotFound, msg) +} + +func newAccessDeniedStatus(msg string) *protostatus.Status { + return newStatus(protostatus.ObjectAccessDenied, msg) +} + +func newStatus(code uint32, msg string) *protostatus.Status { + return &protostatus.Status{Code: code, Message: msg} +} diff --git a/pkg/services/object/put/distributed.go b/pkg/services/object/put/distributed.go index 263d6973a2..e5d4ef42d3 100644 --- a/pkg/services/object/put/distributed.go +++ b/pkg/services/object/put/distributed.go @@ -16,7 +16,9 @@ import ( "github.com/nspcc-dev/neo-go/pkg/core/transaction" iec "github.com/nspcc-dev/neofs-node/internal/ec" + iobject "github.com/nspcc-dev/neofs-node/internal/object" islices "github.com/nspcc-dev/neofs-node/internal/slices" + clientcore "github.com/nspcc-dev/neofs-node/pkg/core/client" netmapcore "github.com/nspcc-dev/neofs-node/pkg/core/netmap" objectcore "github.com/nspcc-dev/neofs-node/pkg/core/object" chaincontainer "github.com/nspcc-dev/neofs-node/pkg/morph/client/container" @@ -70,7 +72,6 @@ type distributedTarget struct { localStorage ObjectStorage clientConstructor ClientConstructor - transport Transport commonPrm *svcutil.CommonPrm keyStorage *svcutil.KeyStorage @@ -694,9 +695,20 @@ func (t *distributedTarget) sendObject(obj object.Object, encObj encodedObject, var sigsRaw []byte var err error if encObj.hdrOff > 0 { - sigsRaw, err = t.transport.SendReplicationRequestToNode(t.opCtx, encObj.b, node.info) - if err != nil { - err = fmt.Errorf("replicate object to remote node (key=%x): %w", node.info.PublicKey(), err) + var conn clientcore.MultiAddressClient + conn, err = t.clientConstructor.Get(t.opCtx, node.info) + if err == nil { + if clientcore.CompareAPIVersion(conn, iobject.ReplicateV2FirstAPIVersion) >= 0 { + payload := encObj.b[encObj.pldOff:] + sigsRaw, err = sendReplicationV2RequestToNode(t.opCtx, t.localNodeSigner, conn, obj, payload, t.metainfoConsistencyAttr != "") + } else { + sigsRaw, err = sendReplicationRequestToNode(t.opCtx, conn, encObj.b) + } + if err != nil { + err = fmt.Errorf("replicate object to remote node (key=%x): %w", node.info.PublicKey(), err) + } + } else { + err = fmt.Errorf("connect to remote node: %w", err) } } else { err = putObjectToNode(t.opCtx, node.info, &obj, t.keyStorage, t.clientConstructor, t.commonPrm) diff --git a/pkg/services/object/put/grpc.go b/pkg/services/object/put/grpc.go new file mode 100644 index 0000000000..6a8e37365c --- /dev/null +++ b/pkg/services/object/put/grpc.go @@ -0,0 +1,5 @@ +package putsvc + +import "google.golang.org/grpc/mem" + +var defaultGRPCBufferPool = mem.DefaultBufferPool() diff --git a/pkg/services/object/put/remote.go b/pkg/services/object/put/remote.go index bbe5bebc04..7d02e088d8 100644 --- a/pkg/services/object/put/remote.go +++ b/pkg/services/object/put/remote.go @@ -2,16 +2,28 @@ package putsvc import ( "context" + "errors" "fmt" "io" + "slices" + clientcore "github.com/nspcc-dev/neofs-node/pkg/core/client" "github.com/nspcc-dev/neofs-node/pkg/services/object/util" "github.com/nspcc-dev/neofs-sdk-go/client" + apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status" + neofscrypto "github.com/nspcc-dev/neofs-sdk-go/crypto" neofsecdsa "github.com/nspcc-dev/neofs-sdk-go/crypto/ecdsa" "github.com/nspcc-dev/neofs-sdk-go/netmap" "github.com/nspcc-dev/neofs-sdk-go/object" oid "github.com/nspcc-dev/neofs-sdk-go/object/id" + protoencoding "github.com/nspcc-dev/neofs-sdk-go/proto/encoding" + protoobject "github.com/nspcc-dev/neofs-sdk-go/proto/object" + "github.com/nspcc-dev/neofs-sdk-go/proto/protobuf" + protorefs "github.com/nspcc-dev/neofs-sdk-go/proto/refs" + protostatus "github.com/nspcc-dev/neofs-sdk-go/proto/status" "github.com/nspcc-dev/neofs-sdk-go/user" + "google.golang.org/grpc" + "google.golang.org/grpc/mem" ) // RemoteSender represents utility for @@ -22,6 +34,8 @@ type RemoteSender struct { clientConstructor ClientConstructor } +const maxReplicateV2PayloadChunkLen = 256 << 10 + func putObjectToNode(ctx context.Context, nodeInfo netmap.NodeInfo, obj *object.Object, keyStorage *util.KeyStorage, clientConstructor ClientConstructor, commonPrm *util.CommonPrm) error { var opts client.PrmObjectPutInit @@ -112,3 +126,120 @@ func (s *RemoteSender) ReplicateObjectToNode(ctx context.Context, id oid.ID, src return nil } + +func sendReplicationRequestToNode(ctx context.Context, conn clientcore.MultiAddressClient, req []byte) ([]byte, error) { + var res []byte + return res, conn.ForAnyGRPCConn(ctx, func(ctx context.Context, conn *grpc.ClientConn) error { + // this will be changed during NeoFS API Go deprecation. Code most likely be + // placed in SDK + var resp protoobject.ReplicateResponse + err := conn.Invoke(ctx, protoobject.ObjectService_Replicate_FullMethodName, mem.SliceBuffer(req), &resp, grpc.ForceCodecV2(protobuf.BufferedCodec{})) + if err != nil { + return newAPICallError(protoobject.ObjectService_Replicate_FullMethodName, err) + } + res, err = replicationResultFromResponse(&resp) + return err + }) +} + +func sendReplicationV2RequestToNode(ctx context.Context, signer neofscrypto.Signer, conn clientcore.MultiAddressClient, hdr object.Object, payload []byte, signObjectMeta bool) ([]byte, error) { + id := hdr.GetID() + + sig, err := signer.Sign(id[:]) + if err != nil { + return nil, fmt.Errorf("sign object ID: %w", err) + } + + hdrMsg := hdr.ProtoMessage() + hdrMsg.Payload = nil + + pubKey := neofscrypto.PublicKeyBytes(signer.Public()) + sigScheme := signer.Scheme() + + hdrLen := hdrMsg.MarshaledSize() + sigLen := protorefs.CalculateSignatureLength(pubKey, sig, sigScheme) + + initLen := protoobject.CalculateReplicateV2InitLength(hdrLen, sigLen, signObjectMeta) + + initReqLen := protoobject.CalculateReplicateV2InitRequestLength(initLen) + + initReqBufItem := defaultGRPCBufferPool.Get(initReqLen) + defer defaultGRPCBufferPool.Put(initReqBufItem) + + initReqBuf := *initReqBufItem + + writeHdrFn := protoencoding.WriteStablyMarshalledMessageFunc(hdrMsg) + writeSigFn := func(buf []byte) int { + return protorefs.WriteSignature(buf, pubKey, sig, sigScheme) + } + protoobject.WriteReplicateV2InitRequest(initReqBuf, hdrLen, writeHdrFn, sigLen, writeSigFn, signObjectMeta) + + var res []byte + + err = conn.ForAnyGRPCConn(ctx, func(ctx context.Context, conn *grpc.ClientConn) error { + stream, err := conn.NewStream(ctx, &grpc.StreamDesc{ClientStreams: true}, protoobject.ObjectService_ReplicateV2_FullMethodName, + grpc.ForceCodecV2(protobuf.BufferedCodec{}), + ) + if err != nil { + return newAPICallError(protoobject.ObjectService_ReplicateV2_FullMethodName, err) + } + + err = stream.SendMsg(mem.SliceBuffer(initReqBuf)) + if err != nil { + return fmt.Errorf("send initial request: %w", err) + } + + for chunk := range slices.Chunk(payload, maxReplicateV2PayloadChunkLen) { + reqLen := protoobject.CalculateReplicateV2ChunkRequestLength(chunk) + bufItem := defaultGRPCBufferPool.Get(reqLen) + protoobject.WriteReplicateV2ChunkRequest(*bufItem, chunk) + + err = stream.SendMsg(mem.NewBuffer(bufItem, defaultGRPCBufferPool)) + if err != nil { + if errors.Is(err, io.EOF) { + res, err = replicationV2ResultFromStream(stream) + return err + } + return fmt.Errorf("send chunk request: %w", err) + } + } + + err = stream.CloseSend() + if err != nil { + return fmt.Errorf("close stream: %w", err) + } + + res, err = replicationV2ResultFromStream(stream) + return err + }) + + return res, err +} + +func newAPICallError(method string, cause error) error { + return fmt.Errorf("API transport (op=%s): %w", method, cause) +} + +func replicationV2ResultFromStream(stream grpc.ClientStream) ([]byte, error) { + var resp protoobject.ReplicateV2Response + + err := stream.RecvMsg(&resp) + if err != nil { + return nil, fmt.Errorf("receive message from stream: %w", err) + } + + return handleReplicationResultFromResponse(resp.Status, resp.ObjectSignature) +} + +func replicationResultFromResponse(m *protoobject.ReplicateResponse) ([]byte, error) { + return handleReplicationResultFromResponse(m.GetStatus(), m.GetObjectSignature()) +} + +func handleReplicationResultFromResponse(st *protostatus.Status, sig []byte) ([]byte, error) { + err := apistatus.ToError(st) + if err != nil { + return nil, err + } + + return sig, nil +} diff --git a/pkg/services/object/put/service.go b/pkg/services/object/put/service.go index 6ce7f47295..322b117d4a 100644 --- a/pkg/services/object/put/service.go +++ b/pkg/services/object/put/service.go @@ -49,19 +49,11 @@ type MaxSizeSource interface { type Service struct { *cfg - transport Transport - neoFSNet NeoFSNetwork + neoFSNet NeoFSNetwork } type Option func(*cfg) -// Transport provides message transmission over NeoFS network. -type Transport interface { - // SendReplicationRequestToNode sends a prepared replication request message to - // the specified remote node. - SendReplicationRequestToNode(ctx context.Context, req []byte, node netmap.NodeInfo) ([]byte, error) -} - type ClientConstructor interface { Get(context.Context, netmap.NodeInfo) (clientcore.MultiAddressClient, error) } @@ -157,7 +149,7 @@ func defaultCfg() *cfg { } } -func NewService(transport Transport, neoFSNet NeoFSNetwork, m *meta.Meta, q QuotaLimiter, p PaymentChecker, opts ...Option) *Service { +func NewService(neoFSNet NeoFSNetwork, m *meta.Meta, q QuotaLimiter, p PaymentChecker, opts ...Option) *Service { c := defaultCfg() for i := range opts { @@ -175,18 +167,16 @@ func NewService(transport Transport, neoFSNet NeoFSNetwork, m *meta.Meta, q Quot c.payments = p return &Service{ - cfg: c, - transport: transport, - neoFSNet: neoFSNet, + cfg: c, + neoFSNet: neoFSNet, } } func (p *Service) Put(ctx context.Context) (*Streamer, error) { return &Streamer{ - cfg: p.cfg, - ctx: ctx, - transport: p.transport, - neoFSNet: p.neoFSNet, + cfg: p.cfg, + ctx: ctx, + neoFSNet: p.neoFSNet, }, nil } diff --git a/pkg/services/object/put/service_test.go b/pkg/services/object/put/service_test.go index 7608f4aa45..9ea9746ab6 100644 --- a/pkg/services/object/put/service_test.go +++ b/pkg/services/object/put/service_test.go @@ -9,6 +9,7 @@ import ( "fmt" "io" "math" + "net" "slices" "strconv" "strings" @@ -45,6 +46,7 @@ import ( objecttest "github.com/nspcc-dev/neofs-sdk-go/object/test" protoobject "github.com/nspcc-dev/neofs-sdk-go/proto/object" protorefs "github.com/nspcc-dev/neofs-sdk-go/proto/refs" + protostatus "github.com/nspcc-dev/neofs-sdk-go/proto/status" "github.com/nspcc-dev/neofs-sdk-go/reputation" "github.com/nspcc-dev/neofs-sdk-go/session" sessionv2 "github.com/nspcc-dev/neofs-sdk-go/session/v2" @@ -56,6 +58,8 @@ import ( "go.uber.org/zap" "go.uber.org/zap/zaptest" "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/encoding/protowire" "google.golang.org/protobuf/proto" ) @@ -106,7 +110,7 @@ func TestPayments(t *testing.T) { p.SetReplicas([]netmap.ReplicaDescriptor{rep}) cnr.SetPlacementPolicy(p) - s := NewService(cluster.nodeServices, &cluster.nodeNetworks[0], nil, + s := NewService(&cluster.nodeNetworks[0], nil, quotas{math.MaxUint64, math.MaxUint64}, payments, WithLogger(zaptest.NewLogger(t)), @@ -195,7 +199,7 @@ func TestQuotas(t *testing.T) { p.SetReplicas([]netmap.ReplicaDescriptor{rep}) cnr.SetPlacementPolicy(p) - s := NewService(cluster.nodeServices, &cluster.nodeNetworks[0], nil, + s := NewService(&cluster.nodeNetworks[0], nil, quotas{hard: hardLimit}, &payments{}, WithLogger(zaptest.NewLogger(t)), @@ -1001,7 +1005,7 @@ func newTestClusterForRepPolicyWithContainer(t *testing.T, repNodes, cnrReserveN expiresAt: cluster.nodeNetworks[i].epoch + 1, } - cluster.nodeServices[i] = NewService(cluster.nodeServices, &cluster.nodeNetworks[i], nil, + cluster.nodeServices[i] = newServiceClient(t, NewService(&cluster.nodeNetworks[i], nil, quotas{math.MaxUint64, math.MaxUint64}, &payments{}, WithSessionsCache(isessions.NewObjectSessionsCache(1)), @@ -1015,7 +1019,7 @@ func newTestClusterForRepPolicyWithContainer(t *testing.T, repNodes, cnrReserveN WithSplitChainVerifier(mockSplitVerifier{}), WithPostPlacementReplicator(mockPostPlacementReplicator{}), WithTombstoneVerifier(mockTombstoneVerifier{}), - ) + )) } return &cluster @@ -1151,9 +1155,9 @@ func (x testPostPlacementReplicator) HandlePostPlacement(obj *object.Object, nod } for i := range nodes { - svc, err := x.services.lookupNode(nodes[i]) + ns, err := x.services.lookupNode(nodes[i]) require.NoError(x.t, err) - require.NoError(x.t, svc.ValidateAndStoreObjectLocally(context.Background(), *obj)) + require.NoError(x.t, ns.svc.ValidateAndStoreObjectLocally(context.Background(), *obj)) } } @@ -1206,11 +1210,11 @@ func (x *inMemLocalStorage) IsLocked(context.Context, oid.Address) (bool, error) panic("unimplemented") } -type nodeServices []*Service +type nodeServices []*serviceClient -func (x nodeServices) lookupNode(node netmap.NodeInfo) (*Service, error) { - ind := slices.IndexFunc(x, func(svc *Service) bool { - return svc.neoFSNet.IsLocalNodePublicKey(node.PublicKey()) +func (x nodeServices) lookupNode(node netmap.NodeInfo) (*serviceClient, error) { + ind := slices.IndexFunc(x, func(c *serviceClient) bool { + return c.svc.neoFSNet.IsLocalNodePublicKey(node.PublicKey()) }) if ind < 0 { return nil, errors.New("unknown node") @@ -1219,48 +1223,122 @@ func (x nodeServices) lookupNode(node netmap.NodeInfo) (*Service, error) { } func (x nodeServices) Get(_ context.Context, node netmap.NodeInfo) (clientcore.MultiAddressClient, error) { - svc, err := x.lookupNode(node) + return x.lookupNode(node) +} + +type testObjectServiceServer struct { + protoobject.UnimplementedObjectServiceServer + svc *Service +} + +func (x testObjectServiceServer) ReplicateV2(stream protoobject.ObjectService_ReplicateV2Server) error { + firstReq, err := stream.Recv() if err != nil { - return nil, err + return err + } + + reqInit, ok := firstReq.StreamPart.(*protoobject.ReplicateV2Request_Init_) + if !ok { + resp := &protoobject.ReplicateV2Response{ + Status: &protostatus.Status{ + Code: protostatus.BadRequest, Message: "first request does not contain init field", + }, + } + return stream.SendAndClose(resp) } - return (*serviceClient)(svc), nil -} -func (x nodeServices) SendReplicationRequestToNode(ctx context.Context, reqBin []byte, node netmap.NodeInfo) ([]byte, error) { - var req protoobject.ReplicateRequest - if err := proto.Unmarshal(reqBin, &req); err != nil { - return nil, fmt.Errorf("invalid request: %w", err) + initPart := reqInit.Init + if initPart == nil { // not expected to ever happen, but better to keep safe + resp := &protoobject.ReplicateV2Response{ + Status: &protostatus.Status{ + Code: protostatus.InternalServerError, Message: "first request contains nil init field", + }, + } + return stream.SendAndClose(resp) } - if req.Object == nil { - return nil, errors.New("missing object in request") + if initPart.Object == nil { + resp := &protoobject.ReplicateV2Response{ + Status: &protostatus.Status{ + Code: protostatus.BadRequest, Message: "object field is missing", + }, + } + return stream.SendAndClose(resp) } var obj object.Object - if err := obj.FromProtoMessage(req.Object); err != nil { - return nil, fmt.Errorf("invalid object in request: %w", err) + if err := obj.FromProtoMessage(initPart.Object); err != nil { + resp := &protoobject.ReplicateV2Response{ + Status: &protostatus.Status{Code: protostatus.BadRequest, Message: fmt.Sprintf("invalid object in request: %v", err)}, + } + return stream.SendAndClose(resp) } - svc, err := x.lookupNode(node) - if err != nil { - return nil, err + for { + req, err := stream.Recv() + if err != nil { + if errors.Is(err, io.EOF) { + break + } + return err + } + + chunkPart, ok := req.StreamPart.(*protoobject.ReplicateV2Request_PayloadChunk) + if !ok { + resp := &protoobject.ReplicateV2Response{ + Status: &protostatus.Status{Code: protostatus.BadRequest, Message: "non-chunk subsequent message"}, + } + return stream.SendAndClose(resp) + } + + obj.SetPayload(append(obj.Payload(), chunkPart.PayloadChunk...)) } - if err := svc.ValidateAndStoreObjectLocally(ctx, obj); err != nil { - return nil, fmt.Errorf("validate and store object locally: %w", err) + if err := x.svc.ValidateAndStoreObjectLocally(stream.Context(), obj); err != nil { + resp := &protoobject.ReplicateV2Response{ + Status: &protostatus.Status{Code: protostatus.InternalServerError, Message: fmt.Sprintf("validate and store object locally: %v", err)}, + } + return stream.SendAndClose(resp) } - return nil, nil + return stream.SendAndClose(new(protoobject.ReplicateV2Response)) } -type serviceClient Service +type serviceClient struct { + svc *Service + grpcConn *grpc.ClientConn +} + +func newServiceClient(t *testing.T, svc *Service) *serviceClient { + srv := grpc.NewServer() + t.Cleanup(srv.GracefulStop) + + protoobject.RegisterObjectServiceServer(srv, testObjectServiceServer{svc: svc}) + + bufConn := bufconn.Listen(100 << 10) + + go func() { _ = srv.Serve(bufConn) }() + + grpcConn, err := grpc.NewClient("localhost:8080", // any + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return bufConn.DialContext(ctx) + }), + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + require.NoError(t, err) + + return &serviceClient{ + svc: svc, + grpcConn: grpcConn, + } +} func (m *serviceClient) APIVersion() *protorefs.Version { return version.Current().ProtoMessage() } func (m *serviceClient) ObjectPutInit(ctx context.Context, hdr object.Object, _ user.Signer, _ client.PrmObjectPutInit) (client.ObjectWriter, error) { - stream, err := (*Service)(m).Put(ctx) + stream, err := m.svc.Put(ctx) if err != nil { return nil, err } @@ -1294,7 +1372,7 @@ func (m *serviceClient) ReplicateObject(ctx context.Context, _ oid.ID, src io.Re if err := obj.FromProtoMessage(&msg); err != nil { return nil, err } - return nil, (*Service)(m).ValidateAndStoreObjectLocally(ctx, obj) + return nil, m.svc.ValidateAndStoreObjectLocally(ctx, obj) } func (m *serviceClient) ObjectDelete(context.Context, cid.ID, oid.ID, user.Signer, client.PrmObjectDelete) (oid.ID, error) { @@ -1330,8 +1408,8 @@ func (m *serviceClient) AnnounceIntermediateTrust(context.Context, uint64, reput panic("unimplemented") } -func (m *serviceClient) ForAnyGRPCConn(context.Context, func(context.Context, *grpc.ClientConn) error) error { - panic("unimplemented") +func (m *serviceClient) ForAnyGRPCConn(ctx context.Context, fn func(context.Context, *grpc.ClientConn) error) error { + return fn(ctx, m.grpcConn) } type testPayloadStream Streamer @@ -1453,11 +1531,11 @@ func newPlacementTestEnv(t *testing.T, cnr container.Container, repRules []uint, }) } - cluster.nodeServices[nodeIdx] = NewService(cluster.nodeServices, &cluster.nodeNetworks[nodeIdx], nil, + cluster.nodeServices[nodeIdx] = newServiceClient(t, NewService(&cluster.nodeNetworks[nodeIdx], nil, quotas{math.MaxUint64, math.MaxUint64}, &payments{}, opts..., - ) + )) nodeIdx++ } @@ -1517,12 +1595,12 @@ func newSessionTokenV2ForNode(t *testing.T, cluster testCluster, nodeLists [][]n return sessionTokenV2 } -func storeObjectWithSession(t *testing.T, svc *Service, obj object.Object, st *session.Object, st2 *sessionv2.Token) { - require.NoError(t, putObjectWithSession(svc, obj, st, st2)) +func storeObjectWithSession(t *testing.T, c *serviceClient, obj object.Object, st *session.Object, st2 *sessionv2.Token) { + require.NoError(t, putObjectWithSession(c, obj, st, st2)) } -func putObjectWithSession(svc *Service, obj object.Object, st *session.Object, st2 *sessionv2.Token) error { - stream, err := svc.Put(context.Background()) +func putObjectWithSession(c *serviceClient, obj object.Object, st *session.Object, st2 *sessionv2.Token) error { + stream, err := c.svc.Put(context.Background()) if err != nil { return fmt.Errorf("init stream: %w", err) } @@ -2352,7 +2430,7 @@ func testInitialPlacement(t *testing.T, repRules []uint, ecRules []iec.Rule, ip expiresAt: cluster.nodeNetworks[nodeIdx].epoch + 1, } - cluster.nodeServices[nodeIdx] = NewService(cluster.nodeServices, &cluster.nodeNetworks[nodeIdx], nil, + cluster.nodeServices[nodeIdx] = newServiceClient(t, NewService(&cluster.nodeNetworks[nodeIdx], nil, quotas{math.MaxUint64, math.MaxUint64}, &payments{}, WithSessionsCache(isessions.NewObjectSessionsCache(1)), @@ -2366,7 +2444,7 @@ func testInitialPlacement(t *testing.T, repRules []uint, ecRules []iec.Rule, ip WithSplitChainVerifier(mockSplitVerifier{}), WithPostPlacementReplicator(mockPostPlacementReplicator{}), WithTombstoneVerifier(mockTombstoneVerifier{}), - ) + )) nodeIdx++ } diff --git a/pkg/services/object/put/streamer.go b/pkg/services/object/put/streamer.go index 5528bbb7ab..1896daa283 100644 --- a/pkg/services/object/put/streamer.go +++ b/pkg/services/object/put/streamer.go @@ -34,8 +34,7 @@ type Streamer struct { maxPayloadSz uint64 // network config - transport Transport - neoFSNet NeoFSNetwork + neoFSNet NeoFSNetwork } var errNotInit = errors.New("stream not initialized") @@ -281,7 +280,6 @@ func (p *Streamer) newDistrubutedWriter(prm *PutInitPrm) *distributedTarget { keyStorage: p.keyStorage, commonPrm: prm.common, clientConstructor: p.clientConstructor, - transport: p.transport, relay: relay, fmt: p.fmtValidator, containerNodes: prm.containerNodes, diff --git a/pkg/services/object/server.go b/pkg/services/object/server.go index 4421f23be5..9f721e59c9 100644 --- a/pkg/services/object/server.go +++ b/pkg/services/object/server.go @@ -72,14 +72,6 @@ type Handlers interface { Delete(context.Context, deletesvc.Prm) error } -// Various NeoFS protocol status codes. -const ( - codeInternal = uint32(1024*protostatus.Section_SECTION_FAILURE_COMMON) + uint32(protostatus.CommonFail_INTERNAL) - codeBadRequest = uint32(1024*protostatus.Section_SECTION_FAILURE_COMMON) + uint32(protostatus.CommonFail_BAD_REQUEST) - codeAccessDenied = uint32(1024*protostatus.Section_SECTION_OBJECT) + uint32(protostatus.Object_ACCESS_DENIED) - codeContainerNotFound = uint32(1024*protostatus.Section_SECTION_CONTAINER) + uint32(protostatus.Container_CONTAINER_NOT_FOUND) -) - // MetricCollector tracks exec statistics for the following ops: // - [stat.MethodObjectPut] // - [stat.MethodObjectGet] @@ -1557,111 +1549,154 @@ func (s *Server) Search(_ *protoobject.SearchRequest, _ protoobject.ObjectServic return grpcstatus.Error(grpccodes.Unimplemented, "no longer supported, use SearchV2") } +func readFirstReplicateV2Request(stream protoobject.ObjectService_ReplicateV2Server) (*protoobject.ReplicateV2Request_Init, *protostatus.Status, error) { + firstReq, err := stream.Recv() + if err != nil { + return nil, nil, err + } + + reqInit, ok := firstReq.StreamPart.(*protoobject.ReplicateV2Request_Init_) + if !ok { + return nil, newBadRequestStatus("first request does not contain init field"), nil + } + + initPart := reqInit.Init + if initPart == nil { // not expected to ever happen, but better to keep safe + return nil, newInternalServerErrorStatus("first request contains nil init field"), nil + } + + if initPart.Object == nil { + return nil, newBadRequestStatus("object field is missing"), nil + } + + if len(initPart.Object.Payload) > 0 { + return nil, newBadRequestStatus("non-empty object payload in init field"), nil + } + + return initPart, nil, nil +} + +// ReplicateV2 serves neo.fs.v2.object.ObjectService/ReplicateV2 RPC. +func (s *Server) ReplicateV2(stream protoobject.ObjectService_ReplicateV2Server) error { + initPart, st, err := readFirstReplicateV2Request(stream) + if err != nil { + return err + } + if st != nil { + return stream.SendAndClose(&protoobject.ReplicateV2Response{Status: st}) + } + + recvChunkFn := func() ([]byte, *protostatus.Status, error) { + req, err := stream.Recv() + if err != nil { + return nil, nil, err + } + chunkPart, ok := req.StreamPart.(*protoobject.ReplicateV2Request_PayloadChunk) + if !ok { + return nil, newBadRequestStatus("non-chunk subsequent message"), nil + } + return chunkPart.PayloadChunk, nil, nil + } + + objSig, st, err := s.replicate(stream.Context(), initPart.Object, initPart.Signature, initPart.SignObject, recvChunkFn) + if err != nil { + return err + } + + resp := &protoobject.ReplicateV2Response{ + Status: st, + ObjectSignature: objSig, + } + return stream.SendAndClose(resp) +} + // Replicate serves neo.fs.v2.object.ObjectService/Replicate RPC. func (s *Server) Replicate(ctx context.Context, req *protoobject.ReplicateRequest) (*protoobject.ReplicateResponse, error) { if req.Object == nil { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, Message: "binary object field is missing/empty", - }}, nil + return &protoobject.ReplicateResponse{ + Status: newBadRequestStatus("binary object field is missing/empty"), + }, nil } - if req.Object.ObjectId == nil || len(req.Object.ObjectId.Value) == 0 { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, Message: "ID field is missing/empty in the object field", - }}, nil + objSig, st, err := s.replicate(ctx, req.Object, req.Signature, req.SignObject, func() ([]byte, *protostatus.Status, error) { + return nil, nil, io.EOF + }) + if err != nil { + return nil, err + } + + return &protoobject.ReplicateResponse{ + Status: st, + ObjectSignature: objSig, + }, nil +} + +func (s *Server) replicate(ctx context.Context, objMsg *protoobject.Object, sig *refs.Signature, signObject bool, recvChunkFn func() ([]byte, *protostatus.Status, error)) ([]byte, *protostatus.Status, error) { + if objMsg.ObjectId == nil || len(objMsg.ObjectId.Value) == 0 { + return nil, newBadRequestStatus("ID field is missing/empty in the object field"), nil } - if req.Signature == nil { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, Message: "missing object signature field", - }}, nil + if sig == nil { + return nil, newBadRequestStatus("missing object signature field"), nil } - if len(req.Signature.Key) == 0 { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, Message: "public key field is missing/empty in the object signature field", - }}, nil + if len(sig.Key) == 0 { + return nil, newBadRequestStatus("public key field is missing/empty in the object signature field"), nil } - if len(req.Signature.Sign) == 0 { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, Message: "signature value is missing/empty in the object signature field", - }}, nil + if len(sig.Sign) == 0 { + return nil, newBadRequestStatus("signature value is missing/empty in the object signature field"), nil } - switch scheme := req.Signature.Scheme; scheme { + switch scheme := sig.Scheme; scheme { default: - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, - Message: "unsupported scheme in the object signature field", - }}, nil + return nil, newBadRequestStatus("unsupported scheme in the object signature field"), nil case refs.SignatureScheme_ECDSA_SHA512, refs.SignatureScheme_ECDSA_RFC6979_SHA256, refs.SignatureScheme_ECDSA_RFC6979_SHA256_WALLET_CONNECT: } - hdr := req.Object.GetHeader() + hdr := objMsg.GetHeader() if hdr == nil { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, - Message: "missing header field in the object field", - }}, nil + return nil, newBadRequestStatus("missing header field in the object field"), nil } gCnrMsg := hdr.GetContainerId() if gCnrMsg == nil { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, - Message: "missing container ID field in the object header field", - }}, nil + return nil, newBadRequestStatus("missing container ID field in the object header field"), nil } var cnr cid.ID err := cnr.FromProtoMessage(gCnrMsg) if err != nil { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, - Message: fmt.Sprintf("invalid container ID in the object header field: %v", err), - }}, nil + return nil, newBadRequestStatus(fmt.Sprintf("invalid container ID in the object header field: %v", err)), nil } var pubKey neofscrypto.PublicKey - switch req.Signature.Scheme { //nolint:exhaustive + switch sig.Scheme { //nolint:exhaustive // other cases already checked above case refs.SignatureScheme_ECDSA_SHA512: pubKey = new(neofsecdsa.PublicKey) - err = pubKey.Decode(req.Signature.Key) + err = pubKey.Decode(sig.Key) if err != nil { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, - Message: "invalid ECDSA public key in the object signature field", - }}, nil + return nil, newBadRequestStatus("invalid ECDSA public key in the object signature field"), nil } case refs.SignatureScheme_ECDSA_RFC6979_SHA256: pubKey = new(neofsecdsa.PublicKeyRFC6979) - err = pubKey.Decode(req.Signature.Key) + err = pubKey.Decode(sig.Key) if err != nil { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, - Message: "invalid ECDSA public key in the object signature field", - }}, nil + return nil, newBadRequestStatus("invalid ECDSA public key in the object signature field"), nil } case refs.SignatureScheme_ECDSA_RFC6979_SHA256_WALLET_CONNECT: pubKey = new(neofsecdsa.PublicKeyWalletConnect) - err = pubKey.Decode(req.Signature.Key) + err = pubKey.Decode(sig.Key) if err != nil { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, - Message: "invalid ECDSA public key in the object signature field", - }}, nil + return nil, newBadRequestStatus("invalid ECDSA public key in the object signature field"), nil } } - if !pubKey.Verify(req.Object.ObjectId.Value, req.Signature.Sign) { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, - Message: "signature mismatch in the object signature field", - }}, nil + if !pubKey.Verify(objMsg.ObjectId.Value, sig.Sign) { + return nil, newBadRequestStatus("signature mismatch in the object signature field"), nil } var serverInCnr bool @@ -1671,77 +1706,75 @@ func (s *Server) Replicate(ctx context.Context, req *protoobject.ReplicateReques }) if err != nil { if errors.Is(err, apistatus.ErrContainerNotFound) { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeContainerNotFound, - Message: "failed to check server's compliance to object's storage policy: object's container not found", - }}, nil + return nil, newContainerNotFoundStatus("failed to check server's compliance to object's storage policy: object's container not found"), nil } - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeInternal, - Message: fmt.Sprintf("failed to apply object's storage policy: %v", err), - }}, nil + return nil, newInternalServerErrorStatus(fmt.Sprintf("failed to apply object's storage policy: %v", err)), nil } else if !serverInCnr { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeAccessDenied, Message: "server does not match the object's storage policy", - }}, nil + return nil, newAccessDeniedStatus("server does not match the object's storage policy"), nil } var clientInCnr bool err = s.fsChain.ForEachContainerNodePublicKeyInLastTwoEpochs(cnr, func(pubKey []byte) bool { - clientInCnr = bytes.Equal(pubKey, req.Signature.Key) + clientInCnr = bytes.Equal(pubKey, sig.Key) return !clientInCnr }) if err != nil { if errors.Is(err, apistatus.ErrContainerNotFound) { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeContainerNotFound, - Message: "failed to check server's compliance to object's storage policy: object's container not found", - }}, nil + return nil, newContainerNotFoundStatus("failed to check server's compliance to object's storage policy: object's container not found"), nil } - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeInternal, - Message: fmt.Sprintf("failed to apply object's storage policy: %v", err), - }}, nil + return nil, newInternalServerErrorStatus(fmt.Sprintf("failed to apply object's storage policy: %v", err)), nil } else if !clientInCnr { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeAccessDenied, Message: "client does not match the object's storage policy", - }}, nil + return nil, newAccessDeniedStatus("client does not match the object's storage policy"), nil } // TODO(@cthulhu-rider): avoid decoding the object completely - obj, err := objectFromMessage(req.Object) + obj, err := objectFromMessage(objMsg) if err != nil { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeBadRequest, - Message: fmt.Sprintf("invalid object field: %v", err), - }}, nil + return nil, newBadRequestStatus(fmt.Sprintf("invalid object field: %v", err)), nil + } + + // TODO: avoid full buffering, copy to local storage stream directly instead + payload := slices.Grow(objMsg.Payload, int(hdr.PayloadLength)-len(objMsg.Payload)) + for { + chunk, st, err := recvChunkFn() + if err != nil { + if errors.Is(err, io.EOF) { + break + } + return nil, nil, err + } + if st != nil { + return nil, st, nil + } + + if len(chunk) == 0 { + return nil, newBadRequestStatus("empty payload chunk"), nil + } + + payload = append(payload, chunk...) } + obj.SetPayload(payload) err = s.storage.VerifyAndStoreObjectLocally(ctx, *obj) if err != nil { if errors.Is(err, apistatus.ErrBusy) { - return &protoobject.ReplicateResponse{Status: apistatus.FromError(err)}, nil + return nil, apistatus.FromError(err), nil } - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeInternal, - Message: fmt.Sprintf("failed to verify and store object locally: %v", err), - }}, nil + return nil, newInternalServerErrorStatus(fmt.Sprintf("failed to verify and store object locally: %v", err)), nil } - resp := new(protoobject.ReplicateResponse) - if req.GetSignObject() { - resp.ObjectSignature, err = s.metaInfoSignature(*obj) + var objSig []byte + if signObject { + objSig, err = s.metaInfoSignature(*obj) if err != nil { - return &protoobject.ReplicateResponse{Status: &protostatus.Status{ - Code: codeInternal, - Message: fmt.Sprintf("failed to sign object meta information: %v", err), - }}, nil + return nil, newInternalServerErrorStatus(fmt.Sprintf("failed to sign object meta information: %v", err)), nil } } - return resp, nil + // nil status corresponds to OK + return objSig, nil, nil } func (s *Server) signSearchResponse(body *protoobject.SearchV2Response_Body, err error, req *protoobject.SearchV2Request) *protoobject.SearchV2Response {