From b4cb1bb7fbbc289e21e338185c37fb08cf0626ef Mon Sep 17 00:00:00 2001 From: binwiederhier Date: Wed, 3 Jun 2026 21:01:52 -0400 Subject: [PATCH] Fix race and a small issue with the access cache --- user/access_cache.go | 33 +++++++++++--- user/manager.go | 13 +++++- user/manager_test.go | 103 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 141 insertions(+), 8 deletions(-) diff --git a/user/access_cache.go b/user/access_cache.go index 6788f5ec..0218b1b6 100644 --- a/user/access_cache.go +++ b/user/access_cache.go @@ -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 } diff --git a/user/manager.go b/user/manager.go index b036df73..34d96403 100644 --- a/user/manager.go +++ b/user/manager.go @@ -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 diff --git a/user/manager_test.go b/user/manager_test.go index 5d6944f2..a3fd07b1 100644 --- a/user/manager_test.go +++ b/user/manager_test.go @@ -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))