perf(admin): cache the suspension gate lookup
CI / changes (pull_request) Successful in 2s
CI / unit (pull_request) Successful in 9s
CI / integration (pull_request) Successful in 15s
CI / ui (pull_request) Successful in 46s
CI / gate (pull_request) Successful in 0s
CI / deploy (pull_request) Successful in 1m4s
CI / changes (pull_request) Successful in 2s
CI / unit (pull_request) Successful in 9s
CI / integration (pull_request) Successful in 15s
CI / ui (pull_request) Successful in 46s
CI / gate (pull_request) Successful in 0s
CI / deploy (pull_request) Successful in 1m4s
The suspension gate runs CurrentSuspension on every authenticated request. Add a write-through in-memory cache on account.Store keyed by account id, invalidated on Suspend/LiftSuspension, with the cached entry re-evaluated against the wall clock so a temporary block lapses without an explicit invalidation. Single-instance, matching the deployment (one shared Store). Keeps the gate off the database on the hot path while a block still takes effect on the next request.
This commit is contained in:
@@ -99,12 +99,15 @@ type Identity struct {
|
||||
type Store struct {
|
||||
db *sql.DB
|
||||
metrics *accountMetrics
|
||||
// suspensions caches each account's current manual block, read by the suspension gate on
|
||||
// every authenticated request and invalidated on Suspend/LiftSuspension. See suspension.go.
|
||||
suspensions *suspensionCache
|
||||
}
|
||||
|
||||
// NewStore constructs a Store wrapping db. Metrics default to a no-op meter until
|
||||
// SetMetrics installs the real one during startup wiring.
|
||||
func NewStore(db *sql.DB) *Store {
|
||||
return &Store{db: db, metrics: defaultAccountMetrics()}
|
||||
return &Store{db: db, metrics: defaultAccountMetrics(), suspensions: newSuspensionCache()}
|
||||
}
|
||||
|
||||
// ProvisionByIdentity returns the account bound to (kind, externalID), creating
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-jet/jet/v2/postgres"
|
||||
@@ -82,6 +83,7 @@ func (s *Store) Suspend(ctx context.Context, accountID uuid.UUID, until *time.Ti
|
||||
if err := stmt.QueryContext(ctx, s.db, &row); err != nil {
|
||||
return Suspension{}, fmt.Errorf("account: suspend %s: %w", accountID, err)
|
||||
}
|
||||
s.invalidateSuspension(accountID)
|
||||
return modelToSuspension(row), nil
|
||||
}
|
||||
|
||||
@@ -100,6 +102,7 @@ func (s *Store) LiftSuspension(ctx context.Context, accountID uuid.UUID) error {
|
||||
if _, err := stmt.ExecContext(ctx, s.db); err != nil {
|
||||
return fmt.Errorf("account: lift suspension %s: %w", accountID, err)
|
||||
}
|
||||
s.invalidateSuspension(accountID)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -115,6 +118,30 @@ func (s *Store) CurrentSuspension(ctx context.Context, accountID uuid.UUID) (Sus
|
||||
return Suspension{}, false, nil
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
if s.suspensions != nil {
|
||||
if e, ok := s.suspensions.get(accountID); ok {
|
||||
if !e.found {
|
||||
return Suspension{}, false, nil // cached: not blocked
|
||||
}
|
||||
if suspensionActiveAt(e.susp, now) {
|
||||
return e.susp, true, nil // cached block still in force
|
||||
}
|
||||
// The cached block has lapsed since it was cached; refresh from the database below.
|
||||
}
|
||||
}
|
||||
susp, found, err := s.queryCurrentSuspension(ctx, accountID, now)
|
||||
if err != nil {
|
||||
return Suspension{}, false, err
|
||||
}
|
||||
if s.suspensions != nil {
|
||||
s.suspensions.put(accountID, suspensionCacheEntry{susp: susp, found: found})
|
||||
}
|
||||
return susp, found, nil
|
||||
}
|
||||
|
||||
// queryCurrentSuspension reads the account's strongest in-force block straight from the database
|
||||
// (no cache), as of now. It backs CurrentSuspension on a cache miss or a lapsed entry.
|
||||
func (s *Store) queryCurrentSuspension(ctx context.Context, accountID uuid.UUID, now time.Time) (Suspension, bool, error) {
|
||||
stmt := postgres.SELECT(table.AccountSuspensions.AllColumns).
|
||||
FROM(table.AccountSuspensions).
|
||||
WHERE(
|
||||
@@ -134,6 +161,61 @@ func (s *Store) CurrentSuspension(ctx context.Context, accountID uuid.UUID) (Sus
|
||||
return modelToSuspension(row), true, nil
|
||||
}
|
||||
|
||||
// invalidateSuspension drops the account's cached block so the next CurrentSuspension re-reads it.
|
||||
// Called after Suspend and LiftSuspension.
|
||||
func (s *Store) invalidateSuspension(accountID uuid.UUID) {
|
||||
if s.suspensions != nil {
|
||||
s.suspensions.invalidate(accountID)
|
||||
}
|
||||
}
|
||||
|
||||
// suspensionActiveAt reports whether a suspension is in force at now: permanent, or not yet
|
||||
// expired. The lifted check is implicit — only non-lifted blocks are ever cached or returned.
|
||||
func suspensionActiveAt(susp Suspension, now time.Time) bool {
|
||||
return susp.BlockedUntil == nil || susp.BlockedUntil.After(now)
|
||||
}
|
||||
|
||||
// suspensionCache is the gate's write-through cache of each account's current block, keyed by
|
||||
// account id. The suspension gate reads it on every authenticated request; Suspend and
|
||||
// LiftSuspension invalidate the account's entry. An entry holds the strongest active block at
|
||||
// query time (or a not-blocked marker), re-evaluated against the wall clock on read, so a
|
||||
// temporary block lapses without an explicit invalidation. It is single-instance, matching the
|
||||
// deployment (one shared Store); a multi-instance deployment would need a shared cache.
|
||||
type suspensionCache struct {
|
||||
mu sync.RWMutex
|
||||
m map[uuid.UUID]suspensionCacheEntry
|
||||
}
|
||||
|
||||
// suspensionCacheEntry is a cached lookup: the strongest active block when found, else a
|
||||
// not-blocked marker (found=false).
|
||||
type suspensionCacheEntry struct {
|
||||
susp Suspension
|
||||
found bool
|
||||
}
|
||||
|
||||
func newSuspensionCache() *suspensionCache {
|
||||
return &suspensionCache{m: make(map[uuid.UUID]suspensionCacheEntry)}
|
||||
}
|
||||
|
||||
func (c *suspensionCache) get(id uuid.UUID) (suspensionCacheEntry, bool) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
e, ok := c.m[id]
|
||||
return e, ok
|
||||
}
|
||||
|
||||
func (c *suspensionCache) put(id uuid.UUID, e suspensionCacheEntry) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.m[id] = e
|
||||
}
|
||||
|
||||
func (c *suspensionCache) invalidate(id uuid.UUID) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
delete(c.m, id)
|
||||
}
|
||||
|
||||
// activeSuspensionPredicate matches the rows of a block that is in force at now: not lifted and
|
||||
// either permanent or not yet expired.
|
||||
func activeSuspensionPredicate(now time.Time) postgres.BoolExpression {
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
package account
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// TestSuspensionActiveAt covers the wall-clock re-evaluation the cache relies on: a permanent
|
||||
// block is always active, a future-dated one is active, a past-dated one is not.
|
||||
func TestSuspensionActiveAt(t *testing.T) {
|
||||
now := time.Date(2026, 6, 14, 12, 0, 0, 0, time.UTC)
|
||||
future := now.Add(time.Hour)
|
||||
past := now.Add(-time.Hour)
|
||||
|
||||
if !suspensionActiveAt(Suspension{BlockedUntil: nil}, now) {
|
||||
t.Error("a permanent block must be active")
|
||||
}
|
||||
if !suspensionActiveAt(Suspension{BlockedUntil: &future}, now) {
|
||||
t.Error("a future-dated block must be active")
|
||||
}
|
||||
if suspensionActiveAt(Suspension{BlockedUntil: &past}, now) {
|
||||
t.Error("a past-dated block must be inactive")
|
||||
}
|
||||
}
|
||||
|
||||
// TestSuspensionCache covers the cache primitives: a miss on an empty cache, a hit after put for
|
||||
// both the not-blocked and blocked markers, and a miss after invalidate.
|
||||
func TestSuspensionCache(t *testing.T) {
|
||||
c := newSuspensionCache()
|
||||
id := uuid.New()
|
||||
|
||||
if _, ok := c.get(id); ok {
|
||||
t.Fatal("empty cache must miss")
|
||||
}
|
||||
|
||||
c.put(id, suspensionCacheEntry{found: false})
|
||||
if e, ok := c.get(id); !ok || e.found {
|
||||
t.Fatalf("cached not-blocked entry = (%+v, ok %v), want hit with found=false", e, ok)
|
||||
}
|
||||
|
||||
c.put(id, suspensionCacheEntry{susp: Suspension{AccountID: id}, found: true})
|
||||
if e, ok := c.get(id); !ok || !e.found {
|
||||
t.Fatalf("cached blocked entry = (%+v, ok %v), want hit with found=true", e, ok)
|
||||
}
|
||||
|
||||
c.invalidate(id)
|
||||
if _, ok := c.get(id); ok {
|
||||
t.Fatal("invalidated entry must miss")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user