fix: Invalidate the cache after write. Moved the db copy from the persist label to the manager.

Author: 2403905

pkg/publicshare/manager/json/json.go

@@ -123,8 +123,6 @@ func New(gwAddr string, pwHashCost, janitorRunInterval int, enableCleanup bool,
 	return m, nil
 }
 
-var _ publicshare.ClosableManager = (*manager)(nil)
-
 type commonConfig struct {
 	GatewayAddr                string `mapstructure:"gateway_addr"`
 	SharePasswordHashCost      int    `mapstructure:"password_hash_cost"`
@@ -152,7 +150,7 @@ func (c *commonConfig) init() {
 		c.SharePasswordHashCost = 11
 	}
 	if c.JanitorRunInterval == 0 {
-		c.JanitorRunInterval = 600
+		c.JanitorRunInterval = 3600 // 1 hour
 	}
 }
 
@@ -174,6 +172,8 @@ type manager struct {
 	janitorDone   chan struct{}
 }
 
+var _ publicshare.ClosableManager = (*manager)(nil)
+
 // init is called at the top of every public method to lazily initialize the
 // persistence layer. It must not take m.mutex: persistence.Init is already
 // idempotent and self-synchronized (it returns immediately once the
@@ -262,19 +262,20 @@ func (m *manager) Load(ctx context.Context, shareChan <-chan *publicshare.WithPa
 	if err != nil {
 		return err
 	}
+	dbCopy := persistence.Copy(db)
 
 	for ps := range shareChan {
 		encShare, err := utils.MarshalProtoV1ToJSON(&ps.PublicShare)
 		if err != nil {
 			return err
 		}
 
-		db[ps.PublicShare.Id.GetOpaqueId()] = map[string]interface{}{
+		dbCopy[ps.PublicShare.Id.GetOpaqueId()] = map[string]interface{}{
 			"share":    string(encShare),
 			"password": ps.Password,
 		}
 	}
-	return m.persistence.Write(ctx, db)
+	return m.persistence.Write(ctx, dbCopy)
 }
 
 // CreatePublicShare adds a new entry to manager.shares
@@ -345,17 +346,18 @@ func (m *manager) CreatePublicShare(ctx context.Context, u *user.User, rInfo *pr
 	if err != nil {
 		return nil, err
 	}
+	dbCopy := persistence.Copy(db)
 
-	if _, ok := db[s.Id.GetOpaqueId()]; !ok {
-		db[s.Id.GetOpaqueId()] = map[string]interface{}{
+	if _, ok := dbCopy[s.Id.GetOpaqueId()]; !ok {
+		dbCopy[s.Id.GetOpaqueId()] = map[string]interface{}{
 			"share":    string(encShare),
 			"password": ps.Password,
 		}
 	} else {
 		return nil, errors.New("key already exists")
 	}
 
-	err = m.persistence.Write(ctx, db)
+	err = m.persistence.Write(ctx, dbCopy)
 	if err != nil {
 		return nil, err
 	}
@@ -433,13 +435,14 @@ func (m *manager) UpdatePublicShare(ctx context.Context, u *user.User, req *link
 	if err != nil {
 		return nil, err
 	}
+	dbCopy := persistence.Copy(db)
 
 	encShare, err := utils.MarshalProtoV1ToJSON(share)
 	if err != nil {
 		return nil, err
 	}
 
-	data, ok := db[share.Id.OpaqueId].(map[string]interface{})
+	data, ok := dbCopy[share.Id.OpaqueId].(map[string]interface{})
 	if !ok {
 		data = map[string]interface{}{}
 	}
@@ -449,9 +452,9 @@ func (m *manager) UpdatePublicShare(ctx context.Context, u *user.User, req *link
 	}
 	data["share"] = string(encShare)
 
-	db[share.Id.OpaqueId] = data
+	dbCopy[share.Id.OpaqueId] = data
 
-	err = m.persistence.Write(ctx, db)
+	err = m.persistence.Write(ctx, dbCopy)
 	if err != nil {
 		return nil, err
 	}
@@ -522,10 +525,9 @@ func (m *manager) ListPublicShares(ctx context.Context, u *user.User, filters []
 
 	m.mutex.RLock()
 
-	// Read returns a copy that shares no mutable state with the persistence
-	// backend (see persistence.Copy), so it's safe to keep using db after
-	// the lock is released below - a concurrent writer can no longer race
-	// what we do with it.
+	// Ranging over db below happens after the lock is released, which is safe
+	// because we never mutate it: writers copy before they mutate, and the
+	// persistence layer publishes a new map instead of changing this one.
 	db, err := m.persistence.Read(ctx)
 	if err != nil {
 		m.mutex.RUnlock()
@@ -627,10 +629,11 @@ func (m *manager) cleanupExpiredShares() error {
 		return err
 	}
 
-	db, err := m.persistence.Read(ctx)
+	read, err := m.persistence.Read(ctx)
 	if err != nil {
 		return err
 	}
+	db := persistence.Copy(read)
 
 	var changed bool
 	for id, v := range db {
@@ -642,9 +645,6 @@ func (m *manager) cleanupExpiredShares() error {
 		}
 
 		if publicshare.IsExpired(&ps) {
-			// db is our own copy (see persistence.Copy), so deleting the
-			// current entry while ranging over it is safe: single goroutine,
-			// no aliasing with the persistence backend's internal state.
 			delete(db, id)
 			changed = true
 		}
@@ -671,10 +671,11 @@ 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)
+	read, err := m.persistence.Read(ctx)
 	if err != nil {
 		return err
 	}
+	db := persistence.Copy(read)
 
 	switch {
 	case ref.GetId() != nil && ref.GetId().OpaqueId != "":

pkg/publicshare/manager/json/json_test.go

@@ -43,26 +43,67 @@ import (
 	. "github.com/onsi/gomega"
 )
 
-// slowReadPersistence is a fake persistence.Persistence whose Read blocks
-// for a fixed delay, standing in for a network round trip such as the cs3
-// persistence layer's Stat/SimpleDownload against metadata.CS3. It has no
-// state of its own to protect - it exists to let a test observe whether
-// concurrent manager calls overlap during that delay.
-type slowReadPersistence struct {
-	delay time.Duration
+// barrierPersistence is a fake persistence.Persistence standing in for one
+// whose Read is a network round trip, such as the cs3 layer's Stat and
+// SimpleDownload against metadata.CS3. Its Read parks until the expected number
+// of callers are inside it at once and records how many ever were, which turns
+// "do these calls overlap?" into a counter the test can assert on rather than
+// an elapsed time it has to interpret. A loaded CI runner slows every caller
+// equally and so cannot change the answer.
+type barrierPersistence struct {
+	expected int
+
+	mu            sync.Mutex
+	concurrent    int
+	maxConcurrent int
+
+	releaseOnce sync.Once
+	release     chan struct{}
 }
 
-func (p *slowReadPersistence) Init(_ context.Context) error { return nil }
+func newBarrierPersistence(expected int) *barrierPersistence {
+	return &barrierPersistence{expected: expected, release: make(chan struct{})}
+}
+
+func (p *barrierPersistence) Init(_ context.Context) error { return nil }
+
+func (p *barrierPersistence) Read(_ context.Context) (persistence.PublicShares, error) {
+	p.mu.Lock()
+	p.concurrent++
+	if p.concurrent > p.maxConcurrent {
+		p.maxConcurrent = p.concurrent
+	}
+	if p.concurrent == p.expected {
+		p.releaseOnce.Do(func() { close(p.release) })
+	}
+	p.mu.Unlock()
+
+	select {
+	case <-p.release:
+	case <-time.After(10 * time.Second):
+		// Nobody joined us, so the callers are being serialized somewhere above.
+		// Open the gate for good, or every queued caller would wait out its own
+		// timeout and the test would take expected*timeout to report it.
+		p.releaseOnce.Do(func() { close(p.release) })
+	}
+
+	p.mu.Lock()
+	p.concurrent--
+	p.mu.Unlock()
 
-func (p *slowReadPersistence) Read(_ context.Context) (persistence.PublicShares, error) {
-	time.Sleep(p.delay)
 	return persistence.PublicShares{}, nil
 }
 
-func (p *slowReadPersistence) Write(_ context.Context, _ persistence.PublicShares) error {
+func (p *barrierPersistence) Write(_ context.Context, _ persistence.PublicShares) error {
 	return nil
 }
 
+func (p *barrierPersistence) peakConcurrency() int {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.maxConcurrent
+}
+
 var _ = Describe("Json", func() {
 	var (
 		user1 = &userpb.User{
@@ -360,17 +401,15 @@ var _ = Describe("Json", func() {
 				// whichever call was already inside persistence.Read - silently
 				// undoing the switch from sync.Mutex to sync.RWMutex. A wrong
 				// re-introduction of that lock wouldn't fail -race (it's a
-				// correctly-used lock), only show up as this test timing out.
-				const (
-					delay       = 150 * time.Millisecond
-					concurrency = 8
-				)
+				// correctly-used lock), only show up here as the reads no longer
+				// managing to be inside persistence.Read at the same time.
+				const concurrency = 8
 
-				slow, err := json.New("https://localhost:9200", 11, 60, false, &slowReadPersistence{delay: delay})
+				barrier := newBarrierPersistence(concurrency)
+				slow, err := json.New("https://localhost:9200", 11, 60, false, barrier)
 				Expect(err).ToNot(HaveOccurred())
 
 				var wg sync.WaitGroup
-				start := time.Now()
 				wg.Add(concurrency)
 				for range concurrency {
 					go func() {
@@ -380,11 +419,9 @@ var _ = Describe("Json", func() {
 				}
 				wg.Wait()
 
-				// Fully serialized would take concurrency*delay (1.2s here).
-				// Overlapping reads should finish close to a single delay - allow
-				// generous slack for scheduling noise without letting a real
-				// regression pass.
-				Expect(time.Since(start)).To(BeNumerically("<", delay*3))
+				Expect(barrier.peakConcurrency()).To(Equal(concurrency),
+					"only %d of %d ListPublicShares calls were ever inside persistence.Read together, so they are being serialized",
+					barrier.peakConcurrency(), concurrency)
 			})
 
 			It("refreshes its cache before writing new data", func() {

pkg/publicshare/manager/json/persistence/cs3/cs3.go

@@ -101,7 +101,7 @@ func (p *cs3) Read(ctx context.Context) (persistence.PublicShares, error) {
 	info, err := p.s.Stat(ctx, "publicshares.json")
 	if err != nil {
 		if _, ok := err.(errtypes.NotFound); ok {
-			return persistence.Copy(p.db.publicShares), nil // Nothing to sync against
+			return p.db.publicShares, nil
 		}
 		return nil, err
 	}
@@ -117,7 +117,7 @@ func (p *cs3) Read(ctx context.Context) (persistence.PublicShares, error) {
 		}
 		p.db.mtime = utils.TSToTime(info.Mtime)
 	}
-	return persistence.Copy(p.db.publicShares), nil
+	return p.db.publicShares, nil
 }
 
 func (p *cs3) Write(ctx context.Context, db persistence.PublicShares) error {
@@ -142,17 +142,9 @@ func (p *cs3) Write(ctx context.Context, db persistence.PublicShares) error {
 		return err
 	}
 
-	// Keep the cache in sync with what was just persisted. This used to
-	// happen implicitly, because Read() handed out a reference to
-	// p.db.publicShares itself and callers mutated it in place before
-	// calling Write() with that same map. Now that Read() returns an
-	// independent copy (see persistence.Copy), it has to be done explicitly
-	// here, or the cache would only pick up our own write once some later
-	// external write advances the remote mtime past our stale one.
-	if info, statErr := p.s.Stat(ctx, "publicshares.json"); statErr == nil {
-		p.db.mtime = utils.TSToTime(info.Mtime)
-	}
-	p.db.publicShares = persistence.Copy(db)
+	// Invalidate the cache to minimize the risk of inconsistency.
+	p.db.mtime = time.Time{}
+	p.db.publicShares = persistence.PublicShares{}
 
 	return nil
 }