Bump reva

Fixes: #1871, #1867
2025-11-20 14:58:30 +01:00
parent 90e4127227
commit e85d8effc1
92 changed files with 11404 additions and 8988 deletions
@@ -28,7 +28,7 @@ import (
 	"go-micro.dev/v4/store"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
-	semconv "go.opentelemetry.io/otel/semconv/v1.10.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.37.0"
 	"go.opentelemetry.io/otel/trace"
 )

@@ -67,7 +67,7 @@ func (bs *Blobstore) Upload(n *node.Node, source, copyTarget string) error {
 		_ = sourceFile.Close()
 	}()

-	tempFile, err := os.OpenFile(tempName, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0700)
+	tempFile, err := os.OpenFile(tempName, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0600)
 	if err != nil {
 		return fmt.Errorf("unable to create temp file '%s': %v", tempName, err)
 	}
@@ -22,6 +22,7 @@ package tree

 import (
 	"fmt"
+	"log/slog"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -31,6 +32,7 @@ import (
 	"github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/options"
 	"github.com/pablodz/inotifywaitgo/inotifywaitgo"
 	"github.com/rs/zerolog"
+	slogzerolog "github.com/samber/slog-zerolog/v2"
 )

 type InotifyWatcher struct {
@@ -56,6 +58,10 @@ func (iw *InotifyWatcher) Watch(path string) {
 			}
 		}()
 	}
+
+	// create a slog logger to be passed to the settings of inotifywatcher to log into
+	logger := slog.New(slogzerolog.Option{Level: slog.LevelDebug, Logger: iw.log}.NewZerologHandler())
+
 	events := make(chan inotifywaitgo.FileEvent)
 	errors := make(chan error)

@@ -76,6 +82,7 @@ func (iw *InotifyWatcher) Watch(path string) {
 			Monitor: true,
 		},
 		Verbose: false,
