fix(decomposedfs): available space calculation

Move the calculation of the available size to the blobstore interface.
We were just returning the available size of the local disk, which is
wrong when using the S3 blobstore.

Also, as the S3 blobstore doesn't have a concept of available size, we
don't return a 'remaining' for S3 backed spaces with unlimited quota.

Fixes: owncloud/ocis#9245

Author: rhafer

changelog/unreleased/fix-available-diskspace.md

@@ -0,0 +1,6 @@
+Bugfix: Fix remaining space calculation for S3 blobstore
+
+The calculation of the remaining space in the S3 blobstore was incorrectly using
+the remaining space of the local disk instead.
+
+https://github.com/cs3org/reva/pull/4867

pkg/storage/fs/ocis/blobstore/blobstore.go

@@ -116,6 +116,11 @@ func (bs *Blobstore) Delete(node *node.Node) error {
 	return nil
 }
 
+// GetAvailableSize returns the available size in the blobstore in bytes
+func (bs *Blobstore) GetAvailableSize(n *node.Node) (uint64, error) {
+	return node.GetAvailableSize(n.InternalPath())
+}
+
 // List lists all blobs in the Blobstore
 func (bs *Blobstore) List() ([]*node.Node, error) {
 	dirs, err := filepath.Glob(filepath.Join(bs.root, "spaces", "*", "*", "blobs", "*", "*", "*", "*", "*"))

pkg/storage/fs/posix/blobstore/blobstore.go

@@ -88,3 +88,8 @@ func (bs *Blobstore) Download(node *node.Node) (io.ReadCloser, error) {
 func (bs *Blobstore) Delete(node *node.Node) error {
 	return nil
 }
+
+// GetAvailableSize returns the available size in the blobstore in bytes
+func (bs *Blobstore) GetAvailableSize(n *node.Node) (uint64, error) {
+	return node.GetAvailableSize(n.InternalPath())
+}

pkg/storage/fs/posix/posix.go

@@ -126,6 +126,7 @@ func New(m map[string]interface{}, stream events.Stream) (storage.FS, error) {
 	aspects := aspects.Aspects{
 		Lookup:            lu,
 		Tree:              tp,
+		Blobstore:         bs,
 		Permissions:       p,
 		EventStream:       stream,
 		UserMapper:        um,

pkg/storage/fs/posix/testhelpers/helpers.go

@@ -178,6 +178,7 @@ func NewTestEnv(config map[string]interface{}) (*TestEnv, error) {
 	)
 
 	bs := &treemocks.Blobstore{}
+	bs.On("GetAvailableSize", mock.Anything).Return(uint64(1000000000), nil)
 	tree, err := tree.New(lu, bs, um, &trashbin.Trashbin{}, o, nil, store.Create())
 	if err != nil {
 		return nil, err
@@ -189,6 +190,7 @@ func NewTestEnv(config map[string]interface{}) (*TestEnv, error) {
 	aspects := aspects.Aspects{
 		Lookup:      lu,
 		Tree:        tree,
+		Blobstore:   bs,
 		Permissions: permissions.NewPermissions(pmock, permissionsSelector),
 		Trashbin:    tb,
 	}

pkg/storage/fs/s3ng/blobstore/blobstore.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/tree"
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
 	"github.com/pkg/errors"
@@ -128,6 +129,12 @@ func (bs *Blobstore) Delete(node *node.Node) error {
 	return nil
 }
 
+// GetAvailableSize returns the available size in the blobstore in bytes
+func (bs *Blobstore) GetAvailableSize(n *node.Node) (uint64, error) {
+	// S3 doen't have a concept of available size
+	return 0, tree.ErrSizeUnlimited
+}
+
 // List lists all blobs in the Blobstore
 func (bs *Blobstore) List() ([]*node.Node, error) {
 	ch := bs.client.ListObjects(context.Background(), bs.bucket, minio.ListObjectsOptions{Recursive: true})

pkg/storage/utils/decomposedfs/aspects/aspects.go

@@ -23,13 +23,15 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/permissions"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/trashbin"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/tree"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/usermapper"
 )
 
 // Aspects holds dependencies for handling aspects of the decomposedfs
 type Aspects struct {
 	Lookup            node.PathLookup
 	Tree              node.Tree
+	Blobstore         tree.Blobstore
 	Trashbin          trashbin.Trashbin
 	Permissions       permissions.Permissions
 	EventStream       events.Stream

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"net/url"
 	"path"
 	"path/filepath"
@@ -112,6 +113,7 @@ type SessionStore interface {
 type Decomposedfs struct {
 	lu           node.PathLookup
 	tp           node.Tree
+	bs           tree.Blobstore
 	trashbin     trashbin.Trashbin
 	o            *options.Options
 	p            permissions.Permissions
@@ -162,6 +164,7 @@ func NewDefault(m map[string]interface{}, bs tree.Blobstore, es events.Stream) (
 	aspects := aspects.Aspects{
 		Lookup:            lu,
 		Tree:              tp,
+		Blobstore:         bs,
 		Permissions:       permissions.NewPermissions(node.NewPermissions(lu), permissionsSelector),
 		EventStream:       es,
 		DisableVersioning: o.DisableVersioning,
@@ -223,6 +226,7 @@ func New(o *options.Options, aspects aspects.Aspects) (storage.FS, error) {
 
 	fs := &Decomposedfs{
 		tp:              aspects.Tree,
+		bs:              aspects.Blobstore,
 		lu:              aspects.Lookup,
 		trashbin:        aspects.Trashbin,
 		o:               o,
@@ -598,9 +602,11 @@ func (fs *Decomposedfs) GetQuota(ctx context.Context, ref *provider.Reference) (
 		quotaStr = string(ri.Opaque.Map["quota"].Value)
 	}
 
-	// FIXME this reads remaining disk size from the local disk, not the blobstore
-	remaining, err = node.GetAvailableSize(n.InternalPath())
-	if err != nil {
+	remaining, err = fs.bs.GetAvailableSize(n)
+	switch {
+	case errors.Is(err, tree.ErrSizeUnlimited):
+		remaining = math.MaxUint64
+	case err != nil:
 		return 0, 0, 0, err
 	}
 

pkg/storage/utils/decomposedfs/decomposedfs_test.go

@@ -58,6 +58,7 @@ var _ = Describe("Decomposed", func() {
 	Describe("NewDefault", func() {
 		It("works", func() {
 			bs := &treemocks.Blobstore{}
+			bs.On("GetAvailableSize", mock.Anything).Return(uint64(1000000000), nil)
 			_, err := decomposedfs.NewDefault(map[string]interface{}{
 				"root":           env.Root,
 				"permissionssvc": "any",

pkg/storage/utils/decomposedfs/spaces.go

@@ -45,6 +45,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/prefixes"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/permissions"
+	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/tree"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/templates"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
 	"github.com/cs3org/reva/v2/pkg/utils"
@@ -1028,9 +1029,11 @@ func (fs *Decomposedfs) StorageSpaceFromNode(ctx context.Context, n *node.Node,
 		quotaStr = quotaInOpaque
 	}
 
-	// FIXME this reads remaining disk size from the local disk, not the blobstore
-	remaining, err := node.GetAvailableSize(n.InternalPath())
-	if err != nil {
+	remaining, err := fs.bs.GetAvailableSize(n)
+	switch {
+	case errors.Is(err, tree.ErrSizeUnlimited):
+		remaining = math.MaxUint64
+	case err != nil:
 		return nil, err
 	}
 	total, used, remaining, err := fs.calculateTotalUsedRemaining(quotaStr, space.GetRootInfo().GetSize(), remaining)
@@ -1039,7 +1042,9 @@ func (fs *Decomposedfs) StorageSpaceFromNode(ctx context.Context, n *node.Node,
 	}
 	space.Opaque = utils.AppendPlainToOpaque(space.Opaque, "quota.total", strconv.FormatUint(total, 10))
 	space.Opaque = utils.AppendPlainToOpaque(space.Opaque, "quota.used", strconv.FormatUint(used, 10))
-	space.Opaque = utils.AppendPlainToOpaque(space.Opaque, "quota.remaining", strconv.FormatUint(remaining, 10))
+	if remaining != math.MaxUint64 {
+		space.Opaque = utils.AppendPlainToOpaque(space.Opaque, "quota.remaining", strconv.FormatUint(remaining, 10))
+	}
 
 	return space, nil
 }

pkg/storage/utils/decomposedfs/testhelpers/helpers.go

@@ -175,6 +175,7 @@ func NewTestEnv(config map[string]interface{}) (*TestEnv, error) {
 	aspects := aspects.Aspects{
 		Lookup:      lu,
 		Tree:        tree,
+		Blobstore:   bs,
 		Permissions: permissions.NewPermissions(pmock, permissionsSelector),
 		Trashbin:    &decomposedfs.DecomposedfsTrashbin{},
 	}
@@ -290,6 +291,7 @@ func (t *TestEnv) CreateTestStorageSpace(typ string, quota *providerv1beta1.Quot
 	if typ == "personal" {
 		owner = t.Owner
 	}
+	t.Blobstore.On("GetAvailableSize", mock.Anything).Return(uint64(1000000000), nil)
 	space, err := t.Fs.CreateStorageSpace(t.Ctx, &providerv1beta1.CreateStorageSpaceRequest{
 		Owner: owner,
 		Type:  typ,

pkg/storage/utils/decomposedfs/tree/mocks/Blobstore.go

@@ -48,7 +48,10 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-var tracer trace.Tracer
+var (
+	tracer           trace.Tracer
+	ErrSizeUnlimited = errors.New("blobstore size unlimited")
+)
 
 func init() {
 	tracer = otel.Tracer("github.com/cs3org/reva/pkg/storage/utils/decomposedfs/tree")
@@ -59,6 +62,7 @@ type Blobstore interface {
 	Upload(node *node.Node, source string) error
 	Download(node *node.Node) (io.ReadCloser, error)
 	Delete(node *node.Node) error
+	GetAvailableSize(node *node.Node) (uint64, error)
 }
 
 // Tree manages a hierarchical tree

pkg/storage/utils/decomposedfs/tree/tree.go

@@ -153,6 +153,7 @@ var _ = Describe("Async file uploads", Ordered, func() {
 			},
 		)
 		bs = &treemocks.Blobstore{}
+		bs.On("GetAvailableSize", mock.Anything).Return(uint64(1000000000), nil)
 
 		// create space uses CheckPermission endpoint
 		cs3permissionsclient.On("CheckPermission", mock.Anything, mock.Anything, mock.Anything).Return(&cs3permissions.CheckPermissionResponse{
@@ -176,6 +177,7 @@ var _ = Describe("Async file uploads", Ordered, func() {
 		aspects := aspects.Aspects{
 			Lookup:      lu,
 			Tree:        tree,
+			Blobstore:   bs,
 			Permissions: permissions.NewPermissions(pmock, permissionsSelector),
 			EventStream: stream.Chan{pub, con},
 			Trashbin:    &DecomposedfsTrashbin{},

pkg/storage/utils/decomposedfs/upload_async_test.go

@@ -136,11 +136,13 @@ var _ = Describe("File uploads", func() {
 			AddGrant: true,
 		}, nil).Times(1)
 		var err error
+		bs.On("GetAvailableSize", mock.Anything).Return(uint64(1000000000), nil).Times(1)
 		tree := tree.New(lu, bs, o, store.Create())
 
 		aspects := aspects.Aspects{
 			Lookup:      lu,
 			Tree:        tree,
+			Blobstore:   bs,
 			Permissions: permissions.NewPermissions(pmock, permissionsSelector),
 			Trashbin:    &decomposedfs.DecomposedfsTrashbin{},
 		}