Limit max number of entries when listing recycle bin

changelog/unreleased/eos-recycle.md

@@ -0,0 +1,5 @@
+Enhancement: Limit max number of entries returned by ListRecycle in eos
+
+The idea is to query first how many entries we'd have from eos recycle ls and bail out if "too many".
+
+https://github.com/cs3org/reva/pull/4455

pkg/eosclient/eosbinary/eosbinary.go

@@ -789,8 +789,8 @@ func (c *Client) WriteFile(ctx context.Context, auth eosclient.Authorization, pa
 
 // ListDeletedEntries returns a list of the deleted entries.
 func (c *Client) ListDeletedEntries(ctx context.Context, auth eosclient.Authorization) ([]*eosclient.DeletedEntry, error) {
-	// TODO(labkode): add protection if slave is configured and alive to count how many files are in the trashbin before
-	// triggering the recycle ls call that could break the instance because of unavailable memory.
+	// Note that this may time out if the recycle has too many items:
+	// the CS3API call ListRecycle includes a check to prevent that
 	args := []string{"recycle", "ls", "-m"}
 	stdout, _, err := c.executeEOS(ctx, args, auth)
 	if err != nil {

pkg/eosclient/eosgrpc/eosgrpc.go

@@ -1389,10 +1389,11 @@ func (c *Client) ListDeletedEntries(ctx context.Context, auth eosclient.Authoriz
 
 	msg := new(erpc.NSRequest_RecycleRequest)
 	msg.Cmd = erpc.NSRequest_RecycleRequest_RECYCLE_CMD(erpc.NSRequest_RecycleRequest_RECYCLE_CMD_value["LIST"])
-
 	rq.Command = &erpc.NSRequest_Recycle{Recycle: msg}
 
 	// Now send the req and see what happens
+	// Note that this may time out if the recycle has too many items:
+	// the CS3API call ListRecycle includes a check to prevent that
 	resp, err := c.cl.Exec(appctx.ContextGetClean(ctx), rq)
 	e := c.getRespError(resp, err)
 	if e != nil {
@@ -1409,10 +1410,6 @@ func (c *Client) ListDeletedEntries(ctx context.Context, auth eosclient.Authoriz
 	} else {
 		log.Debug().Str("func", "ListDeletedEntries").Str("info:", fmt.Sprintf("%#v", resp)).Msg("grpc response")
 	}
-	// TODO(labkode): add protection if slave is configured and alive to count how many files are in the trashbin before
-	// triggering the recycle ls call that could break the instance because of unavailable memory.
-	// FF: I agree with labkode, if we think we may have memory problems then the semantics of the grpc call`and
-	// the semantics if this func will have to change. For now this is not foreseen
 
 	ret := make([]*eosclient.DeletedEntry, 0)
 	for _, f := range resp.Recycle.Recycles {
@@ -1635,7 +1632,10 @@ func (c *Client) grpcMDResponseToFileInfo(ctx context.Context, st *erpc.MDRespon
 			fi.Attrs[strings.TrimPrefix(k, "user.")] = string(v)
 		}
 
-		fi.Size = uint64(st.Cmd.TreeSize)
+		fi.TreeSize = uint64(st.Cmd.TreeSize)
+		fi.Size = fi.TreeSize
+		// TODO(lopresti) this info is missing in the EOS Protobuf, cf. EOS-5974
+		// fi.TreeCount = uint64(st.Cmd.TreeCount)
 
 		log.Debug().Str("stat info - path", fi.File).Uint64("inode", fi.Inode).Uint64("uid", fi.UID).Uint64("gid", fi.GID).Str("etag", fi.ETag).Msg("grpc response")
 	} else {

pkg/storage/utils/eosfs/config.go

@@ -176,4 +176,8 @@ type Config struct {
 
 	// Path of the script to run after an user home folder has been created
 	OnPostCreateHomeHook string `mapstructure:"on_post_create_home_hook"`
+
+	// Maximum entries count a ListRecycle call may return: if exceeded, ListRecycle
+	// will return a BadRequest error
+	MaxRecycleEntries uint64 `mapstructure:"max_recycle_entries"`
 }

pkg/storage/utils/eosfs/eosfs.go

@@ -147,6 +147,10 @@ func (c *Config) ApplyDefaults() {
 		c.TokenExpiry = 3600
 	}
 
+	if c.MaxRecycleEntries == 0 {
+		c.MaxRecycleEntries = 5000
+	}
+
 	c.GatewaySvc = sharedconf.GetGatewaySVC(c.GatewaySvc)
 }
 
@@ -1945,6 +1949,12 @@ func (fs *eosfs) ListRecycle(ctx context.Context, basePath, key, relativePath st
 		}
 	}
 
+	rcount, err := fs.countDeletedEntries(ctx, auth)
+	if err == nil && rcount > fs.conf.MaxRecycleEntries {
+		// ignore errors here and optimistically move on
+		return nil, errtypes.BadRequest("eosfs: too many entries found in listing recycle bin")
+	}
+
 	eosDeletedEntries, err := fs.c.ListDeletedEntries(ctx, auth)
 	if err != nil {
 		return nil, errors.Wrap(err, "eosfs: error listing deleted entries")
@@ -1964,6 +1974,18 @@ func (fs *eosfs) ListRecycle(ctx context.Context, basePath, key, relativePath st
 	return recycleEntries, nil
 }
 
+func (fs *eosfs) countDeletedEntries(ctx context.Context, auth eosclient.Authorization) (uint64, error) {
+	// Look for the recycle path, typically /eos/<instance>/proc/recycle/uid:<UID>
+	recyclePath := "output of eos recycle -m"
+	recyclePath = fmt.Sprintf("%s/uid:%s", recyclePath, auth.Role.UID)
+	// TODO: treeCount is not recursive, if we implement a calendar view we may do this on the '0' bucket
+	eosmd, err := fs.c.GetFileInfoByPath(ctx, auth, recyclePath)
+	if err != nil {
+		return 0, err
+	}
+	return eosmd.TreeCount, nil
+}
+
 func (fs *eosfs) RestoreRecycleItem(ctx context.Context, basePath, key, relativePath string, restoreRef *provider.Reference) error {
 	var auth eosclient.Authorization