From 468d780e24585b9fe818fa387c0e22ed0ea3562c Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Wed, 22 Jul 2026 20:19:11 +0200 Subject: [PATCH 1/3] feat: [OCISDEV-855] receivedsharecache: retry on transient storage errors in CAS loop (#657) * fix: ocisdev-855, cs3/decomposefs transactional concurent upload * fix: ocisdev-855, cs3/decomposefs optimistic-abort concurent upload * fix: ocisdev-855, cs3/decomposefs optimistic-abort concurent upload * feat: logs and cleanup * feat: cleanup * feat: review fixes * feat: cleanup in tracing --- ...ment-receivedsharecache-retry-transient.md | 7 + .../storageprovider/storageprovider.go | 2 + pkg/rhttp/datatx/utils/download/download.go | 3 + .../receivedsharecache/receivedsharecache.go | 198 ++++++++++-------- .../receivedsharecache_test.go | 149 ++++++++++++- pkg/storage/fs/posix/tree/tree_test.go | 5 + pkg/storage/utils/decomposedfs/tree/tree.go | 3 + pkg/storage/utils/decomposedfs/upload.go | 2 + .../utils/decomposedfs/upload/upload.go | 5 +- .../utils/decomposedfs/upload_async_test.go | 8 +- pkg/storage/utils/metadata/cs3.go | 11 + ...torageprovider-ocis-with-dataprovider.toml | 32 +++ .../receivedsharecache_concurrent_test.go | 195 +++++++++++++++++ 13 files changed, 526 insertions(+), 94 deletions(-) create mode 100644 changelog/unreleased/enhancement-receivedsharecache-retry-transient.md 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/changelog/unreleased/enhancement-receivedsharecache-retry-transient.md b/changelog/unreleased/enhancement-receivedsharecache-retry-transient.md new file mode 100644 index 00000000000..ae98901e005 --- /dev/null +++ b/changelog/unreleased/enhancement-receivedsharecache-retry-transient.md @@ -0,0 +1,7 @@ +Enhancement: receivedsharecache: retry on transient storage errors in CAS loop + +The received share cache now retries the compare-and-swap persist loop when +the storage returns transient errors (TooEarly, AlreadyExists, Aborted, +PreconditionFailed), avoiding spurious failures under concurrent writes. + +https://github.com/owncloud/reva/pull/657 diff --git a/internal/grpc/services/storageprovider/storageprovider.go b/internal/grpc/services/storageprovider/storageprovider.go index d790bf2c1d5..4c5418a284f 100644 --- a/internal/grpc/services/storageprovider/storageprovider.go +++ b/internal/grpc/services/storageprovider/storageprovider.go @@ -440,6 +440,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 a35869fd78b..6b7ab6681d6 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go @@ -22,6 +22,7 @@ import ( "context" "encoding/json" "fmt" + "math/rand/v2" "os" "path" "path/filepath" @@ -99,7 +100,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 } @@ -109,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)) - persistFunc := func() error { + err := c.retryPersist(ctx, userID, spaceID, func() error { c.initializeIfNeeded(userID, spaceID) rss, _ := c.ReceivedSpaces.Load(userID) @@ -124,44 +125,12 @@ 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 retries := 100; retries > 0; retries-- { - err = persistFunc() - switch err.(type) { - case nil: - 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 - 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 - 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 - 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") - return 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 err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + } else { + span.SetStatus(codes.Ok, "") } return err } @@ -174,7 +143,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 } @@ -193,11 +162,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 { + err := c.retryPersist(ctx, userID, spaceID, func() error { c.initializeIfNeeded(userID, spaceID) rss, _ := c.ReceivedSpaces.Load(userID) @@ -211,44 +180,12 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err } 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 retries := 100; retries > 0; retries-- { - err = persistFunc() - switch err.(type) { - case nil: - 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 - 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 - 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 - 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") - return 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 err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + } else { + span.SetStatus(codes.Ok, "") } return err } @@ -256,13 +193,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 } @@ -285,8 +223,71 @@ func (c *Cache) List(ctx context.Context, userID string) (map[string]*Space, err return spaces, nil } -func (c *Cache) syncWithLock(ctx context.Context, userID string) error { - ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Sync") +func isSyncTransient(err error) bool { + _, isTooEarly := err.(errtypes.IsTooEarly) + _, isInternal := err.(errtypes.IsInternalError) + return isTooEarly || isInternal +} + +func (c *Cache) retryPersist(ctx context.Context, userID, spaceID string, persistFunc func() error) error { + 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: + 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: + 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.syncIfStale(ctx, userID); serr != nil { + if !isSyncTransient(serr) { + 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 +} + +// 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)) @@ -307,13 +308,20 @@ func (c *Cache) syncWithLock(ctx context.Context, userID string) error { span.AddEvent("updating local cache") case errtypes.NotFound: span.SetStatus(codes.Ok, "") + if err := c.persist(ctx, userID); err != nil { + log.Warn().Err(err).Msg("failed to create empty received share cache file") + } return nil case errtypes.NotModified: 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()) + 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") + } return err } @@ -374,11 +382,21 @@ func (c *Cache) persist(ctx context.Context, userID string) error { return err } rss.etag = res.Etag - span.SetStatus(codes.Ok, "") return nil } +// expBackoff returns full-jitter delay: rand(0, min(100ms, 2^attempt ms)). +// attempt: 0 1 2 3 4 5 6 7+ +// max ms: 1 2 4 8 16 32 64 100 +func expBackoff(attempt int) time.Duration { + base := time.Duration(1< 100*time.Millisecond { + base = 100 * time.Millisecond + } + return time.Duration(rand.Int64N(int64(base) + 1)) +} + func userJSONPath(userID string) string { return filepath.Join("/users", userID, "received.json") } diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go index d7cef33228d..e0047d1e8f3 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go @@ -21,11 +21,16 @@ package receivedsharecache_test import ( "context" "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" @@ -49,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") @@ -71,6 +77,50 @@ var _ = Describe("Cache", func() { } }) + Describe("List", func() { + Context("when no cache file exists yet", func() { + It("creates the cache file on first call", func() { + _, err := c.List(ctx, userID) + Expect(err).ToNot(HaveOccurred()) + + _, err = os.Stat(tmpdir + "/users/" + userID + "/received.json") + Expect(err).ToNot(HaveOccurred()) + }) + + It("returns empty spaces", func() { + spaces, err := c.List(ctx, userID) + Expect(err).ToNot(HaveOccurred()) + Expect(spaces).To(BeEmpty()) + }) + + It("is readable by a fresh cache instance after first call", func() { + _, err := c.List(ctx, userID) + Expect(err).ToNot(HaveOccurred()) + + // a new cache instance must be able to read the bootstrapped file + c2 := receivedsharecache.New(storage, 0*time.Second) + spaces, err := c2.List(ctx, userID) + Expect(err).ToNot(HaveOccurred()) + Expect(spaces).To(BeEmpty()) + }) + + It("allows adding a share after bootstrap", func() { + _, err := c.List(ctx, userID) + Expect(err).ToNot(HaveOccurred()) + + rs := &collaboration.ReceivedShare{ + Share: share, + State: collaboration.ShareState_SHARE_STATE_PENDING, + } + Expect(c.Add(ctx, userID, spaceID, rs)).To(Succeed()) + + spaces, err := c.List(ctx, userID) + Expect(err).ToNot(HaveOccurred()) + Expect(spaces[spaceID].States).To(HaveKey(shareID)) + }) + }) + }) + Describe("Add", func() { It("adds an entry", func() { rs := &collaboration.ReceivedShare{ @@ -100,6 +150,46 @@ var _ = Describe("Cache", func() { }) }) + Describe("concurrent writes from multiple cache instances", func() { + It("preserves the share when 15 replicas write the same file simultaneously", func() { + const numReplicas = 15 + + // 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, numReplicas) + var wg sync.WaitGroup + 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: "share-0"}, + }, + State: collaboration.ShareState_SHARE_STATE_PENDING, + } + errs[idx] = replicas[idx].Add(ctx, userID, spaceID, rs) + }(i) + } + wg.Wait() + for i, err := range errs { + 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()) + Expect(spaces[spaceID].States).To(HaveKey("share-0")) + }) + }) + Describe("with an existing entry", func() { BeforeEach(func() { rs := &collaboration.ReceivedShare{ @@ -154,6 +244,63 @@ var _ = Describe("Cache", func() { Expect(err).ToNot(HaveOccurred()) Expect(s).To(BeNil()) }) + + It("returns context.Canceled immediately when ctx is already canceled", func() { + as := &alwaysFailStorage{Storage: storage} + c2 := receivedsharecache.New(as, 0*time.Second) + + canceled, cancel := context.WithCancel(ctx) + cancel() + + err := c2.Remove(canceled, userID, spaceID, shareID) + Expect(err).To(MatchError(context.Canceled)) + Expect(atomic.LoadInt32(&as.uploads)).To(Equal(int32(0))) + }) + + It("exits the backoff sleep when ctx is canceled", func() { + as := &alwaysFailStorage{Storage: storage} + c2 := receivedsharecache.New(as, 0*time.Second) + + ctx2, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + + start := time.Now() + _ = c2.Remove(ctx2, userID, spaceID, shareID) + Expect(time.Since(start)).To(BeNumerically("<", 200*time.Millisecond)) + }) }) }) }) + +// barrierStorage wraps a Storage and holds Upload calls until n goroutines have +// arrived, then releases them all at once. This makes the concurrent-write race +// reproducible regardless of OS goroutine scheduling. +type barrierStorage struct { + metadata.Storage + arrived int32 + n int32 + ready chan struct{} + closeOnce sync.Once +} + +func newBarrierStorage(s metadata.Storage, n int) *barrierStorage { + return &barrierStorage{Storage: s, n: int32(n), ready: make(chan struct{})} +} + +func (b *barrierStorage) 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) +} + +type alwaysFailStorage struct { + metadata.Storage + uploads int32 +} + +func (a *alwaysFailStorage) Upload(_ context.Context, _ metadata.UploadRequest) (*metadata.UploadResponse, error) { + atomic.AddInt32(&a.uploads, 1) + return nil, errtypes.PreconditionFailed("injected") +} 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 8c4ea410568..5a9559cf181 100644 --- a/pkg/storage/utils/decomposedfs/tree/tree.go +++ b/pkg/storage/utils/decomposedfs/tree/tree.go @@ -704,6 +704,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() diff --git a/pkg/storage/utils/decomposedfs/upload.go b/pkg/storage/utils/decomposedfs/upload.go index 1f08b38636b..fac96b70632 100644 --- a/pkg/storage/utils/decomposedfs/upload.go +++ b/pkg/storage/utils/decomposedfs/upload.go @@ -135,6 +135,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() @@ -335,6 +336,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 2880fa999c0..98f41db563e 100644 --- a/pkg/storage/utils/decomposedfs/upload/upload.go +++ b/pkg/storage/utils/decomposedfs/upload/upload.go @@ -114,12 +114,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 +141,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 +248,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 67d6d34bb29..a78e701eab3 100644 --- a/pkg/storage/utils/decomposedfs/upload_async_test.go +++ b/pkg/storage/utils/decomposedfs/upload_async_test.go @@ -11,6 +11,7 @@ import ( cs3permissions "github.com/cs3org/go-cs3apis/cs3/permissions/v1beta1" v1beta11 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" + "github.com/owncloud/reva/v2/pkg/appctx" ruser "github.com/owncloud/reva/v2/pkg/ctx" "github.com/owncloud/reva/v2/pkg/events" "github.com/owncloud/reva/v2/pkg/events/stream" @@ -68,9 +69,9 @@ var _ = Describe("Async file uploads", Ordered, func() { firstContent = []byte("0123456789") secondContent = []byte("01234567890123456789") - ctx = ruser.ContextSetUser(context.Background(), user) + ctx context.Context - pub chan interface{} + pub chan interface{} con chan interface{} uploadID string @@ -129,6 +130,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()) diff --git a/pkg/storage/utils/metadata/cs3.go b/pkg/storage/utils/metadata/cs3.go index 939b929f59f..f32f758bcbd 100644 --- a/pkg/storage/utils/metadata/cs3.go +++ b/pkg/storage/utils/metadata/cs3.go @@ -39,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" @@ -64,6 +65,7 @@ type CS3 struct { dataGatewayClient *http.Client } + // 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 +164,8 @@ func (cs3 *CS3) SimpleUpload(ctx context.Context, uploadpath string, content []b ctx, span := tracer.Start(ctx, "SimpleUpload") defer span.End() + log := appctx.GetLogger(ctx) + log.Debug().Str("path", uploadpath).Msg("cs3.SimpleUpload.start") _, err := cs3.Upload(ctx, UploadRequest{ Path: uploadpath, Content: content, @@ -174,6 +178,9 @@ func (cs3 *CS3) Upload(ctx context.Context, req UploadRequest) (*UploadResponse, ctx, span := tracer.Start(ctx, "Upload") defer span.End() + log := appctx.GetLogger(ctx) + log.Debug().Str("path", req.Path).Msg("cs3.Upload.start") + client, err := cs3.providerClient() if err != nil { return nil, err @@ -238,6 +245,8 @@ func (cs3 *CS3) Upload(ctx context.Context, req UploadRequest) (*UploadResponse, return nil, errors.New("metadata storage doesn't support the simple upload protocol") } + 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 +262,7 @@ func (cs3 *CS3) Upload(ctx context.Context, req UploadRequest) (*UploadResponse, return nil, err } defer resp.Body.Close() + 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 +270,7 @@ func (cs3 *CS3) Upload(ctx context.Context, req UploadRequest) (*UploadResponse, if ocEtag := resp.Header.Get("OC-ETag"); ocEtag != "" { etag = ocEtag } + 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..e7def94c49c --- /dev/null +++ b/tests/integration/grpc/receivedsharecache_concurrent_test.go @@ -0,0 +1,195 @@ +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 + 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 + 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 a33284dd212a93b4cfed337327804f6fb9299406 Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Wed, 12 Aug 2026 15:55:02 +0200 Subject: [PATCH 2/3] =?UTF-8?q?fix:=20[OCISDEV-855]=20stable-8.0=20build:?= =?UTF-8?q?=20NewTooEarly=E2=86=92NewAborted,=20serialize=20Download=20in?= =?UTF-8?q?=20concurrent=20test?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit status.NewTooEarly was added after stable-8.0 branched; map IsTooEarly to NewAborted instead (same retry semantics for the client). DiskStorage.Upload on stable-8.0 uses os.WriteFile (not renameio), so concurrent Download calls can read a partial file. Add a mutex to barrierStorage that serializes Upload/Download pairs in the test. --- .../grpc/services/storageprovider/storageprovider.go | 2 +- .../receivedsharecache/receivedsharecache_test.go | 12 ++++++++++++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/internal/grpc/services/storageprovider/storageprovider.go b/internal/grpc/services/storageprovider/storageprovider.go index 4c5418a284f..48da8c1076c 100644 --- a/internal/grpc/services/storageprovider/storageprovider.go +++ b/internal/grpc/services/storageprovider/storageprovider.go @@ -441,7 +441,7 @@ func (s *Service) InitiateFileUpload(ctx context.Context, req *provider.Initiate case errtypes.Locked: st = status.NewLocked(ctx, "locked") case errtypes.IsTooEarly: - st = status.NewTooEarly(ctx, err.Error()) + st = status.NewAborted(ctx, err, "upload in progress, retry later") default: st = status.NewInternal(ctx, "error getting upload id: "+err.Error()) } diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go index e0047d1e8f3..cd8ca1c5b58 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go @@ -275,8 +275,12 @@ var _ = Describe("Cache", func() { // barrierStorage wraps a Storage and holds Upload calls until n goroutines have // arrived, then releases them all at once. This makes the concurrent-write race // reproducible regardless of OS goroutine scheduling. +// mu serializes Upload/Download pairs because DiskStorage.Upload is not atomic +// on this branch (os.WriteFile, not renameio) — without it a concurrent Download +// can read a partial file and get a json.SyntaxError. type barrierStorage struct { metadata.Storage + mu sync.Mutex arrived int32 n int32 ready chan struct{} @@ -292,9 +296,17 @@ func (b *barrierStorage) Upload(ctx context.Context, req metadata.UploadRequest) b.closeOnce.Do(func() { close(b.ready) }) } <-b.ready + b.mu.Lock() + defer b.mu.Unlock() return b.Storage.Upload(ctx, req) } +func (b *barrierStorage) Download(ctx context.Context, req metadata.DownloadRequest) (*metadata.DownloadResponse, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.Storage.Download(ctx, req) +} + type alwaysFailStorage struct { metadata.Storage uploads int32 From eede49e8ef9f153ef18e9d715c792c6255baad16 Mon Sep 17 00:00:00 2001 From: Michal Klos Date: Wed, 12 Aug 2026 16:00:23 +0200 Subject: [PATCH 3/3] fix: [OCISDEV-855] stable-8.0: drop IsResourceProcessing (added after branch cut) --- pkg/rhttp/datatx/utils/download/download.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/rhttp/datatx/utils/download/download.go b/pkg/rhttp/datatx/utils/download/download.go index 3fa09787a42..7a0a5655fb2 100644 --- a/pkg/rhttp/datatx/utils/download/download.go +++ b/pkg/rhttp/datatx/utils/download/download.go @@ -275,7 +275,7 @@ 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: + case errtypes.IsTooEarly: log.Debug().Err(err).Str("action", action).Msg("resource is processing") w.WriteHeader(http.StatusTooEarly) default: