Per-user reload

This commit is contained in:
binwiederhier
2026-05-31 14:11:55 -04:00
parent 03d405ed80
commit 301be79f0a
5 changed files with 156 additions and 48 deletions
+78 -9
View File
@@ -64,17 +64,15 @@ func (c *accessCache) reload(d *db.DB, query string) error {
patterns := make(map[string][]aclEntry) patterns := make(map[string][]aclEntry)
for rows.Next() { for rows.Next() {
var username, topic string var username, topic string
var entry aclEntry var read, write bool
if err := rows.Scan(&username, &topic, &entry.read, &entry.write); err != nil { if err := rows.Scan(&username, &topic, &read, &write); err != nil {
return err return err
} }
entry.length = len(topic) entry, isWildcard, err := newACLEntry(topic, read, write)
if strings.Contains(topic, "%") { if err != nil {
re, err := compileLikeToRegex(topic) return err
if err != nil { }
return err if isWildcard {
}
entry.pattern = re
patterns[username] = append(patterns[username], entry) patterns[username] = append(patterns[username], entry)
} else { } else {
if exacts[username] == nil { if exacts[username] == nil {
@@ -93,6 +91,77 @@ func (c *accessCache) reload(d *db.DB, query string) error {
return nil return nil
} }
// reloadUser refreshes just one user's rules from the database and swaps them
// into the cache. Used as a cheaper, owner-cascade-safe alternative to the
// full bulk reload after a mutation that affects a known set of users. The
// caller is expected to invoke reloadUser for every user whose row set may
// have changed -- typically the targeted user plus Everyone, since most
// reservation flows touch both.
//
// The query must return (topic, read, write) for every row whose user_id
// matches the given username. An empty result set is treated as "this user
// has no rules": the user's entries are removed from both maps so the inner
// maps don't grow unbounded under churn.
func (c *accessCache) reloadUser(d *db.DB, query, username string) error {
rows, err := d.Query(query, username)
if err != nil {
return err
}
defer rows.Close()
exact := make(map[string]aclEntry)
var pattern []aclEntry
for rows.Next() {
var topic string
var read, write bool
if err := rows.Scan(&topic, &read, &write); err != nil {
return err
}
entry, isWildcard, err := newACLEntry(topic, read, write)
if err != nil {
return err
}
if isWildcard {
pattern = append(pattern, entry)
} else {
exact[topic] = entry
}
}
if err := rows.Err(); err != nil {
return err
}
c.mu.Lock()
if len(exact) == 0 {
delete(c.exact, username)
} else {
c.exact[username] = exact
}
if len(pattern) == 0 {
delete(c.pattern, username)
} else {
c.pattern[username] = pattern
}
c.mu.Unlock()
return nil
}
// newACLEntry builds an aclEntry from one user_access row's values. The
// isWildcard return tells the caller which storage slot the entry belongs in:
// the per-user wildcard slice if true, the per-user exact map if false.
// Wildcards have their LIKE pattern pre-compiled into entry.pattern; exact
// entries leave entry.pattern nil.
func newACLEntry(topic string, read, write bool) (entry aclEntry, isWildcard bool, err error) {
entry = aclEntry{length: len(topic), read: read, write: write}
if !strings.Contains(topic, "%") {
return entry, false, nil
}
re, err := compileLikeToRegex(topic)
if err != nil {
return entry, true, err
}
entry.pattern = re
return entry, true, nil
}
// Lookup returns the effective (read, write, found) permission for the given // Lookup returns the effective (read, write, found) permission for the given
// (username, topic), preserving the priority ordering of the original SQL query: // (username, topic), preserving the priority ordering of the original SQL query:
// 1. specific user beats Everyone // 1. specific user beats Everyone
+63 -39
View File
@@ -40,9 +40,8 @@ const (
DefaultUserPasswordBcryptCost = 10 DefaultUserPasswordBcryptCost = 10
// DefaultAccessCacheReloadInterval bounds how stale the in-memory ACL snapshot // DefaultAccessCacheReloadInterval bounds how stale the in-memory ACL snapshot
// can be relative to writes made by *other* processes (e.g. a separate `ntfy // can be relative to writes made by *other* processes (e.g. a separate `ntfy
// access` CLI invocation modifying the same database). Mutations performed // access` CLI invocation modifying the same database)
// by this Manager refresh the cache synchronously and do not depend on this. DefaultAccessCacheReloadInterval = 60 * time.Second
DefaultAccessCacheReloadInterval = 5 * time.Second
) )
var ( var (
@@ -87,8 +86,6 @@ func newManager(d *db.DB, queries queries, config *Config) (*Manager, error) {
if err := manager.maybeProvisionUsersAccessAndTokens(); err != nil { if err := manager.maybeProvisionUsersAccessAndTokens(); err != nil {
return nil, err return nil, err
} }
// Populate the ACL cache after provisioning so the initial snapshot includes
// any provisioned access rules. Subsequent mutations call reloadAccessCache.
if err := manager.reloadAccessCache(); err != nil { if err := manager.reloadAccessCache(); err != nil {
return nil, err return nil, err
} }
@@ -99,18 +96,22 @@ func newManager(d *db.DB, queries queries, config *Config) (*Manager, error) {
return manager, nil return manager, nil
} }
// reloadAccessCache rebuilds the in-memory ACL snapshot from the primary // reloadAccessCache rebuilds the in-memory access cache from the primary database
// database. Called once at startup (after provisioning) and after every method
// that mutates user_access (directly or by cascade from "user" deletion).
func (a *Manager) reloadAccessCache() error { func (a *Manager) reloadAccessCache() error {
return a.accessCache.reload(a.db, a.queries.selectAllAccessForCache) return a.accessCache.reload(a.db, a.queries.selectAllAccessForCache)
} }
// asyncAccessCacheReloader periodically refreshes the ACL snapshot so that // reloadAccessCacheUsers refreshes the cache slices for the given usernames
// writes made by other processes against the same database (most notably the func (a *Manager) reloadAccessCacheUsers(usernames ...string) error {
// `ntfy access` CLI subcommand running while a server holds the cache) become for _, username := range usernames {
// visible within the configured interval. This Manager's own writes do not if err := a.accessCache.reloadUser(a.db, a.queries.selectAccessForCacheByUser, username); err != nil {
// depend on the poller -- they refresh the cache synchronously. return err
}
}
return nil
}
// asyncAccessCacheReloader periodically refreshes the access cache
func (a *Manager) asyncAccessCacheReloader(interval time.Duration) { func (a *Manager) asyncAccessCacheReloader(interval time.Duration) {
ticker := time.NewTicker(interval) ticker := time.NewTicker(interval)
defer ticker.Stop() defer ticker.Stop()
@@ -198,13 +199,16 @@ func (a *Manager) RemoveUser(username string) error {
if err := a.CanChangeUser(username); err != nil { if err := a.CanChangeUser(username); err != nil {
return err return err
} }
if err := db.ExecTx(a.db, func(tx *sql.Tx) error { err := db.ExecTx(a.db, func(tx *sql.Tx) error {
return a.removeUserTx(tx, username) return a.removeUserTx(tx, username)
}); err != nil { })
if err != nil {
return err return err
} }
// user_access rows are cascade-deleted along with the user; refresh the snapshot. // user_access rows are cascade-deleted along with the user (both by user_id
return a.reloadAccessCache() // and by owner_user_id). Refresh this user's own slice (now empty) and
// Everyone's slice, since reservations owned by this user landed there too.
return a.reloadAccessCacheUsers(username, Everyone)
} }
// removeUserTx deletes the user with the given username // removeUserTx deletes the user with the given username
@@ -225,7 +229,7 @@ func (a *Manager) MarkUserRemoved(user *User) error {
if !AllowedUsername(user.Name) { if !AllowedUsername(user.Name) {
return ErrInvalidArgument return ErrInvalidArgument
} }
if err := db.ExecTx(a.db, func(tx *sql.Tx) error { err := db.ExecTx(a.db, func(tx *sql.Tx) error {
if err := a.resetUserAccessTx(tx, user.Name); err != nil { if err := a.resetUserAccessTx(tx, user.Name); err != nil {
return err return err
} }
@@ -237,11 +241,14 @@ func (a *Manager) MarkUserRemoved(user *User) error {
return err return err
} }
return nil return nil
}); err != nil { })
if err != nil {
return err return err
} }
// resetUserAccessTx wiped this user's user_access rows; refresh the snapshot. // resetUserAccessTx deleted this user's rows AND any row owned by this user
return a.reloadAccessCache() // (typically the matching Everyone rows from their reservations). Refresh
// both slices to mirror the DB exactly.
return a.reloadAccessCacheUsers(user.Name, Everyone)
} }
// RemoveDeletedUsers deletes all users that have been marked deleted // RemoveDeletedUsers deletes all users that have been marked deleted
@@ -281,15 +288,17 @@ func (a *Manager) ChangeRole(username string, role Role) error {
if err := a.CanChangeUser(username); err != nil { if err := a.CanChangeUser(username); err != nil {
return err return err
} }
if err := db.ExecTx(a.db, func(tx *sql.Tx) error { err := db.ExecTx(a.db, func(tx *sql.Tx) error {
return a.changeRoleTx(tx, username, role) return a.changeRoleTx(tx, username, role)
}); err != nil { })
if err != nil {
return err return err
} }
// Promotion to admin clears user_access rows for the user; refresh the snapshot. // Promotion to admin clears user_access rows for this user AND rows owned
// Other role changes are no-ops for the cache but reloading is cheap and keeps // by this user (Everyone rows from their reservations). Other role changes
// the code path uniform. // are no-ops for the cache but reloading the two affected slices is cheap
return a.reloadAccessCache() // and keeps the code path uniform.
return a.reloadAccessCacheUsers(username, Everyone)
} }
// changeRoleTx changes a user's role // changeRoleTx changes a user's role
@@ -650,12 +659,14 @@ func (a *Manager) resolvePerms(base, perm Permission) error {
// read/write access to a topic. The parameter topicPattern may include wildcards (*). The ACL entry // read/write access to a topic. The parameter topicPattern may include wildcards (*). The ACL entry
// owner may either be a user (username), or the system (empty). // owner may either be a user (username), or the system (empty).
func (a *Manager) AllowAccess(username string, topicPattern string, permission Permission) error { func (a *Manager) AllowAccess(username string, topicPattern string, permission Permission) error {
if err := db.ExecTx(a.db, func(tx *sql.Tx) error { err := db.ExecTx(a.db, func(tx *sql.Tx) error {
return a.allowAccessTx(tx, username, topicPattern, permission, false) return a.allowAccessTx(tx, username, topicPattern, permission, false)
}); err != nil { })
if err != nil {
return err return err
} }
return a.reloadAccessCache() // Only this user's row set changed; refresh their slice only.
return a.reloadAccessCacheUsers(username)
} }
func (a *Manager) allowAccessTx(tx *sql.Tx, username string, topicPattern string, permission Permission, provisioned bool) error { func (a *Manager) allowAccessTx(tx *sql.Tx, username string, topicPattern string, permission Permission, provisioned bool) error {
@@ -671,12 +682,20 @@ func (a *Manager) allowAccessTx(tx *sql.Tx, username string, topicPattern string
// ResetAccess removes an access control list entry for a specific username/topic, or (if topic is // ResetAccess removes an access control list entry for a specific username/topic, or (if topic is
// empty) for an entire user. The parameter topicPattern may include wildcards (*). // empty) for an entire user. The parameter topicPattern may include wildcards (*).
func (a *Manager) ResetAccess(username string, topicPattern string) error { func (a *Manager) ResetAccess(username string, topicPattern string) error {
if err := db.ExecTx(a.db, func(tx *sql.Tx) error { err := db.ExecTx(a.db, func(tx *sql.Tx) error {
return a.resetAccessTx(tx, username, topicPattern) return a.resetAccessTx(tx, username, topicPattern)
}); err != nil { })
if err != nil {
return err return err
} }
return a.reloadAccessCache() // "Delete all access" affects every user; do the bulk reload.
// Otherwise refresh the named user plus Everyone, since resetUserAccessTx
// and deleteTopicAccess both touch rows owned by the user (typically
// Everyone rows from their reservations).
if username == "" {
return a.reloadAccessCache()
}
return a.reloadAccessCacheUsers(username, Everyone)
} }
func (a *Manager) resetAccessTx(tx *sql.Tx, username string, topicPattern string) error { func (a *Manager) resetAccessTx(tx *sql.Tx, username string, topicPattern string) error {
@@ -790,7 +809,7 @@ func (a *Manager) AddReservation(username string, topic string, everyone Permiss
if !AllowedUsername(username) || username == Everyone || !AllowedTopic(topic) { if !AllowedUsername(username) || username == Everyone || !AllowedTopic(topic) {
return ErrInvalidArgument return ErrInvalidArgument
} }
if err := db.ExecTx(a.db, func(tx *sql.Tx) error { err := db.ExecTx(a.db, func(tx *sql.Tx) error {
if limit > 0 { if limit > 0 {
hasReservation, err := a.hasReservationTx(tx, username, topic) hasReservation, err := a.hasReservationTx(tx, username, topic)
if err != nil { if err != nil {
@@ -813,10 +832,12 @@ func (a *Manager) AddReservation(username string, topic string, everyone Permiss
return err return err
} }
return nil return nil
}); err != nil { })
if err != nil {
return err return err
} }
return a.reloadAccessCache() // Both user's and Everyone's rows changed.
return a.reloadAccessCacheUsers(username, Everyone)
} }
// RemoveReservations deletes the access control entries associated with the given username/topic, // RemoveReservations deletes the access control entries associated with the given username/topic,
@@ -831,17 +852,20 @@ func (a *Manager) RemoveReservations(username string, topics ...string) error {
return ErrInvalidArgument return ErrInvalidArgument
} }
} }
if err := db.ExecTx(a.db, func(tx *sql.Tx) error { err := db.ExecTx(a.db, func(tx *sql.Tx) error {
for _, topic := range topics { for _, topic := range topics {
if err := a.removeReservationAccessTx(tx, username, topic); err != nil { if err := a.removeReservationAccessTx(tx, username, topic); err != nil {
return err return err
} }
} }
return nil return nil
}); err != nil { })
if err != nil {
return err return err
} }
return a.reloadAccessCache() // Mirror the DB: rows for this user and any Everyone rows owned by this
// user are gone. Refresh both slices.
return a.reloadAccessCacheUsers(username, Everyone)
} }
// Reservations returns all user-owned topics, and the associated everyone-access // Reservations returns all user-owned topics, and the associated everyone-access
+7
View File
@@ -75,6 +75,12 @@ const (
FROM user_access a FROM user_access a
JOIN "user" u ON u.id = a.user_id JOIN "user" u ON u.id = a.user_id
` `
postgresSelectAccessForCacheByUserQuery = `
SELECT a.topic, a.read, a.write
FROM user_access a
JOIN "user" u ON u.id = a.user_id
WHERE u.user_name = $1
`
postgresSelectUserAllAccessQuery = ` postgresSelectUserAllAccessQuery = `
SELECT user_id, topic, read, write, provisioned SELECT user_id, topic, read, write, provisioned
FROM user_access FROM user_access
@@ -243,6 +249,7 @@ var postgresQueries = queries{
deleteUsersMarked: postgresDeleteUsersMarkedQuery, deleteUsersMarked: postgresDeleteUsersMarkedQuery,
deleteUsersProvisioned: postgresDeleteUsersProvisionedQuery, deleteUsersProvisioned: postgresDeleteUsersProvisionedQuery,
selectAllAccessForCache: postgresSelectAllAccessForCacheQuery, selectAllAccessForCache: postgresSelectAllAccessForCacheQuery,
selectAccessForCacheByUser: postgresSelectAccessForCacheByUserQuery,
selectUserAllAccess: postgresSelectUserAllAccessQuery, selectUserAllAccess: postgresSelectUserAllAccessQuery,
selectUserAccess: postgresSelectUserAccessQuery, selectUserAccess: postgresSelectUserAccessQuery,
selectUserReservations: postgresSelectUserReservationsQuery, selectUserReservations: postgresSelectUserReservationsQuery,
+7
View File
@@ -81,6 +81,12 @@ const (
FROM user_access a FROM user_access a
JOIN user u ON u.id = a.user_id JOIN user u ON u.id = a.user_id
` `
sqliteSelectAccessForCacheByUserQuery = `
SELECT a.topic, a.read, a.write
FROM user_access a
JOIN user u ON u.id = a.user_id
WHERE u.user = ?
`
sqliteSelectUserAllAccessQuery = ` sqliteSelectUserAllAccessQuery = `
SELECT user_id, topic, read, write, provisioned SELECT user_id, topic, read, write, provisioned
FROM user_access FROM user_access
@@ -241,6 +247,7 @@ var sqliteQueries = queries{
deleteUsersMarked: sqliteDeleteUsersMarkedQuery, deleteUsersMarked: sqliteDeleteUsersMarkedQuery,
deleteUsersProvisioned: sqliteDeleteUsersProvisionedQuery, deleteUsersProvisioned: sqliteDeleteUsersProvisionedQuery,
selectAllAccessForCache: sqliteSelectAllAccessForCacheQuery, selectAllAccessForCache: sqliteSelectAllAccessForCacheQuery,
selectAccessForCacheByUser: sqliteSelectAccessForCacheByUserQuery,
selectUserAllAccess: sqliteSelectUserAllAccessQuery, selectUserAllAccess: sqliteSelectUserAllAccessQuery,
selectUserAccess: sqliteSelectUserAccessQuery, selectUserAccess: sqliteSelectUserAccessQuery,
selectUserReservations: sqliteSelectUserReservationsQuery, selectUserReservations: sqliteSelectUserReservationsQuery,
+1
View File
@@ -314,6 +314,7 @@ type queries struct {
// Access queries // Access queries
selectAllAccessForCache string // Bulk load: (user_name, topic, read, write) for the in-memory ACL cache selectAllAccessForCache string // Bulk load: (user_name, topic, read, write) for the in-memory ACL cache
selectAccessForCacheByUser string // Per-user load: (topic, read, write) for one username; used to refresh just one user's slice of the cache after mutation
selectUserAllAccess string selectUserAllAccess string
selectUserAccess string selectUserAccess string
selectUserReservations string selectUserReservations string