Fix race and a small issue with the access cache

2026-10-08 21:05:21 +00:00 · 2026-06-03 21:01:52 -04:00
parent d19617bb6b
commit b4cb1bb7fb
3 changed files with 141 additions and 8 deletions
@@ -23,15 +23,14 @@ import (
 type accessCache struct {
 	exact   map[string]map[string]aclEntry
 	pattern map[string][]aclEntry
-	seq     uint64       // Bumped on every apply; lets a slow full reload notice that a mutation raced its table scan
+	seq     uint64       // Bumped on every reload; lets a full reload detect a per-user reload that raced its scan
 	mu      sync.RWMutex // Protect exact, pattern, and seq
 }

-// maxFullReloadRetries bounds how many extra times a full reload re-scans the
-// table when a per-user mutation keeps landing mid-scan before it gives up for
-// this cycle (the local mutation already kept the cache correct; only external
-// writes are deferred to the next cycle).
-const maxFullReloadRetries = 2
+// testHookReloadScanned, if non-nil, is invoked by Reload after the DB scan but
+// before the result is applied. Tests use it to inject a concurrent mutation
+// into the full-reload race window; it is always nil in production.
+var testHookReloadScanned func()

 // aclEntry mirrors one user_access row. length feeds better()'s "longer
 // pattern wins" tie-break; the stored topic/pattern string itself is not kept
@@ -83,12 +82,20 @@ func (c *accessCache) Lookup(username, topic string) (read, write, found bool) {
 // listed users' slices are touched (a username absent from the result drops
 // them from both maps). Runs against the primary so a reload after a
 // mutation sees the just-written rows.
+//
+// Since Reload can be triggered from different places and for different scopes (full
+// and user-specific), the function may cause races and lost-updates. This is solved
+// with the sequence number.
 func (c *accessCache) Reload(d *db.DB, query string, usernames ...string) error {
 	started := time.Now()
 	scope := "full"
 	if len(usernames) > 0 {
 		scope = "users=" + strings.Join(usernames, ",")
 	}
+	// Read the sequence number before the SQL query so we can detect races later
+	c.mu.RLock()
+	seqBefore := c.seq
+	c.mu.RUnlock()
 	args := make([]any, len(usernames))
 	for i, u := range usernames {
 		args[i] = u
@@ -125,9 +132,20 @@ func (c *accessCache) Reload(d *db.DB, query string, usernames ...string) error
 	if err := rows.Err(); err != nil {
 		return err
 	}
+	if testHookReloadScanned != nil {
+		testHookReloadScanned()
+	}
 	// Replace or update the internal maps
 	c.mu.Lock()
 	if len(usernames) == 0 {
+		if c.seq != seqBefore {
+			c.mu.Unlock()
+			log.Tag(tag).
+				Field("reload_scope", scope).
+				Field("duration_ms", time.Since(started).Milliseconds()).
+				Warn("ACL cache reload skipped due to race")
+			return nil
+		}
 		c.exact = exacts
 		c.pattern = patterns
 	} else {
@@ -144,12 +162,13 @@ func (c *accessCache) Reload(d *db.DB, query string, usernames ...string) error
 			}
 		}
 	}
+	c.seq++
 	c.mu.Unlock()
 	log.Tag(tag).
 		Field("reload_scope", scope).
 		Field("updated_entries", updatedEntries).
 		Field("duration_ms", time.Since(started).Milliseconds()).
-		Debug("Reloaded ACL cache")
+		Debug("ACL cache reloaded")
 	return nil
 }

@@ -987,7 +987,7 @@ func (a *Manager) ReservationOwner(topic string) (string, error) {
 // It returns the list of topics whose reservations were removed. The read and removal are
 // performed atomically in a single transaction to avoid issues with stale replica data.
 func (a *Manager) RemoveExcessReservations(username string, limit int64) ([]string, error) {
-	return db.QueryTx(a.db, func(tx *sql.Tx) ([]string, error) {
+	removedTopics, err := db.QueryTx(a.db, func(tx *sql.Tx) ([]string, error) {
 		reservations, err := a.reservationsTx(tx, username)
 		if err != nil {
 			return nil, err
@@ -1005,6 +1005,17 @@ func (a *Manager) RemoveExcessReservations(username string, limit int64) ([]stri
 		}
 		return removedTopics, nil
 	})
+	if err != nil {
+		return nil, err
+	}
+	if len(removedTopics) > 0 {
+		// removeReservationAccessTx deletes rows owned by this user and the
+		// matching Everyone rows, so we refresh the access cache.
+		if err := a.maybeReloadAccessCache(username, Everyone); err != nil {
+			return nil, err
+		}
+	}
+	return removedTopics, nil
 }

 // otherAccessCount returns the number of access entries for the given topic that are not owned by the user
@@ -2311,6 +2311,109 @@ func TestAccessCacheReloadInterval_PicksUpExternalWrite(t *testing.T) {
 	})
 }

+// TestAccessCache_RemoveExcessReservationsInvalidatesCache models finding #1:
+// RemoveExcessReservations deletes user_access rows but must also refresh the
+// in-memory cache. Otherwise the owner keeps cached read/write access to a
+// reservation that was removed (e.g. on a tier downgrade) until the next
+// periodic reload -- and if another user re-reserves the freed topic in the
+// meantime, the former owner can read/write the new owner's reserved topic.
+func TestAccessCache_RemoveExcessReservationsInvalidatesCache(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, newManager newManagerFunc) {
+		// A deliberately long reload interval ensures the background poller
+		// cannot mask a missing synchronous invalidation: the mutation itself
+		// must refresh the cache.
+		a := newTestManagerFromConfig(t, newManager, &Config{
+			DefaultAccess:             PermissionDenyAll,
+			BcryptCost:                bcrypt.MinCost,
+			AccessCacheEnabled:        true,
+			AccessCacheReloadInterval: time.Hour,
+		})
+		require.Nil(t, a.AddUser("ben", "mypass", RoleUser, false))
+		require.Nil(t, a.AddReservation("ben", "topic1", PermissionDenyAll, 2))
+		require.Nil(t, a.AddReservation("ben", "topic2", PermissionDenyAll, 2))
+
+		// Both reservations grant ben full read/write; confirm the cache agrees.
+		for _, topic := range []string{"topic1", "topic2"} {
+			read, write, found, err := a.authorizeTopicAccess("ben", topic)
+			require.Nil(t, err)
+			require.True(t, found)
+			require.True(t, read)
+			require.True(t, write)
+		}
+
+		// Downgrade ben to a single reservation; one topic is removed from the DB.
+		removed, err := a.RemoveExcessReservations("ben", 1)
+		require.Nil(t, err)
+		require.Len(t, removed, 1)
+
+		// The removed reservation's grant must be gone from the cache, not just
+		// from the database.
+		read, write, found, err := a.authorizeTopicAccess("ben", removed[0])
+		require.Nil(t, err)
+		require.False(t, found, "stale ACL for removed reservation %q still served from cache", removed[0])
+		require.False(t, read)
+		require.False(t, write)
+
+		// The surviving reservation must still be served from the cache.
+		survivor := "topic1"
+		if removed[0] == "topic1" {
+			survivor = "topic2"
+		}
+		read, write, found, err = a.authorizeTopicAccess("ben", survivor)
+		require.Nil(t, err)
+		require.True(t, found)
+		require.True(t, read)
+		require.True(t, write)
+	})
+}
+
+// TestAccessCache_FullReloadDoesNotClobberConcurrentRevoke models finding #2:
+// a periodic full reload scans the whole user_access table outside the cache
+// lock. If a local ACL mutation revokes a grant and refreshes that user's slice
+// while the scan is in flight, applying the now-stale full snapshot must not
+// resurrect the revoked grant. The testHookReloadScanned seam injects the revoke
+// into exactly that race window.
+func TestAccessCache_FullReloadDoesNotClobberConcurrentRevoke(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, newManager newManagerFunc) {
+		a := newTestManagerFromConfig(t, newManager, &Config{
+			DefaultAccess:             PermissionDenyAll,
+			BcryptCost:                bcrypt.MinCost,
+			AccessCacheEnabled:        true,
+			AccessCacheReloadInterval: time.Hour, // keep the background poller out of this test
+		})
+		require.Nil(t, a.AddUser("phil", "mypass", RoleUser, false))
+		require.Nil(t, a.AllowAccess("phil", "secret", PermissionReadWrite))
+
+		// Sanity: the grant is served from the cache.
+		_, _, found, err := a.authorizeTopicAccess("phil", "secret")
+		require.Nil(t, err)
+		require.True(t, found)
+
+		// Arm the seam: when the full reload below finishes scanning (and still
+		// sees the grant), revoke it via a per-user reload before the full reload
+		// applies its now-stale snapshot. The re-entrant per-user reload that
+		// ResetAccess triggers is a no-op here (fired guard), and the whole thing
+		// runs single-threaded in this goroutine.
+		fired := false
+		testHookReloadScanned = func() {
+			if fired {
+				return
+			}
+			fired = true
+			require.Nil(t, a.ResetAccess("phil", "secret"))
+		}
+		defer func() { testHookReloadScanned = nil }()
+
+		// Trigger the full reload. Without the seq guard it would swap in its
+		// stale snapshot and resurrect the grant.
+		require.Nil(t, a.maybeReloadAccessCache())
+
+		_, _, found, err = a.authorizeTopicAccess("phil", "secret")
+		require.Nil(t, err)
+		require.False(t, found, "stale full reload resurrected a revoked grant")
+	})
+}
+
 func TestStoreReservations(t *testing.T) {
 	forEachStoreBackend(t, func(t *testing.T, manager *Manager) {
 		require.Nil(t, manager.AddUser("phil", "mypass", RoleUser, false))