WIP: Encode object into transport message once during PUT

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

go.mod

@@ -19,7 +19,7 @@ require (
 	github.com/nspcc-dev/neo-go v0.104.0
 	github.com/nspcc-dev/neofs-api-go/v2 v2.14.1-0.20231211142108-0f4d49e4804c
 	github.com/nspcc-dev/neofs-contract v0.19.1
-	github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.11.0.20231214115557-e600b997cd80
+	github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.11.0.20231215132359-814b9da96b12
 	github.com/nspcc-dev/tzhash v1.7.1
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/panjf2000/ants/v2 v2.8.2

go.sum

@@ -257,8 +257,8 @@ github.com/nspcc-dev/neofs-contract v0.19.1 h1:U1Uh+MlzfkalO0kRJ2pADZyHrmAOroC6K
 github.com/nspcc-dev/neofs-contract v0.19.1/go.mod h1:ZOGouuwuHpgvYkx/LCGufGncIzEUhYEO18LL4cWEbyw=
 github.com/nspcc-dev/neofs-crypto v0.4.0 h1:5LlrUAM5O0k1+sH/sktBtrgfWtq1pgpDs09fZo+KYi4=
 github.com/nspcc-dev/neofs-crypto v0.4.0/go.mod h1:6XJ8kbXgOfevbI2WMruOtI+qUJXNwSGM/E9eClXxPHs=
-github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.11.0.20231214115557-e600b997cd80 h1:pTDoIhN2QEAvNFDEFhXb5hqPi1ZWyIbzk+tJgOus2r8=
-github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.11.0.20231214115557-e600b997cd80/go.mod h1:d566MJDWH80Xu+1GfyWjMMoRmZ23oVD5+hTrPK1o5Ww=
+github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.11.0.20231215132359-814b9da96b12 h1:cpqxiaNtVEB7b6I7xZ79VqQ0kOpTBbw+SWfqDuSy8vI=
+github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.11.0.20231215132359-814b9da96b12/go.mod h1:d566MJDWH80Xu+1GfyWjMMoRmZ23oVD5+hTrPK1o5Ww=
 github.com/nspcc-dev/rfc6979 v0.2.0 h1:3e1WNxrN60/6N0DW7+UYisLeZJyfqZTNOjeV/toYvOE=
 github.com/nspcc-dev/rfc6979 v0.2.0/go.mod h1:exhIh1PdpDC5vQmyEsGvc4YDM/lyQp/452QxGq/UEso=
 github.com/nspcc-dev/tzhash v1.7.1 h1:6zmexLqdTF/ssbUAh7XJS7RxgKWaw28kdNpE/4UFdEU=

pkg/core/client/client.go

@@ -22,6 +22,7 @@ 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, opts client.ReplicateObjectOptions) error
+	PutFullObjectToNode(context.Context, object.Object, neofscrypto.Signer, client.PutFullObjectToNodeOptions) 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

@@ -241,6 +241,12 @@ func (x *multiClient) ReplicateObject(ctx context.Context, src io.ReadSeeker, si
 	return errSeek
 }
 
+func (x *multiClient) PutFullObjectToNode(ctx context.Context, obj objectSDK.Object, signer neofscrypto.Signer, opts client.PutFullObjectToNodeOptions) error {
+	return x.iterateClients(ctx, func(c clientcore.Client) error {
+		return c.PutFullObjectToNode(ctx, obj, signer, opts)
+	})
+}
+
 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/internal/client/client.go

