diff --git a/changelog/unreleased/purge-orphaned-received-shares.md b/changelog/unreleased/purge-orphaned-received-shares.md new file mode 100644 index 00000000000..80e7c643627 --- /dev/null +++ b/changelog/unreleased/purge-orphaned-received-shares.md @@ -0,0 +1,17 @@ +Enhancement: Add offline purge of orphaned received shares to the jsoncs3 share manager + +The jsoncs3 share manager can accumulate orphaned received-share entries when a +space is deleted but the per-user received.json blocks are not cleaned up (for +example after a missed SpaceDeleted event). These orphans are removed inline on +ListReceivedShares, but that happens on the request hot path and can race a +client deadline. + +This adds `Manager.PurgeOrphanedReceivedShares`, which enumerates users, detects +received-share entries whose share no longer exists in the provider cache and +removes them in a single batched persist per user and space +(`receivedsharecache.Cache.RemoveSpaceShares`). It can be scoped to a single +space and supports a dry-run mode. An entry is only removed when the share is +confirmed absent after a successful space listing, so live shares are never +removed on a transient read failure. + +https://github.com/owncloud/reva/pull/656 diff --git a/pkg/share/manager/jsoncs3/jsoncs3.go b/pkg/share/manager/jsoncs3/jsoncs3.go index ad60134d5ec..06c4cb20316 100644 --- a/pkg/share/manager/jsoncs3/jsoncs3.go +++ b/pkg/share/manager/jsoncs3/jsoncs3.go @@ -20,6 +20,7 @@ package jsoncs3 import ( "context" + "path" "strings" "sync" "time" @@ -1289,3 +1290,125 @@ func (m *Manager) CleanupStaleShares(ctx context.Context) { return true }) } + +// maxPurgeErrorSamples caps how many error messages a PurgeOrphanedReceivedShares +// run collects, so a broadly-failing run cannot produce an unbounded slice. +const maxPurgeErrorSamples = 50 + +// PurgeStats aggregates the counters of a PurgeOrphanedReceivedShares run. +type PurgeStats struct { + UsersScanned int + SpacesScanned int + OrphansFound int + OrphansRemoved int + Errors int + // ErrorSamples holds up to maxPurgeErrorSamples of the encountered error + // messages to give a sense of what went wrong without dumping every error. + ErrorSamples []string +} + +// recordError increments the error counter and keeps a capped sample of messages. +func (s *PurgeStats) recordError(err error) { + s.Errors++ + if len(s.ErrorSamples) < maxPurgeErrorSamples { + s.ErrorSamples = append(s.ErrorSamples, err.Error()) + } +} + +// PurgeOrphanedReceivedShares scans users' received-share state and removes +// entries that point at shares which no longer exist in the provider cache. +// These orphans accumulate when a space is deleted but its per-user +// received.json blocks are not cleaned up (e.g. a missed SpaceDeleted event). +// They are otherwise removed inline on ListReceivedShares, but that path is on +// the request hot path; this offline, batched variant drains them without +// racing a client deadline. +// +// If spaceID is non-empty it must be the "{storageID}:{spaceID}" key as stored +// in received.json (storage and space id joined by shareid.IDDelimiter); only +// that space is considered. When dryRun is true nothing is written; the +// returned stats still reflect what would be removed. +// +// An entry is only treated as an orphan when the share is confirmed absent from +// the provider cache after a successful ListSpace prime. Spaces whose share +// list cannot be read are skipped (counted as errors) so live shares are never +// removed on a transient failure. +func (m *Manager) PurgeOrphanedReceivedShares(ctx context.Context, spaceID string, dryRun bool) (PurgeStats, error) { + log := appctx.GetLogger(ctx) + stats := PurgeStats{} + + if err := m.initialize(ctx); err != nil { + return stats, err + } + + entries, err := m.storage.ReadDir(ctx, "/users") + if err != nil { + return stats, err + } + + for _, entry := range entries { + userID := path.Base(entry) + if userID == "" || userID == "." { + continue + } + stats.UsersScanned++ + + spaces, err := m.UserReceivedStates.List(ctx, userID) + if err != nil { + log.Error().Err(err).Str("userid", userID).Msg("could not list received shares for user") + stats.recordError(err) + continue + } + + for ssid, rspace := range spaces { + if spaceID != "" && ssid != spaceID { + continue + } + stats.SpacesScanned++ + + storageID, spid, _ := shareid.Decode(ssid) + userlog := log.With().Str("userid", userID).Str("storageid", storageID).Str("spaceid", spid).Logger() + + // Prime the provider cache for this space so the subsequent + // Get(skipSync=true) lookups reflect current on-disk state. A + // read error here means we cannot tell live from dead shares, so + // we skip the whole space rather than risk removing live entries. + if _, err := m.Cache.ListSpace(ctx, storageID, spid); err != nil { + userlog.Error().Err(err).Msg("failed to list shares in space, skipping") + stats.recordError(err) + continue + } + + orphans := []string{} + for shareID := range rspace.States { + s, err := m.Cache.Get(ctx, storageID, spid, shareID, true) + if err != nil { + userlog.Error().Err(err).Str("shareid", shareID).Msg("could not retrieve share, skipping") + stats.recordError(err) + continue + } + if s == nil { + orphans = append(orphans, shareID) + } + } + + if len(orphans) == 0 { + continue + } + stats.OrphansFound += len(orphans) + userlog.Info().Int("count", len(orphans)).Bool("dryRun", dryRun).Msg("found orphaned received shares") + userlog.Debug().Strs("shareids", orphans).Msg("orphaned received share ids") + + if dryRun { + continue + } + if err := m.UserReceivedStates.RemoveSpaceShares(ctx, userID, ssid, orphans); err != nil { + userlog.Error().Err(err).Msg("failed to remove orphaned received shares") + stats.recordError(err) + continue + } + stats.OrphansRemoved += len(orphans) + } + } + + return stats, nil +} diff --git a/pkg/share/manager/jsoncs3/jsoncs3_test.go b/pkg/share/manager/jsoncs3/jsoncs3_test.go index 0852275ab42..83354389755 100644 --- a/pkg/share/manager/jsoncs3/jsoncs3_test.go +++ b/pkg/share/manager/jsoncs3/jsoncs3_test.go @@ -1148,5 +1148,110 @@ var _ = Describe("Jsoncs3", func() { }) }) }) + + Describe("PurgeOrphanedReceivedShares", func() { + // ssid is the "{storageID}:{spaceID}" key under which the grantee's + // received state for the existing share is stored in received.json. + ssid := sharedResource.Id.StorageId + ":" + sharedResource.Id.SpaceId + + // orphanProviderCache empties the provider cache for the shared + // space on disk, so the share the grantee received no longer + // resolves - turning the received-state entry into an orphan. + orphanProviderCache := func() { + spaces, ok := m.Cache.Providers.Load(sharedResource.Id.StorageId) + Expect(ok).To(BeTrue()) + cache, ok := spaces.Spaces.Load(sharedResource.Id.SpaceId) + Expect(ok).To(BeTrue()) + delete(cache.Shares, share.Id.OpaqueId) + bytes, err := json.Marshal(cache) + Expect(err).ToNot(HaveOccurred()) + Expect(storage.SimpleUpload(context.Background(), "storages/storageid/spaceid.json", bytes)).To(Succeed()) + cache.Etag = "orphaned" // trigger reload on next sync + } + + It("keeps entries whose share still exists", func() { + stats, err := m.PurgeOrphanedReceivedShares(context.Background(), "", false) + Expect(err).ToNot(HaveOccurred()) + Expect(stats.OrphansFound).To(Equal(0)) + Expect(stats.OrphansRemoved).To(Equal(0)) + + received, err := m.ListReceivedShares(granteeCtx, []*collaboration.Filter{}, nil) + Expect(err).ToNot(HaveOccurred()) + Expect(len(received)).To(Equal(1)) + }) + + It("removes orphaned entries and persists the removal", func() { + orphanProviderCache() + + stats, err := m.PurgeOrphanedReceivedShares(context.Background(), "", false) + Expect(err).ToNot(HaveOccurred()) + Expect(stats.OrphansFound).To(Equal(1)) + Expect(stats.OrphansRemoved).To(Equal(1)) + Expect(stats.Errors).To(Equal(0)) + + // a fresh manager (no in-memory cache) must not see the entry + fresh, err := jsoncs3.New(storage, nil, 0, nil, 0) + Expect(err).ToNot(HaveOccurred()) + spaces, err := fresh.UserReceivedStates.List(context.Background(), grantee.GetId().GetOpaqueId()) + Expect(err).ToNot(HaveOccurred()) + Expect(spaces).ToNot(HaveKey(ssid)) + }) + + It("does not remove anything in dry-run mode", func() { + orphanProviderCache() + + stats, err := m.PurgeOrphanedReceivedShares(context.Background(), "", true) + Expect(err).ToNot(HaveOccurred()) + Expect(stats.OrphansFound).To(Equal(1)) + Expect(stats.OrphansRemoved).To(Equal(0)) + + fresh, err := jsoncs3.New(storage, nil, 0, nil, 0) + Expect(err).ToNot(HaveOccurred()) + spaces, err := fresh.UserReceivedStates.List(context.Background(), grantee.GetId().GetOpaqueId()) + Expect(err).ToNot(HaveOccurred()) + Expect(spaces).To(HaveKey(ssid)) + }) + + It("only considers the requested space when scoped", func() { + orphanProviderCache() + + stats, err := m.PurgeOrphanedReceivedShares(context.Background(), "otherstorage:otherspace", false) + Expect(err).ToNot(HaveOccurred()) + Expect(stats.OrphansFound).To(Equal(0)) + Expect(stats.OrphansRemoved).To(Equal(0)) + + // the real orphan is untouched because it was out of scope + fresh, err := jsoncs3.New(storage, nil, 0, nil, 0) + Expect(err).ToNot(HaveOccurred()) + spaces, err := fresh.UserReceivedStates.List(context.Background(), grantee.GetId().GetOpaqueId()) + Expect(err).ToNot(HaveOccurred()) + Expect(spaces).To(HaveKey(ssid)) + }) + + It("skips the space and keeps entries when the share list cannot be read", func() { + // Corrupt the provider cache file so ListSpace fails. The + // safety predicate must skip the space rather than treat its + // entries as orphans. + Expect(storage.SimpleUpload(context.Background(), "storages/storageid/spaceid.json", []byte("not-json"))).To(Succeed()) + spaces, ok := m.Cache.Providers.Load(sharedResource.Id.StorageId) + Expect(ok).To(BeTrue()) + cache, ok := spaces.Spaces.Load(sharedResource.Id.SpaceId) + Expect(ok).To(BeTrue()) + cache.Etag = "corrupt" // trigger reload from the corrupt file + + stats, err := m.PurgeOrphanedReceivedShares(context.Background(), "", false) + Expect(err).ToNot(HaveOccurred()) + Expect(stats.OrphansFound).To(Equal(0)) + Expect(stats.OrphansRemoved).To(Equal(0)) + Expect(stats.Errors).To(Equal(1)) + Expect(stats.ErrorSamples).To(HaveLen(1)) + + fresh, err := jsoncs3.New(storage, nil, 0, nil, 0) + Expect(err).ToNot(HaveOccurred()) + rspaces, err := fresh.UserReceivedStates.List(context.Background(), grantee.GetId().GetOpaqueId()) + Expect(err).ToNot(HaveOccurred()) + Expect(rspaces).To(HaveKey(ssid)) + }) + }) }) }) diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go index 8c1bf66898c..e59f021c17c 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go @@ -278,6 +278,97 @@ func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) err return err } +// RemoveSpaceShares removes multiple entries of a single space from the cache in +// one persist. It is the batched counterpart of Remove: collecting all shareIDs +// to drop for a user's space avoids the N serial read-modify-write cycles that a +// per-share Remove would incur. Passing an empty shareIDs slice is a no-op. +func (c *Cache) RemoveSpaceShares(ctx context.Context, userID, spaceID string, shareIDs []string) error { + ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "RemoveSpaceShares") + defer span.End() + span.SetAttributes(attribute.String("cs3.userid", userID), attribute.String("cs3.spaceid", spaceID)) + + if len(shareIDs) == 0 { + return nil + } + + unlock := c.lockUser(userID) + defer unlock() + + persistFunc := func() error { + c.initializeIfNeeded(userID, spaceID) + + rss, _ := c.ReceivedSpaces.Load(userID) + receivedSpace := rss.Spaces[spaceID] + if receivedSpace == nil { + // initializeIfNeeded only creates the space entry for a non-empty + // spaceID; guard against a nil entry so we never dereference it. + return nil + } + if receivedSpace.States == nil { + receivedSpace.States = map[string]*State{} + } + for _, shareID := range shareIDs { + delete(receivedSpace.States, shareID) + } + if len(receivedSpace.States) == 0 { + delete(rss.Spaces, spaceID) + } + + 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 removed received shares: 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 removed received shares: 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 removed received shares. retrying...") + // CS3 uses an already exists error instead of precondition failed when using an If-None-Match=* header / IfExists flag in the InitiateFileUpload call. + // That happens when the cache thinks there is no file. + // continue with sync below + default: + span.SetStatus(codes.Error, fmt.Sprintf("persisting removed received shares failed. giving up: %s", err.Error())) + log.Error().Err(err).Msg("persisting removed received shares 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 removed received shares 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) { diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go index bb4e0f35e5d..afc3a5be42f 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache_test.go @@ -227,6 +227,61 @@ var _ = Describe("Cache", func() { }) }) + Describe("RemoveSpaceShares", func() { + var secondShareID = "storageid$spaceid!share2" + + BeforeEach(func() { + rs := &collaboration.ReceivedShare{ + Share: &collaboration.Share{ + Id: &collaboration.ShareId{OpaqueId: secondShareID}, + }, + State: collaboration.ShareState_SHARE_STATE_PENDING, + } + Expect(c.Add(ctx, userID, spaceID, rs)).To(Succeed()) + }) + + It("removes multiple entries in one call", func() { + err := c.RemoveSpaceShares(ctx, userID, spaceID, []string{shareID, secondShareID}) + Expect(err).ToNot(HaveOccurred()) + + spaces, err := c.List(ctx, userID) + Expect(err).ToNot(HaveOccurred()) + Expect(spaces).ToNot(HaveKey(spaceID)) + }) + + It("only removes the requested entries", func() { + err := c.RemoveSpaceShares(ctx, userID, spaceID, []string{shareID}) + Expect(err).ToNot(HaveOccurred()) + + s, err := c.Get(ctx, userID, spaceID, shareID) + Expect(err).ToNot(HaveOccurred()) + Expect(s).To(BeNil()) + + s, err = c.Get(ctx, userID, spaceID, secondShareID) + Expect(err).ToNot(HaveOccurred()) + Expect(s).ToNot(BeNil()) + }) + + It("persists the removal", func() { + err := c.RemoveSpaceShares(ctx, userID, spaceID, []string{shareID, secondShareID}) + Expect(err).ToNot(HaveOccurred()) + + fresh := receivedsharecache.New(storage, 0*time.Second) + spaces, err := fresh.List(ctx, userID) + Expect(err).ToNot(HaveOccurred()) + Expect(spaces).ToNot(HaveKey(spaceID)) + }) + + It("is a no-op for an empty share list", func() { + err := c.RemoveSpaceShares(ctx, userID, spaceID, nil) + Expect(err).ToNot(HaveOccurred()) + + s, err := c.Get(ctx, userID, spaceID, shareID) + Expect(err).ToNot(HaveOccurred()) + Expect(s).ToNot(BeNil()) + }) + }) + Describe("Remove", func() { It("removes the entry", func() { err := c.Remove(ctx, userID, spaceID, shareID)