fstree: serve Stream from the storage (#3431)

Author: roman-khimov

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)

cmd/neofs-cli/modules/acl/extended/create.go

@@ -27,8 +27,8 @@ Action is 'allow' or 'deny'.
 Operation is an object service verb: 'get', 'head', 'put', 'search', 'delete', 'getrange', or 'getrangehash'.
 
 Filter consists of <typ>:<key><match><value>
-  Typ is 'obj' for object applied filter or 'req' for request applied filter. 
-  Key is a valid unicode string corresponding to object or request header key. 
+  Typ is 'obj' for object applied filter or 'req' for request applied filter.
+  Key is a valid unicode string corresponding to object or request header key.
     Well-known system object headers start with '$Object:' prefix.
     User defined headers start without prefix.
     Read more about filter keys at github.com/nspcc-dev/neofs-api/blob/master/proto-docs/acl.md#message-eaclrecordfilter
@@ -38,10 +38,10 @@ Filter consists of <typ>:<key><match><value>
     '>' | '>=' | '<' | '<=' for integer comparison.
   Value is a valid unicode string corresponding to object or request header value. Numeric filters must have base-10 integer values.
 
-Target is 
-  'user' for container owner, 
+Target is
+  'user' for container owner,
   'system' for Storage nodes in container and Inner Ring nodes,
-  'others' for all other request senders, 
+  'others' for all other request senders,
   'address:<adr1>,<adr2>,...' for exact request sender, where <adr> is a base58 25-byte address. Example: NSiVJYZej4XsxG5CUpdwn7VRQk8iiiDMPM.
 
 When both '--rule' and '--file' arguments are used, '--rule' records will be placed higher in resulting extended ACL table.

cmd/neofs-cli/modules/container/create.go

@@ -33,7 +33,7 @@ var (
 var createContainerCmd = &cobra.Command{
 	Use:   "create",
 	Short: "Create new container",
-	Long: `Create new container and register it in the NeoFS. 
+	Long: `Create new container and register it in the NeoFS.
 It will be stored in FS chain when inner ring will accepts it.`,
 	Args: cobra.NoArgs,
 	RunE: func(cmd *cobra.Command, _ []string) error {

cmd/neofs-cli/modules/container/delete.go

@@ -18,7 +18,7 @@ import (
 var deleteContainerCmd = &cobra.Command{
 	Use:   "delete",
 	Short: "Delete existing container",
-	Long: `Delete existing container. 
+	Long: `Delete existing container.
 Only owner of the container has a permission to remove container.`,
 	Args: cobra.NoArgs,
 	RunE: func(cmd *cobra.Command, _ []string) error {

cmd/neofs-cli/modules/session/create.go

@@ -32,8 +32,8 @@ var createCmd = &cobra.Command{
 	Short: "Create session token",
 	Long: `Create session token.
 
-Default lifetime of session token is ` + strconv.Itoa(defaultLifetime) + ` epochs 
-if none of --` + commonflags.ExpireAt + ` or --` + commonflags.Lifetime + ` flags is specified. 
+Default lifetime of session token is ` + strconv.Itoa(defaultLifetime) + ` epochs
+if none of --` + commonflags.ExpireAt + ` or --` + commonflags.Lifetime + ` flags is specified.
 `,
 	Args: cobra.NoArgs,
 	RunE: createSession,

pkg/local_object_storage/blobstor/fstree/fstree.go

@@ -489,6 +489,18 @@ 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.
+// 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) {
+	obj, reader, err := t.getObjectStream(addr)
+	if err != nil {
+		return nil, nil, 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,124 @@
+package fstree
+
+import (
+	"crypto/rand"
+	"fmt"
+	"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()))
+
+	payloadSizes := []int{
+		1,
+		1024,
+		1024 * 1024,
+	}
+
+	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) {
+		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)
+		require.Equal(t, obj.CutPayload(), retrievedObj)
+
+		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("different objects", func(t *testing.T) {
+		for _, size := range payloadSizes {
+			t.Run(fmt.Sprint(size), func(t *testing.T) {
+				testStream(t, size)
+			})
+		}
+	})
+
+	t.Run("compressed object", func(t *testing.T) {
+		compress := compression.Config{Enabled: true}
+		require.NoError(t, compress.Init())
+		tree.Config = &compress
+
+		for _, size := range payloadSizes {
+			t.Run(fmt.Sprint(size), func(t *testing.T) {
+				testStream(t, size)
+			})
+		}
+	})
+}
+
+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)
+	})
+}