Skip to content

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

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 3 commits into from
Aug 13, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
468d780
feat: [OCISDEV-855] receivedsharecache: retry on transient storage er…
mklos-kw Jul 22, 2026
a33284d
fix: [OCISDEV-855] stable-8.0 build: NewTooEarly→NewAborted, serializ…
mklos-kw Aug 12, 2026
eede49e
fix: [OCISDEV-855] stable-8.0: drop IsResourceProcessing (added after…
mklos-kw Aug 12, 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
7 changes: 7 additions & 0 deletions changelog/unreleased/enhancement-receivedsharecache-retry-transient.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
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 @@ -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.NewAborted(ctx, err, "upload in progress, retry later")
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.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
198 changes: 108 additions & 90 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 @@ -22,6 +22,7 @@ import (
"context"
"encoding/json"
"fmt"
"math/rand/v2"
"os"
"path"
"path/filepath"
Expand Down Expand Up @@ -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
}
Expand All @@ -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)
Expand All @@ -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
}
Expand All @@ -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
}
Expand All @@ -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)
Expand All @@ -211,58 +180,27 @@ 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
}

// 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
}
Expand All @@ -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))

Expand All @@ -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
}

Expand Down Expand Up @@ -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<<uint(attempt)) * time.Millisecond
if base > 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")
}
Expand Down
Loading
Loading