feat: OCISDEV-855

internal/grpc/services/storageprovider/storageprovider.go

@@ -452,6 +452,8 @@ func (s *Service) InitiateFileUpload(ctx context.Context, req *provider.Initiate
 			st = status.NewFailedPrecondition(ctx, err, "failed precondition")
 		case errtypes.Locked:
 			st = status.NewLocked(ctx, "locked")
+		case errtypes.IsTooEarly:
+			st = status.NewTooEarly(ctx, err.Error())
 		default:
 			st = status.NewInternal(ctx, "error getting upload id: "+err.Error())
 		}

pkg/rhttp/datatx/utils/download/download.go

@@ -275,6 +275,9 @@ func handleError(w http.ResponseWriter, log *zerolog.Logger, err error, action s
 	case errtypes.Aborted:
 		log.Debug().Err(err).Str("action", action).Msg("etags do not match")
 		w.WriteHeader(http.StatusPreconditionFailed)
+	case errtypes.IsResourceProcessing, errtypes.IsTooEarly:
+		log.Debug().Err(err).Str("action", action).Msg("resource is processing")
+		w.WriteHeader(http.StatusTooEarly)
 	default:
 		log.Error().Err(err).Str("action", action).Msg("unexpected error")
 		w.WriteHeader(http.StatusInternalServerError)

pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go

@@ -157,6 +157,10 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati
 			// CS3 uses an already exists error instead of precondition failed when using an If-None-Match=* header / IfExists flag in the InitiateFileUpload call.
 			// Thas happens when the cache thinks there is no file.
 			// continue with sync below
+		case errtypes.TooEarly:
+			log.Debug().Msg("upload slot busy when persisting received share: retrying...")
+			// storage-system has an upload in progress for this node; wait for it to finish
+			// continue with sync below
 		default:
 			span.SetStatus(codes.Error, fmt.Sprintf("persisting added received share failed. giving up: %s", err.Error()))
 			log.Error().Err(err).Msg("persisting added received share failed")
@@ -169,11 +173,14 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati
 			timer.Stop()
 			return ctx.Err()
 		}
-		if err := c.syncWithLock(ctx, userID); err != nil {
-			span.RecordError(err)
-			span.SetStatus(codes.Error, err.Error())
-			log.Error().Err(err).Msg("persisting added received share failed. giving up.")
-			return err
+		if serr := c.syncWithLock(ctx, userID); serr != nil {
+			if _, ok := serr.(errtypes.IsTooEarly); !ok {
+				span.RecordError(serr)
+				span.SetStatus(codes.Error, serr.Error())
+				log.Error().Err(serr).Msg("persisting added received share failed. giving up.")
+				return serr
+			}
+			log.Debug().Msg("sync skipped: resource is processing, will retry")
 		}
 	}
 	return err
@@ -256,6 +263,10 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err
 			// CS3 uses an already exists error instead of precondition failed when using an If-None-Match=* header / IfExists flag in the InitiateFileUpload call.
 			// Thas happens when the cache thinks there is no file.
 			// continue with sync below
+		case errtypes.TooEarly:
+			log.Debug().Msg("upload slot busy when persisting received share: retrying...")
+			// storage-system has an upload in progress for this node; wait for it to finish
+			// continue with sync below
 		default:
 			span.SetStatus(codes.Error, fmt.Sprintf("persisting added received share failed. giving up: %s", err.Error()))
 			log.Error().Err(err).Msg("persisting added received share failed")
@@ -268,11 +279,14 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err
 			timer.Stop()
 			return ctx.Err()
 		}
-		if err := c.syncWithLock(ctx, userID); err != nil {
-			span.RecordError(err)
-			span.SetStatus(codes.Error, err.Error())
-			log.Error().Err(err).Msg("persisting added received share failed. giving up.")
-			return err
+		if serr := c.syncWithLock(ctx, userID); serr != nil {
+			if _, ok := serr.(errtypes.IsTooEarly); !ok {
+				span.RecordError(serr)
+				span.SetStatus(codes.Error, serr.Error())
+				log.Error().Err(serr).Msg("persisting added received share failed. giving up.")
+				return serr
+			}
+			log.Debug().Msg("sync skipped: resource is processing, will retry")
 		}
 	}
 	return err
