Some error cleanup steps in the decomposed FS (#2511)

* Some error cleanup steps in the decomposed FS

* Hound and changelog added.

* Remove punctuation

* Make CI happy

* Improved error logging.

Co-authored-by: David Christofas <[email protected]>

* Update pkg/storage/utils/decomposedfs/decomposedfs.go

Co-authored-by: David Christofas <[email protected]>

Co-authored-by: David Christofas <[email protected]>
Co-authored-by: Jörn Friedrich Dreyer <[email protected]>

changelog/unreleased/clean-dfs1.md

@@ -0,0 +1,5 @@
+Enhancement: Error handling cleanup in decomposed FS
+
+- Avoid inconsensitencies by cleaning up actions in case of err
+
+https://github.com/cs3org/reva/pull/2511

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -217,6 +217,17 @@ func (fs *Decomposedfs) CreateHome(ctx context.Context) (err error) {
 		}
 		return nil
 	})
+
+	// make sure to delete the created directory if things go wrong
+	defer func() {
+		if err != nil {
+			// do not catch the error to not shadow the original error
+			if tmpErr := fs.tp.Delete(ctx, n); tmpErr != nil {
+				appctx.GetLogger(ctx).Error().Err(tmpErr).Msg("Can not revert file system change after error")
+			}
+		}
+	}()
+
 	if err != nil {
 		return
 	}
@@ -333,6 +344,9 @@ func (fs *Decomposedfs) CreateDir(ctx context.Context, ref *provider.Reference)
 		// mark the home node as the end of propagation
 		if err = xattr.Set(nodePath, xattrs.PropagationAttr, []byte("1")); err != nil {
 			appctx.GetLogger(ctx).Error().Err(err).Interface("node", n).Msg("could not mark node to propagate")
+
+			// FIXME: This does not return an error at all, but results in a severe situation that the
+			// part tree is not marked for propagation
 			return
 		}
 	}
@@ -345,9 +359,10 @@ func (fs *Decomposedfs) TouchFile(ctx context.Context, ref *provider.Reference)
 }
 
 // CreateReference creates a reference as a node folder with the target stored in extended attributes
-// There is no difference between the /Shares folder and normal nodes because the storage is not supposed to be accessible without the storage provider.
-// In effect everything is a shadow namespace.
+// There is no difference between the /Shares folder and normal nodes because the storage is not supposed to be accessible
+// without the storage provider. In effect everything is a shadow namespace.
 // To mimic the eos and owncloud driver we only allow references as children of the "/Shares" folder
+// FIXME: This comment should explain briefly what a reference is in this context.
 func (fs *Decomposedfs) CreateReference(ctx context.Context, p string, targetURI *url.URL) (err error) {
 	ctx, span := rtrace.Provider.Tracer("reva").Start(ctx, "CreateReference")
 	defer span.End()
@@ -368,39 +383,62 @@ func (fs *Decomposedfs) CreateReference(ctx context.Context, p string, targetURI
 	}
 
 	// create Shares folder if it does not exist
-	var n *node.Node
-	if n, err = fs.lu.NodeFromResource(ctx, &provider.Reference{Path: fs.o.ShareFolder}); err != nil {
+	var parentNode *node.Node
+	var parentCreated, childCreated bool // defaults to false
+	if parentNode, err = fs.lu.NodeFromResource(ctx, &provider.Reference{Path: fs.o.ShareFolder}); err != nil {
 		err := errtypes.InternalError(err.Error())
 		span.SetStatus(codes.Error, err.Error())
 		return err
-	} else if !n.Exists {
-		if err = fs.tp.CreateDir(ctx, n); err != nil {
+	} else if !parentNode.Exists {
+		if err = fs.tp.CreateDir(ctx, parentNode); err != nil {
 			span.SetStatus(codes.Error, err.Error())
 			return err
 		}
-	}
+		parentCreated = true
+	}
+
+	var childNode *node.Node
+	// clean up directories created here on error
+	defer func() {
+		if err != nil {
+			// do not catch the error to not shadow the original error
+			if childCreated && childNode != nil {
+				if tmpErr := fs.tp.Delete(ctx, childNode); tmpErr != nil {
+					appctx.GetLogger(ctx).Error().Err(tmpErr).Str("node_id", childNode.ID).Msg("Can not clean up child node after error")
+				}
+			}
+			if parentCreated && parentNode != nil {
+				if tmpErr := fs.tp.Delete(ctx, parentNode); tmpErr != nil {
+					appctx.GetLogger(ctx).Error().Err(tmpErr).Str("node_id", parentNode.ID).Msg("Can not clean up parent node after error")
+				}
+
+			}
+		}
+	}()
 
-	if n, err = n.Child(ctx, parts[1]); err != nil {
+	if childNode, err = parentNode.Child(ctx, parts[1]); err != nil {
 		return errtypes.InternalError(err.Error())
 	}
 
-	if n.Exists {
+	if childNode.Exists {
 		// TODO append increasing number to mountpoint name
 		err := errtypes.AlreadyExists(p)
 		span.SetStatus(codes.Error, err.Error())
 		return err
 	}
 
-	if err := fs.tp.CreateDir(ctx, n); err != nil {
+	if err := fs.tp.CreateDir(ctx, childNode); err != nil {
 		span.SetStatus(codes.Error, err.Error())
 		return err
 	}
+	childCreated = true
 
-	internal := n.InternalPath()
-	if err := xattr.Set(internal, xattrs.ReferenceAttr, []byte(targetURI.String())); err != nil {
+	internalPath := childNode.InternalPath()
+	if err := xattr.Set(internalPath, xattrs.ReferenceAttr, []byte(targetURI.String())); err != nil {
+		// the reference could not be set - that would result in an lost reference?
 		err := errors.Wrapf(err, "Decomposedfs: error setting the target %s on the reference file %s",
 			targetURI.String(),
-			internal,
+			internalPath,
 		)
 		span.SetStatus(codes.Error, err.Error())
 		return err

pkg/storage/utils/decomposedfs/spaces.go

@@ -505,13 +505,15 @@ func (fs *Decomposedfs) createStorageSpace(ctx context.Context, spaceType, space
 	if err != nil {
 		if isAlreadyExists(err) {
 			appctx.GetLogger(ctx).Debug().Err(err).Str("space", spaceID).Str("spacetype", spaceType).Msg("symlink already exists")
+			// FIXME: is it ok to wipe this err if the symlink already exists?
+			err = nil
 		} else {
 			// TODO how should we handle error cases here?
 			appctx.GetLogger(ctx).Error().Err(err).Str("space", spaceID).Str("spacetype", spaceType).Msg("could not create symlink")
 		}
 	}
 
-	return nil
+	return err
 }
 
 func (fs *Decomposedfs) storageSpaceFromNode(ctx context.Context, n *node.Node, spaceType, nodePath string, canListAllSpaces bool) (*provider.StorageSpace, error) {