+		Log:     logger,
 	})

 	for {
@@ -272,7 +272,7 @@ func (tp *Tree) DownloadRevision(ctx context.Context, ref *provider.Reference, r
 	return ri, reader, nil
 }

-func (tp *Tree) RestoreRevision(ctx context.Context, srcNode, targetNode metadata.MetadataNode) error {
+func (tp *Tree) RestoreRevision(ctx context.Context, srcNode, targetNode metadata.MetadataNode, mtime time.Time) error {
 	source := srcNode.InternalPath()
 	target := targetNode.InternalPath()
 	rf, err := os.Open(source)
@@ -305,8 +305,10 @@ func (tp *Tree) RestoreRevision(ctx context.Context, srcNode, targetNode metadat
 		return errtypes.InternalError("failed to copy blob xattrs to old revision to node: " + err.Error())
 	}

-	// always set the node mtime to the current time
-	mtime := time.Now()
+	// set the node mtime to the current time if no mtime was provided
+	if mtime.IsZero() {
+		mtime = time.Now()
+	}
 	err = os.Chtimes(target, mtime, mtime)
 	if err != nil {
 		return errtypes.InternalError("failed to update times:" + err.Error())
@@ -19,7 +19,6 @@ import (

 	"github.com/opencloud-eu/reva/v2/pkg/storage/cache"
 	"github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/metadata/prefixes"
-	"github.com/opencloud-eu/reva/v2/pkg/storage/utils/filelocks"
 )

 var _metadataOffloadedAttr = prefixes.OcPrefix + "metadata_offloaded"
@@ -121,12 +120,11 @@ func (b HybridBackend) list(ctx context.Context, n MetadataNode, acquireLock boo

 	// listing xattrs failed, try again, either with lock or without
 	if acquireLock {
-		f, err := lockedfile.OpenFile(filePath+filelocks.LockFileSuffix, os.O_CREATE|os.O_WRONLY, 0600)
+		unlock, err := b.Lock(n)
 		if err != nil {
 			return nil, err
 		}
-		// Warning: do not remove the lockfile or we may lock the same file more than once, https://github.com/opencloud-eu/opencloud/issues/1793
-		defer f.Close()
+		defer func() { _ = unlock() }()

 	}
 	return xattr.List(filePath)
@@ -375,12 +373,11 @@ func (b HybridBackend) offloadMetadata(ctx context.Context, n MetadataNode) erro
 func (b HybridBackend) Remove(ctx context.Context, n MetadataNode, key string, acquireLock bool) error {
 	path := n.InternalPath()
 	if acquireLock {
-		lockedFile, err := lockedfile.OpenFile(path+filelocks.LockFileSuffix, os.O_CREATE|os.O_WRONLY, 0600)
+		unlock, err := b.Lock(n)
 		if err != nil {
 			return err
 		}
-		// Warning: do not remove the lockfile or we may lock the same file more than once, https://github.com/opencloud-eu/opencloud/issues/1793
-		defer lockedFile.Close()
+		defer func() { _ = unlock() }()
 	}

 	if isOffloadingAttribute(key) {
@@ -129,7 +129,7 @@ type Tree interface {
 	Delete(ctx context.Context, node *Node) (err error)

 	InitNewNode(ctx context.Context, n *Node, fsize uint64) (metadata.UnlockFunc, error)
-	RestoreRevision(ctx context.Context, source, target metadata.MetadataNode) (err error)
+	RestoreRevision(ctx context.Context, source, target metadata.MetadataNode, mtime time.Time) (err error)

 	WriteBlob(node *Node, source string) error
 	ReadBlob(node *Node) (io.ReadCloser, error)
@@ -131,7 +131,7 @@ func (fs *Decomposedfs) RestoreRevision(ctx context.Context, ref *provider.Refer

 	// restore revision
 	restoredRevisionPath := fs.lu.InternalPath(spaceID, revisionKey)
-	if err := fs.tp.RestoreRevision(ctx, revisionNode, n); err != nil {
+	if err := fs.tp.RestoreRevision(ctx, revisionNode, n, time.Now()); err != nil {
 		return err
 	}

@@ -330,7 +330,7 @@ func (tp *Tree) getRevisionNode(ctx context.Context, ref *provider.Reference, re
 	return n, nil
 }

-func (tp *Tree) RestoreRevision(ctx context.Context, sourceNode, targetNode metadata.MetadataNode) error {
+func (tp *Tree) RestoreRevision(ctx context.Context, sourceNode, targetNode metadata.MetadataNode, mtime time.Time) error {
 	err := tp.lookup.CopyMetadata(ctx, sourceNode, targetNode, func(attributeName string, value []byte) (newValue []byte, copy bool) {
 		return value, strings.HasPrefix(attributeName, prefixes.ChecksumPrefix) ||
 			attributeName == prefixes.TypeAttr ||
@@ -340,10 +340,13 @@ func (tp *Tree) RestoreRevision(ctx context.Context, sourceNode, targetNode meta
 	if err != nil {
 		return errtypes.InternalError("failed to copy blob xattrs to old revision to node: " + err.Error())
 	}
-	// always set the node mtime to the current time
+	// set the node mtime to the current time if no mtime was provided
+	if mtime.IsZero() {
+		mtime = time.Now()
+	}
 	err = tp.lookup.MetadataBackend().SetMultiple(ctx, targetNode,
 		map[string][]byte{
-			prefixes.MTimeAttr: []byte(time.Now().UTC().Format(time.RFC3339Nano)),
+			prefixes.MTimeAttr: []byte(mtime.UTC().Format(time.RFC3339Nano)),
 		},
 		false)
 	if err != nil {
@@ -292,7 +292,10 @@ func (session *DecomposedFsSession) Finalize(ctx context.Context) (err error) {
 	revisionNode := node.New(session.SpaceID(), session.NodeID(), "", "", session.Size(), session.ID(),
 		provider.ResourceType_RESOURCE_TYPE_FILE, session.SpaceOwner(), session.store.lu)

-	switch spaceRoot, err := session.store.lu.NodeFromSpaceID(ctx, session.SpaceID()); {
+	var (
+		spaceRoot *node.Node
+	)
+	switch spaceRoot, err = session.store.lu.NodeFromSpaceID(ctx, session.SpaceID()); {
 	case err != nil:
 		return fmt.Errorf("failed to get space root for space id %s: %v", session.SpaceID(), err)
 	case spaceRoot == nil:
@@ -310,6 +313,31 @@ func (session *DecomposedFsSession) Finalize(ctx context.Context) (err error) {
 	}
 	defer func() { _ = unlock() }()

+	isProcessing := revisionNode.IsProcessing(ctx)
+	var procssingID string
+	if isProcessing {
+		procssingID, _ = revisionNode.ProcessingID(ctx)
+	}
+
+	// another upload on this node is in progress or has finished since we started
+	if !isProcessing || procssingID != session.ID() {
+		versionID := revisionNode.ID + node.RevisionIDDelimiter + session.MTime().UTC().Format(time.RFC3339Nano)
+		revisionNode, err = node.ReadNode(ctx, session.store.lu, session.SpaceID(), versionID, false, spaceRoot, false)
+		if err != nil {
+			return fmt.Errorf("failed to read revision node %s for upload finalization: %w", versionID, err)
+		}
+		if !revisionNode.Exists {
+			return fmt.Errorf("revision node %s for upload finalization does not exist", versionID)
+		}
+		// lock this node as well, before writing the blob
+		revisionNodeUnlock, err := session.store.lu.MetadataBackend().Lock(revisionNode)
+		if err != nil {
+			return err
+		}
+		appctx.GetLogger(ctx).Debug().Str("new nodepath", revisionNode.InternalPath()).Msg("uploading to revision node, that was created for us by another upload")
+		defer func() { _ = revisionNodeUnlock() }()
+	}
+
 	// upload the data to the blobstore
 	_, subspan := tracer.Start(ctx, "WriteBlob")
 	err = session.store.tp.WriteBlob(revisionNode, session.binPath())
@@ -343,35 +371,45 @@ func (session *DecomposedFsSession) removeNode(ctx context.Context) {
 // cleanup cleans up after the upload is finished
 func (session *DecomposedFsSession) Cleanup(revertNodeMetadata, cleanBin, cleanInfo bool) {
 	ctx := session.Context(context.Background())
+	sublog := session.store.log.With().Str("cleanup sessionid", session.ID()).Bool("revertNodeMetadata", revertNodeMetadata).Bool("cleanBin", cleanBin).
+		Bool("cleanInfo", cleanInfo).Logger()

 	if revertNodeMetadata {
 		n, err := session.Node(ctx)
 		if err != nil {
-			appctx.GetLogger(ctx).Error().Err(err).Str("sessionid", session.ID()).Msg("reading node for session failed")
+			sublog.Error().Err(err).Msg("reading node for session failed")
 		} else {
 			if session.NodeExists() && session.info.MetaData["versionID"] != "" {
 				versionID := session.info.MetaData["versionID"]
-				revisionNode := node.NewBaseNode(n.SpaceID, versionID, session.store.lu)
+				sublog.Debug().Str("nodepath", n.InternalPath()).Str("versionID", versionID).Msg("restoring revision")
+				revisionNode, err := node.ReadNode(ctx, session.store.lu, session.SpaceID(), versionID, false, n.SpaceRoot, false)
+				if err != nil {
+					sublog.Error().Err(err).Str("versionID", versionID).Msg("reading revision node failed")
+				}

-				if err := session.store.lu.CopyMetadata(ctx, revisionNode, n, func(attributeName string, value []byte) (newValue []byte, copy bool) {
-					return value, strings.HasPrefix(attributeName, prefixes.ChecksumPrefix) ||
-						attributeName == prefixes.TypeAttr ||
-						attributeName == prefixes.BlobIDAttr ||
-						attributeName == prefixes.BlobsizeAttr ||
-						attributeName == prefixes.MTimeAttr
-				}, true); err != nil {
-					appctx.GetLogger(ctx).Info().Str("version", versionID).Str("nodepath", n.InternalPath()).Err(err).Msg("renaming version node failed")
+				if !revisionNode.Exists {
+					sublog.Error().Str("versionID", versionID).Msg("revision node does not exist")
+				}
+
+				// restore the revision
+				mtime, err := revisionNode.GetMTime(ctx)
+				if err != nil {
+					sublog.Error().Err(err).Str("versionID", versionID).Msg("getting mtime of revision node failed")
+					mtime = time.Now()
+				}
+
+				if err := session.store.tp.RestoreRevision(ctx, revisionNode, n, mtime); err != nil {
+					sublog.Error().Err(err).Str("versionID", versionID).Msg("restoring revision node failed")
 				}

 				if err := os.RemoveAll(revisionNode.InternalPath()); err != nil {
-					appctx.GetLogger(ctx).Info().Str("version", versionID).Str("nodepath", n.InternalPath()).Err(err).Msg("error removing version")
+					sublog.Error().Err(err).Str("revisionpath", revisionNode.InternalPath()).Msg("removing restored revision file failed")
 				}
-
 			} else {
 				// if no other upload session is in progress (processing id != session id) or has finished (processing id == "")
 				latestSession, err := n.ProcessingID(ctx)
 				if err != nil {
-					appctx.GetLogger(ctx).Error().Err(err).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("uploadid", session.ID()).Msg("reading processingid for session failed")
+					sublog.Error().Err(err).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("uploadid", session.ID()).Msg("reading processingid for session failed")
 				}
 				if latestSession == session.ID() {
 					// actually delete the node