fstree: serve Stream from the storage

Add a new `GetStream` method to the FSTree storage. This method allows for
retrieving an object's payload as a stream (`io.ReadCloser`), which is
significantly more memory-efficient for large objects as it avoids loading the
entire payload into memory.

Signed-off-by: Andrey Butusov <[email protected]>

Author: End-rey

CHANGELOG.md

@@ -5,6 +5,7 @@ Changelog for NeoFS Node
 
 ### Added
 - `Head` operation for FSTree (#3383)
+- `GetStream` operation for FSTree (#3431)
 
 ### Fixed
 - IR exponentially retries updating SN lists in the Container contract in error cases (#3344)

pkg/local_object_storage/blobstor/fstree/fstree.go

@@ -489,6 +489,36 @@ func (t *FSTree) readFullObject(f io.Reader, initial []byte, size int64) ([]byte
 	return t.Decompress(data)
 }
 
+// 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.
+// If the object is small, the payload may be returned in the object itself,
+// and the reader will be nil.
+// The caller is responsible for closing the returned io.ReadCloser if it is not nil.
+func (t *FSTree) GetStream(addr oid.Address) (*objectSDK.Object, io.ReadCloser, error) {
+	p := t.treePath(addr)
+
+	f, err := os.Open(p)
+	if err != nil {
+		if errors.Is(err, fs.ErrNotExist) {
+			return nil, nil, logicerr.Wrap(apistatus.ObjectNotFound{})
+		}
+		return nil, nil, fmt.Errorf("read file %q: %w", p, err)
+	}
+
+	obj, reader, err := t.extractHeaderAndStream(addr.Object(), f)
+	if err != nil {
+		if reader != nil {
+			_ = reader.Close()
+		}
+		if errors.Is(err, fs.ErrNotExist) {
+			return nil, nil, logicerr.Wrap(apistatus.ErrObjectNotFound)
+		}
+		return nil, nil, fmt.Errorf("extract object stream from %q: %w", p, err)
+	}
+
+	return obj, reader, nil
+}
+
 // GetRange implements common.Storage.
 func (t *FSTree) GetRange(addr oid.Address, from uint64, length uint64) ([]byte, error) {
 	obj, err := t.Get(addr)

pkg/local_object_storage/blobstor/fstree/getstream_test.go

@@ -0,0 +1,117 @@
+package fstree
+
+import (
+	"crypto/rand"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/compression"
+	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
+	oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetStream(t *testing.T) {
+	tree := New(WithPath(t.TempDir()))
+
+	t.Run("object not found", func(t *testing.T) {
+		addr := oidtest.Address()
+		obj, reader, err := tree.GetStream(addr)
+		require.Error(t, err)
+		require.Nil(t, obj)
+		require.Nil(t, reader)
+	})
+
+	testStream := func(t *testing.T, size int, small bool) {
+		payload := make([]byte, size)
+		_, err := rand.Read(payload)
+		require.NoError(t, err)
+
+		addr := oidtest.Address()
+		obj := objectSDK.New()
+		obj.SetID(addr.Object())
+		obj.SetPayload(payload)
+
+		require.NoError(t, tree.Put(addr, obj.Marshal()))
+
+		retrievedObj, reader, err := tree.GetStream(addr)
+		require.NoError(t, err)
+		require.NotNil(t, retrievedObj)
+		if small {
+			require.Nil(t, reader)
+			require.Equal(t, payload, retrievedObj.Payload())
+		} else {
+			require.NotNil(t, reader)
+			streamedPayload, err := io.ReadAll(reader)
+			require.NoError(t, err)
+			require.Equal(t, payload, streamedPayload)
+			require.NoError(t, reader.Close())
+		}
+	}
+
+	t.Run("small object", func(t *testing.T) {
+		testStream(t, 1024, true)
+	})
+
+	t.Run("large object", func(t *testing.T) {
+		testStream(t, 10*1024*1024, false)
+	})
+
+	t.Run("compressed object", func(t *testing.T) {
+		compress := compression.Config{Enabled: true}
+		require.NoError(t, compress.Init())
+		tree.Config = &compress
+
+		testStream(t, 1024, true)
+		testStream(t, 10*1024*1024, false)
+	})
+}
+
+func TestGetStreamAfterErrors(t *testing.T) {
+	tree := New(WithPath(t.TempDir()))
+
+	t.Run("corrupt header", func(t *testing.T) {
+		addr := oidtest.Address()
+
+		objPath := tree.treePath(addr)
+		require.NoError(t, os.MkdirAll(filepath.Dir(objPath), 0755))
+
+		f, err := os.Create(objPath)
+		require.NoError(t, err)
+		_, err = f.Write([]byte("corrupt data that isn't a valid object"))
+		require.NoError(t, err)
+		require.NoError(t, f.Close())
+
+		obj, reader, err := tree.GetStream(addr)
+		require.Error(t, err)
+		require.Nil(t, obj)
+		require.Nil(t, reader)
+	})
+
+	t.Run("corrupt compressed data", func(t *testing.T) {
+		compress := compression.Config{Enabled: true}
+		require.NoError(t, compress.Init())
+		tree.Config = &compress
+
+		addr := oidtest.Address()
+		obj := objectSDK.New()
+		obj.SetID(addr.Object())
+		payload := []byte("test payload")
+		obj.SetPayload(payload)
+
+		require.NoError(t, tree.Put(addr, obj.Marshal()))
+
+		objPath := tree.treePath(addr)
+
+		f, err := os.OpenFile(objPath, os.O_WRONLY|os.O_APPEND, 0644)
+		require.NoError(t, err)
+		_, err = f.Write([]byte("corruption at the end"))
+		require.NoError(t, err)
+		require.NoError(t, f.Close())
+
+		_, _, err = tree.GetStream(addr)
+		require.Error(t, err)
+	})
+}