@@ -365,6 +379,9 @@ func (c *Cache) persist(ctx context.Context, userID string) error {
 	defer span.End()
 	span.SetAttributes(attribute.String("cs3.userid", userID))
 
+	log := appctx.GetLogger(ctx)
+	log.Debug().Str("user", userID).Msg("receivedsharecache:persist:start")
+
 	rss, ok := c.ReceivedSpaces.Load(userID)
 	if !ok {
 		span.SetStatus(codes.Ok, "no received shares")
@@ -395,6 +412,7 @@ func (c *Cache) persist(ctx context.Context, userID string) error {
 		ur.IfNoneMatch = []string{"*"}
 	}
 
+	log.Debug().Str("user", userID).Str("path", jsonPath).Str("etag", ur.IfMatchEtag).Msg("receivedsharecache:persist:upload")
 	res, err := c.storage.Upload(ctx, ur)
 	if err != nil {
 		span.RecordError(err)
@@ -403,6 +421,7 @@ func (c *Cache) persist(ctx context.Context, userID string) error {
 	}
 	rss.etag = res.Etag
 
+	log.Debug().Str("user", userID).Str("etag", res.Etag).Msg("receivedsharecache:persist:done")
 	span.SetStatus(codes.Ok, "")
 	return nil
 }

pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go

@@ -20,16 +20,17 @@ package receivedsharecache_test
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
+	"github.com/owncloud/reva/v2/pkg/appctx"
 	"github.com/owncloud/reva/v2/pkg/errtypes"
 	"github.com/owncloud/reva/v2/pkg/share/manager/jsoncs3/receivedsharecache"
 	"github.com/owncloud/reva/v2/pkg/storage/utils/metadata"
+	"github.com/rs/zerolog"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -53,7 +54,8 @@ var _ = Describe("Cache", func() {
 	)
 
 	BeforeEach(func() {
-		ctx = context.Background()
+		zl := zerolog.New(os.Stdout).Level(zerolog.DebugLevel)
+		ctx = appctx.WithLogger(context.Background(), &zl)
 
 		var err error
 		tmpdir, err = os.MkdirTemp("", "providercache-test")
@@ -149,46 +151,42 @@ var _ = Describe("Cache", func() {
 	})
 
 	Describe("concurrent writes from multiple cache instances", func() {
-		It("preserves all shares when replicas write concurrently for the same user", func() {
-			const numReplicas = 3
-			const numShares = 15
+		It("preserves the share when 15 replicas write the same file simultaneously", func() {
+			const numReplicas = 15
 
-			// barrierStorage holds all Upload calls until numReplicas have arrived,
-			// then releases them simultaneously. This makes the race deterministic
-			// regardless of OS goroutine scheduling or GOMAXPROCS.
+			// barrier releases all 15 Upload calls at once — every replica is a loser
+			// except one, maximising retry pressure on a single shared file.
 			bs := newBarrierStorage(storage, numReplicas)
 			replicas := make([]receivedsharecache.Cache, numReplicas)
 			for i := range replicas {
 				replicas[i] = receivedsharecache.New(bs, 0*time.Second)
 			}
 
-			errs := make([]error, numShares)
+			errs := make([]error, numReplicas)
 			var wg sync.WaitGroup
-			for i := 0; i < numShares; i++ {
+			for i := 0; i < numReplicas; i++ {
 				wg.Add(1)
 				go func(idx int) {
 					defer wg.Done()
 					rs := &collaboration.ReceivedShare{
 						Share: &collaboration.Share{
-							Id: &collaboration.ShareId{OpaqueId: fmt.Sprintf("share-%d", idx)},
+							Id: &collaboration.ShareId{OpaqueId: "share-0"},
 						},
 						State: collaboration.ShareState_SHARE_STATE_PENDING,
 					}
-					errs[idx] = replicas[idx%numReplicas].Add(ctx, userID, spaceID, rs)
+					errs[idx] = replicas[idx].Add(ctx, userID, spaceID, rs)
 				}(i)
 			}
 			wg.Wait()
 			for i, err := range errs {
-				Expect(err).ToNot(HaveOccurred(), "Add failed for share-%d", i)
+				Expect(err).ToNot(HaveOccurred(), "replica %d failed", i)
 			}
 
 			fresh := receivedsharecache.New(storage, 0*time.Second)
 			spaces, err := fresh.List(ctx, userID)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(spaces[spaceID]).ToNot(BeNil())
-			for i := 0; i < numShares; i++ {
-				Expect(spaces[spaceID].States).To(HaveKey(fmt.Sprintf("share-%d", i)))
-			}
+			Expect(spaces[spaceID].States).To(HaveKey("share-0"))
 		})
 	})
 

pkg/storage/utils/decomposedfs/upload.go

@@ -136,6 +136,7 @@ func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Refere
 	_, span := tracer.Start(ctx, "InitiateUpload")
 	defer span.End()
 	log := appctx.GetLogger(ctx)
+	log.Debug().Interface("ref", ref).Msg("decomposedfs:InitiateUpload:start")
 
 	// remember the path from the reference
 	refpath := ref.GetPath()
@@ -336,6 +337,7 @@ func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Refere
 		}
 	}
 
+	log.Debug().Str("uploadid", session.ID()).Msg("decomposedfs:InitiateUpload:complete")
 	return map[string]string{
 		"simple": session.ID(),
 		"tus":    session.ID(),

pkg/storage/utils/decomposedfs/upload/upload.go

@@ -74,6 +74,8 @@ func (session *OcisSession) WriteChunk(ctx context.Context, offset int64, src io
 	}
 	defer file.Close()
 
+	log.Debug().Int64("offset", offset).Msg("decomposedfs:WriteChunk:start")
+
 	// calculate cheksum here? needed for the TUS checksum extension. https://tus.io/protocols/resumable-upload.html#checksum
 	// TODO but how do we get the `Upload-Checksum`? WriteChunk() only has a context, offset and the reader ...
 	// It is sent with the PATCH request, well or in the POST when the creation-with-upload extension is used
@@ -95,6 +97,7 @@ func (session *OcisSession) WriteChunk(ctx context.Context, offset int64, src io
 	// No need to persist the session as the offset is determined by stating the blob in the GetUpload / ReadSession codepath.
 	// The session offset is written to disk in FinishUpload
 	session.info.Offset += n
+	log.Debug().Int64("n", n).Msg("decomposedfs:WriteChunk:complete")
 	return n, nil
 }
 
@@ -114,12 +117,13 @@ func (session *OcisSession) GetReader(ctx context.Context) (io.ReadCloser, error
 // implements tusd.DataStore interface
 // returns tusd errors
 func (session *OcisSession) FinishUpload(ctx context.Context) error {
+	log := appctx.GetLogger(ctx)
+	log.Debug().Msg("decomposedfs:FinishUpload:start")
 	err := session.FinishUploadDecomposed(ctx)
 
 	if err != nil {
 		// this is part of the tusd integration and we might be able to
 		// log the error in another place
-		log := appctx.GetLogger(ctx)
 		log.Error().Err(err).Msg("failed to finish upload")
 	}
 
@@ -140,6 +144,7 @@ func (session *OcisSession) FinishUploadDecomposed(ctx context.Context) error {
 	ctx, span := tracer.Start(session.Context(ctx), "FinishUpload")
 	defer span.End()
 	log := appctx.GetLogger(ctx)
+	log.Debug().Str("session", session.ID()).Msg("decomposedfs:FinishUploadDecomposed:start")
 
 	ctx = ctxpkg.ContextSetInitiator(ctx, session.InitiatorID())
 
@@ -246,6 +251,7 @@ func (session *OcisSession) FinishUploadDecomposed(ctx context.Context) error {
 		metrics.UploadSessionsFinalized.Inc()
 	}
 
+	log.Debug().Str("session", session.ID()).Msg("decomposedfs:FinishUploadDecomposed:complete")
 	return session.store.tp.Propagate(ctx, n, session.SizeDiff())
 }
 

pkg/storage/utils/decomposedfs/upload_async_test.go

@@ -11,6 +11,7 @@ import (
 	cs3permissions "github.com/cs3org/go-cs3apis/cs3/permissions/v1beta1"
 	v1beta11 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/owncloud/reva/v2/pkg/appctx"
 	ruser "github.com/owncloud/reva/v2/pkg/ctx"
 	"github.com/owncloud/reva/v2/pkg/events"
 	"github.com/owncloud/reva/v2/pkg/events/stream"
@@ -68,9 +69,9 @@ var _ = Describe("Async file uploads", Ordered, func() {
 		firstContent  = []byte("0123456789")
 		secondContent = []byte("01234567890123456789")
 
-		ctx = ruser.ContextSetUser(context.Background(), user)
+		ctx context.Context
 
-		pub      chan interface{}
+		pub chan interface{}
 		con      chan interface{}
 		uploadID string
 
@@ -132,6 +133,9 @@ var _ = Describe("Async file uploads", Ordered, func() {
 	)
 
 	BeforeEach(func() {
+		zl := zerolog.New(os.Stdout).Level(zerolog.DebugLevel)
+		ctx = appctx.WithLogger(ruser.ContextSetUser(context.Background(), user), &zl)
+
 		// setup test
 		tmpRoot, err := helpers.TempDir("reva-unit-tests-*-root")
 		Expect(err).ToNot(HaveOccurred())

pkg/storage/utils/metadata/cs3.go

@@ -30,6 +30,7 @@ import (
 	"time"
 
 	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
+	"github.com/rs/zerolog"
 	user "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
@@ -53,6 +54,7 @@ func init() {
 
 // CS3 represents a metadata storage with a cs3 storage backend
 type CS3 struct {
+	log       zerolog.Logger
 	SpaceRoot *provider.ResourceId
 
 	providerAddr      string
@@ -64,6 +66,9 @@ type CS3 struct {
 	dataGatewayClient *http.Client
 }
 
+// SetLogger injects a logger; useful in tests to capture trace output.
+func (cs3 *CS3) SetLogger(log zerolog.Logger) { cs3.log = log }
+
 // NewCS3 returns a new CS3 instance. Use an authenticated context and be sure to define SpaceRoot manually.
 func NewCS3(gwAddr, providerAddr string) (s *CS3) {
 	return &CS3{
@@ -162,6 +167,7 @@ func (cs3 *CS3) SimpleUpload(ctx context.Context, uploadpath string, content []b
 	ctx, span := tracer.Start(ctx, "SimpleUpload")
 	defer span.End()
 
+	cs3.log.Debug().Str("path", uploadpath).Msg("cs3:SimpleUpload:start")
 	_, err := cs3.Upload(ctx, UploadRequest{
 		Path:    uploadpath,
 		Content: content,
@@ -174,6 +180,8 @@ func (cs3 *CS3) Upload(ctx context.Context, req UploadRequest) (*UploadResponse,
 	ctx, span := tracer.Start(ctx, "Upload")
 	defer span.End()
 
+	cs3.log.Debug().Str("path", req.Path).Msg("cs3:Upload:start")
+
 	client, err := cs3.providerClient()
 	if err != nil {
 		return nil, err
@@ -238,6 +246,8 @@ func (cs3 *CS3) Upload(ctx context.Context, req UploadRequest) (*UploadResponse,
 		return nil, errors.New("metadata storage doesn't support the simple upload protocol")
 	}
 
+	cs3.log.Debug().Str("path", req.Path).Str("endpoint", endpoint).Msg("cs3:Upload:initiate_done")
+
 	httpReq, err := http.NewRequest(http.MethodPut, endpoint, bytes.NewReader(req.Content))
 	if err != nil {
 		return nil, err
@@ -253,13 +263,15 @@ func (cs3 *CS3) Upload(ctx context.Context, req UploadRequest) (*UploadResponse,
 		return nil, err
 	}
 	defer resp.Body.Close()
+	cs3.log.Debug().Str("path", req.Path).Int("status", resp.StatusCode).Msg("cs3:Upload:put_done")
 	if err := errtypes.NewErrtypeFromHTTPStatusCode(resp.StatusCode, httpReq.URL.Path); err != nil {
 		return nil, err
 	}
 	etag := resp.Header.Get("Etag")
 	if ocEtag := resp.Header.Get("OC-ETag"); ocEtag != "" {
 		etag = ocEtag
 	}
+	cs3.log.Debug().Str("path", req.Path).Str("etag", etag).Msg("cs3:Upload:complete")
 	return &UploadResponse{
 		Etag:   etag,
 		FileID: resp.Header.Get("OC-Fileid"),

tests/integration/grpc/fixtures/storageprovider-ocis-with-dataprovider.toml

@@ -0,0 +1,32 @@
+[shared]
+jwt_secret = "changemeplease"
+
+[grpc]
+address = "{{grpc_address}}"
+
+[grpc.services.storageprovider]
+driver = "ocis"
+data_server_url = "http://{{grpc_address+1}}/data"
+expose_data_server = true
+
+[grpc.services.storageprovider.drivers.ocis]
+root = "{{root}}/storage"
+treetime_accounting = true
+treesize_accounting = true
+permissionssvc = "{{permissions_address}}"
+
+[grpc.services.storageprovider.drivers.ocis.filemetadatacache]
+cache_store = "noop"
+
+[http]
+address = "{{grpc_address+1}}"
+
+[http.services.dataprovider]
+driver = "ocis"
+
+[http.services.dataprovider.drivers.ocis]
+root = "{{root}}/storage"
+permissionssvc = "{{permissions_address}}"
+
+[http.services.dataprovider.drivers.ocis.filemetadatacache]
+cache_store = "noop"