Skip to content

feat: [OCISDEV-855] receivedsharecache: retry on transient storage errors in CAS loop #657

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mklos-kw merged 7 commits into from
Jul 22, 2026
Merged

feat: [OCISDEV-855] receivedsharecache: retry on transient storage errors in CAS loop #657

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
5446221
fix: ocisdev-855, cs3/decomposefs transactional concurent upload
mklos-kw Jul 15, 2026
217618a
fix: ocisdev-855, cs3/decomposefs optimistic-abort concurent upload
mklos-kw Jul 15, 2026
f61d4ee
fix: ocisdev-855, cs3/decomposefs optimistic-abort concurent upload
mklos-kw Jul 17, 2026
8aabc44
feat: logs and cleanup
mklos-kw Jul 21, 2026
e3701d2
feat: cleanup
mklos-kw Jul 21, 2026
8ee685d
feat: review fixes
mklos-kw Jul 21, 2026
470b5d8
feat: cleanup in tracing
mklos-kw Jul 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions internal/grpc/services/storageprovider/storageprovider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())
}
Expand Down
3 changes: 3 additions & 0 deletions pkg/rhttp/datatx/utils/download/download.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
189 changes: 87 additions & 102 deletions pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,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
}
Expand All @@ -110,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)
Expand All @@ -125,56 +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 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
}
})
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
return err
}
Expand All @@ -187,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
}
Expand All @@ -206,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)
Expand All @@ -224,8 +180,56 @@ 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
// 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 {
log := appctx.GetLogger(ctx).With().
Str("hostname", os.Getenv("HOSTNAME")).
Str("userID", userID).
Expand All @@ -241,24 +245,26 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err
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
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")
log.Error().Int("attempt", attempt).Err(err).Msg("persisting received share failed, giving up")
return err
}
timer := time.NewTimer(expBackoff(attempt))
Expand All @@ -268,50 +274,20 @@ 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) {
log.Error().Int("attempt", attempt).Err(serr).Msg("lost update: re-read failed, aborting")
return serr
Comment on lines +278 to +280

@jvillafanez jvillafanez Jul 22, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it's weird that we only set the error and status code of the span here. If the caller creates the span, it should be the caller the one closing it AND setting the appropriate status code. I think all the errors returned by the function should be registered, not just this one.

}
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))

Expand Down Expand Up @@ -340,8 +316,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
}

Expand All @@ -365,6 +345,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")
Expand Down Expand Up @@ -395,6 +378,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)
Expand All @@ -403,6 +387,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
}
Expand Down
30 changes: 14 additions & 16 deletions pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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")
Expand Down Expand Up @@ -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"))
})
})

Expand Down
Loading
Loading