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>

CHANGELOG.md

@@ -24,6 +24,7 @@ Changelog for NeoFS Node
 - Storage nodes no longer accept objects with header larger than 16KB (#2749)
 - IR sends NeoFS chain GAS to netmap nodes every epoch, not per a configurable blocks number (#2777)
 - Big objects are split with the new split scheme (#2667)
+- Background replicator transfers objects using new `ObjectService.Replicate` RPC (#2317)
 
 ### Removed
 - Object notifications incl. NATS (#2750)

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, id oid.ID, 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"
 
@@ -14,6 +15,7 @@ import (
 	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
 	"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"
@@ -237,6 +239,24 @@ func (x *multiClient) ObjectPutInit(ctx context.Context, header objectSDK.Object
 	return
 }
 
+func (x *multiClient) ReplicateObject(ctx context.Context, id oid.ID, src io.ReadSeeker, signer neofscrypto.Signer) error {
+	var errSeek error
+	err := x.iterateClients(ctx, func(c clientcore.Client) error {
+		err := c.ReplicateObject(ctx, id, 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, id oid.ID, 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, id, 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 = 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, task.addr.Object(), 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()