Locking storage

pkg/appauth/manager/jsoncs3/jsoncs3.go

@@ -24,6 +24,7 @@ import (
 	"github.com/opencloud-eu/reva/v2/pkg/errtypes"
 	"github.com/opencloud-eu/reva/v2/pkg/metadatacache"
 	"github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata"
+	"github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata/locks"
 	"github.com/opencloud-eu/reva/v2/pkg/utils"
 	"github.com/pkg/errors"
 	"github.com/sethvargo/go-diceware/diceware"
@@ -61,6 +62,11 @@ type config struct {
 	// For testing set this -1 to disable automatic updates.
 	UTimeUpdateInterval int `mapstructure:"utime_update_interval_seconds"`
 	UpdateRetryCount    int `mapstructure:"update_retry_count"`
+	// LockBackend selects the locking strategy used to make read-modify-write
+	// cycles atomic across replicas. "disk" (default) uses cross-process file
+	// locks; "memory" uses in-process mutexes. When set, updates use a fully
+	// atomic UploadWithLock instead of etag-based Compare-And-Swap.
+	LockBackend string `mapstructure:"lock_backend"`
 }
 
 const (
@@ -123,7 +129,11 @@ func New(m map[string]any) (appauth.Manager, error) {
 		return nil, fmt.Errorf("appauth jsoncs3 manager: failed initialize password generator: %w", err)
 	}
 
-	cs3, err := metadata.NewCS3Storage(c.ProviderAddr, c.ProviderAddr, c.ServiceUserID, c.ServiceUserIdp, c.MachineAuthAPIKey)
+	var opts []metadata.Option
+	if c.LockBackend == "memory" {
+		opts = append(opts, metadata.WithLocker(locks.NewMemoryLocker()))
+	}
+	cs3, err := metadata.NewCS3Storage(c.ProviderAddr, c.ProviderAddr, c.ServiceUserID, c.ServiceUserIdp, c.MachineAuthAPIKey, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -132,11 +142,17 @@ func New(m map[string]any) (appauth.Manager, error) {
 }
 
 func NewWithOptions(mds metadata.Storage, generator PasswordGenerator, uTimeUpdateInterval time.Duration, updateRetries int) (*manager, error) {
+	var locking metadata.LockingStorage
+	if ls, ok := mds.(metadata.LockingStorage); ok {
+		locking = ls
+	}
+
 	store := metadatacache.New(metadatacache.Options[string, map[string]*apppb.AppPassword]{
-		Storage: mds,
-		Path:    func(userID string) string { return userID + ".json" },
-		Retries: updateRetries,
-		Init:    func() map[string]*apppb.AppPassword { return map[string]*apppb.AppPassword{} },
+		Storage:        mds,
+		Path:           func(userID string) string { return userID + ".json" },
+		Retries:        updateRetries,
+		Init:           func() map[string]*apppb.AppPassword { return map[string]*apppb.AppPassword{} },
+		LockingStorage: locking,
 	})
 
 	return &manager{

pkg/metadatacache/store.go

@@ -57,6 +57,12 @@ type Options[K comparable, V any] struct {
 	// with createIfNotFound=true (e.g. to produce an initialised map rather
 	// than a nil map).
 	Init func() V
+
+	// LockingStorage, when set, makes Update perform a fully-atomic
+	// read-modify-write via LockingStorage.UploadWithLock instead of the
+	// etag-based Compare-And-Swap retry loop. This is safe across replicas
+	// without relying on etag semantics. When nil the etag-CAS path is used.
+	LockingStorage metadata.LockingStorage
 }
 
 // Store is a generic in-memory write-through cache.  V must be
@@ -265,6 +271,13 @@ func (s *Store[K, V]) Update(ctx context.Context, key K, createIfNotFound bool,
 		}
 	}
 
+	// When a locking storage is configured, perform the read-modify-write as a
+	// single atomic operation. The lock held by UploadWithLock guarantees no
+	// other writer can interleave, so no etag-based CAS or retry loop is needed.
+	if s.opts.LockingStorage != nil {
+		return s.updateLocked(ctx, key, createIfNotFound, fn)
+	}
+
 	var lastErr error
 	for attempt := 0; attempt < s.opts.Retries; attempt++ {
 		// Load the current value (or initialise if absent and allowed).
@@ -326,3 +339,44 @@ func (s *Store[K, V]) Update(ctx context.Context, key K, createIfNotFound bool,
 	span.SetStatus(codes.Error, fmt.Sprintf("gave up after %d attempts: %s", s.opts.Retries, lastErr))
 	return fmt.Errorf("metadatacache: update of %v failed after %d attempts: %w", key, s.opts.Retries, lastErr)
 }
+
+// updateLocked performs the read-modify-write using a locking storage. The
+// caller must hold the per-key lock. It mirrors the semantics of Update's
+// etag-CAS path (honouring createIfNotFound and shouldPersist) but replaces the
+// CAS retry loop with a single atomic UploadWithLock call.
+func (s *Store[K, V]) updateLocked(ctx context.Context, key K, createIfNotFound bool, fn func(V) (V, bool, error)) error {
+	p := s.opts.Path(key)
+
+	_, err := s.opts.LockingStorage.UploadWithLock(ctx, metadata.UploadRequest{Path: p}, func(existing []byte) ([]byte, error) {
+		var v V
+		if len(existing) > 0 {
+			if err := json.Unmarshal(existing, &v); err != nil {
+				return nil, err
+			}
+		} else if createIfNotFound && s.opts.Init != nil {
+			v = s.opts.Init()
+		} else {
+			return nil, errtypes.NotFound(fmt.Sprint(key))
+		}
+
+		newV, shouldPersist, err := fn(v)
+		if err != nil {
+			return nil, err
+		}
+
+		// Keep the in-memory cache consistent regardless of whether we persist.
+		s.Set(key, newV)
+
+		if !shouldPersist {
+			// Returning nil signals UploadWithLock to skip the write while
+			// keeping any pre-existing file intact.
+			return nil, nil
+		}
+
+		return json.Marshal(newV)
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}

pkg/metadatacache/store_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/opencloud-eu/reva/v2/pkg/errtypes"
 	"github.com/opencloud-eu/reva/v2/pkg/metadatacache"
 	"github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata"
+	"github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata/locks"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -45,6 +46,21 @@ func newStoreOnDir(dir string) *metadatacache.Store[string, map[string]string] {
 	})
 }
 
+// newLockingStoreOnDir builds a store whose Update uses the fully-atomic
+// LockingStorage.UploadWithLock path (here with an in-memory locker).
+func newLockingStoreOnDir(dir string) *metadatacache.Store[string, map[string]string] {
+	mds, err := metadata.NewDiskStorage(dir, metadata.WithLocker(locks.NewMemoryLocker()))
+	Expect(err).NotTo(HaveOccurred())
+	ctx := context.Background()
+	Expect(mds.Init(ctx, "test")).To(Succeed())
+	return metadatacache.New(metadatacache.Options[string, map[string]string]{
+		Storage:        mds,
+		Path:           func(key string) string { return key + ".json" },
+		Init:           func() map[string]string { return map[string]string{} },
+		LockingStorage: mds.(metadata.LockingStorage),
+	})
+}
+
 var _ = Describe("Store", func() {
 	var (
 		ctx   context.Context
@@ -163,4 +179,72 @@ var _ = Describe("Store", func() {
 			wg.Wait()
 		})
 	})
+
+	Context("with a locking storage", func() {
+		var lockStore *metadatacache.Store[string, map[string]string]
+
+		BeforeEach(func() {
+			lockStore = newLockingStoreOnDir(dir)
+		})
+
+		It("creates an entry when createIfNotFound is true", func() {
+			err := lockStore.Update(ctx, "alice", true, func(m map[string]string) (map[string]string, bool, error) {
+				m["k"] = "v"
+				return m, true, nil
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			unlock := lockStore.Lock("alice")
+			defer unlock()
+			val, ok, err := lockStore.Get(ctx, "alice")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(ok).To(BeTrue())
+			Expect(val).To(HaveKeyWithValue("k", "v"))
+		})
+
+		It("returns NotFound when createIfNotFound is false and key is absent", func() {
+			err := lockStore.Update(ctx, "bob", false, func(m map[string]string) (map[string]string, bool, error) {
+				return m, true, nil
+			})
+			Expect(err).To(BeAssignableToTypeOf(errtypes.NotFound("")))
+		})
+
+		It("does not persist when shouldPersist is false", func() {
+			Expect(lockStore.Update(ctx, "alice", true, func(m map[string]string) (map[string]string, bool, error) {
+				m["a"] = "1"
+				return m, true, nil
+			})).To(Succeed())
+
+			var captured map[string]string
+			Expect(lockStore.Update(ctx, "alice", false, func(m map[string]string) (map[string]string, bool, error) {
+				captured = m
+				return m, false, nil
+			})).To(Succeed())
+			Expect(captured).To(HaveKeyWithValue("a", "1"))
+		})
+
+		It("does not lose updates when many goroutines target the same key", func() {
+			const goroutines = 30
+			var wg sync.WaitGroup
+			wg.Add(goroutines)
+			for i := 0; i < goroutines; i++ {
+				go func(n int) {
+					defer wg.Done()
+					_ = lockStore.Update(ctx, "shared", true, func(m map[string]string) (map[string]string, bool, error) {
+						key := fmt.Sprintf("g%d", n)
+						m[key] = key
+						return m, true, nil
+					})
+				}(i)
+			}
+			wg.Wait()
+
+			unlock := lockStore.Lock("shared")
+			defer unlock()
+			val, ok, err := lockStore.Get(ctx, "shared")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(ok).To(BeTrue())
+			Expect(val).To(HaveLen(goroutines))
+		})
+	})
 })

pkg/publicshare/manager/json/json.go

@@ -50,6 +50,7 @@ import (
 	"github.com/opencloud-eu/reva/v2/pkg/publicshare/manager/registry"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata"
+	"github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata/locks"
 	"github.com/opencloud-eu/reva/v2/pkg/utils"
 	"github.com/pkg/errors"
 )
@@ -98,7 +99,11 @@ func NewCS3(c map[string]interface{}) (publicshare.Manager, error) {
 
 	conf.init()
 
-	s, err := metadata.NewCS3Storage(conf.ProviderAddr, conf.ProviderAddr, conf.ServiceUserID, conf.ServiceUserIdp, conf.MachineAuthAPIKey)
+	var opts []metadata.Option
+	if conf.LockBackend == "memory" {
+		opts = append(opts, metadata.WithLocker(locks.NewMemoryLocker()))
+	}
+	s, err := metadata.NewCS3Storage(conf.ProviderAddr, conf.ProviderAddr, conf.ServiceUserID, conf.ServiceUserIdp, conf.MachineAuthAPIKey, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -142,6 +147,10 @@ type cs3Config struct {
 	ServiceUserID     string `mapstructure:"service_user_id"`
 	ServiceUserIdp    string `mapstructure:"service_user_idp"`
 	MachineAuthAPIKey string `mapstructure:"machine_auth_apikey"`
+	// LockBackend selects the locking strategy used to serialise read-modify-
+	// write cycles. "disk" (default) uses cross-process file locks; "memory"
+	// uses in-process mutexes (useful for single-replica deployments and tests).
+	LockBackend string `mapstructure:"lock_backend"`
 }
 
 func (c *commonConfig) init() {
@@ -306,21 +315,16 @@ func (m *manager) CreatePublicShare(ctx context.Context, u *user.User, rInfo *pr
 		return nil, err
 	}
 
-	db, err := m.persistence.Read(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	if _, ok := db[s.Id.GetOpaqueId()]; !ok {
-		db[s.Id.GetOpaqueId()] = map[string]interface{}{
-			"share":    string(encShare),
-			"password": ps.Password,
+	err = m.persistence.Update(ctx, func(db persistence.PublicShares) (persistence.PublicShares, error) {
+		if _, ok := db[s.Id.GetOpaqueId()]; !ok {
+			db[s.Id.GetOpaqueId()] = map[string]interface{}{
+				"share":    string(encShare),
+				"password": ps.Password,
+			}
+			return db, nil
 		}
-	} else {
 		return nil, errors.New("key already exists")
-	}
-
-	err = m.persistence.Write(ctx, db)
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -394,29 +398,25 @@ func (m *manager) UpdatePublicShare(ctx context.Context, u *user.User, req *link
 		return nil, err
 	}
 
-	db, err := m.persistence.Read(ctx)
-	if err != nil {
-		return nil, err
-	}
-
 	encShare, err := utils.MarshalProtoV1ToJSON(share)
 	if err != nil {
 		return nil, err
 	}
 
-	data, ok := db[share.Id.OpaqueId].(map[string]interface{})
-	if !ok {
-		data = map[string]interface{}{}
-	}
-
-	if ok && passwordChanged {
-		data["password"] = newPasswordEncoded
-	}
-	data["share"] = string(encShare)
+	err = m.persistence.Update(ctx, func(db persistence.PublicShares) (persistence.PublicShares, error) {
+		data, ok := db[share.Id.OpaqueId].(map[string]interface{})
+		if !ok {
+			data = map[string]interface{}{}
+		}
 
-	db[share.Id.OpaqueId] = data
+		if ok && passwordChanged {
+			data["password"] = newPasswordEncoded
+		}
+		data["share"] = string(encShare)
 
-	err = m.persistence.Write(ctx, db)
+		db[share.Id.OpaqueId] = data
+		return db, nil
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -625,29 +625,28 @@ func (m *manager) RevokePublicShare(ctx context.Context, _ *user.User, ref *link
 
 // revokePublicShare doesn't have a lock inside, ensure a lock before call
 func (m *manager) revokePublicShare(ctx context.Context, ref *link.PublicShareReference) error {
-	db, err := m.persistence.Read(ctx)
-	if err != nil {
-		return err
-	}
-
-	switch {
-	case ref.GetId() != nil && ref.GetId().OpaqueId != "":
-		if _, ok := db[ref.GetId().OpaqueId]; ok {
-			delete(db, ref.GetId().OpaqueId)
-		} else {
-			return errors.New("reference does not exist")
-		}
-	case ref.GetToken() != "":
+	if ref.GetToken() != "" && (ref.GetId() == nil || ref.GetId().OpaqueId == "") {
+		// Resolve the token to a share id up front so the atomic update below
+		// only has to perform a simple delete.
 		share, _, err := m.getByToken(ctx, ref.GetToken())
 		if err != nil {
 			return err
 		}
-		delete(db, share.Id.OpaqueId)
-	default:
-		return errors.New("reference does not exist")
+		ref = &link.PublicShareReference{Spec: &link.PublicShareReference_Id{Id: &link.PublicShareId{OpaqueId: share.Id.OpaqueId}}}
 	}
 
-	return m.persistence.Write(ctx, db)
+	return m.persistence.Update(ctx, func(db persistence.PublicShares) (persistence.PublicShares, error) {
+		switch {
+		case ref.GetId() != nil && ref.GetId().OpaqueId != "":
+			if _, ok := db[ref.GetId().OpaqueId]; ok {
+				delete(db, ref.GetId().OpaqueId)
+				return db, nil
+			}
+			return nil, errors.New("reference does not exist")
+		default:
+			return nil, errors.New("reference does not exist")
+		}
+	})
 }
 
 // getByToken doesn't have a lock inside, ensure a lock before call