From 544622108e46fde8cde7c2e74e4bbe9484310bd0 Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Wed, 15 Jul 2026 03:22:20 +0200 Subject: [PATCH 1/7] 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 9669dcc7641..00885588a53 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 217618ab3cb2e91d8a6837cbb8e2e08bcd3056dc Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Wed, 15 Jul 2026 09:48:36 +0200 Subject: [PATCH 2/7] 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 00885588a53..add62e973cb 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 From f61d4ee038a5778ffe2c1adb25c764b06a7b175e Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Fri, 17 Jul 2026 11:59:33 +0200 Subject: [PATCH 3/7] fix: ocisdev-855, cs3/decomposefs optimistic-abort concurent upload --- .../jsoncs3/receivedsharecache/receivedsharecache.go | 12 ++++++++---- pkg/storage/fs/posix/tree/tree_test.go | 5 +++++ pkg/storage/utils/decomposedfs/tree/tree.go | 3 +++ 3 files changed, 16 insertions(+), 4 deletions(-) diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go index 8f4c3704c19..c50eb8858a1 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go @@ -174,13 +174,15 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati return ctx.Err() } if serr := c.syncWithLock(ctx, userID); serr != nil { - if _, ok := serr.(errtypes.IsTooEarly); !ok { + _, isTooEarly := serr.(errtypes.IsTooEarly) + _, isInternal := serr.(errtypes.IsInternalError) + if !isTooEarly && !isInternal { 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") + log.Debug().Err(serr).Msg("sync transient error, will retry") } } return err @@ -280,13 +282,15 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err return ctx.Err() } if serr := c.syncWithLock(ctx, userID); serr != nil { - if _, ok := serr.(errtypes.IsTooEarly); !ok { + _, isTooEarly := serr.(errtypes.IsTooEarly) + _, isInternal := serr.(errtypes.IsInternalError) + if !isTooEarly && !isInternal { 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") + log.Debug().Err(serr).Msg("sync transient error, will retry") } } return err diff --git a/pkg/storage/fs/posix/tree/tree_test.go b/pkg/storage/fs/posix/tree/tree_test.go index e307d497f32..60e94db39dd 100644 --- a/pkg/storage/fs/posix/tree/tree_test.go +++ b/pkg/storage/fs/posix/tree/tree_test.go @@ -5,6 +5,7 @@ import ( "log" "os" "os/exec" + "runtime" "strings" "time" @@ -40,6 +41,10 @@ var ( ) var _ = SynchronizedBeforeSuite(func() { + if runtime.GOOS != "linux" { + Skip("posix/tree tests require inotifywait (Linux only)") + } + var err error env, err = helpers.NewTestEnv(nil) Expect(err).ToNot(HaveOccurred()) diff --git a/pkg/storage/utils/decomposedfs/tree/tree.go b/pkg/storage/utils/decomposedfs/tree/tree.go index c7c400eb3bc..1b2c397a0e3 100644 --- a/pkg/storage/utils/decomposedfs/tree/tree.go +++ b/pkg/storage/utils/decomposedfs/tree/tree.go @@ -706,6 +706,9 @@ func (t *Tree) InitNewNode(ctx context.Context, n *node.Node, fsize uint64) (met h, err := os.OpenFile(n.InternalPath(), os.O_CREATE|os.O_EXCL, 0600) subspan.End() if err != nil { + if errors.Is(err, fs.ErrExist) { + return unlock, errtypes.AlreadyExists(n.Name) + } return unlock, err } h.Close() From 8aabc440d703acf079418ad65d2fe869110c764a Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Tue, 21 Jul 2026 13:40:25 +0200 Subject: [PATCH 4/7] feat: logs and cleanup --- .../receivedsharecache/receivedsharecache.go | 48 +++++++++++-------- pkg/storage/utils/metadata/cs3.go | 10 ++-- 2 files changed, 32 insertions(+), 26 deletions(-) diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go index c50eb8858a1..b3765669ef8 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go @@ -145,25 +145,25 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati span.SetStatus(codes.Ok, "") return nil case errtypes.Aborted: - log.Debug().Msg("aborted when persisting added received share: etag changed. retrying...") // this is the expected status code from the server when the if-match etag check fails // continue with sync below + log.Debug().Int("attempt", attempt).Msg("CAS failed: Aborted (etag changed), retrying") case errtypes.PreconditionFailed: - log.Debug().Msg("precondition failed when persisting added received share: etag changed. retrying...") // actually, this is the wrong status code and we treat it like errtypes.Aborted because of inconsistencies on the server side // continue with sync below + log.Debug().Int("attempt", attempt).Msg("CAS failed: PreconditionFailed (etag changed), retrying") case errtypes.AlreadyExists: - log.Debug().Msg("already exists when persisting added received share. retrying...") // 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 + log.Debug().Int("attempt", attempt).Msg("CAS failed: AlreadyExists (file created concurrently), retrying") 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 + log.Debug().Int("attempt", attempt).Msg("CAS failed: TooEarly (upload in progress), retrying") 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") + span.SetStatus(codes.Error, fmt.Sprintf("persisting received share failed, giving up: %s", err.Error())) + log.Error().Int("attempt", attempt).Err(err).Msg("persisting received share failed, giving up") return err } timer := time.NewTimer(expBackoff(attempt)) @@ -179,10 +179,10 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati if !isTooEarly && !isInternal { span.RecordError(serr) span.SetStatus(codes.Error, serr.Error()) - log.Error().Err(serr).Msg("persisting added received share failed. giving up.") + log.Error().Int("attempt", attempt).Err(serr).Msg("lost update: re-read failed, aborting") return serr } - log.Debug().Err(serr).Msg("sync transient error, will retry") + log.Warn().Int("attempt", attempt).Err(serr).Msg("lost update: re-read before retry") } } return err @@ -253,25 +253,25 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err span.SetStatus(codes.Ok, "") return nil case errtypes.Aborted: - log.Debug().Msg("aborted when persisting added received share: etag changed. retrying...") // this is the expected status code from the server when the if-match etag check fails // continue with sync below + log.Debug().Int("attempt", attempt).Msg("CAS failed: Aborted (etag changed), retrying") case errtypes.PreconditionFailed: - log.Debug().Msg("precondition failed when persisting added received share: etag changed. retrying...") // actually, this is the wrong status code and we treat it like errtypes.Aborted because of inconsistencies on the server side // continue with sync below + log.Debug().Int("attempt", attempt).Msg("CAS failed: PreconditionFailed (etag changed), retrying") case errtypes.AlreadyExists: - log.Debug().Msg("already exists when persisting added received share. retrying...") // 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 + log.Debug().Int("attempt", attempt).Msg("CAS failed: AlreadyExists (file created concurrently), retrying") 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 + log.Debug().Int("attempt", attempt).Msg("CAS failed: TooEarly (upload in progress), retrying") 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") + span.SetStatus(codes.Error, fmt.Sprintf("persisting received share failed, giving up: %s", err.Error())) + log.Error().Int("attempt", attempt).Err(err).Msg("persisting received share failed, giving up") return err } timer := time.NewTimer(expBackoff(attempt)) @@ -287,10 +287,10 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err if !isTooEarly && !isInternal { span.RecordError(serr) span.SetStatus(codes.Error, serr.Error()) - log.Error().Err(serr).Msg("persisting added received share failed. giving up.") + log.Error().Int("attempt", attempt).Err(serr).Msg("lost update: re-read failed, aborting") return serr } - log.Debug().Err(serr).Msg("sync transient error, will retry") + log.Warn().Int("attempt", attempt).Err(serr).Msg("lost update: re-read before retry") } } return err @@ -358,8 +358,14 @@ func (c *Cache) syncWithLock(ctx context.Context, userID string) error { span.SetStatus(codes.Ok, "") return nil default: - span.SetStatus(codes.Error, fmt.Sprintf("Failed to download the received share: %s", err.Error())) - log.Error().Err(err).Msg("Failed to download the received share") + span.SetStatus(codes.Error, err.Error()) + _, isTooEarly := err.(errtypes.IsTooEarly) + _, isInternal := err.(errtypes.IsInternalError) + if isTooEarly || isInternal { + log.Warn().Err(err).Msg("lost update: re-read transient error") + } else { + log.Error().Err(err).Msg("lost update: re-read failed") + } return err } @@ -384,7 +390,7 @@ func (c *Cache) persist(ctx context.Context, userID string) error { span.SetAttributes(attribute.String("cs3.userid", userID)) log := appctx.GetLogger(ctx) - log.Debug().Str("user", userID).Msg("receivedsharecache:persist:start") + log.Debug().Str("user", userID).Msg("receivedsharecache.persist.start") rss, ok := c.ReceivedSpaces.Load(userID) if !ok { @@ -416,7 +422,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") + 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) @@ -425,7 +431,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") + log.Debug().Str("user", userID).Str("etag", res.Etag).Msg("receivedsharecache.persist.done") span.SetStatus(codes.Ok, "") return nil } diff --git a/pkg/storage/utils/metadata/cs3.go b/pkg/storage/utils/metadata/cs3.go index 517fdcbda10..497936f7f22 100644 --- a/pkg/storage/utils/metadata/cs3.go +++ b/pkg/storage/utils/metadata/cs3.go @@ -167,7 +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") + cs3.log.Debug().Str("path", uploadpath).Msg("cs3.SimpleUpload.start") _, err := cs3.Upload(ctx, UploadRequest{ Path: uploadpath, Content: content, @@ -180,7 +180,7 @@ 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") + cs3.log.Debug().Str("path", req.Path).Msg("cs3.Upload.start") client, err := cs3.providerClient() if err != nil { @@ -246,7 +246,7 @@ 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") + 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 { @@ -263,7 +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") + 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 } @@ -271,7 +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") + cs3.log.Debug().Str("path", req.Path).Str("etag", etag).Msg("cs3.Upload.complete") return &UploadResponse{ Etag: etag, FileID: resp.Header.Get("OC-Fileid"), From e3701d282b843bc071a33ab67e0d045c448c18c1 Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Tue, 21 Jul 2026 14:07:54 +0200 Subject: [PATCH 5/7] feat: cleanup --- .../receivedsharecache/receivedsharecache.go | 149 ++++++------------ .../utils/decomposedfs/upload/upload.go | 3 - 2 files changed, 47 insertions(+), 105 deletions(-) diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go index b3765669ef8..6eec989203d 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go @@ -37,6 +37,7 @@ import ( "github.com/owncloud/reva/v2/pkg/storage/utils/metadata" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" ) // name is the Tracer name used to identify this instrumentation library. @@ -110,7 +111,7 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati defer span.End() span.SetAttributes(attribute.String("cs3.userid", userID), attribute.String("cs3.spaceid", spaceID)) - persistFunc := func() error { + return c.retryPersist(ctx, span, userID, spaceID, func() error { c.initializeIfNeeded(userID, spaceID) rss, _ := c.ReceivedSpaces.Load(userID) @@ -125,67 +126,7 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati } return c.persist(ctx, userID) - } - - log := appctx.GetLogger(ctx).With(). - Str("hostname", os.Getenv("HOSTNAME")). - Str("userID", userID). - Str("spaceID", spaceID).Logger() - - var err error - for attempt := 0; attempt < 20; attempt++ { - select { - case <-ctx.Done(): - return ctx.Err() - default: - } - err = persistFunc() - switch err.(type) { - case nil: - span.SetStatus(codes.Ok, "") - return nil - case errtypes.Aborted: - // this is the expected status code from the server when the if-match etag check fails - // continue with sync below - log.Debug().Int("attempt", attempt).Msg("CAS failed: Aborted (etag changed), retrying") - case errtypes.PreconditionFailed: - // actually, this is the wrong status code and we treat it like errtypes.Aborted because of inconsistencies on the server side - // continue with sync below - log.Debug().Int("attempt", attempt).Msg("CAS failed: PreconditionFailed (etag changed), retrying") - case errtypes.AlreadyExists: - // 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 - log.Debug().Int("attempt", attempt).Msg("CAS failed: AlreadyExists (file created concurrently), retrying") - case errtypes.TooEarly: - // storage-system has an upload in progress for this node; wait for it to finish - // continue with sync below - log.Debug().Int("attempt", attempt).Msg("CAS failed: TooEarly (upload in progress), retrying") - default: - span.SetStatus(codes.Error, fmt.Sprintf("persisting received share failed, giving up: %s", err.Error())) - log.Error().Int("attempt", attempt).Err(err).Msg("persisting received share failed, giving up") - return err - } - timer := time.NewTimer(expBackoff(attempt)) - select { - case <-timer.C: - case <-ctx.Done(): - timer.Stop() - return ctx.Err() - } - if serr := c.syncWithLock(ctx, userID); serr != nil { - _, isTooEarly := serr.(errtypes.IsTooEarly) - _, isInternal := serr.(errtypes.IsInternalError) - if !isTooEarly && !isInternal { - span.RecordError(serr) - span.SetStatus(codes.Error, serr.Error()) - log.Error().Int("attempt", attempt).Err(serr).Msg("lost update: re-read failed, aborting") - return serr - } - log.Warn().Int("attempt", attempt).Err(serr).Msg("lost update: re-read before retry") - } - } - return err + }) } // Get returns one entry from the cache @@ -215,11 +156,11 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err span.SetAttributes(attribute.String("cs3.userid", userID)) defer unlock() - ctx, span = appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Add") + ctx, span = appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Remove") defer span.End() span.SetAttributes(attribute.String("cs3.userid", userID), attribute.String("cs3.spaceid", spaceID)) - persistFunc := func() error { + return c.retryPersist(ctx, span, userID, spaceID, func() error { c.initializeIfNeeded(userID, spaceID) rss, _ := c.ReceivedSpaces.Load(userID) @@ -233,8 +174,48 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err } return c.persist(ctx, userID) + }) +} + +// List returns a list of received shares for a given user +// The return list is guaranteed to be thread-safe +func (c *Cache) List(ctx context.Context, userID string) (map[string]*Space, error) { + ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Grab lock") + unlock := c.lockUser(userID) + span.End() + span.SetAttributes(attribute.String("cs3.userid", userID)) + defer unlock() + + err := c.syncWithLock(ctx, userID) + if err != nil { + return nil, err + } + + spaces := map[string]*Space{} + rss, _ := c.ReceivedSpaces.Load(userID) + for spaceID, space := range rss.Spaces { + spaceCopy := &Space{ + States: map[string]*State{}, + } + for shareID, state := range space.States { + spaceCopy.States[shareID] = &State{ + State: state.State, + MountPoint: state.MountPoint, + Hidden: state.Hidden, + } + } + spaces[spaceID] = spaceCopy } + return spaces, nil +} + +func isSyncTransient(err error) bool { + _, isTooEarly := err.(errtypes.IsTooEarly) + _, isInternal := err.(errtypes.IsInternalError) + return isTooEarly || isInternal +} +func (c *Cache) retryPersist(ctx context.Context, span trace.Span, userID, spaceID string, persistFunc func() error) error { log := appctx.GetLogger(ctx).With(). Str("hostname", os.Getenv("HOSTNAME")). Str("userID", userID). @@ -282,9 +263,7 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err return ctx.Err() } if serr := c.syncWithLock(ctx, userID); serr != nil { - _, isTooEarly := serr.(errtypes.IsTooEarly) - _, isInternal := serr.(errtypes.IsInternalError) - if !isTooEarly && !isInternal { + if !isSyncTransient(serr) { span.RecordError(serr) span.SetStatus(codes.Error, serr.Error()) log.Error().Int("attempt", attempt).Err(serr).Msg("lost update: re-read failed, aborting") @@ -296,38 +275,6 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err return err } -// List returns a list of received shares for a given user -// The return list is guaranteed to be thread-safe -func (c *Cache) List(ctx context.Context, userID string) (map[string]*Space, error) { - ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Grab lock") - unlock := c.lockUser(userID) - span.End() - span.SetAttributes(attribute.String("cs3.userid", userID)) - defer unlock() - - err := c.syncWithLock(ctx, userID) - if err != nil { - return nil, err - } - - spaces := map[string]*Space{} - rss, _ := c.ReceivedSpaces.Load(userID) - for spaceID, space := range rss.Spaces { - spaceCopy := &Space{ - States: map[string]*State{}, - } - for shareID, state := range space.States { - spaceCopy.States[shareID] = &State{ - State: state.State, - MountPoint: state.MountPoint, - Hidden: state.Hidden, - } - } - spaces[spaceID] = spaceCopy - } - return spaces, nil -} - func (c *Cache) syncWithLock(ctx context.Context, userID string) error { ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Sync") defer span.End() @@ -359,9 +306,7 @@ func (c *Cache) syncWithLock(ctx context.Context, userID string) error { return nil default: span.SetStatus(codes.Error, err.Error()) - _, isTooEarly := err.(errtypes.IsTooEarly) - _, isInternal := err.(errtypes.IsInternalError) - if isTooEarly || isInternal { + if isSyncTransient(err) { log.Warn().Err(err).Msg("lost update: re-read transient error") } else { log.Error().Err(err).Msg("lost update: re-read failed") diff --git a/pkg/storage/utils/decomposedfs/upload/upload.go b/pkg/storage/utils/decomposedfs/upload/upload.go index 43c347cbfac..ed50600f56e 100644 --- a/pkg/storage/utils/decomposedfs/upload/upload.go +++ b/pkg/storage/utils/decomposedfs/upload/upload.go @@ -74,8 +74,6 @@ 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 @@ -97,7 +95,6 @@ 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 } From 8ee685da5bbcdadd9b1ae3964f73e12eecaa4d34 Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Tue, 21 Jul 2026 21:37:55 +0200 Subject: [PATCH 6/7] feat: review fixes --- .../receivedsharecache/receivedsharecache.go | 27 ++++++++++--------- pkg/storage/utils/metadata/cs3.go | 17 ++++++------ .../receivedsharecache_concurrent_test.go | 2 -- 3 files changed, 23 insertions(+), 23 deletions(-) diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go index 6eec989203d..6e611c87976 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go @@ -101,7 +101,7 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati defer unlock() if _, ok := c.ReceivedSpaces.Load(userID); !ok { - err := c.syncWithLock(ctx, userID) + err := c.syncIfStale(ctx, userID) if err != nil { return err } @@ -111,7 +111,7 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati defer span.End() span.SetAttributes(attribute.String("cs3.userid", userID), attribute.String("cs3.spaceid", spaceID)) - return c.retryPersist(ctx, span, userID, spaceID, func() error { + return c.retryPersist(ctx, userID, spaceID, func() error { c.initializeIfNeeded(userID, spaceID) rss, _ := c.ReceivedSpaces.Load(userID) @@ -137,7 +137,7 @@ func (c *Cache) Get(ctx context.Context, userID, spaceID, shareID string) (*Stat span.SetAttributes(attribute.String("cs3.userid", userID)) defer unlock() - err := c.syncWithLock(ctx, userID) + err := c.syncIfStale(ctx, userID) if err != nil { return nil, err } @@ -160,7 +160,7 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err defer span.End() span.SetAttributes(attribute.String("cs3.userid", userID), attribute.String("cs3.spaceid", spaceID)) - return c.retryPersist(ctx, span, userID, spaceID, func() error { + return c.retryPersist(ctx, userID, spaceID, func() error { c.initializeIfNeeded(userID, spaceID) rss, _ := c.ReceivedSpaces.Load(userID) @@ -180,13 +180,14 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err // List returns a list of received shares for a given user // The return list is guaranteed to be thread-safe func (c *Cache) List(ctx context.Context, userID string) (map[string]*Space, error) { - ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Grab lock") - unlock := c.lockUser(userID) - span.End() + ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "List") + defer span.End() span.SetAttributes(attribute.String("cs3.userid", userID)) + + unlock := c.lockUser(userID) defer unlock() - err := c.syncWithLock(ctx, userID) + err := c.syncIfStale(ctx, userID) if err != nil { return nil, err } @@ -215,7 +216,8 @@ func isSyncTransient(err error) bool { return isTooEarly || isInternal } -func (c *Cache) retryPersist(ctx context.Context, span trace.Span, userID, spaceID string, persistFunc func() error) error { +func (c *Cache) retryPersist(ctx context.Context, userID, spaceID string, persistFunc func() error) error { + span := trace.SpanFromContext(ctx) log := appctx.GetLogger(ctx).With(). Str("hostname", os.Getenv("HOSTNAME")). Str("userID", userID). @@ -262,7 +264,7 @@ func (c *Cache) retryPersist(ctx context.Context, span trace.Span, userID, space timer.Stop() return ctx.Err() } - if serr := c.syncWithLock(ctx, userID); serr != nil { + if serr := c.syncIfStale(ctx, userID); serr != nil { if !isSyncTransient(serr) { span.RecordError(serr) span.SetStatus(codes.Error, serr.Error()) @@ -275,8 +277,9 @@ func (c *Cache) retryPersist(ctx context.Context, span trace.Span, userID, space return err } -func (c *Cache) syncWithLock(ctx context.Context, userID string) error { - ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Sync") +// syncIfStale pulls the authoritative state from storage when the local replica is stale; caller must hold the user lock. +func (c *Cache) syncIfStale(ctx context.Context, userID string) error { + ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "SyncIfStale") defer span.End() span.SetAttributes(attribute.String("cs3.userid", userID)) diff --git a/pkg/storage/utils/metadata/cs3.go b/pkg/storage/utils/metadata/cs3.go index 497936f7f22..f32f758bcbd 100644 --- a/pkg/storage/utils/metadata/cs3.go +++ b/pkg/storage/utils/metadata/cs3.go @@ -30,7 +30,6 @@ 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" @@ -40,6 +39,7 @@ import ( "google.golang.org/grpc/metadata" "github.com/owncloud/reva/v2/internal/http/services/owncloud/ocdav/net" + "github.com/owncloud/reva/v2/pkg/appctx" ctxpkg "github.com/owncloud/reva/v2/pkg/ctx" "github.com/owncloud/reva/v2/pkg/errtypes" "github.com/owncloud/reva/v2/pkg/rgrpc/todo/pool" @@ -54,7 +54,6 @@ func init() { // CS3 represents a metadata storage with a cs3 storage backend type CS3 struct { - log zerolog.Logger SpaceRoot *provider.ResourceId providerAddr string @@ -66,8 +65,6 @@ 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) { @@ -167,7 +164,8 @@ 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") + log := appctx.GetLogger(ctx) + log.Debug().Str("path", uploadpath).Msg("cs3.SimpleUpload.start") _, err := cs3.Upload(ctx, UploadRequest{ Path: uploadpath, Content: content, @@ -180,7 +178,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") + log := appctx.GetLogger(ctx) + log.Debug().Str("path", req.Path).Msg("cs3.Upload.start") client, err := cs3.providerClient() if err != nil { @@ -246,7 +245,7 @@ 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") + 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 { @@ -263,7 +262,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") + 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 } @@ -271,7 +270,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") + 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/receivedsharecache_concurrent_test.go b/tests/integration/grpc/receivedsharecache_concurrent_test.go index 955015ec91b..e7def94c49c 100644 --- a/tests/integration/grpc/receivedsharecache_concurrent_test.go +++ b/tests/integration/grpc/receivedsharecache_concurrent_test.go @@ -93,7 +93,6 @@ var _ = Describe("receivedsharecache concurrent CS3 writes", func() { newCS3 := func() *metadata.CS3 { cs3 := metadata.NewCS3("", revads["storage"].GrpcAddress) cs3.SpaceRoot = spaceRoot - cs3.SetLogger(*appctx.GetLogger(ctx)) return cs3 } @@ -136,7 +135,6 @@ var _ = Describe("receivedsharecache concurrent CS3 writes", func() { newCS3 := func() *metadata.CS3 { cs3 := metadata.NewCS3("", revads["storage"].GrpcAddress) cs3.SpaceRoot = spaceRoot - cs3.SetLogger(*appctx.GetLogger(ctx)) return cs3 } From 470b5d8551ed1aeb3a15a7cbc7c33ed8fb65d201 Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Wed, 22 Jul 2026 09:53:53 +0200 Subject: [PATCH 7/7] feat: cleanup in tracing --- .../receivedsharecache/receivedsharecache.go | 24 ++++++++++++------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go index 6e611c87976..aa098a6dcf8 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go @@ -37,7 +37,6 @@ import ( "github.com/owncloud/reva/v2/pkg/storage/utils/metadata" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" - "go.opentelemetry.io/otel/trace" ) // name is the Tracer name used to identify this instrumentation library. @@ -111,7 +110,7 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati defer span.End() span.SetAttributes(attribute.String("cs3.userid", userID), attribute.String("cs3.spaceid", spaceID)) - return c.retryPersist(ctx, userID, spaceID, func() error { + err := c.retryPersist(ctx, userID, spaceID, func() error { c.initializeIfNeeded(userID, spaceID) rss, _ := c.ReceivedSpaces.Load(userID) @@ -127,6 +126,13 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati return c.persist(ctx, userID) }) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + } else { + span.SetStatus(codes.Ok, "") + } + return err } // Get returns one entry from the cache @@ -160,7 +166,7 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err defer span.End() span.SetAttributes(attribute.String("cs3.userid", userID), attribute.String("cs3.spaceid", spaceID)) - return c.retryPersist(ctx, userID, spaceID, func() error { + err := c.retryPersist(ctx, userID, spaceID, func() error { c.initializeIfNeeded(userID, spaceID) rss, _ := c.ReceivedSpaces.Load(userID) @@ -175,6 +181,13 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err return c.persist(ctx, userID) }) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + } else { + span.SetStatus(codes.Ok, "") + } + return err } // List returns a list of received shares for a given user @@ -217,7 +230,6 @@ func isSyncTransient(err error) bool { } func (c *Cache) retryPersist(ctx context.Context, userID, spaceID string, persistFunc func() error) error { - span := trace.SpanFromContext(ctx) log := appctx.GetLogger(ctx).With(). Str("hostname", os.Getenv("HOSTNAME")). Str("userID", userID). @@ -233,7 +245,6 @@ func (c *Cache) retryPersist(ctx context.Context, userID, spaceID string, persis err = persistFunc() switch err.(type) { case nil: - span.SetStatus(codes.Ok, "") return nil case errtypes.Aborted: // this is the expected status code from the server when the if-match etag check fails @@ -253,7 +264,6 @@ func (c *Cache) retryPersist(ctx context.Context, userID, spaceID string, persis // continue with sync below log.Debug().Int("attempt", attempt).Msg("CAS failed: TooEarly (upload in progress), retrying") default: - span.SetStatus(codes.Error, fmt.Sprintf("persisting received share failed, giving up: %s", err.Error())) log.Error().Int("attempt", attempt).Err(err).Msg("persisting received share failed, giving up") return err } @@ -266,8 +276,6 @@ func (c *Cache) retryPersist(ctx context.Context, userID, spaceID string, persis } if serr := c.syncIfStale(ctx, userID); serr != nil { if !isSyncTransient(serr) { - span.RecordError(serr) - span.SetStatus(codes.Error, serr.Error()) log.Error().Int("attempt", attempt).Err(serr).Msg("lost update: re-read failed, aborting") return serr }