Add offline purge of orphaned received shares to the jsoncs3 share manager

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

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

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

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