WIP: Encode object into transport message once per replication round

Signed-off-by: Leonard Lyubich <[email protected]>

pkg/services/replicator/process.go

@@ -2,9 +2,11 @@ package replicator
 
 import (
 	"context"
+	"errors"
 	"io"
 
 	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"
 )
@@ -16,6 +18,34 @@ type TaskResult interface {
 	SubmitSuccessfulReplication(netmap.NodeInfo)
 }
 
+type readSeekerClosedOnEOF struct {
+	closed bool
+
+	rs io.ReadSeeker
+	c  io.Closer
+}
+
+func (x *readSeekerClosedOnEOF) Read(p []byte) (int, error) {
+	n, err := x.rs.Read(p)
+	if errors.Is(err, io.EOF) {
+		x.closed = true
+		_ = x.c.Close()
+	}
+	return n, err
+}
+
+func (x *readSeekerClosedOnEOF) Seek(offset int64, whence int) (int64, error) {
+	return x.rs.Seek(offset, whence)
+}
+
+func (x *readSeekerClosedOnEOF) Close() error {
+	if !x.closed {
+		x.closed = true
+		return x.c.Close()
+	}
+	return nil
+}
+
 // HandleTask executes replication task inside invoking goroutine.
 // Passes all the nodes that accepted the replication to the TaskResult.
 func (p *Replicator) HandleTask(ctx context.Context, task Task, res TaskResult) {
@@ -38,6 +68,16 @@ func (p *Replicator) HandleTask(ctx context.Context, task Task, res TaskResult)
 			return
 		}
 
+		if len(task.nodes) > 1 {
+			rs := client.ShareReplicatedObject(binObjStream)
+			// since in this case we read object once it's worth to close the stream insta
+			// after reading finish so that no longer used resources do not hang up
+			binObjStream = &readSeekerClosedOnEOF{
+				rs: rs,
+				c:  binObjStream,
+			}
+		}
+
 		defer func() {
 			if err := binObjStream.Close(); err != nil {
 				p.log.Debug("failed to close replicated object's binary stream from the local storage",
@@ -55,15 +95,6 @@ func (p *Replicator) HandleTask(ctx context.Context, task Task, res TaskResult)
 		default:
 		}
 
-		if i > 0 && binObjStream != nil {
-			_, err = binObjStream.Seek(0, io.SeekStart)
-			if err != nil {
-				p.log.Error("failed to seek start of the replicated object's binary stream from the local storage",
-					zap.Stringer("object", task.addr), zap.Error(err))
-				return
-			}
-		}
-
 		log := p.log.With(
 			zap.String("node", netmap.StringifyPublicKey(task.nodes[i])),
 			zap.Stringer("object", task.addr),
@@ -73,6 +104,9 @@ func (p *Replicator) HandleTask(ctx context.Context, task Task, res TaskResult)
 
 		if binObjStream != nil {
 			err = p.remoteSender.CopyObjectToNode(ctx, task.nodes[i], binObjStream)
+			// note that we don't need to reset binObjStream because it always read once:
+			//  - if len(task.nodes) == 1, we won't come here again
+			//  - otherwise, we use client.ShareReplicatedObject (see above)
 			// FIXME: what if node is old?
 		} else {
 			err = p.remoteSender.PutObject(callCtx, prm.WithObject(task.obj).WithNodeInfo(task.nodes[i]))