Storage GetStream API (#3532)

Author: cthulhu-rider

pkg/local_object_storage/blobstor/common/storage.go

@@ -2,6 +2,7 @@ package common
 
 import (
 	"fmt"
+	"io"
 
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/compression"
 	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
@@ -26,6 +27,7 @@ type Storage interface {
 	GetBytes(oid.Address) ([]byte, error)
 	Get(oid.Address) (*objectSDK.Object, error)
 	GetRange(oid.Address, uint64, uint64) ([]byte, error)
+	GetStream(oid.Address) (*objectSDK.Object, io.ReadCloser, error)
 	Head(oid.Address) (*objectSDK.Object, error)
 	Exists(oid.Address) (bool, error)
 	Put(oid.Address, []byte) error

pkg/local_object_storage/blobstor/fstree/fstree.go

@@ -494,7 +494,8 @@ func (t *FSTree) readFullObject(f io.Reader, initial []byte, size int64) ([]byte
 
 // GetStream returns an object from the storage by address as a stream.
 // It returns the object with header only, and a reader for the payload.
-// The caller is responsible for closing the returned io.ReadCloser if it is not nil.
+// On success, the reader is non-nil and must be closed;
+// a nil reader is only returned with a non‑nil error.
 func (t *FSTree) GetStream(addr oid.Address) (*objectSDK.Object, io.ReadCloser, error) {
 	obj, reader, err := t.getObjectStream(addr)
 	if err != nil {

pkg/local_object_storage/engine/get.go

@@ -2,6 +2,7 @@ package engine
 
 import (
 	"errors"
+	"io"
 
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/util"
@@ -154,3 +155,39 @@ func (e *StorageEngine) GetBytes(addr oid.Address) ([]byte, error) {
 	})
 	return b, err
 }
+
+// GetStream reads an object from local storage as a stream.
+//
+// Returns the object header and a reader for the payload.
+// On success, the reader is non-nil and must be closed;
+// a nil reader is only returned with a non‑nil error.
+//
+// Returns any error encountered that did not allow to completely read the object part.
+// Returns an error of type apistatus.ObjectNotFound if the requested object is missing in local storage.
+// Returns an error of type apistatus.ObjectAlreadyRemoved if the object has been marked as removed.
+//
+// Returns an error if executions are blocked (see BlockExecution).
+func (e *StorageEngine) GetStream(addr oid.Address) (*objectSDK.Object, io.ReadCloser, error) {
+	if e.metrics != nil {
+		defer elapsed(e.metrics.AddGetStreamDuration)()
+	}
+
+	e.blockMtx.RLock()
+	defer e.blockMtx.RUnlock()
+
+	if e.blockErr != nil {
+		return nil, nil, e.blockErr
+	}
+
+	var (
+		err    error
+		obj    *objectSDK.Object
+		reader io.ReadCloser
+	)
+
+	err = e.get(addr, func(s *shard.Shard, ignoreMetadata bool) error {
+		obj, reader, err = s.GetStream(addr, ignoreMetadata)
+		return err
+	})
+	return obj, reader, err
+}

pkg/local_object_storage/engine/get_test.go

@@ -1,6 +1,7 @@
 package engine
 
 import (
+	"io"
 	"testing"
 
 	"github.com/nspcc-dev/neofs-node/pkg/core/object"
@@ -22,3 +23,24 @@ func TestStorageEngine_GetBytes(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, objBin, b)
 }
+
+func TestStorageEngine_GetStream(t *testing.T) {
+	e, _, _ := newEngine(t, t.TempDir())
+	obj := generateObjectWithCID(cidtest.ID())
+	addr := object.AddressOf(obj)
+
+	objBin := obj.Payload()
+
+	err := e.Put(obj, nil)
+	require.NoError(t, err)
+
+	header, reader, err := e.GetStream(addr)
+	require.NoError(t, err)
+	require.Equal(t, obj.CutPayload(), header)
+
+	require.NotNil(t, reader)
+	b, err := io.ReadAll(reader)
+	require.NoError(t, err)
+	require.Equal(t, objBin, b)
+	require.NoError(t, reader.Close())
+}

pkg/local_object_storage/engine/metrics.go

@@ -11,6 +11,7 @@ type MetricRegister interface {
 	AddExistsDuration(d time.Duration)
 	AddGetDuration(d time.Duration)
 	AddHeadDuration(d time.Duration)
+	AddGetStreamDuration(d time.Duration)
 	AddInhumeDuration(d time.Duration)
 	AddPutDuration(d time.Duration)
 	AddRangeDuration(d time.Duration)

pkg/local_object_storage/shard/get.go

@@ -3,6 +3,7 @@ package shard
 import (
 	"errors"
 	"fmt"
+	"io"
 
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/util/logicerr"
@@ -140,3 +141,52 @@ func (s *Shard) getBytesWithMetadataLookup(addr oid.Address, skipMeta bool) ([]b
 	}
 	return b, err
 }
+
+// GetStream reads an object from shard as a stream. skipMeta flag allows to fetch object from
+// the blobstor directly.
+//
+// Returns the object header and a reader for the payload.
+// On success, the reader is non-nil and must be closed;
+// a nil reader is only returned with a non‑nil error.
+//
+// Returns any error encountered that did not allow to completely read the object part.
+// Returns an error of type apistatus.ObjectNotFound if the requested object is missing in shard.
+// Returns an error of type apistatus.ObjectAlreadyRemoved if the requested object has been marked as removed in shard.
+// Returns the object.ErrObjectIsExpired if the object is present but already expired.
+func (s *Shard) GetStream(addr oid.Address, skipMeta bool) (*objectSDK.Object, io.ReadCloser, error) {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	var (
+		res    *objectSDK.Object
+		reader io.ReadCloser
+	)
+
+	cb := func(stor common.Storage) error {
+		obj, r, err := stor.GetStream(addr)
+		if err != nil {
+			return err
+		}
+		res = obj
+		reader = r
+		return nil
+	}
+
+	wc := func(c writecache.Cache) error {
+		o, r, err := c.GetStream(addr)
+		if err != nil {
+			return err
+		}
+		res = o
+		reader = r
+		return nil
+	}
+
+	skipMeta = skipMeta || s.info.Mode.NoMetabase()
+	gotMeta, err := s.fetchObjectData(addr, skipMeta, cb, wc)
+	if err != nil && gotMeta {
+		err = fmt.Errorf("%w, %w", err, ErrMetaWithNoObject)
+	}
+
+	return res, reader, err
+}

pkg/local_object_storage/shard/get_test.go

@@ -3,6 +3,7 @@ package shard_test
 import (
 	"bytes"
 	"errors"
+	"io"
 	"testing"
 	"time"
 
@@ -43,6 +44,7 @@ func testShardGet(t *testing.T, hasWriteCache bool) {
 		require.Equal(t, obj, res)
 
 		testGetBytes(t, sh, addr, obj.Marshal())
+		testGetStream(t, sh, addr, obj)
 	})
 
 	t.Run("big object", func(t *testing.T) {
@@ -60,32 +62,38 @@ func testShardGet(t *testing.T, hasWriteCache bool) {
 		require.Equal(t, obj, res)
 
 		testGetBytes(t, sh, addr, obj.Marshal())
+		testGetStream(t, sh, addr, obj)
 	})
 
 	t.Run("parent object", func(t *testing.T) {
-		obj := generateObject()
-		addAttribute(obj, "foo", "bar")
 		cnr := cidtest.ID()
 		splitID := objectSDK.NewSplitID()
 
 		parent := generateObjectWithCID(cnr)
 		addAttribute(parent, "parent", "attribute")
+		parentAddr := object.AddressOf(parent)
 
 		child := generateObjectWithCID(cnr)
 		child.SetParent(parent)
 		idParent := parent.GetID()
 		child.SetParentID(idParent)
 		child.SetSplitID(splitID)
 		addPayload(child, 1<<5)
+		childAddr := object.AddressOf(child)
 
 		err := sh.Put(child, nil)
 		require.NoError(t, err)
 
-		res, err := testGet(t, sh, object.AddressOf(child), hasWriteCache)
+		res, err := testGet(t, sh, childAddr, hasWriteCache)
 		require.NoError(t, err)
 		require.True(t, binaryEqual(child, res))
 
-		_, err = testGet(t, sh, object.AddressOf(parent), hasWriteCache)
+		testGetStream(t, sh, childAddr, child)
+
+		_, _, streamErr := sh.GetStream(parentAddr, false)
+		require.Error(t, streamErr)
+		_, err = testGet(t, sh, parentAddr, hasWriteCache)
+		require.Equal(t, streamErr, err)
 
 		var si *objectSDK.SplitInfoError
 		require.True(t, errors.As(err, &si))
@@ -122,6 +130,18 @@ func testGetBytes(t testing.TB, sh *shard.Shard, addr oid.Address, objBin []byte
 	require.Equal(t, objBin, b)
 }
 
+func testGetStream(t testing.TB, sh *shard.Shard, addr oid.Address, obj *objectSDK.Object) {
+	header, reader, err := sh.GetStream(addr, false)
+	require.NoError(t, err)
+	require.Equal(t, obj.CutPayload(), header)
+
+	data, err := io.ReadAll(reader)
+	require.NoError(t, err)
+	require.Equal(t, obj.Payload(), data)
+
+	require.NoError(t, reader.Close())
+}
+
 // binary equal is used when object contains empty lists in the structure and
 // require.Equal fails on comparing <nil> and []{} lists.
 func binaryEqual(a, b *objectSDK.Object) bool {

pkg/local_object_storage/writecache/get.go

@@ -2,6 +2,7 @@ package writecache
 
 import (
 	"fmt"
+	"io"
 
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/util/logicerr"
 	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
@@ -50,3 +51,19 @@ func (c *cache) GetBytes(addr oid.Address) ([]byte, error) {
 
 	return b, nil
 }
+
+// GetStream returns an object stream from write-cache.
+// On success, the reader is non-nil and must be closed;
+// a nil reader is only returned with a non‑nil error.
+// Returns an error of type apistatus.ObjectNotFound if the requested object is missing in write-cache.
+func (c *cache) GetStream(addr oid.Address) (*objectSDK.Object, io.ReadCloser, error) {
+	if !c.objCounters.HasAddress(addr) {
+		return nil, nil, logicerr.Wrap(apistatus.ErrObjectNotFound)
+	}
+	stream, reader, err := c.fsTree.GetStream(addr)
+	if err != nil {
+		return nil, nil, logicerr.Wrap(apistatus.ErrObjectNotFound)
+	}
+
+	return stream, reader, nil
+}

pkg/local_object_storage/writecache/get_test.go

@@ -1,8 +1,11 @@
 package writecache
 
 import (
+	"io"
 	"testing"
 
+	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
+	oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test"
 	"github.com/stretchr/testify/require"
 )
 
@@ -24,3 +27,34 @@ func TestCache_GetBytes(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, objBin, b)
 }
+
+func TestCache_GetStream(t *testing.T) {
+	const maxObjSize = 4 << 10
+	c, _ := newCache(t)
+
+	testStream := func(t *testing.T, size int) {
+		o := putObject(t, c, size)
+		objBin := o.obj.Payload()
+
+		header, reader, err := c.GetStream(o.addr)
+		require.NoError(t, err)
+		require.Equal(t, o.obj.CutPayload(), header)
+
+		require.NotNil(t, reader)
+		b, err := io.ReadAll(reader)
+		require.NoError(t, err)
+		require.Equal(t, objBin, b)
+		require.NoError(t, reader.Close())
+	}
+
+	testStream(t, 0)
+	testStream(t, maxObjSize/2)
+	testStream(t, 2*maxObjSize)
+
+	t.Run("not found", func(t *testing.T) {
+		header, reader, err := c.GetStream(oidtest.Address())
+		require.ErrorIs(t, err, apistatus.ErrObjectNotFound)
+		require.Nil(t, header)
+		require.Nil(t, reader)
+	})
+}

pkg/local_object_storage/writecache/writecache.go

@@ -3,6 +3,7 @@ package writecache
 import (
 	"errors"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"sync"
@@ -29,6 +30,8 @@ type Cache interface {
 	// canonical NeoFS binary format. Returns [apistatus.ObjectNotFound] if object
 	// is missing.
 	GetBytes(oid.Address) ([]byte, error)
+	// GetStream returns an object and a stream to read its payload.
+	GetStream(oid.Address) (*object.Object, io.ReadCloser, error)
 	Head(oid.Address) (*object.Object, error)
 	// Delete removes object referenced by the given oid.Address from the
 	// Cache. Returns any error encountered that prevented the object to be