@@ -383,38 +383,25 @@ func (x PutObjectRes) ID() oid.ID {
 //
 // Returns any error which prevented the operation from completing correctly in error return.
 func PutObject(prm PutObjectPrm) (*PutObjectRes, error) {
-	var prmCli client.PrmObjectPutInit
-
-	prmCli.MarkLocal()
+	var opts client.PutFullObjectToNodeOptions
 
 	if prm.tokenSession != nil {
-		prmCli.WithinSession(*prm.tokenSession)
+		opts.WithinSession(*prm.tokenSession)
 	}
 
 	if prm.tokenBearer != nil {
-		prmCli.WithBearerToken(*prm.tokenBearer)
+		opts.WithBearerToken(*prm.tokenBearer)
 	}
 
-	prmCli.WithXHeaders(prm.xHeaders...)
-
-	w, err := prm.cli.ObjectPutInit(prm.ctx, *prm.obj, prm.signer, prmCli)
+	err := prm.cli.PutFullObjectToNode(prm.ctx, *prm.obj, prm.signer, opts)
 	if err != nil {
-		return nil, fmt.Errorf("init object writing on client: %w", err)
+		return nil, fmt.Errorf("put full object node: %w", err)
 	}
 
-	_, err = w.Write(prm.obj.Payload())
-	if err != nil {
-		return nil, fmt.Errorf("write object payload into stream: %w", err)
-	}
-
-	err = w.Close()
-	if err != nil {
-		ReportError(prm.cli, err)
-		return nil, fmt.Errorf("finish object stream: %w", err)
-	}
+	id, _ := prm.obj.ID()
 
 	return &PutObjectRes{
-		id: w.GetResult().StoredObjectID(),
+		id: id,
 	}, nil
 }
 

pkg/services/object/put/distributed.go

@@ -1,6 +1,7 @@
 package putsvc
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -10,16 +11,19 @@ import (
 	"github.com/nspcc-dev/neofs-node/pkg/services/object_manager/placement"
 	"github.com/nspcc-dev/neofs-node/pkg/services/object_manager/transformer"
 	"github.com/nspcc-dev/neofs-node/pkg/util"
+	"github.com/nspcc-dev/neofs-sdk-go/client"
 	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
 	"go.uber.org/zap"
 )
 
 type preparedObjectTarget interface {
-	WriteObject(*objectSDK.Object, object.ContentMeta) error
+	WriteObject(context.Context, *objectSDK.Object, object.ContentMeta) error
 	Close() (*transformer.AccessIdentifiers, error)
 }
 
 type distributedTarget struct {
+	ctx context.Context
+
 	traversal traversal
 
 	remotePool, localPool util.WorkerPool
@@ -145,25 +149,25 @@ func (t *distributedTarget) Close() (*transformer.AccessIdentifiers, error) {
 		t.traversal.extraBroadcastEnabled = true
 	}
 
-	return t.iteratePlacement(t.sendObject)
+	return t.iteratePlacement(client.NewSharedPutFullObjectContext(t.ctx), t.sendObject)
 }
 
-func (t *distributedTarget) sendObject(node nodeDesc) error {
+func (t *distributedTarget) sendObject(ctx context.Context, node nodeDesc) error {
 	if !node.local && t.relay != nil {
 		return t.relay(node)
 	}
 
 	target := t.nodeTargetInitializer(node)
 
-	if err := target.WriteObject(t.obj, t.objMeta); err != nil {
+	if err := target.WriteObject(ctx, t.obj, t.objMeta); err != nil {
 		return fmt.Errorf("could not write header: %w", err)
 	} else if _, err := target.Close(); err != nil {
 		return fmt.Errorf("could not close object stream: %w", err)
 	}
 	return nil
 }
 
-func (t *distributedTarget) iteratePlacement(f func(nodeDesc) error) (*transformer.AccessIdentifiers, error) {
+func (t *distributedTarget) iteratePlacement(ctx context.Context, f func(context.Context, nodeDesc) error) (*transformer.AccessIdentifiers, error) {
 	id, _ := t.obj.ID()
 
 	traverser, err := placement.NewTraverser(
@@ -207,7 +211,7 @@ loop:
 			if err := workerPool.Submit(func() {
 				defer wg.Done()
 
-				err := f(nodeDesc{local: isLocal, info: addr})
+				err := f(ctx, nodeDesc{local: isLocal, info: addr})
 
 				// mark the container node as processed in order to exclude it
 				// in subsequent container broadcast. Note that we don't
@@ -244,7 +248,7 @@ loop:
 
 	// perform additional container broadcast if needed
 	if t.traversal.submitPrimaryPlacementFinish() {
-		_, err = t.iteratePlacement(f)
+		_, err = t.iteratePlacement(ctx, f)
 		if err != nil {
 			t.log.Error("additional container broadcast failure",
 				zap.Error(err),

pkg/services/object/put/local.go

@@ -1,6 +1,7 @@
 package putsvc
 
 import (
+	"context"
 	"fmt"
 
 	objectCore "github.com/nspcc-dev/neofs-node/pkg/core/object"
@@ -31,7 +32,7 @@ type localTarget struct {
 	meta objectCore.ContentMeta
 }
 
-func (t *localTarget) WriteObject(obj *object.Object, meta objectCore.ContentMeta) error {
+func (t *localTarget) WriteObject(_ context.Context, obj *object.Object, meta objectCore.ContentMeta) error {
 	t.obj = obj
 	t.meta = meta
 

pkg/services/object/put/remote.go

@@ -46,7 +46,8 @@ type RemotePutPrm struct {
 	obj *object.Object
 }
 
-func (t *remoteTarget) WriteObject(obj *object.Object, _ objectcore.ContentMeta) error {
+func (t *remoteTarget) WriteObject(ctx context.Context, obj *object.Object, _ objectcore.ContentMeta) error {
+	t.ctx = ctx
 	t.obj = obj
 
 	return nil
@@ -120,7 +121,6 @@ func (p *RemotePutPrm) WithObject(v *object.Object) *RemotePutPrm {
 // PutObject sends object to remote node.
 func (s *RemoteSender) PutObject(ctx context.Context, p *RemotePutPrm) error {
 	t := &remoteTarget{
-		ctx:               ctx,
 		keyStorage:        s.keyStorage,
 		clientConstructor: s.clientConstructor,
 	}
@@ -130,7 +130,7 @@ func (s *RemoteSender) PutObject(ctx context.Context, p *RemotePutPrm) error {
 		return fmt.Errorf("parse client node info: %w", err)
 	}
 
-	if err := t.WriteObject(p.obj, objectcore.ContentMeta{}); err != nil {
+	if err := t.WriteObject(ctx, p.obj, objectcore.ContentMeta{}); err != nil {
 		return fmt.Errorf("(%T) could not send object header: %w", s, err)
 	} else if _, err := t.Close(); err != nil {
 		return fmt.Errorf("(%T) could not send object: %w", s, err)

pkg/services/object/put/streamer.go

@@ -217,6 +217,7 @@ func (p *Streamer) newCommonTarget(prm *PutInitPrm) transformer.ObjectTarget {
 	withBroadcast := !prm.common.LocalOnly() && (typ == object.TypeTombstone || typ == object.TypeLock)
 
 	return &distributedTarget{
+		ctx: p.ctx,
 		traversal: traversal{
 			opts: prm.traverseOpts,
 
@@ -233,7 +234,6 @@ func (p *Streamer) newCommonTarget(prm *PutInitPrm) transformer.ObjectTarget {
 			}
 
 			rt := &remoteTarget{
-				ctx:               p.ctx,
 				keyStorage:        p.keyStorage,
 				commonPrm:         prm.common,
 				clientConstructor: p.clientConstructor,