From ce19554febdddaa80880e14d471543c34a4dc802 Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Wed, 15 Jul 2026 03:22:20 +0200 Subject: [PATCH 1/2] fix: ocisdev-855, cs3/decomposefs transactional concurent upload --- .../storageprovider/storageprovider.go | 2 + pkg/rhttp/datatx/utils/download/download.go | 3 + .../receivedsharecache/receivedsharecache.go | 39 +++- .../receivedsharecache_test.go | 30 ++- pkg/storage/utils/decomposedfs/upload.go | 24 +++ .../utils/decomposedfs/upload/upload.go | 8 +- .../utils/decomposedfs/upload_async_test.go | 70 ++++++- pkg/storage/utils/metadata/cs3.go | 12 ++ ...torageprovider-ocis-with-dataprovider.toml | 32 +++ .../receivedsharecache_concurrent_test.go | 197 ++++++++++++++++++ 10 files changed, 388 insertions(+), 29 deletions(-) create mode 100644 tests/integration/grpc/fixtures/storageprovider-ocis-with-dataprovider.toml create mode 100644 tests/integration/grpc/receivedsharecache_concurrent_test.go diff --git a/internal/grpc/services/storageprovider/storageprovider.go b/internal/grpc/services/storageprovider/storageprovider.go index 01fd4045b7f..b23604a275f 100644 --- a/internal/grpc/services/storageprovider/storageprovider.go +++ b/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()) } diff --git a/pkg/rhttp/datatx/utils/download/download.go b/pkg/rhttp/datatx/utils/download/download.go index d4b5fbd31e9..3fa09787a42 100644 --- a/pkg/rhttp/datatx/utils/download/download.go +++ b/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) diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go index 8c1bf66898c..8f4c3704c19 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go +++ b/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 } diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go index bb4e0f35e5d..e0047d1e8f3 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go +++ b/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")) }) }) diff --git a/pkg/storage/utils/decomposedfs/upload.go b/pkg/storage/utils/decomposedfs/upload.go index abacbd25cd9..abc709b5868 100644 --- a/pkg/storage/utils/decomposedfs/upload.go +++ b/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() @@ -295,6 +296,28 @@ func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Refere return nil, err } + if n.Exists { + // Atomically reserve the upload slot. If another session is already uploading + // this node, reject immediately so the caller retries with the updated etag + // rather than racing to FinishUpload and corrupting the node on cleanup. + f, err := lockedfile.OpenFile(fs.lu.MetadataBackend().LockfilePath(n.InternalPath()), os.O_RDWR|os.O_CREATE, 0600) + if err != nil { + return nil, err + } + n.ResetXattrsCache() + if n.IsProcessing(ctx) { + _ = f.Close() + return nil, errtypes.TooEarly("upload in progress for node " + n.ID) + } + if err := n.SetXattrsWithContext(ctx, node.Attributes{ + prefixes.StatusPrefix: []byte(node.ProcessingStatus + session.ID()), + }, false); err != nil { + _ = f.Close() + return nil, err + } + _ = f.Close() + } + usr := ctxpkg.ContextMustGetUser(ctx) // fill future node info @@ -336,6 +359,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(), diff --git a/pkg/storage/utils/decomposedfs/upload/upload.go b/pkg/storage/utils/decomposedfs/upload/upload.go index 1a42f20878b..43c347cbfac 100644 --- a/pkg/storage/utils/decomposedfs/upload/upload.go +++ b/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()) } diff --git a/pkg/storage/utils/decomposedfs/upload_async_test.go b/pkg/storage/utils/decomposedfs/upload_async_test.go index 35827ec15c0..d3b688cdff2 100644 --- a/pkg/storage/utils/decomposedfs/upload_async_test.go +++ b/pkg/storage/utils/decomposedfs/upload_async_test.go @@ -11,7 +11,9 @@ 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/errtypes" "github.com/owncloud/reva/v2/pkg/events" "github.com/owncloud/reva/v2/pkg/events/stream" "github.com/owncloud/reva/v2/pkg/rgrpc/todo/pool" @@ -68,9 +70,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 +134,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()) @@ -403,6 +408,67 @@ var _ = Describe("Async file uploads", Ordered, func() { }) }) + When("Two uploads are processed sequentially (TooEarly prevents parallel uploads)", func() { + var secondUploadID string + + JustBeforeEach(func() { + // first upload must finish before second can start + succeedPostprocessing(uploadID) + + // upload again - only possible now that ProcessingID is cleared + uploadIds, err := fs.InitiateUpload(ctx, ref, 20, map[string]string{}) + Expect(err).ToNot(HaveOccurred()) + Expect(len(uploadIds)).To(Equal(2)) + Expect(uploadIds["simple"]).ToNot(BeEmpty()) + Expect(uploadIds["tus"]).ToNot(BeEmpty()) + + uploadRef := &provider.Reference{Path: "/" + uploadIds["simple"]} + + _, err = fs.Upload(ctx, storage.UploadRequest{ + Ref: uploadRef, + Body: io.NopCloser(bytes.NewReader(secondContent)), + Length: int64(len(secondContent)), + }, nil) + Expect(err).ToNot(HaveOccurred()) + + secondUploadID = uploadIds["simple"] + + // wait for bytes received event + _, ok := (<-pub).(events.BytesReceived) + Expect(ok).To(BeTrue()) + }) + + It("rejects a concurrent InitiateUpload with TooEarly", func() { + // secondUploadID session is still in-progress (BytesReceived published, not yet finished) + _, err := fs.InitiateUpload(ctx, ref, 20, map[string]string{}) + Expect(err).To(HaveOccurred()) + _, ok := err.(errtypes.IsTooEarly) + Expect(ok).To(BeTrue()) + + // node is visible but locked: status=processing, not yet readable + exists, status, _ := fileStatus() + Expect(exists).To(BeTrue()) + Expect(status).To(Equal("processing")) + }) + + It("succeeds when processed", func() { + succeedPostprocessing(secondUploadID) + + _, status, size := fileStatus() + Expect(status).To(Equal("")) + Expect(size).To(Equal(len(secondContent))) + Expect(parentSize()).To(Equal(len(secondContent))) + }) + + It("restores the previous version when deleted", func() { + failPostprocessing(secondUploadID, events.PPOutcomeDelete) + + _, status, size := fileStatus() + Expect(status).To(Equal("")) + Expect(size).To(Equal(len(firstContent))) + Expect(parentSize()).To(Equal(len(firstContent))) + }) + }) When("Two uploads are processed in parallel", func() { var secondUploadID string diff --git a/pkg/storage/utils/metadata/cs3.go b/pkg/storage/utils/metadata/cs3.go index 939b929f59f..517fdcbda10 100644 --- a/pkg/storage/utils/metadata/cs3.go +++ b/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,6 +263,7 @@ 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 } @@ -260,6 +271,7 @@ func (cs3 *CS3) Upload(ctx context.Context, req UploadRequest) (*UploadResponse, 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"), diff --git a/tests/integration/grpc/fixtures/storageprovider-ocis-with-dataprovider.toml b/tests/integration/grpc/fixtures/storageprovider-ocis-with-dataprovider.toml new file mode 100644 index 00000000000..102c9ac9af6 --- /dev/null +++ b/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" diff --git a/tests/integration/grpc/receivedsharecache_concurrent_test.go b/tests/integration/grpc/receivedsharecache_concurrent_test.go new file mode 100644 index 00000000000..955015ec91b --- /dev/null +++ b/tests/integration/grpc/receivedsharecache_concurrent_test.go @@ -0,0 +1,197 @@ +package grpc_test + +import ( + "context" + "fmt" + "os" + "sync" + "sync/atomic" + + grpcMetadata "google.golang.org/grpc/metadata" + + userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" + collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1" + provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" + "github.com/owncloud/reva/v2/pkg/appctx" + "github.com/owncloud/reva/v2/pkg/auth/scope" + ctxpkg "github.com/owncloud/reva/v2/pkg/ctx" + "github.com/owncloud/reva/v2/pkg/rgrpc/todo/pool" + "github.com/owncloud/reva/v2/pkg/share/manager/jsoncs3/receivedsharecache" + "github.com/owncloud/reva/v2/pkg/storage/utils/metadata" + jwt "github.com/owncloud/reva/v2/pkg/token/manager/jwt" + "github.com/rs/zerolog" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("receivedsharecache concurrent CS3 writes", func() { + var ( + revads map[string]*Revad + ctx context.Context + spaceRoot *provider.ResourceId + + csUser = &userpb.User{ + Id: &userpb.UserId{ + Idp: "0.0.0.0:19000", + OpaqueId: "f7fbf8c8-139b-4376-b307-cf0a8c2d0d9c", + Type: userpb.UserType_USER_TYPE_PRIMARY, + }, + Username: "einstein", + } + + csUserID = "user" + csSpaceID = "spaceid" + ) + + BeforeEach(func() { + var err error + zl := zerolog.New(os.Stdout).Level(zerolog.DebugLevel) + ctx = appctx.WithLogger(context.Background(), &zl) + + tokenManager, err := jwt.New(map[string]interface{}{"secret": "changemeplease"}) + Expect(err).ToNot(HaveOccurred()) + sc, err := scope.AddOwnerScope(nil) + Expect(err).ToNot(HaveOccurred()) + t, err := tokenManager.MintToken(ctx, csUser, sc) + Expect(err).ToNot(HaveOccurred()) + ctx = ctxpkg.ContextSetToken(ctx, t) + ctx = grpcMetadata.AppendToOutgoingContext(ctx, ctxpkg.TokenHeader, t) + ctx = ctxpkg.ContextSetUser(ctx, csUser) + + revads, err = startRevads([]RevadConfig{ + {Name: "storage", Config: "storageprovider-ocis-with-dataprovider.toml"}, + {Name: "permissions", Config: "permissions-ocis-ci.toml"}, + }, nil) + Expect(err).ToNot(HaveOccurred()) + + spacesClient, err := pool.GetSpacesProviderServiceClient(revads["storage"].GrpcAddress) + Expect(err).ToNot(HaveOccurred()) + res, err := spacesClient.CreateStorageSpace(ctx, &provider.CreateStorageSpaceRequest{ + Owner: csUser, + Type: "metadata", + Name: "Metadata", + }) + Expect(err).ToNot(HaveOccurred()) + Expect(res.Status.Code.String()).To(Equal("CODE_OK")) + spaceRoot = res.StorageSpace.Root + + // decomposedfs CreateContainer requires parent to exist; pre-create /users. + setup := metadata.NewCS3("", revads["storage"].GrpcAddress) + setup.SpaceRoot = spaceRoot + Expect(setup.MakeDirIfNotExist(ctx, "/users")).To(Succeed()) + }) + + AfterEach(func() { + for _, r := range revads { + r.Cleanup(CurrentSpecReport().Failed()) //nolint:errcheck + } + pool.RemoveSelector("StorageProviderSelector" + revads["storage"].GrpcAddress) + }) + + It("preserves all shares when 2 replicas write concurrently (OCISDEV-855)", func() { + newCS3 := func() *metadata.CS3 { + cs3 := metadata.NewCS3("", revads["storage"].GrpcAddress) + cs3.SpaceRoot = spaceRoot + cs3.SetLogger(*appctx.GetLogger(ctx)) + return cs3 + } + + const numShares = 15 + replicas := [2]receivedsharecache.Cache{ + receivedsharecache.New(newCS3(), 0), + receivedsharecache.New(newCS3(), 0), + } + + errs := make([]error, numShares) + var wg sync.WaitGroup + for i := 0; i < numShares; 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)}, + }, + State: collaboration.ShareState_SHARE_STATE_PENDING, + } + errs[idx] = replicas[idx%2].Add(ctx, csUserID, csSpaceID, rs) + }(i) + } + wg.Wait() + for i, err := range errs { + Expect(err).ToNot(HaveOccurred(), "Add failed for share-%d", i) + } + + fresh := receivedsharecache.New(newCS3(), 0) + spaces, err := fresh.List(ctx, csUserID) + Expect(err).ToNot(HaveOccurred()) + Expect(spaces[csSpaceID]).ToNot(BeNil()) + for i := 0; i < numShares; i++ { + Expect(spaces[csSpaceID].States).To(HaveKey(fmt.Sprintf("share-%d", i))) + } + }) + + It("both replicas recover when writes are forced simultaneous (OCISDEV-855)", func() { + newCS3 := func() *metadata.CS3 { + cs3 := metadata.NewCS3("", revads["storage"].GrpcAddress) + cs3.SpaceRoot = spaceRoot + cs3.SetLogger(*appctx.GetLogger(ctx)) + return cs3 + } + + bs := newBarrierStorageCS3(newCS3(), 2) + replicas := [2]receivedsharecache.Cache{ + receivedsharecache.New(bs, 0), + receivedsharecache.New(bs, 0), + } + + errs := make([]error, 2) + var wg sync.WaitGroup + for i := 0; i < 2; 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)}, + }, + State: collaboration.ShareState_SHARE_STATE_PENDING, + } + errs[idx] = replicas[idx].Add(ctx, csUserID, csSpaceID, rs) + }(i) + } + wg.Wait() + Expect(errs[0]).ToNot(HaveOccurred()) + Expect(errs[1]).ToNot(HaveOccurred()) + + fresh := receivedsharecache.New(newCS3(), 0) + spaces, err := fresh.List(ctx, csUserID) + Expect(err).ToNot(HaveOccurred()) + Expect(spaces[csSpaceID]).ToNot(BeNil()) + Expect(spaces[csSpaceID].States).To(HaveKey("share-0")) + Expect(spaces[csSpaceID].States).To(HaveKey("share-1")) + }) +}) + +// barrierStorageCS3 holds Upload calls until n goroutines have arrived, then +// releases all simultaneously — forcing the concurrent-write race deterministically. +type barrierStorageCS3 struct { + metadata.Storage + arrived int32 + n int32 + ready chan struct{} + closeOnce sync.Once +} + +func newBarrierStorageCS3(s metadata.Storage, n int) *barrierStorageCS3 { + return &barrierStorageCS3{Storage: s, n: int32(n), ready: make(chan struct{})} +} + +func (b *barrierStorageCS3) Upload(ctx context.Context, req metadata.UploadRequest) (*metadata.UploadResponse, error) { + if atomic.AddInt32(&b.arrived, 1) >= b.n { + b.closeOnce.Do(func() { close(b.ready) }) + } + <-b.ready + return b.Storage.Upload(ctx, req) +} From 4fe054030e8f2b1ba63491326fd77ec45c686044 Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Wed, 15 Jul 2026 09:48:36 +0200 Subject: [PATCH 2/2] fix: ocisdev-855, cs3/decomposefs optimistic-abort concurent upload --- pkg/storage/utils/decomposedfs/upload.go | 22 ------- .../utils/decomposedfs/upload_async_test.go | 62 ------------------- 2 files changed, 84 deletions(-) diff --git a/pkg/storage/utils/decomposedfs/upload.go b/pkg/storage/utils/decomposedfs/upload.go index abc709b5868..049dfdebc5d 100644 --- a/pkg/storage/utils/decomposedfs/upload.go +++ b/pkg/storage/utils/decomposedfs/upload.go @@ -296,28 +296,6 @@ func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Refere return nil, err } - if n.Exists { - // Atomically reserve the upload slot. If another session is already uploading - // this node, reject immediately so the caller retries with the updated etag - // rather than racing to FinishUpload and corrupting the node on cleanup. - f, err := lockedfile.OpenFile(fs.lu.MetadataBackend().LockfilePath(n.InternalPath()), os.O_RDWR|os.O_CREATE, 0600) - if err != nil { - return nil, err - } - n.ResetXattrsCache() - if n.IsProcessing(ctx) { - _ = f.Close() - return nil, errtypes.TooEarly("upload in progress for node " + n.ID) - } - if err := n.SetXattrsWithContext(ctx, node.Attributes{ - prefixes.StatusPrefix: []byte(node.ProcessingStatus + session.ID()), - }, false); err != nil { - _ = f.Close() - return nil, err - } - _ = f.Close() - } - usr := ctxpkg.ContextMustGetUser(ctx) // fill future node info diff --git a/pkg/storage/utils/decomposedfs/upload_async_test.go b/pkg/storage/utils/decomposedfs/upload_async_test.go index d3b688cdff2..1ede4b72d61 100644 --- a/pkg/storage/utils/decomposedfs/upload_async_test.go +++ b/pkg/storage/utils/decomposedfs/upload_async_test.go @@ -13,7 +13,6 @@ import ( 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/errtypes" "github.com/owncloud/reva/v2/pkg/events" "github.com/owncloud/reva/v2/pkg/events/stream" "github.com/owncloud/reva/v2/pkg/rgrpc/todo/pool" @@ -408,67 +407,6 @@ var _ = Describe("Async file uploads", Ordered, func() { }) }) - When("Two uploads are processed sequentially (TooEarly prevents parallel uploads)", func() { - var secondUploadID string - - JustBeforeEach(func() { - // first upload must finish before second can start - succeedPostprocessing(uploadID) - - // upload again - only possible now that ProcessingID is cleared - uploadIds, err := fs.InitiateUpload(ctx, ref, 20, map[string]string{}) - Expect(err).ToNot(HaveOccurred()) - Expect(len(uploadIds)).To(Equal(2)) - Expect(uploadIds["simple"]).ToNot(BeEmpty()) - Expect(uploadIds["tus"]).ToNot(BeEmpty()) - - uploadRef := &provider.Reference{Path: "/" + uploadIds["simple"]} - - _, err = fs.Upload(ctx, storage.UploadRequest{ - Ref: uploadRef, - Body: io.NopCloser(bytes.NewReader(secondContent)), - Length: int64(len(secondContent)), - }, nil) - Expect(err).ToNot(HaveOccurred()) - - secondUploadID = uploadIds["simple"] - - // wait for bytes received event - _, ok := (<-pub).(events.BytesReceived) - Expect(ok).To(BeTrue()) - }) - - It("rejects a concurrent InitiateUpload with TooEarly", func() { - // secondUploadID session is still in-progress (BytesReceived published, not yet finished) - _, err := fs.InitiateUpload(ctx, ref, 20, map[string]string{}) - Expect(err).To(HaveOccurred()) - _, ok := err.(errtypes.IsTooEarly) - Expect(ok).To(BeTrue()) - - // node is visible but locked: status=processing, not yet readable - exists, status, _ := fileStatus() - Expect(exists).To(BeTrue()) - Expect(status).To(Equal("processing")) - }) - - It("succeeds when processed", func() { - succeedPostprocessing(secondUploadID) - - _, status, size := fileStatus() - Expect(status).To(Equal("")) - Expect(size).To(Equal(len(secondContent))) - Expect(parentSize()).To(Equal(len(secondContent))) - }) - - It("restores the previous version when deleted", func() { - failPostprocessing(secondUploadID, events.PPOutcomeDelete) - - _, status, size := fileStatus() - Expect(status).To(Equal("")) - Expect(size).To(Equal(len(firstContent))) - Expect(parentSize()).To(Equal(len(firstContent))) - }) - }) When("Two uploads are processed in parallel", func() { var secondUploadID string