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..6e611c87976 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. @@ -100,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 } @@ -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, userID, spaceID, func() error { c.initializeIfNeeded(userID, spaceID) rss, _ := c.ReceivedSpaces.Load(userID) @@ -125,58 +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: - 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 - } - timer := time.NewTimer(expBackoff(attempt)) - select { - case <-timer.C: - case <-ctx.Done(): - 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 - } - } - return err + }) } // Get returns one entry from the cache @@ -187,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 } @@ -206,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, userID, spaceID, func() error { c.initializeIfNeeded(userID, spaceID) rss, _ := c.ReceivedSpaces.Load(userID) @@ -224,8 +174,50 @@ 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, "List") + defer span.End() + span.SetAttributes(attribute.String("cs3.userid", userID)) + + unlock := c.lockUser(userID) + defer unlock() + + err := c.syncIfStale(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, userID, spaceID string, persistFunc func() error) error { + span := trace.SpanFromContext(ctx) log := appctx.GetLogger(ctx).With(). Str("hostname", os.Getenv("HOSTNAME")). Str("userID", userID). @@ -244,21 +236,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: + // 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)) @@ -268,50 +264,22 @@ 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 - } - } - 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, + 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 } + log.Warn().Int("attempt", attempt).Err(serr).Msg("lost update: re-read before retry") } - spaces[spaceID] = spaceCopy } - return spaces, nil + 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)) @@ -340,8 +308,12 @@ 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()) + 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 } @@ -365,6 +337,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 +370,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 +379,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/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() diff --git a/pkg/storage/utils/decomposedfs/upload.go b/pkg/storage/utils/decomposedfs/upload.go index 9669dcc7641..add62e973cb 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() @@ -336,6 +337,7 @@ func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Refere } } + log.Debug().Str("uploadid", session.ID()).Msg("decomposedfs:InitiateUpload:complete") return map[string]string{ "simple": session.ID(), "tus": session.ID(), diff --git a/pkg/storage/utils/decomposedfs/upload/upload.go b/pkg/storage/utils/decomposedfs/upload/upload.go index 1a42f20878b..ed50600f56e 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 35827ec15c0..1ede4b72d61 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 @@ -132,6 +133,9 @@ var _ = Describe("Async file uploads", Ordered, func() { ) BeforeEach(func() { + zl := zerolog.New(os.Stdout).Level(zerolog.DebugLevel) + ctx = appctx.WithLogger(ruser.ContextSetUser(context.Background(), user), &zl) + // setup test tmpRoot, err := helpers.TempDir("reva-unit-tests-*-root") Expect(err).ToNot(HaveOccurred()) 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) +}