Initial support for ReplicateV2

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

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),

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

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=

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)

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)
+}

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}
+}

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)

pkg/services/object/put/grpc.go

@@ -0,0 +1,5 @@
+package putsvc
+
+import "google.golang.org/grpc/mem"
+
+var defaultGRPCBufferPool = mem.DefaultBufferPool()

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
+}