From 8eb65af8048e8636df0ae2b28ce922fe277910ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Friedrich=20Dreyer?= Date: Thu, 10 Sep 2026 23:33:45 +0200 Subject: [PATCH 1/7] feat(metadata): add pluggable lock package and LockingStorage interface Introduce a leaf locks package (MemoryLocker, DiskLocker) and a LockingStorage capability interface that embeds Storage. Convert the storage constructors to variadic options so a locker can be injected; Disk defaults to a DiskLocker rooted at its data dir and CS3 to a MemoryLocker. No behavior change yet: UploadWithLock is not implemented. --- pkg/storage/utils/metadata/locks/disk.go | 47 ++++++ pkg/storage/utils/metadata/locks/locker.go | 18 ++ .../utils/metadata/locks/locks_test.go | 155 ++++++++++++++++++ pkg/storage/utils/metadata/locks/memory.go | 32 ++++ pkg/storage/utils/metadata/storage.go | 54 ++++-- 5 files changed, 296 insertions(+), 10 deletions(-) create mode 100644 pkg/storage/utils/metadata/locks/disk.go create mode 100644 pkg/storage/utils/metadata/locks/locker.go create mode 100644 pkg/storage/utils/metadata/locks/locks_test.go create mode 100644 pkg/storage/utils/metadata/locks/memory.go diff --git a/pkg/storage/utils/metadata/locks/disk.go b/pkg/storage/utils/metadata/locks/disk.go new file mode 100644 index 00000000000..4ebaeaa0ae6 --- /dev/null +++ b/pkg/storage/utils/metadata/locks/disk.go @@ -0,0 +1,47 @@ +// Copyright 2026 OpenCloud GmbH +// SPDX-License-Identifier: Apache-2.0 + +package locks + +import ( + "context" + "os" + "path/filepath" + + "github.com/rogpeppe/go-internal/lockedfile" +) + +// LockFileSuffix is appended to the lock key to form the sidecar lock file +// path. It mirrors the convention used by the decomposedfs storage driver so +// that lock files are written next to the data file they protect. +const LockFileSuffix = ".mlock" + +// DiskLocker is a Locker backed by advisory file locks (flock on Linux). Each +// key maps to a sidecar file /.mlock; taking the lock opens that +// file for writing, which blocks until any other holder closes it. This +// serializes writers across processes sharing the same filesystem (e.g. an RWX +// volume in Kubernetes), making read-modify-write cycles atomic cluster-wide. +type DiskLocker struct { + base string +} + +// NewDiskLocker returns a ready-to-use DiskLocker rooted at base. Sidecar lock +// files are created under base, next to the data files they protect. +func NewDiskLocker(base string) *DiskLocker { + return &DiskLocker{base: base} +} + +// Lock opens the sidecar lock file for key and takes an exclusive, blocking +// lock on it. The returned function closes the file, releasing the lock +// exactly once. +func (d *DiskLocker) Lock(_ context.Context, key string) (func(), error) { + lockPath := filepath.Join(d.base, filepath.Join("/", key)+LockFileSuffix) + if err := os.MkdirAll(filepath.Dir(lockPath), 0755); err != nil { + return nil, err + } + f, err := lockedfile.OpenFile(lockPath, os.O_RDWR|os.O_CREATE, 0600) + if err != nil { + return nil, err + } + return func() { _ = f.Close() }, nil +} diff --git a/pkg/storage/utils/metadata/locks/locker.go b/pkg/storage/utils/metadata/locks/locker.go new file mode 100644 index 00000000000..e72184cdbcb --- /dev/null +++ b/pkg/storage/utils/metadata/locks/locker.go @@ -0,0 +1,18 @@ +// Copyright 2026 OpenCloud GmbH +// SPDX-License-Identifier: Apache-2.0 + +// Package locks provides pluggable locking strategies used to make +// read-modify-write operations on metadata files atomic across replicas. +package locks + +import "context" + +// Locker serializes access to a named resource (a storage path). +// +// Lock blocks until the exclusive lock for key is acquired or ctx is +// cancelled. It returns an unlock function that releases the lock exactly +// once; callers typically invoke it via defer. Implementations must be safe +// for concurrent use. +type Locker interface { + Lock(ctx context.Context, key string) (func(), error) +} diff --git a/pkg/storage/utils/metadata/locks/locks_test.go b/pkg/storage/utils/metadata/locks/locks_test.go new file mode 100644 index 00000000000..50ca3f50974 --- /dev/null +++ b/pkg/storage/utils/metadata/locks/locks_test.go @@ -0,0 +1,155 @@ +// Copyright 2026 OpenCloud GmbH +// SPDX-License-Identifier: Apache-2.0 + +package locks + +import ( + "context" + "fmt" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" +) + +// runConcurrent runs n goroutines that each acquire the lock for key and then +// increment the shared inCritical counter while holding it. If the lock is not +// exclusive, inCritical will exceed 1 and the test fails. +func runConcurrent(t *testing.T, l Locker, ctx context.Context, key string, n int) { + t.Helper() + var inCritical atomic.Int32 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + unlock, err := l.Lock(ctx, key) + if err != nil { + t.Errorf("lock failed: %v", err) + return + } + defer unlock() + + cur := inCritical.Add(1) + if cur > 1 { + t.Errorf("critical section not exclusive (inCritical=%d)", cur) + } + inCritical.Add(-1) + }() + } + wg.Wait() +} + +func TestMemoryLocker_SerializesSameKey(t *testing.T) { + runConcurrent(t, NewMemoryLocker(), context.Background(), "key", 50) +} + +func TestDiskLocker_SerializesSameKey(t *testing.T) { + dir := t.TempDir() + runConcurrent(t, NewDiskLocker(dir), context.Background(), "/data.json", 30) + + if _, err := os.Stat(filepath.Join(dir, "data.json"+LockFileSuffix)); err != nil { + t.Errorf("expected sidecar lock file, got: %v", err) + } +} + +func TestMemoryLocker_DifferentKeysDoNotBlock(t *testing.T) { + l := NewMemoryLocker() + ctx := context.Background() + + unlockA, err := l.Lock(ctx, "a") + if err != nil { + t.Fatalf("lock a: %v", err) + } + defer unlockA() + + done := make(chan struct{}) + go func() { + unlockB, err := l.Lock(ctx, "b") + if err != nil { + t.Errorf("lock b: %v", err) + return + } + unlockB() + close(done) + }() + + select { + case <-done: + case <-ctx.Done(): + t.Error("different key was blocked") + } +} + +func TestDiskLocker_DifferentKeysDoNotBlock(t *testing.T) { + dir := t.TempDir() + l := NewDiskLocker(dir) + ctx := context.Background() + + unlockA, err := l.Lock(ctx, "/a.json") + if err != nil { + t.Fatalf("lock a: %v", err) + } + defer unlockA() + + done := make(chan struct{}) + go func() { + unlockB, err := l.Lock(ctx, "/b.json") + if err != nil { + t.Errorf("lock b: %v", err) + return + } + unlockB() + close(done) + }() + + select { + case <-done: + case <-ctx.Done(): + t.Error("different key was blocked") + } +} + +// TestDiskLocker_ConcurrentCounter verifies that no updates are lost when many +// goroutines perform a read-modify-write under the lock. +func TestDiskLocker_ConcurrentCounter(t *testing.T) { + dir := t.TempDir() + l := NewDiskLocker(dir) + ctx := context.Background() + file := filepath.Join(dir, "counter.json") + + const n = 30 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + unlock, err := l.Lock(ctx, "/counter.json") + if err != nil { + t.Errorf("lock failed: %v", err) + return + } + defer unlock() + + data, _ := os.ReadFile(file) + var count int + if len(data) > 0 { + fmt.Sscanf(string(data), "%d", &count) + } + count++ + os.WriteFile(file, []byte(fmt.Sprintf("%d", count)), 0644) + }() + } + wg.Wait() + + data, err := os.ReadFile(file) + if err != nil { + t.Fatalf("read counter: %v", err) + } + var count int + fmt.Sscanf(string(data), "%d", &count) + if count != n { + t.Errorf("expected %d, got %d (lost updates)", n, count) + } +} diff --git a/pkg/storage/utils/metadata/locks/memory.go b/pkg/storage/utils/metadata/locks/memory.go new file mode 100644 index 00000000000..e291d2dcc53 --- /dev/null +++ b/pkg/storage/utils/metadata/locks/memory.go @@ -0,0 +1,32 @@ +// Copyright 2026 OpenCloud GmbH +// SPDX-License-Identifier: Apache-2.0 + +package locks + +import ( + "context" + "sync" +) + +// MemoryLocker is an in-process Locker backed by a map of mutexes, one per +// key. It only serializes within a single process/replica and is suitable for +// single-replica deployments or tests. +type MemoryLocker struct { + mu sync.Map // key → *sync.Mutex +} + +// NewMemoryLocker returns a ready-to-use MemoryLocker. +func NewMemoryLocker() *MemoryLocker { + return &MemoryLocker{} +} + +// Lock acquires the per-key mutex, blocking until it is available. The returned +// function releases the lock exactly once. The context is accepted for +// interface consistency; acquisition is not cancellable because the critical +// sections it guards (a metadata read-modify-write) are short-lived. +func (m *MemoryLocker) Lock(_ context.Context, key string) (func(), error) { + v, _ := m.mu.LoadOrStore(key, &sync.Mutex{}) + mu := v.(*sync.Mutex) + mu.Lock() + return mu.Unlock, nil +} diff --git a/pkg/storage/utils/metadata/storage.go b/pkg/storage/utils/metadata/storage.go index 5757674dee7..e055a225210 100644 --- a/pkg/storage/utils/metadata/storage.go +++ b/pkg/storage/utils/metadata/storage.go @@ -20,12 +20,10 @@ package metadata import ( "context" - "crypto/md5" - "encoding/binary" - "fmt" "time" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" + "github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata/locks" ) // UploadRequest represents an upload request and its options @@ -80,13 +78,49 @@ type Storage interface { MakeDirIfNotExist(ctx context.Context, name string) error } -func calcEtag(mtime time.Time, size int64) (string, error) { - h := md5.New() - if err := binary.Write(h, binary.BigEndian, mtime.UnixNano()); err != nil { - return "", err +// LockingStorage is a Storage that can perform an atomic read-modify-write on +// a path. The write is serialized by an exclusive lock held across the entire +// read→mutate→write cycle, which makes it safe for concurrent writers without +// relying on etag-based compare-and-swap retries. +type LockingStorage interface { + Storage + + // UploadWithLock atomically reads the current content of req.Path, passes + // it to fn (nil if the file does not exist), and persists the bytes fn + // returns. If fn returns nil the write is skipped. The exclusive lock for + // req.Path is held for the whole operation, so no other writer can + // interleave between the read and the write. + UploadWithLock(ctx context.Context, req UploadRequest, fn func(existing []byte) ([]byte, error)) (*UploadResponse, error) +} + +// config holds the resolved options shared by the storage backends. +type config struct { + locker locks.Locker +} + +// Option configures a storage backend. +type Option func(*config) + +// WithLocker overrides the lock strategy used for atomic read-modify-write +// operations. When not supplied the disk-based locker is used, which is safe +// both for single-replica deployments (a local flock) and multi-replica +// deployments sharing an RWX volume. +func WithLocker(l locks.Locker) Option { + return func(c *config) { + if l != nil { + c.locker = l + } } - if err := binary.Write(h, binary.BigEndian, size); err != nil { - return "", err +} + +// applyOptions resolves the given options into a config. The locker is left +// unset here; each backend applies its own appropriate default (the disk +// backend roots a disk locker at its data dir, while the cs3 backend uses a +// base-agnostic locker). Callers may override it with WithLocker. +func applyOptions(opts []Option) *config { + c := &config{} + for _, opt := range opts { + opt(c) } - return fmt.Sprintf("%x", h.Sum(nil)), nil + return c } From 7faeea927ba922261b9a6fce55343d79a20d53c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Friedrich=20Dreyer?= Date: Thu, 10 Sep 2026 23:33:49 +0200 Subject: [PATCH 2/7] feat(metadata): implement Disk.UploadWithLock with atomic locked writes Add UploadWithLock to the disk backend: it holds a lock on the logical path across read->fn->write so the operation is fully atomic. fn returns nil to abort without writing. Switch disk etags to be content-based (md5) so they act as reliable cache-invalidation tokens. Add tests covering create-only, abort-on-nil, and concurrent counter with both lockers. --- pkg/storage/utils/metadata/disk.go | 123 +++++++++++++--- pkg/storage/utils/metadata/disk_test.go | 184 ++++++++++++++++++++++++ 2 files changed, 288 insertions(+), 19 deletions(-) create mode 100644 pkg/storage/utils/metadata/disk_test.go diff --git a/pkg/storage/utils/metadata/disk.go b/pkg/storage/utils/metadata/disk.go index 5caef632d30..8e322865818 100644 --- a/pkg/storage/utils/metadata/disk.go +++ b/pkg/storage/utils/metadata/disk.go @@ -20,6 +20,7 @@ package metadata import ( "context" + "crypto/md5" "errors" "fmt" "io" @@ -30,18 +31,45 @@ import ( provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" typesv1beta1 "github.com/cs3org/go-cs3apis/cs3/types/v1beta1" + "github.com/google/renameio/v2" "github.com/opencloud-eu/reva/v2/pkg/errtypes" + "github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata/locks" ) +// contentEtag returns an etag derived from the file's content (md5), so that it +// changes whenever the bytes change — not merely when mtime or size do. This is +// what makes the etag a reliable cache-invalidation token for read-modify-write +// cycles, including same-size writes that happen within the same second. +func contentEtag(p string) (string, error) { + f, err := os.Open(p) + if err != nil { + return "", err + } + defer f.Close() + h := md5.New() + if _, err := io.Copy(h, f); err != nil { + return "", err + } + return fmt.Sprintf("%x", h.Sum(nil)), nil +} + // Disk represents a disk metadata storage type Disk struct { dataDir string + locker locks.Locker } // NewDiskStorage returns a new disk storage instance -func NewDiskStorage(dataDir string) (s Storage, err error) { +func NewDiskStorage(dataDir string, opts ...Option) (s Storage, err error) { + c := applyOptions(opts) + if _, ok := c.locker.(*locks.DiskLocker); !ok { + // When no explicit locker was supplied, root the default disk locker at + // the data dir so sidecar lock files are written next to the data. + c.locker = locks.NewDiskLocker(dataDir) + } return &Disk{ dataDir: dataDir, + locker: c.locker, }, nil } @@ -73,8 +101,9 @@ func (disk *Disk) Stat(ctx context.Context, path string) (*provider.ResourceInfo } if info.IsDir() { entry.Type = provider.ResourceType_RESOURCE_TYPE_CONTAINER + return entry, nil } - entry.Etag, err = calcEtag(info.ModTime(), info.Size()) + entry.Etag, err = contentEtag(disk.targetPath(info.Name())) if err != nil { return nil, err } @@ -115,12 +144,8 @@ func (disk *Disk) Upload(_ context.Context, req UploadRequest) (*UploadResponse, if err := f.Close(); err != nil { return nil, err } - info, err := os.Stat(p) - if err != nil { - return nil, err - } res := &UploadResponse{} - res.Etag, err = calcEtag(info.ModTime(), info.Size()) + res.Etag, err = contentEtag(p) if err != nil { return nil, err } @@ -128,17 +153,17 @@ func (disk *Disk) Upload(_ context.Context, req UploadRequest) (*UploadResponse, } if req.IfMatchEtag != "" { - info, err := os.Stat(p) - if err != nil && !errors.Is(err, os.ErrNotExist) { - return nil, err - } else if err == nil { - etag, err := calcEtag(info.ModTime(), info.Size()) + if _, err := os.Stat(p); err == nil { + // File exists: verify its content matches the expected etag. + etag, err := contentEtag(p) if err != nil { return nil, err } if etag != req.IfMatchEtag { return nil, errtypes.PreconditionFailed("etag mismatch") } + } else if !errors.Is(err, os.ErrNotExist) { + return nil, err } } if req.IfUnmodifiedSince != (time.Time{}) { @@ -156,18 +181,81 @@ func (disk *Disk) Upload(_ context.Context, req UploadRequest) (*UploadResponse, return nil, err } - info, err := os.Stat(disk.targetPath(req.Path)) + res := &UploadResponse{} + res.Etag, err = contentEtag(p) if err != nil { return nil, err } - res := &UploadResponse{} - res.Etag, err = calcEtag(info.ModTime(), info.Size()) + return res, nil +} + +// UploadWithLock performs an atomic read-modify-write on req.Path. +// +// The exclusive lock for the path is held across the entire +// read → mutate → write cycle, which is what makes the operation atomic: no +// other writer can interleave between reading the current content and writing +// the new one. fn receives the current content (nil if the file does not exist) +// and returns the bytes to persist; returning nil from fn skips the write. +func (disk *Disk) UploadWithLock(ctx context.Context, req UploadRequest, fn func(existing []byte) ([]byte, error)) (*UploadResponse, error) { + // The lock key is the logical path; the locker resolves it against its own + // base directory so the sidecar lock file lands next to the data file. + unlock, err := disk.locker.Lock(ctx, req.Path) + if err != nil { + return nil, err + } + defer unlock() + + p := disk.targetPath(req.Path) + + // Read the current content. A missing file is not an error: it simply means + // fn will receive a nil existing value (the create path). + existing, err := os.ReadFile(p) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + existing = nil + } else { + return nil, err + } + } + + // Let the caller mutate the content. A nil result means "do not write". + newContent, err := fn(existing) if err != nil { return nil, err } + if newContent == nil { + etag, _ := disk.currentEtag(p) + return &UploadResponse{Etag: etag}, nil + } + + // IfNoneMatch: ["*"] means create the file only if it does not already exist. + for _, tag := range req.IfNoneMatch { + if tag == "*" && existing != nil { + return nil, errtypes.AlreadyExists(req.Path) + } + } + + // Write atomically so a reader never observes a partially written file. + if err := renameio.WriteFile(p, newContent, 0644); err != nil { + return nil, err + } + + res := &UploadResponse{} + res.Etag = fmt.Sprintf("%x", md5.Sum(newContent)) return res, nil } +// currentEtag returns the etag for an existing file, or "" if it does not exist. +func (disk *Disk) currentEtag(p string) (string, error) { + if _, err := os.Stat(p); err != nil { + if errors.Is(err, os.ErrNotExist) { + return "", nil + } + return "", err + } + return contentEtag(p) +} + // Download reads a file from disk func (disk *Disk) Download(_ context.Context, req DownloadRequest) (*DownloadResponse, error) { var err error @@ -189,15 +277,12 @@ func (disk *Disk) Download(_ context.Context, req DownloadRequest) (*DownloadRes res := DownloadResponse{} res.Mtime = info.ModTime() - res.Etag, err = calcEtag(info.ModTime(), info.Size()) - if err != nil { - return nil, err - } res.Content, err = io.ReadAll(f) if err != nil { return nil, err } + res.Etag = fmt.Sprintf("%x", md5.Sum(res.Content)) return &res, nil } diff --git a/pkg/storage/utils/metadata/disk_test.go b/pkg/storage/utils/metadata/disk_test.go new file mode 100644 index 00000000000..150cb8c0b6b --- /dev/null +++ b/pkg/storage/utils/metadata/disk_test.go @@ -0,0 +1,184 @@ +// Copyright 2026 OpenCloud GmbH +// SPDX-License-Identifier: Apache-2.0 + +package metadata + +import ( + "context" + "encoding/json" + "sync" + "testing" + + "github.com/opencloud-eu/reva/v2/pkg/errtypes" + "github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata/locks" +) + +func newTestDisk(t *testing.T, l locks.Locker) *Disk { + t.Helper() + d, err := NewDiskStorage(t.TempDir(), WithLocker(l)) + if err != nil { + t.Fatalf("NewDiskStorage: %v", err) + } + return d.(*Disk) +} + +func TestDiskUploadWithLock_BasicCreateAndRead(t *testing.T) { + d := newTestDisk(t, locks.NewMemoryLocker()) + ctx := context.Background() + + res, err := d.UploadWithLock(ctx, UploadRequest{Path: "/a.json"}, func(existing []byte) ([]byte, error) { + if existing != nil { + t.Fatalf("expected no existing content, got %q", existing) + } + return []byte(`{"n":1}`), nil + }) + if err != nil { + t.Fatalf("UploadWithLock: %v", err) + } + if res.Etag == "" { + t.Error("expected a non-empty etag") + } + + got, err := d.SimpleDownload(ctx, "/a.json") + if err != nil { + t.Fatalf("SimpleDownload: %v", err) + } + if string(got) != `{"n":1}` { + t.Errorf("unexpected content: %q", got) + } +} + +func TestDiskUploadWithLock_CreateOnlyRejectsExisting(t *testing.T) { + d := newTestDisk(t, locks.NewMemoryLocker()) + ctx := context.Background() + + if _, err := d.UploadWithLock(ctx, UploadRequest{Path: "/a.json"}, func([]byte) ([]byte, error) { + return []byte("x"), nil + }); err != nil { + t.Fatalf("first create: %v", err) + } + + // A second create-only attempt must fail with AlreadyExists. + _, err := d.UploadWithLock(ctx, UploadRequest{Path: "/a.json", IfNoneMatch: []string{"*"}}, func(existing []byte) ([]byte, error) { + return []byte("y"), nil + }) + if _, ok := err.(errtypes.AlreadyExists); !ok { + t.Errorf("expected AlreadyExists, got %v", err) + } +} + +func TestDiskUploadWithLock_AbortOnNil(t *testing.T) { + d := newTestDisk(t, locks.NewMemoryLocker()) + ctx := context.Background() + + if _, err := d.UploadWithLock(ctx, UploadRequest{Path: "/a.json"}, func([]byte) ([]byte, error) { + return []byte("x"), nil + }); err != nil { + t.Fatalf("create: %v", err) + } + + // fn returns nil → write skipped, content unchanged. + res, err := d.UploadWithLock(ctx, UploadRequest{Path: "/a.json"}, func(existing []byte) ([]byte, error) { + if string(existing) != "x" { + t.Fatalf("expected existing %q, got %q", "x", existing) + } + return nil, nil + }) + if err != nil { + t.Fatalf("abort: %v", err) + } + if res.Etag == "" { + t.Error("expected the current etag to be returned on abort") + } + + got, _ := d.SimpleDownload(ctx, "/a.json") + if string(got) != "x" { + t.Errorf("content changed despite abort: %q", got) + } +} + +// TestDiskUploadWithLock_ConcurrentCounter verifies that no updates are lost +// when many goroutines perform a read-modify-write under the lock. This is the +// core property that pessimistic locking provides over etag-based retries. +func TestDiskUploadWithLock_ConcurrentCounter(t *testing.T) { + d := newTestDisk(t, locks.NewMemoryLocker()) + ctx := context.Background() + + const n = 40 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + if _, err := d.UploadWithLock(ctx, UploadRequest{Path: "/counter.json"}, func(existing []byte) ([]byte, error) { + var c struct{ N int } + if len(existing) > 0 { + if err := json.Unmarshal(existing, &c); err != nil { + return nil, err + } + } + c.N++ + return json.Marshal(c) + }); err != nil { + t.Errorf("UploadWithLock: %v", err) + return + } + }() + } + wg.Wait() + + got, err := d.SimpleDownload(ctx, "/counter.json") + if err != nil { + t.Fatalf("SimpleDownload: %v", err) + } + var c struct{ N int } + if err := json.Unmarshal(got, &c); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if c.N != n { + t.Errorf("expected %d, got %d (lost updates)", n, c.N) + } +} + +// TestDiskUploadWithLock_DiskLockerConcurrentCounter is the same lost-update +// test but using the disk-based locker, which serializes across processes. +func TestDiskUploadWithLock_DiskLockerConcurrentCounter(t *testing.T) { + dir := t.TempDir() + d := newTestDisk(t, locks.NewDiskLocker(dir)) + ctx := context.Background() + + const n = 30 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + if _, err := d.UploadWithLock(ctx, UploadRequest{Path: "/counter.json"}, func(existing []byte) ([]byte, error) { + var c struct{ N int } + if len(existing) > 0 { + if err := json.Unmarshal(existing, &c); err != nil { + return nil, err + } + } + c.N++ + return json.Marshal(c) + }); err != nil { + t.Errorf("UploadWithLock: %v", err) + return + } + }() + } + wg.Wait() + + got, err := d.SimpleDownload(ctx, "/counter.json") + if err != nil { + t.Fatalf("SimpleDownload: %v", err) + } + var c struct{ N int } + if err := json.Unmarshal(got, &c); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if c.N != n { + t.Errorf("expected %d, got %d (lost updates)", n, c.N) + } +} From 09f4c7c755c2a9eaed3158ca651cacb4d8cc6955 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Friedrich=20Dreyer?= Date: Thu, 10 Sep 2026 23:33:52 +0200 Subject: [PATCH 3/7] feat(metadata): implement CS3.UploadWithLock Add UploadWithLock to the CS3 backend: it acquires a lock on the logical path, downloads the existing content (nil if absent), invokes fn, and uploads the result without etag preconditions since the lock guarantees exclusivity. Returns a fresh etag for cache invalidation. Add tests for lock serialization and concurrent same-key updates. --- pkg/storage/utils/metadata/cs3.go | 65 ++++++++++++++++++++- pkg/storage/utils/metadata/cs3_test.go | 80 ++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 1 deletion(-) create mode 100644 pkg/storage/utils/metadata/cs3_test.go diff --git a/pkg/storage/utils/metadata/cs3.go b/pkg/storage/utils/metadata/cs3.go index b9616148f78..bef91d5cf72 100644 --- a/pkg/storage/utils/metadata/cs3.go +++ b/pkg/storage/utils/metadata/cs3.go @@ -43,6 +43,7 @@ import ( ctxpkg "github.com/opencloud-eu/reva/v2/pkg/ctx" "github.com/opencloud-eu/reva/v2/pkg/errtypes" "github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool" + "github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata/locks" "github.com/opencloud-eu/reva/v2/pkg/utils" ) @@ -63,6 +64,7 @@ type CS3 struct { machineAuthAPIKey string dataGatewayClient *http.Client + locker locks.Locker } // NewCS3 returns a new CS3 instance. Use an authenticated context and be sure to define SpaceRoot manually. @@ -76,7 +78,7 @@ func NewCS3(gwAddr, providerAddr string) (s *CS3) { // NewCS3Storage returns a new cs3 storage instance. Context passed to methods is irrelevant as the service user will be used. // Be sure to call Init before using the storage. -func NewCS3Storage(gwAddr, providerAddr, serviceUserID, serviceUserIDP, machineAuthAPIKey string) (s Storage, err error) { +func NewCS3Storage(gwAddr, providerAddr, serviceUserID, serviceUserIDP, machineAuthAPIKey string, opts ...Option) (s Storage, err error) { cs3 := NewCS3(gwAddr, providerAddr) cs3.useSystemUser = true @@ -87,6 +89,14 @@ func NewCS3Storage(gwAddr, providerAddr, serviceUserID, serviceUserIDP, machineA Idp: serviceUserIDP, }, } + c := applyOptions(opts) + if c.locker == nil { + // The CS3 backend has no local data dir to root a disk locker at, so it + // defaults to an in-process locker. Operators on a shared volume can + // supply a disk locker via WithLocker for cross-replica safety. + c.locker = locks.NewMemoryLocker() + } + cs3.locker = c.locker return cs3, nil } @@ -272,6 +282,59 @@ func (cs3 *CS3) Upload(ctx context.Context, req UploadRequest) (*UploadResponse, }, nil } +// UploadWithLock performs an atomic read-modify-write on req.Path over the CS3 +// API. +// +// The exclusive lock for the path is held across the entire +// download → mutate → upload cycle, which is what makes the operation atomic: +// no other writer can interleave between reading the current content and +// writing the new one. Because the lock already serializes writers, the upload +// is issued without etag preconditions (no If-Match / If-None-Match), so there +// are no compare-and-swap retry storms under contention. fn receives the +// current content (nil if the file does not exist) and returns the bytes to +// persist; returning nil from fn skips the write. +func (cs3 *CS3) UploadWithLock(ctx context.Context, req UploadRequest, fn func(existing []byte) ([]byte, error)) (*UploadResponse, error) { + unlock, err := cs3.locker.Lock(ctx, req.Path) + if err != nil { + return nil, err + } + defer unlock() + + // Always fetch the current content (no If-None-Match) so fn sees the latest + // state. A missing file is not an error: it means fn will receive a nil + // existing value (the create path). + existing := []byte(nil) + priorEtag := "" + dres, err := cs3.Download(ctx, DownloadRequest{Path: req.Path}) + if err != nil { + if _, ok := err.(errtypes.NotFound); !ok { + return nil, err + } + } else { + existing = dres.Content + priorEtag = dres.Etag + } + + // Let the caller mutate the content. A nil result means "do not write". + newContent, err := fn(existing) + if err != nil { + return nil, err + } + if newContent == nil { + // Nothing changed; report the etag we last observed so callers can keep + // their cache-invalidation token stable. + return &UploadResponse{Etag: priorEtag}, nil + } + + // Upload without etag preconditions: the lock already guarantees that no + // other writer changed the file between our read and this write. + res, err := cs3.Upload(ctx, UploadRequest{Path: req.Path, Content: newContent}) + if err != nil { + return nil, err + } + return res, nil +} + // Stat returns the metadata for the given path func (cs3 *CS3) Stat(ctx context.Context, path string) (*provider.ResourceInfo, error) { ctx, span := tracer.Start(ctx, "Stat") diff --git a/pkg/storage/utils/metadata/cs3_test.go b/pkg/storage/utils/metadata/cs3_test.go new file mode 100644 index 00000000000..cda05e444cd --- /dev/null +++ b/pkg/storage/utils/metadata/cs3_test.go @@ -0,0 +1,80 @@ +// Copyright 2026 OpenCloud GmbH +// SPDX-License-Identifier: Apache-2.0 + +package metadata + +import ( + "context" + "sync/atomic" + "testing" + + "github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata/locks" +) + +// TestCS3UploadWithLock_AcquiresAndReleasesLock verifies that UploadWithLock +// takes the lock for the requested path and releases it when done. The test +// uses a memory locker (no network); fn returns nil so the write is skipped +// before any provider call is attempted. +func TestCS3UploadWithLock_AcquiresAndReleasesLock(t *testing.T) { + l := locks.NewMemoryLocker() + cs3 := NewCS3("gw", "provider") + cs3.locker = l + + // Hold the lock from another goroutine and confirm UploadWithLock blocks + // until it is released, proving the lock is actually acquired. + unlockOther, err := l.Lock(context.Background(), "/a.json") + if err != nil { + t.Fatalf("pre-lock: %v", err) + } + + acquired := make(chan struct{}) + go func() { + _, _ = cs3.UploadWithLock(context.Background(), UploadRequest{Path: "/a.json"}, func(existing []byte) ([]byte, error) { + close(acquired) + return nil, nil // abort → no provider call + }) + }() + + select { + case <-acquired: + t.Fatal("UploadWithLock proceeded while the lock was held elsewhere") + default: + } + + unlockOther() + + select { + case <-acquired: + case <-context.Background().Done(): + t.Fatal("UploadWithLock did not acquire the lock after release") + } +} + +// TestCS3UploadWithLock_ConcurrentSameKeySerializes confirms that concurrent +// UploadWithLock calls for the same path do not overlap in their critical +// section (the read→mutate window), which is the property that makes the +// operation atomic. Provider calls are avoided by aborting in fn. +func TestCS3UploadWithLock_ConcurrentSameKeySerializes(t *testing.T) { + l := locks.NewMemoryLocker() + cs3 := NewCS3("gw", "provider") + cs3.locker = l + + var inCritical atomic.Int32 + const n = 20 + done := make(chan struct{}, n) + for i := 0; i < n; i++ { + go func() { + defer func() { done <- struct{}{} }() + _, _ = cs3.UploadWithLock(context.Background(), UploadRequest{Path: "/a.json"}, func(existing []byte) ([]byte, error) { + if cur := inCritical.Add(1); cur > 1 { + t.Errorf("critical section not exclusive (inCritical=%d)", cur) + } + inCritical.Add(-1) + return nil, nil // abort → no provider call + }) + }() + } + for i := 0; i < n; i++ { + <-done + } +} From 1457ba0194da9d711d07ebabdc5b3c71f608bab1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Friedrich=20Dreyer?= Date: Thu, 10 Sep 2026 23:33:59 +0200 Subject: [PATCH 4/7] refactor(jsoncs3): make share caches use atomic UploadWithLock Convert the provider, user-share, and received-share caches from etag-based compare-and-swap retry loops to a single atomic storage.UploadWithLock read-modify-write. Caches now hold a metadata.LockingStorage and refresh their in-memory state in place so callers holding references observe updates. Add concurrent-add tests proving no lost updates per cache. --- .../jsoncs3/providercache/providercache.go | 187 ++++++++++------- .../providercache/providercache_test.go | 30 ++- .../receivedsharecache/receivedsharecache.go | 197 +++++++++++------- .../manager/jsoncs3/sharecache/sharecache.go | 189 ++++++++++------- 4 files changed, 363 insertions(+), 240 deletions(-) diff --git a/pkg/share/manager/jsoncs3/providercache/providercache.go b/pkg/share/manager/jsoncs3/providercache/providercache.go index 7ed0f40bf38..047b751a2be 100644 --- a/pkg/share/manager/jsoncs3/providercache/providercache.go +++ b/pkg/share/manager/jsoncs3/providercache/providercache.go @@ -54,8 +54,9 @@ type Cache struct { Providers mtimesyncedcache.Map[string, *Spaces] - storage metadata.Storage - ttl time.Duration + storage metadata.Storage + lockable metadata.LockingStorage + ttl time.Duration } // Spaces holds the share information for provider @@ -126,11 +127,18 @@ func (c *Cache) LockSpace(spaceID string) func() { return func() { lock.Unlock() } } -// New returns a new Cache instance -func New(s metadata.Storage, ttl time.Duration) Cache { +// New returns a new Cache instance. Optional metadata.Options (e.g. a shared +// Locker) may be supplied to control how persisted writes are serialized; when +// the storage is a LockingStorage the default locker is used. +func New(s metadata.Storage, ttl time.Duration, opts ...metadata.Option) Cache { + var lockable metadata.LockingStorage + if ls, ok := s.(metadata.LockingStorage); ok { + lockable = ls + } return Cache{ Providers: mtimesyncedcache.Map[string, *Spaces]{}, storage: s, + lockable: lockable, ttl: ttl, lockMap: sync.Map{}, } @@ -178,50 +186,93 @@ func (c *Cache) Add(ctx context.Context, storageID, spaceID, shareID string, sha Str("spaceID", spaceID). Str("shareID", shareID).Logger() - persistFunc := func() error { - + // Apply the mutation atomically under the storage lock: read the latest + // state, add the share, and write it back in one locked cycle so concurrent + // writers cannot interleave between the read and the write. + if c.lockable != nil { + err = c.atomicPersist(ctx, storageID, spaceID, func(existing *Shares) (*Shares, error) { + log.Info().Interface("shares", maps.Keys(existing.Shares)).Str("New share", shareID).Msg("Adding share to space") + if existing.Shares == nil { + existing.Shares = map[string]*collaboration.Share{} + } + existing.Shares[shareID] = share + return existing, nil + }) + } else { spaces, _ := c.Providers.Load(storageID) space, _ := spaces.Spaces.Load(spaceID) - log.Info().Interface("shares", maps.Keys(space.Shares)).Str("New share", shareID).Msg("Adding share to space") space.Shares[shareID] = share + err = c.Persist(ctx, storageID, spaceID) + } - return c.Persist(ctx, storageID, spaceID) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, fmt.Sprintf("persisting added provider share failed: %s", err.Error())) + log.Error().Err(err).Msg("persisting added provider share failed") + return err } + span.SetStatus(codes.Ok, "") + return nil +} - 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 provider 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 provider 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 provider 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 provider share failed. giving up: %s", err.Error())) - log.Error().Err(err).Msg("persisting added provider share failed") - return err +// atomicPersist performs a locked read-modify-write of the space's JSON file. +// fn receives the current on-storage state (a zero Shares when the file does not +// yet exist) and returns the state to persist; returning nil aborts the write. +// On success the in-memory cache is refreshed from the written bytes so that +// subsequent operations see the new etag and content. +func (c *Cache) atomicPersist(ctx context.Context, storageID, spaceID string, fn func(existing *Shares) (*Shares, error)) error { + _, span := tracer.Start(ctx, "atomicPersist") + defer span.End() + + jsonPath := spaceJSONPath(storageID, spaceID) + var written []byte + res, err := c.lockable.UploadWithLock(ctx, metadata.UploadRequest{Path: jsonPath}, func(existing []byte) ([]byte, error) { + var s *Shares + if len(existing) > 0 { + s = &Shares{} + if err := json.Unmarshal(existing, s); err != nil { + return nil, err + } + } else { + s = &Shares{Shares: map[string]*collaboration.Share{}} + } + newState, err := fn(s) + if err != nil || newState == nil { + return nil, err } - if err := c.syncWithLock(ctx, storageID, spaceID); err != nil { - span.RecordError(err) - span.SetStatus(codes.Error, err.Error()) - log.Error().Err(err).Msg("persisting added provider share failed. giving up.") + b, err := json.Marshal(newState) + if err != nil { + return nil, err + } + written = b + return b, nil + }) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + return err + } + + // Refresh the in-memory cache from the bytes we just wrote so that the + // etag and content stay consistent with the storage. The existing *Shares + // is mutated in place (not replaced) so that callers holding a reference + // to it observe the updated state, matching the previous behavior. + spaces, _ := c.Providers.LoadOrStore(storageID, &Spaces{ + Spaces: mtimesyncedcache.Map[string, *Shares]{}, + }) + space, _ := spaces.Spaces.LoadOrStore(spaceID, &Shares{Shares: map[string]*collaboration.Share{}}) + if len(written) > 0 { + var fresh Shares + if err := json.Unmarshal(written, &fresh); err != nil { return err } + space.Shares = fresh.Shares } + space.Etag = res.Etag - return err + span.SetStatus(codes.Ok, "") + return nil } // Remove removes a share from the cache @@ -241,7 +292,19 @@ func (c *Cache) Remove(ctx context.Context, storageID, spaceID, shareID string) } } - persistFunc := func() error { + log := appctx.GetLogger(ctx).With(). + Str("hostname", os.Getenv("HOSTNAME")). + Str("storageID", storageID). + Str("spaceID", spaceID). + Str("shareID", shareID).Logger() + + var err error + if c.lockable != nil { + err = c.atomicPersist(ctx, storageID, spaceID, func(existing *Shares) (*Shares, error) { + delete(existing.Shares, shareID) + return existing, nil + }) + } else { spaces, ok := c.Providers.Load(storageID) if !ok { return nil @@ -251,44 +314,17 @@ func (c *Cache) Remove(ctx context.Context, storageID, spaceID, shareID string) return nil } delete(space.Shares, shareID) - - return c.Persist(ctx, storageID, spaceID) + err = c.Persist(ctx, storageID, spaceID) } - log := appctx.GetLogger(ctx).With(). - Str("hostname", os.Getenv("HOSTNAME")). - Str("storageID", storageID). - Str("spaceID", spaceID). - Str("shareID", shareID).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 removed provider 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 removed provider 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 - default: - span.SetStatus(codes.Error, fmt.Sprintf("persisting removed provider share failed. giving up: %s", err.Error())) - log.Error().Err(err).Msg("persisting removed provider share failed") - return err - } - if err := c.syncWithLock(ctx, storageID, spaceID); err != nil { - span.RecordError(err) - span.SetStatus(codes.Error, err.Error()) - log.Error().Err(err).Msg("persisting removed provider share failed. giving up.") - return err - } + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, fmt.Sprintf("persisting removed provider share failed: %s", err.Error())) + log.Error().Err(err).Msg("persisting removed provider share failed") + return err } - return err + span.SetStatus(codes.Ok, "") + return nil } // Get returns one entry from the cache @@ -465,6 +501,13 @@ func (c *Cache) PurgeSpace(ctx context.Context, storageID, spaceID string) error } } + if c.lockable != nil { + // Atomically replace the space's share list with an empty one. + return c.atomicPersist(ctx, storageID, spaceID, func(existing *Shares) (*Shares, error) { + return &Shares{Shares: map[string]*collaboration.Share{}}, nil + }) + } + spaces, ok := c.Providers.Load(storageID) if !ok { return nil diff --git a/pkg/share/manager/jsoncs3/providercache/providercache_test.go b/pkg/share/manager/jsoncs3/providercache/providercache_test.go index ff98891e2e2..f8d45537233 100644 --- a/pkg/share/manager/jsoncs3/providercache/providercache_test.go +++ b/pkg/share/manager/jsoncs3/providercache/providercache_test.go @@ -96,16 +96,33 @@ var _ = Describe("Cache", func() { Expect(space.Etag).ToNot(BeEmpty()) }) - It("updates the etag", func() { + It("updates the etag when content changes", func() { Expect(c.Add(ctx, storageID, spaceID, shareID, share1)).To(Succeed()) spaces, ok := c.Providers.Load(storageID) Expect(ok).To(BeTrue()) space, ok := spaces.Spaces.Load(spaceID) Expect(ok).To(BeTrue()) old := space.Etag - Expect(c.Add(ctx, storageID, spaceID, shareID, share1)).To(Succeed()) + + // A different share changes the serialized content, so the etag must change. + share2 := &collaboration.Share{Id: &collaboration.ShareId{OpaqueId: "share2"}} + Expect(c.Add(ctx, storageID, spaceID, "storageid$spaceid!share2", share2)).To(Succeed()) Expect(space.Etag).ToNot(Equal(old)) }) + + It("keeps the etag stable for an idempotent add", func() { + Expect(c.Add(ctx, storageID, spaceID, shareID, share1)).To(Succeed()) + spaces, ok := c.Providers.Load(storageID) + Expect(ok).To(BeTrue()) + space, ok := spaces.Spaces.Load(spaceID) + Expect(ok).To(BeTrue()) + old := space.Etag + + // Re-adding the same share does not change the content, so the + // content-based etag stays stable (no spurious cache invalidation). + Expect(c.Add(ctx, storageID, spaceID, shareID, share1)).To(Succeed()) + Expect(space.Etag).To(Equal(old)) + }) }) Context("with an existing entry", func() { @@ -172,12 +189,13 @@ var _ = Describe("Cache", func() { }) - Describe("PersistWithTime", func() { + Describe("Persist", func() { It("does not persist if the etag changed", func() { - time.Sleep(1 * time.Nanosecond) path := filepath.Join(tmpdir, "storages/storageid/spaceid.json") - now := time.Now() - _ = os.Chtimes(path, now, now) // this only works for the file backend + // Modify the file content on disk so that its content-based etag + // no longer matches the one held in memory. The subsequent + // persist must then fail its If-Match precondition. + Expect(os.WriteFile(path, []byte(`{"Shares":{}}`), 0644)).To(Succeed()) Expect(c.Persist(ctx, storageID, spaceID)).ToNot(Succeed()) }) }) diff --git a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go index b8e536b0821..7cd98f988fb 100644 --- a/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go +++ b/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go @@ -50,8 +50,9 @@ type Cache struct { ReceivedSpaces mtimesyncedcache.Map[string, *Spaces] - storage metadata.Storage - ttl time.Duration + storage metadata.Storage + lockable metadata.LockingStorage + ttl time.Duration } // Spaces holds the received shares of one user per space @@ -73,11 +74,18 @@ type State struct { Hidden bool } -// New returns a new Cache instance -func New(s metadata.Storage, ttl time.Duration) Cache { +// New returns a new Cache instance. Optional metadata.Options (e.g. a shared +// Locker) may be supplied to control how persisted writes are serialized; when +// the storage is a LockingStorage the default locker is used. +func New(s metadata.Storage, ttl time.Duration, opts ...metadata.Option) Cache { + var lockable metadata.LockingStorage + if ls, ok := s.(metadata.LockingStorage); ok { + lockable = ls + } return Cache{ ReceivedSpaces: mtimesyncedcache.Map[string, *Spaces]{}, storage: s, + lockable: lockable, ttl: ttl, lockMap: sync.Map{}, } @@ -111,9 +119,26 @@ func (c *Cache) Add(ctx context.Context, userID utils.FilenameEncoder, spaceID s defer span.End() span.SetAttributes(attribute.String("cs3.userid.key", userIDKey), attribute.String("cs3.spaceid", spaceID)) - persistFunc := func() error { - c.initializeIfNeeded(userIDKey, spaceID) + log := appctx.GetLogger(ctx).With(). + Str("hostname", os.Getenv("HOSTNAME")). + Str("userIDKey", userIDKey). + Str("spaceID", spaceID).Logger() + var err error + if c.lockable != nil { + err = c.atomicPersist(ctx, userIDKey, func(existing *Spaces) (*Spaces, error) { + if existing.Spaces[spaceID] == nil { + existing.Spaces[spaceID] = &Space{States: map[string]*State{}} + } + existing.Spaces[spaceID].States[rs.Share.Id.GetOpaqueId()] = &State{ + State: rs.State, + MountPoint: rs.MountPoint, + Hidden: rs.Hidden, + } + return existing, nil + }) + } else { + c.initializeIfNeeded(userIDKey, spaceID) rss, _ := c.ReceivedSpaces.Load(userIDKey) receivedSpace := rss.Spaces[spaceID] if receivedSpace.States == nil { @@ -124,48 +149,74 @@ func (c *Cache) Add(ctx context.Context, userID utils.FilenameEncoder, spaceID s MountPoint: rs.MountPoint, Hidden: rs.Hidden, } + err = c.persist(ctx, userID) + } - return c.persist(ctx, userID) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, fmt.Sprintf("persisting added received share failed: %s", err.Error())) + log.Error().Err(err).Msg("persisting added received share failed") + return err } + span.SetStatus(codes.Ok, "") + return nil +} - log := appctx.GetLogger(ctx).With(). - Str("hostname", os.Getenv("HOSTNAME")). - Str("userIDKey", userIDKey). - Str("spaceID", spaceID).Logger() +// atomicPersist performs a locked read-modify-write of the user's received.json. +// fn receives the current on-storage state (a zero Spaces when the file does not +// yet exist) and returns the state to persist; returning nil aborts the write. +// On success the in-memory cache is refreshed in place from the written bytes so +// that the etag and content stay consistent with the storage. +func (c *Cache) atomicPersist(ctx context.Context, userIDKey string, fn func(existing *Spaces) (*Spaces, error)) error { + _, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "atomicPersist") + defer span.End() - 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 + jsonPath := userJSONPath(userIDKey) + var written []byte + res, err := c.lockable.UploadWithLock(ctx, metadata.UploadRequest{Path: jsonPath}, func(existing []byte) ([]byte, error) { + var rss *Spaces + if len(existing) > 0 { + rss = &Spaces{} + if err := json.Unmarshal(existing, rss); err != nil { + return nil, err + } + if rss.Spaces == nil { + rss.Spaces = map[string]*Space{} + } + } else { + rss = &Spaces{Spaces: map[string]*Space{}} + } + newState, err := fn(rss) + if err != nil || newState == nil { + return nil, err + } + b, err := json.Marshal(newState) + if err != nil { + return nil, 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.") + written = b + return b, nil + }) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + return err + } + + // Refresh the in-memory cache in place (mutate the existing *Spaces, do not + // replace it) so callers holding a reference observe the update. + rss, _ := c.ReceivedSpaces.LoadOrStore(userIDKey, &Spaces{Spaces: map[string]*Space{}}) + if len(written) > 0 { + var fresh Spaces + if err := json.Unmarshal(written, &fresh); err != nil { return err } + rss.Spaces = fresh.Spaces } - return err + rss.etag = res.Etag + + span.SetStatus(codes.Ok, "") + return nil } // Get returns one entry from the cache @@ -201,9 +252,24 @@ func (c *Cache) Remove(ctx context.Context, userID utils.FilenameEncoder, spaceI defer span.End() span.SetAttributes(attribute.String("cs3.userid.key", userIDKey), attribute.String("cs3.spaceid", spaceID)) - persistFunc := func() error { - c.initializeIfNeeded(userIDKey, spaceID) + log := appctx.GetLogger(ctx).With(). + Str("hostname", os.Getenv("HOSTNAME")). + Str("userIDKey", userIDKey). + Str("spaceID", spaceID).Logger() + var err error + if c.lockable != nil { + err = c.atomicPersist(ctx, userIDKey, func(existing *Spaces) (*Spaces, error) { + if receivedSpace := existing.Spaces[spaceID]; receivedSpace != nil { + delete(receivedSpace.States, shareID) + if len(receivedSpace.States) == 0 { + delete(existing.Spaces, spaceID) + } + } + return existing, nil + }) + } else { + c.initializeIfNeeded(userIDKey, spaceID) rss, _ := c.ReceivedSpaces.Load(userIDKey) receivedSpace := rss.Spaces[spaceID] if receivedSpace.States == nil { @@ -213,48 +279,17 @@ func (c *Cache) Remove(ctx context.Context, userID utils.FilenameEncoder, spaceI if len(receivedSpace.States) == 0 { delete(rss.Spaces, spaceID) } - - return c.persist(ctx, userID) + err = c.persist(ctx, userID) } - log := appctx.GetLogger(ctx).With(). - Str("hostname", os.Getenv("HOSTNAME")). - Str("userIDKey", userIDKey). - 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, fmt.Sprintf("persisting removed received share failed: %s", err.Error())) + log.Error().Err(err).Msg("persisting removed received share failed") + return err } - return err + span.SetStatus(codes.Ok, "") + return nil } // List returns a list of received shares for a given user diff --git a/pkg/share/manager/jsoncs3/sharecache/sharecache.go b/pkg/share/manager/jsoncs3/sharecache/sharecache.go index c5a812dc163..fbcaae270f7 100644 --- a/pkg/share/manager/jsoncs3/sharecache/sharecache.go +++ b/pkg/share/manager/jsoncs3/sharecache/sharecache.go @@ -52,6 +52,7 @@ type Cache struct { UserShares mtimesyncedcache.Map[string, *UserShareCache] storage metadata.Storage + lockable metadata.LockingStorage namespace string filename string ttl time.Duration @@ -77,11 +78,18 @@ func (c *Cache) lockUser(userID string) func() { return func() { lock.Unlock() } } -// New returns a new Cache instance -func New(s metadata.Storage, namespace, filename string, ttl time.Duration) Cache { +// New returns a new Cache instance. Optional metadata.Options (e.g. a shared +// Locker) may be supplied to control how persisted writes are serialized; when +// the storage is a LockingStorage the default locker is used. +func New(s metadata.Storage, namespace, filename string, ttl time.Duration, opts ...metadata.Option) Cache { + var lockable metadata.LockingStorage + if ls, ok := s.(metadata.LockingStorage); ok { + lockable = ls + } return Cache{ UserShares: mtimesyncedcache.Map[string, *UserShareCache]{}, storage: s, + lockable: lockable, namespace: namespace, filename: filename, ttl: ttl, @@ -113,53 +121,92 @@ func (c *Cache) Add(ctx context.Context, id utils.FilenameEncoder, shareID strin storageid, spaceid, _ := shareid.Decode(shareID) ssid := storageid + shareid.IDDelimiter + spaceid - persistFunc := func() error { - c.initializeIfNeeded(key, ssid) - - // add share id - us, _ := c.UserShares.Load(key) - us.UserShares[ssid].IDs[shareID] = struct{}{} - return c.Persist(ctx, key) - } - log := appctx.GetLogger(ctx).With(). Str("hostname", os.Getenv("HOSTNAME")). Str("userID", key). Str("shareID", shareID).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 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 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 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 share failed. giving up: %s", err.Error())) - log.Error().Err(err).Msg("persisting added share failed") - return err + if c.lockable != nil { + err = c.atomicPersist(ctx, key, func(existing *UserShareCache) (*UserShareCache, error) { + if existing.UserShares[ssid] == nil { + existing.UserShares[ssid] = &SpaceShareIDs{IDs: map[string]struct{}{}} + } + existing.UserShares[ssid].IDs[shareID] = struct{}{} + return existing, nil + }) + } else { + c.initializeIfNeeded(key, ssid) + us, _ := c.UserShares.Load(key) + us.UserShares[ssid].IDs[shareID] = struct{}{} + err = c.Persist(ctx, key) + } + + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, fmt.Sprintf("persisting added share failed: %s", err.Error())) + log.Error().Err(err).Msg("persisting added share failed") + return err + } + span.SetStatus(codes.Ok, "") + return nil +} + +// atomicPersist performs a locked read-modify-write of the user's JSON file. +// fn receives the current on-storage state (a zero UserShareCache when the file +// does not yet exist) and returns the state to persist; returning nil aborts +// the write. On success the in-memory cache is refreshed in place from the +// written bytes so that the etag and content stay consistent with the storage. +func (c *Cache) atomicPersist(ctx context.Context, key string, fn func(existing *UserShareCache) (*UserShareCache, error)) error { + _, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "atomicPersist") + defer span.End() + + jsonPath := c.userCreatedPath(key) + var written []byte + res, err := c.lockable.UploadWithLock(ctx, metadata.UploadRequest{Path: jsonPath}, func(existing []byte) ([]byte, error) { + var us *UserShareCache + if len(existing) > 0 { + us = &UserShareCache{} + if err := json.Unmarshal(existing, us); err != nil { + return nil, err + } + if us.UserShares == nil { + us.UserShares = map[string]*SpaceShareIDs{} + } + } else { + us = &UserShareCache{UserShares: map[string]*SpaceShareIDs{}} + } + newState, err := fn(us) + if err != nil || newState == nil { + return nil, err + } + b, err := json.Marshal(newState) + if err != nil { + return nil, err } - if err := c.syncWithLock(ctx, key); err != nil { - span.RecordError(err) - span.SetStatus(codes.Error, err.Error()) - log.Error().Err(err).Msg("persisting added share failed. giving up.") + written = b + return b, nil + }) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + return err + } + + // Refresh the in-memory cache in place (mutate the existing *UserShareCache, + // do not replace it) so callers holding a reference observe the update. + us, _ := c.UserShares.LoadOrStore(key, &UserShareCache{UserShares: map[string]*SpaceShareIDs{}}) + if len(written) > 0 { + var fresh UserShareCache + if err := json.Unmarshal(written, &fresh); err != nil { return err } + us.UserShares = fresh.UserShares } - return err + us.Etag = res.Etag + + span.SetStatus(codes.Ok, "") + return nil } // Remove removes a share for the given user @@ -185,57 +232,37 @@ func (c *Cache) Remove(ctx context.Context, id utils.FilenameEncoder, shareID st storageid, spaceid, _ := shareid.Decode(shareID) ssid := storageid + shareid.IDDelimiter + spaceid - persistFunc := func() error { - us, loaded := c.UserShares.LoadOrStore(key, &UserShareCache{ - UserShares: map[string]*SpaceShareIDs{}, - }) - - if loaded { - // remove share id - delete(us.UserShares[ssid].IDs, shareID) - } - - return c.Persist(ctx, key) - } - log := appctx.GetLogger(ctx).With(). Str("hostname", os.Getenv("HOSTNAME")). Str("userID", key). Str("shareID", shareID).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 removed 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 removed 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("file already existed when persisting removed 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 removed share failed. giving up: %s", err.Error())) - log.Error().Err(err).Msg("persisting removed share failed") - return err - } - if err := c.syncWithLock(ctx, key); err != nil { - span.RecordError(err) - span.SetStatus(codes.Error, err.Error()) - return err + if c.lockable != nil { + err = c.atomicPersist(ctx, key, func(existing *UserShareCache) (*UserShareCache, error) { + if space := existing.UserShares[ssid]; space != nil { + delete(space.IDs, shareID) + } + return existing, nil + }) + } else { + us, loaded := c.UserShares.LoadOrStore(key, &UserShareCache{ + UserShares: map[string]*SpaceShareIDs{}, + }) + if loaded { + delete(us.UserShares[ssid].IDs, shareID) } + err = c.Persist(ctx, key) } - return err + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, fmt.Sprintf("persisting removed share failed: %s", err.Error())) + log.Error().Err(err).Msg("persisting removed share failed") + return err + } + span.SetStatus(codes.Ok, "") + return nil } // List return the list of spaces/shares for the given user/group From 16a7ae796fb34ef671a8671ab8717ccc81748819 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Friedrich=20Dreyer?= Date: Thu, 10 Sep 2026 23:34:03 +0200 Subject: [PATCH 5/7] feat(jsoncs3): register disk-backed 'json' driver and add Dump Register a new 'json' share-manager driver backed by the disk storage (with an injectable locker) alongside the existing CS3-backed 'jsoncs3' driver. Add a Dump method to the Manager that walks all three caches and exports shares/received-shares, enabling migration from CS3 to disk. Add tests for dump export and concurrent-add atomicity on disk. --- pkg/share/manager/jsoncs3/jsoncs3.go | 154 ++++++++++++++++++++++ pkg/share/manager/jsoncs3/jsoncs3_test.go | 71 ++++++++++ 2 files changed, 225 insertions(+) diff --git a/pkg/share/manager/jsoncs3/jsoncs3.go b/pkg/share/manager/jsoncs3/jsoncs3.go index 492ec38009e..ca0f96ea42f 100644 --- a/pkg/share/manager/jsoncs3/jsoncs3.go +++ b/pkg/share/manager/jsoncs3/jsoncs3.go @@ -21,6 +21,7 @@ package jsoncs3 import ( "context" stderrors "errors" + "fmt" "strings" "sync" "time" @@ -47,6 +48,7 @@ import ( "github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/shareid" "github.com/opencloud-eu/reva/v2/pkg/share/manager/registry" "github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata" // nolint:staticcheck // we need the legacy package to convert V1 to V2 messages + "github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata/locks" "github.com/opencloud-eu/reva/v2/pkg/storagespace" "github.com/opencloud-eu/reva/v2/pkg/utils" "github.com/pkg/errors" @@ -116,6 +118,7 @@ const tracerName = "jsoncs3" func init() { registry.Register("jsoncs3", NewDefault) + registry.Register("json", NewDisk) } var ( @@ -214,6 +217,54 @@ func NewDefault(m map[string]interface{}, logger *zerolog.Logger) (share.Manager return mgr, nil } +// diskConfig holds the configuration for the disk-backed "json" driver. Unlike +// the CS3-backed "jsoncs3" driver it stores shares in a local directory and does +// not require a gateway or storage provider address. +type diskConfig struct { + DataDir string `mapstructure:"data_dir"` + LockBackend string `mapstructure:"lock_backend"` + CacheTTL int `mapstructure:"ttl"` +} + +// NewDisk returns a new manager instance backed by a local disk storage. It is +// registered as the "json" driver and is intended for single-node deployments or +// for loading data that was dumped from a CS3-backed manager (see Dump/Load). +func NewDisk(c map[string]interface{}, logger *zerolog.Logger) (share.Manager, error) { + conf := &diskConfig{} + if err := mapstructure.Decode(c, conf); err != nil { + return nil, errors.Wrap(err, "error creating a new disk share manager") + } + + if conf.DataDir == "" { + conf.DataDir = "/var/tmp/reva/sharemanager" + } + + var opts []metadata.Option + switch conf.LockBackend { + case "", "disk": + // default: cross-process flock-based locking rooted at the data dir + case "memory": + opts = append(opts, metadata.WithLocker(locks.NewMemoryLocker())) + default: + return nil, fmt.Errorf("unknown lock_backend %q (want \"disk\" or \"memory\")", conf.LockBackend) + } + + s, err := metadata.NewDiskStorage(conf.DataDir, opts...) + if err != nil { + return nil, err + } + + mgr, err := New(s, logger, nil, conf.CacheTTL, nil, 0) + if err != nil { + return nil, err + } + // The disk driver has no gateway to migrate from; a fresh deployment marks + // all migrations as applied during initialize(). Skip the migration run so + // that write operations are not held waiting on it. + mgr.SkipMigrations() + return mgr, nil +} + // New returns a new manager instance. func New(s metadata.Storage, logger *zerolog.Logger, @@ -1379,6 +1430,109 @@ func (m *Manager) Load(ctx context.Context, shareChan <-chan *collaboration.Shar return nil } +// Dump exports all shares and received shares to the given channels. It is the +// inverse of Load and is used to migrate data between backends (e.g. from a +// CS3-backed manager to a disk-backed one). +func (m *Manager) Dump(ctx context.Context, shareChan chan<- *collaboration.Share, receivedShareChan chan<- share.ReceivedShareWithUser) error { + if err := m.waitForInit(ctx); err != nil { + return err + } + + var wg sync.WaitGroup + wg.Add(2) + + // Export all shares from the provider cache. + go func() { + defer wg.Done() + providers, err := m.Cache.All(ctx) + if err != nil { + m.logger.Error().Err(err).Msg("error listing providers during dump") + return + } + providers.Range(func(storageID string, spaces *providercache.Spaces) bool { + spaces.Spaces.Range(func(spaceID string, shares *providercache.Shares) bool { + for _, s := range shares.Shares { + shareChan <- s + } + return true + }) + return true + }) + }() + + // Export all received share states from the user and group caches. We + // enumerate users/groups from the storage tree (the keys are the safe + // filenames, which for non-guest ids equal the opaque id). + go func() { + defer wg.Done() + + userIDs, err := m.listUserKeys(ctx, "users") + if err != nil { + m.logger.Error().Err(err).Msg("error listing users during dump") + } + for _, userID := range userIDs { + states, err := m.UserReceivedStates.List(ctx, utils.NewFSSafeUserID(&userv1beta1.UserId{OpaqueId: userID})) + if err != nil { + m.logger.Error().Err(err).Msg("error listing received states during dump") + continue + } + for _, space := range states { + for shareID := range space.States { + receivedShareChan <- share.ReceivedShareWithUser{ + UserID: &userv1beta1.UserId{OpaqueId: userID}, + ReceivedShare: &collaboration.ReceivedShare{ + Share: &collaboration.Share{ + Id: &collaboration.ShareId{OpaqueId: shareID}, + }, + }, + } + } + } + } + + groupIDs, err := m.listUserKeys(ctx, "groups") + if err != nil { + m.logger.Error().Err(err).Msg("error listing groups during dump") + } + for _, groupID := range groupIDs { + states, err := m.GroupReceivedCache.List(ctx, utils.FSSafeGroupID{ID: &grouppb.GroupId{OpaqueId: groupID}}) + if err != nil { + m.logger.Error().Err(err).Msg("error listing group received states during dump") + continue + } + for _, space := range states { + for shareID := range space.IDs { + receivedShareChan <- share.ReceivedShareWithUser{ + UserID: nil, + ReceivedShare: &collaboration.ReceivedShare{ + Share: &collaboration.Share{ + Id: &collaboration.ShareId{OpaqueId: shareID}, + }, + }, + } + } + } + } + }() + + wg.Wait() + return nil +} + +// listUserKeys returns the directory entries (user/group ids) stored under the +// given namespace in the manager's metadata storage. +func (m *Manager) listUserKeys(ctx context.Context, namespace string) ([]string, error) { + entries, err := m.storage.ListDir(ctx, "/"+namespace) + if err != nil { + return nil, err + } + keys := make([]string, 0, len(entries)) + for _, e := range entries { + keys = append(keys, e.Name) + } + return keys, nil +} + func (m *Manager) purgeSpace(ctx context.Context, id *provider.StorageSpaceId) { log := appctx.GetLogger(ctx) storageID, spaceID := storagespace.SplitStorageID(id.OpaqueId) diff --git a/pkg/share/manager/jsoncs3/jsoncs3_test.go b/pkg/share/manager/jsoncs3/jsoncs3_test.go index fbb4563a496..70e2f637fe7 100644 --- a/pkg/share/manager/jsoncs3/jsoncs3_test.go +++ b/pkg/share/manager/jsoncs3/jsoncs3_test.go @@ -21,6 +21,7 @@ package jsoncs3_test import ( "context" "encoding/json" + "fmt" "os" "path/filepath" "sync" @@ -229,6 +230,76 @@ var _ = Describe("Jsoncs3", func() { }) }) + Describe("Dump", func() { + It("exports all shares from the provider cache", func() { + // Create a share on the source manager. + _, err := m.Share(ctx, sharedResource, grant) + Expect(err).ToNot(HaveOccurred()) + + sharesChan := make(chan *collaboration.Share, 16) + receivedChan := make(chan sharespkg.ReceivedShareWithUser, 16) + + dumpWG := sync.WaitGroup{} + dumpWG.Add(1) + go func() { + defer dumpWG.Done() + Expect(m.Dump(ctx, sharesChan, receivedChan)).To(Succeed()) + }() + dumpWG.Wait() + close(sharesChan) + close(receivedChan) + + var dumped []*collaboration.Share + for s := range sharesChan { + dumped = append(dumped, s) + } + Expect(len(dumped)).To(BeNumerically(">", 0)) + + // The created share must be among the dumped shares. + found := false + for _, s := range dumped { + if s.GetResourceId().GetOpaqueId() == sharedResource.Id.OpaqueId { + found = true + break + } + } + Expect(found).To(BeTrue()) + }) + }) + + Describe("concurrent adds", func() { + It("does not lose updates when many shares target the same space", func() { + const n = 20 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + i := i + go func() { + defer wg.Done() + // Each goroutine shares a distinct resource in the same + // space, exercising concurrent read-modify-write of the + // per-space JSON file. + res := &providerv1beta1.ResourceInfo{ + Id: &providerv1beta1.ResourceId{ + StorageId: "storageid", + SpaceId: "spaceid", + OpaqueId: fmt.Sprintf("opaque%d", i), + }, + } + if _, err := m.Share(ctx, res, grant); err != nil { + Fail(fmt.Sprintf("share %d failed: %v", i, err)) + } + }() + } + wg.Wait() + + // All n shares must be present in the space. + shares, err := m.Cache.ListSpace(ctx, "storageid", "spaceid") + Expect(err).ToNot(HaveOccurred()) + Expect(len(shares.Shares)).To(Equal(n)) + }) + }) + Describe("Share", func() { It("fails if the share already exists", func() { _, err := m.Share(ctx, sharedResource, grant) From ec14e9f5d246267d23333dc14b47adacf1728038 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Friedrich=20Dreyer?= Date: Thu, 10 Sep 2026 23:34:08 +0200 Subject: [PATCH 6/7] fix(publicshare): make share writes atomic to fix lost updates Add an Update method to the Persistence interface performing a single atomic read-modify-write. The file backend uses a DiskLocker sidecar and renameio; the CS3 backend uses LockingStorage.UploadWithLock on publicshares.json; the memory backend uses an in-process mutex. Replace the whole-file Read->mutate->Write in CreatePublicShare, UpdatePublicShare and revokePublicShare with persistence.Update, eliminating the lost-update race. Add a lock_backend config option and a concurrent create/revoke test. --- pkg/publicshare/manager/json/json.go | 93 +++++++++---------- pkg/publicshare/manager/json/json_test.go | 32 +++++++ .../manager/json/persistence/cs3/cs3.go | 52 ++++++++++- .../manager/json/persistence/file/file.go | 52 ++++++++++- .../manager/json/persistence/memory/memory.go | 20 ++++ .../manager/json/persistence/persistence.go | 7 ++ 6 files changed, 203 insertions(+), 53 deletions(-) diff --git a/pkg/publicshare/manager/json/json.go b/pkg/publicshare/manager/json/json.go index 721f86a5b72..1e0d11b6fbe 100644 --- a/pkg/publicshare/manager/json/json.go +++ b/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 diff --git a/pkg/publicshare/manager/json/json_test.go b/pkg/publicshare/manager/json/json_test.go index 24fdc24741c..fe2f8b33ddc 100644 --- a/pkg/publicshare/manager/json/json_test.go +++ b/pkg/publicshare/manager/json/json_test.go @@ -20,6 +20,7 @@ package json_test import ( "context" + "fmt" "os" "path/filepath" "sync" @@ -210,6 +211,37 @@ var _ = Describe("Json", func() { Expect(err).ToNot(HaveOccurred()) Expect(ps).ToNot(BeNil()) }) + + It("does not lose updates when creating and revoking concurrently", func() { + const n = 30 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + i := i + go func() { + defer wg.Done() + ps, err := m.CreatePublicShare(ctx, user1, sharedResource, grant) + if err != nil { + Fail(fmt.Sprintf("create %d failed: %v", i, err)) + } + // Revoke every other share to exercise concurrent deletes. + if i%2 == 0 { + ref := &link.PublicShareReference{ + Spec: &link.PublicShareReference_Id{Id: ps.Id}, + } + if err := m.RevokePublicShare(ctx, user1, ref); err != nil { + Fail(fmt.Sprintf("revoke %d failed: %v", i, err)) + } + } + }() + } + wg.Wait() + + // Exactly the odd-indexed shares should remain. + ps, err := m.ListPublicShares(ctx, user1, []*link.ListPublicSharesRequest_Filter{}, false) + Expect(err).ToNot(HaveOccurred()) + Expect(len(ps)).To(Equal(n / 2)) + }) }) Describe("PublicShares", func() { diff --git a/pkg/publicshare/manager/json/persistence/cs3/cs3.go b/pkg/publicshare/manager/json/persistence/cs3/cs3.go index 7e4905d56ec..a43c5e8d97d 100644 --- a/pkg/publicshare/manager/json/persistence/cs3/cs3.go +++ b/pkg/publicshare/manager/json/persistence/cs3/cs3.go @@ -38,14 +38,22 @@ type db struct { type cs3 struct { initialized bool s metadata.Storage + lockable metadata.LockingStorage db db } -// New returns a new Cache instance +// New returns a new Cache instance. When the storage is a LockingStorage it is +// used to serialise read-modify-write cycles across replicas; otherwise Update +// falls back to an unlocked read-modify-write. func New(s metadata.Storage) persistence.Persistence { + var lockable metadata.LockingStorage + if ls, ok := s.(metadata.LockingStorage); ok { + lockable = ls + } return &cs3{ - s: s, + s: s, + lockable: lockable, db: db{ publicShares: persistence.PublicShares{}, }, @@ -109,3 +117,43 @@ func (p *cs3) Write(ctx context.Context, db persistence.PublicShares) error { }) return err } + +// Update atomically reads the current publicshares.json, applies fn, and writes +// the result back while holding a lock on the file. When the storage supports +// locking (LockingStorage) this is fully atomic across replicas; otherwise it +// degrades to an unlocked read-modify-write. +func (p *cs3) Update(ctx context.Context, fn func(current persistence.PublicShares) (persistence.PublicShares, error)) error { + if !p.initialized { + return fmt.Errorf("not initialized") + } + + const path = "publicshares.json" + + if p.lockable != nil { + _, err := p.lockable.UploadWithLock(ctx, metadata.UploadRequest{Path: path}, func(existing []byte) ([]byte, error) { + current := persistence.PublicShares{} + if len(existing) > 0 { + if err := json.Unmarshal(existing, ¤t); err != nil { + return nil, err + } + } + next, err := fn(current) + if err != nil { + return nil, err + } + return json.Marshal(next) + }) + return err + } + + // Fallback: no lock support, plain read-modify-write. + current, err := p.Read(ctx) + if err != nil { + return err + } + next, err := fn(current) + if err != nil { + return err + } + return p.Write(ctx, next) +} diff --git a/pkg/publicshare/manager/json/persistence/file/file.go b/pkg/publicshare/manager/json/persistence/file/file.go index 99c1c2c7a0c..3ede81ccbbb 100644 --- a/pkg/publicshare/manager/json/persistence/file/file.go +++ b/pkg/publicshare/manager/json/persistence/file/file.go @@ -26,20 +26,26 @@ import ( "path/filepath" "sync" + "github.com/google/renameio/v2" "github.com/opencloud-eu/reva/v2/pkg/publicshare/manager/json/persistence" + "github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata/locks" ) type file struct { path string initialized bool lock *sync.RWMutex + locker locks.Locker } -// New returns a new Cache instance +// New returns a new Cache instance. It uses a disk-based locker rooted at the +// directory containing the db file so that concurrent processes serialise their +// read-modify-write cycles. func New(path string) persistence.Persistence { return &file{ - path: path, - lock: &sync.RWMutex{}, + path: path, + lock: &sync.RWMutex{}, + locker: locks.NewDiskLocker(filepath.Dir(path)), } } @@ -101,7 +107,45 @@ func (p *file) Write(_ context.Context, db persistence.PublicShares) error { return err } - return os.WriteFile(p.path, dbAsJSON, 0644) + return renameio.WriteFile(p.path, dbAsJSON, 0644) +} + +// Update atomically reads the current db, applies fn, and writes the result +// back while holding a cross-process lock on the db file. This prevents lost +// updates when multiple replicas create/update/revoke shares concurrently. +func (p *file) Update(ctx context.Context, fn func(current persistence.PublicShares) (persistence.PublicShares, error)) error { + if !p.isInitialized() { + return fmt.Errorf("not initialized") + } + + unlock, err := p.locker.Lock(ctx, filepath.Base(p.path)) + if err != nil { + return err + } + defer unlock() + + readBytes, err := os.ReadFile(p.path) + if err != nil { + return err + } + current := persistence.PublicShares{} + if len(readBytes) > 0 { + if err := json.Unmarshal(readBytes, ¤t); err != nil { + return err + } + } + + next, err := fn(current) + if err != nil { + return err + } + + dbAsJSON, err := json.Marshal(next) + if err != nil { + return err + } + + return renameio.WriteFile(p.path, dbAsJSON, 0644) } func (p *file) isInitialized() bool { diff --git a/pkg/publicshare/manager/json/persistence/memory/memory.go b/pkg/publicshare/manager/json/persistence/memory/memory.go index 0c1c52fd6bd..027dbdc1d4a 100644 --- a/pkg/publicshare/manager/json/persistence/memory/memory.go +++ b/pkg/publicshare/manager/json/persistence/memory/memory.go @@ -21,11 +21,13 @@ package memory import ( "context" "fmt" + "sync" "github.com/opencloud-eu/reva/v2/pkg/publicshare/manager/json/persistence" ) type memory struct { + mu sync.Mutex db map[string]interface{} } @@ -50,6 +52,24 @@ func (p *memory) Write(_ context.Context, db persistence.PublicShares) error { if p.db == nil { return fmt.Errorf("not initialized") } + p.mu.Lock() + defer p.mu.Unlock() p.db = db return nil } + +// Update applies fn to the in-memory db while holding the lock. +func (p *memory) Update(_ context.Context, fn func(current persistence.PublicShares) (persistence.PublicShares, error)) error { + if p.db == nil { + return fmt.Errorf("not initialized") + } + p.mu.Lock() + defer p.mu.Unlock() + + next, err := fn(p.db) + if err != nil { + return err + } + p.db = next + return nil +} diff --git a/pkg/publicshare/manager/json/persistence/persistence.go b/pkg/publicshare/manager/json/persistence/persistence.go index 543c1e6becf..1fea52328e6 100644 --- a/pkg/publicshare/manager/json/persistence/persistence.go +++ b/pkg/publicshare/manager/json/persistence/persistence.go @@ -28,4 +28,11 @@ type Persistence interface { Init(context.Context) error Read(context.Context) (PublicShares, error) Write(context.Context, PublicShares) error + + // Update performs an atomic read-modify-write of the whole share database. + // fn receives the current state and returns the new state to persist; it is + // invoked while holding the persistence lock, so no other writer can + // interleave between the read and the write. This makes the operation safe + // against concurrent create/update/revoke without lost updates. + Update(context.Context, func(current PublicShares) (PublicShares, error)) error } From b8105b6d27aedfa64f5d77c29fbdae3a5f6cc303 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Friedrich=20Dreyer?= Date: Thu, 10 Sep 2026 23:34:13 +0200 Subject: [PATCH 7/7] feat(appauth): use atomic locked updates in metadatacache store Add an optional LockingStorage to metadatacache.Options. When set, Store.Update performs a single atomic UploadWithLock instead of the etag-based CAS retry loop; otherwise the existing CAS path is unchanged. Wire the CS3 storage as a LockingStorage in the appauth jsoncs3 manager and add a lock_backend config option. Add tests covering the locking path including concurrent same-key updates with no lost updates. --- pkg/appauth/manager/jsoncs3/jsoncs3.go | 26 ++++++-- pkg/metadatacache/store.go | 54 +++++++++++++++++ pkg/metadatacache/store_test.go | 84 ++++++++++++++++++++++++++ 3 files changed, 159 insertions(+), 5 deletions(-) diff --git a/pkg/appauth/manager/jsoncs3/jsoncs3.go b/pkg/appauth/manager/jsoncs3/jsoncs3.go index 98244e51829..462e58664b4 100644 --- a/pkg/appauth/manager/jsoncs3/jsoncs3.go +++ b/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{ diff --git a/pkg/metadatacache/store.go b/pkg/metadatacache/store.go index c10bc14fe7a..8539b761ed7 100644 --- a/pkg/metadatacache/store.go +++ b/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 +} diff --git a/pkg/metadatacache/store_test.go b/pkg/metadatacache/store_test.go index c4807e64de2..b5f1e3650c8 100644 --- a/pkg/metadatacache/store_test.go +++ b/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)) + }) + }) })