replicator: Send local objects using new replication API

It's more lightweight and supports binary copying without additional
decode-encode round.

Based on the fact that if the object is fixed, the request remains
unchanged. According to this, transport message is encoded once and sent
to all nodes.

Closes #2317. Refs #2316.

Signed-off-by: Leonard Lyubich <leonard@morphbits.io>

pkg/core/client/client.go

@@ -2,12 +2,14 @@ package client
 
 import (
 	"context"
+	"io"
 
 	rawclient "github.com/nspcc-dev/neofs-api-go/v2/rpc/client"
 	"github.com/nspcc-dev/neofs-node/pkg/network"
 	"github.com/nspcc-dev/neofs-sdk-go/client"
 	"github.com/nspcc-dev/neofs-sdk-go/container"
 	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"
 	reputationSDK "github.com/nspcc-dev/neofs-sdk-go/reputation"
@@ -19,6 +21,7 @@ import (
 type Client interface {
 	ContainerAnnounceUsedSpace(ctx context.Context, announcements []container.SizeEstimation, prm client.PrmAnnounceSpace) error
 	ObjectPutInit(ctx context.Context, header object.Object, signer user.Signer, prm client.PrmObjectPutInit) (client.ObjectWriter, error)
+	ReplicateObject(ctx context.Context, src io.ReadSeeker, signer neofscrypto.Signer) error
 	ObjectDelete(ctx context.Context, containerID cid.ID, objectID oid.ID, signer user.Signer, prm client.PrmObjectDelete) (oid.ID, error)
 	ObjectGetInit(ctx context.Context, containerID cid.ID, objectID oid.ID, signer user.Signer, prm client.PrmObjectGet) (object.Object, *client.PayloadReader, error)
 	ObjectHead(ctx context.Context, containerID cid.ID, objectID oid.ID, signer user.Signer, prm client.PrmObjectHead) (*object.Object, error)

pkg/network/cache/multi.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"sync"
 	"time"
 
@@ -13,6 +14,7 @@ import (
 	"github.com/nspcc-dev/neofs-sdk-go/client"
 	"github.com/nspcc-dev/neofs-sdk-go/container"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
+	neofscrypto "github.com/nspcc-dev/neofs-sdk-go/crypto"
 	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
 	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
 	reputationSDK "github.com/nspcc-dev/neofs-sdk-go/reputation"
@@ -231,6 +233,24 @@ func (x *multiClient) ObjectPutInit(ctx context.Context, header objectSDK.Object
 	return
 }
 
+func (x *multiClient) ReplicateObject(ctx context.Context, src io.ReadSeeker, signer neofscrypto.Signer) error {
+	var errSeek error
+	err := x.iterateClients(ctx, func(c clientcore.Client) error {
+		err := c.ReplicateObject(ctx, src, signer)
+		if err != nil {
+			_, errSeek = src.Seek(0, io.SeekStart)
+			if errSeek != nil {
+				return nil // to break the iterator
+			}
+		}
+		return err
+	})
+	if err != nil {
+		return err
+	}
+	return errSeek
+}
+
 func (x *multiClient) ContainerAnnounceUsedSpace(ctx context.Context, announcements []container.SizeEstimation, prm client.PrmAnnounceSpace) error {
 	return x.iterateClients(ctx, func(c clientcore.Client) error {
 		return c.ContainerAnnounceUsedSpace(ctx, announcements, prm)

pkg/services/object/put/remote.go

@@ -3,12 +3,14 @@ package putsvc
 import (
 	"context"
 	"fmt"
+	"io"
 
 	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"
 	internalclient "github.com/nspcc-dev/neofs-node/pkg/services/object/internal/client"
 	"github.com/nspcc-dev/neofs-node/pkg/services/object/util"
+	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"
@@ -134,3 +136,32 @@ func (s *RemoteSender) PutObject(ctx context.Context, p *RemotePutPrm) error {
 
 	return nil
 }
+
+// ReplicateObjectToNode copies binary-encoded NeoFS object from the given
+// [io.ReadSeeker] into local storage of the node described by specified
+// [netmap.NodeInfo].
+func (s *RemoteSender) ReplicateObjectToNode(ctx context.Context, src io.ReadSeeker, nodeInfo netmap.NodeInfo) error {
+	var nodeInfoForCons clientcore.NodeInfo
+
+	err := clientcore.NodeInfoFromRawNetmapElement(&nodeInfoForCons, netmapCore.Node(nodeInfo))
+	if err != nil {
+		return fmt.Errorf("parse remote node info: %w", err)
+	}
+
+	key, err := s.keyStorage.GetKey(nil)
+	if err != nil {
+		return fmt.Errorf("fetch local node's private key: %w", err)
+	}
+
+	c, err := s.clientConstructor.Get(nodeInfoForCons)
+	if err != nil {
+		return fmt.Errorf("init NeoFS API client of the remote node: %w", err)
+	}
+
+	err = c.ReplicateObject(ctx, src, (*neofsecdsa.Signer)(key))
+	if err != nil {
+		return fmt.Errorf("copy object using NeoFS API client of the remote node: %w", err)
+	}
+
+	return nil
+}

pkg/services/replicator/process.go

@@ -1,10 +1,12 @@
 package replicator
 
 import (
+	"bytes"
 	"context"
+	"io"
 
-	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/engine"
 	putsvc "github.com/nspcc-dev/neofs-node/pkg/services/object/put"
+	"github.com/nspcc-dev/neofs-sdk-go/client"
 	"github.com/nspcc-dev/neofs-sdk-go/netmap"
 	"go.uber.org/zap"
 )
@@ -25,21 +27,27 @@ func (p *Replicator) HandleTask(ctx context.Context, task Task, res TaskResult)
 		)
 	}()
 
-	if task.obj == nil {
-		var err error
-		task.obj, err = engine.Get(p.localStorage, task.addr)
+	var err error
+	var prm *putsvc.RemotePutPrm
+	var stream io.ReadSeeker
+	binReplication := task.obj == nil
+	if binReplication {
+		b, err := p.localStorage.GetBytes(task.addr)
 		if err != nil {
 			p.log.Error("could not get object from local storage",
 				zap.Stringer("object", task.addr),
 				zap.Error(err))
 
 			return
 		}
+		stream = io.ReadSeeker(bytes.NewReader(b))
+		if len(task.nodes) > 1 {
+			stream = client.DemuxReplicatedObject(stream)
+		}
+	} else {
+		prm = new(putsvc.RemotePutPrm).WithObject(task.obj)
 	}
 
-	prm := new(putsvc.RemotePutPrm).
-		WithObject(task.obj)
-
 	for i := 0; task.quantity > 0 && i < len(task.nodes); i++ {
 		select {
 		case <-ctx.Done():
@@ -54,7 +62,13 @@ func (p *Replicator) HandleTask(ctx context.Context, task Task, res TaskResult)
 
 		callCtx, cancel := context.WithTimeout(ctx, p.putTimeout)
 
-		err := p.remoteSender.PutObject(callCtx, prm.WithNodeInfo(task.nodes[i]))
+		if binReplication {
+			err = p.remoteSender.ReplicateObjectToNode(callCtx, stream, task.nodes[i])
+			// note that we don't need to reset stream because it is used exactly once
+			// according to the client.DemuxReplicatedObject above
+		} else {
+			err = p.remoteSender.PutObject(callCtx, prm.WithNodeInfo(task.nodes[i]))
+		}
 
 		cancel()
 
@@ -63,7 +77,7 @@ func (p *Replicator) HandleTask(ctx context.Context, task Task, res TaskResult)
 				zap.String("error", err.Error()),
 			)
 		} else {
-			log.Debug("object successfully replicated")
+			log.Info("object successfully replicated")
 
 			task.quantity--